feat: support merge-engine=aggregation (basic mode)

crates/paimon/src/spec/aggregation.rs

@@ -0,0 +1,238 @@
+// 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.
+
+use std::collections::HashMap;
+
+const MERGE_ENGINE_OPTION: &str = "merge-engine";
+const AGGREGATION_ENGINE: &str = "aggregation";
+const IGNORE_DELETE_OPTION: &str = "ignore-delete";
+const IGNORE_DELETE_SUFFIX: &str = ".ignore-delete";
+const AGGREGATION_REMOVE_RECORD_ON_DELETE_OPTION: &str = "aggregation.remove-record-on-delete";
+const FIELDS_DEFAULT_AGG_FUNCTION_OPTION: &str = "fields.default-aggregate-function";
+const FIELDS_PREFIX: &str = "fields.";
+const AGG_FUNCTION_SUFFIX: &str = ".aggregate-function";
+const IGNORE_RETRACT_SUFFIX: &str = ".ignore-retract";
+const DISTINCT_SUFFIX: &str = ".distinct";
+const SEQUENCE_GROUP_SUFFIX: &str = ".sequence-group";
+const NESTED_KEY_SUFFIX: &str = ".nested-key";
+const COUNT_LIMIT_SUFFIX: &str = ".count-limit";
+
+/// Minimal aggregation mode recognized by the current Rust implementation.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum AggregationMode {
+    Basic,
+}
+
+/// Aggregation-merge-engine option inspection and validation.
+///
+/// The basic mode accepts only `merge-engine=aggregation` on a PK table with
+/// the following option keys:
+/// - `fields.default-aggregate-function`
+/// - `fields.<col>.aggregate-function`
+/// - `fields.<col>.list-agg-delimiter`
+///
+/// All other aggregation-specific knobs (`ignore-retract`, `distinct`,
+/// `nested-key`, `count-limit`, `aggregation.remove-record-on-delete`,
+/// `sequence-group`, `ignore-delete`) are rejected.  Retract rows
+/// (DELETE / UPDATE_BEFORE) are rejected at runtime by the merge function.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct AggregationConfig<'a> {
+    options: &'a HashMap<String, String>,
+}
+
+impl<'a> AggregationConfig<'a> {
+    pub(crate) fn new(options: &'a HashMap<String, String>) -> Self {
+        Self { options }
+    }
+
+    pub(crate) fn is_enabled(&self) -> bool {
+        self.options
+            .get(MERGE_ENGINE_OPTION)
+            .is_some_and(|value| value.eq_ignore_ascii_case(AGGREGATION_ENGINE))
+    }
+
+    /// Validate options at CREATE TABLE time.
+    pub(crate) fn validate_create_mode(
+        &self,
+        has_primary_keys: bool,
+    ) -> crate::Result<Option<AggregationMode>> {
+        match self.validated_mode(has_primary_keys) {
+            Ok(mode) => Ok(mode),
+            Err(unsupported_options) => Err(crate::Error::ConfigInvalid {
+                message: format!(
+                    "merge-engine=aggregation only supports the basic mode in this build; unsupported options: {}",
+                    unsupported_options.join(", ")
+                ),
+            }),
+        }
+    }
+
+    /// Validate options at read/write runtime.
+    pub(crate) fn validate_runtime_mode(
+        &self,
+        has_primary_keys: bool,
+        table_name: &str,
+    ) -> crate::Result<Option<AggregationMode>> {
+        match self.validated_mode(has_primary_keys) {
+            Ok(mode) => Ok(mode),
+            Err(unsupported_options) => Err(crate::Error::Unsupported {
+                message: format!(
+                    "Table '{table_name}' uses merge-engine=aggregation options not supported by this build: {}",
+                    unsupported_options.join(", ")
+                ),
+            }),
+        }
+    }
+
+    fn validated_mode(
+        &self,
+        has_primary_keys: bool,
+    ) -> std::result::Result<Option<AggregationMode>, Vec<String>> {
+        if !has_primary_keys || !self.is_enabled() {
+            return Ok(None);
+        }
+
+        let unsupported_options = self.unsupported_option_keys();
+        if !unsupported_options.is_empty() {
+            return Err(unsupported_options);
+        }
+
+        Ok(Some(AggregationMode::Basic))
+    }
+
+    fn unsupported_option_keys(&self) -> Vec<String> {
+        let mut keys: Vec<String> = self
+            .options
+            .keys()
+            .filter(|key| is_unsupported_aggregation_option(key))
+            .cloned()
+            .collect();
+        keys.sort();
+        keys
+    }
+
+    /// Per-field aggregate function configured via `fields.<col>.aggregate-function`.
+    pub(crate) fn agg_function_for_field(&self, field_name: &str) -> Option<&str> {
+        let key = format!("{FIELDS_PREFIX}{field_name}{AGG_FUNCTION_SUFFIX}");
+        self.options.get(&key).map(String::as_str)
+    }
+
+    /// Default aggregate function from `fields.default-aggregate-function`.
+    pub(crate) fn default_agg_function(&self) -> Option<&str> {
+        self.options
+            .get(FIELDS_DEFAULT_AGG_FUNCTION_OPTION)
+            .map(String::as_str)
+    }
+}
+
+fn is_unsupported_aggregation_option(key: &str) -> bool {
+    key == IGNORE_DELETE_OPTION
+        || key.ends_with(IGNORE_DELETE_SUFFIX)
+        || key == AGGREGATION_REMOVE_RECORD_ON_DELETE_OPTION
+        || is_fields_option_with_suffix(key, IGNORE_RETRACT_SUFFIX)
+        || is_fields_option_with_suffix(key, DISTINCT_SUFFIX)
+        || is_fields_option_with_suffix(key, SEQUENCE_GROUP_SUFFIX)
+        || is_fields_option_with_suffix(key, NESTED_KEY_SUFFIX)
+        || is_fields_option_with_suffix(key, COUNT_LIMIT_SUFFIX)
+}
+
+fn is_fields_option_with_suffix(key: &str, suffix: &str) -> bool {
+    key.starts_with(FIELDS_PREFIX) && key.ends_with(suffix)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn aggregation_options(extra: &[(&str, &str)]) -> HashMap<String, String> {
+        let mut options = HashMap::from([(
+            MERGE_ENGINE_OPTION.to_string(),
+            AGGREGATION_ENGINE.to_string(),
+        )]);
+        options.extend(
+            extra
+                .iter()
+                .map(|(key, value)| ((*key).to_string(), (*value).to_string())),
+        );
+        options
+    }
+
+    #[test]
+    fn test_validate_create_mode_accepts_basic_pk_aggregation() {
+        let options = aggregation_options(&[
+            ("fields.price.aggregate-function", "sum"),
+            ("fields.default-aggregate-function", "last_non_null_value"),
+            ("fields.tags.list-agg-delimiter", ";"),
+        ]);
+        let config = AggregationConfig::new(&options);
+
+        assert_eq!(
+            config.validate_create_mode(true).unwrap(),
+            Some(AggregationMode::Basic)
+        );
+    }
+
+    #[test]
+    fn test_validate_create_mode_ignores_non_pk_tables() {
+        let options = aggregation_options(&[("fields.x.ignore-retract", "true")]);
+        let config = AggregationConfig::new(&options);
+
+        assert_eq!(config.validate_create_mode(false).unwrap(), None);
+    }
+
+    #[test]
+    fn test_is_enabled_disabled_for_other_engines() {
+        let options = HashMap::from([(MERGE_ENGINE_OPTION.to_string(), "partial-update".into())]);
+        let config = AggregationConfig::new(&options);
+        assert!(!config.is_enabled());
+        assert_eq!(config.validate_create_mode(true).unwrap(), None);
+    }
+
+    #[test]
+    fn test_validate_create_mode_rejects_unsupported_options() {
+        for key in [
+            IGNORE_DELETE_OPTION,
+            "fields.price.ignore-delete",
+            AGGREGATION_REMOVE_RECORD_ON_DELETE_OPTION,
+            "fields.price.ignore-retract",
+            "fields.tags.distinct",
+            "fields.price.sequence-group",
+            "fields.payload.nested-key",
+            "fields.payload.count-limit",
+        ] {
+            let options = aggregation_options(&[(key, "value")]);
+            let config = AggregationConfig::new(&options);
+            let err = config.validate_create_mode(true).unwrap_err();
+            assert!(
+                matches!(err, crate::Error::ConfigInvalid { ref message } if message.contains(key)),
+                "expected create-time rejection to mention '{key}', got {err:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn test_validate_runtime_mode_rejects_unsupported_options() {
+        let options = aggregation_options(&[("fields.price.ignore-retract", "true")]);
+        let config = AggregationConfig::new(&options);
+        let err = config.validate_runtime_mode(true, "default.t").unwrap_err();
+
+        assert!(
+            matches!(err, crate::Error::Unsupported { ref message } if message.contains("fields.price.ignore-retract")),
+            "expected runtime rejection to mention the unsupported option, got {err:?}"
+        );
+    }
+}

crates/paimon/src/spec/core_options.rs

@@ -73,6 +73,8 @@ pub enum MergeEngine {
     PartialUpdate,
     /// Keep the first row for each key (ignore later updates).
     FirstRow,
+    /// Apply per-field aggregate functions across rows sharing the same key.
+    Aggregation,
 }
 
 /// Format the bucket directory name for a given bucket number.
@@ -131,6 +133,7 @@ impl<'a> CoreOptions<'a> {
                 "deduplicate" => Ok(MergeEngine::Deduplicate),
                 "partial-update" => Ok(MergeEngine::PartialUpdate),
                 "first-row" => Ok(MergeEngine::FirstRow),
+                "aggregation" => Ok(MergeEngine::Aggregation),
                 other => Err(crate::Error::Unsupported {
                     message: format!("Unsupported merge-engine: '{other}'"),
                 }),
@@ -546,6 +549,14 @@ mod tests {
         assert_eq!(core.merge_engine().unwrap(), MergeEngine::PartialUpdate);
     }
 
+    #[test]
+    fn test_merge_engine_accepts_aggregation() {
+        let options = HashMap::from([(MERGE_ENGINE_OPTION.to_string(), "aggregation".into())]);
+        let core = CoreOptions::new(&options);
+
+        assert_eq!(core.merge_engine().unwrap(), MergeEngine::Aggregation);
+    }
+
     #[test]
     fn test_commit_options_defaults() {
         let options = HashMap::new();

crates/paimon/src/spec/mod.rs

@@ -35,6 +35,9 @@ pub use core_options::*;
 mod partial_update;
 pub(crate) use partial_update::PartialUpdateConfig;
 
+mod aggregation;
+pub(crate) use aggregation::AggregationConfig;
+
 mod schema;
 pub use schema::*;
 

crates/paimon/src/spec/schema.rs

@@ -17,6 +17,7 @@
 
 use crate::spec::core_options::CoreOptions;
 use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType};
+use crate::spec::AggregationConfig;
 use crate::spec::PartialUpdateConfig;
 use serde::{Deserialize, Serialize};
 use serde_with::serde_as;
@@ -291,6 +292,7 @@ impl Schema {
         let fields = Self::normalize_fields(&fields, &partition_keys, &primary_keys)?;
         Self::validate_blob_fields(&fields, &partition_keys, &options)?;
         PartialUpdateConfig::new(&options).validate_create_mode(!primary_keys.is_empty())?;
+        AggregationConfig::new(&options).validate_create_mode(!primary_keys.is_empty())?;
 
         Ok(Self {
             fields,
@@ -714,7 +716,7 @@ impl Default for SchemaBuilder {
 
 #[cfg(test)]
 mod tests {
-    use crate::spec::{BlobType, IntType};
+    use crate::spec::{BlobType, IntType, VarCharType};
 
     use super::*;
 
@@ -978,6 +980,51 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_aggregation_schema_validation_accepts_basic_options() {
+        let schema = Schema::builder()
+            .column("id", DataType::Int(IntType::new()))
+            .column("value", DataType::Int(IntType::new()))
+            .column("tags", DataType::VarChar(VarCharType::new(255).unwrap()))
+            .primary_key(["id"])
+            .option("merge-engine", "aggregation")
+            .option("fields.value.aggregate-function", "sum")
+            .option("fields.tags.aggregate-function", "listagg")
+            .option("fields.tags.list-agg-delimiter", ";")
+            .option("fields.default-aggregate-function", "last_non_null_value")
+            .build()
+            .unwrap();
+
+        assert_eq!(schema.fields().len(), 3);
+    }
+
+    #[test]
+    fn test_aggregation_schema_validation_rejects_unsupported_options() {
+        for (key, value) in [
+            ("ignore-delete", "true"),
+            ("aggregation.remove-record-on-delete", "true"),
+            ("fields.value.ignore-retract", "true"),
+            ("fields.value.distinct", "true"),
+            ("fields.value.sequence-group", "g1"),
+            ("fields.value.nested-key", "id"),
+            ("fields.value.count-limit", "10"),
+        ] {
+            let err = Schema::builder()
+                .column("id", DataType::Int(IntType::new()))
+                .column("value", DataType::Int(IntType::new()))
+                .primary_key(["id"])
+                .option("merge-engine", "aggregation")
+                .option(key, value)
+                .build()
+                .unwrap_err();
+
+            assert!(
+                matches!(err, crate::Error::ConfigInvalid { ref message } if message.contains(key)),
+                "aggregation create-time validation should reject '{key}', got {err:?}"
+            );
+        }
+    }
+
     #[test]
     fn test_schema_builder_column_row_type() {
         let row_type = RowType::new(vec![DataField::new(