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 5fa234e1938f18130baf1624c1a99152fda1cb6d Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 9 18:45:41 2026 +0800 tmp --- .../dataregion/memtable/TsFileProcessor.java | 4 ++ .../dataregion/tsfile/TsFileResource.java | 72 ++++++++++++++++++++++ .../dataregion/utils/TsFileResourceUtils.java | 17 ++++- .../dataregion/memtable/TsFileProcessorTest.java | 2 + 4 files changed, 94 insertions(+), 1 deletion(-) 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 4bc4c0d007b..cd4c871a08e 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 @@ -414,6 +414,10 @@ public class TsFileProcessor { private void updateResourceTime( IDeviceID deviceID, long startTime, long endTime, boolean flatDevice) { tsFileResource.updateStartTime(deviceID, startTime); + if (flatDevice) { + 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) { 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 4b33759af1c..be82f6d2166 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,6 +131,9 @@ 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; @@ -257,6 +260,7 @@ 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; @@ -369,6 +373,27 @@ 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() { @@ -399,6 +424,28 @@ public class TsFileResource implements PersistentResource, Cloneable { timeIndex.updateEndTime(device, time); } + 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)); + } + + 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))); + } + + public Optional<Pair<Long, Long>> getTableTime(String tableName) { + return Optional.ofNullable(tableTimeIndex.get(tableName)); + } + public boolean resourceFileExists() { return file != null && fsFactory.getFile(file + RESOURCE_SUFFIX).exists(); } @@ -661,6 +708,12 @@ 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 + ? (ascending ? getFileStartTime() : getFileEndTime()) + : (ascending ? tableTime.left : tableTime.right); + } if (timeIndex instanceof ArrayDeviceTimeIndex) { return ascending ? timeIndex.getStartTime(deviceId).orElse(Long.MIN_VALUE) @@ -672,6 +725,12 @@ 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 + ? (ascending ? getFileStartTime() : getFileEndTime()) + : (ascending ? tableTime.left : tableTime.right); + } if (timeIndex instanceof ArrayDeviceTimeIndex) { if (ascending) { return timeIndex.getStartTime(deviceId).orElse(Long.MIN_VALUE); @@ -1047,6 +1106,11 @@ public class TsFileResource implements PersistentResource, Cloneable { */ @SuppressWarnings("OptionalGetWithoutIsPresent") public boolean isSatisfied(IDeviceID deviceId, Filter timeFilter, boolean isSeq, boolean debug) { + // A flat-device scan reads the physical table-name group, while the resource indexes logical + // devices. Use conservative file bounds for table-only IDs (also valid for all-null TAGs). + if (isTableOnlyDevice(deviceId)) { + deviceId = null; + } if (deviceId != null && definitelyNotContains(deviceId)) { if (debug) { DEBUG_LOGGER.info( @@ -1101,12 +1165,19 @@ public class TsFileResource implements PersistentResource, Cloneable { * exist or out of dated. */ public boolean isDeviceAlive(IDeviceID device, long ttl) { + if (isTableOnlyDevice(device)) { + return !isClosed() || isAlive(getFileEndTime(), ttl); + } if (definitelyNotContains(device)) { return false; } return !isClosed() || timeIndex.isDeviceAlive(device, ttl); } + private static boolean isTableOnlyDevice(IDeviceID deviceID) { + return deviceID != null && deviceID.isTableModel() && deviceID.segmentNum() == 1; + } + public void setProcessor(TsFileProcessor processor) { this.processor = processor; } @@ -1671,6 +1742,7 @@ 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/main/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileResourceUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileResourceUtils.java index 755fe88bb46..40c20510609 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileResourceUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileResourceUtils.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.storageengine.dataregion.utils; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.StorageEngineMessages; @@ -436,7 +437,21 @@ public class TsFileResourceUtils { public static void updateTsFileResource( TsFileSequenceReader reader, TsFileResource tsFileResource) throws IOException { - updateTsFileResource(reader.getAllTimeseriesMetadata(false), tsFileResource, false); + Map<IDeviceID, List<TimeseriesMetadata>> metadata = reader.getAllTimeseriesMetadata(false); + updateTsFileResource(metadata, tsFileResource, false); + Map<String, byte[]> properties = reader.getTsFileProperties(); + for (Map.Entry<IDeviceID, List<TimeseriesMetadata>> entry : metadata.entrySet()) { + String tableName = entry.getKey().getTableName(); + if (!TsFileTableSchemaUtil.isFlatDevice(properties, tableName)) { + continue; + } + for (TimeseriesMetadata timeseriesMetadata : entry.getValue()) { + tsFileResource.updateTableStartTime( + tableName, timeseriesMetadata.getStatistics().getStartTime()); + tsFileResource.updateTableEndTime( + tableName, timeseriesMetadata.getStatistics().getEndTime()); + } + } tsFileResource.updatePlanIndexes(reader.getMinPlanIndex()); tsFileResource.updatePlanIndexes(reader.getMaxPlanIndex()); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index f8c7375b036..7dd344a2b22 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -1563,6 +1563,8 @@ public class TsFileProcessorTest { assertEquals(Set.of(first.getDeviceID(), second.getDeviceID()), resource.getDevices()); assertResourceTimes(resource, first.getDeviceID(), 10, 60); assertResourceTimes(resource, second.getDeviceID(), 30, 80); + assertEquals(10L, resource.getTableTime("flat_table").get().left.longValue()); + assertEquals(80L, resource.getTableTime("flat_table").get().right.longValue()); IDeviceID physicalDevice = first.getMemTableDeviceID(first.getDeviceID()); assertEquals(Set.of(physicalDevice), processor.getWorkMemTable().getMemTableMap().keySet()); assertTrue(resource.isSatisfied(physicalDevice, TimeFilterApi.eq(70), sequence, false));
