branch-4.0: [fix](fe) Reject TopN in correlated scalar subquery #64251

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/SubExprAnalyzer.java

@@ -138,8 +138,16 @@ public Expression visitScalarSubquery(ScalarSubquery scalar, T context) {
         boolean limitOneIsEliminated = false;
         if (isCorrelated) {
             if (analyzedSubqueryPlan instanceof LogicalLimit) {
-                LogicalLimit limit = (LogicalLimit) analyzedSubqueryPlan;
-                if (limit.getOffset() == 0 && limit.getLimit() == 1) {
+                Plan child = ((LogicalLimit<?>) analyzedSubqueryPlan).child();
+                LogicalLimit<?> limit = (LogicalLimit<?>) analyzedSubqueryPlan;
+                // after analysis, if project not contains sort key, FILL_UP_SORT_PROJECT will add a project upper sort
+                // so we must find sort under project here.
+                while (child instanceof LogicalProject) {
+                    child = ((LogicalProject<?>) child).child();
+                }
+                // order by c1 limit 1 is not acceptable
+                if (!(child instanceof LogicalSort)
+                        && limit.getOffset() == 0 && limit.getLimit() == 1) {
                     // skip useless limit node
                     analyzedResult = new AnalyzedResult((LogicalPlan) analyzedSubqueryPlan.child(0),
                             analyzedResult.correlatedSlots);

fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/AnalyzeSubQueryTest.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.nereids.NereidsPlanner;
 import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.glue.translator.PhysicalPlanTranslator;
 import org.apache.doris.nereids.glue.translator.PlanTranslatorContext;
 import org.apache.doris.nereids.parser.NereidsParser;
@@ -246,6 +247,26 @@ public void testScalarSubquerySlotNullable() {
         }
     }
 
+    @Test
+    public void testCorrelatedScalarSubqueryWithTopNProject() {
+        String sql = "select T1.id from T1 where T1.score > "
+                + "(select T2.score + 1 from T2 where T2.id = T1.id order by T2.score limit 1)";
+
+        AnalysisException exception = Assertions.assertThrows(AnalysisException.class,
+                () -> PlanChecker.from(connectContext).analyze(sql));
+        Assertions.assertTrue(exception.getMessage().contains("limit is not supported in correlated subquery"));
+    }
+
+    @Test
+    public void testCorrelatedScalarSubqueryWithTopN() {
+        String sql = "select T1.id from T1 where T1.score > "
+                + "(select T2.score from T2 where T2.id = T1.id order by T2.score limit 1)";
+
+        AnalysisException exception = Assertions.assertThrows(AnalysisException.class,
+                () -> PlanChecker.from(connectContext).analyze(sql));
+        Assertions.assertTrue(exception.getMessage().contains("limit is not supported in correlated subquery"));
+    }
+
     private void checkScalarSubquerySlotNullable(String sql, boolean outputNullable) {
         Plan root = PlanChecker.from(connectContext)
                 .analyze(sql)

regression-test/suites/nereids_p0/subquery/correlated_scalar_subquery.groovy

@@ -139,6 +139,20 @@ suite("correlated_scalar_subquery") {
         exception "limit is not supported in correlated subquery"
     }
 
+    test {
+        sql """
+              select c1 from correlated_scalar_t1 where correlated_scalar_t1.c2 > (select c2 from correlated_scalar_t2 where correlated_scalar_t1.c1 = correlated_scalar_t2.c1 order by c2 limit 1);
+        """
+        exception "limit is not supported in correlated subquery"
+    }
+
+    test {
+        sql """
+              select c1 from correlated_scalar_t1 where correlated_scalar_t1.c2 > (select c2 + 1 from correlated_scalar_t2 where correlated_scalar_t1.c1 = correlated_scalar_t2.c1 order by c2 limit 1);
+        """
+        exception "limit is not supported in correlated subquery"
+    }
+
     test {
         sql """
               select c1 from correlated_scalar_t1 where correlated_scalar_t1.c2 > (select e1 from (select k1 from (select 1 k1 ) as t where correlated_scalar_t1.c1 = k1 ) tt lateral view explode_numbers(5) tmp1 as e1 order by e1);
@@ -248,4 +262,4 @@ suite("correlated_scalar_subquery") {
     qt_select_agg_project2 """select c2 from correlated_scalar_t1 where correlated_scalar_t1.c2 = (select if(sum(c1) is null, 2, 100) from correlated_scalar_t2 where correlated_scalar_t1.c1 = correlated_scalar_t2.c1) order by c2;"""
     qt_select_2_aggs """select c2 from correlated_scalar_t1 where correlated_scalar_t1.c2 > (select count(c1) - min(c1) from correlated_scalar_t2 where correlated_scalar_t1.c1 = correlated_scalar_t2.c1) order by c2;"""
     qt_select_3_aggs """select c2 from correlated_scalar_t1 where correlated_scalar_t1.c2 > (select if(sum(c1) is null, count(c1), max(c2)) from correlated_scalar_t2 where correlated_scalar_t1.c1 = correlated_scalar_t2.c1) order by c2;"""
-}
+}

regression-test/suites/nereids_syntax_p0/sub_query_correlated.groovy

@@ -383,9 +383,12 @@ suite ("sub_query_correlated") {
     """
 
     //----------subquery with order and limit----------
-    order_qt_scalar_subquery_with_order_and_limit """
-        select * from sub_query_correlated_subquery1 where sub_query_correlated_subquery1.k1 > (select sum(sub_query_correlated_subquery3.k3) a from sub_query_correlated_subquery3 where sub_query_correlated_subquery3.v2 = sub_query_correlated_subquery1.k2 order by a limit 1);
-    """
+    test {
+        sql """
+            select * from sub_query_correlated_subquery1 where sub_query_correlated_subquery1.k1 > (select sum(sub_query_correlated_subquery3.k3) a from sub_query_correlated_subquery3 where sub_query_correlated_subquery3.v2 = sub_query_correlated_subquery1.k2 order by a limit 1);
+        """
+        exception """limit is not supported in correlated subquery"""
+    }
 
     //---------subquery with Disjunctions-------------
 	order_qt_scalar_subquery_with_disjunctions """