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 33a29f0901f598cde3e2dbf389a4a6e6a658856a Author: Tian Jiang <[email protected]> AuthorDate: Thu Oct 15 10:15:37 2026 +0800 tmp --- .../util/builder/PipeTableModelTsFileBuilder.java | 57 +++------------------- 1 file changed, 7 insertions(+), 50 deletions(-) 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 db34e46bff6..2ce5c789bef 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 @@ -20,11 +20,8 @@ package org.apache.iotdb.db.pipe.sink.util.builder; import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; -import org.apache.iotdb.commons.schema.table.TsTable; -import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils; -import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.tsfile.enums.ColumnCategory; @@ -169,7 +166,7 @@ public class PipeTableModelTsFileBuilder extends PipeTsFileBuilder { } try { - tryBestToWriteTabletsIntoOneFile(device2TabletsLinkedList, dataBase); + tryBestToWriteTabletsIntoOneFile(device2TabletsLinkedList); } catch (final Exception e) { LOGGER.warn( DataNodePipeMessages.BATCH_ID_FAILED_TO_WRITE_TABLETS_INTO, @@ -300,8 +297,7 @@ public class PipeTableModelTsFileBuilder extends PipeTsFileBuilder { } private <T extends Pair<Tablet, List<Pair<IDeviceID, Integer>>>> - void tryBestToWriteTabletsIntoOneFile( - final Set<LinkedList<T>> device2TabletsLinkedList, final String dataBase) + void tryBestToWriteTabletsIntoOneFile(final Set<LinkedList<T>> device2TabletsLinkedList) throws IOException { final Iterator<LinkedList<T>> iterator = device2TabletsLinkedList.iterator(); @@ -349,30 +345,17 @@ public class PipeTableModelTsFileBuilder extends PipeTsFileBuilder { iterator.remove(); } - boolean isFlatDevice = false; if (tableName != null) { - final TsTable table = DataNodeTableCache.getInstance().getTable(dataBase, tableName, false); - isFlatDevice = table != null && table.isFlatDevice(); - final List<IMeasurementSchema> physicalColumnSchemas = - new ArrayList<>(columnSchemas.size()); - for (int i = 0; i < columnSchemas.size(); i++) { - physicalColumnSchemas.add( - TsFileTableSchemaUtil.getMeasurementSchemaForTsFile( - columnSchemas.get(i), - TsTableColumnCategory.fromTsFileColumnCategory(columnCategories.get(i)), - isFlatDevice)); - } - fileWriter.registerTableSchema( - new TableSchema(tableName, physicalColumnSchemas, columnCategories)); - TsFileTableSchemaUtil.addFlatDeviceProperty( - fileWriter.getIOWriter(), tableName, isFlatDevice); + fileWriter.registerTableSchema(new TableSchema(tableName, columnSchemas, columnCategories)); + // TsFileWriter stores TAGs in logical device IDs even when the source table is flat. + // Describe the actual fallback layout so LOAD and Pipe readers recover those TAGs. + TsFileTableSchemaUtil.addFlatDeviceProperty(fileWriter.getIOWriter(), tableName, false); } for (final Pair<Tablet, List<Pair<IDeviceID, Integer>>> pair : tabletsToWrite) { final Tablet tablet = pair.left; - final Tablet physicalTablet = isFlatDevice ? withPhysicalSchemas(tablet) : tablet; try { - fileWriter.writeTable(physicalTablet, pair.right); + fileWriter.writeTable(tablet, pair.right); } catch (WriteProcessException e) { LOGGER.warn( DataNodePipeMessages.BATCH_ID_FAILED_TO_BUILD_THE_TABLE, currentBatchId.get(), e); @@ -382,32 +365,6 @@ public class PipeTableModelTsFileBuilder extends PipeTsFileBuilder { } } - private static Tablet withPhysicalSchemas(final Tablet tablet) { - final List<IMeasurementSchema> schemas = new ArrayList<>(tablet.getSchemas().size()); - final List<ColumnCategory> columnCategories = tablet.getColumnTypes(); - for (int i = 0; i < tablet.getSchemas().size(); i++) { - final IMeasurementSchema schema = tablet.getSchemas().get(i); - schemas.add( - schema == null - ? null - : TsFileTableSchemaUtil.getMeasurementSchemaForTsFile( - schema, - TsTableColumnCategory.fromTsFileColumnCategory(columnCategories.get(i)), - true)); - } - final Tablet physicalTablet = - new Tablet( - tablet.getTableName(), - schemas, - columnCategories, - tablet.getTimestamps(), - tablet.getValues(), - tablet.getBitMaps(), - tablet.getMaxRowNumber()); - physicalTablet.setRowSize(tablet.getRowSize()); - return physicalTablet; - } - /** * A Map is used to record the maximum time each {@link IDeviceID} is written. {@link Pair} * records the Index+1 of the maximum timestamp of IDevice in each {@link Tablet}.
