You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by am...@apache.org on 2023/07/25 10:00:26 UTC
[ignite-3] 06/11: wip. Use catalog version for cache plans.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-19499
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 065c9a102a1cc8740a28e2d56776e02c516aa6e6
Author: amashenkov <an...@gmail.com>
AuthorDate: Fri Jul 21 02:19:07 2023 +0300
wip. Use catalog version for cache plans.
---
.../apache/ignite/internal/index/IndexManager.java | 1 -
.../internal/sql/engine/SqlQueryProcessor.java | 3 +--
.../sql/engine/exec/ExecutableTableRegistry.java | 3 ++-
.../engine/exec/ExecutableTableRegistryImpl.java | 30 ++++++++++------------
.../exec/ExecutionDependencyResolverImpl.java | 2 +-
.../sql/engine/exec/LogicalRelImplementor.java | 24 ++++++++---------
.../internal/sql/engine/prepare/CacheKey.java | 28 ++++++++++----------
.../sql/engine/prepare/PrepareServiceImpl.java | 4 ++-
.../exec/ExecutableTableRegistrySelfTest.java | 2 +-
.../exec/ExecutionDependencyResolverSelfTest.java | 16 ++++++------
.../engine/exec/NoOpExecutableTableRegistry.java | 2 +-
.../internal/table/distributed/TableManager.java | 9 +++----
12 files changed, 59 insertions(+), 65 deletions(-)
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
index 7e5caf9491..4a4853c2b8 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.index;
-import static java.util.concurrent.CompletableFuture.allOf;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
import static org.apache.ignite.internal.util.ArrayUtils.STRING_EMPTY_ARRAY;
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 4f64f709d3..c0bb72e478 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
@@ -250,8 +250,7 @@ public class SqlQueryProcessor implements QueryProcessor {
var dependencyResolver = new ExecutionDependencyResolverImpl(executableTableRegistry);
- CatalogSqlSchemaManager sqlSchemaManager = new CatalogSqlSchemaManager(catalogManager, 1000);
- this.sqlSchemaManager = sqlSchemaManager;
+ sqlSchemaManager = new CatalogSqlSchemaManager(catalogManager, 1000);
var executionSrvc = registerService(ExecutionServiceImpl.create(
clusterSrvc.topologyService(),
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 36b8cb4eb6..6461cae939 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
@@ -28,9 +28,10 @@ public interface ExecutableTableRegistry {
/**
* Returns an table that can be used for both read and write operations.
*
+ * @param schemaVersion Table schema version.
* @param tableId Table Id.
* @param tableDescriptor Table descriptor.
* @return An operation that returns executable table.
*/
- CompletableFuture<ExecutableTable> getTable(int tableId, TableDescriptor tableDescriptor);
+ CompletableFuture<ExecutableTable> getTable(int schemaVersion, int tableId, TableDescriptor tableDescriptor);
}
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 92b254855d..67c9b5c4fd 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
@@ -61,8 +61,10 @@ public class ExecutableTableRegistryImpl implements ExecutableTableRegistry, Sch
/** {@inheritDoc} */
@Override
- public CompletableFuture<ExecutableTable> getTable(int tableId, TableDescriptor tableDescriptor) {
- return tableCache.computeIfAbsent(tableId, (k) -> loadTable(k, tableDescriptor));
+ public CompletableFuture<ExecutableTable> getTable(int schemaVersion, int tableId, TableDescriptor tableDescriptor) {
+ //TODO: SchemaVersion must a part of cache key, or cache InternalTable only.
+ // return tableCache.computeIfAbsent(tableId, (k) -> loadTable(schemaVersion, k, tableDescriptor));
+ return loadTable(schemaVersion, tableId, tableDescriptor);
}
/** {@inheritDoc} */
@@ -71,27 +73,21 @@ public class ExecutableTableRegistryImpl implements ExecutableTableRegistry, Sch
tableCache.clear();
}
- private CompletableFuture<ExecutableTable> loadTable(int tableId, TableDescriptor tableDescriptor) {
-
- CompletableFuture<Map.Entry<InternalTable, SchemaRegistry>> f = tableManager.tableAsync(tableId)
+ private CompletableFuture<ExecutableTable> loadTable(int schemaVersion, int tableId, TableDescriptor tableDescriptor) {
+ return tableManager.tableAsync(tableId)
.thenApply(table -> {
InternalTable internalTable = table.internalTable();
SchemaRegistry schemaRegistry = schemaManager.schemaRegistry(tableId);
- return Map.entry(internalTable, schemaRegistry);
- });
- return f.thenApply((table) -> {
- SchemaRegistry schemaRegistry = table.getValue();
- SchemaDescriptor schemaDescriptor = schemaRegistry.schema();
- TableRowConverter rowConverter = new TableRowConverterImpl(schemaRegistry, schemaDescriptor, tableDescriptor);
- InternalTable internalTable = table.getKey();
- ScannableTable scannableTable = new ScannableTableImpl(internalTable, rowConverter, tableDescriptor);
+ SchemaDescriptor schemaDescriptor = schemaRegistry.schema(schemaVersion);
+ TableRowConverter rowConverter = new TableRowConverterImpl(schemaRegistry, schemaDescriptor, tableDescriptor);
+ ScannableTable scannableTable = new ScannableTableImpl(internalTable, rowConverter, tableDescriptor);
- UpdatableTableImpl updatableTable = new UpdatableTableImpl(tableId, tableDescriptor, internalTable.partitions(),
- replicaService, clock, rowConverter, schemaDescriptor);
+ UpdatableTableImpl updatableTable = new UpdatableTableImpl(tableId, tableDescriptor, internalTable.partitions(),
+ replicaService, clock, rowConverter, schemaDescriptor);
- return new ExecutableTableImpl(scannableTable, updatableTable);
- });
+ return new ExecutableTableImpl(scannableTable, updatableTable);
+ });
}
private static final class ExecutableTableImpl implements ExecutableTable {
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 4cd0b8c0ba..c07bdcc3f5 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
@@ -82,7 +82,7 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
int tableId = igniteTable.id();
TableDescriptor tableDescriptor = igniteTable.descriptor();
- tableMap.computeIfAbsent(tableId, (id) -> registry.getTable(tableId, tableDescriptor));
+ tableMap.computeIfAbsent(tableId, (id) -> registry.getTable((int) schemaVersion, tableId, tableDescriptor));
}
};
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
index d1bfb1187f..0a0f973698 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
@@ -389,18 +389,6 @@ public class LogicalRelImplementor<RowT> implements IgniteRelVisitor<Node<RowT>>
);
}
- private static RelDataType getIndexRowType(IgniteTypeFactory typeFactory, IgniteTable tbl, IgniteIndex idx) {
- RelDataTypeFactory.Builder b = new RelDataTypeFactory.Builder(typeFactory);
-
- RelDataType rt = tbl.getRowType(typeFactory);
- for (RelFieldCollation fc : idx.collation().getFieldCollations()) {
- b.add(rt.getFieldList().get(fc.getFieldIndex()));
- }
-
- RelDataType idxRowType = b.build();
- return idxRowType;
- }
-
/** {@inheritDoc} */
@Override
public Node<RowT> visit(IgniteTableScan rel) {
@@ -809,4 +797,16 @@ public class LogicalRelImplementor<RowT> implements IgniteRelVisitor<Node<RowT>>
public <T extends Node<RowT>> T go(IgniteRel rel) {
return (T) visit(rel);
}
+
+ private static RelDataType getIndexRowType(IgniteTypeFactory typeFactory, IgniteTable tbl, IgniteIndex idx) {
+ RelDataTypeFactory.Builder b = new RelDataTypeFactory.Builder(typeFactory);
+
+ RelDataType rt = tbl.getRowType(typeFactory);
+ for (RelFieldCollation fc : idx.collation().getFieldCollations()) {
+ b.add(rt.getFieldList().get(fc.getFieldIndex()));
+ }
+
+ RelDataType idxRowType = b.build();
+ return idxRowType;
+ }
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/CacheKey.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/CacheKey.java
index 89cd57d3a2..9decab5523 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/CacheKey.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/CacheKey.java
@@ -28,6 +28,8 @@ import java.util.Objects;
public class CacheKey {
public static final Class[] EMPTY_CLASS_ARRAY = {};
+ private final int version;
+
private final String schemaName;
private final String query;
@@ -39,29 +41,21 @@ public class CacheKey {
/**
* Constructor.
*
+ * @param version Catalog version.
* @param schemaName Schema name.
- * @param query Query string.
- * @param contextKey Optional context key to differ queries with and without/different flags, having an impact on result plan (like
- * LOCAL flag)
+ * @param query Query string.
+ * @param contextKey Optional context key to differ queries with and without/different flags, having an impact on result plan
+ * (like LOCAL flag)
* @param paramTypes Types of all dynamic parameters, no any type can be {@code null}.
*/
- public CacheKey(String schemaName, String query, Object contextKey, Class[] paramTypes) {
+ public CacheKey(int version, String schemaName, String query, Object contextKey, Class[] paramTypes) {
+ this.version = version;
this.schemaName = schemaName;
this.query = query;
this.contextKey = contextKey;
this.paramTypes = paramTypes;
}
- /**
- * Constructor.
- *
- * @param schemaName Schema name.
- * @param query Query string.
- */
- public CacheKey(String schemaName, String query) {
- this(schemaName, query, null, EMPTY_CLASS_ARRAY);
- }
-
/** {@inheritDoc} */
@Override
public boolean equals(Object o) {
@@ -74,6 +68,9 @@ public class CacheKey {
CacheKey cacheKey = (CacheKey) o;
+ if (version != cacheKey.version) {
+ return false;
+ }
if (!schemaName.equals(cacheKey.schemaName)) {
return false;
}
@@ -89,7 +86,8 @@ public class CacheKey {
@Override
public int hashCode() {
- int result = schemaName.hashCode();
+ int result = version;
+ result = 31 * result + schemaName.hashCode();
result = 31 * result + query.hashCode();
result = 31 * result + (contextKey != null ? contextKey.hashCode() : 0);
result = 31 * result + Arrays.deepHashCode(paramTypes);
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 cf04e7a29d..9b501ed331 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
@@ -298,7 +298,9 @@ public class PrepareServiceImpl implements PrepareService, SchemaUpdateListener
? EMPTY_CLASS_ARRAY :
Arrays.stream(ctx.parameters()).map(p -> (p != null) ? p.getClass() : Void.class).toArray(Class[]::new);
- return new CacheKey(ctx.schemaName(), parsedResult.normalizedQuery(), distributed, paramTypes);
+ int version = (int) ctx.unwrap(BaseQueryContext.class).schemaVersion();
+
+ return new CacheKey(version, ctx.schemaName(), parsedResult.normalizedQuery(), distributed, paramTypes);
}
private ResultSetMetadata resultSetMetadata(
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistrySelfTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistrySelfTest.java
index c98ee0c454..740a1cba71 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistrySelfTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistrySelfTest.java
@@ -150,7 +150,7 @@ public class ExecutableTableRegistrySelfTest {
when(schemaManager.schemaRegistry(tableId)).thenReturn(schemaRegistry);
when(schemaRegistry.schema()).thenReturn(schemaDescriptor);
- return registry.getTable(tableId, descriptor);
+ return registry.getTable(schemaVersion, tableId, descriptor);
}
}
}
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 6a73dff178..75c1d14f66 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
@@ -98,8 +98,8 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
TableDescriptor td1 = tester.tableDescriptor("TEST1");
TableDescriptor td2 = tester.tableDescriptor("TEST2");
- verify(registry, times(1)).getTable(eq(t1Id), same(td1));
- verify(registry, times(1)).getTable(eq(t2Id), same(td2));
+ verify(registry, times(1)).getTable(-1, eq(t1Id), same(td1));
+ verify(registry, times(1)).getTable(-1, eq(t2Id), same(td2));
}
/**
@@ -124,7 +124,7 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
TableDescriptor td1 = tester.tableDescriptor("TEST1");
- verify(registry, times(1)).getTable(eq(t1Id), same(td1));
+ verify(registry, times(1)).getTable(-1, eq(t1Id), same(td1));
}
/**
@@ -155,8 +155,8 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
TableDescriptor td1 = tester.tableDescriptor("TEST1");
TableDescriptor td2 = tester.tableDescriptor("TEST2");
- verify(registry, times(1)).getTable(eq(t1Id), same(td1));
- verify(registry, times(1)).getTable(eq(t2Id), same(td2));
+ verify(registry, times(1)).getTable(-1, eq(t1Id), same(td1));
+ verify(registry, times(1)).getTable(-1, eq(t2Id), same(td2));
}
/**
@@ -179,7 +179,7 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
ResolvedDependencies deps = f.join();
tester.checkDependencies(deps, t1Id);
- verify(registry, times(1)).getTable(anyInt(), any(TableDescriptor.class));
+ verify(registry, times(1)).getTable(-1, anyInt(), any(TableDescriptor.class));
}
/**
@@ -229,14 +229,14 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
CompletableFuture<ExecutableTable> f = CompletableFuture.completedFuture(executableTable);
- when(registry.getTable(eq(tableId), any(TableDescriptor.class))).thenReturn(f);
+ when(registry.getTable(-1, eq(tableId), any(TableDescriptor.class))).thenReturn(f);
}
void setError(int tableId, Throwable err) {
CompletableFuture<ExecutableTable> f = new CompletableFuture<>();
f.completeExceptionally(err);
- when(registry.getTable(eq(tableId), any(TableDescriptor.class))).thenReturn(f);
+ when(registry.getTable(-1, eq(tableId), any(TableDescriptor.class))).thenReturn(f);
}
CompletableFuture<ResolvedDependencies> resolveDependencies(String sql) {
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 0acba776ca..f672fd14b5 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
@@ -25,7 +25,7 @@ public final class NoOpExecutableTableRegistry implements ExecutableTableRegistr
/** {@inheritDoc} */
@Override
- public CompletableFuture<ExecutableTable> getTable(int tableId, TableDescriptor tableDescriptor) {
+ public CompletableFuture<ExecutableTable> getTable(int schemaVersion, int tableId, TableDescriptor tableDescriptor) {
return CompletableFuture.completedFuture(new NoOpExecutableTable(tableId));
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index 5c00af1fc3..ca49abcb65 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -1237,14 +1237,14 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
* @param causalityToken Causality token.
* @param tableDescriptor Catalog table descriptor.
* @param zoneDescriptor Catalog distributed zone descriptor.
- * @param indexesToWait
+ * @param tableIndexes Ids of indexes belongs to the table.
* @return Future that will be completed when local changes related to the table creation are applied.
*/
private CompletableFuture<?> createTableLocally(
long causalityToken,
CatalogTableDescriptor tableDescriptor,
CatalogZoneDescriptor zoneDescriptor,
- int[] indexesToWait,
+ int[] tableIndexes,
List<Set<Assignment>> assignments
) {
String tableName = tableDescriptor.name();
@@ -1265,7 +1265,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
var table = new TableImpl(internalTable, lockMgr);
// TODO: IGNITE-19082 Need another way to wait for indexes
-// table.addIndexesToWait(indexesToWait);
+ table.addIndexesToWait(tableIndexes);
tablesByIdVv.update(causalityToken, (previous, e) -> inBusyLock(busyLock, () -> {
if (e != null) {
@@ -1293,8 +1293,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
pendingTables.remove(tableId);
}));
- createPartsFut
- .thenRunAsync(() -> inBusyLock(busyLock, () -> completeApiCreateFuture(table)));
+ createPartsFut.thenRunAsync(() -> inBusyLock(busyLock, () -> completeApiCreateFuture(table)));
// TODO should be reworked in IGNITE-16763
// We use the event notification future as the result so that dependent components can complete the schema updates.