This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/load-sparse-batched-aligned-chunks in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 8afb2ea42a0902ffe2666e39927b9fc5f23a88ae Author: Caideyipi <[email protected]> AuthorDate: Tue Oct 27 22:39:16 2026 +0800 Fix LOAD splitting of sparse batched aligned chunks --- .../splitter/BatchedAlignedValueChunkData.java | 45 ++---- .../db/storageengine/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 56bff063d70..d29cc2601da 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.chunk.ValueChunkWriter; 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 class BatchedAlignedValueChunkData extends AlignedChunkData { } /** - * 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 @@ public class BatchedAlignedValueChunkData extends AlignedChunkData { 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 @@ public class BatchedAlignedValueChunkData extends AlignedChunkData { || 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<? extends Serializable> 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 d70cb40e814..e993c72a4bf 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.List; 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 class TsFileSplitterTest { } } + @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<File> targets = new ArrayList<>(); + final IDeviceID device = new StringArrayDeviceID("table1", "tagA"); + final Map<Long, Long> 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<ChunkData> chunks = splitSourceTsFile(source); + Assert.assertTrue(chunks.stream().anyMatch(BatchedAlignedValueChunkData.class::isInstance)); + final Set<TTimePartitionSlot> expectedPartitions = new HashSet<>(); + for (final long time : expected.keySet()) { + expectedPartitions.add(TimePartitionUtils.getTimePartitionSlot(time)); + } + Assert.assertEquals(expectedPartitions, groupChunkDataByPartition(chunks).keySet()); + for (final Map.Entry<TTimePartitionSlot, List<ChunkData>> entry : + groupChunkDataByPartition(chunks).entrySet()) { + final Map<Long, Long> partitionRows = new TreeMap<>(); + for (final Map.Entry<Long, Long> 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<Long, Long> 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<Long> nonNullTimes = new ArrayList<>(); + for (final Map.Entry<Long, Long> 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<Long, Long> 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<IDeviceID> deviceIDs,
