You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by am...@apache.org on 2023/06/14 08:32:04 UTC
[ignite-3] branch ignite-17765-2 updated (05cf5031dc -> 134337e248)
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a change to branch ignite-17765-2
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
discard 05cf5031dc fix
new 134337e248 fix
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (05cf5031dc)
\
N -- N -- N refs/heads/ignite-17765-2 (134337e248)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../internal/sql/engine/SqlQueryProcessor.java | 82 +++++++++++-----------
.../internal/sql/engine/prepare/CacheKey.java | 20 +++---
.../sql/engine/schema/CatalogSqlSchemaManager.java | 5 --
.../sql/engine/schema/SqlSchemaManager.java | 5 --
4 files changed, 52 insertions(+), 60 deletions(-)
[ignite-3] 01/01: fix
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-17765-2
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 134337e248147807493af3bc6c9b96d031c2f9a1
Author: amashenkov <an...@gmail.com>
AuthorDate: Tue Jun 13 19:06:06 2023 +0300
fix
---
.../internal/sql/engine/SqlQueryProcessor.java | 281 +++++++++++----------
.../internal/sql/engine/prepare/CacheKey.java | 35 ++-
2 files changed, 175 insertions(+), 141 deletions(-)
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java
index ffa661fe87..b2b98cc3c7 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.sql.engine;
+import static java.util.concurrent.CompletableFuture.failedFuture;
import static org.apache.ignite.internal.sql.engine.SqlQueryType.DML;
import static org.apache.ignite.internal.sql.engine.SqlQueryType.QUERY;
import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
@@ -33,13 +34,11 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.LongFunction;
import java.util.function.Supplier;
@@ -53,6 +52,7 @@ import org.apache.calcite.util.Pair;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.index.IndexManager;
import org.apache.ignite.internal.index.event.IndexEvent;
import org.apache.ignite.internal.index.event.IndexEventParameters;
@@ -90,7 +90,6 @@ import org.apache.ignite.internal.sql.engine.sql.IgniteSqlParser;
import org.apache.ignite.internal.sql.engine.sql.ParseResult;
import org.apache.ignite.internal.sql.engine.sql.StatementParseResult;
import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext.Builder;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.sql.engine.util.TypeUtils;
import org.apache.ignite.internal.storage.DataStorageManager;
@@ -106,11 +105,10 @@ import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.lang.SchemaNotFoundException;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.sql.SqlException;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
- * SqlQueryProcessor.
+ * SqlQueryProcessor.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
public class SqlQueryProcessor implements QueryProcessor {
@@ -260,6 +258,7 @@ public class SqlQueryProcessor implements QueryProcessor {
busyLock
);
+ //TODO IGNITE-17765 Do we need cache clear as cache already aware of catalog versioning.
sqlSchemaManager.registerListener(queryCache::clear);
this.prepareSvc = prepareSvc;
@@ -390,22 +389,125 @@ public class SqlQueryProcessor implements QueryProcessor {
private CompletableFuture<AsyncSqlCursor<List<Object>>> querySingle0(
SessionId sessionId,
- QueryContext context,
+ QueryContext queryContext,
String sql,
Object... params
) {
Session session = sessionManager.session(sessionId);
if (session == null) {
- return CompletableFuture.failedFuture(
+ return failedFuture(
new SqlException(SESSION_NOT_FOUND_ERR, format("Session not found [{}]", sessionId)));
}
+ QueryCancel queryCancel;
+
+ try {
+ queryCancel = createQueryCancel(session);
+ } catch (IllegalStateException ex) {
+ return failedFuture(new IgniteInternalException(SESSION_EXPIRED_ERR,
+ format("Session has been expired [{}]", session.sessionId()), ex));
+ }
+
String schemaName = session.properties().get(QueryProperty.DEFAULT_SCHEMA);
- InternalTransaction outerTx = context.unwrap(InternalTransaction.class);
+ InternalTransaction outerTx = queryContext.unwrap(InternalTransaction.class);
+ long timestamp = outerTx == null ? clock.nowLong() : outerTx.startTimestamp().longValue();
+
+ boolean implicitTx = outerTx == null;
+
+ int plannerCatalogVersion = catalogManager.activeCatalogVersion(timestamp);
+ SchemaPlus schema = sqlSchemaManager.schema(schemaName, plannerCatalogVersion);
+
+ if (schema == null) {
+ return failedFuture(new SchemaNotFoundException(schemaName));
+ }
+
+ int schemaId = catalogManager.schema(plannerCatalogVersion).id();
+
+ CacheKey cacheKey = new CacheKey(plannerCatalogVersion, schemaId, sql, params);
+
+ BaseQueryContext plannerContext = BaseQueryContext.builder()
+ .frameworkConfig(Frameworks.newConfigBuilder(FRAMEWORK_CONFIG).defaultSchema(schema).build())
+ .logger(LOG)
+ .cancel(queryCancel)
+ .parameters(params)
+ .plannerTimeout(PLANNER_TIMEOUT)
+ .build();
+
+ CompletableFuture<QueryPlan> planFuture = queryCache.get(cacheKey);
+
+ if (planFuture == null) {
+ planFuture = CompletableFuture.supplyAsync(() -> {
+ // Parse query
+ StatementParseResult parseResult = IgniteSqlParser.parse(sql, StatementParseResult.MODE);
+ SqlNode sqlNode = parseResult.statement();
+
+ // Validate statement
+ validateParsedStatement(queryContext, parseResult, sqlNode, params);
+
+ return sqlNode;
+ }).thenCompose(sqlNode -> {
+ if (skipCache(Commons.getQueryType(sqlNode))) {
+ // Prepare query plan without caching.
+ //TODO: IGNITE-17765 Can we make a sync callhere ? drop planning pool? or move parsing to planing pool?
+ return prepareSvc.prepareAsync(sqlNode, plannerContext);
+ }
+
+ // Try query plan for normalized query, or create a new one asynchronously.
+ CacheKey normalizedQueryCacheKey = new CacheKey(plannerCatalogVersion, schemaId, sqlNode.toString(), params);
+
+ CompletableFuture<QueryPlan> planFuture0 = queryCache.computeIfAbsent(
+ normalizedQueryCacheKey,
+ k -> prepareSvc.prepareAsync(sqlNode, plannerContext)
+ );
+
+ queryCache.putIfAbsent(cacheKey, planFuture0);
+
+ // Copy shared plan.
+ return planFuture0.thenApply(QueryPlan::copy);
+ });
+ } else {
+ // Copy shared plan.
+ planFuture = planFuture.thenApply(QueryPlan::copy);
+ }
+
+ return planFuture
+ .thenComposeAsync(plan -> {
+ // Validate plan
+ if (SqlQueryType.DDL == plan.type() && outerTx != null) {
+ return failedFuture(new SqlException(UNSUPPORTED_DDL_OPERATION_ERR, "DDL doesn't support transactions."));
+ }
+
+ InternalTransaction tx = implicitTx ? txManager.begin(plan.type() != DML) : outerTx;
+
+ try {
+ int txCatalogVersion = catalogManager.activeCatalogVersion(tx.startTimestamp().longValue());
+
+ if (implicitTx && plannerCatalogVersion != txCatalogVersion) {
+ LOG.info("Retry query planning: plannerCatalogVersion={}, txCatalogVersion={}",
+ plannerCatalogVersion, txCatalogVersion);
+
+ return tx.rollbackAsync()
+ .thenComposeAsync(ignore -> querySingle0(sessionId, queryContext, sql, params), taskExecutor);
+ }
- QueryCancel queryCancel = new QueryCancel();
+ return CompletableFuture.completedFuture(executePlan(session, tx, plan, plannerContext));
+ } catch (Throwable th) {
+ handleQueryException(th, queryCancel, implicitTx ? tx : null);
+
+ return failedFuture(th);
+ }
+ }, taskExecutor);
+ }
+
+ private static boolean skipCache(@Nullable SqlQueryType queryType) {
+ return queryType != QUERY && queryType != DML;
+ }
+
+ private QueryCancel createQueryCancel(Session session) {
+ QueryCancel queryCancel;
+ queryCancel = new QueryCancel();
AsyncCloseable closeableResource = () -> CompletableFuture.runAsync(
queryCancel::cancel,
@@ -413,114 +515,52 @@ public class SqlQueryProcessor implements QueryProcessor {
);
queryCancel.add(() -> session.unregisterResource(closeableResource));
+ session.registerResource(closeableResource);
+ return queryCancel;
+ }
- try {
- session.registerResource(closeableResource);
- } catch (IllegalStateException ex) {
- return CompletableFuture.failedFuture(new IgniteInternalException(SESSION_EXPIRED_ERR,
- format("Session has been expired [{}]", session.sessionId()), ex));
- }
- CompletableFuture<Void> start = new CompletableFuture<>();
-
- boolean implicitTxRequired = outerTx == null;
- AtomicReference<InternalTransaction> tx = new AtomicReference<>();
-
- CompletableFuture<AsyncSqlCursor<List<Object>>> stage = start
- .thenCompose(v -> {
- Builder contextBuilder = BaseQueryContext.builder()
- .logger(LOG)
- .cancel(queryCancel)
- .parameters(params)
- .plannerTimeout(PLANNER_TIMEOUT);
-
- CompletableFuture<PlanWithContext>[] newPlanHolder = new CompletableFuture[1];
-
- CompletableFuture<QueryPlan> cachedPlan = queryCache.computeIfAbsent(new CacheKey(schemaName, sql, params), (k) -> {
- // Parse query.
- StatementParseResult parseResult = IgniteSqlParser.parse(sql, StatementParseResult.MODE);
- SqlNode sqlNode = parseResult.statement();
-
- validateParsedStatement(context, outerTx, parseResult, sqlNode, params);
-
- // Build context.
- tx.set(implicitTxRequired ? txManager.begin(!dataModificationOp(sqlNode)) : outerTx);
- SchemaPlus schema = resolveSchema(schemaName);
- BaseQueryContext ctx = contextBuilder.frameworkConfig(
- Frameworks.newConfigBuilder(FRAMEWORK_CONFIG).defaultSchema(schema).build()).build();
-
- CompletableFuture<QueryPlan> planFuture = prepareSvc.prepareAsync(sqlNode, ctx);
-
- SqlQueryType queryType = Commons.getQueryType(sqlNode);
- boolean putPlanIntoCache = queryType == QUERY || queryType == DML;
-
- newPlanHolder[0] = planFuture.thenApply(plan -> new PlanWithContext(putPlanIntoCache ? plan.copy() : plan, ctx));
-
- return putPlanIntoCache ? planFuture : null;
- });
-
- return Objects.requireNonNullElseGet(
- newPlanHolder[0],
- () -> cachedPlan.thenApply(plan -> {
- // Build query context for execution.
- tx.set(implicitTxRequired ? txManager.begin(plan.type() != DML) : outerTx);
- SchemaPlus schema = resolveSchema(schemaName);
- BaseQueryContext ctx = contextBuilder.frameworkConfig(
- Frameworks.newConfigBuilder(FRAMEWORK_CONFIG).defaultSchema(schema).build()).build();
-
- return new PlanWithContext(plan.copy(), ctx);
- })
- ).thenApply(planWithContext -> {
- QueryPlan plan = planWithContext.plan;
- BaseQueryContext ctx = planWithContext.context;
-
- var dataCursor = executionSrvc.executePlan(tx.get(), plan, ctx);
-
- SqlQueryType queryType = plan.type();
- assert queryType != null : "Expected a full plan but got a fragment: " + plan;
-
- return new AsyncSqlCursorImpl<>(
- queryType,
- plan.metadata(),
- implicitTxRequired ? tx.get() : null,
- new AsyncCursor<List<Object>>() {
- @Override
- public CompletableFuture<BatchedResult<List<Object>>> requestNextAsync(int rows) {
- session.touch();
-
- return dataCursor.requestNextAsync(rows);
- }
-
- @Override
- public CompletableFuture<Void> closeAsync() {
- session.touch();
-
- return dataCursor.closeAsync();
- }
- }
- );
- });
- });
+ private AsyncSqlCursor<List<Object>> executePlan(Session session, InternalTransaction tx, QueryPlan plan, BaseQueryContext ctx) {
+ var dataCursor = executionSrvc.executePlan(tx, plan, ctx);
- stage.whenComplete((cur, ex) -> {
- if (ex instanceof CancellationException) {
- queryCancel.cancel();
- }
+ SqlQueryType queryType = plan.type();
+ assert queryType != null : "Expected a full plan but got a fragment: " + plan;
+
+ return new AsyncSqlCursorImpl<>(
+ queryType,
+ plan.metadata(),
+ tx,
+ new AsyncCursor<>() {
+ @Override
+ public CompletableFuture<BatchedResult<List<Object>>> requestNextAsync(int rows) {
+ session.touch();
+
+ return dataCursor.requestNextAsync(rows);
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ session.touch();
- if (ex != null && outerTx == null) {
- InternalTransaction tx0 = tx.get();
- if (tx0 != null) {
- tx0.rollback();
+ return dataCursor.closeAsync();
+ }
}
- }
- });
+ );
+ }
- start.completeAsync(() -> null, taskExecutor);
+ private static void handleQueryException(Throwable th, QueryCancel queryCancel, @Nullable InternalTransaction tx) {
+ if (th instanceof CancellationException) {
+ queryCancel.cancel();
+ }
- return stage;
+ if (th != null) {
+ if (tx != null) {
+ tx.rollback();
+ }
+ }
}
- private SchemaPlus resolveSchema(String schemaName) {
+ private SchemaPlus resolveSchema(String schemaName, HybridTimestamp timestamp) {
SchemaPlus schema = sqlSchemaManager.schema(schemaName);
if (schema == null) {
@@ -553,7 +593,7 @@ public class SqlQueryProcessor implements QueryProcessor {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
+ public CompletableFuture<Boolean> notify(TableEventParameters parameters, @Nullable Throwable exception) {
return schemaHolder.onTableCreated(
// TODO: https://issues.apache.org/jira/browse/IGNITE-17694 Hardcoded schemas
DEFAULT_SCHEMA_NAME,
@@ -571,7 +611,7 @@ public class SqlQueryProcessor implements QueryProcessor {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
+ public CompletableFuture<Boolean> notify(TableEventParameters parameters, @Nullable Throwable exception) {
return schemaHolder.onTableUpdated(
// TODO: https://issues.apache.org/jira/browse/IGNITE-17694 Hardcoded schemas
DEFAULT_SCHEMA_NAME,
@@ -589,7 +629,7 @@ public class SqlQueryProcessor implements QueryProcessor {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
+ public CompletableFuture<Boolean> notify(TableEventParameters parameters, @Nullable Throwable exception) {
return schemaHolder.onTableDropped(
// TODO: https://issues.apache.org/jira/browse/IGNITE-17694 Hardcoded schemas
DEFAULT_SCHEMA_NAME,
@@ -607,7 +647,7 @@ public class SqlQueryProcessor implements QueryProcessor {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> notify(@NotNull IndexEventParameters parameters, @Nullable Throwable exception) {
+ public CompletableFuture<Boolean> notify(IndexEventParameters parameters, @Nullable Throwable exception) {
return schemaHolder.onIndexDropped(
// TODO: https://issues.apache.org/jira/browse/IGNITE-17694 Hardcoded schemas
DEFAULT_SCHEMA_NAME,
@@ -626,7 +666,7 @@ public class SqlQueryProcessor implements QueryProcessor {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> notify(@NotNull IndexEventParameters parameters, @Nullable Throwable exception) {
+ public CompletableFuture<Boolean> notify(IndexEventParameters parameters, @Nullable Throwable exception) {
return schemaHolder.onIndexCreated(
parameters.tableId(),
parameters.indexId(),
@@ -645,7 +685,6 @@ public class SqlQueryProcessor implements QueryProcessor {
/** Performs additional validation of a parsed statement. **/
private static void validateParsedStatement(
QueryContext context,
- InternalTransaction outerTx,
ParseResult parseResult,
SqlNode node,
Object[] params
@@ -665,10 +704,6 @@ public class SqlQueryProcessor implements QueryProcessor {
throw new QueryValidationException(message);
}
- if (SqlQueryType.DDL == queryType && outerTx != null) {
- throw new SqlException(UNSUPPORTED_DDL_OPERATION_ERR, "DDL doesn't support transactions.");
- }
-
if (parseResult.dynamicParamsCount() != params.length) {
String message = format(
"Unexpected number of query parameters. Provided {} but there is only {} dynamic parameter(s).",
@@ -687,14 +722,4 @@ public class SqlQueryProcessor implements QueryProcessor {
}
}
}
-
- private static final class PlanWithContext {
- private final QueryPlan plan;
- private final BaseQueryContext context;
-
- private PlanWithContext(QueryPlan plan, BaseQueryContext context) {
- this.plan = plan;
- this.context = context;
- }
- }
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/CacheKey.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/CacheKey.java
index b6a1c023aa..58039814e9 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/CacheKey.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/CacheKey.java
@@ -21,27 +21,33 @@ import java.util.Arrays;
/**
* CacheKey.
- * The class uses to distinguish different query plans which could be various for the same query text but different context. As example such
- * context could be schema name, dynamic parameters, and so on...
+ * The class uses to distinguish different query plans which could be various for the same query text but different context.
+ * For example, the context could contains a catalog version, a default schema id, dynamic parameters, and so on.
*/
public class CacheKey {
private static final Class[] EMPTY_CLASS_ARRAY = {};
- private final String schemaName;
+ private final int catalogVersion;
+
+ private final int schemaId;
private final String query;
- private final Class[] paramTypes;
+ private final Class<?>[] paramTypes;
+
/**
* Constructor.
*
- * @param schemaName Schema name.
- * @param query Query string.
- * @param params Dynamic parameters.
+ * @param catalogVersion Catalog version.
+ * @param schemaId Schema id.
+ * @param query Query string.
+ * @param params Dynamic parameters.
*/
- public CacheKey(String schemaName, String query, Object[] params) {
- this.schemaName = schemaName;
+ public CacheKey(int catalogVersion, int schemaId, String query, Object[] params) {
+ //TODO: IGNITE-17765 use schema id instead of name.
+ this.schemaId = schemaId;
+ this.catalogVersion = catalogVersion;
this.query = query;
this.paramTypes = params.length == 0
? EMPTY_CLASS_ARRAY
@@ -60,21 +66,24 @@ public class CacheKey {
CacheKey cacheKey = (CacheKey) o;
- if (!schemaName.equals(cacheKey.schemaName)) {
+ if (catalogVersion != cacheKey.catalogVersion){
+ return false;
+ }
+ if (schemaId != cacheKey.schemaId) {
return false;
}
if (!query.equals(cacheKey.query)) {
return false;
}
- return Arrays.deepEquals(paramTypes, cacheKey.paramTypes);
+ return Arrays.equals(paramTypes, cacheKey.paramTypes);
}
@Override
public int hashCode() {
- int result = schemaName.hashCode();
+ int result = schemaId;
result = 31 * result + query.hashCode();
- result = 31 * result + Arrays.deepHashCode(paramTypes);
+ result = 31 * result + Arrays.hashCode(paramTypes);
return result;
}
}