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 d57cac1 feat: support rowkind.field write semantics (#507)
d57cac1 is described below
commit d57cac12301cf8f269cd85e1fb9d883712a95797
Author: Pandas <[email protected]>
AuthorDate: Mon Jul 13 13:09:55 2026 +0800
feat: support rowkind.field write semantics (#507)
---
crates/paimon/src/spec/core_options.rs | 45 ++++++
crates/paimon/src/spec/mod.rs | 3 +
crates/paimon/src/spec/row_kind.rs | 83 ++++++++++
crates/paimon/src/spec/row_kind_filter.rs | 79 +++++++++
crates/paimon/src/spec/schema.rs | 160 ++++++++++++++++++-
crates/paimon/src/table/mod.rs | 1 +
crates/paimon/src/table/row_kind_generator.rs | 105 ++++++++++++
crates/paimon/src/table/table_write.rs | 117 ++++++++++++--
crates/paimon/tests/common/rowkind_helpers.rs | 206 ++++++++++++++++++++++++
crates/paimon/tests/rowkind_field_test.rs | 222 ++++++++++++++++++++++++++
10 files changed, 1006 insertions(+), 15 deletions(-)
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index b77ab95..1e1df3e 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -71,6 +71,13 @@ pub(crate) const
DISABLE_ALTER_COLUMN_NULL_TO_NOT_NULL_OPTION: &str =
const MERGE_ENGINE_OPTION: &str = "merge-engine";
const CHANGELOG_PRODUCER_OPTION: &str = "changelog-producer";
const ROWKIND_FIELD_OPTION: &str = "rowkind.field";
+const IGNORE_DELETE_OPTION: &str = "ignore-delete";
+const IGNORE_UPDATE_BEFORE_OPTION: &str = "ignore-update-before";
+const IGNORE_DELETE_FALLBACK_KEYS: &[&str] = &[
+ "first-row.ignore-delete",
+ "deduplicate.ignore-delete",
+ "partial-update.ignore-delete",
+];
const DEFAULT_COMMIT_MAX_RETRIES: u32 = 10;
const DEFAULT_COMMIT_TIMEOUT_MS: u64 = 120_000;
const DEFAULT_COMMIT_MIN_RETRY_WAIT_MS: u64 = 1_000;
@@ -384,6 +391,25 @@ impl<'a> CoreOptions<'a> {
self.options.get(ROWKIND_FIELD_OPTION).map(String::as_str)
}
+ /// Whether to ignore delete records (and all retracts when used by
`RowKindFilter`).
+ pub fn ignore_delete(&self) -> bool {
+ for key in
+
std::iter::once(IGNORE_DELETE_OPTION).chain(IGNORE_DELETE_FALLBACK_KEYS.iter().copied())
+ {
+ if let Some(v) = self.options.get(key) {
+ return v.eq_ignore_ascii_case("true");
+ }
+ }
+ false
+ }
+
+ /// Whether to ignore update-before records at write time.
+ pub fn ignore_update_before(&self) -> bool {
+ self.options
+ .get(IGNORE_UPDATE_BEFORE_OPTION)
+ .is_some_and(|v| v.eq_ignore_ascii_case("true"))
+ }
+
pub fn data_evolution_enabled(&self) -> bool {
self.options
.get(DATA_EVOLUTION_ENABLED_OPTION)
@@ -1637,4 +1663,23 @@ mod tests {
let options = HashMap::from([(SCAN_SNAPSHOT_ID_OPTION.to_string(),
"1".to_string())]);
assert!(CoreOptions::new(&options).validate_scan_options().is_ok());
}
+
+ #[test]
+ fn ignore_delete_reads_primary_and_fallback_keys() {
+ let fallback =
+ HashMap::from([("deduplicate.ignore-delete".to_string(),
"true".to_string())]);
+ let opts = CoreOptions::new(&fallback);
+ assert!(opts.ignore_delete());
+
+ let primary = HashMap::from([("ignore-delete".to_string(),
"true".to_string())]);
+ let opts = CoreOptions::new(&primary);
+ assert!(opts.ignore_delete());
+ }
+
+ #[test]
+ fn ignore_update_before_defaults_false() {
+ let options = HashMap::new();
+ let opts = CoreOptions::new(&options);
+ assert!(!opts.ignore_update_before());
+ }
}
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 4614e89..13f9c3d 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -110,3 +110,6 @@ pub use partition_statistics::PartitionStatistics;
mod row_kind;
pub use row_kind::RowKind;
+
+mod row_kind_filter;
+pub(crate) use row_kind_filter::RowKindFilter;
diff --git a/crates/paimon/src/spec/row_kind.rs
b/crates/paimon/src/spec/row_kind.rs
index 1e3ee5d..d6b8965 100644
--- a/crates/paimon/src/spec/row_kind.rs
+++ b/crates/paimon/src/spec/row_kind.rs
@@ -48,4 +48,87 @@ impl RowKind {
pub fn is_add(&self) -> bool {
matches!(self, RowKind::Insert | RowKind::UpdateAfter)
}
+
+ /// Byte value for serialization, matching Java `toByteValue()`.
+ pub fn to_value(self) -> i8 {
+ self as i8
+ }
+
+ /// Short string form: `"+I"`, `"-U"`, `"+U"`, `"-D"`.
+ pub fn short_string(self) -> &'static str {
+ match self {
+ RowKind::Insert => "+I",
+ RowKind::UpdateBefore => "-U",
+ RowKind::UpdateAfter => "+U",
+ RowKind::Delete => "-D",
+ }
+ }
+
+ /// Parse from short string; case-insensitive (Java
`RowKind.fromShortString`).
+ pub fn from_short_string(value: &str) -> crate::Result<Self> {
+ if value.len() != 2 {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Unsupported short string '{value}' for row
kind"),
+ source: None,
+ });
+ }
+ let bytes = value.as_bytes();
+ let sign = bytes[0].to_ascii_uppercase();
+ let letter = bytes[1].to_ascii_uppercase();
+ match (sign, letter) {
+ (b'+', b'I') => Ok(RowKind::Insert),
+ (b'-', b'U') => Ok(RowKind::UpdateBefore),
+ (b'+', b'U') => Ok(RowKind::UpdateAfter),
+ (b'-', b'D') => Ok(RowKind::Delete),
+ _ => Err(crate::Error::DataInvalid {
+ message: format!("Unsupported short string '{value}' for row
kind"),
+ source: None,
+ }),
+ }
+ }
+
+ /// Whether this is `UPDATE_BEFORE` or `DELETE` (Java `isRetract()`).
+ pub fn is_retract(self) -> bool {
+ matches!(self, RowKind::UpdateBefore | RowKind::Delete)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::RowKind;
+
+ #[test]
+ fn from_short_string_accepts_case_insensitive_tokens() {
+ assert_eq!(RowKind::from_short_string("+i").unwrap(), RowKind::Insert);
+ assert_eq!(
+ RowKind::from_short_string("-u").unwrap(),
+ RowKind::UpdateBefore
+ );
+ assert_eq!(
+ RowKind::from_short_string("+U").unwrap(),
+ RowKind::UpdateAfter
+ );
+ assert_eq!(RowKind::from_short_string("-D").unwrap(), RowKind::Delete);
+ }
+
+ #[test]
+ fn from_short_string_rejects_invalid() {
+ assert!(RowKind::from_short_string("INSERT").is_err());
+ assert!(RowKind::from_short_string("").is_err());
+ }
+
+ #[test]
+ fn short_string_round_trip() {
+ for kind in [
+ RowKind::Insert,
+ RowKind::UpdateBefore,
+ RowKind::UpdateAfter,
+ RowKind::Delete,
+ ] {
+ assert_eq!(
+ RowKind::from_short_string(kind.short_string()).unwrap(),
+ kind
+ );
+ }
+ }
}
diff --git a/crates/paimon/src/spec/row_kind_filter.rs
b/crates/paimon/src/spec/row_kind_filter.rs
new file mode 100644
index 0000000..78364cc
--- /dev/null
+++ b/crates/paimon/src/spec/row_kind_filter.rs
@@ -0,0 +1,79 @@
+// 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.
+
+//! Filter row kinds at write time (`ignore-delete`, `ignore-update-before`).
+//!
+//! Reference: Java `org.apache.paimon.utils.RowKindFilter`.
+
+use crate::spec::{CoreOptions, RowKind};
+
+pub struct RowKindFilter {
+ ignore_all_retracts: bool,
+ ignore_update_before: bool,
+}
+
+impl RowKindFilter {
+ pub fn of(options: CoreOptions<'_>) -> Option<Self> {
+ let ignore_all_retracts = options.ignore_delete();
+ let ignore_update_before = options.ignore_update_before();
+ if !ignore_all_retracts && !ignore_update_before {
+ return None;
+ }
+ Some(Self {
+ ignore_all_retracts,
+ ignore_update_before,
+ })
+ }
+
+ pub fn test(&self, row_kind: RowKind) -> bool {
+ match row_kind {
+ RowKind::Delete if self.ignore_all_retracts => false,
+ RowKind::UpdateBefore if self.ignore_update_before ||
self.ignore_all_retracts => false,
+ _ => true,
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::collections::HashMap;
+
+ #[test]
+ fn ignore_delete_drops_delete_and_update_before() {
+ let filter = RowKindFilter::of(CoreOptions::new(&HashMap::from([(
+ "ignore-delete".to_string(),
+ "true".to_string(),
+ )])))
+ .unwrap();
+ assert!(!filter.test(RowKind::Delete));
+ assert!(!filter.test(RowKind::UpdateBefore));
+ assert!(filter.test(RowKind::Insert));
+ }
+
+ #[test]
+ fn ignore_update_before_drops_only_update_before() {
+ let filter = RowKindFilter::of(CoreOptions::new(&HashMap::from([(
+ "ignore-update-before".to_string(),
+ "true".to_string(),
+ )])))
+ .unwrap();
+ assert!(filter.test(RowKind::Delete));
+ assert!(!filter.test(RowKind::UpdateBefore));
+ assert!(filter.test(RowKind::Insert));
+ }
+}
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 626889e..c00d6de 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -16,11 +16,11 @@
// under the License.
use crate::spec::core_options::{
- first_row_supports_changelog_producer, CoreOptions,
BLOB_DESCRIPTOR_FIELD_OPTION,
- BLOB_FIELD_OPTION, BLOB_VIEW_FIELD_OPTION, BUCKET_KEY_OPTION,
QUERY_AUTH_ENABLED_OPTION,
- SEQUENCE_FIELD_OPTION,
+ first_row_supports_changelog_producer, ChangelogProducer, CoreOptions,
MergeEngine,
+ BLOB_DESCRIPTOR_FIELD_OPTION, BLOB_FIELD_OPTION, BLOB_VIEW_FIELD_OPTION,
BUCKET_KEY_OPTION,
+ QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION,
};
-use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType};
+use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType,
VarCharType};
use crate::spec::{
remove_field_scoped_options, rename_field_scoped_options,
AggregationConfig, BlobType,
ColumnMove, ColumnMoveType, PartialUpdateConfig,
@@ -470,6 +470,11 @@ impl TableSchema {
AggregationConfig::new(&new_schema.options)
.validate_create_mode(&new_schema.primary_keys,
&new_schema.fields)?;
Schema::validate_first_row_changelog_producer(&new_schema.options)?;
+ Schema::validate_rowkind_field(
+ &new_schema.options,
+ &new_schema.primary_keys,
+ &new_schema.fields,
+ )?;
Ok(new_schema)
}
@@ -910,6 +915,7 @@ impl Schema {
PartialUpdateConfig::new(&options).validate_create_mode(!primary_keys.is_empty())?;
AggregationConfig::new(&options).validate_create_mode(&primary_keys,
&fields)?;
Self::validate_first_row_changelog_producer(&options)?;
+ Self::validate_rowkind_field(&options, &primary_keys, &fields)?;
Ok(Self {
fields,
@@ -1270,6 +1276,60 @@ impl Schema {
})
}
+ fn validate_rowkind_field(
+ options: &HashMap<String, String>,
+ primary_keys: &[String],
+ fields: &[DataField],
+ ) -> crate::Result<()> {
+ let core = CoreOptions::new(options);
+ let Some(field_name) = core.rowkind_field() else {
+ return Ok(());
+ };
+
+ if primary_keys.is_empty() {
+ return Err(crate::Error::ConfigInvalid {
+ message: "rowkind.field requires a primary-key
table".to_string(),
+ });
+ }
+
+ let merge_engine = core
+ .merge_engine()
+ .map_err(Self::options_error_to_config_invalid)?;
+ if merge_engine != MergeEngine::Deduplicate {
+ return Err(crate::Error::ConfigInvalid {
+ message: "rowkind.field only supports
merge-engine=deduplicate".to_string(),
+ });
+ }
+
+ let producer = core
+ .try_changelog_producer()
+ .map_err(Self::options_error_to_config_invalid)?;
+ if producer == ChangelogProducer::Input {
+ return Err(crate::Error::ConfigInvalid {
+ message: "rowkind.field cannot be used with
changelog-producer=input".to_string(),
+ });
+ }
+
+ let Some(field) = fields.iter().find(|f| f.name() == field_name) else {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!("Rowkind field '{field_name}' can not be
found in table schema."),
+ });
+ };
+
+ if !matches!(
+ field.data_type(),
+ DataType::VarChar(v) if v.length() == VarCharType::MAX_LENGTH
+ ) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "rowkind.field '{field_name}' must be STRING (unbounded
VARCHAR) type"
+ ),
+ });
+ }
+
+ Ok(())
+ }
+
fn options_error_to_config_invalid(error: crate::Error) -> crate::Error {
match error {
crate::Error::Unsupported { message } =>
crate::Error::ConfigInvalid { message },
@@ -2849,4 +2909,96 @@ mod tests {
panic!("expected Row type");
}
}
+
+ #[test]
+ fn rowkind_field_requires_primary_key_table() {
+ let err = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .option("rowkind.field", "op")
+ .build()
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message } if
message.contains("rowkind.field")),
+ "got {err:?}"
+ );
+ }
+
+ #[test]
+ fn rowkind_field_rejects_non_deduplicate_merge_engine() {
+ let err = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .option("merge-engine", "partial-update")
+ .option("rowkind.field", "op")
+ .build()
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("deduplicate")),
+ "got {err:?}"
+ );
+ }
+
+ #[test]
+ fn rowkind_field_rejects_changelog_producer_input() {
+ let err = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .option("changelog-producer", "input")
+ .option("rowkind.field", "op")
+ .build()
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("changelog-producer=input")),
+ "got {err:?}"
+ );
+ }
+
+ #[test]
+ fn rowkind_field_rejects_non_string_field() {
+ let err = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("op", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("rowkind.field", "op")
+ .build()
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("STRING")),
+ "got {err:?}"
+ );
+ }
+
+ #[test]
+ fn rowkind_field_rejects_missing_field() {
+ let err = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("rowkind.field", "op")
+ .build()
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("can not be found")),
+ "got {err:?}"
+ );
+ }
+
+ #[test]
+ fn rowkind_field_accepts_valid_deduplicate_table() {
+ Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .option("rowkind.field", "op")
+ .build()
+ .unwrap();
+ }
}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index e454e98..5a7030a 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -60,6 +60,7 @@ mod read_builder;
pub mod referenced_files;
pub(crate) mod rest_env;
pub(crate) mod row_id_predicate;
+mod row_kind_generator;
mod scan_trace;
pub(crate) mod schema_manager;
pub(crate) mod snapshot_commit;
diff --git a/crates/paimon/src/table/row_kind_generator.rs
b/crates/paimon/src/table/row_kind_generator.rs
new file mode 100644
index 0000000..07e8985
--- /dev/null
+++ b/crates/paimon/src/table/row_kind_generator.rs
@@ -0,0 +1,105 @@
+// 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.
+
+//! Parse row kind from a user STRING column (`rowkind.field` table option).
+//!
+//! Reference: Java `org.apache.paimon.table.sink.RowKindGenerator`.
+
+use arrow_array::{Array, RecordBatch, StringArray};
+
+use crate::spec::{RowKind, TableSchema};
+
+pub struct RowKindGenerator {
+ index: usize,
+}
+
+impl RowKindGenerator {
+ pub fn create(schema: &TableSchema, field_name: &str) ->
crate::Result<Self> {
+ let index = schema
+ .fields()
+ .iter()
+ .position(|f| f.name() == field_name)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "Can not find rowkind {field_name} in table schema: {:?}",
+ schema.fields().iter().map(|f|
f.name()).collect::<Vec<_>>()
+ ),
+ source: None,
+ })?;
+ Ok(Self { index })
+ }
+
+ pub fn generate(&self, batch: &RecordBatch, row: usize) ->
crate::Result<RowKind> {
+ let col = batch.column(self.index);
+ if col.is_null(row) {
+ return Err(crate::Error::DataInvalid {
+ message: "Row kind cannot be null.".to_string(),
+ source: None,
+ });
+ }
+ let strings = col.as_any().downcast_ref::<StringArray>().ok_or_else(||
{
+ crate::Error::DataInvalid {
+ message: "rowkind.field column must be
Utf8/String".to_string(),
+ source: None,
+ }
+ })?;
+ RowKind::from_short_string(strings.value(row))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Arc;
+
+ use arrow_array::{RecordBatch, StringArray};
+ use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
+
+ use crate::spec::{DataType, IntType, RowKind, Schema, TableSchema,
VarCharType};
+
+ use super::RowKindGenerator;
+
+ fn test_schema(op_col: &str) -> TableSchema {
+ TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(op_col, DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .option("rowkind.field", op_col)
+ .build()
+ .unwrap(),
+ )
+ }
+
+ #[test]
+ fn generate_parses_short_string_column() {
+ let schema = test_schema("op");
+ let gen = RowKindGenerator::create(&schema, "op").unwrap();
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ Arc::new(ArrowField::new("id", ArrowDataType::Int32, false)),
+ Arc::new(ArrowField::new("op", ArrowDataType::Utf8, false)),
+ ])),
+ vec![
+ Arc::new(arrow_array::Int32Array::from(vec![1])),
+ Arc::new(StringArray::from(vec!["-D"])),
+ ],
+ )
+ .unwrap();
+ assert_eq!(gen.generate(&batch, 0).unwrap(), RowKind::Delete);
+ }
+}
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 034e634..375dfa3 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -24,7 +24,8 @@ use crate::arrow::build_target_arrow_schema;
use crate::spec::PartitionComputer;
use crate::spec::{
first_row_supports_changelog_producer, BinaryRow, ChangelogProducer,
CoreOptions, DataField,
- DataType, MergeEngine, EMPTY_SERIALIZED_ROW, POSTPONE_BUCKET,
+ DataType, MergeEngine, RowKindFilter, EMPTY_SERIALIZED_ROW,
POSTPONE_BUCKET,
+ VALUE_KIND_FIELD_NAME,
};
use crate::table::bucket_assigner::{BucketAssignerEnum, PartitionBucketKey};
use crate::table::bucket_assigner_constant::ConstantBucketAssigner;
@@ -39,6 +40,7 @@ use crate::table::kv_file_writer::{KeyValueFileWriter,
KeyValueWriteConfig};
use crate::table::partition_filter::PartitionFilter;
use crate::table::postpone_file_writer::{PostponeFileWriter,
PostponeWriteConfig};
use crate::table::prepared_files::PreparedFiles;
+use crate::table::row_kind_generator::RowKindGenerator;
use crate::table::{SnapshotManager, Table, TableScan};
use crate::Result;
use arrow_array::RecordBatch;
@@ -121,6 +123,8 @@ pub struct TableWrite {
vector_file_format: Option<String>,
/// Whether the table has VECTOR fields requiring dedicated vector files.
has_dedicated_vector_fields: bool,
+ row_kind_generator: Option<RowKindGenerator>,
+ row_kind_filter: Option<RowKindFilter>,
}
impl TableWrite {
@@ -278,11 +282,13 @@ impl TableWrite {
});
}
- if has_primary_keys && core_options.rowkind_field().is_some() {
- return Err(crate::Error::Unsupported {
- message: "KeyValueFileWriter does not support
rowkind.field".to_string(),
- });
- }
+ let row_kind_generator = match core_options.rowkind_field() {
+ Some(field_name) if has_primary_keys => {
+ Some(RowKindGenerator::create(schema, field_name)?)
+ }
+ _ => None,
+ };
+ let row_kind_filter = RowKindFilter::of(core_options);
let target_bucket_row_number =
core_options.dynamic_bucket_target_row_num();
let bucket_function_type = core_options.bucket_function_type()?;
@@ -366,6 +372,8 @@ impl TableWrite {
has_blob_fields,
vector_file_format,
has_dedicated_vector_fields,
+ row_kind_generator,
+ row_kind_filter,
})
}
@@ -430,7 +438,12 @@ impl TableWrite {
return Ok(());
}
- let grouped = self.divide_by_partition_bucket(batch).await?;
+ let batch = self.enrich_rowkind_batch(batch)?;
+ if batch.num_rows() == 0 {
+ return Ok(());
+ }
+
+ let grouped = self.divide_by_partition_bucket(&batch).await?;
for ((partition_bytes, bucket), sub_batch) in grouped {
self.write_bucket(partition_bytes, bucket, sub_batch)
.await?;
@@ -475,16 +488,20 @@ impl TableWrite {
}
let mut result = Vec::with_capacity(groups.len());
+ let batch_has_value_kind = batch
+ .schema()
+ .column_with_name(VALUE_KIND_FIELD_NAME)
+ .is_some();
// Cross-partition writers must always include _VALUE_KIND to keep the
// Arrow schema stable across batches (some batches may have deletes,
// others may not — KeyValueFileWriter's concat_batches requires a
// consistent schema).
- let needs_value_kind =
- matches!(self.bucket_assigner,
BucketAssignerEnum::CrossPartition(_))
- || !output.deletes.is_empty();
+ let needs_value_kind = batch_has_value_kind
+ || matches!(self.bucket_assigner,
BucketAssignerEnum::CrossPartition(_))
+ || !output.deletes.is_empty();
for (key, row_indices) in groups {
let sub_batch = Self::take_rows(batch, &row_indices)?;
- let sub_batch = if needs_value_kind {
+ let sub_batch = if needs_value_kind && !batch_has_value_kind {
Self::add_value_kind_column(&sub_batch, 0)?
} else {
sub_batch
@@ -557,6 +574,84 @@ impl TableWrite {
})
}
+ fn enrich_rowkind_batch(&self, batch: &RecordBatch) -> Result<RecordBatch>
{
+ let Some(generator) = &self.row_kind_generator else {
+ return Ok(batch.clone());
+ };
+ if batch
+ .schema()
+ .column_with_name(VALUE_KIND_FIELD_NAME)
+ .is_some()
+ {
+ return Err(crate::Error::DataInvalid {
+ message: "batch must not contain _VALUE_KIND when
rowkind.field is configured"
+ .to_string(),
+ source: None,
+ });
+ }
+
+ let mut keep_rows = Vec::new();
+ let mut kinds = Vec::new();
+ for row in 0..batch.num_rows() {
+ let kind = generator.generate(batch, row)?;
+ if let Some(filter) = &self.row_kind_filter {
+ if !filter.test(kind) {
+ continue;
+ }
+ }
+ keep_rows.push(row);
+ kinds.push(kind.to_value());
+ }
+ if keep_rows.is_empty() {
+ return Ok(RecordBatch::new_empty(batch.schema()));
+ }
+ let filtered = Self::take_rows(batch, &keep_rows)?;
+ Self::add_per_row_value_kind_column(&filtered, kinds)
+ }
+
+ fn add_per_row_value_kind_column(
+ batch: &RecordBatch,
+ value_kinds: Vec<i8>,
+ ) -> Result<RecordBatch> {
+ use arrow_array::Int8Array;
+ use arrow_schema::{DataType as ArrowDataType, Field as ArrowField};
+
+ if batch.num_rows() != value_kinds.len() {
+ return Err(crate::Error::DataInvalid {
+ message: "rowkind batch row count must match value kind array
length".to_string(),
+ source: None,
+ });
+ }
+ if batch
+ .schema()
+ .column_with_name(VALUE_KIND_FIELD_NAME)
+ .is_some()
+ {
+ return Err(crate::Error::DataInvalid {
+ message: "batch already contains _VALUE_KIND
column".to_string(),
+ source: None,
+ });
+ }
+
+ let vk_array = Arc::new(Int8Array::from(value_kinds));
+ let vk_field = Arc::new(ArrowField::new(
+ VALUE_KIND_FIELD_NAME,
+ ArrowDataType::Int8,
+ false,
+ ));
+
+ let mut fields = batch.schema().fields().to_vec();
+ let mut columns: Vec<Arc<dyn arrow_array::Array>> =
batch.columns().to_vec();
+ fields.push(vk_field);
+ columns.push(vk_array);
+
+ let schema = Arc::new(arrow_schema::Schema::new(fields));
+ RecordBatch::try_new(schema, columns).map_err(|e|
crate::Error::DataInvalid {
+ message: format!("Failed to add per-row _VALUE_KIND column: {e}"),
+ source: None,
+ })
+ }
+
/// Write a batch directly to the writer for the given (partition, bucket).
async fn write_bucket(
&mut self,
diff --git a/crates/paimon/tests/common/rowkind_helpers.rs
b/crates/paimon/tests/common/rowkind_helpers.rs
new file mode 100644
index 0000000..4ff3c3a
--- /dev/null
+++ b/crates/paimon/tests/common/rowkind_helpers.rs
@@ -0,0 +1,206 @@
+// 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.
+
+//! Minimal helpers for `rowkind.field` integration tests (no compact APIs).
+
+#![allow(dead_code)]
+
+use arrow_array::{Array, Int32Array, Int8Array, RecordBatch, StringArray};
+use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
+use futures::StreamExt;
+use paimon::catalog::Identifier;
+use paimon::io::FileIOBuilder;
+use paimon::spec::{DataType, IntType, Schema, TableSchema, VarCharType,
VALUE_KIND_FIELD_NAME};
+use paimon::table::Table;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+
+pub async fn setup_dirs(file_io: &paimon::io::FileIO, table_path: &str) {
+ file_io
+ .mkdirs(&format!("{table_path}/schema"))
+ .await
+ .unwrap();
+ file_io
+ .mkdirs(&format!("{table_path}/snapshot"))
+ .await
+ .unwrap();
+}
+
+pub async fn persist_table_schema(
+ file_io: &paimon::io::FileIO,
+ table_path: &str,
+ schema: &TableSchema,
+) {
+ use bytes::Bytes;
+
+ let path = format!("{table_path}/schema/schema-{}", schema.id());
+ let json = serde_json::to_vec(schema).unwrap();
+ file_io
+ .new_output(&path)
+ .unwrap()
+ .write(Bytes::from(json))
+ .await
+ .unwrap();
+}
+
+pub fn rowkind_field_schema(kind_col: &str, options: &[(&str, &str)]) ->
TableSchema {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .column(kind_col, DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("rowkind.field", kind_col);
+ for (k, v) in options {
+ builder = builder.option(*k, *v);
+ }
+ TableSchema::new(0, &builder.build().unwrap())
+}
+
+pub fn memory_table(path: &str, schema: TableSchema) -> (paimon::io::FileIO,
Table) {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "rowkind_test"),
+ path.to_string(),
+ schema,
+ None,
+ );
+ (file_io, table)
+}
+
+pub fn make_batch_with_rowkind(
+ ids: Vec<i32>,
+ values: Vec<i32>,
+ kinds: Vec<&str>,
+ kind_col: &str,
+) -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, false),
+ ArrowField::new(kind_col, ArrowDataType::Utf8, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(Int32Array::from(values)),
+ Arc::new(StringArray::from(kinds)),
+ ],
+ )
+ .unwrap()
+}
+
+pub fn make_batch_with_rowkind_and_value_kind(
+ ids: Vec<i32>,
+ values: Vec<i32>,
+ kinds: Vec<&str>,
+ kind_col: &str,
+) -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, false),
+ ArrowField::new(kind_col, ArrowDataType::Utf8, false),
+ ArrowField::new(VALUE_KIND_FIELD_NAME, ArrowDataType::Int8, false),
+ ]));
+ let n = kinds.len();
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(Int32Array::from(values)),
+ Arc::new(StringArray::from(kinds)),
+ Arc::new(Int8Array::from(vec![0i8; n])),
+ ],
+ )
+ .unwrap()
+}
+
+pub async fn write_batch(table: &Table, batch: &RecordBatch) {
+ let builder = table.new_write_builder();
+ let mut w = builder.new_write().unwrap();
+ w.write_arrow_batch(batch).await.unwrap();
+ let msgs = w.prepare_commit().await.unwrap();
+ builder.new_commit().commit(msgs).await.unwrap();
+}
+
+pub async fn write_batch_expect_err(table: &Table, batch: &RecordBatch) ->
paimon::Error {
+ let builder = table.new_write_builder();
+ let mut w = builder.new_write().unwrap();
+ w.write_arrow_batch(batch).await.unwrap_err()
+}
+
+pub async fn scan_id_values(table: &Table) -> BTreeMap<i32, i32> {
+ let scan = table.new_read_builder().new_scan();
+ let plan = scan.plan().await.unwrap();
+ let read = table.new_read_builder().new_read().unwrap();
+ let mut stream = read.to_arrow(plan.splits()).unwrap();
+ let mut out = BTreeMap::new();
+ while let Some(batch) = stream.next().await {
+ let batch = batch.unwrap();
+ let ids = batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batch
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ for i in 0..ids.len() {
+ out.insert(ids.value(i), values.value(i));
+ }
+ }
+ out
+}
+
+pub async fn scan_pk_value_kind(table: &Table, kind_col: &str) -> Vec<(i32,
i32, String)> {
+ let scan = table.new_read_builder().new_scan();
+ let plan = scan.plan().await.unwrap();
+ let read = table.new_read_builder().new_read().unwrap();
+ let mut stream = read.to_arrow(plan.splits()).unwrap();
+ let mut out = Vec::new();
+ while let Some(batch) = stream.next().await {
+ let batch = batch.unwrap();
+ let ids = batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batch
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let kinds = batch
+ .column_by_name(kind_col)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ for i in 0..ids.len() {
+ out.push((ids.value(i), values.value(i),
kinds.value(i).to_string()));
+ }
+ }
+ out.sort_unstable();
+ out
+}
diff --git a/crates/paimon/tests/rowkind_field_test.rs
b/crates/paimon/tests/rowkind_field_test.rs
new file mode 100644
index 0000000..c7f9f08
--- /dev/null
+++ b/crates/paimon/tests/rowkind_field_test.rs
@@ -0,0 +1,222 @@
+// 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.
+
+//! Integration tests for `rowkind.field` (mirrors Java
`BatchFileStoreITCase`).
+
+#[path = "common/rowkind_helpers.rs"]
+mod rowkind_helpers;
+
+use rowkind_helpers::{
+ make_batch_with_rowkind, make_batch_with_rowkind_and_value_kind,
memory_table,
+ persist_table_schema, rowkind_field_schema, scan_id_values,
scan_pk_value_kind, setup_dirs,
+ write_batch, write_batch_expect_err,
+};
+use std::collections::HashMap;
+
+fn table_with_options(
+ table: &paimon::table::Table,
+ options: HashMap<String, String>,
+) -> paimon::table::Table {
+ table.copy_with_options(options)
+}
+
+#[tokio::test]
+async fn rowkind_field_insert_then_delete() {
+ let table_path = "memory:/rowkind_field/insert_delete";
+ let schema = rowkind_field_schema("rf", &[]);
+ let (file_io, table) = memory_table(table_path, schema);
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![1], vec![1], vec!["+I"], "rf"),
+ )
+ .await;
+ assert_eq!(
+ scan_pk_value_kind(&table, "rf").await,
+ vec![(1, 1, "+I".to_string())]
+ );
+
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![1], vec![2], vec!["-D"], "rf"),
+ )
+ .await;
+ assert!(scan_id_values(&table).await.is_empty());
+}
+
+#[tokio::test]
+async fn rowkind_field_update_tokens() {
+ let table_path = "memory:/rowkind_field/update_tokens";
+ let schema = rowkind_field_schema("rf", &[]);
+ let (file_io, table) = memory_table(table_path, schema);
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![1], vec![1], vec!["+I"], "rf"),
+ )
+ .await;
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![1], vec![1], vec!["-U"], "rf"),
+ )
+ .await;
+ assert!(scan_id_values(&table).await.is_empty());
+
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![1], vec![10], vec!["+U"], "rf"),
+ )
+ .await;
+ assert_eq!(
+ scan_pk_value_kind(&table, "rf").await,
+ vec![(1, 10, "+U".to_string())]
+ );
+}
+
+#[tokio::test]
+async fn rowkind_field_ignore_delete() {
+ let table_path = "memory:/rowkind_field/ignore_delete";
+ let schema = rowkind_field_schema("kind", &[("ignore-delete", "true")]);
+ let (file_io, table) = memory_table(table_path, schema);
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![1], vec![10], vec!["+I"], "kind"),
+ )
+ .await;
+ assert_eq!(
+ scan_pk_value_kind(&table, "kind").await,
+ vec![(1, 10, "+I".to_string())]
+ );
+
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![1], vec![10], vec!["-D"], "kind"),
+ )
+ .await;
+ assert_eq!(
+ scan_pk_value_kind(&table, "kind").await,
+ vec![(1, 10, "+I".to_string())]
+ );
+
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![1], vec![20], vec!["+I"], "kind"),
+ )
+ .await;
+ assert_eq!(
+ scan_pk_value_kind(&table, "kind").await,
+ vec![(1, 20, "+I".to_string())]
+ );
+}
+
+#[tokio::test]
+async fn rowkind_field_ignore_update_before() {
+ let table_path = "memory:/rowkind_field/ignore_update_before";
+ let schema = rowkind_field_schema("kind", &[]);
+ let (file_io, table) = memory_table(table_path, schema);
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![1, 2], vec![10, 20], vec!["+I", "+I"],
"kind"),
+ )
+ .await;
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![2], vec![20], vec!["-U"], "kind"),
+ )
+ .await;
+ assert_eq!(
+ scan_pk_value_kind(&table, "kind").await,
+ vec![(1, 10, "+I".to_string())]
+ );
+
+ let table = table_with_options(
+ &table,
+ HashMap::from([("ignore-update-before".to_string(),
"true".to_string())]),
+ );
+ assert_eq!(
+ scan_pk_value_kind(&table, "kind").await,
+ vec![(1, 10, "+I".to_string())],
+ "after option change before filtered -U"
+ );
+
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![1], vec![10], vec!["-U"], "kind"),
+ )
+ .await;
+ assert_eq!(
+ scan_pk_value_kind(&table, "kind").await,
+ vec![(1, 10, "+I".to_string())]
+ );
+
+ write_batch(
+ &table,
+ &make_batch_with_rowkind(vec![1], vec![10], vec!["-D"], "kind"),
+ )
+ .await;
+ assert!(scan_id_values(&table).await.is_empty());
+}
+
+#[tokio::test]
+async fn rowkind_field_rejects_illegal_token() {
+ let table_path = "memory:/rowkind_field/illegal_token";
+ let schema = rowkind_field_schema("rf", &[]);
+ let (file_io, table) = memory_table(table_path, schema);
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ let err = write_batch_expect_err(
+ &table,
+ &make_batch_with_rowkind(vec![1], vec![1], vec!["INSERT"], "rf"),
+ )
+ .await;
+ assert!(
+ matches!(err, paimon::Error::DataInvalid { ref message, .. }
+ if message.contains("Unsupported short string")),
+ "got {err:?}"
+ );
+}
+
+#[tokio::test]
+async fn rowkind_field_rejects_value_kind_conflict() {
+ let table_path = "memory:/rowkind_field/value_kind_conflict";
+ let schema = rowkind_field_schema("rf", &[]);
+ let (file_io, table) = memory_table(table_path, schema);
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ let err = write_batch_expect_err(
+ &table,
+ &make_batch_with_rowkind_and_value_kind(vec![1], vec![1], vec!["+I"],
"rf"),
+ )
+ .await;
+ assert!(
+ matches!(err, paimon::Error::DataInvalid { ref message, .. }
+ if message.contains("_VALUE_KIND")),
+ "got {err:?}"
+ );
+}
