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 2021/06/11 13:52:49 UTC
[ignite-3] 01/01: Add AlterTable operation. Add SchemaChange
operation. Add addColumn test.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-14863
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit a2ca15d3a8b7f6020a02a97e15830da8ea3c06e7
Author: Andrew Mashenkov <an...@gmail.com>
AuthorDate: Fri Jun 11 16:52:34 2021 +0300
Add AlterTable operation.
Add SchemaChange operation.
Add addColumn test.
---
.../apache/ignite/table/manager/IgniteTables.java | 9 +-
.../runner/app/DynamicTableCreationTest.java | 6 +
...ableCreationTest.java => SchemaChangeTest.java} | 150 ++++------
.../ignite/internal/schema/SchemaManager.java | 63 +++-
.../ignite/internal/schema/SchemaRegistry.java | 17 +-
.../ignite/internal/schema/event/SchemaEvent.java | 3 +
.../schema/registry/SchemaRegistryImpl.java | 21 +-
.../ignite/distributed/ITDistributedTableTest.java | 4 +
.../internal/table/distributed/TableManager.java | 330 +++++++++++++++------
.../ignite/internal/table/event/TableEvent.java | 3 +
.../table/impl/DummySchemaManagerImpl.java | 5 +
11 files changed, 387 insertions(+), 224 deletions(-)
diff --git a/modules/api/src/main/java/org/apache/ignite/table/manager/IgniteTables.java b/modules/api/src/main/java/org/apache/ignite/table/manager/IgniteTables.java
index 780e9f1..aab5f3e 100644
--- a/modules/api/src/main/java/org/apache/ignite/table/manager/IgniteTables.java
+++ b/modules/api/src/main/java/org/apache/ignite/table/manager/IgniteTables.java
@@ -31,7 +31,6 @@ import org.apache.ignite.table.Table;
public interface IgniteTables {
/**
* Creates a cluster table.
- * The table changes if already exists.
*
* @param name Table name.
* @param tableInitChange Table changer.
@@ -40,6 +39,14 @@ public interface IgniteTables {
Table createTable(String name, Consumer<TableChange> tableInitChange);
/**
+ * Alter a cluster table.
+ *
+ * @param name Table name.
+ * @param tableChange Table changer.
+ */
+ void alterTable(String name, Consumer<TableChange> tableChange);
+
+ /**
* Drops a table with the name specified.
*
* @param name Table name.
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/DynamicTableCreationTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/DynamicTableCreationTest.java
index 6c76f04..0406f3c 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/DynamicTableCreationTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/DynamicTableCreationTest.java
@@ -26,6 +26,7 @@ import org.apache.ignite.app.Ignite;
import org.apache.ignite.app.IgnitionManager;
import org.apache.ignite.internal.schema.SchemaManager;
import org.apache.ignite.internal.schema.configuration.SchemaConfigurationConverter;
+import org.apache.ignite.internal.table.SchemaMismatchException;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.schema.ColumnType;
import org.apache.ignite.schema.SchemaBuilders;
@@ -128,6 +129,11 @@ class DynamicTableCreationTest {
assertEquals(111, (Integer)kvView2.get(keyTuple1).value("val"));
assertEquals(222, (Integer)tbl2.get(keyTuple2).value("val"));
assertEquals(222, (Integer)kvView2.get(keyTuple2).value("val"));
+
+ assertThrows(SchemaMismatchException.class, () -> tbl1.get(keyTuple1).value("key"));
+ assertThrows(SchemaMismatchException.class, () -> kvView1.get(keyTuple1).value("key"));
+ assertThrows(SchemaMismatchException.class, () -> tbl1.get(keyTuple1).value("val"));
+ assertThrows(SchemaMismatchException.class, () -> kvView1.get(keyTuple1).value("val"));
}
/**
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/DynamicTableCreationTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/SchemaChangeTest.java
similarity index 52%
copy from modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/DynamicTableCreationTest.java
copy to modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/SchemaChangeTest.java
index 6c76f04..b0af76f 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/DynamicTableCreationTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/SchemaChangeTest.java
@@ -21,30 +21,29 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.UUID;
import org.apache.ignite.app.Ignite;
import org.apache.ignite.app.IgnitionManager;
import org.apache.ignite.internal.schema.SchemaManager;
-import org.apache.ignite.internal.schema.configuration.SchemaConfigurationConverter;
+import org.apache.ignite.internal.table.ColumnNotFoundException;
import org.apache.ignite.lang.IgniteLogger;
+import org.apache.ignite.schema.Column;
import org.apache.ignite.schema.ColumnType;
import org.apache.ignite.schema.SchemaBuilders;
import org.apache.ignite.schema.SchemaTable;
import org.apache.ignite.table.KeyValueBinaryView;
import org.apache.ignite.table.Table;
import org.apache.ignite.table.Tuple;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
+import static org.apache.ignite.internal.schema.configuration.SchemaConfigurationConverter.convert;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
/**
* Ignition interface tests.
*/
-@Disabled("https://issues.apache.org/jira/browse/IGNITE-14581")
-class DynamicTableCreationTest {
+//@Disabled("https://issues.apache.org/jira/browse/IGNITE-14581")
+class SchemaChangeTest {
/** The logger. */
private static final IgniteLogger LOG = IgniteLogger.forClass(SchemaManager.class);
@@ -85,7 +84,7 @@ class DynamicTableCreationTest {
* Check dynamic table creation.
*/
@Test
- void testDynamicSimpleTableCreation() {
+ void addNewColumn() {
List<Ignite> clusterNodes = new ArrayList<>();
for (Map.Entry<String, String> nodeBootstrapCfg : nodesBootstrapCfg.entrySet())
@@ -96,38 +95,71 @@ class DynamicTableCreationTest {
// Create table on node 0.
SchemaTable schTbl1 = SchemaBuilders.tableBuilder("PUBLIC", "tbl1").columns(
SchemaBuilders.column("key", ColumnType.INT64).asNonNull().build(),
- SchemaBuilders.column("val", ColumnType.INT32).asNullable().build()
+ SchemaBuilders.column("val1", ColumnType.INT32).asNullable().build()
).withPrimaryKey("key").build();
clusterNodes.get(0).tables().createTable(schTbl1.canonicalName(), tblCh ->
- SchemaConfigurationConverter.convert(schTbl1, tblCh)
+ convert(schTbl1, tblCh)
.changeReplicas(1)
.changePartitions(10)
);
- // Put data on node 1.
Table tbl1 = waitForTable(schTbl1.canonicalName(), clusterNodes.get(1));
+ Table tbl2 = waitForTable(schTbl1.canonicalName(), clusterNodes.get(2));
+
+ // Put data on node 1.
KeyValueBinaryView kvView1 = tbl1.kvView();
- tbl1.insert(tbl1.tupleBuilder().set("key", 1L).set("val", 111).build());
- kvView1.put(tbl1.tupleBuilder().set("key", 2L).build(), tbl1.tupleBuilder().set("val", 222).build());
+ tbl1.insert(tbl1.tupleBuilder().set("key", 1L).set("val1", 111).build());
+ kvView1.put(tbl1.tupleBuilder().set("key", 2L).build(), tbl1.tupleBuilder().set("val1", 222).build());
+
+ assertThrows(ColumnNotFoundException.class, () -> kvView1.put(tbl1.tupleBuilder().set("key", 2L).build(), tbl1.tupleBuilder().set("val1", 222).set("val2", "str").build()));
+
+ clusterNodes.get(1).tables().alterTable(schTbl1.canonicalName(),
+ chng -> chng.changeColumns(cols -> {
+ final int colIdx = chng.columns().size();
+ final Column col = SchemaBuilders.column("val2", ColumnType.string())
+ .asNullable()
+ .withDefaultValue("default")
+ .build();
+
+ //TODO: avoid 'colIdx' or replace with correct last colIdx.
+ cols.create(String.valueOf(colIdx), colChg -> convert(col, colChg));
+ }));
+
+ kvView1.put(tbl1.tupleBuilder().set("key", 3L).build(), tbl1.tupleBuilder().set("val1", 333).set("val2", "str").build());
// Get data on node 2.
- Table tbl2 = waitForTable(schTbl1.canonicalName(), clusterNodes.get(2));
KeyValueBinaryView kvView2 = tbl2.kvView();
+ final Tuple keyTuple3 = tbl2.tupleBuilder().set("key", 3L).build();
+
+ assertEquals(3, (Long)tbl2.get(keyTuple3).value("key"));
+ assertEquals(333, (Integer)tbl2.get(keyTuple3).value("val1"));
+ assertEquals(333, (Integer)kvView2.get(keyTuple3).value("val1"));
+ assertEquals("str", tbl2.get(keyTuple3).value("val2"));
+ assertEquals("str", kvView2.get(keyTuple3).value("val2"));
+
+ // Check old row conversion.
final Tuple keyTuple1 = tbl2.tupleBuilder().set("key", 1L).build();
final Tuple keyTuple2 = kvView2.tupleBuilder().set("key", 2L).build();
- assertThrows(IllegalArgumentException.class, () -> kvView2.get(keyTuple1).value("key"));
- assertThrows(IllegalArgumentException.class, () -> kvView2.get(keyTuple1).value("key"));
assertEquals(1, (Long)tbl2.get(keyTuple1).value("key"));
assertEquals(2, (Long)tbl2.get(keyTuple2).value("key"));
- assertEquals(111, (Integer)tbl2.get(keyTuple1).value("val"));
- assertEquals(111, (Integer)kvView2.get(keyTuple1).value("val"));
- assertEquals(222, (Integer)tbl2.get(keyTuple2).value("val"));
- assertEquals(222, (Integer)kvView2.get(keyTuple2).value("val"));
+ assertEquals(111, (Integer)tbl2.get(keyTuple1).value("val1"));
+ assertEquals(111, (Integer)kvView2.get(keyTuple1).value("val1"));
+
+ //TODO: https://issues.apache.org/jira/browse/IGNITE-14896 Add evolution converter for default values.
+// assertEquals("default", tbl2.get(keyTuple1).value("val2"));
+// assertEquals("default", kvView2.get(keyTuple1).value("val2"));
+
+ assertEquals(222, (Integer)tbl2.get(keyTuple2).value("val1"));
+ assertEquals(222, (Integer)kvView2.get(keyTuple2).value("val1"));
+
+ //TODO: https://issues.apache.org/jira/browse/IGNITE-14896 Add evolution converter for default values.
+// assertEquals("default", tbl2.get(keyTuple2).value("val2"));
+// assertEquals("default", kvView2.get(keyTuple2).value("val2"));
}
/**
@@ -150,84 +182,4 @@ class DynamicTableCreationTest {
return ign.tables().table(tableName);
}
- /**
- * Check dynamic table creation.
- */
- @Test
- void testDynamicTableCreation() {
- List<Ignite> clusterNodes = new ArrayList<>();
-
- for (Map.Entry<String, String> nodeBootstrapCfg : nodesBootstrapCfg.entrySet())
- clusterNodes.add(IgnitionManager.start(nodeBootstrapCfg.getKey(), nodeBootstrapCfg.getValue()));
-
- assertEquals(3, clusterNodes.size());
-
- // Create table on node 0.
- SchemaTable scmTbl1 = SchemaBuilders.tableBuilder("PUBLIC", "tbl1").columns(
- SchemaBuilders.column("key", ColumnType.UUID).asNonNull().build(),
- SchemaBuilders.column("affKey", ColumnType.INT64).asNonNull().build(),
- SchemaBuilders.column("valStr", ColumnType.string()).asNullable().build(),
- SchemaBuilders.column("valInt", ColumnType.INT32).asNullable().build(),
- SchemaBuilders.column("valNull", ColumnType.INT16).asNullable().build()
- ).withIndex(
- SchemaBuilders.pkIndex()
- .addIndexColumn("key").done()
- .addIndexColumn("affKey").done()
- .withAffinityColumns("affKey")
- .build()
- ).build();
-
- clusterNodes.get(0).tables().createTable(scmTbl1.canonicalName(), tblCh ->
- SchemaConfigurationConverter.convert(scmTbl1, tblCh)
- .changeReplicas(1)
- .changePartitions(10));
-
- final UUID uuid = UUID.randomUUID();
- final UUID uuid2 = UUID.randomUUID();
-
- // Put data on node 1.
- Table tbl1 = waitForTable(scmTbl1.canonicalName(), clusterNodes.get(1));
- KeyValueBinaryView kvView1 = tbl1.kvView();
-
- tbl1.insert(tbl1.tupleBuilder().set("key", uuid).set("affKey", 42L)
- .set("valStr", "String value").set("valInt", 73).set("valNull", null).build());
-
- kvView1.put(kvView1.tupleBuilder().set("key", uuid2).set("affKey", 4242L).build(),
- kvView1.tupleBuilder().set("valStr", "String value 2").set("valInt", 7373).set("valNull", null).build());
-
- // Get data on node 2.
- Table tbl2 = waitForTable(scmTbl1.canonicalName(), clusterNodes.get(2));
- KeyValueBinaryView kvView2 = tbl2.kvView();
-
- final Tuple keyTuple1 = tbl2.tupleBuilder().set("key", uuid).set("affKey", 42L).build();
- final Tuple keyTuple2 = tbl2.tupleBuilder().set("key", uuid2).set("affKey", 4242L).build();
-
- // KV view must NOT return key columns in value.
- assertThrows(IllegalArgumentException.class, () -> kvView2.get(keyTuple1).value("key"));
- assertThrows(IllegalArgumentException.class, () -> kvView2.get(keyTuple1).value("affKey"));
- assertThrows(IllegalArgumentException.class, () -> kvView2.get(keyTuple2).value("key"));
- assertThrows(IllegalArgumentException.class, () -> kvView2.get(keyTuple2).value("affKey"));
-
- // Record binary view MUST return key columns in value.
- assertEquals(uuid, tbl2.get(keyTuple1).value("key"));
- assertEquals(42L, (Long)tbl2.get(keyTuple1).value("affKey"));
- assertEquals(uuid2, tbl2.get(keyTuple2).value("key"));
- assertEquals(4242L, (Long)tbl2.get(keyTuple2).value("affKey"));
-
- assertEquals("String value", tbl2.get(keyTuple1).value("valStr"));
- assertEquals(73, (Integer)tbl2.get(keyTuple1).value("valInt"));
- assertNull(tbl2.get(keyTuple1).value("valNull"));
-
- assertEquals("String value 2", tbl2.get(keyTuple2).value("valStr"));
- assertEquals(7373, (Integer)tbl2.get(keyTuple2).value("valInt"));
- assertNull(tbl2.get(keyTuple2).value("valNull"));
-
- assertEquals("String value", kvView2.get(keyTuple1).value("valStr"));
- assertEquals(73, (Integer)kvView2.get(keyTuple1).value("valInt"));
- assertNull(kvView2.get(keyTuple1).value("valNull"));
-
- assertEquals("String value 2", kvView2.get(keyTuple2).value("valStr"));
- assertEquals(7373, (Integer)kvView2.get(keyTuple2).value("valInt"));
- assertNull(kvView2.get(keyTuple2).value("valNull"));
- }
}
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
index e25521d..682353a 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
@@ -29,6 +29,12 @@ import org.apache.ignite.configuration.schemas.table.TableConfiguration;
import org.apache.ignite.configuration.schemas.table.TablesConfiguration;
import org.apache.ignite.internal.manager.Producer;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.client.Conditions;
+import org.apache.ignite.internal.metastorage.client.Entry;
+import org.apache.ignite.internal.metastorage.client.EntryEvent;
+import org.apache.ignite.internal.metastorage.client.Operations;
+import org.apache.ignite.internal.metastorage.client.WatchEvent;
+import org.apache.ignite.internal.metastorage.client.WatchListener;
import org.apache.ignite.internal.schema.configuration.SchemaConfigurationConverter;
import org.apache.ignite.internal.schema.configuration.SchemaDescriptorConverter;
import org.apache.ignite.internal.schema.event.SchemaEvent;
@@ -40,18 +46,12 @@ import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteLogger;
-import org.apache.ignite.internal.metastorage.client.Conditions;
-import org.apache.ignite.internal.metastorage.client.Entry;
-import org.apache.ignite.internal.metastorage.client.EntryEvent;
-import org.apache.ignite.internal.metastorage.client.Operations;
-import org.apache.ignite.internal.metastorage.client.WatchEvent;
-import org.apache.ignite.internal.metastorage.client.WatchListener;
import org.apache.ignite.schema.SchemaTable;
import org.jetbrains.annotations.NotNull;
/**
* Schema Manager.
- *
+ * <p>
* Schemas MUST be registered in a version ascending order incrementing by {@code 1} with NO gaps,
* otherwise an exception will be thrown. The version numbering starts from the {@code 1}.
* <p>
@@ -111,7 +111,7 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
if (verPos == -1) {
final UUID tblId = UUID.fromString(keyTail);
- SchemaRegistry reg = schemaRegistryForTable(tblId);
+ SchemaRegistryImpl reg = schemaRegistryForTable(tblId);
assert reg != null : "Table schema was not initialized or table has been dropped: " + tblId;
@@ -122,6 +122,13 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
onEvent(SchemaEvent.DROPPED, new SchemaEventParameters(tblId, null), null);
}
+ else {
+ int v = (int)ByteUtils.bytesToLong(evt.newEntry().value(),0);
+
+ assert reg.lastSchemaVersion() == v;
+
+ onEvent(SchemaEvent.CHANGED, new SchemaEventParameters(tblId, reg), null);
+ }
return true; // Ignore last table schema version.
}
@@ -189,6 +196,42 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
}
/**
+ * Creates schema registry for the table with existed schema or
+ * registers initial schema from configuration.
+ *
+ * @param tblId Table id.
+ * @param tblName Table name.
+ * @return Operation future.
+ */
+ public CompletableFuture<Boolean> updateSchemaForTable(final UUID tblId, String tblName) {
+ return vaultMgr.get(ByteArray.fromString(INTERNAL_PREFIX + tblId)).
+ thenCompose(entry -> {
+ TableConfiguration tblConfig = configurationMgr.configurationRegistry().
+ getConfiguration(TablesConfiguration.KEY).tables().get(tblName);
+
+ assert !entry.empty();
+
+ final int oldVer = (int)ByteUtils.bytesToLong(entry.value(), 0);
+ final int newVer = oldVer + 1;
+
+ final ByteArray lastVerKey = new ByteArray(INTERNAL_PREFIX + tblId);
+ final ByteArray schemaKey = new ByteArray(INTERNAL_PREFIX + tblId + INTERNAL_VER_SUFFIX + newVer);
+
+ SchemaTable schemaTable = SchemaConfigurationConverter.convert(tblConfig);
+ final SchemaDescriptor desc = SchemaDescriptorConverter.convert(tblId, newVer, schemaTable);
+
+ return metaStorageMgr.invoke(Conditions.notExists(schemaKey),
+ Operations.put(schemaKey, ByteUtils.toBytes(desc)),
+ Operations.noop())
+ //TODO: IGNITE-14679 Serialize schema.
+ .thenCompose(res -> metaStorageMgr.invoke(
+ Conditions.value(lastVerKey).eq(ByteUtils.longToBytes(oldVer)),
+ Operations.put(lastVerKey, ByteUtils.longToBytes(newVer)),
+ Operations.noop()));
+ });
+ }
+
+ /**
* Return table schema of certain version from history.
*
* @param tblId Table id.
@@ -229,8 +272,8 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
* @param tableId Table id.
* @return Schema registry for the table.
*/
- private SchemaRegistry schemaRegistryForTable(UUID tableId) {
- final SchemaRegistry reg = schemaRegs.get(tableId);
+ private SchemaRegistryImpl schemaRegistryForTable(UUID tableId) {
+ final SchemaRegistryImpl reg = schemaRegs.get(tableId);
if (reg == null)
throw new SchemaRegistryException("No schema was ever registered for the table: " + tableId);
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaRegistry.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaRegistry.java
index 777adf1..a46489a 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaRegistry.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaRegistry.java
@@ -25,14 +25,23 @@ import org.jetbrains.annotations.NotNull;
*/
public interface SchemaRegistry {
/**
- * @return Current schema.
+ * Gets schema descriptor for the latest version if initialized.
+ *
+ * @return Schema descriptor if initialized, {@code null} otherwise.
*/
SchemaDescriptor schema();
/**
- * @param ver Schema version.
- * @return Schema of given version.
- * @throws SchemaRegistryException If schema was not found.
+ * @return Last registereg schema version.
+ */
+ public int lastSchemaVersion();
+
+ /**
+ * Gets schema descriptor for given version.
+ *
+ * @param ver Schema version to get descriptor for.
+ * @return Schema descriptor of given version.
+ * @throws SchemaRegistryException If no schema found for given version.
*/
@NotNull SchemaDescriptor schema(int ver) throws SchemaRegistryException;
}
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/event/SchemaEvent.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/event/SchemaEvent.java
index d81e0d2..6d27985 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/event/SchemaEvent.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/event/SchemaEvent.java
@@ -26,6 +26,9 @@ public enum SchemaEvent implements Event {
/** This event is fired when a schema was initialized. */
INITIALIZED,
+ /** This event is fired when a schema was changed. */
+ CHANGED,
+
/** This event is fired when a schema was dropped. */
DROPPED
}
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/registry/SchemaRegistryImpl.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/registry/SchemaRegistryImpl.java
index f9d9e64..c51bdfd 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/registry/SchemaRegistryImpl.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/registry/SchemaRegistryImpl.java
@@ -60,13 +60,7 @@ public class SchemaRegistryImpl implements SchemaRegistry {
this.history = history;
}
- /**
- * Gets schema descriptor for given version.
- *
- * @param ver Schema version to get descriptor for.
- * @return Schema descriptor.
- * @throws SchemaRegistryException If no schema found for given version.
- */
+ /** {@inheritDoc} */
@Override public SchemaDescriptor schema(int ver) {
SchemaDescriptor desc = schemaCache.get(ver);
@@ -87,12 +81,7 @@ public class SchemaRegistryImpl implements SchemaRegistry {
throw new SchemaRegistryException("Failed to find schema: ver=" + ver);
}
- /**
- * Gets schema descriptor for the latest version if initialized.
- *
- * @return Schema descriptor if initialized, {@code null} otherwise.
- * @throws SchemaRegistryException If failed.
- */
+ /** {@inheritDoc} */
@Override public @Nullable SchemaDescriptor schema() {
final int lastVer0 = lastVer;
@@ -102,10 +91,8 @@ public class SchemaRegistryImpl implements SchemaRegistry {
return schema(lastVer0);
}
- /**
- * @return Last known schema version.
- */
- public int lastSchemaVersion() {
+ /** {@inheritDoc} */
+ @Override public int lastSchemaVersion() {
return lastVer;
}
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java
index 428dd91..7b57cff 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java
@@ -265,6 +265,10 @@ public class ITDistributedTableTest {
return SCHEMA;
}
+ @Override public int lastSchemaVersion() {
+ return SCHEMA.version();
+ }
+
@Override public SchemaDescriptor schema(int ver) {
return SCHEMA;
}
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 4291047..c575c4f 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
@@ -21,7 +21,6 @@ import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -31,15 +30,19 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.function.BiPredicate;
import java.util.function.Consumer;
+import java.util.stream.Collectors;
import org.apache.ignite.configuration.internal.ConfigurationManager;
import org.apache.ignite.configuration.schemas.table.TableChange;
import org.apache.ignite.configuration.schemas.table.TableView;
import org.apache.ignite.configuration.schemas.table.TablesConfiguration;
+import org.apache.ignite.configuration.tree.NamedListView;
import org.apache.ignite.internal.affinity.AffinityManager;
import org.apache.ignite.internal.affinity.event.AffinityEvent;
import org.apache.ignite.internal.affinity.event.AffinityEventParameters;
import org.apache.ignite.internal.manager.Producer;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.client.Conditions;
+import org.apache.ignite.internal.metastorage.client.Operations;
import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.schema.SchemaManager;
import org.apache.ignite.internal.schema.SchemaRegistry;
@@ -53,8 +56,6 @@ import org.apache.ignite.internal.table.event.TableEventParameters;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteLogger;
-import org.apache.ignite.internal.metastorage.client.Conditions;
-import org.apache.ignite.internal.metastorage.client.Operations;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.apache.ignite.table.Table;
@@ -197,123 +198,233 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
//TODO: IGNITE-14652 Change a metastorage update in listeners to multi-invoke
configurationMgr.configurationRegistry().getConfiguration(TablesConfiguration.KEY).tables().listen(ctx -> {
Set<String> tablesToStart = (ctx.newValue() == null || ctx.newValue().namedListKeys() == null) ?
- Collections.emptySet() : new HashSet<>(ctx.newValue().namedListKeys());
+ Collections.emptySet() :
+ ctx.newValue().namedListKeys().stream().filter(t -> !ctx.oldValue().namedListKeys().contains(t)).collect(Collectors.toSet());
- tablesToStart.removeAll(ctx.oldValue().namedListKeys());
-
- long revision = ctx.storageRevision();
+ Set<String> tablesToStop = (ctx.oldValue() == null || ctx.oldValue().namedListKeys() == null) ?
+ Collections.emptySet() :
+ ctx.oldValue().namedListKeys().stream().filter(t -> !ctx.newValue().namedListKeys().contains(t)).collect(Collectors.toSet());
List<CompletableFuture<Boolean>> futs = new ArrayList<>();
- boolean hasMetastorageLocally = metaStorageMgr.hasMetastorageLocally(configurationMgr);
+ final Set<String> schemaChanged =
+ (ctx.oldValue() != null && ctx.oldValue().namedListKeys() != null && ctx.newValue() != null && ctx.newValue().namedListKeys() != null) ?
+ ctx.oldValue().namedListKeys().stream()
+ .filter(tblName -> ctx.newValue().namedListKeys().contains(tblName)) // Filter changed tables.
+ .filter(tblName -> {
+ final TableView newTbl = ctx.newValue().get(tblName);
+ final TableView oldTbl = ctx.oldValue().get(tblName);
- for (String tblName : tablesToStart) {
- TableView tableView = ctx.newValue().get(tblName);
+ assert newTbl.columns().namedListKeys() != null && oldTbl.columns().namedListKeys() != null;
- UUID tblId = new UUID(revision, 0L);
+ return newTbl.columns().namedListKeys().stream().anyMatch(c -> !oldTbl.columns().namedListKeys().contains(c)) ||
+ oldTbl.columns().namedListKeys().stream().anyMatch(c -> !newTbl.columns().namedListKeys().contains(c));
+ }).collect(Collectors.toSet()) :
+ Collections.emptySet();
- if (hasMetastorageLocally) {
- var key = new ByteArray(INTERNAL_PREFIX + tblId);
- futs.add(metaStorageMgr.invoke(
- Conditions.notExists(key),
- Operations.put(key, tableView.name().getBytes(StandardCharsets.UTF_8)),
- Operations.noop())
- .thenCompose(res -> schemaMgr.initSchemaForTable(tblId, tableView.name()))
- .thenCompose(res -> affMgr.calculateAssignments(tblId, tableView.name())));
- }
+ if (!tablesToStart.isEmpty())
+ futs.addAll(startTables(tablesToStart, ctx.storageRevision(), ctx.newValue()));
- final CompletableFuture<AffinityEventParameters> affinityReadyFut = new CompletableFuture<>();
- final CompletableFuture<SchemaEventParameters> schemaReadyFut = new CompletableFuture<>();
+ if (!schemaChanged.isEmpty())
+ futs.addAll(changeSchema(schemaChanged));
- CompletableFuture.allOf(affinityReadyFut, schemaReadyFut)
- .exceptionally(e -> {
- LOG.error("Failed to create a new table [name=" + tblName + ", id=" + tblId + ']', e);
+ if (!tablesToStop.isEmpty())
+ futs.addAll(stopTables(tablesToStop));
- onEvent(TableEvent.CREATE, new TableEventParameters(
- tblId,
- tblName,
- null,
- null
- ), e);
+ return CompletableFuture.allOf(futs.toArray(CompletableFuture[]::new));
+ });
+ }
- return null;
- })
- .thenRun(() -> createTableLocally(
- tblName,
+ /**
+ * Start tables routine.
+ *
+ * @param tbls Tables to start.
+ * @param rev Metastore revision.
+ * @param cfgs Table configurations.
+ * @return Table creation futures.
+ */
+ private List<CompletableFuture<Boolean>> startTables(Set<String> tbls, long rev, NamedListView<TableView> cfgs) {
+ boolean hasMetastorageLocally = metaStorageMgr.hasMetastorageLocally(configurationMgr);
+
+ List<CompletableFuture<Boolean>> futs = new ArrayList<>();
+
+ for (String tblName : tbls) {
+ TableView tableView = cfgs.get(tblName);
+
+ UUID tblId = new UUID(rev, 0L);
+
+ if (hasMetastorageLocally) {
+ var key = new ByteArray(INTERNAL_PREFIX + tblId);
+ futs.add(metaStorageMgr.invoke(
+ Conditions.notExists(key),
+ Operations.put(key, tableView.name().getBytes(StandardCharsets.UTF_8)),
+ Operations.noop())
+ .thenCompose(res -> schemaMgr.initSchemaForTable(tblId, tableView.name()))
+ .thenCompose(res -> affMgr.calculateAssignments(tblId, tableView.name())));
+ }
+
+ final CompletableFuture<AffinityEventParameters> affinityReadyFut = new CompletableFuture<>();
+ final CompletableFuture<SchemaEventParameters> schemaReadyFut = new CompletableFuture<>();
+
+ CompletableFuture.allOf(affinityReadyFut, schemaReadyFut)
+ .exceptionally(e -> {
+ LOG.error("Failed to create a new table [name=" + tblName + ", id=" + tblId + ']', e);
+
+ onEvent(TableEvent.CREATE, new TableEventParameters(
tblId,
- affinityReadyFut.join().assignment(),
- schemaReadyFut.join().schemaRegistry()
- ));
+ tblName,
+ null,
+ null
+ ), e);
+
+ return null;
+ })
+ .thenRun(() -> createTableLocally(
+ tblName,
+ tblId,
+ affinityReadyFut.join().assignment(),
+ schemaReadyFut.join().schemaRegistry()
+ ));
+
+ affMgr.listen(AffinityEvent.CALCULATED, (parameters, e) -> {
+ if (!tblId.equals(parameters.tableId()))
+ return false;
+
+ if (e == null)
+ affinityReadyFut.complete(parameters);
+ else
+ affinityReadyFut.completeExceptionally(e);
- affMgr.listen(AffinityEvent.CALCULATED, (parameters, e) -> {
- if (!tblId.equals(parameters.tableId()))
- return false;
+ return true;
+ });
- if (e == null)
- affinityReadyFut.complete(parameters);
- else
- affinityReadyFut.completeExceptionally(e);
+ schemaMgr.listen(SchemaEvent.INITIALIZED, (parameters, e) -> {
+ if (!tblId.equals(parameters.tableId()) && parameters.schemaRegistry().lastSchemaVersion() >= 1)
+ return false;
- return true;
- });
+ if (e == null)
+ schemaReadyFut.complete(parameters);
+ else
+ schemaReadyFut.completeExceptionally(e);
- schemaMgr.listen(SchemaEvent.INITIALIZED, (parameters, e) -> {
- if (!tblId.equals(parameters.tableId()))
- return false;
+ return true;
+ });
+ }
- if (e == null)
- schemaReadyFut.complete(parameters);
- else
- schemaReadyFut.completeExceptionally(e);
+ return futs;
+ }
- return true;
- });
- }
+ /**
+ * Drop tables routine.
+ *
+ * @param tbls Tables to drop.
+ * @return Table drop futures.
+ */
+ private List<CompletableFuture<Boolean>> stopTables(Set<String> tbls) {
+ boolean hasMetastorageLocally = metaStorageMgr.hasMetastorageLocally(configurationMgr);
- Set<String> tablesToStop = (ctx.oldValue() == null || ctx.oldValue().namedListKeys() == null) ?
- Collections.emptySet() : new HashSet<>(ctx.oldValue().namedListKeys());
+ List<CompletableFuture<Boolean>> futs = new ArrayList<>();
- tablesToStop.removeAll(ctx.newValue().namedListKeys());
+ for (String tblName : tbls) {
+ TableImpl t = tables.get(tblName);
- for (String tblName : tablesToStop) {
- TableImpl t = tables.get(tblName);
+ UUID tblId = t.internalTable().tableId();
- UUID tblId = t.internalTable().tableId();
+ if (hasMetastorageLocally) {
+ var key = new ByteArray(INTERNAL_PREFIX + tblId);
- if (hasMetastorageLocally) {
- var key = new ByteArray(INTERNAL_PREFIX + tblId);
+ futs.add(affMgr.removeAssignment(tblId)
+ .thenCompose(res -> schemaMgr.unregisterSchemas(tblId))
+ .thenCompose(res ->
+ metaStorageMgr.invoke(Conditions.exists(key),
+ Operations.remove(key),
+ Operations.noop())));
+ }
- futs.add(affMgr.removeAssignment(tblId)
- .thenCompose(res -> schemaMgr.unregisterSchemas(tblId))
- .thenCompose(res ->
- metaStorageMgr.invoke(Conditions.exists(key),
- Operations.remove(key),
- Operations.noop())));
+ affMgr.listen(AffinityEvent.REMOVED, (parameters, e) -> {
+ if (!tblId.equals(parameters.tableId()))
+ return false;
+
+ if (e == null)
+ dropTableLocally(tblName, tblId, parameters.assignment());
+ else {
+ LOG.error("Failed to drop a table [name=" + tblName + ", id=" + tblId + ']', e);
+
+ onEvent(TableEvent.DROP, new TableEventParameters(
+ tblId,
+ tblName,
+ null,
+ null
+ ), e);
}
- affMgr.listen(AffinityEvent.REMOVED, (parameters, e) -> {
- if (!tblId.equals(parameters.tableId()))
- return false;
-
- if (e == null)
- dropTableLocally(tblName, tblId, parameters.assignment());
- else {
- LOG.error("Failed to drop a table [name=" + tblName + ", id=" + tblId + ']', e);
-
- onEvent(TableEvent.DROP, new TableEventParameters(
- tblId,
- tblName,
- null,
- null
- ), e);
- }
-
- return true;
- });
- }
+ return true;
+ });
+ }
- return CompletableFuture.allOf(futs.toArray(CompletableFuture[]::new));
- });
+ return futs;
+ }
+
+ /**
+ * Start tables routine.
+ *
+ * @param tbls Tables to start.
+ * @return Table creation futures.
+ */
+ private List<CompletableFuture<Boolean>> changeSchema(Set<String> tbls) {
+ boolean hasMetastorageLocally = metaStorageMgr.hasMetastorageLocally(configurationMgr);
+
+ List<CompletableFuture<Boolean>> futs = new ArrayList<>();
+
+ for (String tblName : tbls) {
+ TableImpl tbl = tables.get(tblName);
+
+ UUID tblId = tbl.internalTable().tableId();
+
+ final int ver = tbl.schemaView().lastSchemaVersion() + 1;
+
+ if (hasMetastorageLocally)
+ futs.add(schemaMgr.updateSchemaForTable(tblId, tblName));
+
+ final CompletableFuture<SchemaEventParameters> schemaReadyFut = new CompletableFuture<>();
+
+ CompletableFuture.allOf(schemaReadyFut)
+ .exceptionally(e -> {
+ LOG.error("Failed to upgrade schema for a table [name=" + tblName + ", id=" + tblId + ']', e);
+
+ onEvent(TableEvent.ALTER, new TableEventParameters(
+ tblId,
+ tblName,
+ null,
+ null
+ ), e);
+
+ return null;
+ })
+ .thenRun(() -> onEvent(
+ TableEvent.ALTER,
+ new TableEventParameters(
+ tblId,
+ tblName,
+ schemaReadyFut.join().schemaRegistry(),
+ tbl.internalTable()),
+ null
+ ));
+
+ schemaMgr.listen(SchemaEvent.CHANGED, (parameters, e) -> {
+ if (!tblId.equals(parameters.tableId()) && parameters.schemaRegistry().lastSchemaVersion() < ver)
+ return false;
+
+ if (e == null)
+ schemaReadyFut.complete(parameters);
+ else
+ schemaReadyFut.completeExceptionally(e);
+
+ return true;
+ });
+ }
+
+ return futs;
}
/** {@inheritDoc} */
@@ -350,6 +461,39 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
}
/** {@inheritDoc} */
+ @Override public void alterTable(String name, Consumer<TableChange> tableChange) {
+ CompletableFuture<Void> tblFut = new CompletableFuture<>();
+
+ listen(TableEvent.ALTER, (params, e) -> {
+ String tableName = params.tableName();
+
+ if (!name.equals(tableName))
+ return false;
+
+ if (e == null) {
+ tblFut.complete(null);
+ }
+ else
+ tblFut.completeExceptionally(e);
+
+ return true;
+ });
+
+ try {
+ configurationMgr.configurationRegistry()
+ .getConfiguration(TablesConfiguration.KEY).tables().change(change ->
+ change.update(name, tableChange)).get();
+ }
+ catch (InterruptedException | ExecutionException e) {
+ LOG.error("Table wasn't created [name=" + name + ']', e);
+
+ tblFut.completeExceptionally(e);
+ }
+
+ tblFut.join();
+ }
+
+ /** {@inheritDoc} */
@Override public void dropTable(String name) {
CompletableFuture<Void> dropTblFut = new CompletableFuture<>();
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/event/TableEvent.java b/modules/table/src/main/java/org/apache/ignite/internal/table/event/TableEvent.java
index f2a737e..94d508a 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/event/TableEvent.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/event/TableEvent.java
@@ -26,6 +26,9 @@ public enum TableEvent implements Event {
/** This event is fired when a table was created. */
CREATE,
+ /** This event is fired when a table config was changed. */
+ ALTER,
+
/** This event is fired when a table was dropped. */
DROP
}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummySchemaManagerImpl.java b/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummySchemaManagerImpl.java
index 9ecaec4..506b615 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummySchemaManagerImpl.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummySchemaManagerImpl.java
@@ -52,4 +52,9 @@ public class DummySchemaManagerImpl implements SchemaRegistry {
return schema;
}
+
+ /** {@inheritDoc} */
+ @Override public int lastSchemaVersion() {
+ return schema.version();
+ }
}