From 3c29fa15a8e9e3d24eec147b4b4e50ff33f9bf18 Mon Sep 17 00:00:00 2001 From: naivedogger Date: Tue, 8 Sep 2026 11:50:00 +0800 Subject: [PATCH] [c++] Avoid redundant copies during row type resolution Borrow unchanged STRING and BYTES values from the input row while preserving type conversions and validation. Add regression tests for borrowed storage, mixed conversions, nested rows, and invalid values. Fixes apache/fluss#4239 --- fluss-rust/bindings/cpp/src/types.rs | 168 +++++++++++++++++++++++++-- 1 file changed, 159 insertions(+), 9 deletions(-) diff --git a/fluss-rust/bindings/cpp/src/types.rs b/fluss-rust/bindings/cpp/src/types.rs index 84b88b98406..f3d31657193 100644 --- a/fluss-rust/bindings/cpp/src/types.rs +++ b/fluss-rust/bindings/cpp/src/types.rs @@ -457,11 +457,12 @@ pub fn core_database_info_to_ffi(info: &fcore::metadata::DatabaseInfo) -> ffi::F /// Resolve types in a GenericRow using schema metadata. /// Narrows Int32 → Int8/Int16, parses decimal strings, etc. -/// Used by both AppendWriter and UpsertWriter. -pub fn resolve_row_types( - row: &fcore::row::GenericRow<'_>, +/// Unchanged STRING and BYTES values borrow the input row's storage. +/// Used by append, upsert, delete, lookup, and prefix lookup. +pub fn resolve_row_types<'a>( + row: &'a fcore::row::GenericRow<'_>, schema: Option<&fcore::metadata::Schema>, -) -> Result> { +) -> Result> { let mut out = fcore::row::GenericRow::new(row.values.len()); for (idx, datum) in row.values.iter().enumerate() { @@ -477,11 +478,11 @@ pub fn resolve_row_types( /// Resolve a single datum against its (optional) target column type, recursing /// into nested ROW values. Narrows Int32 → Int8/Int16, parses decimal strings, /// and leaves already-typed ARRAY/MAP binaries (built by the writers) untouched. -fn resolve_datum( - datum: &fcore::row::Datum<'_>, +fn resolve_datum<'a>( + datum: &'a fcore::row::Datum<'_>, target: Option<&fcore::metadata::DataType>, idx: usize, -) -> Result> { +) -> Result> { Ok(match datum { Datum::Null => Datum::Null, Datum::Bool(v) => Datum::Bool(*v), @@ -509,9 +510,9 @@ fn resolve_datum( .map_err(|e| anyhow!("Column {idx}: {e}"))?; Datum::Decimal(decimal) } - _ => Datum::String(Cow::Owned(cow.to_string())), + _ => Datum::String(Cow::Borrowed(cow.as_ref())), }, - Datum::Blob(cow) => Datum::Blob(Cow::Owned(cow.to_vec())), + Datum::Blob(cow) => Datum::Blob(Cow::Borrowed(cow.as_ref())), Datum::Decimal(d) => Datum::Decimal(d.clone()), Datum::Date(d) => Datum::Date(*d), Datum::Time(t) => Datum::Time(*t), @@ -664,3 +665,152 @@ pub fn core_scan_batches_to_ffi( batches: ffi_batches, }) } + +#[cfg(test)] +mod tests { + use super::resolve_row_types; + use fluss::metadata::{DataField, DataType, DataTypes, DecimalType, RowType, Schema}; + use fluss::row::{Datum, Decimal, GenericRow}; + use std::borrow::Cow; + + fn assert_borrowed_values(input: &[Datum<'_>], resolved: &[Datum<'_>]) { + assert_eq!(resolved, input); + for (input, resolved) in input.iter().zip(resolved) { + match (input, resolved) { + (Datum::String(input), Datum::String(Cow::Borrowed(resolved))) => { + assert_eq!(input.as_ptr(), resolved.as_ptr()); + } + (Datum::Blob(input), Datum::Blob(Cow::Borrowed(resolved))) => { + assert_eq!(input.as_ptr(), resolved.as_ptr()); + } + _ => panic!("expected borrowed STRING or BYTES, got {resolved:?}"), + } + } + } + + #[test] + fn test_resolve_row_types_borrows_strings_and_bytes() { + // Include both setter-owned values and values already borrowing external storage. + let string = String::from("borrowed string"); + let bytes = vec![0, 1, 255]; + let row = GenericRow { + values: vec![ + Datum::String(Cow::Owned(String::from("owned string"))), + Datum::Blob(Cow::Owned(vec![2, 3, 255])), + Datum::String(Cow::Borrowed(&string)), + Datum::Blob(Cow::Borrowed(&bytes)), + Datum::String(Cow::Owned(String::new())), + Datum::Blob(Cow::Owned(Vec::new())), + ], + }; + let schema = Schema::builder() + .column("owned_string", DataTypes::string()) + .column("owned_bytes", DataTypes::bytes()) + .column("borrowed_string", DataTypes::string()) + .column("borrowed_bytes", DataTypes::bytes()) + .column("empty_string", DataTypes::string()) + .column("empty_bytes", DataTypes::bytes()) + .build() + .unwrap(); + + for schema in [Some(&schema), None] { + let resolved = resolve_row_types(&row, schema).unwrap(); + assert_borrowed_values(&row.values, &resolved.values); + } + } + + #[test] + fn test_resolve_row_types_preserves_conversions() { + let schema = Schema::builder() + .column("tiny", DataTypes::tinyint()) + .column("small", DataTypes::smallint()) + .column( + "decimal", + DataType::Decimal(DecimalType::new(5, 2).unwrap()), + ) + .column("string", DataTypes::string()) + .column("bytes", DataTypes::bytes()) + .column("nullable", DataTypes::string()) + .build() + .unwrap(); + let row = GenericRow { + values: vec![ + Datum::Int32(127), + Datum::Int32(-32768), + Datum::String(Cow::Owned(String::from("123.45"))), + Datum::String(Cow::Owned(String::from("unchanged"))), + Datum::Blob(Cow::Owned(vec![1, 2, 3])), + Datum::Null, + ], + }; + let resolved = resolve_row_types(&row, Some(&schema)).unwrap(); + assert_eq!(resolved.values[0], Datum::Int8(127)); + assert_eq!(resolved.values[1], Datum::Int16(-32768)); + assert_eq!( + resolved.values[2], + Datum::Decimal(Decimal::from_unscaled_long(12345, 5, 2).unwrap()) + ); + assert_borrowed_values(&row.values[3..5], &resolved.values[3..5]); + assert_eq!(resolved.values[5], Datum::Null); + } + + #[test] + fn test_resolve_row_types_borrows_nested_values() { + let schema = Schema::builder() + .column( + "nested", + DataType::Row(RowType::new(vec![ + DataField::new("string", DataTypes::string(), None), + DataField::new("bytes", DataTypes::bytes(), None), + ])), + ) + .build() + .unwrap(); + let row = GenericRow { + values: vec![Datum::Row(Box::new(GenericRow { + values: vec![ + Datum::String(Cow::Owned(String::from("nested string"))), + Datum::Blob(Cow::Owned(vec![1, 2, 3])), + ], + }))], + }; + let resolved = resolve_row_types(&row, Some(&schema)).unwrap(); + let (Datum::Row(input), Datum::Row(nested)) = (&row.values[0], &resolved.values[0]) else { + panic!("expected nested rows"); + }; + assert_borrowed_values(&input.values, &nested.values); + } + + #[test] + fn test_resolve_row_types_preserves_validation() { + let cases = [ + (DataTypes::tinyint(), Datum::Int32(128), "overflows TinyInt"), + ( + DataTypes::smallint(), + Datum::Int32(32768), + "overflows SmallInt", + ), + ( + DataType::Decimal(DecimalType::new(5, 2).unwrap()), + Datum::String(Cow::Borrowed("invalid")), + "invalid decimal string", + ), + ( + DataType::Decimal(DecimalType::new(5, 2).unwrap()), + Datum::String(Cow::Borrowed("1234.56")), + "Decimal precision overflow", + ), + ]; + for (data_type, datum, expected_error) in cases { + let schema = Schema::builder() + .column("value", data_type) + .build() + .unwrap(); + let row = GenericRow { + values: vec![datum], + }; + let error = resolve_row_types(&row, Some(&schema)).unwrap_err(); + assert!(error.to_string().contains(expected_error), "{error}"); + } + } +}