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 2021/12/01 11:25:17 UTC
[iceberg] branch master updated: Core: Handle metrics-based deletes (#3600)
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 b67f317 Core: Handle metrics-based deletes (#3600)
b67f317 is described below
commit b67f317042135783bfa8e4c0a122def819b9ca31
Author: Anton Okolnychyi <ao...@apple.com>
AuthorDate: Wed Dec 1 13:25:04 2021 +0200
Core: Handle metrics-based deletes (#3600)
Co-authored-by: Ryan Blue <bl...@apache.org>
---
.../iceberg/expressions/ExpressionVisitors.java | 26 +++--
.../expressions/InclusiveMetricsEvaluator.java | 9 ++
.../expressions/StrictMetricsEvaluator.java | 9 ++
.../org/apache/iceberg/ManifestFilterManager.java | 88 ++++++++++-----
.../java/org/apache/iceberg/TestDeleteFiles.java | 122 ++++++++++++++++++++-
.../apache/iceberg/spark/sql/TestDeleteFrom.java | 115 ++++++++++---------
6 files changed, 271 insertions(+), 98 deletions(-)
diff --git a/api/src/main/java/org/apache/iceberg/expressions/ExpressionVisitors.java b/api/src/main/java/org/apache/iceberg/expressions/ExpressionVisitors.java
index cbec748..318b2dc 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/ExpressionVisitors.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/ExpressionVisitors.java
@@ -60,12 +60,6 @@ public class ExpressionVisitors {
}
}
- /**
- * This base class is used by existing visitors that have not been updated to extend BoundExpressionVisitor.
- *
- * @deprecated use {@link BoundVisitor} instead
- */
- @Deprecated
public abstract static class BoundExpressionVisitor<R> extends ExpressionVisitor<R> {
public <T> R isNull(BoundReference<T> ref) {
return null;
@@ -119,10 +113,26 @@ public class ExpressionVisitors {
throw new UnsupportedOperationException("Unsupported operation.");
}
+ /**
+ * Handle a non-reference value in this visitor.
+ * <p>
+ * Visitors that require {@link BoundReference references} and not {@link Bound terms} can use this method to
+ * return a default value for expressions with non-references. The default implementation will throw a validation
+ * exception because the non-reference is not supported.
+ *
+ * @param term a non-reference bound expression
+ * @param <T> a Java return type
+ * @return a return value for the visitor
+ */
+ public <T> R handleNonReference(Bound<T> term) {
+ throw new ValidationException("Visitor %s does not support non-reference: %s", this, term);
+ }
+
@Override
public <T> R predicate(BoundPredicate<T> pred) {
- ValidationException.check(pred.term() instanceof BoundReference,
- "Visitor %s does not support expression: %s", this, pred.term());
+ if (!(pred.term() instanceof BoundReference)) {
+ return handleNonReference(pred.term());
+ }
if (pred.isLiteralPredicate()) {
BoundLiteralPredicate<T> literalPred = pred.asLiteralPredicate();
diff --git a/api/src/main/java/org/apache/iceberg/expressions/InclusiveMetricsEvaluator.java b/api/src/main/java/org/apache/iceberg/expressions/InclusiveMetricsEvaluator.java
index a6c3c65..a87a38d 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/InclusiveMetricsEvaluator.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/InclusiveMetricsEvaluator.java
@@ -108,6 +108,15 @@ public class InclusiveMetricsEvaluator {
}
@Override
+ public <T> Boolean handleNonReference(Bound<T> term) {
+ // If the term in any expression is not a direct reference, assume that rows may match. This happens when
+ // transforms or other expressions are passed to this evaluator. For example, bucket16(x) = 0 can't be determined
+ // because this visitor operates on data metrics and not partition values. It may be possible to un-transform
+ // expressions for order preserving transforms in the future, but this is not currently supported.
+ return ROWS_MIGHT_MATCH;
+ }
+
+ @Override
public Boolean alwaysTrue() {
return ROWS_MIGHT_MATCH; // all rows match
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/StrictMetricsEvaluator.java b/api/src/main/java/org/apache/iceberg/expressions/StrictMetricsEvaluator.java
index 6ea293a..1eb23ea 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/StrictMetricsEvaluator.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/StrictMetricsEvaluator.java
@@ -98,6 +98,15 @@ public class StrictMetricsEvaluator {
}
@Override
+ public <T> Boolean handleNonReference(Bound<T> term) {
+ // If the term in any expression is not a direct reference, assume that rows may not match. This happens when
+ // transforms or other expressions are passed to this evaluator. For example, bucket16(x) = 0 can't be determined
+ // because this visitor operates on data metrics and not partition values. It may be possible to un-transform
+ // expressions for order preserving transforms in the future, but this is not currently supported.
+ return ROWS_MIGHT_NOT_MATCH;
+ }
+
+ @Override
public Boolean alwaysTrue() {
return ROWS_MUST_MATCH; // all rows match
}
diff --git a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
index 8ecebe3..1057ed4 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
@@ -26,11 +26,11 @@ import java.util.Map;
import java.util.Set;
import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.exceptions.ValidationException;
-import org.apache.iceberg.expressions.Evaluator;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.InclusiveMetricsEvaluator;
import org.apache.iceberg.expressions.ManifestEvaluator;
-import org.apache.iceberg.expressions.Projections;
+import org.apache.iceberg.expressions.ResidualEvaluator;
import org.apache.iceberg.expressions.StrictMetricsEvaluator;
import org.apache.iceberg.relocated.com.google.common.base.Joiner;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -42,7 +42,9 @@ import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.util.CharSequenceSet;
import org.apache.iceberg.util.CharSequenceWrapper;
import org.apache.iceberg.util.ManifestFileUtil;
+import org.apache.iceberg.util.Pair;
import org.apache.iceberg.util.PartitionSet;
+import org.apache.iceberg.util.StructLikeMap;
import org.apache.iceberg.util.Tasks;
import org.apache.iceberg.util.ThreadPools;
import org.slf4j.Logger;
@@ -170,16 +172,13 @@ abstract class ManifestFilterManager<F extends ContentFile<F>> {
return ImmutableList.of();
}
- // use a common metrics evaluator for all manifests because it is bound to the table schema
- StrictMetricsEvaluator metricsEvaluator = new StrictMetricsEvaluator(tableSchema, deleteExpression);
-
ManifestFile[] filtered = new ManifestFile[manifests.size()];
// open all of the manifest files in parallel, use index to avoid reordering
Tasks.range(filtered.length)
.stopOnFailure().throwFailureWhenFinished()
.executeWith(ThreadPools.getWorkerPool())
.run(index -> {
- ManifestFile manifest = filterManifest(metricsEvaluator, manifests.get(index));
+ ManifestFile manifest = filterManifest(tableSchema, manifests.get(index));
filtered[index] = manifest;
});
@@ -276,7 +275,7 @@ abstract class ManifestFilterManager<F extends ContentFile<F>> {
/**
* @return a ManifestReader that is a filtered version of the input manifest.
*/
- private ManifestFile filterManifest(StrictMetricsEvaluator metricsEvaluator, ManifestFile manifest) {
+ private ManifestFile filterManifest(Schema tableSchema, ManifestFile manifest) {
ManifestFile cached = filteredManifests.get(manifest);
if (cached != null) {
return cached;
@@ -289,16 +288,19 @@ abstract class ManifestFilterManager<F extends ContentFile<F>> {
}
try (ManifestReader<F> reader = newManifestReader(manifest)) {
+ PartitionSpec spec = reader.spec();
+ PartitionAndMetricsEvaluator evaluator = new PartitionAndMetricsEvaluator(tableSchema, spec, deleteExpression);
+
// this assumes that the manifest doesn't have files to remove and streams through the
// manifest without copying data. if a manifest does have a file to remove, this will break
// out of the loop and move on to filtering the manifest.
- boolean hasDeletedFiles = manifestHasDeletedFiles(metricsEvaluator, reader);
+ boolean hasDeletedFiles = manifestHasDeletedFiles(evaluator, reader);
if (!hasDeletedFiles) {
filteredManifests.put(manifest, manifest);
return manifest;
}
- return filterManifestWithDeletedFiles(metricsEvaluator, manifest, reader);
+ return filterManifestWithDeletedFiles(evaluator, manifest, reader);
} catch (IOException e) {
throw new RuntimeIOException(e, "Failed to close manifest: %s", manifest);
@@ -339,20 +341,17 @@ abstract class ManifestFilterManager<F extends ContentFile<F>> {
}
@SuppressWarnings("CollectionUndefinedEquality")
- private boolean manifestHasDeletedFiles(
- StrictMetricsEvaluator metricsEvaluator, ManifestReader<F> reader) {
+ private boolean manifestHasDeletedFiles(PartitionAndMetricsEvaluator evaluator, ManifestReader<F> reader) {
boolean isDelete = reader.isDeleteManifestReader();
- Evaluator inclusive = inclusiveDeleteEvaluator(reader.spec());
- Evaluator strict = strictDeleteEvaluator(reader.spec());
boolean hasDeletedFiles = false;
for (ManifestEntry<F> entry : reader.liveEntries()) {
F file = entry.file();
boolean fileDelete = deletePaths.contains(file.path()) ||
dropPartitions.contains(file.specId(), file.partition()) ||
(isDelete && entry.sequenceNumber() > 0 && entry.sequenceNumber() < minSequenceNumber);
- if (fileDelete || inclusive.eval(file.partition())) {
+ if (fileDelete || evaluator.rowsMightMatch(file)) {
ValidationException.check(
- fileDelete || strict.eval(file.partition()) || metricsEvaluator.eval(file),
+ fileDelete || evaluator.rowsMustMatch(file),
"Cannot delete file where some, but not all, rows match filter %s: %s",
this.deleteExpression, file.path());
@@ -366,12 +365,10 @@ abstract class ManifestFilterManager<F extends ContentFile<F>> {
return hasDeletedFiles;
}
- @SuppressWarnings("CollectionUndefinedEquality")
- private ManifestFile filterManifestWithDeletedFiles(
- StrictMetricsEvaluator metricsEvaluator, ManifestFile manifest, ManifestReader<F> reader) {
+ @SuppressWarnings({"CollectionUndefinedEquality", "checkstyle:CyclomaticComplexity"})
+ private ManifestFile filterManifestWithDeletedFiles(PartitionAndMetricsEvaluator evaluator,
+ ManifestFile manifest, ManifestReader<F> reader) {
boolean isDelete = reader.isDeleteManifestReader();
- Evaluator inclusive = inclusiveDeleteEvaluator(reader.spec());
- Evaluator strict = strictDeleteEvaluator(reader.spec());
// when this point is reached, there is at least one file that will be deleted in the
// manifest. produce a copy of the manifest with all deleted files removed.
List<F> deletedFiles = Lists.newArrayList();
@@ -386,9 +383,9 @@ abstract class ManifestFilterManager<F extends ContentFile<F>> {
dropPartitions.contains(file.specId(), file.partition()) ||
(isDelete && entry.sequenceNumber() > 0 && entry.sequenceNumber() < minSequenceNumber);
if (entry.status() != ManifestEntry.Status.DELETED) {
- if (fileDelete || inclusive.eval(file.partition())) {
+ if (fileDelete || evaluator.rowsMightMatch(file)) {
ValidationException.check(
- fileDelete || strict.eval(file.partition()) || metricsEvaluator.eval(file),
+ fileDelete || evaluator.rowsMustMatch(file),
"Cannot delete file where some, but not all, rows match filter %s: %s",
this.deleteExpression, file.path());
@@ -429,13 +426,46 @@ abstract class ManifestFilterManager<F extends ContentFile<F>> {
}
}
- private Evaluator strictDeleteEvaluator(PartitionSpec spec) {
- Expression strictExpr = Projections.strict(spec).project(deleteExpression);
- return new Evaluator(spec.partitionType(), strictExpr);
- }
+ // an evaluator that checks whether rows in a file may/must match a given expression
+ // this class first partially evaluates the provided expression using the partition tuple
+ // and then checks the remaining part of the expression using metrics evaluators
+ private class PartitionAndMetricsEvaluator {
+ private final Schema tableSchema;
+ private final ResidualEvaluator residualEvaluator;
+ private final StructLikeMap<Pair<InclusiveMetricsEvaluator, StrictMetricsEvaluator>> metricsEvaluators;
+
+ // TODO: support case sensitive flags
+ PartitionAndMetricsEvaluator(Schema tableSchema, PartitionSpec spec, Expression expr) {
+ this.tableSchema = tableSchema;
+ this.residualEvaluator = ResidualEvaluator.of(spec, expr, true);
+ this.metricsEvaluators = StructLikeMap.create(spec.partitionType());
+ }
- private Evaluator inclusiveDeleteEvaluator(PartitionSpec spec) {
- Expression inclusiveExpr = Projections.inclusive(spec).project(deleteExpression);
- return new Evaluator(spec.partitionType(), inclusiveExpr);
+ boolean rowsMightMatch(F file) {
+ Pair<InclusiveMetricsEvaluator, StrictMetricsEvaluator> evaluators = metricsEvaluators(file);
+ InclusiveMetricsEvaluator inclusiveMetricsEvaluator = evaluators.first();
+ return inclusiveMetricsEvaluator.eval(file);
+ }
+
+ boolean rowsMustMatch(F file) {
+ Pair<InclusiveMetricsEvaluator, StrictMetricsEvaluator> evaluators = metricsEvaluators(file);
+ StrictMetricsEvaluator strictMetricsEvaluator = evaluators.second();
+ return strictMetricsEvaluator.eval(file);
+ }
+
+ private Pair<InclusiveMetricsEvaluator, StrictMetricsEvaluator> metricsEvaluators(F file) {
+ // ResidualEvaluator removes predicates in the expression using strict/inclusive projections
+ // if strict projection returns true -> the pred would return true -> replace the pred with true
+ // if inclusive projection returns false -> the pred would return false -> replace the pred with false
+ // otherwise, keep the original predicate and proceed to other predicates in the expression
+ // in other words, ResidualEvaluator returns a part of the expression that needs to be evaluated
+ // for rows in the given partition using metrics
+ return metricsEvaluators.computeIfAbsent(file.partition(), partition -> {
+ Expression residual = residualEvaluator.residualFor(partition);
+ InclusiveMetricsEvaluator inclusive = new InclusiveMetricsEvaluator(tableSchema, residual);
+ StrictMetricsEvaluator strict = new StrictMetricsEvaluator(tableSchema, residual);
+ return Pair.of(inclusive, strict);
+ });
+ }
}
}
diff --git a/core/src/test/java/org/apache/iceberg/TestDeleteFiles.java b/core/src/test/java/org/apache/iceberg/TestDeleteFiles.java
index a862ddc..c1b034b 100644
--- a/core/src/test/java/org/apache/iceberg/TestDeleteFiles.java
+++ b/core/src/test/java/org/apache/iceberg/TestDeleteFiles.java
@@ -22,15 +22,47 @@ package org.apache.iceberg;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import org.apache.iceberg.ManifestEntry.Status;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.junit.Assert;
+import org.junit.Assume;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@RunWith(Parameterized.class)
public class TestDeleteFiles extends TableTestBase {
+
+ private static final DataFile DATA_FILE_BUCKET_0_IDS_0_2 = DataFiles.builder(SPEC)
+ .withPath("/path/to/data-1.parquet")
+ .withFileSizeInBytes(10)
+ .withPartitionPath("data_bucket=0")
+ .withMetrics(new Metrics(5L,
+ null, // no column sizes
+ ImmutableMap.of(1, 5L, 2, 5L), // value count
+ ImmutableMap.of(1, 0L, 2, 0L), // null count
+ null, // no nan value counts
+ ImmutableMap.of(1, longToBuffer(0L)), // lower bounds
+ ImmutableMap.of(1, longToBuffer(2L)) // upper bounds
+ ))
+ .build();
+
+ private static final DataFile DATA_FILE_BUCKET_0_IDS_8_10 = DataFiles.builder(SPEC)
+ .withPath("/path/to/data-2.parquet")
+ .withFileSizeInBytes(10)
+ .withPartitionPath("data_bucket=0")
+ .withMetrics(new Metrics(5L,
+ null, // no column sizes
+ ImmutableMap.of(1, 5L, 2, 5L), // value count
+ ImmutableMap.of(1, 0L, 2, 0L), // null count
+ null, // no nan value counts
+ ImmutableMap.of(1, longToBuffer(8L)), // lower bounds
+ ImmutableMap.of(1, longToBuffer(10L)) // upper bounds
+ ))
+ .build();
+
@Parameterized.Parameters(name = "formatVersion = {0}")
public static Object[] parameters() {
return new Object[] { 1, 2 };
@@ -148,7 +180,95 @@ public class TestDeleteFiles extends TableTestBase {
statuses(Status.DELETED));
}
- private ByteBuffer longToBuffer(long value) {
+ @Test
+ public void testDeleteSomeFilesByRowFilterWithoutPartitionPredicates() {
+ // add both data files
+ table.newFastAppend()
+ .appendFile(DATA_FILE_BUCKET_0_IDS_0_2)
+ .appendFile(DATA_FILE_BUCKET_0_IDS_8_10)
+ .commit();
+
+ Snapshot initialSnapshot = table.currentSnapshot();
+ Assert.assertEquals("Should have 1 manifest", 1, initialSnapshot.allManifests().size());
+ validateManifestEntries(initialSnapshot.allManifests().get(0),
+ ids(initialSnapshot.snapshotId(), initialSnapshot.snapshotId()),
+ files(DATA_FILE_BUCKET_0_IDS_0_2, DATA_FILE_BUCKET_0_IDS_8_10),
+ statuses(Status.ADDED, Status.ADDED));
+
+ // delete the second one using a metrics filter (no partition filter)
+ table.newDelete()
+ .deleteFromRowFilter(Expressions.greaterThan("id", 5))
+ .commit();
+
+ Snapshot deleteSnapshot = table.currentSnapshot();
+ Assert.assertEquals("Should have 1 manifest", 1, deleteSnapshot.allManifests().size());
+ validateManifestEntries(deleteSnapshot.allManifests().get(0),
+ ids(initialSnapshot.snapshotId(), deleteSnapshot.snapshotId()),
+ files(DATA_FILE_BUCKET_0_IDS_0_2, DATA_FILE_BUCKET_0_IDS_8_10),
+ statuses(Status.EXISTING, Status.DELETED));
+ }
+
+ @Test
+ public void testDeleteSomeFilesByRowFilterWithCombinedPredicates() {
+ // add both data files
+ table.newFastAppend()
+ .appendFile(DATA_FILE_BUCKET_0_IDS_0_2)
+ .appendFile(DATA_FILE_BUCKET_0_IDS_8_10)
+ .commit();
+
+ Snapshot initialSnapshot = table.currentSnapshot();
+ Assert.assertEquals("Should have 1 manifest", 1, initialSnapshot.allManifests().size());
+ validateManifestEntries(initialSnapshot.allManifests().get(0),
+ ids(initialSnapshot.snapshotId(), initialSnapshot.snapshotId()),
+ files(DATA_FILE_BUCKET_0_IDS_0_2, DATA_FILE_BUCKET_0_IDS_8_10),
+ statuses(Status.ADDED, Status.ADDED));
+
+ // delete the second one using a filter that relies on metrics and partition data
+ Expression partPredicate = Expressions.equal(Expressions.bucket("data", 16), 0);
+ Expression rowPredicate = Expressions.greaterThan("id", 5);
+ Expression predicate = Expressions.and(partPredicate, rowPredicate);
+ table.newDelete()
+ .deleteFromRowFilter(predicate)
+ .commit();
+
+ Snapshot deleteSnapshot = table.currentSnapshot();
+ Assert.assertEquals("Should have 1 manifest", 1, deleteSnapshot.allManifests().size());
+ validateManifestEntries(deleteSnapshot.allManifests().get(0),
+ ids(initialSnapshot.snapshotId(), deleteSnapshot.snapshotId()),
+ files(DATA_FILE_BUCKET_0_IDS_0_2, DATA_FILE_BUCKET_0_IDS_8_10),
+ statuses(Status.EXISTING, Status.DELETED));
+ }
+
+ @Test
+ public void testCannotDeleteFileWhereNotAllRowsMatchPartitionFilter() {
+ Assume.assumeTrue(formatVersion == 2);
+
+ table.updateSpec()
+ .removeField(Expressions.bucket("data", 16))
+ .addField(Expressions.truncate("data", 2))
+ .commit();
+
+ PartitionSpec spec = table.spec();
+
+ DataFile dataFile = DataFiles.builder(spec)
+ .withPath("/path/to/data-1.parquet")
+ .withRecordCount(10)
+ .withFileSizeInBytes(10)
+ .withPartitionPath("data_trunc_2=aa")
+ .build();
+
+ table.newFastAppend()
+ .appendFile(dataFile)
+ .commit();
+
+ AssertHelpers.assertThrows("Should reject as not all rows match filter",
+ ValidationException.class, "Cannot delete file where some, but not all, rows match filter",
+ () -> table.newDelete()
+ .deleteFromRowFilter(Expressions.equal("data", "aa"))
+ .commit());
+ }
+
+ private static ByteBuffer longToBuffer(long value) {
return ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN).putLong(0, value);
}
}
diff --git a/spark/v3.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestDeleteFrom.java b/spark/v3.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestDeleteFrom.java
index eb852fb..6016dd2 100644
--- a/spark/v3.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestDeleteFrom.java
+++ b/spark/v3.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestDeleteFrom.java
@@ -19,16 +19,20 @@
package org.apache.iceberg.spark.sql;
+import java.util.List;
import java.util.Map;
import org.apache.iceberg.AssertHelpers;
import org.apache.iceberg.Table;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
-import org.apache.iceberg.spark.Spark3VersionUtil;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.spark.SparkCatalogTestBase;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
import org.junit.After;
import org.junit.Assert;
-import org.junit.Assume;
import org.junit.Test;
public class TestDeleteFrom extends SparkCatalogTestBase {
@@ -42,67 +46,58 @@ public class TestDeleteFrom extends SparkCatalogTestBase {
}
@Test
- public void testDeleteFromUnpartitionedTable() {
- // This test fails in Spark 3.1. `canDeleteWhere` was added to `SupportsDelete` in Spark 3.1,
- // but logic to rewrite the query if `canDeleteWhere` returns false was left to be implemented
- // later.
- Assume.assumeTrue(Spark3VersionUtil.isSpark30());
- // set the shuffle partitions to 1 to force the write to use a single task and produce 1 file
- String originalParallelism = spark.conf().get("spark.sql.shuffle.partitions");
- spark.conf().set("spark.sql.shuffle.partitions", "1");
- try {
- sql("CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg", tableName);
- sql("INSERT INTO TABLE %s VALUES (1, 'a'), (2, 'b'), (3, 'c')", tableName);
-
- assertEquals("Should have expected rows",
- ImmutableList.of(row(1L, "a"), row(2L, "b"), row(3L, "c")),
- sql("SELECT * FROM %s ORDER BY id", tableName));
-
- AssertHelpers.assertThrows("Should not delete when not all rows of a file match the filter",
- IllegalArgumentException.class, "Failed to cleanly delete data files",
- () -> sql("DELETE FROM %s WHERE id < 2", tableName));
-
- sql("DELETE FROM %s WHERE id < 4", tableName);
-
- Assert.assertEquals("Should have no rows after successful delete",
- 0L, scalarSql("SELECT count(1) FROM %s", tableName));
-
- } finally {
- spark.conf().set("spark.sql.shuffle.partitions", originalParallelism);
- }
+ public void testDeleteFromUnpartitionedTable() throws NoSuchTableException {
+ sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName);
+
+ List<SimpleRecord> records = Lists.newArrayList(
+ new SimpleRecord(1, "a"),
+ new SimpleRecord(2, "b"),
+ new SimpleRecord(3, "c")
+ );
+ Dataset<Row> df = spark.createDataFrame(records, SimpleRecord.class);
+ df.coalesce(1).writeTo(tableName).append();
+
+ assertEquals("Should have expected rows",
+ ImmutableList.of(row(1L, "a"), row(2L, "b"), row(3L, "c")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+
+ AssertHelpers.assertThrows("Should not delete when not all rows of a file match the filter",
+ IllegalArgumentException.class, "Failed to cleanly delete data files",
+ () -> sql("DELETE FROM %s WHERE id < 2", tableName));
+
+ sql("DELETE FROM %s WHERE id < 4", tableName);
+
+ Assert.assertEquals("Should have no rows after successful delete",
+ 0L, scalarSql("SELECT count(1) FROM %s", tableName));
}
@Test
- public void testDeleteFromPartitionedTable() {
- // This test fails in Spark 3.1. `canDeleteWhere` was added to `SupportsDelete` in Spark 3.1,
- // but logic to rewrite the query if `canDeleteWhere` returns false was left to be implemented
- // later.
- Assume.assumeTrue(Spark3VersionUtil.isSpark30());
- // set the shuffle partitions to 1 to force the write to use a single task and produce 1 file per partition
- String originalParallelism = spark.conf().get("spark.sql.shuffle.partitions");
- spark.conf().set("spark.sql.shuffle.partitions", "1");
- try {
- sql("CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg " +
- "PARTITIONED BY (truncate(id, 2))", tableName);
- sql("INSERT INTO TABLE %s VALUES (1, 'a'), (2, 'b'), (3, 'c')", tableName);
-
- assertEquals("Should have 3 rows in 2 partitions",
- ImmutableList.of(row(1L, "a"), row(2L, "b"), row(3L, "c")),
- sql("SELECT * FROM %s ORDER BY id", tableName));
-
- AssertHelpers.assertThrows("Should not delete when not all rows of a file match the filter",
- IllegalArgumentException.class, "Failed to cleanly delete data files",
- () -> sql("DELETE FROM %s WHERE id > 2", tableName));
-
- sql("DELETE FROM %s WHERE id < 2", tableName);
-
- assertEquals("Should have two rows in the second partition",
- ImmutableList.of(row(2L, "b"), row(3L, "c")),
- sql("SELECT * FROM %s ORDER BY id", tableName));
-
- } finally {
- spark.conf().set("spark.sql.shuffle.partitions", originalParallelism);
- }
+ public void testDeleteFromPartitionedTable() throws NoSuchTableException {
+ sql("CREATE TABLE %s (id bigint, data string) " +
+ "USING iceberg " +
+ "PARTITIONED BY (truncate(id, 2))", tableName);
+
+ List<SimpleRecord> records = Lists.newArrayList(
+ new SimpleRecord(1, "a"),
+ new SimpleRecord(2, "b"),
+ new SimpleRecord(3, "c")
+ );
+ Dataset<Row> df = spark.createDataFrame(records, SimpleRecord.class);
+ df.coalesce(1).writeTo(tableName).append();
+
+ assertEquals("Should have 3 rows in 2 partitions",
+ ImmutableList.of(row(1L, "a"), row(2L, "b"), row(3L, "c")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+
+ AssertHelpers.assertThrows("Should not delete when not all rows of a file match the filter",
+ IllegalArgumentException.class, "Failed to cleanly delete data files",
+ () -> sql("DELETE FROM %s WHERE id > 2", tableName));
+
+ sql("DELETE FROM %s WHERE id < 2", tableName);
+
+ assertEquals("Should have two rows in the second partition",
+ ImmutableList.of(row(2L, "b"), row(3L, "c")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
}
@Test