diff --git a/fluss-rust/crates/fluss/src/client/table/append.rs b/fluss-rust/crates/fluss/src/client/table/append.rs index b00c6052108..7f3b2993ad8 100644 --- a/fluss-rust/crates/fluss/src/client/table/append.rs +++ b/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()), diff --git a/fluss-rust/crates/fluss/src/record/arrow.rs b/fluss-rust/crates/fluss/src/record/arrow.rs index 4ca2cc5de03..8cd6c589b7c 100644 --- a/fluss-rust/crates/fluss/src/record/arrow.rs +++ b/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>, 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> { 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 { let arrow_batch_builder: Box = { 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 { let fields: Result> = 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", diff --git a/fluss-rust/crates/fluss/tests/integration/log_table.rs b/fluss-rust/crates/fluss/tests/integration/log_table.rs index 69f4ec18be7..0cca2e3e178 100644 --- a/fluss-rust/crates/fluss/tests/integration/log_table.rs +++ b/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");