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

crates/paimon/src/spec/core_options.rs

@@ -78,6 +78,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,
 }
 
 /// Changelog producer for table writes.
@@ -169,6 +171,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}'"),
                 }),
@@ -637,6 +640,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_changelog_producer_defaults_to_none() {
         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::{first_row_supports_changelog_producer, 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;
@@ -293,6 +294,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(), &fields)?;
         Self::validate_first_row_changelog_producer(&options)?;
 
         Ok(Self {
@@ -751,7 +753,7 @@ impl Default for SchemaBuilder {
 
 #[cfg(test)]
 mod tests {
-    use crate::spec::{BlobType, IntType};
+    use crate::spec::{BlobType, IntType, VarCharType};
 
     use super::*;
 
@@ -1015,6 +1017,107 @@ 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_aggregation_schema_validation_rejects_unknown_field() {
+        let err = Schema::builder()
+            .column("id", DataType::Int(IntType::new()))
+            .column("amount", DataType::Int(IntType::new()))
+            .primary_key(["id"])
+            .option("merge-engine", "aggregation")
+            // typo: `amout` instead of `amount`
+            .option("fields.amout.aggregate-function", "sum")
+            .build()
+            .unwrap_err();
+
+        assert!(
+            matches!(err, crate::Error::ConfigInvalid { ref message }
+                if message.contains("amout") && message.contains("amount")),
+            "expected unknown-field rejection at CREATE TABLE, got {err:?}"
+        );
+    }
+
+    #[test]
+    fn test_aggregation_schema_validation_rejects_unknown_function() {
+        let err = Schema::builder()
+            .column("id", DataType::Int(IntType::new()))
+            .column("amount", DataType::Int(IntType::new()))
+            .primary_key(["id"])
+            .option("merge-engine", "aggregation")
+            .option("fields.amount.aggregate-function", "sume")
+            .build()
+            .unwrap_err();
+
+        assert!(
+            matches!(err, crate::Error::ConfigInvalid { ref message }
+                if message.contains("sume")),
+            "expected unknown-function rejection at CREATE TABLE, got {err:?}"
+        );
+    }
+
+    #[test]
+    fn test_aggregation_schema_validation_rejects_incompatible_function_type() {
+        let err = Schema::builder()
+            .column("id", DataType::Int(IntType::new()))
+            .column("tag", DataType::VarChar(VarCharType::new(255).unwrap()))
+            .primary_key(["id"])
+            .option("merge-engine", "aggregation")
+            // sum on a VarChar column
+            .option("fields.tag.aggregate-function", "sum")
+            .build()
+            .unwrap_err();
+
+        assert!(
+            matches!(err, crate::Error::ConfigInvalid { ref message }
+                if message.contains("sum") && message.contains("tag")),
+            "expected incompatible-type rejection at CREATE TABLE, got {err:?}"
+        );
+    }
+
     #[test]
     fn test_first_row_schema_validation_accepts_supported_changelog_producers() {
         for producer in ["none", "lookup"] {

crates/paimon/src/table/aggregator/bool_agg.rs

@@ -0,0 +1,216 @@
+// 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.
+
+//! Boolean reducers: `bool_and` and `bool_or`.
+//!
+//! Both accumulate over BOOLEAN columns, skipping NULL inputs.  The output
+//! cell is NULL when no non-NULL input was observed for the key.
+//!
+//! Reference: Java `FieldBoolAndAgg`, `FieldBoolOrAgg` under
+//! `org.apache.paimon.mergetree.compact.aggregate`.
+
+use std::sync::Arc;
+
+use arrow_array::{Array, ArrayRef, BooleanArray};
+
+use super::{unsupported_type_error, FieldAggregator};
+use crate::spec::DataType;
+
+fn ensure_boolean(field_name: &str, data_type: &DataType, op: &str) -> crate::Result<()> {
+    match data_type {
+        DataType::Boolean(_) => Ok(()),
+        other => Err(unsupported_type_error(op, field_name, other)),
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct BoolAndAgg {
+    field_name: String,
+    acc: Option<bool>,
+}
+
+impl BoolAndAgg {
+    pub(crate) fn new(field_name: &str, data_type: &DataType) -> crate::Result<Self> {
+        ensure_boolean(field_name, data_type, "bool_and")?;
+        Ok(Self {
+            field_name: field_name.to_string(),
+            acc: None,
+        })
+    }
+}
+
+impl FieldAggregator for BoolAndAgg {
+    fn name(&self) -> &'static str {
+        "bool_and"
+    }
+
+    fn reset(&mut self) {
+        self.acc = None;
+    }
+
+    fn agg(&mut self, array: &dyn Array, row_idx: usize) -> crate::Result<()> {
+        if array.is_null(row_idx) {
+            return Ok(());
+        }
+        let arr = array
+            .as_any()
+            .downcast_ref::<BooleanArray>()
+            .ok_or_else(|| crate::Error::DataInvalid {
+                message: format!(
+                    "bool_and column '{}' received non-Boolean Arrow array {:?}",
+                    self.field_name,
+                    array.data_type()
+                ),
+                source: None,
+            })?;
+        let v = arr.value(row_idx);
+        self.acc = Some(self.acc.map_or(v, |prev| prev && v));
+        Ok(())
+    }
+
+    fn result(&self) -> crate::Result<ArrayRef> {
+        Ok(Arc::new(BooleanArray::from(vec![self.acc])))
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct BoolOrAgg {
+    field_name: String,
+    acc: Option<bool>,
+}
+
+impl BoolOrAgg {
+    pub(crate) fn new(field_name: &str, data_type: &DataType) -> crate::Result<Self> {
+        ensure_boolean(field_name, data_type, "bool_or")?;
+        Ok(Self {
+            field_name: field_name.to_string(),
+            acc: None,
+        })
+    }
+}
+
+impl FieldAggregator for BoolOrAgg {
+    fn name(&self) -> &'static str {
+        "bool_or"
+    }
+
+    fn reset(&mut self) {
+        self.acc = None;
+    }
+
+    fn agg(&mut self, array: &dyn Array, row_idx: usize) -> crate::Result<()> {
+        if array.is_null(row_idx) {
+            return Ok(());
+        }
+        let arr = array
+            .as_any()
+            .downcast_ref::<BooleanArray>()
+            .ok_or_else(|| crate::Error::DataInvalid {
+                message: format!(
+                    "bool_or column '{}' received non-Boolean Arrow array {:?}",
+                    self.field_name,
+                    array.data_type()
+                ),
+                source: None,
+            })?;
+        let v = arr.value(row_idx);
+        self.acc = Some(self.acc.map_or(v, |prev| prev || v));
+        Ok(())
+    }
+
+    fn result(&self) -> crate::Result<ArrayRef> {
+        Ok(Arc::new(BooleanArray::from(vec![self.acc])))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::spec::{BooleanType, IntType};
+
+    fn collect(arr: ArrayRef) -> Option<bool> {
+        let a = arr.as_any().downcast_ref::<BooleanArray>().unwrap();
+        if a.is_null(0) {
+            None
+        } else {
+            Some(a.value(0))
+        }
+    }
+
+    #[test]
+    fn test_bool_and_all_true_returns_true() {
+        let mut agg = BoolAndAgg::new("b", &DataType::Boolean(BooleanType::new())).unwrap();
+        let arr = BooleanArray::from(vec![Some(true), Some(true), None, Some(true)]);
+        for i in 0..arr.len() {
+            agg.agg(&arr, i).unwrap();
+        }
+        assert_eq!(collect(agg.result().unwrap()), Some(true));
+    }
+
+    #[test]
+    fn test_bool_and_short_circuits_false() {
+        let mut agg = BoolAndAgg::new("b", &DataType::Boolean(BooleanType::new())).unwrap();
+        let arr = BooleanArray::from(vec![Some(true), Some(false), Some(true)]);
+        for i in 0..arr.len() {
+            agg.agg(&arr, i).unwrap();
+        }
+        assert_eq!(collect(agg.result().unwrap()), Some(false));
+    }
+
+    #[test]
+    fn test_bool_or_any_true() {
+        let mut agg = BoolOrAgg::new("b", &DataType::Boolean(BooleanType::new())).unwrap();
+        let arr = BooleanArray::from(vec![Some(false), None, Some(true), Some(false)]);
+        for i in 0..arr.len() {
+            agg.agg(&arr, i).unwrap();
+        }
+        assert_eq!(collect(agg.result().unwrap()), Some(true));
+    }
+
+    #[test]
+    fn test_bool_and_or_all_null_returns_null() {
+        let mut and_agg = BoolAndAgg::new("b", &DataType::Boolean(BooleanType::new())).unwrap();
+        let arr = BooleanArray::from(vec![None::<bool>, None]);
+        for i in 0..arr.len() {
+            and_agg.agg(&arr, i).unwrap();
+        }
+        assert_eq!(collect(and_agg.result().unwrap()), None);
+
+        let mut or_agg = BoolOrAgg::new("b", &DataType::Boolean(BooleanType::new())).unwrap();
+        for i in 0..arr.len() {
+            or_agg.agg(&arr, i).unwrap();
+        }
+        assert_eq!(collect(or_agg.result().unwrap()), None);
+    }
+
+    #[test]
+    fn test_bool_and_rejects_non_boolean_type() {
+        let err = BoolAndAgg::new("b", &DataType::Int(IntType::new())).unwrap_err();
+        assert!(
+            matches!(err, crate::Error::ConfigInvalid { message } if message.contains("bool_and"))
+        );
+    }
+
+    #[test]
+    fn test_reset_clears_state() {
+        let mut agg = BoolOrAgg::new("b", &DataType::Boolean(BooleanType::new())).unwrap();
+        let arr = BooleanArray::from(vec![Some(true)]);
+        agg.agg(&arr, 0).unwrap();
+        agg.reset();
+        assert_eq!(collect(agg.result().unwrap()), None);
+    }
+}