From 5f2aae1523c2e030c0affcb8cc0ff963c1abf0f0 Mon Sep 17 00:00:00 2001 From: Dongyang Geng Date: Tue, 4 Aug 2026 15:28:02 +0800 Subject: [PATCH] feat: implement ByteColumnReader and RunLenByteDecoder --- cpp/pixels-common/include/utils/Constants.h | 6 +- cpp/pixels-common/lib/utils/Constants.cpp | 6 +- .../lib/encoding/RunLenIntDecoder.cpp | 6 +- .../lib/encoding/RunLenIntEncoder.cpp | 48 +- .../pixels/common/utils/Constants.java | 9 +- .../core/encoding/RunLenByteDecoder.java | 86 +++- .../core/encoding/RunLenByteEncoder.java | 24 +- .../core/encoding/RunLenIntDecoder.java | 6 +- .../core/encoding/RunLenIntEncoder.java | 42 +- .../pixels/core/reader/ByteColumnReader.java | 267 ++++++++++- .../pixels/core/encoding/TestEncoding.java | 56 +++ .../core/reader/TestByteColumnReader.java | 452 ++++++++++++++++++ 12 files changed, 927 insertions(+), 81 deletions(-) create mode 100644 pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestByteColumnReader.java diff --git a/cpp/pixels-common/include/utils/Constants.h b/cpp/pixels-common/include/utils/Constants.h index bcce3a7721..3895593b08 100644 --- a/cpp/pixels-common/include/utils/Constants.h +++ b/cpp/pixels-common/include/utils/Constants.h @@ -40,9 +40,9 @@ class Constants static int REDIS_BUFFER_SIZE; static int GCS_BUFFER_SIZE; - static int MIN_REPEAT; - static int MAX_SCOPE; - static int MAX_SHORT_REPEAT_LENGTH; + static int RLE_MIN_REPEAT; + static int INT_RLE_MAX_SCOPE; + static int INT_RLE_MAX_SHORT_REPEAT; static float DICT_KEY_SIZE_THRESHOLD; static int INIT_DICT_SIZE; diff --git a/cpp/pixels-common/lib/utils/Constants.cpp b/cpp/pixels-common/lib/utils/Constants.cpp index de35d3de05..e97c4321fa 100644 --- a/cpp/pixels-common/lib/utils/Constants.cpp +++ b/cpp/pixels-common/lib/utils/Constants.cpp @@ -34,9 +34,9 @@ int Constants::S3_BUFFER_SIZE = 8 * 1024 * 1024; int Constants::REDIS_BUFFER_SIZE = 8 * 1024 * 1024; int Constants::GCS_BUFFER_SIZE = 8 * 1024 * 1024; -int Constants::MIN_REPEAT = 3; -int Constants::MAX_SCOPE = 512; -int Constants::MAX_SHORT_REPEAT_LENGTH = 10; +int Constants::RLE_MIN_REPEAT = 3; +int Constants::INT_RLE_MAX_SCOPE = 512; +int Constants::INT_RLE_MAX_SHORT_REPEAT = 10; float Constants::DICT_KEY_SIZE_THRESHOLD = 0.1F; int Constants::INIT_DICT_SIZE = 4096; diff --git a/cpp/pixels-core/lib/encoding/RunLenIntDecoder.cpp b/cpp/pixels-core/lib/encoding/RunLenIntDecoder.cpp index 1e6f17b1d0..f4b834cdde 100644 --- a/cpp/pixels-core/lib/encoding/RunLenIntDecoder.cpp +++ b/cpp/pixels-core/lib/encoding/RunLenIntDecoder.cpp @@ -26,7 +26,7 @@ RunLenIntDecoder::RunLenIntDecoder(const std::shared_ptr &bb, bool isSigned) { - literals = new long[Constants::MAX_SCOPE]; + literals = new long[Constants::INT_RLE_MAX_SCOPE]; inputStream = bb; this->isSigned = isSigned; numLiterals = 0; @@ -308,8 +308,8 @@ void RunLenIntDecoder::readShortRepeatValues(int firstByte) // read the run length int len = firstByte & 0x07; - // run length values are stored only after MIN_REPEAT value is met - len += Constants::MIN_REPEAT; + // run length values are stored only after RLE_MIN_REPEAT value is met + len += Constants::RLE_MIN_REPEAT; // read the repeated value which is stored using fixed bytes long val = bytesToLongBE(inputStream, size); diff --git a/cpp/pixels-core/lib/encoding/RunLenIntEncoder.cpp b/cpp/pixels-core/lib/encoding/RunLenIntEncoder.cpp index 82828f7b0f..a068cb0533 100644 --- a/cpp/pixels-core/lib/encoding/RunLenIntEncoder.cpp +++ b/cpp/pixels-core/lib/encoding/RunLenIntEncoder.cpp @@ -48,11 +48,11 @@ RunLenIntEncoder::RunLenIntEncoder(bool isSigned, bool isAlignedBitPacking) : // PENDING: will the byte buffer be used in a buffer pool // so that we do not need to create it here outputStream = std::make_shared(); - literals = new long[Constants::MAX_SCOPE]; - zigzagLiterals = new long[Constants::MAX_SCOPE]; - baseRedLiterals = new long[Constants::MAX_SCOPE]; - adjDeltas = new long[Constants::MAX_SCOPE]; - gapVsPatchList = new long[Constants::MAX_SCOPE]; + literals = new long[Constants::INT_RLE_MAX_SCOPE]; + zigzagLiterals = new long[Constants::INT_RLE_MAX_SCOPE]; + baseRedLiterals = new long[Constants::INT_RLE_MAX_SCOPE]; + adjDeltas = new long[Constants::INT_RLE_MAX_SCOPE]; + gapVsPatchList = new long[Constants::INT_RLE_MAX_SCOPE]; clear(); } @@ -137,9 +137,9 @@ void RunLenIntEncoder::determineEncoding() zzBits100p = percentileBits(zigzagLiterals, 0, numLiterals, 1.0); // less than min repeat num so direct encoding - if (numLiterals <= Constants::MIN_REPEAT) + if (numLiterals <= Constants::RLE_MIN_REPEAT) { - // std::cout << "numLiterals <= Constants::MIN_REPEAT" << std::endl; + // std::cout << "numLiterals <= Constants::RLE_MIN_REPEAT" << std::endl; encodingType = EncodingType::DIRECT; return; } @@ -467,7 +467,7 @@ void RunLenIntEncoder::writeShortRepeatValues() header |= ((numBytesRepeatVal - 1) << 3); // repeat count (3 bits, 3~10 values) - fixedRunLength -= Constants::MIN_REPEAT; + fixedRunLength -= Constants::RLE_MIN_REPEAT; header |= fixedRunLength; // write header @@ -602,7 +602,7 @@ void RunLenIntEncoder::writeDeltaValues() // if fixed run length is greater than threshold then it will be fixed // delta sequence with delta value 0 else fixed delta sequence with // non-zero delta value - if (fixedRunLength > Constants::MIN_REPEAT) + if (fixedRunLength > Constants::RLE_MIN_REPEAT) { // ex. sequence: 2 2 2 2 2 2 2 2 len = fixedRunLength - 1; @@ -815,26 +815,26 @@ void RunLenIntEncoder::write(long value) fixedRunLength += 1; // if fixed run len meets the minimum repeat threshold, and variable len is non-zero - if (fixedRunLength >= Constants::MIN_REPEAT && variableRunLength > 0) + if (fixedRunLength >= Constants::RLE_MIN_REPEAT && variableRunLength > 0) { - numLiterals -= Constants::MIN_REPEAT; + numLiterals -= Constants::RLE_MIN_REPEAT; // before entering this branch, last (min_repeat - 1) same values are counted into variable run - variableRunLength -= (Constants::MIN_REPEAT - 1); - long *tailVals = new long[Constants::MIN_REPEAT]; + variableRunLength -= (Constants::RLE_MIN_REPEAT - 1); + long *tailVals = new long[Constants::RLE_MIN_REPEAT]; // copy out the current fixed run part // PENDING: can we use memcpy here? - std::memcpy(tailVals, literals + numLiterals, Constants::MIN_REPEAT * sizeof(long)); + std::memcpy(tailVals, literals + numLiterals, Constants::RLE_MIN_REPEAT * sizeof(long)); // flush the variable run determineEncoding(); writeValues(); // shift the tail fixed runs to the start of the buffer - memcpy(literals + numLiterals, tailVals, Constants::MIN_REPEAT * sizeof(long)); - numLiterals += Constants::MIN_REPEAT; + memcpy(literals + numLiterals, tailVals, Constants::RLE_MIN_REPEAT * sizeof(long)); + numLiterals += Constants::RLE_MIN_REPEAT; delete[] tailVals; } - if (fixedRunLength == Constants::MAX_SCOPE) + if (fixedRunLength == Constants::INT_RLE_MAX_SCOPE) { determineEncoding(); writeValues(); @@ -844,10 +844,10 @@ void RunLenIntEncoder::write(long value) else { // if fixed run length meets the minimum repeat threshold - if (fixedRunLength >= Constants::MIN_REPEAT) + if (fixedRunLength >= Constants::RLE_MIN_REPEAT) { // if meets the short repeat condition, write values as short repeats - if (fixedRunLength <= Constants::MAX_SHORT_REPEAT_LENGTH) + if (fixedRunLength <= Constants::INT_RLE_MAX_SHORT_REPEAT) { encodingType = EncodingType::SHORT_REPEAT; writeValues(); @@ -864,7 +864,7 @@ void RunLenIntEncoder::write(long value) // if fixed run length is smaller than the minimum repeat threshold // and current value is different from previous one // it is a variable run - if (fixedRunLength > 0 && fixedRunLength < Constants::MIN_REPEAT) + if (fixedRunLength > 0 && fixedRunLength < Constants::RLE_MIN_REPEAT) { if (value != literals[numLiterals - 1]) { @@ -886,7 +886,7 @@ void RunLenIntEncoder::write(long value) variableRunLength += 1; // flush variable run if it reaches the max scope - if (variableRunLength == Constants::MAX_SCOPE) + if (variableRunLength == Constants::INT_RLE_MAX_SCOPE) { determineEncoding(); writeValues(); @@ -908,15 +908,15 @@ void RunLenIntEncoder::flush() } else if (fixedRunLength != 0) { - if (fixedRunLength < Constants::MIN_REPEAT) + if (fixedRunLength < Constants::RLE_MIN_REPEAT) { variableRunLength = fixedRunLength; fixedRunLength = 0; determineEncoding(); writeValues(); } - else if (fixedRunLength >= Constants::MIN_REPEAT - && fixedRunLength <= Constants::MAX_SHORT_REPEAT_LENGTH) + else if (fixedRunLength >= Constants::RLE_MIN_REPEAT + && fixedRunLength <= Constants::INT_RLE_MAX_SHORT_REPEAT) { encodingType = EncodingType::SHORT_REPEAT; writeValues(); diff --git a/pixels-common/src/main/java/io/pixelsdb/pixels/common/utils/Constants.java b/pixels-common/src/main/java/io/pixelsdb/pixels/common/utils/Constants.java index e1bb4a315e..cb860900b4 100644 --- a/pixels-common/src/main/java/io/pixelsdb/pixels/common/utils/Constants.java +++ b/pixels-common/src/main/java/io/pixelsdb/pixels/common/utils/Constants.java @@ -39,9 +39,12 @@ public final class Constants public static final int STREAM_READER_RG_BUFFER_SIZE = 1024 * 1024; public static final int STREAM_READER_RG_FOOTER_BUFFER_SIZE = 1024; - public static final int MIN_REPEAT = 3; - public static final int MAX_SCOPE = 512; - public static final int MAX_SHORT_REPEAT_LENGTH = 10; + public static final int RLE_MIN_REPEAT = 3; + public static final int INT_RLE_MAX_SCOPE = 512; + public static final int INT_RLE_MAX_SHORT_REPEAT = 10; + public static final int BYTE_RLE_MAX_LITERAL_SIZE = 128; + public static final int BYTE_RLE_MAX_REPEAT_SIZE = 127 + RLE_MIN_REPEAT; + public static final float DICT_KEY_SIZE_THRESHOLD = 0.1F; public static final int INIT_DICT_SIZE = 4096; public static final int MAX_STREAM_RETRY_COUNT = 100; diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenByteDecoder.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenByteDecoder.java index e7d6277778..6b870ae899 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenByteDecoder.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenByteDecoder.java @@ -19,27 +19,105 @@ */ package io.pixelsdb.pixels.core.encoding; +import io.pixelsdb.pixels.common.utils.Constants; + +import java.io.EOFException; import java.io.IOException; +import java.io.InputStream; /** + * A decoder for a sequence of bytes encoded by {@link RunLenByteEncoder}. + * A control byte is read before each run with positive values 0 to 127 meaning 3 to 130 + * repetitions. If the byte is -1 to -128, 1 to 128 literal byte values follow. + * * @author guodong * @author hank */ public class RunLenByteDecoder extends Decoder { + private final InputStream inputStream; + private final byte[] literals = new byte[Constants.BYTE_RLE_MAX_LITERAL_SIZE]; + private int numLiterals = 0; + private int used = 0; + private boolean repeat = false; + + public RunLenByteDecoder(InputStream inputStream) + { + this.inputStream = inputStream; + } + + public byte next() throws IOException + { + if (used == numLiterals) + { + readValues(); + } + if (repeat) + { + used += 1; + return literals[0]; + } + else + { + return literals[used++]; + } + } + @Override public boolean hasNext() throws IOException { - return false; + return used != numLiterals || inputStream.available() > 0; } @Override - public void close() + public void close() throws IOException { + if (inputStream != null) + { + inputStream.close(); + } } - public byte next() + private void readValues() throws IOException { - return (byte) 1; + int nextByte = inputStream.read(); + if (nextByte == -1) + { + throw new EOFException("Read past end of buffer RLE byte"); + } + + int control = (byte) nextByte; + int runLength; + if (control >= 0) + { + // repeat: control 0..127 means 3..130 repetitions + int val = inputStream.read(); + if (val == -1) + { + throw new EOFException("Reading RLE byte got EOF"); + } + literals[0] = (byte) val; + runLength = control + Constants.RLE_MIN_REPEAT; + } + else + { + // literal: control -1..-128 means 1..128 literal bytes + runLength = -control; + int bytes = 0; + while (bytes < runLength) + { + int result = inputStream.read(literals, bytes, runLength - bytes); + if (result <= 0) + { + throw new EOFException("Reading RLE byte literal got EOF"); + } + bytes += result; + } + } + + // Publish a run only after its complete payload has been read. + repeat = control >= 0; + used = 0; + numLiterals = runLength; } } diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenByteEncoder.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenByteEncoder.java index ec759b3709..6ce7988204 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenByteEncoder.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenByteEncoder.java @@ -19,24 +19,22 @@ */ package io.pixelsdb.pixels.core.encoding; +import io.pixelsdb.pixels.common.utils.Constants; + import java.io.ByteArrayOutputStream; import java.io.IOException; /** * A encoder for a sequence of bytes. - * A control byte is written before each run with positive values 0 to 127 meaning 2 to 129 repetitions. + * A control byte is written before each run with positive values 0 to 127 meaning 3 to 130 repetitions. * If the bytes is -1 to -128, 1 to 128 literal byte values follow. * * @author guodong */ public class RunLenByteEncoder extends Encoder { - private static final int MIN_REPEAT_SIZE = 3; - private static final int MAX_LITERAL_SIZE = 128; - private static final int MAX_REPEAT_SIZE = 127 + MIN_REPEAT_SIZE; - private final ByteArrayOutputStream output; - private final byte[] literals = new byte[MAX_LITERAL_SIZE]; + private final byte[] literals = new byte[Constants.BYTE_RLE_MAX_LITERAL_SIZE]; private int numLiterals = 0; private boolean repeat = false; private int tailRunLength = 0; @@ -87,7 +85,7 @@ private void writeValues() { if (repeat) { - output.write(numLiterals - MIN_REPEAT_SIZE); + output.write(numLiterals - Constants.RLE_MIN_REPEAT); output.write(literals, 0, 1); } else @@ -120,7 +118,7 @@ else if (repeat) if (value == literals[0]) { numLiterals += 1; - if (numLiterals == MAX_REPEAT_SIZE) + if (numLiterals == Constants.BYTE_RLE_MAX_REPEAT_SIZE) { writeValues(); } @@ -142,26 +140,26 @@ else if (repeat) { tailRunLength = 1; } - if (tailRunLength == MIN_REPEAT_SIZE) + if (tailRunLength == Constants.RLE_MIN_REPEAT) { - if (numLiterals + 1 == MIN_REPEAT_SIZE) + if (numLiterals + 1 == Constants.RLE_MIN_REPEAT) { repeat = true; numLiterals += 1; } else { - numLiterals -= MIN_REPEAT_SIZE - 1; + numLiterals -= Constants.RLE_MIN_REPEAT - 1; writeValues(); literals[0] = value; repeat = true; - numLiterals = MIN_REPEAT_SIZE; + numLiterals = Constants.RLE_MIN_REPEAT; } } else { literals[numLiterals++] = value; - if (numLiterals == MAX_LITERAL_SIZE) + if (numLiterals == Constants.BYTE_RLE_MAX_LITERAL_SIZE) { writeValues(); } diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenIntDecoder.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenIntDecoder.java index 651f8e72a1..376a0a71b9 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenIntDecoder.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenIntDecoder.java @@ -41,7 +41,7 @@ public class RunLenIntDecoder extends IntDecoder private final InputStream inputStream; private final boolean isSigned; - private final long[] literals = new long[Constants.MAX_SCOPE]; + private final long[] literals = new long[Constants.INT_RLE_MAX_SCOPE]; private final EncodingUtils encodingUtils = new EncodingUtils(); private boolean isRepeating = false; @@ -128,8 +128,8 @@ private void readShortRepeatValues(int firstByte) throws IOException // read the run length int len = firstByte & 0x07; - // run length values are stored only after MIN_REPEAT value is met - len += Constants.MIN_REPEAT; + // run length values are stored only after RLE_MIN_REPEAT value is met + len += Constants.RLE_MIN_REPEAT; // read the repeated value which is stored using fixed bytes long val = bytesToLongBE(inputStream, size); diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenIntEncoder.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenIntEncoder.java index b591ed4b49..ec67ea5c34 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenIntEncoder.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenIntEncoder.java @@ -60,10 +60,10 @@ public class RunLenIntEncoder extends Encoder private long min; - private final long[] literals = new long[Constants.MAX_SCOPE]; - private final long[] zigzagLiterals = new long[Constants.MAX_SCOPE]; - private final long[] baseRedLiterals = new long[Constants.MAX_SCOPE]; - private final long[] adjDeltas = new long[Constants.MAX_SCOPE]; + private final long[] literals = new long[Constants.INT_RLE_MAX_SCOPE]; + private final long[] zigzagLiterals = new long[Constants.INT_RLE_MAX_SCOPE]; + private final long[] baseRedLiterals = new long[Constants.INT_RLE_MAX_SCOPE]; + private final long[] adjDeltas = new long[Constants.INT_RLE_MAX_SCOPE]; private ByteArrayOutputStream outputStream; private EncodingUtils encodingUtils; @@ -140,7 +140,7 @@ private void determineEncoding() zzBits100p = percentileBits(zigzagLiterals, 0, numLiterals, 1.0); // not a big win for shorter runs to determine encoding - if (numLiterals <= Constants.MIN_REPEAT) + if (numLiterals <= Constants.RLE_MIN_REPEAT) { encodingType = EncodingType.DIRECT; return; @@ -462,12 +462,12 @@ private void write(long value) throws IOException // if fixed run meets the minimum repeat condition and if variable run is non-zero, // then flush the variable run and shift the tail fixed runs to the start of the buffer - if (fixedRunLength >= Constants.MIN_REPEAT && variableRunLength > 0) + if (fixedRunLength >= Constants.RLE_MIN_REPEAT && variableRunLength > 0) { - numLiterals -= Constants.MIN_REPEAT; - variableRunLength -= Constants.MIN_REPEAT - 1; - long[] tailVals = new long[Constants.MIN_REPEAT]; - System.arraycopy(literals, numLiterals, tailVals, 0, Constants.MIN_REPEAT); + numLiterals -= Constants.RLE_MIN_REPEAT; + variableRunLength -= Constants.RLE_MIN_REPEAT - 1; + long[] tailVals = new long[Constants.RLE_MIN_REPEAT]; + System.arraycopy(literals, numLiterals, tailVals, 0, Constants.RLE_MIN_REPEAT); determineEncoding(); writeValues(); @@ -478,7 +478,7 @@ private void write(long value) throws IOException } } - if (fixedRunLength == Constants.MAX_SCOPE) + if (fixedRunLength == Constants.INT_RLE_MAX_SCOPE) { determineEncoding(); writeValues(); @@ -490,9 +490,9 @@ private void write(long value) throws IOException // if fixed run length is non-zero and if it satisfies the minimum repeat // and the short repeat condition, then write the values as short repeats // else use delta encoding - if (fixedRunLength >= Constants.MIN_REPEAT) + if (fixedRunLength >= Constants.RLE_MIN_REPEAT) { - if (fixedRunLength <= Constants.MAX_SHORT_REPEAT_LENGTH) + if (fixedRunLength <= Constants.INT_RLE_MAX_SHORT_REPEAT) { encodingType = EncodingType.SHORT_REPEAT; writeValues(); @@ -505,10 +505,10 @@ private void write(long value) throws IOException } } - // if fixed run length is smaller than MIN_REPEAT + // if fixed run length is smaller than RLE_MIN_REPEAT // and current value is different from previous // then treat it as variable run - if (fixedRunLength > 0 && fixedRunLength < Constants.MIN_REPEAT) + if (fixedRunLength > 0 && fixedRunLength < Constants.RLE_MIN_REPEAT) { if (value != literals[numLiterals - 1]) { @@ -530,7 +530,7 @@ private void write(long value) throws IOException variableRunLength += 1; // if variable run length reach the max scope, write it - if (variableRunLength == Constants.MAX_SCOPE) + if (variableRunLength == Constants.INT_RLE_MAX_SCOPE) { determineEncoding(); writeValues(); @@ -552,15 +552,15 @@ private void flush() throws IOException } else if (fixedRunLength != 0) { - if (fixedRunLength < Constants.MIN_REPEAT) + if (fixedRunLength < Constants.RLE_MIN_REPEAT) { variableRunLength = fixedRunLength; fixedRunLength = 0; determineEncoding(); writeValues(); } - else if (fixedRunLength >= Constants.MIN_REPEAT - && fixedRunLength <= Constants.MAX_SHORT_REPEAT_LENGTH) + else if (fixedRunLength >= Constants.RLE_MIN_REPEAT + && fixedRunLength <= Constants.INT_RLE_MAX_SHORT_REPEAT) { encodingType = EncodingType.SHORT_REPEAT; writeValues(); @@ -598,7 +598,7 @@ private void writeShortRepeatValues() // repeat count(3 bytes, 3 to 10 values) int header = getOpcode(); header |= ((numBytesRepeatVal - 1) << 3); - fixedRunLength -= Constants.MIN_REPEAT; + fixedRunLength -= Constants.RLE_MIN_REPEAT; header |= fixedRunLength; // write header @@ -739,7 +739,7 @@ private void writeDeltaValues() throws IOException // if fixed run length is greater than threshold then it will be fixed // delta sequence with delta value 0 else fixed delta sequence with // non-zero delta value - if (fixedRunLength > Constants.MIN_REPEAT) + if (fixedRunLength > Constants.RLE_MIN_REPEAT) { // ex. sequence: 2 2 2 2 2 2 2 2 len = fixedRunLength - 1; diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/ByteColumnReader.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/ByteColumnReader.java index 3e0cb717d5..012e6fa290 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/ByteColumnReader.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/ByteColumnReader.java @@ -21,17 +21,31 @@ import io.pixelsdb.pixels.core.PixelsProto; import io.pixelsdb.pixels.core.TypeDescription; +import io.pixelsdb.pixels.core.encoding.RunLenByteDecoder; +import io.pixelsdb.pixels.core.utils.BitUtils; import io.pixelsdb.pixels.core.utils.Bitmap; +import io.pixelsdb.pixels.core.utils.ByteBufferInputStream; +import io.pixelsdb.pixels.core.vector.ByteColumnVector; import io.pixelsdb.pixels.core.vector.ColumnVector; import java.io.IOException; +import java.io.InputStream; import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.Arrays; /** + * Column reader for byte / tinyint columns. + * * @author guodong + * @author hank */ public class ByteColumnReader extends ColumnReader { + private RunLenByteDecoder decoder; + private ByteBuffer inputBuffer; + private InputStream inputStream; + ByteColumnReader(TypeDescription type) { super(type); @@ -53,6 +67,13 @@ public class ByteColumnReader extends ColumnReader @Override public void close() throws IOException { + if (this.decoder != null) + { + // inputStream is closed inside decoder.close(); + this.decoder.close(); + this.decoder = null; + } + this.inputBuffer = null; } /** @@ -70,9 +91,99 @@ public void close() throws IOException @Override public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, int offset, int size, int pixelStride, final int vectorIndex, - ColumnVector vector, PixelsProto.ColumnChunkIndex chunkIndex) + ColumnVector vector, PixelsProto.ColumnChunkIndex chunkIndex) throws IOException { - throw new UnsupportedOperationException("Not implemented yet."); + ByteColumnVector columnVector = (ByteColumnVector) vector; + boolean decoding = encoding.getKind().equals(PixelsProto.ColumnEncoding.Kind.RUNLENGTH); + boolean nullsPadding = chunkIndex.hasNullsPadding() && chunkIndex.getNullsPadding(); + boolean littleEndian = chunkIndex.hasLittleEndian() && chunkIndex.getLittleEndian(); + // if read from start, init the stream and decoder + if (offset == 0) + { + if (inputStream != null) + { + inputStream.close(); + } + this.inputBuffer = input; + this.inputBuffer.order(littleEndian ? ByteOrder.LITTLE_ENDIAN : ByteOrder.BIG_ENDIAN); + inputStream = new ByteBufferInputStream(inputBuffer, inputBuffer.position(), inputBuffer.limit()); + decoder = new RunLenByteDecoder(inputStream); + // isNull + isNullOffset = inputBuffer.position() + chunkIndex.getIsNullOffset(); + isNullSkipBits = 0; + // re-init + hasNull = true; + elementIndex = 0; + } + + // read without copying the de-compacted content and isNull + int numLeft = size, numToRead, bytesToDeCompact; + boolean endOfPixel; + for (int i = vectorIndex; numLeft > 0; ) + { + if (elementIndex / pixelStride < (elementIndex + numLeft) / pixelStride) + { + // read to the end of the current pixel + numToRead = pixelStride - elementIndex % pixelStride; + endOfPixel = true; + } + else + { + numToRead = numLeft; + endOfPixel = false; + } + bytesToDeCompact = (numToRead + isNullSkipBits + (endOfPixel ? 7 : 0)) / 8; + // read isNull + int pixelId = elementIndex / pixelStride; + hasNull = chunkIndex.getPixelStatistics(pixelId).getStatistic().getHasNull(); + if (hasNull) + { + BitUtils.bitWiseDeCompact(columnVector.isNull, i, numToRead, + inputBuffer, isNullOffset, isNullSkipBits, littleEndian); + isNullOffset += bytesToDeCompact; + isNullSkipBits = endOfPixel ? 0 : (numToRead + isNullSkipBits) % 8; + columnVector.noNulls = false; + } + else + { + Arrays.fill(columnVector.isNull, i, i + numToRead, false); + } + // read content + if (decoding) + { + for (int j = i; j < i + numToRead; ++j) + { + if (!(hasNull && columnVector.isNull[j])) + { + columnVector.vector[j] = decoder.next(); + } + } + } + else + { + if (nullsPadding) + { + for (int j = i; j < i + numToRead; ++j) + { + columnVector.vector[j] = inputBuffer.get(); + } + } + else + { + for (int j = i; j < i + numToRead; ++j) + { + if (!(hasNull && columnVector.isNull[j])) + { + columnVector.vector[j] = inputBuffer.get(); + } + } + } + } + // update variables + numLeft -= numToRead; + elementIndex += numToRead; + i += numToRead; + } } /** @@ -92,8 +203,156 @@ public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, @Override public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, int offset, int size, int pixelStride, final int vectorIndex, - ColumnVector vector, PixelsProto.ColumnChunkIndex chunkIndex, Bitmap selected) + ColumnVector vector, PixelsProto.ColumnChunkIndex chunkIndex, Bitmap selected) throws IOException { - throw new UnsupportedOperationException("Not implemented yet."); + ByteColumnVector columnVector = (ByteColumnVector) vector; + boolean decoding = encoding.getKind().equals(PixelsProto.ColumnEncoding.Kind.RUNLENGTH); + boolean nullsPadding = chunkIndex.hasNullsPadding() && chunkIndex.getNullsPadding(); + boolean littleEndian = chunkIndex.hasLittleEndian() && chunkIndex.getLittleEndian(); + // if read from start, init the stream and decoder + if (offset == 0) + { + if (inputStream != null) + { + inputStream.close(); + } + this.inputBuffer = input; + this.inputBuffer.order(littleEndian ? ByteOrder.LITTLE_ENDIAN : ByteOrder.BIG_ENDIAN); + inputStream = new ByteBufferInputStream(inputBuffer, inputBuffer.position(), inputBuffer.limit()); + decoder = new RunLenByteDecoder(inputStream); + // isNull + isNullOffset = inputBuffer.position() + chunkIndex.getIsNullOffset(); + isNullSkipBits = 0; + // re-init + hasNull = true; + elementIndex = 0; + } + + // read without copying the de-compacted content and isNull + int numLeft = size, numToRead, bytesToDeCompact, vectorWriteIndex = vectorIndex; + boolean[] isNull = null; + boolean endOfPixel; + if (decoding || !nullsPadding) + { + isNull = new boolean[size]; + } + for (int i = vectorIndex; numLeft > 0; ) + { + if (elementIndex / pixelStride < (elementIndex + numLeft) / pixelStride) + { + // read to the end of the current pixel + numToRead = pixelStride - elementIndex % pixelStride; + endOfPixel = true; + } + else + { + numToRead = numLeft; + endOfPixel = false; + } + bytesToDeCompact = (numToRead + isNullSkipBits + (endOfPixel ? 7 : 0)) / 8; + + // read isNull + int pixelId = elementIndex / pixelStride; + hasNull = chunkIndex.getPixelStatistics(pixelId).getStatistic().getHasNull(); + if (hasNull) + { + if (!decoding && nullsPadding) + { + // read isNull directly into the vector of the column chunk + BitUtils.bitWiseDeCompact(columnVector.isNull, vectorWriteIndex, numToRead, inputBuffer, + isNullOffset, isNullSkipBits, littleEndian, selected, i - vectorIndex); + } + else + { + // need to keep isNull for later use + BitUtils.bitWiseDeCompact(isNull, i - vectorIndex, numToRead, inputBuffer, + isNullOffset, isNullSkipBits, littleEndian); + // update columnVector.isNull + int k = vectorWriteIndex; + for (int j = i; j < i + numToRead; ++j) + { + if (selected.get(j - vectorIndex)) + { + columnVector.isNull[k++] = isNull[j - vectorIndex]; + } + } + } + isNullOffset += bytesToDeCompact; + isNullSkipBits = endOfPixel ? 0 : (numToRead + isNullSkipBits) % 8; + columnVector.noNulls = false; + } + else + { + if (decoding || !nullsPadding) + { + Arrays.fill(isNull, i - vectorIndex, i - vectorIndex + numToRead, false); + } + // update columnVector.isNull later to avoid bitmap unnecessary traversal + } + + // read content + int originalVectorWriteIndex = vectorWriteIndex; + if (decoding) + { + for (int j = i; j < i + numToRead; ++j) + { + if (!(hasNull && isNull[j - vectorIndex])) + { + byte value = decoder.next(); + if (selected.get(j - vectorIndex)) + { + columnVector.vector[vectorWriteIndex++] = value; + } + } + else if (selected.get(j - vectorIndex)) + { + vectorWriteIndex++; + } + } + } + else + { + if (nullsPadding) + { + for (int j = i; j < i + numToRead; ++j) + { + byte value = inputBuffer.get(); + if (selected.get(j - vectorIndex)) + { + columnVector.vector[vectorWriteIndex++] = value; + } + } + } + else + { + for (int j = i; j < i + numToRead; ++j) + { + if (!(hasNull && isNull[j - vectorIndex])) + { + byte value = inputBuffer.get(); + if (selected.get(j - vectorIndex)) + { + columnVector.vector[vectorWriteIndex++] = value; + } + } + else if (selected.get(j - vectorIndex)) + { + vectorWriteIndex++; + } + } + } + } + + // update columnVector.isNull if has no nulls + if (!hasNull) + { + Arrays.fill(columnVector.isNull, originalVectorWriteIndex, vectorWriteIndex, false); + } + + // update variables + numLeft -= numToRead; + elementIndex += numToRead; + i += numToRead; + } } } diff --git a/pixels-core/src/test/java/io/pixelsdb/pixels/core/encoding/TestEncoding.java b/pixels-core/src/test/java/io/pixelsdb/pixels/core/encoding/TestEncoding.java index ea039b83f0..8e08830dcc 100644 --- a/pixels-core/src/test/java/io/pixelsdb/pixels/core/encoding/TestEncoding.java +++ b/pixels-core/src/test/java/io/pixelsdb/pixels/core/encoding/TestEncoding.java @@ -26,9 +26,12 @@ import java.io.ByteArrayInputStream; import java.io.IOException; import java.nio.ByteOrder; +import java.util.Arrays; import java.util.Random; import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; /** * pixels @@ -69,6 +72,59 @@ public void runLengthTest() } } + @Test + public void runLengthByteTest() throws IOException + { + System.out.println("=== RunLenByte empty ==="); + RunLenByteEncoder emptyEncoder = new RunLenByteEncoder(); + byte[] emptyEncoded = emptyEncoder.encode(new byte[0]); + emptyEncoder.close(); + System.out.println("encoded length: " + emptyEncoded.length); + assertEquals(0, emptyEncoded.length); + try (RunLenByteDecoder emptyDecoder = new RunLenByteDecoder(new ByteArrayInputStream(emptyEncoded))) + { + assertFalse(emptyDecoder.hasNext()); + System.out.println("decoder.hasNext(): false (ok)"); + } + + System.out.println("=== RunLenByte mixed round-trip ==="); + // mixed: literals, repeats across BYTE_RLE_MAX_REPEAT_SIZE, and boundary values + byte[] values = new byte[200]; + Arrays.fill(values, 0, 5, (byte) 1); + Arrays.fill(values, 5, 140, (byte) 7); // crosses 130-repeat boundary + values[140] = (byte) -128; + values[141] = 0; + values[142] = 127; + for (int i = 143; i < values.length; i++) + { + values[i] = (byte) i; + } + System.out.println("input rows: " + values.length + + " (5x1, 135x7 across 130-repeat boundary, then -128/0/127 + literals)"); + System.out.println("input sample [0..9]: " + Arrays.toString(Arrays.copyOfRange(values, 0, 10))); + System.out.println("input sample [138..149]: " + Arrays.toString(Arrays.copyOfRange(values, 138, 150))); + + RunLenByteEncoder encoder = new RunLenByteEncoder(); + byte[] encoded = encoder.encode(values); + encoder.close(); + System.out.println("encoded bytes: " + encoded.length + + " (compression ratio " + String.format("%.2f", (double) encoded.length / values.length) + ")"); + + byte[] decoded = new byte[values.length]; + try (RunLenByteDecoder decoder = new RunLenByteDecoder(new ByteArrayInputStream(encoded))) + { + for (int i = 0; i < values.length; i++) + { + decoded[i] = decoder.next(); + } + assertFalse(decoder.hasNext()); + } + System.out.println("decoded sample [0..9]: " + Arrays.toString(Arrays.copyOfRange(decoded, 0, 10))); + System.out.println("decoded sample [138..149]: " + Arrays.toString(Arrays.copyOfRange(decoded, 138, 150))); + assertArrayEquals(values, decoded); + System.out.println("round-trip OK"); + } + @Test public void longTest() { diff --git a/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestByteColumnReader.java b/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestByteColumnReader.java new file mode 100644 index 0000000000..f5021c4b95 --- /dev/null +++ b/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestByteColumnReader.java @@ -0,0 +1,452 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the Affero GNU General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.core.reader; + +import io.pixelsdb.pixels.core.PixelsProto; +import io.pixelsdb.pixels.core.TypeDescription; +import io.pixelsdb.pixels.core.encoding.EncodingLevel; +import io.pixelsdb.pixels.core.utils.Bitmap; +import io.pixelsdb.pixels.core.vector.ByteColumnVector; +import io.pixelsdb.pixels.core.writer.ByteColumnWriter; +import io.pixelsdb.pixels.core.writer.PixelsWriterOption; +import org.junit.Test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; + +import static org.junit.Assert.assertEquals; + +/** + * @author hank + */ +public class TestByteColumnReader +{ + private static String formatCell(ByteColumnVector vector, int i) + { + if (!vector.noNulls && vector.isNull[i]) + { + return "NULL"; + } + return Byte.toString(vector.vector[i]); + } + + private static void printVector(String label, ByteColumnVector vector, int numRows) + { + StringBuilder sb = new StringBuilder(label).append(" ["); + for (int i = 0; i < numRows; ++i) + { + if (i > 0) + { + sb.append(", "); + } + sb.append(formatCell(vector, i)); + } + sb.append(']'); + System.out.println(sb); + } + + private static void assertVectorsEqual(String caseName, ByteColumnVector expected, + ByteColumnVector actual, int numRows) + { + System.out.println("--- " + caseName + " ---"); + System.out.println("rows: " + numRows + ", expected.noNulls=" + expected.noNulls + + ", actual.noNulls=" + actual.noNulls); + printVector("expected", expected, numRows); + printVector("actual ", actual, numRows); + assertEquals(expected.noNulls, actual.noNulls); + for (int i = 0; i < numRows; ++i) + { + assertEquals("isNull mismatch at row " + i, expected.isNull[i], actual.isNull[i]); + if (expected.noNulls || !expected.isNull[i]) + { + assertEquals("value mismatch at row " + i, expected.vector[i], actual.vector[i]); + } + } + System.out.println("OK: all " + numRows + " rows match"); + } + + private static ByteColumnVector createSampleVector(int numRows) + { + ByteColumnVector vector = new ByteColumnVector(numRows); + vector.add((byte) 100); + vector.add((byte) 103); + vector.add((byte) 106); + vector.add((byte) 34); + vector.addNull(); + vector.add((byte) 54); + vector.add((byte) 55); + vector.add((byte) 67); + vector.addNull(); + vector.add((byte) 34); + vector.add((byte) 55); + vector.add((byte) 56); + vector.add((byte) -34); + vector.add((byte) 67); + vector.add((byte) 23); + vector.add((byte) 34); + vector.addNull(); + vector.add((byte) 6); + vector.add((byte) 7); + vector.add((byte) 65); + vector.add((byte) 34); + vector.add((byte) 78); + return vector; + } + + private static void assertSelectedRoundTrip(EncodingLevel encodingLevel, boolean nullsPadding, + PixelsProto.ColumnEncoding.Kind expectedEncoding) + throws IOException + { + int pixelsStride = 10; + int numRows = 22; + int vectorIndex = 3; + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelsStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(encodingLevel).nullsPadding(nullsPadding); + ByteColumnWriter columnWriter = new ByteColumnWriter( + TypeDescription.createByte(), writerOption); + ByteColumnVector originVector = createSampleVector(numRows); + columnWriter.write(originVector, numRows); + columnWriter.flush(); + columnWriter.close(); + + byte[] content = columnWriter.getColumnChunkContent(); + PixelsProto.ColumnChunkIndex chunkIndex = columnWriter.getColumnChunkIndex().build(); + PixelsProto.ColumnEncoding encoding = columnWriter.getColumnChunkEncoding().build(); + assertEquals(expectedEncoding, encoding.getKind()); + System.out.println("--- selected round-trip encodingLevel=" + encodingLevel + + ", nullsPadding=" + nullsPadding + + ", encoding=" + encoding.getKind() + + ", chunkBytes=" + content.length + + ", vectorIndex=" + vectorIndex + " ---"); + printVector("origin ", originVector, numRows); + + Bitmap selected = new Bitmap(numRows, true); + // Skip non-null values so the reader must still consume their encoded payload. + selected.clear(0); + selected.clear(2); + selected.clear(5); + selected.clear(10); + selected.clear(14); + selected.clear(20); + System.out.print("selected rows: "); + for (int i = 0; i < numRows; ++i) + { + if (selected.get(i)) + { + System.out.print(i + " "); + } + } + System.out.println(); + + ByteColumnReader columnReader = new ByteColumnReader(TypeDescription.createByte()); + ByteColumnVector targetVector = new ByteColumnVector(vectorIndex + numRows); + columnReader.readSelected(ByteBuffer.wrap(content), encoding, 0, numRows, + pixelsStride, vectorIndex, targetVector, chunkIndex, selected); + columnReader.close(); + + int targetIndex = vectorIndex; + StringBuilder expectedSel = new StringBuilder("expected selected ["); + StringBuilder actualSel = new StringBuilder("actual selected ["); + boolean first = true; + for (int i = 0; i < numRows; ++i) + { + if (selected.get(i)) + { + if (!first) + { + expectedSel.append(", "); + actualSel.append(", "); + } + first = false; + expectedSel.append(formatCell(originVector, i)); + actualSel.append(formatCell(targetVector, targetIndex)); + assertEquals("isNull mismatch at selected src=" + i + " dst=" + targetIndex, + originVector.isNull[i], targetVector.isNull[targetIndex]); + if (!originVector.isNull[i]) + { + assertEquals("value mismatch at selected src=" + i + " dst=" + targetIndex, + originVector.vector[i], targetVector.vector[targetIndex]); + } + targetIndex++; + } + } + System.out.println(expectedSel.append(']')); + System.out.println(actualSel.append(']')); + System.out.println("OK: selected rows match, written from vectorIndex=" + vectorIndex); + } + + @Test + public void testNullsPadding() throws IOException + { + int pixelsStride = 10; + int numRows = 22; + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelsStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL0).nullsPadding(true); + ByteColumnWriter columnWriter = new ByteColumnWriter( + TypeDescription.createByte(), writerOption); + ByteColumnVector originVector = createSampleVector(numRows); + columnWriter.write(originVector, numRows); + columnWriter.flush(); + columnWriter.close(); + + byte[] content = columnWriter.getColumnChunkContent(); + PixelsProto.ColumnChunkIndex chunkIndex = columnWriter.getColumnChunkIndex().build(); + PixelsProto.ColumnEncoding encoding = columnWriter.getColumnChunkEncoding().build(); + assertEquals(PixelsProto.ColumnEncoding.Kind.NONE, encoding.getKind()); + System.out.println("encoding=" + encoding.getKind() + ", nullsPadding=true, chunkBytes=" + content.length); + ByteColumnReader columnReader = new ByteColumnReader(TypeDescription.createByte()); + ByteColumnVector targetVector = new ByteColumnVector(numRows); + columnReader.read(ByteBuffer.wrap(content), encoding, 0, numRows, + pixelsStride, 0, targetVector, chunkIndex); + columnReader.close(); + + assertVectorsEqual("NONE + nullsPadding", originVector, targetVector, numRows); + } + + @Test + public void testWithoutNullsPadding() throws IOException + { + int pixelsStride = 10; + int numRows = 22; + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelsStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL0).nullsPadding(false); + ByteColumnWriter columnWriter = new ByteColumnWriter( + TypeDescription.createByte(), writerOption); + ByteColumnVector originVector = createSampleVector(numRows); + columnWriter.write(originVector, numRows); + columnWriter.flush(); + columnWriter.close(); + + byte[] content = columnWriter.getColumnChunkContent(); + PixelsProto.ColumnChunkIndex chunkIndex = columnWriter.getColumnChunkIndex().build(); + PixelsProto.ColumnEncoding encoding = columnWriter.getColumnChunkEncoding().build(); + assertEquals(PixelsProto.ColumnEncoding.Kind.NONE, encoding.getKind()); + System.out.println("encoding=" + encoding.getKind() + ", nullsPadding=false, chunkBytes=" + content.length); + ByteColumnReader columnReader = new ByteColumnReader(TypeDescription.createByte()); + ByteColumnVector targetVector = new ByteColumnVector(numRows); + columnReader.read(ByteBuffer.wrap(content), encoding, 0, numRows, + pixelsStride, 0, targetVector, chunkIndex); + columnReader.close(); + + assertVectorsEqual("NONE without nullsPadding", originVector, targetVector, numRows); + } + + @Test + public void testRunLength() throws IOException + { + int pixelsStride = 10; + int numRows = 22; + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelsStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL2).nullsPadding(false); + ByteColumnWriter columnWriter = new ByteColumnWriter( + TypeDescription.createByte(), writerOption); + ByteColumnVector originVector = new ByteColumnVector(numRows); + // include repeats and nulls so RLE is exercised + originVector.add((byte) 5); + originVector.add((byte) 5); + originVector.add((byte) 5); + originVector.add((byte) 5); + originVector.addNull(); + originVector.add((byte) 5); + originVector.add((byte) 5); + originVector.add((byte) 7); + originVector.addNull(); + originVector.add((byte) 7); + originVector.add((byte) 1); + originVector.add((byte) 2); + originVector.add((byte) 3); + originVector.add((byte) 9); + originVector.add((byte) 9); + originVector.add((byte) 9); + originVector.addNull(); + originVector.add((byte) 9); + originVector.add((byte) 9); + originVector.add((byte) -128); + originVector.add((byte) 127); + originVector.add((byte) 0); + columnWriter.write(originVector, numRows); + columnWriter.flush(); + columnWriter.close(); + + byte[] content = columnWriter.getColumnChunkContent(); + PixelsProto.ColumnChunkIndex chunkIndex = columnWriter.getColumnChunkIndex().build(); + PixelsProto.ColumnEncoding encoding = columnWriter.getColumnChunkEncoding().build(); + assertEquals(PixelsProto.ColumnEncoding.Kind.RUNLENGTH, encoding.getKind()); + System.out.println("encoding=" + encoding.getKind() + ", EL2, chunkBytes=" + content.length); + ByteColumnReader columnReader = new ByteColumnReader(TypeDescription.createByte()); + ByteColumnVector targetVector = new ByteColumnVector(numRows); + columnReader.read(ByteBuffer.wrap(content), encoding, 0, numRows, + pixelsStride, 0, targetVector, chunkIndex); + columnReader.close(); + + assertVectorsEqual("RUNLENGTH", originVector, targetVector, numRows); + } + + @Test + public void testSelected() throws IOException + { + int pixelsStride = 10; + int numRows = 22; + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelsStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL0).nullsPadding(true); + ByteColumnWriter columnWriter = new ByteColumnWriter( + TypeDescription.createByte(), writerOption); + ByteColumnVector originVector = createSampleVector(numRows); + columnWriter.write(originVector, numRows); + columnWriter.flush(); + columnWriter.close(); + + byte[] content = columnWriter.getColumnChunkContent(); + PixelsProto.ColumnChunkIndex chunkIndex = columnWriter.getColumnChunkIndex().build(); + PixelsProto.ColumnEncoding encoding = columnWriter.getColumnChunkEncoding().build(); + System.out.println("--- selected (skip every 10th row) encoding=" + encoding.getKind() + + ", chunkBytes=" + content.length + " ---"); + printVector("origin ", originVector, numRows); + ByteColumnReader columnReader = new ByteColumnReader(TypeDescription.createByte()); + ByteColumnVector targetVector = new ByteColumnVector(numRows); + Bitmap selected = new Bitmap(numRows, true); + selected.clear(0); + selected.clear(10); + selected.clear(20); + columnReader.readSelected(ByteBuffer.wrap(content), encoding, 0, numRows, + pixelsStride, 0, targetVector, chunkIndex, selected); + columnReader.close(); + + StringBuilder expectedSel = new StringBuilder("expected selected ["); + StringBuilder actualSel = new StringBuilder("actual selected ["); + for (int i = 0, j = 0; i < numRows; ++i) + { + if (i % 10 != 0) + { + if (j > 0) + { + expectedSel.append(", "); + actualSel.append(", "); + } + expectedSel.append(formatCell(originVector, i)); + actualSel.append(formatCell(targetVector, j)); + assertEquals(originVector.noNulls, targetVector.noNulls); + assertEquals("isNull mismatch at src=" + i + " dst=" + j, + originVector.isNull[i], targetVector.isNull[j]); + if (originVector.noNulls || !originVector.isNull[i]) + { + assertEquals("value mismatch at src=" + i + " dst=" + j, + originVector.vector[i], targetVector.vector[j]); + } + j++; + } + } + System.out.println(expectedSel.append(']')); + System.out.println(actualSel.append(']')); + System.out.println("OK: selected rows match"); + } + + @Test + public void testSelectedWithoutNullsPaddingAtNonZeroVectorIndex() throws IOException + { + assertSelectedRoundTrip(EncodingLevel.EL0, false, PixelsProto.ColumnEncoding.Kind.NONE); + } + + @Test + public void testSelectedRunLengthAtNonZeroVectorIndex() throws IOException + { + assertSelectedRoundTrip(EncodingLevel.EL2, false, PixelsProto.ColumnEncoding.Kind.RUNLENGTH); + } + + @Test + public void testLargeFragmented() throws IOException + { + int numBatches = 15; + int numRows = 1024; + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(10000).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL2).nullsPadding(false); + ByteColumnWriter columnWriter = new ByteColumnWriter( + TypeDescription.createByte(), writerOption); + + ByteColumnVector originVector = new ByteColumnVector(numRows); + for (int j = 0; j < numRows; j++) + { + if (j % 100 == 0) + { + originVector.addNull(); + } + else + { + originVector.add((byte) ((j / 200) % 4)); + } + } + + for (int i = 0; i < numBatches; i++) + { + columnWriter.write(originVector, numRows); + } + columnWriter.flush(); + columnWriter.close(); + + byte[] content = columnWriter.getColumnChunkContent(); + PixelsProto.ColumnChunkIndex chunkIndex = columnWriter.getColumnChunkIndex().build(); + PixelsProto.ColumnEncoding encoding = columnWriter.getColumnChunkEncoding().build(); + assertEquals(PixelsProto.ColumnEncoding.Kind.RUNLENGTH, encoding.getKind()); + int totalRows = numBatches * numRows; + System.out.println("--- large fragmented ---"); + System.out.println("encoding=" + encoding.getKind() + + ", batches=" + numBatches + + ", rowsPerBatch=" + numRows + + ", totalRows=" + totalRows + + ", chunkBytes=" + content.length); + System.out.println("read ranges: [0,123), [123,579), [579," + totalRows + ")"); + ByteColumnReader columnReader = new ByteColumnReader(TypeDescription.createByte()); + ByteColumnVector targetVector = new ByteColumnVector(totalRows); + ByteBuffer buffer = ByteBuffer.wrap(content); + columnReader.read(buffer, encoding, 0, 123, + 10000, 0, targetVector, chunkIndex); + columnReader.read(buffer, encoding, 123, 456, + 10000, 123, targetVector, chunkIndex); + columnReader.read(buffer, encoding, 123 + 456, totalRows - 123 - 456, + 10000, 123 + 456, targetVector, chunkIndex); + columnReader.close(); + + int mismatches = 0; + for (int i = 0; i < totalRows; i++) + { + assertEquals("isNull mismatch at row " + i, + originVector.isNull[i % numRows], targetVector.isNull[i]); + if (targetVector.noNulls || !targetVector.isNull[i]) + { + if (originVector.vector[i % numRows] != targetVector.vector[i]) + { + mismatches++; + } + assertEquals("value mismatch at row " + i, + originVector.vector[i % numRows], targetVector.vector[i]); + } + } + System.out.println("sample actual [0..15]: " + + java.util.Arrays.toString(java.util.Arrays.copyOf(targetVector.vector, 16))); + System.out.println("OK: all " + totalRows + " rows match, mismatches=" + mismatches); + } +}