You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/05/13 11:53:12 UTC
(fluss-rust) 01/02: feat: add end-to-end ROW (nested struct) column serialization support
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss-rust.git
commit d4a0bb6c784568183f6ddac10a7ba493fb37cfb4
Author: hemanthsavasere <he...@gmail.com>
AuthorDate: Sun Mar 15 18:27:30 2026 +0000
feat: add end-to-end ROW (nested struct) column serialization support
- Add `Datum::Row(Box<GenericRow>)` variant with `as_row()` accessor
- Add `get_row()` to `InternalRow` trait with default error impl
- Implement `GenericRow::get_row()` and `CompactedRow::get_row()` delegation
- Implement `ColumnarRow::get_row()` with Arrow StructArray extraction + OnceLock caching
- Add `InnerValueWriter::Row(RowType)` and write path via nested CompactedRowWriter
- Add `DataType::Row` arm in `CompactedRowDeserializer` for eager nested decode
- Add `InnerFieldGetter::Row` and hook up FieldGetter/ValueWriter pipeline
- Handle `Datum::Row` in `resolve_row_types` (C++ bindings)
- Add round-trip tests: simple nesting, deep nesting, nullable fields, ROW as primary key
Wire format matches Java: varint-length-prefixed blob of a complete CompactedRow.
---
bindings/cpp/src/types.rs | 1 +
crates/fluss/src/row/binary/binary_writer.rs | 25 +-
crates/fluss/src/row/column.rs | 300 ++++++++++++++++++++-
crates/fluss/src/row/compacted/compacted_row.rs | 4 +
.../src/row/compacted/compacted_row_reader.rs | 151 +++++++++++
crates/fluss/src/row/datum.rs | 15 ++
.../fluss/src/row/encode/compacted_key_encoder.rs | 48 +++-
crates/fluss/src/row/field_getter.rs | 12 +-
crates/fluss/src/row/mod.rs | 19 +-
9 files changed, 563 insertions(+), 12 deletions(-)
diff --git a/bindings/cpp/src/types.rs b/bindings/cpp/src/types.rs
index f61711b5..19aa9a36 100644
--- a/bindings/cpp/src/types.rs
+++ b/bindings/cpp/src/types.rs
@@ -529,6 +529,7 @@ pub fn resolve_row_types(
Datum::TimestampNtz(ts) => Datum::TimestampNtz(*ts),
Datum::TimestampLtz(ts) => Datum::TimestampLtz(*ts),
Datum::Array(a) => Datum::Array(a.clone()),
+ Datum::Row(r) => Datum::Row(Box::new(resolve_row_types(r, None)?)),
};
out.set_field(idx, resolved);
}
diff --git a/crates/fluss/src/row/binary/binary_writer.rs b/crates/fluss/src/row/binary/binary_writer.rs
index f51a6e80..fd7da4c9 100644
--- a/crates/fluss/src/row/binary/binary_writer.rs
+++ b/crates/fluss/src/row/binary/binary_writer.rs
@@ -17,7 +17,7 @@
use crate::error::Error::IllegalArgument;
use crate::error::Result;
-use crate::metadata::DataType;
+use crate::metadata::{DataType, RowType};
use crate::row::Datum;
use crate::row::binary::BinaryRowFormat;
@@ -136,7 +136,7 @@ pub enum InnerValueWriter {
TimestampNtz(u32), // precision
TimestampLtz(u32), // precision
Array,
- // TODO Row
+ Row(RowType),
}
/// Accessor for writing the fields/elements of a binary writer during runtime, the
@@ -176,6 +176,7 @@ impl InnerValueWriter {
Ok(InnerValueWriter::TimestampLtz(t.precision()))
}
DataType::Array(_) => Ok(InnerValueWriter::Array),
+ DataType::Row(row_type) => Ok(InnerValueWriter::Row(row_type.clone())),
_ => unimplemented!(
"ValueWriter for DataType {:?} is currently not implemented",
data_type
@@ -241,6 +242,26 @@ impl InnerValueWriter {
(InnerValueWriter::Array, Datum::Array(arr)) => {
writer.write_array(arr.as_bytes());
}
+ (InnerValueWriter::Row(row_type), Datum::Row(inner_row)) => {
+ use crate::row::compacted::CompactedRowWriter;
+ let field_count = row_type.fields().len();
+ let mut nested = CompactedRowWriter::new(field_count);
+ for (i, field) in row_type.fields().iter().enumerate() {
+ let datum = &inner_row.values[i];
+ if datum.is_null() {
+ if field.data_type.is_nullable() {
+ nested.set_null_at(i);
+ }
+ } else {
+ let vw =
+ InnerValueWriter::create_inner_value_writer(&field.data_type, None)
+ .expect("create_inner_value_writer failed for nested row field");
+ vw.write_value(&mut nested, i, datum)
+ .expect("write_value failed for nested row field");
+ }
+ }
+ writer.write_bytes(nested.buffer());
+ }
_ => {
return Err(IllegalArgument {
message: format!("{self:?} used to write value {value:?}"),
diff --git a/crates/fluss/src/row/column.rs b/crates/fluss/src/row/column.rs
index be5b850d..95cd3395 100644
--- a/crates/fluss/src/row/column.rs
+++ b/crates/fluss/src/row/column.rs
@@ -17,8 +17,8 @@
use crate::error::Error::IllegalArgument;
use crate::error::Result;
-use crate::row::InternalRow;
-use crate::row::datum::{Date, Time, TimestampLtz, TimestampNtz};
+use crate::row::{GenericRow, InternalRow};
+use crate::row::datum::{Date, Datum, Time, TimestampLtz, TimestampNtz};
use arrow::array::{
Array, AsArray, BinaryArray, BooleanArray, FixedSizeBinaryArray, ListArray, RecordBatch,
StringArray,
@@ -35,25 +35,33 @@ use std::sync::Arc;
pub struct ColumnarRow {
record_batch: Arc<RecordBatch>,
row_id: usize,
+ nested_rows: Vec<std::sync::OnceLock<GenericRow<'static>>>,
}
impl ColumnarRow {
pub fn new(batch: Arc<RecordBatch>) -> Self {
+ let num_cols = batch.num_columns();
ColumnarRow {
record_batch: batch,
row_id: 0,
+ nested_rows: (0..num_cols).map(|_| std::sync::OnceLock::new()).collect(),
}
}
pub fn new_with_row_id(bach: Arc<RecordBatch>, row_id: usize) -> Self {
+ let num_cols = bach.num_columns();
ColumnarRow {
record_batch: bach,
row_id,
+ nested_rows: (0..num_cols).map(|_| std::sync::OnceLock::new()).collect(),
}
}
pub fn set_row_id(&mut self, row_id: usize) {
- self.row_id = row_id
+ self.row_id = row_id;
+ for lock in &mut self.nested_rows {
+ *lock = std::sync::OnceLock::new();
+ }
}
pub fn get_row_id(&self) -> usize {
@@ -212,6 +220,168 @@ impl ColumnarRow {
}),
}
}
+
+ /// Extract a `GenericRow<'static>` from a column in the RecordBatch at the given row_id.
+ fn extract_struct_at(
+ batch: &RecordBatch,
+ pos: usize,
+ row_id: usize,
+ ) -> Result<GenericRow<'static>> {
+ let col = batch.column(pos);
+ Self::extract_struct_from_array(col.as_ref(), row_id)
+ }
+
+ /// Recursively extract a `GenericRow<'static>` from a `StructArray` at row_id.
+ fn extract_struct_from_array(array: &dyn Array, row_id: usize) -> Result<GenericRow<'static>> {
+ use arrow::array::StructArray;
+ let sa = array
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .ok_or_else(|| IllegalArgument {
+ message: format!("expected StructArray, got {:?}", array.data_type()),
+ })?;
+ let mut values = Vec::with_capacity(sa.num_columns());
+ for i in 0..sa.num_columns() {
+ let child = sa.column(i);
+ values.push(Self::arrow_value_to_datum(child.as_ref(), row_id)?);
+ }
+ Ok(GenericRow { values })
+ }
+
+ /// Convert a single element at `row_id` in an Arrow array to a `Datum<'static>`.
+ fn arrow_value_to_datum(array: &dyn Array, row_id: usize) -> Result<Datum<'static>> {
+ use arrow::array::{
+ BooleanArray, Decimal128Array, Float32Array, Float64Array, Int8Array, Int16Array,
+ Int32Array, Int64Array, Time32MillisecondArray, Time32SecondArray,
+ Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
+ };
+ use crate::row::Decimal;
+
+ if array.is_null(row_id) {
+ return Ok(Datum::Null);
+ }
+
+ match array.data_type() {
+ ArrowDataType::Boolean => {
+ let a = array.as_any().downcast_ref::<BooleanArray>().unwrap();
+ Ok(Datum::Bool(a.value(row_id)))
+ }
+ ArrowDataType::Int8 => {
+ let a = array.as_any().downcast_ref::<Int8Array>().unwrap();
+ Ok(Datum::Int8(a.value(row_id)))
+ }
+ ArrowDataType::Int16 => {
+ let a = array.as_any().downcast_ref::<Int16Array>().unwrap();
+ Ok(Datum::Int16(a.value(row_id)))
+ }
+ ArrowDataType::Int32 => {
+ let a = array.as_any().downcast_ref::<Int32Array>().unwrap();
+ Ok(Datum::Int32(a.value(row_id)))
+ }
+ ArrowDataType::Int64 => {
+ let a = array.as_any().downcast_ref::<Int64Array>().unwrap();
+ Ok(Datum::Int64(a.value(row_id)))
+ }
+ ArrowDataType::Float32 => {
+ let a = array.as_any().downcast_ref::<Float32Array>().unwrap();
+ Ok(Datum::Float32(a.value(row_id).into()))
+ }
+ ArrowDataType::Float64 => {
+ let a = array.as_any().downcast_ref::<Float64Array>().unwrap();
+ Ok(Datum::Float64(a.value(row_id).into()))
+ }
+ ArrowDataType::Utf8 => {
+ let a = array.as_any().downcast_ref::<StringArray>().unwrap();
+ Ok(Datum::String(std::borrow::Cow::Owned(a.value(row_id).to_owned())))
+ }
+ ArrowDataType::Binary => {
+ let a = array.as_any().downcast_ref::<BinaryArray>().unwrap();
+ Ok(Datum::Blob(std::borrow::Cow::Owned(a.value(row_id).to_vec())))
+ }
+ ArrowDataType::Decimal128(p, s) => {
+ let (p, s) = (*p, *s);
+ let a = array.as_any().downcast_ref::<Decimal128Array>().unwrap();
+ let i128_val = a.value(row_id);
+ Ok(Datum::Decimal(Decimal::from_arrow_decimal128(
+ i128_val,
+ s as i64,
+ p as u32,
+ s as u32,
+ )?))
+ }
+ ArrowDataType::Date32 => {
+ let a = array.as_any().downcast_ref::<arrow::array::Date32Array>().unwrap();
+ Ok(Datum::Date(Date::new(a.value(row_id))))
+ }
+ ArrowDataType::Time32(TimeUnit::Second) => {
+ let a = array.as_any().downcast_ref::<Time32SecondArray>().unwrap();
+ Ok(Datum::Time(Time::new(a.value(row_id) * 1000)))
+ }
+ ArrowDataType::Time32(TimeUnit::Millisecond) => {
+ let a = array.as_any().downcast_ref::<Time32MillisecondArray>().unwrap();
+ Ok(Datum::Time(Time::new(a.value(row_id))))
+ }
+ ArrowDataType::Time64(TimeUnit::Microsecond) => {
+ let a = array.as_any().downcast_ref::<Time64MicrosecondArray>().unwrap();
+ Ok(Datum::Time(Time::new((a.value(row_id) / 1000) as i32)))
+ }
+ ArrowDataType::Time64(TimeUnit::Nanosecond) => {
+ let a = array.as_any().downcast_ref::<Time64NanosecondArray>().unwrap();
+ Ok(Datum::Time(Time::new((a.value(row_id) / 1_000_000) as i32)))
+ }
+ ArrowDataType::Timestamp(time_unit, tz) => {
+ let value: i64 = match time_unit {
+ TimeUnit::Second => {
+ array.as_any().downcast_ref::<TimestampSecondArray>().unwrap().value(row_id)
+ }
+ TimeUnit::Millisecond => {
+ array.as_any().downcast_ref::<TimestampMillisecondArray>().unwrap().value(row_id)
+ }
+ TimeUnit::Microsecond => {
+ array.as_any().downcast_ref::<TimestampMicrosecondArray>().unwrap().value(row_id)
+ }
+ TimeUnit::Nanosecond => {
+ array.as_any().downcast_ref::<TimestampNanosecondArray>().unwrap().value(row_id)
+ }
+ };
+ let (millis, nanos) = match time_unit {
+ TimeUnit::Second => (value * 1000, 0i32),
+ TimeUnit::Millisecond => (value, 0i32),
+ TimeUnit::Microsecond => {
+ let millis = value.div_euclid(1000);
+ let nanos = (value.rem_euclid(1000) * 1000) as i32;
+ (millis, nanos)
+ }
+ TimeUnit::Nanosecond => {
+ let millis = value.div_euclid(1_000_000);
+ let nanos = value.rem_euclid(1_000_000) as i32;
+ (millis, nanos)
+ }
+ };
+ if tz.is_some() {
+ if nanos == 0 {
+ Ok(Datum::TimestampLtz(TimestampLtz::new(millis)))
+ } else {
+ Ok(Datum::TimestampLtz(TimestampLtz::from_millis_nanos(millis, nanos)?))
+ }
+ } else if nanos == 0 {
+ Ok(Datum::TimestampNtz(TimestampNtz::new(millis)))
+ } else {
+ Ok(Datum::TimestampNtz(TimestampNtz::from_millis_nanos(millis, nanos)?))
+ }
+ }
+ ArrowDataType::Struct(_) => {
+ let nested = Self::extract_struct_from_array(array, row_id)?;
+ Ok(Datum::Row(Box::new(nested)))
+ }
+ other => Err(IllegalArgument {
+ message: format!(
+ "unsupported Arrow data type for nested row extraction: {other:?}"
+ ),
+ }),
+ }
+ }
}
impl InternalRow for ColumnarRow {
@@ -433,6 +603,18 @@ impl InternalRow for ColumnarRow {
write_arrow_values_to_fluss_array(&*values, &element_fluss_type, &mut writer)?;
writer.complete()
}
+
+ fn get_row(&self, pos: usize) -> Result<&GenericRow<'_>> {
+ let lock = self.nested_rows.get(pos).ok_or_else(|| IllegalArgument {
+ message: format!("column index {pos} out of bounds for get_row"),
+ })?;
+ let batch = Arc::clone(&self.record_batch);
+ let row_id = self.row_id;
+ Ok(lock.get_or_init(|| {
+ Self::extract_struct_at(&batch, pos, row_id)
+ .expect("failed to extract nested row from StructArray")
+ }))
+ }
}
/// Downcast to a primitive Arrow array type, then loop with null checks calling a writer method.
@@ -794,9 +976,9 @@ mod tests {
use arrow::array::{
ArrayRef, BinaryArray, BooleanArray, Decimal128Array, Float32Array, Float64Array,
Int8Array, Int16Array, Int32Array, Int32Builder, Int64Array, ListBuilder, StringArray,
- UInt32Builder,
+ StructArray, UInt32Builder,
};
- use arrow::datatypes::{DataType, Field, Schema};
+ use arrow::datatypes::{DataType, Field, Fields, Schema};
fn single_column_row(array: ArrayRef) -> ColumnarRow {
let batch =
@@ -1011,4 +1193,112 @@ mod tests {
"unexpected error: {err}"
);
}
+
+ fn make_struct_batch(
+ field_name: &str,
+ child_fields: Fields,
+ child_arrays: Vec<Arc<dyn Array>>,
+ _num_rows: usize,
+ ) -> Arc<RecordBatch> {
+ let struct_array = StructArray::new(child_fields.clone(), child_arrays, None);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ field_name,
+ DataType::Struct(child_fields),
+ false,
+ )]));
+ Arc::new(
+ RecordBatch::try_new(schema, vec![Arc::new(struct_array)])
+ .expect("record batch"),
+ )
+ }
+
+ #[test]
+ fn columnar_row_reads_nested_row() {
+ // Build a RecordBatch with a Struct column: {i32, string}
+ let child_fields = Fields::from(vec![
+ Field::new("x", DataType::Int32, false),
+ Field::new("s", DataType::Utf8, false),
+ ]);
+ let child_arrays: Vec<Arc<dyn Array>> = vec![
+ Arc::new(Int32Array::from(vec![42, 99])),
+ Arc::new(StringArray::from(vec!["hello", "world"])),
+ ];
+ let batch = make_struct_batch("nested", child_fields, child_arrays, 2);
+
+ let mut row = ColumnarRow::new(batch);
+
+ // row_id = 0
+ let nested = row.get_row(0).unwrap();
+ assert_eq!(nested.get_field_count(), 2);
+ assert_eq!(nested.get_int(0).unwrap(), 42);
+ assert_eq!(nested.get_string(1).unwrap(), "hello");
+
+ // row_id = 1
+ row.set_row_id(1);
+ let nested = row.get_row(0).unwrap();
+ assert_eq!(nested.get_int(0).unwrap(), 99);
+ assert_eq!(nested.get_string(1).unwrap(), "world");
+ }
+
+ #[test]
+ fn columnar_row_reads_deeply_nested_row() {
+ // Build: outer struct { i32, inner struct { string } }
+ let inner_fields = Fields::from(vec![Field::new("s", DataType::Utf8, false)]);
+ let inner_array = Arc::new(StructArray::new(
+ inner_fields.clone(),
+ vec![Arc::new(StringArray::from(vec!["deep", "deeper"])) as Arc<dyn Array>],
+ None,
+ ));
+
+ let outer_fields = Fields::from(vec![
+ Field::new("n", DataType::Int32, false),
+ Field::new("inner", DataType::Struct(inner_fields), false),
+ ]);
+ let outer_array = Arc::new(StructArray::new(
+ outer_fields.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])) as Arc<dyn Array>,
+ inner_array as Arc<dyn Array>,
+ ],
+ None,
+ ));
+
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "outer",
+ DataType::Struct(outer_fields),
+ false,
+ )]));
+ let batch = Arc::new(
+ RecordBatch::try_new(schema, vec![outer_array]).expect("record batch"),
+ );
+
+ let row = ColumnarRow::new(batch);
+
+ // Access outer struct at column 0, row 0
+ let outer = row.get_row(0).unwrap();
+ assert_eq!(outer.get_int(0).unwrap(), 1);
+
+ // Access inner struct (column 1 of outer)
+ let inner = outer.get_row(1).unwrap();
+ assert_eq!(inner.get_string(0).unwrap(), "deep");
+ }
+
+ #[test]
+ fn columnar_row_get_row_cache_invalidated_on_set_row_id() {
+ let child_fields = Fields::from(vec![Field::new("x", DataType::Int32, false)]);
+ let child_arrays: Vec<Arc<dyn Array>> =
+ vec![Arc::new(Int32Array::from(vec![10, 20]))];
+ let batch = make_struct_batch("s", child_fields, child_arrays, 2);
+
+ let mut row = ColumnarRow::new(batch);
+
+ // row_id = 0: nested x = 10
+ let nested_0 = row.get_row(0).unwrap();
+ assert_eq!(nested_0.get_int(0).unwrap(), 10);
+
+ // After set_row_id(1), cache is cleared → nested x = 20
+ row.set_row_id(1);
+ let nested_1 = row.get_row(0).unwrap();
+ assert_eq!(nested_1.get_int(0).unwrap(), 20);
+ }
}
diff --git a/crates/fluss/src/row/compacted/compacted_row.rs b/crates/fluss/src/row/compacted/compacted_row.rs
index 267ae13c..24e4bd79 100644
--- a/crates/fluss/src/row/compacted/compacted_row.rs
+++ b/crates/fluss/src/row/compacted/compacted_row.rs
@@ -171,6 +171,10 @@ impl<'a> InternalRow for CompactedRow<'a> {
self.decoded_row()?.get_array(pos)
}
+ fn get_row(&self, pos: usize) -> Result<&GenericRow<'_>> {
+ self.decoded_row().get_row(pos)
+ }
+
fn as_encoded_bytes(&self, write_format: WriteFormat) -> Option<&[u8]> {
match write_format {
WriteFormat::CompactedKv => Some(self.as_bytes()),
diff --git a/crates/fluss/src/row/compacted/compacted_row_reader.rs b/crates/fluss/src/row/compacted/compacted_row_reader.rs
index 4ae442fc..9d7fb6cb 100644
--- a/crates/fluss/src/row/compacted/compacted_row_reader.rs
+++ b/crates/fluss/src/row/compacted/compacted_row_reader.rs
@@ -187,6 +187,18 @@ impl<'a> CompactedRowDeserializer<'a> {
let array = crate::row::binary_array::FlussArray::from_bytes(bytes)?;
(Datum::Array(array), next)
}
+ DataType::Row(row_type) => {
+ let (nested_bytes, next) = reader.read_bytes(cursor)?;
+ let nested_reader = CompactedRowReader::new(
+ row_type.fields().len(),
+ nested_bytes,
+ 0,
+ nested_bytes.len(),
+ );
+ let nested_deser = CompactedRowDeserializer::new_from_owned(row_type.clone());
+ let nested_row = nested_deser.deserialize(&nested_reader)?;
+ (Datum::Row(Box::new(nested_row)), next)
+ }
_ => {
return Err(IllegalArgument {
message: format!(
@@ -326,3 +338,142 @@ impl<'a> CompactedRowReader<'a> {
Ok((s, next_pos))
}
}
+
+#[cfg(test)]
+mod row_type_tests {
+ use crate::metadata::{DataType, DataTypes, RowType};
+ use crate::row::compacted::compacted_row_reader::{CompactedRowDeserializer, CompactedRowReader};
+ use crate::row::compacted::compacted_row_writer::CompactedRowWriter;
+ use crate::row::binary::ValueWriter;
+ use crate::row::field_getter::FieldGetter;
+ use crate::row::{Datum, GenericRow, InternalRow};
+
+ fn round_trip<F>(outer_row_type: &RowType, outer_row: &GenericRow, verify: F)
+ where
+ F: FnOnce(&GenericRow),
+ {
+ // Write
+ let field_getters = FieldGetter::create_field_getters(outer_row_type);
+ let value_writers: Vec<ValueWriter> = outer_row_type
+ .fields()
+ .iter()
+ .map(|f| ValueWriter::create_value_writer(f.data_type(), None).unwrap())
+ .collect();
+ let mut writer = CompactedRowWriter::new(outer_row_type.fields().len());
+ for (i, (getter, vw)) in field_getters.iter().zip(value_writers.iter()).enumerate() {
+ let datum = getter.get_field(outer_row as &dyn InternalRow).unwrap();
+ vw.write_value(&mut writer, i, &datum).unwrap();
+ }
+ let bytes = writer.to_bytes();
+
+ // Read
+ let deser = CompactedRowDeserializer::new(outer_row_type);
+ let reader = CompactedRowReader::new(
+ outer_row_type.fields().len(),
+ bytes.as_ref(),
+ 0,
+ bytes.len(),
+ );
+ let result = deser.deserialize(&reader);
+ verify(&result);
+ }
+
+ #[test]
+ fn test_row_simple_nesting() {
+ // ROW<INT, STRING> nested inside an outer row
+ let inner_row_type = RowType::with_data_types_and_field_names(
+ vec![DataTypes::int(), DataTypes::string()],
+ vec!["x", "label"],
+ );
+ let outer_row_type = RowType::with_data_types_and_field_names(
+ vec![DataTypes::int(), DataType::Row(inner_row_type.clone())],
+ vec!["id", "nested"],
+ );
+
+ let mut inner = GenericRow::new(2);
+ inner.set_field(0, 42_i32);
+ inner.set_field(1, "hello");
+
+ let mut outer = GenericRow::new(2);
+ outer.set_field(0, 1_i32);
+ outer.set_field(1, Datum::Row(Box::new(inner)));
+
+ round_trip(&outer_row_type, &outer, |result| {
+ assert_eq!(result.get_int(0).unwrap(), 1);
+ let nested = result.get_row(1).unwrap();
+ assert_eq!(nested.get_int(0).unwrap(), 42);
+ assert_eq!(nested.get_string(1).unwrap(), "hello");
+ });
+ }
+
+ #[test]
+ fn test_row_deep_nesting() {
+ // ROW<ROW<INT>> — two levels of nesting
+ let inner_inner_row_type = RowType::with_data_types_and_field_names(
+ vec![DataTypes::int()],
+ vec!["n"],
+ );
+ let inner_row_type = RowType::with_data_types_and_field_names(
+ vec![DataType::Row(inner_inner_row_type.clone())],
+ vec!["inner"],
+ );
+ let outer_row_type = RowType::with_data_types_and_field_names(
+ vec![DataType::Row(inner_row_type.clone())],
+ vec!["outer"],
+ );
+
+ let mut innermost = GenericRow::new(1);
+ innermost.set_field(0, 99_i32);
+
+ let mut middle = GenericRow::new(1);
+ middle.set_field(0, Datum::Row(Box::new(innermost)));
+
+ let mut outer = GenericRow::new(1);
+ outer.set_field(0, Datum::Row(Box::new(middle)));
+
+ round_trip(&outer_row_type, &outer, |result| {
+ let mid = result.get_row(0).unwrap();
+ let inner = mid.get_row(0).unwrap();
+ assert_eq!(inner.get_int(0).unwrap(), 99);
+ });
+ }
+
+ #[test]
+ fn test_row_with_nullable_fields() {
+ // Outer nullable ROW column; nested row with a nullable STRING field set to null
+ let inner_row_type = RowType::with_data_types_and_field_names(
+ vec![DataTypes::int(), DataTypes::string()],
+ vec!["id", "optional_name"],
+ );
+ let outer_row_type = RowType::with_data_types_and_field_names(
+ vec![DataTypes::int(), DataType::Row(inner_row_type.clone())],
+ vec!["k", "nested"],
+ );
+
+ // Case 1: non-null nested row with a null field inside
+ let mut inner = GenericRow::new(2);
+ inner.set_field(0, 7_i32);
+ inner.set_field(1, Datum::Null);
+
+ let mut outer = GenericRow::new(2);
+ outer.set_field(0, 10_i32);
+ outer.set_field(1, Datum::Row(Box::new(inner)));
+
+ round_trip(&outer_row_type, &outer, |result| {
+ assert_eq!(result.get_int(0).unwrap(), 10);
+ let nested = result.get_row(1).unwrap();
+ assert_eq!(nested.get_int(0).unwrap(), 7);
+ assert!(nested.is_null_at(1).unwrap());
+ });
+
+ // Case 2: outer ROW column is null
+ let mut outer_null = GenericRow::new(2);
+ outer_null.set_field(0, 20_i32);
+ outer_null.set_field(1, Datum::Null);
+
+ round_trip(&outer_row_type, &outer_null, |result2| {
+ assert_eq!(result2.get_int(0).unwrap(), 20);
+ assert!(result2.is_null_at(1).unwrap());
+ });
+ }
+}
diff --git a/crates/fluss/src/row/datum.rs b/crates/fluss/src/row/datum.rs
index 78dc5498..865d4b4f 100644
--- a/crates/fluss/src/row/datum.rs
+++ b/crates/fluss/src/row/datum.rs
@@ -18,6 +18,7 @@
use crate::error::Error::RowConvertError;
use crate::error::Result;
use crate::row::Decimal;
+use crate::row::GenericRow;
use crate::row::binary_array::FlussArray;
use arrow::array::{
ArrayBuilder, BinaryBuilder, BooleanBuilder, Date32Builder, Decimal128Builder,
@@ -72,6 +73,8 @@ pub enum Datum<'a> {
TimestampLtz(TimestampLtz),
#[display("{0}")]
Array(FlussArray),
+ #[display("{0:?}")]
+ Row(Box<GenericRow<'a>>),
}
impl Datum<'_> {
@@ -134,6 +137,13 @@ impl Datum<'_> {
_ => panic!("not an array: {self:?}"),
}
}
+
+ pub fn as_row(&self) -> &GenericRow<'_> {
+ match self {
+ Self::Row(r) => r.as_ref(),
+ _ => panic!("not a row: {self:?}"),
+ }
+ }
}
// ----------- implement from
@@ -878,6 +888,11 @@ impl Datum<'_> {
Datum::Array(arr) => {
return append_fluss_array_to_list_builder(arr, builder, data_type);
}
+ Datum::Row(_) => {
+ return Err(RowConvertError {
+ message: "append_to is not supported for Row type".to_string(),
+ });
+ }
}
Err(RowConvertError {
diff --git a/crates/fluss/src/row/encode/compacted_key_encoder.rs b/crates/fluss/src/row/encode/compacted_key_encoder.rs
index 877b3ecd..a6ba580f 100644
--- a/crates/fluss/src/row/encode/compacted_key_encoder.rs
+++ b/crates/fluss/src/row/encode/compacted_key_encoder.rs
@@ -109,7 +109,7 @@ impl KeyEncoder for CompactedKeyEncoder {
#[cfg(test)]
mod tests {
use super::*;
- use crate::metadata::DataTypes;
+ use crate::metadata::{DataType, DataTypes};
use crate::row::binary_array::FlussArrayWriter;
use crate::row::{Datum, GenericRow};
@@ -493,4 +493,50 @@ mod tests {
encoded.iter().as_slice()
);
}
+
+ #[test]
+ fn test_row_as_primary_key() {
+ // ROW<INT, STRING> as a primary key column
+ let inner_row_type = RowType::with_data_types_and_field_names(
+ vec![DataTypes::int(), DataTypes::string()],
+ vec!["x", "label"],
+ );
+ let row_type = RowType::with_data_types_and_field_names(
+ vec![
+ DataTypes::int(),
+ DataType::Row(inner_row_type.clone()),
+ ],
+ vec!["id", "nested"],
+ );
+
+ let mut inner = GenericRow::new(2);
+ inner.set_field(0, 42_i32);
+ inner.set_field(1, "hello");
+
+ let mut row = GenericRow::new(2);
+ row.set_field(0, 1_i32);
+ row.set_field(1, Datum::Row(Box::new(inner)));
+
+ let mut encoder = for_test_row_type(&row_type);
+ let encoded = encoder.encode_key(&row).unwrap();
+
+ // Verify it encodes without error and produces non-empty bytes
+ assert!(!encoded.is_empty());
+
+ // Encode the same row again to verify determinism
+ let encoded2 = encoder.encode_key(&row).unwrap();
+ assert_eq!(encoded, encoded2);
+
+ // Encode a different nested row and verify different output
+ let mut inner2 = GenericRow::new(2);
+ inner2.set_field(0, 99_i32);
+ inner2.set_field(1, "world");
+
+ let mut row2 = GenericRow::new(2);
+ row2.set_field(0, 1_i32);
+ row2.set_field(1, Datum::Row(Box::new(inner2)));
+
+ let encoded3 = encoder.encode_key(&row2).unwrap();
+ assert_ne!(encoded, encoded3);
+ }
}
diff --git a/crates/fluss/src/row/field_getter.rs b/crates/fluss/src/row/field_getter.rs
index 69e08602..a1ea378f 100644
--- a/crates/fluss/src/row/field_getter.rs
+++ b/crates/fluss/src/row/field_getter.rs
@@ -82,8 +82,9 @@ impl FieldGetter {
pos,
precision: t.precision(),
},
- // TODO: add Map and Row variants when get_map/get_row are available in InternalRow.
+ // TODO: add Map variant when get_map is available in InternalRow.
DataType::Array(_) => InnerFieldGetter::Array { pos },
+ DataType::Row(_) => InnerFieldGetter::Row { pos },
_ => unimplemented!("DataType {:?} is currently unimplemented", data_type),
};
@@ -154,6 +155,9 @@ pub enum InnerFieldGetter {
Array {
pos: usize,
},
+ Row {
+ pos: usize,
+ },
}
impl InnerFieldGetter {
@@ -183,8 +187,9 @@ impl InnerFieldGetter {
InnerFieldGetter::TimestampLtz { pos, precision } => {
Datum::TimestampLtz(row.get_timestamp_ltz(*pos, *precision)?)
}
- // TODO: add Map and Row field getter support once their binary forms are implemented.
+ // TODO: add Map field getter support once its binary form is implemented.
InnerFieldGetter::Array { pos } => Datum::Array(row.get_array(*pos)?),
+ InnerFieldGetter::Row { pos } => Datum::Row(Box::new(row.get_row(*pos)?.clone())),
})
}
@@ -206,7 +211,8 @@ impl InnerFieldGetter {
| Self::Time { pos }
| Self::Timestamp { pos, .. }
| Self::TimestampLtz { pos, .. }
- | Self::Array { pos } => *pos,
+ | Self::Array { pos }
+ | Self::Row { pos } => *pos,
}
}
}
diff --git a/crates/fluss/src/row/mod.rs b/crates/fluss/src/row/mod.rs
index c6bdd7c5..fb599fc9 100644
--- a/crates/fluss/src/row/mod.rs
+++ b/crates/fluss/src/row/mod.rs
@@ -34,6 +34,7 @@ mod row_decoder;
use crate::client::WriteFormat;
pub use binary_array::FlussArray;
use bytes::Bytes;
+use serde::Serialize;
pub use column::*;
pub use compacted::CompactedRow;
pub use datum::*;
@@ -130,13 +131,20 @@ pub trait InternalRow: Send + Sync {
/// Returns the array value at the given position
fn get_array(&self, pos: usize) -> Result<FlussArray>;
+ /// Returns the nested row value at the given position
+ fn get_row(&self, pos: usize) -> Result<&GenericRow<'_>> {
+ Err(crate::error::Error::IllegalArgument {
+ message: format!("get_row not supported at position {pos}"),
+ })
+ }
+
/// Returns encoded bytes if already encoded
fn as_encoded_bytes(&self, _write_format: WriteFormat) -> Option<&[u8]> {
None
}
}
-#[derive(Debug)]
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize)]
pub struct GenericRow<'a> {
pub values: Vec<Datum<'a>>,
}
@@ -294,6 +302,15 @@ impl<'a> InternalRow for GenericRow<'a> {
}),
}
}
+
+ fn get_row(&self, pos: usize) -> Result<&GenericRow<'_>> {
+ match self.get_value(pos)? {
+ Datum::Row(r) => Ok(r.as_ref()),
+ other => Err(IllegalArgument {
+ message: format!("type mismatch at position {pos}: expected Row, got {other:?}"),
+ }),
+ }
+ }
}
impl<'a> GenericRow<'a> {