[core] Align write validation and file metadata with Java

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(

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).

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)),

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)`: