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