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 3ae85d3f [core] Align write validation and file metadata with Java
(#1172)
3ae85d3f is described below
commit 3ae85d3fc4e56125dda33418ff86761a6e28c5a9
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 22:09:13 2026 +0800
[core] Align write validation and file metadata with Java (#1172)
---
bindings/python/src/arrow_input.rs | 170 +++++++++
bindings/python/src/write.rs | 81 ++++-
crates/paimon/src/catalog/partition_listing.rs | 7 +-
crates/paimon/src/common/date_time.rs | 18 +
crates/paimon/src/spec/data_file.rs | 76 ++++
crates/paimon/src/spec/partition_utils.rs | 134 ++++---
crates/paimon/src/table/data_file_writer.rs | 5 +-
crates/paimon/src/table/kv_file_writer.rs | 5 +-
crates/paimon/src/table/postpone_file_writer.rs | 80 +---
crates/paimon/src/table/row_kind_generator.rs | 16 +-
crates/paimon/src/table/table_commit.rs | 74 +++-
crates/paimon/src/table/table_write.rs | 107 ++++--
crates/paimon/tests/creation_time_write_test.rs | 405 +++++++++++++++++++++
.../paimon/tests/filtered_write_validation_test.rs | 337 +++++++++++++++++
14 files changed, 1320 insertions(+), 195 deletions(-)
diff --git a/bindings/python/src/arrow_input.rs
b/bindings/python/src/arrow_input.rs
index 18cb5427..c2bd7a4f 100644
--- a/bindings/python/src/arrow_input.rs
+++ b/bindings/python/src/arrow_input.rs
@@ -79,6 +79,44 @@ pub(crate) fn batch_from_pyarrow(value: &Bound<'_, PyAny>)
-> PyResult<RecordBat
.map_err(|e| PyValueError::new_err(e.to_string()))
}
+pub(crate) fn write_batch_from_pyarrow(value: &Bound<'_, PyAny>) ->
PyResult<RecordBatch> {
+ let schema = Schema::from_pyarrow_bound(&value.getattr("schema")?)?;
+ let mut fields = schema.fields().to_vec();
+ let mut nullable_transport = false;
+ for (index, field) in schema.fields().iter().enumerate() {
+ if !field.is_nullable()
+ && value
+ .call_method1("column", (index,))?
+ .getattr("null_count")?
+ .extract::<usize>()?
+ > 0
+ {
+ fields[index] =
Arc::new(field.as_ref().clone().with_nullable(true));
+ nullable_transport = true;
+ }
+ }
+ if !nullable_transport {
+ return batch_from_pyarrow(value);
+ }
+
+ // PyArrow permits NULLs under a top-level NOT NULL transport field. A
+ // DELETE may legitimately omit non-key values, so retain those NULLs for
+ // core's declared-schema and per-event checks instead of rejecting them
+ // in Arrow's RecordBatch constructor. Nested fields stay unchanged.
+ let columns = value
+ .getattr("columns")?
+ .try_iter()?
+ .map(|column| array_from_pyarrow(&column?))
+ .collect::<PyResult<Vec<_>>>()?;
+ let rows = value.getattr("num_rows")?.extract::<usize>()?;
+ RecordBatch::try_new_with_options(
+ Arc::new(Schema::new_with_metadata(fields, schema.metadata().clone())),
+ columns,
+ &RecordBatchOptions::default().with_row_count(Some(rows)),
+ )
+ .map_err(|e| PyValueError::new_err(e.to_string()))
+}
+
fn normalize_sparse_unions(data: ArrayData) -> Result<ArrayData, ArrowError> {
if !contains_sparse_union(data.data_type()) {
return Ok(data);
@@ -150,6 +188,138 @@ mod tests {
};
use arrow::buffer::{NullBuffer, ScalarBuffer};
use arrow::datatypes::{Field, UnionFields};
+ use pyo3::{ffi::c_str, types::PyDict};
+
+ #[test]
+ fn write_batch_import_preserves_nullable_transport_for_core() {
+ 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('payload', pa.int32(), nullable=False, metadata={'field':
'value'}),
+ pa.field('nested', pa.struct([pa.field('child', pa.int32(),
nullable=False)])),
+], metadata={'scope': 'write'})
+batch = pa.RecordBatch.from_pylist([
+ {'id': 0, 'payload': 9, 'nested': {'child': 9}},
+ {'id': 1, 'payload': None, 'nested': {'child': 10}},
+ {'id': 2, 'payload': 42, 'nested': {'child': 11}},
+], schema=schema).slice(1, 2)
+"#
+ ),
+ None,
+ Some(&locals),
+ )
+ .unwrap();
+ let source = locals.get_item("batch").unwrap().unwrap();
+ let imported = write_batch_from_pyarrow(&source).unwrap();
+ assert_eq!(imported.num_rows(), 2);
+ let schema = imported.schema();
+ assert!(!schema.field(0).is_nullable());
+ assert!(schema.field(1).is_nullable());
+ assert_eq!(schema.metadata().get("scope").unwrap(), "write");
+ assert_eq!(schema.field(1).metadata().get("field").unwrap(),
"value");
+ let values = imported
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert!(values.is_null(0));
+ assert_eq!(values.value(1), 42);
+ let nested = imported
+ .column(2)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert!(!nested.fields()[0].is_nullable());
+ let children = nested
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(children.values().as_ref(), &[10, 11]);
+ // The input's declared schema stays untouched; the table
declaration
+ // remains the authority for the core's per-event nullability
checks.
+ let original =
Schema::from_pyarrow_bound(&source.getattr("schema").unwrap()).unwrap();
+ assert!(!original.field(1).is_nullable());
+ });
+ }
+
+ #[test]
+ fn write_batch_import_leaves_nested_value_validation_to_core() {
+ use arrow::array::{ListArray, MapArray};
+
+ Python::attach(|py| {
+ let locals = PyDict::new(py);
+ py.run(
+ c_str!(
+ r#"
+import pyarrow as pa
+child = pa.field('value', pa.int32(), nullable=False)
+schema = pa.schema([
+ pa.field('id', pa.int32(), nullable=False),
+ pa.field('items', pa.list_(child)),
+ pa.field('row', pa.struct([child])),
+ pa.field('map', pa.map_(pa.string(), child)),
+])
+batch = pa.RecordBatch.from_pylist([
+ {'id': 0, 'items': [0], 'row': {'value': 0}, 'map': [('key', 0)]},
+ {'id': 1, 'items': [None], 'row': {'value': None}, 'map': [('key', None)]},
+ {'id': 2, 'items': [42], 'row': {'value': 42}, 'map': [('key', 42)]},
+], schema=schema).slice(1, 2)
+"#
+ ),
+ None,
+ Some(&locals),
+ )
+ .unwrap();
+ let source = locals.get_item("batch").unwrap().unwrap();
+ let imported = write_batch_from_pyarrow(&source).unwrap();
+ let original =
Schema::from_pyarrow_bound(&source.getattr("schema").unwrap()).unwrap();
+ assert_eq!(imported.schema().as_ref(), &original);
+ assert_eq!(imported.num_rows(), 2);
+ let items = imported
+ .column(1)
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .unwrap();
+ assert!(!items.value_type().is_nested());
+ assert!(items.value(0).is_null(0));
+ assert_eq!(
+ items
+ .value(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 42
+ );
+ let row = imported
+ .column(2)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert!(row.column(0).is_null(0));
+ let map = imported
+ .column(3)
+ .as_any()
+ .downcast_ref::<MapArray>()
+ .unwrap();
+ assert!(map.value(0).column(1).is_null(0));
+ assert_eq!(
+ map.value(1)
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 42
+ );
+ });
+ }
fn input() -> ArrayData {
UnionArray::try_new(
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index 0ad668c5..3e68e80e 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -863,7 +863,7 @@ impl WriteState {
batch: &Bound<'_, PyAny>,
bucket: Option<i32>,
) -> PyResult<()> {
- let batch = RecordBatch::from_pyarrow_bound(batch)?;
+ let batch = crate::arrow_input::write_batch_from_pyarrow(batch)?;
let inner = self
.inner
.as_mut()
@@ -1651,6 +1651,85 @@ impl PyCommitMessage {
}
}
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use paimon::catalog::Identifier;
+ use paimon::io::FileIOBuilder;
+ use paimon::spec::{IntType, Schema, TableSchema, VarCharType};
+ use pyo3::ffi::c_str;
+
+ #[test]
+ fn ignored_delete_null_payload_reaches_core_nullability_checks() {
+ Python::attach(|py| {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .column("payload",
DataType::Int(IntType::with_nullable(false)))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("rowkind.field", "op")
+ .option("ignore-delete", "true")
+ .build()
+ .unwrap();
+ let table = Arc::new(Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "nullable_transport"),
+ "memory:/nullable-transport".into(),
+ TableSchema::new(0, &schema),
+ None,
+ ));
+ let locals = PyDict::new(py);
+ for large_string in [false, true] {
+ locals.set_item("large_string", large_string).unwrap();
+ for (kind, id, payload, error_column) in [
+ ("-D", Some(1), None, None),
+ ("-U", Some(1), None, Some("payload")),
+ ("+I", Some(1), None, Some("payload")),
+ ("-D", None, Some(42), Some("id")),
+ ] {
+ locals.set_item("kind", kind).unwrap();
+ locals.set_item("row_id", id).unwrap();
+ locals.set_item("payload", payload).unwrap();
+ py.run(
+ c_str!(
+ r#"
+import pyarrow as pa
+schema = pa.schema([
+ pa.field('id', pa.int32(), nullable=False),
+ pa.field('op', pa.large_string() if large_string else pa.string()),
+ pa.field('payload', pa.int32(), nullable=False),
+])
+batch = pa.RecordBatch.from_pylist([
+ {'id': row_id, 'op': kind, 'payload': payload},
+], schema=schema)
+"#
+ ),
+ None,
+ Some(&locals),
+ )
+ .unwrap();
+ let batch = locals.get_item("batch").unwrap().unwrap();
+ let mut state =
WriteContext::new(table.clone()).new_write(false).unwrap();
+ let result = state.write_arrow(py, &batch, None);
+ if let Some(column) = error_column {
+ let error = result.unwrap_err().to_string();
+ assert!(
+ error.contains(&format!(
+ "Cannot write null to non-null
column({column})"
+ )),
+ "{error}"
+ );
+ } else {
+ result.unwrap();
+ }
+ assert!(state.prepare_commit(py).unwrap().is_empty());
+ }
+ }
+ });
+ }
+}
+
#[pymethods]
impl PyCommitMessage {
/// Export the Java `CommitMessageSerializer` v14 body (no version header).
diff --git a/crates/paimon/src/catalog/partition_listing.rs
b/crates/paimon/src/catalog/partition_listing.rs
index 5557cf3b..1fcdbfde 100644
--- a/crates/paimon/src/catalog/partition_listing.rs
+++ b/crates/paimon/src/catalog/partition_listing.rs
@@ -129,12 +129,7 @@ fn aggregate_partition_entries_in_zone(
agg.file_size_in_bytes =
agg.file_size_in_bytes.wrapping_add(file.file_size);
agg.file_count += 1;
agg.total_buckets = entry.total_buckets();
- if let Some(time) = file.creation_time {
- let millis = time.timestamp_millis();
- let time =
-
crate::common::date_time::local_to_utc_seconds(millis.div_euclid(1000), zone)?
- * 1000
- + millis.rem_euclid(1000);
+ if let Some(time) = file.creation_time_epoch_millis_in_zone(zone)? {
agg.last_file_creation_time = Some(
agg.last_file_creation_time
.map_or(time, |old| old.max(time)),
diff --git a/crates/paimon/src/common/date_time.rs
b/crates/paimon/src/common/date_time.rs
index e80b1dfc..44fdcbe5 100644
--- a/crates/paimon/src/common/date_time.rs
+++ b/crates/paimon/src/common/date_time.rs
@@ -19,6 +19,24 @@
use chrono::{Datelike, NaiveDateTime, Timelike};
+/// Java Timestamp.now stores system-local wall fields, without a time zone.
+/// The UTC label is only the carrier used by manifest timestamp-millis fields.
+pub(crate) fn current_local_timestamp() -> chrono::DateTime<chrono::Utc> {
+ // DataFileMeta persists TIMESTAMP(3), as Java now().toMillisTimestamp().
+ let now = chrono::Utc::now();
+ let now = now
+ .with_nanosecond(now.timestamp_subsec_millis() * 1_000_000)
+ .unwrap();
+ local_timestamp(now, &jiff::tz::TimeZone::system())
+}
+
+pub(crate) fn local_timestamp(
+ instant: chrono::DateTime<chrono::Utc>,
+ zone: &jiff::tz::TimeZone,
+) -> chrono::DateTime<chrono::Utc> {
+ instant +
chrono::Duration::seconds(i64::from(local_offset(instant.timestamp(), zone)))
+}
+
/// Format millis-since-midnight to `HH:mm:ss[.f...]` using Java
`DateTimeUtils` semantics.
///
/// Matches Java `DateTimeUtils.formatTimestampMillis(time, precision)`:
diff --git a/crates/paimon/src/spec/data_file.rs
b/crates/paimon/src/spec/data_file.rs
index 97e2d2d0..842daae6 100644
--- a/crates/paimon/src/spec/data_file.rs
+++ b/crates/paimon/src/spec/data_file.rs
@@ -76,6 +76,8 @@ pub struct DataFileMeta {
deserialize_with = "from_millis_opt",
default
)]
+ /// Local wall-clock TIMESTAMP(3), matching Java Timestamp.now().
+ /// The UTC label carries the serialized fields; it does not identify an
instant.
pub creation_time: Option<DateTime<Utc>>,
#[serde(rename = "_DELETE_ROW_COUNT")]
// rowCount = add_row_count + delete_row_count.
@@ -250,6 +252,25 @@ fn read_compact_millis_as_utc(
}
impl DataFileMeta {
+ /// Resolve the stored local creation timestamp in the system time zone,
+ /// matching Java DataFileMeta.creationTimeEpochMillis().
+ pub fn creation_time_epoch_millis(&self) -> crate::Result<Option<i64>> {
+ self.creation_time_epoch_millis_in_zone(&jiff::tz::TimeZone::system())
+ }
+
+ pub(crate) fn creation_time_epoch_millis_in_zone(
+ &self,
+ zone: &jiff::tz::TimeZone,
+ ) -> crate::Result<Option<i64>> {
+ self.creation_time
+ .map(|time| {
+ let millis = time.timestamp_millis();
+
crate::common::date_time::local_to_utc_seconds(millis.div_euclid(1000), zone)
+ .map(|seconds| seconds * 1000 + millis.rem_euclid(1000))
+ })
+ .transpose()
+ }
+
/// Placeholder for `row_count` when the producer cannot know how many rows
/// a file holds.
///
@@ -682,6 +703,61 @@ mod tests {
}
}
+ #[test]
+ fn creation_time_is_local_wall_time_in_every_wire_format() {
+ use crate::common::date_time::local_timestamp;
+ for (zone, instant, expected_wall) in [
+ (
+ "UTC",
+ "1969-12-31T23:59:59.999Z",
+ "1969-12-31T23:59:59.999Z",
+ ),
+ (
+ "Asia/Shanghai",
+ "2024-01-02T04:34:56.789Z",
+ "2024-01-02T12:34:56.789Z",
+ ),
+ (
+ "America/New_York",
+ "2024-01-02T17:34:56.789Z",
+ "2024-01-02T12:34:56.789Z",
+ ),
+ (
+ "America/New_York",
+ "2024-11-03T05:30:00.456Z",
+ "2024-11-03T01:30:00.456Z",
+ ),
+ ] {
+ let zone = jiff::tz::TimeZone::get(zone).unwrap();
+ let instant = instant.parse::<DateTime<Utc>>().unwrap();
+ let wall = expected_wall.parse::<DateTime<Utc>>().unwrap();
+ let mut metadata = sample_full_data_file_meta();
+ metadata.creation_time = Some(local_timestamp(instant, &zone));
+ assert_eq!(metadata.creation_time, Some(wall));
+ for decoded in [
+
serde_json::from_slice::<DataFileMeta>(&serde_json::to_vec(&metadata).unwrap())
+ .unwrap(),
+ DataFileMeta::from_serialized_row_data(
+ &metadata.to_serialized_row_data().unwrap(),
+ DataFileMetaRowLayout::CURRENT,
+ )
+ .unwrap(),
+ ] {
+ assert_eq!(decoded.creation_time, Some(wall));
+ assert_eq!(
+ decoded.creation_time_epoch_millis_in_zone(&zone).unwrap(),
+ Some(instant.timestamp_millis())
+ );
+ }
+ }
+ assert_eq!(
+ sample_minimal_data_file_meta()
+ .creation_time_epoch_millis()
+ .unwrap(),
+ None
+ );
+ }
+
#[test]
fn data_file_meta_row_data_round_trips() {
for meta in [
diff --git a/crates/paimon/src/spec/partition_utils.rs
b/crates/paimon/src/spec/partition_utils.rs
index 85c1bf3a..f9da9b8b 100644
--- a/crates/paimon/src/spec/partition_utils.rs
+++ b/crates/paimon/src/spec/partition_utils.rs
@@ -408,9 +408,10 @@ fn format_partition_value(
s.to_string()
}
- DataType::Binary(_) | DataType::VarBinary(_) if !legacy => {
- // Java's BinaryToStringCastRule wraps the raw bytes in a
- // BinaryString, whose toString decodes them as UTF-8.
+ DataType::Binary(_) | DataType::VarBinary(_) => {
+ // Java uses BinaryToStringCastRule even for legacy names: a raw
+ // byte[].toString() contains a new allocation identity on each
read.
+ // BinaryString.toString decodes the bytes as UTF-8.
let value = decode_java_utf8(row.get_binary(pos)?);
if is_java_whitespace_only(&value) {
return Ok(default_partition_name.to_string());
@@ -467,9 +468,7 @@ fn format_partition_value(
}
}
- DataType::Binary(_)
- | DataType::VarBinary(_)
- | DataType::Variant(_)
+ DataType::Variant(_)
| DataType::Blob(_)
| DataType::Array(_)
| DataType::Map(_)
@@ -1435,14 +1434,6 @@ mod tests {
#[test]
fn test_unsupported_types() {
- // Legacy binary uses byte[].toString() in Java, which contains an
- // allocation-specific identity hash and cannot name a stable path.
- assert_single_partition_err(
- "data",
- DataType::Binary(BinaryType::new(10).unwrap()),
- |b| b.write_int(0, 0),
- true,
- );
// Array
assert_single_partition_err(
"arr",
@@ -1505,70 +1496,75 @@ mod tests {
#[test]
fn test_binary_partition_uses_utf8_cast_and_path_escaping() {
- assert_single_partition(
- "bin",
- DataType::Binary(BinaryType::new(16).unwrap()),
- |b| b.write_string(0, "a/b=c"),
- "bin=a%2Fb%3Dc/",
- false,
- );
- assert_single_partition(
- "bin",
- DataType::VarBinary(VarBinaryType::new(16).unwrap()),
- |b| b.write_string(0, "汉字"),
- "bin=汉字/",
- false,
- );
- assert_single_partition(
- "bin",
- DataType::VarBinary(VarBinaryType::new(16).unwrap()),
- |b| b.write_string(0, " "),
- "bin=__DEFAULT_PARTITION__/",
- false,
- );
- }
-
- #[test]
- fn test_binary_partition_matches_java_whitespace() {
- // Modern JVMs (Java 11/17) treat U+001C as whitespace but U+180E as a
- // format character, and never fold U+00A0 or U+2007. Rust str::trim
- // differs for controls and NBSP, so we match Java explicitly.
- for (value, expected) in [
- (b"\x1c".as_slice(), "bin=__DEFAULT_PARTITION__/"),
- // U+180E is whitespace only on legacy JDK 8; a modern JVM keeps
it.
- ("\u{180E}".as_bytes(), "bin=\u{180E}/"),
- ("\u{00A0}".as_bytes(), "bin=\u{00A0}/"),
- ("\u{2007}".as_bytes(), "bin=\u{2007}/"),
- ] {
+ for legacy in [false, true] {
+ assert_single_partition(
+ "bin",
+ DataType::Binary(BinaryType::new(16).unwrap()),
+ |b| b.write_string(0, "a/b=c"),
+ "bin=a%2Fb%3Dc/",
+ legacy,
+ );
assert_single_partition(
"bin",
DataType::VarBinary(VarBinaryType::new(16).unwrap()),
- |b| b.write_bytes(0, value),
- expected,
- false,
+ |b| b.write_string(0, "汉字"),
+ "bin=汉字/",
+ legacy,
);
+ assert_single_partition(
+ "bin",
+ DataType::VarBinary(VarBinaryType::new(16).unwrap()),
+ |b| b.write_string(0, " "),
+ "bin=__DEFAULT_PARTITION__/",
+ legacy,
+ );
+ }
+ }
+
+ #[test]
+ fn test_binary_partition_matches_java_whitespace() {
+ // Modern JVMs treat U+001C as whitespace but U+180E as a format
+ // character, and never fold U+00A0 or U+2007. Rust str::trim differs
+ // for controls and NBSP, so match Java for both naming modes.
+ for legacy in [false, true] {
+ for (value, expected) in [
+ (b"\x1c".as_slice(), "bin=__DEFAULT_PARTITION__/"),
+ ("\u{180E}".as_bytes(), "bin=\u{180E}/"),
+ ("\u{00A0}".as_bytes(), "bin=\u{00A0}/"),
+ ("\u{2007}".as_bytes(), "bin=\u{2007}/"),
+ ] {
+ assert_single_partition(
+ "bin",
+ DataType::VarBinary(VarBinaryType::new(16).unwrap()),
+ |b| b.write_bytes(0, value),
+ expected,
+ legacy,
+ );
+ }
}
}
#[test]
fn test_binary_partition_matches_java_malformed_utf8() {
- // Expected replacements were checked against JDK 8 new String(bytes,
UTF_8).
- for (value, expected) in [
- (b"\xED\xA0\x80".as_slice(), "bin=\u{FFFD}/"),
- (b"\xED\xA0".as_slice(), "bin=\u{FFFD}/"),
- (b"\xED\xA0A".as_slice(), "bin=\u{FFFD}A/"),
- (b"\xED\xA0\xFF".as_slice(), "bin=\u{FFFD}\u{FFFD}/"),
- (b"\xE0\x80\x80".as_slice(), "bin=\u{FFFD}\u{FFFD}\u{FFFD}/"),
- (b"\xE2\x82".as_slice(), "bin=\u{FFFD}/"),
- (b"\xE2(\xA1".as_slice(), "bin=\u{FFFD}(\u{FFFD}/"),
- ] {
- assert_single_partition(
- "bin",
- DataType::VarBinary(VarBinaryType::new(16).unwrap()),
- |b| b.write_bytes(0, value),
- expected,
- false,
- );
+ // Expected replacements were checked against new String(bytes, UTF_8).
+ for legacy in [false, true] {
+ for (value, expected) in [
+ (b"\xED\xA0\x80".as_slice(), "bin=\u{FFFD}/"),
+ (b"\xED\xA0".as_slice(), "bin=\u{FFFD}/"),
+ (b"\xED\xA0A".as_slice(), "bin=\u{FFFD}A/"),
+ (b"\xED\xA0\xFF".as_slice(), "bin=\u{FFFD}\u{FFFD}/"),
+ (b"\xE0\x80\x80".as_slice(), "bin=\u{FFFD}\u{FFFD}\u{FFFD}/"),
+ (b"\xE2\x82".as_slice(), "bin=\u{FFFD}/"),
+ (b"\xE2(\xA1".as_slice(), "bin=\u{FFFD}(\u{FFFD}/"),
+ ] {
+ assert_single_partition(
+ "bin",
+ DataType::VarBinary(VarBinaryType::new(16).unwrap()),
+ |b| b.write_bytes(0, value),
+ expected,
+ legacy,
+ );
+ }
}
}
diff --git a/crates/paimon/src/table/data_file_writer.rs
b/crates/paimon/src/table/data_file_writer.rs
index 099b9dea..bb7bea28 100644
--- a/crates/paimon/src/table/data_file_writer.rs
+++ b/crates/paimon/src/table/data_file_writer.rs
@@ -36,7 +36,6 @@ use crate::spec::stats::BinaryTableStats;
use crate::spec::{CoreOptions, DataField, DataFileMeta, EMPTY_SERIALIZED_ROW};
use crate::Result;
use arrow_array::RecordBatch;
-use chrono::Utc;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::task::JoinSet;
@@ -452,6 +451,8 @@ impl DataFileWriter {
if let Some(name) = sidecar_name {
meta.extra_files.push(name);
}
+ // Java constructs the result after all auxiliary outputs finish.
+ meta.creation_time =
Some(crate::common::date_time::current_local_timestamp());
Ok(meta)
})
}
@@ -600,7 +601,7 @@ impl DataFileWriter {
schema_id,
level: 0,
extra_files: vec![],
- creation_time: Some(Utc::now()),
+ creation_time: None,
delete_row_count: Some(0),
embedded_index: None,
file_source,
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index cdd8b33c..02d35d64 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -52,7 +52,6 @@ use arrow_array::{Array, BooleanArray, Int64Array, Int8Array,
RecordBatch, UInt3
use arrow_ord::sort::{lexsort_to_indices, SortColumn, SortOptions};
use arrow_row::{RowConverter, SortField};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
-use chrono::Utc;
use std::collections::HashMap;
use std::sync::Arc;
@@ -774,7 +773,7 @@ impl KeyValueFileWriter {
schema_id: self.config.schema_id,
level: 0,
extra_files: vec![],
- creation_time: Some(Utc::now()),
+ creation_time: None,
delete_row_count: Some(write.delete_row_count),
embedded_index: None,
file_source: Some(0), // FileSource.APPEND
@@ -829,6 +828,8 @@ impl KeyValueFileWriter {
)
.await?;
+ // Java creates file metadata after the index and Blob references
finish.
+ meta.creation_time =
Some(crate::common::date_time::current_local_timestamp());
Ok(meta)
}
diff --git a/crates/paimon/src/table/postpone_file_writer.rs
b/crates/paimon/src/table/postpone_file_writer.rs
index 449d4471..952e6ca8 100644
--- a/crates/paimon/src/table/postpone_file_writer.rs
+++ b/crates/paimon/src/table/postpone_file_writer.rs
@@ -40,9 +40,7 @@ use
crate::table::data_file_index_writer::{DataFileIndexWriter, FileIndexOptions
use crate::table::kv_file_writer::build_physical_schema;
use crate::Result;
use arrow_array::{Int64Array, Int8Array, RecordBatch};
-use chrono::{DateTime, Utc};
use std::sync::Arc;
-use tokio::task::JoinSet;
/// Configuration for [`PostponeFileWriter`].
pub(crate) struct PostponeWriteConfig {
@@ -84,12 +82,8 @@ pub(crate) struct PostponeFileWriter {
current_file_name: Option<String>,
current_file_path: Option<DataFilePath>,
current_stats: PostponeFileStats,
- /// Timestamp captured when the current file was opened (used for
deterministic replay order).
- current_file_creation_time: DateTime<Utc>,
written_files: Vec<DataFileMeta>,
created_paths: Vec<String>,
- /// Background file close tasks spawned during rolling.
- in_flight_closes: JoinSet<Result<DataFileMeta>>,
resources: Option<ResourceContext>,
}
@@ -130,10 +124,8 @@ impl PostponeFileWriter {
current_file_name: None,
current_file_path: None,
current_stats: PostponeFileStats::default(),
- current_file_creation_time: Utc::now(),
written_files: Vec::new(),
created_paths: Vec::new(),
- in_flight_closes: JoinSet::new(),
resources: None,
})
}
@@ -220,11 +212,13 @@ impl PostponeFileWriter {
self.current_stats
.add_batch(&batch, &self.config, start_seq, end_seq)?;
- // Roll to a new file if target size is reached — close in background
+ // Java creates metadata at result/close time and replays pending files
+ // by that time. Finish closes in arrival order, including equal-millis
+ // ties, so asynchronous completion cannot reorder one writer's rows.
if self.current_writer.as_ref().unwrap().num_bytes() as i64 >=
self.config.target_file_size
|| self.current_stats.row_count >= self.config.target_file_row_num
{
- self.roll_file();
+ self.close_current_file().await?;
}
// Flush row group if in-progress buffer exceeds write_buffer_size
@@ -245,7 +239,6 @@ impl PostponeFileWriter {
self.current_blob_references = None;
self.current_file_name = None;
self.current_file_path = None;
- while self.in_flight_closes.join_next().await.is_some() {}
for path in self.created_paths.drain(..) {
let _ = self.file_io.delete_file(&path).await;
}
@@ -263,62 +256,11 @@ impl PostponeFileWriter {
async fn finish(&mut self) -> Result<Vec<DataFileMeta>> {
self.close_current_file().await?;
- while let Some(result) = self.in_flight_closes.join_next().await {
- let meta = result.map_err(|e| crate::Error::DataInvalid {
- message: format!("Background file close task panicked: {e}"),
- source: None,
- })??;
- self.written_files.push(meta);
- }
- // Java replays postpone files by millisecond creation time, retaining
- // manifest order for ties. Background closes must not reorder
arrivals.
- self.written_files
- .sort_by_key(|file| file.min_sequence_number);
self.managed_blob_writer.prepare_commit().await?;
self.created_paths.clear();
Ok(std::mem::take(&mut self.written_files))
}
- /// Spawn the current writer's close in the background for non-blocking
rolling.
- fn roll_file(&mut self) {
- let writer = match self.current_writer.take() {
- Some(w) => w,
- None => return,
- };
- let file_name = self.current_file_name.take().unwrap();
- let index = self.current_index.take();
- let blob_references = self.current_blob_references.take();
- let file_io = self.file_io.clone();
- let location = self.current_file_path.take().unwrap();
- let bucket_dir = location.parent().to_string();
- let threshold = self
- .config
- .file_index_options
- .as_ref()
- .map(|options| options.in_manifest_threshold);
- let stats = std::mem::take(&mut self.current_stats);
- let schema_id = self.config.schema_id;
- // Capture creation_time from when the file was opened, not when the
async close finishes.
- // Java's postpone compaction sorts by creationTime for replay order.
- let creation_time = self.current_file_creation_time;
-
- self.in_flight_closes.spawn(async move {
- let file_size = writer.close().await?.file_size as i64;
- let mut meta = build_meta(file_name, file_size, stats, schema_id,
creation_time);
- ManagedBlobReferences::finish(
- blob_references,
- &file_io,
- &location.path,
- &bucket_dir,
- &mut meta,
- )
- .await?;
- meta.external_path = location.external_path;
- write_index(index, threshold, &file_io, &bucket_dir, &mut
meta).await?;
- Ok(meta)
- });
- }
-
async fn open_new_file(&mut self, user_schema: arrow_schema::SchemaRef) ->
Result<()> {
let index = self
.config
@@ -373,7 +315,6 @@ impl PostponeFileWriter {
self.current_file_name = Some(file_name);
self.current_file_path = Some(location);
self.current_stats = PostponeFileStats::default();
- self.current_file_creation_time = Utc::now();
Ok(())
}
@@ -388,13 +329,7 @@ impl PostponeFileWriter {
let stats = std::mem::take(&mut self.current_stats);
let file_size = writer.close().await?.file_size as i64;
- let mut meta = build_meta(
- file_name,
- file_size,
- stats,
- self.config.schema_id,
- self.current_file_creation_time,
- );
+ let mut meta = build_meta(file_name, file_size, stats,
self.config.schema_id);
let location = self.current_file_path.take().unwrap();
let bucket_dir = location.parent().to_string();
let threshold = self
@@ -412,6 +347,8 @@ impl PostponeFileWriter {
.await?;
meta.external_path = location.external_path;
write_index(index, threshold, &self.file_io, &bucket_dir, &mut
meta).await?;
+ // Java constructs the result after auxiliary files have also finished.
+ meta.creation_time =
Some(crate::common::date_time::current_local_timestamp());
self.written_files.push(meta);
Ok(())
}
@@ -498,7 +435,6 @@ fn build_meta(
file_size: i64,
stats: PostponeFileStats,
schema_id: i64,
- creation_time: DateTime<Utc>,
) -> DataFileMeta {
DataFileMeta {
file_name,
@@ -521,7 +457,7 @@ fn build_meta(
schema_id,
level: 0,
extra_files: vec![],
- creation_time: Some(creation_time),
+ creation_time: None,
delete_row_count: Some(stats.delete_row_count),
embedded_index: None,
file_source: Some(0), // FileSource.APPEND
diff --git a/crates/paimon/src/table/row_kind_generator.rs
b/crates/paimon/src/table/row_kind_generator.rs
index 07e89858..f037dabe 100644
--- a/crates/paimon/src/table/row_kind_generator.rs
+++ b/crates/paimon/src/table/row_kind_generator.rs
@@ -19,7 +19,7 @@
//!
//! Reference: Java `org.apache.paimon.table.sink.RowKindGenerator`.
-use arrow_array::{Array, RecordBatch, StringArray};
+use arrow_array::{Array, LargeStringArray, RecordBatch, StringArray};
use crate::spec::{RowKind, TableSchema};
@@ -51,13 +51,17 @@ impl RowKindGenerator {
source: None,
});
}
- let strings = col.as_any().downcast_ref::<StringArray>().ok_or_else(||
{
- crate::Error::DataInvalid {
+ let value = if let Some(strings) =
col.as_any().downcast_ref::<StringArray>() {
+ strings.value(row)
+ } else if let Some(strings) =
col.as_any().downcast_ref::<LargeStringArray>() {
+ strings.value(row)
+ } else {
+ return Err(crate::Error::DataInvalid {
message: "rowkind.field column must be
Utf8/String".to_string(),
source: None,
- }
- })?;
- RowKind::from_short_string(strings.value(row))
+ });
+ };
+ RowKind::from_short_string(value)
}
}
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index ff20f128..df9ec40a 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -3268,8 +3268,7 @@ impl TableCommit {
let file = entry.file();
let file_creation_time = file
- .creation_time
- .map(|t| t.timestamp_millis())
+ .creation_time_epoch_millis()?
.unwrap_or_else(|| current_time_millis() as i64);
let stats = match stats_map.entry(partition_bytes) {
@@ -3288,7 +3287,9 @@ impl TableCommit {
record_count: 0,
file_size_in_bytes: 0,
file_count: 0,
- last_file_creation_time: 0,
+ // Java starts with the first PartitionEntry, so
+ // pre-epoch creation times must remain negative.
+ last_file_creation_time: file_creation_time,
total_buckets: entry.total_buckets(),
})
}
@@ -4032,6 +4033,73 @@ mod tests {
TableCommit::new(table, "test-user".to_string())
}
+ #[test]
+ fn partition_creation_time_uses_java_local_zone_and_keeps_millis() {
+ const CHILD: &str = "PAIMON_CREATION_TIME_ZONE_CHILD";
+ if let Ok(zone) = std::env::var(CHILD) {
+ let commit = TableCommit::new(
+ test_table(&test_file_io(), "memory:/creation-time"),
+ "user".into(),
+ );
+ let cases: &[(&str, &str)] = match zone.as_str() {
+ "UTC" => &[
+ ("2024-01-02T12:34:56.789Z", "2024-01-02T12:34:56.789Z"),
+ ("1970-01-01T00:00:00.000Z", "1970-01-01T00:00:00.000Z"),
+ ("1969-12-31T23:59:59.999Z", "1969-12-31T23:59:59.999Z"),
+ ],
+ "Asia/Shanghai" => &[
+ ("2024-01-02T12:34:56.789Z", "2024-01-02T04:34:56.789Z"),
+ ("1970-01-01T07:59:59.999Z", "1969-12-31T23:59:59.999Z"),
+ ],
+ "America/New_York" => &[
+ ("2024-01-02T12:34:56.789Z", "2024-01-02T17:34:56.789Z"),
+ ("2024-03-10T02:30:00.123Z", "2024-03-10T07:30:00.123Z"),
+ ("2024-11-03T01:30:00.456Z", "2024-11-03T05:30:00.456Z"),
+ ],
+ _ => unreachable!(),
+ };
+ for (wall, expected) in cases {
+ let mut file = test_data_file("data.parquet", 1);
+ file.creation_time =
Some(wall.parse::<DateTime<Utc>>().unwrap());
+ let mut earlier = file.clone();
+ earlier.file_name = "earlier.parquet".into();
+ earlier.creation_time = file
+ .creation_time
+ .map(|wall| wall - chrono::Duration::milliseconds(1));
+ let mut entries = [
+ ManifestEntry::new(FileKind::Add, vec![], 0, 1, file, 2),
+ ManifestEntry::new(FileKind::Add, vec![], 0, 1, earlier,
2),
+ ];
+ for _ in 0..2 {
+ let stats =
commit.generate_partition_statistics(&entries).unwrap();
+ assert_eq!(stats[0].file_count, 2);
+ assert_eq!(stats[0].record_count, 2);
+ assert_eq!(
+ stats[0].last_file_creation_time,
+ expected
+ .parse::<DateTime<Utc>>()
+ .unwrap()
+ .timestamp_millis(),
+ "{zone} {wall}"
+ );
+ entries.reverse();
+ }
+ }
+ return;
+ }
+ for zone in ["UTC", "Asia/Shanghai", "America/New_York"] {
+ let output =
std::process::Command::new(std::env::current_exe().unwrap())
+ .args(["--exact",
"table::table_commit::tests::partition_creation_time_uses_java_local_zone_and_keeps_millis",
"--nocapture"])
+ .env(CHILD, zone).env("TZ", zone).output().unwrap();
+ assert!(
+ output.status.success(),
+ "{zone}: {}\n{}",
+ String::from_utf8_lossy(&output.stdout),
+ String::from_utf8_lossy(&output.stderr)
+ );
+ }
+ }
+
#[test]
fn partition_statistics_use_java_partition_values() {
use crate::spec::{DateType, DecimalType, Schema, VarBinaryType,
VarCharType};
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 8513c2be..7a5abef7 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -806,15 +806,11 @@ impl TableWrite {
}
pub(super) fn normalize_write_batch(&self, batch: &RecordBatch) ->
Result<Option<RecordBatch>> {
- let batch = self.validate_write_batch_schema(batch)?;
- // Java filters row kinds before extracting Blob values. Ignored rows
- // need neither valid descriptor bytes nor physical files.
+ let filtered = self.filter_write_input(batch)?;
+ let batch =
self.validate_write_batch_schema(filtered.as_ref().unwrap_or(batch))?;
+ // Filtering already precedes value normalization. Generate the system
+ // column now so every surviving PK batch has the same buffer schema.
let batch = self.generate_rowkind_batch(&batch)?;
- let selection = self.rowkind_selection(&batch)?;
- let batch = match &selection {
- Some(rows) => take_rows(&batch, rows)?,
- None => batch,
- };
if batch.num_rows() == 0 {
return Ok(None);
}
@@ -870,6 +866,75 @@ 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.
+ 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);
+ }
+ // Reject malformed schemas even when every event would be ignored.
+ // Empty arrays check equivalent layouts without examining row values.
+
self.validate_write_batch_schema(&RecordBatch::new_empty(batch.schema()))?;
+ let kinds = if let Some(generator) = &self.row_kind_generator {
+ (0..batch.num_rows())
+ .map(|row| generator.generate(batch, row))
+ .collect::<Result<Vec<_>>>()?
+ } else if let Some(column) =
batch.column_by_name(VALUE_KIND_FIELD_NAME) {
+ column
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .unwrap()
+ .iter()
+ .map(|kind|
RowKind::from_value(kind.unwrap_or(RowKind::Insert.to_value())))
+ .collect::<Result<Vec<_>>>()?
+ } else {
+ 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();
+ 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());
+ for (row, kind) in kinds.iter().enumerate() {
+ if batch.column(index).is_null(row) && (*kind !=
RowKind::Delete || delete_required)
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Cannot write null to non-null
column({})", field.name()),
+ source: None,
+ });
+ }
+ }
+ }
+ let rows = kinds
+ .into_iter()
+ .enumerate()
+ .filter_map(|(row, kind)| filter.test(kind).then_some(row))
+ .collect::<Vec<_>>();
+ if rows.len() == batch.num_rows() {
+ Ok(None)
+ } else {
+ take_rows(batch, &rows).map(Some)
+ }
+ }
+
pub(super) async fn write_partition_bucket_batch(
&mut self,
partition: Vec<u8>,
@@ -1197,32 +1262,6 @@ impl TableWrite {
Self::add_per_row_value_kind_column(batch, kinds)
}
- /// Both generated row kinds and explicit caller-provided kinds use the
- /// same Java RowKindFilter before bucket assignment and changelog writing.
- fn rowkind_selection(&self, batch: &RecordBatch) ->
Result<Option<Vec<usize>>> {
- let Some(filter) = &self.row_kind_filter else {
- return Ok(None);
- };
- let Some(column) = batch.column_by_name(VALUE_KIND_FIELD_NAME) else {
- return Ok(None);
- };
- let kinds = column
- .as_any()
- .downcast_ref::<arrow_array::Int8Array>()
- .ok_or_else(|| crate::Error::DataInvalid {
- message: "_VALUE_KIND column must be Int8".to_string(),
- source: None,
- })?;
- let mut keep_rows = Vec::new();
- for (row, kind) in kinds.iter().enumerate() {
- let kind = RowKind::from_value(kind.unwrap_or(RowKind::Insert as
i8))?;
- if filter.test(kind) {
- keep_rows.push(row);
- }
- }
- Ok((keep_rows.len() != batch.num_rows()).then_some(keep_rows))
- }
-
fn add_per_row_value_kind_column(
batch: &RecordBatch,
value_kinds: Vec<i8>,
diff --git a/crates/paimon/tests/creation_time_write_test.rs
b/crates/paimon/tests/creation_time_write_test.rs
new file mode 100644
index 00000000..ce038cea
--- /dev/null
+++ b/crates/paimon/tests/creation_time_write_test.rs
@@ -0,0 +1,405 @@
+// 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::{ArrayRef, Int32Array, LargeBinaryArray, RecordBatch};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use opendal::Operator;
+use paimon::spec::{BlobType, DataType, IntType, Manifest, ManifestList,
Schema, TableSchema};
+use paimon::{
+ catalog::Identifier,
+ io::{FileIOBuilder, FileIOProvider},
+ table::Table,
+};
+use std::sync::{Arc, Mutex};
+
+fn local_wall_millis() -> i64 {
+ // Chrono Local ignores TZ on Windows. Resolve the explicit test zone
+ // independently, so the same wall-time expectations cover every platform.
+ let zone = std::env::var("TZ")
+ .ok()
+ .and_then(|name| jiff::tz::TimeZone::get(&name).ok())
+ .unwrap_or_else(jiff::tz::TimeZone::system);
+ jiff::Timestamp::now()
+ .to_zoned(zone)
+ .datetime()
+ .to_zoned(jiff::tz::TimeZone::UTC)
+ .unwrap()
+ .timestamp()
+ .as_millisecond()
+}
+
+#[derive(Debug)]
+struct IndexOutputProvider {
+ operator: Operator,
+ index_output_times: Mutex<Vec<i64>>,
+}
+
+#[async_trait::async_trait]
+impl FileIOProvider for IndexOutputProvider {
+ async fn create(&self, path: &str) -> paimon::Result<(Operator, String)> {
+ let relative = path.strip_prefix("s3://bucket/").unwrap().to_string();
+ if relative.ends_with(".index") {
+ // Force a distinct millisecond at the actual auxiliary output,
+ // without relying on a storage delay or scheduling a sleep.
+ let before = local_wall_millis();
+ let deadline = std::time::Instant::now() +
std::time::Duration::from_secs(10);
+ let completed = loop {
+ let now = local_wall_millis();
+ if now > before {
+ break now;
+ }
+ assert!(
+ std::time::Instant::now() < deadline,
+ "wall clock did not advance"
+ );
+ tokio::task::yield_now().await;
+ };
+ self.index_output_times.lock().unwrap().push(completed);
+ }
+ Ok((self.operator.clone(), relative))
+ }
+}
+
+async fn check_creation_time_after_index_output(primary_key: bool) {
+ let provider = Arc::new(IndexOutputProvider {
+ operator:
Operator::from_config(opendal::services::MemoryConfig::default()).unwrap(),
+ index_output_times: Mutex::new(Vec::new()),
+ });
+ let io = FileIOBuilder::new("unused")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .option("file-index.bitmap.columns", "id")
+ .option("file-index.in-manifest-threshold", "0b");
+ if primary_key {
+ schema = schema.primary_key(["id"]).option("bucket", "1");
+ }
+ let path = "s3://bucket/creation-time-index";
+ let table = Table::new(
+ io.clone(),
+ Identifier::new("default", "creation_time_index"),
+ path.into(),
+ TableSchema::new(0, &schema.build().unwrap()),
+ None,
+ );
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ let batch =
+ RecordBatch::try_from_iter([("id", Arc::new(Int32Array::from(vec![1,
2])) as ArrayRef)])
+ .unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let file = &messages[0].new_files[0];
+ let index_output_time = {
+ let index_times = provider.index_output_times.lock().unwrap();
+ assert_eq!(index_times.len(), 1);
+ index_times[0]
+ };
+ assert!(
+ file.creation_time.unwrap().timestamp_millis() >= index_output_time,
+ "creation time must be assigned after the auxiliary index output: {:?}
vs {}",
+ file.creation_time,
+ index_output_time,
+ );
+ assert_eq!(file.extra_files, vec![format!("{}.index", file.file_name)]);
+ let index = format!("{path}/bucket-0/{}", file.extra_files[0]);
+ assert!(io.new_input(&index).unwrap().metadata().await.unwrap().size > 0);
+}
+
+#[tokio::test]
+async fn append_creation_time_is_assigned_after_index_output() {
+ check_creation_time_after_index_output(false).await;
+}
+
+#[tokio::test]
+async fn pk_creation_time_is_assigned_after_index_output() {
+ check_creation_time_after_index_output(true).await;
+}
+
+#[test]
+fn writers_persist_local_creation_times_in_all_data_and_changelog_files() {
+ const CHILD: &str = "PAIMON_WRITER_CREATION_TIME_ZONE_CHILD";
+ if std::env::var_os(CHILD).is_some() {
+ tokio::runtime::Runtime::new()
+ .unwrap()
+ .block_on(check_writers());
+ return;
+ }
+ for zone in ["Asia/Shanghai", "America/New_York"] {
+ let output =
std::process::Command::new(std::env::current_exe().unwrap())
+ .args([
+ "--exact",
+
"writers_persist_local_creation_times_in_all_data_and_changelog_files",
+ "--nocapture",
+ ])
+ .env(CHILD, "1")
+ .env("TZ", zone)
+ .output()
+ .unwrap();
+ assert!(
+ output.status.success(),
+ "{zone}: {}\n{}",
+ String::from_utf8_lossy(&output.stdout),
+ String::from_utf8_lossy(&output.stderr)
+ );
+ }
+}
+
+async fn check_writers() {
+ for mode in [
+ "append",
+ "pk",
+ "input-changelog",
+ "postpone",
+ "dedicated-blob",
+ ] {
+ let mut schema = Schema::builder().column("id",
DataType::Int(IntType::new()));
+ let payload_type = if mode == "dedicated-blob" {
+ DataType::Blob(BlobType::new())
+ } else {
+ DataType::Int(IntType::new())
+ };
+ schema = schema
+ .column("payload", payload_type)
+ .option("file.format", "parquet");
+ if matches!(mode, "pk" | "input-changelog" | "postpone") {
+ schema = schema
+ .primary_key(["id"])
+ .option("bucket", if mode == "postpone" { "-2" } else { "1" });
+ }
+ if mode == "input-changelog" {
+ schema = schema
+ .option("changelog-producer", "input")
+ .option("changelog-file.format", "parquet");
+ }
+ if mode == "dedicated-blob" {
+ schema = schema
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .option("blob-field", "payload");
+ }
+ let path = format!("memory:/creation_time_{mode}");
+ 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;
+ let payload: ArrayRef = if mode == "dedicated-blob" {
+ Arc::new(LargeBinaryArray::from(vec![Some(b"blob".as_slice())]))
+ } else {
+ Arc::new(Int32Array::from(vec![10]))
+ };
+ let batch = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1])) as ArrayRef),
+ ("payload", payload),
+ ])
+ .unwrap();
+ let before = local_wall_millis();
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let after = local_wall_millis();
+ let files = messages
+ .iter()
+ .flat_map(|message|
message.new_files.iter().chain(&message.new_changelog_files))
+ .collect::<Vec<_>>();
+ assert!(!files.is_empty(), "{mode}");
+ for file in &files {
+ let wall = file.creation_time.unwrap();
+ assert!(
+ (before..=after).contains(&wall.timestamp_millis()),
+ "{mode}: {} outside local wall range {before}..={after}",
+ wall.timestamp_millis()
+ );
+ assert_eq!(
+ wall.timestamp_subsec_nanos() % 1_000_000,
+ 0,
+ "{mode}: creation time must have TIMESTAMP(3) precision"
+ );
+ assert!(file.creation_time_epoch_millis().unwrap().is_some());
+ }
+ if mode == "input-changelog" {
+ assert!(messages
+ .iter()
+ .any(|message| !message.new_changelog_files.is_empty()));
+ }
+ if mode == "dedicated-blob" {
+ assert!(files.iter().any(|file|
file.file_name.ends_with(".blob")));
+ }
+ let mut expected_times = messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .map(|file| file.creation_time)
+ .collect::<Vec<_>>();
+ expected_times.sort();
+ builder.new_commit().commit(messages).await.unwrap();
+ writer.close().await;
+ let snapshot = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let manifest_dir = format!("{}/manifest", table.location());
+ let manifests = ManifestList::read(
+ &io,
+ &format!("{manifest_dir}/{}", snapshot.delta_manifest_list()),
+ )
+ .await
+ .unwrap();
+ let mut persisted_times = Vec::new();
+ for manifest in manifests {
+ persisted_times.extend(
+ Manifest::read(&io, &format!("{manifest_dir}/{}",
manifest.file_name()))
+ .await
+ .unwrap()
+ .iter()
+ .map(|entry| entry.file().creation_time),
+ );
+ }
+ persisted_times.sort();
+ assert_eq!(
+ persisted_times, expected_times,
+ "{mode}: commit must retain local wall time"
+ );
+ if mode == "postpone" {
+ // Pending files enter normal reads only after a fixed-bucket
rewrite.
+ // Their committed metadata must still have the same Java
timestamp.
+ continue;
+ }
+ let mut read = table.new_read_builder();
+ read.with_projection(&["id"]).unwrap();
+ let plan = read.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
+ 1,
+ "{mode}"
+ );
+ }
+}
+
+#[tokio::test]
+async fn postpone_creation_time_is_assigned_when_file_result_is_finished() {
+ use std::time::{Duration, Instant};
+
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "-2")
+ .build()
+ .unwrap();
+ let path = "memory:/postpone_creation_at_result";
+ let (io, table) = memory_table(path, TableSchema::new(0, &schema));
+ setup_dirs(&io, path).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ let batch = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1])) as ArrayRef),
+ ("payload", Arc::new(Int32Array::from(vec![10])) as ArrayRef),
+ ])
+ .unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let opened = local_wall_millis();
+ // Advance one wall-clock millisecond after opening the file, without a
+ // sleep or any process-global time-zone change.
+ let started = Instant::now();
+ let before_finish = loop {
+ let current = local_wall_millis();
+ if current > opened {
+ break current;
+ }
+ assert!(
+ started.elapsed() < Duration::from_secs(2),
+ "wall clock did not advance"
+ );
+ tokio::task::yield_now().await;
+ };
+ let messages = writer.prepare_commit().await.unwrap();
+ let finished = local_wall_millis();
+ let file = &messages[0].new_files[0];
+ assert!(
+
(before_finish..=finished).contains(&file.creation_time.unwrap().timestamp_millis()),
+ "Postpone metadata must use result time, not the earlier file-open
time"
+ );
+ writer.close().await;
+}
+
+#[tokio::test]
+async fn postpone_rolling_results_keep_arrival_order_and_creation_times() {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "-2")
+ .option("target-file-row-num", "1")
+ .build()
+ .unwrap();
+ let path = "memory:/postpone_creation_roll_order";
+ let (io, table) = memory_table(path, TableSchema::new(0, &schema));
+ setup_dirs(&io, path).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ let mut windows = Vec::new();
+ for value in [30, 10, 20] {
+ let batch = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1])) as ArrayRef),
+ (
+ "payload",
+ Arc::new(Int32Array::from(vec![value])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let before = local_wall_millis();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let after = local_wall_millis();
+ windows.push(before..=after);
+ }
+ let messages = writer.prepare_commit().await.unwrap();
+ let files = &messages[0].new_files;
+ assert_eq!(files.len(), 3);
+ let mut previous_time = i64::MIN;
+ for (index, (file, window)) in files.iter().zip(windows).enumerate() {
+ assert_eq!(file.min_sequence_number, index as i64);
+ assert_eq!(file.max_sequence_number, index as i64);
+ let time = file.creation_time.unwrap().timestamp_millis();
+ assert!(
+ window.contains(&time),
+ "Rolled file must get its timestamp before the next input batch"
+ );
+ assert!(
+ time >= previous_time,
+ "Java creation-time replay must preserve this writer's arrival
order"
+ );
+ previous_time = time;
+ }
+ writer.close().await;
+}
diff --git a/crates/paimon/tests/filtered_write_validation_test.rs
b/crates/paimon/tests/filtered_write_validation_test.rs
new file mode 100644
index 00000000..0a78cd0b
--- /dev/null
+++ b/crates/paimon/tests/filtered_write_validation_test.rs
@@ -0,0 +1,337 @@
+// 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::{
+ make_array, types::Int32Type, Array, ArrayRef, Int32Array, Int8Array,
LargeStringArray,
+ ListArray, MapArray, RecordBatch, StringArray, StructArray,
+};
+use arrow_schema::{DataType as ArrowType, Field, Schema as ArrowSchema};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::spec::{
+ ArrayType, DataField, DataType, IntType, MapType, RowType, Schema,
TableSchema, VarCharType,
+};
+use std::sync::Arc;
+
+fn payload(mode: &str, valid: bool) -> (DataType, ArrayRef) {
+ let values = if valid {
+ vec![Some(10), Some(42)]
+ } else {
+ vec![None, Some(42)]
+ };
+ let (logical, actual): (_, ArrayRef) = if mode == "array" {
+ (
+
DataType::Array(ArrayType::new(DataType::Int(IntType::with_nullable(false)))),
+ Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(
+ values.into_iter().map(|value| Some(vec![value])),
+ )),
+ )
+ } else if mode == "row" {
+ (
+ DataType::Row(RowType::new(vec![DataField::new(
+ 3,
+ "value".into(),
+ DataType::Int(IntType::with_nullable(false)),
+ )])),
+ Arc::new(StructArray::from(vec![(
+ Arc::new(Field::new("value", ArrowType::Int32, true)),
+ Arc::new(Int32Array::from(values)) as ArrayRef,
+ )])),
+ )
+ } else {
+ let entries = StructArray::from(vec![
+ (
+ Arc::new(Field::new("key", ArrowType::Utf8, false)),
+ Arc::new(StringArray::from(vec!["a", "b"])) as ArrayRef,
+ ),
+ (
+ Arc::new(Field::new("value", ArrowType::Int32, true)),
+ Arc::new(Int32Array::from(values)) as ArrayRef,
+ ),
+ ]);
+ (
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Int(IntType::with_nullable(false)),
+ )),
+ Arc::new(
+ MapArray::try_new(
+ Arc::new(Field::new("entries",
entries.data_type().clone(), false)),
+ arrow_buffer::OffsetBuffer::new(vec![0, 1, 2].into()),
+ entries,
+ None,
+ false,
+ )
+ .unwrap(),
+ ),
+ )
+ };
+ // The Arrow C Data interface can carry declared NOT NULL children whose
+ // values do not satisfy that declaration, as PyArrow permits constructing.
+ let expected = paimon::arrow::paimon_type_to_arrow(&logical).unwrap();
+ let mut data = actual.to_data();
+ if let ArrowType::Map(entries, _) = &expected {
+ let entry_data = unsafe {
+ data.child_data()[0]
+ .clone()
+ .into_builder()
+ .data_type(entries.data_type().clone())
+ .build_unchecked()
+ };
+ data = unsafe {
+ data.into_builder()
+ .child_data(vec![entry_data])
+ .build_unchecked()
+ };
+ }
+ let data = unsafe {
data.into_builder().data_type(expected).build_unchecked() };
+ (logical, make_array(data))
+}
+
+async fn check_ignored_children(
+ mode: &str,
+ generated: bool,
+ large_string: bool,
+ bucket: Option<i32>,
+) {
+ let (payload_type, payload) = payload(mode, false);
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .column("payload", payload_type)
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("ignore-delete", "true");
+ if generated {
+ schema = schema.option("rowkind.field", "op");
+ } else {
+ schema = schema.option("changelog-producer", "input");
+ }
+ let path = "memory:/filtered-children";
+ 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;
+ let op: ArrayRef = if large_string {
+ Arc::new(LargeStringArray::from(vec!["-D", "+I"]))
+ } else {
+ Arc::new(StringArray::from(vec!["-D", "+I"]))
+ };
+ let mut batch = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef),
+ ("op", op),
+ ("payload", payload),
+ ])
+ .unwrap();
+ if !generated {
+ let mut fields = batch.schema().fields().to_vec();
+ fields.push(Arc::new(Field::new("_VALUE_KIND", ArrowType::Int8,
false)));
+ let mut columns = batch.columns().to_vec();
+ columns.push(Arc::new(Int8Array::from(vec![3, 0])));
+ batch = RecordBatch::try_new(Arc::new(ArrowSchema::new(fields)),
columns).unwrap();
+ }
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow(&[batch], bucket).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(
+ messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .map(|file| file.row_count)
+ .sum::<i64>(),
+ 1
+ );
+ let files = messages
+ .iter()
+ .flat_map(|message|
message.new_files.iter().chain(&message.new_changelog_files))
+ .collect::<Vec<_>>();
+ assert_eq!(files.len(), if generated { 1 } else { 2 });
+ assert!(files.iter().all(|file| file.row_count == 1
+ && file.min_sequence_number == 0
+ && file.max_sequence_number == 0));
+ builder.new_commit().commit(messages).await.unwrap();
+ let read_builder = table.new_read_builder();
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ let batches: Vec<_> = read_builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ let ids = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(ids.value(0), 2);
+}
+
+#[tokio::test]
+async fn ignored_nested_nulls_do_not_fail_writes() {
+ for mode in ["array", "row", "map"] {
+ for generated in [false, true] {
+ for large_string in [false, true] {
+ for bucket in [None, Some(0)] {
+ check_ignored_children(mode, generated, large_string,
bucket).await;
+ }
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn surviving_nested_nulls_still_fail_writes() {
+ for mode in ["array", "row", "map"] {
+ let (logical, payload) = payload(mode, false);
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .column("payload", logical)
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("rowkind.field", "op")
+ .option("ignore-delete", "true")
+ .build()
+ .unwrap();
+ let (_, table) = memory_table("memory:/retained-children",
TableSchema::new(0, &schema));
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ let batch = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef),
+ (
+ "op",
+ Arc::new(StringArray::from(vec!["+I", "+I"])) as ArrayRef,
+ ),
+ ("payload", payload),
+ ])
+ .unwrap();
+ assert!(writer.write_arrow_batch(&batch).await.is_err(), "{mode}");
+ assert!(writer.prepare_commit().await.unwrap().is_empty());
+ }
+}
+
+#[tokio::test]
+async fn ignored_rows_keep_java_top_level_nullability_and_schema_checks() {
+ for engine in ["deduplicate", "partial-update", "aggregation"] {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .column("seq", DataType::Int(IntType::with_nullable(false)))
+ .column("payload", DataType::Int(IntType::with_nullable(false)))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("merge-engine", engine)
+ .option("sequence.field", "seq")
+ .option("rowkind.field", "op")
+ .option("ignore-delete", "true")
+ .build()
+ .unwrap();
+ let (_, table) = memory_table("memory:/filtered-nullability",
TableSchema::new(0, &schema));
+ for (kind, id, seq, payload, valid) in [
+ ("-D", Some(1), Some(1), None, engine == "deduplicate"),
+ ("-U", Some(1), Some(1), None, false),
+ ("-D", None, Some(1), Some(42), false),
+ ("-D", Some(1), None, Some(42), false),
+ ] {
+ let batch = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![id])) as ArrayRef),
+ ("op", Arc::new(StringArray::from(vec![kind])) as ArrayRef),
+ ("seq", Arc::new(Int32Array::from(vec![seq])) as ArrayRef),
+ (
+ "payload",
+ Arc::new(Int32Array::from(vec![payload])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ let result = writer.write_arrow_batch(&batch).await;
+ assert_eq!(
+ result.is_ok(),
+ valid,
+ "{engine} {kind} {id:?} {seq:?}: {result:?}"
+ );
+ assert!(writer.prepare_commit().await.unwrap().is_empty());
+ }
+ let wrong = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1])) as ArrayRef),
+ ("op", Arc::new(StringArray::from(vec!["-D"])) as ArrayRef),
+ ("seq", Arc::new(Int32Array::from(vec![1])) as ArrayRef),
+ (
+ "payload",
+ Arc::new(StringArray::from(vec!["wrong type"])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ assert!(writer.write_arrow_batch(&wrong).await.is_err());
+ assert!(writer.prepare_commit().await.unwrap().is_empty());
+ }
+}
+
+#[tokio::test]
+async fn empty_and_filtered_rowkind_inputs_preserve_transport_validation() {
+ for generated in [false, true] {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("ignore-delete", "true");
+ if generated {
+ schema = schema.option("rowkind.field", "op");
+ }
+ let (_, table) = memory_table(
+ "memory:/empty-filtered",
+ TableSchema::new(0, &schema.build().unwrap()),
+ );
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ let input = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1])) as ArrayRef),
+ (
+ "op",
+ Arc::new(LargeStringArray::from(vec!["-D"])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ if generated {
+ writer
+ .write_arrow_batch(&RecordBatch::new_empty(input.schema()))
+ .await
+ .unwrap();
+ writer.write_arrow_batch(&input).await.unwrap();
+ }
+ for misplaced in [false, true] {
+ let kind: ArrayRef = if misplaced {
+ Arc::new(Int8Array::from(vec![3]))
+ } else {
+ Arc::new(Int32Array::from(vec![3]))
+ };
+ let mut columns = vec![
+ ("id", input.column(0).clone()),
+ ("op", input.column(1).clone()),
+ ];
+ columns.insert(if misplaced { 0 } else { 2 }, ("_VALUE_KIND",
kind));
+ let wrong = RecordBatch::try_from_iter(columns).unwrap();
+ assert!(writer.write_arrow_batch(&wrong).await.is_err());
+ }
+ assert!(writer.prepare_commit().await.unwrap().is_empty());
+ }
+}
