Spark: Fix RewriteManifestsSparkAction sortBy() for dotted partition field names

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java

@@ -314,10 +314,13 @@ private Column sortColumn() {
           partitionFieldClustering);
 
       // Map the top level partition column names to the column name referenced within the manifest
-      // entry dataframe
+      // entry dataframe. Backtick-quote the partition field name because it may itself contain a
+      // literal '.' (e.g. when partitioning on a nested source column) - the partition struct is
+      // always flat, so unquoted, col() would misparse that dot as a further level of nesting
+      // instead of treating the whole name as one field.
       Column[] partitionColumns =
           partitionFieldClustering.stream()
-              .map(p -> col(DATA_FILE_PARTITION_COLUMN_NAME + "." + p))
+              .map(p -> col(DATA_FILE_PARTITION_COLUMN_NAME + ".`" + p + "`"))
               .toArray(Column[]::new);
 
       // Form a new temporary column to cluster manifests on, based on the custom clustering columns

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java

@@ -756,6 +756,45 @@ public void testRewriteManifestsPartitionedTableWithCustomSorting() throws IOExc
     assertThat(manifestsDf.count()).as("There should be at least 2 manifests").isGreaterThan(1L);
   }
 
+  @TestTemplate
+  public void testRewriteManifestsPartitionedTableWithCustomSortingOnFieldNameContainingDot()
+      throws IOException {
+    // Regression test: partitioning on a nested source column (struct "a" with child field "b")
+    // used to make sortBy() fail at execution time, since the resulting partition field is named
+    // "a.b" but the partition struct is flat - col() misparsed the dot as a further nesting level
+    // instead of treating "a.b" as a single field name.
+    Schema schema =
+        new Schema(
+            optional(1, "a", Types.StructType.of(optional(2, "b", Types.StringType.get()))),
+            optional(3, "ds", Types.StringType.get()));
+    PartitionSpec spec = PartitionSpec.builderFor(schema).identity("a.b").build();
+    Map<String, String> options = Maps.newHashMap();
+    options.put(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion));
+    options.put(TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED, snapshotIdInheritanceEnabled);
+    Table table = TABLES.create(schema, spec, options, tableLocation);
+
+    // write two manifests so the rewrite doesn't short-circuit via the "already optimal" early
+    // return, and sortColumn() actually gets exercised
+    ManifestFile manifest1 =
+        writeManifest(table, ImmutableList.of(newDataFile(table, TestHelpers.Row.of("val1"))));
+    table.newFastAppend().appendManifest(manifest1).commit();
+
+    ManifestFile manifest2 =
+        writeManifest(table, ImmutableList.of(newDataFile(table, TestHelpers.Row.of("val2"))));
+    table.newFastAppend().appendManifest(manifest2).commit();
+
+    RewriteManifests.Result result =
+        SparkActions.get()
+            .rewriteManifests(table)
+            .rewriteIf(manifest -> true)
+            .sortBy(ImmutableList.of("a.b"))
+            .option(RewriteManifestsSparkAction.USE_CACHING, useCaching)
+            .execute();
+
+    assertThat(result.rewrittenManifests()).hasSize(2);
+    assertThat(result.addedManifests()).hasSizeGreaterThanOrEqualTo(1);
+  }
+
   @TestTemplate
   public void testRewriteManifestsWithPredicate() throws IOException {
     PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("c1").truncate("c2", 2).build();