You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "charlesdong1991 (via GitHub)" <gi...@apache.org> on 2026/05/02 14:38:57 UTC
[PR] [tests] Add integration tests for array data type [fluss-rust]
charlesdong1991 opened a new pull request, #523:
URL: https://github.com/apache/fluss-rust/pull/523
### Purpose
Add integration tests for array data type support in Rust core and Python bindings.
c++ has integration tests covered already, so only focus on rust and python
<!-- Linking this pull request to the issue -->
Linked issue: close #441
<!-- What is the purpose of the change -->
### Tests
All tests have passed locally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#discussion_r3178275102
##########
crates/fluss/tests/integration/kv_table.rs:
##########
@@ -880,6 +896,226 @@ mod kv_table_test {
found_row_nulls.is_null_at(16).unwrap(),
"col_binary should be null"
);
+ assert!(
+ found_row_nulls.is_null_at(17).unwrap(),
+ "col_array should be null"
+ );
+
+ admin
+ .drop_table(&table_path, false)
+ .await
+ .expect("Failed to drop table");
+ }
+
+ #[tokio::test]
+ async fn upsert_and_lookup_with_array() {
Review Comment:
hmm, array test coverage is also added to `all_supported_datatypes` for both logs and kv tests. This test cover the patterns that don't fit `all_supported_datetypes` structure that well IMHO, since these require multi-row setups and different null patterns per row.
And if i understand correctly, `all_supported_datetypes` tests one row with values and one row with all nulls.
If we add Map, i think it's best to follow the same approach as here for Array, so basic column incorporated in `all_supported_datatypes` while specific tests for map-specific edge cases.
But i don't have very strong opinion on this, happy to adjust if you prefer the consolidation! @leekeiabstraction 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#discussion_r3191611184
##########
bindings/python/test/test_kv_table.py:
##########
@@ -335,6 +346,228 @@ 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 array columns in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_basic")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("tags", pa.list_(pa.string())),
+ pa.field("scores", pa.list_(pa.int32())),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ # Row 1: standard arrays
+ await _upsert_and_wait(
+ upsert_writer, {"id": 1, "tags": ["hello", "world"], "scores": [10, 20, 30]}
+ )
+
+ # Row 2: null element in array + empty array
+ await _upsert_and_wait(upsert_writer, {"id": 2, "tags": [None], "scores": []})
+
+ # Row 3: null array column
+ await _upsert_and_wait(upsert_writer, {"id": 3, "tags": None, "scores": [42]})
+
+ lookuper = table.new_lookup().create_lookuper()
+
+ result1 = await lookuper.lookup({"id": 1})
+ assert result1 is not None
+ assert result1["tags"] == ["hello", "world"]
+ assert result1["scores"] == [10, 20, 30]
+
+ result2 = await lookuper.lookup({"id": 2})
+ assert result2 is not None
+ assert result2["tags"] == [None]
+ assert result2["scores"] == []
+
+ result3 = await lookuper.lookup({"id": 3})
+ assert result3 is not None
+ assert result3["tags"] is None
+ assert result3["scores"] == [42]
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_upsert_and_lookup_with_nested_array(connection, admin):
Review Comment:
Can we merge this test case with above ask well?
##########
bindings/python/test/test_kv_table.py:
##########
@@ -335,6 +346,228 @@ 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 array columns in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_basic")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("tags", pa.list_(pa.string())),
+ pa.field("scores", pa.list_(pa.int32())),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ # Row 1: standard arrays
+ await _upsert_and_wait(
+ upsert_writer, {"id": 1, "tags": ["hello", "world"], "scores": [10, 20, 30]}
+ )
+
+ # Row 2: null element in array + empty array
+ await _upsert_and_wait(upsert_writer, {"id": 2, "tags": [None], "scores": []})
+
+ # Row 3: null array column
+ await _upsert_and_wait(upsert_writer, {"id": 3, "tags": None, "scores": [42]})
+
+ lookuper = table.new_lookup().create_lookuper()
+
+ result1 = await lookuper.lookup({"id": 1})
+ assert result1 is not None
+ assert result1["tags"] == ["hello", "world"]
+ assert result1["scores"] == [10, 20, 30]
+
+ result2 = await lookuper.lookup({"id": 2})
+ assert result2 is not None
+ assert result2["tags"] == [None]
+ assert result2["scores"] == []
+
+ result3 = await lookuper.lookup({"id": 3})
+ assert result3 is not None
+ assert result3["tags"] is None
+ assert result3["scores"] == [42]
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_upsert_and_lookup_with_nested_array(connection, admin):
+ """Test upsert and lookup with nested array (ARRAY<ARRAY<INT>>) in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_nested")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("matrix", pa.list_(pa.list_(pa.int32()))),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ await _upsert_and_wait(upsert_writer, {"id": 1, "matrix": [[1, 2], [3, 4]]})
+
+ await _upsert_and_wait(upsert_writer, {"id": 2, "matrix": [[], [5], [6, 7, 8]]})
+
+ await _upsert_and_wait(upsert_writer, {"id": 3, "matrix": None})
+
+ await _upsert_and_wait(upsert_writer, {"id": 4, "matrix": [[1, None], None, []]})
+
+ lookuper = table.new_lookup().create_lookuper()
+
+ result1 = await lookuper.lookup({"id": 1})
+ assert result1 is not None
+ assert result1["matrix"] == [[1, 2], [3, 4]]
+
+ result2 = await lookuper.lookup({"id": 2})
+ assert result2 is not None
+ assert result2["matrix"] == [[], [5], [6, 7, 8]]
+
+ result3 = await lookuper.lookup({"id": 3})
+ assert result3 is not None
+ assert result3["matrix"] is None
+
+ result4 = await lookuper.lookup({"id": 4})
+ assert result4 is not None
+ assert result4["matrix"] == [[1, None], None, []]
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_upsert_and_lookup_with_array_rich_types(connection, admin):
Review Comment:
Same here.
##########
crates/fluss/tests/integration/log_table.rs:
##########
@@ -1373,4 +1384,430 @@ mod table_test {
.await
.expect("Failed to drop table");
}
+
+ #[tokio::test]
+ async fn append_and_scan_with_array() {
+ use fluss::row::binary_array::FlussArrayWriter;
+ use fluss::row::{Datum, FlussArray, GenericRow};
+
+ let cluster = get_shared_cluster();
+ let connection = cluster.get_fluss_connection().await;
+ let admin = connection.get_admin().expect("Failed to get admin");
+
+ let table_path = TablePath::new("fluss", "test_log_arrays");
+ let inner_array_type = DataTypes::array(DataTypes::int());
+
+ let schema = Schema::builder()
+ .column("id", DataTypes::int())
+ .column("tags", DataTypes::array(DataTypes::string()))
+ .column("scores", DataTypes::array(DataTypes::int()))
+ .column("matrix", DataTypes::array(inner_array_type.clone()))
+ .build()
+ .expect("Failed to build schema");
+
+ let table_descriptor = TableDescriptor::builder()
+ .schema(schema)
+ .build()
+ .expect("Failed to build table descriptor");
+
+ create_table(&admin, &table_path, &table_descriptor).await;
+
+ let table = connection
+ .get_table(&table_path)
+ .await
+ .expect("Failed to get table");
+
+ let append_writer = table
+ .new_append()
+ .expect("Failed to create append")
+ .create_writer()
+ .expect("Failed to create writer");
+
+ // Row 1: id=1, tags=["hello", "world"], scores=[10, 20, 30], matrix=[[1,2],[3,4]]
+ let mut row1 = GenericRow::new(4);
+ row1.set_field(0, 1_i32);
+ row1.set_field(1, make_string_array(&[Some("hello"), Some("world")]));
+ row1.set_field(2, make_int_array(&[Some(10), Some(20), Some(30)]));
+ let m1 = {
+ let mut w = FlussArrayWriter::new(2, &inner_array_type);
+ w.write_array(0, &make_int_array(&[Some(1), Some(2)]));
+ w.write_array(1, &make_int_array(&[Some(3), Some(4)]));
+ w.complete().expect("matrix1")
+ };
+ row1.set_field(3, m1);
+
+ // Row 2: id=2, tags=[null], scores=[], matrix=[[5], null, []]
+ let mut row2 = GenericRow::new(4);
+ row2.set_field(0, 2_i32);
+ row2.set_field(1, make_string_array(&[None]));
+ row2.set_field(2, make_int_array(&[]));
+ let m2 = {
+ let mut w = FlussArrayWriter::new(3, &inner_array_type);
+ w.write_array(0, &make_int_array(&[Some(5)]));
+ w.set_null_at(1);
+ w.write_array(2, &make_int_array(&[]));
+ w.complete().expect("matrix2")
+ };
+ row2.set_field(3, m2);
+
+ // Row 3: id=3, tags=null, scores=[42], matrix=null
+ let mut row3 = GenericRow::new(4);
+ row3.set_field(0, 3_i32);
+ row3.set_field(1, Datum::Null);
+ row3.set_field(2, make_int_array(&[Some(42)]));
+ row3.set_field(3, Datum::Null);
+
+ append_writer.append(&row1).expect("append row1");
+ append_writer.append(&row2).expect("append row2");
+ append_writer.append(&row3).expect("append row3");
+ append_writer.flush().await.expect("Failed to flush");
+
+ let records = scan_table(&table, |scan| scan).await;
+ assert_eq!(records.len(), 3, "expected three log records");
+
+ // Verify row 1: populated flat arrays + nested array
+ let r0 = records[0].row();
+ assert_eq!(r0.get_int(0).unwrap(), 1);
+ let tags_r0 = r0.get_array(1).unwrap();
+ assert_eq!(tags_r0.size(), 2);
+ assert_eq!(tags_r0.get_string(0).unwrap(), "hello");
+ assert_eq!(tags_r0.get_string(1).unwrap(), "world");
+ let scores_r0 = r0.get_array(2).unwrap();
+ assert_eq!(scores_r0.size(), 3);
+ assert_eq!(scores_r0.get_int(0).unwrap(), 10);
+ assert_eq!(scores_r0.get_int(1).unwrap(), 20);
+ assert_eq!(scores_r0.get_int(2).unwrap(), 30);
+ let matrix_r0: FlussArray = r0.get_array(3).unwrap();
+ assert_eq!(matrix_r0.size(), 2);
+ let mr0_0 = matrix_r0.get_array(0).unwrap();
+ assert_eq!(mr0_0.size(), 2);
+ assert_eq!(mr0_0.get_int(0).unwrap(), 1);
+ assert_eq!(mr0_0.get_int(1).unwrap(), 2);
+ let mr0_1 = matrix_r0.get_array(1).unwrap();
+ assert_eq!(mr0_1.size(), 2);
+ assert_eq!(mr0_1.get_int(0).unwrap(), 3);
+ assert_eq!(mr0_1.get_int(1).unwrap(), 4);
+
+ // Verify row 2: null element in array, empty array, nested with mixed inner
+ let r1 = records[1].row();
+ assert_eq!(r1.get_int(0).unwrap(), 2);
+ let tags_r1 = r1.get_array(1).unwrap();
+ assert_eq!(tags_r1.size(), 1);
+ assert!(tags_r1.is_null_at(0));
+ let scores_r1 = r1.get_array(2).unwrap();
+ assert_eq!(scores_r1.size(), 0);
+ let matrix_r1 = r1.get_array(3).unwrap();
+ assert_eq!(matrix_r1.size(), 3);
+ let mr1_0 = matrix_r1.get_array(0).unwrap();
+ assert_eq!(mr1_0.size(), 1);
+ assert_eq!(mr1_0.get_int(0).unwrap(), 5);
+ assert!(matrix_r1.is_null_at(1));
+ let mr1_2 = matrix_r1.get_array(2).unwrap();
+ assert_eq!(mr1_2.size(), 0);
+
+ // Verify row 3: null flat column, null nested column
+ let r2 = records[2].row();
+ assert_eq!(r2.get_int(0).unwrap(), 3);
+ assert!(r2.is_null_at(1).unwrap());
+ let scores_r2 = r2.get_array(2).unwrap();
+ assert_eq!(scores_r2.size(), 1);
+ assert_eq!(scores_r2.get_int(0).unwrap(), 42);
+ assert!(r2.is_null_at(3).unwrap());
+
+ admin
+ .drop_table(&table_path, false)
+ .await
+ .expect("Failed to drop table");
+ }
+
+ #[tokio::test]
+ async fn append_and_scan_with_array_rich_types() {
Review Comment:
Same here.
##########
bindings/python/test/test_kv_table.py:
##########
@@ -335,6 +346,228 @@ 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 array columns in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_basic")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("tags", pa.list_(pa.string())),
+ pa.field("scores", pa.list_(pa.int32())),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ # Row 1: standard arrays
+ await _upsert_and_wait(
+ upsert_writer, {"id": 1, "tags": ["hello", "world"], "scores": [10, 20, 30]}
+ )
+
+ # Row 2: null element in array + empty array
+ await _upsert_and_wait(upsert_writer, {"id": 2, "tags": [None], "scores": []})
+
+ # Row 3: null array column
+ await _upsert_and_wait(upsert_writer, {"id": 3, "tags": None, "scores": [42]})
+
+ lookuper = table.new_lookup().create_lookuper()
+
+ result1 = await lookuper.lookup({"id": 1})
+ assert result1 is not None
+ assert result1["tags"] == ["hello", "world"]
+ assert result1["scores"] == [10, 20, 30]
+
+ result2 = await lookuper.lookup({"id": 2})
+ assert result2 is not None
+ assert result2["tags"] == [None]
+ assert result2["scores"] == []
+
+ result3 = await lookuper.lookup({"id": 3})
+ assert result3 is not None
+ assert result3["tags"] is None
+ assert result3["scores"] == [42]
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_upsert_and_lookup_with_nested_array(connection, admin):
+ """Test upsert and lookup with nested array (ARRAY<ARRAY<INT>>) in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_nested")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("matrix", pa.list_(pa.list_(pa.int32()))),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ await _upsert_and_wait(upsert_writer, {"id": 1, "matrix": [[1, 2], [3, 4]]})
+
+ await _upsert_and_wait(upsert_writer, {"id": 2, "matrix": [[], [5], [6, 7, 8]]})
+
+ await _upsert_and_wait(upsert_writer, {"id": 3, "matrix": None})
+
+ await _upsert_and_wait(upsert_writer, {"id": 4, "matrix": [[1, None], None, []]})
+
+ lookuper = table.new_lookup().create_lookuper()
+
+ result1 = await lookuper.lookup({"id": 1})
+ assert result1 is not None
+ assert result1["matrix"] == [[1, 2], [3, 4]]
+
+ result2 = await lookuper.lookup({"id": 2})
+ assert result2 is not None
+ assert result2["matrix"] == [[], [5], [6, 7, 8]]
+
+ result3 = await lookuper.lookup({"id": 3})
+ assert result3 is not None
+ assert result3["matrix"] is None
+
+ result4 = await lookuper.lookup({"id": 4})
+ assert result4 is not None
+ assert result4["matrix"] == [[1, None], None, []]
+
+ 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 in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_rich_types")
+ 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))),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ await _upsert_and_wait(
+ upsert_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],
+ },
+ )
+
+ 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]
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_upsert_and_lookup_with_array_encoding_edge_cases(connection, admin):
Review Comment:
Same here. IMO we can reduce / simplify some of these by merging the array tests into a single test case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#discussion_r3176775020
##########
bindings/python/test/test_kv_table.py:
##########
@@ -335,6 +346,228 @@ 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 array columns in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_basic")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("tags", pa.list_(pa.string())),
+ pa.field("scores", pa.list_(pa.int32())),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ # Row 1: standard arrays
+ await _upsert_and_wait(
+ upsert_writer, {"id": 1, "tags": ["hello", "world"], "scores": [10, 20, 30]}
+ )
+
+ # Row 2: null element in array + empty array
+ await _upsert_and_wait(upsert_writer, {"id": 2, "tags": [None], "scores": []})
+
+ # Row 3: null array column
+ await _upsert_and_wait(upsert_writer, {"id": 3, "tags": None, "scores": [42]})
+
+ lookuper = table.new_lookup().create_lookuper()
+
+ result1 = await lookuper.lookup({"id": 1})
+ assert result1 is not None
+ assert result1["tags"] == ["hello", "world"]
+ assert result1["scores"] == [10, 20, 30]
+
+ result2 = await lookuper.lookup({"id": 2})
+ assert result2 is not None
+ assert result2["tags"] == [None]
+ assert result2["scores"] == []
+
+ result3 = await lookuper.lookup({"id": 3})
+ assert result3 is not None
+ assert result3["tags"] is None
+ assert result3["scores"] == [42]
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_upsert_and_lookup_with_nested_array(connection, admin):
+ """Test upsert and lookup with nested array (ARRAY<ARRAY<INT>>) in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_nested")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("matrix", pa.list_(pa.list_(pa.int32()))),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ await _upsert_and_wait(upsert_writer, {"id": 1, "matrix": [[1, 2], [3, 4]]})
+
+ await _upsert_and_wait(upsert_writer, {"id": 2, "matrix": [[], [5], [6, 7, 8]]})
+
+ await _upsert_and_wait(upsert_writer, {"id": 3, "matrix": None})
+
+ await _upsert_and_wait(upsert_writer, {"id": 4, "matrix": [[1, None], None, []]})
+
+ lookuper = table.new_lookup().create_lookuper()
+
+ result1 = await lookuper.lookup({"id": 1})
+ assert result1 is not None
+ assert result1["matrix"] == [[1, 2], [3, 4]]
+
+ result2 = await lookuper.lookup({"id": 2})
+ assert result2 is not None
+ assert result2["matrix"] == [[], [5], [6, 7, 8]]
+
+ result3 = await lookuper.lookup({"id": 3})
+ assert result3 is not None
+ assert result3["matrix"] is None
+
+ result4 = await lookuper.lookup({"id": 4})
+ assert result4 is not None
+ assert result4["matrix"] == [[1, None], None, []]
+
+ 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 in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_rich_types")
+ 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))),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ await _upsert_and_wait(
+ upsert_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],
+ },
+ )
+
+ 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]
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_upsert_and_lookup_with_array_encoding_edge_cases(connection, admin):
+ """Test array encoding edge cases in KV table upsert/lookup."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_edge_cases")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ 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): support PyArrow FixedSizeBinary in schema conversion.
Review Comment:
i think it is nicer to address in a separate PR, so added a new issue: https://github.com/apache/fluss-rust/issues/524
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#issuecomment-4366084229
> Thanks for your comments, addressed all, and rebased! @fresh-borzoni PTAL 🙏
All good, thank you for addressing 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#discussion_r3177006837
##########
crates/fluss/tests/integration/kv_table.rs:
##########
@@ -20,14 +20,41 @@
mod kv_table_test {
use crate::integration::utils::{create_partitions, create_table, get_shared_cluster};
use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath};
- use fluss::row::{GenericRow, InternalRow};
+ use fluss::row::binary_array::FlussArrayWriter;
+ use fluss::row::{FlussArray, GenericRow, InternalRow};
fn make_key(id: i32) -> GenericRow<'static> {
- let mut row = GenericRow::new(3);
+ make_key_with_len(id, 3)
+ }
+
+ fn make_key_with_len(id: i32, field_count: usize) -> GenericRow<'static> {
Review Comment:
nit: `make_key_with_field_count` feels like a better name
##########
Cargo.lock:
##########
@@ -746,6 +746,15 @@ dependencies = [
"rustc_version",
]
+[[package]]
Review Comment:
we can rebase now with Cargo.lock synced
##########
crates/fluss/tests/integration/log_table.rs:
##########
@@ -1373,4 +1388,546 @@ mod table_test {
.await
.expect("Failed to drop table");
}
+
+ #[tokio::test]
+ async fn append_and_scan_with_array() {
+ use fluss::row::binary_array::FlussArrayWriter;
+ use fluss::row::{Datum, GenericRow};
+
+ let cluster = get_shared_cluster();
+ let connection = cluster.get_fluss_connection().await;
+ let admin = connection.get_admin().expect("Failed to get admin");
+
+ let table_path = TablePath::new("fluss", "test_log_arrays_basic");
+ let schema = Schema::builder()
+ .column("id", DataTypes::int())
+ .column("tags", DataTypes::array(DataTypes::string()))
+ .column("scores", DataTypes::array(DataTypes::int()))
+ .build()
+ .expect("Failed to build schema");
+
+ let table_descriptor = TableDescriptor::builder()
+ .schema(schema)
+ .build()
+ .expect("Failed to build table descriptor");
+
+ create_table(&admin, &table_path, &table_descriptor).await;
+
+ let table = connection
+ .get_table(&table_path)
+ .await
+ .expect("Failed to get table");
+
+ let append_writer = table
+ .new_append()
+ .expect("Failed to create append")
+ .create_writer()
+ .expect("Failed to create writer");
+
+ // Row 1: id=1, tags=["hello", "world"], scores=[10, 20, 30]
+ let mut row1 = GenericRow::new(3);
+ row1.set_field(0, 1_i32);
+
+ let tags1 = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::string());
+ w.write_string(0, "hello");
+ w.write_string(1, "world");
+ w.complete().expect("tags1")
+ };
+ let scores1 = {
+ let mut w = FlussArrayWriter::new(3, &DataTypes::int());
+ w.write_int(0, 10);
+ w.write_int(1, 20);
+ w.write_int(2, 30);
+ w.complete().expect("scores1")
+ };
+ row1.set_field(1, tags1);
+ row1.set_field(2, scores1);
+
+ // Row 2: id=2, tags=[null], scores=[]
+ let mut row2 = GenericRow::new(3);
+ row2.set_field(0, 2_i32);
+
+ let tags2 = {
+ let mut w = FlussArrayWriter::new(1, &DataTypes::string());
+ w.set_null_at(0);
+ w.complete().expect("tags2")
+ };
+ let scores2 = {
+ let w = FlussArrayWriter::new(0, &DataTypes::int());
+ w.complete().expect("scores2")
+ };
+ row2.set_field(1, tags2);
+ row2.set_field(2, scores2);
+
+ // Row 3: id=3, tags=null, scores=[42]
+ let mut row3 = GenericRow::new(3);
+ row3.set_field(0, 3_i32);
+ row3.set_field(1, Datum::Null);
+ let scores3 = {
+ let mut w = FlussArrayWriter::new(1, &DataTypes::int());
+ w.write_int(0, 42);
+ w.complete().expect("scores3")
+ };
+ row3.set_field(2, scores3);
+
+ append_writer.append(&row1).expect("append row1");
+ append_writer.append(&row2).expect("append row2");
+ append_writer.append(&row3).expect("append row3");
+ append_writer.flush().await.expect("Failed to flush");
+
+ let records = scan_table(&table, |scan| scan).await;
+ assert_eq!(records.len(), 3, "expected three log records");
+
+ let r0 = records[0].row();
+ assert_eq!(r0.get_int(0).unwrap(), 1);
+ let tags_r0 = r0.get_array(1).unwrap();
+ assert_eq!(tags_r0.size(), 2);
+ assert_eq!(tags_r0.get_string(0).unwrap(), "hello");
+ assert_eq!(tags_r0.get_string(1).unwrap(), "world");
+ let scores_r0 = r0.get_array(2).unwrap();
+ assert_eq!(scores_r0.size(), 3);
+ assert_eq!(scores_r0.get_int(0).unwrap(), 10);
+ assert_eq!(scores_r0.get_int(1).unwrap(), 20);
+ assert_eq!(scores_r0.get_int(2).unwrap(), 30);
+
+ let r1 = records[1].row();
+ assert_eq!(r1.get_int(0).unwrap(), 2);
+ let tags_r1 = r1.get_array(1).unwrap();
+ assert_eq!(tags_r1.size(), 1);
+ assert!(tags_r1.is_null_at(0));
+ let scores_r1 = r1.get_array(2).unwrap();
+ assert_eq!(scores_r1.size(), 0);
+
+ let r2 = records[2].row();
+ assert_eq!(r2.get_int(0).unwrap(), 3);
+ assert!(r2.is_null_at(1).unwrap());
+ let scores_r2 = r2.get_array(2).unwrap();
+ assert_eq!(scores_r2.size(), 1);
+ assert_eq!(scores_r2.get_int(0).unwrap(), 42);
+
+ admin
+ .drop_table(&table_path, false)
+ .await
+ .expect("Failed to drop table");
+ }
+
+ #[tokio::test]
+ async fn append_and_scan_with_nested_array() {
+ use fluss::row::binary_array::FlussArrayWriter;
+ use fluss::row::{FlussArray, GenericRow};
+
+ let cluster = get_shared_cluster();
+ let connection = cluster.get_fluss_connection().await;
+ let admin = connection.get_admin().expect("Failed to get admin");
+
+ let table_path = TablePath::new("fluss", "test_log_arrays_nested");
+ let inner_array_type = DataTypes::array(DataTypes::int());
+ let matrix_type = DataTypes::array(inner_array_type.clone());
+
+ let schema = Schema::builder()
+ .column("id", DataTypes::int())
+ .column("matrix", matrix_type.clone())
+ .build()
+ .expect("Failed to build schema");
+
+ let table_descriptor = TableDescriptor::builder()
+ .schema(schema)
+ .build()
+ .expect("Failed to build table descriptor");
+
+ create_table(&admin, &table_path, &table_descriptor).await;
+
+ let table = connection
+ .get_table(&table_path)
+ .await
+ .expect("Failed to get table");
+
+ let append_writer = table
+ .new_append()
+ .expect("Failed to create append")
+ .create_writer()
+ .expect("Failed to create writer");
+
+ fn build_inner_ints(vals: &[i32]) -> FlussArray {
+ let mut w = FlussArrayWriter::new(vals.len(), &DataTypes::int());
+ for (i, v) in vals.iter().enumerate() {
+ w.write_int(i, *v);
+ }
+ w.complete().expect("inner array")
+ }
+
+ // Row 1: matrix = [[1,2],[3,4]]
+ let mut row1 = GenericRow::new(2);
+ row1.set_field(0, 1_i32);
+ let m1 = {
+ let mut w = FlussArrayWriter::new(2, &inner_array_type);
+ w.write_array(0, &build_inner_ints(&[1, 2]));
+ w.write_array(1, &build_inner_ints(&[3, 4]));
+ w.complete().expect("matrix1")
+ };
+ row1.set_field(1, m1);
+
+ // Row 2: matrix = [[5], null, []]
+ let mut row2 = GenericRow::new(2);
+ row2.set_field(0, 2_i32);
+ let m2 = {
+ let mut w = FlussArrayWriter::new(3, &inner_array_type);
+ w.write_array(0, &build_inner_ints(&[5]));
+ w.set_null_at(1);
+ let empty: FlussArray = FlussArrayWriter::new(0, &DataTypes::int())
+ .complete()
+ .unwrap();
+ w.write_array(2, &empty);
+ w.complete().expect("matrix2")
+ };
+ row2.set_field(1, m2);
+
+ append_writer.append(&row1).expect("append row1");
+ append_writer.append(&row2).expect("append row2");
+ append_writer.flush().await.expect("Failed to flush");
+
+ let records = scan_table(&table, |scan| scan).await;
+ assert_eq!(records.len(), 2);
+
+ let r0 = records[0].row();
+ assert_eq!(r0.get_int(0).unwrap(), 1);
+ let mat0 = r0.get_array(1).unwrap();
+ assert_eq!(mat0.size(), 2);
+ let a00 = mat0.get_array(0).unwrap();
+ assert_eq!(a00.size(), 2);
+ assert_eq!(a00.get_int(0).unwrap(), 1);
+ assert_eq!(a00.get_int(1).unwrap(), 2);
+ let a01 = mat0.get_array(1).unwrap();
+ assert_eq!(a01.size(), 2);
+ assert_eq!(a01.get_int(0).unwrap(), 3);
+ assert_eq!(a01.get_int(1).unwrap(), 4);
+
+ let r1 = records[1].row();
+ assert_eq!(r1.get_int(0).unwrap(), 2);
+ let mat1 = r1.get_array(1).unwrap();
+ assert_eq!(mat1.size(), 3);
+ let b0 = mat1.get_array(0).unwrap();
+ assert_eq!(b0.size(), 1);
+ assert_eq!(b0.get_int(0).unwrap(), 5);
+ assert!(mat1.is_null_at(1));
+ let b2 = mat1.get_array(2).unwrap();
+ assert_eq!(b2.size(), 0);
+
+ admin
+ .drop_table(&table_path, false)
+ .await
+ .expect("Failed to drop table");
+ }
+
+ #[tokio::test]
+ async fn append_and_scan_with_array_rich_types() {
+ use fluss::row::binary_array::FlussArrayWriter;
+ use fluss::row::{Date, Decimal, GenericRow, Time, TimestampNtz};
+
+ let cluster = get_shared_cluster();
+ let connection = cluster.get_fluss_connection().await;
+ let admin = connection.get_admin().expect("Failed to get admin");
+
+ let table_path = TablePath::new("fluss", "test_log_arrays_rich_types");
+ let ts_val = TimestampNtz::from_millis_nanos(1769163227123, 456000).unwrap();
+ let dec_val = Decimal::from_unscaled_long(12345, 10, 2).unwrap();
+
+ let schema = Schema::builder()
+ .column("id", DataTypes::int())
+ .column("arr_bytes", DataTypes::array(DataTypes::bytes()))
+ .column("arr_date", DataTypes::array(DataTypes::date()))
+ .column(
+ "arr_time",
+ DataTypes::array(DataTypes::time_with_precision(3)),
+ )
+ .column(
+ "arr_ts",
+ DataTypes::array(DataTypes::timestamp_with_precision(6)),
+ )
+ .column("arr_decimal", DataTypes::array(DataTypes::decimal(10, 2)))
+ .build()
+ .expect("Failed to build schema");
+
+ let table_descriptor = TableDescriptor::builder()
+ .schema(schema)
+ .build()
+ .expect("Failed to build table descriptor");
+
+ create_table(&admin, &table_path, &table_descriptor).await;
+
+ let table = connection
+ .get_table(&table_path)
+ .await
+ .expect("Failed to get table");
+
+ let append_writer = table
+ .new_append()
+ .expect("Failed to create append")
+ .create_writer()
+ .expect("Failed to create writer");
+
+ let mut row = GenericRow::new(6);
+ row.set_field(0, 1_i32);
+
+ let elem_bytes = &[0_u8, 1, 2, 255];
+ let d = Date::new(20476);
+ let t = Time::new(36827123);
+
+ let arr_bytes = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::bytes());
+ w.write_binary_bytes(0, elem_bytes);
+ w.set_null_at(1);
+ w.complete().expect("arr_bytes")
+ };
+ let arr_date = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::date());
+ w.write_date(0, d);
+ w.set_null_at(1);
+ w.complete().expect("arr_date")
+ };
+ let arr_time = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::time_with_precision(3));
+ w.write_time(0, t);
+ w.set_null_at(1);
+ w.complete().expect("arr_time")
+ };
+ let arr_ts = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::timestamp_with_precision(6));
+ w.write_timestamp_ntz(0, &ts_val, 6);
+ w.set_null_at(1);
+ w.complete().expect("arr_ts")
+ };
+ let arr_decimal = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::decimal(10, 2));
+ w.write_decimal(0, &dec_val, 10);
+ w.set_null_at(1);
+ w.complete().expect("arr_decimal")
+ };
+
+ row.set_field(1, arr_bytes);
+ row.set_field(2, arr_date);
+ row.set_field(3, arr_time);
+ row.set_field(4, arr_ts);
+ row.set_field(5, arr_decimal);
+
+ append_writer.append(&row).expect("append");
+ append_writer.flush().await.expect("Failed to flush");
+
+ let records = scan_table(&table, |scan| scan).await;
+ assert_eq!(records.len(), 1);
+ let r = records[0].row();
+
+ let ab = r.get_array(1).unwrap();
+ assert_eq!(ab.size(), 2);
+ assert_eq!(ab.get_binary(0).unwrap(), elem_bytes);
+ assert!(ab.is_null_at(1));
+
+ let ad = r.get_array(2).unwrap();
+ assert_eq!(ad.size(), 2);
+ assert_eq!(ad.get_date(0).unwrap().get_inner(), d.get_inner());
+ assert!(ad.is_null_at(1));
+
+ let at = r.get_array(3).unwrap();
+ assert_eq!(at.size(), 2);
+ assert_eq!(at.get_time(0).unwrap().get_inner(), t.get_inner());
+ assert!(at.is_null_at(1));
+
+ let ats = r.get_array(4).unwrap();
+ assert_eq!(ats.size(), 2);
+ let read_ts = ats.get_timestamp_ntz(0, 6).unwrap();
+ assert_eq!(read_ts.get_millisecond(), ts_val.get_millisecond());
+ assert_eq!(
+ read_ts.get_nano_of_millisecond(),
+ ts_val.get_nano_of_millisecond()
+ );
+ assert!(ats.is_null_at(1));
+
+ let adc = r.get_array(5).unwrap();
+ assert_eq!(adc.size(), 2);
+ assert_eq!(adc.get_decimal(0, 10, 2).unwrap(), dec_val);
+ assert!(adc.is_null_at(1));
+
+ admin
+ .drop_table(&table_path, false)
+ .await
+ .expect("Failed to drop table");
+ }
+
+ #[tokio::test]
+ async fn append_and_scan_with_array_encoding_edge_cases() {
+ use fluss::row::binary_array::FlussArrayWriter;
+ use fluss::row::{Decimal, FlussArray, GenericRow, TimestampNtz};
+
+ fn assert_f32_special(actual: f32, expected: f32) {
+ if expected.is_nan() {
+ assert!(actual.is_nan(), "expected NaN");
+ } else if expected.is_infinite() {
+ assert_eq!(actual.is_infinite(), true);
+ assert_eq!(actual.signum(), expected.signum());
+ } else {
+ assert!((actual - expected).abs() < f32::EPSILON);
+ }
+ }
+
+ fn assert_f64_special(actual: f64, expected: f64) {
+ if expected.is_nan() {
+ assert!(actual.is_nan(), "expected NaN");
+ } else if expected.is_infinite() {
+ assert_eq!(actual.is_infinite(), true);
Review Comment:
ditto
##########
crates/fluss/tests/integration/kv_table.rs:
##########
@@ -20,14 +20,41 @@
mod kv_table_test {
use crate::integration::utils::{create_partitions, create_table, get_shared_cluster};
use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath};
- use fluss::row::{GenericRow, InternalRow};
+ use fluss::row::binary_array::FlussArrayWriter;
+ use fluss::row::{FlussArray, GenericRow, InternalRow};
fn make_key(id: i32) -> GenericRow<'static> {
- let mut row = GenericRow::new(3);
+ make_key_with_len(id, 3)
+ }
+
+ fn make_key_with_len(id: i32, field_count: usize) -> GenericRow<'static> {
+ let mut row = GenericRow::new(field_count);
row.set_field(0, id);
row
}
+ fn make_string_array(values: &[Option<&str>]) -> FlussArray {
Review Comment:
we have a lot of similar helpers shared between log_table.rs and kv_table.rs
Let's factor them out to utils.rs, wdyt?
##########
crates/fluss/tests/integration/log_table.rs:
##########
@@ -1373,4 +1388,546 @@ mod table_test {
.await
.expect("Failed to drop table");
}
+
+ #[tokio::test]
+ async fn append_and_scan_with_array() {
+ use fluss::row::binary_array::FlussArrayWriter;
+ use fluss::row::{Datum, GenericRow};
+
+ let cluster = get_shared_cluster();
+ let connection = cluster.get_fluss_connection().await;
+ let admin = connection.get_admin().expect("Failed to get admin");
+
+ let table_path = TablePath::new("fluss", "test_log_arrays_basic");
+ let schema = Schema::builder()
+ .column("id", DataTypes::int())
+ .column("tags", DataTypes::array(DataTypes::string()))
+ .column("scores", DataTypes::array(DataTypes::int()))
+ .build()
+ .expect("Failed to build schema");
+
+ let table_descriptor = TableDescriptor::builder()
+ .schema(schema)
+ .build()
+ .expect("Failed to build table descriptor");
+
+ create_table(&admin, &table_path, &table_descriptor).await;
+
+ let table = connection
+ .get_table(&table_path)
+ .await
+ .expect("Failed to get table");
+
+ let append_writer = table
+ .new_append()
+ .expect("Failed to create append")
+ .create_writer()
+ .expect("Failed to create writer");
+
+ // Row 1: id=1, tags=["hello", "world"], scores=[10, 20, 30]
+ let mut row1 = GenericRow::new(3);
+ row1.set_field(0, 1_i32);
+
+ let tags1 = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::string());
+ w.write_string(0, "hello");
+ w.write_string(1, "world");
+ w.complete().expect("tags1")
+ };
+ let scores1 = {
+ let mut w = FlussArrayWriter::new(3, &DataTypes::int());
+ w.write_int(0, 10);
+ w.write_int(1, 20);
+ w.write_int(2, 30);
+ w.complete().expect("scores1")
+ };
+ row1.set_field(1, tags1);
+ row1.set_field(2, scores1);
+
+ // Row 2: id=2, tags=[null], scores=[]
+ let mut row2 = GenericRow::new(3);
+ row2.set_field(0, 2_i32);
+
+ let tags2 = {
+ let mut w = FlussArrayWriter::new(1, &DataTypes::string());
+ w.set_null_at(0);
+ w.complete().expect("tags2")
+ };
+ let scores2 = {
+ let w = FlussArrayWriter::new(0, &DataTypes::int());
+ w.complete().expect("scores2")
+ };
+ row2.set_field(1, tags2);
+ row2.set_field(2, scores2);
+
+ // Row 3: id=3, tags=null, scores=[42]
+ let mut row3 = GenericRow::new(3);
+ row3.set_field(0, 3_i32);
+ row3.set_field(1, Datum::Null);
+ let scores3 = {
+ let mut w = FlussArrayWriter::new(1, &DataTypes::int());
+ w.write_int(0, 42);
+ w.complete().expect("scores3")
+ };
+ row3.set_field(2, scores3);
+
+ append_writer.append(&row1).expect("append row1");
+ append_writer.append(&row2).expect("append row2");
+ append_writer.append(&row3).expect("append row3");
+ append_writer.flush().await.expect("Failed to flush");
+
+ let records = scan_table(&table, |scan| scan).await;
+ assert_eq!(records.len(), 3, "expected three log records");
+
+ let r0 = records[0].row();
+ assert_eq!(r0.get_int(0).unwrap(), 1);
+ let tags_r0 = r0.get_array(1).unwrap();
+ assert_eq!(tags_r0.size(), 2);
+ assert_eq!(tags_r0.get_string(0).unwrap(), "hello");
+ assert_eq!(tags_r0.get_string(1).unwrap(), "world");
+ let scores_r0 = r0.get_array(2).unwrap();
+ assert_eq!(scores_r0.size(), 3);
+ assert_eq!(scores_r0.get_int(0).unwrap(), 10);
+ assert_eq!(scores_r0.get_int(1).unwrap(), 20);
+ assert_eq!(scores_r0.get_int(2).unwrap(), 30);
+
+ let r1 = records[1].row();
+ assert_eq!(r1.get_int(0).unwrap(), 2);
+ let tags_r1 = r1.get_array(1).unwrap();
+ assert_eq!(tags_r1.size(), 1);
+ assert!(tags_r1.is_null_at(0));
+ let scores_r1 = r1.get_array(2).unwrap();
+ assert_eq!(scores_r1.size(), 0);
+
+ let r2 = records[2].row();
+ assert_eq!(r2.get_int(0).unwrap(), 3);
+ assert!(r2.is_null_at(1).unwrap());
+ let scores_r2 = r2.get_array(2).unwrap();
+ assert_eq!(scores_r2.size(), 1);
+ assert_eq!(scores_r2.get_int(0).unwrap(), 42);
+
+ admin
+ .drop_table(&table_path, false)
+ .await
+ .expect("Failed to drop table");
+ }
+
+ #[tokio::test]
+ async fn append_and_scan_with_nested_array() {
+ use fluss::row::binary_array::FlussArrayWriter;
+ use fluss::row::{FlussArray, GenericRow};
+
+ let cluster = get_shared_cluster();
+ let connection = cluster.get_fluss_connection().await;
+ let admin = connection.get_admin().expect("Failed to get admin");
+
+ let table_path = TablePath::new("fluss", "test_log_arrays_nested");
+ let inner_array_type = DataTypes::array(DataTypes::int());
+ let matrix_type = DataTypes::array(inner_array_type.clone());
+
+ let schema = Schema::builder()
+ .column("id", DataTypes::int())
+ .column("matrix", matrix_type.clone())
+ .build()
+ .expect("Failed to build schema");
+
+ let table_descriptor = TableDescriptor::builder()
+ .schema(schema)
+ .build()
+ .expect("Failed to build table descriptor");
+
+ create_table(&admin, &table_path, &table_descriptor).await;
+
+ let table = connection
+ .get_table(&table_path)
+ .await
+ .expect("Failed to get table");
+
+ let append_writer = table
+ .new_append()
+ .expect("Failed to create append")
+ .create_writer()
+ .expect("Failed to create writer");
+
+ fn build_inner_ints(vals: &[i32]) -> FlussArray {
+ let mut w = FlussArrayWriter::new(vals.len(), &DataTypes::int());
+ for (i, v) in vals.iter().enumerate() {
+ w.write_int(i, *v);
+ }
+ w.complete().expect("inner array")
+ }
+
+ // Row 1: matrix = [[1,2],[3,4]]
+ let mut row1 = GenericRow::new(2);
+ row1.set_field(0, 1_i32);
+ let m1 = {
+ let mut w = FlussArrayWriter::new(2, &inner_array_type);
+ w.write_array(0, &build_inner_ints(&[1, 2]));
+ w.write_array(1, &build_inner_ints(&[3, 4]));
+ w.complete().expect("matrix1")
+ };
+ row1.set_field(1, m1);
+
+ // Row 2: matrix = [[5], null, []]
+ let mut row2 = GenericRow::new(2);
+ row2.set_field(0, 2_i32);
+ let m2 = {
+ let mut w = FlussArrayWriter::new(3, &inner_array_type);
+ w.write_array(0, &build_inner_ints(&[5]));
+ w.set_null_at(1);
+ let empty: FlussArray = FlussArrayWriter::new(0, &DataTypes::int())
+ .complete()
+ .unwrap();
+ w.write_array(2, &empty);
+ w.complete().expect("matrix2")
+ };
+ row2.set_field(1, m2);
+
+ append_writer.append(&row1).expect("append row1");
+ append_writer.append(&row2).expect("append row2");
+ append_writer.flush().await.expect("Failed to flush");
+
+ let records = scan_table(&table, |scan| scan).await;
+ assert_eq!(records.len(), 2);
+
+ let r0 = records[0].row();
+ assert_eq!(r0.get_int(0).unwrap(), 1);
+ let mat0 = r0.get_array(1).unwrap();
+ assert_eq!(mat0.size(), 2);
+ let a00 = mat0.get_array(0).unwrap();
+ assert_eq!(a00.size(), 2);
+ assert_eq!(a00.get_int(0).unwrap(), 1);
+ assert_eq!(a00.get_int(1).unwrap(), 2);
+ let a01 = mat0.get_array(1).unwrap();
+ assert_eq!(a01.size(), 2);
+ assert_eq!(a01.get_int(0).unwrap(), 3);
+ assert_eq!(a01.get_int(1).unwrap(), 4);
+
+ let r1 = records[1].row();
+ assert_eq!(r1.get_int(0).unwrap(), 2);
+ let mat1 = r1.get_array(1).unwrap();
+ assert_eq!(mat1.size(), 3);
+ let b0 = mat1.get_array(0).unwrap();
+ assert_eq!(b0.size(), 1);
+ assert_eq!(b0.get_int(0).unwrap(), 5);
+ assert!(mat1.is_null_at(1));
+ let b2 = mat1.get_array(2).unwrap();
+ assert_eq!(b2.size(), 0);
+
+ admin
+ .drop_table(&table_path, false)
+ .await
+ .expect("Failed to drop table");
+ }
+
+ #[tokio::test]
+ async fn append_and_scan_with_array_rich_types() {
+ use fluss::row::binary_array::FlussArrayWriter;
+ use fluss::row::{Date, Decimal, GenericRow, Time, TimestampNtz};
+
+ let cluster = get_shared_cluster();
+ let connection = cluster.get_fluss_connection().await;
+ let admin = connection.get_admin().expect("Failed to get admin");
+
+ let table_path = TablePath::new("fluss", "test_log_arrays_rich_types");
+ let ts_val = TimestampNtz::from_millis_nanos(1769163227123, 456000).unwrap();
+ let dec_val = Decimal::from_unscaled_long(12345, 10, 2).unwrap();
+
+ let schema = Schema::builder()
+ .column("id", DataTypes::int())
+ .column("arr_bytes", DataTypes::array(DataTypes::bytes()))
+ .column("arr_date", DataTypes::array(DataTypes::date()))
+ .column(
+ "arr_time",
+ DataTypes::array(DataTypes::time_with_precision(3)),
+ )
+ .column(
+ "arr_ts",
+ DataTypes::array(DataTypes::timestamp_with_precision(6)),
+ )
+ .column("arr_decimal", DataTypes::array(DataTypes::decimal(10, 2)))
+ .build()
+ .expect("Failed to build schema");
+
+ let table_descriptor = TableDescriptor::builder()
+ .schema(schema)
+ .build()
+ .expect("Failed to build table descriptor");
+
+ create_table(&admin, &table_path, &table_descriptor).await;
+
+ let table = connection
+ .get_table(&table_path)
+ .await
+ .expect("Failed to get table");
+
+ let append_writer = table
+ .new_append()
+ .expect("Failed to create append")
+ .create_writer()
+ .expect("Failed to create writer");
+
+ let mut row = GenericRow::new(6);
+ row.set_field(0, 1_i32);
+
+ let elem_bytes = &[0_u8, 1, 2, 255];
+ let d = Date::new(20476);
+ let t = Time::new(36827123);
+
+ let arr_bytes = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::bytes());
+ w.write_binary_bytes(0, elem_bytes);
+ w.set_null_at(1);
+ w.complete().expect("arr_bytes")
+ };
+ let arr_date = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::date());
+ w.write_date(0, d);
+ w.set_null_at(1);
+ w.complete().expect("arr_date")
+ };
+ let arr_time = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::time_with_precision(3));
+ w.write_time(0, t);
+ w.set_null_at(1);
+ w.complete().expect("arr_time")
+ };
+ let arr_ts = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::timestamp_with_precision(6));
+ w.write_timestamp_ntz(0, &ts_val, 6);
+ w.set_null_at(1);
+ w.complete().expect("arr_ts")
+ };
+ let arr_decimal = {
+ let mut w = FlussArrayWriter::new(2, &DataTypes::decimal(10, 2));
+ w.write_decimal(0, &dec_val, 10);
+ w.set_null_at(1);
+ w.complete().expect("arr_decimal")
+ };
+
+ row.set_field(1, arr_bytes);
+ row.set_field(2, arr_date);
+ row.set_field(3, arr_time);
+ row.set_field(4, arr_ts);
+ row.set_field(5, arr_decimal);
+
+ append_writer.append(&row).expect("append");
+ append_writer.flush().await.expect("Failed to flush");
+
+ let records = scan_table(&table, |scan| scan).await;
+ assert_eq!(records.len(), 1);
+ let r = records[0].row();
+
+ let ab = r.get_array(1).unwrap();
+ assert_eq!(ab.size(), 2);
+ assert_eq!(ab.get_binary(0).unwrap(), elem_bytes);
+ assert!(ab.is_null_at(1));
+
+ let ad = r.get_array(2).unwrap();
+ assert_eq!(ad.size(), 2);
+ assert_eq!(ad.get_date(0).unwrap().get_inner(), d.get_inner());
+ assert!(ad.is_null_at(1));
+
+ let at = r.get_array(3).unwrap();
+ assert_eq!(at.size(), 2);
+ assert_eq!(at.get_time(0).unwrap().get_inner(), t.get_inner());
+ assert!(at.is_null_at(1));
+
+ let ats = r.get_array(4).unwrap();
+ assert_eq!(ats.size(), 2);
+ let read_ts = ats.get_timestamp_ntz(0, 6).unwrap();
+ assert_eq!(read_ts.get_millisecond(), ts_val.get_millisecond());
+ assert_eq!(
+ read_ts.get_nano_of_millisecond(),
+ ts_val.get_nano_of_millisecond()
+ );
+ assert!(ats.is_null_at(1));
+
+ let adc = r.get_array(5).unwrap();
+ assert_eq!(adc.size(), 2);
+ assert_eq!(adc.get_decimal(0, 10, 2).unwrap(), dec_val);
+ assert!(adc.is_null_at(1));
+
+ admin
+ .drop_table(&table_path, false)
+ .await
+ .expect("Failed to drop table");
+ }
+
+ #[tokio::test]
+ async fn append_and_scan_with_array_encoding_edge_cases() {
+ use fluss::row::binary_array::FlussArrayWriter;
+ use fluss::row::{Decimal, FlussArray, GenericRow, TimestampNtz};
+
+ fn assert_f32_special(actual: f32, expected: f32) {
+ if expected.is_nan() {
+ assert!(actual.is_nan(), "expected NaN");
+ } else if expected.is_infinite() {
+ assert_eq!(actual.is_infinite(), true);
Review Comment:
nit: assert!(actual.is_infinite()) ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#discussion_r3178280148
##########
crates/fluss/tests/integration/log_table.rs:
##########
@@ -1373,4 +1384,523 @@ mod table_test {
.await
.expect("Failed to drop table");
}
+
+ #[tokio::test]
+ async fn append_and_scan_with_array() {
Review Comment:
similar reply above 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#issuecomment-4365832385
Thanks for your comments, addressed all, and rebased! @fresh-borzoni PTAL 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#issuecomment-4391149428
Consolidated @leekeiabstraction PTAL 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#discussion_r3186469180
##########
crates/fluss/tests/integration/kv_table.rs:
##########
@@ -880,6 +896,226 @@ mod kv_table_test {
found_row_nulls.is_null_at(16).unwrap(),
"col_binary should be null"
);
+ assert!(
+ found_row_nulls.is_null_at(17).unwrap(),
+ "col_array should be null"
+ );
+
+ admin
+ .drop_table(&table_path, false)
+ .await
+ .expect("Failed to drop table");
+ }
+
+ #[tokio::test]
+ async fn upsert_and_lookup_with_array() {
Review Comment:
Thanks Charles for the response. Let's do this, leave `all_supported_datatypes` as you have already updated and then consolidate the array / nested array and potentially the edge case test cases into a single `upsert_and_lookup_with_array ` test case. That way we can leave `all_supported_datatypes` simple and test cases organised in a way that is mostly focused on features of primary key table.
##########
crates/fluss/tests/integration/log_table.rs:
##########
@@ -1373,4 +1384,523 @@ mod table_test {
.await
.expect("Failed to drop table");
}
+
+ #[tokio::test]
+ async fn append_and_scan_with_array() {
Review Comment:
Similar reply to the above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#issuecomment-4366441760
> i think https://github.com/apache/fluss-rust/actions/runs/25275194644/job/74108466204 is a test flake, and shouldn't be caused by changes in this pr @leekeiabstraction @fresh-borzoni hope rerun will fix.
>
> in the meantime, i can take a quick look to understand why
@charlesdong1991 Yes, it's just flake with python tests, we use single test-cluster across 4 different python version x 4 workers, so under contention this can happen. I need to take a look at this, because it's getting annoying sometimes :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#issuecomment-4366435340
i think https://github.com/apache/fluss-rust/actions/runs/25275194644/job/74108466204 is a test flake, and shouldn't be caused by changes in this pr @leekeiabstraction @fresh-borzoni hope rerun will fix.
in the meantime, i can take a quick look to understand why
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#issuecomment-4364055025
@leekeiabstraction @fresh-borzoni it will be great if you can take a look! 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#discussion_r3176775020
##########
bindings/python/test/test_kv_table.py:
##########
@@ -335,6 +346,228 @@ 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 array columns in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_basic")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("tags", pa.list_(pa.string())),
+ pa.field("scores", pa.list_(pa.int32())),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ # Row 1: standard arrays
+ await _upsert_and_wait(
+ upsert_writer, {"id": 1, "tags": ["hello", "world"], "scores": [10, 20, 30]}
+ )
+
+ # Row 2: null element in array + empty array
+ await _upsert_and_wait(upsert_writer, {"id": 2, "tags": [None], "scores": []})
+
+ # Row 3: null array column
+ await _upsert_and_wait(upsert_writer, {"id": 3, "tags": None, "scores": [42]})
+
+ lookuper = table.new_lookup().create_lookuper()
+
+ result1 = await lookuper.lookup({"id": 1})
+ assert result1 is not None
+ assert result1["tags"] == ["hello", "world"]
+ assert result1["scores"] == [10, 20, 30]
+
+ result2 = await lookuper.lookup({"id": 2})
+ assert result2 is not None
+ assert result2["tags"] == [None]
+ assert result2["scores"] == []
+
+ result3 = await lookuper.lookup({"id": 3})
+ assert result3 is not None
+ assert result3["tags"] is None
+ assert result3["scores"] == [42]
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_upsert_and_lookup_with_nested_array(connection, admin):
+ """Test upsert and lookup with nested array (ARRAY<ARRAY<INT>>) in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_nested")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("matrix", pa.list_(pa.list_(pa.int32()))),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ await _upsert_and_wait(upsert_writer, {"id": 1, "matrix": [[1, 2], [3, 4]]})
+
+ await _upsert_and_wait(upsert_writer, {"id": 2, "matrix": [[], [5], [6, 7, 8]]})
+
+ await _upsert_and_wait(upsert_writer, {"id": 3, "matrix": None})
+
+ await _upsert_and_wait(upsert_writer, {"id": 4, "matrix": [[1, None], None, []]})
+
+ lookuper = table.new_lookup().create_lookuper()
+
+ result1 = await lookuper.lookup({"id": 1})
+ assert result1 is not None
+ assert result1["matrix"] == [[1, 2], [3, 4]]
+
+ result2 = await lookuper.lookup({"id": 2})
+ assert result2 is not None
+ assert result2["matrix"] == [[], [5], [6, 7, 8]]
+
+ result3 = await lookuper.lookup({"id": 3})
+ assert result3 is not None
+ assert result3["matrix"] is None
+
+ result4 = await lookuper.lookup({"id": 4})
+ assert result4 is not None
+ assert result4["matrix"] == [[1, None], None, []]
+
+ 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 in KV tables."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_rich_types")
+ 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))),
+ ]
+ ),
+ primary_keys=["id"],
+ )
+ table_descriptor = fluss.TableDescriptor(schema)
+ await admin.create_table(table_path, table_descriptor, ignore_if_exists=False)
+
+ table = await connection.get_table(table_path)
+ upsert_writer = table.new_upsert().create_writer()
+
+ await _upsert_and_wait(
+ upsert_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],
+ },
+ )
+
+ 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]
+
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
+async def test_upsert_and_lookup_with_array_encoding_edge_cases(connection, admin):
+ """Test array encoding edge cases in KV table upsert/lookup."""
+ table_path = fluss.TablePath("fluss", "py_test_kv_arrays_edge_cases")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ 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): support PyArrow FixedSizeBinary in schema conversion.
Review Comment:
i seems fixed size binary will fail now, and should be supported.
it is nicer to address in a separate PR IMHO, so added a new issue: https://github.com/apache/fluss-rust/issues/524
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#discussion_r3178112967
##########
crates/fluss/tests/integration/kv_table.rs:
##########
@@ -880,6 +896,226 @@ mod kv_table_test {
found_row_nulls.is_null_at(16).unwrap(),
"col_binary should be null"
);
+ assert!(
+ found_row_nulls.is_null_at(17).unwrap(),
+ "col_array should be null"
+ );
+
+ admin
+ .drop_table(&table_path, false)
+ .await
+ .expect("Failed to drop table");
+ }
+
+ #[tokio::test]
+ async fn upsert_and_lookup_with_array() {
Review Comment:
Hmm, this approach increases the number of test cases (eventually when we have map types support as well). Can you try to incorporate these into `all_supported_datatypes`? You can have additional columns there e.g. array of arrays.
##########
crates/fluss/tests/integration/log_table.rs:
##########
@@ -1373,4 +1384,523 @@ mod table_test {
.await
.expect("Failed to drop table");
}
+
+ #[tokio::test]
+ async fn append_and_scan_with_array() {
Review Comment:
Similarly here on incorporating into `all_supported_datatypes` test case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#issuecomment-4381812116
Consolidated the tests, PTAL! @leekeiabstraction thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #523:
URL: https://github.com/apache/fluss-rust/pull/523#issuecomment-4415215689
@charlesdong1991 I thinks it's good enough for now, merging 👍
We have an inconsistency between bindings' tests though, as we have a split: structural tests and rich types test in essence, but C++ and Rust don't have this for KV tables, when Python has it for both KV and log.
@leekeiabstraction Do you mind to file a couple of 'first issues' to consolidate tests and make them more consistent and manageable?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tests] Add integration tests for array data type [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni merged PR #523:
URL: https://github.com/apache/fluss-rust/pull/523
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org