This is an automated email from the ASF dual-hosted git repository.
sdanilov 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 86ee7fdee5 IGNITE-19454 Validate backwards schema compatibility (#2075)
86ee7fdee5 is described below
commit 86ee7fdee5779668708daa3e4e47cdfd61c007a3
Author: Roman Puchkovskiy <[email protected]>
AuthorDate: Fri May 19 11:30:29 2023 +0400
IGNITE-19454 Validate backwards schema compatibility (#2075)
---
.../java/org/apache/ignite/lang/ErrorGroups.java | 3 +
.../asserts/CompletableFutureAssert.java | 10 +-
...tionResult.java => CompatValidationResult.java} | 36 ++-
...tType.java => IncompatibleSchemaException.java} | 50 +----
.../replicator/PartitionReplicaListener.java | 159 +++++++++----
.../replicator/SchemaCompatValidator.java | 79 +++++--
.../distributed/replicator/action/RequestType.java | 38 +++-
.../distributed/schema/NonHistoricSchemas.java | 12 +
.../internal/table/distributed/schema/Schemas.java | 21 ++
.../distributed/storage/InternalTableImpl.java | 4 +-
.../replication/PartitionReplicaListenerTest.java | 247 ++++++++++++++++++---
.../table/impl/DummyInternalTableImpl.java | 15 +-
.../ignite/internal/table/impl/DummySchemas.java | 12 +
13 files changed, 518 insertions(+), 168 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
b/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
index 1f5cbaa0f9..ccefdc3cd2 100755
--- a/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
+++ b/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
@@ -276,6 +276,9 @@ public class ErrorGroups {
/** Error occurred when trying to create a read-only transaction with
a timestamp older than the data available in the tables. */
public static final int TX_READ_ONLY_TOO_OLD_ERR =
TX_ERR_GROUP.registerErrorCode(11);
+
+ /** Failure due to an incompatible schema change. */
+ public static final int TX_INCOMPATIBLE_SCHEMA_ERR =
TX_ERR_GROUP.registerErrorCode(12);
}
/** Replicator error group. */
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
index 10a61937d0..cac1d9fa20 100644
---
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
@@ -45,11 +45,10 @@ public class CompletableFutureAssert {
CompletableFuture<?> future,
Class<X> expectedExceptionClass
) {
- try {
- Object normalResult = future.get(1, TimeUnit.SECONDS);
+ Object normalResult;
- return fail("Expected the future to be completed with an exception
of class " + expectedExceptionClass
- + ", but it completed normally with result " +
normalResult);
+ try {
+ normalResult = future.get(1, TimeUnit.SECONDS);
} 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) {
@@ -65,6 +64,9 @@ public class CompletableFutureAssert {
);
}
}
+
+ return fail("Expected the future to be completed with an exception of
class " + expectedExceptionClass
+ + ", but it completed normally with result " + normalResult);
}
private static Throwable unwrapThrowable(Throwable e) {
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/CompatValidationResult.java
similarity index 71%
rename from
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/ForwardValidationResult.java
rename to
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/CompatValidationResult.java
index 560615dce6..def1742451 100644
---
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/CompatValidationResult.java
@@ -22,12 +22,12 @@ import java.util.UUID;
import org.jetbrains.annotations.Nullable;
/**
- * Result of forward schema compatibility validation.
+ * Result of a schema compatibility validation.
*/
-public class ForwardValidationResult {
- private static final ForwardValidationResult SUCCESS = new
ForwardValidationResult(true, null, -1, -1);
+public class CompatValidationResult {
+ private static final CompatValidationResult SUCCESS = new
CompatValidationResult(true, null, -1, -1);
- private final boolean ok;
+ private final boolean successful;
@Nullable
private final UUID failedTableId;
private final int fromSchemaVersion;
@@ -38,7 +38,7 @@ public class ForwardValidationResult {
*
* @return A successful validation result.
*/
- public static ForwardValidationResult success() {
+ public static CompatValidationResult success() {
return SUCCESS;
}
@@ -50,12 +50,12 @@ public class ForwardValidationResult {
* @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);
+ public static CompatValidationResult failure(UUID failedTableId, int
fromSchemaVersion, int toSchemaVersion) {
+ return new CompatValidationResult(false, failedTableId,
fromSchemaVersion, toSchemaVersion);
}
- private ForwardValidationResult(boolean ok, @Nullable UUID failedTableId,
int fromSchemaVersion, int toSchemaVersion) {
- this.ok = ok;
+ private CompatValidationResult(boolean successful, @Nullable UUID
failedTableId, int fromSchemaVersion, int toSchemaVersion) {
+ this.successful = successful;
this.failedTableId = failedTableId;
this.fromSchemaVersion = fromSchemaVersion;
this.toSchemaVersion = toSchemaVersion;
@@ -67,7 +67,7 @@ public class ForwardValidationResult {
* @return Whether the validation was successful
*/
public boolean isSuccessful() {
- return ok;
+ return successful;
}
/**
@@ -81,31 +81,23 @@ public class ForwardValidationResult {
}
/**
- * 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.
+ * Returns version number of the schema from which an incompatible
transition tried to be made.
*
* @return Version number of the schema from which an incompatible
transition tried to be made.
*/
public int fromSchemaVersion() {
- throwIfSuccessful();
+ assert !successful : "Should not be called on a successful result";
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.
+ * Returns version number of the schema to which an incompatible
transition tried to be made.
*
* @return Version number of the schema to which an incompatible
transition tried to be made.
*/
public int toSchemaVersion() {
- throwIfSuccessful();
+ assert !successful : "Should not be called on a successful result";
return toSchemaVersion;
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/action/RequestType.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/IncompatibleSchemaException.java
similarity index 59%
copy from
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/action/RequestType.java
copy to
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/IncompatibleSchemaException.java
index 6baae3a1c5..f553110951 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/action/RequestType.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/IncompatibleSchemaException.java
@@ -15,47 +15,17 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.table.distributed.replicator.action;
+package org.apache.ignite.internal.table.distributed.replicator;
+
+import org.apache.ignite.lang.ErrorGroups.Transactions;
+import org.apache.ignite.tx.TransactionException;
/**
- * Transaction operation type.
+ * Thrown when, during an attempt to execute a transactional operation, it
turns out that the operation cannot be executed
+ * because an incompatible schema change has happened.
*/
-public enum RequestType {
- RW_GET,
-
- RW_GET_ALL,
-
- RW_DELETE,
-
- RW_DELETE_ALL,
-
- RW_DELETE_EXACT,
-
- RW_DELETE_EXACT_ALL,
-
- RW_INSERT,
-
- RW_INSERT_ALL,
-
- RW_UPSERT,
-
- RW_UPSERT_ALL,
-
- RW_REPLACE,
-
- RW_REPLACE_IF_EXIST,
-
- RW_GET_AND_DELETE,
-
- RW_GET_AND_REPLACE,
-
- RW_GET_AND_UPSERT,
-
- RW_SCAN,
-
- RO_GET,
-
- RO_GET_ALL,
-
- RO_SCAN
+public class IncompatibleSchemaException extends TransactionException {
+ public IncompatibleSchemaException(String message) {
+ super(Transactions.TX_INCOMPATIBLE_SCHEMA_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 a232111407..020a9921d8 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
@@ -438,7 +438,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
BinaryRow candidate =
newestCommitTimestamp == null ||
!readResult.isWriteIntent() ? null : cursor.committed(newestCommitTimestamp);
- resolutionFuts.add(resolveReadResult(readResult, readTimestamp, ()
-> candidate));
+ resolutionFuts.add(resolveRoReadResult(readResult, readTimestamp,
() -> candidate));
}
return allOf(resolutionFuts.toArray(new
CompletableFuture[0])).thenCompose(unused -> {
@@ -655,16 +655,28 @@ public class PartitionReplicaListener implements
ReplicaListener {
@SuppressWarnings("resource") PartitionTimestampCursor cursor =
(PartitionTimestampCursor) cursors.computeIfAbsent(cursorId,
id -> mvDataStorage.scan(HybridTimestamp.MAX_VALUE));
- while (batchRows.size() < batchCount && cursor.hasNext()) {
- BinaryRow resolvedReadResult =
resolveReadResult(cursor.next(), txId);
+ return continueScanBatchRetrieval(cursor, batchCount, txId,
batchRows);
+ });
+ }
- if (resolvedReadResult != null) {
- batchRows.add(resolvedReadResult);
- }
- }
+ private CompletableFuture<List<BinaryRow>> continueScanBatchRetrieval(
+ PartitionTimestampCursor cursor,
+ int batchCount,
+ UUID txId,
+ List<BinaryRow> batchRows
+ ) {
+ if (batchRows.size() < batchCount && cursor.hasNext()) {
+ return resolveAndCheckReadCompatibility(cursor.next(), txId)
+ .thenCompose(resolvedReadResult -> {
+ if (resolvedReadResult != null) {
+ batchRows.add(resolvedReadResult);
+ }
- return completedFuture(batchRows);
- });
+ return continueScanBatchRetrieval(cursor, batchCount,
txId, batchRows);
+ });
+ }
+
+ return completedFuture(batchRows);
}
/**
@@ -839,7 +851,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
ReadResult readResult = mvDataStorage.read(rowId, timestamp);
- return resolveReadResult(readResult, timestamp, () -> {
+ return resolveRoReadResult(readResult, timestamp, () -> {
if (readResult.newestCommitTimestamp() == null) {
return null;
}
@@ -893,14 +905,22 @@ public class PartitionReplicaListener implements
ReplicaListener {
return lockManager.acquire(txId, new LockKey(tableId,
currentRow.rowId()), LockMode.S)
.thenComposeAsync(rowLock -> { // Table row S lock
ReadResult readResult =
mvDataStorage.read(currentRow.rowId(), HybridTimestamp.MAX_VALUE);
- BinaryRow resolvedReadResult =
resolveReadResult(readResult, txId);
-
- if (resolvedReadResult != null) {
- result.add(resolvedReadResult);
- }
-
- // Proceed scan.
- return continueIndexScan(txId, indexLocker,
indexCursor, batchSize, result, isUpperBoundAchieved);
+ return
resolveAndCheckReadCompatibility(readResult, txId)
+ .thenCompose(resolvedReadResult -> {
+ if (resolvedReadResult != null) {
+ result.add(resolvedReadResult);
+ }
+
+ // Proceed scan.
+ return continueIndexScan(
+ txId,
+ indexLocker,
+ indexCursor,
+ batchSize,
+ result,
+ isUpperBoundAchieved
+ );
+ });
}, scanRequestExecutor);
});
}
@@ -920,14 +940,15 @@ public class PartitionReplicaListener implements
ReplicaListener {
return lockManager.acquire(txId, new LockKey(tableId, rowId),
LockMode.S)
.thenComposeAsync(rowLock -> { // Table row S lock
ReadResult readResult = mvDataStorage.read(rowId,
HybridTimestamp.MAX_VALUE);
- BinaryRow resolvedReadResult =
resolveReadResult(readResult, txId);
-
- if (resolvedReadResult != null) {
- result.add(resolvedReadResult);
- }
+ return resolveAndCheckReadCompatibility(readResult, txId)
+ .thenCompose(resolvedReadResult -> {
+ if (resolvedReadResult != null) {
+ result.add(resolvedReadResult);
+ }
- // Proceed lookup.
- return continueIndexLookup(txId, indexCursor, batchSize,
result);
+ // Proceed lookup.
+ return continueIndexLookup(txId, indexCursor,
batchSize, result);
+ });
}, scanRequestExecutor);
}
@@ -945,7 +966,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
ReadResult readResult = mvDataStorage.read(rowId, timestamp);
- return resolveReadResult(readResult, timestamp, () -> {
+ return resolveRoReadResult(readResult, timestamp, () -> {
if (readResult.newestCommitTimestamp() == null) {
return null;
}
@@ -988,10 +1009,10 @@ public class PartitionReplicaListener implements
ReplicaListener {
UUID txId = request.txId();
if (request.commit()) {
- return schemaCompatValidator.validateForwards(txId,
aggregatedGroupIds, request.commitTimestamp())
+ return schemaCompatValidator.validateForward(txId,
aggregatedGroupIds, request.commitTimestamp())
.thenCompose(validationResult -> {
return finishAndCleanup(request,
validationResult.isSuccessful(), aggregatedGroupIds, txId)
- .thenAccept(unused ->
throwIfValidationFailed(validationResult));
+ .thenAccept(unused ->
throwIfSchemaValidationOnCommitFailed(validationResult));
});
} else {
// Aborting.
@@ -999,7 +1020,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
}
}
- private static void throwIfValidationFailed(ForwardValidationResult
validationResult) {
+ private static void
throwIfSchemaValidationOnCommitFailed(CompatValidationResult validationResult) {
if (!validationResult.isSuccessful()) {
throw new IncompatibleSchemaAbortException("Commit failed because
schema "
+ validationResult.fromSchemaVersion() + " is not
forward-compatible with "
@@ -1185,21 +1206,48 @@ public class PartitionReplicaListener implements
ReplicaListener {
return pkLocker.locksForLookup(txId, binaryRow)
.thenCompose(ignored -> {
- try (Cursor<RowId> cursor =
pkIndexStorage.get().get(binaryRow)) {
- for (RowId rowId : cursor) {
- BinaryRow row =
resolveReadResult(mvDataStorage.read(rowId, HybridTimestamp.MAX_VALUE), txId);
- if (row != null) {
- return action.apply(rowId, row);
- }
+ boolean cursorClosureSetUp = false;
+ Cursor<RowId> cursor = null;
+
+ try {
+ cursor = pkIndexStorage.get().get(binaryRow);
+
+ Cursor<RowId> finalCursor = cursor;
+ CompletableFuture<T> resolvingFuture =
continueResolvingByPk(cursor, txId, action)
+ .whenComplete((res, ex) ->
finalCursor.close());
+
+ cursorClosureSetUp = true;
+
+ return resolvingFuture;
+ } finally {
+ if (!cursorClosureSetUp && cursor != null) {
+ cursor.close();
}
+ }
+ });
+ }
+
+ private <T> CompletableFuture<T> continueResolvingByPk(
+ Cursor<RowId> cursor,
+ UUID txId,
+ BiFunction<@Nullable RowId, @Nullable BinaryRow,
CompletableFuture<T>> action
+ ) {
+ if (!cursor.hasNext()) {
+ return action.apply(null, null);
+ }
+
+ RowId rowId = cursor.next();
- return action.apply(null, null);
- } catch (Exception e) {
- throw new
IgniteInternalException(Replicator.REPLICA_COMMON_ERR,
- format("Unable to close cursor [tableId={}]",
tableId), e);
+ return resolveAndCheckReadCompatibility(mvDataStorage.read(rowId,
HybridTimestamp.MAX_VALUE), txId)
+ .thenCompose(row -> {
+ if (row != null) {
+ return action.apply(rowId, row);
+ } else {
+ return continueResolvingByPk(cursor, txId, action);
}
});
+
}
/**
@@ -1358,7 +1406,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
TablePartitionId committedPartitionId = request.commitPartitionId();
assert committedPartitionId != null || request.requestType() ==
RequestType.RW_GET_ALL
- : "Commit partition partition is null [type=" +
request.requestType() + ']';
+ : "Commit partition is null [type=" + request.requestType() +
']';
switch (request.requestType()) {
case RW_GET_ALL: {
@@ -1939,7 +1987,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
BinaryRow expectedRow = request.oldBinaryRow();
TablePartitionId commitPartitionId = request.commitPartitionId();
- assert commitPartitionId != null : "Commit partition partition is null
[type=" + request.requestType() + ']';
+ assert commitPartitionId != null : "Commit partition is null [type=" +
request.requestType() + ']';
UUID txId = request.transactionId();
@@ -2037,6 +2085,26 @@ public class PartitionReplicaListener implements
ReplicaListener {
}
}
+ private CompletableFuture<BinaryRow>
resolveAndCheckReadCompatibility(ReadResult readResult, UUID txId) {
+ BinaryRow row = resolveRwReadResult(readResult, txId);
+
+ if (row == null) {
+ return completedFuture(row);
+ }
+
+ return schemaCompatValidator.validateBackwards(row.schemaVersion(),
tableId, txId)
+ .thenCompose(validationResult -> {
+ if (validationResult.isSuccessful()) {
+ return completedFuture(row);
+ } else {
+ throw new IncompatibleSchemaException("Operation
failed because schema "
+ + validationResult.fromSchemaVersion() + " is
not backward-compatible with "
+ + validationResult.toSchemaVersion() + " for
table " + validationResult.failedTableId());
+ }
+ });
+
+ }
+
/**
* Resolves a read result for RW transaction.
*
@@ -2044,8 +2112,9 @@ public class PartitionReplicaListener implements
ReplicaListener {
* @param txId Transaction id.
* @return Resolved binary row.
*/
- private BinaryRow resolveReadResult(ReadResult readResult, UUID txId) {
- // Here is a safety join (waiting of the future result), because the
resolution for RW transaction cannot lead to a network request.
+ @Nullable
+ private BinaryRow resolveRwReadResult(ReadResult readResult, UUID txId) {
+ // This is a safe join (waiting of the future result), because the
resolution for RW transaction cannot lead to a network request.
return resolveReadResult(readResult, txId, null, null).join();
}
@@ -2057,7 +2126,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
* @param lastCommitted Action to get the latest committed row.
* @return Future to resolved binary row.
*/
- private CompletableFuture<BinaryRow> resolveReadResult(
+ private CompletableFuture<BinaryRow> resolveRoReadResult(
ReadResult readResult,
HybridTimestamp timestamp,
Supplier<BinaryRow> lastCommitted
@@ -2071,7 +2140,7 @@ public class PartitionReplicaListener implements
ReplicaListener {
* <li>If txId is not null (RW request), assert that retrieved tx id
matches proposed one or that retrieved tx id is null
* and return binary row. Currently it's only possible to retrieve
write intents if they belong to the same transaction,
* locks prevent reading write intents created by others.</li>
- * <li>If txId is not null (RO request), perform write intent
resolution if given readResult is a write intent itself
+ * <li>If txId is null (RO request), perform write intent resolution
if given readResult is a write intent itself
* or return binary row otherwise.</li>
* </ol>
*
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
index f16a231bfe..a9c5513c56 100644
---
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
@@ -49,9 +49,9 @@ class SchemaCompatValidator {
* @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.
+ * @return Future of validation result.
*/
- CompletableFuture<ForwardValidationResult> validateForwards(
+ CompletableFuture<CompatValidationResult> validateForward(
UUID txId,
List<TablePartitionId> enlistedGroupIds,
@Nullable HybridTimestamp commitTimestamp
@@ -69,26 +69,26 @@ class SchemaCompatValidator {
assert commitTimestamp.compareTo(beginTimestamp) > 0;
return schemas.waitForSchemasAvailability(commitTimestamp)
- .thenApply(ignored -> validateSchemasCompatibility(tableIds,
commitTimestamp, beginTimestamp));
+ .thenApply(ignored ->
validateForwardSchemasCompatibility(tableIds, commitTimestamp, beginTimestamp));
}
- private ForwardValidationResult validateSchemasCompatibility(
+ private CompatValidationResult validateForwardSchemasCompatibility(
Set<UUID> tableIds,
HybridTimestamp commitTimestamp,
HybridTimestamp beginTimestamp
) {
for (UUID tableId : tableIds) {
- ForwardValidationResult validationResult =
validateSchemaCompatibility(beginTimestamp, commitTimestamp, tableId);
+ CompatValidationResult validationResult =
validateForwardSchemaCompatibility(beginTimestamp, commitTimestamp, tableId);
if (!validationResult.isSuccessful()) {
return validationResult;
}
}
- return ForwardValidationResult.success();
+ return CompatValidationResult.success();
}
- private ForwardValidationResult validateSchemaCompatibility(
+ private CompatValidationResult validateForwardSchemaCompatibility(
HybridTimestamp beginTimestamp,
HybridTimestamp commitTimestamp,
UUID tableId
@@ -98,20 +98,71 @@ class SchemaCompatValidator {
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());
+ FullTableSchema oldSchema = tableSchemas.get(i);
+ FullTableSchema newSchema = tableSchemas.get(i + 1);
+ if (!isForwardCompatible(oldSchema, newSchema)) {
+ return CompatValidationResult.failure(tableId,
oldSchema.schemaVersion(), newSchema.schemaVersion());
}
}
- return ForwardValidationResult.success();
+ return CompatValidationResult.success();
}
- private boolean isCompatible(FullTableSchema prevSchema, FullTableSchema
nextSchema) {
+ private boolean isForwardCompatible(FullTableSchema prevSchema,
FullTableSchema nextSchema) {
TableDefinitionDiff diff = nextSchema.diffFrom(prevSchema);
- // TODO: IGNITE-19229 - more sophisticated logic
+ // TODO: IGNITE-19229 - more sophisticated logic.
return diff.isEmpty();
}
+
+ /**
+ * Performs backward compatibility validation of a tuple that was just
read in the transaction.
+ *
+ * <ul>
+ * <li>If the tuple was written with a schema version earlier or same
as the initial schema version of the transaction,
+ * the read is valid.</li>
+ * <li>If the tuple was written with a schema version later than the
initial schema version of the transaction,
+ * the read is valid only if the initial schema version is backward
compatible with the tuple schema version.</li>
+ * </ul>
+ *
+ * @param tupleSchemaVersion Schema version ID of the tuple.
+ * @param tableId ID of the table to which the tuple belongs.
+ * @param txId ID of the transaction that gets validated.
+ * @return Future of validation result.
+ */
+ CompletableFuture<CompatValidationResult> validateBackwards(int
tupleSchemaVersion, UUID tableId, UUID txId) {
+ HybridTimestamp beginTimestamp = TransactionIds.beginTimestamp(txId);
+
+ return schemas.waitForSchemasAvailability(beginTimestamp)
+ .thenCompose(ignored ->
schemas.waitForSchemaAvailability(tableId, tupleSchemaVersion))
+ .thenApply(ignored ->
validateBackwardSchemaCompatibility(tupleSchemaVersion, tableId,
beginTimestamp));
+ }
+
+ private CompatValidationResult validateBackwardSchemaCompatibility(
+ int tupleSchemaVersion,
+ UUID tableId,
+ HybridTimestamp beginTimestamp
+ ) {
+ List<FullTableSchema> tableSchemas =
schemas.tableSchemaVersionsBetween(tableId, beginTimestamp, tupleSchemaVersion);
+
+ if (tableSchemas.isEmpty()) {
+ // 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.failure(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);
+ }
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/action/RequestType.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/action/RequestType.java
index 6baae3a1c5..ac1a7671d2 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/action/RequestType.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/action/RequestType.java
@@ -57,5 +57,41 @@ public enum RequestType {
RO_GET_ALL,
- RO_SCAN
+ RO_SCAN;
+
+ /**
+ * Returns {@code true} if the operation works with a single row.
+ */
+ public boolean isSingleRow() {
+ switch (this) {
+ case RW_GET:
+ case RW_DELETE:
+ case RW_GET_AND_DELETE:
+ case RW_DELETE_EXACT:
+ case RW_INSERT:
+ case RW_UPSERT:
+ case RW_GET_AND_UPSERT:
+ case RW_GET_AND_REPLACE:
+ case RW_REPLACE_IF_EXIST:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ /**
+ * Returns {@code true} if the operation works with multiple rows.
+ */
+ public boolean isMultipleRows() {
+ switch (this) {
+ case RW_GET_ALL:
+ case RW_DELETE_ALL:
+ case RW_DELETE_EXACT_ALL:
+ case RW_INSERT_ALL:
+ case RW_UPSERT_ALL:
+ return true;
+ default:
+ return false;
+ }
+ }
}
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
index 64c99970fc..66f38fe125 100644
---
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
@@ -56,6 +56,11 @@ public class NonHistoricSchemas implements Schemas {
return completedFuture(null);
}
+ @Override
+ public CompletableFuture<?> waitForSchemaAvailability(UUID tableId, int
schemaVersion) {
+ return completedFuture(null);
+ }
+
@Override
public List<FullTableSchema> tableSchemaVersionsBetween(UUID tableId,
HybridTimestamp fromIncluding, HybridTimestamp toIncluding) {
SchemaRegistry schemaRegistry = schemaManager.schemaRegistry(tableId);
@@ -81,6 +86,13 @@ public class NonHistoricSchemas implements Schemas {
return List.of(fullSchema);
}
+ @Override
+ public List<FullTableSchema> tableSchemaVersionsBetween(UUID tableId,
HybridTimestamp fromIncluding, int toIncluding) {
+ // Returning an empty list makes sure that backward validation never
fails, which is what we want before
+ // we switch to CatalogService completely.
+ return List.of();
+ }
+
/**
* 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.
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
index 42c55a0ce5..2ed9db14ac 100644
---
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
@@ -36,6 +36,15 @@ public interface Schemas {
*/
CompletableFuture<?> waitForSchemasAvailability(HybridTimestamp ts);
+ /**
+ * Obtains a future that completes when the given schema version becomes
available.
+ *
+ * @param tableId ID of the table of interest.
+ * @param schemaVersion ID of the schema version.
+ * @return Future that completes when the given schema version becomes
available.
+ */
+ CompletableFuture<?> waitForSchemaAvailability(UUID tableId, int
schemaVersion);
+
/**
* Returns all schema versions between (including) the two that were
effective at the given timestamps.
*
@@ -45,4 +54,16 @@ public interface Schemas {
* @return All schema versions between (including) the two that were
effective at the given timestamps.
*/
List<FullTableSchema> tableSchemaVersionsBetween(UUID tableId,
HybridTimestamp fromIncluding, HybridTimestamp toIncluding);
+
+ /**
+ * Returns all schema versions between (including) the one that was
effective at the given timestamp and
+ * the one identified by a schema version ID. If the starting schema (the
one effective at fromIncluding)
+ * is actually a later schema than the one identified by toIncluding, then
an empty list is returned.
+ *
+ * @param tableId ID of the table which schemas need to be considered.
+ * @param fromIncluding Start timestamp.
+ * @param toIncluding End schema version ID.
+ * @return All schema versions between (including) the given timestamp and
schema version.
+ */
+ List<FullTableSchema> tableSchemaVersionsBetween(UUID tableId,
HybridTimestamp fromIncluding, int toIncluding);
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
index 57a6774ca9..d037a88baf 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
@@ -217,7 +217,7 @@ public class InternalTableImpl implements InternalTable {
*/
private <R> CompletableFuture<R> enlistInTx(
BinaryRowEx row,
- InternalTransaction tx,
+ @Nullable InternalTransaction tx,
IgniteTetraFunction<TablePartitionId, InternalTransaction,
ReplicationGroupId, Long, ReplicaRequest> op
) {
// Check whether proposed tx is read-only. Complete future
exceptionally if true.
@@ -279,7 +279,7 @@ public class InternalTableImpl implements InternalTable {
*/
private <T> CompletableFuture<T> enlistInTx(
Collection<BinaryRowEx> keyRows,
- InternalTransaction tx,
+ @Nullable InternalTransaction tx,
IgniteFiveFunction<TablePartitionId, Collection<BinaryRow>,
InternalTransaction, ReplicationGroupId, Long, ReplicaRequest> op,
Function<CompletableFuture<Object>[], CompletableFuture<T>> reducer
) {
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 cbf3f6b19b..e8a3b4b896 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
@@ -26,6 +26,7 @@ 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.hamcrest.Matchers.nullValue;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -34,6 +35,7 @@ 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.ArgumentMatchers.anyInt;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -58,6 +60,7 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.IntStream;
+import java.util.stream.Stream;
import org.apache.ignite.distributed.TestPartitionDataStorage;
import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder;
import org.apache.ignite.internal.binarytuple.BinaryTuplePrefixBuilder;
@@ -110,6 +113,7 @@ 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.IncompatibleSchemaException;
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;
@@ -146,6 +150,9 @@ 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.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.junitpioneer.jupiter.cartesian.CartesianTest;
import org.junitpioneer.jupiter.cartesian.CartesianTest.Values;
import org.mockito.Mock;
@@ -158,6 +165,12 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
/** Partition id. */
private static final int partId = 0;
+ private static final int CURRENT_SCHEMA_VERSION = 1;
+
+ private static final int FUTURE_SCHEMA_VERSION = 2;
+
+ private static final int FUTURE_SCHEMA_ROW_INDEXED_VALUE = 0;
+
/** Table id. */
private final UUID tblId = UUID.randomUUID();
@@ -241,9 +254,15 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
/** Schema descriptor for tests. */
private SchemaDescriptor schemaDescriptor;
+ /** Schema descriptor, version 2. */
+ private SchemaDescriptor schemaDescriptorVersion2;
+
/** Key-value marshaller for tests. */
private KvMarshaller<TestKey, TestValue> kvMarshaller;
+ /** Key-value marshaller using schema version 2. */
+ private KvMarshaller<TestKey, TestValue> kvMarshallerVersion2;
+
/** Partition replication listener to test. */
private PartitionReplicaListener partitionReplicaListener;
@@ -315,18 +334,14 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
lenient().when(safeTimeClock.waitFor(any())).thenReturn(completedFuture(null));
lenient().when(schemas.waitForSchemasAvailability(any())).thenReturn(completedFuture(null));
+ lenient().when(schemas.waitForSchemaAvailability(any(),
anyInt())).thenReturn(completedFuture(null));
UUID pkIndexId = UUID.randomUUID();
UUID sortedIndexId = UUID.randomUUID();
UUID hashIndexId = UUID.randomUUID();
- schemaDescriptor = new SchemaDescriptor(1, new Column[]{
- new Column("intKey".toUpperCase(Locale.ROOT),
NativeTypes.INT32, false),
- new Column("strKey".toUpperCase(Locale.ROOT),
NativeTypes.STRING, false),
- }, new Column[]{
- new Column("intVal".toUpperCase(Locale.ROOT),
NativeTypes.INT32, false),
- new Column("strVal".toUpperCase(Locale.ROOT),
NativeTypes.STRING, false),
- });
+ schemaDescriptor = schemaDescriptorWith(CURRENT_SCHEMA_VERSION);
+ schemaDescriptorVersion2 = schemaDescriptorWith(FUTURE_SCHEMA_VERSION);
Function<BinaryRow, BinaryTuple> row2Tuple =
BinaryRowConverter.keyExtractor(schemaDescriptor);
@@ -383,15 +398,30 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
completedFuture(schemaManager)
);
- MarshallerFactory marshallerFactory = new
ReflectionMarshallerFactory();
-
sortedIndexBinarySchema =
BinaryTupleSchema.createSchema(schemaDescriptor, new int[]{2 /* intVal column
*/});
- kvMarshaller = marshallerFactory.create(schemaDescriptor,
TestKey.class, TestValue.class);
+ kvMarshaller = marshallerFor(schemaDescriptor);
+ kvMarshallerVersion2 = marshallerFor(schemaDescriptorVersion2);
reset();
}
+ private static SchemaDescriptor schemaDescriptorWith(int ver) {
+ return new SchemaDescriptor(ver, new Column[]{
+ new Column("intKey".toUpperCase(Locale.ROOT),
NativeTypes.INT32, false),
+ new Column("strKey".toUpperCase(Locale.ROOT),
NativeTypes.STRING, false),
+ }, new Column[]{
+ new Column("intVal".toUpperCase(Locale.ROOT),
NativeTypes.INT32, false),
+ new Column("strVal".toUpperCase(Locale.ROOT),
NativeTypes.STRING, false),
+ });
+ }
+
+ private static KvMarshaller<TestKey, TestValue>
marshallerFor(SchemaDescriptor descriptor) {
+ MarshallerFactory marshallerFactory = new
ReflectionMarshallerFactory();
+
+ return marshallerFactory.create(descriptor, TestKey.class,
TestValue.class);
+ }
+
private TableSchemaAwareIndexStorage pkStorage() {
return Objects.requireNonNull(pkStorageSupplier.get());
}
@@ -935,8 +965,8 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
cleanup(txId);
}
- private void doSingleRowRequest(UUID txId, BinaryRow binaryRow,
RequestType requestType) {
-
partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteSingleRowReplicaRequest()
+ private CompletableFuture<?> doSingleRowRequest(UUID txId, BinaryRow
binaryRow, RequestType requestType) {
+ return
partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteSingleRowReplicaRequest()
.transactionId(txId)
.requestType(requestType)
.binaryRow(binaryRow)
@@ -946,8 +976,8 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
);
}
- private void doMultiRowRequest(UUID txId, Collection<BinaryRow>
binaryRows, RequestType requestType) {
-
partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
+ private CompletableFuture<?> doMultiRowRequest(UUID txId,
Collection<BinaryRow> binaryRows, RequestType requestType) {
+ return
partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
.transactionId(txId)
.requestType(requestType)
.binaryRows(binaryRows)
@@ -1197,7 +1227,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
when(txManager.cleanup(any(), any(), any(), anyBoolean(),
any())).thenReturn(completedFuture(null));
HybridTimestamp beginTimestamp = clock.now();
- UUID txId =
TestTransactionIds.TRANSACTION_ID_GENERATOR.transactionIdFor(beginTimestamp);
+ UUID txId = transactionIdFor(beginTimestamp);
TxFinishReplicaRequest commitRequest =
TX_MESSAGES_FACTORY.txFinishReplicaRequest()
.groupId(grpId)
@@ -1210,11 +1240,15 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
return partitionReplicaListener.invoke(commitRequest);
}
+ private static UUID transactionIdFor(HybridTimestamp beginTimestamp) {
+ return
TestTransactionIds.TRANSACTION_ID_GENERATOR.transactionIdFor(beginTimestamp);
+ }
+
@Test
public void commitsOnSameSchemaSuccessfully() {
when(schemas.tableSchemaVersionsBetween(any(), any(), any()))
.thenReturn(List.of(
- tableSchema(1, List.of(nullableColumn("col")))
+ tableSchema(CURRENT_SCHEMA_VERSION,
List.of(nullableColumn("col")))
));
AtomicReference<Boolean> committed = interceptFinishTxCommand();
@@ -1255,7 +1289,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
when(txManager.cleanup(any(), any(), any(), anyBoolean(),
any())).thenReturn(completedFuture(null));
HybridTimestamp beginTimestamp = clock.now();
- UUID txId =
TestTransactionIds.TRANSACTION_ID_GENERATOR.transactionIdFor(beginTimestamp);
+ UUID txId = transactionIdFor(beginTimestamp);
HybridTimestamp commitTimestamp = clock.now();
@@ -1276,8 +1310,8 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
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")))
+ tableSchema(CURRENT_SCHEMA_VERSION,
List.of(nullableColumn("col1"))),
+ tableSchema(FUTURE_SCHEMA_VERSION,
List.of(nullableColumn("col1"), nullableColumn("col2")))
));
AtomicReference<Boolean> committed = interceptFinishTxCommand();
@@ -1291,11 +1325,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
@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)))
- ));
+ simulateForwardIncompatibleSchemaChange(CURRENT_SCHEMA_VERSION,
FUTURE_SCHEMA_VERSION);
AtomicReference<Boolean> committed = interceptFinishTxCommand();
@@ -1309,6 +1339,158 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
assertThat(committed.get(), is(false));
}
+ private void simulateForwardIncompatibleSchemaChange(int
fromSchemaVersion, int toSchemaVersion) {
+ when(schemas.tableSchemaVersionsBetween(any(), any(), any()))
+ .thenReturn(incompatibleSchemaVersions(fromSchemaVersion,
toSchemaVersion));
+ }
+
+ private void simulateBackwardIncompatibleSchemaChange(int
fromSchemaVersion, int toSchemaVersion) {
+ when(schemas.tableSchemaVersionsBetween(any(), any(), anyInt()))
+ .thenReturn(incompatibleSchemaVersions(fromSchemaVersion,
toSchemaVersion));
+ }
+
+ private static List<FullTableSchema> incompatibleSchemaVersions(int
fromSchemaVersion, int toSchemaVersion) {
+ return List.of(
+ tableSchema(fromSchemaVersion, List.of(defaultedColumn("col",
4))),
+ tableSchema(toSchemaVersion, List.of(defaultedColumn("col",
5)))
+ );
+ }
+
+ @ParameterizedTest
+ @MethodSource("singleRowRequestTypes")
+ public void
failsWhenReadingSingleRowFromFutureIncompatibleSchema(RequestType requestType) {
+ testFailsWhenReadingFromFutureIncompatibleSchema((targetTxId, key) ->
doSingleRowRequest(
+ targetTxId,
+ binaryRow(key, new TestValue(1, "v1"), kvMarshaller),
+ requestType
+ ));
+ }
+
+ private void
testFailsWhenReadingFromFutureIncompatibleSchema(ListenerInvocation
listenerInvocation) {
+ UUID targetTxId = transactionIdFor(clock.now());
+
+ TestKey key = simulateWriteWithSchemaVersionFromFuture();
+
+ simulateBackwardIncompatibleSchemaChange(CURRENT_SCHEMA_VERSION,
FUTURE_SCHEMA_VERSION);
+
+ AtomicReference<Boolean> committed = interceptFinishTxCommand();
+
+ CompletableFuture<?> future = listenerInvocation.invoke(targetTxId,
key);
+
+ assertFailureDueToBackwardIncompatibleSchemaChange(future, committed);
+ }
+
+ private static Stream<Arguments> singleRowRequestTypes() {
+ return Arrays.stream(RequestType.values())
+ .filter(RequestType::isSingleRow)
+ .map(Arguments::of);
+ }
+
+ private TestKey simulateWriteWithSchemaVersionFromFuture() {
+ UUID futureSchemaVersionTxId = transactionIdFor(clock.now());
+
+ TestKey key = nextKey();
+ BinaryRow futureSchemaVersionRow = binaryRow(key, new TestValue(2,
"v2"), kvMarshallerVersion2);
+ var rowId = new RowId(partId);
+
+ BinaryTuple indexedValue = new BinaryTuple(sortedIndexBinarySchema,
+ new BinaryTupleBuilder(1,
false).appendInt(FUTURE_SCHEMA_ROW_INDEXED_VALUE).build()
+ );
+
+ pkStorage().put(futureSchemaVersionRow, rowId);
+ testMvPartitionStorage.addWrite(rowId, futureSchemaVersionRow,
futureSchemaVersionTxId, tblId, partId);
+ sortedIndexStorage.storage().put(new IndexRowImpl(indexedValue,
rowId));
+ testMvPartitionStorage.commitWrite(rowId, clock.now());
+
+ return key;
+ }
+
+ private static void assertFailureDueToBackwardIncompatibleSchemaChange(
+ CompletableFuture<?> future,
+ AtomicReference<Boolean> committed
+ ) {
+ IncompatibleSchemaException ex = assertWillThrowFast(future,
+ IncompatibleSchemaException.class);
+ assertThat(ex.code(), is(Transactions.TX_INCOMPATIBLE_SCHEMA_ERR));
+ assertThat(ex.getMessage(), containsString("Operation failed because
schema 1 is not backward-compatible with 2"));
+
+ // Tx should not be finished.
+ assertThat(committed.get(), is(nullValue()));
+ }
+
+ @ParameterizedTest
+ @MethodSource("multiRowsRequestTypes")
+ public void
failsWhenReadingMultiRowsFromFutureIncompatibleSchema(RequestType requestType) {
+ testFailsWhenReadingFromFutureIncompatibleSchema((targetTxId, key) ->
doMultiRowRequest(
+ targetTxId,
+ List.of(binaryRow(key, new TestValue(1, "v1"), kvMarshaller)),
+ requestType
+ ));
+ }
+
+ private static Stream<Arguments> multiRowsRequestTypes() {
+ return Arrays.stream(RequestType.values())
+ .filter(RequestType::isMultipleRows)
+ .map(Arguments::of);
+ }
+
+ @Test
+ public void failsWhenReplacingOnTupleWithIncompatibleSchemaFromFuture() {
+ testFailsWhenReadingFromFutureIncompatibleSchema(
+ (targetTxId, key) ->
partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteSwapRowReplicaRequest()
+ .transactionId(targetTxId)
+ .requestType(RequestType.RW_REPLACE)
+ .oldBinaryRow(binaryRow(key, new TestValue(1, "v1"),
kvMarshaller))
+ .binaryRow(binaryRow(key, new TestValue(3, "v3"),
kvMarshaller))
+ .term(1L)
+ .commitPartitionId(new
TablePartitionId(UUID.randomUUID(), partId))
+ .build()
+ )
+ );
+ }
+
+ @Test
+ public void
failsWhenScanByExactMatchReadsTupleWithIncompatibleSchemaFromFuture() {
+ testFailsWhenReadingFromFutureIncompatibleSchema(
+ (targetTxId, key) ->
partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
+ .transactionId(targetTxId)
+ .indexToUse(sortedIndexStorage.id())
+ .exactKey(toIndexKey(FUTURE_SCHEMA_ROW_INDEXED_VALUE))
+ .term(1L)
+ .scanId(1)
+ .batchSize(100)
+ .build()
+ )
+ );
+ }
+
+ @Test
+ public void
failsWhenScanByIndexReadsTupleWithIncompatibleSchemaFromFuture() {
+ testFailsWhenReadingFromFutureIncompatibleSchema(
+ (targetTxId, key) ->
partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
+ .transactionId(targetTxId)
+ .indexToUse(sortedIndexStorage.id())
+ .term(1L)
+ .scanId(1)
+ .batchSize(100)
+ .build()
+ )
+ );
+ }
+
+ @Test
+ public void failsWhenFullScanReadsTupleWithIncompatibleSchemaFromFuture() {
+ testFailsWhenReadingFromFutureIncompatibleSchema(
+ (targetTxId, key) ->
partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
+ .transactionId(targetTxId)
+ .term(1L)
+ .scanId(1)
+ .batchSize(100)
+ .build()
+ )
+ );
+ }
+
private static UUID beginTx() {
return TestTransactionIds.newTransactionId();
}
@@ -1383,12 +1565,16 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
private BinaryRow nextBinaryKey() {
try {
- return kvMarshaller.marshal(new TestKey(monotonicInt(), "key " +
monotonicInt()));
+ return kvMarshaller.marshal(nextKey());
} catch (MarshallerException e) {
throw new IgniteException(e);
}
}
+ private static TestKey nextKey() {
+ return new TestKey(monotonicInt(), "key " + monotonicInt());
+ }
+
private static int monotonicInt() {
return nextMonotonicInt.getAndIncrement();
}
@@ -1402,8 +1588,12 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
}
private BinaryRow binaryRow(TestKey key, TestValue value) {
+ return binaryRow(key, value, kvMarshaller);
+ }
+
+ private static BinaryRow binaryRow(TestKey key, TestValue value,
KvMarshaller<TestKey, TestValue> marshaller) {
try {
- return kvMarshaller.marshal(key, value);
+ return marshaller.marshal(key, value);
} catch (MarshallerException e) {
throw new AssertionError(e);
}
@@ -1514,4 +1704,9 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
return S.toString(TestValue.class, this);
}
}
+
+ @FunctionalInterface
+ private interface ListenerInvocation {
+ CompletableFuture<?> invoke(UUID targetTxId, TestKey key);
+ }
}
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 1859462772..a147752931 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
@@ -71,8 +71,6 @@ 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;
@@ -306,18 +304,7 @@ 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()));
- }
- },
+ new DummySchemas(schemaManager),
peer -> true,
completedFuture(schemaManager)
);
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
index 89b112db48..b70bb900c7 100644
---
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
@@ -47,6 +47,11 @@ public class DummySchemas implements Schemas {
return completedFuture(null);
}
+ @Override
+ public CompletableFuture<?> waitForSchemaAvailability(UUID tableId, int
schemaVersion) {
+ return completedFuture(null);
+ }
+
@Override
public List<FullTableSchema> tableSchemaVersionsBetween(UUID tableId,
HybridTimestamp fromIncluding, HybridTimestamp toIncluding) {
SchemaDescriptor schemaDescriptor = schemaRegistry.schema();
@@ -70,4 +75,11 @@ public class DummySchemas implements Schemas {
return List.of(fullSchema);
}
+
+ @Override
+ public List<FullTableSchema> tableSchemaVersionsBetween(UUID tableId,
HybridTimestamp fromIncluding, int toIncluding) {
+ // Returning an empty list makes sure that backward validation never
fails, which is what we want before
+ // we switch to CatalogService completely.
+ return List.of();
+ }
}
