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
The following commit(s) were added to refs/heads/main by this push:
new 0d24bddf [python] map and row complex types support (#601)
0d24bddf is described below
commit 0d24bddfe5b435750de7bf7e7ea9265098df5b6b
Author: Anton Borisov <[email protected]>
AuthorDate: Mon Jun 8 00:10:59 2026 +0100
[python] map and row complex types support (#601)
* [python] map and row complex type support
* add additional types to from_arrow_field
---
bindings/python/fluss/__init__.pyi | 11 +
bindings/python/src/table.rs | 320 +++++++++++++++++++----
bindings/python/src/utils.rs | 78 +-----
bindings/python/test/conftest.py | 333 +++++++++++++++++++++++-
bindings/python/test/test_kv_table.py | 375 +++++++++++++++++----------
bindings/python/test/test_log_table.py | 291 ++++++++++++++++-----
bindings/python/test/test_schema.py | 61 ++++-
crates/fluss/src/record/arrow.rs | 70 ++++-
crates/fluss/src/row/binary_array.rs | 25 +-
crates/fluss/src/row/binary_map.rs | 24 ++
crates/fluss/src/row/column_writer.rs | 12 +-
website/docs/user-guide/python/data-types.md | 86 ++++++
12 files changed, 1338 insertions(+), 348 deletions(-)
diff --git a/bindings/python/fluss/__init__.pyi
b/bindings/python/fluss/__init__.pyi
index b5bfdfab..4fbbc97e 100644
--- a/bindings/python/fluss/__init__.pyi
+++ b/bindings/python/fluss/__init__.pyi
@@ -639,11 +639,22 @@ class AppendWriter:
- Bytes, Binary (binary data)
- Date, Time, Timestamp, TimestampLTZ (temporal)
- Decimal (arbitrary precision)
+ - Array (Python list)
+ - Map (dict, or list of (key, value) tuples)
+ - Row (dict keyed by field name, or list/tuple by position)
- Null values
+ Nested combinations of Array, Map, and Row are supported. On read,
+ Array -> list, Map -> list of (key, value) tuples, Row -> dict.
+
+ When the row is a dict, a nullable column may be omitted (it defaults
to
+ null); a non-nullable or primary-key column must be present.
+
Example:
writer.append({'id': 1, 'name': 'Alice', 'score': 95.5})
writer.append([1, 'Alice', 95.5])
+ writer.append({'id': 2, 'tags': ['a', 'b'],
+ 'attrs': {'k': 1}, 'profile': {'age': 30}})
Note:
For high-throughput bulk loading, prefer write_arrow_batch().
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index 04f65a8b..e18c74d4 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -21,6 +21,10 @@ use arrow::array::RecordBatch as ArrowRecordBatch;
use arrow::record_batch::RecordBatchReader as _;
use arrow_pyarrow::{FromPyArrow, ToPyArrow};
use arrow_schema::SchemaRef;
+use fcore::metadata::{DataField, DataType, MapType, RowType};
+use fcore::row::binary_array::{FlussArray, FlussArrayWriter};
+use fcore::row::binary_map::{FlussMap, FlussMapWriter};
+use fcore::row::{Datum, F32, F64, GenericRow, InternalRow};
use fluss::record::to_arrow_schema;
use indexmap::IndexMap;
use pyo3::IntoPyObjectExt;
@@ -1196,12 +1200,19 @@ fn python_to_generic_row_inner(
for (i, &col_idx) in target_indices.iter().enumerate() {
let name = target_names[i];
let field = &fields[col_idx];
- let value = dict
- .get_item(name)?
- .ok_or_else(|| FlussError::new_err(format!("Missing field:
{name}")))?;
let dest = if sparse { col_idx } else { i };
- datums[dest] = python_value_to_datum(&value, field.data_type())
- .map_err(|e| FlussError::new_err(format!("Field '{name}':
{e}")))?;
+ match dict.get_item(name)? {
+ Some(value) => {
+ datums[dest] = python_value_to_datum(&value,
field.data_type())
+ .map_err(|e| FlussError::new_err(format!("Field
'{name}': {e}")))?;
+ }
+ None if field.data_type().is_nullable() => {}
+ None => {
+ return Err(FlussError::new_err(format!(
+ "Missing value for non-nullable field '{name}'"
+ )));
+ }
+ }
}
}
@@ -1230,22 +1241,17 @@ fn python_to_generic_row_inner(
}
/// Convert Python value to Datum based on data type
-fn python_value_to_datum(
- value: &Bound<PyAny>,
- data_type: &fcore::metadata::DataType,
-) -> PyResult<fcore::row::Datum<'static>> {
- use fcore::row::{Datum, F32, F64};
-
+fn python_value_to_datum(value: &Bound<PyAny>, data_type: &DataType) ->
PyResult<Datum<'static>> {
if value.is_none() {
return Ok(Datum::Null);
}
match data_type {
- fcore::metadata::DataType::Boolean(_) => {
+ DataType::Boolean(_) => {
let v: bool = value.extract()?;
Ok(Datum::Bool(v))
}
- fcore::metadata::DataType::TinyInt(_) => {
+ DataType::TinyInt(_) => {
// Strict type checking: reject bool for int columns
if value.is_instance_of::<PyBool>() {
return Err(FlussError::new_err(
@@ -1255,7 +1261,7 @@ fn python_value_to_datum(
let v: i8 = value.extract()?;
Ok(Datum::Int8(v))
}
- fcore::metadata::DataType::SmallInt(_) => {
+ DataType::SmallInt(_) => {
if value.is_instance_of::<PyBool>() {
return Err(FlussError::new_err(
"Expected int for SmallInt column, got bool. Use 0 or 1
explicitly."
@@ -1265,7 +1271,7 @@ fn python_value_to_datum(
let v: i16 = value.extract()?;
Ok(Datum::Int16(v))
}
- fcore::metadata::DataType::Int(_) => {
+ DataType::Int(_) => {
if value.is_instance_of::<PyBool>() {
return Err(FlussError::new_err(
"Expected int for Int column, got bool. Use 0 or 1
explicitly.".to_string(),
@@ -1274,7 +1280,7 @@ fn python_value_to_datum(
let v: i32 = value.extract()?;
Ok(Datum::Int32(v))
}
- fcore::metadata::DataType::BigInt(_) => {
+ DataType::BigInt(_) => {
if value.is_instance_of::<PyBool>() {
return Err(FlussError::new_err(
"Expected int for BigInt column, got bool. Use 0 or 1
explicitly.".to_string(),
@@ -1283,19 +1289,19 @@ fn python_value_to_datum(
let v: i64 = value.extract()?;
Ok(Datum::Int64(v))
}
- fcore::metadata::DataType::Float(_) => {
+ DataType::Float(_) => {
let v: f32 = value.extract()?;
Ok(Datum::Float32(F32::from(v)))
}
- fcore::metadata::DataType::Double(_) => {
+ DataType::Double(_) => {
let v: f64 = value.extract()?;
Ok(Datum::Float64(F64::from(v)))
}
- fcore::metadata::DataType::String(_) |
fcore::metadata::DataType::Char(_) => {
+ DataType::String(_) | DataType::Char(_) => {
let v: String = value.extract()?;
Ok(v.into())
}
- fcore::metadata::DataType::Bytes(_) |
fcore::metadata::DataType::Binary(_) => {
+ DataType::Bytes(_) | DataType::Binary(_) => {
// Efficient extraction: downcast to specific type and use bulk
copy.
// PyBytes::as_bytes() and PyByteArray::to_vec() are O(n) bulk
copies of the underlying data.
if let Ok(bytes) = value.downcast::<PyBytes>() {
@@ -1309,14 +1315,14 @@ fn python_value_to_datum(
)))
}
}
- fcore::metadata::DataType::Decimal(decimal_type) => {
+ DataType::Decimal(decimal_type) => {
python_decimal_to_datum(value, decimal_type.precision(),
decimal_type.scale())
}
- fcore::metadata::DataType::Date(_) => python_date_to_datum(value),
- fcore::metadata::DataType::Time(_) => python_time_to_datum(value),
- fcore::metadata::DataType::Timestamp(_) =>
python_datetime_to_timestamp_ntz(value),
- fcore::metadata::DataType::TimestampLTz(_) =>
python_datetime_to_timestamp_ltz(value),
- fcore::metadata::DataType::Array(array_type) => {
+ DataType::Date(_) => python_date_to_datum(value),
+ DataType::Time(_) => python_time_to_datum(value),
+ DataType::Timestamp(_) => python_datetime_to_timestamp_ntz(value),
+ DataType::TimestampLTz(_) => python_datetime_to_timestamp_ltz(value),
+ DataType::Array(array_type) => {
let element_type = array_type.get_element_type();
if value.is_instance_of::<PyString>() {
return Err(FlussError::new_err(format!(
@@ -1332,7 +1338,7 @@ fn python_value_to_datum(
})?;
let len = seq.len()?;
- let mut writer =
fcore::row::binary_array::FlussArrayWriter::new(len, element_type);
+ let mut writer = FlussArrayWriter::new(len, element_type);
for i in 0..len {
let item = seq.get_item(i)?;
@@ -1352,29 +1358,27 @@ fn python_value_to_datum(
Datum::String(v) => writer.write_string(i, &v),
Datum::Blob(v) => writer.write_binary_bytes(i,
v.as_ref()),
Datum::Decimal(v) => {
- if let fcore::metadata::DataType::Decimal(dt) =
element_type {
+ 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 fcore::metadata::DataType::Timestamp(dt) =
element_type {
+ if let DataType::Timestamp(dt) = element_type {
writer.write_timestamp_ntz(i, &v,
dt.precision());
}
}
Datum::TimestampLtz(v) => {
- if let fcore::metadata::DataType::TimestampLTz(dt)
= element_type {
+ if let DataType::TimestampLTz(dt) = element_type {
writer.write_timestamp_ltz(i, &v,
dt.precision());
}
}
Datum::Array(v) => writer.write_array(i, &v),
Datum::Map(v) => writer.write_map(i, &v),
- Datum::Row(_) => {
- return Err(FlussError::new_err(
- "Row datum is not supported as an array
element",
- ));
- }
+ Datum::Row(v) => writer
+ .write_row(i, v.as_ref())
+ .map_err(|e| FlussError::from_core_error(&e))?,
}
}
}
@@ -1384,22 +1388,158 @@ fn python_value_to_datum(
.map_err(|e| FlussError::from_core_error(&e))?;
Ok(Datum::Array(array))
}
- _ => Err(FlussError::new_err(format!(
- "Unsupported data type for row-level operations: {data_type}"
- ))),
+ DataType::Map(map_type) => {
+ let key_type = map_type.key_type();
+ let value_type = map_type.value_type();
+ let pairs = python_map_pairs(value)?;
+ let mut writer = FlussMapWriter::new(pairs.len(), key_type,
value_type);
+ for (k, v) in pairs {
+ let key_datum = python_value_to_datum(&k, key_type)?;
+ let value_datum = python_value_to_datum(&v, value_type)?;
+ writer
+ .write_entry(key_datum, value_datum)
+ .map_err(|e| FlussError::from_core_error(&e))?;
+ }
+ let map = writer
+ .complete()
+ .map_err(|e| FlussError::from_core_error(&e))?;
+ Ok(Datum::Map(map))
+ }
+ DataType::Row(row_type) => {
+ let nested = python_value_to_row_datum(value, row_type)?;
+ Ok(Datum::Row(Box::new(nested)))
+ }
+ }
+}
+
+/// Extract `(key, value)` pairs from a Python value representing a MAP column.
+/// Accepts a `dict`, or a sequence of `(key, value)` pairs — matching the
+/// shape pyarrow's `MapArray.to_pylist()` produces, so MAP values round-trip.
+fn python_map_pairs<'py>(
+ value: &Bound<'py, PyAny>,
+) -> PyResult<Vec<(Bound<'py, PyAny>, Bound<'py, PyAny>)>> {
+ if let Ok(dict) = value.downcast::<PyDict>() {
+ return Ok(dict.iter().collect());
+ }
+ if value.is_instance_of::<PyString>() {
+ return Err(FlussError::new_err(format!(
+ "Expected dict or sequence of (key, value) pairs for Map column,
got {}",
+ get_type_name(value)
+ )));
+ }
+ let seq = value.downcast::<PySequence>().map_err(|_| {
+ FlussError::new_err(format!(
+ "Expected dict or sequence of (key, value) pairs for Map column,
got {}",
+ get_type_name(value)
+ ))
+ })?;
+ let len = seq.len()?;
+ let mut pairs = Vec::with_capacity(len);
+ for i in 0..len {
+ let entry = seq.get_item(i)?;
+ let pair = entry.downcast::<PySequence>().map_err(|_| {
+ FlussError::new_err("Map entries must be (key, value)
pairs".to_string())
+ })?;
+ if pair.len()? != 2 {
+ return Err(FlussError::new_err(
+ "Map entries must be (key, value) pairs of length
2".to_string(),
+ ));
+ }
+ pairs.push((pair.get_item(0)?, pair.get_item(1)?));
+ }
+ Ok(pairs)
+}
+
+/// Convert a Python value (`dict` by field name, or `list`/`tuple` by
position)
+/// into a nested `GenericRow` for a ROW column.
+fn python_value_to_row_datum(
+ value: &Bound<PyAny>,
+ row_type: &RowType,
+) -> PyResult<GenericRow<'static>> {
+ let fields = row_type.fields();
+ let mut datums: Vec<Datum<'static>> = vec![Datum::Null; fields.len()];
+
+ let row_input: RowInput = value.extract().map_err(|_| {
+ FlussError::new_err(format!(
+ "Row column must be a dict, list, or tuple; got {}",
+ get_type_name(value)
+ ))
+ })?;
+
+ match row_input {
+ RowInput::Dict(dict) => {
+ for (k, _) in dict.iter() {
+ let key_str = k.extract::<&str>().map_err(|_| {
+ FlussError::new_err(format!(
+ "Row field keys must be strings; got {}",
+ get_type_name(&k)
+ ))
+ })?;
+ if !fields.iter().any(|f| f.name() == key_str) {
+ return Err(FlussError::new_err(format!(
+ "Unknown row field '{}'. Expected: {}",
+ key_str,
+ fields
+ .iter()
+ .map(|f| f.name())
+ .collect::<Vec<_>>()
+ .join(", ")
+ )));
+ }
+ }
+ for (i, field) in fields.iter().enumerate() {
+ match dict.get_item(field.name())? {
+ Some(v) => {
+ datums[i] = python_value_to_datum(&v,
field.data_type()).map_err(|e| {
+ FlussError::new_err(format!("Row field '{}': {}",
field.name(), e))
+ })?;
+ }
+ None if field.data_type().is_nullable() => {}
+ None => {
+ return Err(FlussError::new_err(format!(
+ "Missing value for non-nullable row field '{}'",
+ field.name()
+ )));
+ }
+ }
+ }
+ }
+ RowInput::List(list) => fill_row_fields(list.as_sequence(), fields,
&mut datums)?,
+ RowInput::Tuple(tuple) => fill_row_fields(tuple.as_sequence(), fields,
&mut datums)?,
}
+
+ Ok(GenericRow { values: datums })
+}
+
+/// Fill ROW field datums from a positional Python sequence.
+fn fill_row_fields(
+ seq: &Bound<PySequence>,
+ fields: &[DataField],
+ datums: &mut [Datum<'static>],
+) -> PyResult<()> {
+ if seq.len()? != fields.len() {
+ return Err(FlussError::new_err(format!(
+ "Expected {} row fields, got {}",
+ fields.len(),
+ seq.len()?
+ )));
+ }
+ for (i, field) in fields.iter().enumerate() {
+ let v = seq.get_item(i)?;
+ datums[i] = python_value_to_datum(&v, field.data_type())
+ .map_err(|e| FlussError::new_err(format!("Row field '{}': {}",
field.name(), e)))?;
+ }
+ Ok(())
}
/// Convert Rust Datum to Python value based on data type.
/// This is the reverse of python_value_to_datum.
pub fn datum_to_python_value(
py: Python,
- row: &dyn fcore::row::InternalRow,
+ row: &dyn InternalRow,
pos: usize,
- data_type: &fcore::metadata::DataType,
+ data_type: &DataType,
) -> PyResult<Py<PyAny>> {
- use fcore::metadata::DataType;
-
// Check for null first
if row
.is_null_at(pos)
@@ -1522,19 +1662,95 @@ pub fn datum_to_python_value(
.map_err(|e| FlussError::from_core_error(&e))?
.try_into_binary()
.map_err(|e| FlussError::from_core_error(&e))?;
+ array_to_pylist(py, &array_data, array_type.get_element_type())
+ }
+ DataType::Map(map_type) => {
+ let map_data = row
+ .get_map(pos)
+ .map_err(|e| FlussError::from_core_error(&e))?
+ .try_into_binary()
+ .map_err(|e| FlussError::from_core_error(&e))?;
+ map_to_pylist(py, &map_data, map_type)
+ }
+ DataType::Row(row_type) => {
+ let nested = row
+ .get_row(pos)
+ .map_err(|e| FlussError::from_core_error(&e))?
+ .try_into_generic(row_type)
+ .map_err(|e| FlussError::from_core_error(&e))?;
+ row_to_pydict(py, &nested, row_type)
+ }
+ }
+}
- let element_type = array_type.get_element_type();
- let py_list = pyo3::types::PyList::empty(py);
+/// Convert a binary `FlussArray` to a Python list.
+fn array_to_pylist(py: Python, arr: &FlussArray, element_type: &DataType) ->
PyResult<Py<PyAny>> {
+ let py_list = pyo3::types::PyList::empty(py);
+ for i in 0..arr.size() {
+ py_list.append(array_elem_to_python(py, arr, i, element_type)?)?;
+ }
+ Ok(py_list.into_any().unbind())
+}
- for i in 0..array_data.size() {
- let py_val = datum_to_python_value(py, &array_data, i,
element_type)?;
- py_list.append(py_val)?;
- }
- Ok(py_list.into_any().unbind())
+/// Convert a Fluss `MAP` to a Python list of `(key, value)` tuples — matching
+/// pyarrow's default `MapArray.to_pylist()` shape (preserves duplicate keys
and
+/// ordering, and allows non-hashable keys).
+fn map_to_pylist(py: Python, map_data: &FlussMap, map_type: &MapType) ->
PyResult<Py<PyAny>> {
+ let keys = map_data.key_array();
+ let values = map_data.value_array();
+ let py_list = pyo3::types::PyList::empty(py);
+ for i in 0..map_data.size() {
+ let py_key = array_elem_to_python(py, keys, i, map_type.key_type())?;
+ let py_val = array_elem_to_python(py, values, i,
map_type.value_type())?;
+ py_list.append(pyo3::types::PyTuple::new(py, [py_key, py_val])?)?;
+ }
+ Ok(py_list.into_any().unbind())
+}
+
+/// Convert a Fluss `ROW` to a Python dict keyed by field name — matching
+/// pyarrow's `StructArray.to_pylist()` shape.
+fn row_to_pydict(py: Python, row: &dyn InternalRow, row_type: &RowType) ->
PyResult<Py<PyAny>> {
+ let dict = PyDict::new(py);
+ for (i, field) in row_type.fields().iter().enumerate() {
+ let py_val = datum_to_python_value(py, row, i, field.data_type())?;
+ dict.set_item(field.name(), py_val)?;
+ }
+ Ok(dict.into_any().unbind())
+}
+
+/// Convert element `i` of a binary `FlussArray` to a Python value. A binary
+/// array needs the explicit nested type to decode MAP/ROW elements (the
generic
+/// `DataGetters` trait getters cover only scalars and nested arrays), so this
+/// dispatches through `FlussArray`'s inherent typed getters.
+fn array_elem_to_python(
+ py: Python,
+ arr: &FlussArray,
+ i: usize,
+ dt: &DataType,
+) -> PyResult<Py<PyAny>> {
+ if arr.is_null_at(i) {
+ return Ok(py.None());
+ }
+ match dt {
+ DataType::Array(array_type) => {
+ let nested = arr
+ .get_array(i)
+ .map_err(|e| FlussError::from_core_error(&e))?;
+ array_to_pylist(py, &nested, array_type.get_element_type())
+ }
+ DataType::Map(map_type) => {
+ let nested = arr
+ .get_map(i, map_type.key_type(), map_type.value_type())
+ .map_err(|e| FlussError::from_core_error(&e))?;
+ map_to_pylist(py, &nested, map_type)
+ }
+ DataType::Row(row_type) => {
+ let nested = arr
+ .get_row(i, row_type)
+ .map_err(|e| FlussError::from_core_error(&e))?;
+ row_to_pydict(py, &nested, row_type)
}
- _ => Err(FlussError::new_err(format!(
- "Unsupported data type for conversion to Python: {data_type}"
- ))),
+ _ => datum_to_python_value(py, arr, i, dt),
}
}
diff --git a/bindings/python/src/utils.rs b/bindings/python/src/utils.rs
index f69188f1..78f224c9 100644
--- a/bindings/python/src/utils.rs
+++ b/bindings/python/src/utils.rs
@@ -18,6 +18,7 @@
use crate::*;
use arrow_pyarrow::{FromPyArrow, ToPyArrow};
use arrow_schema::SchemaRef;
+use fcore::record::from_arrow_field;
use std::sync::Arc;
/// Utilities for schema conversion between PyArrow, Arrow, and Fluss
@@ -36,82 +37,13 @@ impl Utils {
})
}
- /// Convert an Arrow Field to a Fluss DataType, preserving nullability.
+ /// Convert an Arrow Field to a Fluss DataType. Delegates to core's
canonical
+ /// Arrow->Fluss converter, which handles scalars, list, map, and struct
+ /// recursively and preserves per-field nullability.
pub fn arrow_field_to_fluss_type(
field: &arrow::datatypes::Field,
) -> PyResult<fcore::metadata::DataType> {
- use arrow::datatypes::DataType as ArrowDataType;
- use fcore::metadata::DataTypes;
-
- let fluss_type = match field.data_type() {
- ArrowDataType::Boolean => DataTypes::boolean(),
- ArrowDataType::Int8 => DataTypes::tinyint(),
- ArrowDataType::Int16 => DataTypes::smallint(),
- ArrowDataType::Int32 => DataTypes::int(),
- ArrowDataType::Int64 => DataTypes::bigint(),
- ArrowDataType::UInt8 => DataTypes::tinyint(),
- ArrowDataType::UInt16 => DataTypes::smallint(),
- ArrowDataType::UInt32 => DataTypes::int(),
- ArrowDataType::UInt64 => DataTypes::bigint(),
- ArrowDataType::Float32 => DataTypes::float(),
- ArrowDataType::Float64 => DataTypes::double(),
- ArrowDataType::Utf8 | ArrowDataType::LargeUtf8 =>
DataTypes::string(),
- ArrowDataType::Binary | ArrowDataType::LargeBinary =>
DataTypes::bytes(),
- ArrowDataType::FixedSizeBinary(n) => DataTypes::binary(*n as
usize),
- ArrowDataType::Date32 => DataTypes::date(),
- ArrowDataType::Date64 => DataTypes::date(),
- ArrowDataType::Time32(unit) => match unit {
- arrow_schema::TimeUnit::Second =>
DataTypes::time_with_precision(0),
- arrow_schema::TimeUnit::Millisecond =>
DataTypes::time_with_precision(3),
- _ => {
- return Err(FlussError::new_err(format!(
- "Unsupported Time32 unit: {unit:?}"
- )));
- }
- },
- ArrowDataType::Time64(unit) => match unit {
- arrow_schema::TimeUnit::Microsecond =>
DataTypes::time_with_precision(6),
- arrow_schema::TimeUnit::Nanosecond =>
DataTypes::time_with_precision(9),
- _ => {
- return Err(FlussError::new_err(format!(
- "Unsupported Time64 unit: {unit:?}"
- )));
- }
- },
- ArrowDataType::Timestamp(unit, tz) => {
- let precision = match unit {
- arrow_schema::TimeUnit::Second => 0,
- arrow_schema::TimeUnit::Millisecond => 3,
- arrow_schema::TimeUnit::Microsecond => 6,
- arrow_schema::TimeUnit::Nanosecond => 9,
- };
- // Arrow Timestamp with timezone -> Fluss TimestampLtz
- // Arrow Timestamp without timezone -> Fluss Timestamp (NTZ)
- if tz.is_some() {
- DataTypes::timestamp_ltz_with_precision(precision)
- } else {
- DataTypes::timestamp_with_precision(precision)
- }
- }
- ArrowDataType::Decimal128(precision, scale) => {
- DataTypes::decimal(*precision as u32, *scale as u32)
- }
- ArrowDataType::List(element_field) => {
- let element_type =
Utils::arrow_field_to_fluss_type(element_field)?;
- DataTypes::array(element_type)
- }
- other => {
- return Err(FlussError::new_err(format!(
- "Unsupported Arrow data type: {other:?}"
- )));
- }
- };
-
- if field.is_nullable() {
- Ok(fluss_type)
- } else {
- Ok(fluss_type.as_non_nullable())
- }
+ from_arrow_field(field).map_err(|e| FlussError::from_core_error(&e))
}
/// Convert Fluss DataType to string representation, appending " NOT NULL"
diff --git a/bindings/python/test/conftest.py b/bindings/python/test/conftest.py
index 8b2bc732..22de4d43 100644
--- a/bindings/python/test/conftest.py
+++ b/bindings/python/test/conftest.py
@@ -17,12 +17,17 @@
import asyncio
import json
+import math
import os
import subprocess
import tempfile
import time
+from datetime import date, datetime, timezone
+from datetime import time as dt_time
+from decimal import Decimal
from pathlib import Path
+import pyarrow as pa
import pytest
import pytest_asyncio
from filelock import FileLock
@@ -156,6 +161,7 @@ async def wait_for_table_ready(admin):
"""
Fixture that returns a helper function to wait for a table or partition to
be ready.
"""
+
async def _wait(table_path, timeout=15, partition_name=None):
start_time = time.monotonic()
while time.monotonic() - start_time < timeout:
@@ -165,12 +171,22 @@ async def wait_for_table_ready(admin):
table_path, partition_name, [0],
fluss.OffsetSpec.earliest()
)
else:
- await admin.list_offsets(table_path, [0],
fluss.OffsetSpec.earliest())
+ await admin.list_offsets(
+ table_path, [0], fluss.OffsetSpec.earliest()
+ )
return
except (fluss.FlussError, Exception) as e:
# Catch "No leader found" or other errors that indicate the
table/partition is still initializing
err_msg = str(e)
- if any(msg in err_msg for msg in ["No leader found", "Table
not ready", "Metadata not ready", "not leader or follower"]):
+ if any(
+ msg in err_msg
+ for msg in [
+ "No leader found",
+ "Table not ready",
+ "Metadata not ready",
+ "not leader or follower",
+ ]
+ ):
await asyncio.sleep(1)
continue
raise
@@ -180,3 +196,316 @@ async def wait_for_table_ready(admin):
)
return _wait
+
+
+# Complex-type (ARRAY/MAP/ROW) helpers shared by the KV and log
+# all_complex_datatypes tests: schema sections plus the full/edge/null row
matrix.
+# Read-back shapes: ARRAY -> list, MAP -> list of (key, value) tuples, ROW ->
dict.
+
+
+def pa_row_seq_label() -> pa.DataType:
+ return pa.struct([("seq", pa.int32()), ("label", pa.string())])
+
+
+def pa_row_deep() -> pa.DataType:
+ return pa.struct([("inner", pa.struct([("n", pa.int32())]))])
+
+
+def pa_row_rich() -> pa.DataType:
+ return pa.struct(
+ [
+ ("f_bool", pa.bool_()),
+ ("f_int", pa.int32()),
+ ("f_long", pa.int64()),
+ ("f_float", pa.float32()),
+ ("f_double", pa.float64()),
+ ("f_str", pa.string()),
+ ("f_bytes", pa.binary()),
+ ("f_decimal", pa.decimal128(10, 2)),
+ ("f_date", pa.date32()),
+ ("f_time", pa.time32("ms")),
+ ("f_ts_ntz", pa.timestamp("us")),
+ ("f_ts_ltz", pa.timestamp("us", tz="UTC")),
+ ("f_binary", pa.binary(4)),
+ ("f_array_int", pa.list_(pa.int32())),
+ ]
+ )
+
+
+def array_basics_fields() -> list:
+ return [
+ pa.field("arr_int", pa.list_(pa.int32())),
+ pa.field("arr_string", pa.list_(pa.string())),
+ pa.field("arr_of_arr", pa.list_(pa.list_(pa.int32()))),
+ pa.field("arr_of_row", pa.list_(pa_row_seq_label())),
+ ]
+
+
+def row_basics_fields() -> list:
+ return [
+ pa.field("row_basic", pa_row_seq_label()),
+ pa.field("row_deep", pa_row_deep()),
+ pa.field("row_rich", pa_row_rich()),
+ ]
+
+
+def map_basics_fields() -> list:
+ return [
+ pa.field("map_string_int", pa.map_(pa.string(), pa.int32())),
+ pa.field("map_of_row", pa.map_(pa.string(), pa_row_seq_label())),
+ pa.field("map_of_map", pa.map_(pa.string(), pa.map_(pa.string(),
pa.int32()))),
+ pa.field("map_of_array", pa.map_(pa.string(), pa.list_(pa.int32()))),
+ pa.field("array_of_map", pa.list_(pa.map_(pa.string(), pa.int32()))),
+ ]
+
+
+def array_rich_fields() -> list:
+ return [
+ pa.field("arr_bytes", pa.list_(pa.binary())),
+ pa.field("arr_date", pa.list_(pa.date32())),
+ pa.field("arr_time", pa.list_(pa.time32("ms"))),
+ pa.field("arr_ts", pa.list_(pa.timestamp("us"))),
+ pa.field("arr_ts_ltz", pa.list_(pa.timestamp("us", tz="UTC"))),
+ pa.field("arr_decimal", pa.list_(pa.decimal128(10, 2))),
+ pa.field("arr_decimal_big", pa.list_(pa.decimal128(22, 5))),
+ pa.field("arr_float", pa.list_(pa.float32())),
+ pa.field("arr_double", pa.list_(pa.float64())),
+ pa.field("arr_binary", pa.list_(pa.binary(4))),
+ ]
+
+
+def map_rich_fields() -> list:
+ return [
+ pa.field("map_bytes", pa.map_(pa.string(), pa.binary())),
+ pa.field("map_decimal", pa.map_(pa.string(), pa.decimal128(10, 2))),
+ pa.field("map_date", pa.map_(pa.string(), pa.date32())),
+ pa.field("map_time", pa.map_(pa.string(), pa.time32("ms"))),
+ pa.field("map_ts", pa.map_(pa.string(), pa.timestamp("us"))),
+ pa.field("map_ts_ltz", pa.map_(pa.string(), pa.timestamp("us",
tz="UTC"))),
+ pa.field("map_float", pa.map_(pa.string(), pa.float32())),
+ pa.field("map_double", pa.map_(pa.string(), pa.float64())),
+ pa.field("map_bool", pa.map_(pa.string(), pa.bool_())),
+ pa.field("map_binary", pa.map_(pa.string(), pa.binary(4))),
+ pa.field("map_int_key", pa.map_(pa.int32(), pa.string())),
+ ]
+
+
+def complex_fields() -> list:
+ """`id` + all complex sections, in section order."""
+ return (
+ [pa.field("id", pa.int32())]
+ + array_basics_fields()
+ + array_rich_fields()
+ + row_basics_fields()
+ + map_basics_fields()
+ + map_rich_fields()
+ )
+
+
+def complex_column_names() -> list:
+ """All complex column names (everything except `id`)."""
+ return [f.name for f in complex_fields() if f.name != "id"]
+
+
+def complex_schema(primary_keys=None) -> "fluss.Schema":
+ return fluss.Schema(pa.schema(complex_fields()), primary_keys=primary_keys)
+
+
+_ROW_RICH_FULL = {
+ "f_bool": True,
+ "f_int": 100_000,
+ "f_long": 9_876_543_210,
+ "f_float": float("inf"),
+ "f_double": math.pi,
+ "f_str": "hello world",
+ "f_bytes": b"binary",
+ "f_decimal": Decimal("123.45"),
+ "f_date": date(2026, 1, 23),
+ "f_time": dt_time(10, 13, 47, 123000),
+ "f_ts_ntz": datetime(2026, 1, 23, 10, 13, 47, 123456),
+ "f_ts_ltz": datetime(2026, 1, 23, 10, 13, 47, 123456, tzinfo=timezone.utc),
+ "f_binary": b"\x01\x02\x03\x04",
+ "f_array_int": [7, None, 11],
+}
+
+
+def complex_full_row(id_: int) -> dict:
+ """Fully-populated row exercising every complex shape (incl. nesting)."""
+ return {
+ "id": id_,
+ "arr_int": [10, 20, 30],
+ "arr_string": ["hello", "world"],
+ "arr_of_arr": [[1, 2], [3, 4]],
+ "arr_of_row": [{"seq": 1, "label": "open"}, {"seq": 2, "label":
"close"}],
+ "arr_bytes": [b"\x10\x20\x30", None],
+ "arr_date": [date(2026, 1, 23), None],
+ "arr_time": [dt_time(10, 13, 47, 123000), None],
+ "arr_ts": [datetime(2026, 1, 23, 10, 13, 47, 123456)],
+ "arr_ts_ltz": [datetime(2026, 1, 23, 10, 13, 47, 123456,
tzinfo=timezone.utc)],
+ "arr_decimal": [Decimal("123.45"), None],
+ "arr_decimal_big": [
+ Decimal("12345678901234567.12345"),
+ Decimal("-99999999999999999.99999"),
+ ],
+ "arr_float": [float("nan"), float("inf"), float("-inf")],
+ "arr_double": [float("nan"), float("inf"), float("-inf")],
+ "arr_binary": [b"\xde\xad\xbe\xef", b"\x00\x01\x02\x03"],
+ "row_basic": {"seq": 42, "label": "hello"},
+ "row_deep": {"inner": {"n": 99}},
+ "row_rich": dict(_ROW_RICH_FULL),
+ "map_string_int": {"a": 1, "b": None, "c": 3},
+ "map_of_row": {
+ "e0": {"seq": 1, "label": "open"},
+ "e1": {"seq": 2, "label": "close"},
+ },
+ "map_of_map": {"g1": {"a": 1, "b": 2}, "g2": {"c": 3}},
+ "map_of_array": {"primes": [2, 3, 5], "squares": [1, 4]},
+ "array_of_map": [{"x": 1, "y": 2}, {"z": 9}],
+ "map_bytes": {"k": b"\x10\x20\x30"},
+ "map_decimal": {"p": Decimal("123.45")},
+ "map_date": {"d": date(2026, 1, 23)},
+ "map_time": {"t": dt_time(10, 13, 47, 123000)},
+ "map_ts": {"t": datetime(2026, 1, 23, 10, 13, 47, 123456)},
+ "map_ts_ltz": {
+ "t": datetime(2026, 1, 23, 10, 13, 47, 123456, tzinfo=timezone.utc)
+ },
+ "map_float": {"nan": float("nan"), "inf": float("inf"), "ninf":
float("-inf")},
+ "map_double": {"nan": float("nan"), "inf": float("inf"), "ninf":
float("-inf")},
+ "map_bool": {"t": True, "f": False},
+ "map_binary": {"k": b"\x01\x02\x03\x04"},
+ "map_int_key": {1: "one", 2: "two"},
+ }
+
+
+# Rich sections appear only in the full row; edge/null rows leave them NULL.
+_RICH_COLUMNS = [f.name for f in array_rich_fields() + map_rich_fields()]
+
+
+def complex_edge_row(id_: int) -> dict:
+ """Edge cases: empty collections, null elements, null nested rows."""
+ return {
+ "id": id_,
+ "arr_int": [],
+ "arr_string": [None],
+ "arr_of_arr": [[5], None, []],
+ "arr_of_row": [{"seq": 7, "label": "x"}, None, {"seq": 8, "label":
"y"}],
+ "row_basic": None,
+ "row_deep": None,
+ "row_rich": None,
+ "map_string_int": {},
+ "map_of_row": {},
+ "map_of_map": {},
+ "map_of_array": {},
+ "array_of_map": [],
+ **{name: None for name in _RICH_COLUMNS},
+ }
+
+
+def complex_null_row(id_: int) -> dict:
+ """Every complex column set to NULL."""
+ return {"id": id_, **{name: None for name in complex_column_names()}}
+
+
+def _map(value) -> dict:
+ """A read-back MAP is a list of (key, value) tuples; turn it into a dict
for
+ order-independent comparison (test maps use unique scalar keys)."""
+ return dict(value)
+
+
+def _assert_float_triplet(values) -> None:
+ """Assert a 3-element sequence holds [NaN, +Inf, -Inf]."""
+ assert math.isnan(values[0])
+ assert math.isinf(values[1]) and values[1] > 0
+ assert math.isinf(values[2]) and values[2] < 0
+
+
+def assert_complex_full(row: dict) -> None:
+ assert row["arr_int"] == [10, 20, 30]
+ assert row["arr_string"] == ["hello", "world"]
+ assert row["arr_of_arr"] == [[1, 2], [3, 4]]
+ assert row["arr_of_row"] == [
+ {"seq": 1, "label": "open"},
+ {"seq": 2, "label": "close"},
+ ]
+ assert row["row_basic"] == {"seq": 42, "label": "hello"}
+ assert row["row_deep"] == {"inner": {"n": 99}}
+ rr = row["row_rich"]
+ assert rr["f_bool"] is True
+ assert rr["f_int"] == 100_000
+ assert rr["f_long"] == 9_876_543_210
+ assert math.isinf(rr["f_float"]) and rr["f_float"] > 0
+ assert math.isclose(rr["f_double"], math.pi, rel_tol=1e-15)
+ assert rr["f_str"] == "hello world"
+ assert rr["f_bytes"] == b"binary"
+ assert rr["f_decimal"] == Decimal("123.45")
+ assert rr["f_date"] == date(2026, 1, 23)
+ assert rr["f_time"] == dt_time(10, 13, 47, 123000)
+ assert rr["f_ts_ntz"] == datetime(2026, 1, 23, 10, 13, 47, 123456)
+ assert rr["f_ts_ltz"] == datetime(
+ 2026, 1, 23, 10, 13, 47, 123456, tzinfo=timezone.utc
+ )
+ assert rr["f_binary"] == b"\x01\x02\x03\x04"
+ assert rr["f_array_int"] == [7, None, 11]
+ assert _map(row["map_string_int"]) == {"a": 1, "b": None, "c": 3}
+ assert _map(row["map_of_row"]) == {
+ "e0": {"seq": 1, "label": "open"},
+ "e1": {"seq": 2, "label": "close"},
+ }
+ assert {k: _map(v) for k, v in row["map_of_map"]} == {
+ "g1": {"a": 1, "b": 2},
+ "g2": {"c": 3},
+ }
+ assert _map(row["map_of_array"]) == {"primes": [2, 3, 5], "squares": [1,
4]}
+ assert [_map(m) for m in row["array_of_map"]] == [{"x": 1, "y": 2}, {"z":
9}]
+ assert row["arr_bytes"] == [b"\x10\x20\x30", None]
+ assert row["arr_date"] == [date(2026, 1, 23), None]
+ assert row["arr_time"] == [dt_time(10, 13, 47, 123000), None]
+ assert row["arr_ts"] == [datetime(2026, 1, 23, 10, 13, 47, 123456)]
+ assert row["arr_ts_ltz"] == [
+ datetime(2026, 1, 23, 10, 13, 47, 123456, tzinfo=timezone.utc)
+ ]
+ assert row["arr_decimal"] == [Decimal("123.45"), None]
+ assert row["arr_decimal_big"] == [
+ Decimal("12345678901234567.12345"),
+ Decimal("-99999999999999999.99999"),
+ ]
+ _assert_float_triplet(row["arr_float"])
+ _assert_float_triplet(row["arr_double"])
+ assert row["arr_binary"] == [b"\xde\xad\xbe\xef", b"\x00\x01\x02\x03"]
+ assert _map(row["map_bytes"]) == {"k": b"\x10\x20\x30"}
+ assert _map(row["map_decimal"]) == {"p": Decimal("123.45")}
+ assert _map(row["map_date"]) == {"d": date(2026, 1, 23)}
+ assert _map(row["map_time"]) == {"t": dt_time(10, 13, 47, 123000)}
+ assert _map(row["map_ts"]) == {"t": datetime(2026, 1, 23, 10, 13, 47,
123456)}
+ assert _map(row["map_ts_ltz"]) == {
+ "t": datetime(2026, 1, 23, 10, 13, 47, 123456, tzinfo=timezone.utc)
+ }
+ mf = _map(row["map_float"])
+ _assert_float_triplet([mf["nan"], mf["inf"], mf["ninf"]])
+ md = _map(row["map_double"])
+ _assert_float_triplet([md["nan"], md["inf"], md["ninf"]])
+ assert _map(row["map_bool"]) == {"t": True, "f": False}
+ assert _map(row["map_binary"]) == {"k": b"\x01\x02\x03\x04"}
+ assert _map(row["map_int_key"]) == {1: "one", 2: "two"}
+
+
+def assert_complex_edge(row: dict) -> None:
+ assert row["arr_int"] == []
+ assert row["arr_string"] == [None]
+ assert row["arr_of_arr"] == [[5], None, []]
+ assert row["arr_of_row"] == [
+ {"seq": 7, "label": "x"},
+ None,
+ {"seq": 8, "label": "y"},
+ ]
+ assert row["row_basic"] is None
+ assert row["row_deep"] is None
+ assert row["row_rich"] is None
+ # empty maps read back as empty lists
+ assert row["map_string_int"] == []
+ assert row["map_of_row"] == []
+ assert row["map_of_map"] == []
+ assert row["map_of_array"] == []
+ assert row["array_of_map"] == []
+ for name in _RICH_COLUMNS:
+ assert row[name] is None, f"{name} should be null in the edge row"
diff --git a/bindings/python/test/test_kv_table.py
b/bindings/python/test/test_kv_table.py
index f3cddf8c..83ceea0e 100644
--- a/bindings/python/test/test_kv_table.py
+++ b/bindings/python/test/test_kv_table.py
@@ -27,6 +27,15 @@ from decimal import Decimal
import pyarrow as pa
import pytest
+from conftest import (
+ assert_complex_edge,
+ assert_complex_full,
+ complex_column_names,
+ complex_edge_row,
+ complex_full_row,
+ complex_null_row,
+ complex_schema,
+)
import fluss
@@ -175,7 +184,9 @@ async def test_composite_primary_keys(connection, admin):
assert result is not None
assert result["score"] == 500
- prefix_lookuper = table.new_lookup().lookup_by(["region",
"user_id"]).create_lookuper()
+ prefix_lookuper = (
+ table.new_lookup().lookup_by(["region", "user_id"]).create_lookuper()
+ )
# Prefix (US, 1) should match 2 rows (event_id 1 and 2)
rows = await prefix_lookuper.lookup({"region": "US", "user_id": 1})
@@ -220,9 +231,7 @@ async def test_partial_update(connection, admin):
# Insert initial record
upsert_writer = table.new_upsert().create_writer()
- handle = upsert_writer.upsert(
- {"id": 1, "name": "Verso", "age": 32, "score": 6942}
- )
+ handle = upsert_writer.upsert({"id": 1, "name": "Verso", "age": 32,
"score": 6942})
await handle.wait()
lookuper = table.new_lookup().create_lookuper()
@@ -272,15 +281,11 @@ async def test_partial_update_by_index(connection, admin):
table = await connection.get_table(table_path)
upsert_writer = table.new_upsert().create_writer()
- handle = upsert_writer.upsert(
- {"id": 1, "name": "Verso", "age": 32, "score": 6942}
- )
+ handle = upsert_writer.upsert({"id": 1, "name": "Verso", "age": 32,
"score": 6942})
await handle.wait()
# Partial update by indices: columns 0=id (PK), 1=name
- partial_writer = (
- table.new_upsert().partial_update_by_index([0, 1]).create_writer()
- )
+ partial_writer = table.new_upsert().partial_update_by_index([0,
1]).create_writer()
handle = partial_writer.upsert([1, "Verso Renamed"])
await handle.wait()
@@ -293,6 +298,82 @@ async def test_partial_update_by_index(connection, admin):
await admin.drop_table(table_path, ignore_if_not_exists=False)
+async def test_partial_update_complex(connection, admin):
+ """Partial updates preserve non-targeted ROW/MAP/ARRAY columns and update
the
+ targeted ones (mirrors the Rust partial_update IT)."""
+ table_path = fluss.TablePath("fluss", "py_test_partial_update_complex")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("name", pa.string()),
+ pa.field(
+ "nested", pa.struct([("seq", pa.int32()), ("label",
pa.string())])
+ ),
+ pa.field("attrs", pa.map_(pa.string(), pa.int32())),
+ pa.field("tags", pa.list_(pa.string())),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ await admin.create_table(
+ table_path, fluss.TableDescriptor(schema), ignore_if_exists=False
+ )
+ table = await connection.get_table(table_path)
+
+ await _upsert_and_wait(
+ table.new_upsert().create_writer(),
+ {
+ "id": 1,
+ "name": "Verso",
+ "nested": {"seq": 10, "label": "alpha"},
+ "attrs": {"a": 1, "b": 2},
+ "tags": ["alpha-tag", "beta-tag"],
+ },
+ )
+
+ lookuper = table.new_lookup().create_lookuper()
+
+ # Update only `name`: ROW/MAP/ARRAY columns must be preserved.
+ await _upsert_and_wait(
+ table.new_upsert().partial_update_by_name(["id",
"name"]).create_writer(),
+ {"id": 1, "name": "Renamed"},
+ )
+ r = await lookuper.lookup({"id": 1})
+ assert r["name"] == "Renamed"
+ assert r["nested"] == {"seq": 10, "label": "alpha"}
+ assert dict(r["attrs"]) == {"a": 1, "b": 2}
+ assert r["tags"] == ["alpha-tag", "beta-tag"]
+
+ # Update only `nested`: scalar + other complex columns preserved.
+ await _upsert_and_wait(
+ table.new_upsert().partial_update_by_name(["id",
"nested"]).create_writer(),
+ {"id": 1, "nested": {"seq": 99, "label": "omega"}},
+ )
+ r = await lookuper.lookup({"id": 1})
+ assert r["nested"] == {"seq": 99, "label": "omega"}
+ assert r["name"] == "Renamed"
+ assert dict(r["attrs"]) == {"a": 1, "b": 2}
+ assert r["tags"] == ["alpha-tag", "beta-tag"]
+
+ # Update `attrs` and `tags`.
+ await _upsert_and_wait(
+ table.new_upsert()
+ .partial_update_by_name(["id", "attrs", "tags"])
+ .create_writer(),
+ {"id": 1, "attrs": {"z": 99}, "tags": ["gamma"]},
+ )
+ r = await lookuper.lookup({"id": 1})
+ assert dict(r["attrs"]) == {"z": 99}
+ assert r["tags"] == ["gamma"]
+ assert r["nested"] == {"seq": 99, "label": "omega"}
+ assert r["name"] == "Renamed"
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
async def test_partitioned_table_upsert_and_lookup(connection, admin):
"""Test upsert/lookup/delete on a partitioned KV table."""
table_path = fluss.TablePath("fluss", "py_test_partitioned_kv_table")
@@ -379,163 +460,179 @@ async def
test_partitioned_table_upsert_and_lookup(connection, admin):
await admin.drop_table(table_path, ignore_if_not_exists=False)
-async def test_upsert_and_lookup_with_array(connection, admin):
- """Test upsert and lookup with flat, nested, and null-pattern arrays in KV
tables."""
- table_path = fluss.TablePath("fluss", "py_test_kv_arrays")
+async def test_omit_nullable_vs_required_fields(connection, admin):
+ """Omitting a nullable field defaults it to null; omitting a non-nullable
(or
+ PK) field is a clear client-side error -- at both top level and inside a
ROW."""
+ table_path = fluss.TablePath("fluss", "py_test_omit_fields")
await admin.drop_table(table_path, ignore_if_not_exists=True)
+ nested_type = pa.struct(
+ [
+ pa.field("a", pa.int32()), # nullable
+ pa.field("b", pa.int32(), nullable=False), # NOT NULL
+ ]
+ )
schema = fluss.Schema(
pa.schema(
[
pa.field("id", pa.int32()),
- pa.field("tags", pa.list_(pa.string())),
- pa.field("scores", pa.list_(pa.int32())),
- pa.field("matrix", pa.list_(pa.list_(pa.int32()))),
+ pa.field("opt", pa.int32()), # nullable
+ pa.field("req", pa.int32(), nullable=False), # NOT NULL
+ pa.field("nested", nested_type),
]
),
primary_keys=["id"],
)
- table_descriptor = fluss.TableDescriptor(schema)
- await admin.create_table(table_path, table_descriptor,
ignore_if_exists=False)
-
+ await admin.create_table(
+ table_path, fluss.TableDescriptor(schema), ignore_if_exists=False
+ )
table = await connection.get_table(table_path)
- upsert_writer = table.new_upsert().create_writer()
+ writer = table.new_upsert().create_writer()
+ lookuper = table.new_lookup().create_lookuper()
- await _upsert_and_wait(
- upsert_writer,
- {
- "id": 1,
- "tags": ["hello", "world"],
- "scores": [10, 20, 30],
- "matrix": [[1, 2], [3, 4]],
- },
- )
- await _upsert_and_wait(
- upsert_writer,
- {"id": 2, "tags": [None], "scores": [], "matrix": None},
- )
- await _upsert_and_wait(
- upsert_writer,
- {"id": 3, "tags": None, "scores": [42], "matrix": [[], [5], [6, 7,
8]]},
- )
- await _upsert_and_wait(
- upsert_writer,
- {"id": 4, "tags": None, "scores": None, "matrix": [[1, None], None,
[]]},
- )
+ # Omit nullable top-level field `opt` -> Null.
+ await _upsert_and_wait(writer, {"id": 1, "req": 5, "nested": {"a": 1, "b":
2}})
+ r = await lookuper.lookup({"id": 1})
+ assert r["opt"] is None
+ assert r["req"] == 5
+ assert r["nested"] == {"a": 1, "b": 2}
- lookuper = table.new_lookup().create_lookuper()
+ # Omit nullable nested field `a` -> Null.
+ await _upsert_and_wait(writer, {"id": 2, "req": 9, "nested": {"b": 3}})
+ r = await lookuper.lookup({"id": 2})
+ assert r["nested"] == {"a": None, "b": 3}
- result1 = await lookuper.lookup({"id": 1})
- assert result1 is not None
- assert result1["tags"] == ["hello", "world"]
- assert result1["scores"] == [10, 20, 30]
- assert result1["matrix"] == [[1, 2], [3, 4]]
-
- result2 = await lookuper.lookup({"id": 2})
- assert result2 is not None
- assert result2["tags"] == [None]
- assert result2["scores"] == []
- assert result2["matrix"] is None
-
- result3 = await lookuper.lookup({"id": 3})
- assert result3 is not None
- assert result3["tags"] is None
- assert result3["scores"] == [42]
- assert result3["matrix"] == [[], [5], [6, 7, 8]]
-
- result4 = await lookuper.lookup({"id": 4})
- assert result4 is not None
- assert result4["tags"] is None
- assert result4["scores"] is None
- assert result4["matrix"] == [[1, None], None, []]
+ # Omit non-nullable top-level field `req` -> error.
+ with pytest.raises(Exception, match="non-nullable field 'req'"):
+ writer.upsert({"id": 3, "opt": 1, "nested": {"a": 1, "b": 2}})
+
+ # Omit non-nullable nested field `b` -> error.
+ with pytest.raises(Exception, match="non-nullable row field 'b'"):
+ writer.upsert({"id": 4, "req": 5, "nested": {"a": 1}})
await admin.drop_table(table_path, ignore_if_not_exists=False)
-async def test_upsert_and_lookup_with_array_rich_types(connection, admin):
- """Test upsert/lookup for arrays with rich element types and encoding edge
cases."""
- table_path = fluss.TablePath("fluss", "py_test_kv_arrays_rich_types")
+async def test_partitioned_complex(connection, admin):
+ """ROW/MAP/ARRAY columns round-trip and update correctly across partitions
+ (mirrors the complex columns in the Rust
partitioned_table_upsert_and_lookup IT)."""
+ table_path = fluss.TablePath("fluss", "py_test_partitioned_complex")
await admin.drop_table(table_path, ignore_if_not_exists=True)
schema = fluss.Schema(
pa.schema(
[
- pa.field("id", pa.int32()),
- pa.field("arr_bytes", pa.list_(pa.binary())),
- pa.field("arr_date", pa.list_(pa.date32())),
- pa.field("arr_time", pa.list_(pa.time32("ms"))),
- pa.field("arr_ts_ntz", pa.list_(pa.timestamp("us"))),
- pa.field("arr_ts_ltz", pa.list_(pa.timestamp("us", tz="UTC"))),
- pa.field("arr_decimal", pa.list_(pa.decimal128(10, 2))),
- pa.field("arr_long_str", pa.list_(pa.string())),
- pa.field("arr_big_decimal", pa.list_(pa.decimal128(22, 5))),
- pa.field("arr_ts_nano", pa.list_(pa.timestamp("ns"))),
- pa.field("arr_float", pa.list_(pa.float32())),
- pa.field("arr_double", pa.list_(pa.float64())),
- # TODO(fluss-python#524): support PyArrow FixedSizeBinary in
schema
- # conversion. Then switch to pa.binary(4).
- pa.field("arr_binary", pa.list_(pa.binary())),
+ pa.field("region", pa.string()),
+ pa.field("user_id", pa.int32()),
+ pa.field(
+ "nested", pa.struct([("seq", pa.int32()), ("label",
pa.string())])
+ ),
+ pa.field("attrs", pa.map_(pa.string(), pa.int32())),
+ pa.field("tags", pa.list_(pa.string())),
]
),
- primary_keys=["id"],
+ primary_keys=["region", "user_id"],
)
- table_descriptor = fluss.TableDescriptor(schema)
- await admin.create_table(table_path, table_descriptor,
ignore_if_exists=False)
+ await admin.create_table(
+ table_path,
+ fluss.TableDescriptor(schema, partition_keys=["region"]),
+ ignore_if_exists=False,
+ )
+ for region in ["US", "EU"]:
+ await admin.create_partition(
+ table_path, {"region": region}, ignore_if_exists=True
+ )
table = await connection.get_table(table_path)
- upsert_writer = table.new_upsert().create_writer()
+ writer = table.new_upsert().create_writer()
+ await _upsert_and_wait(
+ writer,
+ {
+ "region": "US",
+ "user_id": 1,
+ "nested": {"seq": 11, "label": "a"},
+ "attrs": {"x": 1},
+ "tags": ["alpha"],
+ },
+ )
+ await _upsert_and_wait(
+ writer,
+ {
+ "region": "EU",
+ "user_id": 1,
+ "nested": {"seq": 22, "label": "b"},
+ "attrs": {"y": 2},
+ "tags": ["beta"],
+ },
+ )
+ lookuper = table.new_lookup().create_lookuper()
+ r = await lookuper.lookup({"region": "US", "user_id": 1})
+ assert r["nested"] == {"seq": 11, "label": "a"}
+ assert dict(r["attrs"]) == {"x": 1}
+ assert r["tags"] == ["alpha"]
+ r = await lookuper.lookup({"region": "EU", "user_id": 1})
+ assert r["nested"] == {"seq": 22, "label": "b"}
+ assert dict(r["attrs"]) == {"y": 2}
+ assert r["tags"] == ["beta"]
+
+ # Update complex columns within one partition; sibling partition
unaffected.
await _upsert_and_wait(
- upsert_writer,
+ writer,
{
- "id": 1,
- "arr_bytes": [b"\x10\x20\x30", None],
- "arr_date": [date(2026, 1, 23), None],
- "arr_time": [dt_time(10, 13, 47, 123000), None],
- "arr_ts_ntz": [datetime(2026, 1, 23, 10, 13, 47, 123000)],
- "arr_ts_ltz": [
- datetime(2026, 1, 23, 10, 13, 47, 123000, tzinfo=timezone.utc)
- ],
- "arr_decimal": [Decimal("123.45"), None],
- "arr_long_str": [
- "abcdefgh",
- "this is a much longer string that definitely exceeds inline",
- ],
- "arr_big_decimal": [
- Decimal("12345678901234567.12345"),
- Decimal("-99999999999999999.99999"),
- ],
- "arr_ts_nano": [datetime(2026, 1, 23, 10, 13, 47, 123456)],
- "arr_float": [float("nan"), float("inf"), float("-inf")],
- "arr_double": [float("nan"), float("inf"), float("-inf")],
- "arr_binary": [b"\xde\xad\xbe\xef", b"\x00\x01\x02\x03"],
+ "region": "US",
+ "user_id": 1,
+ "nested": {"seq": 99, "label": "updated"},
+ "attrs": {"z": 99},
+ "tags": ["renamed"],
},
)
+ r = await lookuper.lookup({"region": "US", "user_id": 1})
+ assert r["nested"] == {"seq": 99, "label": "updated"}
+ assert dict(r["attrs"]) == {"z": 99}
+ assert r["tags"] == ["renamed"]
+ r = await lookuper.lookup({"region": "EU", "user_id": 1})
+ assert r["nested"] == {"seq": 22, "label": "b"}
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_all_complex_datatypes(connection, admin):
+ """Comprehensive ARRAY/MAP/ROW coverage via upsert+lookup: full, edge,
null rows.
+
+ Mirrors the section-based ``all_supported_datatypes`` structure of the Rust
+ integration tests: the schema is composed from the shared array/row/map
+ sections, and three rows exercise fully-populated, edge-case, and all-null
+ values (see complex_types.py).
+ """
+ table_path = fluss.TablePath("fluss", "py_test_kv_all_complex")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = complex_schema(primary_keys=["id"])
+ await admin.create_table(
+ table_path, fluss.TableDescriptor(schema), ignore_if_exists=False
+ )
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+ await _upsert_and_wait(upsert_writer, complex_full_row(1))
+ await _upsert_and_wait(upsert_writer, complex_edge_row(2))
+ await _upsert_and_wait(upsert_writer, complex_null_row(3))
lookuper = table.new_lookup().create_lookuper()
- result = await lookuper.lookup({"id": 1})
- assert result is not None
- assert result["arr_bytes"] == [b"\x10\x20\x30", None]
- assert result["arr_date"] == [date(2026, 1, 23), None]
- assert result["arr_time"] == [dt_time(10, 13, 47, 123000), None]
- assert result["arr_ts_ntz"] == [datetime(2026, 1, 23, 10, 13, 47, 123000)]
- assert result["arr_ts_ltz"] == [
- datetime(2026, 1, 23, 10, 13, 47, 123000, tzinfo=timezone.utc)
- ]
- assert result["arr_decimal"] == [Decimal("123.45"), None]
- assert result["arr_long_str"] == [
- "abcdefgh",
- "this is a much longer string that definitely exceeds inline",
- ]
- assert result["arr_big_decimal"] == [
- Decimal("12345678901234567.12345"),
- Decimal("-99999999999999999.99999"),
- ]
- assert result["arr_ts_nano"] == [datetime(2026, 1, 23, 10, 13, 47, 123456)]
- _assert_float_specials(result["arr_float"])
- _assert_float_specials(result["arr_double"])
- assert result["arr_binary"] == [b"\xde\xad\xbe\xef", b"\x00\x01\x02\x03"]
+ r1 = await lookuper.lookup({"id": 1})
+ assert r1 is not None
+ assert_complex_full(r1)
+
+ r2 = await lookuper.lookup({"id": 2})
+ assert r2 is not None
+ assert_complex_edge(r2)
+
+ r3 = await lookuper.lookup({"id": 3})
+ assert r3 is not None
+ for col in complex_column_names():
+ assert r3[col] is None, f"{col} should be null"
await admin.drop_table(table_path, ignore_if_not_exists=False)
@@ -677,7 +774,9 @@ async def test_prefix_lookup_validation_errors(connection,
admin):
# Partitioned table: lookup columns must include partition keys first,
# followed by bucket keys.
- partitioned_table_path = fluss.TablePath("fluss",
"py_test_prefix_lookup_validation_pt")
+ partitioned_table_path = fluss.TablePath(
+ "fluss", "py_test_prefix_lookup_validation_pt"
+ )
await admin.drop_table(partitioned_table_path, ignore_if_not_exists=True)
partitioned_schema = fluss.Schema(
@@ -708,13 +807,19 @@ async def
test_prefix_lookup_validation_errors(connection, admin):
# A non-existent partition returns empty list.
partitioned_prefix_lookuper = (
- partitioned_table.new_lookup().lookup_by(["region",
"user_id"]).create_lookuper()
+ partitioned_table.new_lookup()
+ .lookup_by(["region", "user_id"])
+ .create_lookuper()
+ )
+ rows = await partitioned_prefix_lookuper.lookup(
+ {"region": "UNKNOWN_REGION", "user_id": 1}
)
- rows = await partitioned_prefix_lookuper.lookup({"region":
"UNKNOWN_REGION", "user_id": 1})
assert rows == []
# After partition keys, remaining columns must equal bucket keys.
with pytest.raises(fluss.FlussError, match="bucket keys"):
- partitioned_table.new_lookup().lookup_by(["region",
"event_id"]).create_lookuper()
+ partitioned_table.new_lookup().lookup_by(
+ ["region", "event_id"]
+ ).create_lookuper()
await admin.drop_table(partitioned_table_path, ignore_if_not_exists=False)
diff --git a/bindings/python/test/test_log_table.py
b/bindings/python/test/test_log_table.py
index 50b9078b..b6bee545 100644
--- a/bindings/python/test/test_log_table.py
+++ b/bindings/python/test/test_log_table.py
@@ -25,6 +25,15 @@ import time
import pyarrow as pa
import pytest
+from conftest import (
+ assert_complex_edge,
+ assert_complex_full,
+ complex_column_names,
+ complex_edge_row,
+ complex_full_row,
+ complex_null_row,
+ complex_schema,
+)
import fluss
@@ -37,9 +46,7 @@ async def test_append_and_scan(connection, admin):
schema = fluss.Schema(
pa.schema([pa.field("c1", pa.int32()), pa.field("c2", pa.string())])
)
- table_descriptor = fluss.TableDescriptor(
- schema, bucket_count=3, bucket_keys=["c1"]
- )
+ table_descriptor = fluss.TableDescriptor(schema, bucket_count=3,
bucket_keys=["c1"])
await admin.create_table(table_path, table_descriptor,
ignore_if_exists=False)
table = await connection.get_table(table_path)
@@ -266,6 +273,58 @@ async def test_project(connection, admin):
await admin.drop_table(table_path, ignore_if_not_exists=False)
+async def test_project_compound_types(connection, admin):
+ """Projection selects and reorders ROW/MAP/ARRAY columns and drops others
+ (mirrors the Rust projection_with_compound_types IT)."""
+ table_path = fluss.TablePath("fluss", "py_test_project_compound")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field(
+ "nested", pa.struct([("seq", pa.int32()), ("label",
pa.string())])
+ ),
+ pa.field("attrs", pa.map_(pa.string(), pa.int32())),
+ pa.field("tags", pa.list_(pa.string())),
+ pa.field("extra", pa.string()),
+ ]
+ )
+ )
+ await admin.create_table(
+ table_path, fluss.TableDescriptor(schema), ignore_if_exists=False
+ )
+ table = await connection.get_table(table_path)
+
+ aw = table.new_append().create_writer()
+ aw.append(
+ {
+ "id": 7,
+ "nested": {"seq": 42, "label": "hello"},
+ "attrs": {"x": 1, "y": 2},
+ "tags": ["alpha", "beta"],
+ "extra": "ignore-me",
+ }
+ )
+ await aw.flush()
+
+ # Reorder and drop `extra`.
+ scan = table.new_scan().project_by_name(["nested", "attrs", "tags", "id"])
+ scanner = await scan.create_log_scanner()
+ scanner.subscribe_buckets({0: fluss.EARLIEST_OFFSET})
+ records = await _poll_records(scanner, expected_count=1)
+ assert len(records) == 1
+ row = records[0].row
+ assert row["nested"] == {"seq": 42, "label": "hello"}
+ assert dict(row["attrs"]) == {"x": 1, "y": 2}
+ assert row["tags"] == ["alpha", "beta"]
+ assert row["id"] == 7
+ assert "extra" not in row
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
async def test_poll_batches(connection, admin, wait_for_table_ready):
"""Test batch-based scanning with poll_arrow and poll_record_batch."""
table_path = fluss.TablePath("fluss", "py_test_poll_batches")
@@ -334,9 +393,7 @@ async def test_poll_batches(connection, admin,
wait_for_table_ready):
# Projection with batch scanner
proj_scanner = (
- await table.new_scan()
- .project_by_name(["id"])
- .create_record_batch_log_scanner()
+ await
table.new_scan().project_by_name(["id"]).create_record_batch_log_scanner()
)
proj_scanner.subscribe(bucket_id=0, start_offset=0)
batches = await proj_scanner.poll_record_batch(10000)
@@ -623,7 +680,9 @@ async def test_partitioned_table_append_scan(connection,
admin, wait_for_table_r
while len(all_records) < 8 and time.monotonic() < deadline:
scan_records = await scanner.poll(5000)
for bucket, bucket_records in scan_records.items():
- assert bucket.partition_id is not None, "Partitioned table should
have partition_id"
+ assert bucket.partition_id is not None, (
+ "Partitioned table should have partition_id"
+ )
# All records in a bucket should belong to the same partition
regions = {r.row["region"] for r in bucket_records}
assert len(regions) == 1, f"Bucket has mixed regions: {regions}"
@@ -801,9 +860,13 @@ async def
test_scan_records_indexing_and_slicing(connection, admin):
writer = table.new_append().create_writer()
writer.write_arrow_batch(
pa.RecordBatch.from_arrays(
- [pa.array(list(range(1, 9)), type=pa.int32()),
- pa.array([f"v{i}" for i in range(1, 9)])],
- schema=pa.schema([pa.field("id", pa.int32()), pa.field("val",
pa.string())]),
+ [
+ pa.array(list(range(1, 9)), type=pa.int32()),
+ pa.array([f"v{i}" for i in range(1, 9)]),
+ ],
+ schema=pa.schema(
+ [pa.field("id", pa.int32()), pa.field("val", pa.string())]
+ ),
)
)
await writer.flush()
@@ -832,13 +895,13 @@ async def
test_scan_records_indexing_and_slicing(connection, admin):
# Verify slices match the same operation on the offsets reference list
test_slices = [
- slice(1, n - 1), # forward subrange
- slice(None, None, -1), # [::-1] full reverse
- slice(n - 2, 0, -1), # reverse with bounds
- slice(n - 1, 0, -2), # reverse with step
- slice(None, None, 2), # [::2]
- slice(1, None, 3), # [1::3]
- slice(2, 2), # empty
+ slice(1, n - 1), # forward subrange
+ slice(None, None, -1), # [::-1] full reverse
+ slice(n - 2, 0, -1), # reverse with bounds
+ slice(n - 1, 0, -2), # reverse with step
+ slice(None, None, 2), # [::2]
+ slice(1, None, 3), # [1::3]
+ slice(2, 2), # empty
]
for s in test_slices:
result = [r.offset for r in sr[s]]
@@ -863,13 +926,17 @@ async def test_async_iterator(connection, admin):
table = await connection.get_table(table_path)
writer = table.new_append().create_writer()
-
+
# Write 5 records
writer.write_arrow_batch(
pa.RecordBatch.from_arrays(
- [pa.array(list(range(1, 6)), type=pa.int32()),
- pa.array([f"async{i}" for i in range(1, 6)])],
- schema=pa.schema([pa.field("id", pa.int32()), pa.field("val",
pa.string())]),
+ [
+ pa.array(list(range(1, 6)), type=pa.int32()),
+ pa.array([f"async{i}" for i in range(1, 6)]),
+ ],
+ schema=pa.schema(
+ [pa.field("id", pa.int32()), pa.field("val", pa.string())]
+ ),
)
)
await writer.flush()
@@ -879,18 +946,18 @@ async def test_async_iterator(connection, admin):
scanner.subscribe_buckets({i: fluss.EARLIEST_OFFSET for i in
range(num_buckets)})
collected = []
-
+
# Here is the magical Issue #424 async iterator logic at work:
async def consume_scanner():
async for record in scanner:
collected.append(record)
if len(collected) == 5:
break
-
+
await consume_scanner()
-
+
assert len(collected) == 5, f"Expected 5 records, got {len(collected)}"
-
+
collected.sort(key=lambda r: r.row["id"])
for i, record in enumerate(collected):
assert record.row["id"] == i + 1
@@ -932,9 +999,7 @@ async def test_async_iterator_break_no_leak(connection,
admin):
scanner = await table.new_scan().create_log_scanner()
num_buckets = (await admin.get_table_info(table_path)).num_buckets
- scanner.subscribe_buckets(
- {i: fluss.EARLIEST_OFFSET for i in range(num_buckets)}
- )
+ scanner.subscribe_buckets({i: fluss.EARLIEST_OFFSET for i in
range(num_buckets)})
# Phase 1: async for with early break (collect only 3 of 10)
collected_async = []
@@ -987,12 +1052,8 @@ async def
test_async_iterator_multiple_batches(connection, admin):
schema = fluss.Schema(
pa.schema([pa.field("id", pa.int32()), pa.field("val", pa.string())])
)
- table_descriptor = fluss.TableDescriptor(
- schema, bucket_count=3, bucket_keys=["id"]
- )
- await admin.create_table(
- table_path, table_descriptor, ignore_if_exists=False
- )
+ table_descriptor = fluss.TableDescriptor(schema, bucket_count=3,
bucket_keys=["id"])
+ await admin.create_table(table_path, table_descriptor,
ignore_if_exists=False)
table = await connection.get_table(table_path)
writer = table.new_append().create_writer()
@@ -1013,9 +1074,7 @@ async def
test_async_iterator_multiple_batches(connection, admin):
scanner = await table.new_scan().create_log_scanner()
num_buckets = (await admin.get_table_info(table_path)).num_buckets
- scanner.subscribe_buckets(
- {i: fluss.EARLIEST_OFFSET for i in range(num_buckets)}
- )
+ scanner.subscribe_buckets({i: fluss.EARLIEST_OFFSET for i in
range(num_buckets)})
collected = []
@@ -1179,12 +1238,8 @@ async def
test_batch_async_iterator_multiple_batches(connection, admin):
schema = fluss.Schema(
pa.schema([pa.field("id", pa.int32()), pa.field("val", pa.string())])
)
- table_descriptor = fluss.TableDescriptor(
- schema, bucket_count=3, bucket_keys=["id"]
- )
- await admin.create_table(
- table_path, table_descriptor, ignore_if_exists=False
- )
+ table_descriptor = fluss.TableDescriptor(schema, bucket_count=3,
bucket_keys=["id"])
+ await admin.create_table(table_path, table_descriptor,
ignore_if_exists=False)
table = await connection.get_table(table_path)
writer = table.new_append().create_writer()
@@ -1342,8 +1397,6 @@ async def test_append_and_scan_with_array(connection,
admin):
]
-
-
async def test_append_rows_with_array(connection, admin):
"""Test appending rows with array data as Python lists and scanning."""
table_path = fluss.TablePath("fluss", "py_test_append_rows_with_array")
@@ -1368,7 +1421,7 @@ async def test_append_rows_with_array(connection, admin):
append_writer.append({"id": 2, "tags": ["c"], "scores": [30]})
# Append row using list with nested list (null handling)
append_writer.append([3, None, [40, None, 60]])
-
+
await append_writer.flush()
scanner = await table.new_scan().create_log_scanner()
@@ -1394,12 +1447,16 @@ async def
test_append_rows_with_nested_array(connection, admin):
table_path = fluss.TablePath("fluss",
"py_test_append_rows_with_nested_array")
await admin.drop_table(table_path, ignore_if_not_exists=True)
- pa_schema = pa.schema([
- pa.field("id", pa.int32()),
- pa.field("matrix", pa.list_(pa.list_(pa.int32()))),
- ])
+ pa_schema = pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("matrix", pa.list_(pa.list_(pa.int32()))),
+ ]
+ )
schema = fluss.Schema(pa_schema)
- await admin.create_table(table_path, fluss.TableDescriptor(schema),
ignore_if_exists=False)
+ await admin.create_table(
+ table_path, fluss.TableDescriptor(schema), ignore_if_exists=False
+ )
table = await connection.get_table(table_path)
append_writer = table.new_append().create_writer()
@@ -1410,7 +1467,7 @@ async def test_append_rows_with_nested_array(connection,
admin):
append_writer.append({"id": 3, "matrix": None})
append_writer.append({"id": 4, "matrix": [[1, None], None, []]})
append_writer.append({"id": 5, "matrix": [[None, None]]})
-
+
await append_writer.flush()
scanner = await table.new_scan().create_log_scanner()
@@ -1435,12 +1492,16 @@ async def
test_append_rows_with_invalid_array(connection, admin):
table_path = fluss.TablePath("fluss",
"py_test_append_rows_with_invalid_array")
await admin.drop_table(table_path, ignore_if_not_exists=True)
- pa_schema = pa.schema([
- pa.field("id", pa.int32()),
- pa.field("tags", pa.list_(pa.string())),
- ])
+ pa_schema = pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("tags", pa.list_(pa.string())),
+ ]
+ )
schema = fluss.Schema(pa_schema)
- await admin.create_table(table_path, fluss.TableDescriptor(schema),
ignore_if_exists=False)
+ await admin.create_table(
+ table_path, fluss.TableDescriptor(schema), ignore_if_exists=False
+ )
table = await connection.get_table(table_path)
append_writer = table.new_append().create_writer()
@@ -1448,5 +1509,119 @@ async def
test_append_rows_with_invalid_array(connection, admin):
# Appending a string instead of a list should raise an error
with pytest.raises(Exception, match="Expected sequence for Array column"):
append_writer.append({"id": 4, "tags": "not_a_list"})
-
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_all_complex_datatypes(connection, admin):
+ """Comprehensive ARRAY/MAP/ROW coverage on a log table.
+
+ Appends fully-populated, edge-case, and all-null rows (see
complex_types.py)
+ and verifies them through both the record (dict) scan path and the Arrow
+ scan path -- the two paths must agree. Mirrors the section-based
+ ``all_supported_datatypes`` structure of the Rust integration tests.
+ """
+ table_path = fluss.TablePath("fluss", "py_test_log_all_complex")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = complex_schema()
+ await admin.create_table(
+ table_path, fluss.TableDescriptor(schema), ignore_if_exists=False
+ )
+
+ table = await connection.get_table(table_path)
+ append_writer = table.new_append().create_writer()
+ append_writer.append(complex_full_row(1))
+ append_writer.append(complex_edge_row(2))
+ append_writer.append(complex_null_row(3))
+ await append_writer.flush()
+
+ # Record (dict) scan path.
+ scanner = await table.new_scan().create_log_scanner()
+ scanner.subscribe_buckets({0: fluss.EARLIEST_OFFSET})
+ records = await _poll_records(scanner, expected_count=3)
+ assert len(records) == 3
+ poll_rows = sorted([r.row for r in records], key=lambda r: r["id"])
+ assert_complex_full(poll_rows[0])
+ assert_complex_edge(poll_rows[1])
+ for col in complex_column_names():
+ assert poll_rows[2][col] is None
+
+ # Arrow scan path: to_pylist() agrees with the dict path column-for-column,
+ # except NaN-bearing float columns (NaN != NaN), which are checked
elsewhere.
+ scanner2 = await table.new_scan().create_record_batch_log_scanner()
+ scanner2.subscribe_buckets({0: fluss.EARLIEST_OFFSET})
+ result_table = await scanner2.to_arrow()
+ assert result_table.num_rows == 3
+ arrow_rows = result_table.sort_by("id").to_pylist()
+ float_cols = {"arr_float", "arr_double", "map_float", "map_double"}
+ for i in range(3):
+ for col in complex_column_names():
+ if col in float_cols:
+ continue
+ assert arrow_rows[i][col] == poll_rows[i][col], (
+ f"scan-path mismatch at row {i}, column {col}"
+ )
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_append_arrow_batch_complex_types(connection, admin):
+ """Arrow write path: write MAP and ROW columns via write_arrow_batch and
+ verify through both the record and Arrow scan paths."""
+ table_path = fluss.TablePath("fluss", "py_test_arrow_batch_complex")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ row_type = pa.struct([("seq", pa.int32()), ("label", pa.string())])
+ map_type = pa.map_(pa.string(), pa.int32())
+ pa_schema = pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("attrs", map_type),
+ pa.field("nested", row_type),
+ ]
+ )
+ schema = fluss.Schema(pa_schema)
+ await admin.create_table(
+ table_path, fluss.TableDescriptor(schema), ignore_if_exists=False
+ )
+
+ table = await connection.get_table(table_path)
+ append_writer = table.new_append().create_writer()
+ batch = pa.RecordBatch.from_arrays(
+ [
+ pa.array([1, 2], type=pa.int32()),
+ pa.array([[("a", 1), ("b", 2)], []], type=map_type),
+ pa.array(
+ [{"seq": 10, "label": "open"}, {"seq": 20, "label": None}],
+ type=row_type,
+ ),
+ ],
+ schema=pa_schema,
+ )
+ append_writer.write_arrow_batch(batch)
+ await append_writer.flush()
+
+ # Record scan path.
+ scanner = await table.new_scan().create_log_scanner()
+ scanner.subscribe_buckets({0: fluss.EARLIEST_OFFSET})
+ records = await _poll_records(scanner, expected_count=2)
+ assert len(records) == 2
+ rows = sorted([r.row for r in records], key=lambda r: r["id"])
+ assert dict(rows[0]["attrs"]) == {"a": 1, "b": 2}
+ assert rows[0]["nested"] == {"seq": 10, "label": "open"}
+ assert rows[1]["attrs"] == []
+ assert rows[1]["nested"] == {"seq": 20, "label": None}
+
+ # Arrow scan path.
+ scanner2 = await table.new_scan().create_record_batch_log_scanner()
+ scanner2.subscribe_buckets({0: fluss.EARLIEST_OFFSET})
+ result_table = await scanner2.to_arrow()
+ assert result_table.column("id").to_pylist() == [1, 2]
+ assert result_table.column("attrs").to_pylist() == [[("a", 1), ("b", 2)],
[]]
+ assert result_table.column("nested").to_pylist() == [
+ {"seq": 10, "label": "open"},
+ {"seq": 20, "label": None},
+ ]
+
await admin.drop_table(table_path, ignore_if_not_exists=False)
diff --git a/bindings/python/test/test_schema.py
b/bindings/python/test/test_schema.py
index dfd9cf56..1b1f8c81 100644
--- a/bindings/python/test/test_schema.py
+++ b/bindings/python/test/test_schema.py
@@ -23,10 +23,12 @@ import fluss
def test_get_primary_keys():
- fields = pa.schema([
- pa.field("id", pa.int32()),
- pa.field("name", pa.string()),
- ])
+ fields = pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("name", pa.string()),
+ ]
+ )
schema_with_pk = fluss.Schema(fields, primary_keys=["id"])
assert schema_with_pk.get_primary_keys() == ["id"]
@@ -95,3 +97,54 @@ def test_nested_list_nullability():
assert types[2] == "array<int NOT NULL> NOT NULL"
+def test_schema_with_map():
+ # PyArrow models a map as Map(entries: struct<key, value>); Arrow map keys
+ # are always non-nullable, while the value is nullable by default.
+ fields = pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("attrs", pa.map_(pa.string(), pa.int32())),
+ ]
+ )
+ schema = fluss.Schema(fields)
+ assert schema.get_column_names() == ["id", "attrs"]
+ assert schema.get_column_types() == ["int", "map<string NOT NULL,int>"]
+
+
+def test_schema_with_row():
+ fields = pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field(
+ "nested",
+ pa.struct([("seq", pa.int32()), ("label", pa.string())]),
+ ),
+ ]
+ )
+ schema = fluss.Schema(fields)
+ assert schema.get_column_names() == ["id", "nested"]
+ assert schema.get_column_types() == ["int", "row<seq: int, label: string>"]
+
+
+def test_schema_with_nested_complex_types():
+ fields = pa.schema(
+ [
+ # map<string, row<seq int, label string>>
+ pa.field(
+ "m_of_row",
+ pa.map_(
+ pa.string(),
+ pa.struct([("seq", pa.int32()), ("label", pa.string())]),
+ ),
+ ),
+ # array<map<string, int>>
+ pa.field("arr_of_map", pa.list_(pa.map_(pa.string(), pa.int32()))),
+ # row containing an array column
+ pa.field("row_with_arr", pa.struct([("ids",
pa.list_(pa.int32()))])),
+ ]
+ )
+ schema = fluss.Schema(fields)
+ types = schema.get_column_types()
+ assert types[0] == "map<string NOT NULL,row<seq: int, label: string>>"
+ assert types[1] == "array<map<string NOT NULL,int>>"
+ assert types[2] == "row<ids: array<int>>"
diff --git a/crates/fluss/src/record/arrow.rs b/crates/fluss/src/record/arrow.rs
index d2127cbc..5bd75924 100644
--- a/crates/fluss/src/record/arrow.rs
+++ b/crates/fluss/src/record/arrow.rs
@@ -1196,11 +1196,20 @@ pub fn to_arrow_type(fluss_type: &DataType) ->
Result<ArrowDataType> {
});
}
},
+ // TIMESTAMP_LTZ is an instant, so it carries the UTC zone. This keeps
+ // `to_arrow_type` symmetric with `from_arrow_type` (which treats a
+ // zoned Arrow timestamp as LTZ) so the type round-trips losslessly.
DataType::TimestampLTz(timestamp_ltz_type) => match
timestamp_ltz_type.precision() {
- 0 => ArrowDataType::Timestamp(arrow_schema::TimeUnit::Second,
None),
- 1..=3 =>
ArrowDataType::Timestamp(arrow_schema::TimeUnit::Millisecond, None),
- 4..=6 =>
ArrowDataType::Timestamp(arrow_schema::TimeUnit::Microsecond, None),
- 7..=9 =>
ArrowDataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, None),
+ 0 => ArrowDataType::Timestamp(arrow_schema::TimeUnit::Second,
Some("UTC".into())),
+ 1..=3 => {
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Millisecond,
Some("UTC".into()))
+ }
+ 4..=6 => {
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Microsecond,
Some("UTC".into()))
+ }
+ 7..=9 => {
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Nanosecond,
Some("UTC".into()))
+ }
invalid => {
return Err(Error::IllegalArgument {
message: format!(
@@ -1264,7 +1273,7 @@ pub fn to_arrow_type(fluss_type: &DataType) ->
Result<ArrowDataType> {
/// Like `from_arrow_type`, but also reads the Field's nullability —
/// Arrow stores it on the Field wrapper, not the leaf data type.
-pub(crate) fn from_arrow_field(field: &arrow_schema::Field) ->
Result<DataType> {
+pub fn from_arrow_field(field: &arrow_schema::Field) -> Result<DataType> {
let mut dt = from_arrow_type(field.data_type())?;
if !field.is_nullable() {
dt = dt.as_non_nullable();
@@ -1283,11 +1292,16 @@ pub(crate) fn from_arrow_type(arrow_type:
&ArrowDataType) -> Result<DataType> {
ArrowDataType::Int16 => DataTypes::smallint(),
ArrowDataType::Int32 => DataTypes::int(),
ArrowDataType::Int64 => DataTypes::bigint(),
+ // No unsigned types in Fluss; map to the signed type of the same
width.
+ ArrowDataType::UInt8 => DataTypes::tinyint(),
+ ArrowDataType::UInt16 => DataTypes::smallint(),
+ ArrowDataType::UInt32 => DataTypes::int(),
+ ArrowDataType::UInt64 => DataTypes::bigint(),
ArrowDataType::Float32 => DataTypes::float(),
ArrowDataType::Float64 => DataTypes::double(),
- ArrowDataType::Utf8 => DataTypes::string(),
- ArrowDataType::Binary => DataTypes::bytes(),
- ArrowDataType::Date32 => DataTypes::date(),
+ ArrowDataType::Utf8 | ArrowDataType::LargeUtf8 => DataTypes::string(),
+ ArrowDataType::Binary | ArrowDataType::LargeBinary =>
DataTypes::bytes(),
+ ArrowDataType::Date32 | ArrowDataType::Date64 => DataTypes::date(),
ArrowDataType::FixedSizeBinary(len) => {
if *len < 0 {
return Err(Error::IllegalArgument {
@@ -1784,19 +1798,19 @@ mod tests {
);
assert_eq!(
to_arrow_type(&DataTypes::timestamp_ltz_with_precision(0)).unwrap(),
- ArrowDataType::Timestamp(arrow_schema::TimeUnit::Second, None)
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Second,
Some("UTC".into()))
);
assert_eq!(
to_arrow_type(&DataTypes::timestamp_ltz_with_precision(3)).unwrap(),
- ArrowDataType::Timestamp(arrow_schema::TimeUnit::Millisecond, None)
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Millisecond,
Some("UTC".into()))
);
assert_eq!(
to_arrow_type(&DataTypes::timestamp_ltz_with_precision(6)).unwrap(),
- ArrowDataType::Timestamp(arrow_schema::TimeUnit::Microsecond, None)
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Microsecond,
Some("UTC".into()))
);
assert_eq!(
to_arrow_type(&DataTypes::timestamp_ltz_with_precision(9)).unwrap(),
- ArrowDataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, None)
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Nanosecond,
Some("UTC".into()))
);
assert_eq!(
to_arrow_type(&DataTypes::bytes()).unwrap(),
@@ -1882,6 +1896,38 @@ mod tests {
}
}
+ #[test]
+ fn test_from_arrow_type_accepts_unsigned_large_and_date64() {
+ assert!(matches!(
+ from_arrow_type(&ArrowDataType::UInt8).unwrap(),
+ DataType::TinyInt(_)
+ ));
+ assert!(matches!(
+ from_arrow_type(&ArrowDataType::UInt16).unwrap(),
+ DataType::SmallInt(_)
+ ));
+ assert!(matches!(
+ from_arrow_type(&ArrowDataType::UInt32).unwrap(),
+ DataType::Int(_)
+ ));
+ assert!(matches!(
+ from_arrow_type(&ArrowDataType::UInt64).unwrap(),
+ DataType::BigInt(_)
+ ));
+ assert!(matches!(
+ from_arrow_type(&ArrowDataType::LargeUtf8).unwrap(),
+ DataType::String(_)
+ ));
+ assert!(matches!(
+ from_arrow_type(&ArrowDataType::LargeBinary).unwrap(),
+ DataType::Bytes(_)
+ ));
+ assert!(matches!(
+ from_arrow_type(&ArrowDataType::Date64).unwrap(),
+ DataType::Date(_)
+ ));
+ }
+
#[test]
fn test_parse_ipc_message() {
let empty_body: &[u8] = &le_bytes(&[0xFFFFFFFF, 0x00000000]);
diff --git a/crates/fluss/src/row/binary_array.rs
b/crates/fluss/src/row/binary_array.rs
index db15b082..3a43ea77 100644
--- a/crates/fluss/src/row/binary_array.rs
+++ b/crates/fluss/src/row/binary_array.rs
@@ -250,15 +250,24 @@ impl FlussArray {
let mut max_extent = fixed_part_size;
for i in 0..self.size {
- if !self.is_null_at(i) {
- let packed = self.read_i64(i, "extent calculation")? as u64;
- let mark = packed & HIGHEST_FIRST_BIT;
- if mark == 0 {
- let offset = (packed >> 32) as usize;
- let len = (packed & 0xFFFF_FFFF) as usize;
- max_extent = max_extent.max(offset + len);
- }
+ if self.is_null_at(i) {
+ continue;
}
+ let packed = self.read_i64(i, "extent calculation")? as u64;
+ let offset = (packed >> 32) as usize;
+ let var_len = match element_type {
+ // Non-compact timestamps pack `nanoOfMillisecond` in the low
word
+ // (not a byte length) and store a fixed 8-byte millisecond
value
+ // in the variable section (see `write_timestamp_ntz`).
+ DataType::Timestamp(_) | DataType::TimestampLTz(_) =>
round_to_nearest_word(8),
+ _ => {
+ if packed & HIGHEST_FIRST_BIT != 0 {
+ continue;
+ }
+ (packed & 0xFFFF_FFFF) as usize
+ }
+ };
+ max_extent = max_extent.max(offset + var_len);
}
Ok(round_to_nearest_word(max_extent))
diff --git a/crates/fluss/src/row/binary_map.rs
b/crates/fluss/src/row/binary_map.rs
index af6d93a2..9c657788 100644
--- a/crates/fluss/src/row/binary_map.rs
+++ b/crates/fluss/src/row/binary_map.rs
@@ -463,6 +463,7 @@ mod tests {
use super::*;
use crate::metadata::DataTypes;
use crate::row::binary_array::FlussArrayWriter;
+ use crate::row::datum::TimestampNtz;
#[test]
fn fluss_map_dispatches_through_internal_map_trait() {
@@ -511,6 +512,29 @@ mod tests {
assert_eq!(decoded_values.get_string(1).unwrap(), "b");
}
+ #[test]
+ fn test_round_trip_map_with_noncompact_timestamp_value() {
+ // Regression: a non-compact timestamp (precision > 3) packs
+ // nanoOfMillisecond in the low word, which `FlussArray::extent` (used
by
+ // map validation) must not treat as a byte length.
+ let key_type = DataTypes::string();
+ let value_type = DataTypes::timestamp_with_precision(6);
+ let ts = TimestampNtz::from_millis_nanos(1_769_163_227_123,
456_000).unwrap();
+
+ let mut writer = FlussMapWriter::new(2, &key_type, &value_type);
+ writer
+ .write_entry("a".into(), Datum::TimestampNtz(ts))
+ .unwrap();
+ writer.write_entry("b".into(), Datum::Null).unwrap();
+ let map = writer.complete().unwrap();
+
+ // Re-decode from the serialized bytes (the compacted / KV-lookup
path).
+ let decoded = FlussMap::from_bytes(map.as_bytes(), &key_type,
&value_type).unwrap();
+ assert_eq!(decoded.size(), 2);
+ assert_eq!(decoded.value_array().get_timestamp_ntz(0, 6).unwrap(), ts);
+ assert!(decoded.value_array().is_null_at(1));
+ }
+
#[test]
fn test_empty_map() {
let writer = FlussMapWriter::new(0, &DataTypes::int(),
&DataTypes::string());
diff --git a/crates/fluss/src/row/column_writer.rs
b/crates/fluss/src/row/column_writer.rs
index fffbc616..b47ce096 100644
--- a/crates/fluss/src/row/column_writer.rs
+++ b/crates/fluss/src/row/column_writer.rs
@@ -321,25 +321,29 @@ impl ColumnWriter {
ArrowDataType::Timestamp(arrow_schema::TimeUnit::Second,
_) => {
TypedWriter::TimestampLtzSecond {
precision,
- builder:
TimestampSecondBuilder::with_capacity(capacity),
+ builder:
TimestampSecondBuilder::with_capacity(capacity)
+ .with_timezone("UTC"),
}
}
ArrowDataType::Timestamp(arrow_schema::TimeUnit::Millisecond, _) => {
TypedWriter::TimestampLtzMillisecond {
precision,
- builder:
TimestampMillisecondBuilder::with_capacity(capacity),
+ builder:
TimestampMillisecondBuilder::with_capacity(capacity)
+ .with_timezone("UTC"),
}
}
ArrowDataType::Timestamp(arrow_schema::TimeUnit::Microsecond, _) => {
TypedWriter::TimestampLtzMicrosecond {
precision,
- builder:
TimestampMicrosecondBuilder::with_capacity(capacity),
+ builder:
TimestampMicrosecondBuilder::with_capacity(capacity)
+ .with_timezone("UTC"),
}
}
ArrowDataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, _) => {
TypedWriter::TimestampLtzNanosecond {
precision,
- builder:
TimestampNanosecondBuilder::with_capacity(capacity),
+ builder:
TimestampNanosecondBuilder::with_capacity(capacity)
+ .with_timezone("UTC"),
}
}
_ => {
diff --git a/website/docs/user-guide/python/data-types.md
b/website/docs/user-guide/python/data-types.md
index 8e4371e2..bde4e867 100644
--- a/website/docs/user-guide/python/data-types.md
+++ b/website/docs/user-guide/python/data-types.md
@@ -19,6 +19,8 @@ The Python client uses PyArrow types for schema definitions:
| `pa.timestamp("us", tz="UTC")` | TimestampLTZ
| `datetime.datetime` |
| `pa.decimal128(precision, scale)` | Decimal
| `decimal.Decimal` |
| `pa.list_(type)` | Array
| `list` |
+| `pa.map_(key_type, value_type)` | Map
| `list[(key, value)]`|
+| `pa.struct([(name, type), ...])` | Row
| `dict` |
All Python native types (`date`, `time`, `datetime`, `Decimal`) work when
appending rows via dicts.
@@ -71,6 +73,11 @@ row = {
handle = writer.append(row)
```
+When a row is written as a **dict**, a nullable column may be omitted — it
+defaults to `null`. A non-nullable column (including primary keys) must be
+present, otherwise the write is rejected with a clear error. The same rule
+applies to the fields of a `ROW` value.
+
Lists and tuples must have values in column order:
```python
@@ -93,3 +100,82 @@ for record in records:
if row["nickname"] is None:
print("nickname is null")
```
+
+## Complex Types (Array, Map, Row)
+
+`ARRAY`, `MAP`, and `ROW` columns can be nested arbitrarily (for example
+`array<map<string, row<...>>>`). On read they materialize to native Python
+objects; on write they accept the shapes below:
+
+| Fluss type | Read-back value | Write input accepted
|
+|-------------|--------------------------------|---------------------------------------------|
+| `ARRAY<T>` | `list` | `list` / `tuple`
|
+| `MAP<K, V>` | `list` of `(key, value)` tuples| `dict`, or a sequence of
`(key, value)` pairs |
+| `ROW<...>` | `dict` keyed by field name | `dict` (by name) or
`list`/`tuple` (by position) |
+
+The MAP read shape matches pyarrow's `MapArray.to_pylist()` (it preserves
+duplicate keys and ordering); ROW matches `StructArray.to_pylist()`.
+
+### Arrays
+
+```python
+schema = pa.schema([
+ pa.field("id", pa.int32()),
+ pa.field("tags", pa.list_(pa.string())),
+ pa.field("matrix", pa.list_(pa.list_(pa.int32()))), # nested
+])
+writer.append({"id": 1, "tags": ["a", "b"], "matrix": [[1, 2], [3, 4]]})
+
+row = await lookuper.lookup({"id": 1})
+row["tags"] # ["a", "b"]
+row["matrix"] # [[1, 2], [3, 4]]
+```
+
+### Maps
+
+Use `pa.map_(key_type, value_type)`. Write a `dict` or a list of
+`(key, value)` pairs; reads return a list of `(key, value)` tuples (wrap with
+`dict(...)` for keyed access). Map keys must be non-null.
+
+```python
+schema = pa.schema([
+ pa.field("id", pa.int32()),
+ pa.field("attrs", pa.map_(pa.string(), pa.int32())),
+])
+writer.append({"id": 1, "attrs": {"a": 1, "b": None}}) # dict input
+# or a sequence of pairs: {"id": 2, "attrs": [("a", 1), ("b", None)]}
+
+row = await lookuper.lookup({"id": 1})
+row["attrs"] # [("a", 1), ("b", None)]
+dict(row["attrs"]) # {"a": 1, "b": None}
+```
+
+### Rows
+
+Use `pa.struct([...])`. Write a `dict` keyed by field name (or a `list`/`tuple`
+in field order); reads return a `dict`.
+
+```python
+schema = pa.schema([
+ pa.field("id", pa.int32()),
+ pa.field("profile", pa.struct([("age", pa.int32()), ("city",
pa.string())])),
+])
+writer.append({"id": 1, "profile": {"age": 30, "city": "NYC"}})
+
+row = await lookuper.lookup({"id": 1})
+row["profile"] # {"age": 30, "city": "NYC"}
+```
+
+### Constraints
+
+`ARRAY`, `MAP`, and `ROW` may be used as row values and nested inside one
+another, but not as primary-key or bucket-key columns — the server rejects
+complex key types.
+
+### Bulk (Arrow) reads
+
+The per-row paths above (`append`/`upsert` and the record-based scanner's
+`record.row` dict, point `lookup`) materialize each value into a Python object.
+For high-throughput scans, prefer the Arrow path — a record-batch scanner's
+`to_arrow()` / `poll_arrow()` returns nested columns as native pyarrow
+`ListArray` / `MapArray` / `StructArray` with no per-element conversion.
