This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 7a00b2318be [core] Fix ANY_VALUE bytes intermediate round trip (#19667)
7a00b2318be is described below
commit 7a00b2318bee93bb6de8bc8538685fab76cc3c95
Author: Xuanyi Li <[email protected]>
AuthorDate: Sat Oct 3 14:35:25 2026 -0700
[core] Fix ANY_VALUE bytes intermediate round trip (#19667)
---
.../function/AnyValueAggregationFunction.java | 11 ++--
.../function/AnyValueAggregationFunctionTest.java | 59 ++++++++++++++++++++++
2 files changed, 65 insertions(+), 5 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
index e6669a16a3a..c984363d2c5 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
@@ -37,6 +37,7 @@ import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.utils.ByteArray;
/// AnyValue aggregation function returns any arbitrary NON-NULL value from
the column for each group.
@@ -224,7 +225,7 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
case BIG_DECIMAL:
return dict.getBigDecimalValue(dictId);
case BYTES:
- return dict.getBytesValue(dictId);
+ return dict.getByteArrayValue(dictId);
default:
throw new IllegalStateException("Unsupported dictionary type: " +
storedType);
}
@@ -246,7 +247,7 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
case BIG_DECIMAL:
return bvs.getBigDecimalValuesSV()[index];
case BYTES:
- return bvs.getBytesValuesSV()[index];
+ return new ByteArray(bvs.getBytesValuesSV()[index]);
default:
throw new IllegalStateException("Unsupported direct access type: " +
bvs.getValueType().getStoredType());
}
@@ -267,8 +268,8 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
return serializeVariableValue(DataType.STRING, ((String)
value).getBytes(StandardCharsets.UTF_8));
} else if (value instanceof BigDecimal) {
return serializeVariableValue(DataType.BIG_DECIMAL,
value.toString().getBytes(StandardCharsets.UTF_8));
- } else if (value instanceof byte[]) {
- return serializeVariableValue(DataType.BYTES, (byte[]) value);
+ } else if (value instanceof ByteArray) {
+ return serializeVariableValue(DataType.BYTES, ((ByteArray)
value).getBytes());
} else {
throw new IllegalStateException("Unsupported value type for
serialization: " + value.getClass().getName());
}
@@ -310,7 +311,7 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
case BIG_DECIMAL:
return new BigDecimal(new String(deserializeVariableBytes(buffer),
StandardCharsets.UTF_8));
case BYTES:
- return deserializeVariableBytes(buffer);
+ return new ByteArray(deserializeVariableBytes(buffer));
default:
throw new IllegalStateException("Unsupported data type for
deserialization: " + dataType);
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
index ffe3201d705..8ae6327b6c6 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
@@ -18,12 +18,71 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import org.apache.pinot.common.CustomObject;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
+import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import org.apache.pinot.queries.FluentQueryTest;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.utils.ByteArray;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+
public class AnyValueAggregationFunctionTest extends
AbstractAggregationFunctionTest {
+ @DataProvider(name = "bytesValueSources")
+ Object[][] bytesValueSources() {
+ return new Object[][]{{false}, {true}};
+ }
+
+ @Test(dataProvider = "bytesValueSources")
+ void testBytesIntermediateResultRoundTrip(boolean dictionaryEncoded) {
+ ExpressionContext expression = ExpressionContext.forIdentifier("myField");
+ AnyValueAggregationFunction function = new
AnyValueAggregationFunction(List.of(expression), true);
+ byte[] original = new byte[]{1, 2, 3};
+ ByteArray expected = new ByteArray(original);
+ BlockValSet valueSet = mock(BlockValSet.class);
+ when(valueSet.getValueType()).thenReturn(DataType.BYTES);
+ if (dictionaryEncoded) {
+ Dictionary dictionary = mock(Dictionary.class);
+ when(dictionary.getByteArrayValue(0)).thenReturn(expected);
+ when(valueSet.isDictionaryEncoded()).thenReturn(true);
+ when(valueSet.getDictionary()).thenReturn(dictionary);
+ when(valueSet.getDictionaryIdsSV()).thenReturn(new int[]{0});
+ } else {
+ when(valueSet.getBytesValuesSV()).thenReturn(new byte[][]{original});
+ }
+ Map<ExpressionContext, BlockValSet> values = Map.of(expression, valueSet);
+
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(1, holder, values);
+ Object intermediate = function.extractAggregationResult(holder);
+ assertEquals(intermediate, expected);
+ assertEquals(function.extractFinalResult(intermediate), expected);
+
+ GroupByResultHolder groupHolder = function.createGroupByResultHolder(1, 2);
+ function.aggregateGroupBySV(1, new int[]{0}, groupHolder, values);
+ groupHolder.ensureCapacity(2);
+ function.aggregateGroupByMV(1, new int[][]{{1}}, groupHolder, values);
+ assertEquals(function.extractGroupByResult(groupHolder, 0), expected);
+ assertEquals(function.extractGroupByResult(groupHolder, 1), expected);
+
+ AggregationFunction.SerializedIntermediateResult serialized =
function.serializeIntermediateResult(intermediate);
+ Object restored = function.deserializeIntermediateResult(
+ new CustomObject(serialized.getType(),
ByteBuffer.wrap(serialized.getBytes())));
+
+ assertEquals(restored, expected);
+ assertEquals(function.extractFinalResult(restored), expected);
+ assertEquals(function.serializeIntermediateResult(restored).getBytes(),
serialized.getBytes());
+ }
// Constants for standardized test queries and expected results
private static final String STANDARD_GROUP_BY_QUERY_TEMPLATE =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
