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 9430ee89 [core] Align delete and update nullability with Java (#1179)
9430ee89 is described below
commit 9430ee898c8481ebabeab02ec70a295a7444add2
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 23:00:40 2026 +0800
[core] Align delete and update nullability with Java (#1179)
---
bindings/python/src/write.rs | 259 +++++++++++++++-
crates/paimon/src/table/kv_file_writer.rs | 14 +-
crates/paimon/src/table/table_write.rs | 78 +++--
.../tests/retained_delete_nullability_test.rs | 328 +++++++++++++++++++++
4 files changed, 654 insertions(+), 25 deletions(-)
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index 3e68e80e..7d850a81 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -525,13 +525,41 @@ fn arrow_table_batches(table: &Bound<'_, PyAny>) ->
PyResult<Vec<RecordBatch>> {
let mut batches = table
.call_method0("to_batches")?
.try_iter()?
- .map(|batch| crate::arrow_input::batch_from_pyarrow(&batch?))
+ .map(|batch| crate::arrow_input::write_batch_from_pyarrow(&batch?))
.collect::<PyResult<Vec<_>>>()?;
// Preserve empty input schema so core validates columns even with no rows.
if batches.is_empty() {
batches.push(RecordBatch::new_empty(Arc::new(
ArrowSchema::from_pyarrow_bound(&table.getattr("schema")?)?,
)));
+ } else if batches.len() > 1 {
+ // One Arrow Table has one schema, even when NULLs occur in only some
+ // chunks. The write importer relaxes transport fields per batch; union
+ // those flags so core can still require a uniform MERGE source schema.
+ let first = batches[0].schema();
+ let mut fields = first.fields().to_vec();
+ for batch in batches.iter().skip(1) {
+ for (field, actual) in
fields.iter_mut().zip(batch.schema().fields()) {
+ if actual.is_nullable() && !field.is_nullable() {
+ *field =
Arc::new(field.as_ref().clone().with_nullable(true));
+ }
+ }
+ }
+ let schema = Arc::new(ArrowSchema::new_with_metadata(
+ fields,
+ first.metadata().clone(),
+ ));
+ for batch in &mut batches {
+ if batch.schema() != schema {
+ *batch = RecordBatch::try_new_with_options(
+ schema.clone(),
+ batch.columns().to_vec(),
+ &arrow::array::RecordBatchOptions::default()
+ .with_row_count(Some(batch.num_rows())),
+ )
+ .map_err(|error| PyValueError::new_err(error.to_string()))?;
+ }
+ }
}
Ok(batches)
}
@@ -544,7 +572,7 @@ fn arrow_upsert_batches(input: &Bound<'_, PyAny>) ->
PyResult<Vec<RecordBatch>>
}
input
.try_iter()?
- .map(|batch| crate::arrow_input::batch_from_pyarrow(&batch?))
+ .map(|batch| crate::arrow_input::write_batch_from_pyarrow(&batch?))
.collect()
}
@@ -1659,6 +1687,233 @@ mod tests {
use paimon::spec::{IntType, Schema, TableSchema, VarCharType};
use pyo3::ffi::c_str;
+ #[test]
+ fn update_inputs_preserve_source_nulls_until_core_selects_payloads() {
+ use arrow::array::{Array, Int32Array};
+
+ Python::attach(|py| {
+ let locals = PyDict::new(py);
+ py.run(
+ c_str!(
+ r#"
+import pyarrow as pa
+schema = pa.schema([
+ pa.field('id', pa.int32(), nullable=False),
+ pa.field('value', pa.int32(), nullable=False),
+ pa.field('unused', pa.int32(), nullable=False, metadata={'role':
'unselected'}),
+], metadata={'origin': 'update'})
+source = pa.Table.from_pylist([
+ {'id': 0, 'value': 0, 'unused': 0},
+ {'id': 1, 'value': None, 'unused': None},
+ {'id': 1, 'value': 42, 'unused': None},
+], schema=schema).slice(1, 2)
+source.validate(full=True)
+batches = source.to_batches()
+"#
+ ),
+ None,
+ Some(&locals),
+ )
+ .unwrap();
+ let table = locals.get_item("source").unwrap().unwrap();
+ let batches = locals.get_item("batches").unwrap().unwrap();
+ for imported in [
+ arrow_table_batches(&table).unwrap(),
+ arrow_upsert_batches(&batches).unwrap(),
+ ] {
+ assert_eq!(imported.len(), 1);
+ let batch = &imported[0];
+ assert_eq!(batch.num_rows(), 2);
+ assert!(!batch.schema().field(0).is_nullable());
+ assert!(batch.schema().field(1).is_nullable());
+ assert!(batch.schema().field(2).is_nullable());
+ assert_eq!(batch.schema().metadata().get("origin").unwrap(),
"update");
+ assert_eq!(
+ batch.schema().field(2).metadata().get("role").unwrap(),
+ "unselected"
+ );
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert!(values.is_null(0));
+ assert_eq!(values.value(1), 42);
+ assert_eq!(batch.column(2).null_count(), 2);
+ }
+ let original =
+
ArrowSchema::from_pyarrow_bound(&table.getattr("schema").unwrap()).unwrap();
+ assert!(original.fields().iter().all(|field|
!field.is_nullable()));
+ });
+ }
+
+ #[test]
+ fn upsert_binding_validates_only_surviving_target_values() {
+ Python::attach(|py| {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::with_nullable(false)))
+ .column("value", DataType::Int(IntType::with_nullable(false)))
+ .column("keep", DataType::Int(IntType::with_nullable(false)))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let table = Arc::new(Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "upsert_source_nullability"),
+ "memory:/upsert-source-nullability".into(),
+ TableSchema::new(0, &schema),
+ None,
+ ));
+ let context =
UpdateContext::new(&WriteContext::new(table)).unwrap();
+ let locals = PyDict::new(py);
+ py.run(
+ c_str!(
+ r#"
+import pyarrow as pa
+schema = pa.schema([pa.field(name, pa.int32(), nullable=False)
+ for name in ('id', 'value', 'keep')])
+source = pa.Table.from_pylist([
+ {'id': 1, 'value': None, 'keep': None},
+ {'id': 1, 'value': 42, 'keep': 100},
+], schema=schema)
+invalid = source.slice(0, 1)
+"#
+ ),
+ None,
+ Some(&locals),
+ )
+ .unwrap();
+ let source = locals.get_item("source").unwrap().unwrap();
+ let messages = context
+ .upsert_by_arrow_with_key(py, &source, vec!["id".into()])
+ .unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].inner.new_files.len(), 1);
+ assert_eq!(messages[0].inner.new_files[0].row_count, 1);
+ let invalid = locals.get_item("invalid").unwrap().unwrap();
+ let error = context
+ .upsert_by_arrow_with_key(py, &invalid, vec!["id".into()])
+ .err()
+ .expect("a surviving NULL must still fail the declared target
schema")
+ .to_string();
+ assert!(error.contains("null"), "{error}");
+ });
+ }
+
+ #[test]
+ fn merge_binding_preserves_one_schema_across_source_chunks() {
+ use arrow::array::Int32Array;
+
+ Python::attach(|py| {
+ for reverse in [false, true] {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::with_nullable(false)))
+ .column("value",
DataType::Int(IntType::with_nullable(false)))
+ .column("keep",
DataType::Int(IntType::with_nullable(false)))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let table = Arc::new(Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "merge_chunk_nullability"),
+ "memory:/merge-chunk-nullability".into(),
+ TableSchema::new(0, &schema),
+ None,
+ ));
+ let seed = RecordBatch::try_new(
+
paimon::arrow::build_target_arrow_schema(table.schema().fields()).unwrap(),
+ vec![
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int32Array::from(vec![10])),
+ Arc::new(Int32Array::from(vec![100])),
+ ],
+ )
+ .unwrap();
+ py.detach(|| {
+ runtime().block_on(async {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&seed).await.unwrap();
+ builder
+ .new_commit()
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ writer.close().await;
+ })
+ });
+ let context =
UpdateContext::new(&WriteContext::new(table)).unwrap();
+ let locals = PyDict::new(py);
+ locals.set_item("reverse", reverse).unwrap();
+ py.run(
+ c_str!(
+ r#"
+import pyarrow as pa
+schema = pa.schema([
+ pa.field('id', pa.int32(), nullable=False),
+ pa.field('value', pa.int32(), nullable=False),
+ pa.field('keep', pa.int32(), nullable=False, metadata={'role':
'unselected'}),
+], metadata={'origin': 'chunked-merge'})
+def chunk(row):
+ return pa.RecordBatch.from_pylist([
+ {'id': 0, 'value': 0, 'keep': 0}, row], schema=schema).slice(1)
+chunks = [chunk({'id': 1, 'value': 42, 'keep': 100}),
+ chunk({'id': 2, 'value': None, 'keep': None}),
+ chunk({'id': None, 'value': 30, 'keep': 100})]
+source = pa.Table.from_batches(chunks[::-1] if reverse else chunks)
+encoded = pa.chunked_array([pa.DictionaryArray.from_arrays(
+ pa.array([0], type=pa.int8()), pa.array([None], type=pa.int32())) for _ in
chunks])
+source = source.append_column(pa.field('encoded_unused', encoded.type,
nullable=False), encoded)
+assert source.column('encoded_unused').null_count == 0
+assert source.column('encoded_unused').to_pylist() == [None, None, None]
+matched = [dict(assignments=[('value', 'source', 'value')], condition=None,
delete=False)]
+not_matched = []
+"#
+ ),
+ Some(&locals),
+ Some(&locals),
+ )
+ .unwrap();
+ let source = locals.get_item("source").unwrap().unwrap();
+ let matched = locals.get_item("matched").unwrap().unwrap();
+ let not_matched =
locals.get_item("not_matched").unwrap().unwrap();
+ let messages = context
+ .merge_into(
+ py,
+ Some(&source),
+ vec![("id".into(), "id".into())],
+ &matched,
+ ¬_matched,
+ )
+ .unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].inner.new_files.len(), 1);
+ let imported = arrow_table_batches(&source).unwrap();
+ assert_eq!(imported.len(), 3);
+ for batch in &imported {
+ assert_eq!(batch.schema(), imported[0].schema());
+ assert!(batch.schema().fields()[..3]
+ .iter()
+ .all(|field| field.is_nullable()));
+ assert!(!batch.schema().field(3).is_nullable());
+ assert_eq!(
+ batch.schema().metadata().get("origin").unwrap(),
+ "chunked-merge"
+ );
+ assert_eq!(
+
batch.schema().field(2).metadata().get("role").unwrap(),
+ "unselected"
+ );
+ }
+ let original =
+
ArrowSchema::from_pyarrow_bound(&source.getattr("schema").unwrap()).unwrap();
+ assert!(original.fields().iter().all(|field|
!field.is_nullable()));
+ }
+ });
+ }
+
#[test]
fn ignored_delete_null_payload_reaches_core_nullability_checks() {
Python::attach(|py| {
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index 02d35d64..2a81c33c 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -1357,12 +1357,24 @@ fn build_physical_fields(
name => value_fields
.iter()
.find(|field| field.name() == name)
- .cloned()
.ok_or_else(|| crate::Error::DataInvalid {
message: format!(
"Physical file column '{name}' is missing from the
table schema"
),
source: None,
+ })
+ .and_then(|field| {
+ // The table remains logically NOT NULL, but Java's KV
+ // value type makes non-key DELETE payloads nullable.
+ Ok(DataField::new(
+ field.id(),
+ field.name().into(),
+ field
+ .data_type()
+ .copy_with_nullable(arrow_field.is_nullable())?,
+ )
+ .with_description(field.description().map(str::to_owned))
+
.with_default_value(field.default_value().map(str::to_owned)))
}),
})
.collect()
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 7a5abef7..80edf85a 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -866,13 +866,10 @@ impl TableWrite {
Ok(Some(batch))
}
- /// Java resolves the event and checks top-level nullability before its
- /// RowKindFilter. Ignored rows never reach value extraction or nested
- /// writers, so their child values must not be normalized or validated.
+ /// Java resolves the event and checks logical top-level nullability before
+ /// its optional RowKindFilter, independently of physical KV nullability.
+ /// Ignored rows never reach value extraction or nested writers.
fn filter_write_input(&self, batch: &RecordBatch) ->
Result<Option<RecordBatch>> {
- let Some(filter) = &self.row_kind_filter else {
- return Ok(None);
- };
if self.primary_key_indices.is_empty() {
return Ok(None);
}
@@ -895,24 +892,17 @@ impl TableWrite {
vec![RowKind::Insert; batch.num_rows()]
};
- let schema = self.table.schema();
- let options = CoreOptions::new(schema.options());
// Java's deduplicate DELETE can omit non-key values. Sequence fields
// still identify the event; other engines and partition migrations
// require every declared top-level NOT NULL field.
- let partial_delete = self.merge_engine == MergeEngine::Deduplicate
- && schema
- .partition_keys()
- .iter()
- .all(|field| schema.primary_keys().contains(field));
- let sequence_fields = options.sequence_fields();
+ let delete_required_fields = self.delete_not_null_fields();
for (index, field) in self.write_schema.fields().iter().enumerate() {
if field.is_nullable() || batch.column(index).null_count() == 0 {
continue;
}
- let delete_required = !partial_delete
- || schema.primary_keys().contains(field.name())
- || sequence_fields.contains(&field.name().as_str());
+ let delete_required = delete_required_fields
+ .as_ref()
+ .is_none_or(|fields| fields.contains(field.name()));
for (row, kind) in kinds.iter().enumerate() {
if batch.column(index).is_null(row) && (*kind !=
RowKind::Delete || delete_required)
{
@@ -923,6 +913,9 @@ impl TableWrite {
}
}
}
+ let Some(filter) = &self.row_kind_filter else {
+ return Ok(None);
+ };
let rows = kinds
.into_iter()
.enumerate()
@@ -935,6 +928,29 @@ impl TableWrite {
}
}
+ /// PrimaryKeyFileStoreTable uses nullable non-key value fields for
ordinary
+ /// deduplication, while retaining the declared schema for event
validation.
+ fn delete_not_null_fields(&self) -> Option<HashSet<String>> {
+ let schema = self.table.schema();
+ if self.primary_key_indices.is_empty()
+ || self.merge_engine != MergeEngine::Deduplicate
+ || schema
+ .partition_keys()
+ .iter()
+ .any(|field| !schema.primary_keys().contains(field))
+ {
+ return None;
+ }
+ let mut fields: HashSet<_> =
schema.primary_keys().iter().cloned().collect();
+ fields.extend(
+ CoreOptions::new(schema.options())
+ .sequence_fields()
+ .into_iter()
+ .map(str::to_owned),
+ );
+ Some(fields)
+ }
+
pub(super) async fn write_partition_bucket_batch(
&mut self,
partition: Vec<u8>,
@@ -951,7 +967,26 @@ impl TableWrite {
fn validate_write_batch_schema(&self, batch: &RecordBatch) ->
Result<RecordBatch> {
let projected = self.project_full_write_input(batch)?;
- self.normalize_batch_schema(projected.as_ref().unwrap_or(batch),
&self.write_schema)
+ let value_schema = self.delete_not_null_fields().map(|required| {
+ arrow_schema::Schema::new_with_metadata(
+ self.write_schema
+ .fields()
+ .iter()
+ .map(|field| {
+ if required.contains(field.name()) {
+ field.clone()
+ } else {
+
Arc::new(field.as_ref().clone().with_nullable(true))
+ }
+ })
+ .collect::<Vec<_>>(),
+ self.write_schema.metadata().clone(),
+ )
+ });
+ self.normalize_batch_schema(
+ projected.as_ref().unwrap_or(batch),
+ value_schema.as_ref().unwrap_or(&self.write_schema),
+ )
}
fn normalize_batch_schema(
@@ -4795,15 +4830,14 @@ pub(in crate::table) mod tests {
);
let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
- table_write
+ let err = table_write
.write_arrow_batch(&make_batch_with_value_kind(vec![1], vec![10],
vec![4]))
.await
- .unwrap();
-
- let err = table_write.prepare_commit().await.unwrap_err();
+ .unwrap_err();
assert!(
matches!(err, crate::Error::DataInvalid { message, .. } if
message.contains("Invalid RowKind value"))
);
+ assert!(table_write.prepare_commit().await.unwrap().is_empty());
}
#[tokio::test]
diff --git a/crates/paimon/tests/retained_delete_nullability_test.rs
b/crates/paimon/tests/retained_delete_nullability_test.rs
new file mode 100644
index 00000000..b7dd245c
--- /dev/null
+++ b/crates/paimon/tests/retained_delete_nullability_test.rs
@@ -0,0 +1,328 @@
+// 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.
+
+mod common;
+
+use arrow_array::{
+ Array, ArrayRef, Int32Array, Int8Array, LargeBinaryArray,
LargeStringArray, RecordBatch,
+ StringArray,
+};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::spec::{
+ BlobType, DataType, Datum, IntType, Predicate, PredicateBuilder, Schema,
TableSchema,
+ VarCharType,
+};
+use paimon::table::Table;
+use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+use std::sync::Arc;
+
+async fn table(options: &[(&str, &str)]) -> Table {
+ table_with_type(
+ options,
+ DataType::VarChar(VarCharType::with_nullable(false,
VarCharType::MAX_LENGTH).unwrap()),
+ false,
+ )
+ .await
+}
+
+async fn table_with_type(
+ options: &[(&str, &str)],
+ value_type: DataType,
+ cross_partition: bool,
+) -> Table {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::with_nullable(false)))
+ .column("seq", DataType::Int(IntType::with_nullable(false)))
+ .column("value", value_type)
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("ignore-delete", "false");
+ if !cross_partition
+ && !options
+ .iter()
+ .any(|(key, value)| *key == "merge-engine" && *value ==
"first-row")
+ {
+ schema = schema.option("sequence.field", "seq");
+ }
+ if cross_partition {
+ schema = schema.partition_keys(["op"]).option("bucket", "-1");
+ }
+ for (key, value) in options {
+ schema = schema.option(*key, *value);
+ }
+ let path = "memory:/retained-delete";
+ let (io, table) = memory_table(path, TableSchema::new(0,
&schema.build().unwrap()));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ table
+}
+
+fn batch(
+ id: Option<i32>,
+ seq: Option<i32>,
+ value: Option<&str>,
+ kind: &str,
+ generated: bool,
+ large: bool,
+) -> RecordBatch {
+ let mut columns = vec![
+ ("id", Arc::new(Int32Array::from(vec![id])) as ArrayRef),
+ ("seq", Arc::new(Int32Array::from(vec![seq])) as ArrayRef),
+ (
+ "value",
+ if large {
+ Arc::new(LargeStringArray::from(vec![value])) as ArrayRef
+ } else {
+ Arc::new(StringArray::from(vec![value])) as ArrayRef
+ },
+ ),
+ ("op", Arc::new(StringArray::from(vec![kind])) as ArrayRef),
+ ];
+ if !generated {
+ columns.push((
+ "_VALUE_KIND",
+ Arc::new(Int8Array::from(vec![match kind {
+ "-D" => 3,
+ "-U" => 1,
+ "+U" => 2,
+ _ => 0,
+ }])) as ArrayRef,
+ ));
+ }
+ RecordBatch::try_from_iter(columns).unwrap()
+}
+
+async fn read_ids(table: &Table, predicate: Option<Predicate>) -> Vec<i32> {
+ let mut read = table.new_read_builder();
+ if let Some(predicate) = predicate {
+ read.with_filter(predicate);
+ }
+ let plan = read.new_scan().plan().await.unwrap();
+ let batches: Vec<_> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut ids = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect::<Vec<_>>();
+ ids.sort_unstable();
+ ids
+}
+
+#[tokio::test]
+async fn retained_delete_can_omit_non_key_not_null_values() {
+ for generated in [false, true] {
+ for large in [false, true] {
+ for bucket in [None, Some(0)] {
+ let options = if generated {
+ vec![("rowkind.field", "op")]
+ } else {
+ vec![("changelog-producer", "input")]
+ };
+ let table = table(&options).await;
+ for (seq, value, kind, expected) in [
+ (10, Some("before"), "+I", vec![1]),
+ (9, None, "-D", vec![1]),
+ (11, None, "-D", vec![]),
+ ] {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow(
+ &[batch(Some(1), Some(seq), value, kind,
generated, large)],
+ bucket,
+ )
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ for file in messages.iter().flat_map(|message| {
+
message.new_files.iter().chain(&message.new_changelog_files)
+ }) {
+ let path = format!("{}/bucket-0/{}", table.location(),
file.file_name);
+ let bytes = table
+ .file_io()
+ .new_input(&path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ let reader =
ParquetRecordBatchReaderBuilder::try_new(bytes).unwrap();
+ assert!(reader
+ .schema()
+ .field_with_name("value")
+ .unwrap()
+ .is_nullable());
+
assert!(!reader.schema().field_with_name("id").unwrap().is_nullable());
+ assert!(!reader
+ .schema()
+ .field_with_name("seq")
+ .unwrap()
+ .is_nullable());
+ for data in reader.build().unwrap() {
+ let data = data.unwrap();
+ assert_eq!(
+
data.column_by_name("value").unwrap().null_count(),
+ usize::from(value.is_none())
+ );
+ assert_eq!(
+ data.column_by_name("_VALUE_KIND")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .unwrap()
+ .value(0),
+ if kind == "-D" { 3 } else { 0 }
+ );
+ }
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(read_ids(&table, None).await, expected);
+ let predicate =
PredicateBuilder::new(table.schema().fields())
+ .equal("value", Datum::String("before".into()))
+ .unwrap();
+ assert_eq!(read_ids(&table, Some(predicate)).await,
expected);
+
assert!(!table.schema().fields()[2].data_type().is_nullable());
+ }
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn retained_events_keep_required_logical_fields() {
+ for generated in [false, true] {
+ let options = if generated {
+ vec![("rowkind.field", "op")]
+ } else {
+ vec![]
+ };
+ let table = table(&options).await;
+ for (id, seq, value, kind, field) in [
+ (None, Some(1), None, "-D", "id"),
+ (Some(1), None, None, "-D", "seq"),
+ (Some(1), Some(1), None, "+I", "value"),
+ (Some(1), Some(1), None, "-U", "value"),
+ (Some(1), Some(1), None, "+U", "value"),
+ ] {
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ let error = writer
+ .write_arrow_batch(&batch(id, seq, value, kind, generated,
false))
+ .await
+ .unwrap_err();
+ assert!(
+ error
+ .to_string()
+ .contains(&format!("non-null column({field})")),
+ "{error}"
+ );
+ assert!(writer.prepare_commit().await.unwrap().is_empty());
+ }
+ }
+ for merge in ["partial-update", "aggregation", "first-row"] {
+ let table = table(&[("rowkind.field", "op"), ("merge-engine",
merge)]).await;
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ let error = writer
+ .write_arrow_batch(&batch(Some(1), Some(1), None, "-D", true,
false))
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("non-null column(value)"),
+ "{error}"
+ );
+ assert!(writer.prepare_commit().await.unwrap().is_empty());
+ }
+ let table = table_with_type(
+ &[("rowkind.field", "op")],
+ DataType::VarChar(VarCharType::with_nullable(false,
VarCharType::MAX_LENGTH).unwrap()),
+ true,
+ )
+ .await;
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ let error = writer
+ .write_arrow_batch(&batch(Some(1), Some(1), None, "-D", true, false))
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("non-null column(value)"),
+ "{error}"
+ );
+}
+
+#[tokio::test]
+async fn retained_null_blob_delete_does_not_create_payload_files() {
+ let table = table_with_type(
+ &[("rowkind.field", "op")],
+ DataType::Blob(BlobType::with_nullable(false)),
+ false,
+ )
+ .await;
+ let mut managed_files = Vec::new();
+ for (seq, payload) in [(1, Some(b"managed payload".as_slice())), (2,
None)] {
+ let kind = if payload.is_some() { "+I" } else { "-D" };
+ let input = batch(Some(1), Some(seq), Some("unused"), kind, true,
false);
+ let mut columns = input.columns().to_vec();
+ columns[2] = Arc::new(LargeBinaryArray::from(vec![payload]));
+ let mut fields = input.schema().fields().to_vec();
+ fields[2] = Arc::new(arrow_schema::Field::new(
+ "value",
+ arrow_schema::DataType::LargeBinary,
+ true,
+ ));
+ let input =
+ RecordBatch::try_new(Arc::new(arrow_schema::Schema::new(fields)),
columns).unwrap();
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&input).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let files = table
+ .file_io()
+ .list_status_recursive(table.location())
+ .await
+ .unwrap()
+ .into_iter()
+ .filter(|file| file.path.ends_with(".managed.blob"))
+ .map(|file| file.path)
+ .collect::<Vec<_>>();
+ if payload.is_none() {
+ assert_eq!(files, managed_files);
+ } else {
+ assert!(!files.is_empty());
+ managed_files = files;
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(
+ read_ids(&table, None).await.len(),
+ usize::from(payload.is_some())
+ );
+ }
+}
