diff --git a/fluss-gateway/src/protocol/rest/datatype.rs b/fluss-gateway/src/protocol/rest/datatype.rs index 726c2a9f785..4e257282cf3 100644 --- a/fluss-gateway/src/protocol/rest/datatype.rs +++ b/fluss-gateway/src/protocol/rest/datatype.rs @@ -268,9 +268,11 @@ impl From<&DataField> for WireRowField { impl TryFrom for DataType { type Error = GatewayError; - /// Builds the native type using its constructors. + /// Builds and validates the native type using fluss-rs. fn try_from(data_type: WireDataType) -> Result { - to_native(data_type, 0) + let converted = to_native(data_type, 0)?; + converted.validate_row_field_names().map_err(invalid_type)?; + Ok(converted) } } @@ -281,7 +283,7 @@ fn to_native(data_type: WireDataType, depth: usize) -> GatewayResult { "the data type nests deeper than {MAX_TYPE_NESTING} levels" ))); } - // TODO: Delegate length and ROW field validation once fluss-rs supports it. + // TODO: Delegate length validation once fluss-rs supports it. let converted = match data_type { WireDataType::Boolean { nullable } => { DataType::Boolean(BooleanType::with_nullable(nullable)) @@ -361,7 +363,7 @@ fn to_native(data_type: WireDataType, depth: usize) -> GatewayResult { Ok(converted) } -/// A type parameter the native constructor refused, which came from a caller's body. +/// Native data-type validation failed for a value from the caller's body. fn invalid_type(error: fluss::error::Error) -> GatewayError { GatewayError::invalid_argument(format!("invalid data type: {error}")) } @@ -586,7 +588,7 @@ mod tests { } #[test] - fn native_constructors_and_nesting_limit_define_validation() { + fn native_validation_and_nesting_limit_define_validation() { for body in [ json!({"type": "DECIMAL", "precision": 0, "scale": 0}), json!({"type": "DECIMAL", "precision": 2, "scale": 3}), @@ -616,13 +618,37 @@ mod tests { } } - let fields = ["", "a\nb", "id", "id"] - .into_iter() - .map(|name| DataField::new(name, DataType::Int(IntType::new()), None)) - .collect(); - let native = DataType::Row(RowType::new(fields)); + let native = DataType::Row(RowType::new(vec![DataField::new( + "a\nb", + DataType::Int(IntType::new()), + None, + )])); assert_eq!(parse(render(&native)).unwrap(), native); + for body in [ + json!({"type": "ROW", "fields": [ + {"name": " ", "field_type": {"type": "INTEGER"}} + ]}), + json!({"type": "ROW", "fields": [ + {"name": "id", "field_type": {"type": "INTEGER"}}, + {"name": "id", "field_type": {"type": "STRING"}} + ]}), + json!({"type": "ARRAY", "element_type": { + "type": "MAP", + "key_type": {"type": "STRING"}, + "value_type": {"type": "ROW", "fields": [ + {"name": "value", "field_type": {"type": "INTEGER"}}, + {"name": "value", "field_type": {"type": "BIGINT"}} + ]} + }}), + ] { + let wire: WireDataType = + serde_json::from_value(body.clone()).expect("the shape parses"); + let error = DataType::try_from(wire).expect_err("invalid ROW fields are refused"); + assert_eq!(error.kind(), ErrorKind::InvalidArgument, "{body}"); + assert!(error.message().contains("Field names must"), "{body}"); + } + for depth in [MAX_TYPE_NESTING, MAX_TYPE_NESTING + 1] { let mut wire = WireDataType::Int { nullable: true }; for _ in 0..depth { diff --git a/fluss-gateway/src/protocol/rest/ddl.rs b/fluss-gateway/src/protocol/rest/ddl.rs index 48207924154..da9d502a5e6 100644 --- a/fluss-gateway/src/protocol/rest/ddl.rs +++ b/fluss-gateway/src/protocol/rest/ddl.rs @@ -1048,6 +1048,29 @@ mod tests { ); } + for fields in [ + json!([{"name": " ", "field_type": {"type": "INTEGER"}}]), + json!([ + {"name": "value", "field_type": {"type": "INTEGER"}}, + {"name": "value", "field_type": {"type": "STRING"}} + ]), + ] { + let mut invalid = partitioned_table(); + invalid["validate_only"] = json!(true); + invalid["columns"][2]["data_type"] = json!({"type": "ROW", "fields": fields}); + let (status, _, body) = + post(&app, "/v1/clusters/default/databases/sales/tables", invalid).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); + assert_eq!(body["error"]["code"], "invalid_argument"); + assert!( + body["error"]["message"] + .as_str() + .unwrap() + .contains("Field names must"), + "{body}" + ); + } + let mut log_table = partitioned_table(); log_table["validate_only"] = json!(true); log_table.as_object_mut().unwrap().remove("primary_key"); @@ -1341,6 +1364,44 @@ mod tests { } } + #[tokio::test] + async fn invalid_nested_row_fields_are_rejected_before_alter_table() { + let (backend, app) = gateway(); + + for fields in [ + json!([{"name": " ", "field_type": {"type": "INTEGER"}}]), + json!([ + {"name": "value", "field_type": {"type": "INTEGER"}}, + {"name": "value", "field_type": {"type": "STRING"}} + ]), + ] { + let (status, _, body) = send( + &app, + Method::PATCH, + "/v1/clusters/default/databases/sales/tables/orders", + Some(json!({ + "changes": [{ + "kind": "add_column", + "name": "payload", + "data_type": {"type": "ROW", "fields": fields} + }] + })), + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); + assert_eq!(body["error"]["code"], "invalid_argument"); + assert!( + body["error"]["message"] + .as_str() + .unwrap() + .contains("Field names must"), + "{body}" + ); + } + + assert!(backend.calls().is_empty()); + } + #[tokio::test] async fn a_mutation_is_checked_before_its_body() { let (backend, app) = gateway(); diff --git a/fluss-rust/crates/fluss/src/metadata/datatype.rs b/fluss-rust/crates/fluss/src/metadata/datatype.rs index 60a44ba7187..195a5211be5 100644 --- a/fluss-rust/crates/fluss/src/metadata/datatype.rs +++ b/fluss-rust/crates/fluss/src/metadata/datatype.rs @@ -18,6 +18,7 @@ use crate::error::Error::IllegalArgument; use crate::error::Result; use serde::{Deserialize, Serialize}; +use std::collections::HashSet; use std::fmt::{Display, Formatter}; /// Data type for Fluss table. @@ -70,6 +71,47 @@ impl DataType { } } + /// Validates the names of fields in every nested [`RowType`]. + /// + /// Returns an error if a row contains a blank field name or duplicate field names. + pub fn validate_row_field_names(&self) -> Result<()> { + match self { + DataType::Array(array) => array.get_element_type().validate_row_field_names(), + DataType::Map(map) => { + map.key_type().validate_row_field_names()?; + map.value_type().validate_row_field_names() + } + DataType::Row(row) => { + let mut seen = HashSet::with_capacity(row.fields().len()); + let mut duplicates = HashSet::new(); + for field in row.fields() { + if field.name().trim().is_empty() { + return Err(IllegalArgument { + message: + "Field names must contain at least one non-whitespace character." + .to_string(), + }); + } + if !seen.insert(field.name()) { + duplicates.insert(field.name()); + } + } + if !duplicates.is_empty() { + return Err(IllegalArgument { + message: format!( + "Field names must be unique. Found duplicates: {duplicates:?}" + ), + }); + } + for field in row.fields() { + field.data_type().validate_row_field_names()?; + } + Ok(()) + } + _ => Ok(()), + } + } + pub fn as_non_nullable(&self) -> Self { match self { DataType::Boolean(v) => DataType::Boolean(v.as_non_nullable()), @@ -1545,6 +1587,52 @@ fn test_row_display() { assert_eq!(row_type_non_null.to_string(), "ROW NOT NULL"); } +#[test] +fn test_validate_row_field_names() { + DataTypes::row(vec![ + DataTypes::field("id", DataTypes::int()), + DataTypes::field( + "payload", + DataTypes::array(DataTypes::row(vec![DataTypes::field( + "value", + DataTypes::string(), + )])), + ), + ]) + .validate_row_field_names() + .expect("valid ROW field names"); + + for (data_type, expected_message) in [ + ( + DataTypes::row(vec![DataTypes::field(" \t", DataTypes::int())]), + "Field names must contain at least one non-whitespace character.", + ), + ( + DataTypes::row(vec![ + DataTypes::field("id", DataTypes::int()), + DataTypes::field("id", DataTypes::string()), + ]), + "Field names must be unique. Found duplicates:", + ), + ( + DataTypes::array(DataTypes::map( + DataTypes::string(), + DataTypes::row(vec![ + DataTypes::field("value", DataTypes::int()), + DataTypes::field("value", DataTypes::bigint()), + ]), + )), + "Field names must be unique. Found duplicates:", + ), + ] { + let err = data_type.validate_row_field_names().unwrap_err(); + assert!( + err.to_string().contains(expected_message), + "unexpected error: {err}" + ); + } +} + #[test] fn test_datatype_display() { assert_eq!(DataTypes::boolean().to_string(), "BOOLEAN"); diff --git a/fluss-rust/crates/fluss/src/metadata/table.rs b/fluss-rust/crates/fluss/src/metadata/table.rs index fe56fd2cf27..e2da8b19252 100644 --- a/fluss-rust/crates/fluss/src/metadata/table.rs +++ b/fluss-rust/crates/fluss/src/metadata/table.rs @@ -542,6 +542,15 @@ impl SchemaBuilder { "Duplicate column names found: {duplicates:?}" ))); } + for column in columns { + column + .data_type() + .validate_row_field_names() + .map_err(|error| match error { + IllegalArgument { message } => Error::invalid_table(message), + other => other, + })?; + } let Some(pk) = primary_key else { return Ok(columns.to_vec()); @@ -1725,6 +1734,29 @@ mod tests { ); } + #[test] + fn invalid_nested_row_field_names_are_rejected() { + let err = Schema::builder() + .column( + "payload", + DataTypes::array(DataTypes::map( + DataTypes::string(), + DataTypes::row(vec![ + DataTypes::field("value", DataTypes::int()), + DataTypes::field("value", DataTypes::bigint()), + ]), + )), + ) + .build() + .unwrap_err(); + + assert!( + err.to_string() + .contains("Field names must be unique. Found duplicates:"), + "unexpected error: {err}" + ); + } + #[test] fn auto_increment_column_requires_a_primary_key_table() { let err = Schema::builder()