fix(dataset): add missing currency_code_column to DatasetPostSchema (#38853)

(cherry picked from commit 9c288d66b5)

tests/integration_tests/datasets/api_tests.py

@@ -861,6 +861,29 @@ class TestDatasetApi(SupersetTestCase):
         assert alpha in model.owners
         self.items_to_delete = [model]
 
+    @pytest.mark.usefixtures("load_energy_table_with_slice")
+    def test_create_dataset_with_currency_code_column(self):
+        """
+        Dataset API: Test create dataset persists currency_code_column
+        """
+
+        energy_usage_ds = self.get_energy_usage_dataset()
+        self.login(ALPHA_USERNAME)
+        table_data = {
+            "database": energy_usage_ds.database_id,
+            "table_name": "energy_usage_virtual_currency_column",
+            "sql": "select * from energy_usage",
+            "currency_code_column": "currency",
+        }
+        if schema := get_example_default_schema():
+            table_data["schema"] = schema
+        rv = self.post_assert_metric("/api/v1/dataset/", table_data, "post")
+        assert rv.status_code == 201
+        data = json.loads(rv.data.decode("utf-8"))
+        model = db.session.query(SqlaTable).get(data.get("id"))
+        assert model.currency_code_column == "currency"
+        self.items_to_delete = [model]
+
     @unittest.skip("test is failing stochastically")
     def test_create_dataset_same_name_different_schema(self):
         if backend() == "sqlite":

tests/unit_tests/datasets/schema_tests.py

@@ -48,35 +48,60 @@ def test_validate_python_date_format_raises(payload) -> None:
         validate_python_date_format(payload)
 
 
-def test_dataset_put_schema_includes_currency_code_column() -> None:
-    """Test that DatasetPutSchema properly handles currency_code_column field."""
-    from superset.datasets.schemas import DatasetPutSchema
+def test_dataset_post_schema_has_all_put_scalar_fields() -> None:
+    """
+    Every scalar model field accepted by DatasetPutSchema should also be accepted
+    by DatasetPostSchema, unless it is intentionally excluded (fields that only
+    make sense after a dataset already exists).
 
-    schema = DatasetPutSchema()
+    This prevents the class of bug where a new column is added to the update
+    schema but forgotten in the create schema.
+    """
+    from superset.datasets.schemas import DatasetPostSchema, DatasetPutSchema
 
-    # Dataset with currency code column
+    # Fields that are intentionally only on Put: they require an existing dataset
+    # or are populated server-side during creation.
+    put_only_fields = {
+        "columns",
+        "metrics",
+        "folders",
+        "database_id",  # Post uses "database" (integer id) instead
+        "description",
+        "main_dttm_col",
+        "filter_select_enabled",
+        "fetch_values_predicate",
+        "offset",
+        "default_endpoint",
+        "cache_timeout",
+        "is_sqllab_view",
+        "extra",
+    }
+
+    put_fields = set(DatasetPutSchema().fields.keys())
+    post_fields = set(DatasetPostSchema().fields.keys())
+
+    missing = put_fields - post_fields - put_only_fields
+    assert missing == set(), (
+        f"Fields {missing} are in DatasetPutSchema but missing from "
+        f"DatasetPostSchema. Either add them to DatasetPostSchema or to "
+        f"the put_only_fields exclusion list in this test."
+    )
+
+
+def test_dataset_post_schema_includes_currency_code_column() -> None:
+    """Test that DatasetPostSchema accepts currency_code_column."""
+    from superset.datasets.schemas import DatasetPostSchema
+
+    schema = DatasetPostSchema()
     data = {
+        "database": 1,
+        "table_name": "virtual_dataset",
         "currency_code_column": "currency",
     }
     result = schema.load(data)
     assert result["currency_code_column"] == "currency"
 
 
-def test_dataset_put_schema_currency_code_column_optional() -> None:
-    """Test that currency_code_column is optional in DatasetPutSchema."""
-    from superset.datasets.schemas import DatasetPutSchema
-
-    schema = DatasetPutSchema()
-
-    # Dataset without currency code column (should not fail)
-    data: dict[str, str | None] = {}
-    result = schema.load(data)
-    assert (
-        "currency_code_column" not in result
-        or result.get("currency_code_column") is None
-    )
-
-
 def test_dataset_metrics_put_schema_parses_currency_string() -> None:
     """Test that DatasetMetricsPutSchema parses string currency payloads."""
     from superset.datasets.schemas import DatasetMetricsPutSchema