Skip to content

Commit

Permalink
[refactor](nereids)use nereids to implement TableQueryPlanAction
Browse files Browse the repository at this point in the history
  • Loading branch information
@starocean999 @morrySnow
starocean999 authored and morrySnow committed Sep 11, 2024
1 parent c8b5386 commit 8be2b8a
Showing 1 changed file with 62 additions and 42 deletions.
104 changes: 62 additions & 42 deletions fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/TableQueryPlanAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,7 @@

package org.apache.doris.httpv2.rest;

import org.apache.doris.analysis.InlineViewRef;
import org.apache.doris.analysis.SelectStmt;
import org.apache.doris.analysis.StatementBase;
import org.apache.doris.analysis.TableName;
import org.apache.doris.analysis.TableRef;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.Table;
Expand All @@ -33,20 +29,31 @@
import org.apache.doris.httpv2.entity.ResponseEntityBuilder;
import org.apache.doris.httpv2.rest.manager.HttpUtils;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.nereids.NereidsPlanner;
import org.apache.doris.nereids.analyzer.UnboundRelation;
import org.apache.doris.nereids.glue.LogicalPlanAdapter;
import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.trees.plans.commands.Command;
import org.apache.doris.nereids.trees.plans.commands.ExplainCommand;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
import org.apache.doris.nereids.trees.plans.logical.LogicalLimit;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalSort;
import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
import org.apache.doris.nereids.trees.plans.logical.LogicalTopN;
import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
import org.apache.doris.nereids.util.RelationUtil;
import org.apache.doris.planner.PlanFragment;
import org.apache.doris.planner.Planner;
import org.apache.doris.planner.ScanNode;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.GlobalVariable;
import org.apache.doris.qe.OriginStatement;
import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.thrift.TDataSink;
import org.apache.doris.thrift.TDataSinkType;
import org.apache.doris.thrift.TMemoryScratchSink;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TPaloScanRange;
import org.apache.doris.thrift.TPlanFragment;
import org.apache.doris.thrift.TQueryOptions;
import org.apache.doris.thrift.TQueryPlanInfo;
import org.apache.doris.thrift.TScanRangeLocations;
import org.apache.doris.thrift.TTabletVersionInfo;
Expand Down Expand Up @@ -163,65 +170,78 @@ public Object query_plan(
*/
private void handleQuery(ConnectContext context, String requestDb, String requestTable, String sql,
Map<String, Object> result) throws DorisHttpException {
// use SE to resolve sql
StmtExecutor stmtExecutor = new StmtExecutor(context, new OriginStatement(sql, 0), false);
List<StatementBase> stmts = null;
try {
TQueryOptions tQueryOptions = context.getSessionVariable().toThrift();
// Conduct Planner create SingleNodePlan#createPlanFragments
tQueryOptions.num_nodes = 1;
// analyze sql
stmtExecutor.analyze(tQueryOptions);
stmts = new NereidsParser().parseSQL(sql, context.getSessionVariable());
} catch (Exception e) {
throw new DorisHttpException(HttpResponseStatus.BAD_REQUEST, e.getMessage());
}
// the parsed logical statement
StatementBase query = stmtExecutor.getParsedStmt();
// only process select semantic
if (!(query instanceof SelectStmt)) {
StatementBase query = stmts.get(0);
if (!(query instanceof LogicalPlanAdapter)) {
throw new DorisHttpException(HttpResponseStatus.BAD_REQUEST,
"Select statement needed, but found [" + sql + " ]");
}
SelectStmt stmt = (SelectStmt) query;
// just only process sql like `select * from table where <predicate>`, only support executing scan semantic
if (stmt.hasAggInfo() || stmt.hasAnalyticInfo()
|| stmt.hasOrderByClause() || stmt.hasOffset() || stmt.hasLimit() || stmt.isExplain()) {
LogicalPlan parsedPlan = ((LogicalPlanAdapter) query).getLogicalPlan();
// only process select semantic
if (parsedPlan instanceof Command) {
throw new DorisHttpException(HttpResponseStatus.BAD_REQUEST,
"only support single table filter-prune-scan, but found [ " + sql + "]");
"Select statement needed, but found [" + sql + " ]");
}
// process only one table by one http query
List<TableRef> fromTables = stmt.getTableRefs();
if (fromTables.size() != 1) {

if (!parsedPlan.collectToList(LogicalSubQueryAlias.class::isInstance).isEmpty()) {
throw new DorisHttpException(HttpResponseStatus.BAD_REQUEST,
"Select statement must have only one table");
"Select statement must not embed another statement");
}

TableRef fromTable = fromTables.get(0);
if (fromTable instanceof InlineViewRef) {
List<UnboundRelation> unboundRelations = parsedPlan.collectToList(UnboundRelation.class::isInstance);
if (unboundRelations.size() != 1) {
throw new DorisHttpException(HttpResponseStatus.BAD_REQUEST,
"Select statement must not embed another statement");
"Select statement must have only one table");
}

// check consistent http requested resource with sql referenced
// if consistent in this way, can avoid check privilege
TableName tableAndDb = fromTables.get(0).getName();
int lower = GlobalVariable.lowerCaseTableNames;
//Determine whether table names are case-sensitive
if (lower == 0) {
if (!(tableAndDb.getDb().equals(requestDb) && tableAndDb.getTbl().equals(requestTable))) {
List<String> tableQualifier = RelationUtil.getQualifierName(context,
unboundRelations.get(0).getNameParts());
if (tableQualifier.size() != 3) {
throw new DorisHttpException(HttpResponseStatus.BAD_REQUEST,
"can't find table " + String.join(",", tableQualifier));
}
String dbName = tableQualifier.get(1);
String tableName = tableQualifier.get(2);

if (GlobalVariable.lowerCaseTableNames == 0) {
if (!(dbName.equals(requestDb) && tableName.equals(requestTable))) {
throw new DorisHttpException(HttpResponseStatus.BAD_REQUEST,
"requested database and table must consistent with sql: request [ "
+ requestDb + "." + requestTable + "]" + "and sql [" + tableAndDb.toString() + "]");
+ requestDb + "." + requestTable + "]" + "and sql [" + dbName + "." + tableName + "]");
}
} else {
if (!(tableAndDb.getDb().equalsIgnoreCase(requestDb)
&& tableAndDb.getTbl().equalsIgnoreCase(requestTable))) {
if (!(dbName.equalsIgnoreCase(requestDb)
&& tableName.equalsIgnoreCase(requestTable))) {
throw new DorisHttpException(HttpResponseStatus.BAD_REQUEST,
"requested database and table must consistent with sql: request [ "
+ requestDb + "." + requestTable + "]" + "and sql [" + tableAndDb.toString() + "]");
+ requestDb + "." + requestTable + "]" + "and sql [" + dbName + "." + tableName + "]");
}
}
NereidsPlanner nereidsPlanner = new NereidsPlanner(context.getStatementContext());
LogicalPlan rewrittenPlan = (LogicalPlan) nereidsPlanner.planWithLock(parsedPlan,
PhysicalProperties.GATHER, ExplainCommand.ExplainLevel.REWRITTEN_PLAN);
if (rewrittenPlan.anyMatch(planTreeNode -> planTreeNode instanceof LogicalAggregate
|| planTreeNode instanceof LogicalWindow || planTreeNode instanceof LogicalSort
|| planTreeNode instanceof LogicalTopN || planTreeNode instanceof LogicalLimit)) {
throw new DorisHttpException(HttpResponseStatus.BAD_REQUEST,
"only support single table filter-prune-scan, but found [ " + sql + "]");
}
NereidsPlanner planner = new NereidsPlanner(context.getStatementContext());
try {
planner.plan(query, context.getSessionVariable().toThrift());
} catch (Exception ex) {
throw new DorisHttpException(HttpResponseStatus.BAD_REQUEST,
"only support single table filter-prune-scan, but found [ " + sql + "]");
}

// acquired Planner to get PlanNode and fragment templates
Planner planner = stmtExecutor.planner();
// acquire ScanNode to obtain pruned tablet
// in this way, just retrieve only one scannode
List<ScanNode> scanNodes = planner.getScanNodes();
Expand Down Expand Up @@ -249,7 +269,7 @@ private void handleQuery(ConnectContext context, String requestDb, String reques
tPlanFragment.output_sink = tDataSink;

tQueryPlanInfo.plan_fragment = tPlanFragment;
tQueryPlanInfo.desc_tbl = query.getAnalyzer().getDescTbl().toThrift();
tQueryPlanInfo.desc_tbl = planner.getDescTable().toThrift();
// set query_id
UUID uuid = UUID.randomUUID();
tQueryPlanInfo.query_id = new TUniqueId(uuid.getMostSignificantBits(), uuid.getLeastSignificantBits());
Expand Down

0 comments on commit 8be2b8a

Please sign in to comment.