fix(rest): skip serializing unset optional fields in CreateTableRequest

Unset optional fields of CreateTableRequest (location, partition-spec,
write-order, stage-create) and of the embedded UnboundPartitionSpec and
UnboundPartitionField (spec-id, field-id) were serialized as explicit
JSON nulls. Strict REST catalog servers such as Apache Polaris reject
those create-table requests with 400 Bad Request. Skip serializing
these fields when they are unset. Deserialization of explicit nulls is
unchanged.

Fixes #2135

Signed-off-by: rahulsmahadev <rahul.mahadev@databricks.com>

Author: rahulsmahadev

crates/catalog/rest/src/types.rs

@@ -251,14 +251,18 @@ pub struct CreateTableRequest {
     /// Name of the table to create
     pub name: String,
     /// Optional table location. If not provided, the server will choose a location.
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub location: Option<String>,
     /// Table schema
     pub schema: Schema,
     /// Optional partition specification. If not provided, the table will be unpartitioned.
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub partition_spec: Option<UnboundPartitionSpec>,
     /// Optional sort order for the table
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub write_order: Option<SortOrder>,
     /// Whether to stage the create for a transaction (true) or create immediately (false)
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub stage_create: Option<bool>,
     /// Optional properties to set on the table
     #[serde(default, skip_serializing_if = "HashMap::is_empty")]
@@ -355,4 +359,117 @@ mod tests {
             json_no_props
         );
     }
+
+    fn test_create_table_request_schema() -> Schema {
+        serde_json::from_value(serde_json::json!({
+            "type": "struct",
+            "schema-id": 1,
+            "fields": [
+                {
+                    "id": 1,
+                    "name": "foo",
+                    "required": false,
+                    "type": "string"
+                },
+                {
+                    "id": 2,
+                    "name": "bar",
+                    "required": true,
+                    "type": "int"
+                }
+            ],
+            "identifier-field-ids": [2]
+        }))
+        .expect("Failed to deserialize test schema")
+    }
+
+    #[test]
+    fn test_create_table_request_minimal_serialization() {
+        let request = CreateTableRequest {
+            name: "tbl1".to_string(),
+            location: None,
+            schema: test_create_table_request_schema(),
+            partition_spec: None,
+            write_order: None,
+            stage_create: None,
+            properties: HashMap::new(),
+        };
+
+        let serialized = serde_json::to_value(&request).expect("Serialization failed");
+        let object = serialized.as_object().expect("Expected a JSON object");
+        assert!(object.contains_key("name"));
+        assert!(object.contains_key("schema"));
+        assert!(!object.contains_key("location"));
+        assert!(!object.contains_key("partition-spec"));
+        assert!(!object.contains_key("write-order"));
+        assert!(!object.contains_key("stage-create"));
+        assert!(!object.contains_key("properties"));
+    }
+
+    #[test]
+    fn test_create_table_request_full_serialization() {
+        let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+            "name": "tbl1",
+            "location": "s3://warehouse/tbl1",
+            "schema": test_create_table_request_schema(),
+            "partition-spec": {
+                "spec-id": 1,
+                "fields": [
+                    {
+                        "source-id": 2,
+                        "field-id": 1000,
+                        "name": "bar",
+                        "transform": "identity"
+                    }
+                ]
+            },
+            "write-order": {
+                "order-id": 1,
+                "fields": [
+                    {
+                        "transform": "identity",
+                        "source-id": 2,
+                        "direction": "asc",
+                        "null-order": "nulls-first"
+                    }
+                ]
+            },
+            "stage-create": true,
+            "properties": {
+                "owner": "test"
+            }
+        }))
+        .expect("Deserialization failed");
+
+        let serialized = serde_json::to_value(&request).expect("Serialization failed");
+        let object = serialized.as_object().expect("Expected a JSON object");
+        assert_eq!(
+            object.get("location"),
+            Some(&serde_json::json!("s3://warehouse/tbl1"))
+        );
+        assert!(object.contains_key("partition-spec"));
+        assert!(object.contains_key("write-order"));
+        assert_eq!(object.get("stage-create"), Some(&serde_json::json!(true)));
+        assert!(object.contains_key("properties"));
+    }
+
+    #[test]
+    fn test_create_table_request_deserialize_explicit_nulls() {
+        let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+            "name": "tbl1",
+            "location": null,
+            "schema": test_create_table_request_schema(),
+            "partition-spec": null,
+            "write-order": null,
+            "stage-create": null
+        }))
+        .expect("Deserialization failed");
+
+        assert_eq!(request.name, "tbl1");
+        assert_eq!(request.location, None);
+        assert_eq!(request.partition_spec, None);
+        assert_eq!(request.write_order, None);
+        assert_eq!(request.stage_create, None);
+        assert!(request.properties.is_empty());
+    }
 }

crates/iceberg/src/spec/partition.rs

@@ -246,6 +246,7 @@ pub struct UnboundPartitionField {
     /// A partition field id that is used to identify a partition field and is unique within a partition spec.
     /// In v2 table metadata, it is unique across all partition specs.
     #[builder(default, setter(strip_option(fallback = field_id_opt)))]
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub field_id: Option<i32>,
     /// A partition name.
     pub name: String,
@@ -260,6 +261,7 @@ pub struct UnboundPartitionField {
 #[serde(rename_all = "kebab-case")]
 pub struct UnboundPartitionSpec {
     /// Identifier for PartitionSpec
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub(crate) spec_id: Option<i32>,
     /// Details of the partition spec
     pub(crate) fields: Vec<UnboundPartitionField>,
@@ -912,6 +914,38 @@ mod tests {
         assert_eq!(Transform::Day, partition_spec.fields[0].transform);
     }
 
+    #[test]
+    fn test_unbound_partition_spec_serialization_skips_none_fields() {
+        let spec = UnboundPartitionSpec::builder()
+            .add_partition_field(4, "ts_day".to_string(), Transform::Day)
+            .unwrap()
+            .build();
+
+        let value = serde_json::to_value(&spec).unwrap();
+        let object = value.as_object().unwrap();
+        assert!(!object.contains_key("spec-id"));
+        let field = object["fields"][0].as_object().unwrap();
+        assert!(!field.contains_key("field-id"));
+
+        let value = serde_json::to_value(spec.with_spec_id(1)).unwrap();
+        let object = value.as_object().unwrap();
+        assert_eq!(Some(&serde_json::json!(1)), object.get("spec-id"));
+
+        // Explicit nulls must still deserialize to `None` for backwards
+        // compatibility.
+        let spec: UnboundPartitionSpec = serde_json::from_str(
+            r#"{
+                "spec-id": null,
+                "fields": [
+                    {"source-id": 4, "name": "ts_day", "transform": "day", "field-id": null}
+                ]
+            }"#,
+        )
+        .unwrap();
+        assert_eq!(None, spec.spec_id);
+        assert_eq!(None, spec.fields[0].field_id);
+    }
+
     #[test]
     fn test_new_unpartition() {
         let schema = Schema::builder()