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 da39b1ad318d9a582beb47eab64180196041e5d5 Author: Tian Jiang <[email protected]> AuthorDate: Tue Sep 8 09:54:47 2026 +0800 load with flat_device --- ...ileInsertionEventTableParserTabletIterator.java | 62 ++++++- .../util/builder/PipeTableModelTsFileBuilder.java | 2 + .../builder/PipeTableModelTsFileBuilderV2.java | 1 + .../plan/analyze/load/LoadTsFileAnalyzer.java | 41 ++++- .../db/storageengine/dataregion/DataRegion.java | 3 +- .../impl/ReadChunkCompactionPerformer.java | 11 ++ .../dataregion/memtable/TsFileProcessor.java | 15 +- .../file/UnsealedTsFileRecoverPerformer.java | 10 +- .../db/storageengine/load/LoadTsFileManager.java | 1 + .../pipe/event/TsFileInsertionEventParserTest.java | 192 +++++++++++++++++++++ .../plan/analyze/load/LoadTsFileAnalyzerTest.java | 27 +++ .../schema/table/TsFileTableSchemaUtil.java | 32 ++++ .../schema/table/TsFileTableSchemaUtilTest.java | 46 +++++ 13 files changed, 423 insertions(+), 20 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java index ab591070d6b..e7aa8387080 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.pipe.event.common.tsfile.parser.table; import org.apache.iotdb.commons.path.PatternTreeMap; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils; import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils.TabletStringInternPool; @@ -105,8 +106,10 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T private List<ColumnCategory> columnTypes; private List<String> measurementList; private List<TSDataType> dataTypeList; + private List<IMeasurementSchema> tagSchemaList; private List<IMeasurementSchema> fieldSchemaList; private int deviceIdSize; + private boolean isFlatDeviceTable; private List<ModsOperationUtil.ModsInfo> modsInfoList; @@ -252,6 +255,7 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T dataTypeList = new ArrayList<>(columnSchemaSize); columnTypes = new ArrayList<>(columnSchemaSize); measurementList = new ArrayList<>(columnSchemaSize); + tagSchemaList = new ArrayList<>(columnSchemaSize); fieldSchemaList = new ArrayList<>(columnSchemaSize); for (int i = 0; i < columnSchemaSize; i++) { @@ -265,6 +269,7 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T columnTypes.add(ColumnCategory.TAG); measurementList.add(measurementName); dataTypeList.add(schema.getType()); + tagSchemaList.add(schema); } if (ColumnCategory.FIELD.equals(columnCategory)) { fieldSchemaList.add(schema); @@ -272,6 +277,8 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T } } deviceIdSize = dataTypeList.size(); + isFlatDeviceTable = + TsFileTableSchemaUtil.isFlatDevice(fileMetadata.getTsFileProperties(), tableName); state = State.INIT_CHUNK_METADATA; break; } @@ -331,7 +338,9 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T } if (fillMeasurementValueColumns(batchData, tablet, rowIndex)) { - fillDeviceIdColumns(deviceID, tablet, rowIndex); + if (!isFlatDeviceTable) { + fillDeviceIdColumns(deviceID, tablet, rowIndex); + } PipeTabletUtils.putTimestamp(tablet, rowIndex, batchData.currentTime()); } } @@ -362,7 +371,8 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T long size = timeChunkSize; final int fieldSchemaSize = fieldSchemaList.size(); - final List<Chunk> valueChunkList = new ArrayList<>(fieldSchemaSize); + final int tagSchemaSize = isFlatDeviceTable ? tagSchemaList.size() : 0; + final List<Chunk> valueChunkList = new ArrayList<>(tagSchemaSize + fieldSchemaSize); final Map<String, IChunkMetadata> valueChunkMetadataMap = new HashMap<>((int) (fieldSchemaSize / 0.75f) + 1); for (final IChunkMetadata metadata : alignedChunkMetadata.getValueChunkMetadataList()) { @@ -387,12 +397,42 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T } columnTypes = categories; - // Clean up the remaining non-DeviceID column information - measurementList.subList(deviceIdSize, measurementList.size()).clear(); - dataTypeList.subList(deviceIdSize, dataTypeList.size()).clear(); + if (isFlatDeviceTable) { + // In flat-device files TAGs are value columns, so rebuild the logical prefix from the + // physical TAG schemas instead of taking values from the device ID. + measurementList.clear(); + dataTypeList.clear(); + for (final IMeasurementSchema schema : tagSchemaList) { + measurementList.add(internMeasurementName(schema)); + dataTypeList.add(schema.getType()); + } + } else { + // Clean up the remaining non-DeviceID column information + measurementList.subList(deviceIdSize, measurementList.size()).clear(); + dataTypeList.subList(deviceIdSize, dataTypeList.size()).clear(); + } - boolean hasSelectedField = fieldSchemaList.isEmpty(); + boolean hasSelectedField = fieldSchemaList.isEmpty() || isFlatDeviceTable; boolean hasSelectedNonNullChunk = false; + + if (isFlatDeviceTable) { + // Flat-device TAGs are physical value chunks, but they are part of every row key. Keep them + // in every field batch so memory-based field splitting can never drop a TAG column. + for (final IMeasurementSchema schema : tagSchemaList) { + final String measurementName = internMeasurementName(schema); + final IChunkMetadata metadata = valueChunkMetadataMap.get(measurementName); + Chunk chunk = null; + if (metadata != null) { + chunk = reader.readMemChunk((ChunkMetadata) metadata); + size += PipeMemoryWeightUtil.calculateChunkRamBytesUsed(chunk); + if (size > allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) { + allocatedMemoryBlockForChunk.forceResize(size); + } + } + valueChunkList.add(chunk); + } + } + for (; offset < fieldSchemaList.size(); ++offset) { final IMeasurementSchema schema = fieldSchemaList.get(offset); final String measurementName = internMeasurementName(schema); @@ -481,11 +521,14 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T final TsPrimitiveType[] primitiveTypes = Objects.nonNull(data.getVector()) ? data.getVector() : new TsPrimitiveType[0]; boolean needFillTime = false; - boolean hasNonDeletedField = dataTypeList.size() == deviceIdSize; + final int valueColumnOffset = isFlatDeviceTable ? 0 : deviceIdSize; + boolean hasNonDeletedField = dataTypeList.size() == valueColumnOffset; - for (int i = deviceIdSize, size = dataTypeList.size(); i < size; i++) { + for (int i = valueColumnOffset, size = dataTypeList.size(); i < size; i++) { final TsPrimitiveType primitiveType = - i - deviceIdSize < primitiveTypes.length ? primitiveTypes[i - deviceIdSize] : null; + i - valueColumnOffset < primitiveTypes.length + ? primitiveTypes[i - valueColumnOffset] + : null; final boolean isDeleted = ModsOperationUtil.isDelete(data.currentTime(), modsInfoList.get(i)); if (!isDeleted) { hasNonDeletedField = true; @@ -568,6 +611,7 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T tablet, rowIndex, i - 1, dataTypeList.get(i - 1), deviceIdSegments[i]); } + // StringArrayDeviceID omits trailing null segments; preserve their logical null values. while (i <= deviceIdSize) { PipeTabletUtils.putValue( tablet, rowIndex, i - 1, dataTypeList.get(i - 1), Binary.EMPTY_VALUE); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java index 0f38aa23b62..db34e46bff6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java @@ -364,6 +364,8 @@ public class PipeTableModelTsFileBuilder extends PipeTsFileBuilder { } fileWriter.registerTableSchema( new TableSchema(tableName, physicalColumnSchemas, columnCategories)); + TsFileTableSchemaUtil.addFlatDeviceProperty( + fileWriter.getIOWriter(), tableName, isFlatDevice); } for (final Pair<Tablet, List<Pair<IDeviceID, Integer>>> pair : tabletsToWrite) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java index 7d247200b08..847a6bbe17d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java @@ -223,6 +223,7 @@ public class PipeTableModelTsFileBuilderV2 extends PipeTsFileBuilder { distinctIndices.stream() .map(aggregatedColumnCategories::get) .collect(Collectors.toList()))); + TsFileTableSchemaUtil.addFlatDeviceProperty(writer, tableName, isFlatDevice); } for (int i = 0, size = tabletList.size(); i < size; ++i) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java index 3cf619d7130..af096b26b37 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.queryengine.common.SessionInfo; import org.apache.iotdb.commons.queryengine.common.SqlDialect; import org.apache.iotdb.commons.queryengine.utils.TimestampPrecisionUtils; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadAnalyzeException; import org.apache.iotdb.db.exception.load.LoadAnalyzeMissingSchemaException; @@ -456,8 +457,17 @@ public class LoadTsFileAnalyzer implements AutoCloseable { this.isTableModelTsFile.set(i, isTableModelFile); this.isTableModelTsFileReliableIndex = i; - if (0 <= tabletConversionThresholdBytes - && tsFile.length() <= tabletConversionThresholdBytes + if (isTableModelFile) { + initializeDatabaseForTableModelIfNecessary(); + } + + // Flat-device tables rely on the tablet write path to preserve their row-key columns and + // sorting metadata. Convert these files regardless of the configured size threshold. + final boolean forceTabletConversion = + isTableModelFile && containsFlatDeviceTable(reader.getTsFileProperties(), tableSchemaMap); + if ((forceTabletConversion + || (0 <= tabletConversionThresholdBytes + && tsFile.length() <= tabletConversionThresholdBytes)) && handleSingleMiniFile(i)) { return; } @@ -496,6 +506,33 @@ public class LoadTsFileAnalyzer implements AutoCloseable { } } + private void initializeDatabaseForTableModelIfNecessary() { + if (Objects.nonNull(databaseForTableData)) { + return; + } + context + .getDatabaseName() + .ifPresent( + database -> { + databaseForTableData = database; + if (isTableModelStatement) { + loadTsFileTableStatement.setDatabase(database); + } else { + loadTsFileTreeStatement.setDatabase(database); + } + }); + } + + private boolean containsFlatDeviceTable( + final Map<String, byte[]> tsFileProperties, final Map<String, TableSchema> tableSchemaMap) { + for (final String tableName : tableSchemaMap.keySet()) { + if (TsFileTableSchemaUtil.isFlatDevice(tsFileProperties, tableName)) { + return true; + } + } + return false; + } + private boolean handleSingleMiniFile(final int i) throws FileNotFoundException { final long startTime = System.nanoTime(); try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index b07b4d6a3ee..07308ec0467 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -1797,7 +1797,8 @@ public class DataRegion implements IDataRegionForQuery { TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(tsTable); cacheTableSchema(database, t, currentVersion, schema); return schema; - }); + }, + node.isFlatDevice()); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java index da2e15db500..9f08759c996 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performe import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.commons.utils.MetadataUtils; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.conf.IoTDBDescriptor; @@ -28,6 +29,7 @@ import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ISeqCompactionPerformer; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CompactionTaskSummary; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionTableSchema; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionTableSchemaCollector; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.MultiTsFileDeviceIterator; @@ -221,6 +223,15 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer { summary.recordTargetTsFileTableSizeMap( targetResources.get(currentTargetFileIndex), currentWriter.getTableSizeMap()); currentWriter.setSchema(CompactionTableSchemaCollector.copySchema(schema)); + schema + .getTableSchemaMap() + .forEach( + (tableName, tableSchema) -> + TsFileTableSchemaUtil.addFlatDeviceProperty( + currentWriter, + tableName, + tableSchema instanceof CompactionTableSchema compactionTableSchema + && compactionTableSchema.isFlatDevice())); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index cb5b92f11d1..809e696ade9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -28,6 +28,7 @@ import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.path.AlignedPath; import org.apache.iotdb.commons.path.IFullPath; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.commons.utils.CommonDateTimeUtils; import org.apache.iotdb.commons.utils.TestOnly; @@ -2737,10 +2738,16 @@ public class TsFileProcessor { public void registerToTsFile( String tableName, Function<String, TableSchema> tableSchemaFunction) { - getWriter() - .getSchema() - .getTableSchemaMap() - .put(tableName, tableSchemaFunction.apply(tableName)); + registerToTsFile(tableName, tableSchemaFunction, false); + } + + public void registerToTsFile( + String tableName, Function<String, TableSchema> tableSchemaFunction, boolean isFlatDevice) { + final TableSchema tableSchema = tableSchemaFunction.apply(tableName); + getWriter().getSchema().getTableSchemaMap().put(tableName, tableSchema); + if (tableSchema != null) { + TsFileTableSchemaUtil.addFlatDeviceProperty(getWriter(), tableName, isFlatDevice); + } } public void writeLock() { 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 f9696cc2617..0ae6e38e5f6 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 @@ -21,6 +21,7 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.recover.file; import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.commons.schema.table.TsTable; import org.apache.iotdb.db.exception.DataRegionException; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; @@ -240,14 +241,15 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform private void registerToTsFile(String tableName) { if (tableName != null) { + final TsTable table = DataNodeTableCache.getInstance().getTable(databaseName, tableName); writer .getSchema() .getTableSchemaMap() .computeIfAbsent( - tableName, - t -> - TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute( - DataNodeTableCache.getInstance().getTable(databaseName, t))); + tableName, t -> TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table)); + if (table != null) { + TsFileTableSchemaUtil.addFlatDeviceProperty(writer, tableName, table.isFlatDevice()); + } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java index 220be0cd0c6..5777ba8b545 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java @@ -533,6 +533,7 @@ public class LoadTsFileManager { .getTableSchemaMap() .computeIfAbsent( tableName, t -> TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table)); + TsFileTableSchemaUtil.addFlatDeviceProperty(writer, tableName, table.isFlatDevice()); } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java index 0a4e0421abe..7a77a44b7fa 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java @@ -27,6 +27,7 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern; import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern; import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParser; @@ -97,6 +98,7 @@ import java.util.Objects; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; import static org.junit.Assert.fail; @@ -964,6 +966,196 @@ public class TsFileInsertionEventParserTest { } } + @Test + public void testTableParserPreservesAllTagColumns() throws Exception { + alignedTsFile = new File("table-parser-all-tags.tsfile"); + if (alignedTsFile.exists()) { + Assert.assertTrue(alignedTsFile.delete()); + } + + final List<IMeasurementSchema> schemaList = + Arrays.asList( + new MeasurementSchema("tag0", TSDataType.STRING), + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("tag2", TSDataType.STRING), + new MeasurementSchema("s0", TSDataType.INT64)); + final List<String> columnNameList = Arrays.asList("tag0", "tag1", "tag2", "s0"); + final List<TSDataType> dataTypeList = + Arrays.asList(TSDataType.STRING, TSDataType.STRING, TSDataType.STRING, TSDataType.INT64); + final List<ColumnCategory> columnCategoryList = + Arrays.asList( + ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.FIELD); + + final Tablet sourceTablet = + new Tablet("all_tags", columnNameList, dataTypeList, columnCategoryList, 1); + sourceTablet.addTimestamp(0, 1); + sourceTablet.addValue(0, 0, "v0"); + sourceTablet.addValue(0, 1, "v1"); + sourceTablet.addValue(0, 2, "v2"); + sourceTablet.addValue(0, 3, 10L); + + try (final TsFileWriter writer = new TsFileWriter(alignedTsFile)) { + writer.registerTableSchema(new TableSchema("all_tags", schemaList, columnCategoryList)); + writer.writeTable(sourceTablet); + } + + try (final TsFileInsertionEventTableParser parser = + new TsFileInsertionEventTableParser( + alignedTsFile, + new TablePattern(true, null, null), + Long.MIN_VALUE, + Long.MAX_VALUE, + null, + null, + null, + false)) { + final Iterator<TabletInsertionEvent> iterator = parser.toTabletInsertionEvents().iterator(); + Assert.assertTrue(iterator.hasNext()); + final Tablet parsedTablet = ((PipeRawTabletInsertionEvent) iterator.next()).convertToTablet(); + Assert.assertEquals( + columnNameList, + parsedTablet.getSchemas().stream() + .map(IMeasurementSchema::getMeasurementName) + .collect(Collectors.toList())); + Assert.assertEquals(columnCategoryList, parsedTablet.getColumnTypes()); + Assert.assertEquals( + new Binary("v0", TSFileConfig.STRING_CHARSET), + ((Binary[]) parsedTablet.getValues()[0])[0]); + Assert.assertEquals( + new Binary("v1", TSFileConfig.STRING_CHARSET), + ((Binary[]) parsedTablet.getValues()[1])[0]); + Assert.assertEquals( + new Binary("v2", TSFileConfig.STRING_CHARSET), + ((Binary[]) parsedTablet.getValues()[2])[0]); + Assert.assertEquals(10L, ((long[]) parsedTablet.getValues()[3])[0]); + Assert.assertFalse(iterator.hasNext()); + } + } + + @Test + public void testTableParserPreservesTrailingNullTagColumns() throws Exception { + alignedTsFile = new File("table-parser-trailing-tag.tsfile"); + if (alignedTsFile.exists()) { + Assert.assertTrue(alignedTsFile.delete()); + } + + final List<IMeasurementSchema> schemaList = + Arrays.asList( + new MeasurementSchema("tag0", TSDataType.STRING), + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("tag2", TSDataType.STRING), + new MeasurementSchema("s0", TSDataType.INT64)); + final List<String> columnNameList = Arrays.asList("tag0", "tag1", "tag2", "s0"); + final List<TSDataType> dataTypeList = + Arrays.asList(TSDataType.STRING, TSDataType.STRING, TSDataType.STRING, TSDataType.INT64); + final List<ColumnCategory> columnCategoryList = + Arrays.asList( + ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.FIELD); + + final Tablet sourceTablet = + new Tablet("trailing_tag", columnNameList, dataTypeList, columnCategoryList, 1); + sourceTablet.addTimestamp(0, 1); + sourceTablet.addValue(0, 0, "v0"); + sourceTablet.addValue(0, 1, "v1"); + sourceTablet.addValue(0, 3, 10L); + + try (final TsFileWriter writer = new TsFileWriter(alignedTsFile)) { + writer.registerTableSchema(new TableSchema("trailing_tag", schemaList, columnCategoryList)); + writer.writeTable(sourceTablet); + } + + try (final TsFileInsertionEventTableParser parser = + new TsFileInsertionEventTableParser( + alignedTsFile, + new TablePattern(true, null, null), + Long.MIN_VALUE, + Long.MAX_VALUE, + null, + null, + null, + false)) { + final Iterator<TabletInsertionEvent> iterator = parser.toTabletInsertionEvents().iterator(); + Assert.assertTrue(iterator.hasNext()); + final Tablet parsedTablet = ((PipeRawTabletInsertionEvent) iterator.next()).convertToTablet(); + Assert.assertTrue(parsedTablet.getBitMaps()[2].isMarked(0)); + Assert.assertFalse(iterator.hasNext()); + } + } + + @Test + public void testTableParserReadsFlatDeviceTagChunks() throws Exception { + alignedTsFile = new File("table-parser-flat-device-tags.tsfile"); + if (alignedTsFile.exists()) { + Assert.assertTrue(alignedTsFile.delete()); + } + + final List<IMeasurementSchema> schemaList = + Arrays.asList( + new MeasurementSchema("tag0", TSDataType.STRING, TSEncoding.DICTIONARY), + new MeasurementSchema("tag1", TSDataType.STRING, TSEncoding.DICTIONARY), + new MeasurementSchema("s0", TSDataType.INT64, TSEncoding.PLAIN)); + final List<ColumnCategory> columnCategories = + Arrays.asList(ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.FIELD); + final TableSchema tableSchema = new TableSchema("flat_device", schemaList, columnCategories); + + try (final TsFileIOWriter writer = new TsFileIOWriter(alignedTsFile)) { + writer.getSchema().registerTableSchema(tableSchema); + TsFileTableSchemaUtil.addFlatDeviceProperty(writer, "flat_device", true); + writer.startChunkGroup( + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"flat_device"})); + final AlignedChunkWriterImpl chunkWriter = new AlignedChunkWriterImpl(schemaList); + for (int row = 0; row < 2; row++) { + chunkWriter.getTimeChunkWriter().write(row); + chunkWriter + .getValueChunkWriterByIndex(0) + .write(row, new Binary("tag0-" + row, TSFileConfig.STRING_CHARSET), false); + chunkWriter + .getValueChunkWriterByIndex(1) + .write(row, new Binary("tag1-" + row, TSFileConfig.STRING_CHARSET), false); + chunkWriter.getValueChunkWriterByIndex(2).write(row, (long) row, false); + } + chunkWriter.writeToFileWriter(writer); + writer.endChunkGroup(); + writer.endFile(); + } + + try (final TsFileSequenceReader reader = + new TsFileSequenceReader(alignedTsFile.getAbsolutePath())) { + Assert.assertTrue( + TsFileTableSchemaUtil.isFlatDevice(reader.getTsFileProperties(), "flat_device")); + } + + try (final TsFileInsertionEventTableParser parser = + new TsFileInsertionEventTableParser( + alignedTsFile, + new TablePattern(true, null, null), + Long.MIN_VALUE, + Long.MAX_VALUE, + null, + null, + null, + false)) { + final Iterator<TabletInsertionEvent> iterator = parser.toTabletInsertionEvents().iterator(); + int rowCount = 0; + while (iterator.hasNext()) { + final Tablet tablet = ((PipeRawTabletInsertionEvent) iterator.next()).convertToTablet(); + Assert.assertEquals(columnCategories, tablet.getColumnTypes()); + Assert.assertEquals(3, tablet.getSchemas().size()); + for (int row = 0; row < tablet.getRowSize(); row++) { + Assert.assertEquals( + new Binary("tag0-" + rowCount, TSFileConfig.STRING_CHARSET), + ((Binary[]) tablet.getValues()[0])[row]); + Assert.assertEquals( + new Binary("tag1-" + rowCount, TSFileConfig.STRING_CHARSET), + ((Binary[]) tablet.getValues()[1])[row]); + Assert.assertEquals(rowCount, ((long[]) tablet.getValues()[2])[row]); + rowCount++; + } + } + Assert.assertEquals(2, rowCount); + } + } + @Test public void testTableParserWithTablePatternReportsLastNonEmptyTablet() throws Exception { final int originalPipeDataStructureTabletRowSize = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java index bf248d7a600..966aec2d555 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.queryengine.plan.analyze.load; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadAnalyzeException; @@ -61,6 +62,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; public class LoadTsFileAnalyzerTest { @@ -478,6 +480,31 @@ public class LoadTsFileAnalyzerTest { .anyMatch(device -> Arrays.equals(device.getSegments(), expectedSegments)); } + @Test + public void testFlatDevicePropertyIsReadWithoutTableCache() throws Exception { + final Map<String, byte[]> properties = + Collections.singletonMap( + TsFileTableSchemaUtil.getFlatDevicePropertyKey("table1"), new byte[] {1}); + final Method method = + LoadTsFileAnalyzer.class.getDeclaredMethod("containsFlatDeviceTable", Map.class, Map.class); + method.setAccessible(true); + final File tempFile = File.createTempFile("flat-device", ".tsfile"); + try (LoadTsFileAnalyzer analyzer = + new LoadTsFileAnalyzer( + LoadTsFileStatement.createUnchecked(tempFile.getAbsolutePath()), + true, + new MPPQueryContext(new QueryId("load_flat_device_property_test")))) { + Assert.assertTrue( + (Boolean) + method.invoke( + analyzer, + properties, + Collections.singletonMap("table1", new TableSchema("table1")))); + } finally { + Assert.assertTrue(tempFile.delete()); + } + } + private static class TrackingLoadTsFileTableSchemaCache extends LoadTsFileTableSchemaCache { private final Set<List<Object>> verifiedDevices = new HashSet<>(); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtil.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtil.java index b641fdfb05b..1d8f9980b36 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtil.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtil.java @@ -29,20 +29,27 @@ import org.apache.tsfile.file.metadata.enums.TSEncoding; import org.apache.tsfile.utils.ReadWriteIOUtils; import org.apache.tsfile.write.schema.IMeasurementSchema; import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.Objects; import static org.apache.iotdb.commons.schema.table.TsTable.TIME_COLUMN_NAME; /** Utility class for converting between TsTable and TSFile TableSchema */ public class TsFileTableSchemaUtil { + private static final String FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX = + TsTable.FLAT_DEVICE_PROPERTY + "."; private static final byte[] FLAT_DEVICE_PROPERTY_BYTES = TsTable.FLAT_DEVICE_PROPERTY.getBytes(StandardCharsets.UTF_8); + private static final byte[] TRUE_BYTES = {1}; + private static final byte[] FALSE_BYTES = {0}; private TsFileTableSchemaUtil() { // Utility class, prevent instantiation @@ -247,6 +254,31 @@ public class TsFileTableSchemaUtil { schema.getProps()); } + /** + * Add the table's physical layout marker to the TsFile footer. A file may contain several tables, + * so the table name is part of the property key. The value is a single byte: 0 for false, 1 for + * true. + */ + public static void addFlatDeviceProperty( + final TsFileIOWriter writer, final String tableName, final boolean isFlatDevice) { + writer.addTsFileProperty( + getFlatDevicePropertyKey(tableName), isFlatDevice ? TRUE_BYTES : FALSE_BYTES); + } + + public static String getFlatDevicePropertyKey(final String tableName) { + return FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX + tableName; + } + + /** Returns the flat-device marker for a table, defaulting to false for old TsFiles. */ + public static boolean isFlatDevice( + final Map<String, byte[]> tsFileProperties, final String tableName) { + if (Objects.isNull(tsFileProperties)) { + return false; + } + final byte[] value = tsFileProperties.get(getFlatDevicePropertyKey(tableName)); + return value != null && value.length == 1 && value[0] == 1; + } + /** * Skip a string in ByteBuffer without reading it. This is more efficient than reading and * discarding the string. diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtilTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtilTest.java index 6fbba6bacf8..07a83a4838d 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtilTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtilTest.java @@ -30,11 +30,16 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.TableSchema; 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.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; import org.junit.Assert; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; import java.io.ByteArrayOutputStream; +import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; @@ -46,6 +51,47 @@ import java.util.stream.Collectors; public class TsFileTableSchemaUtilTest { + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void testFlatDevicePropertiesRoundTrip() throws IOException { + final File file = temporaryFolder.newFile("table-properties.tsfile"); + try (final TsFileIOWriter writer = new TsFileIOWriter(file)) { + TsFileTableSchemaUtil.addFlatDeviceProperty(writer, "flat_table", true); + TsFileTableSchemaUtil.addFlatDeviceProperty(writer, "regular_table", false); + writer.endFile(); + } + + try (final TsFileSequenceReader reader = new TsFileSequenceReader(file.getAbsolutePath())) { + final Map<String, byte[]> properties = reader.getTsFileProperties(); + Assert.assertTrue(TsFileTableSchemaUtil.isFlatDevice(properties, "flat_table")); + Assert.assertFalse(TsFileTableSchemaUtil.isFlatDevice(properties, "regular_table")); + Assert.assertArrayEquals( + new byte[] {1}, + properties.get(TsFileTableSchemaUtil.getFlatDevicePropertyKey("flat_table"))); + Assert.assertArrayEquals( + new byte[] {0}, + properties.get(TsFileTableSchemaUtil.getFlatDevicePropertyKey("regular_table"))); + Assert.assertFalse(TsFileTableSchemaUtil.isFlatDevice(properties, "missing_table")); + } + } + + @Test + public void testMissingFlatDevicePropertyDefaultsToFalse() { + Assert.assertFalse(TsFileTableSchemaUtil.isFlatDevice(null, "table")); + Assert.assertFalse(TsFileTableSchemaUtil.isFlatDevice(new HashMap<>(), "table")); + } + + @Test + public void testInvalidFlatDevicePropertyDefaultsToFalse() { + final Map<String, byte[]> properties = new HashMap<>(); + final String key = TsFileTableSchemaUtil.getFlatDevicePropertyKey("table"); + for (final byte[] value : new byte[][] {null, {}, {2}, {1, 0}}) { + properties.put(key, value); + Assert.assertFalse(TsFileTableSchemaUtil.isFlatDevice(properties, "table")); + } + } + @Test public void testToTsFileTableSchemaNoAttribute() { // Create a TsTable with all column types
