You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ao...@apache.org on 2023/02/03 00:23:59 UTC
[iceberg] branch master updated: Spark 3.3: Handle no-op for rewrite manifests procedure/action (#6695)
This is an automated email from the ASF dual-hosted git repository.
aokolnychyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 6165a7d816 Spark 3.3: Handle no-op for rewrite manifests procedure/action (#6695)
6165a7d816 is described below
commit 6165a7d816519b96294453eebfde7d80dd941c14
Author: Ajantha Bhat <aj...@gmail.com>
AuthorDate: Fri Feb 3 05:53:54 2023 +0530
Spark 3.3: Handle no-op for rewrite manifests procedure/action (#6695)
---
.../extensions/TestRewriteManifestsProcedure.java | 22 ++++++++++++++++++++
.../spark/actions/RewriteManifestsSparkAction.java | 4 ++++
.../spark/actions/TestRewriteManifestsAction.java | 24 +++++++++++++++++-----
3 files changed, 45 insertions(+), 5 deletions(-)
diff --git a/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java b/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java
index 784e5ed1e7..40625b5e34 100644
--- a/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java
+++ b/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java
@@ -78,6 +78,28 @@ public class TestRewriteManifestsProcedure extends SparkExtensionsTestBase {
"Must have 4 manifests", 4, table.currentSnapshot().allManifests(table.io()).size());
}
+ @Test
+ public void testRewriteManifestsNoOp() {
+ sql(
+ "CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg PARTITIONED BY (data)",
+ tableName);
+ sql("INSERT INTO TABLE %s VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd')", tableName);
+
+ Table table = validationCatalog.loadTable(tableIdent);
+
+ Assert.assertEquals(
+ "Must have 1 manifest", 1, table.currentSnapshot().allManifests(table.io()).size());
+
+ List<Object[]> output = sql("CALL %s.system.rewrite_manifests('%s')", catalogName, tableIdent);
+ // should not rewrite any manifests for no-op (output of rewrite is same as before and after)
+ assertEquals("Procedure output must match", ImmutableList.of(row(0, 0)), output);
+
+ table.refresh();
+
+ Assert.assertEquals(
+ "Must have 1 manifests", 1, table.currentSnapshot().allManifests(table.io()).size());
+ }
+
@Test
public void testRewriteLargeManifestsOnDatePartitionedTableWithJava8APIEnabled() {
withSQLConf(
diff --git a/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java b/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java
index b598bcb19b..42977b0be5 100644
--- a/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java
+++ b/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java
@@ -172,6 +172,10 @@ public class RewriteManifestsSparkAction
int targetNumManifests = targetNumManifests(totalSizeBytes);
int targetNumManifestEntries = targetNumManifestEntries(numEntries, targetNumManifests);
+ if (targetNumManifests == 1 && matchingManifests.size() == 1) {
+ return BaseRewriteManifestsActionResult.empty();
+ }
+
Dataset<Row> manifestEntryDF = buildManifestEntryDF(matchingManifests);
List<ManifestFile> newManifests;
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
index 95f2f12d5f..9382846e1e 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
@@ -341,6 +341,10 @@ public class TestRewriteManifestsAction extends SparkTestBase {
SparkTableUtil.importSparkTable(
spark, new TableIdentifier("parquet_table"), table, stagingDir.toString());
+ // add some more data to create more than one manifest for the rewrite
+ inputDF.select("c1", "c2", "c3").write().format("iceberg").mode("append").save(tableLocation);
+ table.refresh();
+
Snapshot snapshot = table.currentSnapshot();
SparkActions actions = SparkActions.get();
@@ -431,6 +435,8 @@ public class TestRewriteManifestsAction extends SparkTestBase {
new ThreeColumnRecord(1, null, "AAAA"), new ThreeColumnRecord(1, "BBBBBBBBBB", "BBBB"));
writeRecords(records1);
+ writeRecords(records1);
+
List<ThreeColumnRecord> records2 =
Lists.newArrayList(
new ThreeColumnRecord(2, "CCCCCCCCCC", "CCCC"),
@@ -440,7 +446,7 @@ public class TestRewriteManifestsAction extends SparkTestBase {
table.refresh();
List<ManifestFile> manifests = table.currentSnapshot().allManifests(table.io());
- Assert.assertEquals("Should have 2 manifests before rewrite", 2, manifests.size());
+ Assert.assertEquals("Should have 3 manifests before rewrite", 3, manifests.size());
SparkActions actions = SparkActions.get();
@@ -448,13 +454,16 @@ public class TestRewriteManifestsAction extends SparkTestBase {
RewriteManifests.Result result =
actions
.rewriteManifests(table)
- .rewriteIf(manifest -> manifest.path().equals(manifests.get(0).path()))
+ .rewriteIf(
+ manifest ->
+ (manifest.path().equals(manifests.get(0).path())
+ || (manifest.path().equals(manifests.get(1).path()))))
.stagingLocation(temp.newFolder().toString())
.option("use-caching", "false")
.execute();
Assert.assertEquals(
- "Action should rewrite 1 manifest", 1, Iterables.size(result.rewrittenManifests()));
+ "Action should rewrite 2 manifest", 2, Iterables.size(result.rewrittenManifests()));
Assert.assertEquals(
"Action should add 1 manifests", 1, Iterables.size(result.addedManifests()));
@@ -464,11 +473,16 @@ public class TestRewriteManifestsAction extends SparkTestBase {
Assert.assertEquals("Should have 2 manifests after rewrite", 2, newManifests.size());
Assert.assertFalse("First manifest must be rewritten", newManifests.contains(manifests.get(0)));
+ Assert.assertFalse(
+ "Second manifest must be rewritten", newManifests.contains(manifests.get(1)));
Assert.assertTrue(
- "Second manifest must not be rewritten", newManifests.contains(manifests.get(1)));
+ "Third manifest must not be rewritten", newManifests.contains(manifests.get(2)));
List<ThreeColumnRecord> expectedRecords = Lists.newArrayList();
- expectedRecords.addAll(records1);
+ expectedRecords.add(records1.get(0));
+ expectedRecords.add(records1.get(0));
+ expectedRecords.add(records1.get(1));
+ expectedRecords.add(records1.get(1));
expectedRecords.addAll(records2);
Dataset<Row> resultDF = spark.read().format("iceberg").load(tableLocation);