Add Delta table support for filesystem destination

.github/workflows/test_destinations.yml

@@ -75,7 +75,7 @@ jobs:
 
       - name: Install dependencies
         # if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true'
-        run: poetry install --no-interaction -E redshift -E gs -E s3 -E az -E parquet -E duckdb -E cli --with sentry-sdk --with pipeline
+        run: poetry install --no-interaction -E redshift -E gs -E s3 -E az -E parquet -E duckdb -E cli --with sentry-sdk --with pipeline -E deltalake
 
       - name: create secrets.toml
         run: pwd && echo "$DLT_SECRETS_TOML" > tests/.dlt/secrets.toml

.github/workflows/test_local_destinations.yml

@@ -90,7 +90,7 @@ jobs:
           key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('**/poetry.lock') }}-local-destinations
 
       - name: Install dependencies
-        run: poetry install --no-interaction -E postgres -E duckdb -E parquet -E filesystem -E cli -E weaviate --with sentry-sdk --with pipeline
+        run: poetry install --no-interaction -E postgres -E duckdb -E parquet -E filesystem -E cli -E weaviate --with sentry-sdk --with pipeline -E deltalake
 
       - name: create secrets.toml
         run: pwd && echo "$DLT_SECRETS_TOML" > tests/.dlt/secrets.toml

dlt/common/destination/reference.py

@@ -214,6 +214,20 @@ def exception(self) -> str:
         pass
 
 
+class DirectoryLoadJob:
+    """Job that loads a directory of files in a single transaction."""
+
+    def __init__(self, dir_name: str) -> None:
+        self._dir_name = dir_name
+
+    def dir_name(self) -> str:
+        """Returns name of directory containing the job files."""
+        return self._dir_name
+
+    def job_id(self) -> str:
+        return "hacked_job_id"
+
+
 class NewLoadJob(LoadJob):
     """Adds a trait that allows to save new job file"""
 
@@ -309,8 +323,12 @@ def restore_file_load(self, file_path: str) -> LoadJob:
         """Finds and restores already started loading job identified by `file_path` if destination supports it."""
         pass
 
+    def can_do_logical_replace(self, table: TTableSchema) -> bool:
+        """Returns True if `replace` can be done without physically deleting data."""
+        return table["table_format"] == "delta"
+
     def should_truncate_table_before_load(self, table: TTableSchema) -> bool:
-        return table["write_disposition"] == "replace"
+        return table["write_disposition"] == "replace" and not self.can_do_logical_replace(table)
 
     def create_table_chain_completed_followup_jobs(
         self, table_chain: Sequence[TTableSchema]

dlt/common/libs/pyarrow.py

@@ -37,6 +37,8 @@
 
 TAnyArrowItem = Union[pyarrow.Table, pyarrow.RecordBatch]
 
+ARROW_DECIMAL_MAX_PRECISION = 76
+
 
 def get_py_arrow_datatype(
     column: TColumnType,
@@ -397,6 +399,25 @@ def pq_stream_with_new_columns(
             yield tbl
 
 
+def adjust_arrow_schema(
+    schema: pyarrow.Schema,
+    type_map: Dict[Callable[[pyarrow.DataType], bool], Callable[..., pyarrow.DataType]],
+) -> pyarrow.Schema:
+    """Returns adjusted Arrow schema.
+
+    Replaces data types for fields matching a type check in `type_map`.
+    Type check functions in `type_map` are assumed to be mutually exclusive, i.e.
+    a data type does not match more than one type check function.
+    """
+    for i, e in enumerate(schema.types):
+        for type_check, cast_type in type_map.items():
+            if type_check(e):
+                adjusted_field = schema.field(i).with_type(cast_type)
+                schema = schema.set(i, adjusted_field)
+                break  # if type matches type check, do not do other type checks
+    return schema
+
+
 class NameNormalizationClash(ValueError):
     def __init__(self, reason: str) -> None:
         msg = f"Arrow column name clash after input data normalization. {reason}"

dlt/common/schema/typing.py

@@ -64,7 +64,7 @@
     "dedup_sort",
 ]
 """Known hints of a column used to declare hint regexes."""
-TTableFormat = Literal["iceberg", "parquet", "jsonl"]
+TTableFormat = Literal["iceberg", "parquet", "jsonl", "delta"]
 TTypeDetections = Literal[
     "timestamp", "iso_timestamp", "iso_date", "large_integer", "hexbytes_to_text", "wei_to_double"
 ]

dlt/common/storages/data_item_storage.py

@@ -1,10 +1,11 @@
 from pathlib import Path
-from typing import Dict, Any, List, Sequence
+from typing import Dict, Any, List, Optional
 from abc import ABC, abstractmethod
 
 from dlt.common import logger
 from dlt.common.schema import TTableSchemaColumns
 from dlt.common.typing import StrAny, TDataItems
+from dlt.common.storages.load_package import PackageStorage
 from dlt.common.data_writers import (
     BufferedDataWriter,
     DataWriter,

dlt/common/storages/load_package.py

@@ -22,6 +22,7 @@
     Any,
     Tuple,
     TypedDict,
+    Union,
 )
 from typing_extensions import NotRequired
 
@@ -177,6 +178,15 @@ def __str__(self) -> str:
         return self.job_id()
 
 
+class ParsedLoadJobDirectoryName(NamedTuple):
+    table_name: str
+
+    @staticmethod
+    def parse(dir_name: str) -> "ParsedLoadJobDirectoryName":
+        table_name = Path(dir_name).name
+        return ParsedLoadJobDirectoryName(table_name=table_name)
+
+
 class LoadJobInfo(NamedTuple):
     state: TJobState
     file_path: str
@@ -316,11 +326,18 @@ def __init__(self, storage: FileStorage, initial_state: TLoadPackageStatus) -> N
     def get_package_path(self, load_id: str) -> str:
         return load_id
 
-    def get_job_folder_path(self, load_id: str, folder: TJobState) -> str:
-        return os.path.join(self.get_package_path(load_id), folder)
+    def get_job_folder_path(
+        self, load_id: str, folder: TJobState, subfolder: Optional[str] = None
+    ) -> str:
+        if subfolder is None:
+            return os.path.join(self.get_package_path(load_id), folder)
+        else:
+            return os.path.join(self.get_package_path(load_id), folder, subfolder)
 
-    def get_job_file_path(self, load_id: str, folder: TJobState, file_name: str) -> str:
-        return os.path.join(self.get_job_folder_path(load_id, folder), file_name)
+    def get_job_file_path(
+        self, load_id: str, folder: TJobState, file_name: str, subfolder: Optional[str] = None
+    ) -> str:
+        return os.path.join(self.get_job_folder_path(load_id, folder, subfolder), file_name)
 
     def list_packages(self) -> Sequence[str]:
         """Lists all load ids in storage, earliest first
@@ -331,11 +348,17 @@ def list_packages(self) -> Sequence[str]:
         # start from the oldest packages
         return sorted(loads)
 
-    def list_new_jobs(self, load_id: str) -> Sequence[str]:
-        new_jobs = self.storage.list_folder_files(
-            self.get_job_folder_path(load_id, PackageStorage.NEW_JOBS_FOLDER)
-        )
-        return new_jobs
+    def list_new_jobs(self, load_id: str, root_only: bool = False) -> Sequence[str]:
+        root_dir = self.get_job_folder_path(load_id, PackageStorage.NEW_JOBS_FOLDER)
+        if root_only:
+            return self.storage.list_folder_files(root_dir)
+        sub_dirs = self.storage.list_folder_dirs(root_dir)
+        dirs = [root_dir] + sub_dirs
+        return [file for dir_ in dirs for file in self.storage.list_folder_files(dir_)]
+
+    def list_new_dir_jobs(self, load_id: str) -> Sequence[str]:
+        root_dir = self.get_job_folder_path(load_id, PackageStorage.NEW_JOBS_FOLDER)
+        return self.storage.list_folder_dirs(root_dir)
 
     def list_started_jobs(self, load_id: str) -> Sequence[str]:
         return self.storage.list_folder_files(
@@ -382,17 +405,19 @@ def import_job(
         """Adds new job by moving the `job_file_path` into `new_jobs` of package `load_id`"""
         self.storage.atomic_import(job_file_path, self.get_job_folder_path(load_id, job_state))
 
-    def start_job(self, load_id: str, file_name: str) -> str:
+    def start_job(self, load_id: str, job: Union["LoadJob", "DirectoryLoadJob"]) -> str:  # type: ignore[name-defined] # noqa: F821
         return self._move_job(
-            load_id, PackageStorage.NEW_JOBS_FOLDER, PackageStorage.STARTED_JOBS_FOLDER, file_name
+            load_id, PackageStorage.NEW_JOBS_FOLDER, PackageStorage.STARTED_JOBS_FOLDER, job
         )
 
-    def fail_job(self, load_id: str, file_name: str, failed_message: Optional[str]) -> str:
+    def fail_job(
+        self, load_id: str, job: Union["LoadJob", "DirectoryLoadJob"], failed_message: Optional[str]  # type: ignore[name-defined] # noqa: F821
+    ) -> str:
         # save the exception to failed jobs
         if failed_message:
             self.storage.save(
                 self.get_job_file_path(
-                    load_id, PackageStorage.FAILED_JOBS_FOLDER, file_name + ".exception"
+                    load_id, PackageStorage.FAILED_JOBS_FOLDER, job.file_name() + ".exception"
                 ),
                 failed_message,
             )
@@ -401,28 +426,30 @@ def fail_job(self, load_id: str, file_name: str, failed_message: Optional[str])
             load_id,
             PackageStorage.STARTED_JOBS_FOLDER,
             PackageStorage.FAILED_JOBS_FOLDER,
-            file_name,
+            job.file_name(),
         )
 
-    def retry_job(self, load_id: str, file_name: str) -> str:
+    def retry_job(self, load_id: str, job: Union["LoadJob", "DirectoryLoadJob"]) -> str:  # type: ignore[name-defined] # noqa: F821
         # when retrying job we must increase the retry count
-        source_fn = ParsedLoadJobFileName.parse(file_name)
+        source_fn = ParsedLoadJobFileName.parse(job.file_name())
         dest_fn = source_fn.with_retry()
         # move it directly to new file name
         return self._move_job(
             load_id,
             PackageStorage.STARTED_JOBS_FOLDER,
             PackageStorage.NEW_JOBS_FOLDER,
-            file_name,
+            job.file_name(),
             dest_fn.file_name(),
         )
 
-    def complete_job(self, load_id: str, file_name: str) -> str:
+    def complete_job(
+        self, load_id: str, job: Union["LoadJob", "DirectoryLoadJob"]  # type: ignore[name-defined] # noqa: F821
+    ) -> str:
         return self._move_job(
             load_id,
             PackageStorage.STARTED_JOBS_FOLDER,
             PackageStorage.COMPLETED_JOBS_FOLDER,
-            file_name,
+            job,
         )
 
     #
@@ -601,15 +628,25 @@ def _move_job(
         load_id: str,
         source_folder: TJobState,
         dest_folder: TJobState,
-        file_name: str,
+        job: Union["LoadJob", "DirectoryLoadJob"],  # type: ignore[name-defined] # noqa: F821
         new_file_name: str = None,
     ) -> str:
-        # ensure we move file names, not paths
-        assert file_name == FileStorage.get_file_name_from_file_path(file_name)
+        from dlt.common.destination.reference import LoadJob, DirectoryLoadJob
+
         load_path = self.get_package_path(load_id)
-        dest_path = os.path.join(load_path, dest_folder, new_file_name or file_name)
-        self.storage.atomic_rename(os.path.join(load_path, source_folder, file_name), dest_path)
-        # print(f"{join(load_path, source_folder, file_name)} -> {dest_path}")
+
+        if isinstance(job, LoadJob):
+            source_name = job.file_name()
+            # ensure we move file names, not paths
+            assert source_name == FileStorage.get_file_name_from_file_path(source_name)
+            dest_name = new_file_name or source_name
+        elif isinstance(job, DirectoryLoadJob):
+            source_name = job.dir_name()
+            dest_name = job.dir_name()
+
+        source_path = os.path.join(load_path, source_folder, source_name)
+        dest_path = os.path.join(load_path, dest_folder, dest_name)
+        self.storage.atomic_rename(source_path, dest_path)
         return self.storage.make_full_path(dest_path)
 
     def _load_schema(self, load_id: str) -> DictStrAny:

dlt/common/storages/load_storage.py

@@ -34,11 +34,23 @@ def __init__(self, package_storage: PackageStorage, writer_spec: FileWriterSpec)
     def _get_data_item_path_template(self, load_id: str, _: str, table_name: str) -> str:
         # implements DataItemStorage._get_data_item_path_template
         file_name = PackageStorage.build_job_file_name(table_name, "%s")
+        subfolder = self._get_data_item_subfolder(load_id, table_name)
         file_path = self.package_storage.get_job_file_path(
-            load_id, PackageStorage.NEW_JOBS_FOLDER, file_name
+            load_id, PackageStorage.NEW_JOBS_FOLDER, file_name, subfolder
         )
         return self.package_storage.storage.make_full_path(file_path)
 
+    def _get_data_item_subfolder(self, load_id: str, table_name: str) -> Optional[str]:
+        """Returns name of subfolder for `table_name`.
+
+        Returns None if subfolder is not used.
+        """
+        subfolder = self.package_storage.get_job_folder_path(
+            load_id, PackageStorage.NEW_JOBS_FOLDER, table_name
+        )
+        subfolder_exists = self.package_storage.storage.has_folder(subfolder)
+        return table_name if subfolder_exists else None
+
 
 class LoadStorage(VersionedStorage):
     STORAGE_VERSION = "1.0.0"
@@ -95,19 +107,20 @@ def import_extracted_package(
 
     def list_new_jobs(self, load_id: str) -> Sequence[str]:
         """Lists all jobs in new jobs folder of normalized package storage and checks if file formats are supported"""
-        new_jobs = self.normalized_packages.list_new_jobs(load_id)
+        file_jobs = self.normalized_packages.list_new_jobs(load_id, root_only=True)
+        dir_jobs = self.normalized_packages.list_new_dir_jobs(load_id)
         # make sure all jobs have supported writers
         wrong_job = next(
             (
                 j
-                for j in new_jobs
+                for j in file_jobs
                 if ParsedLoadJobFileName.parse(j).file_format not in self.supported_job_file_formats
             ),
             None,
         )
         if wrong_job is not None:
             raise JobFileFormatUnsupported(load_id, self.supported_job_file_formats, wrong_job)
-        return new_jobs
+        return file_jobs + dir_jobs  # type: ignore[no-any-return, operator]
 
     def commit_new_load_package(self, load_id: str) -> None:
         self.storage.rename_tree(