[Spark] Support ALTER TABLE ... REPLACE PARTITIONED BY WITH CLUSTER BY

.devcontainer/devcontainer.json

@@ -0,0 +1,20 @@
+{
+  "name": "Delta Lake Dev",
+  "image": "mcr.microsoft.com/devcontainers/java:17-bookworm",
+  "features": {
+    "ghcr.io/devcontainers/features/python:1": {
+      "version": "3.10"
+    }
+  },
+  "customizations": {
+    "vscode": {
+      "extensions": [
+        "scalameta.metals"
+      ],
+      "settings": {
+        "metals.sbtScript": "${containerWorkspaceFolder}/build/sbt"
+      }
+    }
+  },
+  "postCreateCommand": "echo 'Dev container ready. Run: build/sbt compile'"
+}

.github/workflows/build.yaml

@@ -39,12 +39,12 @@ jobs:
           key: delta-sbt-cache-cross-spark-${{ hashFiles('project/scripts/setup_unitycatalog_main.sh') }}
 
       # publishM2 compiles every aggregated project, including storage, which has
-      # unitycatalog-client as a compile-scope dependency. test_cross_spark_publish.py also
+      # unitycatalog-client as a compile-scope dependency. test_published_artifacts.py also
       # iterates over released Spark versions (sbt -DsparkVersion=<X.Y>), so we need UC's
       # spark connector published for each variant Delta will resolve. Discover the version list
       # at runtime from project/spark-versions.json (same source the matrix workflows use) and
       # invoke the setup script per variant; cache hits (the steady state) make the script a
-      # no-op. Snapshot versions are skipped by the cross-Spark test, so only released ones are
+      # no-op. Snapshot versions are skipped by the published-artifact test, so only released ones are
       # published here.
       - name: Set up pinned Unity Catalog for each released Spark version
         run: |
@@ -54,8 +54,8 @@ jobs:
             echo "::endgroup::"
           done
 
-      - name: Run cross-Spark build test
-        run: python project/tests/test_cross_spark_publish.py
+      - name: Test published artifacts
+        run: python project/tests/test_published_artifacts.py
 
       - name: Save SBT cache
         if: github.event_name == 'push' && github.ref == 'refs/heads/master'

.github/workflows/flink_test.yaml

@@ -6,13 +6,19 @@ on:
     paths:
       - 'flink/**'
       - 'kernel/**'
+      - 'build.sbt'
+      - 'project/CrossFlinkVersions.scala'
+      - '.github/workflows/flink_test.yaml'
       - '!**/*.md'
       - '!**/*.txt'
   pull_request:
     branches: [master, branch-*]
     paths:
       - 'flink/**'
       - 'kernel/**'
+      - 'build.sbt'
+      - 'project/CrossFlinkVersions.scala'
+      - '.github/workflows/flink_test.yaml'
       - '!**/*.md'
       - '!**/*.txt'
 
@@ -26,9 +32,34 @@ env:
   SBT_OPTS: "-Dsbt.coursier.home-dir=/home/runner/.cache/coursier -Dsbt.ivy.home=/home/runner/.ivy2"
 
 jobs:
+  generate-matrix:
+    name: "Generate Flink versions matrix"
+    runs-on: ubuntu-24.04
+    outputs:
+      flink-versions: ${{ steps.versions.outputs.flink-versions }}
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
+      - name: Install Java
+        uses: actions/setup-java@c1e323688fd81a25caa38c78aa6df2d33d3e20d9 # v4.8.0
+        with:
+          distribution: "zulu"
+          java-version: "17"
+      - name: Generate matrix
+        id: versions
+        run: |
+          build/sbt exportFlinkVersionsJson
+          versions="$(jq -c '[.[].fullVersion]' target/flink-versions.json)"
+          echo "flink-versions=$versions" >> "$GITHUB_OUTPUT"
+
   test:
-    name: "DF"
+    name: "DF: Flink ${{ matrix.flink }}"
+    needs: generate-matrix
     runs-on: ubuntu-24.04
+    strategy:
+      fail-fast: false
+      matrix:
+        flink: ${{ fromJson(needs.generate-matrix.outputs.flink-versions) }}
     steps:
       - name: Show runner specs
         run: |
@@ -56,7 +87,7 @@ jobs:
             ~/.sbt
             ~/.ivy2
             ~/.cache/coursier
-          key: sbt-flink
+          key: sbt-flink-${{ matrix.flink }}
       - name: Check cache status
         run: |
           if [ "${{ steps.cache-sbt.outputs.cache-hit }}" == "true" ]; then
@@ -70,7 +101,7 @@ jobs:
         uses: ./.github/actions/setup-unitycatalog
       - name: Run unit tests
         run: |
-          build/sbt flinkGroup/test
+          build/sbt -DflinkVersion=${{ matrix.flink }} flinkGroup/test
       - name: Save SBT cache
         if: github.event_name == 'push' && github.ref == 'refs/heads/master'
         uses: actions/cache/save@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
@@ -79,4 +110,4 @@ jobs:
             ~/.sbt
             ~/.ivy2
             ~/.cache/coursier
-          key: sbt-flink
+          key: sbt-flink-${{ matrix.flink }}

PROTOCOL.md

@@ -110,6 +110,7 @@
   - [Generated Columns](#generated-columns)
   - [Default Columns](#default-columns)
   - [Identity Columns](#identity-columns)
+  - [Materialize Partition Columns](#materialize-partition-columns)
   - [Writer Version Requirements](#writer-version-requirements)
 - [Requirements for Readers](#requirements-for-readers)
   - [Reader Version Requirements](#reader-version-requirements)
@@ -621,7 +622,7 @@ Field Name | Data Type | Description | optional/required
 path| String | A relative path to a file from the root of the table or an absolute path to a file that should be removed from the table. The path is a URI as specified by [RFC 2396 URI Generic Syntax](https://www.ietf.org/rfc/rfc2396.txt), which needs to be decoded to get the data file path. | required
 deletionTimestamp | Option[Long] | The time the deletion occurred, represented as milliseconds since the epoch | optional
 dataChange | Boolean | When `false` the records in the removed file must be contained in one or more `add` file actions in the same version | required
-extendedFileMetadata | Boolean | When `true` the fields `partitionValues`, `size`, and `tags` are present | optional
+extendedFileMetadata | Boolean | When `true` the fields `partitionValues` and `size` are present | optional
 partitionValues| Map[String, String] | A map from partition column to value for this file. See also [Partition Value Serialization](#Partition-Value-Serialization) | optional
 size| Long | The size of this data file in bytes | optional
 stats | [Statistics Struct](#Per-file-Statistics) | Contains statistics (e.g., count, min/max values for columns) about the data in this logical file | optional
@@ -1836,7 +1837,7 @@ The following is an example for the `domainMetadata` action definition of a tabl
 {
   "domainMetadata": {
     "domain": "delta.clustering",
-    "configuration": "{\"clusteringColumns\":[\"col-daadafd7-7c20-4697-98f8-bff70199b1f9\", \"col-5abe0e80-cf57-47ac-9ffc-a861a3d1077e\"]}",
+    "configuration": "{\"clusteringColumns\":[[\"col-daadafd7-7c20-4697-98f8-bff70199b1f9\"], [\"col-5abe0e80-cf57-47ac-9ffc-a861a3d1077e\"]]}",
     "removed": false
   }
 }
@@ -1845,11 +1846,12 @@ The example above converts `configuration` field into JSON format, including esc
 ```json
 {
   "clusteringColumns": [
-    "col-daadafd7-7c20-4697-98f8-bff70199b1f9",
-    "col-5abe0e80-cf57-47ac-9ffc-a861a3d1077e"
+    ["col-daadafd7-7c20-4697-98f8-bff70199b1f9"],
+    ["col-5abe0e80-cf57-47ac-9ffc-a861a3d1077e"]
   ]
 }
 ```
+Each entry in `clusteringColumns` is the name path of a clustering column: a single segment for a top-level column, and multiple segments for a nested column (for example, `["user", "address", "city"]`). If [Column Mapping](#column-mapping) is enabled, physical names are used for each segment.
 
 
 # Variant Data Type
@@ -2490,6 +2492,34 @@ When `delta.identity.allowExplicitInsert` is false, writers should meet the foll
 - Overflow when calculating generated Identity values should be detected and such writes should not be allowed.
 - `delta.identity.highWaterMark` should be updated to the new highest value when the write operation commits.
 
+## Materialize Partition Columns
+
+When this feature is supported, partition columns are physically written to Parquet files alongside the data columns. To support this feature:
+ - The table must be on Writer Version 7, and a feature name `materializePartitionColumns` must exist in the table `protocol`'s `writerFeatures`.
+
+Unlike most writer features, `materializePartitionColumns` has no associated `delta.enable*` table property and defines no additional metadata requirements. It is therefore [active](#active-features) whenever it is [supported](#supported-features): its presence in the `protocol`'s `writerFeatures` alone forces the writer requirements below. Hence, for this feature, the terms *supported*, *enabled*, and *active* (including in the decision matrix below) all refer to the same state.
+
+When supported:
+ - When the writer feature `materializePartitionColumns` is set in the protocol, writers must materialize partition columns into every newly added Parquet data file referenced by an `AddFile` action. Each materialized partition column value must equal the corresponding logical partition value recorded in that file's `AddFile` `partitionValues`. This mimics the same partition column materialization requirement from [IcebergCompatV1](#iceberg-compatibility-v1) and [IcebergCompatV2](#iceberg-compatibility-v2). As such, the `materializePartitionColumns` feature can be seen as a subset of the requirements imposed by those features, providing the partition column materialization guarantee independently without requiring full Iceberg compatibility.
+ - When the writer feature `materializePartitionColumns` is not set in the table protocol, writers are not required to write partition columns to data files. Note that other features might still require materialization of partition values, such as [IcebergCompatV1](#iceberg-compatibility-v1).
+
+When [Column Mapping](#column-mapping) is enabled, materialized partition columns are written to the Parquet data file using their assigned physical column names and field IDs, the same as data columns.
+
+This feature does not impose any requirements on readers. All Delta readers must be able to read the table regardless of whether partition columns are materialized in the data files. If partition values are present in both parquet and AddFile metadata, Delta readers should continue to read partition values from AddFile metadata. Also, [file-level statistics](#per-file-statistics) should not be written for the partition column as it would repeat information already present in an AddFile's `partitionValues`.
+
+Note that this table feature, as well as [IcebergCompatV1](#iceberg-compatibility-v1) (and related table features that require partition column materialization), if enabled, take priority over the `delta.writePartitionColumnsToParquet` table property. In other words, if a table feature is enabled that requires materialization of partition columns, and table metadata contains a `false` value for `delta.writePartitionColumnsToParquet`, partition columns must be materialized.
+
+| Table feature enablement | Value of `delta.writePartitionColumnsToParquet` table property | Writer requirement |
+| ------------------------ | -------------------------------------------------------------- | ------------------ |
+| A writer feature requiring partition column materialization (eg. `materializePartitionColumns`) is *enabled* | `false` | Partition columns *must* be materialized in parquet data files |
+| A writer feature requiring partition column materialization (eg. `materializePartitionColumns`) is *enabled* | `true` | Partition columns *must* be materialized in parquet data files |
+| A writer feature requiring partition column materialization (eg. `materializePartitionColumns`) is *enabled* | unset | Partition columns *must* be materialized in parquet data files |
+| No writer feature requiring partition column materialization is enabled | `false` | Partition columns *should not* be materialized in parquet data files |
+| No writer feature requiring partition column materialization is enabled | `true` | Partition columns *should* be materialized in parquet data files |
+| No writer feature requiring partition column materialization is enabled | unset | No requirement on partition column materialization |
+
+The value of having both the table feature `materializePartitionColumns` and the table property `delta.writePartitionColumnsToParquet` supported is that not every table is going to need the heightened requirement of only allowing writes from writers that understand `materializePartitionColumns`. In other words, `materializePartitionColumns` imposes a writer compatibility edge that `delta.writePartitionColumnsToParquet` does not.
+
 ## Writer Version Requirements
 
 The requirements of the writers according to the protocol versions are summarized in the table below. Each row inherits the requirements from the preceding row.
@@ -2525,6 +2555,7 @@ Property | Description | Details
 `delta.parquet.compression.codec` | Compression codec writers SHOULD use for new Parquet data and checkpoint files. Changing this property does not affect existing files; a table may contain files written with different codecs, which is a normal and expected state. | Widely supported values (matched case-insensitively): `uncompressed`/`none` (no compression), `snappy`, `gzip`, `lz4` (deprecated, Hadoop framing), `lz4_raw` ([LZ4 block format](https://parquet.apache.org/docs/file-format/data-pages/compression/#lz4_raw)), `zstd`.<br><br>When absent, writers SHOULD default to `zstd`. If a writer does not support or recognize the specified codec, it SHOULD abort with an appropriate error or fall back to a default codec.<br><br>Readers SHOULD support all codecs listed above regardless of the current property value. Parquet files written with other [parquet-supported codecs](https://parquet.apache.org/docs/file-format/data-pages/compression/) may also exist; readers MAY support reading these files.
 `delta.parquet.format.version` | Parquet data page format writers SHOULD use for new data and checkpoint files. This property is a directive to writers only; readers do not need to consult it, as Parquet pages are self-describing via the `PageType` field in each page header. Changing this property does not affect existing files; a table MAY contain files written with different data page versions, which is a normal and expected state. | Valid values: `1.0.0` (DataPageV1) and `2.x.x` (DataPageV2, where `x.x` is any minor.patch version). Recommended values are `1.0.0` and `2.12.0`.<br><br>When absent, writers SHOULD default to `1.0.0`. Writers SHOULD validate this property and abort if the value does not match `1.0.0` or `2.MINOR.PATCH`.<br><br>Readers SHOULD support both DataPageV1 and DataPageV2 pages regardless of this property's value. Tables intended for access by engines beyond the Delta Lake connectors SHOULD use `1.0.0`, as DataPageV2 support varies across the broader Parquet ecosystem.
 `delta.enableVariantShredding` | When `true`, writers could write variant data to parquet files in [shredded](#variant-shredding) format. | Valid values: `true` (shredding allowed) and `false` (shredding not allowed).<br><br>When enabled, writers must ensure that the `variantShredding` table feature is present in the table `protocol`'s `readerFeatures` and `writerFeatures`.
+`delta.writePartitionColumnsToParquet` | Controls whether writers SHOULD write partition columns in newly written data parquet files, in the absence of any writer features that necessitate writing of partition columns (eg. `IcebergCompatV1`). In other words, if no writer feature is enabled that requires materialization of partition columns, writers should read this property to decide whether to materialize partition columns in data parquet files or not. Writer features requirements take precedence over this property's value. Readers should continue to read partition values off of AddFile actions, regardless of the presence of partition values in data files. File-level statistics should not be present for partition columns in partitioned tables in any case. This setting does not apply to writers of files of any other file format. | Boolean field, with valid values `false` and `true`.
 
 # Appendix
 
@@ -2551,6 +2582,7 @@ Feature | Name | Readers or Writers?
 [Clustered Table](#clustered-table) | `clustering` | Writers only
 [VACUUM Protocol Check](#vacuum-protocol-check) | `vacuumProtocolCheck` | Readers and Writers
 [In-Commit Timestamps](#in-commit-timestamps) | `inCommitTimestamp` | Writers only
+[Materialize Partition Columns](#materialize-partition-columns) | `materializePartitionColumns` | Writers only
 
 ## Deletion Vector Format