This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_flat_device in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b438cfaa687c622c9c9f0b3c11e4e3fbbc47f07e Author: Tian Jiang <[email protected]> AuthorDate: Thu Oct 15 12:10:13 2026 +0800 multiple fixes --- .../recent/IoTDBReadTsFileTableFunctionIT.java | 98 +++++++++++++++++++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 2 + .../iotdb/db/i18n/DataNodeQueryMessages.java | 2 + .../relational/ExternalTsFileSeriesScanUtil.java | 30 +++++- .../ExternalTsFileTableScanOperator.java | 6 ++ .../planner/DataNodeTableOperatorGenerator.java | 23 +++++ .../read_tsfile/ExternalTsFileQueryResource.java | 15 +++ .../tvf/read_tsfile/ReadTsFileTableFunction.java | 6 ++ .../tvf/read_tsfile/TsFileSchemaCollector.java | 22 ++++- .../plan/relational/metadata/FlatDeviceEntry.java | 14 ++- .../plan/relational/planner/RelationPlanner.java | 14 ++- .../planner/node/ExternalTsFileScanNode.java | 5 + .../writer/AbstractInnerCompactionWriter.java | 4 + .../file/UnsealedTsFileRecoverPerformer.java | 104 +++++++++++++++++++++ .../utils/writer/AbstractCompactionWriterTest.java | 89 ++++++++++++++++++ .../wal/recover/file/FlatDeviceRecoveryTest.java | 49 ++++++++++ 16 files changed, 476 insertions(+), 7 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBReadTsFileTableFunctionIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBReadTsFileTableFunctionIT.java index 33c562b3c6d..78abc3e08f0 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBReadTsFileTableFunctionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBReadTsFileTableFunctionIT.java @@ -19,6 +19,9 @@ package org.apache.iotdb.relational.it.query.recent; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter; +import org.apache.iotdb.db.utils.EncryptDBUtils; import org.apache.iotdb.it.env.EnvFactory; import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; import org.apache.iotdb.it.framework.IoTDBTestRunner; @@ -26,12 +29,18 @@ import org.apache.iotdb.itbase.category.TableClusterIT; import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; import org.apache.iotdb.itbase.env.BaseEnv; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.exception.write.WriteProcessException; +import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.utils.Binary; 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.apache.tsfile.write.writer.TsFileIOWriter; import org.junit.After; import org.junit.AfterClass; import org.junit.Before; @@ -86,6 +95,95 @@ public class IoTDBReadTsFileTableFunctionIT { EnvFactory.getEnv().cleanClusterEnvironment(); } + // Two external flat files contain five logical devices, a null TAG, equal timestamps on + // different devices, a duplicate TAG/time key across files, and time regression between devices. + // TAG filters, key-preserving projection pruning, counts and ORDER BY/LIMIT must operate on rows + // rather than the physical table ID; merge the duplicate without losing equal-time other devices. + @Test + public void testReadFlatTsFiles() throws Exception { + File first = new File(tmpDir, "flat-first.tsfile"); + File second = new File(tmpDir, "flat-second.tsfile"); + writeFlatTable(first, new String[] {null, "a", "b"}, new long[] {5, 100, 10}); + writeFlatTable(second, new String[] {"a", "c", "z"}, new long[] {100, 10, 1}); + String source = + " FROM read_tsfile(PATHS => '" + toSqlPath(tmpDir) + "', TABLE_NAME => 'table1')"; + tableResultSetEqualTest( + "SELECT tag1, s1" + source + " WHERE tag1 = 'b'", + new String[] {"tag1", "s1"}, + new String[] {"b,10,"}, + DATABASE_NAME); + tableResultSetEqualTest( + "SELECT s1" + source + " WHERE tag1 IS NULL", + new String[] {"s1"}, + new String[] {"5,"}, + DATABASE_NAME); + tableResultSetEqualTest( + "SELECT count(*) AS row_count, count(s1) AS values_count," + + " count(DISTINCT tag1) AS devices" + + source, + new String[] {"row_count", "values_count", "devices"}, + new String[] {"5,5,4,"}, + DATABASE_NAME); + tableResultSetEqualTest( + "SELECT count(s1) AS cnt" + source + " WHERE time < 11", + new String[] {"cnt"}, + new String[] {"4,"}, + DATABASE_NAME); + tableResultSetEqualTest( + "SELECT count(*) AS cnt" + source, + new String[] {"cnt"}, + new String[] {"5,"}, + DATABASE_NAME); + tableResultSetEqualTest( + "SELECT tag1, s1" + source + " ORDER BY time LIMIT 2", + new String[] {"tag1", "s1"}, + new String[] {"z,1,", "null,5,"}, + DATABASE_NAME); + } + + // The same table appears once with TAGs in device IDs and once with TAGs in chunks. Reject the + // incompatible physical projections explicitly instead of silently dropping or corrupting rows. + @Test + public void testRejectMixedFlatAndOrdinaryTsFiles() throws Exception { + writeFlatTable(new File(tmpDir, "flat.tsfile"), new String[] {"a"}, new long[] {1}); + try (TsFileWriter writer = new TsFileWriter(new File(tmpDir, "ordinary.tsfile"))) { + generateTable(writer, "table1", List.of("tag1"), List.of("s1"), 1, 1); + } + tableAssertTestFail( + "SELECT * FROM read_tsfile(PATHS => '" + toSqlPath(tmpDir) + "', TABLE_NAME => 'table1')", + "read_tsfile cannot combine ordinary and flat_device layouts for the same table", + DATABASE_NAME); + } + + private static void writeFlatTable(File file, String[] tags, long[] times) throws Exception { + List<IMeasurementSchema> schemas = + List.of( + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("s1", TSDataType.INT64)); + try (TsFileIOWriter writer = new TsFileIOWriter(file)) { + writer + .getSchema() + .registerTableSchema( + new TableSchema( + "table1", schemas, List.of(ColumnCategory.TAG, ColumnCategory.FIELD))); + TsFileTableSchemaUtil.addFlatDeviceProperty(writer, "table1", true); + writer.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1"})); + FlatDeviceAlignedChunkWriter chunk = + new FlatDeviceAlignedChunkWriter(schemas, EncryptDBUtils.getDefaultFirstEncryptParam()); + for (int rowIndex = 0; rowIndex < tags.length; rowIndex++) { + chunk.write( + times[rowIndex], + tags[rowIndex] == null ? null : new Binary(tags[rowIndex], TSFileConfig.STRING_CHARSET), + tags[rowIndex] == null); + chunk.write(times[rowIndex], times[rowIndex], false); + chunk.write(times[rowIndex]); + } + chunk.writeToFileWriter(writer); + writer.endChunkGroup(); + writer.endFile(); + } + } + @Test public void testReadSingleTsFile() throws Exception { File tsFile = new File(tmpDir, "single.tsfile"); diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index c029f31653e..c8e8acf39a2 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -20,6 +20,8 @@ package org.apache.iotdb.db.i18n; public final class DataNodeQueryMessages { + public static final String EXCEPTION_READ_TSFILE_CANNOT_COMBINE_ORDINARY_AND_FLAT_DEVICE_LAYOUTS_FOR_THE_SAME_TABLE_2BF02E80 = + "read_tsfile cannot combine ordinary and flat_device layouts for the same table"; // --- Common --- diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index 4280f43c83d..191ed716403 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -20,6 +20,8 @@ package org.apache.iotdb.db.i18n; public final class DataNodeQueryMessages { + public static final String EXCEPTION_READ_TSFILE_CANNOT_COMBINE_ORDINARY_AND_FLAT_DEVICE_LAYOUTS_FOR_THE_SAME_TABLE_2BF02E80 = + "read_tsfile 不能同时读取同一张表的普通布局和 flat_device 布局"; // --- Common --- diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileSeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileSeriesScanUtil.java index e907218fe8a..94bc178e44b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileSeriesScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileSeriesScanUtil.java @@ -35,6 +35,7 @@ import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.read.filter.basic.Filter; import java.io.IOException; +import java.util.Collections; import java.util.List; public class ExternalTsFileSeriesScanUtil extends AlignedSeriesScanUtil { @@ -49,7 +50,34 @@ public class ExternalTsFileSeriesScanUtil extends AlignedSeriesScanUtil { boolean queryAllSensors, List<TSDataType> givenDataTypes, ExternalTsFileMetadataLoader metadataLoader) { - super(seriesPath, scanOrder, scanOptions, context, queryAllSensors, givenDataTypes); + this( + seriesPath, + scanOrder, + scanOptions, + context, + queryAllSensors, + givenDataTypes, + Collections.emptyList(), + metadataLoader); + } + + public ExternalTsFileSeriesScanUtil( + AlignedFullPath seriesPath, + Ordering scanOrder, + SeriesScanOptions scanOptions, + FragmentInstanceContext context, + boolean queryAllSensors, + List<TSDataType> givenDataTypes, + List<Integer> rowKeyColumnIndices, + ExternalTsFileMetadataLoader metadataLoader) { + super( + seriesPath, + scanOrder, + scanOptions, + context, + queryAllSensors, + givenDataTypes, + rowKeyColumnIndices); this.metadataLoader = metadataLoader; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileTableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileTableScanOperator.java index a38fcacd046..8079cbd612f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileTableScanOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileTableScanOperator.java @@ -27,6 +27,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.function.tvf.read_tsfile. import org.apache.iotdb.db.queryengine.plan.relational.function.tvf.read_tsfile.ExternalTsFileQueryResource; import org.apache.iotdb.db.queryengine.plan.relational.function.tvf.read_tsfile.ExternalTsFileQueryResource.DeviceTaskRunReader; import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.FlatDeviceEntry; import org.apache.iotdb.db.storageengine.dataregion.read.IQueryDataSource; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -35,6 +36,8 @@ import org.apache.tsfile.file.metadata.AbstractAlignedTimeSeriesMetadata; import org.apache.tsfile.utils.RamUsageEstimator; import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; public class ExternalTsFileTableScanOperator extends TableScanOperator { private static final long INSTANCE_SIZE = @@ -70,6 +73,9 @@ public class ExternalTsFileTableScanOperator extends TableScanOperator { ((OperatorContext) operatorContext).getInstanceContext(), true, measurementColumnTSDataTypes, + deviceEntry instanceof FlatDeviceEntry flatEntry + ? Arrays.stream(flatEntry.getTagMeasurementColumnIndices()).boxed().toList() + : Collections.emptyList(), this::loadTimeSeriesMetadata); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java index 1909cc21c95..5a3ac272e31 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java @@ -1109,6 +1109,15 @@ public class DataNodeTableOperatorGenerator if (node.getDeviceEntries().isEmpty() || !node.isFlatDevice()) { return node.getDeviceEntries(); } + if (node instanceof ExternalTsFileScanNode externalNode) { + int[] tagIndices = + externalNode.getExternalTsFileQueryResource().getFlatDeviceTagColumnNames().stream() + .mapToInt(measurementColumnNames::indexOf) + .toArray(); + return node.getDeviceEntries().stream() + .map(entry -> (DeviceEntry) new FlatDeviceEntry(entry.getDeviceID(), tagIndices)) + .collect(Collectors.toList()); + } final QualifiedObjectName qualifiedObjectName = node.getQualifiedObjectName(); final TsTable table = DataNodeTableCache.getInstance() @@ -1160,6 +1169,20 @@ public class DataNodeTableOperatorGenerator if (!node.isFlatDevice()) { return; } + if (node instanceof ExternalTsFileScanNode externalNode) { + // Projection pruning may remove every TAG (for example SELECT count(s)). Keep complete + // row keys for merging, obtaining their schema from the external file rather than cache. + for (String tagName : + externalNode.getExternalTsFileQueryResource().getFlatDeviceTagColumnNames()) { + if (!parameter.measurementColumnNames.contains(tagName)) { + parameter.measurementColumnsIndexMap.put( + tagName, parameter.measurementColumnNames.size()); + parameter.measurementColumnNames.add(tagName); + parameter.measurementSchemas.add(new MeasurementSchema(tagName, TSDataType.STRING)); + } + } + return; + } final QualifiedObjectName qualifiedObjectName = node.getQualifiedObjectName(); final TsTable table = DataNodeTableCache.getInstance() diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java index 3333d99e933..03b069bb333 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java @@ -94,6 +94,21 @@ public class ExternalTsFileQueryResource { private final List<DeviceEntry> sharedDeviceEntries = new ArrayList<>(); private final List<DeviceTaskPartition> deviceTaskPartitions = new ArrayList<>(); private Comparator<DeviceEntry> deviceEntryComparator; + private boolean flatDevice; + private List<String> flatDeviceTagColumnNames = Collections.emptyList(); + + public void setFlatDevice(boolean flatDevice, List<String> tagColumnNames) { + this.flatDevice = flatDevice; + this.flatDeviceTagColumnNames = List.copyOf(tagColumnNames); + } + + public boolean isFlatDevice() { + return flatDevice; + } + + public List<String> getFlatDeviceTagColumnNames() { + return flatDeviceTagColumnNames; + } // Counts FragmentInstances that have entered runtime datasource initialization and may read // files under queryTempRoot. QueryExecution cleanup only closes this resource when the count is diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ReadTsFileTableFunction.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ReadTsFileTableFunction.java index 71ee6aa267f..f0fa2ab321b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ReadTsFileTableFunction.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ReadTsFileTableFunction.java @@ -120,6 +120,7 @@ public class ReadTsFileTableFunction implements TableFunction { .collect(Collectors.toList()), outputSchema, deviceMetadataInfoSwapThreshold); + handle.flatDevice = schemaCollector.isFlatDevice(); return TableFunctionAnalysis.builder().properColumnSchema(outputSchema).handle(handle).build(); } @@ -243,6 +244,11 @@ public class ReadTsFileTableFunction implements TableFunction { private List<Type> outputColumnTypes; private List<TsTableColumnCategory> outputColumnCategories; private long deviceMetadataInfoSwapThreshold; + private boolean flatDevice; + + public boolean isFlatDevice() { + return flatDevice; + } public ReadTsFileTableFunctionHandle() { this( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/TsFileSchemaCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/TsFileSchemaCollector.java index 609581ebbd6..05fae712a6b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/TsFileSchemaCollector.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/TsFileSchemaCollector.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.queryengine.plan.relational.function.tvf.read_tsfile; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.udf.api.exception.UDFArgumentNotValidException; @@ -53,6 +54,11 @@ final class TsFileSchemaCollector { private final List<File> tsFiles = new ArrayList<>(); private MergedTableSchemaBuilder schemaBuilder; private TableSchema mergedTableSchema; + private Boolean flatDevice; + + boolean isFlatDevice() { + return Boolean.TRUE.equals(flatDevice); + } TsFileSchemaCollector(String specifiedTableName, MPPQueryContext mppQueryContext) { this.specifiedTableName = @@ -176,7 +182,21 @@ final class TsFileSchemaCollector { private TableSchema readTableSchemaFromValidTsFile( String specifiedTableName, File tsFile, TsFileSequenceReader reader) { try { - return selectTableSchema(specifiedTableName, tsFile, reader.getTableSchemaMap()); + TableSchema schema = + selectTableSchema(specifiedTableName, tsFile, reader.getTableSchemaMap()); + if (schema != null) { + boolean currentFlatDevice = + TsFileTableSchemaUtil.isFlatDevice(reader.getTsFileProperties(), schema.getTableName()); + // Ordinary TAGs live in device IDs, while flat TAGs live in chunks. One scan cannot + // interpret both layouts using the same physical projection. + if (flatDevice != null && flatDevice != currentFlatDevice) { + throw new UDFArgumentNotValidException( + DataNodeQueryMessages + .EXCEPTION_READ_TSFILE_CANNOT_COMBINE_ORDINARY_AND_FLAT_DEVICE_LAYOUTS_FOR_THE_SAME_TABLE_2BF02E80); + } + flatDevice = currentFlatDevice; + } + return schema; } catch (UDFArgumentNotValidException e) { throw e; } catch (Exception e) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/FlatDeviceEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/FlatDeviceEntry.java index 66777fc47d0..d18dca99ff0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/FlatDeviceEntry.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/FlatDeviceEntry.java @@ -35,6 +35,14 @@ public class FlatDeviceEntry extends AlignedDeviceEntry { private final Map<IDeviceID, Binary[]> logicalDeviceAttributes; private final int[] tagMeasurementColumnIndices; + // External files have no schema-region device whitelist or ATTRIBUTE values. All logical + // devices in their rows are visible; only the TAG indices are needed for compound-key merging. + public FlatDeviceEntry(IDeviceID physicalDeviceID, int[] tagMeasurementColumnIndices) { + super(physicalDeviceID, new Binary[0]); + this.logicalDeviceAttributes = null; + this.tagMeasurementColumnIndices = tagMeasurementColumnIndices.clone(); + } + public FlatDeviceEntry( IDeviceID physicalDeviceID, List<DeviceEntry> logicalDeviceEntries, @@ -49,6 +57,9 @@ public class FlatDeviceEntry extends AlignedDeviceEntry { } public Binary getAttributeValue(TsBlock measurementDataBlock, int rowIndex, int attributeIndex) { + if (logicalDeviceAttributes == null) { + return null; + } IDeviceID logicalDeviceID = getLogicalDeviceID(measurementDataBlock, rowIndex); Binary[] attributes = logicalDeviceAttributes.get(logicalDeviceID); return attributes == null || attributeIndex >= attributes.length @@ -57,7 +68,8 @@ public class FlatDeviceEntry extends AlignedDeviceEntry { } public boolean containsLogicalDevice(TsBlock measurementDataBlock, int rowIndex) { - return logicalDeviceAttributes.containsKey(getLogicalDeviceID(measurementDataBlock, rowIndex)); + return logicalDeviceAttributes == null + || logicalDeviceAttributes.containsKey(getLogicalDeviceID(measurementDataBlock, rowIndex)); } public Column getTagColumn(TsBlock measurementDataBlock, int tagIndex) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java index 224c6725d7e..aa6011a35a5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java @@ -1635,25 +1635,28 @@ public class RelationPlanner implements AstVisitor<RelationPlan, Void> { private RelationPlan planExternalTsFileScan( TableFunctionInvocation node, TableFunctionInvocationAnalysis functionAnalysis) { if (!(functionAnalysis.getTableFunctionHandle() - instanceof ReadTsFileTableFunction.ReadTsFileTableFunctionHandle)) { + instanceof ReadTsFileTableFunction.ReadTsFileTableFunctionHandle handle)) { throw new IllegalStateException( DataNodeQueryMessages.READ_TSFILE_TABLE_FUNCTION_HANDLE_IS_INVALID); } - ReadTsFileTableFunction.ReadTsFileTableFunctionHandle handle = - (ReadTsFileTableFunction.ReadTsFileTableFunctionHandle) - functionAnalysis.getTableFunctionHandle(); Scope scope = analysis.getScope(node); RelationType relationType = scope.getRelationType(); ImmutableList.Builder<Symbol> outputSymbolsBuilder = ImmutableList.builder(); ImmutableMap.Builder<Symbol, ColumnSchema> assignmentsBuilder = ImmutableMap.builder(); Map<Symbol, Integer> tagAndAttributeIndexMap = new HashMap<>(); + List<String> flatTagColumnNames = new ArrayList<>(); int tagIndex = 0; for (int i = 0; i < relationType.getAllFieldCount(); i++) { Field field = relationType.getFieldByIndex(i); Symbol symbol = symbolAllocator.newSymbol(field); TsTableColumnCategory columnCategory = handle.getOutputColumnCategories().get(i); + if (handle.isFlatDevice() && columnCategory == TsTableColumnCategory.TAG) { + // Flat TAG predicates must be evaluated against row values, never physical device IDs. + flatTagColumnNames.add(field.getName().orElseThrow()); + columnCategory = TsTableColumnCategory.FIELD; + } outputSymbolsBuilder.add(symbol); assignmentsBuilder.put( symbol, @@ -1681,6 +1684,9 @@ public class RelationPlanner implements AstVisitor<RelationPlan, Void> { handle.getTsFilePaths(), assignments, handle.getDeviceMetadataInfoSwapThreshold())); + scanNode + .getExternalTsFileQueryResource() + .setFlatDevice(handle.isFlatDevice(), flatTagColumnNames); return new RelationPlan(scanNode, scope, outputSymbols, outerContext); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/ExternalTsFileScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/ExternalTsFileScanNode.java index 160d54351e6..66185fdbbb0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/ExternalTsFileScanNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/ExternalTsFileScanNode.java @@ -49,6 +49,11 @@ public class ExternalTsFileScanNode extends DeviceTableScanNode { protected ExternalTsFileScanNode() {} + @Override + public boolean isFlatDevice() { + return externalTsFileQueryResource != null && externalTsFileQueryResource.isFlatDevice(); + } + public ExternalTsFileScanNode( PlanNodeId id, QualifiedObjectName qualifiedObjectName, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java index c94ea1a7c27..44f7fa6b249 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java @@ -104,7 +104,11 @@ public abstract class AbstractInnerCompactionWriter extends AbstractCompactionWr } private void rollCompactionFileWriter() throws IOException { + // Footer statistics belong to this output file, not to the entire compaction task. + writeFlatDeviceStatistics(fileWriter); fileWriter.endFile(); + flatDeviceLogicalDevices.clear(); + flatDeviceRowCounts.clear(); endedFileSize += fileWriter.getFile().length(); if (fileWriter.isEmptyTargetFile()) { targetResources.get(currentFileIndex).forceMarkDeleted(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java index 34429520a88..910dcce8f9b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java @@ -48,10 +48,19 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeInd import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry; import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALRecoverException; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALRecoverListener; +import org.apache.iotdb.db.utils.EncryptDBUtils; +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.ChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.Chunk; import org.apache.tsfile.read.common.TimeRange; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.reader.chunk.TableChunkReader; import org.apache.tsfile.write.writer.RestorableTsFileIOWriter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -59,9 +68,12 @@ import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.function.Consumer; @@ -337,6 +349,7 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform // if we put following codes in the 'if' clause above, this file can be continued writing // into it // currently, we close this file anyway + restoreFlatDeviceStatistics(); writer.endFile(); tsFileResource.serialize(); FileTimeIndexCacheRecorder.getInstance().logFileTimeIndex(tsFileResource); @@ -356,6 +369,97 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform return tsFileResource; } + private void restoreFlatDeviceStatistics() throws IOException { + if (!isTableModel) { + return; + } + writer.flush(); + // The footer was lost in the crash. Rebuild statistics from all surviving chunks, including + // the replay flush; counting only the recovery MemTable would omit completed flushes. + try (TsFileSequenceReader reader = + new TsFileSequenceReader( + tsFileResource.getTsFilePath(), + EncryptDBUtils.getFirstEncryptParamFromTSFilePath(tsFileResource.getTsFilePath()), + false)) { + for (var entry : writer.getDeviceChunkMetadataMap().entrySet()) { + String tableName = entry.getKey().getTableName(); + TsTable table = DataNodeTableCache.getInstance().getTable(databaseName, tableName); + if (table == null || !table.isFlatDevice()) { + continue; + } + TableSchema schema = writer.getSchema().getTableSchemaMap().get(tableName); + List<String> tagNames = new ArrayList<>(); + for (int columnIndex = 0; columnIndex < schema.getColumnTypes().size(); columnIndex++) { + if (schema.getColumnTypes().get(columnIndex) == ColumnCategory.TAG) { + tagNames.add(schema.getColumnSchemas().get(columnIndex).getMeasurementName()); + } + } + Set<IDeviceID> devices = new HashSet<>(); + long rowCount = 0; + ChunkMetadata timeChunk = null; + Map<String, ChunkMetadata> tagChunks = new HashMap<>(); + for (ChunkMetadata metadata : entry.getValue()) { + if (metadata.getMeasurementUid().isEmpty()) { + if (timeChunk != null) { + collectFlatDevices(reader, tableName, tagNames, timeChunk, tagChunks, devices); + } + timeChunk = metadata; + rowCount += metadata.getNumOfPoints(); + tagChunks.clear(); + } else if (tagNames.contains(metadata.getMeasurementUid())) { + tagChunks.put(metadata.getMeasurementUid(), metadata); + } + } + if (timeChunk != null) { + collectFlatDevices(reader, tableName, tagNames, timeChunk, tagChunks, devices); + } + TsFileTableSchemaUtil.addFlatDeviceStatistics(writer, tableName, devices.size(), rowCount); + } + } + } + + private void collectFlatDevices( + TsFileSequenceReader reader, + String tableName, + List<String> tagNames, + ChunkMetadata timeChunk, + Map<String, ChunkMetadata> tagChunks, + Set<IDeviceID> devices) + throws IOException { + if (tagNames.isEmpty()) { + if (timeChunk.getNumOfPoints() > 0) { + devices.add(IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {tableName})); + } + return; + } + // Decode only TAGs and time. FIELD payloads and mods do not affect physical row statistics. + List<Chunk> values = new ArrayList<>(tagNames.size()); + for (String tagName : tagNames) { + ChunkMetadata metadata = tagChunks.get(tagName); + values.add(metadata == null ? null : reader.readMemChunk(metadata)); + } + TableChunkReader chunkReader = + new TableChunkReader(reader.readMemChunk(timeChunk), values, null); + for (var page : chunkReader.loadPageReaderList()) { + page.initTsBlockBuilder(Collections.nCopies(tagNames.size(), TSDataType.STRING)); + TsBlock block = page.getAllSatisfiedData(); + for (int rowIndex = 0; rowIndex < block.getPositionCount(); rowIndex++) { + String[] segments = new String[tagNames.size() + 1]; + segments[0] = tableName; + for (int tagColumnIndex = 0; tagColumnIndex < tagNames.size(); tagColumnIndex++) { + if (!block.getColumn(tagColumnIndex).isNull(rowIndex)) { + segments[tagColumnIndex + 1] = + block + .getColumn(tagColumnIndex) + .getBinary(rowIndex) + .getStringValue(TSFileConfig.STRING_CHARSET); + } + } + devices.add(IDeviceID.Factory.DEFAULT_FACTORY.create(segments)); + } + } + } + public RestorableTsFileIOWriter getWriter() { return writer; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java index 1e3a2827a8b..464e71f21b9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java @@ -19,30 +19,43 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionTableSchema; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.AlignedPageElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.ChunkMetadataElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer.flushcontroller.AbstractCompactionFlushController; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ArrayDeviceTimeIndex; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndex; +import org.apache.iotdb.db.utils.EncryptDBUtils; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.exception.write.PageException; import org.apache.tsfile.file.header.PageHeader; import org.apache.tsfile.file.metadata.ChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.file.metadata.TimeseriesMetadata; import org.apache.tsfile.file.metadata.statistics.Statistics; import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.TsFileSequenceReader; import org.apache.tsfile.read.common.Chunk; import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.common.block.TsBlockBuilder; +import org.apache.tsfile.utils.Binary; import org.apache.tsfile.write.chunk.ChunkWriterImpl; +import org.apache.tsfile.write.schema.IMeasurementSchema; import org.apache.tsfile.write.schema.MeasurementSchema; import org.apache.tsfile.write.schema.Schema; import org.junit.Assert; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; import java.io.IOException; import java.nio.ByteBuffer; @@ -60,6 +73,82 @@ import static org.mockito.Mockito.when; public class AbstractCompactionWriterTest { private static final int SUB_TASK_ID = 0; + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + // Force rollover between flat tables a and b with two output files. Each footer must contain + // exactly its own table's one logical device and one row, including the file closed by rollover. + @Test + public void testInnerRolloverKeepsFlatStatisticsPerFile() throws Exception { + long previousLimit = IoTDBDescriptor.getInstance().getConfig().getTargetCompactionFileSize(); + List<TsFileResource> targets = + List.of( + new TsFileResource( + temporaryFolder.getRoot().toPath().resolve("1-1-0-0.tsfile").toFile()), + new TsFileResource( + temporaryFolder.getRoot().toPath().resolve("2-1-0-0.tsfile").toFile())); + List<IMeasurementSchema> measurements = + List.of( + new MeasurementSchema("tag", TSDataType.STRING), + new MeasurementSchema("s", TSDataType.INT64)); + Schema schema = new Schema(); + for (String table : List.of("a", "b")) { + CompactionTableSchema tableSchema = new CompactionTableSchema(table); + tableSchema.merge( + new TableSchema(table, measurements, List.of(ColumnCategory.TAG, ColumnCategory.FIELD))); + tableSchema.setFlatDevice(true); + schema.registerTableSchema(tableSchema); + } + IoTDBDescriptor.getInstance().getConfig().setTargetCompactionFileSize(1); + try (ReadPointInnerCompactionWriter writer = new ReadPointInnerCompactionWriter(targets)) { + writer.setSchemaForAllTargetFile(List.of(schema)); + for (String table : List.of("a", "b")) { + writer.startChunkGroup( + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {table}), true, true); + writer.setRowKeyColumnIndices(List.of(0)); + writer.startMeasurement( + "", + new FlatDeviceAlignedChunkWriter( + measurements, EncryptDBUtils.getDefaultFirstEncryptParam()), + 0); + TsBlockBuilder block = new TsBlockBuilder(List.of(TSDataType.STRING, TSDataType.INT64)); + block.getTimeColumnBuilder().writeLong(1); + block.getColumnBuilder(0).writeBinary(new Binary("x", TSFileConfig.STRING_CHARSET)); + block.getColumnBuilder(1).writeLong(1); + block.declarePosition(); + writer.write(block.build(), 0); + writer.endMeasurement(0); + writer.endChunkGroup(); + } + writer.endFile(); + } finally { + IoTDBDescriptor.getInstance().getConfig().setTargetCompactionFileSize(previousLimit); + } + for (int fileIndex = 0; fileIndex < targets.size(); fileIndex++) { + String table = fileIndex == 0 ? "a" : "b"; + String otherTable = fileIndex == 0 ? "b" : "a"; + try (TsFileSequenceReader reader = + new TsFileSequenceReader(targets.get(fileIndex).getTsFilePath())) { + Map<String, byte[]> properties = reader.getTsFileProperties(); + Assert.assertEquals( + 1, + ByteBuffer.wrap( + properties.get( + TsFileTableSchemaUtil.getFlatDeviceLogicalDeviceCountPropertyKey(table))) + .getLong()); + Assert.assertEquals( + 1, + ByteBuffer.wrap( + properties.get(TsFileTableSchemaUtil.getFlatDevicePointCountPropertyKey(table))) + .getLong()); + Assert.assertFalse( + properties.containsKey( + TsFileTableSchemaUtil.getFlatDeviceLogicalDeviceCountPropertyKey(otherTable))); + Assert.assertFalse( + properties.containsKey( + TsFileTableSchemaUtil.getFlatDevicePointCountPropertyKey(otherTable))); + } + } + } // Use the same single-segment ID for a flat table and an ordinary all-null-TAG device. // The explicit flat flag selects table/file bounds; ordinary mode reads actual device metadata diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/FlatDeviceRecoveryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/FlatDeviceRecoveryTest.java index 437890c7a44..77062b8121d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/FlatDeviceRecoveryTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/FlatDeviceRecoveryTest.java @@ -70,6 +70,7 @@ import org.junit.Before; import org.junit.Test; import java.io.File; +import java.nio.ByteBuffer; import java.nio.file.Files; import java.util.ArrayList; import java.util.Arrays; @@ -330,6 +331,54 @@ public class FlatDeviceRecoveryTest { resource.updateTableEndTime(TABLE, 100); } + // A crash loses the footer after a completed three-row A/B flush. With no remaining WAL, + // recovery must restore two logical devices and three physical rows in the replacement footer. + @Test + public void testRecoveryRestoresStatisticsWithoutWal() throws Exception { + writeUnsealedFile(); + try (UnsealedTsFileRecoverPerformer performer = + new UnsealedTsFileRecoverPerformer(resource, false, ignored -> {})) { + performer.startRecovery(); + performer.endRecovery(); + } + assertFlatStatistics(2, 3); + } + + // Replay A's duplicate key, a new B row and a null-TAG row after a three-row A/B flush. + // Count six physical rows across flushes, while deduplicating devices to A, B and null TAG. + @Test + public void testRecoveryStatisticsIncludeFlushedAndReplayedRows() throws Exception { + writeUnsealedFile(); + try (UnsealedTsFileRecoverPerformer performer = + new UnsealedTsFileRecoverPerformer(resource, false, ignored -> {})) { + performer.startRecovery(); + performer.redoLog(new WALInfoEntry(1, row("A", 100, 3))); + performer.redoLog(new WALInfoEntry(1, row("B", 50, 4))); + performer.redoLog(new WALInfoEntry(1, row(null, 60, 5))); + performer.endRecovery(); + } + assertFlatStatistics(3, 6); + } + + private void assertFlatStatistics(long devices, long rows) throws Exception { + try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getPath())) { + assertEquals( + devices, + ByteBuffer.wrap( + reader + .getTsFileProperties() + .get(TsFileTableSchemaUtil.getFlatDeviceLogicalDeviceCountPropertyKey(TABLE))) + .getLong()); + assertEquals( + rows, + ByteBuffer.wrap( + reader + .getTsFileProperties() + .get(TsFileTableSchemaUtil.getFlatDevicePointCountPropertyKey(TABLE))) + .getLong()); + } + } + private void writeUnsealedFile() throws Exception { writeUnsealedFile(TABLE); }
