You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ko...@apache.org on 2024/05/21 08:18:55 UTC
(ignite-3) branch main updated: IGNITE-22278 Sql. Refactor base and planning contexts (#3787)
This is an automated email from the ASF dual-hosted git repository.
korlov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 32295284ee IGNITE-22278 Sql. Refactor base and planning contexts (#3787)
32295284ee is described below
commit 32295284ee3caf212619bfe283a7763c135a309b
Author: korlov42 <ko...@gridgain.com>
AuthorDate: Tue May 21 11:18:49 2024 +0300
IGNITE-22278 Sql. Refactor base and planning contexts (#3787)
---
.../internal/sql/engine/SqlOperationContext.java | 189 +++++++++++++
.../internal/sql/engine/SqlQueryProcessor.java | 95 ++++---
.../sql/engine/exec/ExecutableTableRegistry.java | 4 +-
.../engine/exec/ExecutableTableRegistryImpl.java | 4 +-
.../engine/exec/ExecutionDependencyResolver.java | 2 +-
.../exec/ExecutionDependencyResolverImpl.java | 14 +-
.../internal/sql/engine/exec/ExecutionService.java | 6 +-
.../sql/engine/exec/ExecutionServiceImpl.java | 108 ++++----
.../sql/engine/externalize/RelJsonReader.java | 10 +-
.../sql/engine/externalize/RelOptSchemaImpl.java | 174 ++++++++++++
.../sql/engine/message/QueryStartRequest.java | 20 +-
.../sql/engine/prepare/KeyValueGetPlan.java | 8 +-
.../sql/engine/prepare/KeyValueModifyPlan.java | 8 +-
.../internal/sql/engine/prepare/MultiStepPlan.java | 10 +-
.../sql/engine/prepare/PlanningContext.java | 154 ++++++++++-
.../sql/engine/prepare/PrepareService.java | 4 +-
.../sql/engine/prepare/PrepareServiceImpl.java | 69 +++--
.../internal/sql/engine/schema/IgniteSchema.java | 12 +-
.../sql/engine/schema/SqlSchemaManager.java | 18 +-
.../sql/engine/schema/SqlSchemaManagerImpl.java | 24 +-
.../internal/sql/engine/util/BaseQueryContext.java | 304 ---------------------
.../ignite/internal/sql/engine/util/Commons.java | 16 +-
.../exec/ExecutionDependencyResolverSelfTest.java | 2 +-
.../sql/engine/exec/ExecutionServiceImplTest.java | 131 ++++-----
.../engine/exec/NoOpExecutableTableRegistry.java | 2 +-
.../engine/exec/mapping/FragmentMappingTest.java | 6 +-
.../sql/engine/exec/mapping/MappingTestRunner.java | 2 +-
.../engine/framework/PredefinedSchemaManager.java | 8 +-
.../sql/engine/framework/TestBuilders.java | 8 +-
.../internal/sql/engine/framework/TestNode.java | 21 +-
.../sql/engine/planner/AbstractPlannerTest.java | 105 +++----
.../internal/sql/engine/planner/PlannerTest.java | 15 +-
.../sql/engine/planner/PlannerTimeoutTest.java | 15 +-
.../sql/engine/prepare/PrepareServiceImplTest.java | 58 ++--
.../ddl/AbstractDdlSqlToCommandConverterTest.java | 9 +-
.../engine/schema/SqlSchemaManagerImplTest.java | 2 +-
.../sql/metrics/PlanningCacheMetricsTest.java | 28 +-
37 files changed, 922 insertions(+), 743 deletions(-)
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlOperationContext.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlOperationContext.java
new file mode 100644
index 0000000000..0fce23b795
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlOperationContext.java
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine;
+
+import static java.util.Objects.requireNonNull;
+
+import java.time.ZoneId;
+import java.util.UUID;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.sql.engine.SqlQueryProcessor.PrefetchCallback;
+import org.apache.ignite.internal.util.ArrayUtils;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Context of the sql operation.
+ *
+ * <p>Encloses parameters provided by user as well as default substitutions of omitted parameters required for initialization and
+ * kick-starting of the statement execution.
+ */
+public final class SqlOperationContext {
+ private final UUID queryId;
+ private final ZoneId timeZoneId;
+ private final Object[] parameters;
+ private final HybridTimestamp operationTime;
+
+ private final @Nullable QueryCancel cancel;
+ private final @Nullable String defaultSchemaName;
+ private final @Nullable PrefetchCallback prefetchCallback;
+
+ /**
+ * Private constructor, used by a builder.
+ */
+ private SqlOperationContext(
+ UUID queryId,
+ ZoneId timeZoneId,
+ Object[] parameters,
+ HybridTimestamp operationTime,
+ @Nullable QueryCancel cancel,
+ @Nullable String defaultSchemaName,
+ @Nullable PrefetchCallback prefetchCallback
+ ) {
+ this.queryId = queryId;
+ this.cancel = cancel;
+ this.parameters = parameters;
+ this.prefetchCallback = prefetchCallback;
+ this.timeZoneId = timeZoneId;
+ this.operationTime = operationTime;
+ this.defaultSchemaName = defaultSchemaName;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /** Returns unique identifier of the query. */
+ public UUID queryId() {
+ return queryId;
+ }
+
+ /** Returns parameters provided by user required to execute the statement. May be empty but never null. */
+ public Object[] parameters() {
+ return parameters;
+ }
+
+ /**
+ * Returns callback to notify about readiness of the first page of the results.
+ *
+ * <p>May be null on remote side, but never null on node initiator.
+ */
+ public @Nullable PrefetchCallback prefetchCallback() {
+ return prefetchCallback;
+ }
+
+ /**
+ * Returns handler to cancel running operation, as well as add listeners to be notified about cancellation of the query.
+ *
+ * <p>May be null on remote side, but never null on node initiator.
+ */
+ public @Nullable QueryCancel cancel() {
+ return cancel;
+ }
+
+ /** Returns zone id provided by user, to adjust temporal values to match user's time zone during to string conversion. */
+ public ZoneId timeZoneId() {
+ return timeZoneId;
+ }
+
+ /**
+ * Returns name of the schema to use to resolve schema objects, like tables or system views, for which name of the schema was omitted.
+ *
+ * <p>May be null on remote side, but never null on node initiator.
+ */
+ public @Nullable String defaultSchemaName() {
+ return defaultSchemaName;
+ }
+
+ /**
+ * Returns the operation time.
+ *
+ * <p>The time the operation started is the logical time it runs, and all the time readings during the execution time as well as all
+ * time-related operations (like acquiring an active schema) should reflect the same time.
+ */
+ public HybridTimestamp operationTime() {
+ return operationTime;
+ }
+
+ /**
+ * Query context builder.
+ */
+ @SuppressWarnings("PublicInnerClass")
+ public static class Builder {
+ private QueryCancel cancel = new QueryCancel();
+
+ private UUID queryId;
+
+ private Object[] parameters = ArrayUtils.OBJECT_EMPTY_ARRAY;
+
+ private ZoneId timeZoneId;
+
+ private @Nullable PrefetchCallback prefetchCallback;
+
+ private HybridTimestamp operationTime;
+
+ private String defaultSchemaName;
+
+ public Builder cancel(QueryCancel cancel) {
+ this.cancel = requireNonNull(cancel);
+ return this;
+ }
+
+ public Builder queryId(UUID queryId) {
+ this.queryId = requireNonNull(queryId);
+ return this;
+ }
+
+ public Builder prefetchCallback(PrefetchCallback prefetchCallback) {
+ this.prefetchCallback = prefetchCallback;
+ return this;
+ }
+
+ public Builder parameters(Object... parameters) {
+ this.parameters = requireNonNull(parameters);
+ return this;
+ }
+
+ public Builder timeZoneId(ZoneId timeZoneId) {
+ this.timeZoneId = timeZoneId;
+ return this;
+ }
+
+ public Builder defaultSchemaName(String defaultSchemaName) {
+ this.defaultSchemaName = defaultSchemaName;
+ return this;
+ }
+
+ public Builder operationTime(HybridTimestamp operationTime) {
+ this.operationTime = operationTime;
+ return this;
+ }
+
+ /** Creates new context. */
+ public SqlOperationContext build() {
+ return new SqlOperationContext(
+ requireNonNull(queryId, "queryId"),
+ requireNonNull(timeZoneId, "timeZoneId"),
+ requireNonNull(parameters, "parameters"),
+ requireNonNull(operationTime, "operationTime"),
+ cancel,
+ defaultSchemaName,
+ prefetchCallback
+ );
+ }
+ }
+}
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 2d5d7d7793..c857f25e62 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
@@ -23,7 +23,6 @@ import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.lang.SqlExceptionMapperUtil.mapToPublicSqlException;
import static org.apache.ignite.internal.sql.engine.tx.ScriptTransactionContext.NOOP_TX_WRAPPER;
-import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
import static org.apache.ignite.internal.table.distributed.storage.InternalTableImpl.AWAIT_PRIMARY_REPLICA_TIMEOUT;
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
@@ -55,8 +54,6 @@ import java.util.function.Function;
import java.util.function.LongFunction;
import java.util.function.LongSupplier;
import java.util.stream.Collectors;
-import org.apache.calcite.schema.SchemaPlus;
-import org.apache.calcite.tools.Frameworks;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
import org.apache.ignite.internal.failure.FailureProcessor;
@@ -115,7 +112,6 @@ import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapper;
import org.apache.ignite.internal.sql.engine.tx.ScriptTransactionContext;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.sql.engine.util.TypeUtils;
import org.apache.ignite.internal.sql.engine.util.cache.Cache;
@@ -133,7 +129,6 @@ import org.apache.ignite.internal.util.AsyncWrapper;
import org.apache.ignite.internal.util.ExceptionUtils;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.lang.ErrorGroups.Sql;
-import org.apache.ignite.lang.SchemaNotFoundException;
import org.apache.ignite.sql.ResultSetMetadata;
import org.apache.ignite.sql.SqlException;
import org.apache.ignite.tx.IgniteTransactions;
@@ -295,7 +290,8 @@ public class SqlQueryProcessor implements QueryProcessor {
dataStorageManager,
metricManager,
clusterCfg,
- nodeCfg
+ nodeCfg,
+ sqlSchemaManager
));
var msgSrvc = registerService(new MessageServiceImpl(
@@ -579,7 +575,17 @@ public class SqlQueryProcessor implements QueryProcessor {
));
}
- return executeParsedStatement(schemaName, result, txWrapper, queryCancel, timeZoneId, params, null)
+ SqlOperationContext operationContext = SqlOperationContext.builder()
+ .queryId(UUID.randomUUID())
+ .cancel(queryCancel)
+ .prefetchCallback(new PrefetchCallback())
+ .parameters(params)
+ .timeZoneId(timeZoneId)
+ .defaultSchemaName(schemaName)
+ .operationTime(tx.startTimestamp())
+ .build();
+
+ return executeParsedStatement(operationContext, result, txWrapper, null)
.handle((executionResult, e) -> {
if (tx.isReadOnly()) {
transactionInflights.removeInflight(txWrapper.unwrap().id());
@@ -625,11 +631,15 @@ public class SqlQueryProcessor implements QueryProcessor {
QueryCancel queryCancel,
Object[] params) {
- return waitForActualSchema(schemaName, timestamp)
+ return waitForMetadata(timestamp)
.thenCompose(schema -> {
- BaseQueryContext ctx = BaseQueryContext.builder()
- .frameworkConfig(Frameworks.newConfigBuilder(FRAMEWORK_CONFIG).defaultSchema(schema).build())
+ SqlOperationContext ctx = SqlOperationContext.builder()
.queryId(UUID.randomUUID())
+ // time zone is used in execution phase,
+ // so we may use any time zone for preparation only
+ .timeZoneId(DEFAULT_TIME_ZONE_ID)
+ .defaultSchemaName(schemaName)
+ .operationTime(timestamp)
.cancel(queryCancel)
.parameters(params)
.build();
@@ -639,30 +649,16 @@ public class SqlQueryProcessor implements QueryProcessor {
}
private CompletableFuture<AsyncSqlCursor<InternalSqlRow>> executeParsedStatement(
- String schemaName,
+ SqlOperationContext operationContext,
ParsedResult parsedResult,
QueryTransactionWrapper txWrapper,
- QueryCancel queryCancel,
- ZoneId timeZoneId,
- Object[] params,
@Nullable CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextStatement
) {
- return waitForActualSchema(schemaName, txWrapper.unwrap().startTimestamp())
- .thenCompose(schema -> {
- PrefetchCallback callback = new PrefetchCallback();
-
- BaseQueryContext ctx = BaseQueryContext.builder()
- .frameworkConfig(Frameworks.newConfigBuilder(FRAMEWORK_CONFIG).defaultSchema(schema).build())
- .queryId(UUID.randomUUID())
- .cancel(queryCancel)
- .prefetchCallback(callback)
- .parameters(params)
- .timeZoneId(timeZoneId)
- .build();
+ HybridTimestamp operationTime = txWrapper.unwrap().startTimestamp();
- return prepareSvc.prepareAsync(parsedResult, ctx)
- .thenCompose(plan -> executePlan(txWrapper, ctx, callback, plan, nextStatement));
- })
+ return waitForMetadata(operationTime)
+ .thenCompose(none -> prepareSvc.prepareAsync(parsedResult, operationContext)
+ .thenCompose(plan -> executePlan(txWrapper, operationContext, plan, nextStatement)))
.whenComplete((res, ex) -> {
if (ex != null) {
txWrapper.rollback(ex);
@@ -670,26 +666,13 @@ public class SqlQueryProcessor implements QueryProcessor {
});
}
- private CompletableFuture<SchemaPlus> waitForActualSchema(String schemaName, HybridTimestamp timestamp) {
- try {
- return schemaSyncService.waitForMetadataCompleteness(timestamp).thenApply(unused -> {
- SchemaPlus schema = sqlSchemaManager.schema(timestamp.longValue()).getSubSchema(schemaName);
-
- if (schema == null) {
- throw new SchemaNotFoundException(schemaName);
- }
-
- return schema;
- });
- } catch (Throwable t) {
- return failedFuture(t);
- }
+ private CompletableFuture<Void> waitForMetadata(HybridTimestamp timestamp) {
+ return schemaSyncService.waitForMetadataCompleteness(timestamp);
}
private CompletableFuture<AsyncSqlCursor<InternalSqlRow>> executePlan(
QueryTransactionWrapper txWrapper,
- BaseQueryContext ctx,
- PrefetchCallback callback,
+ SqlOperationContext ctx,
QueryPlan plan,
@Nullable CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextStatement
) {
@@ -703,12 +686,16 @@ public class SqlQueryProcessor implements QueryProcessor {
SqlQueryType queryType = plan.type();
UUID queryId = ctx.queryId();
+ PrefetchCallback prefetchCallback = ctx.prefetchCallback();
+
+ assert prefetchCallback != null;
+
AsyncSqlCursorImpl<InternalSqlRow> cursor = new AsyncSqlCursorImpl<>(
queryType,
plan.metadata(),
txWrapper,
dataCursor,
- callback.prefetchFuture(),
+ prefetchCallback.prefetchFuture(),
nextStatement
);
@@ -911,7 +898,17 @@ public class SqlQueryProcessor implements QueryProcessor {
txCtx.registerCursorFuture(parsedResult.queryType(), cursorFuture);
- fut = executeParsedStatement(schemaName, parsedResult, txWrapper, new QueryCancel(), timeZoneId, params, nextCurFut);
+ SqlOperationContext operationContext = SqlOperationContext.builder()
+ .queryId(UUID.randomUUID())
+ .cancel(new QueryCancel())
+ .prefetchCallback(new PrefetchCallback())
+ .parameters(params)
+ .timeZoneId(timeZoneId)
+ .defaultSchemaName(schemaName)
+ .operationTime(txWrapper.unwrap().startTimestamp())
+ .build();
+
+ fut = executeParsedStatement(operationContext, parsedResult, txWrapper, nextCurFut);
}
fut.whenComplete((cursor, ex) -> {
@@ -1049,7 +1046,7 @@ public class SqlQueryProcessor implements QueryProcessor {
}
/** Completes the provided future when the callback is called. */
- private static class PrefetchCallback implements QueryPrefetchCallback {
+ public static class PrefetchCallback implements QueryPrefetchCallback {
private final CompletableFuture<Void> prefetchFuture = new CompletableFuture<>();
@Override
@@ -1061,7 +1058,7 @@ public class SqlQueryProcessor implements QueryProcessor {
}
}
- CompletableFuture<Void> prefetchFuture() {
+ public CompletableFuture<Void> prefetchFuture() {
return prefetchFuture;
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistry.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistry.java
index b33c667aed..af0c11cb61 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistry.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistry.java
@@ -27,9 +27,9 @@ public interface ExecutableTableRegistry {
/**
* Returns an table that can be used for both read and write operations.
*
- * @param schemaVersion Version of the schema.
+ * @param catalogVersion Version of the catalog.
* @param tableId Table Id.
* @return An operation that returns executable table.
*/
- CompletableFuture<ExecutableTable> getTable(int schemaVersion, int tableId);
+ CompletableFuture<ExecutableTable> getTable(int catalogVersion, int tableId);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistryImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistryImpl.java
index 3a7da6dd1a..5d85417746 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistryImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistryImpl.java
@@ -74,8 +74,8 @@ public class ExecutableTableRegistryImpl implements ExecutableTableRegistry {
/** {@inheritDoc} */
@Override
- public CompletableFuture<ExecutableTable> getTable(int schemaVersion, int tableId) {
- IgniteTable sqlTable = sqlSchemaManager.table(schemaVersion, tableId);
+ public CompletableFuture<ExecutableTable> getTable(int catalogVersion, int tableId) {
+ IgniteTable sqlTable = sqlSchemaManager.table(catalogVersion, tableId);
return tableCache.computeIfAbsent(cacheKey(tableId, sqlTable.version()), (k) -> loadTable(sqlTable));
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolver.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolver.java
index c86475a561..35def703ea 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolver.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolver.java
@@ -29,5 +29,5 @@ public interface ExecutionDependencyResolver {
/**
* Resolves dependencies required to execute the given list of relations.
*/
- CompletableFuture<ResolvedDependencies> resolveDependencies(Iterable<IgniteRel> rels, int schemaVersion);
+ CompletableFuture<ResolvedDependencies> resolveDependencies(Iterable<IgniteRel> rels, int catalogVersion);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java
index 804e073f89..135441e809 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java
@@ -59,7 +59,7 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
* {@inheritDoc}
*/
@Override
- public CompletableFuture<ResolvedDependencies> resolveDependencies(Iterable<IgniteRel> rels, int schemaVersion) {
+ public CompletableFuture<ResolvedDependencies> resolveDependencies(Iterable<IgniteRel> rels, int catalogVersion) {
Map<Integer, CompletableFuture<ExecutableTable>> tableMap = new HashMap<>();
Map<Integer, ScannableDataSource> dataSources = new HashMap<>();
@@ -86,7 +86,7 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
public IgniteRel visit(IgniteTableModify rel) {
IgniteTable igniteTable = rel.getTable().unwrapOrThrow(IgniteTable.class);
- resolveTable(schemaVersion, igniteTable.id());
+ resolveTable(catalogVersion, igniteTable.id());
return super.visit(rel);
}
@@ -95,7 +95,7 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
public IgniteRel visit(IgniteIndexScan rel) {
IgniteTable igniteTable = rel.getTable().unwrapOrThrow(IgniteTable.class);
- resolveTable(schemaVersion, igniteTable.id());
+ resolveTable(catalogVersion, igniteTable.id());
return rel;
}
@@ -104,7 +104,7 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
public IgniteRel visit(IgniteTableScan rel) {
IgniteTable igniteTable = rel.getTable().unwrapOrThrow(IgniteTable.class);
- resolveTable(schemaVersion, igniteTable.id());
+ resolveTable(catalogVersion, igniteTable.id());
return rel;
}
@@ -126,12 +126,12 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
if (function.affinity()) {
int tableId = ((AffinityDistribution) function).tableId();
- resolveTable(schemaVersion, tableId);
+ resolveTable(catalogVersion, tableId);
}
}
- private void resolveTable(int schemaVersion, int tableId) {
- tableMap.computeIfAbsent(tableId, (id) -> registry.getTable(schemaVersion, tableId));
+ private void resolveTable(int catalogVersion, int tableId) {
+ tableMap.computeIfAbsent(tableId, (id) -> registry.getTable(catalogVersion, tableId));
}
};
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionService.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionService.java
index 4351e508b8..960c028edf 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionService.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionService.java
@@ -18,8 +18,8 @@
package org.apache.ignite.internal.sql.engine.exec;
import org.apache.ignite.internal.sql.engine.InternalSqlRow;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.prepare.QueryPlan;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.util.AsyncCursor;
@@ -27,5 +27,7 @@ import org.apache.ignite.internal.util.AsyncCursor;
* SQL query plan execution interface.
*/
public interface ExecutionService extends LifecycleAware {
- AsyncCursor<InternalSqlRow> executePlan(InternalTransaction tx, QueryPlan plan, BaseQueryContext ctx);
+ AsyncCursor<InternalSqlRow> executePlan(
+ InternalTransaction tx, QueryPlan plan, SqlOperationContext operationContext
+ );
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
index 8936c00b1c..67643a7cd4 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.sql.engine.exec;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.sql.engine.externalize.RelJsonReader.fromJson;
-import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
@@ -55,9 +54,10 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collectors;
-import org.apache.calcite.tools.Frameworks;
+import org.apache.calcite.schema.SchemaPlus;
import org.apache.ignite.configuration.ConfigurationChangeException;
import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.IgniteBiTuple;
import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
import org.apache.ignite.internal.lang.IgniteInternalException;
@@ -70,8 +70,10 @@ import org.apache.ignite.internal.sql.engine.InternalSqlRowImpl;
import org.apache.ignite.internal.sql.engine.InternalSqlRowSingleBoolean;
import org.apache.ignite.internal.sql.engine.InternalSqlRowSingleString;
import org.apache.ignite.internal.sql.engine.NodeLeftException;
+import org.apache.ignite.internal.sql.engine.QueryCancel;
import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.QueryPrefetchCallback;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
import org.apache.ignite.internal.sql.engine.exec.ddl.DdlCommandHandler;
import org.apache.ignite.internal.sql.engine.exec.exp.func.TableFunctionRegistry;
@@ -103,7 +105,6 @@ import org.apache.ignite.internal.sql.engine.rel.IgniteTableScan;
import org.apache.ignite.internal.sql.engine.rel.SourceAwareIgniteRel;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.sql.engine.util.TypeUtils;
import org.apache.ignite.internal.tx.InternalTransaction;
@@ -275,37 +276,41 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private AsyncCursor<InternalSqlRow> executeQuery(
InternalTransaction tx,
- BaseQueryContext ctx,
+ SqlOperationContext operationContext,
MultiStepPlan plan
) {
- DistributedQueryManager queryManager = new DistributedQueryManager(localNode.name(), true, ctx);
+ DistributedQueryManager queryManager = new DistributedQueryManager(localNode.name(), true, operationContext);
- DistributedQueryManager old = queryManagerMap.put(ctx.queryId(), queryManager);
+ DistributedQueryManager old = queryManagerMap.put(operationContext.queryId(), queryManager);
assert old == null;
- ctx.cancel().add(() -> queryManager.close(true));
+ QueryCancel cancelHandler = operationContext.cancel();
+
+ assert cancelHandler != null;
+
+ cancelHandler.add(() -> queryManager.close(true));
return queryManager.execute(tx, plan);
}
- private BaseQueryContext createQueryContext(UUID queryId, int schemaVersion, ZoneId timeZoneId, Object[] params) {
- return BaseQueryContext.builder()
+ private static SqlOperationContext createOperationContext(
+ UUID queryId, ZoneId timeZoneId, Object[] params, HybridTimestamp operationTime
+ ) {
+ return SqlOperationContext.builder()
.queryId(queryId)
.parameters(params)
- .frameworkConfig(
- Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(sqlSchemaManager.schema(schemaVersion))
- .build()
- )
.timeZoneId(timeZoneId)
+ .operationTime(operationTime)
.build();
}
- private IgniteRel relationalTreeFromJsonString(int schemaVersion, String jsonFragment, BaseQueryContext ctx) {
+ private IgniteRel relationalTreeFromJsonString(int catalogVersion, String jsonFragment) {
+ SchemaPlus rootSchema = sqlSchemaManager.schema(catalogVersion);
+
return physNodesCache.computeIfAbsent(
- new FragmentCacheKey(schemaVersion, jsonFragment),
- key -> fromJson(ctx, key.fragmentString)
+ new FragmentCacheKey(catalogVersion, jsonFragment),
+ key -> fromJson(rootSchema, key.fragmentString)
);
}
@@ -313,7 +318,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
@Override
@SuppressWarnings("CastConflictsWithInstanceof") // IDEA incorrectly highlights casts in EXPLAIN and DDL branches
public AsyncCursor<InternalSqlRow> executePlan(
- InternalTransaction tx, QueryPlan plan, BaseQueryContext ctx
+ InternalTransaction tx, QueryPlan plan, SqlOperationContext operationContext
) {
SqlQueryType queryType = plan.type();
@@ -321,16 +326,16 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
case DML:
case QUERY:
if (plan instanceof ExecutablePlan) {
- return executeExecutablePlan(tx, ctx, (ExecutablePlan) plan, ctx.prefetchCallback());
+ return executeExecutablePlan(tx, operationContext, (ExecutablePlan) plan, operationContext.prefetchCallback());
}
assert plan instanceof MultiStepPlan : plan.getClass();
- return executeQuery(tx, ctx, (MultiStepPlan) plan);
+ return executeQuery(tx, operationContext, (MultiStepPlan) plan);
case EXPLAIN:
- return executeExplain((ExplainPlan) plan, ctx.prefetchCallback());
+ return executeExplain((ExplainPlan) plan, operationContext.prefetchCallback());
case DDL:
- return executeDdl((DdlPlan) plan, ctx.prefetchCallback());
+ return executeDdl((DdlPlan) plan, operationContext.prefetchCallback());
default:
throw new AssertionError("Unexpected query type: " + plan);
@@ -350,20 +355,20 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private AsyncCursor<InternalSqlRow> executeExecutablePlan(
InternalTransaction tx,
- BaseQueryContext ctx,
+ SqlOperationContext operationContext,
ExecutablePlan plan,
@Nullable QueryPrefetchCallback callback
) {
ExecutionContext<RowT> ectx = new ExecutionContext<>(
taskExecutor,
- ctx.queryId(),
+ operationContext.queryId(),
localNode,
localNode.name(),
DUMMY_DESCRIPTION,
handler,
- Commons.parametersMap(ctx.parameters()),
+ Commons.parametersMap(operationContext.parameters()),
TxAttributes.fromTx(tx),
- ctx.timeZoneId()
+ operationContext.timeZoneId()
);
return plan.execute(ectx, tx, tableRegistry, callback);
@@ -415,7 +420,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private void onMessage(String nodeName, QueryStartRequest msg) {
assert nodeName != null && msg != null;
- CompletableFuture<Void> fut = sqlSchemaManager.schemaReadyFuture(msg.schemaVersion());
+ CompletableFuture<Void> fut = sqlSchemaManager.schemaReadyFuture(msg.catalogVersion());
if (fut.isDone()) {
submitFragment(nodeName, msg);
@@ -519,7 +524,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private void submitFragment(String nodeName, QueryStartRequest msg) {
DistributedQueryManager queryManager = getOrCreateQueryManager(nodeName, msg);
- queryManager.submitFragment(nodeName, msg.schemaVersion(), msg.root(), msg.fragmentDescription(), msg.txAttributes());
+ queryManager.submitFragment(nodeName, msg.catalogVersion(), msg.root(), msg.fragmentDescription(), msg.txAttributes());
}
private void handleError(Throwable ex, String nodeName, QueryStartRequest msg) {
@@ -530,9 +535,11 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private DistributedQueryManager getOrCreateQueryManager(String coordinatorNodeName, QueryStartRequest msg) {
return queryManagerMap.computeIfAbsent(msg.queryId(), key -> {
- BaseQueryContext ctx = createQueryContext(key, msg.schemaVersion(), ZoneId.of(msg.timeZoneId()), msg.parameters());
+ SqlOperationContext operationContext = createOperationContext(
+ key, ZoneId.of(msg.timeZoneId()), msg.parameters(), msg.operationTime()
+ );
- return new DistributedQueryManager(coordinatorNodeName, ctx);
+ return new DistributedQueryManager(coordinatorNodeName, operationContext);
});
}
@@ -669,7 +676,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private final String coordinatorNodeName;
- private final BaseQueryContext ctx;
+ private final SqlOperationContext ctx;
private final CompletableFuture<Void> cancelFut = new CompletableFuture<>();
@@ -686,7 +693,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private DistributedQueryManager(
String coordinatorNodeName,
boolean coordinator,
- BaseQueryContext ctx
+ SqlOperationContext ctx
) {
this.ctx = ctx;
this.coordinator = coordinator;
@@ -713,7 +720,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
}
}
- private DistributedQueryManager(String coordinatorNodeName, BaseQueryContext ctx) {
+ private DistributedQueryManager(String coordinatorNodeName, SqlOperationContext ctx) {
this(coordinatorNodeName, false, ctx);
}
@@ -722,7 +729,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
}
private CompletableFuture<Void> sendFragment(
- String targetNodeName, String serialisedFragment, FragmentDescription desc, TxAttributes txAttributes
+ String targetNodeName, String serialisedFragment, FragmentDescription desc, TxAttributes txAttributes, int catalogVersion
) {
QueryStartRequest request = FACTORY.queryStartRequest()
.queryId(ctx.queryId())
@@ -731,8 +738,9 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
.fragmentDescription(desc)
.parameters(ctx.parameters())
.txAttributes(txAttributes)
- .schemaVersion(ctx.schemaVersion())
+ .catalogVersion(catalogVersion)
.timeZoneId(ctx.timeZoneId().getId())
+ .operationTimeLong(ctx.operationTime().longValue())
.timestampLong(clockService.nowLong())
.build();
@@ -827,7 +835,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private void submitFragment(
String initiatorNode,
- int schemaVersion,
+ int catalogVersion,
String fragmentString,
FragmentDescription desc,
TxAttributes txAttributes
@@ -838,9 +846,9 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
// that is going to be used for fragment execution.
ExecutionContext<RowT> context = createContext(initiatorNode, desc, txAttributes);
Executor exec = (r) -> context.execute(r::run, err -> handleError(err, initiatorNode, desc.fragmentId()));
- IgniteRel treeRoot = relationalTreeFromJsonString(schemaVersion, fragmentString, ctx);
+ IgniteRel treeRoot = relationalTreeFromJsonString(catalogVersion, fragmentString);
- dependencyResolver.resolveDependencies(List.of(treeRoot), schemaVersion)
+ dependencyResolver.resolveDependencies(List.of(treeRoot), catalogVersion)
.thenComposeAsync(deps -> executeFragment(treeRoot, deps, context), exec)
.exceptionally(ex -> {
handleError(ex, initiatorNode, desc.fragmentId());
@@ -936,7 +944,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
for (String nodeName : mappedFragment.nodes()) {
CompletableFuture<Void> resultOfSending =
- sendFragment(nodeName, fragment.serialized(), fragmentDesc, attributes);
+ sendFragment(nodeName, fragment.serialized(), fragmentDesc, attributes, multiStepPlan.catalogVersion());
resultsOfFragmentSending.add(
resultOfSending.handle((ignored, t) -> {
@@ -1123,10 +1131,14 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
queryManagerMap.remove(ctx.queryId());
- try {
- ctx.cancel().cancel();
- } catch (Exception th) {
- LOG.debug("Exception raised while cancel", th);
+ QueryCancel cancelHandler = ctx.cancel();
+
+ if (cancelHandler != null) {
+ try {
+ cancelHandler.cancel();
+ } catch (Exception th) {
+ LOG.debug("Exception raised while cancel", th);
+ }
}
cancelFut.complete(null);
@@ -1231,11 +1243,11 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
}
private static class FragmentCacheKey {
- private final int schemaVersion;
+ private final int catalogVersion;
private final String fragmentString;
- FragmentCacheKey(int schemaVersion, String fragmentString) {
- this.schemaVersion = schemaVersion;
+ FragmentCacheKey(int catalogVersion, String fragmentString) {
+ this.catalogVersion = catalogVersion;
this.fragmentString = fragmentString;
}
@@ -1251,13 +1263,13 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
FragmentCacheKey that = (FragmentCacheKey) o;
- return schemaVersion == that.schemaVersion
+ return catalogVersion == that.catalogVersion
&& fragmentString.equals(that.fragmentString);
}
@Override
public int hashCode() {
- return Objects.hash(schemaVersion, fragmentString);
+ return Objects.hash(catalogVersion, fragmentString);
}
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/externalize/RelJsonReader.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/externalize/RelJsonReader.java
index f5842cb6b8..130a2d908c 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/externalize/RelJsonReader.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/externalize/RelJsonReader.java
@@ -47,13 +47,13 @@ import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.runtime.SqlFunctions;
+import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.sql.SqlAggFunction;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.Pair;
import org.apache.calcite.util.Util;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.sql.engine.prepare.bounds.SearchBounds;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.Commons;
/**
@@ -80,8 +80,8 @@ public class RelJsonReader {
* FromJson.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- public static <T extends RelNode> T fromJson(BaseQueryContext ctx, String json) {
- RelJsonReader reader = new RelJsonReader(ctx.catalogReader());
+ public static <T extends RelNode> T fromJson(SchemaPlus rootSchema, String json) {
+ RelJsonReader reader = new RelJsonReader(rootSchema);
return (T) reader.read(json);
}
@@ -105,8 +105,8 @@ public class RelJsonReader {
* Constructor.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- public RelJsonReader(RelOptSchema relOptSchema) {
- this.relOptSchema = relOptSchema;
+ public RelJsonReader(SchemaPlus rootSchema) {
+ this.relOptSchema = new RelOptSchemaImpl(rootSchema);
relJson = new RelJson();
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/externalize/RelOptSchemaImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/externalize/RelOptSchemaImpl.java
new file mode 100644
index 0000000000..94e289f54c
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/externalize/RelOptSchemaImpl.java
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.externalize;
+
+import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
+
+import java.util.List;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelOptSchema;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelReferentialConstraint;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.schema.ColumnStrategy;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.Table;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.ignite.internal.sql.engine.schema.IgniteDataSource;
+import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.lang.ErrorGroups.Common;
+import org.apache.ignite.sql.SqlException;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Minimal implementation of {@link RelOptSchema} interface.
+ *
+ * <p>This is utility class for deserialization purpose. It's used to resolve {@link RelOptTable}'s by its qualified names. Returns
+ * minimal implementations of {@link RelOptTable} which can be used to {@link RelOptTable#unwrap} instance of {@link IgniteTable} which
+ * is required for composing execution tree.
+ */
+class RelOptSchemaImpl implements RelOptSchema {
+ private final SchemaPlus root;
+
+ RelOptSchemaImpl(SchemaPlus root) {
+ this.root = root;
+ }
+
+ @Override
+ public RelOptTable getTableForMember(List<String> names) {
+ if (nullOrEmpty(names) || names.size() != 2) {
+ throw new SqlException(Common.INTERNAL_ERR, "Expected name of exactly two parts, but was " + names);
+ }
+
+ SchemaPlus schema = root.getSubSchema(names.get(0));
+
+ if (schema == null) {
+ throw new SqlException(Common.INTERNAL_ERR, "Schema with name \"" + names.get(0) + "\" not found");
+ }
+
+ Table table = schema.getTable(names.get(1));
+
+ if (table == null) {
+ throw new SqlException(Common.INTERNAL_ERR, "Table with name " + names + " not found");
+ }
+
+ return new RelOptTableImpl(names, (IgniteDataSource) table);
+ }
+
+ @Override
+ public RelDataTypeFactory getTypeFactory() {
+ return Commons.typeFactory();
+ }
+
+ @Override
+ public void registerRules(RelOptPlanner planner) {
+ throw new AssertionError("Should not be called");
+ }
+
+ private static class RelOptTableImpl implements RelOptTable {
+ private final List<String> qualifiedName;
+ private final IgniteDataSource igniteDataSource;
+
+ private RelOptTableImpl(List<String> qualifiedName, IgniteDataSource igniteDataSource) {
+ this.qualifiedName = qualifiedName;
+ this.igniteDataSource = igniteDataSource;
+ }
+
+ @Override
+ public List<String> getQualifiedName() {
+ // used within explainTerms that is widely used in printing out plan or in equality
+ return qualifiedName;
+ }
+
+ @Override
+ public double getRowCount() {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public RelDataType getRowType() {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public @Nullable RelOptSchema getRelOptSchema() {
+ return null;
+ }
+
+ @Override
+ public RelNode toRel(ToRelContext context) {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public @Nullable List<RelCollation> getCollationList() {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public @Nullable RelDistribution getDistribution() {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public boolean isKey(ImmutableBitSet columns) {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public @Nullable List<ImmutableBitSet> getKeys() {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public @Nullable List<RelReferentialConstraint> getReferentialConstraints() {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public @Nullable Expression getExpression(Class clazz) {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public RelOptTable extend(List<RelDataTypeField> extendedFields) {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public List<ColumnStrategy> getColumnStrategies() {
+ throw new AssertionError("Should not be called");
+ }
+
+ @Override
+ public <C> @Nullable C unwrap(Class<C> clazz) {
+ if (clazz.isAssignableFrom(igniteDataSource.getClass())) {
+ return clazz.cast(igniteDataSource);
+ }
+
+ return null;
+ }
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/QueryStartRequest.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/QueryStartRequest.java
index 05b7c9012e..4d213602b0 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/QueryStartRequest.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/QueryStartRequest.java
@@ -17,6 +17,9 @@
package org.apache.ignite.internal.sql.engine.message;
+import static org.apache.ignite.internal.hlc.HybridTimestamp.hybridTimestamp;
+
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.network.annotations.Marshallable;
import org.apache.ignite.internal.network.annotations.Transferable;
import org.apache.ignite.internal.replicator.message.TimestampAware;
@@ -52,12 +55,25 @@ public interface QueryStartRequest extends TimestampAware, ExecutionContextAware
TxAttributes txAttributes();
/**
- * Return last schema version, just a stub, need to be removed after IGNITE-20633.
+ * Returns the version of the catalog the enclosed fragment was prepared on. Must be used to resolve schema objects
+ * (like tables or system views) during deserialization.
*/
- int schemaVersion();
+ int catalogVersion();
/**
* Session time-zone ID.
*/
String timeZoneId();
+
+ /**
+ * Time of the operation.
+ */
+ long operationTimeLong();
+
+ /**
+ * Time of the operation as {@link HybridTimestamp}.
+ */
+ default HybridTimestamp operationTime() {
+ return hybridTimestamp(operationTimeLong());
+ }
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueGetPlan.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueGetPlan.java
index e61f437255..8bb80bcdd4 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueGetPlan.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueGetPlan.java
@@ -63,20 +63,20 @@ public class KeyValueGetPlan implements ExplainablePlan, ExecutablePlan {
private static final IgniteLogger LOG = Loggers.forClass(KeyValueGetPlan.class);
private final PlanId id;
- private final int schemaVersion;
+ private final int catalogVersion;
private final IgniteKeyValueGet lookupNode;
private final ResultSetMetadata meta;
private final ParameterMetadata parameterMetadata;
KeyValueGetPlan(
PlanId id,
- int schemaVersion,
+ int catalogVersion,
IgniteKeyValueGet lookupNode,
ResultSetMetadata meta,
ParameterMetadata parameterMetadata
) {
this.id = id;
- this.schemaVersion = schemaVersion;
+ this.catalogVersion = catalogVersion;
this.lookupNode = lookupNode;
this.meta = meta;
this.parameterMetadata = parameterMetadata;
@@ -135,7 +135,7 @@ public class KeyValueGetPlan implements ExplainablePlan, ExecutablePlan {
) {
IgniteTable sqlTable = table();
- CompletableFuture<Iterator<InternalSqlRow>> result = tableRegistry.getTable(schemaVersion, sqlTable.id())
+ CompletableFuture<Iterator<InternalSqlRow>> result = tableRegistry.getTable(catalogVersion, sqlTable.id())
.thenCompose(execTable -> {
ImmutableBitSet requiredColumns = lookupNode.requiredColumns();
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueModifyPlan.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueModifyPlan.java
index 0a6df52446..dad27b43ee 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueModifyPlan.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueModifyPlan.java
@@ -49,20 +49,20 @@ import org.jetbrains.annotations.Nullable;
*/
public class KeyValueModifyPlan implements ExplainablePlan, ExecutablePlan {
private final PlanId id;
- private final int schemaVersion;
+ private final int catalogVersion;
private final IgniteKeyValueModify modifyNode;
private final ResultSetMetadata meta;
private final ParameterMetadata parameterMetadata;
KeyValueModifyPlan(
PlanId id,
- int schemaVersion,
+ int catalogVersion,
IgniteKeyValueModify modifyNode,
ResultSetMetadata meta,
ParameterMetadata parameterMetadata
) {
this.id = id;
- this.schemaVersion = schemaVersion;
+ this.catalogVersion = catalogVersion;
this.modifyNode = modifyNode;
this.meta = meta;
this.parameterMetadata = parameterMetadata;
@@ -121,7 +121,7 @@ public class KeyValueModifyPlan implements ExplainablePlan, ExecutablePlan {
) {
IgniteTable sqlTable = table();
- CompletableFuture<Iterator<InternalSqlRow>> result = tableRegistry.getTable(schemaVersion, sqlTable.id())
+ CompletableFuture<Iterator<InternalSqlRow>> result = tableRegistry.getTable(catalogVersion, sqlTable.id())
.thenCompose(execTable -> {
List<RexNode> expressions = modifyNode.expressions();
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/MultiStepPlan.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/MultiStepPlan.java
index 6221733931..eabfbf3c96 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/MultiStepPlan.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/MultiStepPlan.java
@@ -40,14 +40,18 @@ public class MultiStepPlan implements ExplainablePlan {
private final IgniteRel root;
private final ParameterMetadata parameterMetadata;
+ private final int catalogVersion;
/** Constructor. */
- public MultiStepPlan(PlanId id, SqlQueryType type, IgniteRel root, ResultSetMetadata meta, ParameterMetadata parameterMetadata) {
+ public MultiStepPlan(
+ PlanId id, SqlQueryType type, IgniteRel root, ResultSetMetadata meta, ParameterMetadata parameterMetadata, int catalogVersion
+ ) {
this.id = id;
this.type = type;
this.root = root;
this.meta = meta;
this.parameterMetadata = parameterMetadata;
+ this.catalogVersion = catalogVersion;
}
/** {@inheritDoc} */
@@ -82,6 +86,10 @@ public class MultiStepPlan implements ExplainablePlan {
return RelOptUtil.toString(clonedRoot, SqlExplainLevel.ALL_ATTRIBUTES);
}
+ public int catalogVersion() {
+ return catalogVersion;
+ }
+
/** Returns root of the query tree. */
public IgniteRel root() {
return root;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlanningContext.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlanningContext.java
index 6091dff773..46bbda666f 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlanningContext.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlanningContext.java
@@ -17,29 +17,126 @@
package org.apache.ignite.internal.sql.engine.prepare;
+import static org.apache.calcite.tools.Frameworks.createRootSchema;
+import static org.apache.ignite.internal.sql.engine.util.Commons.DISTRIBUTED_TRAITS_SET;
+import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
+
+import com.google.common.collect.Multimap;
import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
import it.unimi.dsi.fastutil.ints.Int2ObjectMaps;
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
+import org.apache.calcite.config.CalciteConnectionConfig;
+import org.apache.calcite.config.CalciteConnectionConfigImpl;
+import org.apache.calcite.config.CalciteConnectionProperty;
+import org.apache.calcite.jdbc.CalciteSchema;
import org.apache.calcite.plan.Context;
-import org.apache.calcite.plan.Contexts;
import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptSchema;
+import org.apache.calcite.plan.RelTraitDef;
+import org.apache.calcite.plan.volcano.VolcanoPlanner;
import org.apache.calcite.prepare.CalciteCatalogReader;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.metadata.Metadata;
+import org.apache.calcite.rel.metadata.MetadataDef;
+import org.apache.calcite.rel.metadata.MetadataHandler;
+import org.apache.calcite.rel.metadata.RelMetadataProvider;
+import org.apache.calcite.rel.metadata.UnboundMetadata;
+import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.sql.SqlOperatorTable;
import org.apache.calcite.sql.validate.SqlConformance;
import org.apache.calcite.tools.FrameworkConfig;
+import org.apache.calcite.tools.Frameworks;
import org.apache.calcite.tools.RuleSet;
import org.apache.calcite.util.CancelFlag;
+import org.apache.ignite.internal.sql.engine.metadata.cost.IgniteCostFactory;
+import org.apache.ignite.internal.sql.engine.rex.IgniteRexBuilder;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
+import org.jetbrains.annotations.Nullable;
/**
* Planning context.
*/
public final class PlanningContext implements Context {
+ private static final CalciteConnectionConfig CALCITE_CONNECTION_CONFIG;
+
+ public static final RelOptCluster CLUSTER;
+
+ private static final IgniteCostFactory COST_FACTORY = new IgniteCostFactory();
+
+ static {
+ Properties props = new Properties();
+
+ props.setProperty(CalciteConnectionProperty.CASE_SENSITIVE.camelName(),
+ String.valueOf(FRAMEWORK_CONFIG.getParserConfig().caseSensitive()));
+ props.setProperty(CalciteConnectionProperty.CONFORMANCE.camelName(),
+ String.valueOf(FRAMEWORK_CONFIG.getParserConfig().conformance()));
+ props.setProperty(CalciteConnectionProperty.MATERIALIZATIONS_ENABLED.camelName(),
+ String.valueOf(true));
+
+ CALCITE_CONNECTION_CONFIG = new CalciteConnectionConfigImpl(props);
+
+ RexBuilder defaultRexBuilder = IgniteRexBuilder.INSTANCE;
+
+ PlanningContext emptyContext = builder().build();
+ VolcanoPlanner planner = new VolcanoPlanner(COST_FACTORY, emptyContext) {
+ @Override
+ public void registerSchema(RelOptSchema schema) {
+ // This method in VolcanoPlanner stores schema in hash map. It can be invoked during relational
+ // operators cloning, so, can be executed even with empty context. Override it for empty context to
+ // prevent memory leaks.
+ }
+ };
+
+ // Dummy planner must contain all trait definitions to create singleton cluster with all default traits.
+ for (RelTraitDef<?> def : DISTRIBUTED_TRAITS_SET) {
+ planner.addRelTraitDef(def);
+ }
+
+ RelOptCluster cluster = RelOptCluster.create(planner, defaultRexBuilder);
+
+ // Forbid using the empty cluster in any planning or mapping procedures to prevent memory leaks.
+ String cantBeUsedMsg = "Empty cluster can't be used for planning or mapping";
+
+ cluster.setMetadataProvider(
+ new RelMetadataProvider() {
+ @Override
+ public <M extends Metadata> UnboundMetadata<M> apply(
+ Class<? extends RelNode> relCls,
+ Class<? extends M> metadataCls
+ ) {
+ throw new AssertionError(cantBeUsedMsg);
+ }
+
+ @Override
+ public <M extends Metadata> Multimap<Method, MetadataHandler<M>> handlers(MetadataDef<M> def) {
+ throw new AssertionError(cantBeUsedMsg);
+ }
+
+ @Override
+ public List<MetadataHandler<?>> handlers(Class<? extends MetadataHandler<?>> hndCls) {
+ throw new AssertionError(cantBeUsedMsg);
+ }
+ }
+ );
+
+ cluster.setMetadataQuerySupplier(() -> {
+ throw new AssertionError(cantBeUsedMsg);
+ });
+
+ CLUSTER = cluster;
+ }
+
private final Context parentCtx;
+ private final FrameworkConfig cfg;
+
private final String qry;
/** CancelFlag is used to post and check cancellation requests. */
@@ -58,15 +155,21 @@ public final class PlanningContext implements Context {
private final Int2ObjectMap<Object> parameters;
+ private @Nullable CalciteCatalogReader catalogReader;
+
/** Private constructor, used by a builder. */
private PlanningContext(
- Context parentCtx,
+ FrameworkConfig config,
String qry,
long plannerTimeout,
Int2ObjectMap<Object> parameters
) {
+ this.parentCtx = config.getContext();
+
+ // link frameworkConfig#context() to this.
+ this.cfg = Frameworks.newConfigBuilder(config).context(this).build();
+
this.qry = qry;
- this.parentCtx = parentCtx;
this.plannerTimeout = plannerTimeout;
this.parameters = parameters;
@@ -74,7 +177,7 @@ public final class PlanningContext implements Context {
/** Get framework config. */
public FrameworkConfig config() {
- return unwrap(BaseQueryContext.class).config();
+ return cfg;
}
/** Get query. */
@@ -120,7 +223,11 @@ public final class PlanningContext implements Context {
/** Get schema. */
public SchemaPlus schema() {
- return config().getDefaultSchema();
+ return Objects.requireNonNull(config().getDefaultSchema());
+ }
+
+ public int catalogVersion() {
+ return Objects.requireNonNull(schema().unwrap(IgniteSchema.class)).catalogVersion();
}
/** Get type factory. */
@@ -128,9 +235,26 @@ public final class PlanningContext implements Context {
return IgniteTypeFactory.INSTANCE;
}
- /** Get new catalog reader. */
+ /**
+ * Returns calcite catalog reader.
+ */
public CalciteCatalogReader catalogReader() {
- return unwrap(BaseQueryContext.class).catalogReader();
+ if (catalogReader != null) {
+ return catalogReader;
+ }
+
+ SchemaPlus dfltSchema = schema();
+ SchemaPlus rootSchema = dfltSchema;
+
+ while (rootSchema.getParentSchema() != null) {
+ rootSchema = rootSchema.getParentSchema();
+ }
+
+ //noinspection NestedAssignment
+ return catalogReader = new CalciteCatalogReader(
+ CalciteSchema.from(rootSchema),
+ CalciteSchema.from(dfltSchema).path(null),
+ IgniteTypeFactory.INSTANCE, CALCITE_CONNECTION_CONFIG);
}
/** Get cluster based on a planner and its configuration. */
@@ -181,8 +305,13 @@ public final class PlanningContext implements Context {
* Planner context builder.
*/
public static class Builder {
+ private static final FrameworkConfig EMPTY_CONFIG =
+ Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
+ .defaultSchema(createRootSchema(false))
+ .traitDefs(DISTRIBUTED_TRAITS_SET)
+ .build();
- private Context parentCtx = Contexts.empty();
+ private FrameworkConfig frameworkConfig = EMPTY_CONFIG;
private String qry;
@@ -190,9 +319,8 @@ public final class PlanningContext implements Context {
private Int2ObjectMap<Object> parameters = Int2ObjectMaps.emptyMap();
- /** Parent context. */
- public Builder parentContext(Context parentCtx) {
- this.parentCtx = parentCtx;
+ public Builder frameworkConfig(FrameworkConfig frameworkCfg) {
+ this.frameworkConfig = Objects.requireNonNull(frameworkCfg);
return this;
}
@@ -220,7 +348,7 @@ public final class PlanningContext implements Context {
* @return Planner context.
*/
public PlanningContext build() {
- return new PlanningContext(parentCtx, qry, plannerTimeout, parameters);
+ return new PlanningContext(frameworkConfig, qry, plannerTimeout, parameters);
}
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareService.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareService.java
index 769bedfc20..0cd83ea8e9 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareService.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareService.java
@@ -18,9 +18,9 @@
package org.apache.ignite.internal.sql.engine.prepare;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.exec.LifecycleAware;
import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
/**
* Preparation service that accepts an AST of the query and returns a prepared query plan.
@@ -35,5 +35,5 @@ public interface PrepareService extends LifecycleAware {
*
* @return Future that contains prepared query plan when completes.
*/
- CompletableFuture<QueryPlan> prepareAsync(ParsedResult parsedResult, BaseQueryContext ctx);
+ CompletableFuture<QueryPlan> prepareAsync(ParsedResult parsedResult, SqlOperationContext ctx);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java
index c6af9da1bc..c527a59b35 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.sql.engine.prepare;
import static org.apache.ignite.internal.sql.engine.prepare.CacheKey.EMPTY_CLASS_ARRAY;
import static org.apache.ignite.internal.sql.engine.prepare.PlannerHelper.optimize;
import static org.apache.ignite.internal.sql.engine.trait.TraitUtils.distributionPresent;
+import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
import static org.apache.ignite.internal.thread.ThreadOperation.NOTHING_ALLOWED;
import static org.apache.ignite.lang.ErrorGroups.Sql.PLANNING_TIMEOUT_ERR;
@@ -38,10 +39,12 @@ import java.util.function.Function;
import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.sql.SqlDdl;
import org.apache.calcite.sql.SqlExplain;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.tools.Frameworks;
import org.apache.ignite.internal.lang.SqlExceptionMapperUtil;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
@@ -50,13 +53,14 @@ import org.apache.ignite.internal.sql.ColumnMetadataImpl;
import org.apache.ignite.internal.sql.ResultSetMetadataImpl;
import org.apache.ignite.internal.sql.configuration.distributed.SqlDistributedConfiguration;
import org.apache.ignite.internal.sql.configuration.local.SqlLocalConfiguration;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
import org.apache.ignite.internal.sql.engine.prepare.ddl.DdlSqlToCommandConverter;
import org.apache.ignite.internal.sql.engine.rel.IgniteKeyValueGet;
import org.apache.ignite.internal.sql.engine.rel.IgniteKeyValueModify;
import org.apache.ignite.internal.sql.engine.rel.IgniteRel;
+import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.Cloner;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.sql.engine.util.TypeUtils;
@@ -68,6 +72,7 @@ import org.apache.ignite.internal.thread.IgniteThreadFactory;
import org.apache.ignite.internal.type.NativeTypeSpec;
import org.apache.ignite.internal.util.ExceptionUtils;
import org.apache.ignite.lang.ErrorGroups.Sql;
+import org.apache.ignite.lang.SchemaNotFoundException;
import org.apache.ignite.sql.ColumnMetadata;
import org.apache.ignite.sql.ColumnType;
import org.apache.ignite.sql.ResultSetMetadata;
@@ -100,8 +105,6 @@ public class PrepareServiceImpl implements PrepareService {
private final String nodeName;
- private volatile ThreadPoolExecutor planningPool;
-
private final long plannerTimeout;
private final int plannerThreadCount;
@@ -110,6 +113,10 @@ public class PrepareServiceImpl implements PrepareService {
private final SqlPlanCacheMetricSource sqlPlanCacheMetricSource;
+ private final SqlSchemaManager schemaManager;
+
+ private volatile ThreadPoolExecutor planningPool;
+
/**
* Factory method.
*
@@ -119,6 +126,7 @@ public class PrepareServiceImpl implements PrepareService {
* @param metricManager Metric manager.
* @param clusterCfg Cluster SQL configuration.
* @param nodeCfg Node SQL configuration.
+ * @param schemaManager Schema manager to use on validation phase to bind identifiers in AST with particular schema objects.
*/
public static PrepareServiceImpl create(
String nodeName,
@@ -126,7 +134,8 @@ public class PrepareServiceImpl implements PrepareService {
DataStorageManager dataStorageManager,
MetricManager metricManager,
SqlDistributedConfiguration clusterCfg,
- SqlLocalConfiguration nodeCfg
+ SqlLocalConfiguration nodeCfg,
+ SqlSchemaManager schemaManager
) {
return new PrepareServiceImpl(
nodeName,
@@ -135,7 +144,8 @@ public class PrepareServiceImpl implements PrepareService {
new DdlSqlToCommandConverter(),
clusterCfg.planner().maxPlanningTime().value(),
nodeCfg.planner().threadCount().value(),
- metricManager
+ metricManager,
+ schemaManager
);
}
@@ -148,6 +158,7 @@ public class PrepareServiceImpl implements PrepareService {
* @param ddlConverter A converter of the DDL-related AST to the actual command.
* @param plannerTimeout Timeout in milliseconds to planning.
* @param metricManager Metric manager.
+ * @param schemaManager Schema manager to use on validation phase to bind identifiers in AST with particular schema objects.
*/
public PrepareServiceImpl(
String nodeName,
@@ -156,17 +167,18 @@ public class PrepareServiceImpl implements PrepareService {
DdlSqlToCommandConverter ddlConverter,
long plannerTimeout,
int plannerThreadCount,
- MetricManager metricManager
+ MetricManager metricManager,
+ SqlSchemaManager schemaManager
) {
this.nodeName = nodeName;
this.ddlConverter = ddlConverter;
this.plannerTimeout = plannerTimeout;
this.metricManager = metricManager;
this.plannerThreadCount = plannerThreadCount;
+ this.schemaManager = schemaManager;
sqlPlanCacheMetricSource = new SqlPlanCacheMetricSource();
cache = cacheFactory.create(cacheSize, sqlPlanCacheMetricSource);
-
}
/** {@inheritDoc} */
@@ -197,14 +209,27 @@ public class PrepareServiceImpl implements PrepareService {
/** {@inheritDoc} */
@Override
- public CompletableFuture<QueryPlan> prepareAsync(ParsedResult parsedResult, BaseQueryContext ctx) {
+ public CompletableFuture<QueryPlan> prepareAsync(
+ ParsedResult parsedResult, SqlOperationContext operationContext
+ ) {
CompletableFuture<QueryPlan> result;
+ String schemaName = operationContext.defaultSchemaName();
+
+ assert schemaName != null;
+
+ SchemaPlus schema = schemaManager.schema(operationContext.operationTime().longValue())
+ .getSubSchema(schemaName);
+
+ if (schema == null) {
+ return CompletableFuture.failedFuture(new SchemaNotFoundException(schemaName));
+ }
+
PlanningContext planningContext = PlanningContext.builder()
- .parentContext(ctx)
+ .frameworkConfig(Frameworks.newConfigBuilder(FRAMEWORK_CONFIG).defaultSchema(schema).build())
.query(parsedResult.originalQuery())
.plannerTimeout(plannerTimeout)
- .parameters(Commons.arrayToMap(ctx.parameters()))
+ .parameters(Commons.arrayToMap(operationContext.parameters()))
.build();
result = prepareAsync0(parsedResult, planningContext);
@@ -339,15 +364,17 @@ public class PrepareServiceImpl implements PrepareService {
ResultSetMetadata resultSetMetadata = resultSetMetadata(validated.dataType(), validated.origins(), validated.aliases());
- if (clonedTree instanceof IgniteKeyValueGet) {
- int schemaVersion = ctx.unwrap(BaseQueryContext.class).schemaVersion();
+ int catalogVersion = ctx.catalogVersion();
+ if (clonedTree instanceof IgniteKeyValueGet) {
return new KeyValueGetPlan(
- nextPlanId(), schemaVersion, (IgniteKeyValueGet) clonedTree, resultSetMetadata, parameterMetadata
+ nextPlanId(), catalogVersion, (IgniteKeyValueGet) clonedTree, resultSetMetadata, parameterMetadata
);
}
- var plan = new MultiStepPlan(nextPlanId(), SqlQueryType.QUERY, clonedTree, resultSetMetadata, parameterMetadata);
+ var plan = new MultiStepPlan(
+ nextPlanId(), SqlQueryType.QUERY, clonedTree, resultSetMetadata, parameterMetadata, catalogVersion
+ );
if (LOG.isDebugEnabled()) {
LOG.debug("Plan prepared: \n{}\n\n{}", parsedResult.originalQuery(), plan.explain());
@@ -408,15 +435,17 @@ public class PrepareServiceImpl implements PrepareService {
// before storing tree in plan cache
IgniteRel clonedTree = Cloner.clone(igniteRel, Commons.emptyCluster());
- if (clonedTree instanceof IgniteKeyValueModify) {
- int schemaVersion = ctx.unwrap(BaseQueryContext.class).schemaVersion();
+ int catalogVersion = ctx.catalogVersion();
+ if (clonedTree instanceof IgniteKeyValueModify) {
return new KeyValueModifyPlan(
- nextPlanId(), schemaVersion, (IgniteKeyValueModify) clonedTree, DML_METADATA, parameterMetadata
+ nextPlanId(), catalogVersion, (IgniteKeyValueModify) clonedTree, DML_METADATA, parameterMetadata
);
}
- return new MultiStepPlan(nextPlanId(), SqlQueryType.DML, clonedTree, DML_METADATA, parameterMetadata);
+ return new MultiStepPlan(
+ nextPlanId(), SqlQueryType.DML, clonedTree, DML_METADATA, parameterMetadata, catalogVersion
+ );
}, planningPool));
return planFut.thenApply(Function.identity());
@@ -474,7 +503,7 @@ public class PrepareServiceImpl implements PrepareService {
// See IgniteSqlValidator::validateInferredDynamicParameters
boolean distributed = distributionPresent(ctx.config().getTraitDefs());
- int catalogVersion = ctx.unwrap(BaseQueryContext.class).schemaVersion();
+ int catalogVersion = ctx.catalogVersion();
ColumnType[] paramTypes;
if (parameters.isEmpty()) {
@@ -503,7 +532,7 @@ public class PrepareServiceImpl implements PrepareService {
ParameterMetadata parameterMetadata) {
boolean distributed = distributionPresent(ctx.config().getTraitDefs());
- int catalogVersion = ctx.unwrap(BaseQueryContext.class).schemaVersion();
+ int catalogVersion = ctx.catalogVersion();
ColumnType[] paramTypes;
List<ParameterType> parameterTypes = parameterMetadata.parameterTypes();
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java
index ff5f521d49..109b1ff431 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java
@@ -35,16 +35,16 @@ public class IgniteSchema extends AbstractSchema {
private final String name;
- private final int version;
+ private final int catalogVersion;
private final Map<String, IgniteDataSource> tableByName;
private final Int2ObjectMap<IgniteDataSource> tableById;
/** Constructor. */
- public IgniteSchema(String name, int version, Collection<? extends IgniteDataSource> tables) {
+ public IgniteSchema(String name, int catalogVersion, Collection<? extends IgniteDataSource> tables) {
this.name = name;
- this.version = version;
+ this.catalogVersion = catalogVersion;
this.tableByName = tables.stream().collect(Collectors.toMap(IgniteDataSource::name, Function.identity()));
this.tableById = tables.stream().collect(CollectionUtils.toIntMapCollector(IgniteDataSource::id, Function.identity()));
}
@@ -54,9 +54,9 @@ public class IgniteSchema extends AbstractSchema {
return name;
}
- /** Schema version. */
- public int version() {
- return version;
+ /** Version of the catalog to which this schema correspond. */
+ public int catalogVersion() {
+ return catalogVersion;
}
/** {@inheritDoc} */
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java
index 2490f88a48..a6d82dff82 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java
@@ -25,28 +25,28 @@ import org.apache.calcite.schema.SchemaPlus;
*/
public interface SqlSchemaManager {
/**
- * Returns root schema of the given version.
+ * Returns root schema derived from catalog of the given version.
*/
- SchemaPlus schema(int version);
+ SchemaPlus schema(int catalogVersion);
/**
- * Returns root schema by the given timestamp.
+ * Returns root schema derived from catalog of version which was considered active at the given timestamp.
*/
SchemaPlus schema(long timestamp);
/**
- * Returns table by given id, which version correspond to the one from schema of given version.
+ * Returns table by given id, which version correspond to the one from catalog of given version.
*
- * @param schemaVersion Version of the schema.
+ * @param catalogVersion Version of the catalog.
* @param tableId An identifier of a table of interest.
* @return A table.
*/
- IgniteTable table(int schemaVersion, int tableId);
+ IgniteTable table(int catalogVersion, int tableId);
/**
- * Returns a future to wait for given SQL schema version readiness.
+ * Returns a future to wait for given catalog version readiness.
*
- * @param version SQL schema version to wait.
+ * @param catalogVersion version of the catalog to wait.
*/
- CompletableFuture<Void> schemaReadyFuture(int version);
+ CompletableFuture<Void> schemaReadyFuture(int catalogVersion);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java
index 2ab61fbe20..f2357c911f 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java
@@ -78,9 +78,9 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
/** {@inheritDoc} */
@Override
- public SchemaPlus schema(int schemaVersion) {
+ public SchemaPlus schema(int catalogVersion) {
return schemaCache.get(
- schemaVersion,
+ catalogVersion,
version -> createRootSchema(catalogManager.catalog(version))
);
}
@@ -96,19 +96,19 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> schemaReadyFuture(int version) {
+ public CompletableFuture<Void> schemaReadyFuture(int catalogVersion) {
// SqlSchemaManager creates SQL schema lazily on-demand, thus waiting for Catalog version is enough.
- if (catalogManager.latestCatalogVersion() >= version) {
+ if (catalogManager.latestCatalogVersion() >= catalogVersion) {
return nullCompletedFuture();
}
- return catalogManager.catalogReadyFuture(version);
+ return catalogManager.catalogReadyFuture(catalogVersion);
}
@Override
- public IgniteTable table(int schemaVersion, int tableId) {
- return tableCache.get(tableCacheKey(schemaVersion, tableId), key -> {
- SchemaPlus rootSchema = schemaCache.get(schemaVersion);
+ public IgniteTable table(int catalogVersion, int tableId) {
+ return tableCache.get(tableCacheKey(catalogVersion, tableId), key -> {
+ SchemaPlus rootSchema = schemaCache.get(catalogVersion);
if (rootSchema != null) {
for (String name : rootSchema.getSubSchemaNames()) {
@@ -128,10 +128,10 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
}
}
- Catalog catalog = catalogManager.catalog(schemaVersion);
+ Catalog catalog = catalogManager.catalog(catalogVersion);
if (catalog == null) {
- throw new IgniteInternalException(Common.INTERNAL_ERR, "Catalog of given version not found: " + schemaVersion);
+ throw new IgniteInternalException(Common.INTERNAL_ERR, "Catalog of given version not found: " + catalogVersion);
}
CatalogTableDescriptor tableDescriptor = catalog.table(tableId);
@@ -144,8 +144,8 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
});
}
- private static long tableCacheKey(int schemaVersion, int tableId) {
- long cacheKey = schemaVersion;
+ private static long tableCacheKey(int catalogVersion, int tableId) {
+ long cacheKey = catalogVersion;
cacheKey <<= 32;
return cacheKey | tableId;
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/BaseQueryContext.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/BaseQueryContext.java
deleted file mode 100644
index 3337a58dbd..0000000000
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/BaseQueryContext.java
+++ /dev/null
@@ -1,304 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.util;
-
-import static org.apache.calcite.tools.Frameworks.createRootSchema;
-import static org.apache.ignite.internal.sql.engine.util.Commons.DISTRIBUTED_TRAITS_SET;
-import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
-
-import com.google.common.collect.Multimap;
-import java.lang.reflect.Method;
-import java.time.ZoneId;
-import java.util.List;
-import java.util.Objects;
-import java.util.Properties;
-import java.util.UUID;
-import org.apache.calcite.config.CalciteConnectionConfig;
-import org.apache.calcite.config.CalciteConnectionConfigImpl;
-import org.apache.calcite.config.CalciteConnectionProperty;
-import org.apache.calcite.jdbc.CalciteSchema;
-import org.apache.calcite.plan.Context;
-import org.apache.calcite.plan.Contexts;
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptSchema;
-import org.apache.calcite.plan.RelTraitDef;
-import org.apache.calcite.plan.volcano.VolcanoPlanner;
-import org.apache.calcite.prepare.CalciteCatalogReader;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.metadata.Metadata;
-import org.apache.calcite.rel.metadata.MetadataDef;
-import org.apache.calcite.rel.metadata.MetadataHandler;
-import org.apache.calcite.rel.metadata.RelMetadataProvider;
-import org.apache.calcite.rel.metadata.UnboundMetadata;
-import org.apache.calcite.rex.RexBuilder;
-import org.apache.calcite.schema.SchemaPlus;
-import org.apache.calcite.tools.FrameworkConfig;
-import org.apache.calcite.tools.Frameworks;
-import org.apache.ignite.internal.sql.engine.QueryCancel;
-import org.apache.ignite.internal.sql.engine.QueryPrefetchCallback;
-import org.apache.ignite.internal.sql.engine.metadata.cost.IgniteCostFactory;
-import org.apache.ignite.internal.sql.engine.rex.IgniteRexBuilder;
-import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
-import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
-import org.apache.ignite.internal.util.ArrayUtils;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Base query context.
- */
-public final class BaseQueryContext implements Context {
- private static final CalciteConnectionConfig CALCITE_CONNECTION_CONFIG;
-
- public static final RelOptCluster CLUSTER;
-
- private static final IgniteCostFactory COST_FACTORY = new IgniteCostFactory();
-
- static {
- Properties props = new Properties();
-
- props.setProperty(CalciteConnectionProperty.CASE_SENSITIVE.camelName(),
- String.valueOf(FRAMEWORK_CONFIG.getParserConfig().caseSensitive()));
- props.setProperty(CalciteConnectionProperty.CONFORMANCE.camelName(),
- String.valueOf(FRAMEWORK_CONFIG.getParserConfig().conformance()));
- props.setProperty(CalciteConnectionProperty.MATERIALIZATIONS_ENABLED.camelName(),
- String.valueOf(true));
-
- CALCITE_CONNECTION_CONFIG = new CalciteConnectionConfigImpl(props);
-
- RexBuilder defaultRexBuilder = IgniteRexBuilder.INSTANCE;
-
- BaseQueryContext emptyContext = builder().queryId(new UUID(0L, 0L)).build();
- VolcanoPlanner planner = new VolcanoPlanner(COST_FACTORY, emptyContext) {
- @Override
- public void registerSchema(RelOptSchema schema) {
- // This method in VolcanoPlanner stores schema in hash map. It can be invoked during relational
- // operators cloning, so, can be executed even with empty context. Override it for empty context to
- // prevent memory leaks.
- }
- };
-
- // Dummy planner must contain all trait definitions to create singleton cluster with all default traits.
- for (RelTraitDef<?> def : DISTRIBUTED_TRAITS_SET) {
- planner.addRelTraitDef(def);
- }
-
- RelOptCluster cluster = RelOptCluster.create(planner, defaultRexBuilder);
-
- // Forbid using the empty cluster in any planning or mapping procedures to prevent memory leaks.
- String cantBeUsedMsg = "Empty cluster can't be used for planning or mapping";
-
- cluster.setMetadataProvider(
- new RelMetadataProvider() {
- @Override
- public <M extends Metadata> UnboundMetadata<M> apply(
- Class<? extends RelNode> relCls,
- Class<? extends M> metadataCls
- ) {
- throw new AssertionError(cantBeUsedMsg);
- }
-
- @Override
- public <M extends Metadata> Multimap<Method, MetadataHandler<M>> handlers(MetadataDef<M> def) {
- throw new AssertionError(cantBeUsedMsg);
- }
-
- @Override
- public List<MetadataHandler<?>> handlers(Class<? extends MetadataHandler<?>> hndCls) {
- throw new AssertionError(cantBeUsedMsg);
- }
- }
- );
-
- cluster.setMetadataQuerySupplier(() -> {
- throw new AssertionError(cantBeUsedMsg);
- });
-
- CLUSTER = cluster;
- }
-
- private final Context parentCtx;
-
- private final FrameworkConfig cfg;
-
- private final QueryCancel cancel;
-
- private final UUID queryId;
-
- private final Object[] parameters;
-
- private final QueryPrefetchCallback prefetchCallback;
-
- private final ZoneId timeZoneId;
-
- private CalciteCatalogReader catalogReader;
-
- /**
- * Private constructor, used by a builder.
- */
- private BaseQueryContext(
- UUID queryId,
- FrameworkConfig cfg,
- QueryCancel cancel,
- Object[] parameters,
- QueryPrefetchCallback prefetchCallback,
- ZoneId timeZoneId
- ) {
- this.parentCtx = Contexts.chain(cfg.getContext());
-
- // link frameworkConfig#context() to this.
- this.cfg = Frameworks.newConfigBuilder(cfg).context(this).build();
-
- this.queryId = queryId;
- this.cancel = cancel;
- this.parameters = parameters;
- this.prefetchCallback = prefetchCallback;
- this.timeZoneId = timeZoneId;
- }
-
- public static Builder builder() {
- return new Builder();
- }
-
- /** {@inheritDoc} */
- @Override public <C> @Nullable C unwrap(Class<C> cls) {
- if (cls == getClass()) {
- return cls.cast(this);
- }
-
- return parentCtx.unwrap(cls);
- }
-
- public UUID queryId() {
- return queryId;
- }
-
- public Object[] parameters() {
- return parameters;
- }
-
- public FrameworkConfig config() {
- return cfg;
- }
-
- public SchemaPlus schema() {
- return cfg.getDefaultSchema();
- }
-
- public int schemaVersion() {
- return Objects.requireNonNull(schema().unwrap(IgniteSchema.class)).version();
- }
-
- public QueryPrefetchCallback prefetchCallback() {
- return prefetchCallback;
- }
-
- /**
- * Returns calcite catalog reader.
- */
- public CalciteCatalogReader catalogReader() {
- if (catalogReader != null) {
- return catalogReader;
- }
-
- SchemaPlus dfltSchema = schema();
- SchemaPlus rootSchema = dfltSchema;
-
- while (rootSchema.getParentSchema() != null) {
- rootSchema = rootSchema.getParentSchema();
- }
-
- return catalogReader = new CalciteCatalogReader(
- CalciteSchema.from(rootSchema),
- CalciteSchema.from(dfltSchema).path(null),
- IgniteTypeFactory.INSTANCE, CALCITE_CONNECTION_CONFIG);
- }
-
- public QueryCancel cancel() {
- return cancel;
- }
-
- public ZoneId timeZoneId() {
- return timeZoneId;
- }
-
- /**
- * Query context builder.
- */
- @SuppressWarnings("PublicInnerClass")
- public static class Builder {
- private static final FrameworkConfig EMPTY_CONFIG =
- Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(createRootSchema(false))
- .traitDefs(DISTRIBUTED_TRAITS_SET)
- .build();
-
- private FrameworkConfig frameworkCfg = EMPTY_CONFIG;
-
- private QueryCancel cancel = new QueryCancel();
-
- private UUID queryId;
-
- private Object[] parameters = ArrayUtils.OBJECT_EMPTY_ARRAY;
-
- private ZoneId timeZoneId;
-
- private QueryPrefetchCallback prefetchCallback;
-
- public Builder frameworkConfig(FrameworkConfig frameworkCfg) {
- this.frameworkCfg = Objects.requireNonNull(frameworkCfg);
- return this;
- }
-
- public Builder cancel(QueryCancel cancel) {
- this.cancel = Objects.requireNonNull(cancel);
- return this;
- }
-
- public Builder queryId(UUID queryId) {
- this.queryId = Objects.requireNonNull(queryId);
- return this;
- }
-
- public Builder prefetchCallback(QueryPrefetchCallback prefetchCallback) {
- this.prefetchCallback = prefetchCallback;
- return this;
- }
-
- public Builder parameters(Object... parameters) {
- this.parameters = Objects.requireNonNull(parameters);
- return this;
- }
-
- public Builder timeZoneId(ZoneId timeZoneId) {
- this.timeZoneId = timeZoneId;
- return this;
- }
-
- /** Creates new context. */
- public BaseQueryContext build() {
- return new BaseQueryContext(
- Objects.requireNonNull(queryId, "queryId"),
- frameworkCfg,
- cancel,
- parameters,
- prefetchCallback,
- timeZoneId
- );
- }
- }
-}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/Commons.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/Commons.java
index 2f7f9b185f..3134b788f9 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/Commons.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/Commons.java
@@ -20,7 +20,7 @@ package org.apache.ignite.internal.sql.engine.util;
import static org.apache.calcite.rel.hint.HintPredicates.AGGREGATE;
import static org.apache.calcite.rel.hint.HintPredicates.JOIN;
import static org.apache.ignite.internal.sql.engine.QueryProperty.ALLOWED_QUERY_TYPES;
-import static org.apache.ignite.internal.sql.engine.util.BaseQueryContext.CLUSTER;
+import static org.apache.ignite.internal.sql.engine.prepare.PlanningContext.CLUSTER;
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import static org.apache.ignite.sql.ColumnMetadata.UNDEFINED_PRECISION;
@@ -57,7 +57,6 @@ import org.apache.calcite.DataContexts;
import org.apache.calcite.config.CalciteSystemProperty;
import org.apache.calcite.config.NullCollation;
import org.apache.calcite.linq4j.Ord;
-import org.apache.calcite.plan.Context;
import org.apache.calcite.plan.Contexts;
import org.apache.calcite.plan.ConventionTraitDef;
import org.apache.calcite.plan.RelOptCluster;
@@ -267,22 +266,15 @@ public final class Commons {
/**
* Extracts query context.
*/
- public static BaseQueryContext context(RelNode rel) {
+ public static PlanningContext context(RelNode rel) {
return context(rel.getCluster());
}
/**
* Extracts query context.
*/
- public static BaseQueryContext context(RelOptCluster cluster) {
- return Objects.requireNonNull(cluster.getPlanner().getContext().unwrap(BaseQueryContext.class));
- }
-
- /**
- * Extracts planner context.
- */
- public static PlanningContext context(Context ctx) {
- return Objects.requireNonNull(ctx.unwrap(PlanningContext.class));
+ public static PlanningContext context(RelOptCluster cluster) {
+ return Objects.requireNonNull(cluster.getPlanner().getContext().unwrap(PlanningContext.class));
}
/**
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverSelfTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverSelfTest.java
index 767abc83c9..3ea585858f 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverSelfTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverSelfTest.java
@@ -219,7 +219,7 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
throw new IllegalStateException("Unable to plan: " + sql, e);
}
- return resolver.resolveDependencies(List.of(rel), igniteSchema.version());
+ return resolver.resolveDependencies(List.of(rel), igniteSchema.catalogVersion());
}
void checkDependencies(ResolvedDependencies dependencies, int tableId) {
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
index c21a5778b2..c33c4b7794 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
@@ -20,7 +20,6 @@ package org.apache.ignite.internal.sql.engine.exec;
import static java.util.concurrent.CompletableFuture.allOf;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
-import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrow;
@@ -41,8 +40,6 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
@@ -65,11 +62,9 @@ import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-import org.apache.calcite.jdbc.CalciteSchema;
-import org.apache.calcite.schema.SchemaPlus;
-import org.apache.calcite.tools.Frameworks;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
import org.apache.ignite.internal.failure.FailureProcessor;
@@ -89,7 +84,9 @@ import org.apache.ignite.internal.sql.engine.NodeLeftException;
import org.apache.ignite.internal.sql.engine.QueryCancel;
import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.QueryPrefetchCallback;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.SqlQueryProcessor;
+import org.apache.ignite.internal.sql.engine.SqlQueryProcessor.PrefetchCallback;
import org.apache.ignite.internal.sql.engine.exec.ExecutionServiceImplTest.TestCluster.TestNode;
import org.apache.ignite.internal.sql.engine.exec.ddl.DdlCommandHandler;
import org.apache.ignite.internal.sql.engine.exec.exp.func.TableFunctionRegistry;
@@ -105,6 +102,7 @@ import org.apache.ignite.internal.sql.engine.exec.rel.Outbox;
import org.apache.ignite.internal.sql.engine.exec.rel.ScanNode;
import org.apache.ignite.internal.sql.engine.framework.ArrayRowHandler;
import org.apache.ignite.internal.sql.engine.framework.NoOpTransaction;
+import org.apache.ignite.internal.sql.engine.framework.PredefinedSchemaManager;
import org.apache.ignite.internal.sql.engine.framework.TestBuilders;
import org.apache.ignite.internal.sql.engine.message.ExecutionContextAwareMessage;
import org.apache.ignite.internal.sql.engine.message.MessageListener;
@@ -120,12 +118,10 @@ import org.apache.ignite.internal.sql.engine.rel.IgniteTableScan;
import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
import org.apache.ignite.internal.sql.engine.schema.IgniteSystemView;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
-import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
import org.apache.ignite.internal.sql.engine.sql.ParserService;
import org.apache.ignite.internal.sql.engine.sql.ParserServiceImpl;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.EmptyCacheFactory;
import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
@@ -137,6 +133,7 @@ import org.apache.ignite.lang.ErrorGroups.Common;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.TopologyService;
+import org.apache.ignite.sql.SqlException;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -157,7 +154,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
/** Timeout in ms for stopping execution service.*/
private static final long SHUTDOWN_TIMEOUT = 5_000;
- private static final int SCHEMA_VERSION = -1;
+ private static final int CATALOG_VERSION = 1;
private final List<String> nodeNames = List.of("node_1", "node_2", "node_3");
@@ -175,7 +172,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
.size(1_000_000)
.build();
- private final IgniteSchema schema = new IgniteSchema(SqlCommon.DEFAULT_SCHEMA_NAME, SCHEMA_VERSION, List.of(table));
+ private final IgniteSchema schema = new IgniteSchema(SqlCommon.DEFAULT_SCHEMA_NAME, CATALOG_VERSION, List.of(table));
private final List<CapturingMailboxRegistry> mailboxes = new ArrayList<>();
@@ -200,7 +197,8 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
null,
PLANNING_TIMEOUT,
PLANNING_THREAD_COUNT,
- new MetricManagerImpl()
+ new MetricManagerImpl(),
+ new PredefinedSchemaManager(schema)
);
parserService = new ParserServiceImpl();
@@ -234,7 +232,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
public void testCloseByCursor() throws Exception {
ExecutionService execService = executionServices.get(0);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
@@ -274,7 +272,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
public void testCancelOnInitiator() throws InterruptedException {
ExecutionServiceImpl<?> execService = executionServices.get(0);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
@@ -314,7 +312,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
public void testInitializationFailedOnRemoteNode() throws InterruptedException {
ExecutionService execService = executionServices.get(0);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
@@ -341,7 +339,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
CompletionStage<?> batchFut = cursor.requestNextAsync(1);
- assertTrue(waitForCondition(() -> batchFut.toCompletableFuture().isDone(), TIMEOUT_IN_MS));
+ assertTrue(waitForCondition(() -> batchFut.toCompletableFuture().isDone(), TIMEOUT_IN_MS * 100));
// try gather all possible nodes.
List<AbstractNode<?>> execNodes = executionServices.stream()
@@ -371,7 +369,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
mappingException = new IllegalStateException("Query mapping error");
ExecutionService execService = executionServices.get(0);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
@@ -395,7 +393,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
void cursorCloseCompletesSuccessfullyIfRootWasInitializedWithError() throws InterruptedException {
ExecutionService execService = executionServices.get(0);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
@@ -439,7 +437,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
public void testCancelOnRemote() throws InterruptedException {
ExecutionService execService = executionServices.get(0);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
@@ -481,7 +479,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
public void testErrorOnContextInitialization() {
ExecutionService execService = executionServices.get(0);
- BaseQueryContext ctx = spy(createContext());
+ SqlOperationContext ctx = spy(createContext());
when(ctx.timeZoneId())
.thenCallRealMethod()
.thenThrow(new ExceptionInInitializerError());
@@ -500,7 +498,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
public void testCursorIsClosedAfterAllDataRead() throws InterruptedException {
ExecutionService execService = executionServices.get(0);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
InternalTransaction tx = new NoOpTransaction(nodeNames.get(0));
@@ -527,7 +525,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
public void testCursorIsClosedAfterAllDataRead2() throws InterruptedException {
ExecutionService execService = executionServices.get(0);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
InternalTransaction tx = new NoOpTransaction(nodeNames.get(0));
@@ -549,7 +547,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
public void testCursorIsClosedAfterAllDataReadWithNodeFailure() throws InterruptedException {
ExecutionServiceImpl execService = executionServices.get(0);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
InternalTransaction tx = new NoOpTransaction(nodeNames.get(0));
@@ -617,7 +615,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
public void exceptionArrivingBeforeRootFragmentExecutesDoesNotLeaveQueryHanging() {
ExecutionService execService = executionServices.get(0);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
CountDownLatch queryFailedLatch = new CountDownLatch(1);
@@ -681,33 +679,39 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
AtomicReference<AssertionError> errHolder = new AtomicReference<>();
ExecutionService execService = executionServices.get(0);
- QueryPrefetchCallback prefetchListener = new QueryPrefetchCallback() {
+ Supplier<PrefetchCallback> callbackSupplier = new Supplier<>() {
@Override
- public void onPrefetchComplete(@Nullable Throwable err) {
- try {
- assertThat(err, nullValue());
+ public PrefetchCallback get() {
+ PrefetchCallback prefetchCallback = new PrefetchCallback();
- String sql = queriesQueue.poll();
+ prefetchCallback.prefetchFuture().whenComplete((ignored, err) -> {
+ try {
+ assertThat(err, nullValue());
- assertThat(sql, notNullValue());
+ String sql = queriesQueue.poll();
- BaseQueryContext ctx = createContext(queriesQueue.isEmpty() ? null : this);
- InternalTransaction tx = new NoOpTransaction(nodeNames.get(0));
- QueryPlan plan = prepare(sql, ctx);
+ assertThat(sql, notNullValue());
- resultCursors.add(
- execService.executePlan(tx, plan, ctx)
- );
- } catch (AssertionError e) {
- errHolder.set(e);
- } catch (Throwable t) {
- errHolder.set(new AssertionError(t));
- }
+ SqlOperationContext ctx = createContext(queriesQueue.isEmpty() ? null : get());
+ InternalTransaction tx = new NoOpTransaction(nodeNames.get(0));
+ QueryPlan plan = prepare(sql, ctx);
+
+ resultCursors.add(
+ execService.executePlan(tx, plan, ctx)
+ );
+ } catch (AssertionError e) {
+ errHolder.set(e);
+ } catch (Throwable t) {
+ errHolder.set(new AssertionError(t));
+ }
+ });
+
+ return prefetchCallback;
}
};
// Start statements execution.
- prefetchListener.onPrefetchComplete(null);
+ callbackSupplier.get().onPrefetchComplete(null);
waitForCondition(() -> resultCursors.size() == queries.size(), TIMEOUT_IN_MS);
@@ -730,9 +734,9 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
public void testErrorIsPropagatedToPrefetchCallback() {
ExecutionService execService = executionServices.get(0);
- CompletableFuture<Void> prefetchFut = new CompletableFuture<>();
- IgniteInternalException expectedException = new IgniteInternalException(Common.INTERNAL_ERR, "Expected exception");
- BaseQueryContext ctx = createContext(prefetchFut::completeExceptionally);
+ PrefetchCallback prefetchCallback = new PrefetchCallback();
+ SqlException expectedException = new SqlException(Common.INTERNAL_ERR, "Expected exception");
+ SqlOperationContext ctx = createContext(prefetchCallback);
testCluster.node(nodeNames.get(2)).interceptor((nodeName, msg, original) -> {
if (msg instanceof QueryStartRequest) {
@@ -752,7 +756,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
AsyncCursor<InternalSqlRow> cursor = execService.executePlan(new NoOpTransaction(nodeNames.get(0)), plan, ctx);
- assertThat(prefetchFut, willThrow(equalTo(expectedException)));
+ assertThat(prefetchCallback.prefetchFuture(), willThrow(equalTo(expectedException)));
assertThat(cursor.closeAsync(), willCompleteSuccessfully());
}
@@ -764,7 +768,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
@Test
public void testDebugInfoFormat() throws InterruptedException {
ExecutionServiceImpl<?> execService = executionServices.get(0);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
CountDownLatch startResponseLatch = new CountDownLatch(4);
@@ -862,8 +866,6 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
var exchangeService = new ExchangeServiceImpl(mailboxRegistry, messageService, clockService);
- var schemaManagerMock = mock(SqlSchemaManager.class);
-
var clusterNode = new ClusterNodeImpl(UUID.randomUUID().toString(), nodeName, NetworkAddress.from("127.0.0.1:1111"));
if (nodeName.equals(nodeNames.get(0))) {
@@ -874,18 +876,10 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
when(topologyService.localMember()).thenReturn(clusterNode);
- when(schemaManagerMock.schemaReadyFuture(isA(int.class))).thenReturn(nullCompletedFuture());
-
NoOpExecutableTableRegistry executableTableRegistry = new NoOpExecutableTableRegistry();
ExecutionDependencyResolver dependencyResolver = new ExecutionDependencyResolverImpl(executableTableRegistry, null);
- CalciteSchema rootSch = CalciteSchema.createRootSchema(false);
- rootSch.add(schema.getName(), schema);
- SchemaPlus plus = rootSch.plus().getSubSchema(schema.getName());
-
- when(schemaManagerMock.schema(anyInt())).thenReturn(plus);
-
var targetProvider = new ExecutionTargetProvider() {
@Override
public CompletableFuture<ExecutionTarget> forTable(ExecutionTargetFactory factory, IgniteTable table) {
@@ -916,7 +910,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
messageService,
topologyService,
mappingService,
- schemaManagerMock,
+ new PredefinedSchemaManager(schema),
mock(DdlCommandHandler.class),
taskExecutor,
ArrayRowHandler.INSTANCE,
@@ -934,33 +928,22 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
return executionService;
}
- private BaseQueryContext createContext() {
+ private SqlOperationContext createContext() {
return createContext(null);
}
- private BaseQueryContext createContext(@Nullable QueryPrefetchCallback prefetchCallback) {
- return BaseQueryContext.builder()
+ private SqlOperationContext createContext(@Nullable PrefetchCallback prefetchCallback) {
+ return SqlOperationContext.builder()
.queryId(UUID.randomUUID())
.cancel(new QueryCancel())
.prefetchCallback(prefetchCallback)
- .frameworkConfig(
- Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(wrap(schema))
- .build()
- )
+ .operationTime(new HybridClockImpl().now())
+ .defaultSchemaName(SqlCommon.DEFAULT_SCHEMA_NAME)
.timeZoneId(SqlQueryProcessor.DEFAULT_TIME_ZONE_ID)
.build();
}
- private SchemaPlus wrap(IgniteSchema schema) {
- var schemaPlus = Frameworks.createRootSchema(false);
-
- schemaPlus.add(schema.getName(), schema);
-
- return schemaPlus.getSubSchema(schema.getName());
- }
-
- private QueryPlan prepare(String query, BaseQueryContext ctx) {
+ private QueryPlan prepare(String query, SqlOperationContext ctx) {
ParsedResult parsedResult = parserService.parse(query);
assertEquals(ctx.parameters().length, parsedResult.dynamicParamsCount(), "Invalid number of dynamic parameters");
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java
index c300103102..a06fbcb68a 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java
@@ -26,7 +26,7 @@ import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
public final class NoOpExecutableTableRegistry implements ExecutableTableRegistry {
/** {@inheritDoc} */
@Override
- public CompletableFuture<ExecutableTable> getTable(int schemaVersion, int tableId) {
+ public CompletableFuture<ExecutableTable> getTable(int catalogVersion, int tableId) {
return CompletableFuture.completedFuture(new NoOpExecutableTable(tableId));
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/mapping/FragmentMappingTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/mapping/FragmentMappingTest.java
index 1ddbcc924b..27078417d2 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/mapping/FragmentMappingTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/mapping/FragmentMappingTest.java
@@ -17,7 +17,9 @@
package org.apache.ignite.internal.sql.engine.exec.mapping;
+import static org.apache.calcite.tools.Frameworks.newConfigBuilder;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
+import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import java.nio.file.Path;
@@ -293,7 +295,9 @@ public class FragmentMappingTest extends AbstractPlannerTest {
private IgniteRel parseQuery(IgniteSchema schema, String sqlStmt) {
try {
PlanningContext ctx = PlanningContext.builder()
- .parentContext(baseQueryContext(List.of(schema), null))
+ .frameworkConfig(newConfigBuilder(FRAMEWORK_CONFIG)
+ .defaultSchema(createRootSchema(List.of(schema)).getSubSchema(schema.getName()))
+ .build())
.query(sqlStmt)
.build();
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/mapping/MappingTestRunner.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/mapping/MappingTestRunner.java
index 69700be973..94d1518959 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/mapping/MappingTestRunner.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/mapping/MappingTestRunner.java
@@ -169,7 +169,7 @@ final class MappingTestRunner {
ResultSetMetadataImpl resultSetMetadata = new ResultSetMetadataImpl(Collections.emptyList());
ParameterMetadata parameterMetadata = new ParameterMetadata(Collections.emptyList());
MultiStepPlan multiStepPlan = new MultiStepPlan(new PlanId(UUID.randomUUID(), 1), sqlQueryType, rel,
- resultSetMetadata, parameterMetadata);
+ resultSetMetadata, parameterMetadata, schema.catalogVersion());
String actualText = produceMapping(testDef.nodeName, targetProvider, snapshot, multiStepPlan);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
index 2486358fa5..bccb19189c 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
@@ -46,7 +46,7 @@ public class PredefinedSchemaManager implements SqlSchemaManager {
private final Int2ObjectMap<IgniteTable> tableById;
/** Constructs schema manager from a single schema. */
- PredefinedSchemaManager(IgniteSchema schema) {
+ public PredefinedSchemaManager(IgniteSchema schema) {
this(List.of(schema));
}
@@ -69,7 +69,7 @@ public class PredefinedSchemaManager implements SqlSchemaManager {
/** {@inheritDoc} */
@Override
- public SchemaPlus schema(int version) {
+ public SchemaPlus schema(int catalogVersion) {
return root;
}
@@ -81,12 +81,12 @@ public class PredefinedSchemaManager implements SqlSchemaManager {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> schemaReadyFuture(int version) {
+ public CompletableFuture<Void> schemaReadyFuture(int catalogVersion) {
return nullCompletedFuture();
}
@Override
- public IgniteTable table(int schemaVersion, int tableId) {
+ public IgniteTable table(int catalogVersion, int tableId) {
IgniteTable table = tableById.get(tableId);
if (table == null) {
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
index edc55cd842..25b8bce350 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
@@ -623,9 +623,10 @@ public class TestBuilders {
CatalogManager catalogManager = CatalogTestUtils.createCatalogManagerWithTestUpdateLog(clusterName, clock);
var parserService = new ParserServiceImpl();
+ var schemaManager = new SqlSchemaManagerImpl(catalogManager, CaffeineCacheFactory.INSTANCE, 0);
var prepareService = new PrepareServiceImpl(clusterName, 0, CaffeineCacheFactory.INSTANCE,
new DdlSqlToCommandConverter(), PLANNING_TIMEOUT, PLANNING_THREAD_COUNT,
- mock(MetricManagerImpl.class));
+ mock(MetricManagerImpl.class), schemaManager);
Map<String, List<String>> owningNodesByTableName = new HashMap<>();
for (Entry<String, Map<String, ScannableTable>> entry : nodeName2tableName2table.entrySet()) {
@@ -645,7 +646,6 @@ public class TestBuilders {
ClockWaiter clockWaiter = new ClockWaiter("test", clock);
var ddlHandler = new DdlCommandHandler(catalogManager, new TestClockService(clock, clockWaiter), () -> 100);
- var schemaManager = new SqlSchemaManagerImpl(catalogManager, CaffeineCacheFactory.INSTANCE, 0);
Runnable initClosure = () -> {
assertThat(clockWaiter.startAsync(), willCompleteSuccessfully());
@@ -1356,8 +1356,8 @@ public class TestBuilders {
}
@Override
- public CompletableFuture<ExecutableTable> getTable(int schemaVersion, int tableId) {
- IgniteTable table = schemaManager.table(schemaVersion, tableId);
+ public CompletableFuture<ExecutableTable> getTable(int catalogVersion, int tableId) {
+ IgniteTable table = schemaManager.table(catalogVersion, tableId);
assert table != null;
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
index 245df30048..46523261d2 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.sql.engine.framework;
-import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -27,7 +26,6 @@ import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
-import org.apache.calcite.tools.Frameworks;
import org.apache.ignite.internal.failure.FailureProcessor;
import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridClock;
@@ -39,6 +37,7 @@ import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.sql.SqlCommon;
import org.apache.ignite.internal.sql.engine.InternalSqlRow;
import org.apache.ignite.internal.sql.engine.QueryCancel;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.SqlQueryProcessor;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
import org.apache.ignite.internal.sql.engine.exec.ExchangeService;
@@ -64,7 +63,6 @@ import org.apache.ignite.internal.sql.engine.prepare.QueryPlan;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
import org.apache.ignite.internal.sql.engine.sql.ParserService;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.systemview.api.SystemViewManager;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.util.AsyncCursor;
@@ -81,7 +79,6 @@ import org.jetbrains.annotations.Nullable;
*/
public class TestNode implements LifecycleAware {
private final String nodeName;
- private final SqlSchemaManager schemaManager;
private final PrepareService prepareService;
private final ExecutionService executionService;
private final ParserService parserService;
@@ -114,7 +111,6 @@ public class TestNode implements LifecycleAware {
this.nodeName = nodeName;
this.parserService = parserService;
this.prepareService = prepareService;
- this.schemaManager = schemaManager;
TopologyService topologyService = clusterService.topologyService();
MessagingService messagingService = clusterService.messagingService();
@@ -233,7 +229,7 @@ public class TestNode implements LifecycleAware {
*/
public QueryPlan prepare(String query) {
ParsedResult parsedResult = parserService.parse(query);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
return await(prepareService.prepareAsync(parsedResult, ctx));
}
@@ -265,7 +261,7 @@ public class TestNode implements LifecycleAware {
}
ParsedResult parsedResult = parserService.parse(statement);
- BaseQueryContext ctx = createContext();
+ SqlOperationContext ctx = createContext();
QueryPlan plan = await(prepareService.prepareAsync(parsedResult, ctx));
@@ -279,15 +275,12 @@ public class TestNode implements LifecycleAware {
}
}
- private BaseQueryContext createContext() {
- return BaseQueryContext.builder()
+ private SqlOperationContext createContext() {
+ return SqlOperationContext.builder()
.queryId(UUID.randomUUID())
.cancel(new QueryCancel())
- .frameworkConfig(
- Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(schemaManager.schema(Long.MAX_VALUE).getSubSchema(SqlCommon.DEFAULT_SCHEMA_NAME))
- .build()
- )
+ .operationTime(clock.now())
+ .defaultSchemaName(SqlCommon.DEFAULT_SCHEMA_NAME)
.timeZoneId(SqlQueryProcessor.DEFAULT_TIME_ZONE_ID)
.build();
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java
index d044c18473..1f7482fe1a 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.sql.engine.planner;
-import static org.apache.calcite.tools.Frameworks.createRootSchema;
import static org.apache.calcite.tools.Frameworks.newConfigBuilder;
import static org.apache.ignite.internal.sql.engine.externalize.RelJsonWriter.toJson;
import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
@@ -30,6 +29,7 @@ import static org.junit.jupiter.api.Assertions.fail;
import com.google.common.collect.ImmutableList;
import it.unimi.dsi.fastutil.ints.Int2ObjectArrayMap;
+import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -73,10 +73,13 @@ import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.type.BasicSqlType;
import org.apache.calcite.sql2rel.InitializerContext;
import org.apache.calcite.sql2rel.SqlToRelConverter;
+import org.apache.calcite.tools.Frameworks;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.Util;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.lang.IgniteStringBuilder;
import org.apache.ignite.internal.sql.SqlCommon;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.exec.mapping.IdGenerator;
import org.apache.ignite.internal.sql.engine.exec.mapping.QuerySplitter;
import org.apache.ignite.internal.sql.engine.externalize.RelJsonReader;
@@ -105,7 +108,6 @@ import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
import org.apache.ignite.internal.sql.engine.trait.TraitUtils;
import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.sql.engine.util.StatementChecker;
import org.apache.ignite.internal.testframework.IgniteAbstractTest;
@@ -169,40 +171,6 @@ public abstract class AbstractPlannerTest extends IgniteAbstractTest {
};
}
- interface TestVisitor {
- void visit(RelNode node, int ordinal, RelNode parent);
- }
-
- /**
- * TestRelVisitor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
- */
- public static class TestRelVisitor extends RelVisitor {
- final TestVisitor visitor;
-
- /**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
- */
- TestRelVisitor(TestVisitor visitor) {
- this.visitor = visitor;
- }
-
- /** {@inheritDoc} */
- @Override
- public void visit(RelNode node, int ordinal, RelNode parent) {
- visitor.visit(node, ordinal, parent);
-
- super.visit(node, ordinal, parent);
- }
- }
-
- protected static void relTreeVisit(RelNode n, TestVisitor v) {
- v.visit(n, -1, null);
-
- n.childrenAccept(new TestRelVisitor(v));
- }
-
protected static IgniteDistribution someAffinity() {
return IgniteDistributions.affinity(0, nextTableId(), DEFAULT_ZONE_ID);
}
@@ -285,8 +253,28 @@ public abstract class AbstractPlannerTest extends IgniteAbstractTest {
}
}
+ SqlToRelConverter.Config relConvCfg = FRAMEWORK_CONFIG.getSqlToRelConverterConfig();
+
+ if (hintStrategies != null) {
+ relConvCfg = relConvCfg.withHintStrategyTable(hintStrategies);
+ }
+
+ SchemaPlus rootSchema = createRootSchema(schemas);
+ SchemaPlus defaultSchema = rootSchema.getSubSchema(DEFAULT_SCHEMA);
+
+ if (defaultSchema == null && !schemas.isEmpty()) {
+ defaultSchema = rootSchema.getSubSchema(schemas.iterator().next().getName());
+ }
+
+ assertNotNull(defaultSchema);
+
PlanningContext ctx = PlanningContext.builder()
- .parentContext(baseQueryContext(schemas, hintStrategies, params.toArray()))
+ .frameworkConfig(
+ newConfigBuilder(FRAMEWORK_CONFIG)
+ .defaultSchema(defaultSchema)
+ .sqlToRelConverterConfig(relConvCfg)
+ .build()
+ )
.query(sql)
.parameters(paramsMap)
.build();
@@ -300,37 +288,24 @@ public abstract class AbstractPlannerTest extends IgniteAbstractTest {
return ctx;
}
- protected BaseQueryContext baseQueryContext(
- Collection<IgniteSchema> schemas,
- @Nullable HintStrategyTable hintStrategies,
- Object... params
- ) {
- SchemaPlus rootSchema = createRootSchema(false);
- SchemaPlus dfltSchema = null;
-
- for (IgniteSchema igniteSchema : schemas) {
- SchemaPlus schema = rootSchema.add(igniteSchema.getName(), igniteSchema);
+ protected static SchemaPlus createRootSchema(Collection<IgniteSchema> subSchemas) {
+ SchemaPlus rootSchema = Frameworks.createRootSchema(false);
- if (dfltSchema == null || DEFAULT_SCHEMA.equals(schema.getName())) {
- dfltSchema = schema;
- }
+ for (IgniteSchema igniteSchema : subSchemas) {
+ rootSchema.add(igniteSchema.getName(), igniteSchema);
}
- SqlToRelConverter.Config relConvCfg = FRAMEWORK_CONFIG.getSqlToRelConverterConfig();
+ return rootSchema;
+ }
- if (hintStrategies != null) {
- relConvCfg = relConvCfg.withHintStrategyTable(hintStrategies);
- }
-
-
- return BaseQueryContext.builder()
+ protected static SqlOperationContext operationContext(
+ Object... params
+ ) {
+ return SqlOperationContext.builder()
.queryId(UUID.randomUUID())
- .frameworkConfig(
- newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(dfltSchema)
- .sqlToRelConverterConfig(relConvCfg)
- .build()
- )
+ .timeZoneId(ZoneId.systemDefault())
+ .operationTime(new HybridClockImpl().now())
+ .defaultSchemaName(SqlCommon.DEFAULT_SCHEMA_NAME)
.parameters(params)
.build();
}
@@ -719,10 +694,10 @@ public abstract class AbstractPlannerTest extends IgniteAbstractTest {
List<RelNode> deserializedNodes = new ArrayList<>();
- BaseQueryContext ctx = baseQueryContext(schemas, null);
+ SchemaPlus rootSchema = createRootSchema(schemas);
for (String s : serialized) {
- RelJsonReader reader = new RelJsonReader(ctx.catalogReader());
+ RelJsonReader reader = new RelJsonReader(rootSchema);
deserializedNodes.add(reader.read(s));
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTest.java
index a43123a395..ce5fbf507f 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTest.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.sql.engine.planner;
-import static org.apache.calcite.tools.Frameworks.createRootSchema;
import static org.apache.calcite.tools.Frameworks.newConfigBuilder;
import static org.apache.ignite.internal.sql.engine.planner.CorrelatedSubqueryPlannerTest.createTestTable;
import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
@@ -30,7 +29,6 @@ import static org.junit.jupiter.params.provider.Arguments.arguments;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.UUID;
import java.util.function.Predicate;
import java.util.function.UnaryOperator;
import java.util.stream.Stream;
@@ -58,7 +56,6 @@ import org.apache.ignite.internal.sql.engine.rel.IgniteTableScan;
import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.type.NativeTypes;
import org.junit.jupiter.api.BeforeAll;
@@ -123,20 +120,16 @@ public class PlannerTest extends AbstractPlannerTest {
departmentTable(IgniteDistributions.broadcast())
);
- SchemaPlus schema = createRootSchema(false)
- .add("PUBLIC", publicSchema);
+ SchemaPlus schema = createRootSchema(List.of(publicSchema));
String sql = "select d.deptno, e.deptno "
+ "from dept d, emp e "
+ "where d.deptno + e.deptno = 2";
PlanningContext ctx = PlanningContext.builder()
- .parentContext(BaseQueryContext.builder()
- .queryId(UUID.randomUUID())
- .frameworkConfig(newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(schema)
- .costFactory(new IgniteCostFactory(1, 100, 1, 1))
- .build())
+ .frameworkConfig(newConfigBuilder(FRAMEWORK_CONFIG)
+ .defaultSchema(schema)
+ .costFactory(new IgniteCostFactory(1, 100, 1, 1))
.build())
.query(sql)
.build();
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java
index c7933d3828..2fb6cd2d56 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.sql.engine.planner;
+import static org.apache.calcite.tools.Frameworks.newConfigBuilder;
+import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCause;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.apache.ignite.lang.ErrorGroups.Sql.PLANNING_TIMEOUT_ERR;
@@ -26,7 +28,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import java.time.Duration;
-import java.util.Collections;
+import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.plan.volcano.VolcanoPlanner;
@@ -34,6 +36,8 @@ import org.apache.calcite.plan.volcano.VolcanoTimeoutException;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.RelVisitor;
import org.apache.ignite.internal.metrics.MetricManagerImpl;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
+import org.apache.ignite.internal.sql.engine.framework.PredefinedSchemaManager;
import org.apache.ignite.internal.sql.engine.framework.TestBuilders;
import org.apache.ignite.internal.sql.engine.prepare.IgnitePlanner;
import org.apache.ignite.internal.sql.engine.prepare.PlanningContext;
@@ -46,7 +50,6 @@ import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
import org.apache.ignite.internal.sql.engine.sql.ParserService;
import org.apache.ignite.internal.sql.engine.sql.ParserServiceImpl;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.SqlTestUtils;
import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
import org.apache.ignite.internal.type.NativeTypes;
@@ -62,10 +65,10 @@ public class PlannerTimeoutTest extends AbstractPlannerTest {
public void testPlannerTimeout() throws Exception {
long plannerTimeout = 1L;
IgniteSchema schema = createSchema(createTestTable("T1"));
- BaseQueryContext ctx = baseQueryContext(Collections.singletonList(schema), null);
+ SqlOperationContext ctx = operationContext();
PrepareService prepareService = new PrepareServiceImpl("test", 0,
- CaffeineCacheFactory.INSTANCE, null, plannerTimeout, 1, new MetricManagerImpl());
+ CaffeineCacheFactory.INSTANCE, null, plannerTimeout, 1, new MetricManagerImpl(), new PredefinedSchemaManager(schema));
prepareService.start();
try {
ParserService parserService = new ParserServiceImpl();
@@ -93,8 +96,10 @@ public class PlannerTimeoutTest extends AbstractPlannerTest {
String sql = "SELECT * FROM T1 JOIN T2 ON T1.A = T2.A";
PlanningContext ctx = PlanningContext.builder()
- .parentContext(baseQueryContext(Collections.singletonList(schema), null))
.plannerTimeout(plannerTimeout)
+ .frameworkConfig(newConfigBuilder(FRAMEWORK_CONFIG)
+ .defaultSchema(createRootSchema(List.of(schema)).getSubSchema(schema.getName()))
+ .build())
.query(sql)
.build();
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImplTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImplTest.java
index 1527410a57..bc410aa408 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImplTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImplTest.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.sql.engine.prepare;
-import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
import static org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -29,14 +28,17 @@ import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
+import java.time.ZoneId;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import org.apache.calcite.schema.SchemaPlus;
-import org.apache.calcite.tools.Frameworks;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.metrics.MetricManagerImpl;
+import org.apache.ignite.internal.sql.SqlCommon;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
+import org.apache.ignite.internal.sql.engine.framework.PredefinedSchemaManager;
import org.apache.ignite.internal.sql.engine.framework.TestBuilders;
import org.apache.ignite.internal.sql.engine.prepare.ddl.DdlSqlToCommandConverter;
import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
@@ -44,7 +46,6 @@ import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
import org.apache.ignite.internal.sql.engine.sql.ParserServiceImpl;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.SqlTestUtils;
import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
@@ -62,7 +63,6 @@ import org.junit.jupiter.params.provider.MethodSource;
/**
* Tests to verify {@link PrepareServiceImpl}.
*/
-@SuppressWarnings("DataFlowIssue")
public class PrepareServiceImplTest extends BaseIgniteAbstractTest {
private static final List<PrepareService> createdServices = new ArrayList<>();
@@ -187,7 +187,7 @@ public class PrepareServiceImplTest extends BaseIgniteAbstractTest {
"Ambiguous operator <UNKNOWN> + <UNKNOWN>. Dynamic parameter requires adding explicit type cast",
() -> {
ParsedResult parsedResult = parse("SELECT ? + ?");
- BaseQueryContext context = createContext();
+ SqlOperationContext context = createContext();
await(service.prepareAsync(parsedResult, context));
}
);
@@ -202,14 +202,15 @@ public class PrepareServiceImplTest extends BaseIgniteAbstractTest {
.distribution(IgniteDistributions.single())
.build();
- PrepareService service = createPlannerService();
-
IgniteSchema schema = new IgniteSchema("PUBLIC", 0, List.of(table));
+
+ PrepareService service = createPlannerService(schema);
+
Object paramValue = SqlTestUtils.generateValueByType(nativeType.spec().asColumnType());
QueryPlan queryPlan = await(service.prepareAsync(
parse("SELECT * FROM t WHERE c = ?"),
- createContext(schema, paramValue)
+ createContext(paramValue)
));
ParameterType parameterType = queryPlan.parameterMetadata().parameterTypes().get(0);
@@ -248,38 +249,16 @@ public class PrepareServiceImplTest extends BaseIgniteAbstractTest {
return new ParserServiceImpl().parse(query);
}
- private static BaseQueryContext createContext(Object... params) {
- return BaseQueryContext.builder()
+ private static SqlOperationContext createContext(Object... params) {
+ return SqlOperationContext.builder()
.queryId(UUID.randomUUID())
- .frameworkConfig(
- Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(wrap(createSchema()))
- .build()
- )
+ .timeZoneId(ZoneId.systemDefault())
+ .operationTime(new HybridClockImpl().now())
+ .defaultSchemaName(SqlCommon.DEFAULT_SCHEMA_NAME)
.parameters(params)
.build();
}
- private static BaseQueryContext createContext(IgniteSchema schema, Object... params) {
- return BaseQueryContext.builder()
- .queryId(UUID.randomUUID())
- .frameworkConfig(
- Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(wrap(schema))
- .build()
- )
- .parameters(params)
- .build();
- }
-
- private static SchemaPlus wrap(IgniteSchema schema) {
- var schemaPlus = Frameworks.createRootSchema(false);
-
- schemaPlus.add(schema.getName(), schema);
-
- return schemaPlus.getSubSchema(schema.getName());
- }
-
private static IgniteSchema createSchema() {
IgniteTable table = TestBuilders.table()
.name("T")
@@ -292,8 +271,13 @@ public class PrepareServiceImplTest extends BaseIgniteAbstractTest {
}
private static PrepareService createPlannerService() {
+ return createPlannerService(createSchema());
+ }
+
+ private static PrepareService createPlannerService(IgniteSchema schema) {
PrepareService service = new PrepareServiceImpl("test", 1_000, CaffeineCacheFactory.INSTANCE,
- mock(DdlSqlToCommandConverter.class), 5_000, 2, mock(MetricManagerImpl.class));
+ mock(DdlSqlToCommandConverter.class), 5_000, 2, mock(MetricManagerImpl.class),
+ new PredefinedSchemaManager(schema));
createdServices.add(service);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/ddl/AbstractDdlSqlToCommandConverterTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/ddl/AbstractDdlSqlToCommandConverterTest.java
index 9831167e26..ed32ca5d82 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/ddl/AbstractDdlSqlToCommandConverterTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/ddl/AbstractDdlSqlToCommandConverterTest.java
@@ -25,7 +25,6 @@ import static org.hamcrest.Matchers.not;
import static org.mockito.Mockito.mock;
import java.util.List;
-import java.util.UUID;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.parser.SqlParseException;
import org.apache.calcite.sql.parser.SqlParser;
@@ -37,7 +36,6 @@ import org.apache.ignite.internal.generated.query.calcite.sql.IgniteSqlParserImp
import org.apache.ignite.internal.sql.SqlCommon;
import org.apache.ignite.internal.sql.engine.prepare.PlanningContext;
import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.hamcrest.Matchers;
@@ -68,11 +66,8 @@ class AbstractDdlSqlToCommandConverterTest extends BaseIgniteAbstractTest {
var schema = Frameworks.createRootSchema(false).add(schemaName, publicSchema);
return PlanningContext.builder()
- .parentContext(BaseQueryContext.builder()
- .queryId(UUID.randomUUID())
- .frameworkConfig(newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(schema)
- .build())
+ .frameworkConfig(newConfigBuilder(FRAMEWORK_CONFIG)
+ .defaultSchema(schema)
.build())
.query("")
.build();
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImplTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImplTest.java
index c9809e80a3..f74b5e648a 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImplTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImplTest.java
@@ -174,7 +174,7 @@ public class SqlSchemaManagerImplTest extends BaseIgniteAbstractTest {
IgniteSchema schema = unwrapSchema(schemaPlus);
- assertThat(schema.version(), equalTo(versionAfter));
+ assertThat(schema.catalogVersion(), equalTo(versionAfter));
CatalogSchemaDescriptor schemaDescriptor = catalogManager.schema(PUBLIC_SCHEMA_NAME, versionAfter);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/metrics/PlanningCacheMetricsTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/metrics/PlanningCacheMetricsTest.java
index 352e02d69f..d302eedff5 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/metrics/PlanningCacheMetricsTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/metrics/PlanningCacheMetricsTest.java
@@ -20,10 +20,11 @@ package org.apache.ignite.internal.sql.metrics;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import java.util.Collections;
import org.apache.ignite.internal.metrics.MetricManager;
import org.apache.ignite.internal.metrics.MetricManagerImpl;
import org.apache.ignite.internal.metrics.MetricSet;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
+import org.apache.ignite.internal.sql.engine.framework.PredefinedSchemaManager;
import org.apache.ignite.internal.sql.engine.framework.TestBuilders;
import org.apache.ignite.internal.sql.engine.planner.AbstractPlannerTest;
import org.apache.ignite.internal.sql.engine.prepare.PrepareService;
@@ -34,7 +35,6 @@ import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
import org.apache.ignite.internal.sql.engine.sql.ParserService;
import org.apache.ignite.internal.sql.engine.sql.ParserServiceImpl;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
-import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.cache.CacheFactory;
import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
import org.apache.ignite.internal.type.NativeTypes;
@@ -50,7 +50,19 @@ public class PlanningCacheMetricsTest extends AbstractPlannerTest {
MetricManager metricManager = new MetricManagerImpl();
// Run clean up tasks in the current thread, so no eviction event is delayed.
CacheFactory cacheFactory = CaffeineCacheFactory.create(Runnable::run);
- PrepareService prepareService = new PrepareServiceImpl("test", 2, cacheFactory, null, 15_000L, 2, metricManager);
+
+ IgniteTable table = TestBuilders.table()
+ .name("T")
+ .addColumn("A", NativeTypes.INT32, false)
+ .addColumn("B", NativeTypes.INT32, false)
+ .distribution(IgniteDistributions.single())
+ .build();
+
+ IgniteSchema schema = createSchema(table);
+
+ PrepareService prepareService = new PrepareServiceImpl(
+ "test", 2, cacheFactory, null, 15_000L, 2, metricManager, new PredefinedSchemaManager(schema)
+ );
prepareService.start();
@@ -77,15 +89,7 @@ public class PlanningCacheMetricsTest extends AbstractPlannerTest {
}
private void checkCachePlanStatistics(String qry, PrepareService prepareService, MetricSet metricSet, int hits, int misses) {
- IgniteTable table = TestBuilders.table()
- .name("T")
- .addColumn("A", NativeTypes.INT32, false)
- .addColumn("B", NativeTypes.INT32, false)
- .distribution(IgniteDistributions.single())
- .build();
-
- IgniteSchema schema = createSchema(table);
- BaseQueryContext ctx = baseQueryContext(Collections.singletonList(schema), null);
+ SqlOperationContext ctx = operationContext();
ParserService parserService = new ParserServiceImpl();
ParsedResult parsedResult = parserService.parse(qry);