fix: detect missing fields in CSV/Parquet files correctly

datacontract/engines/data_contract_checks.py

@@ -109,7 +109,12 @@ def to_schema_checks(schema_object: SchemaObject, server: Server) -> List[Check]
         property_name = prop.name
         logical_type = prop.logicalType
 
-        checks.append(check_property_is_present(schema_name, property_name, quoting_config))
+        checks.append(
+            check_property_is_present(
+                schema_name, property_name, quoting_config,
+                use_raw_model=(server_type in ["local", "s3", "gcs", "azure"] and server.format in ["csv", "parquet"]),
+            )
+        )
         if check_types and logical_type is not None:
             sql_type: str = convert_to_sql_type(prop, server_type)
             checks.append(check_property_type(schema_name, property_name, sql_type, quoting_config))
@@ -192,11 +197,17 @@ def to_schema_name(schema_object: SchemaObject, server_type: str) -> str:
     return schema_object.name
 
 
-def check_property_is_present(model_name, field_name, quoting_config: QuotingConfig = QuotingConfig()) -> Check:
+def check_property_is_present(
+    model_name, field_name, quoting_config: QuotingConfig = QuotingConfig(), use_raw_model: bool = False
+) -> Check:
     check_type = "field_is_present"
     check_key = f"{model_name}__{field_name}__{check_type}"
+    # For CSV/Parquet on DuckDB, use the _raw view to detect truly missing columns.
+    # The unioned table fills missing columns with NULLs, making the check always pass.
+    # See https://github.com/datacontract/datacontract-cli/issues/1065
+    table_name = f"{model_name}_raw" if use_raw_model else model_name
     sodacl_check_dict = {
-        checks_for(model_name, quoting_config, check_type): [
+        checks_for(table_name, quoting_config, check_type): [
             {
                 "schema": {
                     "name": check_key,

datacontract/engines/soda/connections/duckdb_connection.py

@@ -96,6 +96,15 @@ def create_view_with_schema_union(con, schema_obj: SchemaObject, model_path: str
         create_empty_table = f"""CREATE TABLE "{model_name}" ({", ".join(columns_def)});"""
         con.sql(create_empty_table)
 
+        # Also create a raw view for field_is_present checks, so missing columns
+        # are actually absent (not filled with NULLs from the unioned table).
+        # See https://github.com/datacontract/datacontract-cli/issues/1065
+        raw_view_name = f"{model_name}_raw"
+        con.sql(
+            f"""CREATE VIEW "{raw_view_name}" AS
+                SELECT * FROM {read_function}('{model_path}', union_by_name=true, hive_partitioning=1);"""
+        )
+
         # Read columns existing in both current data contract and data
         intersecting_columns = con.sql(f"""SELECT column_name
             FROM (DESCRIBE SELECT * FROM {read_function}('{model_path}', union_by_name=true, hive_partitioning=1))

tests/test_test_schema_evolution.py

@@ -2,15 +2,21 @@
 
 
 def test_csv_optional_field_missing_from_old_data():
-    """Optional field not present in historical CSV data should pass validation"""
+    """Optional field not present in historical CSV data is correctly detected as missing"""
     data_contract = DataContract(
         data_contract_file="fixtures/schema-evolution/odcs-datacontract-cities-version-2.yaml", server="historical-csv"
     )
 
     run = data_contract.test()
 
-    assert run.result == "passed"
-    assert all(check.result == "passed" for check in run.checks)
+    # field_is_present should detect that 'population' is missing from historical CSV
+    assert run.result == "failed"
+    missing_field_checks = [c for c in run.checks if c.type == "field_is_present" and c.field == "population"]
+    assert len(missing_field_checks) == 1
+    assert missing_field_checks[0].result == "failed"
+    # Other field_is_present checks should still pass
+    present_field_checks = [c for c in run.checks if c.type == "field_is_present" and c.field != "population"]
+    assert all(c.result == "passed" for c in present_field_checks)
 
 
 def test_csv_optional_field_present_in_new_data():
@@ -67,16 +73,22 @@ def test_csv_required_field_missing_fails():
 
 
 def test_parquet_optional_field_missing_from_old_data():
-    """Optional field not present in historical Parquet data should pass validation"""
+    """Optional field not present in historical Parquet data is correctly detected as missing"""
     data_contract = DataContract(
         data_contract_file="fixtures/schema-evolution/odcs-datacontract-cities-version-2.yaml",
         server="historical-parquet",
     )
 
     run = data_contract.test()
 
-    assert run.result == "passed"
-    assert all(check.result == "passed" for check in run.checks)
+    # field_is_present should detect that 'population' is missing from historical Parquet
+    assert run.result == "failed"
+    missing_field_checks = [c for c in run.checks if c.type == "field_is_present" and c.field == "population"]
+    assert len(missing_field_checks) == 1
+    assert missing_field_checks[0].result == "failed"
+    # Other field_is_present checks should still pass
+    present_field_checks = [c for c in run.checks if c.type == "field_is_present" and c.field != "population"]
+    assert all(c.result == "passed" for c in present_field_checks)
 
 
 def test_parquet_optional_field_present_in_new_data():