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 ebb0c957af9abf04ec25c27a3d56c702c4fcfc54 Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 23 16:51:34 2026 +0800 tmp --- .../operator/source/FlatDevicePageReaderQueue.java | 170 +++++++++++++++++ .../read/reader/common/FlatDevicePointReader.java | 95 ++++++++++ .../common/FlatDevicePriorityMergeReader.java | 52 +++++ .../reader/common/FlatDevicePointReaderTest.java | 211 +++++++++++++++++++++ 4 files changed, 528 insertions(+) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueue.java new file mode 100644 index 00000000000..4a8166544d6 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueue.java @@ -0,0 +1,170 @@ +/* + * 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.queryengine.execution.operator.source; + +import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.AlignedPriorityMergeReader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.FlatDevicePointReader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.MergeReaderPriority; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.PriorityMergeReader; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata; +import org.apache.tsfile.file.metadata.IChunkMetadata; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.filter.basic.Filter; +import org.apache.tsfile.read.reader.IPageReader; +import org.apache.tsfile.utils.TsPrimitiveType; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.PriorityQueue; +import java.util.stream.Collectors; + +/** Opens flat pages by their first compound row key, independently of their time ranges. */ +final class FlatDevicePageReaderQueue { + private final FragmentInstanceContext context; + private final List<TSDataType> dataTypes; + private final List<Integer> keyIndices; + private final boolean ascending; + private final Filter timeFilter; + private final PriorityMergeReader mergeReader; + private final PriorityQueue<PendingPage> pages; + + FlatDevicePageReaderQueue( + FragmentInstanceContext context, + List<TSDataType> dataTypes, + List<Integer> keyIndices, + boolean ascending, + Filter timeFilter, + PriorityMergeReader mergeReader) { + this.context = context; + this.dataTypes = dataTypes; + this.keyIndices = keyIndices; + this.ascending = ascending; + this.timeFilter = timeFilter; + this.mergeReader = mergeReader; + pages = new PriorityQueue<>((left, right) -> compare(left.firstKey, right.firstKey)); + } + + private int compare(TimeValuePair left, TimeValuePair right) { + int result = + AlignedPriorityMergeReader.compareFlatDeviceRowKey(left, right, dataTypes, keyIndices); + return ascending ? result : -result; + } + + void addChunk(AbstractAlignedChunkMetadata metadata, Filter deletionFilter) throws IOException { + List<IChunkMetadata> keyMetadata = new ArrayList<>(); + for (int index : keyIndices) { + keyMetadata.add(metadata.getValueChunkMetadataList().get(index)); + } + AbstractAlignedChunkMetadata keyChunk = + metadata.createNewChunkMetadata(metadata.getTimeChunkMetadata(), keyMetadata); + keyChunk.setChunkLoader(metadata.getChunkLoader()); + List<TSDataType> keyTypes = + keyIndices.stream().map(dataTypes::get).collect(Collectors.toList()); + List<IPageReader> keyPages = + FileLoaderUtils.loadPageReaderList(keyChunk, timeFilter, keyTypes, context); + ChunkPages chunkPages = new ChunkPages(metadata, deletionFilter); + // TsFile statistics contain no compound TAG/time bounds. Decode only TAG and time columns + // to build the pending-page queue, retaining each page's first key in scan order. + // Defer FIELD decoding until the page needs to participate in the merge. + for (int pageIndex = 0; pageIndex < keyPages.size(); pageIndex++) { + IPageReader keyPage = keyPages.set(pageIndex, null); + keyPage.initTsBlockBuilder(keyTypes); + try (FlatDevicePointReader keys = + new FlatDevicePointReader(keyPage.getAllSatisfiedData(), ascending)) { + if (!keys.hasNextTimeValuePair()) { + continue; + } + TimeValuePair first = keys.nextTimeValuePair(); + TsPrimitiveType[] projected = new TsPrimitiveType[dataTypes.size()]; + for (int i = 0; i < keyIndices.size(); i++) { + projected[keyIndices.get(i)] = first.getValue().getVector()[i]; + } + pages.add( + new PendingPage( + new TimeValuePair(first.getTimestamp(), new TsPrimitiveType.TsVector(projected)), + chunkPages, + pageIndex)); + } + } + } + + boolean hasNext() throws IOException { + // A time endpoint cannot prove that a flat page is exhausted or irrelevant: another TAG may + // restart at an earlier timestamp. Open every page whose compound first key can beat or equal + // the heap head, including equal keys needed for overwrite/null-fill resolution. + while (!pages.isEmpty() + && (!mergeReader.hasNextTimeValuePair() + || compare(pages.peek().firstKey, mergeReader.currentTimeValuePair()) <= 0)) { + PendingPage page = pages.poll(); + page.chunk.open(page.pageIndex); + } + return mergeReader.hasNextTimeValuePair(); + } + + void clear() { + pages.clear(); + } + + private class ChunkPages { + private final AbstractAlignedChunkMetadata metadata; + private final Filter deletionFilter; + private List<IPageReader> readers; + + private ChunkPages(AbstractAlignedChunkMetadata metadata, Filter deletionFilter) { + this.metadata = metadata; + this.deletionFilter = deletionFilter; + } + + private void open(int index) throws IOException { + if (readers == null) { + readers = FileLoaderUtils.loadPageReaderList(metadata, timeFilter, dataTypes, context); + } + IPageReader page = readers.set(index, null); + page.initTsBlockBuilder(dataTypes); + if (deletionFilter != null) { + page.addRecordFilter(deletionFilter); + } + TsBlock block = page.getAllSatisfiedData(); + mergeReader.addReader( + new FlatDevicePointReader(block, ascending), + new MergeReaderPriority( + FileLoaderUtils.getTimestampInFileName(metadata), + metadata.getVersion(), + metadata.getOffsetOfChunkHeader(), + metadata.isSeq()), + ascending ? metadata.getEndTime() : metadata.getStartTime()); + } + } + + private static class PendingPage { + private final TimeValuePair firstKey; + private final ChunkPages chunk; + private final int pageIndex; + + private PendingPage(TimeValuePair firstKey, ChunkPages chunk, int pageIndex) { + this.firstKey = firstKey; + this.chunk = chunk; + this.pageIndex = pageIndex; + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDevicePointReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDevicePointReader.java new file mode 100644 index 00000000000..2396c7d8804 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDevicePointReader.java @@ -0,0 +1,95 @@ +/* + * 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.read.reader.common; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.reader.IPointReader; +import org.apache.tsfile.utils.RamUsageEstimator; +import org.apache.tsfile.utils.TsPrimitiveType; + +/** + * Reads a TAG-first flat-device page in either direction, materializing only the current row. + * + * <p>The input page must already be ordered by {@code (TAG..., time)}. Timestamps can go backwards + * when TAGs change; reading in reverse produces descending compound keys without sorting the page. + */ +public final class FlatDevicePointReader implements IPointReader, AutoCloseable { + + private TsBlock block; + private final int rowCount; + private final boolean ascending; + private final long usedMemorySize; + private int rowIndex; + + public FlatDevicePointReader(TsBlock block, boolean ascending) { + this.block = block; + this.ascending = ascending; + rowCount = block.getPositionCount(); + usedMemorySize = + RamUsageEstimator.shallowSizeOfInstance(FlatDevicePointReader.class) + + block.getRetainedSizeInBytes(); + } + + @Override + public boolean hasNextTimeValuePair() { + return rowIndex < rowCount; + } + + @Override + public TimeValuePair nextTimeValuePair() { + TimeValuePair row = currentTimeValuePair(); + rowIndex++; + if (!hasNextTimeValuePair()) { + // The merge heap owns the returned row, so the consumed page can be released immediately. + block = null; + } + return row; + } + + @Override + public TimeValuePair currentTimeValuePair() { + if (!hasNextTimeValuePair()) { + return null; + } + int sourceRowIndex = ascending ? rowIndex : rowCount - 1 - rowIndex; + TsPrimitiveType[] values = new TsPrimitiveType[block.getValueColumnCount()]; + for (int columnIndex = 0; columnIndex < values.length; columnIndex++) { + Column column = block.getColumn(columnIndex); + if (!column.isNull(sourceRowIndex)) { + values[columnIndex] = column.getTsPrimitiveType(sourceRowIndex); + } + } + return new TimeValuePair( + block.getTimeByIndex(sourceRowIndex), new TsPrimitiveType.TsVector(values)); + } + + @Override + public long getUsedMemorySize() { + // PriorityMergeReader subtracts the same reservation after consuming the last row. + return usedMemorySize; + } + + @Override + public void close() { + block = null; + rowIndex = rowCount; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDevicePriorityMergeReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDevicePriorityMergeReader.java new file mode 100644 index 00000000000..7dd34810147 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDevicePriorityMergeReader.java @@ -0,0 +1,52 @@ +/* + * 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.read.reader.common; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.TimeValuePair; + +import java.util.ArrayList; +import java.util.List; + +/** Priority merge reader for flat-device rows whose physical pages are TAG-first. */ +public class FlatDevicePriorityMergeReader extends AlignedPriorityMergeReader { + + private List<TSDataType> dataTypes = List.of(); + private List<Integer> rowKeyColumnIndices = List.of(); + + public FlatDevicePriorityMergeReader() {} + + public FlatDevicePriorityMergeReader( + List<TSDataType> dataTypes, List<Integer> rowKeyColumnIndices) { + configureRowKeyColumns(dataTypes, rowKeyColumnIndices); + } + + @Override + public void configureRowKeyColumns( + List<TSDataType> dataTypes, List<Integer> rowKeyColumnIndices) { + this.dataTypes = new ArrayList<>(dataTypes); + this.rowKeyColumnIndices = new ArrayList<>(rowKeyColumnIndices); + super.configureRowKeyColumns(dataTypes, rowKeyColumnIndices); + } + + @Override + protected int compareRowKey(TimeValuePair left, TimeValuePair right) { + return compareFlatDeviceRowKey(left, right, dataTypes, rowKeyColumnIndices); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDevicePointReaderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDevicePointReaderTest.java new file mode 100644 index 00000000000..80aefd4106d --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDevicePointReaderTest.java @@ -0,0 +1,211 @@ +/* + * 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.read.reader.common; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.common.block.TsBlockBuilder; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.TsPrimitiveType; +import org.junit.Test; + +import java.io.IOException; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +public class FlatDevicePointReaderTest { + + private static final List<TSDataType> DATA_TYPES = + List.of(TSDataType.STRING, TSDataType.INT64, TSDataType.STRING); + + @Test + public void testTagFirstPageInBothDirections() { + // TAGs surround a FIELD column, and changing either TAG can move time backwards. + TsBlock block = + block( + new long[] {70, 90, 10, 20, 1, 0}, + new String[] {null, "a", "a", "a", "a", "b"}, + new Long[] {70L, null, 10L, 20L, 1L, 0L}, + new String[] {"x", null, "x", "x", "y", null}); + for (boolean ascending : new boolean[] {true, false}) { + try (FlatDevicePointReader reader = new FlatDevicePointReader(block, ascending)) { + long reservedMemory = reader.getUsedMemorySize(); + assertTrue(reservedMemory >= block.getRetainedSizeInBytes()); + for (int rowIndex = 0; rowIndex < block.getPositionCount(); rowIndex++) { + int sourceRowIndex = ascending ? rowIndex : block.getPositionCount() - 1 - rowIndex; + assertTrue(reader.hasNextTimeValuePair()); + assertRow(block, sourceRowIndex, reader.currentTimeValuePair()); + assertRow(block, sourceRowIndex, reader.currentTimeValuePair()); + assertRow(block, sourceRowIndex, reader.nextTimeValuePair()); + } + assertFalse(reader.hasNextTimeValuePair()); + assertNull(reader.currentTimeValuePair()); + // The heap releases this reservation after consuming the last row. + assertEquals(reservedMemory, reader.getUsedMemorySize()); + } + } + } + + @Test + public void testEmptyPageInBothDirections() { + TsBlock block = new TsBlockBuilder(DATA_TYPES).build(); + for (boolean ascending : new boolean[] {true, false}) { + try (FlatDevicePointReader reader = new FlatDevicePointReader(block, ascending)) { + assertFalse(reader.hasNextTimeValuePair()); + assertNull(reader.currentTimeValuePair()); + assertNull(reader.nextTimeValuePair()); + } + } + } + + @Test + public void testSingleRowInBothDirections() { + TsBlock block = + block(new long[] {1}, new String[] {"a"}, new Long[] {null}, new String[] {"x"}); + for (boolean ascending : new boolean[] {true, false}) { + try (FlatDevicePointReader reader = new FlatDevicePointReader(block, ascending)) { + assertTrue(reader.hasNextTimeValuePair()); + assertRow(block, 0, reader.nextTimeValuePair()); + assertFalse(reader.hasNextTimeValuePair()); + assertNull(reader.currentTimeValuePair()); + } + } + } + + @Test + public void testCloseBeforePageIsExhausted() { + TsBlock block = + block( + new long[] {100, 1}, + new String[] {"a", "b"}, + new Long[] {100L, 1L}, + new String[] {"x", "x"}); + for (boolean ascending : new boolean[] {true, false}) { + FlatDevicePointReader reader = new FlatDevicePointReader(block, ascending); + long reservedMemory = reader.getUsedMemorySize(); + reader.nextTimeValuePair(); + reader.close(); + assertFalse(reader.hasNextTimeValuePair()); + assertNull(reader.currentTimeValuePair()); + assertEquals(reservedMemory, reader.getUsedMemorySize()); + reader.close(); + } + } + + @Test + public void testOverlappingPagesInBothDirections() throws IOException { + TsBlock older = + block( + new long[] {10, 20, 1, 50}, + new String[] {"a", "a", "b", "b"}, + new Long[] {10L, 20L, 30L, 40L}, + new String[] {"x", "x", "x", "x"}); + TsBlock newer = + block( + new long[] {20, 30, 1, 2}, + new String[] {"a", "a", "b", "c"}, + new Long[] {null, 300L, 301L, 400L}, + new String[] {"x", "x", "x", "x"}); + // Duplicate keys retain the newer value, filling its null FIELD from the older page. + TsBlock expected = + block( + new long[] {10, 20, 30, 1, 50, 2}, + new String[] {"a", "a", "a", "b", "b", "c"}, + new Long[] {10L, 20L, 300L, 301L, 40L, 400L}, + new String[] {"x", "x", "x", "x", "x", "x"}); + for (boolean ascending : new boolean[] {true, false}) { + PriorityMergeReader mergeReader; + if (ascending) { + mergeReader = new FlatDevicePriorityMergeReader(DATA_TYPES, List.of(0, 2)); + } else { + AlignedDescPriorityMergeReader descendingReader = new FlatDeviceDescPriorityMergeReader(); + descendingReader.configureRowKeyColumns(DATA_TYPES, List.of(0, 2)); + mergeReader = descendingReader; + } + try { + mergeReader.addReader( + new FlatDevicePointReader(older, ascending), + new MergeReaderPriority(1, 1, 0, false), + ascending ? 50 : 1); + mergeReader.addReader( + new FlatDevicePointReader(newer, ascending), + new MergeReaderPriority(2, 2, 0, false), + ascending ? 30 : 1); + for (int rowIndex = 0; rowIndex < expected.getPositionCount(); rowIndex++) { + assertTrue(mergeReader.hasNextTimeValuePair()); + int sourceRowIndex = ascending ? rowIndex : expected.getPositionCount() - 1 - rowIndex; + assertRow(expected, sourceRowIndex, mergeReader.nextTimeValuePair()); + } + assertFalse(mergeReader.hasNextTimeValuePair()); + assertEquals(0, mergeReader.getUsedMemorySize()); + } finally { + mergeReader.close(); + } + } + } + + private static TsBlock block( + long[] times, String[] firstTags, Long[] fields, String[] secondTags) { + TsBlockBuilder builder = new TsBlockBuilder(DATA_TYPES); + for (int rowIndex = 0; rowIndex < times.length; rowIndex++) { + builder.getTimeColumnBuilder().writeLong(times[rowIndex]); + if (firstTags[rowIndex] == null) { + builder.getColumnBuilder(0).appendNull(); + } else { + builder + .getColumnBuilder(0) + .writeBinary(new Binary(firstTags[rowIndex], TSFileConfig.STRING_CHARSET)); + } + if (fields[rowIndex] == null) { + builder.getColumnBuilder(1).appendNull(); + } else { + builder.getColumnBuilder(1).writeLong(fields[rowIndex]); + } + if (secondTags[rowIndex] == null) { + builder.getColumnBuilder(2).appendNull(); + } else { + builder + .getColumnBuilder(2) + .writeBinary(new Binary(secondTags[rowIndex], TSFileConfig.STRING_CHARSET)); + } + builder.declarePosition(); + } + return builder.build(); + } + + private static void assertRow(TsBlock expected, int rowIndex, TimeValuePair actual) { + assertEquals(expected.getTimeByIndex(rowIndex), actual.getTimestamp()); + TsPrimitiveType[] values = actual.getValue().getVector(); + assertEquals(expected.getValueColumnCount(), values.length); + for (int columnIndex = 0; columnIndex < values.length; columnIndex++) { + if (expected.getColumn(columnIndex).isNull(rowIndex)) { + assertNull(values[columnIndex]); + } else { + assertEquals( + expected.getColumn(columnIndex).getTsPrimitiveType(rowIndex), values[columnIndex]); + } + } + } +}
