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/07/15 12:41:58 UTC
(fluss) branch main updated: [elixir] Support complex types (Array/Map/Row) and nullability (#3636)
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.git
The following commit(s) were added to refs/heads/main by this push:
new 8b078d818 [elixir] Support complex types (Array/Map/Row) and nullability (#3636)
8b078d818 is described below
commit 8b078d818a825c87ca73ef8b9695d139a95307b0
Author: Nicoleta Lazar <ni...@fresha.com>
AuthorDate: Wed Jul 15 15:41:53 2026 +0300
[elixir] Support complex types (Array/Map/Row) and nullability (#3636)
* [elixir] Support complex types (Array/Map/Row) and nullability
Add nested types and per-node nullability to the Elixir bindings, in both
schema definitions and row values.
- DataType gains recursive Array/Map/Row variants with recursive converters.
- Nullability uses an additive {:not_null, dt} wrapper (types nullable by
default, composable at any depth). The read path now preserves the flag,
so primary-key columns read back as {:not_null, ...}.
- Row value shapes: array -> list; map -> %{k => v}; row -> map keyed by field-name atom.
- Adds unit and integration tests; updates existing PK read assertions.
* [elixir] Add tests for deeply-nested complex types
---
fluss-rust/bindings/elixir/lib/fluss/schema.ex | 10 +
.../elixir/native/fluss_nif/src/row_convert.rs | 272 +++++++++++++++++----
.../bindings/elixir/native/fluss_nif/src/schema.rs | 69 ++++--
.../elixir/test/integration/admin_test.exs | 42 +++-
.../elixir/test/integration/log_table_test.exs | 136 +++++++++++
.../bindings/elixir/test/table_descriptor_test.exs | 12 +
6 files changed, 474 insertions(+), 67 deletions(-)
diff --git a/fluss-rust/bindings/elixir/lib/fluss/schema.ex b/fluss-rust/bindings/elixir/lib/fluss/schema.ex
index e11911eee..0b05af75d 100644
--- a/fluss-rust/bindings/elixir/lib/fluss/schema.ex
+++ b/fluss-rust/bindings/elixir/lib/fluss/schema.ex
@@ -24,6 +24,12 @@ defmodule Fluss.Schema do
Parameterized types: `{:decimal, precision, scale}`, `{:char, length}`, `{:binary, length}`
+ Complex types: `{:array, dt}`, `{:map, key_dt, value_dt}`, `{:row, [{name, dt}]}`
+ (e.g. `{:row, [{"x", :int}, {"y", :string}]}`).
+
+ Nullability: types are nullable by default; wrap as `{:not_null, dt}` to mark
+ non-null at any depth — e.g. `{:array, {:not_null, :int}}`.
+
## Examples
schema =
@@ -53,6 +59,10 @@ defmodule Fluss.Schema do
| {:decimal, non_neg_integer(), non_neg_integer()}
| {:char, non_neg_integer()}
| {:binary, non_neg_integer()}
+ | {:array, data_type()}
+ | {:map, data_type(), data_type()}
+ | {:row, [{String.t(), data_type()}]}
+ | {:not_null, data_type()}
@type t :: %__MODULE__{
columns: [{String.t(), data_type()}],
diff --git a/fluss-rust/bindings/elixir/native/fluss_nif/src/row_convert.rs b/fluss-rust/bindings/elixir/native/fluss_nif/src/row_convert.rs
index c72395e90..70dd7bfa0 100644
--- a/fluss-rust/bindings/elixir/native/fluss_nif/src/row_convert.rs
+++ b/fluss-rust/bindings/elixir/native/fluss_nif/src/row_convert.rs
@@ -17,10 +17,13 @@
use std::str::FromStr;
-use fluss::metadata::{Column, DataType};
-use fluss::row::{Date, Decimal, GenericRow, InternalRow, Time, TimestampLtz, TimestampNtz};
+use fluss::metadata::{Column, DataType, MapType, RowType};
+use fluss::row::{
+ DataGetters, Date, Datum, Decimal, FlussArray, FlussArrayWriter, FlussMap, FlussMapWriter,
+ GenericRow, InternalRow, Time, TimestampLtz, TimestampNtz,
+};
use rustler::types::binary::NewBinary;
-use rustler::{Encoder, Env, Term};
+use rustler::{Encoder, Env, MapIterator, Term};
use crate::atoms;
@@ -57,62 +60,62 @@ pub fn row_to_term<'a>(
fn field_to_term<'a>(
env: Env<'a>,
- row: &dyn InternalRow,
+ getters: &dyn DataGetters,
pos: usize,
data_type: &DataType,
) -> Result<Term<'a>, String> {
- if row.is_null_at(pos).map_err(|e| e.to_string())? {
+ if getters.is_null_at(pos).map_err(|e| e.to_string())? {
return Ok(atoms::nil().encode(env));
}
match data_type {
DataType::Boolean(_) => {
- let v = row.get_boolean(pos).map_err(|e| e.to_string())?;
+ let v = getters.get_boolean(pos).map_err(|e| e.to_string())?;
Ok(v.encode(env))
}
DataType::TinyInt(_) => {
- let v = row.get_byte(pos).map_err(|e| e.to_string())?;
+ let v = getters.get_byte(pos).map_err(|e| e.to_string())?;
Ok(v.encode(env))
}
DataType::SmallInt(_) => {
- let v = row.get_short(pos).map_err(|e| e.to_string())?;
+ let v = getters.get_short(pos).map_err(|e| e.to_string())?;
Ok(v.encode(env))
}
DataType::Int(_) => {
- let v = row.get_int(pos).map_err(|e| e.to_string())?;
+ let v = getters.get_int(pos).map_err(|e| e.to_string())?;
Ok(v.encode(env))
}
DataType::BigInt(_) => {
- let v = row.get_long(pos).map_err(|e| e.to_string())?;
+ let v = getters.get_long(pos).map_err(|e| e.to_string())?;
Ok(v.encode(env))
}
DataType::Float(_) => {
- let v = row.get_float(pos).map_err(|e| e.to_string())?;
+ let v = getters.get_float(pos).map_err(|e| e.to_string())?;
Ok(v.encode(env))
}
DataType::Double(_) => {
- let v = row.get_double(pos).map_err(|e| e.to_string())?;
+ let v = getters.get_double(pos).map_err(|e| e.to_string())?;
Ok(v.encode(env))
}
DataType::String(_) => {
- let v = row.get_string(pos).map_err(|e| e.to_string())?;
+ let v = getters.get_string(pos).map_err(|e| e.to_string())?;
Ok(v.encode(env))
}
DataType::Char(ct) => {
- let v = row
+ let v = getters
.get_char(pos, ct.length() as usize)
.map_err(|e| e.to_string())?;
Ok(v.encode(env))
}
DataType::Bytes(_) => {
- let v = row.get_bytes(pos).map_err(|e| e.to_string())?;
+ let v = getters.get_bytes(pos).map_err(|e| e.to_string())?;
let mut bin = NewBinary::new(env, v.len());
bin.as_mut_slice().copy_from_slice(v);
let binary: rustler::Binary = bin.into();
Ok(binary.encode(env))
}
DataType::Binary(bt) => {
- let v = row
+ let v = getters
.get_binary(pos, bt.length())
.map_err(|e| e.to_string())?;
let mut bin = NewBinary::new(env, v.len());
@@ -121,35 +124,131 @@ fn field_to_term<'a>(
Ok(binary.encode(env))
}
DataType::Date(_) => {
- let v = row.get_date(pos).map_err(|e| e.to_string())?;
+ let v = getters.get_date(pos).map_err(|e| e.to_string())?;
Ok(v.get_inner().encode(env))
}
DataType::Time(_) => {
- let v = row.get_time(pos).map_err(|e| e.to_string())?;
+ let v = getters.get_time(pos).map_err(|e| e.to_string())?;
Ok(v.get_inner().encode(env))
}
DataType::Timestamp(ts) => {
- let v = row
+ let v = getters
.get_timestamp_ntz(pos, ts.precision())
.map_err(|e| e.to_string())?;
Ok((v.get_millisecond(), v.get_nano_of_millisecond()).encode(env))
}
DataType::TimestampLTz(ts) => {
- let v = row
+ let v = getters
.get_timestamp_ltz(pos, ts.precision())
.map_err(|e| e.to_string())?;
Ok((v.get_epoch_millisecond(), v.get_nano_of_millisecond()).encode(env))
}
DataType::Decimal(dt) => {
- let v = row
+ let v = getters
.get_decimal(pos, dt.precision() as usize, dt.scale() as usize)
.map_err(|e| e.to_string())?;
Ok(v.to_string().encode(env))
}
- _ => Err(format!("unsupported data type: {data_type:?}")),
+ DataType::Array(dt) => {
+ let arr = getters
+ .get_array(pos)
+ .map_err(|e| e.to_string())?
+ .try_into_binary()
+ .map_err(|e| e.to_string())?;
+ array_to_term(env, &arr, dt.get_element_type())
+ }
+ DataType::Map(dt) => {
+ let map = getters
+ .get_map(pos)
+ .map_err(|e| e.to_string())?
+ .try_into_binary()
+ .map_err(|e| e.to_string())?; // → FlussMap
+ map_to_term(env, &map, dt)
+ }
+ DataType::Row(dt) => {
+ let nested = getters
+ .get_row(pos)
+ .map_err(|e| e.to_string())?
+ .try_into_generic(dt)
+ .map_err(|e| e.to_string())?; // → GenericRow
+ row_fields_to_term(env, &nested, dt)
+ }
}
}
+pub fn array_to_term<'a>(
+ env: Env<'a>,
+ arr: &FlussArray,
+ element_type: &DataType,
+) -> Result<Term<'a>, String> {
+ let items = (0..arr.size())
+ .map(|i| array_elem_to_term(env, arr, i, element_type))
+ .collect::<Result<Vec<Term<'a>>, String>>()?;
+ Ok(items.encode(env))
+}
+
+fn array_elem_to_term<'a>(
+ env: Env<'a>,
+ arr: &FlussArray,
+ i: usize,
+ element_type: &DataType,
+) -> Result<Term<'a>, String> {
+ if arr.is_null_at(i) {
+ return Ok(atoms::nil().encode(env));
+ }
+ match element_type {
+ DataType::Array(at) => {
+ let nested = arr.get_array(i).map_err(|e| e.to_string())?;
+ array_to_term(env, &nested, at.get_element_type())
+ }
+ DataType::Map(mt) => {
+ let nested = arr
+ .get_map(i, mt.key_type(), mt.value_type())
+ .map_err(|e| e.to_string())?;
+ map_to_term(env, &nested, mt)
+ }
+ DataType::Row(rt) => {
+ let nested = arr.get_row(i, rt).map_err(|e| e.to_string())?;
+ row_fields_to_term(env, &nested, rt)
+ }
+ _ => field_to_term(env, arr, i, element_type),
+ }
+}
+
+pub fn map_to_term<'a>(env: Env<'a>, map: &FlussMap, mt: &MapType) -> Result<Term<'a>, String> {
+ let keys = map.key_array();
+ let values = map.value_array();
+
+ let pairs = (0..map.size())
+ .map(|i| {
+ let k = array_elem_to_term(env, keys, i, mt.key_type())?;
+ let v = array_elem_to_term(env, values, i, mt.value_type())?;
+ Ok((k, v))
+ })
+ .collect::<Result<Vec<(Term<'a>, Term<'a>)>, String>>()?;
+
+ Term::map_from_pairs(env, &pairs).map_err(|_| "failed to build map".to_string())
+}
+
+pub fn row_fields_to_term<'a>(
+ env: Env<'a>,
+ row: &dyn DataGetters,
+ rt: &RowType,
+) -> Result<Term<'a>, String> {
+ let pairs: Vec<(Term<'a>, Term<'a>)> = rt
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(i, field)| {
+ let key = rustler::Atom::from_str(env, field.name())
+ .map_err(|_| "invalid field name".to_string())?
+ .encode(env);
+ let value = field_to_term(env, row, i, field.data_type())?;
+ Ok((key, value))
+ })
+ .collect::<Result<_, String>>()?;
+ Term::map_from_pairs(env, &pairs).map_err(|_| "failed to build row map".to_string())
+}
pub fn term_to_row<'a>(
env: Env<'a>,
values: Term<'a>,
@@ -174,80 +273,87 @@ pub fn term_to_row<'a>(
{
continue; // leave as null
}
- set_field_from_term(env, &mut row, i, *term, col.data_type())?;
+ row.set_field(i, term_to_datum(env, *term, col.data_type())?);
}
Ok(row)
}
-fn set_field_from_term<'a>(
- _env: Env<'a>,
- row: &mut GenericRow<'static>,
- pos: usize,
+fn term_to_datum<'a>(
+ env: Env<'a>,
term: Term<'a>,
data_type: &DataType,
-) -> Result<(), String> {
+) -> Result<Datum<'static>, String> {
+ // A nil term maps to Null regardless of the declared type. This is what lets
+ // the Array/Map/Row arms recurse without special-casing null elements.
+ if term.is_atom()
+ && let Ok(atom) = term.decode::<rustler::Atom>()
+ && atom == atoms::nil()
+ {
+ return Ok(Datum::Null);
+ }
+
match data_type {
DataType::Boolean(_) => {
let v: bool = term.decode().map_err(|_| "expected boolean")?;
- row.set_field(pos, v);
+ Ok(v.into())
}
DataType::TinyInt(_) => {
let v: i8 = term
.decode()
.map_err(|_| "expected integer in range -128..127 for tinyint")?;
- row.set_field(pos, v);
+ Ok(v.into())
}
DataType::SmallInt(_) => {
let v: i16 = term
.decode()
.map_err(|_| "expected integer in range -32768..32767 for smallint")?;
- row.set_field(pos, v);
+ Ok(v.into())
}
DataType::Int(_) => {
let v: i32 = term.decode().map_err(|_| "expected integer")?;
- row.set_field(pos, v);
+ Ok(v.into())
}
DataType::BigInt(_) => {
let v: i64 = term.decode().map_err(|_| "expected integer")?;
- row.set_field(pos, v);
+ Ok(v.into())
}
DataType::Date(_) => {
let v: i32 = term
.decode()
.map_err(|_| "expected integer (days since epoch)")?;
- row.set_field(pos, Date::new(v));
+ Ok(Date::new(v).into())
}
DataType::Time(_) => {
let v: i32 = term
.decode()
.map_err(|_| "expected integer (millis since midnight)")?;
- row.set_field(pos, Time::new(v));
+ Ok(Time::new(v).into())
}
DataType::Timestamp(_) => {
let (millis, nanos): (i64, i32) = term
.decode()
.map_err(|_| "expected {millis, nanos} tuple for timestamp")?;
let ts = TimestampNtz::from_millis_nanos(millis, nanos).map_err(|e| e.to_string())?;
- row.set_field(pos, ts);
+ Ok(ts.into())
}
DataType::TimestampLTz(_) => {
let (millis, nanos): (i64, i32) = term
.decode()
.map_err(|_| "expected {millis, nanos} tuple for timestamp_ltz")?;
let ts = TimestampLtz::from_millis_nanos(millis, nanos).map_err(|e| e.to_string())?;
- row.set_field(pos, ts);
+ Ok(ts.into())
}
DataType::Float(_) => {
let v: f64 = term.decode().map_err(|_| "expected number for float")?;
- row.set_field(pos, v as f32);
+ Ok((v as f32).into())
}
DataType::Double(_) => {
let v: f64 = term.decode().map_err(|_| "expected number for double")?;
- row.set_field(pos, v);
+ Ok(v.into())
}
DataType::String(_) | DataType::Char(_) => {
let v: String = term.decode().map_err(|_| "expected string")?;
- row.set_field(pos, v);
+ Ok(v.into())
}
DataType::Decimal(dt) => {
let v: String = term.decode().map_err(|_| "expected string for decimal")?;
@@ -255,13 +361,93 @@ fn set_field_from_term<'a>(
.map_err(|e| format!("failed to parse decimal '{v}': {e}"))?;
let decimal = Decimal::from_big_decimal(bd, dt.precision(), dt.scale())
.map_err(|e| e.to_string())?;
- row.set_field(pos, decimal);
+ Ok(decimal.into())
}
DataType::Bytes(_) | DataType::Binary(_) => {
let bin: rustler::Binary = term.decode().map_err(|_| "expected binary")?;
- row.set_field(pos, bin.as_slice().to_vec());
+ Ok(bin.as_slice().to_vec().into())
+ }
+ DataType::Array(at) => {
+ let et = at.get_element_type();
+ let list: Vec<Term<'a>> = term
+ .decode()
+ .map_err(|_| "expected a list for array".to_string())?;
+ let mut writer = FlussArrayWriter::new(list.len(), et);
+ for (i, elem) in list.iter().enumerate() {
+ let datum = term_to_datum(env, *elem, et)?;
+ write_datum_into_array(&mut writer, i, datum, et)?;
+ }
+ Ok(Datum::Array(writer.complete().map_err(|e| e.to_string())?))
+ }
+ DataType::Map(mt) => {
+ let kt = mt.key_type();
+ let vt = mt.value_type();
+ let size = term
+ .map_size()
+ .map_err(|_| "expected a map for map column".to_string())?;
+ let iter = MapIterator::new(term)
+ .ok_or_else(|| "expected a map for map column".to_string())?;
+ let mut writer = FlussMapWriter::new(size, kt, vt);
+ for (k_term, v_term) in iter {
+ let key = term_to_datum(env, k_term, kt)?;
+ let value = term_to_datum(env, v_term, vt)?;
+ writer.write_entry(key, value).map_err(|e| e.to_string())?;
+ }
+ Ok(Datum::Map(writer.complete().map_err(|e| e.to_string())?))
+ }
+ DataType::Row(rt) => {
+ let fields = rt.fields();
+ let mut nested = GenericRow::new(fields.len());
+ for (i, field) in fields.iter().enumerate() {
+ let key = rustler::Atom::from_str(env, field.name())
+ .map_err(|_| "invalid field name".to_string())?;
+ let value_term = term
+ .map_get(key)
+ .map_err(|_| format!("row is missing field '{}'", field.name()))?;
+ nested.set_field(i, term_to_datum(env, value_term, field.data_type())?);
+ }
+ Ok(Datum::Row(Box::new(nested)))
+ }
+ }
+}
+
+pub fn write_datum_into_array(
+ writer: &mut FlussArrayWriter,
+ i: usize,
+ datum: Datum<'static>,
+ element_type: &DataType,
+) -> Result<(), String> {
+ match datum {
+ Datum::Null => writer.set_null_at(i),
+ Datum::Bool(v) => writer.write_boolean(i, v),
+ Datum::Int8(v) => writer.write_byte(i, v),
+ Datum::Int16(v) => writer.write_short(i, v),
+ Datum::Int32(v) => writer.write_int(i, v),
+ Datum::Int64(v) => writer.write_long(i, v),
+ Datum::Float32(v) => writer.write_float(i, v.into_inner()),
+ Datum::Float64(v) => writer.write_double(i, v.into_inner()),
+ Datum::String(v) => writer.write_string(i, &v),
+ Datum::Blob(v) => writer.write_binary_bytes(i, v.as_ref()),
+ Datum::Decimal(v) => {
+ if let DataType::Decimal(dt) = element_type {
+ writer.write_decimal(i, &v, dt.precision());
+ }
+ }
+ Datum::Date(v) => writer.write_date(i, v),
+ Datum::Time(v) => writer.write_time(i, v),
+ Datum::TimestampNtz(v) => {
+ if let DataType::Timestamp(dt) = element_type {
+ writer.write_timestamp_ntz(i, &v, dt.precision());
+ }
+ }
+ Datum::TimestampLtz(v) => {
+ if let DataType::TimestampLTz(dt) = element_type {
+ writer.write_timestamp_ltz(i, &v, dt.precision());
+ }
}
- _ => return Err(format!("unsupported data type for writing: {data_type:?}")),
+ Datum::Array(v) => writer.write_array(i, &v),
+ Datum::Map(v) => writer.write_map(i, &v),
+ Datum::Row(v) => writer.write_row(i, v.as_ref()).map_err(|e| e.to_string())?,
}
Ok(())
}
diff --git a/fluss-rust/bindings/elixir/native/fluss_nif/src/schema.rs b/fluss-rust/bindings/elixir/native/fluss_nif/src/schema.rs
index 95b31bf20..375fee881 100644
--- a/fluss-rust/bindings/elixir/native/fluss_nif/src/schema.rs
+++ b/fluss-rust/bindings/elixir/native/fluss_nif/src/schema.rs
@@ -52,6 +52,10 @@ pub enum DataType {
Decimal(u32, u32),
Char(u32),
Binary(usize),
+ Array(Box<DataType>),
+ Map(Box<DataType>, Box<DataType>),
+ Row(Vec<(String, DataType)>),
+ NotNull(Box<DataType>),
}
fn to_fluss_type(dt: &DataType) -> metadata::DataType {
@@ -72,32 +76,55 @@ fn to_fluss_type(dt: &DataType) -> metadata::DataType {
DataType::Decimal(precision, scale) => DataTypes::decimal(*precision, *scale),
DataType::Char(length) => DataTypes::char(*length),
DataType::Binary(length) => DataTypes::binary(*length),
+ DataType::Array(item) => DataTypes::array(to_fluss_type(item)),
+ DataType::Map(k, v) => DataTypes::map(to_fluss_type(k), to_fluss_type(v)),
+ DataType::Row(fields) => DataTypes::row(
+ fields
+ .iter()
+ .map(|(name, dt)| DataTypes::field(name, to_fluss_type(dt)))
+ .collect(),
+ ),
+ DataType::NotNull(dt) => to_fluss_type(dt).as_non_nullable(),
}
}
fn from_fluss_type(dt: &metadata::DataType) -> Result<DataType, Error> {
- match dt {
- metadata::DataType::Boolean(_) => Ok(DataType::Boolean),
- metadata::DataType::TinyInt(_) => Ok(DataType::Tinyint),
- metadata::DataType::SmallInt(_) => Ok(DataType::Smallint),
- metadata::DataType::Int(_) => Ok(DataType::Int),
- metadata::DataType::BigInt(_) => Ok(DataType::Bigint),
- metadata::DataType::Float(_) => Ok(DataType::Float),
- metadata::DataType::Double(_) => Ok(DataType::Double),
- metadata::DataType::String(_) => Ok(DataType::String),
- metadata::DataType::Bytes(_) => Ok(DataType::Bytes),
- metadata::DataType::Date(_) => Ok(DataType::Date),
- metadata::DataType::Time(_) => Ok(DataType::Time),
- metadata::DataType::Timestamp(_) => Ok(DataType::Timestamp),
- metadata::DataType::TimestampLTz(_) => Ok(DataType::TimestampLtz),
- metadata::DataType::Decimal(d) => Ok(DataType::Decimal(d.precision(), d.scale())),
- metadata::DataType::Char(c) => Ok(DataType::Char(c.length())),
- metadata::DataType::Binary(b) => Ok(DataType::Binary(b.length())),
- metadata::DataType::Array(_) | metadata::DataType::Map(_) | metadata::DataType::Row(_) => {
- Err(Error::UnsupportedOperation {
- message: format!("data type {dt:?} is not supported by the Elixir bindings"),
- })
+ let base = match dt {
+ metadata::DataType::Boolean(_) => DataType::Boolean,
+ metadata::DataType::TinyInt(_) => DataType::Tinyint,
+ metadata::DataType::SmallInt(_) => DataType::Smallint,
+ metadata::DataType::Int(_) => DataType::Int,
+ metadata::DataType::BigInt(_) => DataType::Bigint,
+ metadata::DataType::Float(_) => DataType::Float,
+ metadata::DataType::Double(_) => DataType::Double,
+ metadata::DataType::String(_) => DataType::String,
+ metadata::DataType::Bytes(_) => DataType::Bytes,
+ metadata::DataType::Date(_) => DataType::Date,
+ metadata::DataType::Time(_) => DataType::Time,
+ metadata::DataType::Timestamp(_) => DataType::Timestamp,
+ metadata::DataType::TimestampLTz(_) => DataType::TimestampLtz,
+ metadata::DataType::Decimal(d) => DataType::Decimal(d.precision(), d.scale()),
+ metadata::DataType::Char(c) => DataType::Char(c.length()),
+ metadata::DataType::Binary(b) => DataType::Binary(b.length()),
+ metadata::DataType::Array(a) => {
+ DataType::Array(Box::new(from_fluss_type(a.get_element_type())?))
}
+ metadata::DataType::Map(m) => DataType::Map(
+ Box::new(from_fluss_type(m.key_type())?),
+ Box::new(from_fluss_type(m.value_type())?),
+ ),
+ metadata::DataType::Row(r) => DataType::Row(
+ r.fields()
+ .iter()
+ .map(|f| Ok((f.name().to_string(), from_fluss_type(f.data_type())?)))
+ .collect::<Result<Vec<_>, Error>>()?,
+ ),
+ };
+
+ if dt.is_nullable() {
+ Ok(base)
+ } else {
+ Ok(DataType::NotNull(Box::new(base)))
}
}
diff --git a/fluss-rust/bindings/elixir/test/integration/admin_test.exs b/fluss-rust/bindings/elixir/test/integration/admin_test.exs
index 1de7a1e74..c25ce1695 100644
--- a/fluss-rust/bindings/elixir/test/integration/admin_test.exs
+++ b/fluss-rust/bindings/elixir/test/integration/admin_test.exs
@@ -262,7 +262,7 @@ defmodule Fluss.Integration.AdminTest do
# Schema round-trips: columns in definition order, PK preserved.
assert info.schema == %Fluss.Schema{
- columns: [{"id", :int}, {"name", :string}],
+ columns: [{"id", {:not_null, :int}}, {"name", :string}],
primary_key: ["id"]
}
@@ -307,7 +307,11 @@ defmodule Fluss.Integration.AdminTest do
assert info.physical_primary_keys == ["id", "region"]
assert info.schema == %Fluss.Schema{
- columns: [{"id", :int}, {"region", :string}, {"name", :string}],
+ columns: [
+ {"id", {:not_null, :int}},
+ {"region", {:not_null, :string}},
+ {"name", :string}
+ ],
primary_key: ["id", "region"]
}
end
@@ -363,13 +367,45 @@ defmodule Fluss.Integration.AdminTest do
Fluss.Admin.get_table_schema(admin, @database, table)
assert info.schema == %Fluss.Schema{
- columns: [{"id", :int}, {"name", :string}],
+ columns: [{"id", {:not_null, :int}}, {"name", :string}],
primary_key: ["id"]
}
assert is_integer(info.schema_id) and info.schema_id >= 1
end
+ test "round-trips complex types and nullability", %{admin: admin} do
+ table = "fluss_table_#{:rand.uniform(100_000)}"
+
+ schema =
+ Fluss.Schema.new()
+ |> Fluss.Schema.column("id", :int)
+ |> Fluss.Schema.column("tags", {:array, :string})
+ |> Fluss.Schema.column("scores", {:array, {:not_null, :int}})
+ |> Fluss.Schema.column("attrs", {:map, :string, :int})
+ |> Fluss.Schema.column("point", {:row, [{"x", :int}, {"y", :int}]})
+ |> Fluss.Schema.primary_key(["id"])
+
+ descriptor = Fluss.TableDescriptor.new!(schema)
+ :ok = Fluss.Admin.create_table(admin, @database, table, descriptor, true)
+ on_exit(fn -> Fluss.Admin.drop_table(admin, @database, table, true) end)
+
+ assert {:ok, info} = Fluss.Admin.get_table_schema(admin, @database, table)
+
+ assert info.schema == %Fluss.Schema{
+ columns: [
+ {"id", {:not_null, :int}},
+ {"tags", {:array, :string}},
+ {"scores", {:array, {:not_null, :int}}},
+ # map keys are non-null by Fluss spec, so the `:string`
+ # key reads back wrapped even though it was written bare.
+ {"attrs", {:map, {:not_null, :string}, :int}},
+ {"point", {:row, [{"x", :int}, {"y", :int}]}}
+ ],
+ primary_key: ["id"]
+ }
+ end
+
test "fetches a specific schema version by id", %{admin: admin} do
table = "fluss_table_#{:rand.uniform(100_000)}"
diff --git a/fluss-rust/bindings/elixir/test/integration/log_table_test.exs b/fluss-rust/bindings/elixir/test/integration/log_table_test.exs
index d3eb55e43..d78662eee 100644
--- a/fluss-rust/bindings/elixir/test/integration/log_table_test.exs
+++ b/fluss-rust/bindings/elixir/test/integration/log_table_test.exs
@@ -166,6 +166,142 @@ defmodule Fluss.Integration.LogTableTest do
cleanup_table(admin, table_name)
end
+
+ test "round-trips array, map, and row values", %{conn: conn, admin: admin} do
+ table_name = "ex_test_complex_#{:rand.uniform(100_000)}"
+ cleanup_table(admin, table_name)
+
+ schema =
+ Fluss.Schema.new()
+ |> Fluss.Schema.column("id", :int)
+ |> Fluss.Schema.column("tags", {:array, :string})
+ |> Fluss.Schema.column("attrs", {:map, :string, :int})
+ |> Fluss.Schema.column("point", {:row, [{"x", :int}, {"y", :int}]})
+
+ descriptor = Fluss.TableDescriptor.new!(schema)
+ :ok = Fluss.Admin.create_table(admin, @database, table_name, descriptor, false)
+
+ table = Fluss.Table.get!(conn, @database, table_name)
+ writer = Fluss.AppendWriter.new!(table)
+
+ # A MAP<string,int> takes string keys (data); a ROW takes atom keys
+ # (field names — the write side resolves them via Atom.from_str/2).
+ {:ok, _} =
+ Fluss.AppendWriter.append(writer, [
+ 1,
+ ["a", "b", "c"],
+ %{"x" => 1, "y" => 2},
+ %{x: 10, y: 20}
+ ])
+
+ :ok = Fluss.AppendWriter.flush(writer)
+
+ scanner = Fluss.LogScanner.new!(table)
+ :ok = Fluss.LogScanner.subscribe(scanner, 0, Fluss.earliest_offset())
+
+ records = poll_records(scanner, 1)
+ assert length(records) == 1
+
+ row = Enum.at(records, 0)[:row]
+ assert row[:id] == 1
+ assert row[:tags] == ["a", "b", "c"]
+ assert row[:attrs] == %{"x" => 1, "y" => 2}
+ assert row[:point] == %{x: 10, y: 20}
+
+ cleanup_table(admin, table_name)
+ end
+
+ test "round-trips a deeply-nested array<map<string, row>>", %{conn: conn, admin: admin} do
+ table_name = "ex_test_deep_nested_#{:rand.uniform(100_000)}"
+ cleanup_table(admin, table_name)
+
+ # array<map<string, row<n:int, label:string>>> — three levels of nesting.
+ inner_row = {:row, [{"n", :int}, {"label", :string}]}
+
+ schema =
+ Fluss.Schema.new()
+ |> Fluss.Schema.column("id", :int)
+ |> Fluss.Schema.column("data", {:array, {:map, :string, inner_row}})
+
+ descriptor = Fluss.TableDescriptor.new!(schema)
+ :ok = Fluss.Admin.create_table(admin, @database, table_name, descriptor, false)
+
+ table = Fluss.Table.get!(conn, @database, table_name)
+ writer = Fluss.AppendWriter.new!(table)
+
+ # A list of maps; each map value is a row (atom-keyed by field name).
+ data = [
+ %{"first" => %{n: 1, label: "one"}, "second" => %{n: 2, label: "two"}},
+ %{"third" => %{n: 3, label: "three"}}
+ ]
+
+ {:ok, _} = Fluss.AppendWriter.append(writer, [1, data])
+ :ok = Fluss.AppendWriter.flush(writer)
+
+ scanner = Fluss.LogScanner.new!(table)
+ :ok = Fluss.LogScanner.subscribe(scanner, 0, Fluss.earliest_offset())
+
+ records = poll_records(scanner, 1)
+ assert length(records) == 1
+
+ row = Enum.at(records, 0)[:row]
+ assert row[:id] == 1
+ assert row[:data] == data
+
+ cleanup_table(admin, table_name)
+ end
+
+ test "round-trips nulls at every depth", %{conn: conn, admin: admin} do
+ table_name = "ex_test_nested_nulls_#{:rand.uniform(100_000)}"
+ cleanup_table(admin, table_name)
+
+ # All columns are nullable by default (no {:not_null, ...} wrapper), so
+ # nulls are allowed both for a whole complex column and for a value nested
+ # inside an array element, a map value, or a row field.
+ schema =
+ Fluss.Schema.new()
+ |> Fluss.Schema.column("id", :int)
+ |> Fluss.Schema.column("whole_array", {:array, :int})
+ |> Fluss.Schema.column("arr_with_null", {:array, :int})
+ |> Fluss.Schema.column("map_with_null", {:map, :string, :int})
+ |> Fluss.Schema.column("row_with_null", {:row, [{"x", :int}, {"y", :int}]})
+
+ descriptor = Fluss.TableDescriptor.new!(schema)
+ :ok = Fluss.Admin.create_table(admin, @database, table_name, descriptor, false)
+
+ table = Fluss.Table.get!(conn, @database, table_name)
+ writer = Fluss.AppendWriter.new!(table)
+
+ {:ok, _} =
+ Fluss.AppendWriter.append(writer, [
+ 1,
+ # whole complex column is null
+ nil,
+ # null element inside an array
+ [10, nil, 30],
+ # null value inside a map
+ %{"a" => 1, "b" => nil},
+ # null field inside a row
+ %{x: nil, y: 2}
+ ])
+
+ :ok = Fluss.AppendWriter.flush(writer)
+
+ scanner = Fluss.LogScanner.new!(table)
+ :ok = Fluss.LogScanner.subscribe(scanner, 0, Fluss.earliest_offset())
+
+ records = poll_records(scanner, 1)
+ assert length(records) == 1
+
+ row = Enum.at(records, 0)[:row]
+ assert row[:id] == 1
+ assert row[:whole_array] == nil
+ assert row[:arr_with_null] == [10, nil, 30]
+ assert row[:map_with_null] == %{"a" => 1, "b" => nil}
+ assert row[:row_with_null] == %{x: nil, y: 2}
+
+ cleanup_table(admin, table_name)
+ end
end
describe "subscribe_buckets" do
diff --git a/fluss-rust/bindings/elixir/test/table_descriptor_test.exs b/fluss-rust/bindings/elixir/test/table_descriptor_test.exs
index 3c3913198..9e7713c26 100644
--- a/fluss-rust/bindings/elixir/test/table_descriptor_test.exs
+++ b/fluss-rust/bindings/elixir/test/table_descriptor_test.exs
@@ -82,5 +82,17 @@ defmodule Fluss.TableDescriptorTest do
TableDescriptor.new!(schema, bogus: 1)
end
end
+
+ test "accepts complex types and nullability wrappers" do
+ schema =
+ Schema.new()
+ |> Schema.column("tags", {:array, :string})
+ |> Schema.column("scores", {:array, {:not_null, :int}})
+ |> Schema.column("attrs", {:map, :string, :int})
+ |> Schema.column("point", {:row, [{"x", :int}, {"y", :int}]})
+ |> Schema.column("deep", {:array, {:map, :string, {:row, [{"a", :int}]}}})
+
+ assert is_reference(TableDescriptor.new!(schema))
+ end
end
end