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 c4444c4bf55f..9cb250c599ee 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"; 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 56bff063d705..073229ae75f8 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; @@ -33,7 +34,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; @@ -53,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); @@ -64,9 +70,9 @@ 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 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(); @@ -112,28 +118,31 @@ 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); + for (final 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; + 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; } - // 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( @@ -144,44 +153,34 @@ 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; - 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) { + // 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)); @@ -190,4 +189,22 @@ 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) { + for (final PageBuffer page : chunkPages) { + // 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; + } } 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 d70cb40e8141..fd92e1014b4d 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; @@ -87,7 +88,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 +911,212 @@ 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); + } + + @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"); + 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)); + 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)); + } + 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()); + 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()); + } + } + + 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,