diff --git a/LICENSE b/LICENSE index 2fe515bb8cd..1fa828e40ad 100644 --- a/LICENSE +++ b/LICENSE @@ -404,12 +404,14 @@ Apache Paimon ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldAggregatorFactory.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldBoolAndAggFactory.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldBoolOrAggFactory.java +./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldCollectAggFactory.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldFirstNonNullValueAggFactory.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldFirstValueAggFactory.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldLastNonNullValueAggFactory.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldLastValueAggFactory.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldListaggAggFactory.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldMaxAggFactory.java +./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldMergeMapAggFactory.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldMinAggFactory.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldProductAggFactory.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldRoaringBitmap32AggFactory.java @@ -419,12 +421,14 @@ Apache Paimon ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldAggregator.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldBoolAndAgg.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldBoolOrAgg.java +./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldCollectAgg.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldFirstNonNullValueAgg.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldFirstValueAgg.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldLastNonNullValueAgg.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldLastValueAgg.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldListaggAgg.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldMaxAgg.java +./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldMergeMapAgg.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldMinAgg.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldProductAgg.java ./fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldRoaringBitmap32Agg.java diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctionType.java b/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctionType.java index 79ac0575df6..aa02e38c8c5 100644 --- a/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctionType.java +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctionType.java @@ -58,6 +58,10 @@ public enum AggFunctionType { BOOL_AND, BOOL_OR, + // Collection aggregation + COLLECT, + MERGE_MAP, + // Roaring bitmap aggregation RBM32, RBM64, @@ -95,6 +99,9 @@ public enum AggFunctionType { DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE }; + /** Parameter name for distinct flag used in COLLECT function. */ + public static final String PARAM_DISTINCT = "distinct"; + /** * Returns the set of supported parameter names for this aggregation function. * @@ -106,6 +113,8 @@ public Set getSupportedParameters() { case STRING_AGG: // LISTAGG and STRING_AGG support optional "delimiter" parameter return Collections.singleton(PARAM_DELIMITER); + case COLLECT: + return Collections.singleton(PARAM_DISTINCT); case LAST_N: return Collections.singleton(PARAM_SIZE); default: @@ -143,7 +152,10 @@ public DataTypeRoot[] getSupportedDataTypeRoots() { case RBM64: return new DataTypeRoot[] {DataTypeRoot.BYTES}; case LAST_N: + case COLLECT: return new DataTypeRoot[] {DataTypeRoot.ARRAY}; + case MERGE_MAP: + return new DataTypeRoot[] {DataTypeRoot.MAP}; case LISTAGG: case STRING_AGG: return new DataTypeRoot[] {DataTypeRoot.STRING, DataTypeRoot.CHAR}; @@ -200,6 +212,18 @@ public void validateParameter(String parameterName, String parameterValue) { } } break; + case COLLECT: + if (PARAM_DISTINCT.equals(parameterName)) { + if (parameterValue == null + || (!parameterValue.equalsIgnoreCase("true") + && !parameterValue.equalsIgnoreCase("false"))) { + throw new IllegalArgumentException( + String.format( + "Parameter '%s' for aggregation function '%s' must be 'true' or 'false'", + parameterName, this)); + } + } + break; case LAST_N: int size; try { diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctions.java b/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctions.java index e43f3ab3a8c..087744e3539 100644 --- a/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctions.java +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctions.java @@ -322,6 +322,57 @@ public static AggFunction LAST_N(int size) { return new AggFunction(AggFunctionType.LAST_N, params); } + // =================================================================================== + // Collection Aggregation Functions + // =================================================================================== + + /** + * Creates a COLLECT aggregation function that collects multiple elements into an array. + * + *

Supported data types: ARRAY<T> + * + *

Null handling: Null values are included in the collection + * + * @return a COLLECT aggregation function + */ + public static AggFunction COLLECT() { + return new AggFunction(AggFunctionType.COLLECT, null); + } + + /** + * Creates a COLLECT aggregation function with distinct behavior. + * + *

Collects multiple elements into an array, optionally removing duplicates. Elements retain + * their first occurrence order when duplicates are removed. + * + *

Supported data types: ARRAY<T> + * + *

Null handling: Null values are included in the collection + * + * @param distinct whether to remove duplicates + * @return a COLLECT aggregation function + */ + public static AggFunction COLLECT(boolean distinct) { + Map params = new HashMap<>(); + params.put("distinct", String.valueOf(distinct)); + return new AggFunction(AggFunctionType.COLLECT, params); + } + + /** + * Creates a MERGE_MAP aggregation function that merges multiple maps. + * + *

Merges maps by combining their key-value pairs. For duplicate keys, the latest value wins. + * + *

Supported data types: MAP<K, V> + * + *

Null handling: Null keys are not allowed; null values are allowed + * + * @return a MERGE_MAP aggregation function + */ + public static AggFunction MERGE_MAP() { + return new AggFunction(AggFunctionType.MERGE_MAP, null); + } + // =================================================================================== // Internal Factory Methods // =================================================================================== diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java index 5ab3528d5c2..2cf8515ed12 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java @@ -1692,6 +1692,40 @@ void testLastNAggregation() throws Exception { assertResultsIgnoreOrder(rowIter, expectedRows, true); } + @Test + void testCollectionAggregation() throws Exception { + tEnv.executeSql( + "create table collection_agg (" + + "id int not null primary key not enforced, " + + "tags array, " + + "events array, " + + "attributes map" + + ") with (" + + "'table.merge-engine' = 'aggregation', " + + "'fields.tags.agg' = 'collect', " + + "'fields.tags.collect.distinct' = 'true', " + + "'fields.events.agg' = 'collect', " + + "'fields.attributes.agg' = 'merge_map')"); + + tEnv.executeSql( + "INSERT INTO collection_agg VALUES " + + "(1, ARRAY['b', 'a', 'b'], ARRAY['x', 'x'], MAP['a', 1])") + .await(); + tEnv.executeSql( + "INSERT INTO collection_agg VALUES " + + "(1, ARRAY['a', 'c'], ARRAY['y'], MAP['a', 2])") + .await(); + + CloseableIterator rowIter = tEnv.executeSql("SELECT * FROM collection_agg").collect(); + assertResultsIgnoreOrder( + rowIter, + Arrays.asList( + "+I[1, [b, a], [x, x], {a=1}]", + "-U[1, [b, a], [x, x], {a=1}]", + "+U[1, [b, a, c], [x, x, y], {a=2}]"), + true); + } + @Test void testPartialUpdateOnAggregationMergeEngine() throws Exception { tEnv.executeSql( diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkAggFunctionParserTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkAggFunctionParserTest.java index fe019a08add..ba4d82f970e 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkAggFunctionParserTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkAggFunctionParserTest.java @@ -168,4 +168,18 @@ void testRoundTripConversion() { assertThat(col2Func).isPresent(); assertThat(col2Func.get()).isEqualTo(AggFunctions.LISTAGG(";")); } + + @Test + void testRejectInvalidCollectDistinctParameter() { + Configuration options = new Configuration(); + options.setString("fields.tags.agg", "collect"); + options.setString("fields.tags.collect.distinct", "yes"); + assertThatThrownBy( + () -> + FlinkAggFunctionParser.parseAggFunction( + "tags", DataTypes.ARRAY(DataTypes.INT()), options)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("distinct") + .hasMessageContaining("must be 'true' or 'false'"); + } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldCollectAggFactory.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldCollectAggFactory.java new file mode 100644 index 00000000000..6cf5e8429bf --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldCollectAggFactory.java @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.fluss.server.kv.rowmerger.aggregate.factory; + +/* This file is based on source code of Apache Paimon Project (https://paimon.apache.org/), licensed by the Apache + * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for + * additional information regarding copyright ownership. */ + +import org.apache.fluss.metadata.AggFunction; +import org.apache.fluss.metadata.AggFunctionType; +import org.apache.fluss.server.kv.rowmerger.aggregate.functions.FieldCollectAgg; +import org.apache.fluss.types.ArrayType; +import org.apache.fluss.types.DataType; + +import static org.apache.fluss.utils.Preconditions.checkArgument; + +/** Factory for {@link FieldCollectAgg}. */ +public class FieldCollectAggFactory implements FieldAggregatorFactory { + + @Override + public FieldCollectAgg create(DataType fieldType, AggFunction aggFunction) { + checkArgument( + fieldType instanceof ArrayType, + "Data type for collect column must be 'ArrayType' but was '%s'.", + fieldType); + + boolean distinct = + Boolean.parseBoolean(aggFunction.getParameter(AggFunctionType.PARAM_DISTINCT)); + + return new FieldCollectAgg((ArrayType) fieldType, distinct); + } + + @Override + public String identifier() { + return AggFunctionType.COLLECT.toString(); + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldMergeMapAggFactory.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldMergeMapAggFactory.java new file mode 100644 index 00000000000..f4dfd20ef88 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldMergeMapAggFactory.java @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.fluss.server.kv.rowmerger.aggregate.factory; + +/* This file is based on source code of Apache Paimon Project (https://paimon.apache.org/), licensed by the Apache + * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for + * additional information regarding copyright ownership. */ + +import org.apache.fluss.metadata.AggFunction; +import org.apache.fluss.metadata.AggFunctionType; +import org.apache.fluss.server.kv.rowmerger.aggregate.functions.FieldMergeMapAgg; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.MapType; + +import static org.apache.fluss.utils.Preconditions.checkArgument; + +/** Factory for {@link FieldMergeMapAgg}. */ +public class FieldMergeMapAggFactory implements FieldAggregatorFactory { + + @Override + public FieldMergeMapAgg create(DataType fieldType, AggFunction aggFunction) { + checkArgument( + fieldType instanceof MapType, + "Data type for merge_map column must be 'MapType' but was '%s'.", + fieldType); + + return new FieldMergeMapAgg((MapType) fieldType); + } + + @Override + public String identifier() { + return AggFunctionType.MERGE_MAP.toString(); + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldCollectAgg.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldCollectAgg.java new file mode 100644 index 00000000000..e959b901255 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldCollectAgg.java @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.fluss.server.kv.rowmerger.aggregate.functions; + +/* This file is based on source code of Apache Paimon Project (https://paimon.apache.org/), licensed by the Apache + * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for + * additional information regarding copyright ownership. */ + +import org.apache.fluss.row.GenericArray; +import org.apache.fluss.row.InternalArray; +import org.apache.fluss.types.ArrayType; +import org.apache.fluss.utils.ByteArrayWrapper; + +import java.util.LinkedHashSet; +import java.util.Set; + +/** Collects array elements, retaining their first occurrence order when distinct is enabled. */ +public class FieldCollectAgg extends FieldAggregator { + + private static final long serialVersionUID = 1L; + + private final boolean distinct; + private final InternalArray.ElementGetter elementGetter; + + /** Creates a collector for the given array type and duplicate handling. */ + public FieldCollectAgg(ArrayType dataType, boolean distinct) { + super(dataType); + this.distinct = distinct; + this.elementGetter = InternalArray.createElementGetter(dataType.getElementType()); + } + + @Override + public Object agg(Object accumulator, Object inputField) { + if (inputField == null) { + return accumulator; + } + if (accumulator == null && !distinct) { + return inputField; + } + + InternalArray accumulatorArray = (InternalArray) accumulator; + InternalArray inputArray = (InternalArray) inputField; + int expectedSize = + (accumulatorArray == null ? 0 : accumulatorArray.size()) + inputArray.size(); + if (expectedSize == 0) { + return inputField; + } + if (!distinct) { + if (inputArray.size() == 0) { + return accumulator; + } + if (accumulatorArray.size() == 0) { + return inputField; + } + Object[] result = new Object[expectedSize]; + int accumulatorSize = accumulatorArray.size(); + for (int i = 0; i < accumulatorSize; i++) { + result[i] = elementGetter.getElementOrNull(accumulatorArray, i); + } + for (int i = 0; i < inputArray.size(); i++) { + result[accumulatorSize + i] = elementGetter.getElementOrNull(inputArray, i); + } + return new GenericArray(result); + } + + Set elements = new LinkedHashSet<>(expectedSize); + collectDistinct(elements, accumulatorArray); + int accumulatorDistinctSize = elements.size(); + collectDistinct(elements, inputArray); + if (accumulatorArray != null + && accumulatorDistinctSize == accumulatorArray.size() + && elements.size() == accumulatorDistinctSize) { + return accumulator; + } + Object[] result = elements.toArray(); + for (int i = 0; i < result.length; i++) { + if (result[i] instanceof ByteArrayWrapper) { + result[i] = ((ByteArrayWrapper) result[i]).getData(); + } + } + return new GenericArray(result); + } + + @Override + public boolean requiresAggOnFirstWrite() { + return distinct; + } + + private void collectDistinct(Set elements, InternalArray array) { + if (array == null) { + return; + } + for (int i = 0; i < array.size(); i++) { + Object element = elementGetter.getElementOrNull(array, i); + elements.add( + element instanceof byte[] ? new ByteArrayWrapper((byte[]) element) : element); + } + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldMergeMapAgg.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldMergeMapAgg.java new file mode 100644 index 00000000000..61cd85ee50c --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldMergeMapAgg.java @@ -0,0 +1,117 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.fluss.server.kv.rowmerger.aggregate.functions; + +/* This file is based on source code of Apache Paimon Project (https://paimon.apache.org/), licensed by the Apache + * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for + * additional information regarding copyright ownership. */ + +import org.apache.fluss.row.GenericArray; +import org.apache.fluss.row.GenericMap; +import org.apache.fluss.row.InternalArray; +import org.apache.fluss.row.InternalMap; +import org.apache.fluss.row.columnar.ColumnarMap; +import org.apache.fluss.types.DataTypeRoot; +import org.apache.fluss.types.MapType; +import org.apache.fluss.utils.ByteArrayWrapper; + +import java.util.HashMap; +import java.util.Map; + +/** Merge two maps. */ +public class FieldMergeMapAgg extends FieldAggregator { + + private static final long serialVersionUID = 1L; + private final boolean binaryKeys; + private final boolean binaryValues; + private final InternalArray.ElementGetter keyGetter; + private final InternalArray.ElementGetter valueGetter; + + /** Creates a map merger whose duplicate keys are resolved in favor of the input. */ + public FieldMergeMapAgg(MapType dataType) { + super(dataType); + this.binaryKeys = dataType.getKeyType().isAnyOf(DataTypeRoot.BINARY, DataTypeRoot.BYTES); + this.binaryValues = + dataType.getValueType().isAnyOf(DataTypeRoot.BINARY, DataTypeRoot.BYTES); + this.keyGetter = InternalArray.createElementGetter(dataType.getKeyType()); + this.valueGetter = InternalArray.createElementGetter(dataType.getValueType()); + } + + @Override + public Object agg(Object accumulator, Object inputField) { + if (accumulator == null || inputField == null) { + return accumulator == null ? inputField : accumulator; + } + + InternalMap accumulatorMap = (InternalMap) accumulator; + InternalMap inputMap = (InternalMap) inputField; + if (accumulatorMap.size() == 0) { + return inputField; + } + if (inputMap.size() == 0) { + return accumulator; + } + + int expectedSize = accumulatorMap.size() + inputMap.size(); + int capacity = (int) (expectedSize / 0.75f) + 1; + Map resultMap = new HashMap<>(capacity); + if (binaryValues) { + putToMap(resultMap, inputMap); + putMissingToMap(resultMap, accumulatorMap); + } else { + putToMap(resultMap, accumulatorMap); + putToMap(resultMap, inputMap); + } + if (binaryKeys) { + Object[] keys = new Object[resultMap.size()]; + Object[] values = new Object[resultMap.size()]; + int i = 0; + for (Map.Entry entry : resultMap.entrySet()) { + keys[i] = ((ByteArrayWrapper) entry.getKey()).getData(); + values[i] = entry.getValue(); + i++; + } + return new ColumnarMap(new GenericArray(keys), new GenericArray(values)); + } + return new GenericMap(resultMap); + } + + private void putToMap(Map map, InternalMap mapData) { + InternalArray keyArray = mapData.keyArray(); + InternalArray valueArray = mapData.valueArray(); + for (int i = 0; i < keyArray.size(); i++) { + Object key = keyGetter.getElementOrNull(keyArray, i); + Object value = valueGetter.getElementOrNull(valueArray, i); + map.put(binaryKeys ? new ByteArrayWrapper((byte[]) key) : key, value); + } + } + + private void putMissingToMap(Map map, InternalMap mapData) { + InternalArray keyArray = mapData.keyArray(); + InternalArray valueArray = mapData.valueArray(); + // Scan backwards so duplicate stored keys retain their last value. + for (int i = keyArray.size() - 1; i >= 0; i--) { + Object key = keyGetter.getElementOrNull(keyArray, i); + Object mapKey = binaryKeys ? new ByteArrayWrapper((byte[]) key) : key; + if (!map.containsKey(mapKey)) { + map.put(mapKey, valueGetter.getElementOrNull(valueArray, i)); + } + } + } +} diff --git a/fluss-server/src/main/resources/META-INF/services/org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldAggregatorFactory b/fluss-server/src/main/resources/META-INF/services/org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldAggregatorFactory index 7fdfe278786..e39ca587569 100644 --- a/fluss-server/src/main/resources/META-INF/services/org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldAggregatorFactory +++ b/fluss-server/src/main/resources/META-INF/services/org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldAggregatorFactory @@ -31,3 +31,5 @@ org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldStringAggFactory org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldRoaringBitmap32AggFactory org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldRoaringBitmap64AggFactory org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldLastNAggFactory +org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldCollectAggFactory +org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldMergeMapAggFactory diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/aggregate/FieldCollectionAggregatorTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/aggregate/FieldCollectionAggregatorTest.java new file mode 100644 index 00000000000..b6f7126036c --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/aggregate/FieldCollectionAggregatorTest.java @@ -0,0 +1,395 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.fluss.server.kv.rowmerger.aggregate; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.TableConfig; +import org.apache.fluss.metadata.AggFunction; +import org.apache.fluss.metadata.AggFunctions; +import org.apache.fluss.metadata.KvFormat; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.SchemaInfo; +import org.apache.fluss.record.BinaryValue; +import org.apache.fluss.record.TestingSchemaGetter; +import org.apache.fluss.row.BinaryString; +import org.apache.fluss.row.GenericArray; +import org.apache.fluss.row.GenericMap; +import org.apache.fluss.row.GenericRow; +import org.apache.fluss.row.InternalArray; +import org.apache.fluss.row.InternalMap; +import org.apache.fluss.row.columnar.ColumnarMap; +import org.apache.fluss.row.encode.RowEncoder; +import org.apache.fluss.server.kv.rowmerger.AggregateRowMerger; +import org.apache.fluss.server.kv.rowmerger.aggregate.functions.FieldCollectAgg; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.types.MapType; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.stream.Stream; + +import static org.apache.fluss.testutils.InternalMapAssert.assertThatMap; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.entry; + +/** Tests for collection aggregation semantics and encoding in both KV formats. */ +class FieldCollectionAggregatorTest { + + @ParameterizedTest + @EnumSource(KvFormat.class) + void testDistinctCollectOnFirstWrite(KvFormat format) { + Schema schema = schema(DataTypes.ARRAY(DataTypes.INT()), AggFunctions.COLLECT(true)); + AggregateRowMerger merger = merger(schema, format); + BinaryValue input = value(schema, format, GenericArray.of(3, 1, 3, null, null)); + BinaryValue result = merger.merge(null, input); + assertCollectedValues(result, DataTypes.INT(), 3, 1, null); + assertCollectedValues( + merger.merge(result, value(schema, format, null)), DataTypes.INT(), 3, 1, null); + } + + @ParameterizedTest + @EnumSource(KvFormat.class) + void testDistinctCollectAfterNull(KvFormat format) { + Schema schema = schema(DataTypes.ARRAY(DataTypes.INT()), AggFunctions.COLLECT(true)); + BinaryValue result = + merger(schema, format) + .merge( + value(schema, format, null), + value(schema, format, GenericArray.of(3, 1, 3, null, null))); + assertCollectedValues(result, DataTypes.INT(), 3, 1, null); + } + + @ParameterizedTest + @EnumSource(KvFormat.class) + void testDistinctCollectKeepsFirstOccurrenceOrder(KvFormat format) { + Schema schema = schema(DataTypes.ARRAY(DataTypes.INT()), AggFunctions.COLLECT(true)); + BinaryValue result = + merger(schema, format) + .merge( + value(schema, format, GenericArray.of(3, 1)), + value(schema, format, GenericArray.of(1, 2, 3, null))); + assertCollectedValues(result, DataTypes.INT(), 3, 1, 2, null); + } + + @ParameterizedTest + @EnumSource(KvFormat.class) + void testDistinctCollectWithRepeatedInput(KvFormat format) { + Schema schema = schema(DataTypes.ARRAY(DataTypes.INT()), AggFunctions.COLLECT(true)); + AggregateRowMerger merger = merger(schema, format); + BinaryValue result = + merger.merge( + value(schema, format, GenericArray.of(3, 1, null)), + value(schema, format, GenericArray.of(null, 1, 3, 1))); + assertCollectedValues(result, DataTypes.INT(), 3, 1, null); + result = + merger.merge( + value(schema, format, GenericArray.of(3, 3, null, null)), + value(schema, format, GenericArray.of(3, null))); + assertCollectedValues(result, DataTypes.INT(), 3, null); + } + + @ParameterizedTest + @MethodSource("collectModes") + void testCollectWithEmptyArrays(KvFormat format, boolean distinct) { + Schema schema = schema(DataTypes.ARRAY(DataTypes.INT()), AggFunctions.COLLECT(distinct)); + AggregateRowMerger merger = merger(schema, format); + BinaryValue empty = value(schema, format, GenericArray.of()); + BinaryValue input = value(schema, format, GenericArray.of(3, 3, null)); + Object[] expected = distinct ? new Object[] {3, null} : new Object[] {3, 3, null}; + assertCollectedValues(merger.merge(null, empty), DataTypes.INT()); + assertCollectedValues(merger.merge(value(schema, format, null), empty), DataTypes.INT()); + assertCollectedValues(merger.merge(empty, empty), DataTypes.INT()); + assertCollectedValues(merger.merge(empty, input), DataTypes.INT(), expected); + assertCollectedValues(merger.merge(input, empty), DataTypes.INT(), expected); + } + + static Stream collectModes() { + return Stream.of(KvFormat.values()) + .flatMap( + format -> + Stream.of(false, true) + .map(distinct -> Arguments.of(format, distinct))); + } + + @ParameterizedTest(name = "{0}, distinct={1}, element={2}") + @MethodSource("collectElements") + void testCollectElements( + KvFormat format, + boolean distinct, + DataType elementType, + Object first, + Object second, + Object third) { + Schema schema = schema(DataTypes.ARRAY(elementType), AggFunctions.COLLECT(distinct)); + BinaryValue result = + merger(schema, format) + .merge( + value(schema, format, GenericArray.of(first, second, null)), + value(schema, format, GenericArray.of(first, third, null, first))); + Object[] expected = + distinct + ? new Object[] {first, second, null, third} + : new Object[] {first, second, null, first, third, null, first}; + assertCollectedValues(result, elementType, expected); + if (distinct) { + BinaryValue repeated = + merger(schema, format) + .merge( + value(schema, format, new GenericArray(expected)), + value( + schema, + format, + GenericArray.of(third, first, null, second))); + assertCollectedValues(repeated, elementType, expected); + } + } + + static Stream collectElements() { + Stream.Builder cases = Stream.builder(); + for (KvFormat format : KvFormat.values()) { + for (boolean distinct : new boolean[] {false, true}) { + cases.add(Arguments.of(format, distinct, DataTypes.INT(), 1, 2, 3)); + cases.add( + Arguments.of( + format, + distinct, + DataTypes.STRING(), + BinaryString.fromString("b"), + BinaryString.fromString("a"), + BinaryString.fromString("c"))); + cases.add( + Arguments.of( + format, + distinct, + DataTypes.BYTES(), + new byte[] {2, 3}, + new byte[] {4}, + new byte[] {1, 4, 5})); + cases.add( + Arguments.of( + format, + distinct, + DataTypes.ARRAY(DataTypes.INT()), + GenericArray.of(1, null), + GenericArray.of(1, 3), + GenericArray.of(1, 4))); + cases.add( + Arguments.of( + format, + distinct, + DataTypes.ROW( + DataTypes.FIELD("id", DataTypes.INT()), + DataTypes.FIELD("name", DataTypes.STRING())), + GenericRow.of(1, BinaryString.fromString("a")), + GenericRow.of(1, BinaryString.fromString("b")), + GenericRow.of(1, BinaryString.fromString("c")))); + } + } + return cases.build(); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testReversedCollect(boolean distinct) { + FieldCollectAgg aggregator = + new FieldCollectAgg(DataTypes.ARRAY(DataTypes.INT()), distinct); + InternalArray result = + (InternalArray) + aggregator.aggReversed(GenericArray.of(3, 4), GenericArray.of(1, 2)); + assertThat(result.size()).isEqualTo(4); + for (int i = 0; i < result.size(); i++) { + assertThat(result.getInt(i)).isEqualTo(i + 1); + } + } + + @ParameterizedTest + @EnumSource(KvFormat.class) + void testMergeMap(KvFormat format) { + MapType type = DataTypes.MAP(DataTypes.INT(), DataTypes.STRING()); + Schema schema = schema(type, AggFunctions.MERGE_MAP()); + BinaryValue result = + merger(schema, format) + .merge( + value( + schema, + format, + GenericMap.of( + 1, + BinaryString.fromString("a"), + 2, + BinaryString.fromString("b"))), + value( + schema, + format, + GenericMap.of(2, BinaryString.fromString("c"), 3, null))); + InternalMap map = result.row.getMap(1); + assertThat(map.size()).isEqualTo(3); + assertThatMap(map) + .withMapType(type) + .isEqualTo( + GenericMap.of( + 1, + BinaryString.fromString("a"), + 2, + BinaryString.fromString("c"), + 3, + null)); + } + + @ParameterizedTest + @MethodSource("binaryTypes") + void testMergeMapBinaryKeys(KvFormat format, DataType keyType) { + Schema schema = + schema(DataTypes.MAP(keyType, DataTypes.STRING()), AggFunctions.MERGE_MAP()); + AggregateRowMerger merger = merger(schema, format); + BinaryValue result = + merger.merge( + value( + schema, + format, + GenericMap.of( + new byte[] {1, 2}, BinaryString.fromString("old"), + new byte[] {3, 4}, BinaryString.fromString("kept"))), + value( + schema, + format, + GenericMap.of( + new byte[] {1, 2}, BinaryString.fromString("new"), + new byte[] {5, 6}, BinaryString.fromString("added")))); + InternalMap map = result.row.getMap(1); + assertThat(map.size()).isEqualTo(3); + assertThat(binaryMapEntries(map)) + .containsOnly( + entry("[1, 2]", "new"), entry("[3, 4]", "kept"), entry("[5, 6]", "added")); + + result = + merger.merge(result, value(schema, format, GenericMap.of(new byte[] {1, 2}, null))); + assertThat(result.row.getMap(1).size()).isEqualTo(3); + assertThat(binaryMapEntries(result.row.getMap(1))) + .containsOnly( + entry("[1, 2]", null), entry("[3, 4]", "kept"), entry("[5, 6]", "added")); + } + + @ParameterizedTest + @MethodSource("binaryTypes") + void testMergeMapKeepsLastStoredValue(KvFormat format, DataType keyType) { + Schema schema = schema(DataTypes.MAP(keyType, DataTypes.BYTES()), AggFunctions.MERGE_MAP()); + InternalMap old = + new ColumnarMap( + GenericArray.of(new byte[] {1, 2}, new byte[] {1, 2}), + GenericArray.of(new byte[] {1}, new byte[] {2})); + BinaryValue result = + merger(schema, format) + .merge( + value(schema, format, old), + value( + schema, + format, + GenericMap.of(new byte[] {3, 4}, new byte[] {3}))); + InternalMap map = result.row.getMap(1); + assertThat(map.size()).isEqualTo(2); + Map entries = new HashMap<>(); + for (int i = 0; i < map.size(); i++) { + entries.put( + Arrays.toString(map.keyArray().getBytes(i)), + Arrays.toString(map.valueArray().getBytes(i))); + } + assertThat(entries).containsOnly(entry("[1, 2]", "[2]"), entry("[3, 4]", "[3]")); + } + + @ParameterizedTest + @MethodSource("binaryTypes") + void testMergeMapNullValueOverridesStoredBytes(KvFormat format, DataType valueType) { + Schema schema = schema(DataTypes.MAP(DataTypes.INT(), valueType), AggFunctions.MERGE_MAP()); + BinaryValue result = + merger(schema, format) + .merge( + value(schema, format, GenericMap.of(1, new byte[] {1, 2})), + value(schema, format, GenericMap.of(1, null))); + InternalMap map = result.row.getMap(1); + assertThat(map.size()).isEqualTo(1); + assertThat(map.keyArray().getInt(0)).isEqualTo(1); + assertThat(map.valueArray().isNullAt(0)).isTrue(); + } + + static Stream binaryTypes() { + return Stream.of(KvFormat.values()) + .flatMap( + format -> + Stream.of(DataTypes.BYTES(), DataTypes.BINARY(2)) + .map(type -> Arguments.of(format, type))); + } + + private static Map binaryMapEntries(InternalMap map) { + Map entries = new HashMap<>(); + for (int i = 0; i < map.size(); i++) { + entries.put( + Arrays.toString(map.keyArray().getBytes(i)), + map.valueArray().isNullAt(i) ? null : map.valueArray().getString(i).toString()); + } + return entries; + } + + private static void assertCollectedValues( + BinaryValue value, DataType elementType, Object... expected) { + assertThat(value.row.isNullAt(1)).isFalse(); + InternalArray array = value.row.getArray(1); + InternalArray.ElementGetter getter = + InternalArray.createDeepElementGetter(elementType, true); + Object[] actual = new Object[array.size()]; + for (int i = 0; i < array.size(); i++) { + actual[i] = getter.getElementOrNull(array, i); + } + assertThat(actual).containsExactly(expected); + } + + private static Schema schema(DataType type, AggFunction function) { + return Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("value", type, function) + .primaryKey("id") + .build(); + } + + private static AggregateRowMerger merger(Schema schema, KvFormat format) { + AggregateRowMerger merger = + new AggregateRowMerger( + new TableConfig(new Configuration()), + format, + new TestingSchemaGetter(new SchemaInfo(schema, (short) 1))); + merger.configureTargetColumns(null, (short) 1, schema); + return merger; + } + + private static BinaryValue value(Schema schema, KvFormat format, Object field) { + RowEncoder encoder = RowEncoder.create(format, schema.getRowType()); + encoder.startNewRow(); + encoder.encodeField(0, 1); + encoder.encodeField(1, field); + return new BinaryValue((short) 1, encoder.finishRow()); + } +} diff --git a/website/docs/table-design/merge-engines/aggregation.md b/website/docs/table-design/merge-engines/aggregation.md index cc5150e4c2e..7628e229e9c 100644 --- a/website/docs/table-design/merge-engines/aggregation.md +++ b/website/docs/table-design/merge-engines/aggregation.md @@ -1154,6 +1154,118 @@ TableDescriptor.builder() +### collect + +Collects array elements into a single array. + +- **Supported Data Types**: `ARRAY` +- **Behavior**: Concatenates array elements; with `distinct=true`, duplicates are removed +- **Ordering**: Elements follow the order writes reach the server. With `distinct=true`, the first occurrence of each element is retained +- **Null Handling**: Null elements are retained; null inputs are ignored +- **Parameters**: `distinct` (`true` or `false`, default is `false`) + +:::note +The collected array has no size limit. Every update rewrites the whole stored array. With the default `'table.changelog.image' = 'full'`, both the old and the new array are also written to the changelog on every update. As the array grows, so do memory usage, write costs, and changelog volume. Use `last_n` when only a bounded number of recent elements is needed. +::: + +**Example:** + + + +```sql +CREATE TABLE test_collect ( + id BIGINT, + tags ARRAY, + PRIMARY KEY (id) NOT ENFORCED +) WITH ( + 'table.merge-engine' = 'aggregation', + 'fields.tags.agg' = 'collect', + 'fields.tags.collect.distinct' = 'true' +); + +INSERT INTO test_collect VALUES + (1, ARRAY[1, 2, 2]), + (1, ARRAY[2, 3, NULL]); + +SELECT * FROM test_collect; +-- Result: +-- (1, [1, 2, 3, NULL]) +``` + + + + +```java +Schema schema = Schema.newBuilder() + .column("id", DataTypes.BIGINT()) + .column("tags", DataTypes.ARRAY(DataTypes.INT()), AggFunctions.COLLECT(true)) + .primaryKey("id") + .build(); + +TableDescriptor.builder() + .schema(schema) + .property("table.merge-engine", "aggregation") + .build(); + +// Input: (1, [1, 2, 2]), (1, [2, 3, null]) +// Result: (1, [1, 2, 3, null]) +``` + + + + +### merge_map + +Merges map values by combining key-value pairs. For duplicate keys, the latest value wins. + +- **Supported Data Types**: `MAP` +- **Behavior**: Merges maps, overriding earlier values for duplicate keys +- **Null Handling**: Null keys are not allowed; null values are allowed + +**Example:** + + + +```sql +CREATE TABLE test_merge_map ( + id BIGINT, + attributes MAP, + PRIMARY KEY (id) NOT ENFORCED +) WITH ( + 'table.merge-engine' = 'aggregation', + 'fields.attributes.agg' = 'merge_map' +); + +INSERT INTO test_merge_map VALUES + (1, MAP[1, 'a', 2, 'b']), + (1, MAP[2, 'c', 3, NULL]); + +SELECT * FROM test_merge_map; +-- Result: (1, {1='a', 2='c', 3=NULL}) +``` + + + + +```java +Schema schema = Schema.newBuilder() + .column("id", DataTypes.BIGINT()) + .column("attributes", DataTypes.MAP(DataTypes.INT(), DataTypes.STRING()), AggFunctions.MERGE_MAP()) + .primaryKey("id") + .build(); + +TableDescriptor.builder() + .schema(schema) + .property("table.merge-engine", "aggregation") + .build(); + +// Input: (1, {1='a', 2='b'}), (1, {2='c', 3=null}) +// Result: (1, {1='a', 2='c', 3=null}) +``` + + + + ## Delete Behavior The aggregation merge engine provides limited support for delete operations. You can configure the behavior using the `'table.delete.behavior'` option: