You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fresh-borzoni (via GitHub)" <gi...@apache.org> on 2026/05/23 22:12:09 UTC
[PR] [rust] integration tests for MAP dt + restructure tests around comple… [fluss-rust]
fresh-borzoni opened a new pull request, #560:
URL: https://github.com/apache/fluss-rust/pull/560
closes #549
Added Map integration tests, consolidated dt tests to avoid overhead in table creations and exponential tests per complex dts.
--
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] [rust] integration tests for MAP dt + restructure ITs for complex dts [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #560:
URL: https://github.com/apache/fluss-rust/pull/560#issuecomment-4532861918
Merging to continue working on typed writers refactoring
--
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] [rust] integration tests for MAP dt + restructure ITs for complex dts [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #560:
URL: https://github.com/apache/fluss-rust/pull/560#issuecomment-4526711804
@charlesdong1991 @leekeiabstraction @luoyuxia PTAL 🙏 , I need this as a preparation to refactor complex types to use typed writers, it's mostly tests consolidation and new map tests.
Appreciate if you can take a quick 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] [rust] integration tests for MAP dt + restructure ITs for complex dts [fluss-rust]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #560:
URL: https://github.com/apache/fluss-rust/pull/560#discussion_r3296921496
##########
crates/fluss/src/row/column.rs:
##########
@@ -661,25 +661,19 @@ impl InternalRow for ColumnarRow {
};
let column = self.column(pos)?;
- let element_field = match column.data_type() {
- ArrowDataType::List(field) => field,
+ match column.data_type() {
+ ArrowDataType::List(_) => {}
other => {
return Err(IllegalArgument {
message: format!("expected List array at position {pos}, got {other:?}"),
});
}
- };
-
- let actual_element_type = from_arrow_field(element_field)?;
- if actual_element_type != *element_fluss_type {
- return Err(IllegalArgument {
- message: format!(
- "Arrow list element type {:?} does not match expected Fluss type {:?}",
- actual_element_type, element_fluss_type
- ),
- });
}
+ // `to_arrow_type` is lossy (e.g. TIMESTAMP_LTZ → plain Arrow Timestamp);
Review Comment:
so to confirm since not mentioned in PR description: i guess this will be a behaviour, correct? that users won't know schema mismatch when calling get_array on position
##########
crates/fluss/tests/integration/kv_table.rs:
##########
@@ -2175,4 +1251,522 @@ mod kv_table_test {
.await
.expect("Failed to drop table");
}
+
+ /// KV upsert + lookup against a schema covering every supported data type.
+ #[tokio::test]
+ async fn all_supported_datatypes() {
Review Comment:
Very nit: somehow i am a bit conservative to have such huge test, i think now if something breaks, developers will see all unrelated assertions and not sure if there is meaningful tie with the failing column index? We can revisit in future if encounter such issue in development
##########
crates/fluss/tests/integration/utils.rs:
##########
@@ -176,3 +178,200 @@ pub async fn create_partitions(
.expect("Failed to create partition");
}
}
+
+pub fn dt_array_int() -> DataType {
+ DataTypes::array(DataTypes::int())
+}
+
+pub fn dt_map_string_int() -> DataType {
+ DataTypes::map(DataTypes::string(), DataTypes::int())
+}
+
+pub fn dt_row_seq_label() -> DataType {
+ DataTypes::row(vec![
+ DataField::new("seq", DataTypes::int(), None),
+ DataField::new("label", DataTypes::string(), None),
+ ])
+}
+
+pub fn as_row_type(dt: &DataType) -> RowType {
+ match dt {
+ DataType::Row(rt) => rt.clone(),
+ other => panic!("expected DataType::Row, got {other:?}"),
+ }
+}
+
+pub fn dt_row_deep() -> DataType {
+ let inner = DataTypes::row(vec![DataField::new("n", DataTypes::int(), None)]);
+ DataTypes::row(vec![DataField::new("inner", inner, None)])
+}
+
+pub fn dt_row_rich() -> DataType {
+ DataTypes::row(vec![
+ DataField::new("f_bool", DataTypes::boolean(), None),
+ DataField::new("f_int", DataTypes::int(), None),
+ DataField::new("f_long", DataTypes::bigint(), None),
+ DataField::new("f_float", DataTypes::float(), None),
+ DataField::new("f_double", DataTypes::double(), None),
+ DataField::new("f_str", DataTypes::string(), None),
+ DataField::new("f_bytes", DataTypes::bytes(), None),
+ DataField::new("f_decimal", DataTypes::decimal(10, 2), None),
+ DataField::new("f_date", DataTypes::date(), None),
+ DataField::new("f_time", DataTypes::time_with_precision(3), None),
+ DataField::new("f_ts_ntz", DataTypes::timestamp_with_precision(6), None),
+ DataField::new("f_ts_ltz", DataTypes::timestamp_ltz_with_precision(6), None),
+ DataField::new("f_binary_fixed", DataTypes::binary(4), None),
+ DataField::new("f_array_int", DataTypes::array(DataTypes::int()), None),
+ ])
+}
+
+pub fn array_dt_basics_columns() -> Vec<(&'static str, DataType)> {
+ vec![
+ ("arr_int", DataTypes::array(DataTypes::int())),
+ ("arr_string", DataTypes::array(DataTypes::string())),
+ ("arr_of_arr", DataTypes::array(dt_array_int())),
+ ("arr_of_row", DataTypes::array(dt_row_seq_label())),
+ ]
+}
+
+pub fn row_dt_basics_columns() -> Vec<(&'static str, DataType)> {
+ vec![
+ ("row_basic", dt_row_seq_label()),
+ ("row_deep", dt_row_deep()),
+ ("row_rich", dt_row_rich()),
+ ]
+}
+
+pub fn map_dt_basics_columns() -> Vec<(&'static str, DataType)> {
+ vec![
+ ("map_string_int", dt_map_string_int()),
+ (
+ "map_of_row",
+ DataTypes::map(DataTypes::string(), dt_row_seq_label()),
+ ),
+ (
+ "map_of_map",
+ DataTypes::map(DataTypes::string(), dt_map_string_int()),
+ ),
+ (
+ "map_of_array",
+ DataTypes::map(DataTypes::string(), dt_array_int()),
+ ),
+ ("array_of_map", DataTypes::array(dt_map_string_int())),
+ ]
+}
+
+pub fn scalar_dt_columns() -> Vec<(&'static str, DataType)> {
+ vec![
+ ("col_tinyint", DataTypes::tinyint()),
+ ("col_smallint", DataTypes::smallint()),
+ ("col_bigint", DataTypes::bigint()),
+ ("col_float", DataTypes::float()),
+ ("col_double", DataTypes::double()),
+ ("col_boolean", DataTypes::boolean()),
+ ("col_char", DataTypes::char(10)),
+ ("col_string", DataTypes::string()),
+ ("col_decimal", DataTypes::decimal(10, 2)),
+ ("col_date", DataTypes::date()),
+ ("col_time_s", DataTypes::time_with_precision(0)),
+ ("col_time_ms", DataTypes::time_with_precision(3)),
+ ("col_time_us", DataTypes::time_with_precision(6)),
+ ("col_time_ns", DataTypes::time_with_precision(9)),
+ ("col_ts_s", DataTypes::timestamp_with_precision(0)),
+ ("col_ts_ms", DataTypes::timestamp_with_precision(3)),
+ ("col_ts_us", DataTypes::timestamp_with_precision(6)),
+ ("col_ts_ns", DataTypes::timestamp_with_precision(9)),
+ ("col_ts_ltz_s", DataTypes::timestamp_ltz_with_precision(0)),
+ ("col_ts_ltz_ms", DataTypes::timestamp_ltz_with_precision(3)),
+ ("col_ts_ltz_us", DataTypes::timestamp_ltz_with_precision(6)),
+ ("col_ts_ltz_ns", DataTypes::timestamp_ltz_with_precision(9)),
+ ("col_bytes_top", DataTypes::bytes()),
+ ("col_binary_top", DataTypes::binary(4)),
+ ("col_ts_us_neg", DataTypes::timestamp_with_precision(6)),
+ ("col_ts_ns_neg", DataTypes::timestamp_with_precision(9)),
+ (
+ "col_ts_ltz_us_neg",
+ DataTypes::timestamp_ltz_with_precision(6),
+ ),
+ (
+ "col_ts_ltz_ns_neg",
+ DataTypes::timestamp_ltz_with_precision(9),
+ ),
+ ]
+}
+
+#[derive(Default)]
Review Comment:
nice idea on ColumnPlan that helps reduce redundancy a lot!
Just a nit: do we need `derive[Default]` for internal test only feature?
--
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] [rust] integration tests for MAP dt + restructure ITs for complex dts [fluss-rust]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #560:
URL: https://github.com/apache/fluss-rust/pull/560#discussion_r3294015975
##########
crates/fluss/tests/integration/kv_table.rs:
##########
@@ -2175,4 +1249,429 @@ mod kv_table_test {
.await
.expect("Failed to drop table");
}
+
+ /// KV upsert + lookup against a schema covering every supported data type.
+ #[tokio::test]
+ async fn all_supported_datatypes() {
+ 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_kv_complex_types");
+
+ let row_seq_label_owned = DataTypes::row(vec![
+ DataField::new("seq", DataTypes::int(), None),
+ DataField::new("label", DataTypes::string(), None),
+ ]);
+ let row_seq_label = match &row_seq_label_owned {
+ DataType::Row(rt) => rt.clone(),
+ _ => unreachable!(),
+ };
+ let row_deep_inner = DataTypes::row(vec![DataField::new("n", DataTypes::int(), None)]);
+ let row_deep_owned = DataTypes::row(vec![DataField::new("inner", row_deep_inner, None)]);
+ let row_rich_owned = DataTypes::row(vec![
+ DataField::new("f_bool", DataTypes::boolean(), None),
+ DataField::new("f_int", DataTypes::int(), None),
+ DataField::new("f_long", DataTypes::bigint(), None),
+ DataField::new("f_double", DataTypes::double(), None),
+ DataField::new("f_str", DataTypes::string(), None),
+ DataField::new("f_decimal", DataTypes::decimal(10, 2), None),
+ DataField::new("f_date", DataTypes::date(), None),
+ DataField::new("f_ts_ntz", DataTypes::timestamp_with_precision(6), None),
+ DataField::new("f_ts_ltz", DataTypes::timestamp_ltz_with_precision(6), None),
+ DataField::new("f_array_int", DataTypes::array(DataTypes::int()), None),
+ ]);
+ let inner_array_int = DataTypes::array(DataTypes::int());
+ let inner_map_string_int = DataTypes::map(DataTypes::string(), DataTypes::int());
+
+ let table_descriptor = TableDescriptor::builder()
+ .schema(
+ Schema::builder()
+ .column("id", DataTypes::int())
+ .column("arr_int", DataTypes::array(DataTypes::int()))
+ .column("arr_string", DataTypes::array(DataTypes::string()))
+ .column("arr_of_arr", DataTypes::array(inner_array_int.clone()))
+ .column("arr_of_row", DataTypes::array(row_seq_label_owned.clone()))
+ .column("row_basic", row_seq_label_owned.clone())
+ .column("row_deep", row_deep_owned)
+ .column("row_rich", row_rich_owned)
+ .column("map_string_int", inner_map_string_int.clone())
+ .column(
+ "map_of_row",
+ DataTypes::map(DataTypes::string(), row_seq_label_owned.clone()),
+ )
+ .column(
+ "map_of_map",
+ DataTypes::map(DataTypes::string(), inner_map_string_int.clone()),
+ )
+ .column(
+ "map_of_array",
+ DataTypes::map(DataTypes::string(), inner_array_int.clone()),
+ )
+ .column(
+ "array_of_map",
+ DataTypes::array(inner_map_string_int.clone()),
+ )
+ // Top-level scalar coverage (indices 13..=27).
+ .column("col_tinyint", DataTypes::tinyint())
+ .column("col_smallint", DataTypes::smallint())
+ .column("col_bigint", DataTypes::bigint())
+ .column("col_float", DataTypes::float())
+ .column("col_double", DataTypes::double())
+ .column("col_boolean", DataTypes::boolean())
+ .column("col_char", DataTypes::char(10))
+ .column("col_string", DataTypes::string())
+ .column("col_decimal", DataTypes::decimal(10, 2))
+ .column("col_date", DataTypes::date())
+ .column("col_time", DataTypes::time())
+ .column("col_timestamp", DataTypes::timestamp())
+ .column("col_timestamp_ltz", DataTypes::timestamp_ltz())
+ .column("col_bytes", DataTypes::bytes())
+ .column("col_binary", DataTypes::binary(20))
+ .primary_key(vec!["id"])
+ .build()
+ .expect("schema"),
+ )
+ .build()
+ .expect("table descriptor");
+
+ create_table(&admin, &table_path, &table_descriptor).await;
+
+ let table = connection.get_table(&table_path).await.expect("table");
+ let upsert_writer = table
+ .new_upsert()
+ .expect("upsert")
+ .create_writer()
+ .expect("writer");
+
+ // Row 1 (id=1) — comprehensive: every column populated.
+ let column_count = 28; // id + 12 compound + 15 scalar
+ let mut row1 = GenericRow::new(column_count);
+ row1.set_field(0, 1_i32);
Review Comment:
`column_count` is hard-coded to 28 even though the schema is defined immediately above. This is brittle and can silently go out of sync if columns are added/removed. Prefer deriving the count from the schema/table info to keep the test resilient.
##########
crates/fluss/tests/integration/log_table.rs:
##########
@@ -2088,263 +999,1047 @@ mod table_test {
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");
-
- // Compact types: DECIMAL(10,2) precision<=18, TIMESTAMP(6) precision<=3 for millis
- let dec_compact = Decimal::from_unscaled_long(12345, 10, 2).unwrap();
- let ts_compact = TimestampNtz::from_millis_nanos(1769163227123, 456000).unwrap();
-
- // Non-compact types: DECIMAL(22,5) precision>18, TIMESTAMP(9) precision>3
- let dec_big = Decimal::from_unscaled_bytes(&[66, 237, 18, 59, 11, 216, 31, 4, 244], 22, 5)
- .expect("big decimal");
- let ts_nano = TimestampNtz::from_millis_nanos(1769163227123, 999_999).unwrap();
+ let table_path = TablePath::new("fluss", "test_log_complex_types");
- let d = Date::new(20476);
- let t = Time::new(36827123);
- let elem_bytes = &[0_u8, 1, 2, 255];
- let fixed_a: Vec<u8> = vec![0xDE, 0xAD, 0xBE, 0xEF];
- let fixed_b: Vec<u8> = vec![0x01, 0x02, 0x03, 0x04];
+ // Shared compound type definitions
+ let row_seq_label_owned = DataTypes::row(vec![
+ DataField::new("seq", DataTypes::int(), None),
+ DataField::new("label", DataTypes::string(), None),
+ ]);
+ let row_seq_label = match &row_seq_label_owned {
+ DataType::Row(rt) => rt.clone(),
+ _ => unreachable!(),
+ };
+ let row_deep_inner = DataTypes::row(vec![DataField::new("n", DataTypes::int(), None)]);
+ let row_deep_owned = DataTypes::row(vec![DataField::new("inner", row_deep_inner, None)]);
+ let row_rich_owned = DataTypes::row(vec![
+ DataField::new("f_bool", DataTypes::boolean(), None),
+ DataField::new("f_int", DataTypes::int(), None),
+ DataField::new("f_long", DataTypes::bigint(), None),
+ DataField::new("f_float", DataTypes::float(), None),
+ DataField::new("f_double", DataTypes::double(), None),
+ DataField::new("f_str", DataTypes::string(), None),
+ DataField::new("f_bytes", DataTypes::bytes(), None),
+ DataField::new("f_decimal", DataTypes::decimal(10, 2), None),
+ DataField::new("f_date", DataTypes::date(), None),
+ DataField::new("f_time", DataTypes::time_with_precision(3), None),
+ DataField::new("f_ts_ntz", DataTypes::timestamp_with_precision(6), None),
+ DataField::new("f_ts_ltz", DataTypes::timestamp_ltz_with_precision(6), None),
+ DataField::new("f_binary_fixed", DataTypes::binary(4), None),
+ DataField::new("f_array_int", DataTypes::array(DataTypes::int()), None),
+ ]);
+ let inner_array_int = DataTypes::array(DataTypes::int());
+ let inner_map_string_int = DataTypes::map(DataTypes::string(), DataTypes::int());
let schema = Schema::builder()
.column("id", DataTypes::int())
+ // ARRAY basics
+ .column("arr_int", DataTypes::array(DataTypes::int()))
+ .column("arr_string", DataTypes::array(DataTypes::string()))
+ .column("arr_of_arr", DataTypes::array(inner_array_int.clone()))
+ .column("arr_of_row", DataTypes::array(row_seq_label_owned.clone()))
+ // ROW basics
+ .column("row_basic", row_seq_label_owned.clone())
+ .column("row_deep", row_deep_owned)
+ .column("row_rich", row_rich_owned)
+ // MAP basics
+ .column("map_string_int", inner_map_string_int.clone())
+ .column(
+ "map_of_row",
+ DataTypes::map(DataTypes::string(), row_seq_label_owned.clone()),
+ )
+ .column(
+ "map_of_map",
+ DataTypes::map(DataTypes::string(), inner_map_string_int.clone()),
+ )
+ .column(
+ "map_of_array",
+ DataTypes::map(DataTypes::string(), inner_array_int.clone()),
+ )
+ .column(
+ "array_of_map",
+ DataTypes::array(inner_map_string_int.clone()),
+ )
+ // ARRAY rich types
.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_compact",
+ "arr_ts",
DataTypes::array(DataTypes::timestamp_with_precision(6)),
)
.column(
- "arr_ts_nano",
- DataTypes::array(DataTypes::timestamp_with_precision(9)),
- )
- .column(
- "arr_decimal_compact",
- DataTypes::array(DataTypes::decimal(10, 2)),
+ "arr_ts_ltz",
+ DataTypes::array(DataTypes::timestamp_ltz_with_precision(3)),
)
+ .column("arr_decimal", DataTypes::array(DataTypes::decimal(10, 2)))
.column(
"arr_decimal_big",
DataTypes::array(DataTypes::decimal(22, 5)),
)
- .column("arr_long_str", DataTypes::array(DataTypes::string()))
.column("arr_float", DataTypes::array(DataTypes::float()))
.column("arr_double", DataTypes::array(DataTypes::double()))
.column("arr_binary", DataTypes::array(DataTypes::binary(4)))
+ // MAP rich types
+ .column(
+ "map_bytes",
+ DataTypes::map(DataTypes::string(), DataTypes::bytes()),
+ )
+ .column(
+ "map_decimal",
+ DataTypes::map(DataTypes::string(), DataTypes::decimal(10, 2)),
+ )
+ .column(
+ "map_date",
+ DataTypes::map(DataTypes::string(), DataTypes::date()),
+ )
+ .column(
+ "map_time",
+ DataTypes::map(DataTypes::string(), DataTypes::time_with_precision(3)),
+ )
+ .column(
+ "map_ts",
+ DataTypes::map(DataTypes::string(), DataTypes::timestamp_with_precision(6)),
+ )
+ .column(
+ "map_ts_ltz",
+ DataTypes::map(
+ DataTypes::string(),
+ DataTypes::timestamp_ltz_with_precision(3),
+ ),
+ )
+ .column(
+ "map_float",
+ DataTypes::map(DataTypes::string(), DataTypes::float()),
+ )
+ .column(
+ "map_double",
+ DataTypes::map(DataTypes::string(), DataTypes::double()),
+ )
+ .column(
+ "map_bool",
+ DataTypes::map(DataTypes::string(), DataTypes::boolean()),
+ )
+ .column(
+ "map_binary",
+ DataTypes::map(DataTypes::string(), DataTypes::binary(4)),
+ )
+ .column(
+ "map_int_key",
+ DataTypes::map(DataTypes::int(), DataTypes::string()),
+ )
+ // Top-level scalar coverage (every type + every time/ts precision +
+ // negative-epoch timestamps). Lives at indices 34..=61.
+ .column("col_tinyint", DataTypes::tinyint())
+ .column("col_smallint", DataTypes::smallint())
+ .column("col_bigint", DataTypes::bigint())
+ .column("col_float", DataTypes::float())
+ .column("col_double", DataTypes::double())
+ .column("col_boolean", DataTypes::boolean())
+ .column("col_char", DataTypes::char(10))
+ .column("col_string", DataTypes::string())
+ .column("col_decimal", DataTypes::decimal(10, 2))
+ .column("col_date", DataTypes::date())
+ .column("col_time_s", DataTypes::time_with_precision(0))
+ .column("col_time_ms", DataTypes::time_with_precision(3))
+ .column("col_time_us", DataTypes::time_with_precision(6))
+ .column("col_time_ns", DataTypes::time_with_precision(9))
+ .column("col_ts_s", DataTypes::timestamp_with_precision(0))
+ .column("col_ts_ms", DataTypes::timestamp_with_precision(3))
+ .column("col_ts_us", DataTypes::timestamp_with_precision(6))
+ .column("col_ts_ns", DataTypes::timestamp_with_precision(9))
+ .column("col_ts_ltz_s", DataTypes::timestamp_ltz_with_precision(0))
+ .column("col_ts_ltz_ms", DataTypes::timestamp_ltz_with_precision(3))
+ .column("col_ts_ltz_us", DataTypes::timestamp_ltz_with_precision(6))
+ .column("col_ts_ltz_ns", DataTypes::timestamp_ltz_with_precision(9))
+ .column("col_bytes_top", DataTypes::bytes())
+ .column("col_binary_top", DataTypes::binary(4))
+ .column("col_ts_us_neg", DataTypes::timestamp_with_precision(6))
+ .column("col_ts_ns_neg", DataTypes::timestamp_with_precision(9))
+ .column(
+ "col_ts_ltz_us_neg",
+ DataTypes::timestamp_ltz_with_precision(6),
+ )
+ .column(
+ "col_ts_ltz_ns_neg",
+ DataTypes::timestamp_ltz_with_precision(9),
+ )
.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;
+ .expect("schema");
+ let column_count = 62; // id + 33 compound + 28 scalar
Review Comment:
`column_count` is hard-coded to 62 even though the schema is built just above. This is brittle (adding/removing a column will cause mismatched row sizes or incorrect assertions). Prefer deriving the count from the schema (e.g., `schema.columns().len()`) or from `table.get_table_info().schema.columns().len()` after creation.
--
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] [rust] integration tests for MAP dt + restructure ITs for complex dts [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #560:
URL: https://github.com/apache/fluss-rust/pull/560#discussion_r3298212650
##########
crates/fluss/tests/integration/kv_table.rs:
##########
@@ -2175,4 +1251,522 @@ mod kv_table_test {
.await
.expect("Failed to drop table");
}
+
+ /// KV upsert + lookup against a schema covering every supported data type.
+ #[tokio::test]
+ async fn all_supported_datatypes() {
Review Comment:
Each row's assertions include the column name in their failure messages, so a break should at least point to the offending column.
but we can revisit later if it becomes a concern 👍
--
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] [rust] integration tests for MAP dt + restructure ITs for complex dts [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #560:
URL: https://github.com/apache/fluss-rust/pull/560#discussion_r3293581648
##########
crates/fluss/src/row/column_writer.rs:
##########
@@ -940,6 +939,57 @@ fn write_map_into(
Ok(())
}
+// FlussArray carries no schema; nested row/map elements need the typed
Review Comment:
will be fixed with refactoring, now I need this for test to pass
--
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] [rust] integration tests for MAP dt + restructure ITs for complex dts [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni merged PR #560:
URL: https://github.com/apache/fluss-rust/pull/560
--
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] [rust] integration tests for MAP dt + restructure ITs for complex dts [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #560:
URL: https://github.com/apache/fluss-rust/pull/560#issuecomment-4527902487
@luoyuxia addressed Copilot comments, added couple of helpers to help with boundaries and shared schema between log and kv for "all datatypes" tests, extended kv tests to include more precisions(it was required to unify).
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] [rust] integration tests for MAP dt + restructure ITs for complex dts [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #560:
URL: https://github.com/apache/fluss-rust/pull/560#discussion_r3298230138
##########
crates/fluss/tests/integration/utils.rs:
##########
@@ -176,3 +178,200 @@ pub async fn create_partitions(
.expect("Failed to create partition");
}
}
+
+pub fn dt_array_int() -> DataType {
+ DataTypes::array(DataTypes::int())
+}
+
+pub fn dt_map_string_int() -> DataType {
+ DataTypes::map(DataTypes::string(), DataTypes::int())
+}
+
+pub fn dt_row_seq_label() -> DataType {
+ DataTypes::row(vec![
+ DataField::new("seq", DataTypes::int(), None),
+ DataField::new("label", DataTypes::string(), None),
+ ])
+}
+
+pub fn as_row_type(dt: &DataType) -> RowType {
+ match dt {
+ DataType::Row(rt) => rt.clone(),
+ other => panic!("expected DataType::Row, got {other:?}"),
+ }
+}
+
+pub fn dt_row_deep() -> DataType {
+ let inner = DataTypes::row(vec![DataField::new("n", DataTypes::int(), None)]);
+ DataTypes::row(vec![DataField::new("inner", inner, None)])
+}
+
+pub fn dt_row_rich() -> DataType {
+ DataTypes::row(vec![
+ DataField::new("f_bool", DataTypes::boolean(), None),
+ DataField::new("f_int", DataTypes::int(), None),
+ DataField::new("f_long", DataTypes::bigint(), None),
+ DataField::new("f_float", DataTypes::float(), None),
+ DataField::new("f_double", DataTypes::double(), None),
+ DataField::new("f_str", DataTypes::string(), None),
+ DataField::new("f_bytes", DataTypes::bytes(), None),
+ DataField::new("f_decimal", DataTypes::decimal(10, 2), None),
+ DataField::new("f_date", DataTypes::date(), None),
+ DataField::new("f_time", DataTypes::time_with_precision(3), None),
+ DataField::new("f_ts_ntz", DataTypes::timestamp_with_precision(6), None),
+ DataField::new("f_ts_ltz", DataTypes::timestamp_ltz_with_precision(6), None),
+ DataField::new("f_binary_fixed", DataTypes::binary(4), None),
+ DataField::new("f_array_int", DataTypes::array(DataTypes::int()), None),
+ ])
+}
+
+pub fn array_dt_basics_columns() -> Vec<(&'static str, DataType)> {
+ vec![
+ ("arr_int", DataTypes::array(DataTypes::int())),
+ ("arr_string", DataTypes::array(DataTypes::string())),
+ ("arr_of_arr", DataTypes::array(dt_array_int())),
+ ("arr_of_row", DataTypes::array(dt_row_seq_label())),
+ ]
+}
+
+pub fn row_dt_basics_columns() -> Vec<(&'static str, DataType)> {
+ vec![
+ ("row_basic", dt_row_seq_label()),
+ ("row_deep", dt_row_deep()),
+ ("row_rich", dt_row_rich()),
+ ]
+}
+
+pub fn map_dt_basics_columns() -> Vec<(&'static str, DataType)> {
+ vec![
+ ("map_string_int", dt_map_string_int()),
+ (
+ "map_of_row",
+ DataTypes::map(DataTypes::string(), dt_row_seq_label()),
+ ),
+ (
+ "map_of_map",
+ DataTypes::map(DataTypes::string(), dt_map_string_int()),
+ ),
+ (
+ "map_of_array",
+ DataTypes::map(DataTypes::string(), dt_array_int()),
+ ),
+ ("array_of_map", DataTypes::array(dt_map_string_int())),
+ ]
+}
+
+pub fn scalar_dt_columns() -> Vec<(&'static str, DataType)> {
+ vec![
+ ("col_tinyint", DataTypes::tinyint()),
+ ("col_smallint", DataTypes::smallint()),
+ ("col_bigint", DataTypes::bigint()),
+ ("col_float", DataTypes::float()),
+ ("col_double", DataTypes::double()),
+ ("col_boolean", DataTypes::boolean()),
+ ("col_char", DataTypes::char(10)),
+ ("col_string", DataTypes::string()),
+ ("col_decimal", DataTypes::decimal(10, 2)),
+ ("col_date", DataTypes::date()),
+ ("col_time_s", DataTypes::time_with_precision(0)),
+ ("col_time_ms", DataTypes::time_with_precision(3)),
+ ("col_time_us", DataTypes::time_with_precision(6)),
+ ("col_time_ns", DataTypes::time_with_precision(9)),
+ ("col_ts_s", DataTypes::timestamp_with_precision(0)),
+ ("col_ts_ms", DataTypes::timestamp_with_precision(3)),
+ ("col_ts_us", DataTypes::timestamp_with_precision(6)),
+ ("col_ts_ns", DataTypes::timestamp_with_precision(9)),
+ ("col_ts_ltz_s", DataTypes::timestamp_ltz_with_precision(0)),
+ ("col_ts_ltz_ms", DataTypes::timestamp_ltz_with_precision(3)),
+ ("col_ts_ltz_us", DataTypes::timestamp_ltz_with_precision(6)),
+ ("col_ts_ltz_ns", DataTypes::timestamp_ltz_with_precision(9)),
+ ("col_bytes_top", DataTypes::bytes()),
+ ("col_binary_top", DataTypes::binary(4)),
+ ("col_ts_us_neg", DataTypes::timestamp_with_precision(6)),
+ ("col_ts_ns_neg", DataTypes::timestamp_with_precision(9)),
+ (
+ "col_ts_ltz_us_neg",
+ DataTypes::timestamp_ltz_with_precision(6),
+ ),
+ (
+ "col_ts_ltz_ns_neg",
+ DataTypes::timestamp_ltz_with_precision(9),
+ ),
+ ]
+}
+
+#[derive(Default)]
Review Comment:
well, it was easier this way, another option - inline Self { cols: vec![], index: HashMap::new(), sections: vec![] }, but it's all the same. Arguably derive is a bit cleaner overall
--
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] [rust] integration tests for MAP dt + restructure ITs for complex dts [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #560:
URL: https://github.com/apache/fluss-rust/pull/560#discussion_r3298218506
##########
crates/fluss/src/row/column.rs:
##########
@@ -661,25 +661,19 @@ impl InternalRow for ColumnarRow {
};
let column = self.column(pos)?;
- let element_field = match column.data_type() {
- ArrowDataType::List(field) => field,
+ match column.data_type() {
+ ArrowDataType::List(_) => {}
other => {
return Err(IllegalArgument {
message: format!("expected List array at position {pos}, got {other:?}"),
});
}
- };
-
- let actual_element_type = from_arrow_field(element_field)?;
- if actual_element_type != *element_fluss_type {
- return Err(IllegalArgument {
- message: format!(
- "Arrow list element type {:?} does not match expected Fluss type {:?}",
- actual_element_type, element_fluss_type
- ),
- });
}
+ // `to_arrow_type` is lossy (e.g. TIMESTAMP_LTZ → plain Arrow Timestamp);
Review Comment:
Intentional. The old strict check tripped on lossy Arrow round-trips (e.g. TIMESTAMP_LTZ -> plain Arrow Timestamp) and rejected valid schemas.
Real shape mismatches still error from the per-element conversion below
--
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