Make sure that column types are not inferred to VARCHAR (#1006)

The `UNION BY NAME` approach introduced as a fix for issue #977 led to
the problem that if one of the files contained only empty values for an
optional column, that column would be inferred to be of type `VARCHAR`
by DuckDB.

The solution is to create the empty table with strict types based on
the data contract and then insert the data into the table in a separate
step, taking only into account those columns that exist in both the data
contract and the data, which might be missing newly introduced columns
or still have columns that have been removed from the contract by now.

Author: dwestheide

datacontract/engines/soda/connections/duckdb_connection.py

@@ -78,15 +78,21 @@ def create_view_with_schema_union(con, schema_obj: SchemaObject, model_path: str
     if converted_types:
         # Create empty table with contract schema
         columns_def = [f'"{col_name}" {col_type}' for col_name, col_type in converted_types.items()]
-        create_empty_table = f"""CREATE TABLE "{model_name}_schema" ({', '.join(columns_def)});"""
+        create_empty_table = f"""CREATE TABLE "{model_name}" ({', '.join(columns_def)});"""
         con.sql(create_empty_table)
 
-        # Create view as UNION of empty schema table and data
-        create_view_sql = f"""CREATE VIEW "{model_name}" AS
-            SELECT * FROM "{model_name}_schema"
-            UNION ALL BY NAME
-            SELECT * FROM {read_function}('{model_path}', union_by_name=true, hive_partitioning=1);"""
-        con.sql(create_view_sql)
+        # 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))
+            INTERSECT SELECT column_name
+            FROM information_schema.columns
+            WHERE table_name = '{model_name}'""").fetchall()
+        selected_columns = ', '.join([column[0] for column in intersecting_columns])
+
+        # Insert data into table by name, but only columns existing in contract and data
+        insert_data_sql = f"""INSERT INTO {model_name} BY NAME
+            (SELECT {selected_columns} FROM {read_function}('{model_path}', union_by_name=true, hive_partitioning=1));"""
+        con.sql(insert_data_sql)
     else:
         # Fallback
         con.sql(

tests/fixtures/schema-evolution/cities-version-3.csv

@@ -0,0 +1,3 @@
+city_id,name,population
+c5193fd1-6b9b-4abc-81c5-ddb30a52d8f6,Babylon,
+cdcc7194-8f4d-40a8-84bb-92b32e0fb684,Tokyo,

tests/test_test_schema_evolution.py

@@ -9,7 +9,6 @@ def test_csv_optional_field_missing_from_old_data():
 
     run = data_contract.test()
 
-    print(run)
     assert run.result == "passed"
     assert all(check.result == "passed" for check in run.checks)
 
@@ -21,7 +20,6 @@ def test_csv_optional_field_present_in_new_data():
 
     run = data_contract.test()
 
-    print(run)
     assert run.result == "passed"
     assert all(check.result == "passed" for check in run.checks)
 
@@ -32,7 +30,6 @@ def test_data_from_historical_and_current_schema_csv_mixed():
 
     run = data_contract.test()
 
-    print(run)
     assert run.result == "passed"
     assert all(check.result == "passed" for check in run.checks)
 
@@ -43,7 +40,6 @@ def test_csv_optional_field_with_invalid_values():
 
     run = data_contract.test()
 
-    print(run)
     assert run.result == "failed"
     # Should have at least one failed check for constraint violation
     assert any(check.result == "failed" for check in run.checks)
@@ -55,7 +51,6 @@ def test_csv_required_field_missing_fails():
 
     run = data_contract.test()
 
-    print(run)
     assert run.result == "failed"
 
 
@@ -70,7 +65,6 @@ def test_parquet_optional_field_missing_from_old_data():
 
     run = data_contract.test()
 
-    print(run)
     assert run.result == "passed"
     assert all(check.result == "passed" for check in run.checks)
 
@@ -82,7 +76,6 @@ def test_parquet_optional_field_present_in_new_data():
 
     run = data_contract.test()
 
-    print(run)
     assert run.result == "passed"
     assert all(check.result == "passed" for check in run.checks)
 
@@ -93,7 +86,6 @@ def test_data_from_historical_and_current_schema_parquet_mixed():
 
     run = data_contract.test()
 
-    print(run)
     assert run.result == "passed"
     assert all(check.result == "passed" for check in run.checks)
 
@@ -104,7 +96,6 @@ def test_parquet_optional_field_with_invalid_values():
 
     run = data_contract.test()
 
-    print(run)
     assert run.result == "failed"
     # Should have at least one failed check for constraint violation
     assert any(check.result == "failed" for check in run.checks)
@@ -116,5 +107,4 @@ def test_parquet_required_field_missing_fails():
 
     run = data_contract.test()
 
-    print(run)
     assert run.result == "failed"