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 4daba224d60e65e9df1789a25aa07ab6cef509b6 Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 16 16:58:41 2026 +0800 fix compaction --- .../execute/utils/FlatDeviceCompactionUtils.java | 119 +++++++++ .../utils/reader/SeriesDataBlockReader.java | 20 +- .../writer/RepairUnsortedFileCompactionWriter.java | 14 +- .../dataregion/memtable/TsFileProcessor.java | 33 ++- .../dataregion/tsfile/TsFileResource.java | 61 ++--- .../tablemodel/FlatDeviceCompactionTest.java | 277 +++++++++++++++++++++ .../schema/table/TsFileTableSchemaUtil.java | 23 ++ 7 files changed, 491 insertions(+), 56 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/FlatDeviceCompactionUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/FlatDeviceCompactionUtils.java new file mode 100644 index 00000000000..57ca4e45dbb --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/FlatDeviceCompactionUtils.java @@ -0,0 +1,119 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils; + +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.utils.TsPrimitiveType; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.Schema; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +public class FlatDeviceCompactionUtils { + private FlatDeviceCompactionUtils() {} + + public static boolean containsFlatDevice(List<TsFileResource> resources) throws IOException { + for (TsFileResource resource : resources) { + TsFileSequenceReader reader = + FileReaderManager.getInstance() + .get(resource.getTsFilePath(), resource.getTsFileID(), true); + Map<String, TableSchema> tableSchemaMap = reader.getTableSchemaMap(); + // Tree-model and legacy TsFiles do not carry table schema metadata. + if (tableSchemaMap == null) { + continue; + } + for (String table : tableSchemaMap.keySet()) { + if (TsFileTableSchemaUtil.isFlatDevice(reader.getTsFileProperties(), table)) { + return true; + } + } + } + return false; + } + + public static boolean containsFlatDevice(Schema schema) { + return schema.getTableSchemaMap().values().stream() + .anyMatch(FlatDeviceCompactionUtils::isFlatDevice); + } + + public static boolean isFlatDevice(TableSchema schema) { + return schema instanceof CompactionTableSchema compactionSchema + && compactionSchema.isFlatDevice(); + } + + public static List<Integer> getRowKeyColumnIndices( + TableSchema tableSchema, List<IMeasurementSchema> measurements) { + if (!isFlatDevice(tableSchema)) { + return Collections.emptyList(); + } + List<Integer> indices = new ArrayList<>(); + for (int i = 0; i < tableSchema.getColumnTypes().size(); i++) { + if (tableSchema.getColumnTypes().get(i) != ColumnCategory.TAG) { + continue; + } + IMeasurementSchema tag = tableSchema.getColumnSchemas().get(i); + int index = 0; + while (index < measurements.size() + && !measurements.get(index).getMeasurementName().equals(tag.getMeasurementName())) { + index++; + } + // An added TAG can be absent in older files. Read it as a null column, preserving TAG order. + if (index == measurements.size()) { + measurements.add(tag); + } + indices.add(index); + } + return indices; + } + + public static void updateResource( + TsFileResource resource, IDeviceID physicalDevice, List<Integer> tagIndices, TsBlock block) { + for (int row = 0; row < block.getPositionCount(); row++) { + resource.updateTableStartTime(physicalDevice.getTableName(), block.getTimeByIndex(row)); + resource.updateTableEndTime(physicalDevice.getTableName(), block.getTimeByIndex(row)); + } + } + + public static IDeviceID getLogicalDevice( + IDeviceID physicalDevice, List<Integer> tagIndices, TimeValuePair row) { + String[] segments = new String[tagIndices.size() + 1]; + segments[0] = physicalDevice.getTableName(); + TsPrimitiveType[] values = row.getValue().getVector(); + for (int tag = 0; tag < tagIndices.size(); tag++) { + TsPrimitiveType value = values[tagIndices.get(tag)]; + segments[tag + 1] = + value == null ? null : value.getBinary().getStringValue(TSFileConfig.STRING_CHARSET); + } + return IDeviceID.Factory.DEFAULT_FACTORY.create(segments); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/SeriesDataBlockReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/SeriesDataBlockReader.java index f89f8510072..735ccaddb89 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/SeriesDataBlockReader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/SeriesDataBlockReader.java @@ -36,6 +36,7 @@ import com.google.common.collect.Sets; import org.apache.tsfile.read.common.block.TsBlock; import java.io.IOException; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Optional; @@ -55,16 +56,29 @@ public class SeriesDataBlockReader implements IDataBlockReader { FragmentInstanceContext context, QueryDataSource dataSource, boolean ascending) { + this(seriesPath, allSensors, context, dataSource, ascending, Collections.emptyList()); + } + + public SeriesDataBlockReader( + IFullPath seriesPath, + Set<String> allSensors, + FragmentInstanceContext context, + QueryDataSource dataSource, + boolean ascending, + List<Integer> rowKeyColumnIndices) { SeriesScanOptions.Builder scanOptionsBuilder = new SeriesScanOptions.Builder(); scanOptionsBuilder.withAllSensors(allSensors); - if (seriesPath instanceof AlignedFullPath) { + if (seriesPath instanceof AlignedFullPath alignedPath) { this.seriesScanUtil = new AlignedSeriesScanUtil( - (AlignedFullPath) seriesPath, + alignedPath, ascending ? Ordering.ASC : Ordering.DESC, scanOptionsBuilder.build(), - context); + context, + false, + null, + rowKeyColumnIndices); } else if (seriesPath instanceof NonAlignedFullPath) { this.seriesScanUtil = new SeriesScanUtil( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java index 08d1740e351..e459e2a4f57 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java @@ -27,7 +27,6 @@ import org.apache.iotdb.db.utils.EncryptDBUtils; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.enums.TSDataType; -import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.common.block.TsBlock; import org.apache.tsfile.read.reader.IPointReader; @@ -39,6 +38,7 @@ import org.apache.tsfile.write.chunk.ValueChunkWriter; import java.io.IOException; import java.util.ArrayList; import java.util.Comparator; +import java.util.HashSet; import java.util.List; import java.util.stream.Collectors; @@ -105,6 +105,18 @@ public class RepairUnsortedFileCompactionWriter extends ReadPointInnerCompaction private void writeToChunkWriter(TimeValuePair timeValuePair, int subTaskId) throws IOException { writeDataPoint( timeValuePair.getTimestamp(), timeValuePair.getValue(), chunkWriters[subTaskId], subTaskId); + if (FlatDeviceCompactionUtils.isFlatDevice( + schemas.get(0).getTableSchemaMap().get(deviceId.getTableName()))) { + // Count only rows emitted after sorting and deduplication. The resource time index is + // updated with the physical table ID by endChunkGroup(), never with logical TAG IDs. + flatDeviceLogicalDevices + .computeIfAbsent(deviceId.getTableName(), key -> new HashSet<>()) + .add( + FlatDeviceCompactionUtils.getLogicalDevice( + deviceId, rowKeyColumnIndices, timeValuePair) + .toString()); + flatDeviceRowCounts.merge(deviceId.getTableName(), 1L, Long::sum); + } checkChunkSizeAndMayOpenANewChunk(fileWriter, chunkWriters[subTaskId], subTaskId); } 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 cd4c871a08e..fd52cc3b7e2 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 @@ -214,6 +214,9 @@ public class TsFileProcessor { /** Flush file listener. */ private final List<FlushListener> flushListeners = new ArrayList<>(); + private final Map<String, Set<String>> flatDeviceLogicalDevices = new HashMap<>(); + private final Map<String, Long> flatDeviceRowCounts = new HashMap<>(); + private final QueryExecutionMetricSet QUERY_EXECUTION_METRICS = QueryExecutionMetricSet.getInstance(); private final QueryResourceMetricSet QUERY_RESOURCE_METRICS = @@ -413,15 +416,23 @@ public class TsFileProcessor { private void updateResourceTime( IDeviceID deviceID, long startTime, long endTime, boolean flatDevice) { - tsFileResource.updateStartTime(deviceID, startTime); + updateResourceTime(deviceID, startTime, endTime, flatDevice, 1); + } + + private void updateResourceTime( + IDeviceID deviceID, long startTime, long endTime, boolean flatDevice, long rowCount) { if (flatDevice) { + flatDeviceLogicalDevices + .computeIfAbsent(deviceID.getTableName(), key -> new HashSet<>()) + .add(deviceID.toString()); + flatDeviceRowCounts.merge(deviceID.getTableName(), rowCount, Long::sum); tsFileResource.updateTableStartTime(deviceID.getTableName(), startTime); tsFileResource.updateTableEndTime(deviceID.getTableName(), endTime); - } - // A flat-device MemTable only retains physical device groups, so capture logical end times - // while the insert plan still provides each device's range, even for sequence files. - if (!sequence || flatDevice) { - tsFileResource.updateEndTime(deviceID, endTime); + } else { + tsFileResource.updateStartTime(deviceID, startTime); + if (!sequence) { + tsFileResource.updateEndTime(deviceID, endTime); + } } } @@ -752,7 +763,8 @@ public class TsFileProcessor { deviceEndOffset.left, insertTabletNode.getTimes()[deviceStartOffset], insertTabletNode.getTimes()[deviceEndOffset.right - 1], - insertTabletNode.isFlatDevice()); + insertTabletNode.isFlatDevice(), + deviceEndOffset.right - deviceStartOffset); deviceStartOffset = deviceEndOffset.right; } } @@ -2152,6 +2164,13 @@ public class TsFileProcessor { if (logger.isDebugEnabled()) { logger.debug(StorageEngineMessages.START_TO_END_FILE, tsFileResource); } + for (Map.Entry<String, Set<String>> entry : flatDeviceLogicalDevices.entrySet()) { + TsFileTableSchemaUtil.addFlatDeviceStatistics( + writer, + entry.getKey(), + entry.getValue().size(), + flatDeviceRowCounts.getOrDefault(entry.getKey(), 0L)); + } writer.endFile(); // Listen after "endFile" to avoid unnecessary waiting for tsFile close diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java index be82f6d2166..8c95d67eb3e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java @@ -131,9 +131,6 @@ public class TsFileResource implements PersistentResource, Cloneable { /** time index */ private volatile ITimeIndex timeIndex; - /** Start/end times aggregated by table for flat-device resource ordering. */ - private final Map<String, Pair<Long, Long>> tableTimeIndex = new HashMap<>(); - private final AtomicReference<Boolean> isEmpty = new AtomicReference<>(); private Future<ModificationFile> exclusiveModFileFuture; @@ -260,7 +257,6 @@ public class TsFileResource implements PersistentResource, Cloneable { throws IOException { this.file = originTsFileResource.file; this.timeIndex = originTsFileResource.timeIndex; - this.tableTimeIndex.putAll(originTsFileResource.tableTimeIndex); this.pathToReadOnlyMemChunkMap = pathToReadOnlyMemChunkMap; this.pathToChunkMetadataListMap = pathToChunkMetadataListMap; this.originTsFileResource = originTsFileResource; @@ -373,27 +369,6 @@ public class TsFileResource implements PersistentResource, Cloneable { } } } - rebuildTableTimeIndex(); - } - - private void rebuildTableTimeIndex() { - tableTimeIndex.clear(); - // The resource format only stores device times, without a flat-device marker. Aggregate - // table-model devices by table name; tree-model devices do not contribute to this index. - // A file-only index cannot recover per-table bounds, so ordering uses file bounds instead. - if (timeIndex instanceof FileTimeIndex) { - return; - } - for (IDeviceID deviceID : getDevices()) { - if (deviceID.isTableModel()) { - timeIndex - .getStartTime(deviceID) - .ifPresent(time -> updateTableStartTime(deviceID.getTableName(), time)); - timeIndex - .getEndTime(deviceID) - .ifPresent(time -> updateTableEndTime(deviceID.getTableName(), time)); - } - } } public static int getFileTimeIndexSerializedSize() { @@ -425,25 +400,22 @@ public class TsFileResource implements PersistentResource, Cloneable { } public void updateTableStartTime(String tableName, long time) { - tableTimeIndex.compute( - tableName, - (key, value) -> - value == null - ? new Pair<>(time, Long.MIN_VALUE) - : new Pair<>(Math.min(value.left, time), value.right)); + timeIndex.updateStartTime( + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {tableName}), time); } public void updateTableEndTime(String tableName, long time) { - tableTimeIndex.compute( - tableName, - (key, value) -> - value == null - ? new Pair<>(Long.MAX_VALUE, time) - : new Pair<>(value.left, Math.max(value.right, time))); + timeIndex.updateEndTime( + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {tableName}), time); } public Optional<Pair<Long, Long>> getTableTime(String tableName) { - return Optional.ofNullable(tableTimeIndex.get(tableName)); + IDeviceID tableDevice = IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {tableName}); + Optional<Long> start = timeIndex.getStartTime(tableDevice); + Optional<Long> end = timeIndex.getEndTime(tableDevice); + return start.isPresent() && end.isPresent() + ? Optional.of(new Pair<>(start.get(), end.get())) + : Optional.empty(); } public boolean resourceFileExists() { @@ -709,10 +681,10 @@ public class TsFileResource implements PersistentResource, Cloneable { // cannot use FileTimeIndex public long getOrderTimeForSeq(IDeviceID deviceId, boolean ascending) { if (isTableOnlyDevice(deviceId)) { - Pair<Long, Long> tableTime = tableTimeIndex.get(deviceId.getTableName()); - return tableTime == null + Optional<Pair<Long, Long>> tableTime = getTableTime(deviceId.getTableName()); + return tableTime.isEmpty() ? (ascending ? getFileStartTime() : getFileEndTime()) - : (ascending ? tableTime.left : tableTime.right); + : (ascending ? tableTime.get().left : tableTime.get().right); } if (timeIndex instanceof ArrayDeviceTimeIndex) { return ascending @@ -726,10 +698,10 @@ public class TsFileResource implements PersistentResource, Cloneable { // can use FileTimeIndex public long getOrderTimeForUnseq(IDeviceID deviceId, boolean ascending) { if (isTableOnlyDevice(deviceId)) { - Pair<Long, Long> tableTime = tableTimeIndex.get(deviceId.getTableName()); - return tableTime == null + Optional<Pair<Long, Long>> tableTime = getTableTime(deviceId.getTableName()); + return tableTime.isEmpty() ? (ascending ? getFileStartTime() : getFileEndTime()) - : (ascending ? tableTime.left : tableTime.right); + : (ascending ? tableTime.get().left : tableTime.get().right); } if (timeIndex instanceof ArrayDeviceTimeIndex) { if (ascending) { @@ -1742,7 +1714,6 @@ public class TsFileResource implements PersistentResource, Cloneable { TsFileResource cloned = new TsFileResource(); cloned.file = this.file; cloned.timeIndex = this.timeIndex; - cloned.tableTimeIndex.putAll(this.tableTimeIndex); cloned.maxPlanIndex = this.maxPlanIndex; cloned.minPlanIndex = this.minPlanIndex; cloned.exclusiveModFileFuture = this.exclusiveModFileFuture; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/FlatDeviceCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/FlatDeviceCompactionTest.java new file mode 100644 index 00000000000..5d60777593c --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/FlatDeviceCompactionTest.java @@ -0,0 +1,277 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.iotdb.db.storageengine.dataregion.compaction.tablemodel; + +import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.path.AlignedFullPath; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; +import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanUtil; +import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions; +import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; +import org.apache.iotdb.db.storageengine.dataregion.compaction.AbstractCompactionTest; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadChunkCompactionPerformer; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadPointCompactionPerformer; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.RepairUnsortedFileCompactionTask; +import org.apache.iotdb.db.storageengine.dataregion.compaction.repair.RepairDataFileScanUtil; +import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileRepairStatus; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; + +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.StringArrayDeviceID; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.reader.IPointReader; +import org.apache.tsfile.utils.TsPrimitiveType; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; + +public class FlatDeviceCompactionTest extends AbstractCompactionTest { + + private final String oldThreadName = Thread.currentThread().getName(); + + @Before + public void setUp() + throws IOException, + org.apache.tsfile.exception.write.WriteProcessException, + MetadataException, + InterruptedException { + super.setUp(); + Thread.currentThread().setName("pool-1-IoTDB-Compaction-Worker-1"); + } + + @After + public void tearDown() throws IOException, StorageEngineException { + super.tearDown(); + Thread.currentThread().setName(oldThreadName); + } + + @Test + public void testReadPointCompactionKeepsFlatDeviceRowsWithSameTimestamp() throws Exception { + assertFlatDeviceRowsSurviveCompaction(new ReadPointCompactionPerformer()); + } + + @Test + public void testRepairChecksFlatDeviceRowKeys() throws Exception { + // TAGs are separated by a FIELD in the physical schema. Compare both TAGs, including + // across chunk boundaries, while resource validation uses only the physical table ID. + for (String secondTag : Arrays.asList("z", "a", "m")) { + TsFileResource resource = createEmptyFileAndResource(true); + seqResources.add(resource); + try (CompactionTableModelTestFileWriter writer = + new CompactionTableModelTestFileWriter(resource)) { + writer.registerFlatDeviceSchema("flat_table"); + writer.startChunkGroup("flat_table", Collections.emptyList()); + writer.generateFlatDeviceAlignedRows( + new long[] {1}, new String[] {"a"}, new long[] {10}, new String[] {"m"}); + writer.generateFlatDeviceAlignedRows( + new long[] {1}, new String[] {"a"}, new long[] {20}, new String[] {secondTag}); + writer.endChunkGroup(); + writer.endFile(); + } + for (boolean checkResource : Arrays.asList(false, true)) { + RepairDataFileScanUtil scan = new RepairDataFileScanUtil(resource); + scan.scanTsFile(checkResource); + Assert.assertFalse(scan.isBrokenFile()); + Assert.assertEquals(!"z".equals(secondTag), scan.hasUnsortedDataOrWrongStatistics()); + } + } + } + + @Test + public void testRepairWritesOnlyTableTimeIndexAndCountsDeduplicatedRows() throws Exception { + TsFileResource source = createEmptyFileAndResource(true); + seqResources.add(source); + try (CompactionTableModelTestFileWriter writer = + new CompactionTableModelTestFileWriter(source)) { + writer.registerFlatDeviceSchema("flat_table"); + writer.startChunkGroup("flat_table", Collections.emptyList()); + // Include descending times, different TAGs at the same time, and a duplicate full row key. + writer.generateFlatDeviceAlignedRows( + new long[] {3, 1, 1, 1, 2}, + new String[] {"b", "b", "a", "a", "a"}, + new long[] {30, 11, 10, 10, 20}, + new String[] {"y", "y", "x", "x", "x"}); + writer.endChunkGroup(); + writer.endFile(); + } + tsFileManager.addAll(seqResources, true); + source.setTsFileRepairStatus(TsFileRepairStatus.NEED_TO_REPAIR_BY_REWRITE); + Assert.assertTrue( + new RepairUnsortedFileCompactionTask(0, tsFileManager, source, true, 0).start()); + + Assert.assertEquals(1, tsFileManager.getTsFileList(false).size()); + TsFileResource target = tsFileManager.getTsFileList(false).get(0); + IDeviceID tableDevice = new StringArrayDeviceID(new String[] {"flat_table"}); + Assert.assertEquals(Collections.singleton(tableDevice), target.getDevices()); + Assert.assertEquals(1L, target.getStartTime(tableDevice).get().longValue()); + Assert.assertEquals(3L, target.getEndTime(tableDevice).get().longValue()); + Assert.assertEquals( + Arrays.asList("1:a:10:x", "1:b:11:y", "2:a:20:x", "3:b:30:y"), readFlatDeviceRows(target)); + try (TsFileSequenceReader reader = new TsFileSequenceReader(target.getTsFilePath())) { + Assert.assertEquals( + 2L, + ByteBuffer.wrap( + reader + .getTsFileProperties() + .get( + TsFileTableSchemaUtil.getFlatDeviceLogicalDeviceCountPropertyKey( + "flat_table"))) + .getLong()); + Assert.assertEquals( + 4L, + ByteBuffer.wrap( + reader + .getTsFileProperties() + .get(TsFileTableSchemaUtil.getFlatDevicePointCountPropertyKey("flat_table"))) + .getLong()); + } + RepairDataFileScanUtil scan = new RepairDataFileScanUtil(target); + scan.scanTsFile(true); + Assert.assertFalse(scan.isBrokenFile()); + Assert.assertFalse(scan.hasUnsortedDataOrWrongStatistics()); + } + + @Test + public void testFastCompactionKeepsFlatDeviceRowsWithSameTimestamp() throws Exception { + assertFlatDeviceRowsSurviveCompaction(new FastCompactionPerformer(false)); + } + + @Test + public void testReadChunkCompactionKeepsFlatDeviceRowsWithSameTimestamp() throws Exception { + assertFlatDeviceRowsSurviveCompaction(new ReadChunkCompactionPerformer()); + } + + private void assertFlatDeviceRowsSurviveCompaction(ICompactionPerformer performer) + throws Exception { + TsFileResource first = + createFlatDeviceFile(true, 10, new String[] {"a", "b"}, new String[] {"x", "y"}, 1); + TsFileResource second = + createFlatDeviceFile(true, 20, new String[] {"a", "b"}, new String[] {"z", "x"}, 1); + seqResources.add(first); + seqResources.add(second); + tsFileManager.addAll(seqResources, true); + + InnerSpaceCompactionTask task = + new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, performer, 0); + Assert.assertTrue(task.start()); + + TsFileResource target = tsFileManager.getTsFileList(true).get(0); + try (TsFileSequenceReader reader = new TsFileSequenceReader(target.getTsFilePath())) { + Assert.assertTrue( + TsFileTableSchemaUtil.isFlatDevice(reader.getTsFileProperties(), "flat_table")); + TableSchema tableSchema = reader.getTableSchemaMap().get("flat_table"); + Assert.assertNotNull(tableSchema); + Assert.assertEquals( + Arrays.asList("tag1", "tag2", "field"), + tableSchema.getColumnSchemas().stream() + .map(IMeasurementSchema::getMeasurementName) + .toList()); + } + + Assert.assertEquals( + Arrays.asList("1:a:10:x", "1:a:20:z", "1:b:21:x", "1:b:11:y"), readFlatDeviceRows(target)); + } + + private TsFileResource createFlatDeviceFile( + boolean sequence, long fieldOffset, String[] tag1, String[] tag2, long timestamp) + throws IOException { + TsFileResource resource = createEmptyFileAndResource(sequence); + try (CompactionTableModelTestFileWriter writer = + new CompactionTableModelTestFileWriter(resource)) { + writer.registerFlatDeviceSchema("flat_table"); + writer.startChunkGroup("flat_table", Collections.emptyList()); + writer.generateFlatDeviceAlignedRows( + new long[] {timestamp, timestamp}, tag1, new long[] {fieldOffset, fieldOffset + 1}, tag2); + writer.endChunkGroup(); + writer.endFile(); + } + return resource; + } + + private List<String> readFlatDeviceRows(TsFileResource resource) throws IOException { + IDeviceID physicalDevice = new StringArrayDeviceID(new String[] {"flat_table"}); + List<IMeasurementSchema> schemas = + Arrays.asList( + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("tag2", TSDataType.STRING), + new MeasurementSchema("field", TSDataType.INT64)); + AlignedFullPath path = + new AlignedFullPath(physicalDevice, Arrays.asList("tag1", "tag2", "field"), schemas); + SeriesScanOptions.Builder options = new SeriesScanOptions.Builder(); + options.withAllSensors(new HashSet<>(Arrays.asList("tag1", "tag2", "field"))); + AlignedSeriesScanUtil scan = + new AlignedSeriesScanUtil( + path, + Ordering.ASC, + options.build(), + FragmentInstanceContext.createFragmentInstanceContextForCompaction(0L), + true, + Arrays.asList(TSDataType.STRING, TSDataType.STRING, TSDataType.INT64), + Arrays.asList(0, 1)); + scan.initQueryDataSource( + new QueryDataSource(Collections.singletonList(resource), Collections.emptyList())); + List<String> rows = new ArrayList<>(); + while (scan.hasNextFile().orElse(true)) { + while (scan.hasNextChunk().orElse(true)) { + while (scan.hasNextPage()) { + TsBlock block = scan.nextPage(); + if (block == null) { + continue; + } + IPointReader reader = block.getTsBlockAlignedRowIterator(); + while (reader.hasNextTimeValuePair()) { + TimeValuePair pair = reader.nextTimeValuePair(); + TsPrimitiveType[] vector = pair.getValue().getVector(); + rows.add( + pair.getTimestamp() + + ":" + + vector[0].getBinary().getStringValue(TSFileConfig.STRING_CHARSET) + + ":" + + vector[2].getLong() + + ":" + + vector[1].getBinary().getStringValue(TSFileConfig.STRING_CHARSET)); + } + } + } + } + return rows; + } +} 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 1d8f9980b36..17c2acb0168 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 @@ -32,6 +32,7 @@ import org.apache.tsfile.write.schema.MeasurementSchema; import org.apache.tsfile.write.writer.TsFileIOWriter; import java.nio.ByteBuffer; +import java.nio.ByteOrder; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.HashMap; @@ -269,6 +270,28 @@ public class TsFileTableSchemaUtil { return FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX + tableName; } + public static String getFlatDeviceLogicalDeviceCountPropertyKey(final String tableName) { + return FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX + "logical_device_count_" + tableName; + } + + public static String getFlatDevicePointCountPropertyKey(final String tableName) { + return FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX + "point_count_" + tableName; + } + + public static void addFlatDeviceStatistics( + final TsFileIOWriter writer, + final String tableName, + final long logicalDeviceCount, + final long pointCount) { + writer.addTsFileProperty( + getFlatDeviceLogicalDeviceCountPropertyKey(tableName), longBytes(logicalDeviceCount)); + writer.addTsFileProperty(getFlatDevicePointCountPropertyKey(tableName), longBytes(pointCount)); + } + + private static byte[] longBytes(long value) { + return ByteBuffer.allocate(Long.BYTES).order(ByteOrder.BIG_ENDIAN).putLong(value).array(); + } + /** 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) {
