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 2022/08/26 14:03:16 UTC
[ignite-3] 01/02: Make calcite schema objects immutable.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-17431
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit a47ffbe3f9da945e60810283991b6890f6ae2e42
Author: amashenkov <an...@gmail.com>
AuthorDate: Thu Aug 25 14:04:57 2022 +0300
Make calcite schema objects immutable.
---
.../internal/sql/engine/schema/IgniteSchema.java | 24 +---
.../sql/engine/schema/IgniteTableImpl.java | 18 ++-
.../sql/engine/schema/SqlSchemaManagerImpl.java | 148 ++++++++++++++-------
.../engine/exec/schema/SqlSchemaManagerTest.java | 71 +++++++++-
4 files changed, 190 insertions(+), 71 deletions(-)
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 f983e3c5f1..243fa682d3 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
@@ -54,6 +54,10 @@ public class IgniteSchema extends AbstractSchema {
this(schemaName, null, null);
}
+ public static IgniteSchema copy(IgniteSchema old) {
+ return new IgniteSchema(old.schemaName, old.tblMap, old.idxMap);
+ }
+
/**
* Get schema name.
*
@@ -99,10 +103,6 @@ public class IgniteSchema extends AbstractSchema {
* @param index Index.
*/
public void addIndex(UUID indexId, IgniteIndex index) {
- // TODO: https://issues.apache.org/jira/browse/IGNITE-17474
- // Decouple tables and indices.
- index.table().addIndex(index);
-
idxMap.put(indexId, index);
}
@@ -110,20 +110,10 @@ public class IgniteSchema extends AbstractSchema {
* Remove index.
*
* @param indexId Index id.
- * @return {@code True} if index was removed or {@code false} otherwise.
+ * @return Removed index.
*/
- public boolean removeIndex(UUID indexId) {
- IgniteIndex rmv = idxMap.remove(indexId);
-
- if (rmv == null) {
- return false;
- }
-
- // TODO: https://issues.apache.org/jira/browse/IGNITE-17474
- // Decouple tables and indices.
- rmv.table().removeIndex(rmv.name());
-
- return true;
+ public IgniteIndex removeIndex(UUID indexId) {
+ return idxMap.remove(indexId);
}
/**
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
index cd9258cad5..5529b215a7 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
@@ -22,6 +22,7 @@ import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -81,7 +82,7 @@ public class IgniteTableImpl extends AbstractTable implements InternalIgniteTabl
private final Statistic statistic;
- private final Map<String, IgniteIndex> indexes = new ConcurrentHashMap<>();
+ private Map<String, IgniteIndex> indexes = new HashMap<>();
private final List<ColumnDescriptor> columnsOrderedByPhysSchema;
@@ -115,6 +116,21 @@ public class IgniteTableImpl extends AbstractTable implements InternalIgniteTabl
statistic = new StatisticsImpl();
}
+ private IgniteTableImpl(IgniteTableImpl t) {
+ this.desc = t.desc;
+ this.ver = t.ver;
+ this.table = t.table;
+ this.schemaRegistry = t.schemaRegistry;
+ this.schemaDescriptor = t.schemaDescriptor;
+ this.statistic = t.statistic;
+ this.columnsOrderedByPhysSchema = t.columnsOrderedByPhysSchema;
+ this.indexes.putAll(t.indexes);
+ }
+
+ public static IgniteTableImpl copyOf(IgniteTableImpl v) {
+ return new IgniteTableImpl(v);
+ }
+
/** {@inheritDoc} */
@Override
public UUID id() {
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 861500261a..72aefa5be9 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
@@ -26,6 +26,7 @@ import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
@@ -66,7 +67,7 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
private final VersionedValue<Map<String, IgniteSchema>> schemasVv;
- private final VersionedValue<Map<UUID, IgniteTable>> tablesVv;
+ private final VersionedValue<Map<UUID, InternalIgniteTable>> tablesVv;
private final VersionedValue<Map<UUID, IgniteIndex>> indicesVv;
@@ -213,7 +214,8 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
Map<String, IgniteSchema> res = new HashMap<>(schemas);
- IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
+ IgniteSchema schema = res.compute(schemaName,
+ (k, v) -> v == null ? new IgniteSchema(schemaName) : IgniteSchema.copy(v));
CompletableFuture<IgniteTableImpl> igniteTableFuture = convert(causalityToken, table);
@@ -223,7 +225,7 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
return failedFuture(ex);
}
- Map<UUID, IgniteTable> resTbls = new HashMap<>(tables);
+ Map<UUID, InternalIgniteTable> resTbls = new HashMap<>(tables);
return igniteTableFuture
.thenApply(igniteTable -> inBusyLock(busyLock, () -> {
@@ -281,7 +283,8 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
Map<String, IgniteSchema> res = new HashMap<>(schemas);
- IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
+ IgniteSchema schema = res.compute(schemaName,
+ (k, v) -> v == null ? new IgniteSchema(schemaName) : IgniteSchema.copy(v));
String calciteTableName = objectSimpleName(schemaName, tableName);
@@ -295,7 +298,7 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
return failedFuture(ex);
}
- Map<UUID, IgniteTable> resTbls = new HashMap<>(tables);
+ Map<UUID, InternalIgniteTable> resTbls = new HashMap<>(tables);
resTbls.remove(table.id());
@@ -363,15 +366,9 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
);
}
- private CompletableFuture<IgniteIndex> convert(long causalityToken, Index<?> index) {
- return schemaManager.schemaRegistry(causalityToken, index.tableId())
- .thenApply(schemaRegistry -> inBusyLock(busyLock, () -> convert(index, schemaRegistry)));
- }
-
- private IgniteIndex convert(Index<?> index, SchemaRegistry schemaRegistry) {
+ private IgniteIndex convert(Index<?> index, SchemaRegistry schemaRegistry, IgniteTable table) {
IndexDescriptor desc = index.descriptor();
SchemaDescriptor schema = schemaRegistry.schema();
- IgniteTable table = tablesVv.latest().get(index.tableId());
assert table != null;
@@ -430,36 +427,54 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
Map<String, IgniteSchema> res = new HashMap<>(schemas);
- IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
-
- CompletableFuture<IgniteIndex> igniteIndexFuture = convert(causalityToken, index);
-
- return indicesVv
- .update(
- causalityToken,
- (indices, ex) -> inBusyLock(busyLock, () -> {
- if (ex != null) {
- return failedFuture(ex);
- }
-
- Map<UUID, IgniteIndex> resIdxs = new HashMap<>(indices);
-
- return igniteIndexFuture
- .thenApply(igniteIndex -> {
- resIdxs.put(index.id(), igniteIndex);
+ IgniteSchema schema = res.compute(schemaName,
+ (k, v) -> v == null ? new IgniteSchema(schemaName) : IgniteSchema.copy(v));
- return resIdxs;
- });
- })
- )
- .thenCombine(
- igniteIndexFuture,
- (v, igniteIndex) -> inBusyLock(busyLock, () -> {
- schema.addIndex(index.id(), igniteIndex);
+ return tablesVv.update(causalityToken, (tables, tblEx) -> inBusyLock(busyLock, () -> {
+ if (tblEx != null) {
+ return failedFuture(tblEx);
+ }
- return null;
- })
- )
+ Map<UUID, InternalIgniteTable> resTbls = new HashMap<>(tables);
+
+ InternalIgniteTable table = resTbls.compute(index.tableId(),
+ (k, v) -> IgniteTableImpl.copyOf((IgniteTableImpl) v));
+
+ CompletableFuture<IgniteIndex> igniteIndexFuture = schemaManager.schemaRegistry(causalityToken, index.tableId())
+ .thenApply(reg -> convert(index, reg, table));
+
+ return indicesVv
+ .update(
+ causalityToken,
+ (indices, idxEx) -> inBusyLock(busyLock, () -> {
+ if (idxEx != null) {
+ return failedFuture(idxEx);
+ }
+
+ Map<UUID, IgniteIndex> resIdxs = new HashMap<>(indices);
+
+ return igniteIndexFuture
+ .thenApply(igniteIndex -> {
+ resIdxs.put(index.id(), igniteIndex);
+
+ return resIdxs;
+ });
+ })
+ )
+ .thenCombine(
+ igniteIndexFuture,
+ (v, igniteIndex) -> inBusyLock(busyLock, () -> {
+ String tblName = tableNameById(schema, index.tableId());
+
+ table.addIndex(igniteIndex);
+ schema.addTable(tblName, (InternalIgniteTable) table);
+ schema.addIndex(index.id(), igniteIndex);
+
+ return v;
+ })
+ )
+ .thenCompose(v -> inBusyLock(busyLock, () -> completedFuture(resTbls)));
+ }))
.thenCompose(v -> inBusyLock(busyLock, () -> completedFuture(res)));
}));
@@ -469,6 +484,18 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
}
}
+ @NotNull
+ private static String tableNameById(IgniteSchema schema, UUID tableId) {
+ String tblName = schema.getTableMap()
+ .entrySet()
+ .stream()
+ .filter(entry -> tableId
+ .equals(((InternalIgniteTable) entry.getValue()).id()))
+ .map(Entry::getKey)
+ .findFirst().get();
+ return tblName;
+ }
+
/**
* Index dropped callback method deregisters index from Calcite schema.
*
@@ -489,21 +516,42 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
Map<String, IgniteSchema> res = new HashMap<>(schemas);
- IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
+ IgniteSchema schema = res.compute(schemaName,
+ (k, v) -> v == null ? new IgniteSchema(schemaName) : IgniteSchema.copy(v));
- if (schema.removeIndex(indexId)) {
- return indicesVv.update(causalityToken, (indices, ex) -> inBusyLock(busyLock, () -> {
- if (ex != null) {
- return failedFuture(ex);
+ IgniteIndex rmvIndex = schema.removeIndex(indexId);
+ if (rmvIndex != null) {
+ return tablesVv.update(causalityToken, (tables, tlbEx) -> inBusyLock(busyLock, () -> {
+ if (tlbEx != null) {
+ return failedFuture(tlbEx);
}
- Map<UUID, IgniteIndex> resIdxs = new HashMap<>(indices);
+ Map<UUID, InternalIgniteTable> resTbls = new HashMap<>(tables);
+
+ InternalIgniteTable table = resTbls.compute(rmvIndex.table().id(),
+ (k, v) -> IgniteTableImpl.copyOf((IgniteTableImpl) v));
- resIdxs.remove(indexId);
+ table.removeIndex(rmvIndex.name());
- return completedFuture(resIdxs);
- }
- )).thenCompose(v -> inBusyLock(busyLock, () -> completedFuture(res)));
+ return indicesVv.update(causalityToken, (indices, idxEx) -> inBusyLock(busyLock, () -> {
+ if (idxEx != null) {
+ return failedFuture(idxEx);
+ }
+
+ Map<UUID, IgniteIndex> resIdxs = new HashMap<>(indices);
+
+ IgniteIndex rmvIdx = resIdxs.remove(indexId);
+
+ assert table.id().equals(rmvIdx.table().id());
+
+ String tblName = tableNameById(schema, rmvIdx.table().id());
+ schema.addTable(tblName, table);
+
+ return completedFuture(resIdxs);
+ }
+ )).thenCompose(v -> inBusyLock(busyLock, () -> completedFuture(resTbls)));
+ })
+ ).thenCompose(v -> inBusyLock(busyLock, () -> completedFuture(res)));
}
return completedFuture(res);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java
index b9d2f64351..84bf63c766 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java
@@ -24,6 +24,7 @@ import static org.hamcrest.Matchers.containsString;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -42,6 +43,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Function;
+import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.Table;
import org.apache.ignite.internal.index.Index;
import org.apache.ignite.internal.index.IndexDescriptor;
@@ -54,6 +56,7 @@ import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.schema.IgniteTableImpl;
+import org.apache.ignite.internal.sql.engine.schema.InternalIgniteTable;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManagerImpl;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.TableImpl;
@@ -83,7 +86,7 @@ public class SqlSchemaManagerTest {
tableVer,
new Column[]{new Column(0, "ID", NativeTypes.INT64, false)},
new Column[]{new Column(1, "VAL", NativeTypes.INT64, false)}
- );
+ );
@Mock
private TableManager tableManager;
@@ -312,6 +315,68 @@ public class SqlSchemaManagerTest {
verifyNoMoreInteractions(tableManager);
}
+
+ @Test
+ public void testIndexEventsProcessed() {
+ InternalTable mock = mock(InternalTable.class);
+ when(mock.tableId()).thenReturn(tableId);
+
+ when(table.name()).thenReturn("TEST_SCHEMA.T");
+ when(table.internalTable()).thenReturn(mock);
+ when(schemaRegistry.schema()).thenReturn(schemaDescriptor);
+ when(schemaRegistry.lastSchemaVersion()).thenReturn(schemaDescriptor.version());
+ when(schemaManager.schemaRegistry(anyLong(), any())).thenReturn(completedFuture(schemaRegistry));
+
+ sqlSchemaManager.onTableCreated("TEST_SCHEMA", table, testRevisionRegister.actualToken() + 1);
+ testRevisionRegister.moveForward();
+
+ IndexDescriptor descMock = mock(IndexDescriptor.class);
+ when(descMock.columns()).thenReturn(List.of());
+
+ when(index.name()).thenReturn("I");
+ when(index.id()).thenReturn(indexId);
+ when(index.tableId()).thenReturn(tableId);
+ when(index.descriptor()).thenReturn(descMock);
+
+ {
+ SchemaPlus schema1 = sqlSchemaManager.schema("TEST_SCHEMA");
+
+ sqlSchemaManager.onIndexCreated("TEST_SCHEMA", index, testRevisionRegister.actualToken() + 1);
+ testRevisionRegister.moveForward();
+
+ SchemaPlus schema2 = sqlSchemaManager.schema("TEST_SCHEMA");
+
+ // Validate schema snapshot.
+ assertNotSame(schema1, schema2);
+ assertNotSame(schema1.getTable("T"), schema2.getTable("T"));
+
+ assertNull(schema1.unwrap(IgniteSchema.class).index(indexId));
+ assertNotNull(schema2.unwrap(IgniteSchema.class).index(indexId));
+
+ assertNull(((InternalIgniteTable)schema1.getTable("T")).getIndex("I"));
+ assertNotNull(((InternalIgniteTable)schema2.getTable("T")).getIndex("I"));
+ }
+ {
+ sqlSchemaManager.onIndexDropped("TEST_SCHEMA", indexId, testRevisionRegister.actualToken() + 1);
+ SchemaPlus schema1 = sqlSchemaManager.schema("TEST_SCHEMA");
+ testRevisionRegister.moveForward();
+
+ SchemaPlus schema2 = sqlSchemaManager.schema("TEST_SCHEMA");
+
+ // Validate schema snapshot.
+ assertNotSame(schema1, schema2);
+ assertNotSame(schema1.getTable("T"), schema2.getTable("T"));
+
+ assertNotNull(schema1.unwrap(IgniteSchema.class).index(indexId));
+ assertNull(schema2.unwrap(IgniteSchema.class).index(indexId));
+
+ assertNotNull(((InternalIgniteTable)schema1.getTable("T")).getIndex("I"));
+ assertNull(((InternalIgniteTable)schema2.getTable("T")).getIndex("I"));
+ }
+
+ verifyNoMoreInteractions(tableManager);
+ }
+
/**
* Test revision register.
*/
@@ -346,8 +411,8 @@ public class SqlSchemaManagerTest {
Function<Long, CompletableFuture<?>> old = moveRevision;
moveRevision = rev -> allOf(
- old.apply(rev),
- function.apply(rev)
+ old.apply(rev),
+ function.apply(rev)
);
}
}