You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2024/02/21 10:03:57 UTC
(ignite-3) branch main updated: IGNITE-20117 Implement index backfill process (#3233)
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 bc0d6e9198 IGNITE-20117 Implement index backfill process (#3233)
bc0d6e9198 is described below
commit bc0d6e919830b366ceb2cfa28d437d6d2bb95197
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Wed Feb 21 13:03:51 2024 +0300
IGNITE-20117 Implement index backfill process (#3233)
---
.../ignite/internal/storage/BinaryRowAndRowId.java | 4 +-
.../org/apache/ignite/internal/storage/RowId.java | 25 +--
.../storage/BinaryRowAndRowIdMatcherTest.java | 82 +++++++
.../internal/storage/BinaryRowAndRowIdMatcher.java | 55 +++++
.../distributed/command/BuildIndexCommand.java | 3 +
.../raft/BuildIndexRowVersionChooser.java | 110 ++++++++++
.../table/distributed/raft/PartitionListener.java | 48 +++--
.../replicator/PartitionReplicaListener.java | 1 +
.../raft/BuildIndexRowVersionChooserTest.java | 235 +++++++++++++++++++++
.../raft/PartitionCommandListenerTest.java | 19 +-
10 files changed, 541 insertions(+), 41 deletions(-)
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/BinaryRowAndRowId.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/BinaryRowAndRowId.java
index 1aebb86ca0..fc719e3023 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/BinaryRowAndRowId.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/BinaryRowAndRowId.java
@@ -20,9 +20,7 @@ package org.apache.ignite.internal.storage;
import org.apache.ignite.internal.schema.BinaryRow;
import org.jetbrains.annotations.Nullable;
-/**
- * Wrapper that holds both {@link BinaryRow} and {@link RowId}. {@link BinaryRow} is null for tombstones.
- */
+/** Wrapper that holds both {@link BinaryRow} and {@link RowId}. {@link BinaryRow} is null for tombstones. */
public class BinaryRowAndRowId {
/** Binary row. {@code null} if tombstone. */
private final @Nullable BinaryRow binaryRow;
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/RowId.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/RowId.java
index 8e41c02c3b..7eab59427a 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/RowId.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/RowId.java
@@ -75,30 +75,22 @@ public final class RowId implements Serializable, Comparable<RowId> {
this.uuid = uuid;
}
- /**
- * Returns a partition ID for current row ID.
- */
+ /** Returns a partition ID for current row ID. */
public int partitionId() {
return partitionId & 0xFFFF;
}
- /**
- * Returns the most significant 64 bits of row ID's UUID.
- */
+ /** Returns the most significant 64 bits of row ID's UUID.*/
public long mostSignificantBits() {
return uuid.getMostSignificantBits();
}
- /**
- * Returns the least significant 64 bits of row ID's UUID.
- */
+ /** Returns the least significant 64 bits of row ID's UUID. */
public long leastSignificantBits() {
return uuid.getLeastSignificantBits();
}
- /**
- * Returns the UUID equivalent of {@link #mostSignificantBits()} and {@link #leastSignificantBits()}.
- */
+ /** Returns the UUID equivalent of {@link #mostSignificantBits()} and {@link #leastSignificantBits()}. */
public UUID uuid() {
return uuid;
}
@@ -115,11 +107,8 @@ public final class RowId implements Serializable, Comparable<RowId> {
RowId rowId = (RowId) o;
- if (partitionId != rowId.partitionId) {
- return false;
- }
+ return partitionId == rowId.partitionId && uuid.equals(rowId.uuid);
- return uuid.equals(rowId.uuid);
}
@Override
@@ -140,9 +129,7 @@ public final class RowId implements Serializable, Comparable<RowId> {
return uuid.compareTo(rowId.uuid);
}
- /**
- * Returns the next row ID withing a single partition, or {@code null} if current row ID already has maximal possible value.
- */
+ /** Returns the next row ID withing a single partition, or {@code null} if current row ID already has maximal possible value. */
public @Nullable RowId increment() {
long lsb = uuid.getLeastSignificantBits() + 1;
diff --git a/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/BinaryRowAndRowIdMatcherTest.java b/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/BinaryRowAndRowIdMatcherTest.java
new file mode 100644
index 0000000000..031198b72b
--- /dev/null
+++ b/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/BinaryRowAndRowIdMatcherTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.storage;
+
+import static org.apache.ignite.internal.storage.BinaryRowAndRowIdMatcher.equalToBinaryRowAndRowId;
+import static org.apache.ignite.internal.storage.RowId.highestRowId;
+import static org.apache.ignite.internal.storage.RowId.lowestRowId;
+import static org.apache.ignite.internal.type.NativeTypes.INT32;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.not;
+
+import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.Column;
+import org.apache.ignite.internal.schema.SchemaDescriptor;
+import org.apache.ignite.internal.schema.row.RowAssembler;
+import org.junit.jupiter.api.Test;
+
+/** For {@link BinaryRowAndRowIdMatcher} testing. */
+public class BinaryRowAndRowIdMatcherTest {
+ private static final int PARTITION_ID = 0;
+
+ private static final SchemaDescriptor SCHEMA = new SchemaDescriptor(
+ 1,
+ new Column[]{new Column("key", INT32, false)},
+ new Column[]{new Column("value", INT32, false)}
+ );
+
+ @Test
+ void testNotMatchByRowId() {
+ var binaryRowAndRowId0 = new BinaryRowAndRowId(null, lowestRowId(PARTITION_ID));
+ var binaryRowAndRowId1 = new BinaryRowAndRowId(null, highestRowId(PARTITION_ID));
+
+ assertThat(binaryRowAndRowId0, not(equalToBinaryRowAndRowId(binaryRowAndRowId1)));
+ assertThat(binaryRowAndRowId1, not(equalToBinaryRowAndRowId(binaryRowAndRowId0)));
+ }
+
+ @Test
+ void testNotMatchByBinaryRow() {
+ var binaryRowAndRowId0 = new BinaryRowAndRowId(binaryRow(0, 1), lowestRowId(PARTITION_ID));
+ var binaryRowAndRowId1 = new BinaryRowAndRowId(binaryRow(5, 6), lowestRowId(PARTITION_ID));
+ var binaryRowAndRowId2 = new BinaryRowAndRowId(null, lowestRowId(PARTITION_ID));
+
+ assertThat(binaryRowAndRowId0, not(equalToBinaryRowAndRowId(binaryRowAndRowId1)));
+ assertThat(binaryRowAndRowId1, not(equalToBinaryRowAndRowId(binaryRowAndRowId0)));
+
+ assertThat(binaryRowAndRowId1, not(equalToBinaryRowAndRowId(binaryRowAndRowId2)));
+ assertThat(binaryRowAndRowId2, not(equalToBinaryRowAndRowId(binaryRowAndRowId1)));
+ }
+
+ @Test
+ void testMatch() {
+ var binaryRowAndRowId0 = new BinaryRowAndRowId(binaryRow(0, 1), lowestRowId(PARTITION_ID));
+ var binaryRowAndRowId1 = new BinaryRowAndRowId(binaryRow(0, 1), lowestRowId(PARTITION_ID));
+
+ assertThat(binaryRowAndRowId0, equalToBinaryRowAndRowId(binaryRowAndRowId1));
+ assertThat(binaryRowAndRowId1, equalToBinaryRowAndRowId(binaryRowAndRowId0));
+ }
+
+ private static BinaryRow binaryRow(int key, int val) {
+ RowAssembler rowBuilder = new RowAssembler(SCHEMA, -1);
+
+ rowBuilder.appendInt(key);
+ rowBuilder.appendInt(val);
+
+ return rowBuilder.build();
+ }
+}
diff --git a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BinaryRowAndRowIdMatcher.java b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BinaryRowAndRowIdMatcher.java
new file mode 100644
index 0000000000..8108132cd3
--- /dev/null
+++ b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BinaryRowAndRowIdMatcher.java
@@ -0,0 +1,55 @@
+/*
+ * 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.storage;
+
+import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.hamcrest.Matchers;
+import org.hamcrest.TypeSafeMatcher;
+
+/** Matcher for comparing {@link BinaryRowAndRowId}s. */
+public class BinaryRowAndRowIdMatcher extends TypeSafeMatcher<BinaryRowAndRowId> {
+ private final RowId rowId;
+
+ private final Matcher<BinaryRow> binaryRowMatcher;
+
+ private BinaryRowAndRowIdMatcher(BinaryRowAndRowId expected) {
+ this.rowId = expected.rowId();
+
+ BinaryRow row = expected.binaryRow();
+ this.binaryRowMatcher = row == null ? Matchers.nullValue(BinaryRow.class) : BinaryRowMatcher.equalToRow(row);
+ }
+
+ public static BinaryRowAndRowIdMatcher equalToBinaryRowAndRowId(BinaryRowAndRowId exp) {
+ return new BinaryRowAndRowIdMatcher(exp);
+ }
+
+ @Override
+ protected boolean matchesSafely(BinaryRowAndRowId item) {
+ return rowId.equals(item.rowId()) && binaryRowMatcher.matches(item.binaryRow());
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("was rowId").appendValue(rowId);
+ description.appendText(" binaryRow");
+ binaryRowMatcher.describeTo(description);
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/BuildIndexCommand.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/BuildIndexCommand.java
index 4f3896194e..4bf044361f 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/BuildIndexCommand.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/BuildIndexCommand.java
@@ -34,4 +34,7 @@ public interface BuildIndexCommand extends WriteCommand {
/** Returns {@code true} if this batch is the last one. */
boolean finish();
+
+ /** Returns the catalog version in which the index was created. */
+ int creationCatalogVersion();
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/BuildIndexRowVersionChooser.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/BuildIndexRowVersionChooser.java
new file mode 100644
index 0000000000..6e06567503
--- /dev/null
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/BuildIndexRowVersionChooser.java
@@ -0,0 +1,110 @@
+/*
+ * 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.raft;
+
+import static org.apache.ignite.internal.tx.TransactionIds.beginTimestamp;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
+import org.apache.ignite.internal.storage.BinaryRowAndRowId;
+import org.apache.ignite.internal.storage.ReadResult;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.util.Cursor;
+
+/** Obtains versions of a row to be indexed while building an index. */
+class BuildIndexRowVersionChooser {
+ private final PartitionDataStorage storage;
+
+ /** Timestamp of activation of the catalog version in which the index was created. */
+ private final long createIndexActivationTs;
+
+ /** Timestamp of activation of the catalog version in which the index start building (get {@link CatalogIndexStatus#BUILDING}). */
+ private final long startBuildingIndexActivationTs;
+
+ /**
+ * Constructor.
+ *
+ * @param storage Multi-version partition storage.
+ * @param createIndexActivationTs Timestamp of activation of the catalog version in which the index created.
+ * @param startBuildingIndexActivationTs Timestamp of activation of the catalog version in which the index start building
+ * (get {@link CatalogIndexStatus#BUILDING}).
+ */
+ BuildIndexRowVersionChooser(PartitionDataStorage storage, long createIndexActivationTs, long startBuildingIndexActivationTs) {
+ this.storage = storage;
+ this.createIndexActivationTs = createIndexActivationTs;
+ this.startBuildingIndexActivationTs = startBuildingIndexActivationTs;
+ }
+
+ /**
+ * Collects binary versions of a row to build an index.
+ *
+ * <p>Index selection algorithm:</p>
+ * <ul>
+ * <li>For writeCommitted with commitTs > activationTs(BUILDING), we will ignore.</li>
+ * <li>For writeCommitted with commitTs <= activationTs(BUILDING), we will take the latest of them.</li>
+ * <li>For writeIntent with beginTs >= activationTs(REGISTERED), we will ignore.</li>
+ * <li>For writeIntent with beginTs < activationTs(REGISTERED), we take.</li>
+ * </ul>
+ *
+ * @param rowId Row ID of interest.
+ */
+ List<BinaryRowAndRowId> chooseForBuildIndex(RowId rowId) {
+ try (Cursor<ReadResult> rowVersionCursor = storage.scanVersions(rowId)) {
+ List<BinaryRowAndRowId> result = new ArrayList<>();
+
+ boolean takenLatestVersionOfWriteCommitted = false;
+
+ // Versions are iterated in the newest-to-oldest order.
+ for (ReadResult readResult : rowVersionCursor) {
+ if (readResult.isEmpty()) {
+ continue;
+ }
+
+ if (readResult.isWriteIntent()) {
+ if (beginTs(readResult) >= createIndexActivationTs) {
+ continue;
+ }
+ } else {
+ if (commitTs(readResult) > startBuildingIndexActivationTs) {
+ continue;
+ } else {
+ takenLatestVersionOfWriteCommitted = true;
+ }
+ }
+
+ // TODO: IGNITE-21546 It may be necessary to return ReadResult for write intent resolution
+ result.add(new BinaryRowAndRowId(readResult.binaryRow(), rowId));
+
+ if (takenLatestVersionOfWriteCommitted) {
+ break;
+ }
+ }
+
+ return result;
+ }
+ }
+
+ private static long beginTs(ReadResult readResult) {
+ return beginTimestamp(readResult.transactionId()).longValue();
+ }
+
+ private static long commitTs(ReadResult readResult) {
+ return readResult.commitTimestamp().longValue();
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
index e418646b3a..2a3c099af1 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
@@ -18,8 +18,6 @@
package org.apache.ignite.internal.table.distributed.raft;
import static java.util.Objects.requireNonNull;
-import static java.util.function.Predicate.not;
-import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.hlc.HybridTimestamp.CLOCK_SKEW;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.table.distributed.TableUtils.indexIdsAtRwTxBeginTs;
@@ -39,8 +37,11 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.function.Consumer;
+import java.util.stream.IntStream;
import java.util.stream.Stream;
+import org.apache.ignite.internal.catalog.Catalog;
import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.lang.SafeTimeReorderException;
@@ -58,7 +59,6 @@ import org.apache.ignite.internal.replicator.command.SafeTimeSyncCommand;
import org.apache.ignite.internal.storage.BinaryRowAndRowId;
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.MvPartitionStorage.Locker;
-import org.apache.ignite.internal.storage.ReadResult;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand;
@@ -72,7 +72,6 @@ import org.apache.ignite.internal.tx.TxMeta;
import org.apache.ignite.internal.tx.TxState;
import org.apache.ignite.internal.tx.TxStateMeta;
import org.apache.ignite.internal.tx.storage.state.TxStateStorage;
-import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.internal.util.TrackerClosedException;
import org.jetbrains.annotations.Nullable;
@@ -521,13 +520,15 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
// It's important to not block any thread while being inside of the "runConsistently" section.
storageUpdateHandler.getIndexUpdateHandler().waitForIndex(cmd.indexId());
+ BuildIndexRowVersionChooser rowVersionChooser = createBuildIndexRowVersionChooser(cmd);
+
storage.runConsistently(locker -> {
List<UUID> rowUuids = new ArrayList<>(cmd.rowIds());
// Natural UUID order matches RowId order within the same partition.
Collections.sort(rowUuids);
- Stream<BinaryRowAndRowId> buildIndexRowStream = createBuildIndexRowStream(rowUuids, locker);
+ Stream<BinaryRowAndRowId> buildIndexRowStream = createBuildIndexRowStream(rowUuids, locker, rowVersionChooser);
RowId nextRowIdToBuild = cmd.finish() ? null : toRowId(requireNonNull(last(rowUuids))).increment();
@@ -577,19 +578,15 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
}
}
- private Stream<BinaryRowAndRowId> createBuildIndexRowStream(List<UUID> rowUuids, Locker locker) {
+ private Stream<BinaryRowAndRowId> createBuildIndexRowStream(
+ List<UUID> rowUuids,
+ Locker locker,
+ BuildIndexRowVersionChooser rowVersionChooser
+ ) {
return rowUuids.stream()
.map(this::toRowId)
.peek(locker::lock)
- .map(rowId -> {
- try (Cursor<ReadResult> cursor = storage.scanVersions(rowId)) {
- return cursor.stream()
- .filter(not(ReadResult::isEmpty))
- .map(ReadResult::binaryRow)
- .map(binaryRow -> new BinaryRowAndRowId(binaryRow, rowId))
- .collect(toList());
- }
- })
+ .map(rowVersionChooser::chooseForBuildIndex)
.flatMap(Collection::stream);
}
@@ -614,4 +611,25 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
commit ? commitTimestamp : null
));
}
+
+ // TODO: IGNITE-21560 Add schema sync to BuildIndexCommand#startBuildingCatalogVersion
+ // TODO: IGNITE-21560 Skip command if index was removed
+ private BuildIndexRowVersionChooser createBuildIndexRowVersionChooser(BuildIndexCommand command) {
+ int creationCatalogVersion = command.creationCatalogVersion();
+ Catalog creationIndexCatalog = catalogService.catalog(creationCatalogVersion);
+
+ assert creationIndexCatalog != null : "indexId=" + command.indexId() + ", catalogVersion=" + creationCatalogVersion;
+
+ int latestCatalogVersion = catalogService.latestCatalogVersion();
+ Catalog startBuildingIndexCatalog = IntStream.rangeClosed(creationCatalogVersion, latestCatalogVersion)
+ .mapToObj(catalogService::catalog)
+ .filter(catalog -> catalog.index(command.indexId()).status() == CatalogIndexStatus.BUILDING)
+ .findFirst()
+ .orElse(null);
+
+ assert startBuildingIndexCatalog != null : format("indexId={}, catalogVersionFrom={}, catalogVersionTo={}",
+ command.indexId(), creationCatalogVersion, latestCatalogVersion);
+
+ return new BuildIndexRowVersionChooser(storage, creationIndexCatalog.time(), startBuildingIndexCatalog.time());
+ }
}
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 1135c711ac..879c551822 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
@@ -3721,6 +3721,7 @@ public class PartitionReplicaListener implements ReplicaListener {
.indexId(request.indexId())
.rowIds(request.rowIds())
.finish(request.finish())
+ .creationCatalogVersion(request.creationCatalogVersion())
.build();
}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/BuildIndexRowVersionChooserTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/BuildIndexRowVersionChooserTest.java
new file mode 100644
index 0000000000..166d3351f8
--- /dev/null
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/BuildIndexRowVersionChooserTest.java
@@ -0,0 +1,235 @@
+/*
+ * 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.raft;
+
+import static org.apache.ignite.internal.hlc.HybridTimestamp.hybridTimestamp;
+import static org.apache.ignite.internal.storage.BinaryRowAndRowIdMatcher.equalToBinaryRowAndRowId;
+import static org.apache.ignite.internal.storage.RowId.lowestRowId;
+import static org.apache.ignite.internal.tx.TransactionIds.transactionId;
+import static org.apache.ignite.internal.type.NativeTypes.INT32;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.empty;
+import static org.mockito.Mockito.spy;
+
+import java.util.UUID;
+import org.apache.ignite.distributed.TestPartitionDataStorage;
+import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.Column;
+import org.apache.ignite.internal.schema.SchemaDescriptor;
+import org.apache.ignite.internal.schema.row.RowAssembler;
+import org.apache.ignite.internal.storage.BinaryRowAndRowId;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.storage.impl.TestMvPartitionStorage;
+import org.apache.ignite.internal.testframework.IgniteAbstractTest;
+import org.hamcrest.Matcher;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.Test;
+
+/** For {@link BuildIndexRowVersionChooser} testing. */
+public class BuildIndexRowVersionChooserTest extends IgniteAbstractTest {
+ private static final int TABLE_ID = 1;
+
+ private static final int PARTITION_ID = 0;
+
+ private static final long CREATE_INDEX_ACTIVATION_TS_MILLS = 100;
+
+ private static final long START_BUILDING_INDEX_ACTIVATION_TS_MILLS = 200;
+
+ private static final SchemaDescriptor SCHEMA = new SchemaDescriptor(
+ 1,
+ new Column[]{new Column("key", INT32, false)},
+ new Column[]{new Column("value", INT32, false)}
+ );
+
+ private final MvPartitionStorage mvPartitionStorage = spy(new TestMvPartitionStorage(PARTITION_ID));
+
+ private final PartitionDataStorage partitionDataStorage = spy(new TestPartitionDataStorage(TABLE_ID, PARTITION_ID, mvPartitionStorage));
+
+ private final BuildIndexRowVersionChooser chooser = new BuildIndexRowVersionChooser(
+ partitionDataStorage,
+ CREATE_INDEX_ACTIVATION_TS_MILLS,
+ START_BUILDING_INDEX_ACTIVATION_TS_MILLS
+ );
+
+ @Test
+ void testEmptyStorage() {
+ assertThat(chooser.chooseForBuildIndex(lowestRowId(PARTITION_ID)), empty());
+ }
+
+ @Test
+ void testWriteIntentOnCreateIndexActivationTs() {
+ RowId rowId = lowestRowId(PARTITION_ID);
+ BinaryRow row = binaryRow(1, 2);
+ UUID txId = txId(CREATE_INDEX_ACTIVATION_TS_MILLS);
+
+ addWrite(rowId, row, txId);
+
+ assertThat(chooser.chooseForBuildIndex(rowId), empty());
+ }
+
+ @Test
+ void testWriteIntentAfterCreateIndexActivationTs() {
+ RowId rowId = lowestRowId(PARTITION_ID);
+ BinaryRow row = binaryRow(1, 2);
+ UUID txId = txId(CREATE_INDEX_ACTIVATION_TS_MILLS + 1);
+
+ addWrite(rowId, row, txId);
+
+ assertThat(chooser.chooseForBuildIndex(rowId), empty());
+ }
+
+ @Test
+ void testWriteIntentBeforeCreateIndexActivationTs() {
+ RowId rowId = lowestRowId(PARTITION_ID);
+ BinaryRow row = binaryRow(1, 2);
+ UUID txId = txId(CREATE_INDEX_ACTIVATION_TS_MILLS - 1);
+
+ addWrite(rowId, row, txId);
+
+ assertThat(chooser.chooseForBuildIndex(rowId), contains(expBinaryRowAndRowId(rowId, row)));
+ }
+
+ @Test
+ void testWriteCommittedAfterStartBuildingActivationTs() {
+ RowId rowId = lowestRowId(PARTITION_ID);
+ BinaryRow row = binaryRow(1, 2);
+ long commitTs = START_BUILDING_INDEX_ACTIVATION_TS_MILLS + 1;
+
+ addWriteCommitted(rowId, row, commitTs);
+
+ assertThat(chooser.chooseForBuildIndex(rowId), empty());
+ }
+
+ @Test
+ void testWriteCommittedOnStartBuildingActivationTs() {
+ RowId rowId = lowestRowId(PARTITION_ID);
+ BinaryRow row = binaryRow(1, 2);
+ long commitTs = START_BUILDING_INDEX_ACTIVATION_TS_MILLS;
+
+ addWriteCommitted(rowId, row, commitTs);
+
+ assertThat(chooser.chooseForBuildIndex(rowId), contains(expBinaryRowAndRowId(rowId, row)));
+ }
+
+ @Test
+ void testWriteCommittedBeforeStartBuildingActivationTs() {
+ RowId rowId = lowestRowId(PARTITION_ID);
+ BinaryRow row = binaryRow(1, 2);
+ long commitTs = START_BUILDING_INDEX_ACTIVATION_TS_MILLS - 1;
+
+ addWriteCommitted(rowId, row, commitTs);
+
+ assertThat(chooser.chooseForBuildIndex(rowId), contains(expBinaryRowAndRowId(rowId, row)));
+ }
+
+ @Test
+ void testWriteCommittedBeforeStartBuildingActivationTsLatest() {
+ RowId rowId = lowestRowId(PARTITION_ID);
+
+ BinaryRow row0 = binaryRow(1, 2);
+ BinaryRow row1 = binaryRow(3, 4);
+ BinaryRow row2 = binaryRow(5, 6);
+
+ long commitTs0 = START_BUILDING_INDEX_ACTIVATION_TS_MILLS - 2;
+ long commitTs1 = START_BUILDING_INDEX_ACTIVATION_TS_MILLS - 1;
+ long commitTs2 = START_BUILDING_INDEX_ACTIVATION_TS_MILLS;
+
+ addWriteCommitted(rowId, row0, commitTs0);
+ addWriteCommitted(rowId, row1, commitTs1);
+ addWriteCommitted(rowId, row2, commitTs2);
+
+ assertThat(chooser.chooseForBuildIndex(rowId), contains(expBinaryRowAndRowId(rowId, row2)));
+ }
+
+ @Test
+ void testWriteCommittedMixed() {
+ RowId rowId = lowestRowId(PARTITION_ID);
+
+ BinaryRow row0 = binaryRow(1, 2);
+ BinaryRow row1 = binaryRow(3, 4);
+ BinaryRow row2 = binaryRow(5, 6);
+
+ long commitTs0 = START_BUILDING_INDEX_ACTIVATION_TS_MILLS - 1;
+ long commitTs1 = START_BUILDING_INDEX_ACTIVATION_TS_MILLS;
+ long commitTs2 = START_BUILDING_INDEX_ACTIVATION_TS_MILLS + 1;
+
+ addWriteCommitted(rowId, row0, commitTs0);
+ addWriteCommitted(rowId, row1, commitTs1);
+ addWriteCommitted(rowId, row2, commitTs2);
+
+ assertThat(chooser.chooseForBuildIndex(rowId), contains(expBinaryRowAndRowId(rowId, row1)));
+ }
+
+ @Test
+ void testWriteMixed() {
+ RowId rowId = lowestRowId(PARTITION_ID);
+
+ BinaryRow row0 = binaryRow(1, 2);
+ BinaryRow row1 = binaryRow(3, 4);
+
+ long commitTs = START_BUILDING_INDEX_ACTIVATION_TS_MILLS;
+ long beginTs = CREATE_INDEX_ACTIVATION_TS_MILLS - 1;
+
+ addWriteCommitted(rowId, row0, commitTs);
+ addWrite(rowId, row1, txId(beginTs));
+
+ assertThat(
+ chooser.chooseForBuildIndex(rowId),
+ contains(expBinaryRowAndRowId(rowId, row1), expBinaryRowAndRowId(rowId, row0))
+ );
+ }
+
+ private void addWrite(RowId rowId, BinaryRow row, UUID txId) {
+ partitionDataStorage.runConsistently(locker -> {
+ locker.lock(rowId);
+
+ partitionDataStorage.addWrite(rowId, row, txId, TABLE_ID, PARTITION_ID);
+
+ return null;
+ });
+ }
+
+ private void addWriteCommitted(RowId rowId, BinaryRow row, long commitTs) {
+ partitionDataStorage.runConsistently(locker -> {
+ locker.lock(rowId);
+
+ partitionDataStorage.addWriteCommitted(rowId, row, hybridTimestamp(commitTs));
+
+ return null;
+ });
+ }
+
+ private static UUID txId(long beginTs) {
+ return transactionId(hybridTimestamp(beginTs), 1);
+ }
+
+ private static BinaryRow binaryRow(int key, int val) {
+ RowAssembler rowBuilder = new RowAssembler(SCHEMA, -1);
+
+ rowBuilder.appendInt(key);
+ rowBuilder.appendInt(val);
+
+ return rowBuilder.build();
+ }
+
+ private static Matcher<BinaryRowAndRowId> expBinaryRowAndRowId(RowId rowId, @Nullable BinaryRow row) {
+ return equalToBinaryRowAndRowId(new BinaryRowAndRowId(row, rowId));
+ }
+}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java
index 705ed4eb9f..5e26174b6c 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.table.distributed.raft;
+import static org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus.BUILDING;
import static org.apache.ignite.internal.util.ArrayUtils.asList;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -56,6 +57,7 @@ import java.util.stream.Stream;
import org.apache.ignite.distributed.TestPartitionDataStorage;
import org.apache.ignite.internal.TestHybridClock;
import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder;
+import org.apache.ignite.internal.catalog.Catalog;
import org.apache.ignite.internal.catalog.CatalogService;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
@@ -183,6 +185,8 @@ public class PartitionCommandListenerTest extends BaseIgniteAbstractTest {
private CatalogService catalogService;
+ private CatalogIndexDescriptor indexDescriptor;
+
/**
* Initializes a table listener before tests.
*/
@@ -196,8 +200,10 @@ public class PartitionCommandListenerTest extends BaseIgniteAbstractTest {
safeTimeTracker = new PendingComparableValuesTracker<>(new HybridTimestamp(1, 0));
+ int indexId = pkStorage.id();
+
indexUpdateHandler = spy(new IndexUpdateHandler(
- DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(pkStorage.id(), pkStorage))
+ DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(indexId, pkStorage))
));
storageUpdateHandler = spy(new StorageUpdateHandler(
@@ -209,10 +215,14 @@ public class PartitionCommandListenerTest extends BaseIgniteAbstractTest {
catalogService = mock(CatalogService.class);
- CatalogIndexDescriptor indexDescriptor = mock(CatalogIndexDescriptor.class);
+ Catalog catalog = mock(Catalog.class);
+
+ indexDescriptor = mock(CatalogIndexDescriptor.class);
- lenient().when(indexDescriptor.id()).thenReturn(pkStorage.id());
+ lenient().when(indexDescriptor.id()).thenReturn(indexId);
+ lenient().when(catalog.index(indexId)).thenReturn(indexDescriptor);
lenient().when(catalogService.indexes(anyInt(), anyInt())).thenReturn(List.of(indexDescriptor));
+ lenient().when(catalogService.catalog(anyInt())).thenReturn(catalog);
commandListener = new PartitionListener(
mock(TxManager.class),
@@ -493,9 +503,10 @@ public class PartitionCommandListenerTest extends BaseIgniteAbstractTest {
@Test
void testBuildIndexCommand() {
- int indexId = 1;
+ int indexId = pkStorage.id();
doNothing().when(indexUpdateHandler).buildIndex(eq(indexId), any(Stream.class), any());
+ when(indexDescriptor.status()).thenReturn(BUILDING);
RowId row0 = new RowId(PARTITION_ID);
RowId row1 = new RowId(PARTITION_ID);