You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "hemanthsavasere (via GitHub)" <gi...@apache.org> on 2026/03/15 19:08:09 UTC
[PR] feat: add end-to-end ROW (nested struct) column serialization support [fluss-rust]
hemanthsavasere opened a new pull request, #442:
URL: https://github.com/apache/fluss-rust/pull/442
### Purpose
Linked issue: close #388
`DataType::Row` was defined in the schema layer but the serialization stack was entirely missing — `CompactedRowWriter`/`Reader` panicked, `FieldGetter`/`ValueWriter` hit `unimplemented!()`, and `Datum` had no `Row` variant. This PR wires up the full stack.
### Brief change log
- Add `Datum::Row(Box<GenericRow<'a>>)` variant and `as_row()` accessor
- Add `get_row()` to `InternalRow` trait with a default error impl (non-breaking)
- Implement `get_row()` on `GenericRow` and `CompactedRow` (delegates to decoded row)
- Implement `ColumnarRow::get_row()` via Arrow `StructArray` extraction with `OnceLock` caching per column, invalidated on `set_row_id()`
- Add `InnerValueWriter::Row(RowType)` — serializes into a temporary `CompactedRowWriter` then calls `write_bytes`, matching Java's `writeRow` wire format
- Add `DataType::Row` arm in `CompactedRowDeserializer` using `read_bytes` + recursive deserialize
- Add `InnerFieldGetter::Row` — automatically enables ROW in `CompactedKeyEncoder` with no further changes
- Handle `Datum::Row` in C++ `resolve_row_types` (recursive)
### Tests
- `test_row_simple_nesting` — round-trip `ROW<INT, STRING>`
- `test_row_deep_nesting` — round-trip `ROW<ROW<INT>>` (two levels)
- `test_row_with_nullable_fields` — null field inside nested row; null outer ROW column
- `test_row_as_primary_key` — ROW through `CompactedKeyEncoder`; asserts non-empty, deterministic, and distinguishable output
- `columnar_row_reads_nested_row`, `columnar_row_reads_deeply_nested_row`, `columnar_row_get_row_cache_invalidated_on_set_row_id` — Arrow `StructArray` extraction
All 257 tests pass. No new clippy warnings.
### API and Format
**API**: `Datum::Row` is a new variant — exhaustive `match` on `Datum` will require a new arm. `InternalRow::get_row()` is additive (default impl).
**Wire format**: Unchanged. ROW uses varint-length + CompactedRow blob, identical to `String`/`Bytes` — matches Java reference byte-for-byte.
### Documentation
No new documentation needed.
### How to verify it
- [x] `cargo test --workspace` — all 257 tests pass
- [x] `cargo clippy --workspace -- -D warnings` — no new warnings
- [ ] Manual: review the diff for correctness against the linked plan/research doc
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat : Row data type support in Rust [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #442:
URL: https://github.com/apache/fluss-rust/pull/442#issuecomment-4070413463
Additionally, please can you add to existing integration test? TY
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat : Row data type support in Rust [fluss-rust]
Posted by "hemanthsavasere (via GitHub)" <gi...@apache.org>.
hemanthsavasere commented on PR #442:
URL: https://github.com/apache/fluss-rust/pull/442#issuecomment-4063724780
Hi @fresh-borzoni,
Can you please review the PR. Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat : Row data type support in Rust [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni closed pull request #442: Feat : Row data type support in Rust
URL: https://github.com/apache/fluss-rust/pull/442
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat : Row data type support in Rust [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #442:
URL: https://github.com/apache/fluss-rust/pull/442#discussion_r2942537008
##########
crates/fluss/src/row/binary/binary_writer.rs:
##########
@@ -237,6 +238,26 @@ impl InnerValueWriter {
(InnerValueWriter::TimestampLtz(p), Datum::TimestampLtz(ts)) => {
writer.write_timestamp_ltz(ts, *p);
}
+ (InnerValueWriter::Row(row_type), Datum::Row(inner_row)) => {
Review Comment:
Why don't we delegate like on Java side?
https://github.com/apache/fluss/blob/main/fluss-common/src/main/java/org/apache/fluss/row/BinaryWriter.java#L176
##########
crates/fluss/src/row/column.rs:
##########
@@ -533,4 +715,112 @@ mod tests {
.unwrap()
);
}
+
+ fn make_struct_batch(
Review Comment:
Is this for test only? If so, move under '#[test]'/mod?
##########
crates/fluss/src/row/column.rs:
##########
@@ -533,4 +715,112 @@ mod tests {
.unwrap()
);
}
+
+ 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);
Review Comment:
Not: should we assert second row as well?
##########
crates/fluss/src/row/compacted/compacted_row_reader.rs:
##########
@@ -160,6 +160,18 @@ impl<'a> CompactedRowDeserializer<'a> {
(Datum::TimestampLtz(timestamp_ltz), 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());
Review Comment:
Can we use from 'new' which borrows instead? Seems like nested_deser does not live beyond current scope anyway.
##########
crates/fluss/src/row/column.rs:
##########
@@ -209,6 +217,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();
Review Comment:
Let's error with appropriate message instead of unwrapping/panic?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat : Row data type support in Rust [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #442:
URL: https://github.com/apache/fluss-rust/pull/442#issuecomment-4070081355
@charlesdong1991 would be good if you can review this as well as you've worked on the array type.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat : Row data type support in Rust [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #442:
URL: https://github.com/apache/fluss-rust/pull/442#issuecomment-4234367236
@hemanthsavasere Bump, are you able to address the comments? Thank you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat : Row data type support in Rust [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #442:
URL: https://github.com/apache/fluss-rust/pull/442#issuecomment-4416005707
Reviving this as #544 - rebased on current main, addressed review feedback, fixes and improvements.
@hemanthsavasere your commit is preserved with original authorship, thanks for the original implementation 🙇
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org