Honor normalize loader_file_format config

dlt/normalize/configuration.py

@@ -2,7 +2,9 @@
 
 from dlt.common.configuration import configspec
 from dlt.common.configuration.specs import BaseConfiguration
+from dlt.common.exceptions import ValueErrorWithKnownValues
 from dlt.common.destination import DestinationCapabilitiesContext
+from dlt.common.destination.capabilities import LOADER_FILE_FORMATS, TLoaderFileFormat
 from dlt.common.runners.configuration import PoolRunnerConfiguration, TPoolType
 from dlt.common.storages import (
     LoadStorageConfiguration,
@@ -22,6 +24,7 @@ class ItemsNormalizerConfiguration(BaseConfiguration):
 @configspec
 class NormalizeConfiguration(PoolRunnerConfiguration):
     pool_type: TPoolType = "process"
+    loader_file_format: Optional[TLoaderFileFormat] = None
     destination_capabilities: DestinationCapabilitiesContext = None  # injectable
     _schema_storage_config: SchemaStorageConfiguration = None
     _normalize_storage_config: NormalizeStorageConfiguration = None
@@ -41,3 +44,7 @@ class NormalizeConfiguration(PoolRunnerConfiguration):
 
     def on_resolved(self) -> None:
         self.pool_type = "none" if self.workers == 1 else "process"
+        if self.loader_file_format and self.loader_file_format not in LOADER_FILE_FORMATS:
+            raise ValueErrorWithKnownValues(
+                "loader_file_format", self.loader_file_format, LOADER_FILE_FORMATS
+            )

dlt/normalize/worker.py

@@ -77,7 +77,8 @@ def w_normalize_files(
     item_normalizers: Dict[str, ItemsNormalizer] = {}
 
     preferred_file_format = (
-        destination_caps.preferred_loader_file_format
+        config.loader_file_format
+        or destination_caps.preferred_loader_file_format
         or destination_caps.preferred_staging_file_format
     )
     # TODO: capabilities.supported_*_formats can be None, it should have defaults

tests/normalize/test_normalize.py

@@ -158,6 +158,26 @@ def test_normalize_single_user_event_insert(
     assert "(7005479104644416710," in event_text
 
 
+@pytest.mark.parametrize("caps", [INSERT_CAPS[0]], indirect=True)
+def test_normalize_loader_file_format_config_overrides_destination_preference(
+    caps: DestinationCapabilitiesContext, raw_normalize: Normalize
+) -> None:
+    assert caps.preferred_loader_file_format == "insert_values"
+    raw_normalize.config.loader_file_format = "jsonl"
+
+    load_id = extract_and_normalize_cases(raw_normalize, ["event.event.user_load_1"])
+
+    _, load_files = expect_load_package(
+        raw_normalize.load_storage,
+        "jsonl",
+        load_id,
+        EXPECTED_USER_TABLES,
+    )
+    event_text, lines = get_line_from_file(raw_normalize.load_storage, load_files["event"], 0)
+    assert lines == 1
+    assert json.loads(event_text)["event"] == "user"
+
+
 @pytest.mark.parametrize("caps", JSONL_CAPS, indirect=True)
 def test_normalize_filter_user_event(
     caps: DestinationCapabilitiesContext, rasa_normalize: Normalize

tests/pipeline/test_pipeline.py

@@ -83,7 +83,12 @@
 from dlt.pipeline.typing import TPipelineStep
 
 from tests.common.utils import TEST_SENTRY_DSN
-from tests.utils import capture_dlt_logger, get_test_storage_root, skipifwindows
+from tests.utils import (
+    TEST_DICT_CONFIG_PROVIDER,
+    capture_dlt_logger,
+    get_test_storage_root,
+    skipifwindows,
+)
 from tests.extract.utils import expect_extracted_file
 from tests.pipeline.utils import (
     PIPELINE_TEST_CASES_PATH,
@@ -4219,6 +4224,24 @@ def _data():
     assert pipeline.default_schema.get_table("_datax")["file_format"] == "jsonl"
 
 
+def test_normalize_loader_file_format_config_sets_default_format() -> None:
+    pipeline = dlt.pipeline(
+        pipeline_name="test_normalize_loader_file_format_config_sets_default_format",
+        destination="duckdb",
+        dev_mode=True,
+    )
+    pipeline.config.restore_from_destination = False
+
+    with TEST_DICT_CONFIG_PROVIDER().values({"normalize": {"loader_file_format": "jsonl"}}):
+        pipeline.extract([1, 2, 3], table_name="numbers")
+        normalize_info = pipeline.normalize()
+
+    jobs = pipeline._get_load_storage().normalized_packages.list_new_jobs(
+        normalize_info.loads_ids[0]
+    )
+    assert ParsedLoadJobFileName.parse(jobs[0]).file_format == "jsonl"
+
+
 @pytest.mark.parametrize(
     "use_single_dataset",
     [True, False],