Skip to content

Spark 3.5, 4.0: Add sort_by parameter to rewrite_manifests procedure #18065

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
huaxingao merged 3 commits into from
Sep 14, 2026
Merged

Spark 3.5, 4.0: Add sort_by parameter to rewrite_manifests procedure #18065

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5b83203
Spark 3.5, 4.0: Add sort_by parameter to rewrite_manifests procedure
hemanthboyina Sep 11, 2026
70b8e45
fix test errors
hemanthboyina Sep 11, 2026
9a61573
fix checkstyle too
hemanthboyina Sep 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 97 additions & 0 deletions ...ions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,6 +394,103 @@ public void testWriteManifestWithSpecId() {
sql("SELECT partition_spec_id FROM %s.manifests order by 1 asc", tableName));
}

@TestTemplate
public void testRewriteManifestsWithSortBy() {
sql(
"CREATE TABLE %s (id bigint NOT NULL, data string, category string) USING iceberg PARTITIONED BY (data, category)",
tableName);

sql("INSERT INTO TABLE %s VALUES (1, 'a', 'x')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'b', 'y')", tableName);
sql("INSERT INTO TABLE %s VALUES (3, 'c', 'x')", tableName);
sql("INSERT INTO TABLE %s VALUES (4, 'd', 'y')", tableName);

Table table = validationCatalog.loadTable(tableIdent);

assertThat(table.currentSnapshot().allManifests(table.io()))
.as("Must have 4 manifests")
.hasSize(4);

List<Object[]> output =
sql(
"CALL %s.system.rewrite_manifests(table => '%s', sort_by => array('category', 'data'))",
catalogName, tableIdent);
assertEquals("Procedure output must match", ImmutableList.of(row(4, 1)), output);

table.refresh();

assertThat(table.currentSnapshot().allManifests(table.io()))
.as("Must have 1 manifest")
.hasSize(1);
}

@TestTemplate
public void testRewriteManifestsWithSortBySingleColumn() {
sql(
"CREATE TABLE %s (id bigint NOT NULL, data string, category string) USING iceberg PARTITIONED BY (data, category)",
tableName);

sql("INSERT INTO TABLE %s VALUES (1, 'a', 'x')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'b', 'y')", tableName);
sql("INSERT INTO TABLE %s VALUES (3, 'c', 'x')", tableName);
sql("INSERT INTO TABLE %s VALUES (4, 'd', 'y')", tableName);

Table table = validationCatalog.loadTable(tableIdent);

assertThat(table.currentSnapshot().allManifests(table.io()))
.as("Must have 4 manifests")
.hasSize(4);

List<Object[]> output =
sql(
"CALL %s.system.rewrite_manifests(table => '%s', sort_by => array('category'))",
catalogName, tableIdent);
assertEquals("Procedure output must match", ImmutableList.of(row(4, 1)), output);

table.refresh();

assertThat(table.currentSnapshot().allManifests(table.io()))
.as("Must have 1 manifest")
.hasSize(1);
}

@TestTemplate
public void testRewriteManifestsWithInvalidSortBy() {
sql(
"CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg PARTITIONED BY (data)",
tableName);

sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'b')", tableName);

assertThatThrownBy(
() ->
sql(
"CALL %s.system.rewrite_manifests(table => '%s', sort_by => array('nonexistent'))",
catalogName, tableIdent))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("not found in current partition spec");
}

@TestTemplate
public void testRewriteManifestsWithEmptySortBy() {
sql(
"CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg PARTITIONED BY (data)",
tableName);

sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);

// The Iceberg CALL grammar requires ARRAY to have at least one element, so an empty sort_by
// array is rejected at parse time rather than by the procedure's runtime validation.
assertThatThrownBy(
() ->
sql(
"CALL %s.system.rewrite_manifests(table => '%s', sort_by => array())",
catalogName, tableIdent))
.isInstanceOf(AnalysisException.class)
.hasMessageContaining("no viable alternative at input ')'");
}

@TestTemplate
public void testPartitionStatsIncrementalCompute() throws IOException {
sql(
Expand Down
13 changes: 12 additions & 1 deletion ....5/spark/src/main/java/org/apache/iceberg/spark/procedures/RewriteManifestsProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,10 @@
*/
package org.apache.iceberg.spark.procedures;

import java.util.Arrays;
import org.apache.iceberg.Table;
import org.apache.iceberg.actions.RewriteManifests;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.spark.actions.RewriteManifestsSparkAction;
import org.apache.iceberg.spark.actions.SparkActions;
Expand Down Expand Up @@ -49,9 +51,11 @@ class RewriteManifestsProcedure extends BaseProcedure {
optionalInParameter("use_caching", DataTypes.BooleanType);
private static final ProcedureParameter SPEC_ID_PARAM =
optionalInParameter("spec_id", DataTypes.IntegerType);
private static final ProcedureParameter SORT_BY_PARAM =
optionalInParameter("sort_by", STRING_ARRAY);

private static final ProcedureParameter[] PARAMETERS =
new ProcedureParameter[] {TABLE_PARAM, USE_CACHING_PARAM, SPEC_ID_PARAM};
new ProcedureParameter[] {TABLE_PARAM, USE_CACHING_PARAM, SPEC_ID_PARAM, SORT_BY_PARAM};

// counts are not nullable since the action result is never null
private static final StructType OUTPUT_TYPE =
Expand Down Expand Up @@ -91,6 +95,7 @@ public InternalRow[] call(InternalRow args) {
Identifier tableIdent = input.ident(TABLE_PARAM);
Boolean useCaching = input.asBoolean(USE_CACHING_PARAM, null);
Integer specId = input.asInt(SPEC_ID_PARAM, null);
String[] sortBy = input.asStringArray(SORT_BY_PARAM, null);

return modifyIcebergTable(
tableIdent,
Expand All @@ -105,6 +110,12 @@ public InternalRow[] call(InternalRow args) {
action.specId(specId);
}

if (sortBy != null) {
Preconditions.checkArgument(
sortBy.length > 0, "sort_by must not be empty when provided");
action.sortBy(Arrays.asList(sortBy));
}

RewriteManifests.Result result = action.execute();

return toOutputRows(result);
Expand Down
97 changes: 96 additions & 1 deletion ...ions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -318,7 +318,7 @@ public void testInvalidRewriteManifestsCases() {
() -> sql("CALL %s.system.rewrite_manifests(table => 't', tAbLe => 't')", catalogName))
.isInstanceOf(AnalysisException.class)
.hasMessage(
"[UNRECOGNIZED_PARAMETER_NAME] Cannot invoke routine `rewrite_manifests` because the routine call included a named argument reference for the argument named `tAbLe`, but this routine does not include any signature containing an argument with this name. Did you mean one of the following? [`table` `spec_id` `use_caching`]. SQLSTATE: 4274K");
"[UNRECOGNIZED_PARAMETER_NAME] Cannot invoke routine `rewrite_manifests` because the routine call included a named argument reference for the argument named `tAbLe`, but this routine does not include any signature containing an argument with this name. Did you mean one of the following? [`table` `sort_by` `spec_id`]. SQLSTATE: 4274K");

assertThatThrownBy(() -> sql("CALL %s.system.rewrite_manifests('')", catalogName))
.isInstanceOf(IllegalArgumentException.class)
Expand Down Expand Up @@ -394,6 +394,101 @@ public void testWriteManifestWithSpecId() {
sql("SELECT partition_spec_id FROM %s.manifests order by 1 asc", tableName));
}

@TestTemplate
public void testRewriteManifestsWithSortBy() {
sql(
"CREATE TABLE %s (id bigint NOT NULL, data string, category string) USING iceberg PARTITIONED BY (data, category)",
tableName);

sql("INSERT INTO TABLE %s VALUES (1, 'a', 'x')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'b', 'y')", tableName);
sql("INSERT INTO TABLE %s VALUES (3, 'c', 'x')", tableName);
sql("INSERT INTO TABLE %s VALUES (4, 'd', 'y')", tableName);

Table table = validationCatalog.loadTable(tableIdent);

assertThat(table.currentSnapshot().allManifests(table.io()))
.as("Must have 4 manifests")
.hasSize(4);

List<Object[]> output =
sql(
"CALL %s.system.rewrite_manifests(table => '%s', sort_by => array('category', 'data'))",
catalogName, tableIdent);
assertEquals("Procedure output must match", ImmutableList.of(row(4, 1)), output);

table.refresh();

assertThat(table.currentSnapshot().allManifests(table.io()))
.as("Must have 1 manifest")
.hasSize(1);
}

@TestTemplate
public void testRewriteManifestsWithSortBySingleColumn() {
sql(
"CREATE TABLE %s (id bigint NOT NULL, data string, category string) USING iceberg PARTITIONED BY (data, category)",
tableName);

sql("INSERT INTO TABLE %s VALUES (1, 'a', 'x')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'b', 'y')", tableName);
sql("INSERT INTO TABLE %s VALUES (3, 'c', 'x')", tableName);
sql("INSERT INTO TABLE %s VALUES (4, 'd', 'y')", tableName);

Table table = validationCatalog.loadTable(tableIdent);

assertThat(table.currentSnapshot().allManifests(table.io()))
.as("Must have 4 manifests")
.hasSize(4);

List<Object[]> output =
sql(
"CALL %s.system.rewrite_manifests(table => '%s', sort_by => array('category'))",
catalogName, tableIdent);
assertEquals("Procedure output must match", ImmutableList.of(row(4, 1)), output);

table.refresh();

assertThat(table.currentSnapshot().allManifests(table.io()))
.as("Must have 1 manifest")
.hasSize(1);
}

@TestTemplate
public void testRewriteManifestsWithInvalidSortBy() {
sql(
"CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg PARTITIONED BY (data)",
tableName);

sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'b')", tableName);

assertThatThrownBy(
() ->
sql(
"CALL %s.system.rewrite_manifests(table => '%s', sort_by => array('nonexistent'))",
catalogName, tableIdent))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("not found in current partition spec");
}

@TestTemplate
public void testRewriteManifestsWithEmptySortBy() {
sql(
"CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg PARTITIONED BY (data)",
tableName);

sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);

assertThatThrownBy(
() ->
sql(
"CALL %s.system.rewrite_manifests(table => '%s', sort_by => array())",
catalogName, tableIdent))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("sort_by must not be empty when provided");
}

@TestTemplate
public void testPartitionStatsIncrementalCompute() throws IOException {
sql(
Expand Down
13 changes: 12 additions & 1 deletion ....0/spark/src/main/java/org/apache/iceberg/spark/procedures/RewriteManifestsProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,11 @@
*/
package org.apache.iceberg.spark.procedures;

import java.util.Arrays;
import java.util.Iterator;
import org.apache.iceberg.Table;
import org.apache.iceberg.actions.RewriteManifests;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.spark.actions.RewriteManifestsSparkAction;
import org.apache.iceberg.spark.actions.SparkActions;
Expand Down Expand Up @@ -54,9 +56,11 @@ class RewriteManifestsProcedure extends BaseProcedure {
optionalInParameter("use_caching", DataTypes.BooleanType);
private static final ProcedureParameter SPEC_ID_PARAM =
optionalInParameter("spec_id", DataTypes.IntegerType);
private static final ProcedureParameter SORT_BY_PARAM =
optionalInParameter("sort_by", STRING_ARRAY);

private static final ProcedureParameter[] PARAMETERS =
new ProcedureParameter[] {TABLE_PARAM, USE_CACHING_PARAM, SPEC_ID_PARAM};
new ProcedureParameter[] {TABLE_PARAM, USE_CACHING_PARAM, SPEC_ID_PARAM, SORT_BY_PARAM};

// counts are not nullable since the action result is never null
private static final StructType OUTPUT_TYPE =
Expand Down Expand Up @@ -96,6 +100,7 @@ public Iterator<Scan> call(InternalRow args) {
Identifier tableIdent = input.ident(TABLE_PARAM);
Boolean useCaching = input.asBoolean(USE_CACHING_PARAM, null);
Integer specId = input.asInt(SPEC_ID_PARAM, null);
String[] sortBy = input.asStringArray(SORT_BY_PARAM, null);

return modifyIcebergTable(
tableIdent,
Expand All @@ -110,6 +115,12 @@ public Iterator<Scan> call(InternalRow args) {
action.specId(specId);
}

if (sortBy != null) {
Preconditions.checkArgument(
sortBy.length > 0, "sort_by must not be empty when provided");
action.sortBy(Arrays.asList(sortBy));
}

RewriteManifests.Result result = action.execute();

return asScanIterator(OUTPUT_TYPE, toOutputRows(result));
Expand Down
Loading