[kv] Implement basic Aggregate Merge Engine (#2255)

Co-authored-by: Jark Wu <[email protected]>

Author: platinumhamburg

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java

@@ -52,6 +52,7 @@
 import org.apache.fluss.exception.TooManyPartitionsException;
 import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.fs.FsPathAndFileName;
+import org.apache.fluss.metadata.AggFunctions;
 import org.apache.fluss.metadata.DatabaseDescriptor;
 import org.apache.fluss.metadata.DatabaseInfo;
 import org.apache.fluss.metadata.DeleteBehavior;
@@ -512,6 +513,47 @@ void testCreateTableWithDeleteBehavior() {
         TableInfo tableInfo2 = admin.getTableInfo(tablePath2).join();
         assertThat(tableInfo2.getTableConfig().getDeleteBehavior()).hasValue(DeleteBehavior.IGNORE);
 
+        // Test 2.5: AGGREGATION merge engine - should set delete behavior to IGNORE
+        TablePath tablePathAggregate = TablePath.of("fluss", "test_ignore_delete_for_aggregate");
+        Schema aggregateSchema =
+                Schema.newBuilder()
+                        .column("id", DataTypes.INT())
+                        .column("count", DataTypes.BIGINT(), AggFunctions.SUM())
+                        .primaryKey("id")
+                        .build();
+        Map<String, String> propertiesAggregate = new HashMap<>();
+        propertiesAggregate.put(ConfigOptions.TABLE_MERGE_ENGINE.key(), "aggregation");
+        TableDescriptor tableDescriptorAggregate =
+                TableDescriptor.builder()
+                        .schema(aggregateSchema)
+                        .comment("aggregate merge engine table")
+                        .properties(propertiesAggregate)
+                        .build();
+        admin.createTable(tablePathAggregate, tableDescriptorAggregate, false).join();
+        // Get the table and verify delete behavior is changed to IGNORE
+        TableInfo tableInfoAggregate = admin.getTableInfo(tablePathAggregate).join();
+        assertThat(tableInfoAggregate.getTableConfig().getDeleteBehavior())
+                .hasValue(DeleteBehavior.IGNORE);
+
+        // Test 2.6: AGGREGATION merge engine with delete behavior explicitly set to ALLOW - should
+        // be allowed
+        TablePath tablePathAggregateAllow =
+                TablePath.of("fluss", "test_allow_delete_for_aggregate");
+        Map<String, String> propertiesAggregateAllow = new HashMap<>();
+        propertiesAggregateAllow.put(ConfigOptions.TABLE_MERGE_ENGINE.key(), "aggregation");
+        propertiesAggregateAllow.put(ConfigOptions.TABLE_DELETE_BEHAVIOR.key(), "ALLOW");
+        TableDescriptor tableDescriptorAggregateAllow =
+                TableDescriptor.builder()
+                        .schema(aggregateSchema)
+                        .comment("aggregate merge engine table with allow delete")
+                        .properties(propertiesAggregateAllow)
+                        .build();
+        admin.createTable(tablePathAggregateAllow, tableDescriptorAggregateAllow, false).join();
+        // Get the table and verify delete behavior is set to ALLOW
+        TableInfo tableInfoAggregateAllow = admin.getTableInfo(tablePathAggregateAllow).join();
+        assertThat(tableInfoAggregateAllow.getTableConfig().getDeleteBehavior())
+                .hasValue(DeleteBehavior.ALLOW);
+
         // Test 3: FIRST_ROW merge engine with delete behavior explicitly set to ALLOW
         TablePath tablePath3 = TablePath.of("fluss", "test_allow_delete_for_first_row");
         Map<String, String> properties3 = new HashMap<>();

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -1420,9 +1420,10 @@ public class ConfigOptions {
                     .noDefaultValue()
                     .withDescription(
                             "Defines the merge engine for the primary key table. By default, primary key table doesn't have merge engine. "
-                                    + "The supported merge engines are `first_row` and `versioned`. "
+                                    + "The supported merge engines are `first_row`, `versioned`, and `aggregation`. "
                                     + "The `first_row` merge engine will keep the first row of the same primary key. "
-                                    + "The `versioned` merge engine will keep the row with the largest version of the same primary key.");
+                                    + "The `versioned` merge engine will keep the row with the largest version of the same primary key. "
+                                    + "The `aggregation` merge engine will aggregate rows with the same primary key using field-level aggregate functions.");
 
     public static final ConfigOption<String> TABLE_MERGE_ENGINE_VERSION_COLUMN =
             // we may need to introduce "del-column" in the future to support delete operation
@@ -1440,10 +1441,11 @@ public class ConfigOptions {
                     .withDescription(
                             "Defines the delete behavior for the primary key table. "
                                     + "The supported delete behaviors are `allow`, `ignore`, and `disable`. "
-                                    + "The `allow` behavior allows normal delete operations (default). "
+                                    + "The `allow` behavior allows normal delete operations (default for default merge engine). "
                                     + "The `ignore` behavior silently skips delete requests without error. "
                                     + "The `disable` behavior rejects delete requests with a clear error message. "
-                                    + "For tables with FIRST_ROW or VERSIONED merge engines, this option defaults to `ignore`.");
+                                    + "For tables with FIRST_ROW, VERSIONED, or AGGREGATION merge engines, this option defaults to `ignore`. "
+                                    + "Note: For AGGREGATION merge engine, when set to `allow`, delete operations will remove the entire record.");
 
     public static final ConfigOption<String> TABLE_AUTO_INCREMENT_FIELDS =
             key("table.auto-increment.fields")

fluss-common/src/main/java/org/apache/fluss/metadata/AggFunction.java

@@ -0,0 +1,148 @@
+/*
+ * 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.metadata;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Aggregation function with optional parameters for aggregate merge engine.
+ *
+ * <p>This class represents a parameterized aggregation function that can be applied to non-primary
+ * key columns in aggregation merge engine tables. It encapsulates both the function type and
+ * function-specific parameters (e.g., delimiter for LISTAGG).
+ *
+ * <p>Use {@link AggFunctions} utility class to create instances:
+ *
+ * <pre>{@code
+ * AggFunction sumFunc = AggFunctions.SUM();
+ * AggFunction listaggFunc = AggFunctions.LISTAGG(";");
+ * }</pre>
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public final class AggFunction implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private final AggFunctionType type;
+    private final Map<String, String> parameters;
+
+    /**
+     * Creates an aggregation function with the specified type and parameters.
+     *
+     * @param type the aggregation function type
+     * @param parameters the function parameters (nullable)
+     */
+    AggFunction(AggFunctionType type, @Nullable Map<String, String> parameters) {
+        this.type = Objects.requireNonNull(type, "Aggregation function type must not be null");
+        this.parameters =
+                parameters == null || parameters.isEmpty()
+                        ? Collections.emptyMap()
+                        : Collections.unmodifiableMap(new HashMap<>(parameters));
+    }
+
+    /**
+     * Returns the aggregation function type.
+     *
+     * @return the function type
+     */
+    public AggFunctionType getType() {
+        return type;
+    }
+
+    /**
+     * Returns the function parameters.
+     *
+     * @return an immutable map of parameters
+     */
+    public Map<String, String> getParameters() {
+        return parameters;
+    }
+
+    /**
+     * Gets a specific parameter value.
+     *
+     * @param key the parameter key
+     * @return the parameter value, or null if not found
+     */
+    @Nullable
+    public String getParameter(String key) {
+        return parameters.get(key);
+    }
+
+    /**
+     * Checks if this function has any parameters.
+     *
+     * @return true if parameters are present, false otherwise
+     */
+    public boolean hasParameters() {
+        return !parameters.isEmpty();
+    }
+
+    /**
+     * Validates all parameters of this aggregation function.
+     *
+     * <p>This method checks that:
+     *
+     * <ul>
+     *   <li>All parameter names are supported by the function type
+     *   <li>All parameter values are valid
+     * </ul>
+     *
+     * @throws IllegalArgumentException if any parameter is invalid
+     */
+    public void validate() {
+        for (Map.Entry<String, String> entry : parameters.entrySet()) {
+            type.validateParameter(entry.getKey(), entry.getValue());
+        }
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        AggFunction that = (AggFunction) o;
+        return type == that.type && parameters.equals(that.parameters);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(type, parameters);
+    }
+
+    @Override
+    public String toString() {
+        if (parameters.isEmpty()) {
+            return type.toString();
+        }
+        return type.toString() + parameters;
+    }
+}

fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctionType.java

@@ -0,0 +1,157 @@
+/*
+ * 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.metadata;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+import java.util.Collections;
+import java.util.Locale;
+import java.util.Set;
+
+/**
+ * Aggregation function type for aggregate merge engine.
+ *
+ * <p>This enum represents all supported aggregation function types that can be applied to
+ * non-primary key columns in aggregation merge engine tables.
+ */
+@PublicEvolving
+public enum AggFunctionType {
+    // Numeric aggregation
+    SUM,
+    PRODUCT,
+    MAX,
+    MIN,
+
+    // Value selection
+    LAST_VALUE,
+    LAST_VALUE_IGNORE_NULLS,
+    FIRST_VALUE,
+    FIRST_VALUE_IGNORE_NULLS,
+
+    // String aggregation
+    LISTAGG,
+    STRING_AGG, // Alias for LISTAGG - maps to same factory
+
+    // Boolean aggregation
+    BOOL_AND,
+    BOOL_OR;
+
+    /** Parameter name for delimiter used in LISTAGG and STRING_AGG functions. */
+    public static final String PARAM_DELIMITER = "delimiter";
+
+    /**
+     * Returns the set of supported parameter names for this aggregation function.
+     *
+     * @return an immutable set of parameter names
+     */
+    public Set<String> getSupportedParameters() {
+        switch (this) {
+            case LISTAGG:
+            case STRING_AGG:
+                // LISTAGG and STRING_AGG support optional "delimiter" parameter
+                return Collections.singleton(PARAM_DELIMITER);
+            default:
+                // All other functions do not accept parameters
+                return Collections.emptySet();
+        }
+    }
+
+    /**
+     * Validates a parameter value for this aggregation function.
+     *
+     * @param parameterName the parameter name
+     * @param parameterValue the parameter value
+     * @throws IllegalArgumentException if the parameter value is invalid
+     */
+    public void validateParameter(String parameterName, String parameterValue) {
+        // Check if parameter is supported
+        if (!getSupportedParameters().contains(parameterName)) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Parameter '%s' is not supported for aggregation function '%s'. "
+                                    + "Supported parameters: %s",
+                            parameterName,
+                            this,
+                            getSupportedParameters().isEmpty()
+                                    ? "none"
+                                    : getSupportedParameters()));
+        }
+
+        // Validate parameter value based on function type and parameter name
+        switch (this) {
+            case LISTAGG:
+            case STRING_AGG:
+                if (PARAM_DELIMITER.equals(parameterName)) {
+                    if (parameterValue == null || parameterValue.isEmpty()) {
+                        throw new IllegalArgumentException(
+                                String.format(
+                                        "Parameter '%s' for aggregation function '%s' must be a non-empty string",
+                                        parameterName, this));
+                    }
+                }
+                break;
+            default:
+                // No validation needed for other functions (they don't have parameters)
+                break;
+        }
+    }
+
+    /**
+     * Converts a string to an AggFunctionType enum value.
+     *
+     * <p>This method supports multiple naming formats:
+     *
+     * <ul>
+     *   <li>Underscore format: "last_value_ignore_nulls"
+     *   <li>Hyphen format: "last-value-ignore-nulls"
+     *   <li>Case insensitive matching
+     * </ul>
+     *
+     * <p>Note: For string_agg, this will return STRING_AGG enum, but the server-side factory will
+     * map it to the same implementation as listagg.
+     *
+     * @param name the aggregation function type name
+     * @return the AggFunctionType enum value, or null if not found
+     */
+    public static AggFunctionType fromString(String name) {
+        if (name == null || name.trim().isEmpty()) {
+            return null;
+        }
+
+        // Normalize the input: convert hyphens to underscores and uppercase
+        String normalized = name.replace('-', '_').toUpperCase(Locale.ROOT).trim();
+
+        try {
+            return AggFunctionType.valueOf(normalized);
+        } catch (IllegalArgumentException e) {
+            return null;
+        }
+    }
+
+    /**
+     * Converts this AggFunctionType to its string identifier.
+     *
+     * <p>The identifier is the lowercase name with underscores, e.g., "sum", "last_value".
+     *
+     * @return the identifier string
+     */
+    @Override
+    public String toString() {
+        return name().toLowerCase(Locale.ROOT);
+    }
+}