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 6b51c1ae06df3..dadd5a7b4bae8 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.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.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 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.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 void testLoadWithTwoDataNodesDownFailsWithAMeaningfulMessage() throws Exc } } + @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> 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 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 @@ private static void waitUntilClusterAnswers() { 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 7bfbb9b6df3e9..53c26f77882c6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java @@ -36,6 +36,7 @@ import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.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.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 IDeviceID 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( StorageEngineMessages @@ -264,8 +252,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()) { @@ -273,23 +260,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; @@ -299,49 +273,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() { @@ -399,7 +389,7 @@ public Pair next() { activeQueryTask.measurements, file.getAbsolutePath(), e); - splitOrSkipActiveQueryTask(); + splitOrSkipActiveQueryTask(e); } } @@ -414,15 +404,12 @@ private void recordProgress(final Pair tabletWithIsAligned) { } final IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create(tablet.getDeviceId()); - 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.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 @@ 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)) { @@ -549,15 +484,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()); } } @@ -565,12 +505,6 @@ private List extractMeasurementNames(final Tablet tablet) { return IMeasurementSchema.getMeasurementNameList(tablet.getSchemas()); } - 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 b374880d6a859..2ee8024e2f5a2 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.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.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 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 = 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 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 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 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> 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.flush(); + 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(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> 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(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 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); @@ -215,7 +504,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( @@ -247,7 +536,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); @@ -280,7 +569,7 @@ public void testPipeOutOfMemoryIsTemporaryUnavailable() throws Exception { 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 @@ 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.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 @@ 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(0, measurement, measurement * 10_000L + timestamp); + } + writer.writeAligned(tablet); + } + private void writeDevice( final TsFileWriter writer, final List schemaList, @@ -394,6 +698,27 @@ file, new IoTDBTreePattern(null), Long.MIN_VALUE, Long.MAX_VALUE, null, null, tr } } + private static 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 (!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 pointCountByTimeseries) { Assert.assertTrue(statement instanceof InsertMultiTabletsStatement);