feat(delta-table): add record-level upsert (#543)

amadeusz-ds · web-flow · commit 663254bb219a · 2025-06-26T18:58:16.000+02:00
- `DeltaTableUploadStager`
- injects `RECORD_ID_LABEL` on every row, enabling per-record tracking.

- `DeltaTableUploader.upload_dataframe`
- deletes rows whose `record_id` matches the current file before
appending new data instead of doing a full table overwrite.
- handles concurrent writers by wrapping the delete-then-append sequence
in a tenacity-based retry (`10` attempts, random 0.2-1 s back-off) that
re-runs only on commit-conflict errors (“CommitFailed”, “Metadata
changed”).
- keeps the SIGABRT-work-around: runs the writer in a subprocess unless
executing inside a daemon worker; error propagation unified through a
single `multiprocessing.Queue`.

- refactor
- removed duplicated queue/error-handling blocks, preserved detailed
explanatory comments.
- added `@requires_dependencies(["tenacity"], extras="delta-table")` to
enforce the new optional dependency.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 1.0.44
+
+* **Improve DeltaTable ingestion process and reliability**
+
 ## 1.0.43
 
 * **Fix document limits in Confluence connectr**
diff --git a/requirements/connectors/delta-table.txt b/requirements/connectors/delta-table.txt
@@ -1,3 +1,4 @@
 pandas
 deltalake
 boto3
+tenacity
diff --git a/test/integration/connectors/elasticsearch/test_elasticsearch.py b/test/integration/connectors/elasticsearch/test_elasticsearch.py
@@ -206,7 +206,7 @@ async def test_elasticsearch_source(source_index: str, movies_dataframe: pd.Data
                 validate_downloaded_files=True,
                 predownload_file_data_check=source_filedata_display_name_set_check,
                 postdownload_file_data_check=source_filedata_display_name_set_check,
-                exclude_fields_extend=["display_name"] # includes dynamic ids, might change
+                exclude_fields_extend=["display_name"],  # includes dynamic ids, might change
             ),
         )
 
diff --git a/test/integration/connectors/elasticsearch/test_opensearch.py b/test/integration/connectors/elasticsearch/test_opensearch.py
@@ -196,7 +196,7 @@ async def test_opensearch_source(source_index: str, movies_dataframe: pd.DataFra
                 validate_downloaded_files=True,
                 predownload_file_data_check=source_filedata_display_name_set_check,
                 postdownload_file_data_check=source_filedata_display_name_set_check,
-                exclude_fields_extend=["display_name"]  # includes dynamic ids, might change
+                exclude_fields_extend=["display_name"],  # includes dynamic ids, might change
             ),
         )
 
diff --git a/test/integration/connectors/test_astradb.py b/test/integration/connectors/test_astradb.py
@@ -173,7 +173,7 @@ async def test_astra_search_source(
             validate_downloaded_files=True,
             predownload_file_data_check=source_filedata_display_name_set_check,
             postdownload_file_data_check=source_filedata_display_name_set_check,
-            exclude_fields_extend=["display_name"]  # includes dynamic ids, might change
+            exclude_fields_extend=["display_name"],  # includes dynamic ids, might change
         ),
     )
 
diff --git a/test/integration/connectors/test_confluence.py b/test/integration/connectors/test_confluence.py
@@ -76,6 +76,6 @@ async def test_confluence_source_param(
             test_id=test_id,
             expected_num_files=expected_num_files,
             validate_downloaded_files=validate_downloaded_files,
-            validate_file_data=validate_file_data
+            validate_file_data=validate_file_data,
         ),
-    )
+    )
diff --git a/test/integration/connectors/test_delta_table.py b/test/integration/connectors/test_delta_table.py
@@ -5,6 +5,7 @@
 import pytest
 from deltalake import DeltaTable
 from fsspec import get_filesystem_class
+from pydantic import Secret
 
 from test.integration.connectors.utils.constants import DESTINATION_TAG, SQL_TAG
 from test.integration.utils import requires_env
@@ -27,13 +28,24 @@
 async def test_delta_table_destination_local(upload_file: Path, temp_dir: Path):
     destination_path = str(temp_dir)
     connection_config = DeltaTableConnectionConfig(
-        access_config=DeltaTableAccessConfig(),
+        access_config=Secret(DeltaTableAccessConfig()),
         table_uri=destination_path,
     )
     stager_config = DeltaTableUploadStagerConfig()
     stager = DeltaTableUploadStager(upload_stager_config=stager_config)
+
+    mock_file_data = FileData(
+        identifier="mock file data",
+        connector_type=CONNECTOR_TYPE,
+        source_identifiers=SourceIdentifiers(
+            filename=upload_file.name,
+            fullpath=upload_file.name,
+        ),
+    )
+
     new_upload_file = stager.run(
         elements_filepath=upload_file,
+        file_data=mock_file_data,
         output_dir=temp_dir,
         output_filename=upload_file.name,
     )
@@ -52,11 +64,10 @@ async def test_delta_table_destination_local(upload_file: Path, temp_dir: Path):
         await uploader.run_async(path=new_upload_file, file_data=file_data)
     else:
         uploader.run(path=new_upload_file, file_data=file_data)
-    delta_table_path = os.path.join(destination_path, upload_file.name)
-    delta_table = DeltaTable(table_uri=delta_table_path)
+    delta_table = DeltaTable(table_uri=destination_path)
     df = delta_table.to_pandas()
 
-    EXPECTED_COLUMNS = 10
+    EXPECTED_COLUMNS = 11
     EXPECTED_ROWS = 22
     assert len(df) == EXPECTED_ROWS, (
         f"Number of rows in table vs expected: {len(df)}/{EXPECTED_ROWS}"
@@ -86,17 +97,30 @@ async def test_delta_table_destination_s3(upload_file: Path, temp_dir: Path):
     s3_bucket = "s3://utic-platform-test-destination"
     destination_path = f"{s3_bucket}/destination/test"
     connection_config = DeltaTableConnectionConfig(
-        access_config=DeltaTableAccessConfig(
-            aws_access_key_id=aws_credentials["AWS_ACCESS_KEY_ID"],
-            aws_secret_access_key=aws_credentials["AWS_SECRET_ACCESS_KEY"],
+        access_config=Secret(
+            DeltaTableAccessConfig(
+                aws_access_key_id=aws_credentials["AWS_ACCESS_KEY_ID"],
+                aws_secret_access_key=aws_credentials["AWS_SECRET_ACCESS_KEY"],
+            )
         ),
         aws_region=aws_credentials["AWS_REGION"],
         table_uri=destination_path,
     )
     stager_config = DeltaTableUploadStagerConfig()
     stager = DeltaTableUploadStager(upload_stager_config=stager_config)
+
+    mock_file_data = FileData(
+        identifier="mock file data",
+        connector_type=CONNECTOR_TYPE,
+        source_identifiers=SourceIdentifiers(
+            filename=upload_file.name,
+            fullpath=upload_file.name,
+        ),
+    )
+
     new_upload_file = stager.run(
         elements_filepath=upload_file,
+        file_data=mock_file_data,
         output_dir=temp_dir,
         output_filename=upload_file.name,
     )
@@ -117,11 +141,10 @@ async def test_delta_table_destination_s3(upload_file: Path, temp_dir: Path):
             await uploader.run_async(path=new_upload_file, file_data=file_data)
         else:
             uploader.run(path=new_upload_file, file_data=file_data)
-        delta_table_path = os.path.join(destination_path, upload_file.name)
-        delta_table = DeltaTable(table_uri=delta_table_path, storage_options=aws_credentials)
+        delta_table = DeltaTable(table_uri=destination_path, storage_options=aws_credentials)
         df = delta_table.to_pandas()
 
-        EXPECTED_COLUMNS = 10
+        EXPECTED_COLUMNS = 11
         EXPECTED_ROWS = 22
         assert len(df) == EXPECTED_ROWS, (
             f"Number of rows in table vs expected: {len(df)}/{EXPECTED_ROWS}"
@@ -149,17 +172,30 @@ async def test_delta_table_destination_s3_bad_creds(upload_file: Path, temp_dir:
     s3_bucket = "s3://utic-platform-test-destination"
     destination_path = f"{s3_bucket}/destination/test"
     connection_config = DeltaTableConnectionConfig(
-        access_config=DeltaTableAccessConfig(
-            aws_access_key_id=aws_credentials["AWS_ACCESS_KEY_ID"],
-            aws_secret_access_key=aws_credentials["AWS_SECRET_ACCESS_KEY"],
+        access_config=Secret(
+            DeltaTableAccessConfig(
+                aws_access_key_id=aws_credentials["AWS_ACCESS_KEY_ID"],
+                aws_secret_access_key=aws_credentials["AWS_SECRET_ACCESS_KEY"],
+            )
         ),
         aws_region=aws_credentials["AWS_REGION"],
         table_uri=destination_path,
     )
     stager_config = DeltaTableUploadStagerConfig()
     stager = DeltaTableUploadStager(upload_stager_config=stager_config)
+
+    mock_file_data = FileData(
+        identifier="mock file data",
+        connector_type=CONNECTOR_TYPE,
+        source_identifiers=SourceIdentifiers(
+            filename=upload_file.name,
+            fullpath=upload_file.name,
+        ),
+    )
+
     new_upload_file = stager.run(
         elements_filepath=upload_file,
+        file_data=mock_file_data,
         output_dir=temp_dir,
         output_filename=upload_file.name,
     )
diff --git a/unstructured_ingest/__version__.py b/unstructured_ingest/__version__.py
@@ -1 +1 @@
-__version__ = "1.0.43"  # pragma: no cover
+__version__ = "1.0.44"  # pragma: no cover
diff --git a/unstructured_ingest/processes/connectors/astradb.py b/unstructured_ingest/processes/connectors/astradb.py
@@ -197,8 +197,10 @@ def run(self, **kwargs: Any) -> Generator[AstraDBBatchFileData, None, None]:
         id_batches = batch_generator(ids, self.index_config.batch_size)
         for batch in id_batches:
             batch_items = [BatchItem(identifier=b) for b in batch]
-            display_name = (f"{self.index_config.collection_name}-{self.index_config.keyspace}"
-                            f"-[{batch_items[0].identifier}..{batch_items[-1].identifier}]")
+            display_name = (
+                f"{self.index_config.collection_name}-{self.index_config.keyspace}"
+                f"-[{batch_items[0].identifier}..{batch_items[-1].identifier}]"
+            )
             fd = AstraDBBatchFileData(
                 connector_type=CONNECTOR_TYPE,
                 metadata=FileDataSourceMetadata(
diff --git a/unstructured_ingest/processes/connectors/confluence.py b/unstructured_ingest/processes/connectors/confluence.py
@@ -191,7 +191,7 @@ def _get_docs_ids_within_one_space(self, space_key: str) -> List[dict]:
                 status=None,
             )
         # Limit the number of documents to max_num_of_docs_from_each_space
-        # Note: this is needed because the limit field in client.get_all_pages_from_space does 
+        # Note: this is needed because the limit field in client.get_all_pages_from_space does
         # not seem to work as expected
         limited_pages = pages[: self.index_config.max_num_of_docs_from_each_space]
         doc_ids = [{"space_id": space_key, "doc_id": page["id"]} for page in limited_pages]
diff --git a/unstructured_ingest/processes/connectors/delta_table.py b/unstructured_ingest/processes/connectors/delta_table.py
@@ -1,7 +1,7 @@
-import os
+import logging
 import traceback
 from dataclasses import dataclass, field
-from multiprocessing import Process, Queue
+from multiprocessing import Process, Queue, current_process
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Optional
 from urllib.parse import urlparse
@@ -20,6 +20,7 @@
 )
 from unstructured_ingest.logger import logger
 from unstructured_ingest.processes.connector_registry import DestinationRegistryEntry
+from unstructured_ingest.utils.constants import RECORD_ID_LABEL
 from unstructured_ingest.utils.data_prep import get_data_df, get_json_data
 from unstructured_ingest.utils.dep_check import requires_dependencies
 from unstructured_ingest.utils.table import convert_to_pandas_dataframe
@@ -47,18 +48,17 @@ class DeltaTableAccessConfig(AccessConfig):
 
 class DeltaTableConnectionConfig(ConnectionConfig):
     access_config: Secret[DeltaTableAccessConfig] = Field(
-        default=DeltaTableAccessConfig(), validate_default=True
+        default=Secret(DeltaTableAccessConfig()), validate_default=True
     )
     aws_region: Optional[str] = Field(default=None, description="AWS Region")
     table_uri: str = Field(
-        default=None,
         description=(
             "Local path or path to the target folder in the S3 bucket, "
             "formatted as s3://my-bucket/my-folder/"
         ),
     )
 
-    def update_storage_options(self, storage_options: dict) -> None:
+    def update_storage_options(self, storage_options: dict[str, str]) -> None:
         secrets = self.access_config.get_secret_value()
         if self.aws_region and secrets.aws_access_key_id and secrets.aws_secret_access_key:
             storage_options["AWS_REGION"] = self.aws_region
@@ -80,9 +80,10 @@ class DeltaTableUploadStager(UploadStager):
         default_factory=lambda: DeltaTableUploadStagerConfig()
     )
 
-    def run(
+    def run(  # type: ignore[override]
         self,
         elements_filepath: Path,
+        file_data: FileData,
         output_dir: Path,
         output_filename: str,
         **kwargs: Any,
@@ -91,6 +92,8 @@ def run(
         output_path = Path(output_dir) / Path(f"{output_filename}.parquet")
 
         df = convert_to_pandas_dataframe(elements_dict=elements_contents)
+        # Ensure per-record overwrite/delete semantics: tag each row with the record identifier
+        df[RECORD_ID_LABEL] = file_data.identifier
         df = df.dropna(axis=1, how="all")
         df.to_parquet(output_path)
 
@@ -138,41 +141,92 @@ def precheck(self):
                 logger.error(f"failed to validate connection: {e}", exc_info=True)
                 raise DestinationConnectionError(f"failed to validate connection: {e}")
 
+    @requires_dependencies(["tenacity"], extras="delta-table")
     def upload_dataframe(self, df: "DataFrame", file_data: FileData) -> None:
-        updated_upload_path = os.path.join(
-            self.connection_config.table_uri, file_data.source_identifiers.relative_path
-        )
+        upload_path = self.connection_config.table_uri
         logger.info(
-            f"writing {len(df)} rows to destination table "
-            f"at {updated_upload_path}\ndtypes: {df.dtypes}",
+            f"writing {len(df)} rows to destination table at {upload_path}\ndtypes: {df.dtypes}",
         )
-        storage_options = {}
+        storage_options: dict[str, str] = {}
         self.connection_config.update_storage_options(storage_options=storage_options)
 
+        # Decide whether the Delta table already exists. If it does, we first delete all rows
+        # belonging to the current record and then append the fresh data. Otherwise we will
+        # create a brand-new table via an overwrite.
+
+        mode = "overwrite"
+        try:
+            from deltalake import DeltaTable  # pylint: disable=import-error
+
+            dt = DeltaTable(upload_path, storage_options=storage_options)
+            logger.debug(f"Table exists: deleting rows for {file_data.identifier}")
+            # Table exists – remove any previous rows for this record_id so that appending is
+            # effectively an idempotent overwrite for the record.
+            dt.delete(predicate=f"{RECORD_ID_LABEL} = '{file_data.identifier}'")
+            mode = "append"
+        except Exception:
+            # Table does not exist yet (or cannot be opened) – we will create it below with
+            # mode="overwrite". All other failures will be captured later by the writer.
+            logger.debug("Table does not exist: creating new table")
+
         writer_kwargs = {
-            "table_or_uri": updated_upload_path,
+            "table_or_uri": upload_path,
             "data": df,
-            "mode": "overwrite",
+            "mode": mode,
             "schema_mode": "merge",
             "storage_options": storage_options,
         }
-        queue = Queue()
-        # NOTE: deltalake writer on Linux sometimes can finish but still trigger a SIGABRT and cause
-        # ingest to fail, even though all tasks are completed normally. Putting the writer into a
-        # process mitigates this issue by ensuring python interpreter waits properly for deltalake's
-        # rust backend to finish
-        writer = Process(
-            target=write_deltalake_with_error_handling,
-            kwargs={"queue": queue, **writer_kwargs},
+
+        from tenacity import (
+            before_log,
+            retry,
+            retry_if_exception,
+            stop_after_attempt,
+            wait_random,
         )
-        writer.start()
-        writer.join()
 
-        # Check if the queue has any exception message
-        if not queue.empty():
-            error_message = queue.get()
-            logger.error(f"Exception occurred in write_deltalake: {error_message}")
-            raise RuntimeError(f"Error in write_deltalake: {error_message}")
+        def _is_commit_conflict(exc: BaseException) -> bool:  # noqa: ANN401
+            """Return True if exception looks like a Delta Lake commit conflict."""
+
+            return isinstance(exc, RuntimeError) and (
+                "CommitFailed" in str(exc) or "Metadata changed" in str(exc)
+            )
+
+        @retry(
+            stop=stop_after_attempt(10),
+            wait=wait_random(min=0.2, max=1.0),
+            before=before_log(logger, logging.DEBUG),
+            retry=retry_if_exception(_is_commit_conflict),
+            reraise=True,
+        )
+        def _single_attempt() -> None:
+            """One optimistic transaction: delete old rows, then append new ones."""
+
+            # NOTE: deltalake writer on Linux sometimes can finish but still trigger a SIGABRT and
+            # cause ingest to fail, even though all tasks are completed normally. Putting the writer
+            # into a process mitigates this issue by ensuring python interpreter waits properly for
+            # deltalake's rust backend to finish
+            queue: Queue[str] = Queue()
+
+            if current_process().daemon:
+                # write_deltalake_with_error_handling will push any traceback to our queue
+                write_deltalake_with_error_handling(queue=queue, **writer_kwargs)
+            else:
+                # On non-daemon processes we still guard against SIGABRT by running in a subprocess.
+                writer = Process(
+                    target=write_deltalake_with_error_handling,
+                    kwargs={"queue": queue, **writer_kwargs},
+                )
+                writer.start()
+                writer.join()
+
+            # Check if the queue has any exception message
+            if not queue.empty():
+                error_message = queue.get()
+                logger.error("Exception occurred in write_deltalake: %s", error_message)
+                raise RuntimeError(f"Error in write_deltalake: {error_message}")
+
+        _single_attempt()
 
     @requires_dependencies(["pandas"], extras="delta-table")
     def run_data(self, data: list[dict], file_data: FileData, **kwargs: Any) -> None:
@@ -182,7 +236,7 @@ def run_data(self, data: list[dict], file_data: FileData, **kwargs: Any) -> None
         self.upload_dataframe(df=df, file_data=file_data)
 
     @requires_dependencies(["pandas"], extras="delta-table")
-    def run(self, path: Path, file_data: FileData, **kwargs: Any) -> None:
+    def run(self, path: Path, file_data: FileData, **kwargs: Any) -> None:  # type: ignore[override]
         df = get_data_df(path)
         self.upload_dataframe(df=df, file_data=file_data)
 
diff --git a/unstructured_ingest/processes/connectors/sql/sql.py b/unstructured_ingest/processes/connectors/sql/sql.py

-Original file line number
+Diff line change
@@ @@ -1,3 +1,4 @@ @@
 pandas
 deltalake
 boto3
 +tenacity
Original file line number	Diff line number	Diff line change
`@@ -206,7 +206,7 @@ async def test_elasticsearch_source(source_index: str, movies_dataframe: pd.Data`
`206`	`206`	`validate_downloaded_files=True,`
`207`	`207`	`predownload_file_data_check=source_filedata_display_name_set_check,`
`208`	`208`	`postdownload_file_data_check=source_filedata_display_name_set_check,`
`209`		`- exclude_fields_extend=["display_name"] # includes dynamic ids, might change`
	`209`	`+ exclude_fields_extend=["display_name"], # includes dynamic ids, might change`
`210`	`210`	`),`
`211`	`211`	`)`
`212`	`212`
Original file line number	Diff line number	Diff line change
`@@ -196,7 +196,7 @@ async def test_opensearch_source(source_index: str, movies_dataframe: pd.DataFra`
`196`	`196`	`validate_downloaded_files=True,`
`197`	`197`	`predownload_file_data_check=source_filedata_display_name_set_check,`
`198`	`198`	`postdownload_file_data_check=source_filedata_display_name_set_check,`
`199`		`- exclude_fields_extend=["display_name"] # includes dynamic ids, might change`
	`199`	`+ exclude_fields_extend=["display_name"], # includes dynamic ids, might change`
`200`	`200`	`),`
`201`	`201`	`)`
`202`	`202`
Original file line number	Diff line number	Diff line change
`@@ -173,7 +173,7 @@ async def test_astra_search_source(`
`173`	`173`	`validate_downloaded_files=True,`
`174`	`174`	`predownload_file_data_check=source_filedata_display_name_set_check,`
`175`	`175`	`postdownload_file_data_check=source_filedata_display_name_set_check,`
`176`		`- exclude_fields_extend=["display_name"] # includes dynamic ids, might change`
	`176`	`+ exclude_fields_extend=["display_name"], # includes dynamic ids, might change`
`177`	`177`	`),`
`178`	`178`	`)`
`179`	`179`
Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-__version__ = "1.0.43" # pragma: no cover`
	`1`	`+__version__ = "1.0.44" # pragma: no cover`
Original file line number	Diff line number	Diff line change
`@@ -191,7 +191,7 @@ def _get_docs_ids_within_one_space(self, space_key: str) -> List[dict]:`
`191`	`191`	`status=None,`
`192`	`192`	`)`
`193`	`193`	`# Limit the number of documents to max_num_of_docs_from_each_space`
`194`		`- # Note: this is needed because the limit field in client.get_all_pages_from_space does`
	`194`	`+ # Note: this is needed because the limit field in client.get_all_pages_from_space does`
`195`	`195`	`# not seem to work as expected`
`196`	`196`	`limited_pages = pages[: self.index_config.max_num_of_docs_from_each_space]`
`197`	`197`	`doc_ids = [{"space_id": space_key, "doc_id": page["id"]} for page in limited_pages]`