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/01 23:15:24 UTC
(fluss) branch main updated: [rust] Preserve highest_field_id during Schema JSON deserialization (#4184)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 79c71c24a [rust] Preserve highest_field_id during Schema JSON deserialization (#4184)
79c71c24a is described below
commit 79c71c24a141838c46b020d9fff60ed0c70a1dd0
Author: Junbo Wang <be...@gmail.com>
AuthorDate: Wed Sep 2 07:15:20 2026 +0800
[rust] Preserve highest_field_id during Schema JSON deserialization (#4184)
* [rust] Preserve highest_field_id during Schema JSON deserialization
* [rust] Clamp deserialized highest_field_id to schema maximum
---
fluss-rust/crates/fluss/src/metadata/json_serde.rs | 55 ++++++++++++++++++++++
fluss-rust/crates/fluss/src/metadata/table.rs | 12 ++++-
2 files changed, 66 insertions(+), 1 deletion(-)
diff --git a/fluss-rust/crates/fluss/src/metadata/json_serde.rs b/fluss-rust/crates/fluss/src/metadata/json_serde.rs
index c504660e9..cc636a4ad 100644
--- a/fluss-rust/crates/fluss/src/metadata/json_serde.rs
+++ b/fluss-rust/crates/fluss/src/metadata/json_serde.rs
@@ -543,6 +543,23 @@ impl JsonSerde for Schema {
}
}
+ if let Some(highest_field_id_node) = node.get(Self::HIGHEST_FIELD_ID) {
+ let highest_field_id =
+ highest_field_id_node
+ .as_i64()
+ .ok_or_else(|| Error::JsonSerdeError {
+ message: format!("{} must be an integer", Self::HIGHEST_FIELD_ID),
+ })?;
+ let highest_field_id =
+ i32::try_from(highest_field_id).map_err(|_| Error::JsonSerdeError {
+ message: format!(
+ "{} value {highest_field_id} does not fit in i32",
+ Self::HIGHEST_FIELD_ID
+ ),
+ })?;
+ schema_builder = schema_builder.highest_field_id(highest_field_id);
+ }
+
schema_builder.build()
}
}
@@ -984,6 +1001,44 @@ mod tests {
assert_eq!(round_tripped, original);
}
+ #[test]
+ fn schema_deserialization_preserves_highest_field_id_above_maximum() {
+ let json = json!({
+ "version": 1,
+ "columns": [
+ {"name": "id", "data_type": {"type": "INTEGER"}, "id": 0},
+ {"name": "name", "data_type": {"type": "STRING"}, "id": 1}
+ ],
+ "highest_field_id": 10
+ });
+
+ let schema = Schema::deserialize_json(&json).unwrap();
+ assert_eq!(schema.highest_field_id(), 10);
+ }
+
+ #[test]
+ fn schema_deserialization_clamps_highest_field_id_to_computed_minimum() {
+ let json = json!({
+ "version": 1,
+ "columns": [
+ {"name": "id", "data_type": {"type": "INTEGER"}, "id": 0},
+ {"name": "name", "data_type": {"type": "STRING"}, "id": 1}
+ ],
+ "highest_field_id": 0
+ });
+
+ let schema = Schema::deserialize_json(&json).unwrap();
+ assert_eq!(schema.highest_field_id(), 1);
+
+ let empty_json = json!({
+ "version": 1,
+ "columns": [],
+ "highest_field_id": -3
+ });
+ let empty_schema = Schema::deserialize_json(&empty_json).unwrap();
+ assert_eq!(empty_schema.highest_field_id(), -1);
+ }
+
#[test]
fn schema_rejects_duplicate_nested_field_ids() {
let nested = DataTypes::row(vec![
diff --git a/fluss-rust/crates/fluss/src/metadata/table.rs b/fluss-rust/crates/fluss/src/metadata/table.rs
index 28b94840e..ea014d57e 100644
--- a/fluss-rust/crates/fluss/src/metadata/table.rs
+++ b/fluss-rust/crates/fluss/src/metadata/table.rs
@@ -253,6 +253,7 @@ pub struct SchemaBuilder {
columns: Vec<Column>,
primary_key: Option<PrimaryKey>,
auto_increment_col_names: Vec<String>,
+ highest_field_id: Option<i32>,
}
impl SchemaBuilder {
@@ -330,9 +331,18 @@ impl SchemaBuilder {
Ok(self)
}
+ pub(crate) fn highest_field_id(mut self, highest_field_id: i32) -> Self {
+ self.highest_field_id = Some(highest_field_id);
+ self
+ }
+
pub fn build(&self) -> Result<Schema> {
let columns = Self::normalize_columns(&self.columns, self.primary_key.as_ref())?;
- let (columns_with_ids, highest_field_id) = Self::assign_all_field_ids(columns)?;
+ let (columns_with_ids, maximum_field_id) = Self::assign_all_field_ids(columns)?;
+ let highest_field_id = self
+ .highest_field_id
+ .unwrap_or(maximum_field_id)
+ .max(maximum_field_id);
if !self.auto_increment_col_names.is_empty() && self.primary_key.is_none() {
return Err(IllegalArgument {