Move limit/topN pushdown tests to base connector test

plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/BaseJdbcConnectorTest.java

@@ -937,72 +937,14 @@ protected TestTable createTableWithDoubleAndRealColumns(String name, List<String
     }
 
     @Test
-    public void testLimitPushdown()
+    public void testLimitPushdownWithDistinctAndJoin()
     {
-        if (!hasBehavior(SUPPORTS_LIMIT_PUSHDOWN)) {
-            assertThat(query("SELECT name FROM nation LIMIT 30")).isNotFullyPushedDown(LimitNode.class); // Use high limit for result determinism
-            return;
-        }
-
-        assertThat(query("SELECT name FROM nation LIMIT 30")).isFullyPushedDown(); // Use high limit for result determinism
-        assertThat(query("SELECT name FROM nation LIMIT 3")).skipResultsCorrectnessCheckForPushdown().isFullyPushedDown();
-
-        // with filter over numeric column
-        assertThat(query("SELECT name FROM nation WHERE regionkey = 3 LIMIT 5")).isFullyPushedDown();
-
-        // with filter over varchar column
-        PlanMatchPattern filterOverTableScan = node(FilterNode.class, node(TableScanNode.class));
-        assertConditionallyPushedDown(
-                getSession(),
-                "SELECT name FROM nation WHERE name < 'EEE' LIMIT 5",
-                hasBehavior(SUPPORTS_PREDICATE_PUSHDOWN_WITH_VARCHAR_INEQUALITY),
-                filterOverTableScan);
-
-        // with aggregation
-        PlanMatchPattern aggregationOverTableScan = node(AggregationNode.class, anyTree(node(TableScanNode.class)));
-        assertConditionallyPushedDown(
-                getSession(),
-                "SELECT max(regionkey) FROM nation LIMIT 5", // global aggregation, LIMIT removed
-                hasBehavior(SUPPORTS_AGGREGATION_PUSHDOWN),
-                aggregationOverTableScan);
-        assertConditionallyPushedDown(
-                getSession(),
-                "SELECT regionkey, max(nationkey) FROM nation GROUP BY regionkey LIMIT 5",
-                hasBehavior(SUPPORTS_AGGREGATION_PUSHDOWN),
-                aggregationOverTableScan);
+        // covered by testLimitPushdown
+        skipTestUnless(hasBehavior(SUPPORTS_LIMIT_PUSHDOWN));
 
         // distinct limit can be pushed down even without aggregation pushdown
         assertThat(query("SELECT DISTINCT regionkey FROM nation LIMIT 5")).isFullyPushedDown();
 
-        // with aggregation and filter over numeric column
-        assertConditionallyPushedDown(
-                getSession(),
-                "SELECT regionkey, count(*) FROM nation WHERE nationkey < 5 GROUP BY regionkey LIMIT 3",
-                hasBehavior(SUPPORTS_AGGREGATION_PUSHDOWN),
-                aggregationOverTableScan);
-        // with aggregation and filter over varchar column
-        if (hasBehavior(SUPPORTS_PREDICATE_PUSHDOWN_WITH_VARCHAR_INEQUALITY)) {
-            assertConditionallyPushedDown(
-                    getSession(),
-                    "SELECT regionkey, count(*) FROM nation WHERE name < 'EGYPT' GROUP BY regionkey LIMIT 3",
-                    hasBehavior(SUPPORTS_AGGREGATION_PUSHDOWN),
-                    aggregationOverTableScan);
-        }
-
-        // with TopN over numeric column
-        PlanMatchPattern topnOverTableScan = project(node(TopNNode.class, anyTree(node(TableScanNode.class))));
-        assertConditionallyPushedDown(
-                getSession(),
-                "SELECT * FROM (SELECT regionkey FROM nation ORDER BY nationkey ASC LIMIT 10) LIMIT 5",
-                hasBehavior(SUPPORTS_TOPN_PUSHDOWN),
-                topnOverTableScan);
-        // with TopN over varchar column
-        assertConditionallyPushedDown(
-                getSession(),
-                "SELECT * FROM (SELECT regionkey FROM nation ORDER BY name ASC LIMIT 10) LIMIT 5",
-                hasBehavior(SUPPORTS_TOPN_PUSHDOWN_WITH_VARCHAR),
-                topnOverTableScan);
-
         // with join
         PlanMatchPattern joinOverTableScans = node(JoinNode.class,
                 anyTree(node(TableScanNode.class)),
@@ -1029,69 +971,10 @@ public void testTopNPushdownDisabled()
     }
 
     @Test
-    public void testTopNPushdown()
+    public void testTopNPushdownWithJoin()
     {
-        if (!hasBehavior(SUPPORTS_TOPN_PUSHDOWN)) {
-            assertThat(query("SELECT orderkey FROM orders ORDER BY orderkey LIMIT 10"))
-                    .ordered()
-                    .isNotFullyPushedDown(TopNNode.class);
-            return;
-        }
-
-        assertThat(query("SELECT orderkey FROM orders ORDER BY orderkey LIMIT 10"))
-                .ordered()
-                .isFullyPushedDown();
-
-        assertThat(query("SELECT orderkey FROM orders ORDER BY orderkey DESC LIMIT 10"))
-                .ordered()
-                .isFullyPushedDown();
-
-        // multiple sort columns with different orders
-        assertThat(query("SELECT * FROM orders ORDER BY shippriority DESC, totalprice ASC LIMIT 10"))
-                .ordered()
-                .isFullyPushedDown();
-
-        // TopN over aggregation column
-        if (hasBehavior(SUPPORTS_AGGREGATION_PUSHDOWN)) {
-            assertThat(query("SELECT sum(totalprice) AS total FROM orders GROUP BY custkey ORDER BY total DESC LIMIT 10"))
-                    .ordered()
-                    .isFullyPushedDown();
-        }
-
-        // TopN over TopN
-        assertThat(query("SELECT orderkey, totalprice FROM (SELECT orderkey, totalprice FROM orders ORDER BY 1, 2 LIMIT 10) ORDER BY 2, 1 LIMIT 5"))
-                .ordered()
-                .isFullyPushedDown();
-
-        assertThat(query("" +
-                "SELECT orderkey, totalprice " +
-                "FROM (SELECT orderkey, totalprice FROM (SELECT orderkey, totalprice FROM orders ORDER BY 1, 2 LIMIT 10) " +
-                "ORDER BY 2, 1 LIMIT 5) ORDER BY 1, 2 LIMIT 3"))
-                .ordered()
-                .isFullyPushedDown();
-
-        // TopN over limit - use high limit for deterministic result
-        assertThat(query("SELECT orderkey, totalprice FROM (SELECT orderkey, totalprice FROM orders LIMIT 15000) ORDER BY totalprice ASC LIMIT 5"))
-                .ordered()
-                .isFullyPushedDown();
-
-        // TopN over limit with filter
-        assertThat(query("" +
-                "SELECT orderkey, totalprice " +
-                "FROM (SELECT orderkey, totalprice FROM orders WHERE orderdate = DATE '1995-09-16' LIMIT 20) " +
-                "ORDER BY totalprice ASC LIMIT 5"))
-                .ordered()
-                .isFullyPushedDown();
-
-        // TopN over aggregation with filter
-        if (hasBehavior(SUPPORTS_AGGREGATION_PUSHDOWN)) {
-            assertThat(query("" +
-                    "SELECT * " +
-                    "FROM (SELECT SUM(totalprice) as sum, custkey AS total FROM orders GROUP BY custkey HAVING COUNT(*) > 3) " +
-                    "ORDER BY sum DESC LIMIT 10"))
-                    .ordered()
-                    .isFullyPushedDown();
-        }
+        // covered by testTopNPushdown
+        skipTestUnless(hasBehavior(SUPPORTS_TOPN_PUSHDOWN));
 
         // TopN over LEFT join (enforces SINGLE TopN cannot be pushed below OUTER side of join)
         // We expect PARTIAL TopN on the LEFT side of join to be pushed down.
@@ -1439,19 +1322,6 @@ public void testExplainAnalyzePhysicalReadWallTime()
                 "Physical input time: .*s");
     }
 
-    protected QueryAssert assertConditionallyPushedDown(
-            Session session,
-            @Language("SQL") String query,
-            boolean condition,
-            PlanMatchPattern otherwiseExpected)
-    {
-        QueryAssert queryAssert = assertThat(query(session, query));
-        if (condition) {
-            return queryAssert.isFullyPushedDown();
-        }
-        return queryAssert.isNotFullyPushedDown(otherwiseExpected);
-    }
-
     protected QueryAssert assertJoinConditionallyPushedDown(
             Session session,
             @Language("SQL") String query,

plugin/trino-bigquery/src/test/java/io/trino/plugin/bigquery/BaseBigQueryConnectorTest.java

@@ -103,6 +103,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                     SUPPORTS_CREATE_MATERIALIZED_VIEW,
                     SUPPORTS_CREATE_VIEW,
                     SUPPORTS_DEFAULT_COLUMN_VALUE,
+                    SUPPORTS_LIMIT_PUSHDOWN,
                     SUPPORTS_MAP_TYPE,
                     SUPPORTS_MERGE,
                     SUPPORTS_NEGATIVE_DATE,

plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraConnectorTest.java

@@ -95,6 +95,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                  SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT,
                  SUPPORTS_CREATE_VIEW,
                  SUPPORTS_DEFAULT_COLUMN_VALUE,
+                 SUPPORTS_LIMIT_PUSHDOWN,
                  SUPPORTS_MAP_TYPE,
                  SUPPORTS_MERGE,
                  SUPPORTS_NOT_NULL_CONSTRAINT,

plugin/trino-druid/src/test/java/io/trino/plugin/druid/TestDruidConnectorTest.java

@@ -18,12 +18,8 @@
 import io.trino.plugin.jdbc.BaseJdbcConnectorTest;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.SchemaTableName;
-import io.trino.sql.planner.assertions.PlanMatchPattern;
 import io.trino.sql.planner.plan.AggregationNode;
 import io.trino.sql.planner.plan.FilterNode;
-import io.trino.sql.planner.plan.JoinNode;
-import io.trino.sql.planner.plan.TableScanNode;
-import io.trino.sql.planner.plan.TopNNode;
 import io.trino.testing.MaterializedResult;
 import io.trino.testing.QueryRunner;
 import io.trino.testing.TestingConnectorBehavior;
@@ -39,8 +35,6 @@
 import static io.trino.plugin.druid.DruidQueryRunner.copyAndIngestTpchData;
 import static io.trino.plugin.druid.DruidTpchTables.SELECT_FROM_ORDERS;
 import static io.trino.spi.type.VarcharType.VARCHAR;
-import static io.trino.sql.planner.assertions.PlanMatchPattern.anyTree;
-import static io.trino.sql.planner.assertions.PlanMatchPattern.node;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.output;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.values;
 import static io.trino.testing.MaterializedResult.resultBuilder;
@@ -234,47 +228,6 @@ public void testFilteringForTablesAndColumns()
         assertThat(query("DESCRIBE " + datasourceB)).result().matches(expectedColumns);
     }
 
-    @Test
-    public void testLimitPushDown()
-    {
-        assertThat(query("SELECT name FROM nation LIMIT 30")).isFullyPushedDown(); // Use high limit for result determinism
-
-        // with filter over numeric column
-        assertThat(query("SELECT name FROM nation WHERE regionkey = 3 LIMIT 5")).isFullyPushedDown();
-
-        // with filter over varchar column
-        assertThat(query("SELECT name FROM nation WHERE name < 'EEE' LIMIT 5")).isFullyPushedDown();
-
-        // with aggregation
-        assertThat(query("SELECT max(regionkey) FROM nation LIMIT 5")).isNotFullyPushedDown(AggregationNode.class); // global aggregation, LIMIT removed TODO https://github.com/trinodb/trino/pull/4313
-        assertThat(query("SELECT regionkey, max(name) FROM nation GROUP BY regionkey LIMIT 5")).isNotFullyPushedDown(AggregationNode.class); // TODO https://github.com/trinodb/trino/pull/4313
-
-        // distinct limit can be pushed down even without aggregation pushdown
-        assertThat(query("SELECT DISTINCT regionkey FROM nation LIMIT 5")).isFullyPushedDown();
-
-        // with aggregation and filter over numeric column
-        assertThat(query("SELECT regionkey, count(*) FROM nation WHERE nationkey < 5 GROUP BY regionkey LIMIT 3")).isNotFullyPushedDown(AggregationNode.class); // TODO https://github.com/trinodb/trino/pull/4313
-        // with aggregation and filter over varchar column
-        assertThat(query("SELECT regionkey, count(*) FROM nation WHERE name < 'EGYPT' GROUP BY regionkey LIMIT 3")).isNotFullyPushedDown(AggregationNode.class); // TODO https://github.com/trinodb/trino/pull/4313
-
-        // with TopN over numeric column
-        assertThat(query("SELECT * FROM (SELECT regionkey FROM nation ORDER BY nationkey ASC LIMIT 10) LIMIT 5")).isNotFullyPushedDown(TopNNode.class);
-        // with TopN over varchar column
-        assertThat(query("SELECT * FROM (SELECT regionkey FROM nation ORDER BY name ASC LIMIT 10) LIMIT 5")).isNotFullyPushedDown(TopNNode.class);
-
-        // with join
-        PlanMatchPattern joinOverTableScans = node(JoinNode.class,
-                anyTree(node(TableScanNode.class)),
-                anyTree(node(TableScanNode.class)));
-        assertThat(query(
-                joinPushdownEnabled(getSession()),
-                "SELECT n.name, r.name " +
-                        "FROM nation n " +
-                        "LEFT JOIN region r USING (regionkey) " +
-                        "LIMIT 30"))
-                .isNotFullyPushedDown(joinOverTableScans);
-    }
-
     @Test
     @Override
     public void testInsertNegativeDate()

plugin/trino-elasticsearch/src/test/java/io/trino/plugin/elasticsearch/BaseElasticsearchConnectorTest.java

@@ -18,7 +18,6 @@
 import com.google.common.collect.ImmutableMap;
 import io.trino.Session;
 import io.trino.spi.type.VarcharType;
-import io.trino.sql.planner.plan.LimitNode;
 import io.trino.testing.AbstractTestQueries;
 import io.trino.testing.BaseConnectorTest;
 import io.trino.testing.MaterializedResult;
@@ -1711,13 +1710,6 @@ public void testFiltersCharset()
                 .matches("VALUES (VARCHAR 'Türkiye')");
     }
 
-    @Test
-    public void testLimitPushdown()
-            throws IOException
-    {
-        assertThat(query("SELECT name FROM nation LIMIT 30")).isNotFullyPushedDown(LimitNode.class); // Use high limit for result determinism
-    }
-
     @Test
     public void testDataTypesNested()
             throws IOException

plugin/trino-hive/src/test/java/io/trino/plugin/hive/BaseHiveConnectorTest.java

@@ -287,6 +287,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                  SUPPORTS_CREATE_MATERIALIZED_VIEW,
                  SUPPORTS_DEFAULT_COLUMN_VALUE,
                  SUPPORTS_DROP_FIELD,
+                 SUPPORTS_LIMIT_PUSHDOWN,
                  SUPPORTS_MERGE,
                  SUPPORTS_NOT_NULL_CONSTRAINT,
                  SUPPORTS_REFRESH_VIEW,

plugin/trino-hudi/src/test/java/io/trino/plugin/hudi/TestHudiConnectorTest.java

@@ -49,6 +49,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                  SUPPORTS_DELETE,
                  SUPPORTS_DEREFERENCE_PUSHDOWN,
                  SUPPORTS_INSERT,
+                 SUPPORTS_LIMIT_PUSHDOWN,
                  SUPPORTS_MERGE,
                  SUPPORTS_RENAME_COLUMN,
                  SUPPORTS_RENAME_TABLE,

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java

@@ -278,6 +278,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                  SUPPORTS_REPORTING_WRITTEN_BYTES -> true;
             case SUPPORTS_ADD_COLUMN_NOT_NULL_CONSTRAINT,
                  SUPPORTS_DEFAULT_COLUMN_VALUE,
+                 SUPPORTS_LIMIT_PUSHDOWN,
                  SUPPORTS_REFRESH_VIEW,
                  SUPPORTS_RENAME_MATERIALIZED_VIEW_ACROSS_SCHEMAS,
                  SUPPORTS_TOPN_PUSHDOWN -> false;

plugin/trino-kafka/src/test/java/io/trino/plugin/kafka/TestKafkaConnectorTest.java

@@ -170,6 +170,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                  SUPPORTS_CREATE_VIEW,
                  SUPPORTS_DELETE,
                  SUPPORTS_DEREFERENCE_PUSHDOWN,
+                 SUPPORTS_LIMIT_PUSHDOWN,
                  SUPPORTS_MERGE,
                  SUPPORTS_RENAME_COLUMN,
                  SUPPORTS_RENAME_TABLE,

plugin/trino-lakehouse/src/test/java/io/trino/plugin/lakehouse/TestLakehouseConnectorTest.java

@@ -124,6 +124,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                  SUPPORTS_REPORTING_WRITTEN_BYTES -> true;
             case SUPPORTS_ADD_COLUMN_NOT_NULL_CONSTRAINT,
                  SUPPORTS_DEFAULT_COLUMN_VALUE,
+                 SUPPORTS_LIMIT_PUSHDOWN,
                  SUPPORTS_REFRESH_VIEW,
                  SUPPORTS_RENAME_MATERIALIZED_VIEW_ACROSS_SCHEMAS,
                  SUPPORTS_TOPN_PUSHDOWN -> false;

plugin/trino-memory/src/test/java/io/trino/plugin/memory/TestMemoryConnectorTest.java

@@ -96,7 +96,6 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                  SUPPORTS_DELETE,
                  SUPPORTS_DEREFERENCE_PUSHDOWN,
                  SUPPORTS_DROP_COLUMN,
-                 SUPPORTS_LIMIT_PUSHDOWN,
                  SUPPORTS_MERGE,
                  SUPPORTS_PREDICATE_PUSHDOWN,
                  SUPPORTS_RENAME_FIELD,

plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/TestMongoConnectorTest.java

@@ -102,6 +102,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
         return switch (connectorBehavior) {
             case SUPPORTS_ADD_COLUMN_WITH_POSITION,
                  SUPPORTS_ADD_FIELD,
+                 SUPPORTS_AGGREGATION_PUSHDOWN,
                  SUPPORTS_CREATE_MATERIALIZED_VIEW,
                  SUPPORTS_CREATE_VIEW,
                  SUPPORTS_DEFAULT_COLUMN_VALUE,
@@ -111,6 +112,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                  SUPPORTS_RENAME_FIELD,
                  SUPPORTS_RENAME_SCHEMA,
                  SUPPORTS_SET_FIELD_TYPE,
+                 SUPPORTS_TOPN_PUSHDOWN,
                  SUPPORTS_TRUNCATE,
                  SUPPORTS_UPDATE -> false;
             default -> super.hasBehavior(connectorBehavior);
@@ -980,10 +982,8 @@ public void testNullPredicates()
     }
 
     @Test
-    public void testLimitPushdown()
+    void testLimitWithLowerAndUpperBound()
     {
-        assertThat(query("SELECT name FROM nation LIMIT 30")).isFullyPushedDown(); // Use high limit for result determinism
-
         // Make sure LIMIT 0 returns empty result because cursor.limit(0) means no limit in MongoDB
         assertThat(query("SELECT name FROM nation LIMIT 0")).returnsEmptyResult();
 

plugin/trino-opensearch/src/test/java/io/trino/plugin/opensearch/BaseOpenSearchConnectorTest.java

@@ -19,7 +19,6 @@
 import com.google.common.net.HostAndPort;
 import io.trino.Session;
 import io.trino.spi.type.VarcharType;
-import io.trino.sql.planner.plan.LimitNode;
 import io.trino.sql.planner.plan.ProjectNode;
 import io.trino.testing.AbstractTestQueries;
 import io.trino.testing.BaseConnectorTest;
@@ -1679,13 +1678,6 @@ public void testFiltersCharset()
                 .matches("VALUES (VARCHAR 'Türkiye')");
     }
 
-    @Test
-    public void testLimitPushdown()
-            throws IOException
-    {
-        assertThat(query("SELECT name FROM nation LIMIT 30")).isNotFullyPushedDown(LimitNode.class); // Use high limit for result determinism
-    }
-
     @Test
     public void testDataTypesNested()
             throws IOException

plugin/trino-pinot/src/test/java/io/trino/plugin/pinot/TestPinotConnectorTest.java

@@ -59,6 +59,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                  SUPPORTS_CREATE_VIEW,
                  SUPPORTS_DELETE,
                  SUPPORTS_INSERT,
+                 SUPPORTS_LIMIT_PUSHDOWN,
                  SUPPORTS_MAP_TYPE,
                  SUPPORTS_MERGE,
                  SUPPORTS_RENAME_COLUMN,

plugin/trino-redis/src/test/java/io/trino/plugin/redis/BaseRedisConnectorTest.java

@@ -37,6 +37,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                  SUPPORTS_CREATE_VIEW,
                  SUPPORTS_DELETE,
                  SUPPORTS_INSERT,
+                 SUPPORTS_LIMIT_PUSHDOWN,
                  SUPPORTS_MAP_TYPE,
                  SUPPORTS_MERGE,
                  SUPPORTS_RENAME_COLUMN,

plugin/trino-thrift/src/test/java/io/trino/plugin/thrift/integration/TestThriftConnectorTest.java

@@ -39,6 +39,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                  SUPPORTS_CREATE_VIEW,
                  SUPPORTS_DELETE,
                  SUPPORTS_INSERT,
+                 SUPPORTS_LIMIT_PUSHDOWN,
                  SUPPORTS_MERGE,
                  SUPPORTS_NOT_NULL_CONSTRAINT,
                  SUPPORTS_RENAME_COLUMN,