You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by et...@apache.org on 2023/12/19 07:54:27 UTC
(iceberg) branch 1.4.x updated: Core: Scan only live entries in partitions table (#8969) (#9197)
This is an automated email from the ASF dual-hosted git repository.
etudenhoefner pushed a commit to branch 1.4.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/1.4.x by this push:
new f6fcd45bf8 Core: Scan only live entries in partitions table (#8969) (#9197)
f6fcd45bf8 is described below
commit f6fcd45bf8938121a7ec32c85c84442a20d7f23f
Author: Fokko Driesprong <fo...@apache.org>
AuthorDate: Tue Dec 19 08:54:20 2023 +0100
Core: Scan only live entries in partitions table (#8969) (#9197)
Co-authored-by: Hongyue/Steve Zhang <st...@gmail.com>
Co-authored-by: Steve Zhang <ho...@apple.com>
---
.../java/org/apache/iceberg/PartitionsTable.java | 2 +-
.../iceberg/spark/extensions/TestDelete.java | 32 ++++++++++++++++++++++
.../iceberg/spark/extensions/TestDelete.java | 32 ++++++++++++++++++++++
.../iceberg/spark/extensions/TestDelete.java | 32 ++++++++++++++++++++++
.../iceberg/spark/extensions/TestDelete.java | 32 ++++++++++++++++++++++
5 files changed, 129 insertions(+), 1 deletion(-)
diff --git a/core/src/main/java/org/apache/iceberg/PartitionsTable.java b/core/src/main/java/org/apache/iceberg/PartitionsTable.java
index d93200c7cf..2537f5172b 100644
--- a/core/src/main/java/org/apache/iceberg/PartitionsTable.java
+++ b/core/src/main/java/org/apache/iceberg/PartitionsTable.java
@@ -202,7 +202,7 @@ public class PartitionsTable extends BaseMetadataTable {
ManifestFiles.open(manifest, table.io(), table.specs())
.caseSensitive(scan.isCaseSensitive())
.select(scanColumns(manifest.content())) // don't select stats columns
- .entries(),
+ .liveEntries(),
t ->
(ManifestEntry<? extends ContentFile<?>>)
// defensive copy of manifest entry without stats columns
diff --git a/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java b/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
index 59112912a3..46d3823529 100644
--- a/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
+++ b/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
@@ -158,6 +158,38 @@ public abstract class TestDelete extends SparkRowLevelOperationsTestBase {
sql("SELECT * FROM %s ORDER BY id", tableName));
}
+ @Test
+ public void testDeleteWithPartitionedTable() throws Exception {
+ createAndInitPartitionedTable();
+
+ sql("INSERT INTO TABLE %s VALUES (1, 'hr'), (3, 'hr')", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1, 'hardware'), (2, 'hardware')", tableName);
+
+ // row level delete
+ sql("DELETE FROM %s WHERE id = 1", tableName);
+
+ assertEquals(
+ "Should have expected rows",
+ ImmutableList.of(row(2, "hardware"), row(3, "hr")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ List<Row> rowLevelDeletePartitions =
+ spark.sql("SELECT * FROM " + tableName + ".partitions ").collectAsList();
+ Assert.assertEquals(
+ "row level delete does not reduce number of partition", 2, rowLevelDeletePartitions.size());
+
+ // partition aligned delete
+ sql("DELETE FROM %s WHERE dep = 'hr'", tableName);
+
+ assertEquals(
+ "Should have expected rows",
+ ImmutableList.of(row(2, "hardware")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ List<Row> actualPartitions =
+ spark.sql("SELECT * FROM " + tableName + ".partitions ").collectAsList();
+ Assert.assertEquals(
+ "partition aligned delete results in 1 partition", 1, actualPartitions.size());
+ }
+
@Test
public void testDeleteWithFalseCondition() {
createAndInitUnpartitionedTable();
diff --git a/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java b/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
index 4b60d580ae..c5ff25b1a0 100644
--- a/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
+++ b/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
@@ -167,6 +167,38 @@ public abstract class TestDelete extends SparkRowLevelOperationsTestBase {
sql("SELECT * FROM %s ORDER BY id", selectTarget()));
}
+ @Test
+ public void testDeleteWithPartitionedTable() throws Exception {
+ createAndInitPartitionedTable();
+
+ sql("INSERT INTO TABLE %s VALUES (1, 'hr'), (3, 'hr')", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1, 'hardware'), (2, 'hardware')", tableName);
+
+ // row level delete
+ sql("DELETE FROM %s WHERE id = 1", tableName);
+
+ assertEquals(
+ "Should have expected rows",
+ ImmutableList.of(row(2, "hardware"), row(3, "hr")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ List<Row> rowLevelDeletePartitions =
+ spark.sql("SELECT * FROM " + tableName + ".partitions ").collectAsList();
+ Assert.assertEquals(
+ "row level delete does not reduce number of partition", 2, rowLevelDeletePartitions.size());
+
+ // partition aligned delete
+ sql("DELETE FROM %s WHERE dep = 'hr'", tableName);
+
+ assertEquals(
+ "Should have expected rows",
+ ImmutableList.of(row(2, "hardware")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ List<Row> actualPartitions =
+ spark.sql("SELECT * FROM " + tableName + ".partitions ").collectAsList();
+ Assert.assertEquals(
+ "partition aligned delete results in 1 partition", 1, actualPartitions.size());
+ }
+
@Test
public void testDeleteWithFalseCondition() {
createAndInitUnpartitionedTable();
diff --git a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
index 9a59fbc861..87a98cbae5 100644
--- a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
+++ b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
@@ -335,6 +335,38 @@ public abstract class TestDelete extends SparkRowLevelOperationsTestBase {
sql("SELECT * FROM %s ORDER BY id", selectTarget()));
}
+ @Test
+ public void testDeleteWithPartitionedTable() throws Exception {
+ createAndInitPartitionedTable();
+
+ append(tableName, new Employee(1, "hr"), new Employee(3, "hr"));
+ append(tableName, new Employee(1, "hardware"), new Employee(2, "hardware"));
+
+ // row level delete
+ sql("DELETE FROM %s WHERE id = 1", tableName);
+
+ assertEquals(
+ "Should have expected rows",
+ ImmutableList.of(row(2, "hardware"), row(3, "hr")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ List<Row> rowLevelDeletePartitions =
+ spark.sql("SELECT * FROM " + tableName + ".partitions ").collectAsList();
+ Assert.assertEquals(
+ "row level delete does not reduce number of partition", 2, rowLevelDeletePartitions.size());
+
+ // partition aligned delete
+ sql("DELETE FROM %s WHERE dep = 'hr'", tableName);
+
+ assertEquals(
+ "Should have expected rows",
+ ImmutableList.of(row(2, "hardware")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ List<Row> actualPartitions =
+ spark.sql("SELECT * FROM " + tableName + ".partitions ").collectAsList();
+ Assert.assertEquals(
+ "partition aligned delete results in 1 partition", 1, actualPartitions.size());
+ }
+
@Test
public void testDeleteWithFalseCondition() {
createAndInitUnpartitionedTable();
diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
index a5f62d73e9..bdc4be08b4 100644
--- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
+++ b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
@@ -349,6 +349,38 @@ public abstract class TestDelete extends SparkRowLevelOperationsTestBase {
sql("SELECT * FROM %s ORDER BY id", selectTarget()));
}
+ @Test
+ public void testDeleteWithPartitionedTable() throws Exception {
+ createAndInitPartitionedTable();
+
+ append(tableName, new Employee(1, "hr"), new Employee(3, "hr"));
+ append(tableName, new Employee(1, "hardware"), new Employee(2, "hardware"));
+
+ // row level delete
+ sql("DELETE FROM %s WHERE id = 1", tableName);
+
+ assertEquals(
+ "Should have expected rows",
+ ImmutableList.of(row(2, "hardware"), row(3, "hr")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ List<Row> rowLevelDeletePartitions =
+ spark.sql("SELECT * FROM " + tableName + ".partitions ").collectAsList();
+ Assert.assertEquals(
+ "row level delete does not reduce number of partition", 2, rowLevelDeletePartitions.size());
+
+ // partition aligned delete
+ sql("DELETE FROM %s WHERE dep = 'hr'", tableName);
+
+ assertEquals(
+ "Should have expected rows",
+ ImmutableList.of(row(2, "hardware")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ List<Row> actualPartitions =
+ spark.sql("SELECT * FROM " + tableName + ".partitions ").collectAsList();
+ Assert.assertEquals(
+ "partition aligned delete results in 1 partition", 1, actualPartitions.size());
+ }
+
@Test
public void testDeleteWithFalseCondition() {
createAndInitUnpartitionedTable();