This is an automated email from the ASF dual-hosted git repository.
ibessonov 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 4c61ee327c IGNITE-19229 Schema validation during tx processing:
validators (#2872)
4c61ee327c is described below
commit 4c61ee327cd38483e7f431641f4dbdf1d3e4b3ac
Author: Roman Puchkovskiy <[email protected]>
AuthorDate: Tue Nov 28 11:03:08 2023 +0400
IGNITE-19229 Schema validation during tx processing: validators (#2872)
---
.../commands/AlterTableAlterColumnCommand.java | 30 +-
.../internal/catalog/commands/CatalogUtils.java | 73 +++
.../commands/TypeChangeValidationListener.java | 43 ++
.../ItSchemaForwardCompatibilityTest.java | 175 ++++++
.../replicator/PartitionReplicaListener.java | 2 +
.../replicator/SchemaCompatibilityValidator.java | 191 ++++++-
.../schema/CatalogValidationSchemasSource.java | 1 +
.../distributed/schema/ColumnDefinitionDiff.java | 51 +-
.../table/distributed/schema/FullTableSchema.java | 21 +-
.../distributed/schema/TableDefinitionDiff.java | 48 +-
.../replication/PartitionReplicaListenerTest.java | 4 +-
.../SchemaCompatibilityValidatorTest.java | 594 +++++++++++++++++++++
.../distributed/schema/FullTableSchemaTest.java | 36 +-
.../table/impl/DummyValidationSchemasSource.java | 6 +-
14 files changed, 1188 insertions(+), 87 deletions(-)
diff --git
a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/AlterTableAlterColumnCommand.java
b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/AlterTableAlterColumnCommand.java
index 1906ea3d59..6f7f3c42e1 100644
---
a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/AlterTableAlterColumnCommand.java
+++
b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/AlterTableAlterColumnCommand.java
@@ -44,6 +44,10 @@ public class AlterTableAlterColumnCommand extends
AbstractTableCommand {
return new Builder();
}
+ private static final TypeChangeValidationListener
TYPE_CHANGE_VALIDATION_HANDLER = (pattern, originalType, newType) -> {
+ throw new CatalogValidationException(format(pattern, originalType,
newType));
+ };
+
private final String columnName;
private final @Nullable ColumnType type;
@@ -145,31 +149,7 @@ public class AlterTableAlterColumnCommand extends
AbstractTableCommand {
}
private void validateColumnChange(CatalogTableColumnDescriptor origin) {
- if (type != null && type != origin.type()) {
- if (!CatalogUtils.isSupportedColumnTypeChange(origin.type(),
type)) {
- throw new CatalogValidationException(format("Changing the type
from {} to {} is not allowed", origin.type(), type));
- }
- }
-
- if (precision != null && precision != origin.precision() &&
origin.type() != ColumnType.DECIMAL) {
- throw new CatalogValidationException(format("Changing the
precision for column of type '{}' is not allowed", origin.type()));
- }
-
- if (precision != null && precision < origin.precision()) {
- throw new CatalogValidationException("Decreasing the precision is
not allowed");
- }
-
- if (scale != null && scale != origin.scale()) {
- throw new CatalogValidationException("Changing the scale is not
allowed");
- }
-
- if (length != null && length != origin.length() && origin.type() !=
ColumnType.STRING && origin.type() != ColumnType.BYTE_ARRAY) {
- throw new CatalogValidationException(format("Changing the length
for column of type '{}' is not allowed", origin.type()));
- }
-
- if (length != null && length < origin.length()) {
- throw new CatalogValidationException("Decreasing the length is not
allowed");
- }
+ CatalogUtils.validateColumnChange(origin, type, precision, scale,
length, TYPE_CHANGE_VALIDATION_HANDLER);
if (nullable != null && !nullable && origin.nullable()) {
throw new CatalogValidationException("Adding NOT NULL constraint
is not allowed");
diff --git
a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java
b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java
index 670c8eebb9..408c16647c 100644
---
a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java
+++
b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java
@@ -211,6 +211,79 @@ public class CatalogUtils {
return supportedTransitions != null &&
supportedTransitions.contains(target);
}
+ /**
+ * Validates a column change. If something is not valid, the supplied
listener is invoked with information about the exact reason.
+ *
+ * @param origin Original column definition.
+ * @param newType New type.
+ * @param newPrecision New column precision.
+ * @param newScale New column scale.
+ * @param newLength New column length.
+ * @param listener Listener to invoke on a validation failure.
+ * @return {@code true} iff the proposed change is valid.
+ */
+ static boolean validateColumnChange(
+ CatalogTableColumnDescriptor origin,
+ @Nullable ColumnType newType,
+ @Nullable Integer newPrecision,
+ @Nullable Integer newScale,
+ @Nullable Integer newLength,
+ TypeChangeValidationListener listener
+ ) {
+ if (newType != null && newType != origin.type()) {
+ if (!isSupportedColumnTypeChange(origin.type(), newType)) {
+ listener.onFailure("Changing the type from {} to {} is not
allowed", origin.type(), newType);
+ return false;
+ }
+ }
+
+ if (newPrecision != null && newPrecision != origin.precision() &&
origin.type() != ColumnType.DECIMAL) {
+ listener.onFailure("Changing the precision for column of type '{}'
is not allowed", origin.type(), newType);
+ return false;
+ }
+
+ if (newPrecision != null && newPrecision < origin.precision()) {
+ listener.onFailure("Decreasing the precision is not allowed",
origin.type(), newType);
+ return false;
+ }
+
+ if (newScale != null && newScale != origin.scale()) {
+ listener.onFailure("Changing the scale is not allowed",
origin.type(), newType);
+ return false;
+ }
+
+ if (newLength != null && newLength != origin.length()
+ && origin.type() != ColumnType.STRING && origin.type() !=
ColumnType.BYTE_ARRAY) {
+ listener.onFailure("Changing the length for column of type '{}' is
not allowed", origin.type(), newType);
+ return false;
+ }
+
+ if (newLength != null && newLength < origin.length()) {
+ listener.onFailure("Decreasing the length is not allowed",
origin.type(), newType);
+ return false;
+ }
+
+ return true;
+ }
+
+ /**
+ * Returns whether the proposed column type change is supported.
+ *
+ * @param oldColumn Original column definition.
+ * @param newColumn New column definition.
+ * @return {@code true} iff the proposed change is supported.
+ */
+ public static boolean
isColumnTypeChangeSupported(CatalogTableColumnDescriptor oldColumn,
CatalogTableColumnDescriptor newColumn) {
+ return validateColumnChange(
+ oldColumn,
+ newColumn.type(),
+ newColumn.precision(),
+ newColumn.scale(),
+ newColumn.length(),
+ TypeChangeValidationListener.NO_OP
+ );
+ }
+
/**
* Returns a list of schemas, replacing any schema with {@code newSchema}
if its id equal to {@code newSchema.id()}.
*
diff --git
a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/TypeChangeValidationListener.java
b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/TypeChangeValidationListener.java
new file mode 100644
index 0000000000..8910b48bc1
--- /dev/null
+++
b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/TypeChangeValidationListener.java
@@ -0,0 +1,43 @@
+/*
+ * 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.catalog.commands;
+
+import org.apache.ignite.sql.ColumnType;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Listener that gets invoked when column type change fails for some reason
with the exact reason of the failure.
+ */
+@SuppressWarnings("InterfaceMayBeAnnotatedFunctional")
+public interface TypeChangeValidationListener {
+ /**
+ * A listener that does nothing.
+ */
+ TypeChangeValidationListener NO_OP = (pattern, originalType, newType) -> {
+ // No-op.
+ };
+
+ /**
+ * Gets invoked on a validation failure.
+ *
+ * @param pattern String pattern that denotes the validation failure
reason (placeholders are marked with {}).
+ * @param originalType Original column type.
+ * @param newType New column type.
+ */
+ void onFailure(String pattern, ColumnType originalType, @Nullable
ColumnType newType);
+}
diff --git
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/schemasync/ItSchemaForwardCompatibilityTest.java
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/schemasync/ItSchemaForwardCompatibilityTest.java
new file mode 100644
index 0000000000..563b97b375
--- /dev/null
+++
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/schemasync/ItSchemaForwardCompatibilityTest.java
@@ -0,0 +1,175 @@
+/*
+ * 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.schemasync;
+
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.apache.ignite.internal.Cluster;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.lang.ErrorGroups.Transactions;
+import org.apache.ignite.table.Table;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.tx.Transaction;
+import org.apache.ignite.tx.TransactionException;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+/**
+ * Tests about forward compatibility of table schemas as defined by IEP-110.
+ *
+ * @see <a
href="https://cwiki.apache.org/confluence/display/IGNITE/IEP-110%3A+Schema+synchronization%3A+basic+schema+changes";>IEP-110</a>
+ */
+class ItSchemaForwardCompatibilityTest extends ClusterPerTestIntegrationTest {
+ private static final int NODES_TO_START = 1;
+
+ private static final String TABLE_NAME = "test";
+
+ private IgniteImpl node;
+
+ @Override
+ protected int initialNodes() {
+ return NODES_TO_START;
+ }
+
+ @BeforeEach
+ void assignNode() {
+ node = cluster.node(0);
+ }
+
+ /**
+ * Makes sure forward-compatible schema changes happenning between
transaction operations and
+ * commit do not prevent a commit from happening.
+ */
+ @ParameterizedTest
+ @EnumSource(ForwardCompatibleDdl.class)
+ void forwardCompatibleSchemaChangesAllowCommitting(ForwardCompatibleDdl
ddl) {
+ createTable();
+
+ Transaction tx = node.transactions().begin();
+
+ writeIn(tx);
+
+ ddl.executeOn(cluster);
+
+ assertDoesNotThrow(tx::commit);
+ }
+
+ @SuppressWarnings("resource")
+ private void writeIn(Transaction tx) {
+ putInTx(cluster.node(0).tables().table(TABLE_NAME), tx);
+ }
+
+ /**
+ * Makes sure forward-incompatible schema changes happenning between
transaction operations and
+ * commit prevent a commit from happening: instead, the transaction is
aborted.
+ */
+ @ParameterizedTest
+ @EnumSource(ForwardIncompatibleDdl.class)
+ void
forwardIncompatibleSchemaChangesDoNotAllowCommitting(ForwardIncompatibleDdl
ddl) {
+ createTable();
+
+ Table table = node.tables().table(TABLE_NAME);
+
+ InternalTransaction tx = (InternalTransaction)
node.transactions().begin();
+
+ writeIn(tx);
+
+ ddl.executeOn(cluster);
+
+ int tableId = ((TableViewInternal) table).tableId();
+
+ TransactionException ex = assertThrows(TransactionException.class,
tx::commit);
+ assertThat(
+ ex.getMessage(),
+ containsString(String.format(
+ "Commit failed because schema 1 is not
forward-compatible with 2 for table %d",
+ tableId
+ ))
+ );
+
+ assertThat(ex.code(), is(Transactions.TX_COMMIT_ERR));
+
+ assertThat(tx.state(), is(TxState.ABORTED));
+ }
+
+ private void createTable() {
+ cluster.doInSession(0, session -> {
+ executeUpdate(
+ "CREATE TABLE " + TABLE_NAME + " (id INT PRIMARY KEY,
not_null_int INT NOT NULL, int_with_default INT DEFAULT 1)",
+ session
+ );
+ });
+ }
+
+ private static void putInTx(Table table, Transaction tx) {
+ table.keyValueView().put(tx, Tuple.create().set("id", 1),
Tuple.create().set("not_null_int", 1));
+ }
+
+ private enum ForwardCompatibleDdl {
+ ADD_NULLABLE_COLUMN("ALTER TABLE " + TABLE_NAME + " ADD COLUMN new_col
INT"),
+ ADD_COLUMN_WITH_DEFAULT("ALTER TABLE " + TABLE_NAME + " ADD COLUMN
new_col INT NOT NULL DEFAULT 42"),
+ // TODO: IGNITE-19485, IGNITE-20315 - Uncomment this after column
rename support gets aded.
+ //RENAME_COLUMN("ALTER TABLE " + TABLE_NAME + " RENAME COLUMN
not_null_int to new_col"),
+ DROP_NOT_NULL("ALTER TABLE " + TABLE_NAME + " ALTER COLUMN
not_null_int DROP NOT NULL");
+ // TODO: Uncomment after
https://issues.apache.org/jira/browse/IGNITE-20906 is fixed.
+ //WIDEN_COLUMN_TYPE("ALTER TABLE " + TABLE_NAME + " ALTER COLUMN
not_null_int SET DATA TYPE BIGINT"),
+
+ private final String ddl;
+
+ ForwardCompatibleDdl(String ddl) {
+ this.ddl = ddl;
+ }
+
+ void executeOn(Cluster cluster) {
+ cluster.doInSession(0, session -> {
+ executeUpdate(ddl, session);
+ });
+ }
+ }
+
+ private enum ForwardIncompatibleDdl {
+ // TODO: Enable after
https://issues.apache.org/jira/browse/IGNITE-19484 is fixed.
+ //RENAME_TABLE("RENAME TABLE " + TABLE_NAME + " to new_table"),
+ DROP_COLUMN("ALTER TABLE " + TABLE_NAME + " DROP COLUMN not_null_int"),
+ ADD_DEFAULT("ALTER TABLE " + TABLE_NAME + " ALTER COLUMN not_null_int
SET DEFAULT 102"),
+ CHANGE_DEFAULT("ALTER TABLE " + TABLE_NAME + " ALTER COLUMN
int_with_default SET DEFAULT 102"),
+ DROP_DEFAULT("ALTER TABLE " + TABLE_NAME + " ALTER COLUMN
int_with_default DROP DEFAULT");
+
+ private final String ddl;
+
+ ForwardIncompatibleDdl(String ddl) {
+ this.ddl = ddl;
+ }
+
+ void executeOn(Cluster cluster) {
+ cluster.doInSession(0, session -> {
+ executeUpdate(ddl, session);
+ });
+ }
+ }
+}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
index ed862682fa..29e9483ef0 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
@@ -1472,10 +1472,12 @@ public class PartitionReplicaListener implements
ReplicaListener {
private static void
throwIfSchemaValidationOnCommitFailed(CompatValidationResult validationResult) {
if (!validationResult.isSuccessful()) {
if (validationResult.isTableDropped()) {
+ // TODO: IGNITE-20966 - improve error message.
throw new IncompatibleSchemaAbortException(
format("Commit failed because a table was already
dropped [tableId={}]", validationResult.failedTableId())
);
} else {
+ // TODO: IGNITE-20966 - improve error message.
throw new IncompatibleSchemaAbortException("Commit failed
because schema "
+ validationResult.fromSchemaVersion() + " is not
forward-compatible with "
+ validationResult.toSchemaVersion() + " for table " +
validationResult.failedTableId());
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/SchemaCompatibilityValidator.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/SchemaCompatibilityValidator.java
index 724cf28f1a..c90a8239e7 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/SchemaCompatibilityValidator.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/SchemaCompatibilityValidator.java
@@ -27,9 +27,11 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.ignite.internal.catalog.CatalogService;
+import
org.apache.ignite.internal.catalog.descriptors.CatalogTableColumnDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.replicator.TablePartitionId;
+import
org.apache.ignite.internal.table.distributed.schema.ColumnDefinitionDiff;
import org.apache.ignite.internal.table.distributed.schema.FullTableSchema;
import org.apache.ignite.internal.table.distributed.schema.SchemaSyncService;
import org.apache.ignite.internal.table.distributed.schema.TableDefinitionDiff;
@@ -47,6 +49,13 @@ class SchemaCompatibilityValidator {
// TODO: Remove entries from cache when compacting schemas in
SchemaManager https://issues.apache.org/jira/browse/IGNITE-20789
private final ConcurrentMap<TableDefinitionDiffKey, TableDefinitionDiff>
diffCache = new ConcurrentHashMap<>();
+ private static final List<ForwardCompatibilityValidator>
FORWARD_COMPATIBILITY_VALIDATORS = List.of(
+ new RenameTableValidator(),
+ new AddColumnsValidator(),
+ new DropColumnsValidator(),
+ new ChangeColumnsValidator()
+ );
+
/** Constructor. */
SchemaCompatibilityValidator(
ValidationSchemasSource validationSchemasSource,
@@ -149,8 +158,24 @@ class SchemaCompatibilityValidator {
key -> nextSchema.diffFrom(prevSchema)
);
- // TODO: IGNITE-19229 - more sophisticated logic.
- return diff.isEmpty();
+ boolean accepted = false;
+
+ for (ForwardCompatibilityValidator validator :
FORWARD_COMPATIBILITY_VALIDATORS) {
+ switch (validator.compatible(diff)) {
+ case COMPATIBLE:
+ accepted = true;
+ break;
+ case INCOMPATIBLE:
+ return false;
+ default:
+ break;
+ }
+ }
+
+ assert accepted : "Table schema changed from " +
prevSchema.schemaVersion() + " and " + nextSchema.schemaVersion()
+ + ", but no schema change validator voted for any change. Some
schema validator is missing.";
+
+ return true;
}
/**
@@ -187,25 +212,14 @@ class SchemaCompatibilityValidator {
tupleSchemaVersion
);
- if (tableSchemas.isEmpty()) {
+ if (tableSchemas.size() < 2) {
// The tuple was not written with a future schema.
return CompatValidationResult.success();
}
- for (int i = 0; i < tableSchemas.size() - 1; i++) {
- FullTableSchema oldSchema = tableSchemas.get(i);
- FullTableSchema newSchema = tableSchemas.get(i + 1);
- if (!isBackwardCompatible(oldSchema, newSchema)) {
- return CompatValidationResult.incompatibleChange(tableId,
oldSchema.schemaVersion(), newSchema.schemaVersion());
- }
- }
-
- return CompatValidationResult.success();
- }
-
- private boolean isBackwardCompatible(FullTableSchema oldSchema,
FullTableSchema newSchema) {
- // TODO: IGNITE-19229 - is backward compatibility always symmetric
with the forward compatibility?
- return isForwardCompatible(newSchema, oldSchema);
+ FullTableSchema oldSchema = tableSchemas.get(0);
+ FullTableSchema newSchema = tableSchemas.get(1);
+ return CompatValidationResult.incompatibleChange(tableId,
oldSchema.schemaVersion(), newSchema.schemaVersion());
}
void failIfSchemaChangedAfterTxStart(UUID txId, HybridTimestamp
operationTimestamp, int tableId) {
@@ -259,4 +273,147 @@ class SchemaCompatibilityValidator {
throw new InternalSchemaVersionMismatchException();
}
}
+
+ private enum ValidatorVerdict {
+ /**
+ * Validator accepts a change: it's compatible.
+ */
+ COMPATIBLE,
+ /**
+ * Validator rejects a change: it's incompatible.
+ */
+ INCOMPATIBLE,
+ /**
+ * Validator does not know how to handle a change.
+ */
+ DONT_CARE
+ }
+
+ @SuppressWarnings("InterfaceMayBeAnnotatedFunctional")
+ private interface ForwardCompatibilityValidator {
+ ValidatorVerdict compatible(TableDefinitionDiff diff);
+ }
+
+ private static class RenameTableValidator implements
ForwardCompatibilityValidator {
+ @Override
+ public ValidatorVerdict compatible(TableDefinitionDiff diff) {
+ return diff.nameDiffers() ? ValidatorVerdict.INCOMPATIBLE :
ValidatorVerdict.DONT_CARE;
+ }
+ }
+
+ private static class AddColumnsValidator implements
ForwardCompatibilityValidator {
+ @Override
+ public ValidatorVerdict compatible(TableDefinitionDiff diff) {
+ if (diff.addedColumns().isEmpty()) {
+ return ValidatorVerdict.DONT_CARE;
+ }
+
+ for (CatalogTableColumnDescriptor column : diff.addedColumns()) {
+ if (!column.nullable() && column.defaultValue() == null) {
+ return ValidatorVerdict.INCOMPATIBLE;
+ }
+ }
+
+ return ValidatorVerdict.COMPATIBLE;
+ }
+ }
+
+ private static class DropColumnsValidator implements
ForwardCompatibilityValidator {
+ @Override
+ public ValidatorVerdict compatible(TableDefinitionDiff diff) {
+ return diff.removedColumns().isEmpty() ?
ValidatorVerdict.DONT_CARE : ValidatorVerdict.INCOMPATIBLE;
+ }
+ }
+
+ @SuppressWarnings("InterfaceMayBeAnnotatedFunctional")
+ private interface ColumnChangeCompatibilityValidator {
+ ValidatorVerdict compatible(ColumnDefinitionDiff diff);
+ }
+
+ private static class ChangeColumnsValidator implements
ForwardCompatibilityValidator {
+ private final List<ColumnChangeCompatibilityValidator> validators =
List.of(
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20948 -
add validator that says that column rename is compatible.
+ new ChangeNullabilityValidator(),
+ new ChangeDefaultValueValidator(),
+ new ChangeColumnTypeValidator()
+ );
+
+ @Override
+ public ValidatorVerdict compatible(TableDefinitionDiff diff) {
+ if (diff.changedColumns().isEmpty()) {
+ return ValidatorVerdict.DONT_CARE;
+ }
+
+ boolean accepted = false;
+
+ for (ColumnDefinitionDiff columnDiff : diff.changedColumns()) {
+ switch (compatible(columnDiff)) {
+ case COMPATIBLE:
+ accepted = true;
+ break;
+ case INCOMPATIBLE:
+ return ValidatorVerdict.INCOMPATIBLE;
+ default:
+ break;
+ }
+ }
+
+ assert accepted : "Table schema changed from " +
diff.oldSchemaVersion() + " and " + diff.newSchemaVersion()
+ + ", but no column change validator voted for any change.
Some schema validator is missing.";
+
+ return ValidatorVerdict.COMPATIBLE;
+ }
+
+ private ValidatorVerdict compatible(ColumnDefinitionDiff columnDiff) {
+ boolean accepted = false;
+
+ for (ColumnChangeCompatibilityValidator validator : validators) {
+ switch (validator.compatible(columnDiff)) {
+ case COMPATIBLE:
+ accepted = true;
+ break;
+ case INCOMPATIBLE:
+ return ValidatorVerdict.INCOMPATIBLE;
+ default:
+ break;
+ }
+ }
+
+ return accepted ? ValidatorVerdict.COMPATIBLE :
ValidatorVerdict.DONT_CARE;
+ }
+ }
+
+ private static class ChangeDefaultValueValidator implements
ColumnChangeCompatibilityValidator {
+ @Override
+ public ValidatorVerdict compatible(ColumnDefinitionDiff diff) {
+ return diff.defaultChanged() ? ValidatorVerdict.INCOMPATIBLE :
ValidatorVerdict.DONT_CARE;
+ }
+ }
+
+ private static class ChangeNullabilityValidator implements
ColumnChangeCompatibilityValidator {
+ @Override
+ public ValidatorVerdict compatible(ColumnDefinitionDiff diff) {
+ if (diff.notNullAdded()) {
+ return ValidatorVerdict.INCOMPATIBLE;
+ }
+ if (diff.notNullDropped()) {
+ return ValidatorVerdict.COMPATIBLE;
+ }
+
+ assert !diff.nullabilityChanged() : diff;
+
+ return ValidatorVerdict.DONT_CARE;
+ }
+ }
+
+ private static class ChangeColumnTypeValidator implements
ColumnChangeCompatibilityValidator {
+ @Override
+ public ValidatorVerdict compatible(ColumnDefinitionDiff diff) {
+ if (!diff.typeChanged()) {
+ return ValidatorVerdict.DONT_CARE;
+ }
+
+ return diff.typeChangeIsSupported() ? ValidatorVerdict.COMPATIBLE
: ValidatorVerdict.INCOMPATIBLE;
+ }
+ }
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CatalogValidationSchemasSource.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CatalogValidationSchemasSource.java
index e0ca393a60..83b838b632 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CatalogValidationSchemasSource.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CatalogValidationSchemasSource.java
@@ -131,6 +131,7 @@ public class CatalogValidationSchemasSource implements
ValidationSchemasSource {
return new FullTableSchema(
tableDescriptor.tableVersion(),
tableDescriptor.id(),
+ tableDescriptor.name(),
tableDescriptor.columns()
);
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/ColumnDefinitionDiff.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/ColumnDefinitionDiff.java
index 4a5f79889e..0590ad4f2e 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/ColumnDefinitionDiff.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/ColumnDefinitionDiff.java
@@ -17,21 +17,64 @@
package org.apache.ignite.internal.table.distributed.schema;
+import java.util.Objects;
+import org.apache.ignite.internal.catalog.commands.CatalogUtils;
import
org.apache.ignite.internal.catalog.descriptors.CatalogTableColumnDescriptor;
/**
* Captures a difference between 'old' and 'new' versions of the same column
definition.
*/
public class ColumnDefinitionDiff {
- @SuppressWarnings("PMD.UnusedPrivateField")
private final CatalogTableColumnDescriptor oldColumn;
- @SuppressWarnings("PMD.UnusedPrivateField")
private final CatalogTableColumnDescriptor newColumn;
- // TODO: IGNITE-19229 - extend
-
public ColumnDefinitionDiff(CatalogTableColumnDescriptor oldColumn,
CatalogTableColumnDescriptor newColumn) {
this.oldColumn = oldColumn;
this.newColumn = newColumn;
}
+
+ /**
+ * Returns whether nullability has been changed on the column.
+ */
+ public boolean nullabilityChanged() {
+ return oldColumn.nullable() != newColumn.nullable();
+ }
+
+ /**
+ * Returns whether NOT NULL constraint has been dropped from the column.
+ */
+ public boolean notNullDropped() {
+ return !oldColumn.nullable() && newColumn.nullable();
+ }
+
+ /**
+ * Returns whether NOT NULL constraint has been added to the column.
+ */
+ public boolean notNullAdded() {
+ return oldColumn.nullable() && !newColumn.nullable();
+ }
+
+ /**
+ * Returns whether column type (including precision, scale, length) has
been changed.
+ */
+ public boolean typeChanged() {
+ return oldColumn.type() != newColumn.type()
+ || oldColumn.precision() != newColumn.precision()
+ || oldColumn.scale() != newColumn.scale()
+ || oldColumn.length() != newColumn.length();
+ }
+
+ /**
+ * Returns whether type change is supported.
+ */
+ public boolean typeChangeIsSupported() {
+ return CatalogUtils.isColumnTypeChangeSupported(oldColumn, newColumn);
+ }
+
+ /**
+ * Returns whether the default value has been changed.
+ */
+ public boolean defaultChanged() {
+ return !Objects.equals(oldColumn.defaultValue(),
newColumn.defaultValue());
+ }
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/FullTableSchema.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/FullTableSchema.java
index 630d3716a7..84a00601a5 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/FullTableSchema.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/FullTableSchema.java
@@ -38,15 +38,17 @@ import
org.apache.ignite.internal.catalog.descriptors.CatalogTableColumnDescript
public class FullTableSchema {
private final int schemaVersion;
private final int tableId;
+ private final String tableName;
private final List<CatalogTableColumnDescriptor> columns;
/**
* Constructor.
*/
- public FullTableSchema(int schemaVersion, int tableId,
List<CatalogTableColumnDescriptor> columns) {
+ public FullTableSchema(int schemaVersion, int tableId, String tableName,
List<CatalogTableColumnDescriptor> columns) {
this.schemaVersion = schemaVersion;
this.tableId = tableId;
+ this.tableName = tableName;
this.columns = List.copyOf(columns);
}
@@ -68,6 +70,13 @@ public class FullTableSchema {
return tableId;
}
+ /**
+ * Returns name of the table.
+ */
+ public String tableName() {
+ return tableName;
+ }
+
/**
* Returns definitions of the columns of the table.
*
@@ -101,7 +110,15 @@ public class FullTableSchema {
}
}
- return new TableDefinitionDiff(addedColumns, removedColumns,
changedColumns);
+ return new TableDefinitionDiff(
+ prevSchema.schemaVersion(),
+ this.schemaVersion,
+ prevSchema.tableName(),
+ this.tableName(),
+ addedColumns,
+ removedColumns,
+ changedColumns
+ );
}
private static <T> Map<String, T> toMapByName(List<T> elements,
Function<T, String> nameExtractor) {
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/TableDefinitionDiff.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/TableDefinitionDiff.java
index b4faf5ac64..28e634cc06 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/TableDefinitionDiff.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/TableDefinitionDiff.java
@@ -27,15 +27,17 @@ import
org.apache.ignite.internal.catalog.descriptors.CatalogTableColumnDescript
*/
public class TableDefinitionDiff {
private static final TableDefinitionDiff EMPTY = new TableDefinitionDiff(
- emptyList(), emptyList(), emptyList()
+ -1, -1, "name", "name", emptyList(), emptyList(), emptyList()
);
+ private final int oldSchemaVersion;
+ private final int newSchemaVersion;
+
+ private final boolean nameDiffers;
private final List<CatalogTableColumnDescriptor> addedColumns;
private final List<CatalogTableColumnDescriptor> removedColumns;
private final List<ColumnDefinitionDiff> changedColumns;
- // TODO: IGNITE-19229 - other change types
-
/**
* Returns an empty diff (meaning there is no difference).
*
@@ -49,15 +51,44 @@ public class TableDefinitionDiff {
* Constructor.
*/
public TableDefinitionDiff(
+ int oldSchemaVersion,
+ int newSchemaVersion,
+ String oldName,
+ String newName,
List<CatalogTableColumnDescriptor> addedColumns,
List<CatalogTableColumnDescriptor> removedColumns,
List<ColumnDefinitionDiff> changedColumns
) {
+ this.oldSchemaVersion = oldSchemaVersion;
+ this.newSchemaVersion = newSchemaVersion;
+
+ nameDiffers = !oldName.equals(newName);
this.addedColumns = List.copyOf(addedColumns);
this.removedColumns = List.copyOf(removedColumns);
this.changedColumns = List.copyOf(changedColumns);
}
+ /**
+ * Returns old schema version.
+ */
+ public int oldSchemaVersion() {
+ return oldSchemaVersion;
+ }
+
+ /**
+ * Returns new schema version.
+ */
+ public int newSchemaVersion() {
+ return newSchemaVersion;
+ }
+
+ /**
+ * Returns whether name of the table has been changed.
+ */
+ public boolean nameDiffers() {
+ return nameDiffers;
+ }
+
/**
* Returns columns that were added.
*/
@@ -78,15 +109,4 @@ public class TableDefinitionDiff {
public List<ColumnDefinitionDiff> changedColumns() {
return changedColumns;
}
-
- /**
- * Returns whether this diff is empty (so no difference at all).
- *
- * @return Whether this diff is empty (so no difference at all).
- */
- public boolean isEmpty() {
- return addedColumns.isEmpty()
- && removedColumns.isEmpty()
- && changedColumns.isEmpty();
- }
}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
index 261170da84..500c78ee67 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
@@ -1481,7 +1481,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
}
private static FullTableSchema tableSchema(int schemaVersion,
List<CatalogTableColumnDescriptor> columns) {
- return new FullTableSchema(schemaVersion, 1, columns);
+ return new FullTableSchema(schemaVersion, TABLE_ID, "test", columns);
}
private AtomicReference<Boolean> interceptFinishTxCommand() {
@@ -1518,11 +1518,11 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
}
@Test
- @Disabled("IGNITE-19229")
public void commitsOnCompatibleSchemaChangeSuccessfully() {
when(validationSchemasSource.tableSchemaVersionsBetween(anyInt(),
any(), any(HybridTimestamp.class)))
.thenReturn(List.of(
tableSchema(CURRENT_SCHEMA_VERSION,
List.of(nullableColumn("col1"))),
+ // Addition of a nullable column is forward-compatible.
tableSchema(FUTURE_SCHEMA_VERSION,
List.of(nullableColumn("col1"), nullableColumn("col2")))
));
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replicator/SchemaCompatibilityValidatorTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replicator/SchemaCompatibilityValidatorTest.java
new file mode 100644
index 0000000000..f3072bedc5
--- /dev/null
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replicator/SchemaCompatibilityValidatorTest.java
@@ -0,0 +1,594 @@
+/*
+ * 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.table.distributed.replicator;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.groupingBy;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static
org.apache.ignite.internal.catalog.commands.CatalogUtils.DEFAULT_LENGTH;
+import static
org.apache.ignite.internal.catalog.commands.CatalogUtils.DEFAULT_PRECISION;
+import static
org.apache.ignite.internal.catalog.commands.CatalogUtils.DEFAULT_SCALE;
+import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
+import static org.apache.ignite.sql.ColumnType.BITMASK;
+import static org.apache.ignite.sql.ColumnType.BOOLEAN;
+import static org.apache.ignite.sql.ColumnType.BYTE_ARRAY;
+import static org.apache.ignite.sql.ColumnType.DECIMAL;
+import static org.apache.ignite.sql.ColumnType.DOUBLE;
+import static org.apache.ignite.sql.ColumnType.DURATION;
+import static org.apache.ignite.sql.ColumnType.FLOAT;
+import static org.apache.ignite.sql.ColumnType.INT16;
+import static org.apache.ignite.sql.ColumnType.INT32;
+import static org.apache.ignite.sql.ColumnType.INT64;
+import static org.apache.ignite.sql.ColumnType.INT8;
+import static org.apache.ignite.sql.ColumnType.NUMBER;
+import static org.apache.ignite.sql.ColumnType.PERIOD;
+import static org.apache.ignite.sql.ColumnType.STRING;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.catalog.commands.DefaultValue;
+import
org.apache.ignite.internal.catalog.descriptors.CatalogTableColumnDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.IgniteBiTuple;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import
org.apache.ignite.internal.table.distributed.schema.AlwaysSyncedSchemaSyncService;
+import org.apache.ignite.internal.table.distributed.schema.FullTableSchema;
+import
org.apache.ignite.internal.table.distributed.schema.ValidationSchemasSource;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.internal.tx.TransactionIds;
+import org.apache.ignite.sql.ColumnType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class SchemaCompatibilityValidatorTest extends BaseIgniteAbstractTest {
+ @Mock
+ private ValidationSchemasSource schemasSource;
+
+ @Mock
+ private CatalogService catalogService;
+
+ private SchemaCompatibilityValidator validator;
+
+ private final HybridTimestamp beginTimestamp = new HybridTimestamp(1, 1);
+
+ private final HybridTimestamp commitTimestamp = new HybridTimestamp(2, 2);
+
+ private final UUID txId = TransactionIds.transactionId(beginTimestamp, 0);
+
+ private static final int TABLE_ID = 1;
+ private static final String TABLE_NAME = "test";
+ private static final String ANOTHER_NAME = "another";
+
+ private final TablePartitionId tablePartitionId = new
TablePartitionId(TABLE_ID, 0);
+
+ @BeforeEach
+ void createValidatorAndInitMocks() {
+ lenient().when(catalogService.table(TABLE_ID,
commitTimestamp.longValue())).thenReturn(mock(CatalogTableDescriptor.class));
+
+ validator = new SchemaCompatibilityValidator(schemasSource,
catalogService, new AlwaysSyncedSchemaSyncService());
+ }
+
+ @ParameterizedTest
+ @EnumSource(ForwardCompatibleChange.class)
+ void forwardCompatibleChangesAllowCommitting(ForwardCompatibleChange
change) {
+ assertForwardCompatibleChangeAllowsCommitting(change);
+ }
+
+ private void
assertForwardCompatibleChangeAllowsCommitting(SchemaChangeSource changeSource) {
+ when(schemasSource.tableSchemaVersionsBetween(TABLE_ID,
beginTimestamp, commitTimestamp))
+ .thenReturn(changeSource.schemaVersions());
+
+ CompletableFuture<CompatValidationResult> resultFuture =
validator.validateCommit(txId, List.of(tablePartitionId), commitTimestamp);
+
+ assertThat(resultFuture, willCompleteSuccessfully());
+
+ CompatValidationResult result = resultFuture.getNow(null);
+ assertThat(result, is(notNullValue()));
+
+ assertThat("Change is incompatible", result.isSuccessful(), is(true));
+ }
+
+ @ParameterizedTest
+ @EnumSource(ForwardIncompatibleChange.class)
+ void
forwardIncompatibleChangesDisallowCommitting(ForwardIncompatibleChange change) {
+ assertForwardIncompatibleChangeDisallowsCommitting(change);
+ }
+
+ private void
assertForwardIncompatibleChangeDisallowsCommitting(SchemaChangeSource
changeSource) {
+ when(schemasSource.tableSchemaVersionsBetween(TABLE_ID,
beginTimestamp, commitTimestamp))
+ .thenReturn(changeSource.schemaVersions());
+
+ CompletableFuture<CompatValidationResult> resultFuture =
validator.validateCommit(txId, List.of(tablePartitionId), commitTimestamp);
+
+ assertThat(resultFuture, willCompleteSuccessfully());
+
+ CompatValidationResult result = resultFuture.getNow(null);
+ assertThat(result, is(notNullValue()));
+
+ assertThat("Change is compatible", result.isSuccessful(), is(false));
+ assertThat(result.isTableDropped(), is(false));
+
+ assertThat(result.failedTableId(), is(TABLE_ID));
+ assertThat(result.fromSchemaVersion(), is(1));
+ assertThat(result.toSchemaVersion(), is(2));
+ }
+
+ @ParameterizedTest
+ @MethodSource("exactColumnTypeChanges")
+ void exactColumnTypeChangesAllowCommitting(ColumnTypeChange change) {
+ assertForwardCompatibleChangeAllowsCommitting(change);
+ }
+
+ private static Stream<Arguments> exactColumnTypeChanges() {
+ List<ColumnTypeChange> changes = new ArrayList<>();
+
+ for (IgniteBiTuple<ColumnType, ColumnType> pair :
simpleTypeCompatibleChanges()) {
+ //noinspection ConstantConditions
+ changes.add(new ColumnTypeChange(pair.get1(), pair.get2()));
+ }
+
+ // Increasing precision.
+ changes.add(new ColumnTypeChange(decimal(10, 5), decimal(11, 5)));
+
+ // Increasing length.
+ changes.add(new ColumnTypeChange(string(10), string(20)));
+ changes.add(new ColumnTypeChange(byteArray(10), byteArray(20)));
+
+ return changes.stream().map(Arguments::of);
+ }
+
+ private static List<IgniteBiTuple<ColumnType, ColumnType>>
simpleTypeCompatibleChanges() {
+ List<IgniteBiTuple<ColumnType, ColumnType>> changes = new
ArrayList<>();
+
+ List<ColumnType> intTypes = List.of(INT8, INT16, INT32, INT64);
+
+ // INT8->INT16->INT32->INT64.
+ for (int i = 0; i < intTypes.size() - 1; i++) {
+ ColumnType narrowerType = intTypes.get(i);
+
+ for (int j = i + 1; j < intTypes.size(); j++) {
+ ColumnType widerType = intTypes.get(j);
+ changes.add(new IgniteBiTuple<>(narrowerType, widerType));
+ }
+ }
+
+ changes.add(new IgniteBiTuple<>(FLOAT, DOUBLE));
+
+ return List.copyOf(changes);
+ }
+
+ private static Type decimal(int precision, int scale) {
+ return new Type(DECIMAL, precision, scale, DEFAULT_LENGTH);
+ }
+
+ private static Type number(int precision) {
+ return new Type(NUMBER, precision, DEFAULT_SCALE, DEFAULT_LENGTH);
+ }
+
+ private static Type string(int length) {
+ return new Type(STRING, DEFAULT_PRECISION, DEFAULT_SCALE, length);
+ }
+
+ private static Type byteArray(int length) {
+ return new Type(BYTE_ARRAY, DEFAULT_PRECISION, DEFAULT_SCALE, length);
+ }
+
+ @ParameterizedTest
+ @MethodSource("nonExactColumnTypeChanges")
+ void nonExactColumnTypeChangesDisallowCommitting(ColumnTypeChange change) {
+ assertForwardIncompatibleChangeDisallowsCommitting(change);
+ }
+
+ private static Stream<Arguments> nonExactColumnTypeChanges() {
+ List<ColumnTypeChange> changes = new ArrayList<>();
+
+ List<ColumnType> simpleTypes = Arrays.stream(ColumnType.values())
+ .filter(type -> !type.precisionAllowed() &&
!type.scaleAllowed() && !type.lengthAllowed())
+ .collect(toList());
+
+ List<IgniteBiTuple<ColumnType, ColumnType>>
simpleTypeCompatibleChanges = simpleTypeCompatibleChanges();
+ Map<ColumnType, Set<ColumnType>> typeToTypes =
simpleTypeCompatibleChanges.stream()
+ .collect(groupingBy(IgniteBiTuple::get1,
Collectors.mapping(IgniteBiTuple::get2, toSet())));
+
+ for (ColumnType type1 : simpleTypes) {
+ for (ColumnType type2 : simpleTypes) {
+ if (type1 == type2) {
+ continue;
+ }
+ Set<ColumnType> types = typeToTypes.get(type1);
+ if (types == null || !types.contains(type2)) {
+ changes.add(new ColumnTypeChange(type1, type2));
+ }
+ }
+ }
+
+ changes.add(new ColumnTypeChange(INT8, number(100)));
+ changes.add(new ColumnTypeChange(INT16, number(100)));
+ changes.add(new ColumnTypeChange(INT32, number(100)));
+ changes.add(new ColumnTypeChange(INT64, number(100)));
+ changes.add(new ColumnTypeChange(INT8, decimal(100, 0)));
+ changes.add(new ColumnTypeChange(INT16, decimal(100, 0)));
+ changes.add(new ColumnTypeChange(INT32, decimal(100, 0)));
+ changes.add(new ColumnTypeChange(INT64, decimal(100, 0)));
+
+ changes.add(new ColumnTypeChange(number(10), decimal(100, 0)));
+
+ // Decreasing precision.
+ changes.add(new ColumnTypeChange(decimal(10, 5), decimal(9, 5)));
+ changes.add(new ColumnTypeChange(number(10), number(9)));
+
+ // Decreasing length.
+ changes.add(new ColumnTypeChange(string(10), string(9)));
+ changes.add(new ColumnTypeChange(byteArray(10), byteArray(9)));
+
+ // Conversions to STRING.
+ changes.add(new ColumnTypeChange(INT8, string(100)));
+ changes.add(new ColumnTypeChange(INT16, string(100)));
+ changes.add(new ColumnTypeChange(INT32, string(100)));
+ changes.add(new ColumnTypeChange(INT64, string(100)));
+ changes.add(new ColumnTypeChange(decimal(10, 0), string(100)));
+ changes.add(new ColumnTypeChange(number(10), string(100)));
+ changes.add(new ColumnTypeChange(ColumnType.UUID, string(100)));
+
+ // Conversions from STRING.
+ changes.add(new ColumnTypeChange(string(1), INT8));
+ changes.add(new ColumnTypeChange(string(1), INT16));
+ changes.add(new ColumnTypeChange(string(1), INT32));
+ changes.add(new ColumnTypeChange(string(1), INT64));
+ changes.add(new ColumnTypeChange(string(1), decimal(10, 0)));
+ changes.add(new ColumnTypeChange(string(1), number(10)));
+ changes.add(new ColumnTypeChange(string(36), ColumnType.UUID));
+
+ for (ColumnType columnType : ColumnType.values()) {
+ addInconvertible(columnType, BOOLEAN, changes);
+ addInconvertible(columnType, ColumnType.UUID, changes);
+ addInconvertible(columnType, BITMASK, changes);
+ addInconvertible(columnType, BYTE_ARRAY, changes);
+ addInconvertible(columnType, PERIOD, changes);
+ addInconvertible(columnType, DURATION, changes);
+ }
+
+ return changes.stream().map(Arguments::of);
+ }
+
+ private static void addInconvertible(ColumnType type1, ColumnType type2,
List<ColumnTypeChange> changes) {
+ if (type1 != type2) {
+ changes.add(new ColumnTypeChange(type2, type1));
+ changes.add(new ColumnTypeChange(type1, type2));
+ }
+ }
+
+ @Test
+ void combinationOfForwardCompatibleChangesIsCompatible() {
+ assertForwardCompatibleChangeAllowsCommitting(() -> List.of(
+ tableSchema(1, List.of(column(INT32, false))),
+ // Type is widened, NOT NULL dropped.
+ tableSchema(2, List.of(column(INT64, true)))
+ ));
+ }
+
+ private static CatalogTableColumnDescriptor column(ColumnType type,
boolean nullable) {
+ return new CatalogTableColumnDescriptor(
+ "col",
+ type,
+ nullable,
+ DEFAULT_PRECISION,
+ DEFAULT_SCALE,
+ DEFAULT_LENGTH,
+ null
+ );
+ }
+
+ @Test
+ void oneForwardIncompatibleChangeMakesCombinationIncompatible() {
+ assertForwardIncompatibleChangeDisallowsCommitting(() -> List.of(
+ tableSchema(1, List.of(column(INT32, true))),
+ // Type is widened (compatible), but NOT NULL added
(incompatible).
+ tableSchema(2, List.of(column(INT64, false)))
+ ));
+ }
+
+ @Test
+ void forwardCompatibleChangeIsNotBackwardCompatible() {
+
assertChangeIsNotBackwardCompatible(ForwardCompatibleChange.ADD_NULLABLE_COLUMN);
+ }
+
+ @Test
+ void forwardIncompatibleChangeIsNotBackwardCompatible() {
+
assertChangeIsNotBackwardCompatible(ForwardIncompatibleChange.DROP_COLUMN);
+ }
+
+ @Test
+ void changeOppositeToForwardCompatibleChangeIsNotBackwardCompatible() {
+ assertChangeIsNotBackwardCompatible(() -> List.of(
+ tableSchema(1, List.of(
+ intColumn("col1"),
+ intColumnWithDefault("col2", 42)
+ )),
+ tableSchema(2, List.of(
+ intColumn("col1")
+ ))
+ ));
+ }
+
+ private void assertChangeIsNotBackwardCompatible(SchemaChangeSource
changeSource) {
+ when(schemasSource.tableSchemaVersionsBetween(TABLE_ID,
beginTimestamp, 2))
+ .thenReturn(changeSource.schemaVersions());
+ when(schemasSource.waitForSchemaAvailability(anyInt(),
anyInt())).thenReturn(completedFuture(null));
+
+ CompletableFuture<CompatValidationResult> resultFuture =
validator.validateBackwards(2, TABLE_ID, txId);
+
+ assertThat(resultFuture, willCompleteSuccessfully());
+
+ CompatValidationResult result = resultFuture.getNow(null);
+ assertThat(result, is(notNullValue()));
+
+ assertThat("Change is compatible", result.isSuccessful(), is(false));
+ assertThat(result.isTableDropped(), is(false));
+
+ assertThat(result.failedTableId(), is(TABLE_ID));
+ assertThat(result.fromSchemaVersion(), is(1));
+ assertThat(result.toSchemaVersion(), is(2));
+ }
+
+ private static CatalogTableColumnDescriptor intColumn(String columnName) {
+ return new CatalogTableColumnDescriptor(
+ columnName,
+ INT32,
+ false,
+ DEFAULT_PRECISION,
+ DEFAULT_SCALE,
+ DEFAULT_LENGTH,
+ null
+ );
+ }
+
+ private static CatalogTableColumnDescriptor nullableIntColumn(String
columnName) {
+ return new CatalogTableColumnDescriptor(columnName, INT32, true,
DEFAULT_PRECISION, DEFAULT_SCALE, DEFAULT_LENGTH, null);
+ }
+
+ private static CatalogTableColumnDescriptor intColumnWithDefault(String
columnName, int defaultValue) {
+ return new CatalogTableColumnDescriptor(
+ columnName,
+ INT32,
+ false,
+ DEFAULT_PRECISION,
+ DEFAULT_SCALE,
+ DEFAULT_LENGTH,
+ DefaultValue.constant(defaultValue)
+ );
+ }
+
+ private static FullTableSchema tableSchema(int schemaVersion,
List<CatalogTableColumnDescriptor> columns) {
+ return tableSchema(schemaVersion, TABLE_NAME, columns);
+ }
+
+ private static FullTableSchema tableSchema(int schemaVersion, String name,
List<CatalogTableColumnDescriptor> columns) {
+ return new FullTableSchema(schemaVersion, TABLE_ID, name, columns);
+ }
+
+ @FunctionalInterface
+ private interface SchemaChangeSource {
+ List<FullTableSchema> schemaVersions();
+ }
+
+ private enum ForwardCompatibleChange implements SchemaChangeSource {
+ ADD_NULLABLE_COLUMN(List.of(
+ tableSchema(1, List.of(
+ intColumn("col1")
+ )),
+ tableSchema(2, List.of(
+ intColumn("col1"),
+ nullableIntColumn("col2")
+ ))
+ )),
+ ADD_COLUMN_WITH_DEFAULT(List.of(
+ tableSchema(1, List.of(
+ intColumn("col1")
+ )),
+ tableSchema(2, List.of(
+ intColumn("col1"),
+ intColumnWithDefault("col2", 42)
+ ))
+ )),
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20948 -
uncomment this.
+ //RENAME_COLUMN(List.of(
+ // tableSchema(1, List.of(
+ // intColumn("col1")
+ // )),
+ // tableSchema(2, List.of(
+ // intColumn("col2")
+ // ))
+ //)),
+ DROP_NOT_NULL(List.of(
+ tableSchema(1, List.of(
+ intColumn("col1")
+ )),
+ tableSchema(2, List.of(
+ nullableIntColumn("col1")
+ ))
+ ));
+
+ private final List<FullTableSchema> schemaVersions;
+
+ ForwardCompatibleChange(List<FullTableSchema> schemaVersions) {
+ this.schemaVersions = schemaVersions;
+ }
+
+ @Override
+ public List<FullTableSchema> schemaVersions() {
+ return schemaVersions;
+ }
+ }
+
+ private enum ForwardIncompatibleChange implements SchemaChangeSource {
+ RENAME_TABLE(List.of(
+ tableSchema(1, TABLE_NAME, List.of(
+ intColumn("col1")
+ )),
+ tableSchema(2, ANOTHER_NAME, List.of(
+ intColumn("col1")
+ ))
+ )),
+ DROP_COLUMN(List.of(
+ tableSchema(1, List.of(
+ intColumn("col1"),
+ intColumn("col2")
+ )),
+ tableSchema(2, List.of(
+ intColumn("col1")
+ ))
+ )),
+ ADD_DEFAULT(List.of(
+ tableSchema(1, List.of(
+ intColumn("col1")
+ )),
+ tableSchema(2, List.of(
+ intColumnWithDefault("col1", 42)
+ ))
+ )),
+ CHANGE_DEFAULT(List.of(
+ tableSchema(1, List.of(
+ intColumnWithDefault("col1", 1)
+ )),
+ tableSchema(2, List.of(
+ intColumnWithDefault("col1", 2)
+ ))
+ )),
+ DROP_DEFAULT(List.of(
+ tableSchema(1, List.of(
+ intColumnWithDefault("col1", 42)
+ )),
+ tableSchema(2, List.of(
+ intColumn("col1")
+ ))
+ ));
+
+ private final List<FullTableSchema> schemaVersions;
+
+ ForwardIncompatibleChange(List<FullTableSchema> schemaVersions) {
+ this.schemaVersions = schemaVersions;
+ }
+
+ @Override
+ public List<FullTableSchema> schemaVersions() {
+ return schemaVersions;
+ }
+ }
+
+ private static class Type {
+ private final ColumnType columnType;
+ private final int precision;
+ private final int scale;
+ private final int length;
+
+ private Type(ColumnType columnType) {
+ this(columnType, DEFAULT_PRECISION, DEFAULT_SCALE, DEFAULT_LENGTH);
+ }
+
+ private Type(ColumnType columnType, int precision, int scale, int
length) {
+ this.columnType = columnType;
+ this.precision = precision;
+ this.scale = scale;
+ this.length = length;
+ }
+
+ @Override
+ public String toString() {
+ if (precision == DEFAULT_PRECISION && scale == DEFAULT_SCALE &&
length == DEFAULT_LENGTH) {
+ return columnType.toString();
+ } else if (precision == DEFAULT_PRECISION && scale ==
DEFAULT_SCALE) {
+ return String.format("%s(%d)", columnType, length);
+ } else {
+ return String.format("%s(%d/%d/%d)", columnType, precision,
scale, length);
+ }
+ }
+ }
+
+ private static class ColumnTypeChange implements SchemaChangeSource {
+ private final Type typeBefore;
+ private final Type typeAfter;
+
+ private ColumnTypeChange(ColumnType typeBefore, ColumnType typeAfter) {
+ this(new Type(typeBefore), new Type(typeAfter));
+ }
+
+ private ColumnTypeChange(ColumnType typeBefore, Type typeAfter) {
+ this(new Type(typeBefore), typeAfter);
+ }
+
+ private ColumnTypeChange(Type typeBefore, ColumnType typeAfter) {
+ this(typeBefore, new Type(typeAfter));
+ }
+
+ private ColumnTypeChange(Type typeBefore, Type typeAfter) {
+ this.typeBefore = typeBefore;
+ this.typeAfter = typeAfter;
+ }
+
+ @Override
+ public List<FullTableSchema> schemaVersions() {
+ return List.of(
+ tableSchema(1, List.of(columnFromType(typeBefore))),
+ tableSchema(2, List.of(columnFromType(typeAfter)))
+ );
+ }
+
+ private static CatalogTableColumnDescriptor columnFromType(Type type) {
+ return new CatalogTableColumnDescriptor(
+ "col",
+ type.columnType,
+ true,
+ type.precision,
+ type.scale,
+ type.length,
+ null
+ );
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s -> %s", typeBefore, typeAfter);
+ }
+ }
+}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/schema/FullTableSchemaTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/schema/FullTableSchemaTest.java
index a172121ccb..17cab220a6 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/schema/FullTableSchemaTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/schema/FullTableSchemaTest.java
@@ -29,17 +29,8 @@ import org.apache.ignite.sql.ColumnType;
import org.junit.jupiter.api.Test;
class FullTableSchemaTest {
- @Test
- void sameSchemasHaveEmptyDiff() {
- CatalogTableColumnDescriptor column = someColumn("a");
-
- var schema1 = new FullTableSchema(1, 1, List.of(column));
- var schema2 = new FullTableSchema(2, 1, List.of(column));
-
- TableDefinitionDiff diff = schema2.diffFrom(schema1);
-
- assertThat(diff.isEmpty(), is(true));
- }
+ private static final String TABLE_NAME1 = "test1";
+ private static final String TABLE_NAME2 = "test2";
private static CatalogTableColumnDescriptor someColumn(String columnName) {
return new CatalogTableColumnDescriptor(columnName, ColumnType.INT32,
true, 0, 0, 0, DefaultValue.constant(null));
@@ -51,12 +42,11 @@ class FullTableSchemaTest {
CatalogTableColumnDescriptor column2 = someColumn("b");
CatalogTableColumnDescriptor column3 = someColumn("c");
- var schema1 = new FullTableSchema(1, 1, List.of(column1, column2));
- var schema2 = new FullTableSchema(2, 1, List.of(column2, column3));
+ var schema1 = new FullTableSchema(1, 1, TABLE_NAME1, List.of(column1,
column2));
+ var schema2 = new FullTableSchema(2, 1, TABLE_NAME1, List.of(column2,
column3));
TableDefinitionDiff diff = schema2.diffFrom(schema1);
- assertThat(diff.isEmpty(), is(false));
assertThat(diff.addedColumns(), is(List.of(column3)));
assertThat(diff.removedColumns(), is(List.of(column1)));
assertThat(diff.changedColumns(), is(empty()));
@@ -66,16 +56,26 @@ class FullTableSchemaTest {
void changedColumnsAreReflectedInDiff() {
CatalogTableColumnDescriptor column1 = someColumn("a");
- var schema1 = new FullTableSchema(1, 1, List.of(column1));
- var schema2 = new FullTableSchema(2, 1,
+ var schema1 = new FullTableSchema(1, 1, TABLE_NAME1, List.of(column1));
+ var schema2 = new FullTableSchema(2, 1, TABLE_NAME1,
List.of(new CatalogTableColumnDescriptor("a",
ColumnType.STRING, true, 0, 0, 10, DefaultValue.constant(null)))
);
TableDefinitionDiff diff = schema2.diffFrom(schema1);
- assertThat(diff.isEmpty(), is(false));
-
List<ColumnDefinitionDiff> changedColumns = diff.changedColumns();
assertThat(changedColumns, is(hasSize(1)));
}
+
+ @Test
+ void changedNameIsReflected() {
+ CatalogTableColumnDescriptor column = someColumn("a");
+
+ var schema1 = new FullTableSchema(1, 1, TABLE_NAME1, List.of(column));
+ var schema2 = new FullTableSchema(1, 1, TABLE_NAME2, List.of(column));
+
+ TableDefinitionDiff diff = schema2.diffFrom(schema1);
+
+ assertThat(diff.nameDiffers(), is(true));
+ }
}
diff --git
a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyValidationSchemasSource.java
b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyValidationSchemasSource.java
index 4d171a88fc..2f38eff1f7 100644
---
a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyValidationSchemasSource.java
+++
b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyValidationSchemasSource.java
@@ -67,11 +67,7 @@ public class DummyValidationSchemasSource implements
ValidationSchemasSource {
})
.collect(toList());
- var fullSchema = new FullTableSchema(
- 1,
- 1,
- columns
- );
+ var fullSchema = new FullTableSchema(1, tableId, "test", columns);
return List.of(fullSchema);
}
