feldera · swanandx · Jun 2, 2026 · May 29, 2026
diff --git a/.github/workflows/test-integration-platform.yml b/.github/workflows/test-integration-platform.yml
@@ -227,6 +227,11 @@ jobs:
         env:
           PYTHONPATH: ${{ github.workspace }}/python
           IN_CI: 1 # We use this flag to skip some kafka tests in the python code base
+          # MinIO test-bucket credentials, surfaced under the standard env
+          # names the Delta tests read (see python/tests/utils.py). These are
+          # the only MinIO secrets available to the OSS platform job.
+          CI_K8S_MINIO_ACCESS_KEY_ID: ${{ secrets.CI_K8S_MINIO_TEST_BUCKET_ACCESS_KEY_ID }}
+          CI_K8S_MINIO_SECRET_ACCESS_KEY: ${{ secrets.CI_K8S_MINIO_TEST_BUCKET_SECRET_ACCESS_KEY }}
 
       - name: Download fda binary
         uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8

diff --git a/python/tests/platform/_dv_fixture_builder.py b/python/tests/platform/_dv_fixture_builder.py
@@ -0,0 +1,73 @@
+"""Seed a Delta table with deletion vectors for the snapshot DV test.
+
+Run this under ``delta-spark``, not bare ``pyspark``. The Delta write path and
+deletion-vector support live in the Delta Lake Spark *JARs* that Spark loads at
+runtime; ``delta-spark`` is the standard way to obtain them — it depends on a
+compatible ``pyspark`` and ships ``configure_spark_with_delta_pip``, which
+auto-resolves the matching Delta JAR. Bare ``pyspark`` could write Delta too,
+but only by hardcoding the Delta Maven coordinate and Scala suffix and keeping
+them in lockstep with the pyspark version — fragile, and no cheaper (the JARs
+download at runtime either way). The companion test invokes this file with::
+
+    uv run --with "delta-spark>=4.2,<5" python _dv_fixture_builder.py \\
+        <dest> <total_rows> <expected_active>
+
+It writes ``total_rows`` rows to ``dest`` with DVs enabled, runs a ``DELETE``
+on the even ``id`` rows that produces deletion vectors, then asserts that
+exactly ``expected_active`` rows remain readable.
+
+The ``_`` prefix keeps pytest from collecting this module as a test; it is only
+ever run as a subprocess.
+"""
+
+import sys
+
+from pyspark.sql import SparkSession
+from delta import configure_spark_with_delta_pip
+
+
+def main() -> None:
+    dest = sys.argv[1]
+    total_rows = int(sys.argv[2])
+    expected_active = int(sys.argv[3])
+
+    builder = (
+        SparkSession.builder.appName("feldera_dv_fixture")
+        .master("local[2]")
+        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
+        .config(
+            "spark.sql.catalog.spark_catalog",
+            "org.apache.spark.sql.delta.catalog.DeltaCatalog",
+        )
+        .config(
+            "spark.databricks.delta.properties.defaults.enableDeletionVectors",
+            "true",
+        )
+        .config("spark.ui.showConsoleProgress", "false")
+    )
+    spark = configure_spark_with_delta_pip(builder).getOrCreate()
+    try:
+        spark.sparkContext.setLogLevel("ERROR")
+        (
+            spark.range(1, total_rows + 1)
+            .selectExpr(
+                "cast(id as int) as id",
+                "concat('user_', id) as name",
+                "cast(id * 1.5 as double) as value",
+            )
+            .write.format("delta")
+            .option("delta.enableDeletionVectors", "true")
+            .save(dest)
+        )
+        spark.sql("DELETE FROM delta.`" + dest + "` WHERE id % 2 = 0")
+        active = spark.read.format("delta").load(dest).count()
+        assert active == expected_active, (
+            f"builder expected {expected_active} active rows after DV "
+            f"DELETE, got {active}"
+        )
+    finally:
+        spark.stop()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/python/tests/platform/test_delta_input_deletion_vectors.py b/python/tests/platform/test_delta_input_deletion_vectors.py
@@ -0,0 +1,173 @@
+"""Snapshot read of a Delta table with deletion vectors.
+
+PySpark seeds the fixture (delta-rs cannot write DVs). The builder lives in
+``_dv_fixture_builder.py`` and runs via ``uv run --with delta-spark`` so the
+Spark/JVM wheels are only fetched on cache miss.
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+
+from feldera import PipelineBuilder
+from feldera.runtime_config import RuntimeConfig
+from feldera.testutils import FELDERA_TEST_NUM_HOSTS, FELDERA_TEST_NUM_WORKERS
+
+from tests import TEST_CLIENT
+from tests.utils import DeltaTestLocation
+
+
+TABLE = "dv_data"
+CONNECTOR = "dv_in"
+TOTAL_ROWS = 200
+EXPECTED_ROWS_AFTER_DV = 100
+# Bump to invalidate cached MinIO copies when the fixture definition changes.
+FIXTURE_VERSION = "dv_snapshot_v1"
+
+# Spark builder that writes the DV-enabled table. It runs in a subprocess
+# (see _ensure_dv_snapshot_fixture) rather than being imported here.
+_FIXTURE_BUILDER = Path(__file__).parent / "_dv_fixture_builder.py"
+
+
+def _log_has_dv_entries(loc: DeltaTestLocation) -> bool:
+    """Return True when any Delta log entry carries a deletion vector.
+
+    The Spark builder validates the active row count itself before exiting,
+    so on the Python side we only need to confirm that the fixture exists
+    and is DV-shaped — neither delta-rs nor the deltalake wheel reads DVs.
+    """
+    try:
+        log_paths = loc.log_json_paths()
+    except FileNotFoundError:
+        return False
+    for log_path in log_paths:
+        for line in loc._read_text(log_path).splitlines():
+            if not line.strip():
+                continue
+            try:
+                action = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if (action.get("add") or {}).get("deletionVector"):
+                return True
+    return False
+
+
+def _ensure_dv_snapshot_fixture(loc: DeltaTestLocation) -> None:
+    """Build the DV fixture if absent; reuse the cached copy otherwise.
+
+    The fixture lives at a shared, commit-independent path
+    (``stable_subpath``), so the Spark build runs at most once per
+    ``FIXTURE_VERSION`` and later runs just read the cached table.
+    """
+    if _log_has_dv_entries(loc):
+        return
+
+    if shutil.which("uv") is None:
+        raise RuntimeError(
+            "`uv` is required on PATH to rebuild the DV fixture "
+            "(builder runs via `uv run --with delta-spark`)."
+        )
+
+    # Stage in a temp dir so a half-finished build cannot leak into the upload.
+    # Writing DV-enabled Delta tables needs the Delta Lake Spark JARs;
+    # `delta-spark` is the clean way to pull them plus a matching pyspark
+    # (see _dv_fixture_builder.py for why not bare pyspark). `uv run --with`
+    # installs the stack only on this rare rebuild path.
+    staging = Path(tempfile.mkdtemp(prefix="feldera_dv_stage_"))
+    try:
+        subprocess.run(
+            [
+                "uv",
+                "run",
+                "--with",
+                "delta-spark>=4.2,<5",
+                "python",
+                str(_FIXTURE_BUILDER),
+                str(staging),
+                str(TOTAL_ROWS),
+                str(EXPECTED_ROWS_AFTER_DV),
+            ],
+            check=True,
+        )
+        # Upload data files first, then _delta_log in version order, so a
+        # reader observing mid-upload never sees a log referencing a missing
+        # parquet.
+        if loc.local_dir is not None:
+            if loc.local_dir.exists():
+                shutil.rmtree(loc.local_dir)
+            shutil.copytree(staging, loc.local_dir)
+        else:
+            fs = loc._s3_filesystem()
+            files = [f for f in sorted(staging.rglob("*")) if f.is_file()]
+            for f in sorted(files, key=lambda p: ("_delta_log" in p.parts, p.name)):
+                rel = f.relative_to(staging).as_posix()
+                with fs.open_output_stream(f"{loc.root_path}/{rel}") as out:
+                    out.write(f.read_bytes())
+    finally:
+        shutil.rmtree(staging, ignore_errors=True)
+
+    if not _log_has_dv_entries(loc):
+        raise RuntimeError(
+            f"DV fixture at {loc.uri} has no deletion-vector log entries "
+            "after upload — partial upload, or DV-stripping middleware?"
+        )
+
+
+def _build_sql(loc: DeltaTestLocation) -> str:
+    connectors = json.dumps(
+        [
+            {
+                "name": CONNECTOR,
+                "transport": {
+                    "name": "delta_table_input",
+                    "config": dict(loc.connector_config),
+                },
+            }
+        ]
+    ).replace("'", "''")
+    return (
+        f"CREATE TABLE {TABLE} ("
+        "id INT NOT NULL,"
+        "name VARCHAR,"
+        "value DOUBLE"
+        f") WITH ('materialized' = 'true', 'connectors' = '{connectors}');"
+    )
+
+
+def test_delta_input_snapshot_with_deletion_vectors(pipeline_name):
+    """Snapshot read of a DV-enabled table must skip soft-deleted rows."""
+    loc = DeltaTestLocation.create(
+        pipeline_name,
+        mode="snapshot",
+        stable_subpath=FIXTURE_VERSION,
+    )
+    try:
+        _ensure_dv_snapshot_fixture(loc)
+
+        pipeline = PipelineBuilder(
+            TEST_CLIENT,
+            pipeline_name,
+            sql=_build_sql(loc),
+            runtime_config=RuntimeConfig(
+                workers=FELDERA_TEST_NUM_WORKERS,
+                hosts=FELDERA_TEST_NUM_HOSTS,
+                logging="debug",
+            ),
+        ).create_or_replace()
+        pipeline.start()
+        pipeline.wait_for_completion(force_stop=False, timeout_s=600)
+
+        rows = list(pipeline.query(f"SELECT COUNT(*) AS c FROM {TABLE}"))
+        assert int(rows[0]["c"]) == EXPECTED_ROWS_AFTER_DV, (
+            "snapshot ingest of a DV-enabled table must drop the soft-deleted "
+            f"rows ({TOTAL_ROWS - EXPECTED_ROWS_AFTER_DV} rows have id % 2 = 0)"
+        )
+
+        pipeline.stop(force=True)
+    finally:
+        loc.cleanup()
diff --git a/python/tests/utils.py b/python/tests/utils.py
@@ -46,6 +46,10 @@ class DeltaTestLocation:
     connector_config: dict[str, object]
     root_path: str
     local_dir: pathlib.Path | None = None
+    # True when ``stable_subpath`` was used at construction time. ``cleanup()``
+    # honors this by leaving the directory in place so the next run reuses the
+    # cached fixture instead of paying to rebuild it.
+    stable: bool = False
 
     @classmethod
     def create(
@@ -60,18 +64,22 @@ def create(
         :param mode: Value of the connector's ``mode`` field. Output
             connectors use ``"truncate"`` (the default); input connectors
             should pass ``"snapshot"``.
-        :param stable_subpath: When set, locates the table at a fixed
-            path so cached data persists across test runs. When unset,
-            a fresh random subpath is used (the original behavior). Use
-            a stable path only when the table contents are deterministic
-            and idempotent across runs.
+        :param stable_subpath: When set, locates the table at a fixed,
+            shared path (``_fixtures/<stable_subpath>``) that is *not*
+            namespaced by the pipeline name or commit SHA, so a fixture
+            built once is reused by every later run and every commit.
+            When unset, a fresh random subpath under ``pipeline_name`` is
+            used (the original behavior). Use a stable path only for
+            fixtures whose contents are deterministic across runs.
         """
 
         if runs_in_ci():
             access_key = required_env("CI_K8S_MINIO_ACCESS_KEY_ID")
             secret_key = required_env("CI_K8S_MINIO_SECRET_ACCESS_KEY")
-            tail = stable_subpath if stable_subpath is not None else uuid.uuid4().hex
-            prefix = f"{pipeline_name}/{tail}"
+            if stable_subpath is not None:
+                prefix = f"_fixtures/{stable_subpath}"
+            else:
+                prefix = f"{pipeline_name}/{uuid.uuid4().hex}"
             root_path = f"{MINIO_BUCKET}/{prefix}"
             minio_endpoint = MINIO_ENDPOINT.rstrip("/")
             parsed_endpoint = urlparse(minio_endpoint)
@@ -96,10 +104,11 @@ def create(
                     "aws_allow_http": str(parsed_endpoint.scheme == "http").lower(),
                 },
                 root_path=root_path,
+                stable=stable_subpath is not None,
             )
 
         if stable_subpath is not None:
-            local_dir = pathlib.Path("/tmp") / f"{pipeline_name}_{stable_subpath}"
+            local_dir = pathlib.Path("/tmp/feldera_fixtures") / stable_subpath
             local_dir.mkdir(parents=True, exist_ok=True)
         else:
             local_dir = pathlib.Path(
@@ -113,6 +122,7 @@ def create(
             },
             root_path=str(local_dir),
             local_dir=local_dir,
+            stable=stable_subpath is not None,
         )
 
     def delta_storage_options(self) -> dict[str, str]:
@@ -241,10 +251,17 @@ def row_count(self, missing_ok: bool = False) -> int:
     def cleanup(self) -> None:
         """Remove the local temp directory, if any.
 
-        No-op on the CI/MinIO path: the bucket is ephemeral and the uuid
-        prefix prevents collisions, so leaked keys are acceptable.
+        No-op when ``stable_subpath`` was used at construction time:
+        the whole point of a stable path is to cache contents across
+        runs, so deleting it would defeat the cache. Also a no-op on
+        the CI/MinIO path: there is no local directory to remove, so any
+        objects this run wrote to the bucket are simply left in place.
+        The shared MinIO bucket is long-lived and these tests never delete
+        from it, so those objects accumulate; non-stable runs each use a
+        unique uuid prefix so they never collide, and the leftover volume
+        for this internal CI bucket is accepted rather than swept here.
         """
-        if self.local_dir is not None:
+        if self.local_dir is not None and not self.stable:
             shutil.rmtree(self.local_dir, ignore_errors=True)
             self.local_dir = None