Fix comments, add tests

dlt/common/destination/reference.py

@@ -562,9 +562,9 @@ def with_staging_dataset(self) -> ContextManager["JobClientBase"]:
 class SupportsStagingDestination:
     """Adds capability to support a staging destination for the load"""
 
-    def __init__(self, truncate_table_before_load_on_staging_destination: bool = True) -> None:
+    def __init__(self, config: DestinationClientDwhWithStagingConfiguration) -> None:
         self.truncate_table_before_load_on_staging_destination = (
-            truncate_table_before_load_on_staging_destination
+            config.truncate_table_before_load_on_staging_destination
         )
 
     def should_load_data_to_staging_dataset_on_staging_destination(

dlt/destinations/impl/athena/athena.py

@@ -389,9 +389,7 @@ def __init__(
             config,
             capabilities,
         )
-        SupportsStagingDestination.__init__(
-            self, config.truncate_table_before_load_on_staging_destination
-        )
+        SupportsStagingDestination.__init__(self, config)
         super().__init__(schema, config, sql_client)
         self.sql_client: AthenaSQLClient = sql_client  # type: ignore
         self.config: AthenaClientConfiguration = config
@@ -532,7 +530,7 @@ def should_truncate_table_before_load_on_staging_destination(self, table: TTable
         if table["write_disposition"] == "replace" and not self._is_iceberg_table(
             self.prepare_load_table(table["name"])
         ):
-            return True
+            return self.truncate_table_before_load_on_staging_destination
         return False
 
     def should_load_data_to_staging_dataset_on_staging_destination(

dlt/destinations/impl/clickhouse/clickhouse.py

@@ -283,9 +283,7 @@ def __init__(
             capabilities,
             config,
         )
-        SupportsStagingDestination.__init__(
-            self, config.truncate_table_before_load_on_staging_destination
-        )
+        SupportsStagingDestination.__init__(self, config)
         super().__init__(schema, config, self.sql_client)
         self.config: ClickHouseClientConfiguration = config
         self.active_hints = deepcopy(HINT_TO_CLICKHOUSE_ATTR)

dlt/destinations/impl/databricks/databricks.py

@@ -262,9 +262,7 @@ def __init__(
             config.credentials,
             capabilities,
         )
-        SupportsStagingDestination.__init__(
-            self, config.truncate_table_before_load_on_staging_destination
-        )
+        SupportsStagingDestination.__init__(self, config)
         super().__init__(schema, config, sql_client)
         self.config: DatabricksClientConfiguration = config
         self.sql_client: DatabricksSqlClient = sql_client  # type: ignore[assignment]

dlt/destinations/impl/dremio/dremio.py

@@ -147,9 +147,7 @@ def __init__(
             config.credentials,
             capabilities,
         )
-        SupportsStagingDestination.__init__(
-            self, config.truncate_table_before_load_on_staging_destination
-        )
+        SupportsStagingDestination.__init__(self, config)
         super().__init__(schema, config, sql_client)
         self.config: DremioClientConfiguration = config
         self.sql_client: DremioSqlClient = sql_client  # type: ignore

dlt/destinations/impl/dummy/configuration.py

@@ -34,6 +34,8 @@ class DummyClientConfiguration(DestinationClientConfiguration):
     """raise terminal exception in job init"""
     fail_transiently_in_init: bool = False
     """raise transient exception in job init"""
+    truncate_table_before_load_on_staging_destination: bool = True
+    """truncate tables on staging destination"""
 
     # new jobs workflows
     create_followup_jobs: bool = False

dlt/destinations/impl/dummy/dummy.py

@@ -127,7 +127,7 @@ def __init__(
         config: DummyClientConfiguration,
         capabilities: DestinationCapabilitiesContext,
     ) -> None:
-        SupportsStagingDestination.__init__(self)
+        SupportsStagingDestination.__init__(self, config)
         super().__init__(schema, config, capabilities)
         self.in_staging_context = False
         self.config: DummyClientConfiguration = config

dlt/destinations/impl/redshift/redshift.py

@@ -233,9 +233,7 @@ def __init__(
             config.credentials,
             capabilities,
         )
-        SupportsStagingDestination.__init__(
-            self, config.truncate_table_before_load_on_staging_destination
-        )
+        SupportsStagingDestination.__init__(self, config)
         super().__init__(schema, config, sql_client)
         self.sql_client = sql_client
         self.config: RedshiftClientConfiguration = config

dlt/destinations/impl/snowflake/snowflake.py

@@ -266,9 +266,7 @@ def __init__(
             capabilities,
             config.query_tag,
         )
-        SupportsStagingDestination.__init__(
-            self, config.truncate_table_before_load_on_staging_destination
-        )
+        SupportsStagingDestination.__init__(self, config)
         super().__init__(schema, config, sql_client)
         self.config: SnowflakeClientConfiguration = config
         self.sql_client: SnowflakeSqlClient = sql_client  # type: ignore

dlt/destinations/impl/synapse/synapse.py

@@ -59,6 +59,7 @@ def __init__(
         config: SynapseClientConfiguration,
         capabilities: DestinationCapabilitiesContext,
     ) -> None:
+        SupportsStagingDestination.__init__(self, config)
         super().__init__(schema, config, capabilities)
         self.config: SynapseClientConfiguration = config
         self.sql_client = SynapseSqlClient(

docs/website/docs/dlt-ecosystem/staging.md

@@ -42,9 +42,13 @@ truncate_staging_dataset=true
 `dlt` allows to chain destinations where the first one (`staging`) is responsible for uploading the files from local filesystem to the remote storage. It then generates followup jobs for the second destination that (typically) copy the files from remote storage into destination.
 
 Currently, only one destination the [filesystem](destinations/filesystem.md) can be used as a staging. Following destinations can copy remote files:
-1. [Redshift.](destinations/redshift.md#staging-support)
-2. [Bigquery.](destinations/bigquery.md#staging-support)
-3. [Snowflake.](destinations/snowflake.md#staging-support)
+
+1. [Azure Synapse](destinations/synapse#staging-support)
+1. [Athena](destinations/athena#staging-support)
+1. [Bigquery](destinations/bigquery.md#staging-support)
+1. [Dremio](destinations/dremio#staging-support)
+1. [Redshift](destinations/redshift.md#staging-support)
+1. [Snowflake](destinations/snowflake.md#staging-support)
 
 ### How to use
 In essence, you need to set up two destinations and then pass them to `dlt.pipeline`. Below we'll use `filesystem` staging with `parquet` files to load into `Redshift` destination.
@@ -96,4 +100,21 @@ In essence, you need to set up two destinations and then pass them to `dlt.pipel
 
     Run the pipeline script as usual.
 
-> 💡 Please note that `dlt` does not delete loaded files from the staging storage after the load is complete.
+:::tip
+Please note that `dlt` does not delete loaded files from the staging storage after the load is complete, but it trunkate previously loaded files.
+:::
+
+### How to prevent staging files truncation
+
+Before `dlt` loads data to staging storage it truncates previously loaded files. To prevent it and keep the whole history
+of loaded files you can use the following parameter:
+
+```toml
+[destination.redshift]
+truncate_table_before_load_on_staging_destination=false
+```
+
+:::caution
+[Athena](destinations/athena#staging-support) destination only truncate not iceberg tables with `replace` merge_disposition.
+Therefore, parameter `truncate_table_before_load_on_staging_destination` only control truncation of corresponding files for these tables.
+:::

tests/load/test_dummy_client.py

@@ -512,6 +512,23 @@ def test_completed_loop_with_delete_completed() -> None:
     assert_complete_job(load, should_delete_completed=True)
 
 
+@pytest.mark.parametrize("to_truncate", [True, False])
+def test_truncate_table_before_load_on_stanging(to_truncate) -> None:
+    load = setup_loader(
+        client_config=DummyClientConfiguration(
+            truncate_table_before_load_on_staging_destination=to_truncate
+        )
+    )
+    load_id, schema = prepare_load_package(load.load_storage, NORMALIZED_FILES)
+    destination_client = load.get_destination_client(schema)
+    assert (
+        destination_client.should_truncate_table_before_load_on_staging_destination(
+            schema.tables["_dlt_version"]
+        )
+        == to_truncate
+    )
+
+
 def test_retry_on_new_loop() -> None:
     # test job that retries sitting in new jobs
     load = setup_loader(client_config=DummyClientConfiguration(retry_prob=1.0))