From d232fcf3d231909196746960f11d4d00cb0571c3 Mon Sep 17 00:00:00 2001 From: Dongyang Geng Date: Mon, 3 Aug 2026 15:03:22 +0800 Subject: [PATCH] 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); - } -}