[iceberg] branch master updated: Core: Fix query failure when using projection on top of partitions metadata table (#4720)

[ru...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

russellspitzer 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 0390e1703 Core: Fix query failure when using projection on top of partitions metadata table (#4720)
0390e1703 is described below

commit 0390e170318d7ad624751d985eaca78205848d62
Author: Prashant Singh <35...@users.noreply.github.com>
AuthorDate: Mon May 9 22:42:00 2022 +0530

    Core: Fix query failure when using projection on top of partitions metadata table (#4720)
    
    Co-authored-by: Prashant Singh <ps...@amazon.com>
---
 .../java/org/apache/iceberg/PartitionsTable.java     |  2 +-
 .../org/apache/iceberg/TestMetadataTableScans.java   | 20 ++++++++++++++++++++
 .../iceberg/spark/extensions/TestMetadataTables.java |  7 +++++++
 3 files changed, 28 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 32ae8c9c9..38ba3496d 100644
--- a/core/src/main/java/org/apache/iceberg/PartitionsTable.java
+++ b/core/src/main/java/org/apache/iceberg/PartitionsTable.java
@@ -152,7 +152,7 @@ public class PartitionsTable extends BaseMetadataTable {
 
     LoadingCache<Integer, ManifestEvaluator> evalCache = Caffeine.newBuilder().build(specId -> {
       PartitionSpec spec = table.specs().get(specId);
-      PartitionSpec transformedSpec = transformSpec(scan.schema(), spec);
+      PartitionSpec transformedSpec = transformSpec(scan.tableSchema(), spec);
       return ManifestEvaluator.forRowFilter(scan.filter(), transformedSpec, caseSensitive);
     });
 
diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
index 0d2316046..ee00bb8d8 100644
--- a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
+++ b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
@@ -377,6 +377,26 @@ public class TestMetadataTableScans extends TableTestBase {
     validateIncludesPartitionScan(tasksNoFilter, 3);
   }
 
+  @Test
+  public void testPartitionsTableScanWithProjection() {
+    preparePartitionedTable();
+
+    Table partitionsTable = new PartitionsTable(table.ops(), table);
+    Types.StructType expected = new Schema(
+        required(3, "file_count", Types.IntegerType.get())
+       ).asStruct();
+
+    TableScan scanWithProjection = partitionsTable.newScan().select("file_count");
+    Assert.assertEquals(expected, scanWithProjection.schema().asStruct());
+    CloseableIterable<FileScanTask> tasksWithProjection =
+        PartitionsTable.planFiles((StaticTableScan) scanWithProjection);
+    Assert.assertEquals(4, Iterators.size(tasksWithProjection.iterator()));
+    validateIncludesPartitionScan(tasksWithProjection, 0);
+    validateIncludesPartitionScan(tasksWithProjection, 1);
+    validateIncludesPartitionScan(tasksWithProjection, 2);
+    validateIncludesPartitionScan(tasksWithProjection, 3);
+  }
+
   @Test
   public void testPartitionsTableScanNoStats() {
     table.newFastAppend()
diff --git a/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java b/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java
index ea4cee2a6..60c57e12a 100644
--- a/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java
+++ b/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java
@@ -171,6 +171,13 @@ public class TestMetadataTables extends SparkExtensionsTestBase {
     Assert.assertEquals("Metadata table should return one data file", 1, actualDataFiles.size());
     TestHelpers.assertEqualsSafe(filesTableSchema.asStruct(), expectedDataFiles.get(0), actualDataFiles.get(0));
 
+    List<Row> actualPartitionsWithProjection =
+        spark.sql("SELECT file_count FROM " + tableName + ".partitions ").collectAsList();
+    Assert.assertEquals("Metadata table should return two partitions record", 2, actualPartitionsWithProjection.size());
+    for (int i = 0; i < 2; ++i) {
+      Assert.assertEquals(1, actualPartitionsWithProjection.get(i).get(0));
+    }
+
     // Check files table
     List<Record> expectedFiles = Stream.concat(expectedDataFiles.stream(), expectedDeleteFiles.stream())
         .collect(Collectors.toList());