This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new c816d5f8d81 [To dev/1.3] Preserve LOAD sources after unrecoverable
query fallback (#19090)
c816d5f8d81 is described below
commit c816d5f8d8180386f643840471b8394076a1e6c5
Author: Caideyipi <[email protected]>
AuthorDate: Tue Nov 3 09:30:42 2026 +0800
[To dev/1.3] Preserve LOAD sources after unrecoverable query fallback
(#19090)
* Preserve LOAD sources after unrecoverable query fallback (#19073)
* Fix LOAD fallback error reporting and scan-boundary resumption
* Fix LOAD fallback checkpoints across chunk groups
* Fix LOAD fallback metadata error propagation
* Fix LOAD fallback metadata batches
(cherry picked from commit 9463d011e857ca060c8d1c799f803a54fba9c115)
* Adapt LOAD fallback tests to 1.3 scheduling
* Check all DataNode logs in LOAD fallback IT
* Exercise LOAD query fallback in single-node tests
---
.../org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 279 +++++++++++++++++
.../converter/LoadTreeTsFileTabletIterator.java | 230 +++++---------
...atementDataTypeConvertExecutionVisitorTest.java | 338 ++++++++++++++++++++-
3 files changed, 693 insertions(+), 154 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java
index c2fbbb0e30b..61fe99e9ea1 100644
---
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java
@@ -20,11 +20,13 @@
package org.apache.iotdb.db.it;
import org.apache.iotdb.commons.auth.entity.PrivilegeType;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.db.it.utils.TestUtils;
import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion;
import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
@@ -32,11 +34,20 @@ import org.apache.iotdb.it.utils.TsFileGenerator;
import org.apache.iotdb.itbase.category.ClusterIT;
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
import org.apache.iotdb.jdbc.IoTDBSQLException;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.commons.io.FileUtils;
+import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.PlainDeviceID;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.TsFileSequenceReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.write.TsFileWriter;
+import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.After;
import org.junit.Assert;
@@ -49,6 +60,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
import java.nio.file.Files;
import java.sql.Connection;
import java.sql.ResultSet;
@@ -62,6 +75,10 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -782,6 +799,268 @@ public class IoTDBLoadTsFileIT {
}
}
+ @Test
+ public void testDirectoryQueryFallbackRetainsOnlyFailedSources() throws
Exception {
+ for (final boolean delete : new boolean[] {false, true}) {
+ final File directory = new File(tmpDir, delete ? "batch-delete" :
"batch-none");
+ Assert.assertTrue(directory.mkdir());
+ final String devicePrefix =
+ SchemaConfig.STORAGE_GROUP_0 + (delete ? ".batch_delete" :
".batch_none");
+ final File normal = new File(directory, "401-0-0-0.tsfile");
+ final File recoverable = new File(directory, "402-0-0-0.tsfile");
+ final File failed = new File(directory, "403-0-0-0.tsfile");
+ final File[] sources = {normal, recoverable, failed};
+ try {
+ long recoverableMarkerOffset = -1;
+ long failedMarkerOffset = -1;
+ for (int i = 0; i < sources.length; i++) {
+ final long groupMarkerOffset =
+ writeFallbackSource(sources[i], devicePrefix + ".d" + i, i == 1,
i > 0);
+ if (i == 1) {
+ recoverableMarkerOffset = groupMarkerOffset;
+ } else if (i == 2) {
+ failedMarkerOffset = groupMarkerOffset;
+ }
+ }
+ // All footer metadata stays valid, so LOAD must reach scheduling and
query fallback.
+ replaceByte(recoverable, recoverableMarkerOffset, 0x7f);
+ final long compressionOffset = firstChunkCompressionOffset(failed);
+ final int originalCompression = replaceByte(failed, compressionOffset,
0x7f);
+ final int originalFailedMarker = replaceByte(failed,
failedMarkerOffset, 0x7f);
+
+ try (final Connection connection =
+ EnvFactory.getEnv()
+
.getConnectionWithSpecifiedDataNode(EnvFactory.getEnv().getDataNodeWrapper(0));
+ final Statement statement = connection.createStatement()) {
+ final SQLException failure =
+ Assert.assertThrows(
+ SQLException.class, () ->
statement.execute(fallbackLoadSql(directory, delete)));
+ assertFallbackLoadFailure(failure, failed);
+ Assert.assertFalse(failure.getMessage(),
failure.getMessage().contains(normal.getName()));
+ Assert.assertFalse(
+ failure.getMessage(),
failure.getMessage().contains(recoverable.getName()));
+ assertFallbackRows(statement, devicePrefix + ".d0");
+ assertFallbackRows(statement, devicePrefix + ".d1",
fallbackPartitionTimestamps());
+ boolean foundConversionLog = false;
+ for (final DataNodeWrapper dataNodeWrapper :
+ EnvFactory.getEnv().getDataNodeWrapperList()) {
+ if (dataNodeWrapper.logContains(
+ "Load: Successfully converted TsFile "
+ + recoverable.getAbsolutePath()
+ + " into tablets and inserted.")) {
+ foundConversionLog = true;
+ break;
+ }
+ }
+ Assert.assertTrue(foundConversionLog);
+ assertFallbackSourceExists(normal, !delete);
+ assertFallbackSourceExists(recoverable, !delete);
+ assertFallbackSourceExists(failed, true);
+
+ replaceByte(failed, compressionOffset, originalCompression);
+ replaceByte(failed, failedMarkerOffset, originalFailedMarker);
+ statement.execute(fallbackLoadSql(failed, delete));
+ assertFallbackRows(statement, devicePrefix + ".d2",
fallbackPartitionTimestamps());
+ assertFallbackRows(statement, devicePrefix + ".d0");
+ assertFallbackRows(statement, devicePrefix + ".d1",
fallbackPartitionTimestamps());
+ assertFallbackSourceExists(failed, !delete);
+ }
+ } finally {
+ FileUtils.deleteDirectory(directory);
+ }
+ }
+ }
+
+ @Test
+ public void testConcurrentQueryFallbackFailureRetainsSourceForRetry() throws
Exception {
+ final File good = new File(tmpDir, "301-0-0-0.tsfile");
+ final File bad = new File(tmpDir, "302-0-0-0.tsfile");
+ final String device = SchemaConfig.STORAGE_GROUP_0 + ".concurrent";
+ writeFallbackSource(good, device + ".good", false, false);
+ writeFallbackSource(bad, device + ".bad", false, true);
+ final long compressionOffset = firstChunkCompressionOffset(bad);
+ final int originalCompression = replaceByte(bad, compressionOffset, 0x7f);
+ final ExecutorService loaders = Executors.newFixedThreadPool(2);
+ final CountDownLatch ready = new CountDownLatch(2);
+ final CountDownLatch start = new CountDownLatch(1);
+ try {
+ final List<Future<?>> loads = new ArrayList<>();
+ for (final File source : Arrays.asList(good, bad)) {
+ loads.add(
+ loaders.submit(
+ () -> {
+ try (final Connection connection =
EnvFactory.getEnv().getConnection();
+ final Statement statement =
connection.createStatement()) {
+ ready.countDown();
+ Assert.assertTrue(start.await(30, TimeUnit.SECONDS));
+ if (source.equals(bad)) {
+ final SQLException failure =
+ Assert.assertThrows(
+ SQLException.class,
+ () -> statement.execute(fallbackLoadSql(source,
true)));
+ assertFallbackLoadFailure(failure, bad);
+ } else {
+ statement.execute(fallbackLoadSql(source, true));
+ }
+ }
+ return null;
+ }));
+ }
+ Assert.assertTrue(ready.await(30, TimeUnit.SECONDS));
+ start.countDown();
+ for (final Future<?> load : loads) {
+ load.get(90, TimeUnit.SECONDS);
+ }
+ assertFallbackSourceExists(good, false);
+ assertFallbackSourceExists(bad, true);
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement()) {
+ assertFallbackRows(statement, device + ".good");
+ replaceByte(bad, compressionOffset, originalCompression);
+ statement.execute(fallbackLoadSql(bad, true));
+ assertFallbackRows(statement, device + ".bad",
fallbackPartitionTimestamps());
+ assertFallbackRows(statement, device + ".good");
+ }
+ assertFallbackSourceExists(bad, false);
+ } finally {
+ start.countDown();
+ loaders.shutdownNow();
+ Assert.assertTrue(loaders.awaitTermination(30, TimeUnit.SECONDS));
+ }
+ }
+
+ private static String fallbackLoadSql(final File source, final boolean
delete) {
+ return String.format(
+ "load '%s' with ('database-level'='2','on-success'='%s',"
+ + "'tablet-conversion-threshold'='-1')",
+ source.getAbsolutePath(), delete ? "delete" : "none");
+ }
+
+ private static void assertFallbackLoadFailure(final SQLException failure,
final File source) {
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(),
failure.getErrorCode());
+ Assert.assertTrue(failure.getMessage(),
failure.getMessage().contains(source.getName()));
+ }
+
+ private static long writeFallbackSource(
+ final File file,
+ final String device,
+ final boolean includeAnchor,
+ final boolean spanTimePartitions)
+ throws Exception {
+ final List<MeasurementSchema> schemas =
+ Arrays.asList(
+ new MeasurementSchema(
+ "s1", TSDataType.INT64, TSEncoding.PLAIN,
CompressionType.UNCOMPRESSED),
+ new MeasurementSchema(
+ "s2", TSDataType.INT64, TSEncoding.PLAIN,
CompressionType.UNCOMPRESSED));
+ final String anchor = SchemaConfig.STORAGE_GROUP_0 + ".a_fallback_anchor";
+ final long groupMarkerOffset;
+ try (final TsFileWriter writer = new TsFileWriter(file)) {
+ if (includeAnchor) {
+ // Preserve a valid first group for the encryption-header check, and
keep physical device
+ // order consistent with the metadata index so query fallback can
resume after this group.
+ writer.registerAlignedTimeseries(new Path(anchor), schemas);
+ final Tablet anchorTablet = new Tablet(anchor, schemas);
+ anchorTablet.addTimestamp(0, 0);
+ anchorTablet.addValue("s1", 0, 0L);
+ anchorTablet.addValue("s2", 0, 0L);
+ anchorTablet.rowSize = 1;
+ writer.writeAligned(anchorTablet);
+ writer.flushAllChunkGroups();
+ }
+ groupMarkerOffset = writer.getIOWriter().getPos();
+ writer.registerAlignedTimeseries(new Path(device), schemas);
+ final Tablet tablet = new Tablet(device, schemas);
+ for (int i = 0; i < 4; i++) {
+ final long timestamp =
+ spanTimePartitions && i >= 2
+ ?
CommonDescriptor.getInstance().getConfig().getTimePartitionInterval() + i - 1
+ : i + 1;
+ tablet.addTimestamp(i, timestamp);
+ tablet.addValue("s1", i, (long) i);
+ tablet.addValue("s2", i, (long) i * 10);
+ }
+ tablet.rowSize = 4;
+ writer.writeAligned(tablet);
+ }
+ final TsFileResource resource = new TsFileResource(file);
+ final IDeviceID deviceId = new PlainDeviceID(device);
+ resource.updateStartTime(deviceId, 1);
+ resource.updateEndTime(
+ deviceId,
+ spanTimePartitions
+ ?
CommonDescriptor.getInstance().getConfig().getTimePartitionInterval() + 2
+ : 4);
+ if (includeAnchor) {
+ final IDeviceID anchorId = new PlainDeviceID(anchor);
+ resource.updateStartTime(anchorId, 0);
+ resource.updateEndTime(anchorId, 0);
+ }
+ resource.serialize();
+ try (final ModificationFile mods = resource.getModFile()) {
+ mods.write(new Deletion(new MeasurementPath(device + ".s1"),
Long.MAX_VALUE, 2, 2));
+ }
+ return groupMarkerOffset;
+ }
+
+ private static long firstChunkCompressionOffset(final File file) throws
Exception {
+ try (final TsFileSequenceReader reader = new
TsFileSequenceReader(file.getAbsolutePath())) {
+ reader.position(TSFileConfig.MAGIC_STRING.length() + 1L);
+ reader.readMarker();
+ reader.readChunkGroupHeader();
+ reader.readChunkHeader(reader.readMarker());
+ // The final two header bytes are compression and encoding; corrupt only
compression.
+ return reader.position() - 2;
+ }
+ }
+
+ private static int replaceByte(final File file, final long offset, final int
value)
+ throws IOException {
+ try (final RandomAccessFile bytes = new RandomAccessFile(file, "rw")) {
+ bytes.seek(offset);
+ final int original = bytes.readUnsignedByte();
+ bytes.seek(offset);
+ bytes.writeByte(value);
+ return original;
+ }
+ }
+
+ private static void assertFallbackSourceExists(final File file, final
boolean expected) {
+ Assert.assertEquals(file.toString(), expected, file.exists());
+ Assert.assertEquals(
+ file.toString(), expected, new File(file +
TsFileResource.RESOURCE_SUFFIX).exists());
+ Assert.assertEquals(
+ file.toString(), expected, new File(file +
ModificationFile.FILE_SUFFIX).exists());
+ }
+
+ private static void assertFallbackRows(final Statement statement, final
String device)
+ throws SQLException {
+ assertFallbackRows(statement, device, new long[] {1, 2, 3, 4});
+ }
+
+ private static long[] fallbackPartitionTimestamps() {
+ final long timePartitionInterval =
+ CommonDescriptor.getInstance().getConfig().getTimePartitionInterval();
+ return new long[] {1, 2, timePartitionInterval + 1, timePartitionInterval
+ 2};
+ }
+
+ private static void assertFallbackRows(
+ final Statement statement, final String device, final long[] timestamps)
throws SQLException {
+ try (final ResultSet rows = statement.executeQuery("select s1,s2 from " +
device)) {
+ for (int i = 0; i < 4; i++) {
+ Assert.assertTrue(device, rows.next());
+ Assert.assertEquals(timestamps[i], rows.getLong(1));
+ if (i == 1) {
+ Assert.assertNull(rows.getObject(2));
+ } else {
+ Assert.assertEquals(i, rows.getLong(2));
+ }
+ Assert.assertEquals(i * 10, rows.getLong(3));
+ }
+ Assert.assertFalse(device, rows.next());
+ }
+ }
+
@Test
public void testLoadWithLastCache() throws Exception {
registerSchema();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java
index 8c7e260aa68..3f0d9fe175a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java
@@ -36,6 +36,7 @@ import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.file.metadata.PlainDeviceID;
import org.apache.tsfile.file.metadata.TimeseriesMetadata;
import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.TsFileSequenceReaderTimeseriesMetadataIterator;
import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -49,19 +50,18 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
-import java.util.Set;
import java.util.stream.Collectors;
import static
org.apache.iotdb.db.storageengine.load.LoadTsFilePathUtils.getValidatedDevicePath;
/**
* Load uses scan parsing first for throughput. If scan parsing hits
corruption, fall back to query
- * parsing for the remaining measurements and devices so later data can still
be loaded.
+ * parsing for the remaining measurements and devices so later data can still
be loaded. Report any
+ * unrecoverable query failure after exhausting the readable data so a partial
load cannot succeed.
*/
class LoadTreeTsFileTabletIterator
implements Iterable<Pair<Tablet, Boolean>>, Iterator<Pair<Tablet,
Boolean>>, AutoCloseable {
@@ -91,14 +91,9 @@ class LoadTreeTsFileTabletIterator
private Iterator<Pair<Tablet, Boolean>> activeIterator;
private boolean scanInitialized;
private boolean fallbackTriggered;
+ private RuntimeException unrecoverableQueryFailure;
- private IDeviceID lastEmittedDevice;
- private List<String> lastEmittedMeasurements = Collections.emptyList();
- private long lastEmittedTimestamp = Long.MIN_VALUE;
-
- private IDeviceID lastScanTabletDevice;
- private List<String> lastScanTabletMeasurements = Collections.emptyList();
- private final Map<IDeviceID, Set<String>> fullyEmittedMeasurementsByDevice =
+ private final Map<IDeviceID, Map<String, Long>>
lastEmittedTimestampsByDevice =
new LinkedHashMap<>();
LoadTreeTsFileTabletIterator(final File file, final boolean isWithMod) {
@@ -117,8 +112,7 @@ class LoadTreeTsFileTabletIterator
try {
ensureActiveIterator();
if (Objects.isNull(activeIterator)) {
- close();
- return false;
+ return finishIteration();
}
if (activeIterator.hasNext()) {
@@ -126,8 +120,7 @@ class LoadTreeTsFileTabletIterator
}
if (!switchToNextIterator()) {
- close();
- return false;
+ return finishIteration();
}
} catch (final Exception e) {
if (recoverFromIteratorFailure(e)) {
@@ -222,7 +215,7 @@ class LoadTreeTsFileTabletIterator
activeQueryTask.measurements,
file.getAbsolutePath(),
e);
- splitOrSkipActiveQueryTask();
+ splitOrSkipActiveQueryTask(e);
return true;
}
@@ -237,15 +230,10 @@ class LoadTreeTsFileTabletIterator
fallbackTriggered = true;
final IDeviceID currentDevice =
Objects.nonNull(scanParser) ? scanParser.getCurrentDevice() : null;
- final List<String> currentMeasurements =
- Objects.nonNull(scanParser) ? scanParser.getCurrentMeasurements() :
Collections.emptyList();
-
- markLastScanMeasurementsAsCompletedIfNeeded(currentDevice,
currentMeasurements);
-
closeScanParser();
try {
- pendingQueryTasks.addAll(buildQueryTasks(currentDevice,
currentMeasurements));
+ pendingQueryTasks.addAll(buildQueryTasks());
} catch (final Exception queryInitException) {
LOGGER.warn(
"Load: Failed to initialize query fallback for TsFile {} after scan
parser failure.",
@@ -263,8 +251,7 @@ class LoadTreeTsFileTabletIterator
return true;
}
- private ArrayDeque<QueryTask> buildQueryTasks(
- final IDeviceID currentDevice, final List<String> currentMeasurements)
throws IOException {
+ private ArrayDeque<QueryTask> buildQueryTasks() throws IOException {
final LinkedHashMap<IDeviceID, List<String>> deviceMeasurementsMap =
readDeviceMeasurementsInOrder();
if (deviceMeasurementsMap.isEmpty()) {
@@ -272,23 +259,10 @@ class LoadTreeTsFileTabletIterator
}
final ArrayDeque<QueryTask> tasks = new ArrayDeque<>();
- boolean includeCurrentAndFollowingDevices =
- Objects.isNull(currentDevice) ||
!deviceMeasurementsMap.containsKey(currentDevice);
-
+ // Devices and measurements can recur in later chunk groups. The metadata
index order does
+ // not describe the scan position, so resume every series from its own
emitted prefix.
for (final Map.Entry<IDeviceID, List<String>> entry :
deviceMeasurementsMap.entrySet()) {
- final IDeviceID device = entry.getKey();
- if (!includeCurrentAndFollowingDevices && device.equals(currentDevice)) {
- includeCurrentAndFollowingDevices = true;
- }
- if (!includeCurrentAndFollowingDevices) {
- continue;
- }
-
- if (device.equals(currentDevice)) {
- addCurrentDeviceQueryTasks(tasks, device, entry.getValue(),
currentMeasurements);
- } else {
- addQueryTaskIfNecessary(tasks, device, entry.getValue(),
Long.MIN_VALUE, Long.MAX_VALUE);
- }
+ addQueryTasks(tasks, entry.getKey(), entry.getValue(), Long.MIN_VALUE,
Long.MAX_VALUE);
}
return tasks;
@@ -298,49 +272,65 @@ class LoadTreeTsFileTabletIterator
throws IOException {
final LinkedHashMap<IDeviceID, List<String>> deviceMeasurementsMap = new
LinkedHashMap<>();
try (final TsFileSequenceReader reader = new
TsFileSequenceReader(file.getAbsolutePath())) {
- final Iterator<Pair<IDeviceID, List<TimeseriesMetadata>>>
metadataIterator =
- reader.iterAllTimeseriesMetadata(false, false);
+ // iterAllTimeseriesMetadata treats an IOException as the end of
iteration. Use the
+ // throwing batch iterator so an incomplete metadata index cannot turn a
partial load into a
+ // success without rereading every device's metadata separately.
+ final TsFileSequenceReaderTimeseriesMetadataIterator metadataIterator =
+ new TsFileSequenceReaderTimeseriesMetadataIterator(reader, false);
while (metadataIterator.hasNext()) {
- final Pair<IDeviceID, List<TimeseriesMetadata>> deviceMetadata =
metadataIterator.next();
- deviceMeasurementsMap.put(
- deviceMetadata.getLeft(),
- deviceMetadata.getRight().stream()
- .map(TimeseriesMetadata::getMeasurementId)
- .collect(Collectors.toList()));
+ for (final Map.Entry<IDeviceID, List<TimeseriesMetadata>>
deviceMetadata :
+ metadataIterator.next().entrySet()) {
+ final Map<String, Long> emittedTimestamps =
+ lastEmittedTimestampsByDevice.getOrDefault(
+ deviceMetadata.getKey(), Collections.emptyMap());
+ deviceMeasurementsMap
+ .computeIfAbsent(deviceMetadata.getKey(), key -> new
ArrayList<>())
+ .addAll(
+ deviceMetadata.getValue().stream()
+ .filter(
+ metadata -> {
+ final Long emittedTimestamp =
+
emittedTimestamps.get(metadata.getMeasurementId());
+ return !metadata.getMeasurementId().isEmpty()
+ && (emittedTimestamp == null
+ || emittedTimestamp <
metadata.getStatistics().getEndTime());
+ })
+ .map(TimeseriesMetadata::getMeasurementId)
+ .collect(Collectors.toList()));
+ }
}
}
return deviceMeasurementsMap;
}
- private void addCurrentDeviceQueryTasks(
+ private void addQueryTasks(
final ArrayDeque<QueryTask> tasks,
final IDeviceID device,
- final List<String> allMeasurements,
- final List<String> currentMeasurements) {
- final Set<String> completedMeasurements =
- fullyEmittedMeasurementsByDevice.getOrDefault(device,
Collections.emptySet());
- final Set<String> currentMeasurementSet = new
LinkedHashSet<>(currentMeasurements);
-
- final List<String> currentMeasurementsToResume = new ArrayList<>();
- final List<String> remainingMeasurements = new ArrayList<>();
- for (final String measurement : allMeasurements) {
- if (completedMeasurements.contains(measurement)) {
+ final List<String> measurements,
+ final long startTime,
+ final long endTime) {
+ final Map<String, Long> emittedTimestamps =
+ lastEmittedTimestampsByDevice.getOrDefault(device,
Collections.emptyMap());
+ final Map<Long, List<String>> measurementsByStartTime = new
LinkedHashMap<>();
+ for (final String measurement : measurements) {
+ final Long emittedTimestamp = emittedTimestamps.get(measurement);
+ if (emittedTimestamp != null && emittedTimestamp >= endTime) {
continue;
}
- if (currentMeasurementSet.contains(measurement)) {
- currentMeasurementsToResume.add(measurement);
- } else {
- remainingMeasurements.add(measurement);
+ // A missing checkpoint differs from a checkpoint at Long.MIN_VALUE. A
start time of
+ // Long.MAX_VALUE is also valid when the last unconsumed point has that
timestamp.
+ final long resumeStartTime =
+ emittedTimestamp == null ? startTime : Math.max(startTime,
emittedTimestamp + 1);
+ if (resumeStartTime <= endTime) {
+ measurementsByStartTime
+ .computeIfAbsent(resumeStartTime, key -> new ArrayList<>())
+ .add(measurement);
}
}
- addQueryTaskIfNecessary(
- tasks,
- device,
- currentMeasurementsToResume,
- determineTaskResumeStartTime(device, currentMeasurementsToResume,
Long.MIN_VALUE),
- Long.MAX_VALUE);
- addQueryTaskIfNecessary(tasks, device, remainingMeasurements,
Long.MIN_VALUE, Long.MAX_VALUE);
+ measurementsByStartTime.forEach(
+ (resumeStartTime, taskMeasurements) ->
+ tasks.addLast(new QueryTask(device, taskMeasurements,
resumeStartTime, endTime)));
}
private boolean activateNextQueryParser() {
@@ -396,7 +386,7 @@ class LoadTreeTsFileTabletIterator
activeQueryTask.measurements,
file.getAbsolutePath(),
e);
- splitOrSkipActiveQueryTask();
+ splitOrSkipActiveQueryTask(e);
}
}
@@ -411,15 +401,12 @@ class LoadTreeTsFileTabletIterator
}
final IDeviceID device = new PlainDeviceID(tablet.deviceId);
- final List<String> measurements = extractMeasurementNames(tablet);
-
- if (Objects.isNull(activeQueryParser)) {
- recordScanProgress(device, measurements);
+ final Map<String, Long> emittedTimestamps =
+ lastEmittedTimestampsByDevice.computeIfAbsent(device, key -> new
LinkedHashMap<>());
+ final long timestamp = tablet.timestamps[tablet.rowSize - 1];
+ for (final String measurement : extractMeasurementNames(tablet)) {
+ emittedTimestamps.merge(measurement, timestamp, Math::max);
}
-
- lastEmittedDevice = device;
- lastEmittedMeasurements = measurements;
- lastEmittedTimestamp = tablet.timestamps[tablet.rowSize - 1];
}
private boolean shouldRethrow(final Exception e) {
@@ -478,67 +465,15 @@ class LoadTreeTsFileTabletIterator
pendingQueryTasks.clear();
}
- private void recordScanProgress(final IDeviceID device, final List<String>
measurements) {
- if (Objects.nonNull(lastScanTabletDevice)
- && (!lastScanTabletDevice.equals(device)
- || !measurementsEqual(lastScanTabletMeasurements, measurements))) {
- markMeasurementsFullyEmitted(lastScanTabletDevice,
lastScanTabletMeasurements);
- }
-
- lastScanTabletDevice = device;
- lastScanTabletMeasurements = measurements;
- }
-
- private void markLastScanMeasurementsAsCompletedIfNeeded(
- final IDeviceID currentDevice, final List<String> currentMeasurements) {
- if (Objects.isNull(lastScanTabletDevice) ||
lastScanTabletMeasurements.isEmpty()) {
- return;
- }
-
- if (!lastScanTabletDevice.equals(currentDevice)
- || !currentMeasurements.isEmpty()
- && !measurementsEqual(lastScanTabletMeasurements,
currentMeasurements)) {
- markMeasurementsFullyEmitted(lastScanTabletDevice,
lastScanTabletMeasurements);
+ private boolean finishIteration() {
+ close();
+ if (Objects.nonNull(unrecoverableQueryFailure)) {
+ throw unrecoverableQueryFailure;
}
+ return false;
}
- private void markMeasurementsFullyEmitted(
- final IDeviceID device, final List<String> measurements) {
- if (Objects.isNull(device) || measurements.isEmpty()) {
- return;
- }
-
- fullyEmittedMeasurementsByDevice
- .computeIfAbsent(device, key -> new LinkedHashSet<>())
- .addAll(measurements);
- }
-
- private long determineTaskResumeStartTime(
- final IDeviceID device, final List<String> measurements, final long
defaultStartTime) {
- if (measurements.isEmpty()
- || !device.equals(lastEmittedDevice)
- || lastEmittedTimestamp == Long.MIN_VALUE
- || !measurementsEqual(measurements, lastEmittedMeasurements)) {
- return defaultStartTime;
- }
-
- return lastEmittedTimestamp == Long.MAX_VALUE ? Long.MAX_VALUE :
lastEmittedTimestamp + 1;
- }
-
- private void addQueryTaskIfNecessary(
- final ArrayDeque<QueryTask> tasks,
- final IDeviceID device,
- final List<String> measurements,
- final long startTime,
- final long endTime) {
- if (measurements.isEmpty() || startTime == Long.MAX_VALUE) {
- return;
- }
-
- tasks.addLast(new QueryTask(device, measurements, startTime, endTime));
- }
-
- private void splitOrSkipActiveQueryTask() {
+ private void splitOrSkipActiveQueryTask(final Exception failure) {
final QueryTask failedTask = activeQueryTask;
closeActiveQueryParser();
if (Objects.isNull(failedTask)) {
@@ -546,15 +481,20 @@ class LoadTreeTsFileTabletIterator
}
if (failedTask.measurements.size() <= 1) {
+ // Keep salvaging later tasks, but retain the first failure so the
caller preserves the
+ // source.
+ if (Objects.isNull(unrecoverableQueryFailure)) {
+ unrecoverableQueryFailure = toRuntimeException(failure);
+ }
return;
}
- final long resumeStartTime =
- determineTaskResumeStartTime(
- failedTask.device, failedTask.measurements, failedTask.startTime);
- final List<QueryTask> splitTasks = failedTask.split(resumeStartTime);
- for (int i = splitTasks.size() - 1; i >= 0; --i) {
- pendingQueryTasks.addFirst(splitTasks.get(i));
+ final ArrayDeque<QueryTask> splitTasks = new ArrayDeque<>();
+ for (final QueryTask task : failedTask.split(failedTask.startTime)) {
+ addQueryTasks(splitTasks, task.device, task.measurements,
task.startTime, task.endTime);
+ }
+ while (!splitTasks.isEmpty()) {
+ pendingQueryTasks.addFirst(splitTasks.removeLast());
}
}
@@ -564,12 +504,6 @@ class LoadTreeTsFileTabletIterator
.collect(Collectors.toList());
}
- private boolean measurementsEqual(
- final List<String> leftMeasurements, final List<String>
rightMeasurements) {
- return leftMeasurements.size() == rightMeasurements.size()
- && new LinkedHashSet<>(leftMeasurements).equals(new
LinkedHashSet<>(rightMeasurements));
- }
-
private static class QueryTask {
private final IDeviceID device;
private final List<String> measurements;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitorTest.java
index 0cfbbc364f4..dc5ab073bcd 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitorTest.java
@@ -30,10 +30,14 @@ import
org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.IMetadataIndexEntry;
+import org.apache.tsfile.file.MetaMarker;
import org.apache.tsfile.file.metadata.AlignedChunkMetadata;
import org.apache.tsfile.file.metadata.IChunkMetadata;
import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.MetadataIndexNode;
import org.apache.tsfile.file.metadata.PlainDeviceID;
+import org.apache.tsfile.file.metadata.enums.MetadataIndexNodeType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.read.TsFileSequenceReader;
import org.apache.tsfile.read.common.Path;
@@ -48,6 +52,7 @@ import org.junit.Before;
import org.junit.Test;
import java.io.File;
+import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.Arrays;
import java.util.HashMap;
@@ -101,16 +106,299 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
});
final Optional<TSStatus> status =
-
visitor.visitLoadFile(LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()),
null);
+ visitor.visitLoadFile(
+
LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()).setDeleteAfterLoad(true),
+ null);
Assert.assertTrue(status.isPresent());
- Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.get().getCode());
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(),
status.get().getCode());
+ Assert.assertTrue("A partial load must retain its source",
tsFile.exists());
final int loadedPointCountBeforeCorruption =
pointCountByDevice.getOrDefault(DEVICE_0, 0);
final int loadedPointCountAfterFallback =
pointCountByDevice.getOrDefault(DEVICE_2, 0);
Assert.assertTrue(loadedPointCountBeforeCorruption > 0);
Assert.assertEquals(loadedPointCountBeforeCorruption,
loadedPointCountAfterFallback);
}
+ @Test
+ public void testFullyRecoveredQueryFallbackCanDeleteSource() throws
Exception {
+ tsFile = new File("load-tree-query-fallback-recoverable-marker.tsfile");
+ final long groupMarkerOffset = writeTsFile(tsFile);
+ // Keep the first marker valid for the reader's encryption-header check.
Indexed queries can
+ // recover every point without reading the second device's chunk-group
marker.
+ try (final TsFileSequenceReader reader = new
TsFileSequenceReader(tsFile.getAbsolutePath())) {
+ reader.position(groupMarkerOffset);
+ Assert.assertEquals(MetaMarker.CHUNK_GROUP_HEADER, reader.readMarker());
+ }
+ try (final RandomAccessFile file = new RandomAccessFile(tsFile, "rw")) {
+ file.seek(groupMarkerOffset);
+ file.writeByte(0x7f);
+ }
+ Assert.assertTrue("Expected scan parser to fail after corruption.",
scanParserFails(tsFile));
+
+ final Map<String, Integer> pointCountByTimeseries = new HashMap<>();
+ final LoadTreeStatementDataTypeConvertExecutionVisitor visitor =
+ new LoadTreeStatementDataTypeConvertExecutionVisitor(
+ statement -> {
+ collectLoadedPointsByTimeseries(statement,
pointCountByTimeseries);
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ });
+ final Optional<TSStatus> status =
+ visitor.visitLoadFile(
+
LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()).setDeleteAfterLoad(true),
+ null);
+
+ Assert.assertTrue(status.isPresent());
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.get().getCode());
+ for (final String device : Arrays.asList(DEVICE_0, DEVICE_1, DEVICE_2)) {
+ assertMeasurementLoadedCompletely(pointCountByTimeseries, device, 0);
+ assertMeasurementLoadedCompletely(pointCountByTimeseries, device, 1);
+ }
+ Assert.assertFalse("A fully recovered load can delete its source",
tsFile.exists());
+ }
+
+ @Test
+ public void testFallbackRetainsSourceWhenFirstDeviceMetadataIsUnreadable()
throws Exception {
+ assertFallbackRetainsSourceWhenMetadataIsUnreadable(DEVICE_0);
+ }
+
+ @Test
+ public void testFallbackRetainsSourceWhenLaterDeviceMetadataIsUnreadable()
throws Exception {
+ assertFallbackRetainsSourceWhenMetadataIsUnreadable(DEVICE_1);
+ }
+
+ private void assertFallbackRetainsSourceWhenMetadataIsUnreadable(final
String corruptedDevice)
+ throws Exception {
+ tsFile = File.createTempFile("load-tree-unreadable-metadata-", ".tsfile");
+ final long groupMarkerOffset = writeTsFile(tsFile);
+ final IDeviceID deviceId = new PlainDeviceID(corruptedDevice);
+ final long measurementIndexEndOffsetPosition;
+ try (final TsFileSequenceReader reader = new
TsFileSequenceReader(tsFile.getAbsolutePath())) {
+ final MetadataIndexNode deviceIndex =
reader.readFileMetadata().getMetadataIndex();
+ Assert.assertEquals(MetadataIndexNodeType.LEAF_DEVICE,
deviceIndex.getNodeType());
+ final Pair<IMetadataIndexEntry, Long> deviceEntry =
+ deviceIndex.getChildIndexEntry(deviceId, true);
+ Assert.assertNotNull(deviceEntry);
+ final MetadataIndexNode measurementIndex =
+ reader.readMetadataIndexNode(deviceEntry.left.getOffset(),
deviceEntry.right, false);
+ Assert.assertEquals(MetadataIndexNodeType.LEAF_MEASUREMENT,
measurementIndex.getNodeType());
+ measurementIndexEndOffsetPosition = deviceEntry.right - Long.BYTES -
Byte.BYTES;
+ }
+ // Preserve the footer and device index, but make the measurement metadata
read reach EOF.
+ try (final RandomAccessFile bytes = new RandomAccessFile(tsFile, "rw")) {
+ bytes.seek(measurementIndexEndOffsetPosition);
+ bytes.writeLong(bytes.length() + 1);
+ }
+ try (final TsFileSequenceReader reader = new
TsFileSequenceReader(tsFile.getAbsolutePath())) {
+ Assert.assertThrows(IOException.class, () ->
reader.readDeviceMetadata(deviceId));
+ }
+ corruptGroupMarker(tsFile, groupMarkerOffset);
+ Assert.assertTrue("Expected scan parser to fail after corruption.",
scanParserFails(tsFile));
+
+ final Map<String, Integer> points = new HashMap<>();
+ final LoadTreeStatementDataTypeConvertExecutionVisitor visitor =
+ new LoadTreeStatementDataTypeConvertExecutionVisitor(
+ statement -> {
+ collectLoadedPointsByTimeseries(statement, points);
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ });
+ final Optional<TSStatus> status =
+ visitor.visitLoadFile(
+
LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()).setDeleteAfterLoad(true),
+ null);
+
+ Assert.assertTrue(status.isPresent());
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(),
status.get().getCode());
+ Assert.assertTrue("Unreadable metadata must retain the source",
tsFile.exists());
+ assertMeasurementLoadedCompletely(points, DEVICE_0, 0);
+ assertMeasurementLoadedCompletely(points, DEVICE_0, 1);
+ }
+
+ @Test
+ public void testFallbackResumesRepeatedNonAlignedDevicesAcrossChunkGroups()
throws Exception {
+ assertFallbackResumesRepeatedDevices(false);
+ }
+
+ @Test
+ public void testFallbackResumesRepeatedAlignedDevicesAcrossChunkGroups()
throws Exception {
+ assertFallbackResumesRepeatedDevices(true);
+ }
+
+ @Test
+ public void testFallbackRetainsMeasurementsSplitAcrossMetadataBatches()
throws Exception {
+ tsFile = File.createTempFile("load-tree-metadata-batches-", ".tsfile");
+ final int measurementCount = 4001;
+ final List<MeasurementSchema> schemas = new java.util.ArrayList<>();
+ for (int measurement = 0; measurement < measurementCount; measurement++) {
+ schemas.add(new MeasurementSchema("s" + measurement, TSDataType.INT64,
TSEncoding.PLAIN));
+ }
+
+ final long secondGroupMarkerOffset;
+ try (final TsFileWriter writer = new TsFileWriter(tsFile)) {
+ writer.registerAlignedTimeseries(new Path(DEVICE_0), schemas);
+ writeWideAlignedRow(writer, schemas, 0);
+ writer.flushAllChunkGroups();
+ secondGroupMarkerOffset = writer.getIOWriter().getPos();
+ writeWideAlignedRow(writer, schemas, 1);
+ }
+ corruptGroupMarker(tsFile, secondGroupMarkerOffset);
+ Assert.assertTrue("Expected scan parser to fail after corruption.",
scanParserFails(tsFile));
+
+ final Map<String, Map<Long, Long>> points = new HashMap<>();
+ final LoadTreeStatementDataTypeConvertExecutionVisitor visitor =
+ new LoadTreeStatementDataTypeConvertExecutionVisitor(
+ statement -> {
+ collectLoadedLongPoints(statement, points);
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ });
+ final Optional<TSStatus> status =
+ visitor.visitLoadFile(
+
LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()).setDeleteAfterLoad(true),
+ null);
+
+ Assert.assertTrue(status.isPresent());
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.get().getCode());
+ Assert.assertEquals(measurementCount, points.size());
+ for (int measurement = 0; measurement < measurementCount; measurement++) {
+ final String path = DEVICE_0 + ".s" + measurement;
+ final Map<Long, Long> values = points.get(path);
+ Assert.assertNotNull(path, values);
+ Assert.assertEquals(path, 2, values.size());
+ Assert.assertEquals(Long.valueOf(measurement * 10_000L), values.get(0L));
+ Assert.assertEquals(Long.valueOf(measurement * 10_000L + 1),
values.get(1L));
+ }
+ Assert.assertFalse("A fully recovered load can delete its source",
tsFile.exists());
+ }
+
+ @Test
+ public void testFallbackDoesNotReplayMinimumTimestamp() throws Exception {
+ assertFallbackResumesExtremeTimestamps(Long.MIN_VALUE);
+ }
+
+ @Test
+ public void testFallbackResumesAtMaximumTimestamp() throws Exception {
+ assertFallbackResumesExtremeTimestamps(Long.MAX_VALUE - 1);
+ }
+
+ private void assertFallbackResumesExtremeTimestamps(final long
firstTimestamp) throws Exception {
+ tsFile = File.createTempFile("load-tree-extreme-timestamps-", ".tsfile");
+ final List<MeasurementSchema> schemas =
+ Arrays.asList(new MeasurementSchema("s0", TSDataType.INT64,
TSEncoding.PLAIN));
+ final long secondFlushOffset;
+ try (final TsFileWriter writer = new TsFileWriter(tsFile)) {
+ writer.registerTimeseries(new Path(DEVICE_0), schemas);
+ writeRows(writer, schemas, DEVICE_0, firstTimestamp, 1, false);
+ writer.flushAllChunkGroups();
+ secondFlushOffset = writer.getIOWriter().getPos();
+ writeRows(writer, schemas, DEVICE_0, firstTimestamp + 1, 1, false);
+ }
+ corruptGroupMarker(tsFile, secondFlushOffset);
+
+ final Map<String, Map<Long, Long>> points = new HashMap<>();
+ final LoadTreeStatementDataTypeConvertExecutionVisitor visitor =
+ new LoadTreeStatementDataTypeConvertExecutionVisitor(
+ statement -> {
+ collectLoadedLongPoints(statement, points);
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ });
+ final Optional<TSStatus> status =
+ visitor.visitLoadFile(
+
LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()).setDeleteAfterLoad(true),
+ null);
+ Assert.assertTrue(status.isPresent());
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.get().getCode());
+ final Map<Long, Long> values = points.get(DEVICE_0 + ".s0");
+ Assert.assertNotNull(values);
+ Assert.assertEquals(2, values.size());
+ Assert.assertEquals(Long.valueOf(firstTimestamp),
values.get(firstTimestamp));
+ Assert.assertEquals(Long.valueOf(firstTimestamp + 1),
values.get(firstTimestamp + 1));
+ Assert.assertFalse("A fully recovered load can delete its source",
tsFile.exists());
+ }
+
+ private void assertFallbackResumesRepeatedDevices(final boolean aligned)
throws Exception {
+ tsFile = File.createTempFile("load-tree-repeated-devices-", ".tsfile");
+ final List<MeasurementSchema> schemas =
+ Arrays.asList(
+ new MeasurementSchema("s0", TSDataType.INT64, TSEncoding.PLAIN),
+ new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN));
+ final long secondFlushOffset;
+ try (final TsFileWriter writer = new TsFileWriter(tsFile)) {
+ for (final String device : Arrays.asList(DEVICE_0, DEVICE_1)) {
+ if (aligned) {
+ writer.registerAlignedTimeseries(new Path(device), schemas);
+ } else {
+ writer.registerTimeseries(new Path(device), schemas);
+ }
+ writeRows(writer, schemas, device, 0, 4, aligned);
+ }
+ writer.flushAllChunkGroups();
+ secondFlushOffset = writer.getIOWriter().getPos();
+ for (final String device : Arrays.asList(DEVICE_0, DEVICE_1)) {
+ writeRows(writer, schemas, device, 4, 4, aligned);
+ }
+ }
+ corruptGroupMarker(tsFile, secondFlushOffset);
+
+ final Map<String, Map<Long, Long>> points = new HashMap<>();
+ final LoadTreeStatementDataTypeConvertExecutionVisitor visitor =
+ new LoadTreeStatementDataTypeConvertExecutionVisitor(
+ statement -> {
+ collectLoadedLongPoints(statement, points);
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ });
+ final Optional<TSStatus> status =
+ visitor.visitLoadFile(
+
LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()).setDeleteAfterLoad(true),
+ null);
+ Assert.assertTrue(status.isPresent());
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.get().getCode());
+ for (final String device : Arrays.asList(DEVICE_0, DEVICE_1)) {
+ for (int measurement = 0; measurement < 2; measurement++) {
+ final String path = device + ".s" + measurement;
+ final Map<Long, Long> values = points.get(path);
+ Assert.assertNotNull(path, values);
+ Assert.assertEquals(path, 8, values.size());
+ for (long timestamp = 0; timestamp < 8; timestamp++) {
+ Assert.assertEquals(
+ path, Long.valueOf(timestamp + measurement * 100),
values.get(timestamp));
+ }
+ }
+ }
+ Assert.assertFalse("A fully recovered load can delete its source",
tsFile.exists());
+ }
+
+ private static void corruptGroupMarker(final File tsFile, final long offset)
throws Exception {
+ try (final RandomAccessFile file = new RandomAccessFile(tsFile, "rw")) {
+ file.seek(offset);
+ Assert.assertEquals(MetaMarker.CHUNK_GROUP_HEADER, file.readByte());
+ file.seek(offset);
+ file.writeByte(0x7f);
+ }
+ }
+
+ private static void writeRows(
+ final TsFileWriter writer,
+ final List<MeasurementSchema> schemas,
+ final String device,
+ final long firstTimestamp,
+ final int rowCount,
+ final boolean aligned)
+ throws Exception {
+ final Tablet tablet = new Tablet(device, schemas, rowCount);
+ for (int row = 0; row < rowCount; row++) {
+ tablet.addTimestamp(row, firstTimestamp + row);
+ for (int column = 0; column < schemas.size(); column++) {
+ tablet.addValue(
+ schemas.get(column).getMeasurementId(), row, firstTimestamp + row
+ column * 100);
+ }
+ }
+ tablet.rowSize = rowCount;
+ if (aligned) {
+ writer.writeAligned(tablet);
+ } else {
+ writer.write(tablet);
+ }
+ }
+
@Test
public void testFlushesPendingTabletsWhenIteratorFails() throws Exception {
tsFile = File.createTempFile("load-tree-pending-tablet", ".tsfile");
@@ -158,7 +446,7 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
visitor.visitLoadFile(LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()),
null);
Assert.assertTrue(status.isPresent());
- Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.get().getCode());
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(),
status.get().getCode());
Assert.assertTrue(
pointCountByTimeseries.getOrDefault(DEVICE_0 + ".s0", 0) <
ROW_COUNT_PER_DEVICE);
@@ -190,7 +478,7 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
visitor.visitLoadFile(LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()),
null);
Assert.assertTrue(status.isPresent());
- Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.get().getCode());
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(),
status.get().getCode());
assertMeasurementLoadedCompletely(pointCountByTimeseries, DEVICE_1, 0);
Assert.assertTrue(
@@ -222,7 +510,7 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
visitor.visitLoadFile(LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()),
null);
Assert.assertTrue(status.isPresent());
- Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.get().getCode());
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(),
status.get().getCode());
for (int measurementIndex = 0; measurementIndex < 8; ++measurementIndex) {
assertMeasurementLoadedCompletely(pointCountByTimeseries,
ALIGNED_DEVICE, measurementIndex);
@@ -237,7 +525,7 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
pointCountByTimeseries.getOrDefault(ALIGNED_DEVICE + ".s12", 0) <
ROW_COUNT_PER_DEVICE);
}
- private void writeTsFile(final File file) throws Exception {
+ private long writeTsFile(final File file) throws Exception {
if (file.exists()) {
Assert.assertTrue(file.delete());
}
@@ -247,11 +535,15 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
new MeasurementSchema("s0", TSDataType.INT64, TSEncoding.PLAIN),
new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN));
+ final long secondGroupMarkerOffset;
try (final TsFileWriter writer = new TsFileWriter(file)) {
writeDevice(writer, schemaList, DEVICE_0, 0);
+ writer.flushAllChunkGroups();
+ secondGroupMarkerOffset = writer.getIOWriter().getPos();
writeDevice(writer, schemaList, DEVICE_1, 10_000);
writeDevice(writer, schemaList, DEVICE_2, 20_000);
}
+ return secondGroupMarkerOffset;
}
private void writeWideAlignedTsFile(
@@ -281,6 +573,19 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
}
}
+ private void writeWideAlignedRow(
+ final TsFileWriter writer, final List<MeasurementSchema> schemas, final
long timestamp)
+ throws Exception {
+ final Tablet tablet = new Tablet(DEVICE_0, schemas, 1);
+ tablet.addTimestamp(0, timestamp);
+ for (int measurement = 0; measurement < schemas.size(); measurement++) {
+ tablet.addValue(
+ schemas.get(measurement).getMeasurementId(), 0, measurement *
10_000L + timestamp);
+ }
+ tablet.rowSize = 1;
+ writer.writeAligned(tablet);
+ }
+
private void writeDevice(
final TsFileWriter writer,
final List<MeasurementSchema> schemaList,
@@ -353,6 +658,27 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
}
}
+ private void collectLoadedLongPoints(
+ final Statement statement, final Map<String, Map<Long, Long>> points) {
+ Assert.assertTrue(statement instanceof InsertMultiTabletsStatement);
+ for (final InsertTabletStatement tablet :
+ ((InsertMultiTabletsStatement)
statement).getInsertTabletStatementList()) {
+ for (int column = 0; column < tablet.getMeasurements().length; column++)
{
+ final String path =
+ tablet.getDevicePath().getFullPath() + "." +
tablet.getMeasurements()[column];
+ final Map<Long, Long> values = points.computeIfAbsent(path, key -> new
HashMap<>());
+ for (int row = 0; row < tablet.getRowCount(); row++) {
+ if (!isNull(tablet, row, column)) {
+ final long timestamp = tablet.getTimes()[row];
+ Assert.assertNull(
+ "Duplicate point: " + path + " at " + timestamp,
+ values.put(timestamp, ((long[])
tablet.getColumns()[column])[row]));
+ }
+ }
+ }
+ }
+ }
+
private void collectLoadedPointsByTimeseries(
final Statement statement, final Map<String, Integer>
pointCountByTimeseries) {
Assert.assertTrue(statement instanceof InsertMultiTabletsStatement);
