feat(delta-table): fix pyo3_runtime.PanicException on linux (#545)

Author: amadeusz-ds

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 1.0.46-dev0
+
+* **Fix delta-table `pyo3_runtime.PanicException: Forked process detected` on Linux**
+
 ## 1.0.45
 
 ### Fixes

requirements/connectors/delta-table.txt

@@ -2,3 +2,4 @@ pandas
 deltalake
 boto3
 tenacity
+pyarrow

test/integration/connectors/test_s3.py

@@ -76,18 +76,15 @@ async def test_s3_source(anon_connection_config: S3ConnectionConfig):
             ),
         )
 
+
 @pytest.mark.asyncio
 @pytest.mark.tags(CONNECTOR_TYPE, SOURCE_TAG, BLOB_STORAGE_TAG)
 async def test_s3_source_special_char(anon_connection_config: S3ConnectionConfig):
-    indexer_config = S3IndexerConfig(
-        remote_url="s3://utic-dev-tech-fixtures/special-characters/"
-    )
+    indexer_config = S3IndexerConfig(remote_url="s3://utic-dev-tech-fixtures/special-characters/")
     with tempfile.TemporaryDirectory() as tempdir:
         tempdir_path = Path(tempdir)
         download_config = S3DownloaderConfig(download_dir=tempdir_path)
-        indexer = S3Indexer(
-            connection_config=anon_connection_config, index_config=indexer_config
-        )
+        indexer = S3Indexer(connection_config=anon_connection_config, index_config=indexer_config)
         downloader = S3Downloader(
             connection_config=anon_connection_config, download_config=download_config
         )
@@ -108,26 +105,23 @@ async def test_s3_source_special_char(anon_connection_config: S3ConnectionConfig
             ),
         )
 
+
 @pytest.mark.tags(CONNECTOR_TYPE, SOURCE_TAG, BLOB_STORAGE_TAG)
 def test_s3_source_no_access(anon_connection_config: S3ConnectionConfig):
-    indexer_config = S3IndexerConfig(
-        remote_url="s3://utic-ingest-test-fixtures/destination/"
-    )
-    indexer = S3Indexer(
-        connection_config=anon_connection_config, index_config=indexer_config
-    )
+    indexer_config = S3IndexerConfig(remote_url="s3://utic-ingest-test-fixtures/destination/")
+    indexer = S3Indexer(connection_config=anon_connection_config, index_config=indexer_config)
     with pytest.raises(UserAuthError):
         indexer.precheck()
 
+
 @pytest.mark.tags(CONNECTOR_TYPE, SOURCE_TAG, BLOB_STORAGE_TAG)
 def test_s3_source_no_bucket(anon_connection_config: S3ConnectionConfig):
     indexer_config = S3IndexerConfig(remote_url="s3://fake-bucket")
-    indexer = S3Indexer(
-        connection_config=anon_connection_config, index_config=indexer_config
-    )
+    indexer = S3Indexer(connection_config=anon_connection_config, index_config=indexer_config)
     with pytest.raises(UserError):
         indexer.precheck()
 
+
 @pytest.mark.asyncio
 @pytest.mark.tags(CONNECTOR_TYPE, SOURCE_TAG, "minio", BLOB_STORAGE_TAG)
 async def test_s3_minio_source(anon_connection_config: S3ConnectionConfig):
@@ -139,9 +133,7 @@ async def test_s3_minio_source(anon_connection_config: S3ConnectionConfig):
     ):
         tempdir_path = Path(tempdir)
         download_config = S3DownloaderConfig(download_dir=tempdir_path)
-        indexer = S3Indexer(
-            connection_config=anon_connection_config, index_config=indexer_config
-        )
+        indexer = S3Indexer(connection_config=anon_connection_config, index_config=indexer_config)
         downloader = S3Downloader(
             connection_config=anon_connection_config, download_config=download_config
         )
@@ -167,13 +159,15 @@ async def test_s3_minio_source(anon_connection_config: S3ConnectionConfig):
             ),
         )
 
+
 def get_aws_credentials() -> dict:
     access_key = os.getenv("S3_INGEST_TEST_ACCESS_KEY", None)
     assert access_key
     secret_key = os.getenv("S3_INGEST_TEST_SECRET_KEY", None)
     assert secret_key
     return {"aws_access_key_id": access_key, "aws_secret_access_key": secret_key}
 
+
 @pytest.mark.asyncio
 @pytest.mark.tags(CONNECTOR_TYPE, DESTINATION_TAG, BLOB_STORAGE_TAG)
 @requires_env("S3_INGEST_TEST_ACCESS_KEY", "S3_INGEST_TEST_SECRET_KEY")
@@ -191,9 +185,7 @@ async def test_s3_destination(upload_file: Path):
     uploader = S3Uploader(connection_config=connection_config, upload_config=upload_config)
     s3fs = uploader.fs
     file_data = FileData(
-        source_identifiers=SourceIdentifiers(
-            fullpath=upload_file.name, filename=upload_file.name
-        ),
+        source_identifiers=SourceIdentifiers(fullpath=upload_file.name, filename=upload_file.name),
         connector_type=CONNECTOR_TYPE,
         identifier="mock file data",
     )
@@ -204,14 +196,13 @@ async def test_s3_destination(upload_file: Path):
         else:
             uploader.run(path=upload_file, file_data=file_data)
         uploaded_files = [
-            Path(file)
-            for file in s3fs.ls(path=destination_path)
-            if Path(file).name != "_empty"
+            Path(file) for file in s3fs.ls(path=destination_path) if Path(file).name != "_empty"
         ]
         assert len(uploaded_files) == 1
     finally:
         s3fs.rm(path=destination_path, recursive=True)
 
+
 @pytest.mark.asyncio
 @pytest.mark.tags(CONNECTOR_TYPE, DESTINATION_TAG, BLOB_STORAGE_TAG)
 @requires_env("S3_INGEST_TEST_ACCESS_KEY", "S3_INGEST_TEST_SECRET_KEY")
@@ -252,14 +243,13 @@ async def test_s3_destination_same_filename_different_folders(upload_file: Path)
             uploader.run(path=upload_file, file_data=file_data_1)
             uploader.run(path=upload_file, file_data=file_data_2)
         uploaded_files = [
-            Path(file)
-            for file in s3fs.ls(path=destination_path)
-            if Path(file).name != "_empty"
+            Path(file) for file in s3fs.ls(path=destination_path) if Path(file).name != "_empty"
         ]
         assert len(uploaded_files) == 2
     finally:
         s3fs.rm(path=destination_path, recursive=True)
 
+
 @pytest.mark.asyncio
 @pytest.mark.tags(CONNECTOR_TYPE, DESTINATION_TAG, BLOB_STORAGE_TAG)
 @requires_env("S3_INGEST_TEST_ACCESS_KEY", "S3_INGEST_TEST_SECRET_KEY")
@@ -292,9 +282,7 @@ async def test_s3_destination_different_relative_path_and_full_path(upload_file:
         else:
             uploader.run(path=upload_file, file_data=file_data)
         uploaded_files = [
-            Path(file)
-            for file in s3fs.ls(path=destination_path)
-            if Path(file).name != "_empty"
+            Path(file) for file in s3fs.ls(path=destination_path) if Path(file).name != "_empty"
         ]
         assert len(uploaded_files) == 1
         assert uploaded_files[0].as_posix() == f"{destination_path.lstrip('s3://')}/folder1"

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "1.0.45"  # pragma: no cover
+__version__ = "1.0.46-dev0"  # pragma: no cover

unstructured_ingest/processes/connectors/delta_table.py

@@ -1,7 +1,7 @@
 import logging
 import traceback
 from dataclasses import dataclass, field
-from multiprocessing import Process, Queue, current_process
+from multiprocessing import Queue, current_process
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Optional
 from urllib.parse import urlparse
@@ -186,10 +186,25 @@ def upload_dataframe(self, df: "DataFrame", file_data: FileData) -> None:
         )
 
         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)
+            """Return True if exception looks like a Delta Lake commit conflict.
+
+            Besides the canonical *CommitFailed* / *Metadata changed* errors that
+            deltalake surfaces when two writers clash, we occasionally hit
+            messages such as *Delta transaction failed, version 0 already
+            exists* while multiple processes race to create the very first log
+            entry. These situations are equally safe to retry, so detect them
+            too.
+            """
+
+            return isinstance(exc, RuntimeError) and any(
+                marker in str(exc)
+                for marker in (
+                    "CommitFailed",
+                    "Metadata changed",
+                    "version 0 already exists",
+                    "version already exists",
+                    "Delta transaction failed",
+                )
             )
 
         @retry(
@@ -206,25 +221,39 @@ def _single_attempt() -> None:
             # 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()
+            # Use a multiprocessing context that relies on 'spawn' to avoid inheriting the
+            # parent process' Tokio runtime, which leads to `pyo3_runtime.PanicException`.
+            from multiprocessing import get_context
+
+            ctx = get_context("spawn")
+            queue: "Queue[str]" = ctx.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(
+                # On non-daemon processes we still guard against SIGABRT by running in a
+                # dedicated subprocess created via the 'spawn' method.
+                writer = ctx.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}")
+                # First surface any traceback captured inside the subprocess so users see the real
+                # root-cause instead of a generic non-zero exit code.
+                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}")
+
+                # If the subprocess terminated abnormally but produced no traceback (e.g., SIGABRT),
+                # still raise a helpful error for callers.
+                if not current_process().daemon and writer.exitcode != 0:
+                    raise RuntimeError(
+                        f"write_deltalake subprocess exited with code {writer.exitcode}"
+                    )
 
         _single_attempt()