Fix error in Iceberg predicate pushdown to Parquet files

The Paruqet reader does not support pushdown on fields of a Row type.
The checks in `IcebergPageSourceProvider#getParquetTupleDomain` used
to prevent this, but they stopped working when dereference pushdown
was implemented.

If a row field had the same name as a top level column this would
have resulted in a correctness issue.

Author: Alex Jo

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSourceProvider.java

@@ -1354,7 +1354,7 @@ private static TupleDomain<ColumnDescriptor> getParquetTupleDomain(Map<List<Stri
         effectivePredicate.getDomains().orElseThrow().forEach((columnHandle, domain) -> {
             String baseType = columnHandle.getType().getTypeSignature().getBase();
             // skip looking up predicates for complex types as Parquet only stores stats for primitives
-            if (!baseType.equals(StandardTypes.MAP) && !baseType.equals(StandardTypes.ARRAY) && !baseType.equals(StandardTypes.ROW)) {
+            if (columnHandle.isBaseColumn() && (!baseType.equals(StandardTypes.MAP) && !baseType.equals(StandardTypes.ARRAY) && !baseType.equals(StandardTypes.ROW))) {
                 ColumnDescriptor descriptor = descriptorsByPath.get(ImmutableList.of(columnHandle.getName()));
                 if (descriptor != null) {
                     predicate.put(descriptor, domain);

testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java

@@ -3827,6 +3827,18 @@ public void testUpdateRowType()
         }
     }
 
+    @Test
+    public void testPredicateOnRowTypeField()
+    {
+        skipTestUnless(hasBehavior(SUPPORTS_CREATE_TABLE) && hasBehavior(SUPPORTS_INSERT) && hasBehavior(SUPPORTS_ROW_TYPE));
+
+        try (TestTable table = new TestTable(getQueryRunner()::execute, "test_predicate_on_row_type_field", "(int_t INT, row_t row(varchar_t VARCHAR, int_t INT))")) {
+            assertUpdate("INSERT INTO " + table.getName() + " VALUES (2, row('first', 1)), (20, row('second', 10)), (200, row('third', 100))", 3);
+            assertQuery("SELECT int_t FROM " + table.getName() + " WHERE row_t.int_t = 1", "VALUES 2");
+            assertQuery("SELECT int_t FROM " + table.getName() + " WHERE row_t.int_t > 1", "VALUES 20, 200");
+        }
+    }
+
     @Test
     public void testUpdateAllValues()
     {