From 74097f69dd1e66612b85d5ee903b0333c6863c33 Mon Sep 17 00:00:00 2001 From: Xinyi Lu Date: Wed, 29 Jul 2026 21:46:47 -0700 Subject: [PATCH 1/2] Spark 4.1:Spark: Fix RewriteManifestsSparkAction sortBy() for dotted partition field names --- .../actions/RewriteManifestsSparkAction.java | 7 +++- .../actions/TestRewriteManifestsAction.java | 39 +++++++++++++++++++ 2 files changed, 44 insertions(+), 2 deletions(-) diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java index 668360c06730..907a2a066295 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java +++ b/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 diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java index 0120d4b5fe78..a45547ae90ff 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java +++ b/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 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(); From 2879c3ce85175284340753c6110c7b2ab7964974 Mon Sep 17 00:00:00 2001 From: Xinyi Lu Date: Thu, 30 Jul 2026 09:46:45 -0700 Subject: [PATCH 2/2] resolve comments and apply fix to all spark versions --- .../actions/RewriteManifestsSparkAction.java | 11 ++++- .../actions/TestRewriteManifestsAction.java | 41 +++++++++++++++++++ .../actions/RewriteManifestsSparkAction.java | 11 ++++- .../actions/TestRewriteManifestsAction.java | 41 +++++++++++++++++++ .../actions/RewriteManifestsSparkAction.java | 14 ++++--- .../actions/TestRewriteManifestsAction.java | 20 +++++---- 6 files changed, 120 insertions(+), 18 deletions(-) diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java index 475088b148c6..10c01b959cf3 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java @@ -314,10 +314,17 @@ 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 (escaping any embedded backtick) + // 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( + String.format( + "%s.`%s`", DATA_FILE_PARTITION_COLUMN_NAME, p.replace("`", "``")))) .toArray(Column[]::new); // Form a new temporary column to cluster manifests on, based on the custom clustering columns diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java index 0120d4b5fe78..7f42d351df41 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java @@ -756,6 +756,47 @@ public void testRewriteManifestsPartitionedTableWithCustomSorting() throws IOExc assertThat(manifestsDf.count()).as("There should be at least 2 manifests").isGreaterThan(1L); } + @TestTemplate + public void + testRewriteManifestsPartitionedTableWithCustomSortingOnFieldNameContainingDotAndBacktick() + throws IOException { + // Regression test: partitioning on a nested source column (struct "a" with child field + // "b`c") used to make sortBy() fail at execution time, since the resulting partition field is + // named "a.b`c" but the partition struct is flat - col() misparsed the dot as a further + // nesting level instead of treating "a.b`c" as a single field name, and even once quoted, the + // embedded backtick must itself be escaped or the quoting is malformed. + Schema schema = + new Schema( + optional(1, "a", Types.StructType.of(optional(2, "b`c", Types.StringType.get()))), + optional(3, "ds", Types.StringType.get())); + PartitionSpec spec = PartitionSpec.builderFor(schema).identity("a.b`c").build(); + Map 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`c")) + .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(); diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java index 668360c06730..1716ebf43b17 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java @@ -314,10 +314,17 @@ 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 (escaping any embedded backtick) + // 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( + String.format( + "%s.`%s`", DATA_FILE_PARTITION_COLUMN_NAME, p.replace("`", "``")))) .toArray(Column[]::new); // Form a new temporary column to cluster manifests on, based on the custom clustering columns diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java index 0120d4b5fe78..7f42d351df41 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java @@ -756,6 +756,47 @@ public void testRewriteManifestsPartitionedTableWithCustomSorting() throws IOExc assertThat(manifestsDf.count()).as("There should be at least 2 manifests").isGreaterThan(1L); } + @TestTemplate + public void + testRewriteManifestsPartitionedTableWithCustomSortingOnFieldNameContainingDotAndBacktick() + throws IOException { + // Regression test: partitioning on a nested source column (struct "a" with child field + // "b`c") used to make sortBy() fail at execution time, since the resulting partition field is + // named "a.b`c" but the partition struct is flat - col() misparsed the dot as a further + // nesting level instead of treating "a.b`c" as a single field name, and even once quoted, the + // embedded backtick must itself be escaped or the quoting is malformed. + Schema schema = + new Schema( + optional(1, "a", Types.StructType.of(optional(2, "b`c", Types.StringType.get()))), + optional(3, "ds", Types.StringType.get())); + PartitionSpec spec = PartitionSpec.builderFor(schema).identity("a.b`c").build(); + Map 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`c")) + .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(); diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java index 907a2a066295..1716ebf43b17 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java @@ -314,13 +314,17 @@ private Column sortColumn() { partitionFieldClustering); // Map the top level partition column names to the column name referenced within the manifest - // 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. + // entry dataframe. Backtick-quote the partition field name (escaping any embedded backtick) + // 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( + String.format( + "%s.`%s`", DATA_FILE_PARTITION_COLUMN_NAME, p.replace("`", "``")))) .toArray(Column[]::new); // Form a new temporary column to cluster manifests on, based on the custom clustering columns diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java index a45547ae90ff..7f42d351df41 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java @@ -757,17 +757,19 @@ public void testRewriteManifestsPartitionedTableWithCustomSorting() throws IOExc } @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. + public void + testRewriteManifestsPartitionedTableWithCustomSortingOnFieldNameContainingDotAndBacktick() + throws IOException { + // Regression test: partitioning on a nested source column (struct "a" with child field + // "b`c") used to make sortBy() fail at execution time, since the resulting partition field is + // named "a.b`c" but the partition struct is flat - col() misparsed the dot as a further + // nesting level instead of treating "a.b`c" as a single field name, and even once quoted, the + // embedded backtick must itself be escaped or the quoting is malformed. Schema schema = new Schema( - optional(1, "a", Types.StructType.of(optional(2, "b", Types.StringType.get()))), + optional(1, "a", Types.StructType.of(optional(2, "b`c", Types.StringType.get()))), optional(3, "ds", Types.StringType.get())); - PartitionSpec spec = PartitionSpec.builderFor(schema).identity("a.b").build(); + PartitionSpec spec = PartitionSpec.builderFor(schema).identity("a.b`c").build(); Map options = Maps.newHashMap(); options.put(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)); options.put(TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED, snapshotIdInheritanceEnabled); @@ -787,7 +789,7 @@ public void testRewriteManifestsPartitionedTableWithCustomSortingOnFieldNameCont SparkActions.get() .rewriteManifests(table) .rewriteIf(manifest -> true) - .sortBy(ImmutableList.of("a.b")) + .sortBy(ImmutableList.of("a.b`c")) .option(RewriteManifestsSparkAction.USE_CACHING, useCaching) .execute();