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 6522d54508fd48ec53c37bb54deee4f52bd9f33a Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 23 17:39:45 2026 +0800 tmp --- .../it/insertquery/IoTDBFlatDeviceIT.java | 4 +- .../execution/operator/source/SeriesScanUtil.java | 83 +++------------------- 2 files changed, 10 insertions(+), 77 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java index 50b6e9cdc8d..de9d122af4f 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java @@ -85,10 +85,10 @@ public class IoTDBFlatDeviceIT { public void setUp() throws SQLException { try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); Statement statement = connection.createStatement()) { - statement.execute("CREATE DATABASE " + DATABASE); + statement.execute("CREATE DATABASE IF NOT EXISTS " + DATABASE); statement.execute("USE " + DATABASE); statement.execute( - "CREATE TABLE " + "CREATE TABLE IF NOT EXISTS " + TABLE + " (tag_a STRING TAG, tag_b STRING TAG, attr STRING ATTRIBUTE, value DOUBLE FIELD)" + " WITH (flat_device=true)"); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java index cf58d2d26ca..566fc35f37a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java @@ -73,7 +73,6 @@ import org.apache.tsfile.read.common.block.column.IntColumn; import org.apache.tsfile.read.common.block.column.LongColumn; import org.apache.tsfile.read.controller.IChunkLoader; import org.apache.tsfile.read.filter.basic.Filter; -import org.apache.tsfile.read.filter.factory.FilterFactory; import org.apache.tsfile.read.reader.IPageReader; import org.apache.tsfile.read.reader.IPointReader; import org.apache.tsfile.read.reader.page.AlignedPageReader; @@ -694,14 +693,6 @@ public class SeriesScanUtil implements Accountable { return false; } - // A flat-device page is physically written in TAG-first order. Its timestamp statistics may - // still describe a non-overlapping range, but returning the page directly would expose that - // physical order to the caller. Route every flat-device page through the row-key merge reader - // so that the page adapter can restore the logical (time, TAG...) order first. - if (rowKeyColumnIndices.length > 0) { - return true; - } - long endpointTime = orderUtils.getOverlapCheckTime(firstPageReader.getStatistics()); unpackAllOverlappedTsFilesToTimeSeriesMetadata(endpointTime); unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(endpointTime, false); @@ -730,7 +721,7 @@ public class SeriesScanUtil implements Accountable { boolean hasMeetSeq = false; while (!cachedChunkMetadata.isEmpty() && orderUtils.isOverlapped(endpointTime, cachedChunkMetadata.peek().getStatistics())) { - if (rowKeyColumnIndices.length == 0 && cachedChunkMetadata.peek().isSeq() && hasMeetSeq) { + if (cachedChunkMetadata.peek().isSeq() && hasMeetSeq) { break; } else if (cachedChunkMetadata.peek().isSeq()) { hasMeetSeq = true; @@ -766,33 +757,10 @@ public class SeriesScanUtil implements Accountable { FileLoaderUtils.loadPageReaderList( chunkMetaData, scanOptions.getGlobalTimeFilter(), getTsDataTypeList(), context); - // Flat-device TAG deletions are row-scoped and must be evaluated while decoding each page. - Filter deletionFilter = - rowKeyColumnIndices.length > 0 - && chunkLoader instanceof DiskAlignedChunkLoader diskAlignedLoader - ? getFlatDeviceDeletionFilter(diskAlignedLoader.getResource()) - : null; - // init TsBlockBuilder for each page reader pageReaderList.forEach(p -> p.initTsBlockBuilder(getTsDataTypeList())); - if (rowKeyColumnIndices.length > 0) { - // TAG-first flushes can produce overlapping pages even in a sequence file. Queue them by - // their actual time ranges, but retain the file's sequence flag for version precedence. - for (IPageReader pageReader : pageReaderList) { - unSeqPageReaders.add( - new VersionPageReader( - context, - timestampInFileName, - chunkMetaData.getVersion(), - chunkMetaData.getOffsetOfChunkHeader(), - pageReader, - chunkMetaData.isSeq(), - tsFile, - deletionFilter, - false)); - } - } else if (chunkMetaData.isSeq()) { + if (chunkMetaData.isSeq()) { if (orderUtils.getAscending()) { for (IPageReader iPageReader : pageReaderList) { seqPageReaders.add( @@ -803,8 +771,7 @@ public class SeriesScanUtil implements Accountable { chunkMetaData.getOffsetOfChunkHeader(), iPageReader, true, - tsFile, - deletionFilter)); + tsFile)); } } else { for (int i = pageReaderList.size() - 1; i >= 0; i--) { @@ -816,8 +783,7 @@ public class SeriesScanUtil implements Accountable { chunkMetaData.getOffsetOfChunkHeader(), pageReaderList.get(i), true, - tsFile, - deletionFilter)); + tsFile)); } } } else { @@ -831,8 +797,7 @@ public class SeriesScanUtil implements Accountable { chunkMetaData.getOffsetOfChunkHeader(), pageReader, false, - tsFile, - deletionFilter))); + tsFile))); } if (LOGGER.isDebugEnabled()) { @@ -2283,7 +2248,6 @@ public class SeriesScanUtil implements Accountable { protected final QueryContext context; protected final MergeReaderPriority version; protected final IPageReader data; - private final Filter deletionFilter; protected final boolean isSeq; protected final boolean isAligned; @@ -2298,40 +2262,10 @@ public class SeriesScanUtil implements Accountable { IPageReader data, boolean isSeq, File tsFile) { - this(context, fileTimestamp, version, offset, data, isSeq, tsFile, null); - } - - VersionPageReader( - QueryContext context, - long fileTimestamp, - long version, - long offset, - IPageReader data, - boolean isSeq, - File tsFile, - Filter deletionFilter) { - this(context, fileTimestamp, version, offset, data, isSeq, tsFile, deletionFilter, true); - } - - VersionPageReader( - QueryContext context, - long fileTimestamp, - long version, - long offset, - IPageReader data, - boolean isSeq, - File tsFile, - Filter deletionFilter, - boolean pagesOrderedByTime) { this.context = context; this.version = new MergeReaderPriority(fileTimestamp, version, offset, isSeq); this.data = data; - this.deletionFilter = deletionFilter; - if (deletionFilter != null) { - // Overlapping pages must apply file-scoped deletions before entering the merge reader. - data.addRecordFilter(deletionFilter); - } - this.isSeq = isSeq && pagesOrderedByTime; + this.isSeq = isSeq; this.isAligned = data instanceof AlignedPageReader || data instanceof MemAlignedPageReader @@ -2420,12 +2354,11 @@ public class SeriesScanUtil implements Accountable { } public void addPushDownFilter(Filter pushDownFilter) { - // addRecordFilter replaces the previous filter, including when the new filter is null. - data.addRecordFilter(FilterFactory.and(deletionFilter, pushDownFilter)); + data.addRecordFilter(pushDownFilter); } public boolean isModified() { - return deletionFilter != null || data.isModified(); + return data.isModified(); } public boolean isSeq() {
