You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/10/12 06:50:52 UTC
(iotdb) 17/38: fix compaction
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 <jt...@163.com>
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) {