You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by si...@apache.org on 2022/08/31 17:00:10 UTC
[pinot] branch master updated: [multistage] explain plan to visualization (#9280)
This is an automated email from the ASF dual-hosted git repository.
siddteotia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 1c9528c1f5 [multistage] explain plan to visualization (#9280)
1c9528c1f5 is described below
commit 1c9528c1f54d90a0a62b1795c8785cf5647a676a
Author: Rong Rong <ro...@apache.org>
AuthorDate: Wed Aug 31 10:00:02 2022 -0700
[multistage] explain plan to visualization (#9280)
* support EXPLAIN in multi-stage engine
* remove test
Co-authored-by: Rong Rong <ro...@startree.ai>
---
.../MultiStageBrokerRequestHandler.java | 29 ++++++--
.../org/apache/pinot/query/QueryEnvironment.java | 83 ++++++++++++++++------
.../apache/pinot/query/planner/PlannerUtils.java | 12 ++++
.../apache/pinot/query/QueryCompilationTest.java | 47 +++++++-----
4 files changed, 128 insertions(+), 43 deletions(-)
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
index 5d83a6e329..2554c9ed60 100644
--- a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
@@ -133,13 +133,22 @@ public class MultiStageBrokerRequestHandler extends BaseBrokerRequestHandler {
throws Exception {
LOGGER.debug("SQL query for request {}: {}", requestId, query);
- // Parse the request
- sqlNodeAndOptions = sqlNodeAndOptions != null ? sqlNodeAndOptions : RequestUtils.parseQuery(query, request);
- // Compile the request
- long compilationStartTimeNs = System.nanoTime();
+ long compilationStartTimeNs;
QueryPlan queryPlan;
try {
- queryPlan = _queryEnvironment.planQuery(query, sqlNodeAndOptions);
+ // Parse the request
+ sqlNodeAndOptions = sqlNodeAndOptions != null ? sqlNodeAndOptions : RequestUtils.parseQuery(query, request);
+ // Compile the request
+ compilationStartTimeNs = System.nanoTime();
+ switch (sqlNodeAndOptions.getSqlNode().getKind()) {
+ case EXPLAIN:
+ String plan = _queryEnvironment.explainQuery(query, sqlNodeAndOptions);
+ return constructMultistageExplainPlan(query, plan);
+ case SELECT:
+ default:
+ queryPlan = _queryEnvironment.planQuery(query, sqlNodeAndOptions);
+ break;
+ }
} catch (Exception e) {
LOGGER.info("Caught exception while compiling SQL request {}: {}, {}", requestId, query, e.getMessage());
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.REQUEST_COMPILATION_EXCEPTIONS, 1);
@@ -168,6 +177,16 @@ public class MultiStageBrokerRequestHandler extends BaseBrokerRequestHandler {
return brokerResponse;
}
+ private BrokerResponseNative constructMultistageExplainPlan(String sql, String plan) {
+ BrokerResponseNative brokerResponse = BrokerResponseNative.empty();
+ List<Object[]> rows = new ArrayList<>();
+ rows.add(new Object[]{sql, plan});
+ DataSchema multistageExplainResultSchema = new DataSchema(new String[]{"SQL", "PLAN"},
+ new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING, DataSchema.ColumnDataType.STRING});
+ brokerResponse.setResultTable(new ResultTable(multistageExplainResultSchema, rows));
+ return brokerResponse;
+ }
+
@Override
protected BrokerResponseNative processBrokerRequest(long requestId, BrokerRequest originalBrokerRequest,
BrokerRequest serverBrokerRequest, @Nullable BrokerRequest offlineBrokerRequest,
diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java b/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java
index 516e6815d9..9d512369d1 100644
--- a/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java
+++ b/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java
@@ -38,6 +38,9 @@ import org.apache.calcite.rel.RelRoot;
import org.apache.calcite.rel.hint.HintStrategyTable;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.sql.SqlExplain;
+import org.apache.calcite.sql.SqlExplainFormat;
+import org.apache.calcite.sql.SqlExplainLevel;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
@@ -47,6 +50,7 @@ import org.apache.calcite.sql2rel.StandardConvertletTable;
import org.apache.calcite.tools.FrameworkConfig;
import org.apache.calcite.tools.Frameworks;
import org.apache.pinot.query.context.PlannerContext;
+import org.apache.pinot.query.planner.PlannerUtils;
import org.apache.pinot.query.planner.QueryPlan;
import org.apache.pinot.query.planner.logical.LogicalPlanner;
import org.apache.pinot.query.planner.logical.StagePlanner;
@@ -125,18 +129,39 @@ public class QueryEnvironment {
* @return a dispatchable query plan
*/
public QueryPlan planQuery(String sqlQuery, SqlNodeAndOptions sqlNodeAndOptions) {
- PlannerContext plannerContext = new PlannerContext();
try {
+ PlannerContext plannerContext = new PlannerContext();
plannerContext.setOptions(sqlNodeAndOptions.getOptions());
- SqlNode validated = validate(sqlNodeAndOptions.getSqlNode());
- RelRoot relation = toRelation(validated, plannerContext);
- RelNode optimized = optimize(relation, plannerContext);
- return toDispatchablePlan(optimized, plannerContext);
+ RelNode relRoot = compileQuery(sqlNodeAndOptions.getSqlNode(), plannerContext);
+ return toDispatchablePlan(relRoot, plannerContext);
} catch (Exception e) {
throw new RuntimeException("Error composing query plan for: " + sqlQuery, e);
- } finally {
- _planner.close();
- _planner.reset();
+ }
+ }
+
+ /**
+ * Explain a SQL query.
+ *
+ * Similar to {@link QueryEnvironment#planQuery(String, SqlNodeAndOptions)}, this API runs the query compilation.
+ * But it doesn't run the distributed {@link QueryPlan} generation, instead it only returns the explained logical
+ * plan.
+ *
+ * @param sqlQuery SQL query string.
+ * @param sqlNodeAndOptions parsed SQL query.
+ * @return the explained query plan.
+ */
+ public String explainQuery(String sqlQuery, SqlNodeAndOptions sqlNodeAndOptions) {
+ try {
+ SqlExplain explain = (SqlExplain) sqlNodeAndOptions.getSqlNode();
+ PlannerContext plannerContext = new PlannerContext();
+ plannerContext.setOptions(sqlNodeAndOptions.getOptions());
+ RelNode relRoot = compileQuery(explain.getExplicandum(), plannerContext);
+ SqlExplainFormat format = explain.getFormat() == null ? SqlExplainFormat.DOT : explain.getFormat();
+ SqlExplainLevel level = explain.getDetailLevel() == null ? SqlExplainLevel.DIGEST_ATTRIBUTES
+ : explain.getDetailLevel();
+ return PlannerUtils.explainPlan(relRoot, format, level);
+ } catch (Exception e) {
+ throw new RuntimeException("Error explain query plan for: " + sqlQuery, e);
}
}
@@ -145,20 +170,29 @@ public class QueryEnvironment {
return planQuery(sqlQuery, CalciteSqlParser.compileToSqlNodeAndOptions(sqlQuery));
}
+ @VisibleForTesting
+ public String explainQuery(String sqlQuery) {
+ return explainQuery(sqlQuery, CalciteSqlParser.compileToSqlNodeAndOptions(sqlQuery));
+ }
+
// --------------------------------------------------------------------------
// steps
// --------------------------------------------------------------------------
@VisibleForTesting
- protected SqlNode parse(String query, PlannerContext plannerContext)
+ protected RelNode compileQuery(SqlNode sqlNode, PlannerContext plannerContext)
throws Exception {
- // 1. invoke CalciteSqlParser to parse out SqlNode;
- SqlNodeAndOptions sqlNodeAndOptions = CalciteSqlParser.compileToSqlNodeAndOptions(query);
- plannerContext.setOptions(sqlNodeAndOptions.getOptions());
- return sqlNodeAndOptions.getSqlNode();
+ try {
+ SqlNode validated = validate(sqlNode);
+ RelRoot relation = toRelation(validated, plannerContext);
+ return optimize(relation, plannerContext);
+ } finally {
+ _planner.close();
+ _planner.reset();
+ }
}
- protected SqlNode validate(SqlNode parsed)
+ private SqlNode validate(SqlNode parsed)
throws Exception {
// 2. validator to validate.
SqlNode validated = _validator.validate(parsed);
@@ -169,7 +203,7 @@ public class QueryEnvironment {
return validated;
}
- protected RelRoot toRelation(SqlNode parsed, PlannerContext plannerContext) {
+ private RelRoot toRelation(SqlNode parsed, PlannerContext plannerContext) {
// 3. convert sqlNode to relNode.
RexBuilder rexBuilder = new RexBuilder(_typeFactory);
RelOptCluster cluster = RelOptCluster.create(_relOptPlanner, rexBuilder);
@@ -179,12 +213,7 @@ public class QueryEnvironment {
return sqlToRelConverter.convertQuery(parsed, false, true);
}
- // TODO: add hint strategy table based on plannerContext.
- private HintStrategyTable getHintStrategyTable(PlannerContext plannerContext) {
- return HintStrategyTable.builder().build();
- }
-
- protected RelNode optimize(RelRoot relRoot, PlannerContext plannerContext) {
+ private RelNode optimize(RelRoot relRoot, PlannerContext plannerContext) {
// 4. optimize relNode
// TODO: add support for traits, cost factory.
try {
@@ -196,9 +225,19 @@ public class QueryEnvironment {
}
}
- protected QueryPlan toDispatchablePlan(RelNode relRoot, PlannerContext plannerContext) {
+ private QueryPlan toDispatchablePlan(RelNode relRoot, PlannerContext plannerContext) {
// 5. construct a dispatchable query plan.
StagePlanner queryStagePlanner = new StagePlanner(plannerContext, _workerManager);
return queryStagePlanner.makePlan(relRoot);
}
+
+
+ // --------------------------------------------------------------------------
+ // utils
+ // --------------------------------------------------------------------------
+
+ // TODO: add hint strategy table based on plannerContext.
+ private HintStrategyTable getHintStrategyTable(PlannerContext plannerContext) {
+ return HintStrategyTable.builder().build();
+ }
}
diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/PlannerUtils.java b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/PlannerUtils.java
index d1710a486b..b6ebb214a1 100644
--- a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/PlannerUtils.java
+++ b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/PlannerUtils.java
@@ -23,15 +23,23 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlExplainFormat;
+import org.apache.calcite.sql.SqlExplainLevel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Utilities used by planner.
*/
public class PlannerUtils {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PlannerUtils.class);
+
private PlannerUtils() {
// do not instantiate.
}
@@ -64,4 +72,8 @@ public class PlannerUtils {
public static boolean isRootStage(int stageId) {
return stageId == 0;
}
+
+ public static String explainPlan(RelNode relRoot, SqlExplainFormat format, SqlExplainLevel explainLevel) {
+ return RelOptUtil.dumpPlan("Execution Plan", relRoot, format, explainLevel);
+ }
}
diff --git a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
index 31c018d158..e2332235fa 100644
--- a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
+++ b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
@@ -23,9 +23,7 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import org.apache.calcite.rel.RelDistribution;
-import org.apache.calcite.sql.SqlNode;
import org.apache.pinot.core.transport.ServerInstance;
-import org.apache.pinot.query.context.PlannerContext;
import org.apache.pinot.query.planner.PlannerUtils;
import org.apache.pinot.query.planner.QueryPlan;
import org.apache.pinot.query.planner.StageMetadata;
@@ -43,13 +41,15 @@ import org.testng.annotations.Test;
public class QueryCompilationTest extends QueryEnvironmentTestBase {
- @Test(dataProvider = "testQueryParserDataProvider")
- public void testQueryParser(String query, String digest)
+ @Test(dataProvider = "testQueryPlanDataProvider")
+ public void testQueryPlanExplain(String query, String digest)
throws Exception {
- PlannerContext plannerContext = new PlannerContext();
- SqlNode sqlNode = _queryEnvironment.parse(query, plannerContext);
- _queryEnvironment.validate(sqlNode);
- Assert.assertEquals(sqlNode.toString(), digest);
+ try {
+ String explainedPlan = _queryEnvironment.explainQuery(query);
+ Assert.assertEquals(explainedPlan, digest);
+ } catch (RuntimeException e) {
+ Assert.fail("failed to explain query: " + query, e);
+ }
}
@Test(dataProvider = "testQueryDataProvider")
@@ -169,14 +169,6 @@ public class QueryCompilationTest extends QueryEnvironmentTestBase {
return false;
}
- @DataProvider(name = "testQueryParserDataProvider")
- private Object[][] provideQueriesAndDigest() {
- return new Object[][] {
- new Object[]{"SELECT * FROM a JOIN b ON a.col1 = b.col2 WHERE a.col3 >= 0",
- "SELECT *\n" + "FROM `a`\n" + "INNER JOIN `b` ON `a`.`col1` = `b`.`col2`\n" + "WHERE `a`.`col3` >= 0"},
- };
- }
-
@DataProvider(name = "testQueryExceptionDataProvider")
private Object[][] provideQueriesWithException() {
return new Object[][] {
@@ -186,4 +178,27 @@ public class QueryCompilationTest extends QueryEnvironmentTestBase {
new Object[]{"SELECT a.col1, SUM(a.col3) FROM a", "'a.col1' is not being grouped"},
};
}
+
+ @DataProvider(name = "testQueryPlanDataProvider")
+ private Object[][] provideQueriesWithExplainedPlan() {
+ return new Object[][] {
+ new Object[]{"EXPLAIN PLAN INCLUDING ALL ATTRIBUTES AS JSON FOR SELECT col1, col3 FROM a", "{\n"
+ + " \"rels\": [\n" + " {\n" + " \"id\": \"0\",\n" + " \"relOp\": \"LogicalTableScan\",\n"
+ + " \"table\": [\n" + " \"a\"\n" + " ],\n" + " \"inputs\": []\n" + " },\n"
+ + " {\n" + " \"id\": \"1\",\n" + " \"relOp\": \"LogicalProject\",\n" + " \"fields\": [\n"
+ + " \"col1\",\n" + " \"col3\"\n" + " ],\n" + " \"exprs\": [\n" + " {\n"
+ + " \"input\": 2,\n" + " \"name\": \"$2\"\n" + " },\n" + " {\n"
+ + " \"input\": 1,\n" + " \"name\": \"$1\"\n" + " }\n" + " ]\n" + " }\n"
+ + " ]\n" + "}"},
+ new Object[]{"EXPLAIN PLAN EXCLUDING ATTRIBUTES AS DOT FOR SELECT col1, COUNT(*) FROM a GROUP BY col1",
+ "Execution Plan\n" + "digraph {\n" + "\"LogicalExchange\\n\" -> \"LogicalAggregate\\n\" [label=\"0\"]\n"
+ + "\"LogicalAggregate\\n\" -> \"LogicalExchange\\n\" [label=\"0\"]\n"
+ + "\"LogicalTableScan\\n\" -> \"LogicalAggregate\\n\" [label=\"0\"]\n" + "}\n"},
+ new Object[]{"EXPLAIN PLAN FOR SELECT a.col1, b.col3 FROM a JOIN b ON a.col1 = b.col1", "Execution Plan\n"
+ + "LogicalProject(col1=[$0], col3=[$1])\n" + " LogicalJoin(condition=[=($0, $2)], joinType=[inner])\n"
+ + " LogicalExchange(distribution=[hash[0]])\n" + " LogicalProject(col1=[$2])\n"
+ + " LogicalTableScan(table=[[a]])\n" + " LogicalExchange(distribution=[hash[1]])\n"
+ + " LogicalProject(col3=[$1], col1=[$2])\n" + " LogicalTableScan(table=[[b]])\n"},
+ };
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org