You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ru...@apache.org on 2026/09/22 21:43:00 UTC
(iceberg) branch main updated: Spark: Fix RewriteManifestsSparkAction sortBy() for dotted partition field names (#17428)
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 <12...@users.noreply.github.com>
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();