[fix](lateral view)(subquery) Forbidden directly AGG/SORT on lateral view (#7337)

This PR mainly prohibits operations such as aggregation/sorting/window functions
on lateral views containing subqueries.
For example:
select min(e1) from (select c1 from table group by c1)tmp1 lateral view explode_split(c1, ",") tmp2 as e1
But the query can be written in another way, and the result is the same.
select min(e1) from (select e1 from (select c1 from table group by c1)tmp1 lateral view explode_split(c1, ",") tmp2 as e1) tmp3

The reason is that when the results of a inline view are subjected to a lateral view,
and the outer query performs aggregation or sorting operations on non-table-function columns.
The output slot id of the table function node is empty or has fewer columns.

The essential reason is that when the inner layer contains an inline view,
the outer expression needs to be mapped to the correct tuple through the substitute method
according to the smap instead of the virtual tuple.
But the substitute method of slot ref cannot recurse to its own source exprs.

E.g
SlotRef: c2 <source expr min(c1)> from agg tuple
smap: <c1, c3>
before: c2 <source expr min(c1)>
after: c2 <source expr min(c1)> no changed

fe/fe-core/src/main/java/org/apache/doris/analysis/SelectStmt.java

@@ -564,6 +564,15 @@ public class SelectStmt extends QueryStmt {
         return result;
     }
 
+    public boolean hasInlineView() {
+        for (TableRef ref : fromClause_) {
+            if (ref instanceof InlineViewRef) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     @Override
     public List<TupleId> collectTupleIds() {
         List<TupleId> result = Lists.newArrayList();

fe/fe-core/src/main/java/org/apache/doris/analysis/SlotRef.java

@@ -30,7 +30,6 @@ import org.apache.doris.thrift.TSlotRef;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Objects;
 import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -375,19 +374,6 @@ public class SlotRef extends Expr {
         }
     }
 
-    @Override
-    protected Expr substituteImpl(ExprSubstitutionMap smap, Analyzer analyzer) throws AnalysisException {
-        if (isAnalyzed && desc != null && desc.getParent().getTable() == null && desc.getSourceExprs() != null) {
-            List<Expr> newSourceExprs = Lists.newArrayList();
-            for (int i = 0; i < desc.getSourceExprs().size(); ++i) {
-                newSourceExprs.add(desc.getSourceExprs().get(i).substituteImpl(smap, analyzer));
-            }
-            desc.setSourceExprs(newSourceExprs);
-            return this;
-        }
-        return super.substituteImpl(smap, analyzer);
-    }
-
     public Table getTable() {
         Preconditions.checkState(desc != null);
         Table table = desc.getParent().getTable();

fe/fe-core/src/main/java/org/apache/doris/planner/TableFunctionNode.java

@@ -24,6 +24,7 @@ import org.apache.doris.analysis.SelectStmt;
 import org.apache.doris.analysis.SlotId;
 import org.apache.doris.analysis.SlotRef;
 import org.apache.doris.analysis.TupleId;
+import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.UserException;
 import org.apache.doris.thrift.TExplainLevel;
 import org.apache.doris.thrift.TPlanNode;
@@ -81,7 +82,15 @@ public class TableFunctionNode extends PlanNode {
      * Query: select k1 from table a lateral view explode_split(v1, ",") t1 as c1;
      * The outputSlots: [k1, c1]
      */
-    public void projectSlots(Analyzer analyzer, SelectStmt selectStmt) {
+    public void projectSlots(Analyzer analyzer, SelectStmt selectStmt) throws AnalysisException {
+        // TODO(ml): Support project calculations that include aggregation and sorting in select stmt
+        if ((selectStmt.hasAggInfo() || selectStmt.getSortInfo() != null || selectStmt.hasAnalyticInfo())
+                && selectStmt.hasInlineView()) {
+            // The query must be rewritten like TableFunctionPlanTest.aggColumnInOuterQuery()
+            throw new AnalysisException("Please treat the query containing the lateral view as a inline view"
+                    + "and extract your aggregation/sort/window functions to the outer query."
+                    + "For example select sum(a) from (select a from table lateral view xxx) tmp1");
+        }
         Set<SlotRef> outputSlotRef = Sets.newHashSet();
         // case1
         List<Expr> baseTblResultExprs = selectStmt.getBaseTblResultExprs();

fe/fe-core/src/test/java/org/apache/doris/planner/TableFunctionPlanTest.java

@@ -441,4 +441,33 @@ public class TableFunctionPlanTest {
         Assert.assertTrue(explainString.contains("table function: explode_json_array_double('[1.1, 2.2, 3.3]')"));
         Assert.assertTrue(explainString.contains("output slot id: 0 1"));
     }
+    /*
+    Case4 agg and order column in the same stmt with lateral view
+    select min(c1) from (select k1 as c1, min(k2) as c2 from tbl1 group by k1) tmp1
+    lateral view explode_split(c2, ",") tmp2 as e1 order by min(c1)
+     */
+    @Test
+    public void aggColumnForbidden() throws Exception {
+        String sql = "desc verbose select min(c1) from (select k1 as c1, min(k2) as c2 from db1.tbl1 group by c1) a "
+                + "lateral view explode_split(c2, \",\") tmp1 as e1 order by min(c1)";
+        String errorMsg = UtFrameUtils.getSQLPlanOrErrorMsg(ctx, sql, true);
+        errorMsg.equalsIgnoreCase("lateral view as a inline view");
+    }
+
+    /*
+    Case5 agg and order column in the outer level
+    select min(c1) from (select c1 from (select k1 as c1, min(k2) as c2 from tbl1 group by k1) tmp1
+    lateral view explode_split(c2, ",") tmp2 as e1 ) tmp3
+    */
+    @Test
+    public void aggColumnInOuterQuery() throws Exception {
+        String sql = "desc verbose select min(c1) from (select c1 from (select k1 as c1, min(k2) as c2 from db1.tbl1 group by c1) a "
+                + "lateral view explode_split(c2, \",\") tmp1 as e1) tmp2";
+        String explainString = UtFrameUtils.getSQLPlanOrErrorMsg(ctx, sql, true);
+        Assert.assertTrue(explainString.contains("2:TABLE FUNCTION NODE"));
+        Assert.assertTrue(explainString.contains("table function: explode_split(<slot 3> min(`k2`), ',')"));
+        Assert.assertTrue(explainString.contains("lateral view tuple id: 3"));
+        Assert.assertTrue(explainString.contains("output slot id: 2"));
+        Assert.assertTrue(explainString.contains("tuple ids: 1 3"));
+    }
 }