Core: Fix query failure when using projection on top of partitions me…

…tadata table (apache#4720)

Co-authored-by: Prashant Singh <[email protected]>

core/src/main/java/org/apache/iceberg/PartitionsTable.java

@@ -152,7 +152,7 @@ static CloseableIterable<FileScanTask> planFiles(StaticTableScan scan) {
 
     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);
     });
 

core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java

@@ -377,6 +377,26 @@ public void testPartitionsTableScanNoFilter() {
     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()

spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java

@@ -171,6 +171,13 @@ public void testPartitionedTable() throws Exception {
     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());