From d232fcf3d231909196746960f11d4d00cb0571c3 Mon Sep 17 00:00:00 2001 From: Dongyang Geng Date: Mon, 3 Aug 2026 15:03:22 +0800 Subject: [PATCH 1/2] fix: align Pixels block loading with Trino 466 --- .../pixels/trino/PixelsBlockLoader.java | 32 +- .../pixelsdb/pixels/trino/PixelsPlugin.java | 3 +- .../pixels/trino/block/TimeArrayBlock.java | 356 ------------------ .../trino/block/TimeArrayBlockEncoding.java | 102 ----- 4 files changed, 26 insertions(+), 467 deletions(-) delete mode 100644 connector/src/main/java/io/pixelsdb/pixels/trino/block/TimeArrayBlock.java delete mode 100644 connector/src/main/java/io/pixelsdb/pixels/trino/block/TimeArrayBlockEncoding.java diff --git a/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsBlockLoader.java b/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsBlockLoader.java index d2529ef..b375c18 100644 --- a/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsBlockLoader.java +++ b/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsBlockLoader.java @@ -22,7 +22,6 @@ import io.airlift.slice.Slices; import io.pixelsdb.pixels.core.TypeDescription; import io.pixelsdb.pixels.core.vector.*; -import io.pixelsdb.pixels.trino.block.TimeArrayBlock; import io.pixelsdb.pixels.trino.block.VarcharArrayBlock; import io.pixelsdb.pixels.trino.block.VarcharArrayBlockEncoding; import io.trino.spi.block.*; @@ -70,7 +69,26 @@ public Block load() switch (typeCategory) { case BYTE: + ByteColumnVector bytecv = (ByteColumnVector) vector; + block = new ByteArrayBlock(batchSize, Optional.ofNullable(bytecv.isNull), bytecv.vector); + break; case SHORT: + IntColumnVector shortcv = (IntColumnVector) vector; + short[] shortValues = new short[batchSize]; + for (int i = 0; i < batchSize; ++i) + { + if (!shortcv.isNull[i]) + { + int value = shortcv.vector[i]; + if (value < Short.MIN_VALUE || value > Short.MAX_VALUE) + { + throw new IllegalArgumentException("SMALLINT value out of range: " + value); + } + shortValues[i] = (short) value; + } + } + block = new ShortArrayBlock(batchSize, Optional.ofNullable(shortcv.isNull), shortValues); + break; case INT: IntColumnVector icv = (IntColumnVector) vector; block = new IntArrayBlock(batchSize, Optional.ofNullable(icv.isNull), icv.vector); @@ -146,12 +164,12 @@ public Block load() case TIME: // PIXELS-94: add time type. TimeColumnVector tcv = (TimeColumnVector) vector; - /** - * In Presto, LongArrayBlock is used for time type. However, in Pixels, - * Time value is stored as int, so here we use TimeArrayBlock, which - * accepts int values but provides getLong method same as LongArrayBlock. - */ - block = new TimeArrayBlock(batchSize, tcv.times, !tcv.noNulls, tcv.isNull); + long[] timeValues = new long[batchSize]; + for (int i = 0; i < batchSize; ++i) + { + timeValues[i] = (long) tcv.times[i] * 1_000_000_000L; + } + block = new LongArrayBlock(batchSize, Optional.ofNullable(tcv.isNull), timeValues); break; case TIMESTAMP: TimestampColumnVector tscv = (TimestampColumnVector) vector; diff --git a/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsPlugin.java b/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsPlugin.java index f7bbab9..c78a972 100644 --- a/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsPlugin.java +++ b/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsPlugin.java @@ -20,7 +20,6 @@ package io.pixelsdb.pixels.trino; import com.google.common.collect.ImmutableList; -import io.pixelsdb.pixels.trino.block.TimeArrayBlockEncoding; import io.pixelsdb.pixels.trino.block.VarcharArrayBlockEncoding; import io.trino.spi.Plugin; import io.trino.spi.block.BlockEncoding; @@ -31,7 +30,7 @@ public class PixelsPlugin implements Plugin @Override public Iterable getBlockEncodings() { - return ImmutableList.of(VarcharArrayBlockEncoding.Instance(), TimeArrayBlockEncoding.Instance()); + return ImmutableList.of(VarcharArrayBlockEncoding.Instance()); } @Override diff --git a/connector/src/main/java/io/pixelsdb/pixels/trino/block/TimeArrayBlock.java b/connector/src/main/java/io/pixelsdb/pixels/trino/block/TimeArrayBlock.java deleted file mode 100644 index be3985a..0000000 --- a/connector/src/main/java/io/pixelsdb/pixels/trino/block/TimeArrayBlock.java +++ /dev/null @@ -1,356 +0,0 @@ -/* - * Copyright 2021 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.trino.block; - -import io.trino.spi.block.Block; -import io.trino.spi.block.ByteArrayBlock; -import io.trino.spi.block.ValueBlock; -import org.openjdk.jol.info.ClassLayout; - -import java.util.Optional; -import java.util.OptionalInt; -import java.util.function.ObjLongConsumer; - -import static io.airlift.slice.SizeOf.sizeOf; -import static io.pixelsdb.pixels.trino.block.BlockUtil.*; - -/** - * This class is derived from io.trino.spi.block.IntArrayBlock. - * - * With this class, we use int values to simulate a LongArrayBlock, so that - * we can reduce 50% memory footprint. Int value is enough for time type - * in Pixels. - * - * Modifications: - * 1. add getLong, getShort, getByte, so that this class can be compatible - * with io.trino.spi.block.LongArrayBlock. - * - * 2. change the returned statement of the methods that return Block or - * BlockEncoding. - * - * @author hank - * @create 2021-04-26 - * @update 2024-12-01 adapt to with Trino 465 and add hasNull argument to the constructor. - */ -public class TimeArrayBlock implements ValueBlock -{ - private static final long INSTANCE_SIZE = ClassLayout.parseClass(TimeArrayBlock.class).instanceSize(); - public static final int SIZE_IN_BYTES_PER_POSITION = Integer.BYTES + Byte.BYTES; - /** - * Trino assumes each time value is a long of precision 12, - * so we need to multiply a scale factor for each value. - */ - private static final long SCALE_FACTOR = 1000000000L; - - private final int arrayOffset; - private final int positionCount; - private final int[] values; - private final boolean[] valueIsNull; - private final boolean hasNull; - - private final long sizeInBytes; - private final long retainedSizeInBytes; - - public TimeArrayBlock(int positionCount, int[] values, boolean hasNull, boolean[] valueIsNull) - { - this(0, positionCount, values, hasNull, valueIsNull); - } - - TimeArrayBlock(int arrayOffset, int positionCount, int[] values, boolean hasNull, boolean[] valueIsNull) - { - if (arrayOffset < 0) - { - throw new IllegalArgumentException("arrayOffset is negative"); - } - this.arrayOffset = arrayOffset; - if (positionCount < 0) - { - throw new IllegalArgumentException("positionCount is negative"); - } - this.positionCount = positionCount; - - if (values == null || values.length - arrayOffset < positionCount) - { - throw new IllegalArgumentException("values is null or its length is less than positionCount"); - } - this.values = values; - - this.hasNull = hasNull; - // Issue #123: in Pixels, the isNull bitmap from column vectors always presents even if there is no nulls. - if (valueIsNull == null || valueIsNull.length - arrayOffset < positionCount) - { - throw new IllegalArgumentException("valueIsNull is null or its length is less than positionCount"); - } - this.valueIsNull = valueIsNull; - - sizeInBytes = (Integer.BYTES + Byte.BYTES) * (long) positionCount; - retainedSizeInBytes = INSTANCE_SIZE + sizeOf(valueIsNull) + sizeOf(values); - } - - @Override - public OptionalInt fixedSizeInBytesPerPosition() - { - return OptionalInt.of(SIZE_IN_BYTES_PER_POSITION); - } - - @Override - public long getSizeInBytes() - { - return SIZE_IN_BYTES_PER_POSITION * (long) positionCount; - } - - @Override - public long getRegionSizeInBytes(int position, int length) - { - return SIZE_IN_BYTES_PER_POSITION * (long) length; - } - - @Override - public long getPositionsSizeInBytes(boolean[] positions, int selectedPositionsCount) - { - return (long) SIZE_IN_BYTES_PER_POSITION * selectedPositionsCount; - } - - @Override - public long getRetainedSizeInBytes() - { - return retainedSizeInBytes; - } - - /** - * Returns the estimated in memory data size for stats of position. - * Do not use it for other purpose. - * - * @param position - */ - @Override - public long getEstimatedDataSizeForStats(int position) - { - return isNull(position) ? 0 : Integer.BYTES; - } - - @Override - public void retainedBytesForEachPart(ObjLongConsumer consumer) - { - consumer.accept(values, sizeOf(values)); - consumer.accept(valueIsNull, sizeOf(valueIsNull)); - consumer.accept(this, INSTANCE_SIZE); - } - - @Override - public int getPositionCount() - { - return positionCount; - } - - public long getLong(int position) - { - checkReadablePosition(position); - return values[position + arrayOffset] * SCALE_FACTOR; - } - - public int getInt(int position) - { - checkReadablePosition(position); - return values[position + arrayOffset]; - } - - protected int[] getRawValues() - { - return this.values; - } - - protected int getRawValuesOffset() - { - return this.arrayOffset; - } - - @Override - public boolean isNull(int position) - { - checkReadablePosition(position); - return hasNull && valueIsNull[position + arrayOffset]; - } - - /** - * Returns a block that contains a copy of the contents of the current block, and an appended null at the end. The - * original block will not be modified. The purpose of this method is to leverage the contents of a block and the - * structure of the implementation to efficiently produce a copy of the block with a NULL element inserted - so that - * it can be used as a dictionary. This method is expected to be invoked on completely built {@link Block} instances - * i.e. not on in-progress block builders. - */ - @Override - public TimeArrayBlock copyWithAppendedNull() - { - boolean[] newValueIsNull = copyIsNullAndAppendNull(valueIsNull, arrayOffset, positionCount); - int[] newValues = ensureCapacity(values, arrayOffset + positionCount + 1); - - return new TimeArrayBlock(arrayOffset, positionCount + 1, newValues, true, newValueIsNull); - } - - @Override - public ValueBlock getUnderlyingValueBlock() - { - return this; - } - - @Override - public int getUnderlyingValuePosition(int position) - { - return position; - } - - @Override - public boolean mayHaveNull() - { - return hasNull; - } - - @Override - public Optional getNulls() - { - return BlockUtil.getNulls(valueIsNull, arrayOffset, positionCount); - } - - @Override - public Block getPositions(int[] positions, int offset, int length) - { - return ValueBlock.super.getPositions(positions, offset, length); - } - - @Override - public boolean isLoaded() - { - return true; - } - - @Override - public Block getLoadedBlock() - { - return this; - } - - @Override - public TimeArrayBlock getSingleValueBlock(int position) - { - checkReadablePosition(position); - return new TimeArrayBlock(1, - new int[] {values[position + arrayOffset]}, - hasNull && valueIsNull[position + arrayOffset], - new boolean[] {valueIsNull[position + arrayOffset]}); - } - - @Override - public TimeArrayBlock copyPositions(int[] positions, int offset, int length) - { - checkArrayRange(positions, offset, length); - - boolean[] newValueIsNull = new boolean[length]; - boolean newHasNull = false; - int[] newValues = new int[length]; - for (int i = 0; i < length; i++) - { - int position = positions[offset + i]; - checkReadablePosition(position); - if (hasNull && valueIsNull[position + arrayOffset]) - { - newValueIsNull[i] = true; - newHasNull = true; - } - else - { - newValues[i] = values[position + arrayOffset]; - } - } - return new TimeArrayBlock(length, newValues, newHasNull, newValueIsNull); - } - - @Override - public TimeArrayBlock getRegion(int positionOffset, int length) - { - checkValidRegion(getPositionCount(), positionOffset, length); - - boolean newHasNull = false; - if (hasNull) - { - for (int i = 0; i < length; ++i) - { - if (valueIsNull[i + arrayOffset]) - { - newHasNull = true; - break; - } - } - } - return new TimeArrayBlock(positionOffset + arrayOffset, length, values, newHasNull, valueIsNull); - } - - @Override - public TimeArrayBlock copyRegion(int positionOffset, int length) - { - checkValidRegion(getPositionCount(), positionOffset, length); - - positionOffset += arrayOffset; - boolean[] newValueIsNull = compactArray(valueIsNull, positionOffset, length); - int[] newValues = compactArray(values, positionOffset, length); - - if (newValueIsNull == valueIsNull && newValues == values) - { - return this; - } - - boolean newHasNull = false; - if (hasNull) - { - for (int i = 0; i < length; ++i) - { - if (newValueIsNull[i]) - { - newHasNull = true; - break; - } - } - } - return new TimeArrayBlock(length, newValues, newHasNull, newValueIsNull); - } - - @Override - public String getEncodingName() - { - return TimeArrayBlockEncoding.NAME; - } - - @Override - public String toString() - { - StringBuilder sb = new StringBuilder("TimeArrayBlock{"); - sb.append("positionCount=").append(getPositionCount()); - sb.append('}'); - return sb.toString(); - } - - private void checkReadablePosition(int position) - { - if (position < 0 || position >= getPositionCount()) - { - throw new IllegalArgumentException("position is not valid"); - } - } -} diff --git a/connector/src/main/java/io/pixelsdb/pixels/trino/block/TimeArrayBlockEncoding.java b/connector/src/main/java/io/pixelsdb/pixels/trino/block/TimeArrayBlockEncoding.java deleted file mode 100644 index 7da87f7..0000000 --- a/connector/src/main/java/io/pixelsdb/pixels/trino/block/TimeArrayBlockEncoding.java +++ /dev/null @@ -1,102 +0,0 @@ -/* - * Copyright 2022 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.trino.block; - -import io.trino.spi.block.Block; -import io.trino.spi.block.BlockEncoding; -import io.trino.spi.block.BlockEncodingSerde; -import io.airlift.slice.SliceInput; -import io.airlift.slice.SliceOutput; - -import static io.pixelsdb.pixels.trino.block.EncoderUtil.decodeNullBits; -import static io.pixelsdb.pixels.trino.block.EncoderUtil.encodeNullsAsBits; - -/** - * This class is derived from io.trino.spi.block.IntArrayBlockEncoding. - * - * @author hank - */ -public class TimeArrayBlockEncoding implements BlockEncoding -{ - public static final String NAME = "TIME_ARRAY"; - - private static final TimeArrayBlockEncoding instance = new TimeArrayBlockEncoding(); - - public static TimeArrayBlockEncoding Instance() - { - return instance; - } - - @Override - public String getName() - { - return NAME; - } - - @Override - public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceOutput, Block block) - { - // The down casts here are safe because it is the block itself the provides this encoding implementation. - TimeArrayBlock timeArrayBlock = (TimeArrayBlock) block; - - int positionCount = timeArrayBlock.getPositionCount(); - sliceOutput.appendInt(positionCount); - // hasNull - sliceOutput.appendByte(timeArrayBlock.mayHaveNull() ? 1 : 0); - - encodeNullsAsBits(sliceOutput, timeArrayBlock); - - if (!timeArrayBlock.mayHaveNull()) - { - sliceOutput.writeInts(timeArrayBlock.getRawValues(), - timeArrayBlock.getRawValuesOffset(), timeArrayBlock.getPositionCount()); - } - else - { - for (int position = 0; position < positionCount; position++) - { - if (!timeArrayBlock.isNull(position)) - { - sliceOutput.writeInt(timeArrayBlock.getInt(position)); - } - } - } - } - - @Override - public Block readBlock(BlockEncodingSerde blockEncodingSerde, SliceInput sliceInput) - { - int positionCount = sliceInput.readInt(); - boolean hasNull = sliceInput.readByte() != 0; - - boolean[] valueIsNull = decodeNullBits(sliceInput, positionCount).get(); - - int[] values = new int[positionCount]; - for (int position = 0; position < positionCount; position++) - { - if (!valueIsNull[position]) - { - values[position] = sliceInput.readInt(); - } - } - - return new TimeArrayBlock(positionCount, values, hasNull, valueIsNull); - } -} From 0604b52b739dd219d353342e0358ade9a3bf9cb7 Mon Sep 17 00:00:00 2001 From: Dongyang Geng Date: Tue, 4 Aug 2026 16:53:22 +0800 Subject: [PATCH 2/2] refactor: build VariableWidthBlock directly in PixelsBlockLoader --- .../pixels/trino/PixelsBlockLoader.java | 30 +- .../pixelsdb/pixels/trino/PixelsPlugin.java | 8 - .../pixels/trino/block/BlockUtil.java | 331 ----------- .../pixels/trino/block/EncoderUtil.java | 130 ----- .../pixels/trino/block/VarcharArrayBlock.java | 529 ------------------ .../block/VarcharArrayBlockEncoding.java | 176 ------ .../pixels/trino/TestPixelsBlockLoader.java | 283 ++++++++++ 7 files changed, 308 insertions(+), 1179 deletions(-) delete mode 100644 connector/src/main/java/io/pixelsdb/pixels/trino/block/BlockUtil.java delete mode 100644 connector/src/main/java/io/pixelsdb/pixels/trino/block/EncoderUtil.java delete mode 100644 connector/src/main/java/io/pixelsdb/pixels/trino/block/VarcharArrayBlock.java delete mode 100644 connector/src/main/java/io/pixelsdb/pixels/trino/block/VarcharArrayBlockEncoding.java create mode 100644 connector/src/test/java/io/pixelsdb/pixels/trino/TestPixelsBlockLoader.java diff --git a/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsBlockLoader.java b/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsBlockLoader.java index b375c18..a3c203b 100644 --- a/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsBlockLoader.java +++ b/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsBlockLoader.java @@ -22,8 +22,6 @@ import io.airlift.slice.Slices; import io.pixelsdb.pixels.core.TypeDescription; import io.pixelsdb.pixels.core.vector.*; -import io.pixelsdb.pixels.trino.block.VarcharArrayBlock; -import io.pixelsdb.pixels.trino.block.VarcharArrayBlockEncoding; import io.trino.spi.block.*; import io.trino.spi.type.Type; @@ -127,9 +125,31 @@ public Block load() case VARBINARY: if (vector instanceof BinaryColumnVector scv) { - block = VarcharArrayBlockEncoding.Instance() - .replacementBlockForWrite(new VarcharArrayBlock(batchSize, scv.vector, scv.start, scv.lens, !scv.noNulls, scv.isNull).getLoadedBlock()) - .get(); + int totalLength = 0; + for (int i = 0; i < batchSize; ++i) + { + if (!scv.noNulls && scv.isNull[i]) + { + continue; + } + totalLength += scv.lens[i]; + } + byte[] content = new byte[totalLength]; + int[] offsets = new int[batchSize + 1]; + int curOffset = 0; + for (int i = 0; i < batchSize; ++i) + { + offsets[i] = curOffset; + if (!scv.noNulls && scv.isNull[i]) + { + continue; + } + int len = scv.lens[i]; + System.arraycopy(scv.vector[i], scv.start[i], content, curOffset, len); + curOffset += len; + } + offsets[batchSize] = curOffset; + block = new VariableWidthBlock(batchSize, Slices.wrappedBuffer(content), offsets, Optional.of(scv.isNull)); } else { DictionaryColumnVector dscv = (DictionaryColumnVector) vector; diff --git a/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsPlugin.java b/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsPlugin.java index c78a972..a89ca4c 100644 --- a/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsPlugin.java +++ b/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsPlugin.java @@ -20,19 +20,11 @@ package io.pixelsdb.pixels.trino; import com.google.common.collect.ImmutableList; -import io.pixelsdb.pixels.trino.block.VarcharArrayBlockEncoding; import io.trino.spi.Plugin; -import io.trino.spi.block.BlockEncoding; import io.trino.spi.connector.ConnectorFactory; public class PixelsPlugin implements Plugin { - @Override - public Iterable getBlockEncodings() - { - return ImmutableList.of(VarcharArrayBlockEncoding.Instance()); - } - @Override public Iterable getConnectorFactories() { diff --git a/connector/src/main/java/io/pixelsdb/pixels/trino/block/BlockUtil.java b/connector/src/main/java/io/pixelsdb/pixels/trino/block/BlockUtil.java deleted file mode 100644 index 2d96372..0000000 --- a/connector/src/main/java/io/pixelsdb/pixels/trino/block/BlockUtil.java +++ /dev/null @@ -1,331 +0,0 @@ -/* - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.pixelsdb.pixels.trino.block; - -import io.airlift.slice.Slice; -import io.trino.spi.block.Block; -import io.trino.spi.block.ByteArrayBlock; - -import javax.annotation.Nullable; -import java.util.Arrays; -import java.util.Optional; - -import static java.lang.Math.ceil; -import static java.lang.String.format; -import static java.util.Objects.requireNonNull; - -/** - * This class is copied from io.trino.spi.block.BlockUtil, because it's not public. - */ -final class BlockUtil -{ - private static final double BLOCK_RESET_SKEW = 1.25; - - private static final int DEFAULT_CAPACITY = 64; - // See java.util.ArrayList for an explanation - static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; - - private BlockUtil() - { - } - - static void checkArrayRange(int[] array, int offset, int length) - { - requireNonNull(array, "array is null"); - if (offset < 0 || length < 0 || offset + length > array.length) { - throw new IndexOutOfBoundsException(format("Invalid offset %s and length %s in array with %s elements", - offset, length, array.length)); - } - } - - static void checkArrayRange(boolean[] array, int offset, int length) - { - requireNonNull(array, "array is null"); - if (offset < 0 || length < 0 || offset + length > array.length) { - throw new IndexOutOfBoundsException(format("Invalid offset %s and length %s in array with %s elements", - offset, length, array.length)); - } - } - - static void checkValidRegion(int positionCount, int positionOffset, int length) - { - if (positionOffset < 0 || length < 0 || positionOffset + length > positionCount) { - throw new IndexOutOfBoundsException(format("Invalid position %s and length %s in block with %s positions", - positionOffset, length, positionCount)); - } - } - - static void checkValidPositions(boolean[] positions, int positionCount) - { - if (positions.length != positionCount) { - throw new IllegalArgumentException(format("Invalid positions array size %d, actual position count is %d", - positions.length, positionCount)); - } - } - - static void checkValidPosition(int position, int positionCount) - { - if (position < 0 || position >= positionCount) { - throw new IllegalArgumentException(format("Invalid position %s in block with %s positions", - position, positionCount)); - } - } - - static int calculateNewArraySize(int currentSize) - { - // grow array by 50% - long newSize = (long) currentSize + (currentSize >> 1); - - // verify new size is within reasonable bounds - if (newSize < DEFAULT_CAPACITY) { - newSize = DEFAULT_CAPACITY; - } - else if (newSize > MAX_ARRAY_SIZE) { - newSize = MAX_ARRAY_SIZE; - if (newSize == currentSize) { - throw new IllegalArgumentException(format("Cannot grow array beyond '%s'", MAX_ARRAY_SIZE)); - } - } - return (int) newSize; - } - - static int calculateBlockResetSize(int currentSize) - { - long newSize = (long) ceil(currentSize * BLOCK_RESET_SKEW); - - // verify new size is within reasonable bounds - if (newSize < DEFAULT_CAPACITY) { - newSize = DEFAULT_CAPACITY; - } - else if (newSize > MAX_ARRAY_SIZE) { - newSize = MAX_ARRAY_SIZE; - } - return (int) newSize; - } - - static int calculateBlockResetBytes(int currentBytes) - { - long newBytes = (long) ceil(currentBytes * BLOCK_RESET_SKEW); - if (newBytes > MAX_ARRAY_SIZE) { - return MAX_ARRAY_SIZE; - } - return (int) newBytes; - } - - /** - * Recalculate the offsets array for the specified range. - * The returned offsets array contains length + 1 integers - * with the first value set to 0. - * If the range matches the entire offsets array, the input array will be returned. - */ - static int[] compactOffsets(int[] offsets, int index, int length) - { - if (index == 0 && offsets.length == length + 1) { - return offsets; - } - - int[] newOffsets = new int[length + 1]; - for (int i = 1; i <= length; i++) { - newOffsets[i] = offsets[index + i] - offsets[index]; - } - return newOffsets; - } - - /** - * Returns a slice containing values in the specified range of the specified slice. - * If the range matches the entire slice, the input slice will be returned. - * Otherwise, a copy will be returned. - */ - static Slice compactSlice(Slice slice, int index, int length) - { - if (slice.isCompact() && index == 0 && length == slice.length()) { - return slice; - } - return slice.copy(index, length); - } - - /** - * Returns an array containing elements in the specified range of the specified array. - * If the range matches the entire array, the input array will be returned. - * Otherwise, a copy will be returned. - */ - static boolean[] compactArray(boolean[] array, int index, int length) - { - if (index == 0 && length == array.length) { - return array; - } - return Arrays.copyOfRange(array, index, index + length); - } - - static byte[] compactArray(byte[] array, int index, int length) - { - if (index == 0 && length == array.length) { - return array; - } - return Arrays.copyOfRange(array, index, index + length); - } - - static short[] compactArray(short[] array, int index, int length) - { - if (index == 0 && length == array.length) { - return array; - } - return Arrays.copyOfRange(array, index, index + length); - } - - static int[] compactArray(int[] array, int index, int length) - { - if (index == 0 && length == array.length) { - return array; - } - return Arrays.copyOfRange(array, index, index + length); - } - - static long[] compactArray(long[] array, int index, int length) - { - if (index == 0 && length == array.length) { - return array; - } - return Arrays.copyOfRange(array, index, index + length); - } - - static int countSelectedPositionsFromOffsets(boolean[] positions, int[] offsets, int offsetBase) - { - checkArrayRange(offsets, offsetBase, positions.length); - int used = 0; - for (int i = 0; i < positions.length; i++) { - int offsetStart = offsets[offsetBase + i]; - int offsetEnd = offsets[offsetBase + i + 1]; - used += ((positions[i] ? 1 : 0) * (offsetEnd - offsetStart)); - } - return used; - } - - static int countAndMarkSelectedPositionsFromOffsets(boolean[] positions, int[] offsets, - int offsetBase, boolean[] elementPositions) - { - checkArrayRange(offsets, offsetBase, positions.length); - int used = 0; - for (int i = 0; i < positions.length; i++) { - int offsetStart = offsets[offsetBase + i]; - int offsetEnd = offsets[offsetBase + i + 1]; - if (positions[i]) { - used += (offsetEnd - offsetStart); - Arrays.fill(elementPositions, offsetStart, offsetEnd, true); - } - } - return used; - } - - /** - * Returns true if the two specified arrays contain the same object in every position. - * Unlike the {@link Arrays#equals(Object[], Object[])} method, this method compares using reference equals. - */ - static boolean arraySame(Object[] array1, Object[] array2) - { - if (array1 == null || array2 == null || array1.length != array2.length) { - throw new IllegalArgumentException("array1 and array2 cannot be null and should have same length"); - } - - for (int i = 0; i < array1.length; i++) { - if (array1[i] != array2[i]) { - return false; - } - } - return true; - } - - /** - * Returns the input blocks array if all blocks are already loaded, otherwise returns - * a new blocks array with all blocks loaded - */ - static Block[] ensureBlocksAreLoaded(Block[] blocks) - { - for (int i = 0; i < blocks.length; i++) { - Block loaded = blocks[i].getLoadedBlock(); - if (loaded != blocks[i]) { - // Transition to new block creation mode after the first newly loaded block is encountered - Block[] loadedBlocks = blocks.clone(); - loadedBlocks[i++] = loaded; - for (; i < blocks.length; i++) { - loadedBlocks[i] = blocks[i].getLoadedBlock(); - } - return loadedBlocks; - } - } - // No newly loaded blocks - return blocks; - } - - static boolean[] copyIsNullAndAppendNull(@Nullable boolean[] isNull, int offsetBase, int positionCount) - { - int desiredLength = offsetBase + positionCount + 1; - boolean[] newIsNull = new boolean[desiredLength]; - if (isNull != null) { - checkArrayRange(isNull, offsetBase, positionCount); - System.arraycopy(isNull, 0, newIsNull, 0, desiredLength - 1); - } - // mark the last element to append null - newIsNull[desiredLength - 1] = true; - return newIsNull; - } - - static int[] copyOffsetsAndAppendNull(int[] offsets, int offsetBase, int positionCount) - { - int desiredLength = offsetBase + positionCount + 1; - checkArrayRange(offsets, offsetBase, positionCount); - int[] newOffsets = Arrays.copyOf(offsets, desiredLength); - // Null element does not move the offset forward - newOffsets[desiredLength - 1] = newOffsets[desiredLength - 2]; - return newOffsets; - } - - /** - * Returns a new int array of size capacity if the input buffer is null or - * smaller than the capacity. Returns the original array otherwise. - * Any original values in the input buffer will be preserved in the output. - */ - public static int[] ensureCapacity(@Nullable int[] buffer, int capacity) - { - if (buffer == null) { - buffer = new int[capacity]; - } - else if (buffer.length < capacity) { - buffer = Arrays.copyOf(buffer, capacity); - } - - return buffer; - } - - /** - * Ideally, the underlying nulls array in Block implementations should be a byte array instead of a boolean array. - * This method is used to perform that conversion until the Block implementations are changed. - */ - static Optional getNulls(@Nullable boolean[] valueIsNull, int arrayOffset, int positionCount) - { - if (valueIsNull == null) { - return Optional.empty(); - } - byte[] booleansAsBytes = new byte[positionCount]; - boolean foundAnyNull = false; - for (int i = 0; i < positionCount; i++) { - booleansAsBytes[i] = (byte) (valueIsNull[arrayOffset + i] ? 1 : 0); - foundAnyNull = foundAnyNull || valueIsNull[arrayOffset + i]; - } - if (!foundAnyNull) { - return Optional.empty(); - } - return Optional.of(new ByteArrayBlock(booleansAsBytes.length, Optional.empty(), booleansAsBytes)); - } -} diff --git a/connector/src/main/java/io/pixelsdb/pixels/trino/block/EncoderUtil.java b/connector/src/main/java/io/pixelsdb/pixels/trino/block/EncoderUtil.java deleted file mode 100644 index 4d7e5f4..0000000 --- a/connector/src/main/java/io/pixelsdb/pixels/trino/block/EncoderUtil.java +++ /dev/null @@ -1,130 +0,0 @@ -/* - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.pixelsdb.pixels.trino.block; - -import io.airlift.slice.SliceInput; -import io.airlift.slice.SliceOutput; -import io.trino.spi.block.Block; - -import javax.annotation.Nullable; -import java.io.IOException; -import java.io.UncheckedIOException; -import java.util.Optional; - -/** - * This class is copied from io.trino.spi.block.EncoderUtil, because it's not public. - */ -final class EncoderUtil -{ - private EncoderUtil() - { - } - - /** - * Append null values for the block as a stream of bits. - */ - @SuppressWarnings({"NarrowingCompoundAssignment", "ImplicitNumericConversion"}) - public static void encodeNullsAsBits(SliceOutput sliceOutput, Block block) - { - boolean mayHaveNull = block.mayHaveNull(); - sliceOutput.writeBoolean(mayHaveNull); - if (!mayHaveNull) { - return; - } - - int positionCount = block.getPositionCount(); - byte[] packedIsNull = new byte[((positionCount & ~0b111) + 1) / 8]; - int currentByte = 0; - - for (int position = 0; position < (positionCount & ~0b111); position += 8, currentByte++) { - byte value = 0; - value |= block.isNull(position) ? 0b1000_0000 : 0; - value |= block.isNull(position + 1) ? 0b0100_0000 : 0; - value |= block.isNull(position + 2) ? 0b0010_0000 : 0; - value |= block.isNull(position + 3) ? 0b0001_0000 : 0; - value |= block.isNull(position + 4) ? 0b0000_1000 : 0; - value |= block.isNull(position + 5) ? 0b0000_0100 : 0; - value |= block.isNull(position + 6) ? 0b0000_0010 : 0; - value |= block.isNull(position + 7) ? 0b0000_0001 : 0; - packedIsNull[currentByte] = value; - } - - sliceOutput.writeBytes(packedIsNull); - - // write last null bits - if ((positionCount & 0b111) > 0) { - byte value = 0; - int mask = 0b1000_0000; - for (int position = positionCount & ~0b111; position < positionCount; position++) { - value |= block.isNull(position) ? mask : 0; - mask >>>= 1; - } - sliceOutput.appendByte(value); - } - } - - /** - * Decode the bit stream created by encodeNullsAsBits. - */ - public static Optional decodeNullBits(SliceInput sliceInput, int positionCount) - { - return Optional.ofNullable(retrieveNullBits(sliceInput, positionCount)) - .map(packedIsNull -> decodeNullBits(packedIsNull, positionCount)); - } - - public static boolean[] decodeNullBits(byte[] packedIsNull, int positionCount) - { - // read null bits 8 at a time - boolean[] valueIsNull = new boolean[positionCount]; - int currentByte = 0; - for (int position = 0; position < (positionCount & ~0b111); position += 8, currentByte++) { - byte value = packedIsNull[currentByte]; - valueIsNull[position] = ((value & 0b1000_0000) != 0); - valueIsNull[position + 1] = ((value & 0b0100_0000) != 0); - valueIsNull[position + 2] = ((value & 0b0010_0000) != 0); - valueIsNull[position + 3] = ((value & 0b0001_0000) != 0); - valueIsNull[position + 4] = ((value & 0b0000_1000) != 0); - valueIsNull[position + 5] = ((value & 0b0000_0100) != 0); - valueIsNull[position + 6] = ((value & 0b0000_0010) != 0); - valueIsNull[position + 7] = ((value & 0b0000_0001) != 0); - } - - // read last null bits - if ((positionCount & 0b111) > 0) { - byte value = packedIsNull[packedIsNull.length - 1]; - int mask = 0b1000_0000; - for (int position = positionCount & ~0b111; position < positionCount; position++) { - valueIsNull[position] = ((value & mask) != 0); - mask >>>= 1; - } - } - - return valueIsNull; - } - - @Nullable - public static byte[] retrieveNullBits(SliceInput sliceInput, int positionCount) - { - if (!sliceInput.readBoolean()) { - return null; - } - try { - return sliceInput.readNBytes((positionCount + 7) / 8); - } - catch (IOException e) { - throw new UncheckedIOException(e); - } - } -} - diff --git a/connector/src/main/java/io/pixelsdb/pixels/trino/block/VarcharArrayBlock.java b/connector/src/main/java/io/pixelsdb/pixels/trino/block/VarcharArrayBlock.java deleted file mode 100644 index 36efe78..0000000 --- a/connector/src/main/java/io/pixelsdb/pixels/trino/block/VarcharArrayBlock.java +++ /dev/null @@ -1,529 +0,0 @@ -/* - * Copyright 2022 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.trino.block; - -import io.airlift.slice.Slice; -import io.airlift.slice.Slices; -import io.trino.spi.block.Block; -import io.trino.spi.block.ByteArrayBlock; -import io.trino.spi.block.ValueBlock; -import io.trino.spi.block.VariableWidthBlock; -import org.openjdk.jol.info.ClassLayout; - -import java.util.*; -import java.util.function.ObjLongConsumer; - -import static io.airlift.slice.SizeOf.sizeOf; -import static io.pixelsdb.pixels.trino.block.BlockUtil.copyIsNullAndAppendNull; -import static io.pixelsdb.pixels.trino.block.BlockUtil.copyOffsetsAndAppendNull; - -/** - * This class is derived from io.trino.spi.block.VariableWidthBlock and AbstractVariableWidthBlock. - *

- * Our main modifications: - * 1. we use a byte[][] instead of Slice as the backing storage - * and replaced the implementation of each method; - * 2. add some other methods. - *

- * - * @author hank - * @create 2019-05 - * @update 2024-12-01 adapt to with Trino 465 and add hasNull argument to the constructor. - */ -public class VarcharArrayBlock implements ValueBlock -{ - private static final long INSTANCE_SIZE = ClassLayout.parseClass(VarcharArrayBlock.class).instanceSize(); - - private final int arrayOffset; // start index of the valid items in offsets and length, usually 0. - private final int positionCount; // number of items in this block. - private final byte[][] values; // values of the items. - private final int[] offsets; // start byte offset of the item in each value, \ - // always 0 if this block is deserialized by VarcharArrayBlockEncoding.readBlock. - private final int[] lengths; // byte length of each item. - private final boolean[] valueIsNull; // isNull flag of each item. - private final boolean hasNull; - - private final long retainedSizeInBytes; - /** - * PIXELS-167: - * The actual memory footprint of the member values. - */ - private final long retainedSizeOfValues; - private final long sizeInBytes; - private Optional alternativeBlock; - public VarcharArrayBlock(int positionCount, byte[][] values, int[] offsets, int[] lengths, boolean hasNull, boolean[] valueIsNull) - { - this(0, positionCount, values, offsets, lengths, hasNull, valueIsNull); - } - - VarcharArrayBlock(int arrayOffset, int positionCount, byte[][] values, int[] offsets, int[] lengths, boolean hasNull, boolean[] valueIsNull) - { - if (arrayOffset < 0) - { - throw new IllegalArgumentException("arrayOffset is negative"); - } - this.arrayOffset = arrayOffset; - if (positionCount < 0) - { - throw new IllegalArgumentException("positionCount is negative"); - } - this.positionCount = positionCount; - - if (values == null || values.length - arrayOffset < (positionCount)) - { - throw new IllegalArgumentException("values is null or its length is less than positionCount"); - } - this.values = values; - - if (offsets == null || offsets.length - arrayOffset < (positionCount)) - { - throw new IllegalArgumentException("offsets is null or its length is less than positionCount"); - } - this.offsets = offsets; - - if (lengths == null || lengths.length - arrayOffset < (positionCount)) - { - throw new IllegalArgumentException("lengths is null or its length is less than positionCount"); - } - this.lengths = lengths; - - this.hasNull = hasNull; - // Issue #123: in Pixels, the isNull bitmap from column vectors always presents even if there is no nulls. - if (valueIsNull == null || valueIsNull.length - arrayOffset < positionCount) - { - throw new IllegalArgumentException("valueIsNull is null or its length is less than positionCount"); - } - this.valueIsNull = valueIsNull; - - long size = 0L, retainedSize = 0L; - Set existingValues = new HashSet<>(2); - for (int i = 0; i < positionCount; ++i) - { - size += lengths[arrayOffset + i]; - // retainedSize should count the physical footprint of the values. - if (!valueIsNull[arrayOffset + i]) - { - if (!existingValues.contains(values[arrayOffset + i])) - { - existingValues.add(values[arrayOffset + i]); - retainedSize += values[arrayOffset + i].length; - } - } - } - existingValues.clear(); - sizeInBytes = size; - retainedSizeOfValues = retainedSize + sizeOf(values); - retainedSizeInBytes = INSTANCE_SIZE + retainedSizeOfValues + - sizeOf(valueIsNull) + sizeOf(offsets) + sizeOf(lengths); - } - - /** - * Gets the start offset of the value at the {@code position}. - */ - protected final int getPositionOffset(int position) - { - /** - * PIXELS-132: - * FIX: null must be checked here as offsets (i.e. starts) in column vector - * may be reused in vectorized row batch and is not reset. - */ - if (hasNull && valueIsNull[position + arrayOffset]) - { - return 0; - } - return offsets[position + arrayOffset]; - } - - /** - * Gets the length of the value at the {@code position}. - * This method must be implemented if @{code getSlice} is implemented. - */ - protected int getSliceLength(int position) - { - checkReadablePosition(position); - /** - * PIXELS-132: - * FIX: null must be checked here as lengths (i.e. lens) in column vector - * may be reused in vectorized row batch and is not reset. - */ - if (hasNull && valueIsNull[position + arrayOffset]) - { - return 0; - } - return lengths[position + arrayOffset]; - } - - @Override - public int getPositionCount() - { - return positionCount; - } - - @Override - public long getSizeInBytes() - { - return sizeInBytes; - } - - /** - * Returns the logical size of {@code block.getRegion(position, length)} in memory. - * The method can be expensive. Do not use it outside an implementation of Block. - */ - @Override - public long getRegionSizeInBytes(int position, int length) - { - BlockUtil.checkValidRegion(getPositionCount(), position, length); - long size = 0L; - for (int i = 0; i < length; ++i) - { - // lengths[i] is zero if valueIsNull[i] is true, no need to check. - size += lengths[position + arrayOffset + i]; - } - return size + ((Integer.BYTES * 2 + Byte.BYTES) * (long) length); - } - - @Override - public OptionalInt fixedSizeInBytesPerPosition() - { - return OptionalInt.empty(); // size varies per element and is not fixed - } - - /** - * Returns the size of all positions marked true in the positions array. - * This is equivalent to multiple calls of {@code block.getRegionSizeInBytes(position, length)} - * where you mark all positions for the regions first. - * - * @param positions - */ - @Override - public long getPositionsSizeInBytes(boolean[] positions, int _selectedPositionsCount) - { - long sizeInBytes = 0; - int usedPositionCount = 0; - for (int i = 0; i < positions.length; ++i) - { - if (positions[i]) - { - usedPositionCount++; - sizeInBytes += lengths[arrayOffset + i]; - } - } - return sizeInBytes + (Integer.BYTES * 2 + Byte.BYTES) * (long) usedPositionCount; - } - - /** - * Returns the retained size of this block in memory. - * This method is called from the innermost execution loop and must be fast. - */ - @Override - public long getRetainedSizeInBytes() - { - return retainedSizeInBytes; - } - - /** - * Returns the estimated in memory data size for stats of position. - * Do not use it for other purpose. - * - * @param position - */ - @Override - public long getEstimatedDataSizeForStats(int position) - { - return isNull(position) ? 0 : getSliceLength(position); - } - - /** - * {@code consumer} visits each of the internal data container and accepts the size for it. - * This method can be helpful in cases such as memory counting for internal data structure. - * Also, the method should be non-recursive, only visit the elements at the top level, - * and specifically should not call retainedBytesForEachPart on nested blocks - * {@code consumer} should be called at least once with the current block and - * must include the instance size of the current block - */ - @Override - public void retainedBytesForEachPart(ObjLongConsumer consumer) - { - /** - * PIXELS-167: - * DO NOT calculate the retained size of values by adding up values[i].length. - */ - consumer.accept(values, retainedSizeOfValues); - consumer.accept(offsets, sizeOf(offsets)); - consumer.accept(lengths, sizeOf(lengths)); - consumer.accept(valueIsNull, sizeOf(valueIsNull)); - consumer.accept(this, INSTANCE_SIZE); - } - - /** - * Returns a block containing the specified positions. - * Positions to copy are stored in a subarray within {@code positions} array - * that starts at {@code offset} and has length of {@code length}. - * All specified positions must be valid for this block. - *

- * The returned block must be a compact representation of the original block. - */ - @Override - public VarcharArrayBlock copyPositions(int[] positions, int offset, int length) - { - BlockUtil.checkArrayRange(positions, offset, length); - byte[][] newValues = new byte[length][]; - int[] newStarts = new int[length]; - int[] newLengths = new int[length]; - boolean newHasNull = false; - boolean[] newValueIsNull = new boolean[length]; - - for (int i = 0; i < length; i++) - { - int position = positions[offset + i]; - if (hasNull && valueIsNull[position + arrayOffset]) - { - newValueIsNull[i] = true; - newHasNull = true; - } else - { - // we only copy the valid part of each value. - int from = offsets[position + arrayOffset]; - newLengths[i] = lengths[position + arrayOffset]; - newValues[i] = Arrays.copyOfRange(values[position + arrayOffset], - from, from + newLengths[i]); - // newStarts is 0. - } - } - return new VarcharArrayBlock(length, newValues, newStarts, newLengths, newHasNull, newValueIsNull); - } - - protected Slice getRawSlice(int position) - { - // DO NOT specify the offset and length for wrappedBuffer, - // a raw slice should contain the whole bytes of value at the position. - if (hasNull && valueIsNull[position + arrayOffset]) - { - return Slices.EMPTY_SLICE; - } - return Slices.wrappedBuffer(values[position + arrayOffset]); - } - - protected byte[] getRawValue(int position) - { - if (hasNull && valueIsNull[position + arrayOffset]) - { - return null; - } - return values[position + arrayOffset]; - } - - /** - * Returns a block starting at the specified position and extends for the - * specified length. The specified region must be entirely contained - * within this block. - *

- * The region can be a view over this block. If this block is released - * the region block may also be released. If the region block is released - * this block may also be released. - */ - @Override - public VarcharArrayBlock getRegion(int positionOffset, int length) - { - BlockUtil.checkValidRegion(getPositionCount(), positionOffset, length); - - boolean newHasNull = false; - if (hasNull) - { - for (int i = 0; i < length; ++i) - { - if (valueIsNull[i + arrayOffset]) - { - newHasNull = true; - break; - } - } - } - return new VarcharArrayBlock(positionOffset + arrayOffset, length, values, offsets, lengths, newHasNull, valueIsNull); - } - - /** - * Gets the value at the specified position as a single element block. The method - * must copy the data into a new block. - *

- * This method is useful for operators that hold on to a single value without - * holding on to the entire block. - * - * @throws IllegalArgumentException if this position is not valid - */ - @Override - public VarcharArrayBlock getSingleValueBlock(int position) - { - checkReadablePosition(position); - byte[][] copy = new byte[1][]; - if (isNull(position)) - { - return new VarcharArrayBlock(1, copy, new int[]{0}, new int[]{0}, true, new boolean[]{true}); - } - - int offset = offsets[position + arrayOffset]; - int entrySize = lengths[position + arrayOffset]; - copy[0] = Arrays.copyOfRange(values[position + arrayOffset], - offset, offset + entrySize); - - return new VarcharArrayBlock(1, copy, new int[]{0}, new int[]{entrySize}, false, new boolean[]{false}); - } - - /** - * Returns a block starting at the specified position and extends for the - * specified length. The specified region must be entirely contained - * within this block. - *

- * The region returned must be a compact representation of the original block, unless their internal - * representation will be exactly the same. This method is useful for - * operators that hold on to a range of values without holding on to the - * entire block. - */ - @Override - public VarcharArrayBlock copyRegion(int positionOffset, int length) - { - BlockUtil.checkValidRegion(getPositionCount(), positionOffset, length); - positionOffset += arrayOffset; - - byte[][] newValues = new byte[length][]; - int[] newStarts = new int[length]; - int[] newLengths = new int[length]; - boolean newHasNull = false; - boolean[] newValueIsNull = new boolean[length]; - - for (int i = 0; i < length; i++) - { - if (hasNull && valueIsNull[positionOffset + i]) - { - newValueIsNull[i] = true; - newHasNull = true; - } else - { - // we only copy the valid part of each value. - newLengths[i] = lengths[positionOffset + i]; - newValues[i] = Arrays.copyOfRange(values[positionOffset + i], - offsets[positionOffset + i], offsets[positionOffset + i] + newLengths[i]); - // newStarts is 0. - } - } - return new VarcharArrayBlock(length, newValues, newStarts, newLengths, newHasNull, newValueIsNull); - } - - @Override - public String getEncodingName() - { - return VarcharArrayBlockEncoding.NAME; - } - - @Override - public boolean isNull(int position) - { - checkReadablePosition(position); - return hasNull && valueIsNull[position + arrayOffset]; - } - - /** - * Returns a block that contains a copy of the contents of the current block, and an appended null at the end. The - * original block will not be modified. The purpose of this method is to leverage the contents of a block and the - * structure of the implementation to efficiently produce a copy of the block with a NULL element inserted - so that - * it can be used as a dictionary. This method is expected to be invoked on completely built {@link Block} instances - * i.e. not on in-progress block builders. - */ - @Override - public VarcharArrayBlock copyWithAppendedNull() - { - boolean[] newValueIsNull = copyIsNullAndAppendNull(valueIsNull, arrayOffset, positionCount); - int[] newOffsets = copyOffsetsAndAppendNull(offsets, arrayOffset, positionCount); - int[] newLengths = copyOffsetsAndAppendNull(lengths, arrayOffset, positionCount); - - return new VarcharArrayBlock(arrayOffset, positionCount + 1, values, newOffsets, newLengths, true, newValueIsNull); - } - - @Override - public ValueBlock getUnderlyingValueBlock() - { - //TODO better way to adapt block APPEND operation? - if(alternativeBlock == null) - { - alternativeBlock = VarcharArrayBlockEncoding.Instance().replacementBlockForWrite(this); - } - if(alternativeBlock.isPresent()) - { - return alternativeBlock.get().getUnderlyingValueBlock(); - } - return this; - } - - @Override - public int getUnderlyingValuePosition(int position) - { - return position; - } - - @Override - public Optional getNulls() - { - return BlockUtil.getNulls(valueIsNull, arrayOffset, positionCount); - } - - @Override - public boolean mayHaveNull() - { - return this.hasNull; - } - - @Override - public Block getPositions(int[] positions, int offset, int length) - { - return ValueBlock.super.getPositions(positions, offset, length); - } - - @Override - public boolean isLoaded() - { - return true; - } - - @Override - public Block getLoadedBlock() - { - return this; - } - - protected void checkReadablePosition(int position) - { - BlockUtil.checkValidPosition(position, getPositionCount()); - } - - @Override - public String toString() - { - String sb = "VarcharArrayBlock{" + "positionCount=" + getPositionCount() + - ", size=" + sizeInBytes + - ", retainedSize=" + retainedSizeInBytes + - '}'; - return sb; - } - - public boolean[] getValueIsNull() - { - return valueIsNull; - } -} diff --git a/connector/src/main/java/io/pixelsdb/pixels/trino/block/VarcharArrayBlockEncoding.java b/connector/src/main/java/io/pixelsdb/pixels/trino/block/VarcharArrayBlockEncoding.java deleted file mode 100644 index ed04e77..0000000 --- a/connector/src/main/java/io/pixelsdb/pixels/trino/block/VarcharArrayBlockEncoding.java +++ /dev/null @@ -1,176 +0,0 @@ -/* - * Copyright 2022 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.trino.block; - -import io.airlift.slice.SliceInput; -import io.airlift.slice.SliceOutput; -import io.airlift.slice.Slices; -import io.trino.spi.block.Block; -import io.trino.spi.block.BlockEncoding; -import io.trino.spi.block.BlockEncodingSerde; -import io.trino.spi.block.VariableWidthBlock; - -import java.util.Optional; - -import static io.pixelsdb.pixels.trino.block.EncoderUtil.decodeNullBits; -import static io.pixelsdb.pixels.trino.block.EncoderUtil.encodeNullsAsBits; - -/** - * This class is derived from io.trino.spi.block.VariableWidthBlockEncoding - *

- * We reimplemented writeBlock and readBlock - * - * @author hank - */ -public class VarcharArrayBlockEncoding implements BlockEncoding -{ - public static final String NAME = "VARCHAR_ARRAY"; - - private static final VarcharArrayBlockEncoding instance = new VarcharArrayBlockEncoding(); - - public static VarcharArrayBlockEncoding Instance() - { - return instance; - } - - @Override - public String getName() - { - return NAME; - } - - @Override - public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceOutput, Block block) - { - // The down casts here are safe because it is the block itself the provides this encoding implementation. - VarcharArrayBlock varcharArrayBlock = (VarcharArrayBlock) block; - - int positionCount = varcharArrayBlock.getPositionCount(); - sliceOutput.appendInt(positionCount); - // hasNull - sliceOutput.appendByte(varcharArrayBlock.mayHaveNull() ? 1 : 0); - - // do not encode offsets, they should be 0. - - // lengths - for (int position = 0; position < positionCount; position++) - { - sliceOutput.appendInt(varcharArrayBlock.getSliceLength(position)); - } - - // isNull - encodeNullsAsBits(sliceOutput, varcharArrayBlock); - - // values - // sliceOutput.appendInt((int) varcharArrayBlock.getSizeInBytes()); - for (int position = 0; position < positionCount; position++) - { - byte[] rawValue = varcharArrayBlock.getRawValue(position); - if (rawValue != null) - { - sliceOutput.writeBytes(rawValue, varcharArrayBlock.getPositionOffset(position), - varcharArrayBlock.getSliceLength(position)); - } - } - } - - @Override - public Block readBlock(BlockEncodingSerde blockEncodingSerde, SliceInput sliceInput) - { - int positionCount = sliceInput.readInt(); - boolean hasNull = sliceInput.readByte() != 0; - int[] offsets = new int[positionCount]; - int[] lengths = new int[positionCount]; - - // offsets should be 0, do not read them from sliceInput. - - // destinationIndex should be 0, because we do not need 0 to be the first item in lengths. - sliceInput.readInts(lengths, 0, positionCount); - - boolean[] valueIsNull = decodeNullBits(sliceInput, positionCount).orElse(new boolean[positionCount]); - - // int blockSize = sliceInput.readInt(); - byte[][] values = new byte[positionCount][]; - for (int position = 0; position < positionCount; position++) - { - values[position] = new byte[lengths[position]]; - sliceInput.readBytes(values[position]); - } - - return new VarcharArrayBlock(positionCount, values, offsets, lengths, hasNull, valueIsNull); - } - - /** - * PIXELS-902: Trino expects standard block types like {@code VariableWidthBlock} for - * Varchar values when serializing query results. However, {@code VarcharArrayBlock} - * causes {@link ClassCastException} in paths such as {@code JsonEncodingUtils}. This method provides - * a fallback replacement to ensure compatibility by converting the {@code VarcharArrayBlock} into a supported format. - * - * @param block - * @return - */ - @Override - public Optional replacementBlockForWrite(Block block) - { - if (!(block instanceof VarcharArrayBlock varcharBlock)) - { - return Optional.empty(); - } - - int positionCount = varcharBlock.getPositionCount(); - - int totalLength = 0; - for (int i = 0; i < positionCount; ++i) - { - totalLength += varcharBlock.getSliceLength(i); - } - - byte[] content = new byte[totalLength]; - int[] offsets = new int[positionCount + 1]; - int curOffset = 0; - - for (int i = 0; i < positionCount; ++i) - { - offsets[i] = curOffset; - int len = varcharBlock.getSliceLength(i); - if (!varcharBlock.isNull(i)) - { - System.arraycopy( - varcharBlock.getRawValue(i), - varcharBlock.getPositionOffset(i), - content, - curOffset, - len - ); - } - curOffset += len; - } - offsets[positionCount] = totalLength; - - VariableWidthBlock newBlock = new VariableWidthBlock( - positionCount, - Slices.wrappedBuffer(content), - offsets, - Optional.of(varcharBlock.getValueIsNull()) - ); - - return Optional.of(newBlock); - } -} diff --git a/connector/src/test/java/io/pixelsdb/pixels/trino/TestPixelsBlockLoader.java b/connector/src/test/java/io/pixelsdb/pixels/trino/TestPixelsBlockLoader.java new file mode 100644 index 0000000..a788dbc --- /dev/null +++ b/connector/src/test/java/io/pixelsdb/pixels/trino/TestPixelsBlockLoader.java @@ -0,0 +1,283 @@ +/* + * 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.trino; + +import io.pixelsdb.pixels.core.TypeDescription; +import io.pixelsdb.pixels.core.vector.BinaryColumnVector; +import io.pixelsdb.pixels.core.vector.ByteColumnVector; +import io.pixelsdb.pixels.core.vector.ColumnVector; +import io.pixelsdb.pixels.core.vector.IntColumnVector; +import io.pixelsdb.pixels.core.vector.TimeColumnVector; +import io.trino.spi.Page; +import io.trino.spi.block.Block; +import io.trino.spi.block.ByteArrayBlock; +import io.trino.spi.block.LongArrayBlock; +import io.trino.spi.block.ShortArrayBlock; +import io.trino.spi.block.VariableWidthBlock; +import io.trino.spi.type.SmallintType; +import io.trino.spi.type.TimeType; +import io.trino.spi.type.TinyintType; +import io.trino.spi.type.Type; +import io.trino.spi.type.VarbinaryType; +import io.trino.spi.type.VarcharType; +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class TestPixelsBlockLoader +{ + @Test + public void testTinyintBlock() + { + byte[] values = {-128, -9, 0, 8, 127}; + ByteColumnVector vector = new ByteColumnVector(values.length + 1); + for (byte value : values) + { + vector.add(value); + } + vector.addNull(); + + Block block = load(vector, TinyintType.TINYINT, TypeDescription.Category.BYTE, values.length + 1); + + assertTrue(block instanceof ByteArrayBlock); + for (int i = 0; i < values.length; ++i) + { + assertEquals((long) values[i], TinyintType.TINYINT.getLong(block, i)); + } + assertTrue(block.isNull(values.length)); + } + + @Test + public void testSmallintBlock() + { + int[] values = {Short.MIN_VALUE, -3456, 0, 1234, Short.MAX_VALUE}; + IntColumnVector vector = new IntColumnVector(values.length + 1); + for (int value : values) + { + vector.add(value); + } + vector.addNull(); + + Block block = load(vector, SmallintType.SMALLINT, TypeDescription.Category.SHORT, values.length + 1); + + assertTrue(block instanceof ShortArrayBlock); + for (int i = 0; i < values.length; ++i) + { + assertEquals((long) values[i], SmallintType.SMALLINT.getLong(block, i)); + } + assertTrue(block.isNull(values.length)); + } + + @Test + public void testSmallintRejectsOutOfRangeValue() + { + IntColumnVector vector = new IntColumnVector(1); + vector.add(Short.MAX_VALUE + 1); + + assertThrows(IllegalArgumentException.class, + () -> load(vector, SmallintType.SMALLINT, TypeDescription.Category.SHORT, 1)); + } + + @Test + public void testTimeBlock() + { + int[] millis = {0, 3_723_004, 86_399_999}; + TimeColumnVector vector = new TimeColumnVector(millis.length + 1, 3); + for (int i = 0; i < millis.length; ++i) + { + vector.set(i, millis[i]); + } + vector.setNullValue(millis.length); + + Block block = load(vector, TimeType.TIME_MILLIS, TypeDescription.Category.TIME, millis.length + 1); + + assertTrue(block instanceof LongArrayBlock); + for (int i = 0; i < millis.length; ++i) + { + assertEquals((long) millis[i] * 1_000_000_000L, TimeType.TIME_MILLIS.getLong(block, i)); + } + assertTrue(block.isNull(millis.length)); + } + + @Test + public void testVarcharBlock() + { + // Capacity larger than batchSize: only the logical batch must be emitted. + int batchSize = 7; + BinaryColumnVector vector = new BinaryColumnVector(batchSize + 4); + + // Shared backing buffer with non-zero starts. + byte[] shared = "XXXhelloYYYY中文ZZ😀WW".getBytes(StandardCharsets.UTF_8); + byte[] hello = "hello".getBytes(StandardCharsets.UTF_8); + byte[] chinese = "中文".getBytes(StandardCharsets.UTF_8); + byte[] emoji = "😀".getBytes(StandardCharsets.UTF_8); + + int helloStart = indexOf(shared, hello); + int chineseStart = indexOf(shared, chinese); + int emojiStart = indexOf(shared, emoji); + + vector.setRef(0, shared, helloStart, hello.length); + vector.setRef(1, shared, chineseStart, chinese.length); + vector.setRef(2, shared, emojiStart, emoji.length); + vector.setRef(3, new byte[0], 0, 0); // empty string (non-null) + // NULL with intentionally stale start/lens/vector that must be ignored. + vector.addNull(); + assertEquals(5, vector.getWriteIndex()); + vector.vector[4] = shared; + vector.start[4] = helloStart; + vector.lens[4] = hello.length; + byte[] afterNull = "after-null".getBytes(StandardCharsets.UTF_8); + byte[] ascii = "ascii".getBytes(StandardCharsets.UTF_8); + vector.setRef(5, afterNull, 0, afterNull.length); + vector.setRef(6, ascii, 0, ascii.length); + // Extra capacity row must not appear in the block. + vector.setRef(7, "ignored".getBytes(StandardCharsets.UTF_8), 0, 7); + + Block block = load(vector, VarcharType.VARCHAR, TypeDescription.Category.VARCHAR, batchSize); + + assertTrue(block instanceof VariableWidthBlock); + VariableWidthBlock vw = (VariableWidthBlock) block; + assertEquals(batchSize, vw.getPositionCount()); + + assertSliceEquals(VarcharType.VARCHAR, vw, 0, hello); + assertSliceEquals(VarcharType.VARCHAR, vw, 1, chinese); + assertSliceEquals(VarcharType.VARCHAR, vw, 2, emoji); + assertFalse(vw.isNull(3)); + assertEquals(0, vw.getSliceLength(3)); + assertSliceEquals(VarcharType.VARCHAR, vw, 3, new byte[0]); + assertTrue(vw.isNull(4)); + assertEquals(0, vw.getSliceLength(4)); + assertSliceEquals(VarcharType.VARCHAR, vw, 5, afterNull); + assertSliceEquals(VarcharType.VARCHAR, vw, 6, ascii); + } + + @Test + public void testVarbinaryBlock() + { + int batchSize = 4; + BinaryColumnVector vector = new BinaryColumnVector(batchSize + 2); + + byte[] raw = {(byte) 0x00, (byte) 0x7F, (byte) 0x80, (byte) 0xFE, (byte) 0xFF}; + byte[] high = {(byte) 0xFF, (byte) 0x00, (byte) 0xAB}; + byte[] empty = new byte[0]; + + vector.setRef(0, raw, 0, raw.length); + vector.setRef(1, high, 0, high.length); + vector.setRef(2, empty, 0, 0); + vector.addNull(); + // Stale payload on the NULL row. + vector.vector[3] = raw; + vector.start[3] = 1; + vector.lens[3] = 3; + vector.setRef(4, new byte[] {(byte) 0x11}, 0, 1); + + Block block = load(vector, VarbinaryType.VARBINARY, TypeDescription.Category.VARBINARY, batchSize); + + assertTrue(block instanceof VariableWidthBlock); + VariableWidthBlock vw = (VariableWidthBlock) block; + assertEquals(batchSize, vw.getPositionCount()); + assertSliceEquals(VarbinaryType.VARBINARY, vw, 0, raw); + assertSliceEquals(VarbinaryType.VARBINARY, vw, 1, high); + assertFalse(vw.isNull(2)); + assertEquals(0, vw.getSliceLength(2)); + assertSliceEquals(VarbinaryType.VARBINARY, vw, 2, empty); + assertTrue(vw.isNull(3)); + assertEquals(0, vw.getSliceLength(3)); + } + + private static void assertSliceEquals(Type type, Block block, int position, byte[] expected) + { + assertFalse(block.isNull(position)); + assertArrayEquals(expected, type.getSlice(block, position).getBytes()); + } + + private static int indexOf(byte[] haystack, byte[] needle) + { + outer: + for (int i = 0; i <= haystack.length - needle.length; ++i) + { + for (int j = 0; j < needle.length; ++j) + { + if (haystack[i + j] != needle[j]) + { + continue outer; + } + } + return i; + } + throw new IllegalArgumentException("needle not found in haystack"); + } + + private static Block load(ColumnVector vector, Type type, + TypeDescription.Category typeCategory, int batchSize) + { + return new PixelsBlockLoader(new TestPageSource(), vector, type, typeCategory, batchSize).load(); + } + + private static class TestPageSource implements PixelsPageSource + { + @Override + public int getBatchId() + { + return 0; + } + + @Override + public long getCompletedBytes() + { + return 0; + } + + @Override + public long getReadTimeNanos() + { + return 0; + } + + @Override + public boolean isFinished() + { + return true; + } + + @Override + public Page getNextPage() + { + return null; + } + + @Override + public long getMemoryUsage() + { + return 0; + } + + @Override + public void close() + { + } + } +}