address review comments, squash this commit before merging

swanandx · swanandx · commit b84da384afa4 · 2026-06-05T05:29:00.000Z
diff --git a/python/tests/platform/fixtures/column_mapping.py b/python/tests/platform/fixtures/column_mapping.py
@@ -1,17 +1,24 @@
 """Build a column-mapped, schema-evolved Delta table fixture with PySpark.
 
 ``tests.utils.ensure_delta_spark_fixture`` runs this as a subprocess
-(``uv run --with "delta-spark>=4.2,<5" python column_mapping.py <output_dir>``).
-PySpark is the only writer that can produce a Delta table with column mapping
-enabled *and* perform the rename / drop schema-evolution operations that make
-logical column names diverge from the physical Parquet column names
-(``col-<uuid>``); neither ``delta-rs`` nor the ``deltalake`` Python wheel can.
-PySpark imports stay function-local so importing this module (for
+(``uv run --with "delta-spark>=4.2,<5" python column_mapping.py <output_dir>
+<mode>``), where ``<mode>`` is a ``delta.columnMapping.mode`` value: ``name``
+(physical Parquet columns are renamed to ``col-<uuid>``) or ``id`` (columns are
+matched by Parquet field ID). PySpark is currently the only writer that can
+produce a Delta table with column mapping enabled *and* perform the rename /
+drop schema-evolution operations that make logical column names diverge from
+the physical ones; neither ``delta-rs`` nor the ``deltalake`` Python wheel can.
+DROP COLUMN under column mapping requires delta-spark 4.x (Delta writer v5,
+reader v2). PySpark imports stay function-local so importing this module (for
 ``EXPECTED_ROWS``) never pulls in the JVM/Spark stack.
 
+Changing this builder changes the fixture it produces, but a cached fixture is
+reused based on its path alone — bump ``FIXTURE_VERSION`` in
+``test_delta_input_column_mapping.py`` on any builder change.
+
 The resulting table's history (one commit per step):
 
-* ``v0`` CREATE TABLE with ``delta.columnMapping.mode = 'name'``
+* ``v0`` CREATE TABLE with ``delta.columnMapping.mode = '<mode>'``
 * ``v1`` INSERT two rows under the original ``(id, name, amount)`` schema
 * ``v2`` RENAME COLUMN ``name`` -> ``full_name``
 * ``v3`` ADD COLUMN ``country``
@@ -40,8 +47,13 @@
 ]
 
 
-def build(table_path: str) -> None:
-    """Create the column-mapped, schema-evolved table at ``table_path``."""
+def build(table_path: str, mode: str = "name") -> None:
+    """Create the column-mapped, schema-evolved table at ``table_path``.
+
+    :param mode: ``delta.columnMapping.mode`` for the table: ``name`` or ``id``.
+    """
+    if mode not in ("name", "id"):
+        raise ValueError(f"unsupported column-mapping mode: {mode!r}")
     from delta import configure_spark_with_delta_pip
     from pyspark.sql import SparkSession
 
@@ -64,7 +76,7 @@ def build(table_path: str) -> None:
             f"""
             CREATE TABLE {t} (id BIGINT, name STRING, amount DOUBLE)
             USING delta
-            TBLPROPERTIES ('delta.columnMapping.mode' = 'name',
+            TBLPROPERTIES ('delta.columnMapping.mode' = '{mode}',
                            'delta.minReaderVersion' = '2',
                            'delta.minWriterVersion' = '5')
             """
@@ -80,6 +92,6 @@ def build(table_path: str) -> None:
 
 
 if __name__ == "__main__":
-    if len(sys.argv) != 2:
-        raise SystemExit("usage: column_mapping.py <output_dir>")
-    build(sys.argv[1])
+    if len(sys.argv) not in (2, 3):
+        raise SystemExit("usage: column_mapping.py <output_dir> [name|id]")
+    build(*sys.argv[1:])
diff --git a/python/tests/platform/test_delta_input_column_mapping.py b/python/tests/platform/test_delta_input_column_mapping.py
@@ -1,10 +1,11 @@
 """Snapshot read of a column-mapped Delta table after schema evolution.
 
 A table with ``delta.columnMapping.mode = 'name'`` stores opaque physical
-Parquet column names (``col-<uuid>``) that differ from the logical names, so a
+Parquet column names (``col-<uuid>``) that differ from the logical names; with
+``mode = 'id'`` columns are matched by Parquet field ID instead. Either way, a
 correct snapshot read must resolve every value through the column-mapping
-metadata in the Delta log. PySpark seeds the fixture (it is the only writer
-that supports column-mapping rename/drop); the builder lives in
+metadata in the Delta log. PySpark seeds the fixture (it is currently the only
+writer that supports column-mapping rename/drop); the builder lives in
 ``fixtures/column_mapping.py`` and runs via ``uv run --with delta-spark`` so
 the Spark/JVM wheels are only fetched on cache miss.
 """
@@ -25,7 +26,7 @@
 TABLE = "t"
 CONNECTOR = "delta_in"
 # Bump to invalidate cached MinIO copies when the fixture definition changes.
-FIXTURE_VERSION = "column_mapping_v1"
+FIXTURE_VERSION = "v1"
 
 # Spark builder that writes the column-mapped, schema-evolved table. It runs in
 # a subprocess (see ensure_delta_spark_fixture) rather than being imported here.
@@ -65,16 +66,22 @@ def _snapshot_rows(pipeline) -> list[dict]:
     )
 
 
-def test_delta_input_column_mapping_snapshot(pipeline_name):
+def _run_column_mapping_snapshot_test(pipeline_name: str, mapping_mode: str) -> None:
     """A snapshot read of a column-mapped, schema-evolved table returns the
-    final logical schema and correctly resolves physical column names."""
+    final logical schema and correctly resolves physical column names.
+
+    Driven by one wrapper test per ``delta.columnMapping.mode`` (rather than
+    ``pytest.mark.parametrize``) so each case gets a distinct pipeline name —
+    the ``pipeline_name`` fixture derives the name from the test function, and
+    parametrized cases sharing one name could collide under ``pytest -n``.
+    """
     loc = DeltaTestLocation.create(
         pipeline_name,
         mode="snapshot",
-        stable_subpath=FIXTURE_VERSION,
+        stable_subpath=f"column_mapping_{mapping_mode}_{FIXTURE_VERSION}",
     )
     try:
-        ensure_delta_spark_fixture(loc, _FIXTURE_BUILDER)
+        ensure_delta_spark_fixture(loc, _FIXTURE_BUILDER, builder_args=(mapping_mode,))
 
         pipeline = PipelineBuilder(
             TEST_CLIENT,
@@ -99,3 +106,13 @@ def test_delta_input_column_mapping_snapshot(pipeline_name):
         pipeline.stop(force=True)
     finally:
         loc.cleanup()
+
+
+def test_delta_input_column_mapping_name_snapshot(pipeline_name):
+    """Snapshot read with ``delta.columnMapping.mode = 'name'``."""
+    _run_column_mapping_snapshot_test(pipeline_name, "name")
+
+
+def test_delta_input_column_mapping_id_snapshot(pipeline_name):
+    """Snapshot read with ``delta.columnMapping.mode = 'id'``."""
+    _run_column_mapping_snapshot_test(pipeline_name, "id")
diff --git a/python/tests/utils.py b/python/tests/utils.py
@@ -250,7 +250,7 @@ def row_count(self, missing_ok: bool = False) -> int:
             raise
         return dt.count()
 
-    def table_exists(self) -> bool:
+    def delta_log_exists(self) -> bool:
         """Return True when a Delta log is present at this location.
 
         Used to decide whether a cached fixture (see ``stable_subpath``) can
@@ -262,11 +262,13 @@ def table_exists(self) -> bool:
             return False
 
     def _place_tree(self, staging: pathlib.Path) -> None:
-        """Place a locally-built Delta table tree onto this location's backend.
+        """Copy a locally-built Delta table tree to where this location stores
+        its data: the local directory, or the S3/MinIO bucket, depending on
+        how this location was created.
 
         Some fixtures can only be produced on the local filesystem (e.g. a
-        PySpark-written table) yet must be read back from the configured
-        backend. Local targets are replaced wholesale; S3/MinIO targets get
+        PySpark-written table). For a local target, any existing content at
+        ``self.local_dir`` is deleted before the copy. S3/MinIO targets get
         the data files first and ``_delta_log`` last, so a reader observing
         the upload mid-flight never sees a log referencing a missing parquet.
         """
@@ -316,23 +318,28 @@ def ensure_delta_spark_fixture(
     wheel. This builds such a fixture once and reuses it:
 
     * If the fixture is already present (``is_present``, defaulting to
-      :meth:`DeltaTestLocation.table_exists`), do nothing — so a
+      :meth:`DeltaTestLocation.delta_log_exists`), do nothing — so a
       ``stable_subpath`` cache is reused across runs.
     * Otherwise run ``builder_script`` in an isolated environment
-      (``uv run --with <delta_spark_spec> python <builder_script> <staging>
-      *builder_args``), staging into a temp dir so a half-finished build can
-      never leak into the upload, then place the tree onto ``loc``'s backend.
+      (``uv run --no-project --with <delta_spark_spec> python <builder_script>
+      <staging> *builder_args``), staging into a temp dir so a half-finished
+      build can never leak into the upload, then place the tree onto ``loc``'s
+      backend. ``--no-project`` keeps the builder hermetic: it depends only on
+      ``delta_spark_spec``, never on building the enclosing project.
 
     The heavy PySpark + JVM stack is pulled only on this rare rebuild path.
 
     :param builder_script: Path to a standalone script that writes a Delta
         table to the directory given as its first argument.
     :param builder_args: Extra positional arguments passed after the staging
-        directory (stringified).
+        directory. Each is stringified verbatim with ``str()`` — pass
+        primitives; ``None`` would become the literal string ``"None"``.
     :param is_present: Predicate deciding whether the fixture already exists;
         also re-checked after upload to catch partial uploads.
     """
-    present = is_present if is_present is not None else DeltaTestLocation.table_exists
+    present = (
+        is_present if is_present is not None else DeltaTestLocation.delta_log_exists
+    )
     if present(loc):
         return
 
@@ -348,6 +355,7 @@ def ensure_delta_spark_fixture(
             [
                 "uv",
                 "run",
+                "--no-project",
                 "--with",
                 delta_spark_spec,
                 "python",
@@ -363,8 +371,8 @@ def ensure_delta_spark_fixture(
 
     if not present(loc):
         raise RuntimeError(
-            f"Delta fixture at {loc.uri} is still absent after upload — "
-            "partial upload, or content-stripping middleware?"
+            f"Delta fixture at {loc.uri} is still absent after the builder "
+            "ran and the tree was uploaded."
         )
 
 
