feat(normalize): fail early on unsupported data type for file format

dlt/normalize/configuration.py

@@ -23,6 +23,8 @@ class ItemsNormalizerConfiguration(BaseConfiguration):
 class NormalizeConfiguration(PoolRunnerConfiguration):
     pool_type: TPoolType = "process"
     destination_capabilities: DestinationCapabilitiesContext = None  # injectable
+    destination_type: Optional[str] = None
+    """Type of the destination the data is normalized for, used to report unsupported data types."""
     _schema_storage_config: SchemaStorageConfiguration = None
     _normalize_storage_config: NormalizeStorageConfiguration = None
     _load_storage_config: LoadStorageConfiguration = None

dlt/normalize/worker.py

@@ -10,7 +10,7 @@
     get_best_writer_spec,
     is_native_writer,
 )
-from dlt.common.destination.utils import prepare_load_table
+from dlt.common.destination.utils import prepare_load_table, verify_supported_data_types
 from dlt.common.metrics import DataWriterMetrics
 from dlt.common.schema.utils import new_table
 from dlt.common.typing import TLoaderFileFormat
@@ -271,6 +271,29 @@ def _gather_metrics_and_close(
             raise NormalizeJobFailed(load_id, job_id, str(exc), writer_metrics) from exc
         else:
             writer_metrics = _gather_metrics_and_close(parsed_file_name, in_exception=False)
+            # fail early if any discovered data type is not supported by the destination for the
+            # file format that was actually selected for the job, instead of failing at load time
+            new_jobs = [
+                ParsedLoadJobFileName.parse(metrics.file_path) for metrics in writer_metrics
+            ]
+            # model and reference jobs carry no typed data file, so they have no format to check
+            new_jobs = [job for job in new_jobs if job.file_format not in ("model", "reference")]
+            verified_tables = {job.table_name for job in new_jobs}
+            prepared_tables = [
+                prepare_load_table(schema.tables, schema.tables[table_name], destination_caps)
+                for table_name in verified_tables
+                if table_name in schema.tables
+            ]
+            if exceptions := verify_supported_data_types(
+                prepared_tables,
+                new_jobs,
+                destination_caps,
+                config.destination_type,
+                warnings=False,
+            ):
+                for exception in exceptions:
+                    logger.error(str(exception))
+                raise exceptions[0]
         finally:
             for normalizer in item_normalizers.values():
                 normalizer.close()

dlt/pipeline/pipeline.py

@@ -545,6 +545,7 @@ def normalize(self, workers: int = 1) -> NormalizeInfo:
         # create default normalize config
         normalize_config = NormalizeConfiguration(
             workers=workers,
+            destination_type=self._destination.destination_type,
             _schema_storage_config=self._schema_storage_config,
             _normalize_storage_config=self._normalize_storage_config(),
             _load_storage_config=self._load_storage_config(),

tests/normalize/test_normalize.py

@@ -19,6 +19,7 @@
 from dlt.common.data_types import TDataType
 from dlt.common.storages import NormalizeStorage, LoadStorage, ParsedLoadJobFileName, PackageStorage
 from dlt.common.destination import DestinationCapabilitiesContext
+from dlt.common.destination.exceptions import UnsupportedDataType
 from dlt.common.runtime.collector import DictCollector
 from dlt.common.configuration.container import Container
 
@@ -27,6 +28,7 @@
 from dlt.normalize.validate import validate_and_update_schema
 from dlt.normalize.worker import group_worker_files
 from dlt.normalize.exceptions import NormalizeJobFailed
+from dlt.destinations import bigquery
 
 from tests.cases import JSON_TYPED_DICT, JSON_TYPED_DICT_TYPES
 from tests.utils import (
@@ -729,6 +731,28 @@ def test_collect_empty_metrics_on_exception(raw_normalize: Normalize, pool_worke
     assert len(step_info.load_packages[0].jobs["new_jobs"]) == 1
 
 
+@pytest.mark.parametrize("caps", [bigquery().capabilities], indirect=True)
+def test_normalize_fails_on_unsupported_data_type(
+    caps: DestinationCapabilitiesContext, raw_normalize: Normalize
+) -> None:
+    pytest.importorskip("pyarrow")
+    # bigquery cannot load json columns from parquet, this must surface during normalize not load
+    schema = Schema("event")
+    table = new_table(
+        "data", write_disposition="append", columns=[{"name": "payload", "data_type": "json"}]
+    )
+    table["file_format"] = "parquet"
+    schema.update_table(table)
+
+    extract_items(raw_normalize.normalize_storage, [{"payload": {"a": 1}}], schema, "data")
+
+    with pytest.raises(UnsupportedDataType) as exc:
+        normalize_pending(raw_normalize, schema)
+    assert exc.value.column == "payload"
+    assert exc.value.data_type == "json"
+    assert exc.value.file_format == "parquet"
+
+
 @pytest.mark.parametrize("pool_workers", (1, 2))
 def test_collect_metrics_on_exception(raw_normalize: Normalize, pool_workers: int) -> None:
     # files below have 3 elements and we want to split them as such