Skip to content

feat(partial-update): support ignore-delete semantics #573

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Jul 21, 2026
Merged

feat(partial-update): support ignore-delete semantics #573

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 70 additions & 1 deletion crates/integrations/datafusion/tests/pk_tables.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,9 +32,13 @@ use common::{
collect_id_name, collect_id_value, collect_int_int_str, create_sql_context, create_test_env,
row_count, setup_sql_context, string_value,
};
use datafusion::arrow::array::{Array, Int32Array, Int64Array};
use datafusion::arrow::array::{Array, Int32Array, Int64Array, Int8Array, RecordBatch};
use datafusion::arrow::datatypes::{
DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema,
};
use paimon::catalog::Identifier;
use paimon::Catalog;
use std::sync::Arc;

// ======================= Basic PK Write + Read =======================

Expand Down Expand Up @@ -174,6 +178,71 @@ async fn test_pk_partial_update_fixed_bucket_e2e() {
);
}

#[tokio::test]
async fn test_pk_partial_update_ignore_delete_alias_e2e() {
let (_tmp, catalog) = create_test_env();
let sql_context = create_sql_context(catalog.clone()).await;
sql_context
.sql("CREATE SCHEMA paimon.test_db")
.await
.unwrap();
sql_context
.sql(
"CREATE TABLE paimon.test_db.t_partial_update_ignore_delete (
id INT NOT NULL, value INT,
PRIMARY KEY (id)
) WITH (
'bucket' = '1',
'merge-engine' = 'partial-update',
'partial-update.ignore-delete' = 'true'
)",
)
.await
.unwrap();

let table = catalog
.get_table(&Identifier::new(
"test_db",
"t_partial_update_ignore_delete",
))
.await
.unwrap();
let batch = RecordBatch::try_new(
Arc::new(ArrowSchema::new(vec![
ArrowField::new("id", ArrowDataType::Int32, false),
ArrowField::new("value", ArrowDataType::Int32, true),
ArrowField::new("_VALUE_KIND", ArrowDataType::Int8, false),
])),
vec![
Arc::new(Int32Array::from(vec![1, 1, 2, 1])),
Arc::new(Int32Array::from(vec![
Some(10),
Some(999),
Some(200),
Some(20),
])),
Arc::new(Int8Array::from(vec![0, 3, 1, 2])),
],
)
.unwrap();
let write_builder = table.new_write_builder();
let mut write = write_builder.new_write().unwrap();
write.write_arrow_batch(&batch).await.unwrap();
let messages = write.prepare_commit().await.unwrap();
write_builder.new_commit().commit(messages).await.unwrap();

assert_eq!(
collect_id_value(
&sql_context,
"SELECT id, value
FROM paimon.test_db.t_partial_update_ignore_delete
ORDER BY id",
)
.await,
vec![(1, 20)]
);
}

/// Partial updates of one key within a single INSERT are merged at flush
/// (mirrors Java MergeTreeWriter#flushWriteBuffer): the flushed file holds
/// one row per key, so SELECT and COUNT(*) agree.
Expand Down
17 changes: 17 additions & 0 deletions crates/paimon/src/spec/core_options.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1966,9 +1966,26 @@ mod tests {
let opts = CoreOptions::new(&fallback);
assert!(opts.ignore_delete());

let partial_update = HashMap::from([(
"partial-update.ignore-delete".to_string(),
"true".to_string(),
)]);
let opts = CoreOptions::new(&partial_update);
assert!(opts.ignore_delete());

let primary = HashMap::from([("ignore-delete".to_string(), "true".to_string())]);
let opts = CoreOptions::new(&primary);
assert!(opts.ignore_delete());

let primary_precedence = HashMap::from([
("ignore-delete".to_string(), "false".to_string()),
(
"partial-update.ignore-delete".to_string(),
"true".to_string(),
),
]);
let opts = CoreOptions::new(&primary_precedence);
assert!(!opts.ignore_delete());
}

#[test]
Expand Down
36 changes: 32 additions & 4 deletions crates/paimon/src/spec/partial_update.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ 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_IGNORE_DELETE_OPTION: &str = "partial-update.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 =
Expand Down Expand Up @@ -116,8 +117,9 @@ impl<'a> PartialUpdateConfig<'a> {
}

fn is_unsupported_partial_update_option(key: &str) -> bool {
key == IGNORE_DELETE_OPTION
|| key.ends_with(IGNORE_DELETE_SUFFIX)
(key.ends_with(IGNORE_DELETE_SUFFIX)
&& key != IGNORE_DELETE_OPTION
&& key != PARTIAL_UPDATE_IGNORE_DELETE_OPTION)
|| key == PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION
|| key == PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP_OPTION
|| key == FIELDS_DEFAULT_AGG_FUNCTION_OPTION
Expand Down Expand Up @@ -157,6 +159,32 @@ mod tests {
);
}

#[test]
fn test_validate_create_mode_accepts_partial_update_ignore_delete() {
for value in ["true", "false"] {
let options = partial_update_options(&[(PARTIAL_UPDATE_IGNORE_DELETE_OPTION, value)]);
let config = PartialUpdateConfig::new(&options);

assert_eq!(
config.validate_create_mode(true).unwrap(),
Some(PartialUpdateMode::Basic)
);
}
}

#[test]
fn test_validate_create_mode_accepts_ignore_delete() {
for value in ["true", "false"] {
let options = partial_update_options(&[(IGNORE_DELETE_OPTION, value)]);
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")]);
Expand All @@ -168,10 +196,10 @@ mod tests {
#[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,
"deduplicate.ignore-delete",
"fields.price.ignore-delete",
"fields.price.sequence-group",
"fields.price.aggregate-function",
FIELDS_DEFAULT_AGG_FUNCTION_OPTION,
Expand Down
97 changes: 88 additions & 9 deletions crates/paimon/src/spec/schema.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -453,6 +453,15 @@ impl TableSchema {
new_schema.highest_field_id =
highest_field_id.max(Self::current_highest_field_id(&new_schema.fields));

if PartialUpdateConfig::new(&new_schema.options).is_enabled()

@jerry-024 jerry-024 Jul 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[major] Preserve the one-way ignore-delete invariant across merge-engine changes

This guard only runs when the resulting schema is still partial-update, so the invariant can be bypassed across schema versions:

  1. start with merge-engine=partial-update and ignore-delete=true
  2. set merge-engine=deduplicate (this guard is disabled)
  3. set ignore-delete=false or remove it (this guard is still disabled)
  4. set merge-engine=partial-update again (the previous effective value is already false)

All calls to apply_changes succeed, and historical DELETE / UPDATE_BEFORE rows can become unreadable again. The Java SchemaManager.checkAlterTableOption enforces the true -> false restriction on the option itself, independent of merge engine.

Please preserve the effective true state across engine changes, or reject disabling/removing it whenever the current schema has it enabled. A regression test covering the four-step transition above would catch this.

&& self.core_options().ignore_delete()
&& !CoreOptions::new(&new_schema.options).ignore_delete()
{
return Err(crate::Error::Unsupported {
message: "Cannot change ignore-delete from true to false.".to_string(),
});
}

// Re-run create-time validations on the final schema, mirroring Java
// `SchemaValidation.validateTableSchema` after applying changes.
Schema::validate_key_field_types(
Expand Down Expand Up @@ -2099,7 +2108,7 @@ mod tests {
#[test]
fn test_partial_update_schema_validation_rejects_unsupported_options() {
for (key, value) in [
("ignore-delete", "true"),
("fields.value.ignore-delete", "true"),
("fields.value.sequence-group", "g1"),
("fields.default-aggregate-function", "last_non_null"),
] {
Expand All @@ -2119,6 +2128,27 @@ mod tests {
}
}

#[test]
fn test_partial_update_schema_validation_accepts_ignore_delete_options() {
for (key, value) in [
("ignore-delete", "true"),
("ignore-delete", "false"),
("partial-update.ignore-delete", "true"),
("partial-update.ignore-delete", "false"),
] {
let schema = Schema::builder()
.column("id", DataType::Int(IntType::new()))
.column("value", DataType::Int(IntType::new()))
.primary_key(["id"])
.option("merge-engine", "partial-update")
.option(key, value)
.build()
.unwrap();

assert_eq!(schema.options().get(key).map(String::as_str), Some(value));
}
}

#[test]
fn test_aggregation_schema_validation_accepts_basic_options() {
let schema = Schema::builder()
Expand Down Expand Up @@ -2709,7 +2739,7 @@ mod tests {
}

#[test]
fn test_partial_update_apply_changes_rejects_unsupported_option() {
fn test_partial_update_apply_changes_accepts_ignore_delete_option() {
let table_schema = TableSchema::new(
0,
&Schema::builder()
Expand All @@ -2721,21 +2751,70 @@ mod tests {
.unwrap(),
);

let err = table_schema
let new_schema = table_schema
.apply_changes(vec![crate::spec::SchemaChange::set_option(
"ignore-delete".to_string(),
"true".to_string(),
)])
.unwrap_err();
.unwrap();

assert!(
matches!(err, crate::Error::ConfigInvalid { ref message }
if message.contains("merge-engine=partial-update")
&& message.contains("ignore-delete")),
"partial-update alter should reject unsupported option, got {err:?}"
assert_eq!(
new_schema
.options()
.get("ignore-delete")
.map(String::as_str),
Some("true")
);
}

#[test]
fn test_partial_update_apply_changes_rejects_disabling_ignore_delete() {
for (existing_options, change) in [
(
vec![("ignore-delete", "true")],
crate::spec::SchemaChange::set_option(
"ignore-delete".to_string(),
"false".to_string(),
),
),
(
vec![("ignore-delete", "true")],
crate::spec::SchemaChange::remove_option("ignore-delete".to_string()),
),
(
vec![("partial-update.ignore-delete", "true")],
crate::spec::SchemaChange::set_option(
"ignore-delete".to_string(),
"false".to_string(),
),
),
(
vec![("partial-update.ignore-delete", "true")],
crate::spec::SchemaChange::remove_option(
"partial-update.ignore-delete".to_string(),
),
),
] {
let mut builder = Schema::builder()
.column("id", DataType::Int(IntType::new()))
.column("value", DataType::Int(IntType::new()))
.primary_key(["id"])
.option("merge-engine", "partial-update");
for (key, value) in existing_options {
builder = builder.option(key, value);
}
let table_schema = TableSchema::new(0, &builder.build().unwrap());

let err = table_schema.apply_changes(vec![change]).unwrap_err();

assert!(
matches!(err, crate::Error::Unsupported { ref message }
if message.contains("Cannot change ignore-delete from true to false")),
"got {err:?}"
);
}
}

#[test]
fn test_aggregation_apply_changes_accepts_valid_option() {
let table_schema = TableSchema::new(
Expand Down
Loading
Loading