This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 9463d011e85 Preserve LOAD sources after unrecoverable query fallback
(#19073)
9463d011e85 is described below
commit 9463d011e857ca060c8d1c799f803a54fba9c115
Author: Caideyipi <[email protected]>
AuthorDate: Mon Nov 2 15:09:47 2026 +0800
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
---
.../iotdb/db/it/IoTDBLoadTsFileClusterIT.java | 238 ++++++++++++++-
.../converter/LoadTreeTsFileTabletIterator.java | 230 +++++---------
...atementDataTypeConvertExecutionVisitorTest.java | 337 ++++++++++++++++++++-
3 files changed, 650 insertions(+), 155 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileClusterIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileClusterIT.java
index 6b51c1ae06d..dadd5a7b4ba 100644
---
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileClusterIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileClusterIT.java
@@ -19,8 +19,13 @@
package org.apache.iotdb.db.it;
+import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.env.cluster.ClusterConstant;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
@@ -28,8 +33,12 @@ import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.it.utils.TsFileGenerator;
import org.apache.iotdb.itbase.category.ClusterIT;
import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.rpc.TSStatusCode;
+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.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.TsFileSequenceReader;
@@ -38,6 +47,8 @@ import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
import org.apache.tsfile.read.expression.QueryExpression;
import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.write.TsFileWriter;
+import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.awaitility.Awaitility;
@@ -52,6 +63,7 @@ import org.junit.runner.RunWith;
import java.io.File;
import java.io.IOException;
+import java.io.RandomAccessFile;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.sql.Connection;
@@ -68,6 +80,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -619,6 +632,228 @@ public class IoTDBLoadTsFileClusterIT {
}
}
+ @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 = DATABASE + (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};
+ long recoverableMarkerOffset = -1;
+ for (int i = 0; i < sources.length; i++) {
+ final long groupMarkerOffset =
+ writeFallbackSource(sources[i], devicePrefix + ".d" + i, i == 1);
+ if (i == 1) {
+ recoverableMarkerOffset = 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);
+
+ 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");
+ Assert.assertTrue(
+ EnvFactory.getEnv()
+ .getDataNodeWrapper(0)
+ .logContains(
+ DataNodeQueryMessages
+
.LOAD_SUCCESSFULLY_CONVERTED_TSFILE_ARG_INTO_TABLETS_AND_INSERTED
+ .replace("{}", recoverable.getAbsolutePath())));
+ assertFallbackSourceExists(normal, !delete);
+ assertFallbackSourceExists(recoverable, !delete);
+ assertFallbackSourceExists(failed, true);
+
+ replaceByte(failed, compressionOffset, originalCompression);
+ statement.execute(fallbackLoadSql(failed, delete));
+ assertFallbackRows(statement, devicePrefix + ".d2");
+ assertFallbackRows(statement, devicePrefix + ".d0");
+ assertFallbackRows(statement, devicePrefix + ".d1");
+ assertFallbackSourceExists(failed, !delete);
+ }
+ }
+ }
+
+ @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 = DATABASE + ".concurrent";
+ writeFallbackSource(good, device + ".good", false);
+ writeFallbackSource(bad, device + ".bad", false);
+ 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");
+ 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'='1','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)
throws Exception {
+ final List<IMeasurementSchema> schemas =
+ Arrays.asList(
+ new MeasurementSchema(
+ "s1", TSDataType.INT64, TSEncoding.PLAIN,
CompressionType.UNCOMPRESSED),
+ new MeasurementSchema(
+ "s2", TSDataType.INT64, TSEncoding.PLAIN,
CompressionType.UNCOMPRESSED));
+ final String anchor = DATABASE + ".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(anchor, schemas);
+ final Tablet anchorTablet = new Tablet(anchor, schemas);
+ anchorTablet.addTimestamp(0, 0);
+ anchorTablet.addValue(0, 0, 0L);
+ anchorTablet.addValue(0, 1, 0L);
+ writer.writeAligned(anchorTablet);
+ writer.flush();
+ }
+ groupMarkerOffset = writer.getIOWriter().getPos();
+ writer.registerAlignedTimeseries(device, schemas);
+ final Tablet tablet = new Tablet(device, schemas);
+ for (int i = 0; i < 4; i++) {
+ tablet.addTimestamp(i, i + 1);
+ tablet.addValue(i, 0, (long) i);
+ tablet.addValue(i, 1, (long) i * 10);
+ }
+ writer.writeAligned(tablet);
+ }
+ final TsFileResource resource = new TsFileResource(file);
+ final IDeviceID deviceId =
IDeviceID.Factory.DEFAULT_FACTORY.create(device);
+ resource.updateStartTime(deviceId, 1);
+ resource.updateEndTime(deviceId, 4);
+ if (includeAnchor) {
+ final IDeviceID anchorId =
IDeviceID.Factory.DEFAULT_FACTORY.create(anchor);
+ resource.updateStartTime(anchorId, 0);
+ resource.updateEndTime(anchorId, 0);
+ }
+ resource.serialize();
+ try (final ModificationFile mods = resource.getExclusiveModFile()) {
+ mods.write(new TreeDeletionEntry(new MeasurementPath(device + ".s1"), 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 {
+ 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(i + 1, 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());
+ }
+ }
+
private void createDatabaseAndSchema(final Statement statement) throws
SQLException {
statement.execute("create database " + DATABASE);
for (final String device : DEVICES) {
@@ -644,7 +879,8 @@ public class IoTDBLoadTsFileClusterIT {
try (final Connection connection =
EnvFactory.getEnv().getConnection();
final Statement statement = connection.createStatement();
final ResultSet resultSet = statement.executeQuery("show
databases")) {
- return resultSet.next();
+ // A healthy cluster may have no user databases.
+ return true;
}
});
}
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 7bfbb9b6df3..53c26f77882 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.exception.PathParseException;
import org.apache.tsfile.file.metadata.IDeviceID;
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.IMeasurementSchema;
@@ -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(
StorageEngineMessages
@@ -264,8 +252,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()) {
@@ -273,23 +260,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;
@@ -299,49 +273,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() {
@@ -399,7 +389,7 @@ class LoadTreeTsFileTabletIterator
activeQueryTask.measurements,
file.getAbsolutePath(),
e);
- splitOrSkipActiveQueryTask();
+ splitOrSkipActiveQueryTask(e);
}
}
@@ -414,15 +404,12 @@ class LoadTreeTsFileTabletIterator
}
final IDeviceID device =
IDeviceID.Factory.DEFAULT_FACTORY.create(tablet.getDeviceId());
- 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.getTimestamp(tablet.getRowSize() - 1);
+ for (final String measurement : extractMeasurementNames(tablet)) {
+ emittedTimestamps.merge(measurement, timestamp, Math::max);
}
-
- lastEmittedDevice = device;
- lastEmittedMeasurements = measurements;
- lastEmittedTimestamp = tablet.getTimestamp(tablet.getRowSize() - 1);
}
private boolean shouldRethrow(final Exception e) {
@@ -481,67 +468,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)) {
@@ -549,15 +484,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());
}
}
@@ -565,12 +505,6 @@ class LoadTreeTsFileTabletIterator
return IMeasurementSchema.getMeasurementNameList(tablet.getSchemas());
}
- 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 b374880d6a8..2ee8024e2f5 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
@@ -34,9 +34,13 @@ import
org.apache.iotdb.db.storageengine.load.memory.LoadTsFileMemoryManager;
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.AbstractAlignedChunkMetadata;
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.enums.MetadataIndexNodeType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.read.TsFileSequenceReader;
import org.apache.tsfile.read.common.Path;
@@ -51,6 +55,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;
@@ -104,16 +109,300 @@ 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 =
IDeviceID.Factory.DEFAULT_FACTORY.create(corruptedDevice);
+ final long measurementIndexEndOffsetPosition;
+ try (final TsFileSequenceReader reader = new
TsFileSequenceReader(tsFile.getAbsolutePath())) {
+ final MetadataIndexNode deviceIndex =
+
reader.readFileMetadata().getTableMetadataIndexNode(deviceId.getTableName());
+ 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.getDeviceTimeseriesMetadataWithoutChunkMetadata(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<IMeasurementSchema> 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 PartialPath(DEVICE_0), schemas);
+ writeWideAlignedRow(writer, schemas, 0);
+ writer.flush();
+ 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<IMeasurementSchema> 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.flush();
+ 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<IMeasurementSchema> 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(device, schemas);
+ } else {
+ writer.registerTimeseries(new Path(device), schemas);
+ }
+ writeRows(writer, schemas, device, 0, 4, aligned);
+ }
+ writer.flush();
+ 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<IMeasurementSchema> 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(row, column, firstTimestamp + row + column * 100);
+ }
+ }
+ if (aligned) {
+ writer.writeAligned(tablet);
+ } else {
+ writer.writeTree(tablet);
+ }
+ }
+
@Test
public void testFlushesPendingTabletsWhenIteratorFails() throws Exception {
tsFile = File.createTempFile("load-tree-pending-tablet", ".tsfile");
@@ -183,7 +472,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);
@@ -215,7 +504,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(
@@ -247,7 +536,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);
@@ -280,7 +569,7 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
Assert.assertNotEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(),
status.getCode());
}
- private void writeTsFile(final File file) throws Exception {
+ private long writeTsFile(final File file) throws Exception {
if (file.exists()) {
Assert.assertTrue(file.delete());
}
@@ -290,11 +579,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.flush();
+ secondGroupMarkerOffset = writer.getIOWriter().getPos();
writeDevice(writer, schemaList, DEVICE_1, 10_000);
writeDevice(writer, schemaList, DEVICE_2, 20_000);
}
+ return secondGroupMarkerOffset;
}
private void writeWideAlignedTsFile(
@@ -323,6 +616,17 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
}
}
+ private void writeWideAlignedRow(
+ final TsFileWriter writer, final List<IMeasurementSchema> 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(0, measurement, measurement * 10_000L + timestamp);
+ }
+ writer.writeAligned(tablet);
+ }
+
private void writeDevice(
final TsFileWriter writer,
final List<IMeasurementSchema> schemaList,
@@ -394,6 +698,27 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
}
}
+ private static 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 (!tablet.isNull(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);
