From 99667936233a9d7437f27b8b553a867de0ffc051 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:09:47 +0800 Subject: [PATCH 1/4] 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) --- .../apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 245 +++++++++++++ .../LoadTreeTsFileTabletIterator.java | 230 +++++------- ...ntDataTypeConvertExecutionVisitorTest.java | 338 +++++++++++++++++- 3 files changed, 659 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 c2fbbb0e30b50..94f3fb2843151 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 @@ -25,6 +25,7 @@ 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 +33,20 @@ 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 +59,8 @@ 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 +74,10 @@ 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 +798,235 @@ public void testLoadWithOnSuccess() throws Exception { } } + @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; + 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( + "Load: Successfully converted TsFile " + + recoverable.getAbsolutePath() + + " into tablets and inserted.")); + 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); + } + } 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); + 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> 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'='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) throws Exception { + final List 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++) { + tablet.addTimestamp(i, i + 1); + 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, 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 { + 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()); + } + } + @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 8c7e260aa680c..3f0d9fe175ae5 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.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.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>, Iterator>, AutoCloseable { @@ -91,14 +91,9 @@ public boolean mayOverlapWithDevice(final String device) { private Iterator> activeIterator; private boolean scanInitialized; private boolean fallbackTriggered; + private RuntimeException unrecoverableQueryFailure; - private IDeviceID lastEmittedDevice; - private List lastEmittedMeasurements = Collections.emptyList(); - private long lastEmittedTimestamp = Long.MIN_VALUE; - - private IDeviceID lastScanTabletDevice; - private List lastScanTabletMeasurements = Collections.emptyList(); - private final Map> fullyEmittedMeasurementsByDevice = + private final Map> lastEmittedTimestampsByDevice = new LinkedHashMap<>(); LoadTreeTsFileTabletIterator(final File file, final boolean isWithMod) { @@ -117,8 +112,7 @@ public boolean hasNext() { try { ensureActiveIterator(); if (Objects.isNull(activeIterator)) { - close(); - return false; + return finishIteration(); } if (activeIterator.hasNext()) { @@ -126,8 +120,7 @@ public boolean hasNext() { } if (!switchToNextIterator()) { - close(); - return false; + return finishIteration(); } } catch (final Exception e) { if (recoverFromIteratorFailure(e)) { @@ -222,7 +215,7 @@ private boolean recoverFromIteratorFailure(final Exception e) { activeQueryTask.measurements, file.getAbsolutePath(), e); - splitOrSkipActiveQueryTask(); + splitOrSkipActiveQueryTask(e); return true; } @@ -237,15 +230,10 @@ private boolean switchFromScanToQuery(final Exception e) { fallbackTriggered = true; final IDeviceID currentDevice = Objects.nonNull(scanParser) ? scanParser.getCurrentDevice() : null; - final List 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 @@ private boolean switchFromScanToQuery(final Exception e) { return true; } - private ArrayDeque buildQueryTasks( - final IDeviceID currentDevice, final List currentMeasurements) throws IOException { + private ArrayDeque buildQueryTasks() throws IOException { final LinkedHashMap> deviceMeasurementsMap = readDeviceMeasurementsInOrder(); if (deviceMeasurementsMap.isEmpty()) { @@ -272,23 +259,10 @@ private ArrayDeque buildQueryTasks( } final ArrayDeque 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> 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 @@ private LinkedHashMap> readDeviceMeasurementsInOrder() throws IOException { final LinkedHashMap> deviceMeasurementsMap = new LinkedHashMap<>(); try (final TsFileSequenceReader reader = new TsFileSequenceReader(file.getAbsolutePath())) { - final Iterator>> 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> deviceMetadata = metadataIterator.next(); - deviceMeasurementsMap.put( - deviceMetadata.getLeft(), - deviceMetadata.getRight().stream() - .map(TimeseriesMetadata::getMeasurementId) - .collect(Collectors.toList())); + for (final Map.Entry> deviceMetadata : + metadataIterator.next().entrySet()) { + final Map 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 tasks, final IDeviceID device, - final List allMeasurements, - final List currentMeasurements) { - final Set completedMeasurements = - fullyEmittedMeasurementsByDevice.getOrDefault(device, Collections.emptySet()); - final Set currentMeasurementSet = new LinkedHashSet<>(currentMeasurements); - - final List currentMeasurementsToResume = new ArrayList<>(); - final List remainingMeasurements = new ArrayList<>(); - for (final String measurement : allMeasurements) { - if (completedMeasurements.contains(measurement)) { + final List measurements, + final long startTime, + final long endTime) { + final Map emittedTimestamps = + lastEmittedTimestampsByDevice.getOrDefault(device, Collections.emptyMap()); + final Map> 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 @@ public Pair next() { activeQueryTask.measurements, file.getAbsolutePath(), e); - splitOrSkipActiveQueryTask(); + splitOrSkipActiveQueryTask(e); } } @@ -411,15 +401,12 @@ private void recordProgress(final Pair tabletWithIsAligned) { } final IDeviceID device = new PlainDeviceID(tablet.deviceId); - final List measurements = extractMeasurementNames(tablet); - - if (Objects.isNull(activeQueryParser)) { - recordScanProgress(device, measurements); + final Map 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 @@ public void close() { pendingQueryTasks.clear(); } - private void recordScanProgress(final IDeviceID device, final List 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 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 measurements) { - if (Objects.isNull(device) || measurements.isEmpty()) { - return; - } - - fullyEmittedMeasurementsByDevice - .computeIfAbsent(device, key -> new LinkedHashSet<>()) - .addAll(measurements); - } - - private long determineTaskResumeStartTime( - final IDeviceID device, final List 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 tasks, - final IDeviceID device, - final List 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 @@ private void splitOrSkipActiveQueryTask() { } 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 splitTasks = failedTask.split(resumeStartTime); - for (int i = splitTasks.size() - 1; i >= 0; --i) { - pendingQueryTasks.addFirst(splitTasks.get(i)); + final ArrayDeque 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 @@ private List extractMeasurementNames(final Tablet tablet) { .collect(Collectors.toList()); } - private boolean measurementsEqual( - final List leftMeasurements, final List rightMeasurements) { - return leftMeasurements.size() == rightMeasurements.size() - && new LinkedHashSet<>(leftMeasurements).equals(new LinkedHashSet<>(rightMeasurements)); - } - private static class QueryTask { private final IDeviceID device; private final List 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 0cfbbc364f4ca..dc5ab073bcd65 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.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.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 void testFallbackToQueryForRemainingDevicesWhenScanParserHitsCorruption() }); final Optional 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 pointCountByTimeseries = new HashMap<>(); + final LoadTreeStatementDataTypeConvertExecutionVisitor visitor = + new LoadTreeStatementDataTypeConvertExecutionVisitor( + statement -> { + collectLoadedPointsByTimeseries(statement, pointCountByTimeseries); + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + }); + final Optional 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 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 points = new HashMap<>(); + final LoadTreeStatementDataTypeConvertExecutionVisitor visitor = + new LoadTreeStatementDataTypeConvertExecutionVisitor( + statement -> { + collectLoadedPointsByTimeseries(statement, points); + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + }); + final Optional 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 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> points = new HashMap<>(); + final LoadTreeStatementDataTypeConvertExecutionVisitor visitor = + new LoadTreeStatementDataTypeConvertExecutionVisitor( + statement -> { + collectLoadedLongPoints(statement, points); + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + }); + final Optional 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 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 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> points = new HashMap<>(); + final LoadTreeStatementDataTypeConvertExecutionVisitor visitor = + new LoadTreeStatementDataTypeConvertExecutionVisitor( + statement -> { + collectLoadedLongPoints(statement, points); + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + }); + final Optional 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 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 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> points = new HashMap<>(); + final LoadTreeStatementDataTypeConvertExecutionVisitor visitor = + new LoadTreeStatementDataTypeConvertExecutionVisitor( + statement -> { + collectLoadedLongPoints(statement, points); + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + }); + final Optional 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 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 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 void testFallbackToQueryWhenFirstNonAlignedDeviceIsCorrupted() throws Exc 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 void testFallbackDoesNotReloadCompletedMeasurementsOfCurrentNonAlignedDev 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 void testFallbackToQueryForRemainingMeasurementsOfCurrentAlignedDevice() 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 void testFallbackToQueryForRemainingMeasurementsOfCurrentAlignedDevice() 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 @@ private void writeTsFile(final File file) throws Exception { 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 @@ private void writeWideAlignedTsFile( } } + private void writeWideAlignedRow( + final TsFileWriter writer, final List 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 schemaList, @@ -353,6 +658,27 @@ file, new IoTDBPipePattern(null), Long.MIN_VALUE, Long.MAX_VALUE, null, null, tr } } + private void collectLoadedLongPoints( + final Statement statement, final Map> 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 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 pointCountByTimeseries) { Assert.assertTrue(statement instanceof InsertMultiTabletsStatement); From c7d8cab900041611bf39abb5ddc195ab57c0e7de Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:55:33 +0800 Subject: [PATCH 2/4] Adapt LOAD fallback tests to 1.3 scheduling --- .../apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 47 +++++++++++++++---- 1 file changed, 38 insertions(+), 9 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 94f3fb2843151..55019f8f2157b 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,6 +20,7 @@ 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; @@ -811,17 +812,21 @@ public void testDirectoryQueryFallbackRetainsOnlyFailedSources() throws Exceptio 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); + writeFallbackSource(sources[i], devicePrefix + ".d" + i, i == 1, i == 2); 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() @@ -848,8 +853,9 @@ public void testDirectoryQueryFallbackRetainsOnlyFailedSources() throws Exceptio assertFallbackSourceExists(failed, true); replaceByte(failed, compressionOffset, originalCompression); + replaceByte(failed, failedMarkerOffset, originalFailedMarker); statement.execute(fallbackLoadSql(failed, delete)); - assertFallbackRows(statement, devicePrefix + ".d2"); + assertFallbackRows(statement, devicePrefix + ".d2", fallbackPartitionTimestamps()); assertFallbackRows(statement, devicePrefix + ".d0"); assertFallbackRows(statement, devicePrefix + ".d1"); assertFallbackSourceExists(failed, !delete); @@ -865,8 +871,8 @@ public void testConcurrentQueryFallbackFailureRetainsSourceForRetry() throws Exc 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); - writeFallbackSource(bad, device + ".bad", false); + 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); @@ -907,7 +913,7 @@ public void testConcurrentQueryFallbackFailureRetainsSourceForRetry() throws Exc assertFallbackRows(statement, device + ".good"); replaceByte(bad, compressionOffset, originalCompression); statement.execute(fallbackLoadSql(bad, true)); - assertFallbackRows(statement, device + ".bad"); + assertFallbackRows(statement, device + ".bad", fallbackPartitionTimestamps()); assertFallbackRows(statement, device + ".good"); } assertFallbackSourceExists(bad, false); @@ -931,7 +937,11 @@ private static void assertFallbackLoadFailure(final SQLException failure, final } private static long writeFallbackSource( - final File file, final String device, final boolean includeAnchor) throws Exception { + final File file, + final String device, + final boolean includeAnchor, + final boolean spanTimePartitions) + throws Exception { final List schemas = Arrays.asList( new MeasurementSchema( @@ -957,7 +967,11 @@ private static long writeFallbackSource( writer.registerAlignedTimeseries(new Path(device), schemas); final Tablet tablet = new Tablet(device, schemas); for (int i = 0; i < 4; i++) { - tablet.addTimestamp(i, i + 1); + 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); } @@ -967,7 +981,11 @@ private static long writeFallbackSource( final TsFileResource resource = new TsFileResource(file); final IDeviceID deviceId = new PlainDeviceID(device); resource.updateStartTime(deviceId, 1); - resource.updateEndTime(deviceId, 4); + resource.updateEndTime( + deviceId, + spanTimePartitions + ? CommonDescriptor.getInstance().getConfig().getTimePartitionInterval() + 2 + : 4); if (includeAnchor) { final IDeviceID anchorId = new PlainDeviceID(anchor); resource.updateStartTime(anchorId, 0); @@ -1012,10 +1030,21 @@ private static void assertFallbackSourceExists(final File file, final boolean ex 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(i + 1, rows.getLong(1)); + Assert.assertEquals(timestamps[i], rows.getLong(1)); if (i == 1) { Assert.assertNull(rows.getObject(2)); } else { From 5d75e511c0389a3caf76dc6059598e3a1000decf Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 2 Nov 2026 19:09:07 +0800 Subject: [PATCH 3/4] Check all DataNode logs in LOAD fallback IT --- .../org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 13 +++++++------ 1 file changed, 7 insertions(+), 6 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 55019f8f2157b..e7a902d58260a 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 @@ -842,12 +842,13 @@ public void testDirectoryQueryFallbackRetainsOnlyFailedSources() throws Exceptio assertFallbackRows(statement, devicePrefix + ".d0"); assertFallbackRows(statement, devicePrefix + ".d1"); Assert.assertTrue( - EnvFactory.getEnv() - .getDataNodeWrapper(0) - .logContains( - "Load: Successfully converted TsFile " - + recoverable.getAbsolutePath() - + " into tablets and inserted.")); + EnvFactory.getEnv().getDataNodeWrapperList().stream() + .anyMatch( + dataNodeWrapper -> + dataNodeWrapper.logContains( + "Load: Successfully converted TsFile " + + recoverable.getAbsolutePath() + + " into tablets and inserted."))); assertFallbackSourceExists(normal, !delete); assertFallbackSourceExists(recoverable, !delete); assertFallbackSourceExists(failed, true); From a3f10d79be1845ce6a252547b75e16be3bc17002 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 2 Nov 2026 19:47:11 +0800 Subject: [PATCH 4/4] Exercise LOAD query fallback in single-node tests --- .../apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 26 +++++++++++-------- 1 file changed, 15 insertions(+), 11 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 e7a902d58260a..61fe99e9ea139 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 @@ -815,7 +815,7 @@ public void testDirectoryQueryFallbackRetainsOnlyFailedSources() throws Exceptio long failedMarkerOffset = -1; for (int i = 0; i < sources.length; i++) { final long groupMarkerOffset = - writeFallbackSource(sources[i], devicePrefix + ".d" + i, i == 1, i == 2); + writeFallbackSource(sources[i], devicePrefix + ".d" + i, i == 1, i > 0); if (i == 1) { recoverableMarkerOffset = groupMarkerOffset; } else if (i == 2) { @@ -840,15 +840,19 @@ public void testDirectoryQueryFallbackRetainsOnlyFailedSources() throws Exceptio Assert.assertFalse( failure.getMessage(), failure.getMessage().contains(recoverable.getName())); assertFallbackRows(statement, devicePrefix + ".d0"); - assertFallbackRows(statement, devicePrefix + ".d1"); - Assert.assertTrue( - EnvFactory.getEnv().getDataNodeWrapperList().stream() - .anyMatch( - dataNodeWrapper -> - dataNodeWrapper.logContains( - "Load: Successfully converted TsFile " - + recoverable.getAbsolutePath() - + " into tablets and inserted."))); + 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); @@ -858,7 +862,7 @@ public void testDirectoryQueryFallbackRetainsOnlyFailedSources() throws Exceptio statement.execute(fallbackLoadSql(failed, delete)); assertFallbackRows(statement, devicePrefix + ".d2", fallbackPartitionTimestamps()); assertFallbackRows(statement, devicePrefix + ".d0"); - assertFallbackRows(statement, devicePrefix + ".d1"); + assertFallbackRows(statement, devicePrefix + ".d1", fallbackPartitionTimestamps()); assertFallbackSourceExists(failed, !delete); } } finally {