This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 86adc63f [core] Support configurable sort order for partial-update
sequence groups (#1226)
86adc63f is described below
commit 86adc63f54d34f1889a1aa33513006ff1899c0ed
Author: Ran Tao <[email protected]>
AuthorDate: Tue Oct 27 16:10:21 2026 +0800
[core] Support configurable sort order for partial-update sequence groups
(#1226)
---
bindings/python/tests/test_write.py | 54 ++++++
crates/paimon/src/spec/aggregation.rs | 9 +-
crates/paimon/src/spec/partial_update.rs | 101 +++++++++++
crates/paimon/src/spec/schema.rs | 24 ++-
crates/paimon/src/table/kv_file_reader.rs | 110 +++++++++++
crates/paimon/src/table/kv_file_writer.rs | 292 ++++++++++++++++--------------
crates/paimon/src/table/sort_merge.rs | 159 +++++++++++++++-
docs/src/sql.md | 32 +++-
8 files changed, 626 insertions(+), 155 deletions(-)
diff --git a/bindings/python/tests/test_write.py
b/bindings/python/tests/test_write.py
index 14284b3f..4ee40bbf 100644
--- a/bindings/python/tests/test_write.py
+++ b/bindings/python/tests/test_write.py
@@ -64,6 +64,60 @@ def test_write_commit_read_roundtrip():
assert result == {"id": [1, 2, 3], "name": ["a", "b", "c"]}
[email protected]("direction", [None, "ascending", "descending",
"DeScEnDiNg"])
[email protected]("separate_commits", [False, True])
[email protected]("reverse_input", [False, True])
+def test_partial_update_sequence_group_sort_order(
+ tmp_path, direction, separate_commits, reverse_input):
+ # apache/paimon#10546: retain the latest status and the earliest channel.
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": str(tmp_path)})
+ ctx.sql("CREATE SCHEMA paimon.wdb")
+ sort_option = ("" if direction is None else
+ f", 'fields.first_event_time.sequence-group.sort-order' =
'{direction}'")
+ ctx.sql(
+ "CREATE TABLE paimon.wdb.t (user_id BIGINT, status STRING, "
+ "status_version BIGINT, channel STRING, first_event_time BIGINT, "
+ "PRIMARY KEY (user_id)) WITH ("
+ "'bucket' = '1', 'merge-engine' = 'partial-update', "
+ "'fields.status_version.sequence-group' = 'status', "
+ "'fields.first_event_time.sequence-group' = 'channel'" + sort_option +
")")
+ schema = pa.schema([
+ ("user_id", pa.int64()), ("status", pa.string()),
+ ("status_version", pa.int64()), ("channel", pa.string()),
+ ("first_event_time", pa.int64())])
+ rows = [dict(zip(schema.names, row)) for row in [
+ (1, "active", 200, "organic", 100),
+ (1, "registered", 100, "referral", 50)]]
+ if reverse_input:
+ rows.reverse()
+ table = _get_table(str(tmp_path))
+ for group in ([[row] for row in rows] if separate_commits else [rows]):
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(pa.RecordBatch.from_pylist(group,
schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+
+ descending = direction is not None and direction.lower() == "descending"
+ channel, event_time = ("referral", 50) if descending else ("organic", 100)
+ expected = [dict(zip(schema.names, (1, "active", 200, channel,
event_time)))]
+ assert pa.Table.from_batches(ctx.sql("SELECT * FROM
paimon.wdb.t")).to_pylist() == expected
+ # Both the native read builder and SQL projection must retain ordering
dependencies.
+ read_builder = _get_table(str(tmp_path)).new_read_builder()
+ read_builder.with_projection(["channel"])
+ batches =
read_builder.new_read().read(read_builder.new_scan().plan().splits())
+ assert pa.Table.from_batches(batches).to_pylist() == [{"channel": channel}]
+ assert pa.Table.from_batches(ctx.sql(
+ "SELECT channel FROM paimon.wdb.t")).to_pylist() == [{"channel":
channel}]
+ ctx.sql("ALTER TABLE paimon.wdb.t RENAME COLUMN first_event_time TO
event_time")
+ assert pa.Table.from_batches(ctx.sql(
+ "SELECT channel, event_time FROM paimon.wdb.t")).to_pylist() == [
+ {"channel": channel, "event_time": event_time}]
+
+
@pytest.mark.parametrize("streaming", [False, True])
def test_partial_write_type_roundtrip_and_lifecycle(tmp_path, streaming):
ctx = SQLContext()
diff --git a/crates/paimon/src/spec/aggregation.rs
b/crates/paimon/src/spec/aggregation.rs
index 9a05e68f..fd90da27 100644
--- a/crates/paimon/src/spec/aggregation.rs
+++ b/crates/paimon/src/spec/aggregation.rs
@@ -17,6 +17,7 @@
use std::collections::HashMap;
+use super::partial_update::SEQUENCE_GROUP_SORT_ORDER_SUFFIX;
use crate::spec::core_options::IGNORE_DELETE_FALLBACK_KEYS;
use crate::spec::{CoreOptions, DataField, DataType, VarCharType};
@@ -318,6 +319,7 @@ const FIELD_SCOPED_FIELD_LIST_SUFFIXES: [&str; 2] =
[SEQUENCE_GROUP_SUFFIX, NEST
/// - case 2: `fields.<old>.<suffix>` -> `fields.<new>.<suffix>`, value kept
/// - case 3 (`sequence-group`, `nested-key`): every comma-separated entry
/// equal to `old` is renamed in BOTH the key's field list and the value
+/// - `sequence-group.sort-order`: rename the key's field list only
pub(crate) fn rename_field_scoped_options(
options: &mut HashMap<String, String>,
old: &str,
@@ -336,13 +338,18 @@ pub(crate) fn rename_field_scoped_options(
for (key, value) in options.iter() {
let Some(suffix) = FIELD_SCOPED_FIELD_LIST_SUFFIXES
.iter()
+ .chain(std::iter::once(&SEQUENCE_GROUP_SORT_ORDER_SUFFIX))
.find(|suffix| key.starts_with(FIELDS_PREFIX) &&
key.ends_with(**suffix))
else {
continue;
};
let key_fields = &key[FIELDS_PREFIX.len()..key.len() - suffix.len()];
let new_key_fields = rename_in_field_list(key_fields, old, new);
- let new_value = rename_in_field_list(value, old, new);
+ let new_value = if *suffix == SEQUENCE_GROUP_SORT_ORDER_SUFFIX {
+ value.clone()
+ } else {
+ rename_in_field_list(value, old, new)
+ };
if new_key_fields != key_fields || new_value != *value {
rewritten.push((
key.clone(),
diff --git a/crates/paimon/src/spec/partial_update.rs
b/crates/paimon/src/spec/partial_update.rs
index 88e253bf..447b8123 100644
--- a/crates/paimon/src/spec/partial_update.rs
+++ b/crates/paimon/src/spec/partial_update.rs
@@ -32,6 +32,7 @@ const PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP_OPTION:
&str =
const FIELDS_DEFAULT_AGG_FUNCTION_OPTION: &str =
"fields.default-aggregate-function";
const FIELDS_PREFIX: &str = "fields.";
const SEQUENCE_GROUP_SUFFIX: &str = ".sequence-group";
+pub(super) const SEQUENCE_GROUP_SORT_ORDER_SUFFIX: &str =
".sequence-group.sort-order";
const AGGREGATION_FUNCTION_SUFFIX: &str = ".aggregate-function";
const LIST_AGG_DELIMITER_SUFFIX: &str = ".list-agg-delimiter";
const IGNORE_RETRACT_SUFFIX: &str = ".ignore-retract";
@@ -50,6 +51,7 @@ pub(crate) enum PartialUpdateMode {
pub(crate) struct SequenceGroup {
pub(crate) sequence_fields: Vec<String>,
pub(crate) protected_fields: Vec<String>,
+ pub(crate) ascending: bool,
}
impl SequenceGroup {
@@ -84,6 +86,7 @@ impl<'a> PartialUpdateConfig<'a> {
&self,
has_primary_keys: bool,
) -> crate::Result<Option<PartialUpdateMode>> {
+ self.validate_sequence_group_sort_orders(has_primary_keys)?;
match self.validated_mode(has_primary_keys) {
Ok(mode) => Ok(mode),
Err(unsupported_options) => Err(crate::Error::ConfigInvalid {
@@ -114,6 +117,7 @@ impl<'a> PartialUpdateConfig<'a> {
has_primary_keys: bool,
table_name: &str,
) -> crate::Result<Option<PartialUpdateMode>> {
+ self.validate_sequence_group_sort_orders(has_primary_keys)?;
match self.validated_mode(has_primary_keys) {
Ok(mode) => {
if mode.is_some() { self.validate_delete_config()?; }
@@ -133,6 +137,7 @@ impl<'a> PartialUpdateConfig<'a> {
has_primary_keys: bool,
table_name: &str,
) -> crate::Result<Option<PartialUpdateMode>> {
+ self.validate_sequence_group_sort_orders(has_primary_keys)?;
if !has_primary_keys || !self.is_enabled() {
return Ok(None);
}
@@ -156,6 +161,7 @@ impl<'a> PartialUpdateConfig<'a> {
}
pub(crate) fn sequence_groups(&self) -> crate::Result<Vec<SequenceGroup>> {
+ self.validate_sequence_group_sort_orders(true)?;
let mut options: Vec<(&String, &String)> = self
.options
.iter()
@@ -179,11 +185,48 @@ impl<'a> PartialUpdateConfig<'a> {
Ok(SequenceGroup {
sequence_fields,
protected_fields,
+ ascending: !self
+ .options
+ .get(&format!("{key}.sort-order"))
+ .is_some_and(|value|
value.eq_ignore_ascii_case("descending")),
})
})
.collect()
}
+ fn validate_sequence_group_sort_orders(&self, has_primary_keys: bool) ->
crate::Result<()> {
+ for (key, value) in self
+ .options
+ .iter()
+ .filter(|(key, _)| is_fields_option_with_suffix(key,
SEQUENCE_GROUP_SORT_ORDER_SUFFIX))
+ {
+ if !has_primary_keys || !self.is_enabled() {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Option '{key}' is only supported by primary-key
tables using merge-engine=partial-update"
+ ),
+ });
+ }
+ let group_key = key
+ .strip_suffix(".sort-order")
+ .expect("sort-order suffix checked");
+ if !self.options.contains_key(group_key) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Cannot set '{key}' without defining corresponding
sequence group '{group_key}'"
+ ),
+ });
+ }
+ if !value.eq_ignore_ascii_case("ascending") &&
!value.eq_ignore_ascii_case("descending")
+ {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!("Invalid value '{value}' for option
'{key}': expected ascending or descending"),
+ });
+ }
+ }
+ Ok(())
+ }
+
pub(crate) fn remove_record_on_delete(&self) -> bool {
self.options
.get(PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION)
@@ -763,15 +806,73 @@ mod tests {
SequenceGroup {
sequence_fields: vec!["event_time".to_string(),
"source_order".to_string()],
protected_fields: vec!["price".to_string(),
"quantity".to_string()],
+ ascending: true,
},
SequenceGroup {
sequence_fields: vec!["profile_version".to_string()],
protected_fields: vec!["name".to_string(),
"address".to_string()],
+ ascending: true,
},
]
);
}
+ #[test]
+ fn test_sequence_group_sort_order_validation() {
+ let group_key = "fields.event_time,source_order.sequence-group";
+ let sort_key =
"fields.event_time,source_order.sequence-group.sort-order";
+ for (value, ascending) in [
+ ("ascending", true),
+ ("descending", false),
+ ("DeScEnDiNg", false),
+ ] {
+ let options = partial_update_options(&[(group_key, "price"),
(sort_key, value)]);
+ let config = PartialUpdateConfig::new(&options);
+ assert_eq!(config.sequence_groups().unwrap()[0].ascending,
ascending);
+ assert_eq!(
+ config.validate_create_mode(true).unwrap(),
+ Some(PartialUpdateMode::SequenceGroup)
+ );
+ }
+
+ for (options, has_pk, message) in [
+ (
+ partial_update_options(&[(group_key, "price"), (sort_key,
"desc")]),
+ true,
+ "expected ascending or descending",
+ ),
+ (
+ partial_update_options(&[(sort_key, "descending")]),
+ true,
+ "corresponding sequence group",
+ ),
+ (
+ partial_update_options(&[(group_key, "price"), (sort_key,
"descending")]),
+ false,
+ "primary-key tables",
+ ),
+ (
+ HashMap::from([
+ (group_key.to_string(), "price".to_string()),
+ (sort_key.to_string(), "descending".to_string()),
+ ]),
+ true,
+ "merge-engine=partial-update",
+ ),
+ ] {
+ let config = PartialUpdateConfig::new(&options);
+ for result in [
+ config.validate_create_mode(has_pk),
+ config.validate_read_mode(has_pk, "t"),
+ config.validate_write_mode(has_pk, "t"),
+ ] {
+ assert!(
+ matches!(result, Err(crate::Error::ConfigInvalid {
message: actual }) if actual.contains(message))
+ );
+ }
+ }
+ }
+
#[test]
fn test_parse_sequence_groups_rejects_empty_field_names() {
for (key, value) in [
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index bf257eb7..85dd36a5 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -5147,9 +5147,6 @@ mod tests {
#[test]
fn test_rename_column_rewrites_sequence_group_options() {
- // `sequence-group` is rejected by Rust's create-time merge-engine
- // validation, so the fixture carries the option on a table without
- // `merge-engine` — the shape in which Java-written metadata arrives.
let table_schema = TableSchema::new(
0,
&Schema::builder()
@@ -5158,11 +5155,17 @@ mod tests {
.column("source_order", DataType::Int(IntType::new()))
.column("price", DataType::Int(IntType::new()))
.column("quantity", DataType::Int(IntType::new()))
+ .column("descending", DataType::Int(IntType::new()))
.primary_key(["id"])
+ .option("merge-engine", "partial-update")
.option(
"fields.version,source_order.sequence-group",
"price,quantity",
)
+ .option(
+ "fields.version,source_order.sequence-group.sort-order",
+ "descending",
+ )
.build()
.unwrap(),
);
@@ -5175,6 +5178,11 @@ mod tests {
"order_seq".to_string(),
),
crate::spec::SchemaChange::rename_column("price".to_string(),
"amount".to_string()),
+ // The direction is a literal, even if a column shares its
name.
+ crate::spec::SchemaChange::rename_column(
+ "descending".to_string(),
+ "direction".to_string(),
+ ),
])
.unwrap();
@@ -5191,6 +5199,16 @@ mod tests {
.get("fields.version,source_order.sequence-group"),
None
);
+ assert_eq!(
+ new_schema
+ .options()
+ .get("fields.version,order_seq.sequence-group.sort-order")
+ .map(String::as_str),
+ Some("descending")
+ );
+ assert!(!new_schema
+ .options()
+
.contains_key("fields.version,source_order.sequence-group.sort-order"));
}
#[test]
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index eda7a27c..e5486e9e 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -3477,6 +3477,116 @@ mod tests {
.all(|batch| batch.schema().fields().len() == 1));
}
+ #[tokio::test]
+ async fn kv_read_partial_update_row_sequence_group_sort_order() {
+ use crate::spec::RowType;
+ use arrow_buffer::NullBuffer;
+
+ // None is a top-level null; Some(None) is a non-null ROW(NULL).
+ // The final two entries select the winning row for
ascending/descending.
+ let cases = [
+ (Some(Some(5)), Some(None), 0, 1),
+ (Some(None), Some(Some(5)), 1, 0),
+ (None, Some(Some(5)), 1, 1),
+ (Some(Some(5)), None, 0, 0),
+ (None, Some(None), 1, 1),
+ (Some(None), None, 0, 0),
+ (Some(Some(5)), Some(Some(2)), 0, 1),
+ (Some(None), Some(None), 1, 1),
+ (Some(Some(5)), Some(Some(5)), 1, 1),
+ ];
+ for direction in ["ascending", "descending"] {
+ for separate_commits in [false, true] {
+ for (first, second, ascending_winner, descending_winner) in
cases {
+ let file_io = test_file_io();
+ let table_path =
"memory:/kv_partial_update_row_sequence_group";
+ setup_dirs(&file_io, table_path).await;
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "seq",
+ DataType::Row(RowType::new(vec![DataField::new(
+ 0,
+ "version".into(),
+ DataType::Int(IntType::new()),
+ )])),
+ )
+ .column("value", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("merge-engine", "partial-update")
+ .option("fields.seq.sequence-group", "value")
+ .option("fields.seq.sequence-group.sort-order",
direction)
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io,
+ Identifier::new("default",
"kv_partial_update_row_sequence_group"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let arrow_schema =
build_target_arrow_schema(table.schema().fields()).unwrap();
+ let ArrowDataType::Struct(sequence_fields) =
arrow_schema.field(1).data_type()
+ else {
+ panic!("expected a ROW sequence field");
+ };
+ let sequences = [first, second];
+ let sequence_array = StructArray::new(
+ sequence_fields.clone(),
+ vec![Arc::new(Int32Array::from(
+ sequences
+ .into_iter()
+ .map(Option::flatten)
+ .collect::<Vec<_>>(),
+ ))],
+ Some(NullBuffer::from(
+ sequences
+ .into_iter()
+ .map(|seq| seq.is_some())
+ .collect::<Vec<_>>(),
+ )),
+ );
+ // Equal ordering values must allow clearing the protected
field.
+ let values = vec![Some(50), if first == second { None }
else { Some(99) }];
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 1])),
+ Arc::new(sequence_array),
+ Arc::new(Int32Array::from(values)),
+ ],
+ )
+ .unwrap();
+ if separate_commits {
+ write_commit(&table, &batch.slice(0, 1)).await;
+ write_commit(&table, &batch.slice(1, 1)).await;
+ } else {
+ write_commit(&table, &batch).await;
+ }
+ let winner = if direction == "ascending" {
+ ascending_winner
+ } else {
+ descending_winner
+ };
+ for projection in [None, Some(["value"].as_slice())] {
+ let result = read_rows(&table, projection, None).await;
+ assert_eq!(result.len(), 1);
+ assert_eq!(result[0].num_rows(), 1);
+ for field in result[0].schema().fields() {
+ assert_eq!(
+
result[0].column_by_name(field.name()).unwrap().as_ref(),
+
batch.column_by_name(field.name()).unwrap().slice(winner, 1).as_ref(),
+ "direction={direction},
separate_commits={separate_commits}, sequences={sequences:?},
projection={projection:?}, field={}",
+ field.name(),
+ );
+ }
+ }
+ }
+ }
+ }
+ }
+
#[tokio::test]
async fn
kv_read_partial_update_sequence_group_aggregation_with_projection() {
let file_io = test_file_io();
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index 2a81c33c..35bcb8e7 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -2145,50 +2145,61 @@ mod tests {
#[test]
fn test_flush_partial_update_sequence_group_delete_preserves_tombstone() {
- let schema = Arc::new(ArrowSchema::new(vec![
- ArrowField::new("id", ArrowDataType::Int32, false),
- ArrowField::new("seq", ArrowDataType::Int64, false),
- ArrowField::new(VALUE_KIND_FIELD_NAME, ArrowDataType::Int8, false),
- ArrowField::new("value", ArrowDataType::Int32, true),
- ]));
- let batch = RecordBatch::try_new(
- schema,
- vec![
- Arc::new(Int32Array::from(vec![1, 1])),
- Arc::new(Int64Array::from(vec![10, 20])),
- Arc::new(Int8Array::from(vec![0, 3])),
- Arc::new(Int32Array::from(vec![Some(100), Some(100)])),
- ],
- )
- .unwrap();
- let mut config = test_write_config(MergeEngine::PartialUpdate);
- config
- .table_options
- .insert("fields.seq.sequence-group".into(), "value".into());
- config.table_options.insert(
- "partial-update.remove-record-on-sequence-group".into(),
- "seq".into(),
- );
- let writer =
-
KeyValueFileWriter::new(FileIOBuilder::new("memory").build().unwrap(), config,
0)
- .unwrap();
- let (merged, _) = writer
- .merge_partial_update_rows(
- &batch,
- &Int64Array::from(vec![0, 1]),
- &UInt32Array::from(vec![0, 1]),
+ for (direction, sequences, expected_kind) in [
+ ("ascending", vec![10, 20], 3),
+ ("descending", vec![20, 10], 3),
+ ("descending", vec![10, 20], 0),
+ ("descending", vec![10, 10], 3),
+ ] {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("seq", ArrowDataType::Int64, false),
+ ArrowField::new(VALUE_KIND_FIELD_NAME, ArrowDataType::Int8,
false),
+ ArrowField::new("value", ArrowDataType::Int32, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 1])),
+ Arc::new(Int64Array::from(sequences)),
+ Arc::new(Int8Array::from(vec![0, 3])),
+ Arc::new(Int32Array::from(vec![Some(100), Some(100)])),
+ ],
)
.unwrap();
- assert_eq!(merged.num_rows(), 1);
- assert_eq!(
- merged
- .column(2)
- .as_any()
- .downcast_ref::<Int8Array>()
- .unwrap()
- .value(0),
- 3
- );
+ let mut config = test_write_config(MergeEngine::PartialUpdate);
+ config
+ .table_options
+ .insert("fields.seq.sequence-group".into(), "value".into());
+ config.table_options.insert(
+ "fields.seq.sequence-group.sort-order".into(),
+ direction.into(),
+ );
+ config.table_options.insert(
+ "partial-update.remove-record-on-sequence-group".into(),
+ "seq".into(),
+ );
+ let writer =
+
KeyValueFileWriter::new(FileIOBuilder::new("memory").build().unwrap(), config,
0)
+ .unwrap();
+ let (merged, _) = writer
+ .merge_partial_update_rows(
+ &batch,
+ &Int64Array::from(vec![0, 1]),
+ &UInt32Array::from(vec![0, 1]),
+ )
+ .unwrap();
+ assert_eq!(merged.num_rows(), 1);
+ assert_eq!(
+ merged
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .unwrap()
+ .value(0),
+ expected_kind
+ );
+ }
}
#[test]
@@ -2459,102 +2470,111 @@ mod tests {
#[test]
fn test_flush_partial_update_sequence_group_and_aggregates() {
- let mut config = test_write_config(MergeEngine::PartialUpdate);
- config.table_options.extend([
- (
- "fields.version.sequence-group".into(),
- "price,total,tag".into(),
- ),
- ("fields.total.aggregate-function".into(), "sum".into()),
- ("fields.tag.aggregate-function".into(), "listagg".into()),
- ]);
- config.value_fields = vec![
- DataField::new(0, "id".into(), DataType::Int(IntType::new())),
- DataField::new(1, "arrival".into(),
DataType::BigInt(BigIntType::new())),
- DataField::new(2, "version".into(),
DataType::BigInt(BigIntType::new())),
- DataField::new(3, "price".into(), DataType::Int(IntType::new())),
- DataField::new(4, "total".into(), DataType::Int(IntType::new())),
- DataField::new(
- 5,
- "tag".into(),
- DataType::VarChar(crate::spec::VarCharType::string_type()),
- ),
- ];
- let writer =
-
KeyValueFileWriter::new(FileIOBuilder::new("memory").build().unwrap(), config,
0)
- .unwrap();
- let schema = Arc::new(ArrowSchema::new(vec![
- ArrowField::new("id", ArrowDataType::Int32, false),
- ArrowField::new("arrival", ArrowDataType::Int64, false),
- ArrowField::new("version", ArrowDataType::Int64, true),
- ArrowField::new("price", ArrowDataType::Int32, true),
- ArrowField::new("total", ArrowDataType::Int32, true),
- ArrowField::new("tag", ArrowDataType::Utf8, true),
- ]));
- let batch = RecordBatch::try_new(
- schema,
- vec![
- Arc::new(Int32Array::from(vec![1, 1, 1])),
- Arc::new(Int64Array::from(vec![1, 2, 3])),
- Arc::new(Int64Array::from(vec![10, 5, 20])),
- Arc::new(Int32Array::from(vec![100, 50, 200])),
- Arc::new(Int32Array::from(vec![3, 7, 11])),
- Arc::new(StringArray::from(vec!["a", "b", "c"])),
- ],
- )
- .unwrap();
- let (merged, merged_seq) = writer
- .merge_partial_update_rows(
- &batch,
- &Int64Array::from(vec![100, 101, 102]),
- &UInt32Array::from(vec![0, 1, 2]),
+ for (direction, version, price, tags) in [
+ ("ascending", 20, 200, "b,a,c"),
+ ("descending", 5, 50, "c,a,b"),
+ ] {
+ let mut config = test_write_config(MergeEngine::PartialUpdate);
+ config.table_options.extend([
+ (
+ "fields.version.sequence-group".into(),
+ "price,total,tag".into(),
+ ),
+ (
+ "fields.version.sequence-group.sort-order".into(),
+ direction.into(),
+ ),
+ ("fields.total.aggregate-function".into(), "sum".into()),
+ ("fields.tag.aggregate-function".into(), "listagg".into()),
+ ]);
+ config.value_fields = vec![
+ DataField::new(0, "id".into(), DataType::Int(IntType::new())),
+ DataField::new(1, "arrival".into(),
DataType::BigInt(BigIntType::new())),
+ DataField::new(2, "version".into(),
DataType::BigInt(BigIntType::new())),
+ DataField::new(3, "price".into(),
DataType::Int(IntType::new())),
+ DataField::new(4, "total".into(),
DataType::Int(IntType::new())),
+ DataField::new(
+ 5,
+ "tag".into(),
+ DataType::VarChar(crate::spec::VarCharType::string_type()),
+ ),
+ ];
+ let writer =
+
KeyValueFileWriter::new(FileIOBuilder::new("memory").build().unwrap(), config,
0)
+ .unwrap();
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("arrival", ArrowDataType::Int64, false),
+ ArrowField::new("version", ArrowDataType::Int64, true),
+ ArrowField::new("price", ArrowDataType::Int32, true),
+ ArrowField::new("total", ArrowDataType::Int32, true),
+ ArrowField::new("tag", ArrowDataType::Utf8, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 1, 1])),
+ Arc::new(Int64Array::from(vec![1, 2, 3])),
+ Arc::new(Int64Array::from(vec![10, 5, 20])),
+ Arc::new(Int32Array::from(vec![100, 50, 200])),
+ Arc::new(Int32Array::from(vec![3, 7, 11])),
+ Arc::new(StringArray::from(vec!["a", "b", "c"])),
+ ],
)
.unwrap();
- assert_eq!(merged.num_rows(), 1);
- assert_eq!(
- merged
- .column(2)
- .as_any()
- .downcast_ref::<Int64Array>()
- .unwrap()
- .value(0),
- 20,
- );
- assert_eq!(
- merged
- .column(3)
- .as_any()
- .downcast_ref::<Int32Array>()
- .unwrap()
- .value(0),
- 200,
- );
- assert_eq!(
- merged
- .column(4)
- .as_any()
- .downcast_ref::<Int32Array>()
- .unwrap()
- .value(0),
- 21,
- );
- assert_eq!(
- merged
- .column(5)
- .as_any()
- .downcast_ref::<StringArray>()
- .unwrap()
- .value(0),
- "b,a,c",
- );
- assert_eq!(
- merged_seq
- .as_any()
- .downcast_ref::<Int64Array>()
- .unwrap()
- .value(0),
- 102,
- );
+ let (merged, merged_seq) = writer
+ .merge_partial_update_rows(
+ &batch,
+ &Int64Array::from(vec![100, 101, 102]),
+ &UInt32Array::from(vec![0, 1, 2]),
+ )
+ .unwrap();
+ assert_eq!(merged.num_rows(), 1);
+ assert_eq!(
+ merged
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .value(0),
+ version,
+ );
+ assert_eq!(
+ merged
+ .column(3)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ price,
+ );
+ assert_eq!(
+ merged
+ .column(4)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 21,
+ );
+ assert_eq!(
+ merged
+ .column(5)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ tags,
+ );
+ assert_eq!(
+ merged_seq
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .value(0),
+ 102,
+ );
+ }
}
/// Lock the flush-time merge to the read-side
`PartialUpdateMergeFunction`.
diff --git a/crates/paimon/src/table/sort_merge.rs
b/crates/paimon/src/table/sort_merge.rs
index 29320a8a..579ca39b 100644
--- a/crates/paimon/src/table/sort_merge.rs
+++ b/crates/paimon/src/table/sort_merge.rs
@@ -289,6 +289,7 @@ fn merged_metadata_column(
struct RuntimeSequenceGroup {
sequence_indices: Vec<usize>,
protected_indices: Vec<usize>,
+ ascending: bool,
}
impl PartialUpdateMergeFunction {
@@ -405,6 +406,7 @@ impl PartialUpdateMergeFunction {
Ok(RuntimeSequenceGroup {
sequence_indices,
protected_indices,
+ ascending: group.ascending,
})
})
.collect::<crate::Result<Vec<_>>>()?;
@@ -546,7 +548,7 @@ impl PartialUpdateMergeFunction {
row.row_idx,
current_batch,
current_row,
- &group.sequence_indices,
+ group,
batch_buffer,
source_output_col_indices,
)?,
@@ -654,7 +656,7 @@ impl PartialUpdateMergeFunction {
row.row_idx,
current_batch_idx,
current_row_idx,
- &group.sequence_indices,
+ group,
batch_buffer,
source_output_col_indices,
)?,
@@ -756,11 +758,11 @@ fn compare_sequence_group_rows(
incoming_row_idx: usize,
current_batch_idx: usize,
current_row_idx: usize,
- sequence_indices: &[usize],
+ group: &RuntimeSequenceGroup,
batch_buffer: &[BufferedBatch],
source_output_col_indices: &[usize],
) -> crate::Result<Ordering> {
- for &output_col_idx in sequence_indices {
+ for &output_col_idx in &group.sequence_indices {
let incoming = batch_buffer[incoming_batch_idx]
.column_for_output(output_col_idx, source_output_col_indices);
let current = batch_buffer[current_batch_idx]
@@ -777,7 +779,15 @@ fn compare_sequence_group_rows(
message: format!("Failed to compare partial-update sequence-group
fields: {e}"),
source: Some(Box::new(e)),
})?;
- let ordering = comparator(incoming_row_idx, current_row_idx);
+ let mut ordering = comparator(incoming_row_idx, current_row_idx);
+ // Top-level nulls have lower precedence in either direction.
Descending
+ // order reverses non-null value comparisons, including nulls inside
ROWs.
+ if !group.ascending
+ && !incoming.is_null(incoming_row_idx)
+ && !current.is_null(current_row_idx)
+ {
+ ordering = ordering.reverse();
+ }
if !ordering.is_eq() {
return Ok(ordering);
}
@@ -2865,6 +2875,145 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_partial_update_descending_composite_sequence_group() {
+ let output_schema = Arc::new(Schema::new(vec![
+ Field::new("pk", DataType::Int32, false),
+ Field::new("major", DataType::Int32, true),
+ Field::new("minor", DataType::Int32, true),
+ Field::new("value", DataType::Int32, true),
+ ]));
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("pk", DataType::Int32, false),
+ Field::new("_SEQUENCE_NUMBER", DataType::Int64, false),
+ Field::new("_VALUE_KIND", DataType::Int8, false),
+ Field::new("major", DataType::Int32, true),
+ Field::new("minor", DataType::Int32, true),
+ Field::new("value", DataType::Int32, true),
+ ]));
+ let fields = ["pk", "major", "minor", "value"]
+ .iter()
+ .enumerate()
+ .map(|(id, name)| {
+ DataField::new(
+ id as i32,
+ (*name).into(),
+ PaimonDataType::Int(IntType::new()),
+ )
+ })
+ .collect::<Vec<_>>();
+ let options = HashMap::from([
+ ("merge-engine".into(), "partial-update".into()),
+ ("fields.major,minor.sequence-group".into(), "value".into()),
+ (
+ "fields.major,minor.sequence-group.sort-order".into(),
+ "descending".into(),
+ ),
+ ]);
+ // Each case contains two arrival-ordered rows and the winning
tuple/value.
+ let cases = [
+ // Lexicographic order: each non-null component sorts descending.
+ (
+ [(Some(2), Some(0), Some(10)), (Some(1), Some(9), Some(20))],
+ (Some(1), Some(9), Some(20)),
+ ),
+ (
+ [(Some(1), Some(9), Some(10)), (Some(1), Some(2), Some(20))],
+ (Some(1), Some(2), Some(20)),
+ ),
+ (
+ [(Some(1), Some(2), Some(10)), (Some(1), Some(9), Some(20))],
+ (Some(1), Some(2), Some(10)),
+ ),
+ // Non-null wins over null in either position, regardless of
direction.
+ (
+ [(None, Some(0), Some(10)), (Some(1), None, Some(20))],
+ (Some(1), None, Some(20)),
+ ),
+ (
+ [(Some(1), None, Some(10)), (None, Some(0), Some(20))],
+ (Some(1), None, Some(10)),
+ ),
+ (
+ [(Some(1), None, Some(10)), (Some(1), Some(9), Some(20))],
+ (Some(1), Some(9), Some(20)),
+ ),
+ (
+ [(Some(1), Some(9), Some(10)), (Some(1), None, Some(20))],
+ (Some(1), Some(9), Some(10)),
+ ),
+ // All-null tuples skip updates; ties can clear protected fields.
+ (
+ [(Some(1), Some(2), Some(10)), (None, None, Some(20))],
+ (Some(1), Some(2), Some(10)),
+ ),
+ (
+ [(Some(1), Some(2), Some(10)), (Some(1), Some(2), None)],
+ (Some(1), Some(2), None),
+ ),
+ ];
+ for (rows, expected) in cases {
+ // Separate streams exercise read-time merging of separate files.
+ let streams = rows
+ .iter()
+ .enumerate()
+ .map(|(index, &(major, minor, value))| {
+ stream_from_batches(vec![RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int64Array::from(vec![index as i64])),
+ Arc::new(Int8Array::from(vec![0])),
+ Arc::new(Int32Array::from(vec![major])),
+ Arc::new(Int32Array::from(vec![minor])),
+ Arc::new(Int32Array::from(vec![value])),
+ ],
+ )
+ .unwrap()])
+ })
+ .collect();
+ let result = SortMergeReaderBuilder::new(
+ streams,
+ schema.clone(),
+ vec![0],
+ 1,
+ 2,
+ vec![],
+ vec![3, 4, 5],
+ output_schema.clone(),
+ Box::new(
+ PartialUpdateMergeFunction::new_with_schema(
+ &options,
+ "t",
+ &fields,
+ &fields,
+ &["pk".into()],
+ )
+ .unwrap(),
+ ),
+ )
+ .build()
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(result.len(), 1);
+ assert_eq!(result[0].num_rows(), 1);
+ for (index, value) in [expected.0, expected.1,
expected.2].into_iter().enumerate() {
+ let actual = result[0]
+ .column(index + 1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(
+ actual.iter().collect::<Vec<_>>(),
+ vec![value],
+ "rows: {rows:?}"
+ );
+ }
+ }
+ }
+
#[tokio::test]
async fn
test_partial_update_composite_sequence_group_skips_empty_sequence() {
let schema = Arc::new(Schema::new(vec