Add collect/merge_map aggregations and binary merge optimizations.

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<String> 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 {

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.
+     *
+     * <p>Supported data types: ARRAY&lt;T&gt;
+     *
+     * <p>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.
+     *
+     * <p>Collects multiple elements into an array, optionally removing duplicates. Elements retain
+     * their first occurrence order when duplicates are removed.
+     *
+     * <p>Supported data types: ARRAY&lt;T&gt;
+     *
+     * <p>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<String, String> 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.
+     *
+     * <p>Merges maps by combining their key-value pairs. For duplicate keys, the latest value wins.
+     *
+     * <p>Supported data types: MAP&lt;K, V&gt;
+     *
+     * <p>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
     // ===================================================================================

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<string>, "
+                        + "events array<string>, "
+                        + "attributes map<string, int>"
+                        + ") 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<Row> 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(

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'");
+    }
 }

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();
+    }
+}

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();
+    }
+}

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<Object> 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<Object> 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);
+        }
+    }
+}