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 59ca9e97452a41836d1533ef7f938a870427d6da Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 23 15:12:34 2026 +0800 tmp --- .../db/utils/datastructure/FlatDeviceEncoder.java | 181 ++++++++--- .../utils/datastructure/FlatDeviceEncoderTest.java | 336 +++++++++++++++++++++ 2 files changed, 480 insertions(+), 37 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoder.java index 99f4951668b..b82c9bee802 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoder.java @@ -39,18 +39,23 @@ final class FlatDeviceEncoder { static void encode( List<FlatDeviceAlignedTVList> lists, List<IMeasurementSchema> schemas, - List<Integer> columns, + List<Integer> columnIndices, EncryptParameter encryption, BlockingQueue<Object> output, long chunkLimit, int pageLimit) { + if (lists.size() == 1) { + encodeSingleList( + lists.get(0), schemas, columnIndices, encryption, output, chunkLimit, pageLimit); + return; + } PriorityQueue<Cursor> devices = new PriorityQueue<>( (a, b) -> FlatDeviceAlignedTVList.compareDevices(a.device.id, b.device.id)); PriorityQueue<Cursor> rows = new PriorityQueue<>( (a, b) -> { - int result = a.list.compareWithinDevice(a.row(), b.list, b.row()); + int result = a.list.compareWithinDevice(a.rowIndex(), b.list, b.rowIndex()); // Older lists and earlier inserts are consumed first. A later non-null value wins. return result == 0 ? Integer.compare(a.priority, b.priority) : result; }); @@ -62,8 +67,8 @@ final class FlatDeviceEncoder { devices.add(cursor); } } - FlatDeviceAlignedTVList[] selectedLists = new FlatDeviceAlignedTVList[columns.size()]; - int[] selectedRows = new int[columns.size()]; + FlatDeviceAlignedTVList[] selectedLists = new FlatDeviceAlignedTVList[columnIndices.size()]; + int[] selectedRowIndices = new int[columnIndices.size()]; FlatDeviceAlignedChunkWriter writer = new FlatDeviceAlignedChunkWriter(schemas, encryption); long chunkRows = 0; while (!devices.isEmpty()) { @@ -79,20 +84,22 @@ final class FlatDeviceEncoder { while (!rows.isEmpty()) { Cursor first = rows.peek(); FlatDeviceAlignedTVList keyList = first.list; - int keyRow = first.row(); - long time = keyList.fieldStorage().getTime(keyRow); + int keyRowIndex = first.rowIndex(); + long time = keyList.fieldStorage().getTime(keyRowIndex); Arrays.fill(selectedLists, null); do { Cursor cursor = rows.remove(); - int row = cursor.row(); - for (int i = 0; i < columns.size(); i++) { - int column = columns.get(i); - int tag = cursor.list.tagOrdinal(column); + int rowIndex = cursor.rowIndex(); + for (int i = 0; i < columnIndices.size(); i++) { + int columnIndex = columnIndices.get(i); + int tagColumnIndex = cursor.list.tagOrdinal(columnIndex); boolean nil = - tag >= 0 ? cursor.device.tag(tag) == null : cursor.list.isNullValue(row, column); + tagColumnIndex >= 0 + ? cursor.device.tag(tagColumnIndex) == null + : cursor.list.isNullValue(rowIndex, columnIndex); if (!nil) { selectedLists[i] = cursor.list; - selectedRows[i] = row; + selectedRowIndices[i] = rowIndex; } } if (cursor.nextRow()) { @@ -101,40 +108,105 @@ final class FlatDeviceEncoder { devices.add(cursor); } } while (!rows.isEmpty() - && keyList.compareWithinDevice(keyRow, rows.peek().list, rows.peek().row()) == 0); + && keyList.compareWithinDevice(keyRowIndex, rows.peek().list, rows.peek().rowIndex()) + == 0); // Reuse the ordinary typed chunk writers; only the traversal and duplicate selection // differ. No TimeValuePair, vector or per-row iterator wrapper is materialized. - for (int i = 0; i < columns.size(); i++) { + for (int i = 0; i < columnIndices.size(); i++) { FlatDeviceAlignedTVList list = selectedLists[i]; - int row = selectedRows[i]; - int column = columns.get(i); + int rowIndex = selectedRowIndices[i]; + int columnIndex = columnIndices.get(i); boolean nil = list == null; - switch (schemas.get(i).getType()) { - case BOOLEAN -> - writer.write(time, !nil && list.getBooleanByValueIndex(row, column), nil); - case INT32, DATE -> - writer.write(time, nil ? 0 : list.getIntByValueIndex(row, column), nil); - case INT64, TIMESTAMP -> - writer.write(time, nil ? 0L : list.getLongByValueIndex(row, column), nil); - case FLOAT -> - writer.write(time, nil ? 0F : list.getFloatByValueIndex(row, column), nil); - case DOUBLE -> - writer.write(time, nil ? 0D : list.getDoubleByValueIndex(row, column), nil); - case STRING, TEXT, BLOB, OBJECT -> { - Binary value = Binary.EMPTY_VALUE; - if (!nil) { - int tag = list.tagOrdinal(column); - value = tag >= 0 ? currentDevice.tag(tag) : list.getBinaryByValueIndex(row, column); + int tagColumnIndex = list == null ? -1 : list.tagOrdinal(columnIndex); + writeValue( + writer, + schemas.get(i), + time, + list, + currentDevice, + rowIndex, + columnIndex, + tagColumnIndex, + nil); + } + writer.write(time); + chunkRows++; + // The chunk writer may already have sealed on its byte-size threshold. + if (writer.getTimeChunkWriter().getPageWriter().getPointNumber() >= pageLimit) { + writer.sealCurrentPage(); + } + if (chunkRows >= chunkLimit) { + if (!enqueue(writer, output)) { + return; + } + writer = new FlatDeviceAlignedChunkWriter(schemas, encryption); + chunkRows = 0; + } + } + } + if (chunkRows > 0) { + enqueue(writer, output); + } + } + + private static void encodeSingleList( + FlatDeviceAlignedTVList list, + List<IMeasurementSchema> schemas, + List<Integer> columnIndices, + EncryptParameter encryption, + BlockingQueue<Object> output, + long chunkLimit, + int pageLimit) { + list.sort(); + FlatDeviceAlignedChunkWriter writer = new FlatDeviceAlignedChunkWriter(schemas, encryption); + long chunkRows = 0; + for (DeviceInfo device : list.getDeviceInfoMap().values()) { + int position = 0; + while (position < device.rowCount) { + int start = position++; + int keyRowIndex = device.rowIndices[start]; + if (list.isTimeDeleted(keyRowIndex, false)) { + continue; + } + while (position < device.rowCount + && list.compareWithinDevice(keyRowIndex, list, device.rowIndices[position]) == 0) { + position++; + } + long time = list.fieldStorage().getTime(keyRowIndex); + for (int i = 0; i < columnIndices.size(); i++) { + int columnIndex = columnIndices.get(i); + int tagColumnIndex = list.tagOrdinal(columnIndex); + int rowIndex = keyRowIndex; + boolean nil; + if (tagColumnIndex >= 0) { + nil = device.tag(tagColumnIndex) == null; + } else { + // Equal keys are sorted by insertion order. Select each FIELD from the end without + // allocating merge queues, cursors or per-column source arrays for a single list. + nil = true; + for (int candidate = position - 1; candidate >= start; candidate--) { + rowIndex = device.rowIndices[candidate]; + if (!list.isTimeDeleted(rowIndex, false) + && !list.isNullValue(rowIndex, columnIndex)) { + nil = false; + break; } - writer.write(time, value, nil); } - case VECTOR, UNKNOWN -> throw new IllegalArgumentException(); } + writeValue( + writer, + schemas.get(i), + time, + list, + device, + rowIndex, + columnIndex, + tagColumnIndex, + nil); } writer.write(time); chunkRows++; - // The chunk writer may already have sealed on its byte-size threshold. if (writer.getTimeChunkWriter().getPageWriter().getPointNumber() >= pageLimit) { writer.sealCurrentPage(); } @@ -152,6 +224,41 @@ final class FlatDeviceEncoder { } } + private static void writeValue( + FlatDeviceAlignedChunkWriter writer, + IMeasurementSchema schema, + long time, + FlatDeviceAlignedTVList list, + DeviceInfo device, + int rowIndex, + int columnIndex, + int tagColumnIndex, + boolean nil) { + switch (schema.getType()) { + case BOOLEAN -> + writer.write(time, !nil && list.getBooleanByValueIndex(rowIndex, columnIndex), nil); + case INT32, DATE -> + writer.write(time, nil ? 0 : list.getIntByValueIndex(rowIndex, columnIndex), nil); + case INT64, TIMESTAMP -> + writer.write(time, nil ? 0L : list.getLongByValueIndex(rowIndex, columnIndex), nil); + case FLOAT -> + writer.write(time, nil ? 0F : list.getFloatByValueIndex(rowIndex, columnIndex), nil); + case DOUBLE -> + writer.write(time, nil ? 0D : list.getDoubleByValueIndex(rowIndex, columnIndex), nil); + case STRING, TEXT, BLOB, OBJECT -> { + Binary value = Binary.EMPTY_VALUE; + if (!nil) { + value = + tagColumnIndex >= 0 + ? device.tag(tagColumnIndex) + : list.getBinaryByValueIndex(rowIndex, columnIndex); + } + writer.write(time, value, nil); + } + case VECTOR, UNKNOWN -> throw new IllegalArgumentException(); + } + } + private static boolean enqueue( FlatDeviceAlignedChunkWriter writer, BlockingQueue<Object> output) { writer.sealCurrentPage(); @@ -178,7 +285,7 @@ final class FlatDeviceEncoder { devices = list.getDeviceInfoMap().values().iterator(); } - private int row() { + private int rowIndex() { return device.rowIndices[position]; } @@ -196,7 +303,7 @@ final class FlatDeviceEncoder { private boolean nextRow() { do { position++; - } while (position < device.rowCount && list.isTimeDeleted(row(), false)); + } while (position < device.rowCount && list.isTimeDeleted(rowIndex(), false)); return position < device.rowCount; } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoderTest.java new file mode 100644 index 00000000000..c4bdc712eb0 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoderTest.java @@ -0,0 +1,336 @@ +/* + * 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.utils.datastructure; + +import org.apache.iotdb.db.utils.EncryptDBUtils; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata; +import org.apache.tsfile.file.metadata.ChunkMetadata; +import org.apache.tsfile.file.metadata.IChunkMetadata; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.BatchData; +import org.apache.tsfile.read.common.Chunk; +import org.apache.tsfile.read.reader.chunk.TableChunkReader; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.TsPrimitiveType; +import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.File; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public class FlatDeviceEncoderTest { + @Rule public TemporaryFolder folder = new TemporaryFolder(); + + private static Binary binary(String value) { + return new Binary(value, TSFileConfig.STRING_CHARSET); + } + + private static FlatDeviceAlignedTVList list(List<TSDataType> types, List<Integer> tags) { + return (FlatDeviceAlignedTVList) + AlignedTVListFactory.createFlatDeviceAlignedList( + types, tags, IDeviceID.Factory.DEFAULT_FACTORY.create("table")); + } + + @Test + public void testDeviceMergeLatestNonNullAndPageChunkBoundaries() throws Exception { + List<TSDataType> types = List.of(TSDataType.STRING, TSDataType.INT64, TSDataType.INT64); + FlatDeviceAlignedTVList old = list(types, List.of(0)); + old.putAlignedValue(1, new Object[] {binary("b"), 10L, 100L}); + old.putAlignedValue(100, new Object[] {binary("a"), 1L, 11L}); + old.putAlignedValue(2, new Object[] {binary("b"), 20L, 200L}); + old.putAlignedValue(3, new Object[] {binary("c"), 30L, 300L}); + FlatDeviceAlignedTVList current = list(types, List.of(0)); + current.putAlignedValue(1, new Object[] {binary("b"), 12L, null}); + current.putAlignedValue(1, new Object[] {binary("b"), null, 102L}); + current.putAlignedValue(4, new Object[] {binary("b"), 40L, 400L}); + current.putAlignedValue(2, new Object[] {binary("b"), 999L, 999L}); + current.deleteMatchingRows(2, 2, row -> true, null); + FlatDeviceAlignedTVList last = list(types, List.of(0)); + last.putAlignedValue(1, new Object[] {binary("b"), 13L, null}); + BlockingQueue<Object> queue = encode(List.of(old, current, last), types, 3, 2); + Assert.assertEquals(2, queue.size()); + List<AlignedChunkWriterImpl> writers = + queue.stream().map(AlignedChunkWriterImpl.class::cast).collect(Collectors.toList()); + Assert.assertEquals(3, writers.get(0).getTimeChunkWriter().getStatistics().getCount()); + Assert.assertEquals(1, writers.get(0).getTimeChunkWriter().getStatistics().getStartTime()); + Assert.assertEquals(100, writers.get(0).getTimeChunkWriter().getStatistics().getEndTime()); + Assert.assertEquals(2, writers.get(0).getTimeChunkWriter().getNumOfPages()); + Assert.assertEquals(1, writers.get(1).getTimeChunkWriter().getNumOfPages()); + + List<Row> rows = read(queue); + Assert.assertEquals(5, rows.size()); + assertRow(rows.get(0), 100, "a", 1L, 11L); + assertRow(rows.get(1), 1, "b", 13L, 102L); + assertRow(rows.get(2), 2, "b", 20L, 200L); + assertRow(rows.get(3), 4, "b", 40L, 400L); + assertRow(rows.get(4), 3, "c", 30L, 300L); + } + + @Test + public void testSingleListLatestNonNullDeletionAndPageChunkBoundaries() throws Exception { + List<TSDataType> types = List.of(TSDataType.STRING, TSDataType.INT64, TSDataType.INT64); + FlatDeviceAlignedTVList list = list(types, List.of(0)); + list.putAlignedValue(1, new Object[] {binary("b"), 10L, 100L}); + list.putAlignedValue(100, new Object[] {binary("a"), 1L, 11L}); + list.putAlignedValue(1, new Object[] {binary("b"), 12L, null}); + list.putAlignedValue(1, new Object[] {binary("b"), 999L, 999L}); + list.putAlignedValue(1, new Object[] {binary("b"), null, 102L}); + list.putAlignedValue(1, new Object[] {binary("b"), null, null}); + list.putAlignedValue(1, new Object[] {binary("b"), 999L, 999L}); + list.putAlignedValue(2, new Object[] {binary("b"), null, null}); + list.putAlignedValue(1, new Object[] {binary("c"), 30L, 300L}); + list.putAlignedValue(3, new Object[] {binary("d"), 999L, 999L}); + list.putAlignedValue(0, new Object[] {binary("b"), 999L, 999L}); + list.putAlignedValue(5, new Object[] {null, null, null}); + // Deleted duplicates must not win; fully deleted devices and leading rows emit nothing. + list.deleteMatchingRows( + Long.MIN_VALUE, Long.MAX_VALUE, row -> row == 3 || row == 6 || row == 9 || row == 10, null); + + BlockingQueue<Object> queue = encode(List.of(list), types, 3, 2); + Assert.assertEquals(2, queue.size()); + List<AlignedChunkWriterImpl> writers = + queue.stream().map(AlignedChunkWriterImpl.class::cast).collect(Collectors.toList()); + Assert.assertEquals(3, writers.get(0).getTimeChunkWriter().getStatistics().getCount()); + Assert.assertEquals(1, writers.get(0).getTimeChunkWriter().getStatistics().getStartTime()); + Assert.assertEquals(100, writers.get(0).getTimeChunkWriter().getStatistics().getEndTime()); + Assert.assertEquals(2, writers.get(0).getTimeChunkWriter().getNumOfPages()); + Assert.assertEquals(2, writers.get(1).getTimeChunkWriter().getStatistics().getCount()); + Assert.assertEquals(1, writers.get(1).getTimeChunkWriter().getNumOfPages()); + List<Row> rows = read(queue); + Assert.assertEquals(5, rows.size()); + assertRow(rows.get(0), 5, null, null, null); + assertRow(rows.get(1), 100, "a", 1L, 11L); + assertRow(rows.get(2), 1, "b", 12L, 102L); + assertRow(rows.get(3), 2, "b", null, null); + assertRow(rows.get(4), 1, "c", 30L, 300L); + } + + @Test + public void testSingleListDuplicateFieldKeysWithColumnProjection() throws Exception { + List<TSDataType> types = + List.of( + TSDataType.STRING, + TSDataType.INT32, + TSDataType.INT64, + TSDataType.TEXT, + TSDataType.STRING); + FlatDeviceAlignedTVList list = list(types, List.of(0, 4)); + list.setSortColumnIndices(List.of(1)); + list.putAlignedValue(1, new Object[] {binary("a"), 2, 20L, binary("old"), binary("x")}); + list.putAlignedValue(1, new Object[] {binary("a"), 1, 10L, binary("k1"), binary("x")}); + list.putAlignedValue(1, new Object[] {binary("a"), 2, null, binary("new"), binary("x")}); + list.putAlignedValue(1, new Object[] {binary("a"), 2, 22L, null, binary("x")}); + list.putAlignedValue(1, new Object[] {binary("b"), 2, 30L, binary("b"), binary("x")}); + // The omitted FIELD sort key still distinguishes equal timestamps. Reordered TAGs and a + // missing FIELD must use the output-to-source column mapping, not the output position. + List<Row> rows = + read( + encode( + List.of(list), + List.of( + TSDataType.TEXT, + TSDataType.STRING, + TSDataType.INT64, + TSDataType.STRING, + TSDataType.INT64), + List.of(3, 4, 2, 0, 5), + 2, + 1)); + Assert.assertEquals(3, rows.size()); + assertRow(rows.get(0), 1, "k1", "x", 10L, "a", null); + assertRow(rows.get(1), 1, "new", "x", 22L, "a", null); + assertRow(rows.get(2), 1, "b", "x", 30L, "b", null); + } + + @Test + public void testLateTagMissingFieldAndDeviceKeyOnly() throws Exception { + FlatDeviceAlignedTVList old = list(List.of(TSDataType.STRING, TSDataType.INT64), List.of(0)); + old.putAlignedValue(1, new Object[] {binary("a"), 10L}); + old.putAlignedValue(2, new Object[] {binary("b"), 20L}); + List<TSDataType> types = + List.of(TSDataType.STRING, TSDataType.INT64, TSDataType.STRING, TSDataType.INT64); + FlatDeviceAlignedTVList current = list(types, List.of(0, 2)); + current.putAlignedValue(1, new Object[] {binary("a"), null, null, 11L}); + current.putAlignedValue(1, new Object[] {binary("a"), 12L, binary("x"), null}); + current.putAlignedValue(5, new Object[] {null, null, null, null}); + List<Row> rows = read(encode(List.of(old, current), types, 100, 100)); + Assert.assertEquals(4, rows.size()); + assertRow(rows.get(0), 5, null, null, null, null); + assertRow(rows.get(1), 1, "a", 10L, null, 11L); + assertRow(rows.get(2), 1, "a", 12L, "x", null); + assertRow(rows.get(3), 2, "b", 20L, null, null); + } + + @Test + public void testFieldSortThenTimeAndAllValueTypes() throws Exception { + List<TSDataType> types = + List.of( + TSDataType.STRING, + TSDataType.INT32, + TSDataType.BOOLEAN, + TSDataType.FLOAT, + TSDataType.DOUBLE, + TSDataType.INT64, + TSDataType.TEXT, + TSDataType.STRING); + FlatDeviceAlignedTVList list = list(types, List.of(0)); + list.setSortColumnIndices(List.of(1)); + list.putAlignedValue( + 1, new Object[] {binary("a"), 2, true, 2.5F, 3.5D, 4L, binary("v"), binary("s")}); + list.putAlignedValue( + 9, new Object[] {binary("a"), 1, false, 1.5F, 2.5D, 3L, binary("w"), binary("t")}); + list.putAlignedValue(3, new Object[] {binary("a"), 1, null, null, null, null, null, null}); + list.putAlignedValue( + 3, new Object[] {binary("b"), 0, true, 0.5F, 1.5D, 2L, binary("x"), binary("u")}); + BlockingQueue<Object> queue = encode(List.of(list), types, 100, 100); + Assert.assertEquals( + 1, ((AlignedChunkWriterImpl) queue.peek()).getTimeChunkWriter().getNumOfPages()); + List<Row> rows = read(queue); + Assert.assertEquals(4, rows.size()); + assertRow(rows.get(0), 3, "a", 1, null, null, null, null, null, null); + assertRow(rows.get(1), 9, "a", 1, false, 1.5F, 2.5D, 3L, "w", "t"); + assertRow(rows.get(2), 1, "a", 2, true, 2.5F, 3.5D, 4L, "v", "s"); + assertRow(rows.get(3), 3, "b", 0, true, 0.5F, 1.5D, 2L, "x", "u"); + } + + @Test + public void testTimeOnlyTagOnlyAndEmptyLists() throws Exception { + FlatDeviceAlignedTVList timeOnly = list(List.of(), List.of()); + timeOnly.putAlignedValue(2, new Object[0]); + timeOnly.putAlignedValue(1, new Object[0]); + timeOnly.putAlignedValue(1, new Object[0]); + BlockingQueue<Object> queue = encode(List.of(timeOnly), List.of(), 100, 100); + Assert.assertEquals( + 2, + ((AlignedChunkWriterImpl) queue.remove()).getTimeChunkWriter().getStatistics().getCount()); + FlatDeviceAlignedTVList tagOnly = list(List.of(TSDataType.STRING), List.of(0)); + tagOnly.putAlignedValue(9, new Object[] {binary("a")}); + tagOnly.putAlignedValue(1, new Object[] {binary("b")}); + Assert.assertTrue(tagOnly.getValues().isEmpty()); + List<Row> rows = read(encode(List.of(tagOnly), List.of(TSDataType.STRING), 100, 100)); + assertRow(rows.get(0), 9, "a"); + assertRow(rows.get(1), 1, "b"); + tagOnly.deleteMatchingRows(Long.MIN_VALUE, Long.MAX_VALUE, row -> true, null); + Assert.assertTrue(encode(List.of(tagOnly), List.of(TSDataType.STRING), 100, 100).isEmpty()); + Assert.assertTrue(encode(List.of(), List.of(), 100, 100).isEmpty()); + } + + private static BlockingQueue<Object> encode( + List<FlatDeviceAlignedTVList> lists, List<TSDataType> types, long chunkLimit, int pageLimit) { + return encode( + lists, + types, + IntStream.range(0, types.size()).boxed().collect(Collectors.toList()), + chunkLimit, + pageLimit); + } + + private static BlockingQueue<Object> encode( + List<FlatDeviceAlignedTVList> lists, + List<TSDataType> types, + List<Integer> columns, + long chunkLimit, + int pageLimit) { + List<IMeasurementSchema> schemas = new ArrayList<>(); + for (int i = 0; i < types.size(); i++) { + schemas.add(new MeasurementSchema("s" + i, types.get(i))); + } + BlockingQueue<Object> queue = new LinkedBlockingQueue<>(); + FlatDeviceAlignedTVList.encode( + lists, + schemas, + columns, + EncryptDBUtils.getDefaultFirstEncryptParam(), + queue, + chunkLimit, + pageLimit); + return queue; + } + + private List<Row> read(BlockingQueue<Object> queue) throws Exception { + File file = folder.newFile(); + IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create("root.flat"); + try (TsFileIOWriter writer = new TsFileIOWriter(file)) { + writer.startChunkGroup(device); + for (Object task : queue) { + ((AlignedChunkWriterImpl) task).writeToFileWriter(writer); + } + writer.endChunkGroup(); + writer.endFile(); + } + List<Row> rows = new ArrayList<>(); + try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getPath())) { + for (AbstractAlignedChunkMetadata metadata : reader.getAlignedChunkMetadata(device, true)) { + Chunk time = reader.readMemChunk((ChunkMetadata) metadata.getTimeChunkMetadata()); + List<Chunk> values = new ArrayList<>(); + for (IChunkMetadata value : metadata.getValueChunkMetadataList()) { + values.add(value == null ? null : reader.readMemChunk((ChunkMetadata) value)); + } + // Table readers retain all-null rows, including a logical device whose TAGs are null. + TableChunkReader chunks = new TableChunkReader(time, values, null); + while (chunks.hasNextSatisfiedPage()) { + BatchData batch = chunks.nextPageData(); + while (batch.hasCurrent()) { + rows.add(new Row(batch.currentTime(), batch.getVector())); + batch.next(); + } + } + } + } + return rows; + } + + private static void assertRow(Row actual, long time, Object... expected) { + Assert.assertEquals(time, actual.time); + Assert.assertEquals(expected.length, actual.values.length); + for (int i = 0; i < expected.length; i++) { + Object value = expected[i]; + if (value instanceof String text) { + value = binary(text); + } + Assert.assertEquals(value, actual.values[i] == null ? null : actual.values[i].getValue()); + } + } + + private static final class Row { + private final long time; + private final TsPrimitiveType[] values; + + private Row(long time, TsPrimitiveType[] values) { + this.time = time; + this.values = values; + } + } +}
