feat: support fixed-bucket partial-update merge engine

crates/integrations/datafusion/tests/pk_tables.rs

@@ -30,7 +30,7 @@ mod common;
 use common::{
     collect_id_name, collect_id_value, create_handler, create_test_env, row_count, setup_handler,
 };
-use datafusion::arrow::array::{Int32Array, StringArray};
+use datafusion::arrow::array::{Array, Int32Array, StringArray};
 use paimon::catalog::Identifier;
 use paimon::Catalog;
 
@@ -75,6 +75,106 @@ async fn test_pk_basic_write_read() {
     );
 }
 
+/// Partial-update merge engine: keep latest non-null value for each field.
+#[tokio::test]
+async fn test_pk_partial_update_fixed_bucket_e2e() {
+    let (_tmp, handler) = setup_handler().await;
+
+    handler
+        .sql(
+            "CREATE TABLE paimon.test_db.t_partial_update (
+                id INT NOT NULL, v_int INT, v_str STRING,
+                PRIMARY KEY (id)
+            ) WITH ('bucket' = '1', 'merge-engine' = 'partial-update')",
+        )
+        .await
+        .unwrap();
+
+    handler
+        .sql(
+            "INSERT INTO paimon.test_db.t_partial_update VALUES
+             (1, 10, 'old-1'),
+             (2, 20, 'old-2')",
+        )
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    handler
+        .sql(
+            "INSERT INTO paimon.test_db.t_partial_update VALUES
+             (1, CAST(NULL AS INT), 'new-1'),
+             (2, 200, CAST(NULL AS STRING)),
+             (3, 30, CAST(NULL AS STRING))",
+        )
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    handler
+        .sql(
+            "INSERT INTO paimon.test_db.t_partial_update VALUES
+             (1, 111, CAST(NULL AS STRING))",
+        )
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    let batches = handler
+        .sql("SELECT id, v_int, v_str FROM paimon.test_db.t_partial_update ORDER BY id")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    let mut rows = Vec::new();
+    for batch in &batches {
+        let ids = batch
+            .column_by_name("id")
+            .and_then(|c| c.as_any().downcast_ref::<Int32Array>())
+            .unwrap();
+        let ints = batch
+            .column_by_name("v_int")
+            .and_then(|c| c.as_any().downcast_ref::<Int32Array>())
+            .unwrap();
+        let strs = batch
+            .column_by_name("v_str")
+            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
+            .unwrap();
+        for i in 0..batch.num_rows() {
+            rows.push((
+                ids.value(i),
+                if ints.is_null(i) {
+                    None
+                } else {
+                    Some(ints.value(i))
+                },
+                if strs.is_null(i) {
+                    None
+                } else {
+                    Some(strs.value(i).to_string())
+                },
+            ));
+        }
+    }
+
+    assert_eq!(
+        rows,
+        vec![
+            (1, Some(111), Some("new-1".to_string())),
+            (2, Some(200), Some("old-2".to_string())),
+            (3, Some(30), None),
+        ]
+    );
+}
+
 // ======================= Dedup Within Single Commit =======================
 
 /// Duplicate keys in a single INSERT — last value wins (Deduplicate engine).

crates/paimon/src/spec/core_options.rs

@@ -66,6 +66,8 @@ const DEFAULT_DYNAMIC_BUCKET_TARGET_ROW_NUM: i64 = 200_000;
 pub enum MergeEngine {
     /// Keep the row with the highest sequence number (default).
     Deduplicate,
+    /// Merge same-key rows field-by-field, usually keeping non-null updates.
+    PartialUpdate,
     /// Keep the first row for each key (ignore later updates).
     FirstRow,
 }
@@ -124,6 +126,7 @@ impl<'a> CoreOptions<'a> {
             None => Ok(MergeEngine::Deduplicate),
             Some(v) => match v.to_ascii_lowercase().as_str() {
                 "deduplicate" => Ok(MergeEngine::Deduplicate),
+                "partial-update" => Ok(MergeEngine::PartialUpdate),
                 "first-row" => Ok(MergeEngine::FirstRow),
                 other => Err(crate::Error::Unsupported {
                     message: format!("Unsupported merge-engine: '{other}'"),
@@ -498,6 +501,14 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_merge_engine_accepts_partial_update() {
+        let options = HashMap::from([(MERGE_ENGINE_OPTION.to_string(), "partial-update".into())]);
+        let core = CoreOptions::new(&options);
+
+        assert_eq!(core.merge_engine().unwrap(), MergeEngine::PartialUpdate);
+    }
+
     #[test]
     fn test_commit_options_defaults() {
         let options = HashMap::new();

crates/paimon/src/spec/mod.rs

@@ -29,6 +29,9 @@ mod core_options;
 pub(crate) use core_options::TimeTravelSelector;
 pub use core_options::*;
 
+mod partial_update;
+pub(crate) use partial_update::PartialUpdateConfig;
+
 mod schema;
 pub use schema::*;
 

crates/paimon/src/spec/partial_update.rs

@@ -0,0 +1,202 @@
+// 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 PARTIAL_UPDATE_ENGINE: &str = "partial-update";
+const IGNORE_DELETE_OPTION: &str = "ignore-delete";
+const IGNORE_DELETE_SUFFIX: &str = ".ignore-delete";
+const PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION: &str =
+    "partial-update.remove-record-on-delete";
+const PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP_OPTION: &str =
+    "partial-update.remove-record-on-sequence-group";
+const FIELDS_DEFAULT_AGG_FUNCTION_OPTION: &str = "fields.default-aggregate-function";
+const FIELDS_PREFIX: &str = "fields.";
+const SEQUENCE_GROUP_SUFFIX: &str = ".sequence-group";
+const AGGREGATION_FUNCTION_SUFFIX: &str = ".aggregate-function";
+
+/// Minimal partial-update mode recognized by the current Rust implementation.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum PartialUpdateMode {
+    Basic,
+}
+
+/// Partial-update-specific option inspection and validation.
+///
+/// PR1 only recognizes the basic mode: `merge-engine=partial-update` on a PK
+/// table without delete, sequence-group, or aggregation controls.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct PartialUpdateConfig<'a> {
+    options: &'a HashMap<String, String>,
+}
+
+impl<'a> PartialUpdateConfig<'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(PARTIAL_UPDATE_ENGINE))
+    }
+
+    pub(crate) fn validate_create_mode(
+        &self,
+        has_primary_keys: bool,
+    ) -> crate::Result<Option<PartialUpdateMode>> {
+        match self.validated_mode(has_primary_keys) {
+            Ok(mode) => Ok(mode),
+            Err(unsupported_options) => Err(crate::Error::ConfigInvalid {
+                message: format!(
+                    "merge-engine=partial-update only supports the basic mode in this build; unsupported options: {}",
+                    unsupported_options.join(", ")
+                ),
+            }),
+        }
+    }
+
+    pub(crate) fn validate_runtime_mode(
+        &self,
+        has_primary_keys: bool,
+        table_name: &str,
+    ) -> crate::Result<Option<PartialUpdateMode>> {
+        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=partial-update options not supported by this build: {}",
+                    unsupported_options.join(", ")
+                ),
+            }),
+        }
+    }
+
+    fn validated_mode(
+        &self,
+        has_primary_keys: bool,
+    ) -> std::result::Result<Option<PartialUpdateMode>, 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(PartialUpdateMode::Basic))
+    }
+
+    fn unsupported_option_keys(&self) -> Vec<String> {
+        let mut keys: Vec<String> = self
+            .options
+            .keys()
+            .filter(|key| is_unsupported_partial_update_option(key))
+            .cloned()
+            .collect();
+        keys.sort();
+        keys
+    }
+}
+
+fn is_unsupported_partial_update_option(key: &str) -> bool {
+    key == IGNORE_DELETE_OPTION
+        || key.ends_with(IGNORE_DELETE_SUFFIX)
+        || key == PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION
+        || key == PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP_OPTION
+        || key == FIELDS_DEFAULT_AGG_FUNCTION_OPTION
+        || is_fields_option_with_suffix(key, SEQUENCE_GROUP_SUFFIX)
+        || is_fields_option_with_suffix(key, AGGREGATION_FUNCTION_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 partial_update_options(extra: &[(&str, &str)]) -> HashMap<String, String> {
+        let mut options = HashMap::from([(
+            MERGE_ENGINE_OPTION.to_string(),
+            PARTIAL_UPDATE_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_partial_update() {
+        let options = partial_update_options(&[]);
+        let config = PartialUpdateConfig::new(&options);
+
+        assert_eq!(
+            config.validate_create_mode(true).unwrap(),
+            Some(PartialUpdateMode::Basic)
+        );
+    }
+
+    #[test]
+    fn test_validate_create_mode_ignores_non_pk_tables() {
+        let options = partial_update_options(&[(IGNORE_DELETE_OPTION, "true")]);
+        let config = PartialUpdateConfig::new(&options);
+
+        assert_eq!(config.validate_create_mode(false).unwrap(), None);
+    }
+
+    #[test]
+    fn test_validate_create_mode_rejects_unsupported_partial_update_options() {
+        for key in [
+            IGNORE_DELETE_OPTION,
+            "partial-update.ignore-delete",
+            PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION,
+            PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP_OPTION,
+            "fields.price.sequence-group",
+            "fields.price.aggregate-function",
+            FIELDS_DEFAULT_AGG_FUNCTION_OPTION,
+        ] {
+            let options = partial_update_options(&[(key, "value")]);
+            let config = PartialUpdateConfig::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_partial_update_options() {
+        let options =
+            partial_update_options(&[("fields.price.aggregate-function", "last_non_null")]);
+        let config = PartialUpdateConfig::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.aggregate-function")),
+            "expected runtime rejection to mention the unsupported option, got {err:?}"
+        );
+    }
+}