You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by ji...@apache.org on 2026/09/09 01:27:21 UTC
(tsfile) branch develop updated: Add batched null and bitmap-offset writes to ValuePageWriter (#948)
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/develop by this push:
new 1f5f06837 Add batched null and bitmap-offset writes to ValuePageWriter (#948)
1f5f06837 is described below
commit 1f5f06837db6022c3aca8457ccf16700b352f859
Author: Jiang Tian <jt...@163.com>
AuthorDate: Wed Sep 9 09:27:16 2026 +0800
Add batched null and bitmap-offset writes to ValuePageWriter (#948)
* add ValuePageWriter.writeNull
* add ValuePageWriter.write with Bitmap
---
.../apache/tsfile/write/page/ValuePageWriter.java | 133 +++++++++++++++++++++
.../tsfile/write/writer/ValuePageWriterTest.java | 87 ++++++++++++++
2 files changed, 220 insertions(+)
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java
index 525c3199a..f7f375759 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java
@@ -28,6 +28,7 @@ import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.EncryptionType;
import org.apache.tsfile.file.metadata.statistics.Statistics;
import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.utils.PublicBAOS;
import org.apache.tsfile.utils.ReadWriteForEncodingUtils;
@@ -69,6 +70,8 @@ public class ValuePageWriter {
private static final int MASK = 1 << 7;
+ private static final byte[] NULL_BITMAP_BATCH = new byte[1024];
+
public ValuePageWriter(Encoder valueEncoder, ICompressor compressor, TSDataType dataType) {
this.valueOut = new PublicBAOS();
this.bitmap = 0;
@@ -158,6 +161,34 @@ public class ValuePageWriter {
}
}
+ /** Write consecutive null values without encoding values or updating statistics. */
+ public void writeNull(int count) {
+ if (count <= 0) {
+ return;
+ }
+
+ int remaining = count;
+ while (remaining > 0 && size % 8 != 0) {
+ setBit(true);
+ remaining--;
+ }
+
+ // At a byte boundary, eight nulls are a zero byte. Reuse a bounded buffer for large runs.
+ int fullByteCount = remaining / 8;
+ while (fullByteCount > 0) {
+ int batchSize = Math.min(fullByteCount, NULL_BITMAP_BATCH.length);
+ bitmapOut.write(NULL_BITMAP_BATCH, 0, batchSize);
+ size += batchSize * 8;
+ fullByteCount -= batchSize;
+ remaining -= batchSize * 8;
+ }
+
+ while (remaining > 0) {
+ setBit(true);
+ remaining--;
+ }
+ }
+
private void setBit(boolean isNull) {
if (!isNull) {
bitmap |= (MASK >>> (size % 8));
@@ -241,6 +272,108 @@ public class ValuePageWriter {
}
}
+ public void write(
+ long[] timestamps,
+ boolean[] values,
+ BitMap bitMap,
+ int bitMapOffset,
+ int batchSize,
+ int arrayOffset) {
+ for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
+ boolean isNull = bitMap.isMarked(bitMapOffset + i - arrayOffset);
+ setBit(isNull);
+ if (!isNull) {
+ valueEncoder.encode(values[i], valueOut);
+ statistics.update(timestamps[i], values[i]);
+ }
+ }
+ }
+
+ public void write(
+ long[] timestamps,
+ int[] values,
+ BitMap bitMap,
+ int bitMapOffset,
+ int batchSize,
+ int arrayOffset) {
+ for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
+ boolean isNull = bitMap.isMarked(bitMapOffset + i - arrayOffset);
+ setBit(isNull);
+ if (!isNull) {
+ valueEncoder.encode(values[i], valueOut);
+ statistics.update(timestamps[i], values[i]);
+ }
+ }
+ }
+
+ public void write(
+ long[] timestamps,
+ long[] values,
+ BitMap bitMap,
+ int bitMapOffset,
+ int batchSize,
+ int arrayOffset) {
+ for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
+ boolean isNull = bitMap.isMarked(bitMapOffset + i - arrayOffset);
+ setBit(isNull);
+ if (!isNull) {
+ valueEncoder.encode(values[i], valueOut);
+ statistics.update(timestamps[i], values[i]);
+ }
+ }
+ }
+
+ public void write(
+ long[] timestamps,
+ float[] values,
+ BitMap bitMap,
+ int bitMapOffset,
+ int batchSize,
+ int arrayOffset) {
+ for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
+ boolean isNull = bitMap.isMarked(bitMapOffset + i - arrayOffset);
+ setBit(isNull);
+ if (!isNull) {
+ valueEncoder.encode(values[i], valueOut);
+ statistics.update(timestamps[i], values[i]);
+ }
+ }
+ }
+
+ public void write(
+ long[] timestamps,
+ double[] values,
+ BitMap bitMap,
+ int bitMapOffset,
+ int batchSize,
+ int arrayOffset) {
+ for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
+ boolean isNull = bitMap.isMarked(bitMapOffset + i - arrayOffset);
+ setBit(isNull);
+ if (!isNull) {
+ valueEncoder.encode(values[i], valueOut);
+ statistics.update(timestamps[i], values[i]);
+ }
+ }
+ }
+
+ public void write(
+ long[] timestamps,
+ Binary[] values,
+ BitMap bitMap,
+ int bitMapOffset,
+ int batchSize,
+ int arrayOffset) {
+ for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
+ boolean isNull = bitMap.isMarked(bitMapOffset + i - arrayOffset);
+ setBit(isNull);
+ if (!isNull) {
+ valueEncoder.encode(values[i], valueOut);
+ statistics.update(timestamps[i], values[i]);
+ }
+ }
+ }
+
/** flush all data remained in encoders. */
private void prepareEndWriteOnePage() throws IOException {
valueEncoder.flush(valueOut);
diff --git a/java/tsfile/src/test/java/org/apache/tsfile/write/writer/ValuePageWriterTest.java b/java/tsfile/src/test/java/org/apache/tsfile/write/writer/ValuePageWriterTest.java
index 7d314648e..bac65226f 100644
--- a/java/tsfile/src/test/java/org/apache/tsfile/write/writer/ValuePageWriterTest.java
+++ b/java/tsfile/src/test/java/org/apache/tsfile/write/writer/ValuePageWriterTest.java
@@ -37,11 +37,98 @@ import org.junit.Test;
import java.io.IOException;
import java.nio.ByteBuffer;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
public class ValuePageWriterTest {
+ @Test
+ public void testWriteNull() throws IOException {
+ ICompressor compressor = ICompressor.getCompressor(CompressionType.UNCOMPRESSED);
+ // Exercise every bitmap offset and null runs spanning multiple bytes.
+ for (int prefixSize = 0; prefixSize <= 8; prefixSize++) {
+ for (int nullCount = 0; nullCount <= 24; nullCount++) {
+ ValuePageWriter pageWriter =
+ new ValuePageWriter(
+ new PlainEncoder(TSDataType.FLOAT, 0), compressor, TSDataType.FLOAT);
+ ValuePageWriter expectedWriter =
+ new ValuePageWriter(
+ new PlainEncoder(TSDataType.FLOAT, 0), compressor, TSDataType.FLOAT);
+ for (int i = 0; i < prefixSize; i++) {
+ pageWriter.write(i, (float) i, false);
+ expectedWriter.write(i, (float) i, false);
+ }
+ pageWriter.writeNull(nullCount);
+ for (int i = 0; i < nullCount; i++) {
+ expectedWriter.write(prefixSize + i, 0.0f, true);
+ }
+ assertEquals(prefixSize + nullCount, pageWriter.getSize());
+ assertEquals(prefixSize, pageWriter.getPointNumber());
+
+ int lastTime = prefixSize + nullCount;
+ pageWriter.write(lastTime, 42.0f, false);
+ expectedWriter.write(lastTime, 42.0f, false);
+ assertEquals(expectedWriter.getUncompressedBytes(), pageWriter.getUncompressedBytes());
+ PublicBAOS expectedStatistics = new PublicBAOS();
+ PublicBAOS actualStatistics = new PublicBAOS();
+ expectedWriter.getStatistics().serialize(expectedStatistics);
+ pageWriter.getStatistics().serialize(actualStatistics);
+ assertArrayEquals(expectedStatistics.toByteArray(), actualStatistics.toByteArray());
+ }
+ }
+ }
+
+ @Test
+ public void testWriteNullOnlyPageAndReset() throws IOException {
+ ValuePageWriter pageWriter =
+ new ValuePageWriter(
+ new PlainEncoder(TSDataType.FLOAT, 0),
+ ICompressor.getCompressor(CompressionType.UNCOMPRESSED),
+ TSDataType.FLOAT);
+ PublicBAOS pageBuffer = new PublicBAOS();
+ pageWriter.writeNull(0);
+ assertEquals(0, pageWriter.getSize());
+ assertEquals(0, pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, true));
+ assertEquals(0, pageBuffer.size());
+
+ pageWriter.writeNull(7);
+ pageWriter.writeNull(10);
+ assertEquals(17, pageWriter.getSize());
+ assertEquals(0, pageWriter.getPointNumber());
+ assertEquals(1, pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, true));
+ assertEquals(1, pageBuffer.size());
+ assertEquals(0, pageBuffer.getBuf()[0]);
+
+ pageWriter.reset(TSDataType.FLOAT);
+ pageWriter.writeNull(1);
+ pageWriter.write(1L, 1.0f, false);
+ ByteBuffer buffer = pageWriter.getUncompressedBytes();
+ assertEquals(2, buffer.getInt());
+ assertEquals((byte) 0x40, buffer.get());
+ assertEquals(1.0f, buffer.getFloat(), 0.0f);
+ assertEquals(0, buffer.remaining());
+ }
+
+ @Test
+ public void testWriteLargeNullBatch() throws IOException {
+ ValuePageWriter pageWriter =
+ new ValuePageWriter(
+ new PlainEncoder(TSDataType.FLOAT, 0),
+ ICompressor.getCompressor(CompressionType.UNCOMPRESSED),
+ TSDataType.FLOAT);
+ pageWriter.writeNull(16393);
+
+ assertEquals(16393, pageWriter.getSize());
+ assertEquals(0, pageWriter.getPointNumber());
+ ByteBuffer buffer = pageWriter.getUncompressedBytes();
+ assertEquals(16393, buffer.getInt());
+ assertEquals(2050, buffer.remaining());
+ while (buffer.hasRemaining()) {
+ assertEquals(0, buffer.get());
+ }
+ }
+
@Test
public void testWrite1() {
Encoder valueEncoder = new PlainEncoder(TSDataType.FLOAT, 0);