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 b03dcedc93516766552c81561fc78fb74725d517 Author: Tian Jiang <[email protected]> AuthorDate: Tue Sep 15 18:02:11 2026 +0800 tmp --- .../utils/writer/AbstractCompactionWriter.java | 44 +++++++++++++++++- .../writer/AbstractCrossCompactionWriter.java | 23 ++++++++-- .../writer/AbstractInnerCompactionWriter.java | 1 + .../reader/common/AlignedPriorityMergeReader.java | 2 +- .../utils/writer/AbstractCompactionWriterTest.java | 50 +++++++++++++++++++++ .../CompactionTableModelTestFileWriter.java | 52 ++++++++++++++++++++++ .../common/AlignedPriorityMergeReaderTest.java | 43 ++++++++++++++++++ 7 files changed, 209 insertions(+), 6 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java index 56ee94acf0c..eff07d21260 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java @@ -30,6 +30,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.exe import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer.flushcontroller.AbstractCompactionFlushController; import org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFileWriter; import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.AlignedPriorityMergeReader; import org.apache.tsfile.block.column.Column; import org.apache.tsfile.encrypt.EncryptParameter; @@ -43,6 +44,7 @@ import org.apache.tsfile.file.metadata.statistics.Statistics; import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.common.Chunk; 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.chunk.AlignedChunkWriterImpl; import org.apache.tsfile.write.chunk.ChunkWriterImpl; @@ -52,7 +54,10 @@ import org.apache.tsfile.write.schema.Schema; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Collections; import java.util.List; +import java.util.stream.Collectors; public abstract class AbstractCompactionWriter implements AutoCloseable { protected int subTaskNum = IoTDBDescriptor.getInstance().getConfig().getSubCompactionTaskNum(); @@ -116,6 +121,36 @@ public abstract class AbstractCompactionWriter implements AutoCloseable { protected IDeviceID deviceId; + protected List<Integer> rowKeyColumnIndices = Collections.emptyList(); + private TimeValuePair lastRowKey; + + public void setRowKeyColumnIndices(List<Integer> rowKeyColumnIndices) { + this.rowKeyColumnIndices = rowKeyColumnIndices; + lastRowKey = null; + } + + protected void checkFlatDeviceRowOrder(TsBlock block) throws IOException { + if (rowKeyColumnIndices.isEmpty()) { + return; + } + List<TSDataType> types = + Arrays.stream(block.getValueColumns()) + .map(Column::getDataType) + .collect(Collectors.toList()); + IPointReader rows = block.getTsBlockAlignedRowIterator(); + while (rows.hasNextTimeValuePair()) { + TimeValuePair row = rows.nextTimeValuePair(); + if (lastRowKey != null + && AlignedPriorityMergeReader.compareAlignedRowKey( + lastRowKey, row, types, rowKeyColumnIndices) + >= 0) { + throw new CompactionLastTimeCheckFailedException( + deviceId.toString(), row.getTimestamp(), lastRowKey.getTimestamp()); + } + lastRowKey = row; + } + } + protected String[] measurementId = new String[subTaskNum]; protected ModEntry ttlDeletionForCurrentDevice; @@ -126,6 +161,11 @@ public abstract class AbstractCompactionWriter implements AutoCloseable { public abstract void startChunkGroup(IDeviceID deviceId, boolean isAlign) throws IOException; + public void startChunkGroup(IDeviceID deviceId, boolean isAlign, boolean isFlatDevice) + throws IOException { + startChunkGroup(deviceId, isAlign); + } + public abstract void endChunkGroup() throws IOException; public void setTTLDeletion(ModEntry ttlDeletion) { @@ -478,7 +518,9 @@ public abstract class AbstractCompactionWriter implements AutoCloseable { } protected void checkPreviousTimestamp(long currentWritingTimestamp, int subTaskId) { - if (lastTimeSet[subTaskId] && currentWritingTimestamp <= lastTime[subTaskId]) { + if (lastTimeSet[subTaskId] + && (currentWritingTimestamp < lastTime[subTaskId] + || (currentWritingTimestamp == lastTime[subTaskId] && rowKeyColumnIndices.isEmpty()))) { throw new CompactionLastTimeCheckFailedException( deviceId.toString() + IoTDBConstant.PATH_SEPARATOR + measurementId[subTaskId], currentWritingTimestamp, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java index ad5e6e53c64..f2e518021ce 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java @@ -37,11 +37,13 @@ import org.apache.tsfile.file.metadata.TimeseriesMetadata; 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.Pair; import org.apache.tsfile.utils.TsPrimitiveType; 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; import java.util.Optional; @@ -113,10 +115,17 @@ public abstract class AbstractCrossCompactionWriter extends AbstractCompactionWr @Override public void startChunkGroup(IDeviceID deviceId, boolean isAlign) throws IOException { + startChunkGroup(deviceId, isAlign, false); + } + + @Override + public void startChunkGroup(IDeviceID deviceId, boolean isAlign, boolean isFlatDevice) + throws IOException { this.deviceId = deviceId; this.isAlign = isAlign; + setRowKeyColumnIndices(Collections.emptyList()); this.seqFileIndexArray = new int[subTaskNum]; - checkIsDeviceExistAndGetDeviceEndTime(); + checkIsDeviceExistAndGetDeviceEndTime(isFlatDevice); for (CompactionTsFileWriter targetFileWriter : targetFileWriters) { chunkGroupHeaderSize = targetFileWriter.startChunkGroup(deviceId); } @@ -228,11 +237,16 @@ public abstract class AbstractCrossCompactionWriter extends AbstractCompactionWr } } - private void checkIsDeviceExistAndGetDeviceEndTime() throws IOException { + private void checkIsDeviceExistAndGetDeviceEndTime(boolean isFlatDevice) throws IOException { int fileIndex = 0; while (fileIndex < seqTsFileResources.size()) { ITimeIndex timeIndex = seqTsFileResources.get(fileIndex).getTimeIndex(); - if (timeIndex.getTimeIndexType() != ITimeIndex.FILE_TIME_INDEX_TYPE) { + if (isFlatDevice) { + Optional<Pair<Long, Long>> tableTime = + seqTsFileResources.get(fileIndex).getTableTime(deviceId.getTableName()); + currentDeviceEndTime[fileIndex] = tableTime.map(time -> time.right).orElse(Long.MIN_VALUE); + isCurrentDeviceExistedInSourceSeqFiles[fileIndex] = tableTime.isPresent(); + } else if (timeIndex.getTimeIndexType() != ITimeIndex.FILE_TIME_INDEX_TYPE) { // the timeIndexType of resource is deviceTimeIndex Optional<Long> endTime = timeIndex.getEndTime(deviceId); currentDeviceEndTime[fileIndex] = endTime.orElse(Long.MIN_VALUE); @@ -241,7 +255,8 @@ public abstract class AbstractCrossCompactionWriter extends AbstractCompactionWr long endTime = Long.MIN_VALUE; boolean isDeviceExist = false; // Fast compaction get reader from cache map, while read point compaction get reader from - // FileReaderManager + // FileReaderManager. This path is only used for file-level indexes; flat devices use the + // table-level resource index above because their logical devices are not indexed here. Map<String, TimeseriesMetadata> deviceMetadataMap = getFileReader(seqTsFileResources.get(fileIndex)).readDeviceMetadata(deviceId); for (Map.Entry<String, TimeseriesMetadata> entry : deviceMetadataMap.entrySet()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java index 4e840a83687..13d8a9dac8b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java @@ -85,6 +85,7 @@ public abstract class AbstractInnerCompactionWriter extends AbstractCompactionWr fileWriter.startChunkGroup(deviceId); this.isAlign = isAlign; this.deviceId = deviceId; + this.rowKeyColumnIndices = Collections.emptyList(); } private CompactionTsFileWriter getAvailableWriter() throws IOException { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReader.java index d43c7dc8426..771da6c7698 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReader.java @@ -54,7 +54,7 @@ public class AlignedPriorityMergeReader extends PriorityMergeReader { return compareAlignedRowKey(left, right, dataTypes, rowKeyColumnIndices); } - static int compareAlignedRowKey( + public static int compareAlignedRowKey( TimeValuePair left, TimeValuePair right, List<TSDataType> dataTypes, diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java index de28c629215..11035a2a651 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java @@ -22,6 +22,8 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.wr import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.AlignedPageElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.ChunkMetadataElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer.flushcontroller.AbstractCompactionFlushController; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ArrayDeviceTimeIndex; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.enums.TSDataType; @@ -41,12 +43,60 @@ import org.junit.Test; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Collections; import java.util.List; public class AbstractCompactionWriterTest { private static final int SUB_TASK_ID = 0; + @Test + public void testCrossCompactionUsesExplicitFlatDeviceProperty() throws IOException { + IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1"}); + TsFileResource source = new TsFileResource(); + source.setTimeIndex(new ArrayDeviceTimeIndex()); + // A normal table with all-null TAGs also has a single-segment device ID. Its device + // boundary must not be replaced by the aggregate boundary of the entire table. + source.updateStartTime(device, 1); + source.updateEndTime(device, 20); + source.updateTableStartTime("table1", 1); + source.updateTableEndTime("table1", 100); + try (ReadPointCrossCompactionWriter writer = + new ReadPointCrossCompactionWriter( + Collections.emptyList(), Collections.singletonList(source))) { + writer.startChunkGroup(device, true, true); + Assert.assertEquals(100, writer.currentDeviceEndTime[0]); + Assert.assertTrue(writer.isCurrentDeviceExistedInSourceSeqFiles[0]); + + writer.startChunkGroup(device, true, false); + Assert.assertEquals(20, writer.currentDeviceEndTime[0]); + Assert.assertTrue(writer.isCurrentDeviceExistedInSourceSeqFiles[0]); + + writer.startChunkGroup(device, true); + Assert.assertEquals(20, writer.currentDeviceEndTime[0]); + } + } + + @Test + public void testCrossCompactionFlatDeviceDoesNotRequireDeviceTimeIndexEntry() throws IOException { + IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1"}); + TsFileResource source = new TsFileResource(); + source.setTimeIndex(new ArrayDeviceTimeIndex()); + source.updateTableStartTime("table1", 1); + source.updateTableEndTime("table1", 100); + try (ReadPointCrossCompactionWriter writer = + new ReadPointCrossCompactionWriter( + Collections.emptyList(), Collections.singletonList(source))) { + writer.startChunkGroup(device, true, true); + Assert.assertEquals(100, writer.currentDeviceEndTime[0]); + Assert.assertTrue(writer.isCurrentDeviceExistedInSourceSeqFiles[0]); + + writer.startChunkGroup(device, true, false); + Assert.assertEquals(Long.MIN_VALUE, writer.currentDeviceEndTime[0]); + Assert.assertFalse(writer.isCurrentDeviceExistedInSourceSeqFiles[0]); + } + } + @Test public void testBinarySizeCheckpointTriggersChunkSizeCheckBeforePointCheckpoint() throws IOException, PageException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java index f586f6a5473..302b33d2145 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java @@ -19,10 +19,12 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.tablemodel; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionTableSchema; import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionTestFileWriter; 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.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; @@ -30,6 +32,8 @@ import org.apache.tsfile.file.metadata.StringArrayDeviceID; import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.file.metadata.enums.CompressionType; import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.utils.Binary; +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.schema.Schema; @@ -75,6 +79,54 @@ public class CompactionTableModelTestFileWriter extends CompactionTestFileWriter schema.registerTableSchema(new TableSchema(tableName, measurementSchemas, columnTypes)); } + public void registerFlatDeviceSchema(String tableName) { + List<IMeasurementSchema> measurementSchemas = + List.of( + new MeasurementSchema( + "tag1", TSDataType.STRING, TSEncoding.DICTIONARY, CompressionType.UNCOMPRESSED), + new MeasurementSchema( + "field", TSDataType.INT64, TSEncoding.PLAIN, CompressionType.UNCOMPRESSED), + new MeasurementSchema( + "tag2", TSDataType.STRING, TSEncoding.DICTIONARY, CompressionType.UNCOMPRESSED)); + List<ColumnCategory> columnTypes = + List.of(ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.TAG); + schema.registerTableSchema(new TableSchema(tableName, measurementSchemas, columnTypes)); + TsFileTableSchemaUtil.addFlatDeviceProperty(fileWriter, tableName, true); + } + + public void generateFlatDeviceAlignedRows( + long[] timestamps, String[] tag1Values, long[] fieldValues, String[] tag2Values) + throws IOException { + if (timestamps.length != tag1Values.length + || timestamps.length != fieldValues.length + || timestamps.length != tag2Values.length) { + throw new IllegalArgumentException("Flat-device row columns must have the same length"); + } + List<IMeasurementSchema> measurementSchemas = + List.of( + new MeasurementSchema( + "tag1", TSDataType.STRING, TSEncoding.DICTIONARY, CompressionType.UNCOMPRESSED), + new MeasurementSchema( + "field", TSDataType.INT64, TSEncoding.PLAIN, CompressionType.UNCOMPRESSED), + new MeasurementSchema( + "tag2", TSDataType.STRING, TSEncoding.DICTIONARY, CompressionType.UNCOMPRESSED)); + AlignedChunkWriterImpl chunkWriter = new AlignedChunkWriterImpl(measurementSchemas); + for (int row = 0; row < timestamps.length; row++) { + long timestamp = timestamps[row]; + currentDeviceStartTime = Math.min(timestamp, currentDeviceStartTime); + currentDeviceEndTime = Math.max(timestamp, currentDeviceEndTime); + chunkWriter.getTimeChunkWriter().write(timestamp); + chunkWriter + .getValueChunkWriterByIndex(0) + .write(timestamp, new Binary(tag1Values[row], TSFileConfig.STRING_CHARSET), false); + chunkWriter.getValueChunkWriterByIndex(1).write(timestamp, fieldValues[row], false); + chunkWriter + .getValueChunkWriterByIndex(2) + .write(timestamp, new Binary(tag2Values[row], TSFileConfig.STRING_CHARSET), false); + } + chunkWriter.writeToFileWriter(fileWriter); + } + public IDeviceID startChunkGroup(String tableName, List<String> idColumnFields) throws IOException { int idx = 0; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReaderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReaderTest.java index befced09aa9..ff02f97b064 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReaderTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReaderTest.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.storageengine.dataregion.read.reader.common; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.reader.IPointReader; +import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.TsPrimitiveType; import org.apache.tsfile.utils.TsPrimitiveType.TsInt; import org.apache.tsfile.utils.TsPrimitiveType.TsLong; @@ -28,6 +29,7 @@ import org.apache.tsfile.utils.TsPrimitiveType.TsVector; import org.junit.Test; import java.io.IOException; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -118,6 +120,36 @@ public class AlignedPriorityMergeReaderTest { new long[] {30, 20, 11}); } + @Test + public void testFlatDeviceRowsAreOrderedByTagColumns() throws IOException { + // A flat-device aligned vector may contain fields between TAG columns. The row-key indices + // therefore refer to the projected vector positions, rather than to the TAG ordinal. + AlignedPriorityMergeReader mergeReader = + new AlignedPriorityMergeReader( + Arrays.asList(TSDataType.INT32, TSDataType.TEXT, TSDataType.INT32), + Arrays.asList(0, 2)); + mergeReader.addReader( + new ListPointReader( + Arrays.asList(flatDeviceRow(1, 10, "x", 1), flatDeviceRow(1, 20, "z", 2))), + new MergeReaderPriority(Long.MAX_VALUE, 1, 0, false), + 1); + mergeReader.addReader( + new ListPointReader( + Arrays.asList(flatDeviceRow(1, 10, "x", 3), flatDeviceRow(1, 20, "z", 4))), + new MergeReaderPriority(Long.MAX_VALUE, 2, 0, false), + 1); + + List<String> keys = new ArrayList<>(); + List<String> values = new ArrayList<>(); + while (mergeReader.hasNextTimeValuePair()) { + TsPrimitiveType[] vector = mergeReader.nextTimeValuePair().getValue().getVector(); + keys.add(vector[0].getInt() + ":" + vector[2].getInt()); + values.add(vector[1].getBinary().getStringValue(StandardCharsets.UTF_8)); + } + assertEquals(Arrays.asList("10:1", "10:3", "20:2", "20:4"), keys); + assertEquals(Arrays.asList("x", "x", "z", "z"), values); + } + private static void assertCompoundMerge( PriorityMergeReader mergeReader, List<TimeValuePair> lowPriorityRows, @@ -150,6 +182,17 @@ public class AlignedPriorityMergeReaderTest { 1, new TsVector(new TsPrimitiveType[] {new TsInt(key), new TsLong(value)})); } + private static TimeValuePair flatDeviceRow(long time, int tag1, String field, int tag2) { + return new TimeValuePair( + time, + new TsVector( + new TsPrimitiveType[] { + new TsInt(tag1), + TsPrimitiveType.getByType(TSDataType.TEXT, new Binary(field, StandardCharsets.UTF_8)), + new TsInt(tag2) + })); + } + private static final class ListPointReader implements IPointReader { private final List<TimeValuePair> rows;
