[rust] Validate append_arrow_batch nullability before enqueue

fluss-rust/crates/fluss/src/client/table/append.rs

@@ -21,6 +21,7 @@ use crate::client::{WriteRecord, WriteResultFuture, WriterClient};
 use crate::error::Error::{IllegalArgument, UnexpectedError};
 use crate::error::Result;
 use crate::metadata::{PhysicalTablePath, TableInfo, TablePath};
+use crate::record::validate_append_record_batch;
 use crate::row::encode::{KeyEncoder, KeyEncoderFactory};
 use crate::row::{ColumnarRow, InternalRow};
 use arrow::array::{RecordBatch, UInt32Array};
@@ -174,6 +175,7 @@ impl AppendWriter {
             // Nothing to write; also avoids a keyless send to a bucket-key table.
             return Ok(WriteResultFuture::join(Vec::new()));
         }
+        validate_append_record_batch(&batch, self.table_info.row_type())?;
         let physical_table_path = if self.partition_getter.is_some() {
             let first_row = ColumnarRow::new(
                 Arc::new(batch.clone()),

fluss-rust/crates/fluss/src/record/arrow.rs

@@ -25,7 +25,7 @@ use crate::record::{ChangeType, ScanRecord};
 use crate::row::column_vector::TypedBatch;
 use crate::row::column_writer::{ColumnWriter, round_up_to_8};
 use crate::row::{ColumnarRow, InternalRow};
-use arrow::array::{ArrayBuilder, ArrayRef, new_null_array};
+use arrow::array::{Array, ArrayBuilder, ArrayRef, new_null_array};
 use arrow::{
     array::RecordBatch,
     buffer::Buffer,
@@ -207,12 +207,22 @@ pub trait ArrowRecordBatchInnerBuilder: Send {
     fn estimated_size_in_bytes(&self) -> usize;
 }
 
-#[derive(Default)]
-pub struct PrebuiltRecordBatchBuilder {
+pub(crate) struct PrebuiltRecordBatchBuilder {
+    row_type: RowType,
     arrow_record_batch: Option<Arc<RecordBatch>>,
     records_count: i32,
 }
 
+impl PrebuiltRecordBatchBuilder {
+    fn new(row_type: RowType) -> Self {
+        Self {
+            row_type,
+            arrow_record_batch: None,
+            records_count: 0,
+        }
+    }
+}
+
 impl ArrowRecordBatchInnerBuilder for PrebuiltRecordBatchBuilder {
     fn build_arrow_record_batch(&mut self) -> Result<Arc<RecordBatch>> {
         Ok(self.arrow_record_batch.as_ref().unwrap().clone())
@@ -227,6 +237,7 @@ impl ArrowRecordBatchInnerBuilder for PrebuiltRecordBatchBuilder {
         if self.arrow_record_batch.is_some() {
             return Ok(false);
         }
+        validate_append_record_batch(record_batch.as_ref(), &self.row_type)?;
         self.records_count = record_batch.num_rows() as i32;
         self.arrow_record_batch = Some(record_batch);
         Ok(true)
@@ -373,7 +384,7 @@ impl MemoryLogRecordsArrowBuilder {
     ) -> Result<Self> {
         let arrow_batch_builder: Box<dyn ArrowRecordBatchInnerBuilder> = {
             if to_append_record_batch {
-                Box::new(PrebuiltRecordBatchBuilder::default())
+                Box::new(PrebuiltRecordBatchBuilder::new(row_type.clone()))
             } else {
                 Box::new(RowAppendRecordBatchBuilder::new(row_type)?)
             }
@@ -1179,6 +1190,28 @@ fn parse_ipc_message(
     Ok((batch_metadata, body_buffer, message.version()))
 }
 
+/// Validates a caller-supplied Arrow [`RecordBatch`] against the table's
+/// [`RowType`] before the prebuilt append path serializes it.
+pub(crate) fn validate_append_record_batch(batch: &RecordBatch, row_type: &RowType) -> Result<()> {
+    TypedBatch::build(batch, row_type)?;
+
+    for (i, field) in row_type.fields().iter().enumerate() {
+        if field.data_type().is_nullable() {
+            continue;
+        }
+        let column = batch.column(i);
+        if column.null_count() > 0 {
+            return Err(IllegalArgument {
+                message: format!(
+                    "Column '{}' is declared as non-nullable but contains null values",
+                    field.name()
+                ),
+            });
+        }
+    }
+    Ok(())
+}
+
 pub fn to_arrow_schema(fluss_schema: &RowType) -> Result<SchemaRef> {
     let fields: Result<Vec<Field>> = fluss_schema
         .fields()
@@ -1979,6 +2012,102 @@ mod tests {
         }
     }
 
+    #[test]
+    fn validate_append_record_batch_rejects_nulls_in_not_null_column() {
+        let row_type = RowType::new(vec![DataField::new(
+            "id",
+            DataTypes::bigint().as_non_nullable(),
+            None,
+        )]);
+
+        // Caller marks the Arrow field nullable (common from pyarrow) and includes a null.
+        let batch_schema = Arc::new(arrow_schema::Schema::new(vec![Field::new(
+            "id",
+            arrow_schema::DataType::Int64,
+            true,
+        )]));
+        let poison = RecordBatch::try_new(
+            batch_schema.clone(),
+            vec![Arc::new(arrow::array::Int64Array::from(vec![
+                Some(1_i64),
+                None,
+                Some(3_i64),
+            ])) as arrow::array::ArrayRef],
+        )
+        .expect("nullable arrow batch with nulls");
+
+        let err = validate_append_record_batch(&poison, &row_type)
+            .expect_err("null in NOT NULL column must be rejected");
+        assert!(
+            err.to_string()
+                .contains("declared as non-nullable but contains null values"),
+            "unexpected error: {err}"
+        );
+
+        // Same nullable Arrow metadata but no null values is accepted.
+        let ok_batch = RecordBatch::try_new(
+            batch_schema,
+            vec![Arc::new(arrow::array::Int64Array::from(vec![1_i64, 2_i64]))
+                as arrow::array::ArrayRef],
+        )
+        .expect("nullable arrow batch without nulls");
+        validate_append_record_batch(&ok_batch, &row_type)
+            .expect("nullable metadata alone must not reject a null-free batch");
+    }
+
+    #[test]
+    fn prebuilt_builder_rejects_nulls_in_not_null_column() {
+        let row_type = RowType::new(vec![DataField::new(
+            "id",
+            DataTypes::bigint().as_non_nullable(),
+            None,
+        )]);
+        let table_path = TablePath::new("db".to_string(), "tbl".to_string());
+        let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1));
+        let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path)));
+
+        let mut builder = MemoryLogRecordsArrowBuilder::new(
+            1,
+            &row_type,
+            true,
+            ArrowCompressionInfo {
+                compression_type: ArrowCompressionType::None,
+                compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL,
+            },
+            usize::MAX,
+            Arc::new(ArrowCompressionRatioEstimator::default()),
+        )
+        .expect("NOT NULL prebuilt builder should construct");
+
+        let batch_schema = Arc::new(arrow_schema::Schema::new(vec![Field::new(
+            "id",
+            arrow_schema::DataType::Int64,
+            true,
+        )]));
+        let poison = RecordBatch::try_new(
+            batch_schema,
+            vec![
+                Arc::new(arrow::array::Int64Array::from(vec![Some(1_i64), None]))
+                    as arrow::array::ArrayRef,
+            ],
+        )
+        .expect("poison batch");
+        let record = WriteRecord::for_append_record_batch(
+            Arc::clone(&table_info),
+            physical_table_path,
+            1,
+            poison,
+        );
+        let err = builder
+            .append(&record)
+            .expect_err("prebuilt append must reject null in NOT NULL");
+        assert!(
+            err.to_string()
+                .contains("declared as non-nullable but contains null values"),
+            "unexpected error: {err}"
+        );
+    }
+
     fn single_int_read_context() -> (ReadContext, SchemaRef) {
         let row_type = Arc::new(RowType::new(vec![DataField::new(
             "id",

fluss-rust/crates/fluss/tests/integration/log_table.rs

@@ -77,6 +77,47 @@ mod table_test {
             .create_writer()
             .expect("Failed to create writer");
 
+        // A null in the NOT NULL column c3 must be rejected client-side, before the
+        // bucket split enqueues part of the batch. The exact record count asserted
+        // below proves the rejected batch never partially reached any bucket.
+        {
+            use arrow::array::{Int32Array, Int64Array, StringArray};
+            use arrow::datatypes::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
+            use std::sync::Arc;
+
+            // Arrow metadata marks c3 nullable (as pyarrow often does) and carries a
+            // null; distinct c1 values hash across multiple buckets.
+            let poison_schema = Arc::new(ArrowSchema::new(vec![
+                Field::new("c1", ArrowDataType::Int32, true),
+                Field::new("c2", ArrowDataType::Utf8, true),
+                Field::new("c3", ArrowDataType::Int64, true),
+            ]));
+            let poison = arrow::array::RecordBatch::try_new(
+                poison_schema,
+                vec![
+                    Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6])),
+                    Arc::new(StringArray::from(vec!["a", "b", "c", "d", "e", "f"])),
+                    Arc::new(Int64Array::from(vec![
+                        Some(10),
+                        Some(20),
+                        None,
+                        Some(40),
+                        Some(50),
+                        Some(60),
+                    ])),
+                ],
+            )
+            .expect("poison batch");
+            let err = append_writer
+                .append_arrow_batch(poison)
+                .expect_err("null in NOT NULL column must be rejected before the bucket split");
+            assert!(
+                err.to_string()
+                    .contains("declared as non-nullable but contains null values"),
+                "unexpected error: {err}"
+            );
+        }
+
         let batch1 = record_batch!(
             ("c1", Int32, [1, 2, 3]),
             ("c2", Utf8, ["a1", "a2", "a3"]),
@@ -87,15 +128,25 @@ mod table_test {
             .append_arrow_batch(batch1)
             .expect("Failed to append batch with mixed nullability");
 
-        let batch2 = record_batch!(
-            ("c1", Int32, [4, 5, 6]),
-            ("c2", Utf8, ["a4", "a5", "a6"]),
-            ("c3", Int64, [40, 50, 60])
+        // Arrow schema metadata marks all fields nullable (pyarrow-style), but
+        // null-free values for NOT NULL c3 should still be accepted.
+        let batch2_schema = std::sync::Arc::new(arrow::datatypes::Schema::new(vec![
+            arrow::datatypes::Field::new("c1", arrow::datatypes::DataType::Int32, true),
+            arrow::datatypes::Field::new("c2", arrow::datatypes::DataType::Utf8, true),
+            arrow::datatypes::Field::new("c3", arrow::datatypes::DataType::Int64, true),
+        ]));
+        let batch2 = arrow::array::RecordBatch::try_new(
+            batch2_schema,
+            vec![
+                std::sync::Arc::new(arrow::array::Int32Array::from(vec![4, 5, 6])),
+                std::sync::Arc::new(arrow::array::StringArray::from(vec!["a4", "a5", "a6"])),
+                std::sync::Arc::new(arrow::array::Int64Array::from(vec![40_i64, 50_i64, 60_i64])),
+            ],
         )
-        .unwrap();
+        .expect("nullable-metadata batch without nulls");
         append_writer
             .append_arrow_batch(batch2)
-            .expect("Failed to append batch with mixed nullability");
+            .expect("Failed to append nullable-metadata batch without null values");
 
         // Flush to ensure all writes are acknowledged
         append_writer.flush().await.expect("Failed to flush");