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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
