This is an automated email from the ASF dual-hosted git repository.
RussellSpitzer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 3deb3e2275 Spark: Fix RewriteManifestsSparkAction sortBy() for dotted
partition field names (#17428)
3deb3e2275 is described below
commit 3deb3e2275ee1c6e8e386c10f63a9aeed719d638
Author: Xinyi Lu <[email protected]>
AuthorDate: Tue Sep 22 14:42:54 2026 -0700
Spark: Fix RewriteManifestsSparkAction sortBy() for dotted partition field
names (#17428)
---
.../spark/actions/RewriteManifestsSparkAction.java | 2 +-
.../spark/actions/TestRewriteManifestsAction.java | 36 ++++++++++++++++++++++
.../spark/actions/RewriteManifestsSparkAction.java | 2 +-
.../spark/actions/TestRewriteManifestsAction.java | 36 ++++++++++++++++++++++
.../spark/actions/RewriteManifestsSparkAction.java | 2 +-
.../spark/actions/TestRewriteManifestsAction.java | 36 ++++++++++++++++++++++
.../spark/actions/RewriteManifestsSparkAction.java | 2 +-
.../spark/actions/TestRewriteManifestsAction.java | 36 ++++++++++++++++++++++
8 files changed, 148 insertions(+), 4 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 cb11fa89a6..ec0599aef1 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
@@ -318,7 +318,7 @@ public class RewriteManifestsSparkAction
// entry dataframe
Column[] partitionColumns =
partitionFieldClustering.stream()
- .map(p -> col(DATA_FILE_PARTITION_COLUMN_NAME + "." + p))
+ .map(p -> col(DATA_FILE_PARTITION_COLUMN_NAME).getField(p))
.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 0120d4b5fe..8d9619b6d7 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,42 @@ public class TestRewriteManifestsAction extends TestBase {
assertThat(manifestsDf.count()).as("There should be at least 2
manifests").isGreaterThan(1L);
}
+ @TestTemplate
+ public void
+
testRewriteManifestsPartitionedTableWithCustomSortingOnFieldNameContainingDotAndBacktick()
+ throws IOException {
+ 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<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`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 e9195b46fd..513991f5cc 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
@@ -318,7 +318,7 @@ public class RewriteManifestsSparkAction
// entry dataframe
Column[] partitionColumns =
partitionFieldClustering.stream()
- .map(p -> col(DATA_FILE_PARTITION_COLUMN_NAME + "." + p))
+ .map(p -> col(DATA_FILE_PARTITION_COLUMN_NAME).getField(p))
.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 0120d4b5fe..8d9619b6d7 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,42 @@ public class TestRewriteManifestsAction extends TestBase {
assertThat(manifestsDf.count()).as("There should be at least 2
manifests").isGreaterThan(1L);
}
+ @TestTemplate
+ public void
+
testRewriteManifestsPartitionedTableWithCustomSortingOnFieldNameContainingDotAndBacktick()
+ throws IOException {
+ 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<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`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 e9195b46fd..513991f5cc 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
@@ -318,7 +318,7 @@ public class RewriteManifestsSparkAction
// entry dataframe
Column[] partitionColumns =
partitionFieldClustering.stream()
- .map(p -> col(DATA_FILE_PARTITION_COLUMN_NAME + "." + p))
+ .map(p -> col(DATA_FILE_PARTITION_COLUMN_NAME).getField(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 0120d4b5fe..8d9619b6d7 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,42 @@ public class TestRewriteManifestsAction extends TestBase {
assertThat(manifestsDf.count()).as("There should be at least 2
manifests").isGreaterThan(1L);
}
+ @TestTemplate
+ public void
+
testRewriteManifestsPartitionedTableWithCustomSortingOnFieldNameContainingDotAndBacktick()
+ throws IOException {
+ 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<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`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.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java
index e9195b46fd..513991f5cc 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java
@@ -318,7 +318,7 @@ public class RewriteManifestsSparkAction
// entry dataframe
Column[] partitionColumns =
partitionFieldClustering.stream()
- .map(p -> col(DATA_FILE_PARTITION_COLUMN_NAME + "." + p))
+ .map(p -> col(DATA_FILE_PARTITION_COLUMN_NAME).getField(p))
.toArray(Column[]::new);
// Form a new temporary column to cluster manifests on, based on the
custom clustering columns
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
index 0120d4b5fe..8d9619b6d7 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
@@ -756,6 +756,42 @@ public class TestRewriteManifestsAction extends TestBase {
assertThat(manifestsDf.count()).as("There should be at least 2
manifests").isGreaterThan(1L);
}
+ @TestTemplate
+ public void
+
testRewriteManifestsPartitionedTableWithCustomSortingOnFieldNameContainingDotAndBacktick()
+ throws IOException {
+ 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<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`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();
