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 668360c06730..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,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.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..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 @@ -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();