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 5f9f1843 partial_update: support sequence-group reads (#584)
5f9f1843 is described below
commit 5f9f1843fb80af8e45c616731876cbd5592c44b6
Author: shyjsarah <[email protected]>
AuthorDate: Wed Jul 22 22:47:22 2026 +0800
partial_update: support sequence-group reads (#584)
---
crates/paimon/src/spec/partial_update.rs | 351 +++++++++++++++++++++++-
crates/paimon/src/table/kv_file_reader.rs | 183 ++++++++++++-
crates/paimon/src/table/kv_file_writer.rs | 2 +-
crates/paimon/src/table/sort_merge.rs | 435 +++++++++++++++++++++++++++++-
docs/src/sql.md | 12 +-
5 files changed, 960 insertions(+), 23 deletions(-)
diff --git a/crates/paimon/src/spec/partial_update.rs
b/crates/paimon/src/spec/partial_update.rs
index 07a57d73..f1c84e45 100644
--- a/crates/paimon/src/spec/partial_update.rs
+++ b/crates/paimon/src/spec/partial_update.rs
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
+
+use crate::spec::DataField;
const MERGE_ENGINE_OPTION: &str = "merge-engine";
const PARTIAL_UPDATE_ENGINE: &str = "partial-update";
@@ -31,16 +33,32 @@ 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.
+/// Partial-update mode recognized by the current Rust implementation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum PartialUpdateMode {
Basic,
+ SequenceGroup,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct SequenceGroup {
+ pub(crate) sequence_fields: Vec<String>,
+ pub(crate) protected_fields: Vec<String>,
+}
+
+impl SequenceGroup {
+ fn option_key(&self) -> String {
+ format!(
+ "{FIELDS_PREFIX}{}{SEQUENCE_GROUP_SUFFIX}",
+ self.sequence_fields.join(",")
+ )
+ }
}
/// 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.
+/// Reads support basic partial update and sequence groups. Table creation and
+/// writes remain restricted to basic partial update.
#[derive(Debug, Clone, Copy)]
pub(crate) struct PartialUpdateConfig<'a> {
options: &'a HashMap<String, String>,
@@ -72,7 +90,7 @@ impl<'a> PartialUpdateConfig<'a> {
}
}
- pub(crate) fn validate_runtime_mode(
+ pub(crate) fn validate_write_mode(
&self,
has_primary_keys: bool,
table_name: &str,
@@ -88,6 +106,143 @@ impl<'a> PartialUpdateConfig<'a> {
}
}
+ pub(crate) fn validate_read_mode(
+ &self,
+ has_primary_keys: bool,
+ table_name: &str,
+ ) -> crate::Result<Option<PartialUpdateMode>> {
+ if !has_primary_keys || !self.is_enabled() {
+ return Ok(None);
+ }
+
+ let unsupported_options = self.read_unsupported_option_keys();
+ if !unsupported_options.is_empty() {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Table '{table_name}' uses merge-engine=partial-update
options not supported by this build: {}",
+ unsupported_options.join(", ")
+ ),
+ });
+ }
+
+ Ok(Some(if self.sequence_groups()?.is_empty() {
+ PartialUpdateMode::Basic
+ } else {
+ PartialUpdateMode::SequenceGroup
+ }))
+ }
+
+ pub(crate) fn sequence_groups(&self) -> crate::Result<Vec<SequenceGroup>> {
+ let mut options: Vec<(&String, &String)> = self
+ .options
+ .iter()
+ .filter(|(key, _)| is_fields_option_with_suffix(key,
SEQUENCE_GROUP_SUFFIX))
+ .collect();
+ options.sort_by_key(|(key, _)| *key);
+
+ options
+ .into_iter()
+ .map(|(key, value)| {
+ let sequence_fields = key
+ .strip_prefix(FIELDS_PREFIX)
+ .and_then(|key| key.strip_suffix(SEQUENCE_GROUP_SUFFIX))
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: format!(
+ "Invalid partial-update sequence-group option
'{key}={value}'"
+ ),
+ })
+ .and_then(|fields| parse_field_list(fields, key, value))?;
+ let protected_fields = parse_field_list(value, key, value)?;
+ Ok(SequenceGroup {
+ sequence_fields,
+ protected_fields,
+ })
+ })
+ .collect()
+ }
+
+ pub(crate) fn validated_sequence_groups(
+ &self,
+ fields: &[DataField],
+ primary_keys: &[String],
+ ) -> crate::Result<Vec<SequenceGroup>> {
+ let groups = self.sequence_groups()?;
+ let field_names: HashSet<&str> =
fields.iter().map(DataField::name).collect();
+ let primary_keys: HashSet<&str> =
primary_keys.iter().map(String::as_str).collect();
+ let mut protected_field_owners: HashMap<String, String> =
HashMap::new();
+
+ for group in &groups {
+ let option_key = group.option_key();
+ for field in group
+ .sequence_fields
+ .iter()
+ .chain(group.protected_fields.iter())
+ {
+ if !field_names.contains(field.as_str()) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Field '{field}' referenced by partial-update
sequence-group \
+ option '{option_key}' does not exist in the table
schema"
+ ),
+ });
+ }
+ if primary_keys.contains(field.as_str()) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "The sequence-group '{option_key}' contains
primary key field \
+ '{field}', which is not allowed. Primary key
columns cannot be put \
+ in sequence-group."
+ ),
+ });
+ }
+ }
+ for field in &group.protected_fields {
+ if let Some(previous_option) =
+ protected_field_owners.insert(field.clone(),
option_key.clone())
+ {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Field '{field}' is protected by multiple sequence
groups: \
+ '{previous_option}' and '{option_key}'"
+ ),
+ });
+ }
+ }
+ }
+
+ Ok(groups)
+ }
+
+ pub(crate) fn required_sequence_fields(
+ &self,
+ fields: &[DataField],
+ primary_keys: &[String],
+ projected_fields: &[String],
+ ) -> crate::Result<Vec<String>> {
+ let groups = self.validated_sequence_groups(fields, primary_keys)?;
+ let projected: HashSet<&str> =
projected_fields.iter().map(String::as_str).collect();
+ let mut required = Vec::new();
+ let mut seen = HashSet::new();
+
+ for group in groups {
+ let group_is_projected = group
+ .sequence_fields
+ .iter()
+ .chain(group.protected_fields.iter())
+ .any(|field| projected.contains(field.as_str()));
+ if !group_is_projected {
+ continue;
+ }
+ for field in group.sequence_fields {
+ if seen.insert(field.clone()) {
+ required.push(field);
+ }
+ }
+ }
+
+ Ok(required)
+ }
+
fn validated_mode(
&self,
has_primary_keys: bool,
@@ -114,6 +269,20 @@ impl<'a> PartialUpdateConfig<'a> {
keys.sort();
keys
}
+
+ fn read_unsupported_option_keys(&self) -> Vec<String> {
+ let mut keys: Vec<String> = self
+ .options
+ .keys()
+ .filter(|key| {
+ is_unsupported_partial_update_option(key)
+ && !is_fields_option_with_suffix(key,
SEQUENCE_GROUP_SUFFIX)
+ })
+ .cloned()
+ .collect();
+ keys.sort();
+ keys
+ }
}
fn is_unsupported_partial_update_option(key: &str) -> bool {
@@ -131,9 +300,33 @@ fn is_fields_option_with_suffix(key: &str, suffix: &str)
-> bool {
key.starts_with(FIELDS_PREFIX) && key.ends_with(suffix)
}
+fn parse_field_list(
+ value: &str,
+ option_key: &str,
+ option_value: &str,
+) -> crate::Result<Vec<String>> {
+ value
+ .split(',')
+ .map(str::trim)
+ .map(|field| {
+ if field.is_empty() {
+ Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Invalid partial-update sequence-group option \
+ '{option_key}={option_value}': empty field name"
+ ),
+ })
+ } else {
+ Ok(field.to_string())
+ }
+ })
+ .collect()
+}
+
#[cfg(test)]
mod tests {
use super::*;
+ use crate::spec::{DataField, DataType, IntType};
fn partial_update_options(extra: &[(&str, &str)]) -> HashMap<String,
String> {
let mut options = HashMap::from([(
@@ -216,15 +409,159 @@ mod tests {
}
#[test]
- fn test_validate_runtime_mode_rejects_unsupported_partial_update_options()
{
+ fn test_validate_write_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();
+ let err = config.validate_write_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:?}"
);
}
+
+ #[test]
+ fn test_validate_read_mode_accepts_sequence_group() {
+ let options =
+ partial_update_options(&[("fields.updated_at.sequence-group",
"price,quantity")]);
+ let config = PartialUpdateConfig::new(&options);
+
+ assert_eq!(
+ config.validate_read_mode(true, "default.t").unwrap(),
+ Some(PartialUpdateMode::SequenceGroup)
+ );
+ }
+
+ #[test]
+ fn test_parse_sequence_groups() {
+ let options = partial_update_options(&[
+ (
+ "fields.event_time,source_order.sequence-group",
+ "price,quantity",
+ ),
+ ("fields.profile_version.sequence-group", "name,address"),
+ ]);
+ let config = PartialUpdateConfig::new(&options);
+
+ assert_eq!(
+ config.sequence_groups().unwrap(),
+ vec![
+ SequenceGroup {
+ sequence_fields: vec!["event_time".to_string(),
"source_order".to_string()],
+ protected_fields: vec!["price".to_string(),
"quantity".to_string()],
+ },
+ SequenceGroup {
+ sequence_fields: vec!["profile_version".to_string()],
+ protected_fields: vec!["name".to_string(),
"address".to_string()],
+ },
+ ]
+ );
+ }
+
+ #[test]
+ fn test_parse_sequence_groups_rejects_empty_field_names() {
+ for (key, value) in [
+ ("fields.version,,source.sequence-group", "price"),
+ ("fields.version.sequence-group", "price,,quantity"),
+ ] {
+ let options = partial_update_options(&[(key, value)]);
+ let config = PartialUpdateConfig::new(&options);
+
+ let err = config.sequence_groups().unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains(key) && message.contains("empty field
name")),
+ "expected malformed field list to be rejected, got {err:?}"
+ );
+ }
+ }
+
+ #[test]
+ fn test_validate_sequence_groups_rejects_primary_key() {
+ let options =
partial_update_options(&[("fields.version.sequence-group", "id,price")]);
+ let config = PartialUpdateConfig::new(&options);
+ let fields = vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "version".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(2, "price".to_string(),
DataType::Int(IntType::new())),
+ ];
+
+ let err = config
+ .validated_sequence_groups(&fields, &["id".to_string()])
+ .unwrap_err();
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("fields.version.sequence-group")
+ && message.contains("primary key field 'id'")),
+ "expected primary-key conflict, got {err:?}"
+ );
+ }
+
+ #[test]
+ fn test_validate_sequence_groups_rejects_duplicate_protected_field() {
+ let options = partial_update_options(&[
+ ("fields.version_a.sequence-group", "price"),
+ ("fields.version_b.sequence-group", "price"),
+ ]);
+ let config = PartialUpdateConfig::new(&options);
+ let fields = vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "version_a".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(2, "version_b".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(3, "price".to_string(),
DataType::Int(IntType::new())),
+ ];
+
+ let err = config
+ .validated_sequence_groups(&fields, &["id".to_string()])
+ .unwrap_err();
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("price") && message.contains("multiple
sequence groups")),
+ "expected duplicate protected-field error, got {err:?}"
+ );
+ }
+
+ #[test]
+ fn test_required_sequence_fields_for_projection() {
+ let options = partial_update_options(&[
+ (
+ "fields.event_time,source_order.sequence-group",
+ "price,quantity",
+ ),
+ ("fields.profile_version.sequence-group", "name,address"),
+ ]);
+ let config = PartialUpdateConfig::new(&options);
+ let fields = vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "event_time".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(2, "source_order".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(3, "price".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(4, "quantity".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(
+ 5,
+ "profile_version".to_string(),
+ DataType::Int(IntType::new()),
+ ),
+ DataField::new(6, "name".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(7, "address".to_string(),
DataType::Int(IntType::new())),
+ ];
+
+ assert_eq!(
+ config
+ .required_sequence_fields(
+ &fields,
+ &["id".to_string()],
+ &["price".to_string(), "name".to_string()],
+ )
+ .unwrap(),
+ vec![
+ "event_time".to_string(),
+ "source_order".to_string(),
+ "profile_version".to_string(),
+ ]
+ );
+ }
}
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index db97e66b..bb36e369 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -33,8 +33,8 @@ use super::sort_merge::{
use crate::arrow::build_target_arrow_schema;
use crate::io::FileIO;
use crate::spec::{
- BigIntType, DataField, DataType as PaimonDataType, MergeEngine, Predicate,
TinyIntType,
- SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID,
+ BigIntType, DataField, DataType as PaimonDataType, MergeEngine,
PartialUpdateConfig, Predicate,
+ TinyIntType, SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME,
VALUE_KIND_FIELD_ID,
VALUE_KIND_FIELD_NAME,
};
use crate::table::schema_manager::SchemaManager;
@@ -107,6 +107,45 @@ pub(super) fn retain_primary_key_conjuncts(
.collect()
}
+fn widen_partial_update_sequence_group_fields(
+ merge_engine: MergeEngine,
+ table_options: &HashMap<String, String>,
+ table_fields: &[DataField],
+ primary_keys: &[String],
+ mut user_fields: Vec<DataField>,
+) -> crate::Result<Vec<DataField>> {
+ if merge_engine != MergeEngine::PartialUpdate {
+ return Ok(user_fields);
+ }
+
+ let projected_fields = user_fields
+ .iter()
+ .map(|field| field.name().to_string())
+ .collect::<Vec<_>>();
+ let required_fields =
PartialUpdateConfig::new(table_options).required_sequence_fields(
+ table_fields,
+ primary_keys,
+ &projected_fields,
+ )?;
+ for field_name in required_fields {
+ if user_fields.iter().any(|field| field.name() == field_name) {
+ continue;
+ }
+ let field = table_fields
+ .iter()
+ .find(|field| field.name() == field_name)
+ .cloned()
+ .ok_or_else(|| Error::UnexpectedError {
+ message: format!(
+ "Partial-update sequence field '{field_name}' not found in
table schema"
+ ),
+ source: None,
+ })?;
+ user_fields.push(field);
+ }
+ Ok(user_fields)
+}
+
impl KeyValueFileReader {
pub(crate) fn new(file_io: FileIO, config: KeyValueReadConfig) -> Self {
let pushdown_predicates = retain_primary_key_conjuncts(
@@ -136,16 +175,22 @@ impl KeyValueFileReader {
merge_engine: MergeEngine,
table_options: &HashMap<String, String>,
table_name: &str,
+ table_fields: &[DataField],
merge_output_fields: &[DataField],
primary_keys: &[String],
sequence_fields: &[String],
) -> crate::Result<Box<dyn super::sort_merge::MergeFunction>> {
match merge_engine {
MergeEngine::Deduplicate => Ok(Box::new(DeduplicateMergeFunction)),
- MergeEngine::PartialUpdate =>
Ok(Box::new(PartialUpdateMergeFunction::new(
- table_options,
- table_name,
- )?)),
+ MergeEngine::PartialUpdate => Ok(Box::new(
+ PartialUpdateMergeFunction::new_with_schema(
+ table_options,
+ table_name,
+ table_fields,
+ merge_output_fields,
+ primary_keys,
+ )?,
+ )),
MergeEngine::FirstRow => Err(Error::Unsupported {
message: "KeyValueFileReader does not support
merge-engine=first-row; first-row reads should use the non-KV path".to_string(),
}),
@@ -242,6 +287,13 @@ impl KeyValueFileReader {
&user_fields,
residual_file_predicates.as_ref(),
);
+ let user_fields = widen_partial_update_sequence_group_fields(
+ self.config.merge_engine,
+ &self.config.table_options,
+ &self.config.table_fields,
+ &self.config.primary_keys,
+ user_fields,
+ )?;
// Internal read type: [_SEQ, _VK, user_fields...]
let mut internal_read_type: Vec<DataField> = Vec::new();
@@ -413,6 +465,7 @@ impl KeyValueFileReader {
merge_engine,
&table_options,
&table_name,
+ &table_fields,
&merge_output_fields,
&primary_keys,
&sequence_fields,
@@ -678,6 +731,36 @@ mod tests {
assert!(matches!(&kept[0], Predicate::AlwaysTrue));
}
+ #[test]
+ fn widen_partial_update_projection_with_sequence_fields() {
+ let table_fields = vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "version".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(2, "price".to_string(),
DataType::Int(IntType::new())),
+ ];
+ let options = HashMap::from([
+ ("merge-engine".to_string(), "partial-update".to_string()),
+ (
+ "fields.version.sequence-group".to_string(),
+ "price".to_string(),
+ ),
+ ]);
+
+ let widened = widen_partial_update_sequence_group_fields(
+ MergeEngine::PartialUpdate,
+ &options,
+ &table_fields,
+ &["id".to_string()],
+ vec![table_fields[2].clone()],
+ )
+ .unwrap();
+
+ assert_eq!(
+ widened.iter().map(DataField::name).collect::<Vec<_>>(),
+ vec!["price", "version"]
+ );
+ }
+
#[tokio::test]
async fn
kv_input_decode_honors_read_batch_size_without_changing_merge_batching() {
let file_io = test_file_io();
@@ -1147,6 +1230,94 @@ mod tests {
assert_eq!(int_column(&batches, "b"), vec![7]);
}
+ #[tokio::test]
+ async fn kv_read_partial_update_sequence_groups_with_projection() {
+ let file_io = test_file_io();
+ let table_path = "memory:/kv_partial_update_sequence_groups";
+ setup_dirs(&file_io, table_path).await;
+
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("seq_a", DataType::Int(IntType::new()))
+ .column("value_a", DataType::Int(IntType::new()))
+ .column("seq_b", DataType::Int(IntType::new()))
+ .column("value_b", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("merge-engine", "partial-update")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "kv_partial_update_sequence_groups_t"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("seq_a", ArrowDataType::Int32, true),
+ ArrowField::new("value_a", ArrowDataType::Int32, true),
+ ArrowField::new("seq_b", ArrowDataType::Int32, true),
+ ArrowField::new("value_b", ArrowDataType::Int32, true),
+ ]));
+ let make = |seq_a, value_a, seq_b, value_b| {
+ RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int32Array::from(vec![seq_a])),
+ Arc::new(Int32Array::from(vec![value_a])),
+ Arc::new(Int32Array::from(vec![seq_b])),
+ Arc::new(Int32Array::from(vec![value_b])),
+ ],
+ )
+ .unwrap()
+ };
+
+ write_commit(&table, &make(10, 100, 10, 1000)).await;
+ write_commit(&table, &make(9, 200, 11, 2000)).await;
+
+ let sequence_group_schema =
table.schema().copy_with_options(HashMap::from([
+ (
+ "fields.seq_a.sequence-group".to_string(),
+ "value_a".to_string(),
+ ),
+ (
+ "fields.seq_b.sequence-group".to_string(),
+ "value_b".to_string(),
+ ),
+ ]));
+ let sequence_group_table = Table::new(
+ file_io,
+ Identifier::new("default", "kv_partial_update_sequence_groups_t"),
+ table_path.to_string(),
+ sequence_group_schema,
+ None,
+ );
+
+ let batches = read_rows(
+ &sequence_group_table,
+ Some(&["id", "value_a", "value_b"]),
+ None,
+ )
+ .await;
+
+ assert_eq!(int_column(&batches, "value_a"), vec![100]);
+ assert_eq!(int_column(&batches, "value_b"), vec![2000]);
+ for batch in batches {
+ assert_eq!(
+ batch
+ .schema()
+ .fields()
+ .iter()
+ .map(|field| field.name().as_str())
+ .collect::<Vec<_>>(),
+ vec!["id", "value_a", "value_b"]
+ );
+ }
+ }
+
/// An AlwaysFalse filter on a PK table must return nothing, end to end.
/// Two layers enforce it: scan-side stats pruning treats AlwaysFalse as
/// prune-everything (plans no files), and the post-merge residual masks
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index 86873e8a..b5e94f54 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -110,7 +110,7 @@ impl KeyValueFileWriter {
&& CoreOptions::new(&config.table_options).ignore_delete();
if config.merge_engine == MergeEngine::PartialUpdate {
PartialUpdateConfig::new(&config.table_options)
- .validate_runtime_mode(true, &config.table_name)?;
+ .validate_write_mode(true, &config.table_name)?;
if config.deletion_vectors_enabled {
return Err(crate::Error::Unsupported {
diff --git a/crates/paimon/src/table/sort_merge.rs
b/crates/paimon/src/table/sort_merge.rs
index abbb6513..4a6d5835 100644
--- a/crates/paimon/src/table/sort_merge.rs
+++ b/crates/paimon/src/table/sort_merge.rs
@@ -31,8 +31,9 @@ use crate::table::aggregator::{new_aggregator,
FieldAggregator};
use crate::table::ArrowRecordBatchStream;
use crate::Error;
use arrow_array::{new_null_array, ArrayRef, Int64Array, Int8Array,
RecordBatch};
+use arrow_ord::ord::make_comparator;
use arrow_row::{RowConverter, Rows, SortField};
-use arrow_schema::SchemaRef;
+use arrow_schema::{SchemaRef, SortOptions};
use arrow_select::interleave::interleave;
use async_stream::try_stream;
use futures::StreamExt;
@@ -186,19 +187,99 @@ impl MergeFunction for DeduplicateMergeFunction {
///
/// DELETE / UPDATE_BEFORE rows are ignored when `ignore-delete=true` and
/// treated as unsupported otherwise.
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone)]
pub(crate) struct PartialUpdateMergeFunction {
ignore_delete: bool,
+ sequence_groups: Vec<RuntimeSequenceGroup>,
+ grouped_fields: HashSet<usize>,
+}
+
+#[derive(Debug, Clone)]
+struct RuntimeSequenceGroup {
+ sequence_indices: Vec<usize>,
+ protected_indices: Vec<usize>,
}
impl PartialUpdateMergeFunction {
+ #[cfg(test)]
pub(crate) fn new(
table_options: &HashMap<String, String>,
table_name: &str,
) -> crate::Result<Self> {
- PartialUpdateConfig::new(table_options).validate_runtime_mode(true,
table_name)?;
+ PartialUpdateConfig::new(table_options).validate_write_mode(true,
table_name)?;
Ok(Self {
ignore_delete: CoreOptions::new(table_options).ignore_delete(),
+ sequence_groups: Vec::new(),
+ grouped_fields: HashSet::new(),
+ })
+ }
+
+ pub(crate) fn new_with_schema(
+ table_options: &HashMap<String, String>,
+ table_name: &str,
+ table_fields: &[DataField],
+ output_fields: &[DataField],
+ primary_keys: &[String],
+ ) -> crate::Result<Self> {
+ let config = PartialUpdateConfig::new(table_options);
+ config.validate_read_mode(true, table_name)?;
+ let groups = config.validated_sequence_groups(table_fields,
primary_keys)?;
+ let field_indices: HashMap<&str, usize> = output_fields
+ .iter()
+ .enumerate()
+ .map(|(index, field)| (field.name(), index))
+ .collect();
+ let sequence_groups = groups
+ .into_iter()
+ .filter(|group| {
+ group
+ .sequence_fields
+ .iter()
+ .chain(group.protected_fields.iter())
+ .any(|field| field_indices.contains_key(field.as_str()))
+ })
+ .map(|group| {
+ let sequence_indices = group
+ .sequence_fields
+ .iter()
+ .map(|field| {
+
field_indices.get(field.as_str()).copied().ok_or_else(|| {
+ Error::UnexpectedError {
+ message: format!(
+ "Projected partial-update sequence group
is missing \
+ sequence field '{field}'"
+ ),
+ source: None,
+ }
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ let protected_indices = group
+ .protected_fields
+ .iter()
+ .filter_map(|field|
field_indices.get(field.as_str()).copied())
+ .collect();
+ Ok(RuntimeSequenceGroup {
+ sequence_indices,
+ protected_indices,
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ let grouped_fields = sequence_groups
+ .iter()
+ .flat_map(|group| {
+ group
+ .sequence_indices
+ .iter()
+ .chain(group.protected_indices.iter())
+ })
+ .copied()
+ .collect();
+
+ Ok(Self {
+ ignore_delete: CoreOptions::new(table_options).ignore_delete(),
+ sequence_groups,
+ grouped_fields,
})
}
}
@@ -224,8 +305,10 @@ impl MergeFunction for PartialUpdateMergeFunction {
.then_with(|| lhs_idx.cmp(&rhs_idx))
});
- let mut latest_non_null_by_col: Vec<Option<(usize, usize)>> =
+ let mut selected_by_col: Vec<Option<(usize, usize)>> =
vec![None; output_schema.fields().len()];
+ let mut group_sequence_rows: Vec<Option<(usize, usize)>> =
+ vec![None; self.sequence_groups.len()];
let mut saw_add = false;
for row_idx in ordered_row_indices {
@@ -240,11 +323,50 @@ impl MergeFunction for PartialUpdateMergeFunction {
}
saw_add = true;
- for (output_col_idx, latest_non_null) in
latest_non_null_by_col.iter_mut().enumerate() {
+ for (output_col_idx, selected) in
selected_by_col.iter_mut().enumerate() {
+ if self.grouped_fields.contains(&output_col_idx) {
+ continue;
+ }
let source_array = batch_buffer[row.batch_idx]
.column_for_output(output_col_idx,
source_output_col_indices);
if !source_array.is_null(row.row_idx) {
- *latest_non_null = Some((row.batch_idx, row.row_idx));
+ *selected = Some((row.batch_idx, row.row_idx));
+ }
+ }
+
+ for (group_idx, group) in self.sequence_groups.iter().enumerate() {
+ let sequence_is_empty =
group.sequence_indices.iter().all(|&output_col_idx| {
+ batch_buffer[row.batch_idx]
+ .column_for_output(output_col_idx,
source_output_col_indices)
+ .is_null(row.row_idx)
+ });
+ if sequence_is_empty {
+ continue;
+ }
+
+ let should_advance = match group_sequence_rows[group_idx] {
+ None => true,
+ Some((current_batch_idx, current_row_idx)) =>
compare_sequence_group_rows(
+ row.batch_idx,
+ row.row_idx,
+ current_batch_idx,
+ current_row_idx,
+ &group.sequence_indices,
+ batch_buffer,
+ source_output_col_indices,
+ )?
+ .is_ge(),
+ };
+ if !should_advance {
+ continue;
+ }
+
+ group_sequence_rows[group_idx] = Some((row.batch_idx,
row.row_idx));
+ for &output_col_idx in &group.sequence_indices {
+ selected_by_col[output_col_idx] = Some((row.batch_idx,
row.row_idx));
+ }
+ for &output_col_idx in &group.protected_indices {
+ selected_by_col[output_col_idx] = Some((row.batch_idx,
row.row_idx));
}
}
}
@@ -258,7 +380,7 @@ impl MergeFunction for PartialUpdateMergeFunction {
.iter()
.enumerate()
.map(|(output_col_idx, field)| {
- Ok(match latest_non_null_by_col[output_col_idx] {
+ Ok(match selected_by_col[output_col_idx] {
Some((batch_idx, row_idx)) => batch_buffer[batch_idx]
.column_for_output(output_col_idx,
source_output_col_indices)
.slice(row_idx, 1),
@@ -289,6 +411,40 @@ impl MergeFunction for PartialUpdateMergeFunction {
}
}
+fn compare_sequence_group_rows(
+ incoming_batch_idx: usize,
+ incoming_row_idx: usize,
+ current_batch_idx: usize,
+ current_row_idx: usize,
+ sequence_indices: &[usize],
+ batch_buffer: &[BufferedBatch],
+ source_output_col_indices: &[usize],
+) -> crate::Result<Ordering> {
+ for &output_col_idx in 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]
+ .column_for_output(output_col_idx, source_output_col_indices);
+ let comparator = make_comparator(
+ incoming,
+ current,
+ SortOptions {
+ descending: false,
+ nulls_first: true,
+ },
+ )
+ .map_err(|e| Error::DataInvalid {
+ 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);
+ if !ordering.is_eq() {
+ return Ok(ordering);
+ }
+ }
+ Ok(Ordering::Equal)
+}
+
// ---------------------------------------------------------------------------
// AggregateMergeFunction
// ---------------------------------------------------------------------------
@@ -1882,6 +2038,271 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_partial_update_sequence_groups_advance_independently() {
+ 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("seq_a", DataType::Int32, true),
+ Field::new("value_a", DataType::Int32, true),
+ Field::new("seq_b", DataType::Int32, true),
+ Field::new("value_b", DataType::Int32, true),
+ ]));
+ let output_schema = Arc::new(Schema::new(vec![
+ Field::new("pk", DataType::Int32, false),
+ Field::new("seq_a", DataType::Int32, true),
+ Field::new("value_a", DataType::Int32, true),
+ Field::new("seq_b", DataType::Int32, true),
+ Field::new("value_b", DataType::Int32, true),
+ ]));
+ let output_fields = vec![
+ DataField::new(0, "pk".into(),
crate::spec::DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "seq_a".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ DataField::new(
+ 2,
+ "value_a".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ DataField::new(
+ 3,
+ "seq_b".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ DataField::new(
+ 4,
+ "value_b".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ ];
+ let options = HashMap::from([
+ ("merge-engine".to_string(), "partial-update".to_string()),
+ (
+ "fields.seq_a.sequence-group".to_string(),
+ "value_a".to_string(),
+ ),
+ (
+ "fields.seq_b.sequence-group".to_string(),
+ "value_b".to_string(),
+ ),
+ ]);
+ let old = stream_from_batches(vec![RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int64Array::from(vec![1])),
+ Arc::new(Int8Array::from(vec![0])),
+ Arc::new(Int32Array::from(vec![10])),
+ Arc::new(Int32Array::from(vec![100])),
+ Arc::new(Int32Array::from(vec![10])),
+ Arc::new(Int32Array::from(vec![1000])),
+ ],
+ )
+ .unwrap()]);
+ let mixed = stream_from_batches(vec![RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int64Array::from(vec![2])),
+ Arc::new(Int8Array::from(vec![0])),
+ Arc::new(Int32Array::from(vec![9])),
+ Arc::new(Int32Array::from(vec![200])),
+ Arc::new(Int32Array::from(vec![11])),
+ Arc::new(Int32Array::from(vec![2000])),
+ ],
+ )
+ .unwrap()]);
+
+ let result = SortMergeReaderBuilder::new(
+ vec![old, mixed],
+ schema,
+ vec![0],
+ 1,
+ 2,
+ vec![],
+ vec![3, 4, 5, 6],
+ output_schema,
+ Box::new(
+ PartialUpdateMergeFunction::new_with_schema(
+ &options,
+ "test_table",
+ &output_fields,
+ &output_fields,
+ &["pk".to_string()],
+ )
+ .unwrap(),
+ ),
+ )
+ .build()
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(result.len(), 1);
+ let batch = &result[0];
+ assert_eq!(
+ batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 10
+ );
+ assert_eq!(
+ batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 100
+ );
+ assert_eq!(
+ batch
+ .column(3)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 11
+ );
+ assert_eq!(
+ batch
+ .column(4)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 2000
+ );
+ }
+
+ #[tokio::test]
+ async fn
test_partial_update_composite_sequence_group_skips_empty_sequence() {
+ 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("seq_major", DataType::Int32, true),
+ Field::new("seq_minor", DataType::Int32, true),
+ Field::new("value", DataType::Int32, true),
+ ]));
+ let output_schema = Arc::new(Schema::new(vec![
+ Field::new("pk", DataType::Int32, false),
+ Field::new("seq_major", DataType::Int32, true),
+ Field::new("seq_minor", DataType::Int32, true),
+ Field::new("value", DataType::Int32, true),
+ ]));
+ let output_fields = vec![
+ DataField::new(0, "pk".into(),
crate::spec::DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "seq_major".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ DataField::new(
+ 2,
+ "seq_minor".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ DataField::new(
+ 3,
+ "value".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ ];
+ let options = HashMap::from([
+ ("merge-engine".to_string(), "partial-update".to_string()),
+ (
+ "fields.seq_major,seq_minor.sequence-group".to_string(),
+ "value".to_string(),
+ ),
+ ]);
+ let stream = stream_from_batches(vec![RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 1, 1, 1, 1])),
+ Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5])),
+ Arc::new(Int8Array::from(vec![0, 0, 0, 0, 0])),
+ Arc::new(Int32Array::from(vec![
+ Some(1),
+ Some(2),
+ None,
+ Some(2),
+ Some(3),
+ ])),
+ Arc::new(Int32Array::from(vec![
+ Some(2),
+ Some(1),
+ None,
+ None,
+ Some(0),
+ ])),
+ Arc::new(Int32Array::from(vec![
+ Some(100),
+ Some(200),
+ Some(300),
+ Some(400),
+ None,
+ ])),
+ ],
+ )
+ .unwrap()]);
+
+ let result = SortMergeReaderBuilder::new(
+ vec![stream],
+ schema,
+ vec![0],
+ 1,
+ 2,
+ vec![],
+ vec![3, 4, 5],
+ output_schema,
+ Box::new(
+ PartialUpdateMergeFunction::new_with_schema(
+ &options,
+ "test_table",
+ &output_fields,
+ &output_fields,
+ &["pk".to_string()],
+ )
+ .unwrap(),
+ ),
+ )
+ .build()
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ let batch = &result[0];
+ assert_eq!(
+ batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 3
+ );
+ assert_eq!(
+ batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 0
+ );
+ assert!(batch.column(3).is_null(0));
+ }
+
#[tokio::test]
async fn test_partial_update_merge_rejects_delete_like_rows() {
let schema = make_schema();
diff --git a/docs/src/sql.md b/docs/src/sql.md
index fd95b396..a84a956e 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1886,8 +1886,16 @@ Set either `'ignore-delete' = 'true'` or
`UPDATE_BEFORE` rows during writes and when reading existing files. The default
and an explicit `false` continue to reject these retract rows. Once enabled on
an existing partial-update table, `ignore-delete` cannot be changed back to
-`false`. Advanced partial-update features such as sequence groups, partial
-aggregation, and remove-record-on-delete are not supported.
+`false`.
+
+Rust can read existing partial-update tables that define
+`fields.<sequence-field>.sequence-group=<protected-fields>`. Single and
+composite sequence fields, multiple independent groups, and projected reads are
+supported. Rows whose sequence tuple is entirely null do not update the group;
+an accepted group update can set protected fields to null. Rust table creation
+and writes still reject sequence-group options because write-side group merging
+is not implemented. Partial aggregation inside sequence groups and
+remove-record options are also not supported.
Rust can read fully materialized compacted files from deletion-vector-enabled
partial-update and aggregation tables. Every split must be raw-convertible,
