zstan commented on a change in pull request #712:
URL: https://github.com/apache/ignite-3/pull/712#discussion_r825994401
##########
File path:
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java
##########
@@ -216,106 +214,62 @@ public synchronized void stop() throws Exception {
}
}
+ /**
+ * To be removed.
+ *
+ * @return Always return null.
+ */
public QueryRegistry queryRegistry() {
- return queryRegistry;
+ return null;
}
- private List<SqlCursor<List<?>>> query0(QueryContext context, String
schemaName, String sql, Object... params) {
+ private List<CompletableFuture<AsyncSqlCursor<List<?>>>>
query0(QueryContext context, String schemaName, String sql, Object... params) {
SchemaPlus schema = schemaManager.schema(schemaName);
- assert schema != null : "Schema not found: " + schemaName;
-
- QueryPlan plan = planCache.queryPlan(new CacheKey(schema.getName(),
sql));
-
- if (plan != null) {
- final QueryPlan finalPlan = plan;
-
- context.maybeUnwrap(QueryValidator.class)
- .ifPresent(queryValidator ->
queryValidator.validatePlan(finalPlan));
-
- RootQuery<Object[]> qry = new RootQuery<>(
- sql,
- schema,
- params,
- exchangeService,
- (q) -> queryRegistry.unregister(q.id()),
- LOG
- );
-
- queryRegistry.register(qry);
-
- try {
- return Collections.singletonList(executionSrvc.executePlan(
- qry,
- plan
- ));
- } catch (Exception e) {
- boolean isCanceled = qry.isCancelled();
-
- qry.cancel();
-
- queryRegistry.unregister(qry.id());
-
- if (isCanceled) {
- throw new IgniteInternalException("The query was cancelled
while planning", e);
- } else {
- throw e;
- }
- }
+ if (schema == null) {
+ throw new IgniteInternalException(format("Schema not found
[schemaName={}]", schemaName));
}
- SqlNodeList qryList = Commons.parse(sql,
FRAMEWORK_CONFIG.getParserConfig());
- List<SqlCursor<List<?>>> cursors = new ArrayList<>(qryList.size());
-
- List<RootQuery<Object[]>> qrys = new ArrayList<>(qryList.size());
-
- for (final SqlNode sqlNode : qryList) {
- RootQuery<Object[]> qry = new RootQuery<>(
- sqlNode.toString(),
- schemaManager.schema(schemaName), // Update schema for
each query in multiple statements.
- params,
- exchangeService,
- (q) -> queryRegistry.unregister(q.id()),
- LOG
- );
-
- qrys.add(qry);
+ SqlNodeList nodes = parsingCache.computeIfAbsent(sql, key ->
Commons.parse(key, FRAMEWORK_CONFIG.getParserConfig()));
- queryRegistry.register(qry);
+ List<CompletableFuture<AsyncSqlCursor<List<?>>>> res = new
ArrayList<>(nodes.size());
- try {
- if (qryList.size() == 1) {
- plan = planCache.queryPlan(
- new CacheKey(schemaName, qry.sql()),
- () -> prepareSvc.prepareSingle(sqlNode,
qry.planningContext()));
- } else {
- plan = prepareSvc.prepareSingle(sqlNode,
qry.planningContext());
- }
+ CompletableFuture<Void> start = new CompletableFuture<>();
- final QueryPlan finalPlan = plan;
+ for (SqlNode sqlNode : nodes) {
+ BaseQueryContext ctx = BaseQueryContext.builder()
+ .cancel(new QueryCancel())
+ .frameworkConfig(
+ Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
+ .defaultSchema(schema)
+ .build()
+ )
+ .logger(LOG)
+ .parameters(params)
+ .build();
- context.maybeUnwrap(QueryValidator.class)
- .ifPresent(queryValidator ->
queryValidator.validatePlan(finalPlan));
+ CompletableFuture<AsyncSqlCursor<List<?>>> stage =
start.thenCompose(voidArg -> prepareSvc.prepare(sqlNode, ctx))
+ .thenApply(plan -> {
+ context.maybeUnwrap(QueryValidator.class)
+ .ifPresent(queryValidator ->
queryValidator.validatePlan(plan));
- cursors.add(executionSrvc.executePlan(qry, plan));
- } catch (Exception e) {
- boolean isCanceled = qry.isCancelled();
+ return new AsyncSqlCursorImpl<>(
+ SqlQueryType.mapPlanTypeToSqlType(plan.type()),
+ executionSrvc.executePlan(plan, ctx)
+ );
+ });
- qrys.forEach(RootQuery::cancel);
-
- queryRegistry.unregister(qry.id());
-
- if (isCanceled) {
- throw new IgniteInternalException("The query was cancelled
while planning", e);
- } else {
- throw e;
- }
- }
+ res.add(stage);
}
- return cursors;
+ // TODO: pass to a particular executor
+ start.completeAsync(() -> null);
+
+ return res;
}
+ Map<String, SqlNodeList> parsingCache = new HashMap<>();
Review comment:
move it to appropriate code part.
##########
File path:
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java
##########
@@ -62,43 +64,49 @@ public void stop() throws Exception {
}
/** {@inheritDoc} */
- @Override public QueryPlan prepareSingle(SqlNode sqlNode, PlanningContext
ctx) {
- try {
- assert single(sqlNode);
-
- ctx.planner().reset();
-
- if (SqlKind.DDL.contains(sqlNode.getKind())) {
- return prepareDdl(sqlNode, ctx);
- }
-
- switch (sqlNode.getKind()) {
- case SELECT:
- case ORDER_BY:
- case WITH:
- case VALUES:
- case UNION:
- case EXCEPT:
- case INTERSECT:
- return prepareQuery(sqlNode, ctx);
-
- case INSERT:
- case DELETE:
- case UPDATE:
- case MERGE:
- return prepareDml(sqlNode, ctx);
-
- case EXPLAIN:
- return prepareExplain(sqlNode, ctx);
-
- default:
- throw new IgniteInternalException("Unsupported operation ["
- + "sqlNodeKind=" + sqlNode.getKind() + "; "
- + "querySql=\"" + ctx.query() + "\"]");
- }
- } catch (ValidationException | CalciteContextException e) {
- throw new IgniteInternalException("Failed to validate query. " +
e.getMessage(), e);
- }
+ @Override
+ public CompletableFuture<QueryPlan> prepare(SqlNode sqlNode,
BaseQueryContext ctx) {
+ return CompletableFuture.completedFuture(null)
Review comment:
```suggestion
return CompletableFuture.supplyAsync(() -> {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
