More coverage to TestHivePlans

* Test to ensure filter on build side table is derived from table properties
* Test to ensure query fails if it scans too many partitions

plugin/trino-hive/src/test/java/io/trino/plugin/hive/optimizer/TestHivePlans.java

@@ -31,6 +31,7 @@
 import io.trino.plugin.hive.metastore.MetastoreConfig;
 import io.trino.plugin.hive.metastore.file.FileHiveMetastore;
 import io.trino.plugin.hive.metastore.file.FileHiveMetastoreConfig;
+import io.trino.spi.TrinoException;
 import io.trino.spi.security.PrincipalType;
 import io.trino.sql.planner.assertions.BasePlanTest;
 import io.trino.testing.LocalQueryRunner;
@@ -61,6 +62,7 @@
 import static io.trino.sql.planner.plan.ExchangeNode.Type.REPARTITION;
 import static io.trino.sql.planner.plan.JoinNode.Type.INNER;
 import static io.trino.testing.TestingSession.testSessionBuilder;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 public class TestHivePlans
         extends BasePlanTest
@@ -104,7 +106,7 @@ protected LocalQueryRunner createLocalQueryRunner()
     protected LocalQueryRunner createQueryRunner(Session session, HiveMetastore metastore)
     {
         LocalQueryRunner queryRunner = LocalQueryRunner.create(session);
-        queryRunner.createCatalog(HIVE_CATALOG_NAME, new TestingHiveConnectorFactory(metastore), Map.of());
+        queryRunner.createCatalog(HIVE_CATALOG_NAME, new TestingHiveConnectorFactory(metastore), Map.of("hive.max-partitions-per-scan", "5"));
         return queryRunner;
     }
 
@@ -122,6 +124,9 @@ public void setUp()
         // partitioned on varchar
         queryRunner.execute("CREATE TABLE table_str_partitioned WITH (partitioned_by = ARRAY['str_part']) AS SELECT int_col, str_part FROM (" + values + ") t(str_part, int_col)");
 
+        // with too many partitions
+        queryRunner.execute("CREATE TABLE table_int_with_too_many_partitions WITH (partitioned_by = ARRAY['int_part']) AS SELECT str_col, int_part FROM (" + values + ", ('six', 6)) t(str_col, int_part)");
+
         // unpartitioned
         queryRunner.execute("CREATE TABLE table_unpartitioned AS SELECT str_col, int_col FROM (" + values + ") t(str_col, int_col)");
     }
@@ -267,6 +272,36 @@ public void testSubsumePartitionFilterNotConvertibleToTupleDomain()
                                                                         tableScan("table_unpartitioned", Map.of("R_STR_COL", "str_col", "R_INT_COL", "int_col"))))))))));
     }
 
+    @Test
+    public void testFilterDerivedFromTableProperties()
+    {
+        // Test that the filter is on build side table is derived from table properties
+        assertDistributedPlan(
+                "SELECT l.str_col, r.str_col FROM table_int_partitioned l JOIN table_unpartitioned r ON l.int_part = r.int_col",
+                noJoinReordering(),
+                output(
+                        exchange(REMOTE, GATHER,
+                                join(INNER, List.of(equiJoinClause("L_INT_PART", "R_INT_COL")),
+                                        exchange(REMOTE, REPARTITION,
+                                                project(
+                                                        filter("true", //dynamic filter
+                                                                tableScan("table_int_partitioned", Map.of("L_INT_PART", "int_part", "L_STR_COL", "str_col"))))),
+                                        exchange(LOCAL,
+                                                exchange(REMOTE, REPARTITION,
+                                                        project(
+                                                                filter("R_INT_COL IN (1, 2, 3, 4, 5)",
+                                                                        tableScan("table_unpartitioned", Map.of("R_STR_COL", "str_col", "R_INT_COL", "int_col"))))))))));
+    }
+
+    @Test
+    public void testQueryScanningForTooManyPartitions()
+    {
+        assertThatThrownBy(() -> plan("SELECT l.str_col, r.str_col FROM table_int_with_too_many_partitions l JOIN table_unpartitioned r ON l.int_part = r.int_col"))
+                .getCause()
+                .isInstanceOf(TrinoException.class)
+                .hasMessage("Query over table 'test_schema.table_int_with_too_many_partitions' can potentially read more than 5 partitions");
+    }
+
     // Disable join ordering so that expected plans are well defined.
     private Session noJoinReordering()
     {