[rust] Preserve highest_field_id during Schema JSON deserialization

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,41 @@ 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_rejects_highest_field_id_below_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": 0
+        });
+
+        let error = Schema::deserialize_json(&json).unwrap_err();
+        assert!(
+            error
+                .to_string()
+                .contains("must be greater than or equal to the maximum field id (1)"),
+            "{error}"
+        );
+    }
+
     #[test]
     fn schema_rejects_duplicate_nested_field_ids() {
         let nested = DataTypes::row(vec![

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,23 @@ 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);
+
+        if !columns_with_ids.is_empty() && highest_field_id < maximum_field_id {
+            return Err(IllegalArgument {
+                message: format!(
+                    "Highest field id ({highest_field_id}) must be greater than or equal to the maximum field id ({maximum_field_id})"
+                ),
+            });
+        }
 
         if !self.auto_increment_col_names.is_empty() && self.primary_key.is_none() {
             return Err(IllegalArgument {