From 08c57bfbfdfbaa552d964b501455de5ca92aaa96 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 27 Oct 2026 22:39:16 +0800 Subject: [PATCH 1/4] Fix LOAD splitting of sparse batched aligned chunks --- .../BatchedAlignedValueChunkData.java | 45 ++--- .../load/TsFileSplitterTest.java | 163 ++++++++++++++++++ 2 files changed, 176 insertions(+), 32 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java index 56bff063d7053..d29cc2601da07 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java @@ -33,7 +33,6 @@ import org.apache.tsfile.write.schema.MeasurementSchema; import java.io.IOException; -import java.io.Serializable; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collections; @@ -64,9 +63,8 @@ public BatchedAlignedValueChunkData(IDeviceID device, TTimePartitionSlot timePar } /** - * Override the encoding process to handle standalone Value Chunks without a Time Chunk. It uses - * ValueChunkWriter directly, writes a dummy time (0), and manually corrects the start/end time in - * the statistics. + * Encodes standalone value chunks while preserving the row and page positions of their time + * chunks. */ protected void encodeAndBuildChunks() throws IOException, PageException { final long startTime = timePartitionSlot.getStartTime(); @@ -120,20 +118,14 @@ protected void encodeAndBuildChunks() throws IOException, PageException { boolean hasBufferedData = false; for (PageBuffer page : chunkPages) { if (!page.needDecode) { - // An undecoded page may carry no data at all (for aligned series an all-null column is - // stored as an empty page). Writing its page header alone would produce a chunk whose - // data size does not describe a parsable page, so such a page is dropped. - if (page.pageData.remaining() > 0) { - valueChunkWriter.writePageHeaderAndDataIntoBuff(page.pageData, page.pageHeader); - hasBufferedData = true; - } + // Empty pages still occupy a slot opposite the corresponding time page. If every page + // is empty, the whole chunk is emitted without a body below. + hasBufferedData |= page.pageData.remaining() > 0; + valueChunkWriter.writePageHeaderAndDataIntoBuff(page.pageData, page.pageHeader); continue; } - // Keep track of actual bounds since we will write '0' as dummy time - long pageStartTime = Long.MAX_VALUE; - long pageEndTime = Long.MIN_VALUE; - boolean pageHasValue = false; + boolean pageHasRows = false; final TypeServices.TsPrimitiveValueChunkWriter valueChunkValueWriter = TypeServices.StorageEngine.TS_PRIMITIVE_VALUE_CHUNK_WRITER_SERVICE.call( @@ -148,27 +140,16 @@ protected void encodeAndBuildChunks() throws IOException, PageException { || page.valueBatch[j] == null); if (!isNull) { hasBufferedData = true; - pageHasValue = true; - pageStartTime = Math.min(pageStartTime, time); - pageEndTime = Math.max(pageEndTime, time); - - // Write 0 as a dummy time, valueChunkWriter will ignore it anyway for value columns - valueChunkValueWriter.write(valueChunkWriter, 0, page.valueBatch[j], false); } - // A standalone value chunk cannot store nulls: for aligned series a missing value is - // represented by the absence of the point, the time chunk still carries the row. + // The bitmap has one position per timestamp, including nulls. Use the real timestamp + // so the writer also computes the statistics of the non-null values correctly. + valueChunkValueWriter.write( + valueChunkWriter, time, isNull ? null : page.valueBatch[j], isNull); + pageHasRows = true; } } - // Correct the page statistics with the actual Start/End time - Statistics statistics = - valueChunkWriter.getPageWriter().getStatistics(); - if (pageStartTime <= pageEndTime) { - statistics.setStartTime(pageStartTime); - statistics.setEndTime(pageEndTime); - } - - if (pageHasValue) { + if (pageHasRows) { valueChunkWriter.sealCurrentPage(); } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/TsFileSplitterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/TsFileSplitterTest.java index d70cb40e8141c..e993c72a4bf6e 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/TsFileSplitterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/TsFileSplitterTest.java @@ -87,7 +87,9 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.TreeMap; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.LongPredicate; public class TsFileSplitterTest { @@ -908,6 +910,167 @@ public void testSplitBatchedAlignedChunkOfMultipleDevices() throws Exception { } } + @Test + public void testSplitBatchedAlignedChunkPreservesNullRows() throws Exception { + final long interval = TimePartitionUtils.getTimePartitionInterval(); + assertSparseBatchedAlignedChunkSplit( + new long[][] {{0, 1, 2, interval - 1, interval, interval + 1, interval + 2}}, + time -> time % 2 == 0); + } + + @Test + public void testSplitBatchedAlignedChunkPreservesCopiedEmptyPage() throws Exception { + final long interval = TimePartitionUtils.getTimePartitionInterval(); + assertSparseBatchedAlignedChunkSplit( + new long[][] {{0, 1}, {interval - 2, interval - 1, interval, interval + 1}}, + time -> time < 2); + } + + @Test + public void testSplitBatchedAlignedChunkPreservesDecodedEmptyPage() throws Exception { + final long interval = TimePartitionUtils.getTimePartitionInterval(); + assertSparseBatchedAlignedChunkSplit( + new long[][] { + {interval - 1, interval, interval + 1}, + {interval + 2, 2 * interval - 1, 2 * interval} + }, + time -> time <= interval + 1); + } + + private void assertSparseBatchedAlignedChunkSplit( + final long[][] pages, final LongPredicate isNull) throws Exception { + final java.nio.file.Path directory = Files.createTempDirectory("split-sparse-batched"); + final File source = directory.resolve("source.tsfile").toFile(); + final List targets = new ArrayList<>(); + final IDeviceID device = new StringArrayDeviceID("table1", "tagA"); + final Map expected = new TreeMap<>(); + final long interval = TimePartitionUtils.getTimePartitionInterval(); + try { + try (final TsFileIOWriter writer = new TsFileIOWriter(source)) { + writer.setSchema(createBatchedSchema()); + writer.startChunkGroup(device); + // Compaction writes the first series batch with all time chunks, then appends the + // remaining value chunks. Each value chunk must retain the rows and pages of its time + // chunk. + for (int window = 0; window < 2; window++) { + final long offset = window * 3 * interval; + final AlignedChunkWriterImpl alignedWriter = + new AlignedChunkWriterImpl( + Collections.singletonList(new MeasurementSchema("s1", TSDataType.INT64))); + for (final long[] page : pages) { + for (final long localTime : page) { + final long time = offset + localTime; + alignedWriter.getTimeChunkWriter().write(time); + alignedWriter.getValueChunkWriterByIndex(0).write(time, time, false); + expected.put(time, isNull.test(localTime) ? null : time + 1); + } + alignedWriter.sealCurrentPage(); + } + alignedWriter.writeToFileWriter(writer); + } + for (int window = 0; window < 2; window++) { + final long offset = window * 3 * interval; + final ValueChunkWriter valueWriter = + new ValueChunkWriter( + "s2", + CompressionType.LZ4, + TSDataType.INT64, + TSEncoding.PLAIN, + new PlainEncoder(TSDataType.INT64, 8)); + for (final long[] page : pages) { + for (final long localTime : page) { + final long time = offset + localTime; + valueWriter.write(time, time + 1, isNull.test(localTime)); + } + valueWriter.sealCurrentPage(); + } + valueWriter.writeToFileWriter(writer); + } + writer.endChunkGroup(); + writer.endFile(); + } + // Check that the source itself has the expected sparse rows before exercising LOAD. + assertSparseAlignedRows(source, device, expected); + final List chunks = splitSourceTsFile(source); + Assert.assertTrue(chunks.stream().anyMatch(BatchedAlignedValueChunkData.class::isInstance)); + final Set expectedPartitions = new HashSet<>(); + for (final long time : expected.keySet()) { + expectedPartitions.add(TimePartitionUtils.getTimePartitionSlot(time)); + } + Assert.assertEquals(expectedPartitions, groupChunkDataByPartition(chunks).keySet()); + for (final Map.Entry> entry : + groupChunkDataByPartition(chunks).entrySet()) { + final Map partitionRows = new TreeMap<>(); + for (final Map.Entry row : expected.entrySet()) { + if (TimePartitionUtils.getTimePartitionSlot(row.getKey()).equals(entry.getKey())) { + partitionRows.put(row.getKey(), row.getValue()); + } + } + final File target = + directory.resolve("target-" + entry.getKey().getStartTime() + ".tsfile").toFile(); + targets.add(target); + assertSparseAlignedRows(writeTargetTsFile(target, entry.getValue()), device, partitionRows); + } + } finally { + for (final File target : targets) { + Files.deleteIfExists(target.toPath()); + } + Files.deleteIfExists(source.toPath()); + Files.deleteIfExists(directory); + } + } + + private void assertSparseAlignedRows( + final File file, final IDeviceID device, final Map expected) throws Exception { + try (final TsFileSequenceReader reader = new TsFileSequenceReader(file.getAbsolutePath())) { + long valueCount = 0; + long start = Long.MAX_VALUE; + long end = Long.MIN_VALUE; + for (final AbstractAlignedChunkMetadata metadata : + reader.getAlignedChunkMetadata(device, false)) { + for (final IChunkMetadata valueMetadata : metadata.getValueChunkMetadataList()) { + if (valueMetadata != null && "s2".equals(valueMetadata.getMeasurementUid())) { + valueCount += valueMetadata.getStatistics().getCount(); + start = Math.min(start, valueMetadata.getStartTime()); + end = Math.max(end, valueMetadata.getEndTime()); + } + } + } + final List nonNullTimes = new ArrayList<>(); + for (final Map.Entry row : expected.entrySet()) { + if (row.getValue() != null) { + nonNullTimes.add(row.getKey()); + } + } + Assert.assertEquals(nonNullTimes.size(), valueCount); + if (!nonNullTimes.isEmpty()) { + Assert.assertEquals(nonNullTimes.get(0).longValue(), start); + Assert.assertEquals(nonNullTimes.get(nonNullTimes.size() - 1).longValue(), end); + } + } + try (final TsFileReader reader = new TsFileReader(file)) { + final QueryDataSet rows = + reader.query( + QueryExpression.create( + Arrays.asList(new Path(device, "s1", true), new Path(device, "s2", true)), null)); + for (final Map.Entry expectedRow : expected.entrySet()) { + Assert.assertTrue("missing row at " + expectedRow.getKey(), rows.hasNext()); + final RowRecord row = rows.next(); + Assert.assertEquals(expectedRow.getKey().longValue(), row.getTimestamp()); + Assert.assertEquals(row.getTimestamp(), row.getField(0).getLongV()); + final Field value = row.getField(1); + if (expectedRow.getValue() == null) { + Assert.assertTrue(value == null || value.getDataType() == null); + } else { + Assert.assertNotNull(value); + Assert.assertNotNull(value.getDataType()); + Assert.assertEquals(expectedRow.getValue().longValue(), value.getLongV()); + } + } + Assert.assertFalse(rows.hasNext()); + } + } + private void writeBatchedAlignedTsFile( final File tsFile, final List deviceIDs, From c768d8d7b00ef0936aaab47c87faa79fada21ddb Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Mon, 9 Nov 2026 11:55:25 +0800 Subject: [PATCH 2/4] Fix LOAD splitting of sparse batched aligned value chunks The LOAD splitter re-encodes the standalone value chunks of a batched aligned table with a ValueChunkWriter, and the chunks it builds have to keep the row positions and the page positions of the time chunks they belong to. - Write every row, nulls included, with its real timestamp, so that the value page bitmap keeps one position per row and the statistics cover the non-null values only. - Keep one page per page slot of the time chunk, even when the page is all null: the aligned reader pairs the pages of a value chunk with the pages of its time chunk by position, so a missing page shifts every later page onto the timestamps of an earlier one. - Normalize the header of an empty page. The page writer decides on the presence of the page statistics instead of the page size, so the statistics carried by the placeholder page of an empty chunk would be written into a page that has no data and the reader would parse a phantom page out of it. - Choose between the mirrored layout and the zero-page layout of an all-null column with one rule - the counterpart of the rule the base class applies to the chunks it writes itself - instead of conflating "has a non-null value" with "has rows of this partition". Cover the cases the previous tests missed: a copied page that carries data followed by a decoded page, an all-null page behind the pages that carry data, and an assertion that every emitted chunk declares exactly the number of bytes it carries. Based on the branch of apache/iotdb#19032. --- .../BatchedAlignedValueChunkData.java | 84 ++++++++++++++----- .../load/TsFileSplitterTest.java | 48 ++++++++++- 2 files changed, 112 insertions(+), 20 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java index d29cc2601da07..15492d2079ad6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java @@ -25,6 +25,7 @@ import org.apache.tsfile.exception.write.PageException; import org.apache.tsfile.file.header.ChunkHeader; +import org.apache.tsfile.file.header.PageHeader; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.statistics.Statistics; import org.apache.tsfile.read.common.Chunk; @@ -52,6 +53,12 @@ public class BatchedAlignedValueChunkData extends AlignedChunkData { private static final ByteBuffer EMPTY_CHUNK_DATA = ByteBuffer.allocate(0); + // An empty page holds no page body and no page statistics: it serializes to the single 0 varint + // that stands for it. The header must therefore not carry statistics, otherwise the page writer + // appends a compressed-size varint to a page without data and the reader sees a phantom page. + private static final ByteBuffer EMPTY_PAGE_DATA = ByteBuffer.allocate(0); + private static final PageHeader EMPTY_PAGE_HEADER = new PageHeader(0, 0, null); + // Used for splitter public BatchedAlignedValueChunkData(AlignedChunkData alignedChunkData) { super(alignedChunkData); @@ -63,8 +70,9 @@ public BatchedAlignedValueChunkData(IDeviceID device, TTimePartitionSlot timePar } /** - * Encodes standalone value chunks while preserving the row and page positions of their time - * chunks. + * Encodes standalone value chunks while preserving the row and the page positions of their time + * chunks: a row whose value is null still occupies a bitmap position, and an all-null page still + * occupies a page slot. */ protected void encodeAndBuildChunks() throws IOException, PageException { final long startTime = timePartitionSlot.getStartTime(); @@ -110,17 +118,26 @@ protected void encodeAndBuildChunks() throws IOException, PageException { measurementSchema.getEncodingType(), measurementSchema.getValueEncoder()); - // Filter and process the pages that belong to the current value chunk (index 'i'). - // A chunk header can exist for a physical chunk whose points all belong to another time - // partition. Such a chunk stays empty here and must be emitted without any page: sealing an - // untouched writer would produce a zero-point page whose data size does not match the bytes - // actually written, and the resulting TsFile could not be parsed back. - boolean hasBufferedData = false; - for (PageBuffer page : chunkPages) { + // Whether this chunk covers any row of this time partition. Such a chunk writes one page for + // every page slot of its time chunk, all-null pages included: the aligned reader pairs the + // pages of a value chunk with the pages of its time chunk by position, so a missing page + // shifts every later page onto the wrong timestamps. A chunk that covers no row at all keeps + // the zero-page layout of an all-null column, which the metadata reader reports as null. + final boolean chunkHasRows = hasPageWithRows(chunkPages, startTime, endTime); + for (final PageBuffer page : chunkPages) { if (!page.needDecode) { - // Empty pages still occupy a slot opposite the corresponding time page. If every page - // is empty, the whole chunk is emitted without a body below. - hasBufferedData |= page.pageData.remaining() > 0; + if (page.pageData.remaining() == 0) { + // An all-null page is stored as an empty page and keeps its slot. Its header is + // normalized because the page writer decides on the presence of the statistics instead + // of the page size: the placeholder page built by + // TsFileSplitter#handleEmptyValueChunk carries the statistics of an empty chunk, and + // those would turn the empty page into two bytes that the reader parses as a page + // followed by a phantom one. + if (chunkHasRows) { + valueChunkWriter.writePageHeaderAndDataIntoBuff(EMPTY_PAGE_DATA, EMPTY_PAGE_HEADER); + } + continue; + } valueChunkWriter.writePageHeaderAndDataIntoBuff(page.pageData, page.pageHeader); continue; } @@ -136,11 +153,8 @@ protected void encodeAndBuildChunks() throws IOException, PageException { if (time >= startTime && time <= endTime) { boolean isNull = (page.valueBatch == null - || page.valueBatch.length == 0 + || j >= page.valueBatch.length || page.valueBatch[j] == null); - if (!isNull) { - hasBufferedData = true; - } // The bitmap has one position per timestamp, including nulls. Use the real timestamp // so the writer also computes the statistics of the non-null values correctly. valueChunkValueWriter.write( @@ -149,20 +163,24 @@ protected void encodeAndBuildChunks() throws IOException, PageException { } } + // Sealing an untouched writer would count a page that holds no byte. if (pageHasRows) { valueChunkWriter.sealCurrentPage(); } } - // Directly assemble the final standard Chunk into memory + // Directly assemble the final standard Chunk into memory. A chunk without any row of this + // partition stays empty: sealing its untouched writer would produce a page whose data size + // does not match the bytes actually written, and the resulting TsFile could not be parsed + // back. final ByteBuffer valueData = - hasBufferedData ? valueChunkWriter.getByteBuffer() : EMPTY_CHUNK_DATA; + chunkHasRows ? valueChunkWriter.getByteBuffer() : EMPTY_CHUNK_DATA; final Statistics valueStat = valueChunkWriter.getStatistics(); final ChunkHeader encodedHeader = createEncodedChunkHeader( chunkHeader, valueData.remaining(), - hasBufferedData ? valueChunkWriter.getNumOfPages() : 0, + chunkHasRows ? valueChunkWriter.getNumOfPages() : 0, VALUE_COLUMN_MASK); entireChunks.add(new Chunk(encodedHeader, valueData, null, valueStat)); @@ -171,4 +189,32 @@ protected void encodeAndBuildChunks() throws IOException, PageException { dataSize += valueData.remaining() + valueStat.getSerializedSize(); } } + + /** + * Whether any page of the chunk holds rows of this time partition. Only such a chunk mirrors the + * page slots of its time chunk, because only then does the time chunk have a page to pair with; a + * chunk that holds no row at all keeps the zero-page layout. This is the counterpart of the rule + * applied by the base class to the chunks it writes itself. + */ + private static boolean hasPageWithRows( + final List chunkPages, final long startTime, final long endTime) { + for (final PageBuffer page : chunkPages) { + if (!page.needDecode) { + // A page copied as it is was read from this partition, so a page with a body always + // carries rows of the partition and only an all-null page has no body. + if (page.pageData.remaining() > 0) { + return true; + } + continue; + } + // A decoded page may be entirely outside the partition, and its timestamps are the only + // reliable record of the rows it contributes here. + for (final long time : page.timeBatch) { + if (time >= startTime && time <= endTime) { + return true; + } + } + } + return false; + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/TsFileSplitterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/TsFileSplitterTest.java index e993c72a4bf6e..fd92e1014b4d8 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/TsFileSplitterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/TsFileSplitterTest.java @@ -77,6 +77,7 @@ import java.io.File; import java.io.IOException; import java.lang.management.ManagementFactory; +import java.nio.file.DirectoryNotEmptyException; import java.nio.file.Files; import java.util.ArrayList; import java.util.Arrays; @@ -937,6 +938,27 @@ public void testSplitBatchedAlignedChunkPreservesDecodedEmptyPage() throws Excep time -> time <= interval + 1); } + @Test + public void testSplitBatchedAlignedChunkPreservesCopiedPageBeforeDecodedPage() throws Exception { + final long interval = TimePartitionUtils.getTimePartitionInterval(); + // The first page stays inside its own time partition and is copied as it is, the second page + // crosses the boundary and is decoded: the statistics of the copied page have to be spliced + // into the chunk that the value chunk writer assembles from then on. + assertSparseBatchedAlignedChunkSplit( + new long[][] {{0, 1, 2}, {interval - 2, interval - 1, interval, interval + 1}}, + time -> time % 2 == 0); + } + + @Test + public void testSplitBatchedAlignedChunkPreservesEmptyPageAfterDataPages() throws Exception { + final long interval = TimePartitionUtils.getTimePartitionInterval(); + // The third page is entirely null in both partitions it spans, so it is an empty page whose + // slot has to be kept behind the pages that carry data. + assertSparseBatchedAlignedChunkSplit( + new long[][] {{0, 1}, {2, 3}, {interval - 2, interval - 1, interval, interval + 1}}, + time -> time >= interval - 2); + } + private void assertSparseBatchedAlignedChunkSplit( final long[][] pages, final LongPredicate isNull) throws Exception { final java.nio.file.Path directory = Files.createTempDirectory("split-sparse-batched"); @@ -993,6 +1015,11 @@ private void assertSparseBatchedAlignedChunkSplit( assertSparseAlignedRows(source, device, expected); final List chunks = splitSourceTsFile(source); Assert.assertTrue(chunks.stream().anyMatch(BatchedAlignedValueChunkData.class::isInstance)); + for (final List chunkDataList : groupChunkDataByPartition(chunks).values()) { + for (final ChunkData chunkData : chunkDataList) { + assertChunkBytesMatchHeader(chunkData); + } + } final Set expectedPartitions = new HashSet<>(); for (final long time : expected.keySet()) { expectedPartitions.add(TimePartitionUtils.getTimePartitionSlot(time)); @@ -1016,7 +1043,26 @@ private void assertSparseBatchedAlignedChunkSplit( Files.deleteIfExists(target.toPath()); } Files.deleteIfExists(source.toPath()); - Files.deleteIfExists(directory); + try { + Files.deleteIfExists(directory); + } catch (final DirectoryNotEmptyException e) { + // A leftover file must not hide the failure that left it behind. + } + } + } + + /** + * A chunk header has to declare exactly the bytes the chunk carries: a header that advertises + * fewer bytes than its pages hold produces a split TsFile whose chunks cannot be parsed back. + */ + private void assertChunkBytesMatchHeader(final ChunkData chunkData) { + for (final Chunk chunk : chunkData.getChunks()) { + Assert.assertEquals( + "chunk of measurement '" + + chunk.getHeader().getMeasurementID() + + "' declares a data size that does not match the bytes it carries", + chunk.getHeader().getDataSize(), + chunk.getData().remaining()); } } From 9b3a8f7e99ad146f1db20640094190204dbd312f Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Mon, 9 Nov 2026 12:04:26 +0800 Subject: [PATCH 3/4] Add an IT that loads a sparse batched aligned TsFile The LOAD splitter re-encodes the value chunks of a batched aligned chunk group when the file crosses time partitions. The new IT builds such a file with the compaction layout - one time chunk and two value chunks, the second one behind the time chunk - whose rows include null values and an entirely null page, loads it, and checks that every row keeps its timestamp and its value. The file spans three time partitions of the interval the test configures, so LOAD decodes and splits it instead of storing it as it is. --- .../it/db/it/IoTDBLoadTsFileIT.java | 122 ++++++++++++++++++ 1 file changed, 122 insertions(+) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTsFileIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTsFileIT.java index c4444c4bf55f8..9cb250c599ee1 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTsFileIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTsFileIT.java @@ -26,14 +26,22 @@ import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; import org.apache.iotdb.itbase.env.BaseEnv; +import org.apache.tsfile.encoding.encoder.PlainEncoder; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; +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.Pair; import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; +import org.apache.tsfile.write.chunk.ValueChunkWriter; import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.IMeasurementSchema; import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.schema.Schema; +import org.apache.tsfile.write.writer.TsFileIOWriter; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -51,6 +59,7 @@ import java.sql.Statement; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Objects; import java.util.concurrent.TimeUnit; @@ -560,6 +569,119 @@ public void testLoadWithAllFieldsNullRows() throws Exception { } } + /** + * The compaction writer emits the series batches behind the first one as value chunks that follow + * all time chunks of the chunk group. When LOAD splits such a chunk by time partition it has to + * re-encode it, and the rows of the chunk keep their positions: a null value still occupies a + * bitmap position and an all-null page still occupies the page slot opposite its time page. + * + *

The rows span three time partitions of the interval configured in {@code setUp}, so LOAD has + * to decode the file and split it instead of storing it as it is. + */ + @Test + public void testLoadSparseBatchedAlignedTsFile() throws Exception { + // The first page stays inside one time partition and is copied as it is, the second one crosses + // a partition boundary and holds a null row, and the third one is entirely null in both + // partitions it spans. + final long[][] pages = { + {0, 1, 2}, + {PARTITION_INTERVAL - 2, PARTITION_INTERVAL - 1, PARTITION_INTERVAL, PARTITION_INTERVAL + 1}, + { + 2 * PARTITION_INTERVAL - 2, + 2 * PARTITION_INTERVAL - 1, + 2 * PARTITION_INTERVAL, + 2 * PARTITION_INTERVAL + 1 + } + }; + final File file = new File(tmpDir, "1-0-0-0.tsfile"); + writeSparseBatchedAlignedTsFile(file, pages); + + try (final Connection connection = + EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + statement.execute(String.format("create database if not exists %s", SchemaConfig.DATABASE_0)); + statement.execute(String.format("use %s", SchemaConfig.DATABASE_0)); + statement.execute( + String.format( + "load '%s' with ('database'='%s')", file.getAbsolutePath(), SchemaConfig.DATABASE_0)); + + try (final ResultSet resultSet = + statement.executeQuery( + String.format("select time, f1, f2 from %s order by time", SchemaConfig.TABLE_0))) { + for (final long[] page : pages) { + for (final long time : page) { + Assert.assertTrue(resultSet.next()); + Assert.assertEquals(time, resultSet.getLong("time")); + Assert.assertEquals(time, resultSet.getLong("f1")); + if (isNullField2(time)) { + resultSet.getLong("f2"); + Assert.assertTrue(resultSet.wasNull()); + } else { + Assert.assertEquals(time + 1, resultSet.getLong("f2")); + } + } + } + Assert.assertFalse(resultSet.next()); + } + } + } + + private void writeSparseBatchedAlignedTsFile(final File file, final long[][] pages) + throws Exception { + final Schema schema = new Schema(); + schema.registerTableSchema( + new TableSchema( + SchemaConfig.TABLE_0, + Arrays.asList( + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("f1", TSDataType.INT64), + new MeasurementSchema("f2", TSDataType.INT64)), + Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.FIELD))); + + try (final TsFileIOWriter writer = new TsFileIOWriter(file)) { + writer.setSchema(schema); + writer.startChunkGroup(new StringArrayDeviceID(SchemaConfig.TABLE_0, "tagA")); + + // The first field shares the chunk group of the time chunk. + final AlignedChunkWriterImpl alignedWriter = + new AlignedChunkWriterImpl( + Collections.singletonList(new MeasurementSchema("f1", TSDataType.INT64))); + for (final long[] page : pages) { + for (final long time : page) { + alignedWriter.getTimeChunkWriter().write(time); + alignedWriter.getValueChunkWriterByIndex(0).write(time, time, false); + } + alignedWriter.sealCurrentPage(); + } + alignedWriter.writeToFileWriter(writer); + + // The second field follows the time chunk as a value chunk of its own. + final ValueChunkWriter valueWriter = + new ValueChunkWriter( + "f2", + CompressionType.LZ4, + TSDataType.INT64, + TSEncoding.PLAIN, + new PlainEncoder(TSDataType.INT64, 8)); + for (final long[] page : pages) { + for (final long time : page) { + valueWriter.write(time, time + 1, isNullField2(time)); + } + valueWriter.sealCurrentPage(); + } + valueWriter.writeToFileWriter(writer); + + writer.endChunkGroup(); + writer.endFile(); + } + } + + private static boolean isNullField2(final long time) { + // The row in the middle of the page that crosses the first boundary and the rows of the last + // page carry no value. + return time == PARTITION_INTERVAL - 1 || time >= 2 * PARTITION_INTERVAL - 2; + } + private static class SchemaConfig { private static final String DATABASE_0 = "root"; private static final String TABLE_0 = "test"; From 5941397363ff7b48379642739f16c6adb8253a52 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Mon, 9 Nov 2026 16:37:03 +0800 Subject: [PATCH 4/4] Use the recorded row count when deciding the chunk layout PageBuffer.satisfiedLength already holds the number of rows the page contributes to the time partition, so the layout decision no longer scans the decoded timestamps a second time. --- .../BatchedAlignedValueChunkData.java | 24 ++++++------------- 1 file changed, 7 insertions(+), 17 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java index 15492d2079ad6..073229ae75f86 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java @@ -123,7 +123,7 @@ protected void encodeAndBuildChunks() throws IOException, PageException { // pages of a value chunk with the pages of its time chunk by position, so a missing page // shifts every later page onto the wrong timestamps. A chunk that covers no row at all keeps // the zero-page layout of an all-null column, which the metadata reader reports as null. - final boolean chunkHasRows = hasPageWithRows(chunkPages, startTime, endTime); + final boolean chunkHasRows = hasPageWithRows(chunkPages); for (final PageBuffer page : chunkPages) { if (!page.needDecode) { if (page.pageData.remaining() == 0) { @@ -196,23 +196,13 @@ protected void encodeAndBuildChunks() throws IOException, PageException { * chunk that holds no row at all keeps the zero-page layout. This is the counterpart of the rule * applied by the base class to the chunks it writes itself. */ - private static boolean hasPageWithRows( - final List chunkPages, final long startTime, final long endTime) { + private static boolean hasPageWithRows(final List chunkPages) { for (final PageBuffer page : chunkPages) { - if (!page.needDecode) { - // A page copied as it is was read from this partition, so a page with a body always - // carries rows of the partition and only an all-null page has no body. - if (page.pageData.remaining() > 0) { - return true; - } - continue; - } - // A decoded page may be entirely outside the partition, and its timestamps are the only - // reliable record of the rows it contributes here. - for (final long time : page.timeBatch) { - if (time >= startTime && time <= endTime) { - return true; - } + // A decoded page already records how many of its rows belong to this partition, and a page + // copied as it is was read from this partition, so a body means rows and only an all-null + // page has no body. + if (page.needDecode ? page.satisfiedLength > 0 : page.pageData.remaining() > 0) { + return true; } } return false;