feat: implement FlussMap support for compacted rows and binary encoding by qzyu999 · Pull Request #530 · apache/fluss-rust

qzyu999 · 2026-05-05T05:20:08Z

Purpose

Linked issue: close #387

This pull request introduces support for the MAP data type in the Fluss Rust client. It allows developers to define associative arrays (key-value pairs) in their table schemas and provides the underlying binary representation (FlussMap) necessary to process these fields, ensuring wire compatibility with the Java client's BinaryMap.

Brief change log

Metadata & Schema: Added MapType to the DataType enum and implemented the DataTypes::map() factory. Updated eq_ignore_nullable to support recursive map comparison.
Core Data Type: Implemented FlussMap in crates/fluss/src/row/binary_map.rs. This provides a zero-copy, wire-compatible wrapper for the [4B key_size] + [keys] + [values] binary layout.
Row Integration: Extended the InternalRow trait with get_map() and implemented it across GenericRow, CompactedRow, LookupRow, ProjectedRow, and nested FlussArray accessors.
Serialization/Deserialization:
- Implemented write_map in BinaryWriter, CompactedRowWriter, and CompactedKeyWriter.
- Updated CompactedRowDeserializer to handle MAP data type reconstruction.
- Added FieldGetter and InnerFieldGetter support for dynamically accessing map fields.
Arrow Interoperability:
- Implemented ColumnarRow::get_map() to convert Arrow MapArray structures into FlussMap.
- Implemented Datum::Map::append_to using Arrow's MapBuilder for seamless integration with the Arrow ecosystem.
- Updated ColumnWriter to support building Arrow MapArray outputs from Fluss rows.
Bindings: Updated Python, C++, and Elixir binding layers to support Datum::Map during serialization and row processing.
Key Constraints: Explicitly enforced the rejection of MAP types as primary or bucketing keys in the CompactedKeyWriter.

Tests

Added unit tests covering the new functionality:

Unit Tests: Added comprehensive coverage in binary_map.rs for round-trips, empty maps, null values, and malformed buffer validation.
Integration Tests: Added 4 end-to-end integration tests in compacted_row.rs covering basic maps, nullable map columns, nested maps (Map<String, Array<Int>>), and empty maps.
Interoperability Tests: Added a ColumnWriter unit test to verify that MAP columns are correctly serialized into valid Arrow MapArray structures.
Regression Tests: Added test_encode_map_rejected to ensure MAP cannot be incorrectly used as a key type.

API and Format

API: Yes. Users can now use DataTypes::map() for schemas and row.get_map() for data retrieval.
Format: Yes. This implements the memory/wire layout for map serialization, maintaining compatibility with the Java implementation.

Documentation

Yes. This adds the MAP data type. User guides will need updating to reflect that MAP is now a supported type in the Rust client.

leekeiabstraction · 2026-05-05T06:41:37Z

@charlesdong1991 @fresh-borzoni Appreciate if you guys can have a look as well.

leekeiabstraction

TY for the PR! Left an early comment.

leekeiabstraction · 2026-05-05T07:04:57Z

    fn write_array(&mut self, value: &[u8]);

+    fn write_map(&mut self, value: &[u8]);
+


Appreciate that you are following convention from write_array.

I'm thinking about the ease of read/use of this signature. The caller would have to know in advance how to get the map into the form of a single byte array.

Why not pass FlussMap? Save some .as_bytes() calls and get the benefit of stronger compile time check vs. passing byte array.

Hi @leekeiabstraction, thanks for the suggestion, made appropriate changes in 9f995f8.

leekeiabstraction

TY for the PR! Added comments, PTAL @qzyu999

leekeiabstraction · 2026-05-06T06:49:28Z

+        use crate::record::from_arrow_type;
+        use crate::row::binary_array::FlussArrayWriter;
+        use crate::row::binary_map::FlussMap;
+        use arrow::array::MapArray;


Let's follow current code convention and have use at the top of the file.

Hi @leekeiabstraction, this has been addressed in 558b7ee.

leekeiabstraction · 2026-05-06T06:49:53Z

Let's follow current code convention and have use at the top of the file.

Hi @leekeiabstraction, this has been addressed in 558b7ee.

leekeiabstraction · 2026-05-06T06:51:12Z

        writer.complete()
    }
+
+    fn get_map(&self, pos: usize) -> Result<crate::row::binary_map::FlussMap> {


We can make this less verbose by declaring use at the top of the file

fn get_map(&self, pos: usize) -> Result<FlussMap>

Hi @leekeiabstraction, this has been addressed in 558b7ee.

leekeiabstraction · 2026-05-06T06:51:35Z

We can make this less verbose by declaring use at the top of the file

fn get_array(&self, pos: usize) -> Result<FlussArray>

Hi @leekeiabstraction, this has been addressed in 558b7ee.

leekeiabstraction · 2026-05-06T06:52:12Z

+    fn write_array(&mut self, value: &crate::row::FlussArray);
+
+    fn write_map(&mut self, value: &crate::row::FlussMap);


We can make this less verbose by declaring use at the top of the file.

fn write_array(&mut self, value: &FlussArray); fn write_map(&mut self, value: &FlussMap);

Hi @leekeiabstraction, this has been addressed in 558b7ee.

leekeiabstraction · 2026-05-06T07:07:06Z

+        let mut bad_data2 = vec![];
+        bad_data2.extend_from_slice(&large_size);
+        bad_data2.extend_from_slice(&[0, 0, 0, 0]);
+        assert!(FlussMap::from_bytes(&bad_data2).is_err());


What's the error message in this case?

Hi @leekeiabstraction, this has been addressed in 471016f.

leekeiabstraction · 2026-05-06T07:16:01Z

+        // Get the entries for this specific row
+        let entries = map_arr.value(self.row_id);
+        let struct_arr = entries
+            .as_any()
+            .downcast_ref::<arrow::array::StructArray>()
+            .ok_or_else(|| IllegalArgument {
+                message: "Expected StructArray inside MapArray entries".to_string(),
+            })?;
+
+        if struct_arr.num_columns() != 2 {
+            return Err(IllegalArgument {
+                message: format!(
+                    "Expected 2 columns in Map entries struct, got {}",
+                    struct_arr.num_columns()
+                ),
+            });
+        }
+
+        let keys_arrow = struct_arr.column(0);
+        let values_arrow = struct_arr.column(1);
+        let key_fluss_type = from_arrow_type(keys_arrow.data_type())?;
+        let value_fluss_type = from_arrow_type(values_arrow.data_type())?;
+
+        let len = keys_arrow.len();
+
+        // Convert Arrow keys → FlussArray
+        let mut key_writer = FlussArrayWriter::new(len, &key_fluss_type);
+        write_arrow_values_to_fluss_array(&**keys_arrow, &key_fluss_type, &mut key_writer)?;
+        let key_array = key_writer.complete()?;
+
+        // Convert Arrow values → FlussArray
+        let mut value_writer = FlussArrayWriter::new(len, &value_fluss_type);
+        write_arrow_values_to_fluss_array(&**values_arrow, &value_fluss_type, &mut value_writer)?;
+        let value_array = value_writer.complete()?;
+
+        FlussMap::from_arrays(&key_array, &value_array)


These lines which parse from StructArray to FlussMap seem duplicated, let's apply DRY and refactor.

Hi @leekeiabstraction, this has been addressed in 558b7ee.

leekeiabstraction · 2026-05-06T07:16:12Z

+                    let entries = map_arr.value(i);
+                    let struct_arr = entries
+                        .as_any()
+                        .downcast_ref::<arrow::array::StructArray>()
+                        .ok_or_else(|| IllegalArgument {
+                            message: "Expected StructArray inside MapArray entries".to_string(),
+                        })?;
+                    let keys_arrow = struct_arr.column(0);
+                    let values_arrow = struct_arr.column(1);
+                    let key_fluss_type = from_arrow_type(keys_arrow.data_type())?;
+                    let value_fluss_type = from_arrow_type(values_arrow.data_type())?;
+                    let entry_len = keys_arrow.len();
+
+                    let mut key_w = FlussArrayWriter::new(entry_len, &key_fluss_type);
+                    write_arrow_values_to_fluss_array(&**keys_arrow, &key_fluss_type, &mut key_w)?;
+                    let key_array = key_w.complete()?;
+
+                    let mut val_w = FlussArrayWriter::new(entry_len, &value_fluss_type);
+                    write_arrow_values_to_fluss_array(
+                        &**values_arrow,
+                        &value_fluss_type,
+                        &mut val_w,
+                    )?;
+                    let value_array = val_w.complete()?;
+
+                    let fluss_map =
+                        crate::row::binary_map::FlussMap::from_arrays(&key_array, &value_array)?;


These lines which parse from StructArray to FlussMap seem duplicated, let's apply DRY and refactor.

+1, parsing logic seems very similar if not identical, can be wrapped to a helper function

Hi @leekeiabstraction and @charlesdong1991, this has been addressed in 558b7ee.

leekeiabstraction · 2026-05-06T07:20:58Z

+    pub fn from_bytes(data: &[u8]) -> Result<Self> {
+        let (key_array, value_array) = Self::validate(data)?;
+        Ok(FlussMap {
+            data: Bytes::copy_from_slice(data),
+            key_array,
+            value_array,
+        })
+    }
+
+    /// Creates a FlussMap from owned bytes without copying.
+    pub fn from_owned_bytes(data: Bytes) -> Result<Self> {


Do these need to be pub? Can they be pub(crate)?

Hi @leekeiabstraction, this has been addressed in 471016f.

leekeiabstraction · 2026-05-06T07:24:51Z

+    /// Creates a FlussMap by combining a key array and a value array.
+    ///
+    /// Copies both arrays into a new contiguous buffer.
+    pub fn from_arrays(key_array: &FlussArray, value_array: &FlussArray) -> Result<Self> {


IIUC, this is the main method through which we expect end user to use FlussMap. This method signature is rather clunky for end user to use because they need to initialise two FlussArray.

Let's use a more user friendly signature e.g. something that accepts HashMap

Hi @leekeiabstraction, thanks for your reply I think you bring up a valid point.

My understanding is that because FlussMap writes to a strict binary layout, we need the exact DataType schema (like DataTypes::int()) at runtime to initialize the FlussArrayWriters.

If we tried to accept a standard HashMap<K, V>, we wouldn't have that Fluss schema metadata inside the method to know how to properly pack the bytes. Because of that, it seemed like from_arrays had to be the required low-level API since the caller has already explicitly built the arrays with the correct schemas.

IIUC, it would make sense to keep from_arrays for the core API, but maybe add a helper method in test_utils like from_hashmap(map, key_schema, val_schema) to make the test UX nicer? Please let me know what your thoughts are about this.

Hello @qzyu999 ,

I realise I should elaborate more. Suppose a client wants to set a map field on generic row with current pattern, if I understand correctly the pattern that they can use with current shape of API is below:

let mut row = GenericRow::new(2); row.set_field(0, i as i32); let keyWriter = FlussArrayWriter::new(2, &DataType::String()); keyWriter.write_string(0,"hello"); keyWriter.write_string(1,"bonjour"); let valueWriter = FlussArrayWriter::new(2, &DataType::Int()); valueWriter.write_int(0,1); valueWriter.write_int(1,2); row.set_field(1, FlussMap::from_arrays(keyWriter.complete()?, valueWriter.complete()?));

This is rather verbose and error prone. We can instead we have a FlussMapWriter, that allows clients to use a more Map like write pattern e.g. mapWriter.write_entry(A, B) or mapWriter.write_map(map). Type checking, conversion to FlussArray etc. can happen within FlussMapWriter.

fresh-borzoni

@qzyu999 Ty for the PR, GH is acting weird, so I'll paste my review like this:

crates/fluss/src/row/column_writer.rs:563(append_null)
When Datum::Null is appended to a Map column, the path falls through with_builder! and crashes at runtime.
Probably we need an arm for Map as well

crates/fluss/src/row/binary_map.rs:172 and probably in datatype.rs
Shall we check that key is not null?

charlesdong1991

Thanks for the nice PR 👍 left some comments!

charlesdong1991 · 2026-05-06T19:36:08Z

    }
+
+    #[test]
+    fn test_write_map_type() {


can you add a test where a NULL Map value is written?

Hi @charlesdong1991, this has been addressed in 21e8ca9.

charlesdong1991 · 2026-05-06T19:40:28Z

+                    let entries = map_arr.value(i);
+                    let struct_arr = entries
+                        .as_any()
+                        .downcast_ref::<arrow::array::StructArray>()
+                        .ok_or_else(|| IllegalArgument {
+                            message: "Expected StructArray inside MapArray entries".to_string(),
+                        })?;
+                    let keys_arrow = struct_arr.column(0);
+                    let values_arrow = struct_arr.column(1);
+                    let key_fluss_type = from_arrow_type(keys_arrow.data_type())?;
+                    let value_fluss_type = from_arrow_type(values_arrow.data_type())?;
+                    let entry_len = keys_arrow.len();
+
+                    let mut key_w = FlussArrayWriter::new(entry_len, &key_fluss_type);
+                    write_arrow_values_to_fluss_array(&**keys_arrow, &key_fluss_type, &mut key_w)?;
+                    let key_array = key_w.complete()?;
+
+                    let mut val_w = FlussArrayWriter::new(entry_len, &value_fluss_type);
+                    write_arrow_values_to_fluss_array(
+                        &**values_arrow,
+                        &value_fluss_type,
+                        &mut val_w,
+                    )?;
+                    let value_array = val_w.complete()?;
+
+                    let fluss_map =
+                        crate::row::binary_map::FlussMap::from_arrays(&key_array, &value_array)?;


+1, parsing logic seems very similar if not identical, can be wrapped to a helper function

charlesdong1991 · 2026-05-06T19:43:22Z

+    builder: &mut dyn ArrayBuilder,
+    data_type: &arrow_schema::DataType,
+) -> Result<()> {
+    use crate::record::from_arrow_type;


Does Arrow type support Map? can we add a test to see if it does or not?

Hi @charlesdong1991, this has been addressed in fa3adfa.

charlesdong1991 · 2026-05-06T19:44:59Z

    }

+    fn write_map(&mut self, _value: &crate::row::FlussMap) {
+        panic!("Iceberg key columns do not support map values");


With a second thought now, i feel it might be better to return an error, instead of panicing, the same applies to array.

Hi @charlesdong1991, I addressed this in 94fccea where I return an unreachable!() instead, which I believe should work since the function returns void. To return an error my understanding is that it would require rewriting the BinaryWriter trait which would lead to a much larger refactor of all other code that touches it.

charlesdong1991 · 2026-05-06T19:46:27Z

+
+        assert!(FlussMap::from_arrays(&key_array, &value_array).is_err());
+    }
+}


i wonder if we can add a test for nested Map?

Hi @charlesdong1991, this has been addressed in 471016f.

…p parsing - Move inline `use` statements for `FlussArray`, `FlussMap`, and Arrow types to the top of `column.rs` and `binary_writer.rs`. - Simplify method signatures for `get_map`, `get_array`, `write_map`, and `write_array` to avoid verbose module paths. - Extract duplicated logic for converting Arrow `StructArray` into `FlussMap` into a new `arrow_map_entry_to_fluss_map` helper in `column.rs`. - Optimize the new helper to accept `&StructArray` directly, removing redundant downcasts.

…files

- Changed the visibility of from_bytes and from_owned_bytes to pub(crate). - Added a validation loop inside from_arrays that checks for null keys, safely returning an IllegalArgument error. - Refined test_invalid_data and test_mismatched_array_sizes to strictly assert their exact error messages. - Hardcoded the generated map byte sequences in test_round_trip_int_to_string_map and test_map_with_null_values to lock them down from regression. - Added a comprehensive test_nested_map case that validates a MAP<STRING, MAP<INT, STRING>> structure.

qzyu999 · 2026-05-07T03:56:56Z

crates/fluss/src/row/binary_map.rs:172 ... Shall we check that key is not null?

Hi @fresh-borzoni, this has been addressed in 471016f.

- Updated MapType::with_nullable to force the key_type to be non-nullable via Box::new(key_type.as_non_nullable()). - Updated the Display traits in datatype.rs tests and the Arrow conversion tests in arrow.rs to correctly assert the updated schema behavior (NOT NULL map keys).

- Added the missing TypedWriter::Map arm inside append_null() to properly write an empty map entry (updating offsets and pushing a false validity) whenever Datum::Null is provided for a map type. - Implemented test_write_null_map_type to verify that Datum::Null successfully converts into a null Map field correctly.

qzyu999 · 2026-05-07T04:22:06Z

crates/fluss/src/row/column_writer.rs:563(append_null) When Datum::Null is appended to a Map column, the path falls through with_builder! and crashes at runtime. Probably we need an arm for Map as well

Hi @fresh-borzoni, this has been addressed in 21e8ca9.

qzyu999 · 2026-05-07T19:21:45Z

... probably in datatype.rs

Hi @fresh-borzoni, this has been addressed in 4c06cd9.

qzyu999 · 2026-05-07T19:22:41Z

Hi @leekeiabstraction, @fresh-borzoni, and @charlesdong1991, thank you all for the review suggestions. I've replied to all of your comments and have made corresponding changes, PTAL.

fresh-borzoni

@qzyu999 Ty for the changes, looked through, looks good, left comments, PTAL

Also can we add simple e2e integration test to see if everything is wired correctly?
Scenario:

Create a table with one Map column
Insert one row
Fetch it back
Assert the map round-trips (size + at least one key/value)

fresh-borzoni · 2026-05-08T20:52:52Z

-    fn write_array(&mut self, _value: &[u8]) {
-        panic!("Iceberg key columns do not support array values");
+    fn write_array(&mut self, _value: &FlussArray) {
+        unreachable!("Array/Map types are rejected during value writer creation");


It's the same panic under the hood

Hi @fresh-borzoni, I responded to @charlesdong1991 regarding this with the following:

Hi @charlesdong1991, I addressed this in 94fccea where I return an unreachable!() instead, which I believe should work since the function returns void. To return an error my understanding is that it would require rewriting the BinaryWriter trait which would lead to a much larger refactor of all other code that touches it.

fresh-borzoni · 2026-05-08T20:55:55Z

+impl FlussMap {
+    /// Validates the raw bytes and extracts the sub-arrays.
+    fn validate(data: &[u8]) -> Result<(FlussArray, FlussArray)> {
+        if data.len() < 4 {


validate accepts trailing bytes - FlussArray::from_bytes only checks the header, not that the slice is fully consumed

Could we assert 4 + key_extent + value_extent == data.len() at the end of validate?

fresh-borzoni · 2026-05-08T20:58:17Z

+            });
+        }
+
+        for i in 0..key_array.size() {


The null-key check is in from_arrays but not from_bytes. With non-nullable keys, column_writerskips the null check and writes garbage to Arrow.

Could the check move into validate so all paths share it?

fresh-borzoni · 2026-05-08T21:00:26Z

                Field::new("key", key_type, map_type.key_type().is_nullable()),
                Field::new("value", value_type, map_type.value_type().is_nullable()),
            ];
            ArrowDataType::Map(


Arrow spec requires the entries field of Map to be non-nullable
Should be Field::new("entries", ..., false).

fresh-borzoni · 2026-05-08T21:09:38Z

+        }
+    };
+
+    let key_fluss_type = from_arrow_type(&key_arrow_type)?;


Key/value Fluss types come from the Arrow type, not the FlussMap. What if is they mismatch?

qzyu999 added 2 commits May 4, 2026 21:56

feat: implement FlussMap support for compacted rows and binary encoding

f31ccb0

style: run formatting

2ae968b

leekeiabstraction reviewed May 5, 2026

View reviewed changes

refactor: use strongly-typed FlussArray and FlussMap in BinaryWriter

9f995f8

leekeiabstraction reviewed May 6, 2026

View reviewed changes

fresh-borzoni reviewed May 6, 2026

View reviewed changes

charlesdong1991 reviewed May 6, 2026

View reviewed changes

qzyu999 added 3 commits May 6, 2026 14:50

refactor: reduce signature verbosity etc. by declaring use at top of …

94fccea

…files

qzyu999 added 4 commits May 6, 2026 21:14

style: run formatting

6254414

style: run formatting

7a4cc8d

test: Add test Map for Arrow

fa3adfa

fresh-borzoni reviewed May 8, 2026

View reviewed changes

		fn write_array(&mut self, value: &[u8]);

		fn write_map(&mut self, value: &[u8]);

		fn write_array(&mut self, value: &crate::row::FlussArray);

		fn write_map(&mut self, value: &crate::row::FlussMap);

Conversation

qzyu999 commented May 5, 2026

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

leekeiabstraction commented May 5, 2026

Uh oh!

leekeiabstraction left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leekeiabstraction left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fresh-borzoni left a comment

Choose a reason for hiding this comment

Uh oh!

charlesdong1991 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fresh-borzoni left a comment •

edited

Loading