This is an automated email from the ASF dual-hosted git repository.
tkalkirill 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 3fc5202a78 IGNITE-19228 Schema validation during tx processing: common
framework (#2055)
3fc5202a78 is described below
commit 3fc5202a785b6c22ea9cda911d242d297c072067
Author: Roman Puchkovskiy <[email protected]>
AuthorDate: Fri May 12 16:29:29 2023 +0400
IGNITE-19228 Schema validation during tx processing: common framework
(#2055)
---
.../catalog/descriptors/TableColumnDescriptor.java | 46 +++++-
.../handler/requests/table/ClientTableCommon.java | 55 +-------
.../ignite/internal/util/CollectionUtils.java | 12 ++
.../ignite/internal/util/CollectionUtilsTest.java | 8 ++
.../asserts/CompletableFutureAssert.java | 77 ++++++++++
.../ignite/internal/schema/NativeTypeSpec.java | 80 +++++++++++
modules/table/build.gradle | 2 +
.../distributed/ItTxDistributedTestSingleNode.java | 2 +
.../internal/table/distributed/TableManager.java | 3 +
.../raft/RebalanceRaftGroupEventsListener.java | 14 +-
.../replicator/ForwardValidationResult.java | 112 +++++++++++++++
.../IncompatibleSchemaAbortException.java | 31 ++++
.../replicator/PartitionReplicaListener.java | 45 +++++-
.../replicator/SchemaCompatValidator.java | 117 ++++++++++++++++
.../distributed/schema/ColumnDefinitionDiff.java | 37 +++++
.../table/distributed/schema/FullTableSchema.java | 144 +++++++++++++++++++
.../distributed/schema/NonHistoricSchemas.java | 99 +++++++++++++
.../internal/table/distributed/schema/Schemas.java | 48 +++++++
.../distributed/schema/TableDefinitionDiff.java | 116 +++++++++++++++
.../table/distributed/schema/package-info.java | 22 +++
.../ignite/internal/utils/RebalanceUtil.java | 26 +---
.../RepeatedFinishReadWriteTransactionTest.java | 2 +-
.../PartitionReplicaListenerIndexLockingTest.java | 2 +
.../replication/PartitionReplicaListenerTest.java | 156 ++++++++++++++++++++-
.../distributed/schema/FullTableSchemaTest.java | 108 ++++++++++++++
.../table/impl/DummyInternalTableImpl.java | 21 ++-
.../ignite/internal/table/impl/DummySchemas.java | 73 ++++++++++
.../org/apache/ignite/internal/tx/TxManager.java | 4 +-
.../ignite/internal/tx/impl/TxManagerImpl.java | 3 +-
.../internal/tx/test/TestTransactionIds.java | 2 +-
30 files changed, 1369 insertions(+), 98 deletions(-)
diff --git
a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/TableColumnDescriptor.java
b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/TableColumnDescriptor.java
index 47133fb964..be1d86177c 100644
---
a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/TableColumnDescriptor.java
+++
b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/TableColumnDescriptor.java
@@ -36,7 +36,7 @@ public class TableColumnDescriptor implements Serializable {
private int length;
private int precision;
private int scale;
- private DefaultValue defaultValue;
+ private final DefaultValue defaultValue;
/**
* Constructor.
@@ -80,6 +80,50 @@ public class TableColumnDescriptor implements Serializable {
return defaultValue;
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ TableColumnDescriptor that = (TableColumnDescriptor) o;
+
+ if (nullable != that.nullable) {
+ return false;
+ }
+ if (length != that.length) {
+ return false;
+ }
+ if (precision != that.precision) {
+ return false;
+ }
+ if (scale != that.scale) {
+ return false;
+ }
+ if (!name.equals(that.name)) {
+ return false;
+ }
+ if (type != that.type) {
+ return false;
+ }
+ return defaultValue != null ? defaultValue.equals(that.defaultValue) :
that.defaultValue == null;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = name.hashCode();
+ result = 31 * result + type.hashCode();
+ result = 31 * result + (nullable ? 1 : 0);
+ result = 31 * result + length;
+ result = 31 * result + precision;
+ result = 31 * result + scale;
+ result = 31 * result + (defaultValue != null ? defaultValue.hashCode()
: 0);
+ return result;
+ }
+
/** {@inheritDoc} */
@Override
public String toString() {
diff --git
a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java
b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java
index 70354ff864..ac6c45a5a0 100644
---
a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java
+++
b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java
@@ -373,58 +373,13 @@ public class ClientTableCommon {
* @return Client type code.
*/
public static ColumnType getColumnType(NativeTypeSpec spec) {
- switch (spec) {
- case INT8:
- return ColumnType.INT8;
+ ColumnType columnType = spec.asColumnTypeOrNull();
- case INT16:
- return ColumnType.INT16;
-
- case INT32:
- return ColumnType.INT32;
-
- case INT64:
- return ColumnType.INT64;
-
- case FLOAT:
- return ColumnType.FLOAT;
-
- case DOUBLE:
- return ColumnType.DOUBLE;
-
- case DECIMAL:
- return ColumnType.DECIMAL;
-
- case NUMBER:
- return ColumnType.NUMBER;
-
- case UUID:
- return ColumnType.UUID;
-
- case STRING:
- return ColumnType.STRING;
-
- case BYTES:
- return ColumnType.BYTE_ARRAY;
-
- case BITMASK:
- return ColumnType.BITMASK;
-
- case DATE:
- return ColumnType.DATE;
-
- case TIME:
- return ColumnType.TIME;
-
- case DATETIME:
- return ColumnType.DATETIME;
-
- case TIMESTAMP:
- return ColumnType.TIMESTAMP;
-
- default:
- throw new IgniteException(PROTOCOL_ERR, "Unsupported native
type: " + spec);
+ if (columnType == null) {
+ throw new IgniteException(PROTOCOL_ERR, "Unsupported native type:
" + spec);
}
+
+ return columnType;
}
/**
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/util/CollectionUtils.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/CollectionUtils.java
index facc79eb6a..c1e182faee 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/util/CollectionUtils.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/util/CollectionUtils.java
@@ -22,6 +22,7 @@ import static java.util.Collections.emptyIterator;
import static java.util.Collections.emptyList;
import static java.util.Collections.unmodifiableCollection;
import static java.util.Collections.unmodifiableSet;
+import static java.util.stream.Collectors.toSet;
import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
import it.unimi.dsi.fastutil.ints.IntSet;
@@ -596,4 +597,15 @@ public final class CollectionUtils {
public static IntSet setOf(Collection<Integer> coll) {
return IntSets.unmodifiable(new IntOpenHashSet(coll));
}
+
+ /**
+ * Returns an intersection of two sets.
+ *
+ * @param op1 First operand.
+ * @param op2 Second operand.
+ * @return Result of the intersection.
+ */
+ public static <T> Set<T> intersect(Set<T> op1, Set<T> op2) {
+ return op1.stream().filter(op2::contains).collect(toSet());
+ }
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/util/CollectionUtilsTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/util/CollectionUtilsTest.java
index f4e226e159..cbd508f211 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/util/CollectionUtilsTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/util/CollectionUtilsTest.java
@@ -22,11 +22,13 @@ import static java.util.function.Function.identity;
import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.util.CollectionUtils.concat;
import static org.apache.ignite.internal.util.CollectionUtils.difference;
+import static org.apache.ignite.internal.util.CollectionUtils.intersect;
import static org.apache.ignite.internal.util.CollectionUtils.setOf;
import static org.apache.ignite.internal.util.CollectionUtils.union;
import static org.apache.ignite.internal.util.CollectionUtils.viewReadOnly;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -359,4 +361,10 @@ public class CollectionUtilsTest {
assertThrows(UnsupportedOperationException.class, () -> copy.add(42));
assertThrows(UnsupportedOperationException.class, () ->
copy.remove(3));
}
+
+ @Test
+ void testIntersect() {
+ assertThat(intersect(Set.of(1, 2), Set.of(2, 3)), is(Set.of(2)));
+ assertThat(intersect(Set.of(1, 2), Set.of(3, 4)), is(Set.of()));
+ }
}
diff --git
a/modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/asserts/CompletableFutureAssert.java
b/modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/asserts/CompletableFutureAssert.java
new file mode 100644
index 0000000000..10a61937d0
--- /dev/null
+++
b/modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/asserts/CompletableFutureAssert.java
@@ -0,0 +1,77 @@
+/*
+ * 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.testframework.asserts;
+
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Assertions related to {@link CompletableFuture}.
+ */
+public class CompletableFutureAssert {
+ /**
+ * Asserts that the given future completes with an exception being an
instance of the given class and returns
+ * that exception for further examination.
+ *
+ * <p>Unlike {@link
org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher#willThrowFast(Class)},
+ * this method allows to examine the actual exception thrown further in
the test.
+ *
+ * @param future Future to work on.
+ * @param expectedExceptionClass Expected class of the exception.
+ * @param <X> Exception type.
+ * @return Matched exception.
+ */
+ public static <X extends Throwable> X assertWillThrowFast(
+ CompletableFuture<?> future,
+ Class<X> expectedExceptionClass
+ ) {
+ try {
+ Object normalResult = future.get(1, TimeUnit.SECONDS);
+
+ return fail("Expected the future to be completed with an exception
of class " + expectedExceptionClass
+ + ", but it completed normally with result " +
normalResult);
+ } catch (TimeoutException e) {
+ return fail("Expected the future to be completed with an exception
of class in 1 second, but it did not complete in time");
+ } catch (Throwable e) {
+ Throwable unwrapped = unwrapThrowable(e);
+
+ if (expectedExceptionClass.isInstance(unwrapped)) {
+ return (X) unwrapped;
+ } else {
+ return fail(
+ "Expected the future to be completed with an exception
of class " + expectedExceptionClass.getName() + ", but got "
+ + e.getClass().getName(),
+ e
+ );
+ }
+ }
+ }
+
+ private static Throwable unwrapThrowable(Throwable e) {
+ while (e instanceof ExecutionException || e instanceof
CompletionException) {
+ e = e.getCause();
+ }
+
+ return e;
+ }
+}
diff --git
a/modules/schema/src/main/java/org/apache/ignite/internal/schema/NativeTypeSpec.java
b/modules/schema/src/main/java/org/apache/ignite/internal/schema/NativeTypeSpec.java
index a87a651f13..b7b6e732a2 100644
---
a/modules/schema/src/main/java/org/apache/ignite/internal/schema/NativeTypeSpec.java
+++
b/modules/schema/src/main/java/org/apache/ignite/internal/schema/NativeTypeSpec.java
@@ -26,6 +26,10 @@ import java.time.LocalTime;
import java.util.BitSet;
import org.apache.ignite.internal.schema.row.InternalTuple;
import org.apache.ignite.internal.tostring.S;
+import org.apache.ignite.lang.ErrorGroups.Common;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.sql.ColumnType;
+import org.jetbrains.annotations.Nullable;
/**
* Base class for storage built-in data types definition. The class contains
predefined values for fixed-sized types and some of the
@@ -355,6 +359,82 @@ public enum NativeTypeSpec {
}
}
+ /**
+ * Gets client type corresponding to this server type.
+ *
+ * @return Client type code.
+ */
+ public ColumnType asColumnType() {
+ ColumnType columnType = asColumnTypeOrNull();
+
+ if (columnType == null) {
+ throw new IgniteException(Common.UNEXPECTED_ERR, "Unsupported
native type: " + this);
+ }
+
+ return columnType;
+ }
+
+ /**
+ * Gets client type corresponding to this server type.
+ *
+ * @return Client type code.
+ */
+ @Nullable
+ public ColumnType asColumnTypeOrNull() {
+ switch (this) {
+ case INT8:
+ return ColumnType.INT8;
+
+ case INT16:
+ return ColumnType.INT16;
+
+ case INT32:
+ return ColumnType.INT32;
+
+ case INT64:
+ return ColumnType.INT64;
+
+ case FLOAT:
+ return ColumnType.FLOAT;
+
+ case DOUBLE:
+ return ColumnType.DOUBLE;
+
+ case DECIMAL:
+ return ColumnType.DECIMAL;
+
+ case NUMBER:
+ return ColumnType.NUMBER;
+
+ case UUID:
+ return ColumnType.UUID;
+
+ case STRING:
+ return ColumnType.STRING;
+
+ case BYTES:
+ return ColumnType.BYTE_ARRAY;
+
+ case BITMASK:
+ return ColumnType.BITMASK;
+
+ case DATE:
+ return ColumnType.DATE;
+
+ case TIME:
+ return ColumnType.TIME;
+
+ case DATETIME:
+ return ColumnType.DATETIME;
+
+ case TIMESTAMP:
+ return ColumnType.TIMESTAMP;
+
+ default:
+ return null;
+ }
+ }
+
/**
* Maps object to native type.
*
diff --git a/modules/table/build.gradle b/modules/table/build.gradle
index 70611297f3..d66a209095 100644
--- a/modules/table/build.gradle
+++ b/modules/table/build.gradle
@@ -42,6 +42,7 @@ dependencies {
implementation project(':ignite-distribution-zones')
implementation project(':ignite-vault')
implementation project(':ignite-cluster-management')
+ implementation project(':ignite-catalog')
implementation libs.jetbrains.annotations
implementation libs.fastutil.core
implementation libs.auto.service.annotations
@@ -81,6 +82,7 @@ dependencies {
testFixturesImplementation project(':ignite-raft-api')
testFixturesImplementation project(':ignite-replicator')
testFixturesImplementation project(':ignite-configuration')
+ testFixturesImplementation project(':ignite-catalog')
testFixturesImplementation(testFixtures(project(':ignite-core')))
testFixturesImplementation(testFixtures(project(':ignite-storage-api')))
testFixturesImplementation(testFixtures(project(':ignite-transactions')))
diff --git
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
index 195e11ec86..b1b0ac602f 100644
---
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
+++
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
@@ -96,6 +96,7 @@ import
org.apache.ignite.internal.table.distributed.replicator.PlacementDriver;
import org.apache.ignite.internal.table.distributed.storage.InternalTableImpl;
import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.table.impl.DummySchemaManagerImpl;
+import org.apache.ignite.internal.table.impl.DummySchemas;
import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.TxManager;
@@ -503,6 +504,7 @@ public class ItTxDistributedTestSingleNode extends
TxAbstractTest {
txStateStorage,
placementDriver,
storageUpdateHandler,
+ new
DummySchemas(schemaManager),
peer ->
assignment.equals(peer.consistentId()),
completedFuture(schemaManager)
),
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 a10ab28812..dda37dce6c 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
@@ -143,6 +143,7 @@ import
org.apache.ignite.internal.table.distributed.raft.snapshot.outgoing.Outgo
import
org.apache.ignite.internal.table.distributed.raft.snapshot.outgoing.SnapshotAwarePartitionDataStorage;
import
org.apache.ignite.internal.table.distributed.replicator.PartitionReplicaListener;
import org.apache.ignite.internal.table.distributed.replicator.PlacementDriver;
+import org.apache.ignite.internal.table.distributed.schema.NonHistoricSchemas;
import org.apache.ignite.internal.table.distributed.storage.InternalTableImpl;
import org.apache.ignite.internal.table.distributed.storage.PartitionStorages;
import org.apache.ignite.internal.table.event.TableEvent;
@@ -826,6 +827,7 @@ public class TableManager extends Producer<TableEvent,
TableEventParameters> imp
txStateStorage,
placementDriver,
storageUpdateHandler,
+ new
NonHistoricSchemas(schemaManager),
this::isLocalPeer,
schemaManager.schemaRegistry(causalityToken, tblId)
),
@@ -2098,6 +2100,7 @@ public class TableManager extends Producer<TableEvent,
TableEventParameters> imp
txStatePartitionStorage,
placementDriver,
storageUpdateHandler,
+ new
NonHistoricSchemas(schemaManager),
this::isLocalPeer,
completedFuture(schemaManager.schemaRegistry(tblId))
),
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/RebalanceRaftGroupEventsListener.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/RebalanceRaftGroupEventsListener.java
index a8324a8cd2..b7fb7d9bc0 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/RebalanceRaftGroupEventsListener.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/RebalanceRaftGroupEventsListener.java
@@ -24,11 +24,11 @@ import static
org.apache.ignite.internal.metastorage.dsl.Operations.ops;
import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
import static org.apache.ignite.internal.metastorage.dsl.Operations.remove;
import static org.apache.ignite.internal.metastorage.dsl.Statements.iif;
-import static org.apache.ignite.internal.utils.RebalanceUtil.intersect;
+import static org.apache.ignite.internal.util.CollectionUtils.difference;
+import static org.apache.ignite.internal.util.CollectionUtils.intersect;
import static
org.apache.ignite.internal.utils.RebalanceUtil.pendingPartAssignmentsKey;
import static
org.apache.ignite.internal.utils.RebalanceUtil.plannedPartAssignmentsKey;
import static
org.apache.ignite.internal.utils.RebalanceUtil.stablePartAssignmentsKey;
-import static org.apache.ignite.internal.utils.RebalanceUtil.subtract;
import static org.apache.ignite.internal.utils.RebalanceUtil.switchAppendKey;
import static org.apache.ignite.internal.utils.RebalanceUtil.switchReduceKey;
import static org.apache.ignite.internal.utils.RebalanceUtil.union;
@@ -336,20 +336,20 @@ public class RebalanceRaftGroupEventsListener implements
RaftGroupEventsListener
Set<Assignment> stable = createAssignments(configuration);
// Were reduced
- Set<Assignment> reducedNodes = subtract(retrievedSwitchReduce,
stable);
+ Set<Assignment> reducedNodes = difference(retrievedSwitchReduce,
stable);
// Were added
- Set<Assignment> addedNodes = subtract(stable, retrievedStable);
+ Set<Assignment> addedNodes = difference(stable, retrievedStable);
// For further reduction
- Set<Assignment> calculatedSwitchReduce =
subtract(retrievedSwitchReduce, reducedNodes);
+ Set<Assignment> calculatedSwitchReduce =
difference(retrievedSwitchReduce, reducedNodes);
// For further addition
Set<Assignment> calculatedSwitchAppend =
union(retrievedSwitchAppend, reducedNodes);
- calculatedSwitchAppend = subtract(calculatedSwitchAppend,
addedNodes);
+ calculatedSwitchAppend = difference(calculatedSwitchAppend,
addedNodes);
calculatedSwitchAppend = intersect(calculatedAssignments,
calculatedSwitchAppend);
- Set<Assignment> calculatedPendingReduction = subtract(stable,
retrievedSwitchReduce);
+ Set<Assignment> calculatedPendingReduction = difference(stable,
retrievedSwitchReduce);
Set<Assignment> calculatedPendingAddition = union(stable,
reducedNodes);
calculatedPendingAddition = intersect(calculatedAssignments,
calculatedPendingAddition);
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/ForwardValidationResult.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/ForwardValidationResult.java
new file mode 100644
index 0000000000..560615dce6
--- /dev/null
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/ForwardValidationResult.java
@@ -0,0 +1,112 @@
+/*
+ * 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 java.util.Objects;
+import java.util.UUID;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Result of forward schema compatibility validation.
+ */
+public class ForwardValidationResult {
+ private static final ForwardValidationResult SUCCESS = new
ForwardValidationResult(true, null, -1, -1);
+
+ private final boolean ok;
+ @Nullable
+ private final UUID failedTableId;
+ private final int fromSchemaVersion;
+ private final int toSchemaVersion;
+
+ /**
+ * Returns a successful validation result.
+ *
+ * @return A successful validation result.
+ */
+ public static ForwardValidationResult success() {
+ return SUCCESS;
+ }
+
+ /**
+ * Creates a validation result denoting validation failure.
+ *
+ * @param failedTableId Table which schema change is incompatible.
+ * @param fromSchemaVersion Version number of the schema from which an
incompatible transition tried to be made.
+ * @param toSchemaVersion Version number of the schema to which an
incompatible transition tried to be made.
+ * @return A validation result for a failure.
+ */
+ public static ForwardValidationResult failure(UUID failedTableId, int
fromSchemaVersion, int toSchemaVersion) {
+ return new ForwardValidationResult(false, failedTableId,
fromSchemaVersion, toSchemaVersion);
+ }
+
+ private ForwardValidationResult(boolean ok, @Nullable UUID failedTableId,
int fromSchemaVersion, int toSchemaVersion) {
+ this.ok = ok;
+ this.failedTableId = failedTableId;
+ this.fromSchemaVersion = fromSchemaVersion;
+ this.toSchemaVersion = toSchemaVersion;
+ }
+
+ /**
+ * Returns whether the validation was successful.
+ *
+ * @return Whether the validation was successful
+ */
+ public boolean isSuccessful() {
+ return ok;
+ }
+
+ /**
+ * Returns ID of the table for which the validation has failed. Should
only be called for a failed validation result,
+ * otherwise an exception is thrown.
+ *
+ * @return Table ID.
+ */
+ public UUID failedTableId() {
+ return Objects.requireNonNull(failedTableId);
+ }
+
+ /**
+ * Returns version number of the schema from which an incompatible
transition tried to be made. Should only be called for a failed
+ * validation result, otherwise an exception is thrown.
+ *
+ * @return Version number of the schema from which an incompatible
transition tried to be made.
+ */
+ public int fromSchemaVersion() {
+ throwIfSuccessful();
+
+ return fromSchemaVersion;
+ }
+
+ private void throwIfSuccessful() {
+ if (ok) {
+ throw new IllegalStateException("Should not be called on a
successful result");
+ }
+ }
+
+ /**
+ * Returns version number of the schema to which an incompatible
transition tried to be made. Should only be called for a failed
+ * * validation result, otherwise an exception is thrown.
+ *
+ * @return Version number of the schema to which an incompatible
transition tried to be made.
+ */
+ public int toSchemaVersion() {
+ throwIfSuccessful();
+
+ return toSchemaVersion;
+ }
+}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/IncompatibleSchemaAbortException.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/IncompatibleSchemaAbortException.java
new file mode 100644
index 0000000000..4855ee00bf
--- /dev/null
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/IncompatibleSchemaAbortException.java
@@ -0,0 +1,31 @@
+/*
+ * 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 org.apache.ignite.lang.ErrorGroups.Transactions;
+import org.apache.ignite.tx.TransactionException;
+
+/**
+ * Thrown when, during an attempt to commit a transaction, it turns out that
the transaction cannot be committed
+ * because an incompatible schema change has happened.
+ */
+public class IncompatibleSchemaAbortException extends TransactionException {
+ public IncompatibleSchemaAbortException(String message) {
+ super(Transactions.TX_COMMIT_ERR, message);
+ }
+}
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 6cafe2c33f..a232111407 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
@@ -21,6 +21,7 @@ import static java.util.Objects.requireNonNull;
import static java.util.concurrent.CompletableFuture.allOf;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
import static org.apache.ignite.internal.util.IgniteUtils.filter;
import static org.apache.ignite.internal.util.IgniteUtils.findAny;
@@ -56,7 +57,6 @@ import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.raft.Peer;
import org.apache.ignite.internal.raft.service.RaftGroupService;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.TablePartitionId;
import
org.apache.ignite.internal.replicator.exception.PrimaryReplicaMissException;
import org.apache.ignite.internal.replicator.exception.ReplicationException;
@@ -102,6 +102,7 @@ import
org.apache.ignite.internal.table.distributed.replication.request.ReadWrit
import
org.apache.ignite.internal.table.distributed.replication.request.ReadWriteSingleRowReplicaRequest;
import
org.apache.ignite.internal.table.distributed.replication.request.ReadWriteSwapRowReplicaRequest;
import
org.apache.ignite.internal.table.distributed.replicator.action.RequestType;
+import org.apache.ignite.internal.table.distributed.schema.Schemas;
import org.apache.ignite.internal.tx.Lock;
import org.apache.ignite.internal.tx.LockKey;
import org.apache.ignite.internal.tx.LockManager;
@@ -204,6 +205,8 @@ public class PartitionReplicaListener implements
ReplicaListener {
private final CompletableFuture<SchemaRegistry> schemaFut;
+ private final SchemaCompatValidator schemaCompatValidator;
+
/**
* The constructor.
*
@@ -240,6 +243,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
TxStateStorage txStateStorage,
PlacementDriver placementDriver,
StorageUpdateHandler storageUpdateHandler,
+ Schemas schemas,
Function<Peer, Boolean> isLocalPeerChecker,
CompletableFuture<SchemaRegistry> schemaFut
) {
@@ -264,6 +268,8 @@ public class PartitionReplicaListener implements
ReplicaListener {
this.replicationGroupId = new TablePartitionId(tableId, partId);
cursors = new
ConcurrentSkipListMap<>(IgniteUuid.globalOrderComparator());
+
+ schemaCompatValidator = new SchemaCompatValidator(schemas);
}
@Override
@@ -964,6 +970,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
* Processes transaction finish request.
* <ol>
* <li>Get commit timestamp from finish replica request.</li>
+ * <li>If attempting a commit, validate commit (and, if not valid,
switch to abort)</li>
* <li>Run specific raft {@code FinishTxCommand} command, that will
apply txn state to corresponding txStateStorage.</li>
* <li>Send cleanup requests to all enlisted primary replicas.</li>
* </ol>
@@ -976,13 +983,37 @@ public class PartitionReplicaListener implements
ReplicaListener {
List<TablePartitionId> aggregatedGroupIds =
request.groups().values().stream()
.flatMap(List::stream)
.map(IgniteBiTuple::get1)
- .collect(Collectors.toList());
+ .collect(toList());
UUID txId = request.txId();
- boolean commit = request.commit();
+ if (request.commit()) {
+ return schemaCompatValidator.validateForwards(txId,
aggregatedGroupIds, request.commitTimestamp())
+ .thenCompose(validationResult -> {
+ return finishAndCleanup(request,
validationResult.isSuccessful(), aggregatedGroupIds, txId)
+ .thenAccept(unused ->
throwIfValidationFailed(validationResult));
+ });
+ } else {
+ // Aborting.
+ return finishAndCleanup(request, false, aggregatedGroupIds, txId);
+ }
+ }
+
+ private static void throwIfValidationFailed(ForwardValidationResult
validationResult) {
+ if (!validationResult.isSuccessful()) {
+ throw new IncompatibleSchemaAbortException("Commit failed because
schema "
+ + validationResult.fromSchemaVersion() + " is not
forward-compatible with "
+ + validationResult.toSchemaVersion() + " for table " +
validationResult.failedTableId());
+ }
+ }
- CompletableFuture<Object> changeStateFuture =
finishTransaction(aggregatedGroupIds, txId, commit);
+ private CompletableFuture<Void> finishAndCleanup(
+ TxFinishReplicaRequest request,
+ boolean commit,
+ List<TablePartitionId> aggregatedGroupIds,
+ UUID txId
+ ) {
+ CompletableFuture<?> changeStateFuture =
finishTransaction(aggregatedGroupIds, txId, commit);
// TODO: https://issues.apache.org/jira/browse/IGNITE-17578 Cleanup
process should be asynchronous.
CompletableFuture<?>[] cleanupFutures = new
CompletableFuture[request.groups().size()];
@@ -1028,7 +1059,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
.tablePartitionIds(
aggregatedGroupIds.stream()
.map(PartitionReplicaListener::tablePartitionId)
- .collect(Collectors.toList())
+ .collect(toList())
);
if (commit) {
@@ -2118,7 +2149,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
* @return Future with boolean value, indicating whether the transaction
was committed before timestamp.
*/
private CompletableFuture<Boolean> resolveTxState(
- ReplicationGroupId commitGrpId,
+ TablePartitionId commitGrpId,
UUID txId,
HybridTimestamp timestamp
) {
@@ -2162,7 +2193,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
* @param txId Transaction ID.
* @return Constructed {@link UpdateCommand} object.
*/
- private UpdateCommand updateCommand(TablePartitionId tablePartId, UUID
rowUuid, ByteBuffer rowBuf, UUID txId) {
+ private UpdateCommand updateCommand(TablePartitionId tablePartId, UUID
rowUuid, @Nullable ByteBuffer rowBuf, UUID txId) {
UpdateCommandBuilder bldr = MSG_FACTORY.updateCommand()
.tablePartitionId(tablePartitionId(tablePartId))
.rowUuid(rowUuid)
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/SchemaCompatValidator.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/SchemaCompatValidator.java
new file mode 100644
index 0000000000..f16a231bfe
--- /dev/null
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/SchemaCompatValidator.java
@@ -0,0 +1,117 @@
+/*
+ * 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.stream.Collectors.toSet;
+
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.distributed.schema.FullTableSchema;
+import org.apache.ignite.internal.table.distributed.schema.Schemas;
+import org.apache.ignite.internal.table.distributed.schema.TableDefinitionDiff;
+import org.apache.ignite.internal.tx.TransactionIds;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Validates schema compatibility.
+ */
+class SchemaCompatValidator {
+ private final Schemas schemas;
+
+ SchemaCompatValidator(Schemas schemas) {
+ this.schemas = schemas;
+ }
+
+ /**
+ * Performs commit forward compatibility validation. That is, for each
table enlisted in the transaction, checks to see whether the
+ * initial schema (identified by the begin timestamp) is
forward-compatible with the commit schema (identified by the commit
+ * timestamp).
+ *
+ * @param txId ID of the transaction that gets validated.
+ * @param enlistedGroupIds IDs of the partitions that are enlisted with
the transaction.
+ * @param commitTimestamp Commit timestamp (or {@code null} if it's an
abort).
+ * @return Future completed with validation result.
+ */
+ CompletableFuture<ForwardValidationResult> validateForwards(
+ UUID txId,
+ List<TablePartitionId> enlistedGroupIds,
+ @Nullable HybridTimestamp commitTimestamp
+ ) {
+ HybridTimestamp beginTimestamp = TransactionIds.beginTimestamp(txId);
+
+ Set<UUID> tableIds = enlistedGroupIds.stream()
+ .map(TablePartitionId::tableId)
+ .collect(toSet());
+
+ assert commitTimestamp != null;
+ // Using compareTo() instead of after()/begin() because the latter
methods take clock skew into account
+ // which only makes sense when comparing 'unrelated' timestamps.
beginTs and commitTs have a causal relationship,
+ // so we don't need to account for clock skew.
+ assert commitTimestamp.compareTo(beginTimestamp) > 0;
+
+ return schemas.waitForSchemasAvailability(commitTimestamp)
+ .thenApply(ignored -> validateSchemasCompatibility(tableIds,
commitTimestamp, beginTimestamp));
+ }
+
+ private ForwardValidationResult validateSchemasCompatibility(
+ Set<UUID> tableIds,
+ HybridTimestamp commitTimestamp,
+ HybridTimestamp beginTimestamp
+ ) {
+ for (UUID tableId : tableIds) {
+ ForwardValidationResult validationResult =
validateSchemaCompatibility(beginTimestamp, commitTimestamp, tableId);
+
+ if (!validationResult.isSuccessful()) {
+ return validationResult;
+ }
+ }
+
+ return ForwardValidationResult.success();
+ }
+
+ private ForwardValidationResult validateSchemaCompatibility(
+ HybridTimestamp beginTimestamp,
+ HybridTimestamp commitTimestamp,
+ UUID tableId
+ ) {
+ List<FullTableSchema> tableSchemas =
schemas.tableSchemaVersionsBetween(tableId, beginTimestamp, commitTimestamp);
+
+ assert !tableSchemas.isEmpty();
+
+ for (int i = 0; i < tableSchemas.size() - 1; i++) {
+ FullTableSchema from = tableSchemas.get(i);
+ FullTableSchema to = tableSchemas.get(i + 1);
+ if (!isCompatible(from, to)) {
+ return ForwardValidationResult.failure(tableId,
from.schemaVersion(), to.schemaVersion());
+ }
+ }
+
+ return ForwardValidationResult.success();
+ }
+
+ private boolean isCompatible(FullTableSchema prevSchema, FullTableSchema
nextSchema) {
+ TableDefinitionDiff diff = nextSchema.diffFrom(prevSchema);
+
+ // TODO: IGNITE-19229 - more sophisticated logic
+ return diff.isEmpty();
+ }
+}
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
new file mode 100644
index 0000000000..7ed0cd4929
--- /dev/null
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/ColumnDefinitionDiff.java
@@ -0,0 +1,37 @@
+/*
+ * 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.schema;
+
+import org.apache.ignite.internal.catalog.descriptors.TableColumnDescriptor;
+
+/**
+ * Captures a difference between 'old' and 'new' versions of the same column
definition.
+ */
+public class ColumnDefinitionDiff {
+ @SuppressWarnings("PMD.UnusedPrivateField")
+ private final TableColumnDescriptor oldColumn;
+ @SuppressWarnings("PMD.UnusedPrivateField")
+ private final TableColumnDescriptor newColumn;
+
+ // TODO: IGNITE-19229 - extend
+
+ public ColumnDefinitionDiff(TableColumnDescriptor oldColumn,
TableColumnDescriptor newColumn) {
+ this.oldColumn = oldColumn;
+ this.newColumn = newColumn;
+ }
+}
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
new file mode 100644
index 0000000000..fbe9e88aca
--- /dev/null
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/FullTableSchema.java
@@ -0,0 +1,144 @@
+/*
+ * 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.schema;
+
+import static java.util.function.Function.identity;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toMap;
+import static org.apache.ignite.internal.util.CollectionUtils.intersect;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.function.Function;
+import org.apache.ignite.internal.catalog.descriptors.IndexDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.TableColumnDescriptor;
+
+/**
+ * Represents a full table schema: that is, the definition of the table and
all objects (indexes, constraints, etc)
+ * that belong to the table.
+ */
+public class FullTableSchema {
+ private final int schemaVersion;
+ private final int tableId;
+
+ private final List<TableColumnDescriptor> columns;
+
+ private final List<IndexDescriptor> indexes;
+
+ /**
+ * Constructor.
+ */
+ public FullTableSchema(
+ int schemaVersion,
+ int tableId,
+ List<TableColumnDescriptor> columns,
+ List<IndexDescriptor> indexes
+ ) {
+ this.schemaVersion = schemaVersion;
+ this.tableId = tableId;
+ this.columns = columns;
+ this.indexes = indexes;
+ }
+
+ /**
+ * Returns version of the table definition.
+ *
+ * @return Version of the table definition.
+ */
+ public int schemaVersion() {
+ return schemaVersion;
+ }
+
+ /**
+ * Returns ID of the table.
+ *
+ * @return ID of the table
+ */
+ public int tableId() {
+ return tableId;
+ }
+
+ /**
+ * Returns definitions of the columns of the table.
+ *
+ * @return Definitions of the columns of the table.
+ */
+ public List<TableColumnDescriptor> columns() {
+ return columns;
+ }
+
+ /**
+ * Returns definitions of indexes belonging to the table.
+ *
+ * @return Definitions of indexes belonging to the table.
+ */
+ public List<IndexDescriptor> indexes() {
+ return indexes;
+ }
+
+ /**
+ * Computes a diff between this and a previous schema.
+ *
+ * @param prevSchema Previous table schema.
+ * @return Difference between the schemas.
+ */
+ public TableDefinitionDiff diffFrom(FullTableSchema prevSchema) {
+ Map<String, TableColumnDescriptor> prevColumnsByName =
toMapByName(prevSchema.columns, TableColumnDescriptor::name);
+ Map<String, TableColumnDescriptor> thisColumnsByName =
toMapByName(this.columns, TableColumnDescriptor::name);
+
+ List<TableColumnDescriptor> addedColumns =
subtractKeyed(thisColumnsByName, prevColumnsByName);
+ List<TableColumnDescriptor> removedColumns =
subtractKeyed(prevColumnsByName, thisColumnsByName);
+
+ Set<String> intersectionColumnNames =
intersect(thisColumnsByName.keySet(), prevColumnsByName.keySet());
+ List<ColumnDefinitionDiff> changedColumns = new ArrayList<>();
+ for (String commonColumnName : intersectionColumnNames) {
+ TableColumnDescriptor prevColumn =
prevColumnsByName.get(commonColumnName);
+ TableColumnDescriptor thisColumn =
thisColumnsByName.get(commonColumnName);
+
+ if (columnChanged(prevColumn, thisColumn)) {
+ changedColumns.add(new ColumnDefinitionDiff(prevColumn,
thisColumn));
+ }
+ }
+
+ Map<String, IndexDescriptor> prevIndexesByName =
toMapByName(prevSchema.indexes, IndexDescriptor::name);
+ Map<String, IndexDescriptor> thisIndexesByName =
toMapByName(this.indexes, IndexDescriptor::name);
+
+ List<IndexDescriptor> addedIndexes = subtractKeyed(thisIndexesByName,
prevIndexesByName);
+ List<IndexDescriptor> removedIndexes =
subtractKeyed(prevIndexesByName, thisIndexesByName);
+
+ return new TableDefinitionDiff(addedColumns, removedColumns,
changedColumns, addedIndexes, removedIndexes);
+ }
+
+ private static <T> Map<String, T> toMapByName(List<T> elements,
Function<T, String> nameExtractor) {
+ return elements.stream().collect(toMap(nameExtractor, identity()));
+ }
+
+ private static <T> List<T> subtractKeyed(Map<String, T> minuend,
Map<String, T> subtrahend) {
+ return minuend.entrySet().stream()
+ .filter(entry -> !subtrahend.containsKey(entry.getKey()))
+ .map(Entry::getValue)
+ .collect(toList());
+ }
+
+ private static boolean columnChanged(TableColumnDescriptor prevColumn,
TableColumnDescriptor newColumn) {
+ return !prevColumn.equals(newColumn);
+ }
+}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/NonHistoricSchemas.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/NonHistoricSchemas.java
new file mode 100644
index 0000000000..64c99970fc
--- /dev/null
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/NonHistoricSchemas.java
@@ -0,0 +1,99 @@
+/*
+ * 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.schema;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.catalog.commands.DefaultValue;
+import org.apache.ignite.internal.catalog.descriptors.TableColumnDescriptor;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.schema.Column;
+import org.apache.ignite.internal.schema.SchemaDescriptor;
+import org.apache.ignite.internal.schema.SchemaManager;
+import org.apache.ignite.internal.schema.SchemaRegistry;
+
+/**
+ * A dummy implementation over {@link SchemaManager}. It is dummy because:
+ *
+ * <ul>
+ * <li>It imitates historicity, but always takes the latest known
schema</li>
+ * <li>{@link #tableSchemaVersionsBetween(UUID, HybridTimestamp,
HybridTimestamp)} always returns a single schema to avoid
+ * validation failures</li>
+ * </ul>
+ *
+ * <p>The point of this implementation is to allow the system work in the
pre-SchemaSync fashion before the switch to CatalogService
+ * is possible.
+ */
+// TODO: IGNITE-19447 - remove when switched to the CatalogService
+public class NonHistoricSchemas implements Schemas {
+ private final SchemaManager schemaManager;
+
+ public NonHistoricSchemas(SchemaManager schemaManager) {
+ this.schemaManager = schemaManager;
+ }
+
+ @Override
+ public CompletableFuture<?> waitForSchemasAvailability(HybridTimestamp ts)
{
+ return completedFuture(null);
+ }
+
+ @Override
+ public List<FullTableSchema> tableSchemaVersionsBetween(UUID tableId,
HybridTimestamp fromIncluding, HybridTimestamp toIncluding) {
+ SchemaRegistry schemaRegistry = schemaManager.schemaRegistry(tableId);
+ SchemaDescriptor schemaDescriptor = schemaRegistry.schema();
+
+ List<TableColumnDescriptor> columns =
schemaDescriptor.columnNames().stream()
+ .map(colName -> {
+ Column column = schemaDescriptor.column(colName);
+
+ assert column != null;
+
+ return columnDescriptor(column);
+ })
+ .collect(toList());
+
+ var fullSchema = new FullTableSchema(
+ 1,
+ 1,
+ columns,
+ List.of()
+ );
+
+ return List.of(fullSchema);
+ }
+
+ /**
+ * Converts a {@link Column} to a {@link TableColumnDescriptor}. Please
note that the conversion is not full; it's
+ * used in the code that actually doesn't care about columns.
+ *
+ * @param column Column to convert.
+ * @return Conversion result.
+ */
+ public static TableColumnDescriptor columnDescriptor(Column column) {
+ return new TableColumnDescriptor(
+ column.name(),
+ column.type().spec().asColumnType(),
+ column.nullable(),
+ DefaultValue.constant(column.defaultValue())
+ );
+ }
+}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/Schemas.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/Schemas.java
new file mode 100644
index 0000000000..42c55a0ce5
--- /dev/null
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/Schemas.java
@@ -0,0 +1,48 @@
+/*
+ * 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.schema;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+
+/**
+ * Provides access to table schemas.
+ */
+public interface Schemas {
+ /**
+ * Obtains a future that completes when all schemas activating not later
than the given timestamp are available.
+ *
+ * @param ts Timestamp we are interested in. This is the timestamp
transaction processing logic is interested in (like beginTs or
+ * commitTs), not the timestamp after subtraction described in section
'Waiting for safe time in the past' of
+ * <a
href="https://cwiki.apache.org/confluence/display/IGNITE/IEP-98:+Schema+Synchronization";>IEP-98</a>
+ * @return Future that completes when all schemas activating not later
than the given timestamp are available.
+ */
+ CompletableFuture<?> waitForSchemasAvailability(HybridTimestamp ts);
+
+ /**
+ * Returns all schema versions between (including) the two that were
effective at the given timestamps.
+ *
+ * @param tableId ID of the table which schemas need to be considered.
+ * @param fromIncluding Start timestamp.
+ * @param toIncluding End timestamp.
+ * @return All schema versions between (including) the two that were
effective at the given timestamps.
+ */
+ List<FullTableSchema> tableSchemaVersionsBetween(UUID tableId,
HybridTimestamp fromIncluding, HybridTimestamp toIncluding);
+}
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
new file mode 100644
index 0000000000..9981e86710
--- /dev/null
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/TableDefinitionDiff.java
@@ -0,0 +1,116 @@
+/*
+ * 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.schema;
+
+import static java.util.Collections.emptyList;
+
+import java.util.List;
+import org.apache.ignite.internal.catalog.descriptors.IndexDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.TableColumnDescriptor;
+
+/**
+ * Captures the difference between two versions of the same table definition.
+ */
+public class TableDefinitionDiff {
+ private static final TableDefinitionDiff EMPTY = new TableDefinitionDiff(
+ emptyList(), emptyList(), emptyList(), emptyList(), emptyList()
+ );
+
+ private final List<TableColumnDescriptor> addedColumns;
+ private final List<TableColumnDescriptor> removedColumns;
+ private final List<ColumnDefinitionDiff> changedColumns;
+
+ private final List<IndexDescriptor> addedIndexes;
+ private final List<IndexDescriptor> removedIndexes;
+
+ // TODO: IGNITE-19229 - other change types
+
+ /**
+ * Returns an empty diff (meaning there is no difference).
+ *
+ * @return Empty diff (meaning there is no difference).
+ */
+ public static TableDefinitionDiff empty() {
+ return EMPTY;
+ }
+
+ /**
+ * Constructor.
+ */
+ public TableDefinitionDiff(
+ List<TableColumnDescriptor> addedColumns,
+ List<TableColumnDescriptor> removedColumns,
+ List<ColumnDefinitionDiff> changedColumns,
+ List<IndexDescriptor> addedIndexes,
+ List<IndexDescriptor> removedIndexes
+ ) {
+ this.addedColumns = List.copyOf(addedColumns);
+ this.removedColumns = List.copyOf(removedColumns);
+ this.changedColumns = List.copyOf(changedColumns);
+ this.addedIndexes = List.copyOf(addedIndexes);
+ this.removedIndexes = List.copyOf(removedIndexes);
+ }
+
+ /**
+ * Returns columns that were added.
+ */
+ public List<TableColumnDescriptor> addedColumns() {
+ return addedColumns;
+ }
+
+ /**
+ * Returns columns that were removed.
+ */
+ public List<TableColumnDescriptor> removedColumns() {
+ return removedColumns;
+ }
+
+ /**
+ * Returns columns that were changed.
+ */
+ public List<ColumnDefinitionDiff> changedColumns() {
+ return changedColumns;
+ }
+
+ /**
+ * Returns indexes that were added.
+ */
+ public List<IndexDescriptor> addedIndexes() {
+ return addedIndexes;
+ }
+
+ /**
+ * Returns indexes that were removed.
+ */
+ public List<IndexDescriptor> removedIndexes() {
+ return removedIndexes;
+ }
+
+ /**
+ * 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()
+ && addedIndexes.isEmpty()
+ && removedIndexes.isEmpty();
+ }
+}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/package-info.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/package-info.java
new file mode 100644
index 0000000000..1f92efab55
--- /dev/null
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * This is a temporary package to host code needed until CatalogService is
ready
+ * TODO: IGNITE-19447 - remove/rework when switched to full-blown usage of
CatalogService.
+ */
+package org.apache.ignite.internal.table.distributed.schema;
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtil.java
b/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtil.java
index 1dd45571c2..81050182af 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtil.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtil.java
@@ -25,6 +25,7 @@ import static
org.apache.ignite.internal.metastorage.dsl.Conditions.value;
import static org.apache.ignite.internal.metastorage.dsl.Operations.ops;
import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
import static org.apache.ignite.internal.metastorage.dsl.Statements.iif;
+import static org.apache.ignite.internal.util.CollectionUtils.difference;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
@@ -32,7 +33,6 @@ import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import java.util.stream.Collectors;
import org.apache.ignite.internal.affinity.AffinityUtils;
import org.apache.ignite.internal.affinity.Assignment;
import org.apache.ignite.internal.metastorage.Entry;
@@ -250,7 +250,7 @@ public class RebalanceUtil {
ByteArray pendingKey = pendingPartAssignmentsKey(partId);
- Set<Assignment> pendingAssignments = subtract(assignments,
switchReduce);
+ Set<Assignment> pendingAssignments = difference(assignments,
switchReduce);
byte[] pendingByteArray = ByteUtils.toBytes(pendingAssignments);
byte[] assignmentsByteArray = ByteUtils.toBytes(assignments);
@@ -294,17 +294,6 @@ public class RebalanceUtil {
return metaStorageMgr.invoke(resultingOperation).thenApply(unused ->
null);
}
- /**
- * Removes nodes from set of nodes.
- *
- * @param minuend Set to remove nodes from.
- * @param subtrahend Set of nodes to be removed.
- * @return Result of the subtraction.
- */
- public static <T> Set<T> subtract(Set<T> minuend, Set<T> subtrahend) {
- return minuend.stream().filter(v ->
!subtrahend.contains(v)).collect(Collectors.toSet());
- }
-
/**
* Adds nodes to the set of nodes.
*
@@ -319,15 +308,4 @@ public class RebalanceUtil {
return res;
}
-
- /**
- * Returns an intersection of two set of nodes.
- *
- * @param op1 First operand.
- * @param op2 Second operand.
- * @return Result of the intersection.
- */
- public static <T> Set<T> intersect(Set<T> op1, Set<T> op2) {
- return op1.stream().filter(op2::contains).collect(Collectors.toSet());
- }
}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/RepeatedFinishReadWriteTransactionTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/RepeatedFinishReadWriteTransactionTest.java
index af54528561..8c39b45678 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/RepeatedFinishReadWriteTransactionTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/RepeatedFinishReadWriteTransactionTest.java
@@ -268,7 +268,7 @@ public class RepeatedFinishReadWriteTransactionTest {
@Override
public CompletableFuture<Void> cleanup(ClusterNode recipientNode,
List<IgniteBiTuple<TablePartitionId, Long>> tablePartitionIds,
UUID txId, boolean commit,
- HybridTimestamp commitTimestamp) {
+ @Nullable HybridTimestamp commitTimestamp) {
return null;
}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
index d47de41365..44a8fbc65c 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
@@ -72,6 +72,7 @@ import
org.apache.ignite.internal.table.distributed.replicator.PlacementDriver;
import
org.apache.ignite.internal.table.distributed.replicator.action.RequestType;
import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.table.impl.DummySchemaManagerImpl;
+import org.apache.ignite.internal.table.impl.DummySchemas;
import org.apache.ignite.internal.testframework.IgniteAbstractTest;
import org.apache.ignite.internal.tx.Lock;
import org.apache.ignite.internal.tx.LockManager;
@@ -198,6 +199,7 @@ public class PartitionReplicaListenerIndexLockingTest
extends IgniteAbstractTest
safeTime,
mock(LowWatermark.class)
),
+ new DummySchemas(schemaManager),
peer -> true,
CompletableFuture.completedFuture(schemaManager)
);
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 c9bd4bfe02..5079324369 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
@@ -18,10 +18,14 @@
package org.apache.ignite.internal.table.distributed.replication;
import static java.util.concurrent.CompletableFuture.completedFuture;
+import static
org.apache.ignite.internal.hlc.HybridTimestamp.hybridTimestampToLong;
+import static
org.apache.ignite.internal.testframework.asserts.CompletableFutureAssert.assertWillThrowFast;
import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrowFast;
import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
import static org.apache.ignite.internal.util.ArrayUtils.asList;
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.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -29,8 +33,10 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@@ -48,12 +54,15 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.IntStream;
import org.apache.ignite.distributed.TestPartitionDataStorage;
import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder;
import org.apache.ignite.internal.binarytuple.BinaryTuplePrefixBuilder;
+import org.apache.ignite.internal.catalog.commands.DefaultValue;
+import org.apache.ignite.internal.catalog.descriptors.TableColumnDescriptor;
import
org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import
org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
import org.apache.ignite.internal.hlc.HybridClock;
@@ -94,15 +103,19 @@ import
org.apache.ignite.internal.table.distributed.SortedIndexLocker;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import org.apache.ignite.internal.table.distributed.TableMessagesFactory;
import
org.apache.ignite.internal.table.distributed.TableSchemaAwareIndexStorage;
+import org.apache.ignite.internal.table.distributed.command.FinishTxCommand;
import org.apache.ignite.internal.table.distributed.command.PartitionCommand;
import org.apache.ignite.internal.table.distributed.command.TxCleanupCommand;
import org.apache.ignite.internal.table.distributed.command.UpdateCommand;
import org.apache.ignite.internal.table.distributed.raft.PartitionDataStorage;
import
org.apache.ignite.internal.table.distributed.replication.request.ReadWriteReplicaRequest;
+import
org.apache.ignite.internal.table.distributed.replicator.IncompatibleSchemaAbortException;
import org.apache.ignite.internal.table.distributed.replicator.LeaderOrTxState;
import
org.apache.ignite.internal.table.distributed.replicator.PartitionReplicaListener;
import org.apache.ignite.internal.table.distributed.replicator.PlacementDriver;
import
org.apache.ignite.internal.table.distributed.replicator.action.RequestType;
+import org.apache.ignite.internal.table.distributed.schema.FullTableSchema;
+import org.apache.ignite.internal.table.distributed.schema.Schemas;
import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.table.impl.DummySchemaManagerImpl;
import org.apache.ignite.internal.testframework.IgniteAbstractTest;
@@ -113,19 +126,24 @@ import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.TxMeta;
import org.apache.ignite.internal.tx.TxState;
import org.apache.ignite.internal.tx.impl.HeapLockManager;
+import org.apache.ignite.internal.tx.message.TxFinishReplicaRequest;
import org.apache.ignite.internal.tx.message.TxMessagesFactory;
import org.apache.ignite.internal.tx.storage.state.test.TestTxStateStorage;
import org.apache.ignite.internal.tx.test.TestTransactionIds;
import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.internal.util.Lazy;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
+import org.apache.ignite.lang.ErrorGroups.Transactions;
+import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.TopologyService;
+import org.apache.ignite.sql.ColumnType;
import org.apache.ignite.tx.TransactionException;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junitpioneer.jupiter.cartesian.CartesianTest;
@@ -208,12 +226,18 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
@Mock
private RaftGroupService mockRaftClient;
+ @Mock
+ private TxManager txManager;
+
@Mock
private TopologyService topologySrv;
@Mock
private PendingComparableValuesTracker<HybridTimestamp, Void>
safeTimeClock;
+ @Mock
+ private Schemas schemas;
+
/** Schema descriptor for tests. */
private SchemaDescriptor schemaDescriptor;
@@ -290,6 +314,8 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
lenient().when(safeTimeClock.waitFor(any())).thenReturn(completedFuture(null));
+
lenient().when(schemas.waitForSchemasAvailability(any())).thenReturn(completedFuture(null));
+
UUID pkIndexId = UUID.randomUUID();
UUID sortedIndexId = UUID.randomUUID();
UUID hashIndexId = UUID.randomUUID();
@@ -332,7 +358,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
partitionReplicaListener = new PartitionReplicaListener(
testMvPartitionStorage,
mockRaftClient,
- mock(TxManager.class),
+ txManager,
lockManager,
Runnable::run,
partId,
@@ -352,6 +378,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
safeTimeClock,
mock(LowWatermark.class)
),
+ schemas,
peer -> localNode.name().equals(peer.consistentId()),
completedFuture(schemaManager)
);
@@ -1155,6 +1182,133 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
cleanup(tx1);
}
+ @Test
+ public void abortsSuccessfully() {
+ AtomicReference<Boolean> committed = interceptFinishTxCommand();
+
+ CompletableFuture<?> future = beginAndAbortTx();
+
+ assertThat(future, willSucceedFast());
+
+ assertThat(committed.get(), is(false));
+ }
+
+ private CompletableFuture<?> beginAndAbortTx() {
+ when(txManager.cleanup(any(), any(), any(), anyBoolean(),
any())).thenReturn(completedFuture(null));
+
+ HybridTimestamp beginTimestamp = clock.now();
+ UUID txId =
TestTransactionIds.TRANSACTION_ID_GENERATOR.transactionIdFor(beginTimestamp);
+
+ TxFinishReplicaRequest commitRequest =
TX_MESSAGES_FACTORY.txFinishReplicaRequest()
+ .groupId(grpId)
+ .txId(txId)
+ .groups(Map.of(localNode, List.of(new IgniteBiTuple<>(grpId,
1L))))
+ .commit(false)
+ .term(1L)
+ .build();
+
+ return partitionReplicaListener.invoke(commitRequest);
+ }
+
+ @Test
+ public void commitsOnSameSchemaSuccessfully() {
+ when(schemas.tableSchemaVersionsBetween(any(), any(), any()))
+ .thenReturn(List.of(
+ tableSchema(1, List.of(nullableColumn("col")))
+ ));
+
+ AtomicReference<Boolean> committed = interceptFinishTxCommand();
+
+ CompletableFuture<?> future = beginAndCommitTx();
+
+ assertThat(future, willSucceedFast());
+
+ assertThat(committed.get(), is(true));
+ }
+
+ private static TableColumnDescriptor nullableColumn(String colName) {
+ return new TableColumnDescriptor(colName, ColumnType.INT32, true,
DefaultValue.constant(null));
+ }
+
+ private static TableColumnDescriptor defaultedColumn(String colName, int
defaultValue) {
+ return new TableColumnDescriptor(colName, ColumnType.INT32, false,
DefaultValue.constant(defaultValue));
+ }
+
+ private static FullTableSchema tableSchema(int schemaVersion,
List<TableColumnDescriptor> columns) {
+ return new FullTableSchema(schemaVersion, 1, columns, List.of());
+ }
+
+ private AtomicReference<Boolean> interceptFinishTxCommand() {
+ AtomicReference<Boolean> committed = new AtomicReference<>();
+
+ raftClientFutureClosure = command -> {
+ if (command instanceof FinishTxCommand) {
+ committed.set(((FinishTxCommand) command).commit());
+ }
+ return defaultMockRaftFutureClosure.apply(command);
+ };
+
+ return committed;
+ }
+
+ private CompletableFuture<?> beginAndCommitTx() {
+ when(txManager.cleanup(any(), any(), any(), anyBoolean(),
any())).thenReturn(completedFuture(null));
+
+ HybridTimestamp beginTimestamp = clock.now();
+ UUID txId =
TestTransactionIds.TRANSACTION_ID_GENERATOR.transactionIdFor(beginTimestamp);
+
+ HybridTimestamp commitTimestamp = clock.now();
+
+ TxFinishReplicaRequest commitRequest =
TX_MESSAGES_FACTORY.txFinishReplicaRequest()
+ .groupId(grpId)
+ .txId(txId)
+ .groups(Map.of(localNode, List.of(new IgniteBiTuple<>(grpId,
1L))))
+ .commit(true)
+ .commitTimestampLong(hybridTimestampToLong(commitTimestamp))
+ .term(1L)
+ .build();
+
+ return partitionReplicaListener.invoke(commitRequest);
+ }
+
+ @Test
+ @Disabled("IGNITE-19229")
+ public void commitsOnCompatibleSchemaChangeSuccessfully() {
+ when(schemas.tableSchemaVersionsBetween(any(), any(), any()))
+ .thenReturn(List.of(
+ tableSchema(1, List.of(nullableColumn("col1"))),
+ tableSchema(2, List.of(nullableColumn("col1"),
nullableColumn("col2")))
+ ));
+
+ AtomicReference<Boolean> committed = interceptFinishTxCommand();
+
+ CompletableFuture<?> future = beginAndCommitTx();
+
+ assertThat(future, willSucceedFast());
+
+ assertThat(committed.get(), is(true));
+ }
+
+ @Test
+ public void abortsCommitOnIncompatibleSchema() {
+ when(schemas.tableSchemaVersionsBetween(any(), any(), any()))
+ .thenReturn(List.of(
+ tableSchema(1, List.of(defaultedColumn("col", 4))),
+ tableSchema(2, List.of(defaultedColumn("col", 5)))
+ ));
+
+ AtomicReference<Boolean> committed = interceptFinishTxCommand();
+
+ CompletableFuture<?> future = beginAndCommitTx();
+
+ IncompatibleSchemaAbortException ex = assertWillThrowFast(future,
+ IncompatibleSchemaAbortException.class);
+ assertThat(ex.code(), is(Transactions.TX_COMMIT_ERR));
+ assertThat(ex.getMessage(), containsString("Commit failed because
schema 1 is not forward-compatible with 2"));
+
+ assertThat(committed.get(), is(false));
+ }
+
private static UUID beginTx() {
return TestTransactionIds.newTransactionId();
}
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
new file mode 100644
index 0000000000..fbe2ce2830
--- /dev/null
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/schema/FullTableSchemaTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.schema;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+
+import java.util.List;
+import org.apache.ignite.internal.catalog.commands.DefaultValue;
+import org.apache.ignite.internal.catalog.descriptors.HashIndexDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.IndexDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.TableColumnDescriptor;
+import org.apache.ignite.sql.ColumnType;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.Test;
+
+class FullTableSchemaTest {
+ @Test
+ void sameSchemasHaveEmptyDiff() {
+ TableColumnDescriptor column = someColumn("a");
+ IndexDescriptor index = someIndex(1, "ind_a");
+
+ var schema1 = new FullTableSchema(1, 1, List.of(column),
List.of(index));
+ var schema2 = new FullTableSchema(2, 1, List.of(column),
List.of(index));
+
+ TableDefinitionDiff diff = schema2.diffFrom(schema1);
+
+ assertThat(diff.isEmpty(), is(true));
+ }
+
+ @NotNull
+ private static HashIndexDescriptor someIndex(int id, String name) {
+ return new HashIndexDescriptor(id, name, 1, List.of("a"));
+ }
+
+ @NotNull
+ private static TableColumnDescriptor someColumn(String columnName) {
+ return new TableColumnDescriptor(columnName, ColumnType.INT32, true,
DefaultValue.constant(null));
+ }
+
+ @Test
+ void addedRemovedColumnsAreReflectedInDiff() {
+ TableColumnDescriptor column1 = someColumn("a");
+ TableColumnDescriptor column2 = someColumn("b");
+ TableColumnDescriptor column3 = someColumn("c");
+
+ var schema1 = new FullTableSchema(1, 1, List.of(column1, column2),
List.of());
+ var schema2 = new FullTableSchema(2, 1, List.of(column2, column3),
List.of());
+
+ 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()));
+ }
+
+ @Test
+ void changedColumnsAreReflectedInDiff() {
+ TableColumnDescriptor column1 = someColumn("a");
+
+ var schema1 = new FullTableSchema(1, 1, List.of(column1), List.of());
+ var schema2 = new FullTableSchema(2, 1,
+ List.of(new TableColumnDescriptor("a", ColumnType.STRING,
true, DefaultValue.constant(null))),
+ List.of()
+ );
+
+ TableDefinitionDiff diff = schema2.diffFrom(schema1);
+
+ assertThat(diff.isEmpty(), is(false));
+
+ List<ColumnDefinitionDiff> changedColumns = diff.changedColumns();
+ assertThat(changedColumns, is(hasSize(1)));
+ }
+
+ @Test
+ void addedRemovedIndexesAreReflectedInDiff() {
+ IndexDescriptor index1 = someIndex(1, "a");
+ IndexDescriptor index2 = someIndex(2, "b");
+ IndexDescriptor index3 = someIndex(3, "c");
+
+ var schema1 = new FullTableSchema(1, 1, List.of(someColumn("a")),
List.of(index1, index2));
+ var schema2 = new FullTableSchema(2, 1, List.of(someColumn("a")),
List.of(index2, index3));
+
+ TableDefinitionDiff diff = schema2.diffFrom(schema1);
+
+ assertThat(diff.isEmpty(), is(false));
+ assertThat(diff.addedIndexes(), is(List.of(index3)));
+ assertThat(diff.removedIndexes(), is(List.of(index1)));
+ }
+}
diff --git
a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
index efb3727dca..1859462772 100644
---
a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
+++
b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.table.impl;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.lenient;
@@ -70,6 +71,8 @@ import
org.apache.ignite.internal.table.distributed.raft.PartitionDataStorage;
import org.apache.ignite.internal.table.distributed.raft.PartitionListener;
import
org.apache.ignite.internal.table.distributed.replicator.PartitionReplicaListener;
import org.apache.ignite.internal.table.distributed.replicator.PlacementDriver;
+import org.apache.ignite.internal.table.distributed.schema.FullTableSchema;
+import org.apache.ignite.internal.table.distributed.schema.Schemas;
import org.apache.ignite.internal.table.distributed.storage.InternalTableImpl;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.TxManager;
@@ -199,7 +202,7 @@ public class DummyInternalTableImpl extends
InternalTableImpl {
lenient().doReturn(groupId).when(svc).groupId();
Peer leaderPeer = new Peer(UUID.randomUUID().toString());
lenient().doReturn(leaderPeer).when(svc).leader();
- lenient().doReturn(CompletableFuture.completedFuture(new
LeaderWithTerm(leaderPeer, 1L))).when(svc).refreshAndGetLeaderWithTerm();
+ lenient().doReturn(completedFuture(new LeaderWithTerm(leaderPeer,
1L))).when(svc).refreshAndGetLeaderWithTerm();
if (!crossTableUsage) {
// Delegate replica requests directly to replica listener.
@@ -303,8 +306,20 @@ public class DummyInternalTableImpl extends
InternalTableImpl {
txStateStorage().getOrCreateTxStateStorage(PART_ID),
placementDriver,
storageUpdateHandler,
+ new Schemas() {
+ @Override
+ public CompletableFuture<?>
waitForSchemasAvailability(HybridTimestamp ts) {
+ return completedFuture(null);
+ }
+
+ @Override
+ public List<FullTableSchema>
tableSchemaVersionsBetween(UUID tableId, HybridTimestamp fromIncluding,
+ HybridTimestamp toIncluding) {
+ return List.of(new FullTableSchema(1, 1, List.of(),
List.of()));
+ }
+ },
peer -> true,
- CompletableFuture.completedFuture(schemaManager)
+ completedFuture(schemaManager)
);
partitionListener = new PartitionListener(
@@ -398,7 +413,7 @@ public class DummyInternalTableImpl extends
InternalTableImpl {
/** {@inheritDoc} */
@Override
public CompletableFuture<ClusterNode> evaluateReadOnlyRecipientNode(int
partId) {
- return CompletableFuture.completedFuture(mock(ClusterNode.class));
+ return completedFuture(mock(ClusterNode.class));
}
@Override
diff --git
a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummySchemas.java
b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummySchemas.java
new file mode 100644
index 0000000000..89b112db48
--- /dev/null
+++
b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummySchemas.java
@@ -0,0 +1,73 @@
+/*
+ * 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.impl;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.catalog.descriptors.TableColumnDescriptor;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.schema.Column;
+import org.apache.ignite.internal.schema.SchemaDescriptor;
+import org.apache.ignite.internal.schema.SchemaRegistry;
+import org.apache.ignite.internal.table.distributed.schema.FullTableSchema;
+import org.apache.ignite.internal.table.distributed.schema.NonHistoricSchemas;
+import org.apache.ignite.internal.table.distributed.schema.Schemas;
+
+/**
+ * Dummy {@link Schemas} implementation that is not historic and always uses
same {@link SchemaRegistry}.
+ */
+public class DummySchemas implements Schemas {
+ private final SchemaRegistry schemaRegistry;
+
+ public DummySchemas(SchemaRegistry schemaRegistry) {
+ this.schemaRegistry = schemaRegistry;
+ }
+
+ @Override
+ public CompletableFuture<?> waitForSchemasAvailability(HybridTimestamp ts)
{
+ return completedFuture(null);
+ }
+
+ @Override
+ public List<FullTableSchema> tableSchemaVersionsBetween(UUID tableId,
HybridTimestamp fromIncluding, HybridTimestamp toIncluding) {
+ SchemaDescriptor schemaDescriptor = schemaRegistry.schema();
+
+ List<TableColumnDescriptor> columns =
schemaDescriptor.columnNames().stream()
+ .map(colName -> {
+ Column column = schemaDescriptor.column(colName);
+
+ assert column != null;
+
+ return NonHistoricSchemas.columnDescriptor(column);
+ })
+ .collect(toList());
+
+ var fullSchema = new FullTableSchema(
+ 1,
+ 1,
+ columns,
+ List.of()
+ );
+
+ return List.of(fullSchema);
+ }
+}
diff --git
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java
index 5e91960401..3f00db332a 100644
---
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java
+++
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java
@@ -110,7 +110,7 @@ public interface TxManager extends IgniteComponent {
* @param tablePartitionIds Table partition ids with raft terms.
* @param txId Transaction id.
* @param commit {@code True} if a commit requested.
- * @param commitTimestamp Commit timestamp.
+ * @param commitTimestamp Commit timestamp ({@code null} if it's an abort).
* @return Completable future of Void.
*/
CompletableFuture<Void> cleanup(
@@ -118,7 +118,7 @@ public interface TxManager extends IgniteComponent {
List<IgniteBiTuple<TablePartitionId, Long>> tablePartitionIds,
UUID txId,
boolean commit,
- HybridTimestamp commitTimestamp
+ @Nullable HybridTimestamp commitTimestamp
);
/**
diff --git
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
index b6cee11806..16f5235e10 100644
---
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
+++
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
@@ -45,6 +45,7 @@ import
org.apache.ignite.internal.tx.message.TxMessagesFactory;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.network.ClusterNode;
+import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.TestOnly;
/**
@@ -196,7 +197,7 @@ public class TxManagerImpl implements TxManager {
List<IgniteBiTuple<TablePartitionId, Long>> tablePartitionIds,
UUID txId,
boolean commit,
- HybridTimestamp commitTimestamp
+ @Nullable HybridTimestamp commitTimestamp
) {
var cleanupFutures = new CompletableFuture[tablePartitionIds.size()];
diff --git
a/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/TestTransactionIds.java
b/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/TestTransactionIds.java
index 66e4a6814b..4f2ba869fa 100644
---
a/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/TestTransactionIds.java
+++
b/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/TestTransactionIds.java
@@ -33,7 +33,7 @@ public class TestTransactionIds {
/** Hard-coded node ID. */
private static final int SOLE_NODE_ID = 0xdeadbeef;
- private static final TransactionIdGenerator TRANSACTION_ID_GENERATOR = new
TransactionIdGenerator(SOLE_NODE_ID);
+ public static final TransactionIdGenerator TRANSACTION_ID_GENERATOR = new
TransactionIdGenerator(SOLE_NODE_ID);
/**
* Generates new transaction ID using the global clock and hard-coded node
ID.
