You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/09/17 10:02:24 UTC
(fluss) branch release-1.0 updated: [rust] Validate nested ROW field names (#4209) (#4399)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch release-1.0
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/release-1.0 by this push:
new 573ce8c55 [rust] Validate nested ROW field names (#4209) (#4399)
573ce8c55 is described below
commit 573ce8c557cfc29c270a9b74edd7334369d36b26
Author: naivedogger <ya...@gmail.com>
AuthorDate: Thu Sep 17 18:02:18 2026 +0800
[rust] Validate nested ROW field names (#4209) (#4399)
* [rust] Validate nested ROW field names
* [rust] Centralize ROW field name validation
* [rust] Reject blank table column names
(cherry picked from commit 7e28c961aa1cefc35e03a65d71ece04e85d61dab)
---
fluss-gateway/src/protocol/rest/datatype.rs | 46 ++++++++++---
fluss-gateway/src/protocol/rest/ddl.rs | 65 +++++++++++++++++
fluss-rust/crates/fluss/src/metadata/datatype.rs | 88 ++++++++++++++++++++++++
fluss-rust/crates/fluss/src/metadata/table.rs | 51 ++++++++++++++
4 files changed, 240 insertions(+), 10 deletions(-)
diff --git a/fluss-gateway/src/protocol/rest/datatype.rs b/fluss-gateway/src/protocol/rest/datatype.rs
index 726c2a9f7..4e257282c 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<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 {
diff --git a/fluss-gateway/src/protocol/rest/ddl.rs b/fluss-gateway/src/protocol/rest/ddl.rs
index 482079241..30069d727 100644
--- a/fluss-gateway/src/protocol/rest/ddl.rs
+++ b/fluss-gateway/src/protocol/rest/ddl.rs
@@ -1032,6 +1032,10 @@ mod tests {
{"name": "id", "data_type": {"type": "STRING"}}
]),
),
+ (
+ "columns",
+ json!([{"name": " ", "data_type": {"type": "BIGINT"}}]),
+ ),
] {
let mut invalid = partitioned_table();
invalid["validate_only"] = json!(true);
@@ -1048,6 +1052,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 +1368,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 60a44ba71..195a5211b 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<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");
diff --git a/fluss-rust/crates/fluss/src/metadata/table.rs b/fluss-rust/crates/fluss/src/metadata/table.rs
index fe56fd2cf..57e1aacdf 100644
--- a/fluss-rust/crates/fluss/src/metadata/table.rs
+++ b/fluss-rust/crates/fluss/src/metadata/table.rs
@@ -537,11 +537,25 @@ impl SchemaBuilder {
primary_key: Option<&PrimaryKey>,
) -> Result<Vec<Column>> {
let names: Vec<_> = columns.iter().map(|c| &c.name).collect();
+ if names.iter().any(|name| name.trim().is_empty()) {
+ return Err(Error::invalid_table(
+ "Field names must contain at least one non-whitespace character.",
+ ));
+ }
if let Some(duplicates) = Self::find_duplicates(&names) {
return Err(Error::invalid_table(format!(
"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());
@@ -1709,6 +1723,20 @@ mod tests {
}
}
+ #[test]
+ fn blank_column_names_are_rejected() {
+ let err = Schema::builder()
+ .column(" \t", DataTypes::int())
+ .build()
+ .unwrap_err();
+
+ assert!(
+ err.to_string()
+ .contains("Field names must contain at least one non-whitespace character."),
+ "unexpected error: {err}"
+ );
+ }
+
#[test]
fn multiple_primary_keys_are_rejected() {
let err = Schema::builder()
@@ -1725,6 +1753,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()