[rust] Validate nested ROW field names

fluss-gateway/src/protocol/rest/datatype.rs

@@ -268,9 +268,11 @@ impl From<&DataField> for WireRowField {
 impl TryFrom<WireDataType> 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<Self, Self::Error> {
-        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<DataType> {
             "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<DataType> {
     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 {

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();

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<age BIGINT> 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");

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()