sashapolo commented on code in PR #1800:
URL: https://github.com/apache/ignite-3/pull/1800#discussion_r1144549710
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -91,33 +112,58 @@ public class IndexManager extends Producer<IndexEvent,
IndexEventParameters> imp
/** Prevents double stopping of the component. */
private final AtomicBoolean stopGuard = new AtomicBoolean();
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** Index building executor. */
+ private final ExecutorService buildIndexExecutor;
+
/**
* Constructor.
*
- * @param tablesCfg Tables and indexes configuration.
+ * @param nodeName Node name.
+ * @param tablesConfig Tables and indexes configuration.
* @param schemaManager Schema manager.
* @param tableManager Table manager.
+ * @param clusterService Cluster service.
*/
- public IndexManager(TablesConfiguration tablesCfg, SchemaManager
schemaManager, TableManager tableManager) {
- this.tablesCfg = Objects.requireNonNull(tablesCfg, "tablesCfg");
+ public IndexManager(
+ String nodeName,
+ TablesConfiguration tablesConfig,
+ SchemaManager schemaManager,
+ TableManager tableManager,
+ ClusterService clusterService
+ ) {
+ this.tablesConfig = Objects.requireNonNull(tablesConfig, "tablesCfg");
this.schemaManager = Objects.requireNonNull(schemaManager,
"schemaManager");
this.tableManager = tableManager;
+ this.clusterService = clusterService;
+
+ int cpus = Runtime.getRuntime().availableProcessors();
+
+ buildIndexExecutor = new ThreadPoolExecutor(
+ cpus,
+ cpus,
+ 30,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(),
+ new NamedThreadFactory(threadPrefix(nodeName, "build-index"),
LOG)
Review Comment:
You can use `NamedThreadFactory.create` instead
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -534,32 +584,159 @@ public BinaryTuple convert(BinaryRow binaryRow) {
}
private class ConfigurationListener implements
ConfigurationNamedListListener<TableIndexView> {
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onCreate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onCreate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexCreate(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onRename(
+ public CompletableFuture<?> onRename(
String oldName,
String newName,
ConfigurationNotificationEvent<TableIndexView> ctx
) {
return failedFuture(new
UnsupportedOperationException("https://issues.apache.org/jira/browse/IGNITE-16196";));
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onDelete(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onDelete(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexDrop(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onUpdate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onUpdate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return failedFuture(new IllegalStateException("Should not be
called"));
}
}
+
+ /**
+ * Initializes the build of the index.
+ */
+ private void initIndexBuildIfNeeded(TableIndexView tableIndexView,
TableImpl table) {
+ for (int partitionId = 0; partitionId <
table.internalTable().partitions(); partitionId++) {
+ buildIndexExecutor.submit(new BuildIndexTask(table,
tableIndexView, partitionId, true));
Review Comment:
do we need to submit index building tasks for all partitions regardless of
current assignments?
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -534,32 +584,159 @@ public BinaryTuple convert(BinaryRow binaryRow) {
}
private class ConfigurationListener implements
ConfigurationNamedListListener<TableIndexView> {
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onCreate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onCreate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexCreate(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onRename(
+ public CompletableFuture<?> onRename(
String oldName,
String newName,
ConfigurationNotificationEvent<TableIndexView> ctx
) {
return failedFuture(new
UnsupportedOperationException("https://issues.apache.org/jira/browse/IGNITE-16196";));
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onDelete(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onDelete(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexDrop(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onUpdate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onUpdate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return failedFuture(new IllegalStateException("Should not be
called"));
}
}
+
+ /**
+ * Initializes the build of the index.
+ */
+ private void initIndexBuildIfNeeded(TableIndexView tableIndexView,
TableImpl table) {
+ for (int partitionId = 0; partitionId <
table.internalTable().partitions(); partitionId++) {
+ buildIndexExecutor.submit(new BuildIndexTask(table,
tableIndexView, partitionId, true));
+ }
+ }
+
+ /**
+ * Task of building a table index for a partition.
+ *
+ * <p>Only the leader of the raft group will manage the building of the
index. Leader sends batches of row IDs via
+ * {@link BuildIndexCommand}, the next batch will only be send after the
previous batch has been processed.
+ *
+ * <p>Index building itself occurs locally on each node of the raft group
(majority) when processing {@link BuildIndexCommand}. This
+ * ensures that the index build in the raft group is consistent and that
the index build is restored after restarting the raft group
+ * (not from the beginning).
+ */
+ private class BuildIndexTask implements Runnable {
+ private final TableImpl table;
+
+ private final TableIndexView tableIndexView;
+
+ private final int partitionId;
+
+ private final boolean firstBatch;
+
+ private BuildIndexTask(TableImpl table, TableIndexView tableIndexView,
int partitionId, boolean firstBatch) {
+ this.table = table;
+ this.tableIndexView = tableIndexView;
+ this.partitionId = partitionId;
+ this.firstBatch = firstBatch;
+ }
+
+ @Override
+ public void run() {
+ if (!busyLock.enterBusy()) {
+ return;
+ }
+
+ try {
+ InternalTable internalTable = table.internalTable();
+
+ RaftGroupService raftGroupService =
internalTable.partitionRaftGroupService(partitionId);
+
+ if (!isLocalNodeLeader(raftGroupService)) {
+ // TODO: IGNITE-19053 Must handle the change of leader
+ return;
+ }
+
+ RowId lastBuildRowId =
internalTable.storage().getOrCreateIndex(partitionId,
tableIndexView.id()).getLastBuildRowId();
+
+ if (lastBuildRowId == null) {
+ // Index has already been built.
+ return;
+ }
+
+ if (firstBatch) {
+ LOG.info("Start building the index: [{}]",
createCommonTableIndexInfo());
+ }
+
+ List<RowId> batchRowIds = createBatchRowIds(lastBuildRowId,
BUILD_INDEX_ROW_ID_BATCH_SIZE);
+
+ boolean finish = batchRowIds.size() <
BUILD_INDEX_ROW_ID_BATCH_SIZE;
+
+ raftGroupService.run(createBuildIndexCommand(batchRowIds,
finish))
+ .thenAccept(unused -> {
Review Comment:
```suggestion
.thenRun(() -> {
```
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -91,33 +112,58 @@ public class IndexManager extends Producer<IndexEvent,
IndexEventParameters> imp
/** Prevents double stopping of the component. */
private final AtomicBoolean stopGuard = new AtomicBoolean();
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** Index building executor. */
+ private final ExecutorService buildIndexExecutor;
+
/**
* Constructor.
*
- * @param tablesCfg Tables and indexes configuration.
+ * @param nodeName Node name.
+ * @param tablesConfig Tables and indexes configuration.
* @param schemaManager Schema manager.
* @param tableManager Table manager.
+ * @param clusterService Cluster service.
*/
- public IndexManager(TablesConfiguration tablesCfg, SchemaManager
schemaManager, TableManager tableManager) {
- this.tablesCfg = Objects.requireNonNull(tablesCfg, "tablesCfg");
+ public IndexManager(
+ String nodeName,
+ TablesConfiguration tablesConfig,
+ SchemaManager schemaManager,
+ TableManager tableManager,
+ ClusterService clusterService
+ ) {
+ this.tablesConfig = Objects.requireNonNull(tablesConfig, "tablesCfg");
this.schemaManager = Objects.requireNonNull(schemaManager,
"schemaManager");
this.tableManager = tableManager;
+ this.clusterService = clusterService;
+
+ int cpus = Runtime.getRuntime().availableProcessors();
+
+ buildIndexExecutor = new ThreadPoolExecutor(
Review Comment:
Why don't we use `Executors.newFixedThreadPool` here?
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -534,32 +584,159 @@ public BinaryTuple convert(BinaryRow binaryRow) {
}
private class ConfigurationListener implements
ConfigurationNamedListListener<TableIndexView> {
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onCreate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onCreate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexCreate(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onRename(
+ public CompletableFuture<?> onRename(
String oldName,
String newName,
ConfigurationNotificationEvent<TableIndexView> ctx
) {
return failedFuture(new
UnsupportedOperationException("https://issues.apache.org/jira/browse/IGNITE-16196";));
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onDelete(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onDelete(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexDrop(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onUpdate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onUpdate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return failedFuture(new IllegalStateException("Should not be
called"));
}
}
+
+ /**
+ * Initializes the build of the index.
+ */
+ private void initIndexBuildIfNeeded(TableIndexView tableIndexView,
TableImpl table) {
+ for (int partitionId = 0; partitionId <
table.internalTable().partitions(); partitionId++) {
+ buildIndexExecutor.submit(new BuildIndexTask(table,
tableIndexView, partitionId, true));
+ }
+ }
+
+ /**
+ * Task of building a table index for a partition.
+ *
+ * <p>Only the leader of the raft group will manage the building of the
index. Leader sends batches of row IDs via
+ * {@link BuildIndexCommand}, the next batch will only be send after the
previous batch has been processed.
+ *
+ * <p>Index building itself occurs locally on each node of the raft group
(majority) when processing {@link BuildIndexCommand}. This
Review Comment:
why do you use the "majority" word here?
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -534,32 +584,159 @@ public BinaryTuple convert(BinaryRow binaryRow) {
}
private class ConfigurationListener implements
ConfigurationNamedListListener<TableIndexView> {
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onCreate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onCreate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexCreate(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onRename(
+ public CompletableFuture<?> onRename(
String oldName,
String newName,
ConfigurationNotificationEvent<TableIndexView> ctx
) {
return failedFuture(new
UnsupportedOperationException("https://issues.apache.org/jira/browse/IGNITE-16196";));
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onDelete(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onDelete(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexDrop(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onUpdate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onUpdate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return failedFuture(new IllegalStateException("Should not be
called"));
}
}
+
+ /**
+ * Initializes the build of the index.
+ */
+ private void initIndexBuildIfNeeded(TableIndexView tableIndexView,
TableImpl table) {
+ for (int partitionId = 0; partitionId <
table.internalTable().partitions(); partitionId++) {
+ buildIndexExecutor.submit(new BuildIndexTask(table,
tableIndexView, partitionId, true));
+ }
+ }
+
+ /**
+ * Task of building a table index for a partition.
+ *
+ * <p>Only the leader of the raft group will manage the building of the
index. Leader sends batches of row IDs via
+ * {@link BuildIndexCommand}, the next batch will only be send after the
previous batch has been processed.
+ *
+ * <p>Index building itself occurs locally on each node of the raft group
(majority) when processing {@link BuildIndexCommand}. This
+ * ensures that the index build in the raft group is consistent and that
the index build is restored after restarting the raft group
+ * (not from the beginning).
+ */
+ private class BuildIndexTask implements Runnable {
+ private final TableImpl table;
+
+ private final TableIndexView tableIndexView;
+
+ private final int partitionId;
+
+ private final boolean firstBatch;
+
+ private BuildIndexTask(TableImpl table, TableIndexView tableIndexView,
int partitionId, boolean firstBatch) {
+ this.table = table;
+ this.tableIndexView = tableIndexView;
+ this.partitionId = partitionId;
+ this.firstBatch = firstBatch;
+ }
+
+ @Override
+ public void run() {
+ if (!busyLock.enterBusy()) {
+ return;
+ }
+
+ try {
+ InternalTable internalTable = table.internalTable();
+
+ RaftGroupService raftGroupService =
internalTable.partitionRaftGroupService(partitionId);
+
+ if (!isLocalNodeLeader(raftGroupService)) {
+ // TODO: IGNITE-19053 Must handle the change of leader
+ return;
+ }
+
+ RowId lastBuildRowId =
internalTable.storage().getOrCreateIndex(partitionId,
tableIndexView.id()).getLastBuildRowId();
+
+ if (lastBuildRowId == null) {
+ // Index has already been built.
+ return;
+ }
+
+ if (firstBatch) {
+ LOG.info("Start building the index: [{}]",
createCommonTableIndexInfo());
+ }
+
+ List<RowId> batchRowIds = createBatchRowIds(lastBuildRowId,
BUILD_INDEX_ROW_ID_BATCH_SIZE);
+
+ boolean finish = batchRowIds.size() <
BUILD_INDEX_ROW_ID_BATCH_SIZE;
+
+ raftGroupService.run(createBuildIndexCommand(batchRowIds,
finish))
+ .thenAccept(unused -> {
+ if (!finish) {
+ buildIndexExecutor.submit(new
BuildIndexTask(table, tableIndexView, partitionId, false));
+ }
+ });
+ } catch (Throwable t) {
+ LOG.error("Index build error: [{}]", t,
createCommonTableIndexInfo());
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ private boolean isLocalNodeLeader(RaftGroupService raftGroupService) {
+ Peer leader = raftGroupService.leader();
+
+ assert leader != null : "tableId=" + table.tableId() + ",
partitionId=" + partitionId;
+
+ return localNodeConsistentId().equals(leader.consistentId());
+ }
+
+ private List<RowId> createBatchRowIds(RowId lastBuildRowId, int
batchSize) {
+ MvPartitionStorage mvPartition =
table.internalTable().storage().getMvPartition(partitionId);
+
+ assert mvPartition != null : createCommonTableIndexInfo();
+
+ List<RowId> batch = new ArrayList<>(batchSize);
+
+ for (int i = 0; i < batchSize; i++) {
+ lastBuildRowId =
mvPartition.closestRowId(lastBuildRowId.increment());
Review Comment:
Idea complains that `lastBuildRowId.increment()` can return `null`
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -534,32 +584,159 @@ public BinaryTuple convert(BinaryRow binaryRow) {
}
private class ConfigurationListener implements
ConfigurationNamedListListener<TableIndexView> {
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onCreate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onCreate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexCreate(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onRename(
+ public CompletableFuture<?> onRename(
String oldName,
String newName,
ConfigurationNotificationEvent<TableIndexView> ctx
) {
return failedFuture(new
UnsupportedOperationException("https://issues.apache.org/jira/browse/IGNITE-16196";));
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onDelete(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onDelete(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexDrop(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onUpdate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onUpdate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return failedFuture(new IllegalStateException("Should not be
called"));
}
}
+
+ /**
+ * Initializes the build of the index.
+ */
+ private void initIndexBuildIfNeeded(TableIndexView tableIndexView,
TableImpl table) {
Review Comment:
I think we should rename this method to `startIndexBuild`
##########
modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java:
##########
@@ -240,7 +240,7 @@ void before() throws Exception {
tblManager = mockManagers();
- idxManager = new IndexManager(tblsCfg, schemaManager, tblManager);
+ idxManager = new IndexManager("test", tblsCfg, schemaManager,
tblManager, cs);
Review Comment:
```suggestion
idxManager = new IndexManager(NODE_NAME, tblsCfg, schemaManager,
tblManager, cs);
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/BuildIndexCommand.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.command;
+
+import java.util.List;
+import java.util.UUID;
+import org.apache.ignite.internal.raft.WriteCommand;
+import org.apache.ignite.internal.table.distributed.TableMessageGroup;
+import org.apache.ignite.network.annotations.Transferable;
+
+/**
+ * State machine command to build a table index.
+ */
+@Transferable(TableMessageGroup.Commands.BUILD_INDEX)
+public interface BuildIndexCommand extends WriteCommand {
+ TablePartitionIdMessage tablePartitionId();
Review Comment:
Please add javadocs for all these methods
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -534,32 +584,159 @@ public BinaryTuple convert(BinaryRow binaryRow) {
}
private class ConfigurationListener implements
ConfigurationNamedListListener<TableIndexView> {
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onCreate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onCreate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexCreate(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onRename(
+ public CompletableFuture<?> onRename(
String oldName,
String newName,
ConfigurationNotificationEvent<TableIndexView> ctx
) {
return failedFuture(new
UnsupportedOperationException("https://issues.apache.org/jira/browse/IGNITE-16196";));
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onDelete(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onDelete(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexDrop(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onUpdate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onUpdate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return failedFuture(new IllegalStateException("Should not be
called"));
}
}
+
+ /**
+ * Initializes the build of the index.
+ */
+ private void initIndexBuildIfNeeded(TableIndexView tableIndexView,
TableImpl table) {
+ for (int partitionId = 0; partitionId <
table.internalTable().partitions(); partitionId++) {
+ buildIndexExecutor.submit(new BuildIndexTask(table,
tableIndexView, partitionId, true));
+ }
+ }
+
+ /**
+ * Task of building a table index for a partition.
+ *
+ * <p>Only the leader of the raft group will manage the building of the
index. Leader sends batches of row IDs via
+ * {@link BuildIndexCommand}, the next batch will only be send after the
previous batch has been processed.
+ *
+ * <p>Index building itself occurs locally on each node of the raft group
(majority) when processing {@link BuildIndexCommand}. This
+ * ensures that the index build in the raft group is consistent and that
the index build is restored after restarting the raft group
+ * (not from the beginning).
+ */
+ private class BuildIndexTask implements Runnable {
Review Comment:
Can we extract this class and the related executor to a separate class?
Something like an `IndexBuilder`
##########
modules/storage-api/src/main/java/org/apache/ignite/internal/storage/index/IndexStorage.java:
##########
@@ -54,4 +55,22 @@ public interface IndexStorage {
* @throws StorageException If failed to remove data.
*/
void remove(IndexRow row) throws StorageException;
+
+ /**
+ * Returns last row ID for which the index was built, {@code null} means
that the index was built.
Review Comment:
```suggestion
* Returns the last row ID that has been processed by an ongoing index
build process or {@code null} if the process has finished.
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -437,4 +439,33 @@ public void onShutdown() {
public MvPartitionStorage getMvStorage() {
return storage.getStorage();
}
+
+ /**
+ * Handler for the {@link BuildIndexCommand}.
+ *
+ * @param cmd Command.
+ * @param commandIndex RAFT index of the command.
+ * @param commandTerm RAFT term of the command.
+ */
+ void handleBuildIndexCommand(BuildIndexCommand cmd, long commandIndex,
long commandTerm) {
+ // Skips the write command because the storage has already executed it.
+ if (commandIndex <= storage.lastAppliedIndex()) {
+ return;
+ }
+
+ storage.runConsistently(() -> {
+ storage.lastApplied(commandIndex, commandTerm);
+
+ storageUpdateHandler.buildIndex(cmd.indexId(), cmd.rowIds(),
cmd.finish());
+
+ return null;
+ });
+
+ if (cmd.finish()) {
+ LOG.info(
+ "Finish building the index: [tableId={}, partitionId={},
indexId={}]",
Review Comment:
It's strange that we print the "start building index" message in
`IndexManager` but print the "finish" message here
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionDataStorage.java:
##########
@@ -166,6 +167,17 @@ public interface PartitionDataStorage extends
ManuallyCloseable {
*/
Cursor<ReadResult> scanVersions(RowId rowId) throws StorageException;
+ /**
+ * Scans the partition and returns a cursor of values at the given
timestamp. This cursor filters out committed tombstones, but not
+ * tombstones in the write-intent state.
+ *
+ * @param timestamp Timestamp. Can't be {@code null}.
+ * @return Cursor.
+ * @throws TxIdMismatchException If there's another pending update
associated with different transaction id.
Review Comment:
Is this a copy-paste mistake or what updates are you talking about?
##########
modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMetaStorage.java:
##########
@@ -103,13 +112,85 @@ void putPartitionId(int partitionId) {
}
}
+ /**
+ * Puts last row ID for which the index was built, {@code null} means
index building is finished.
+ *
+ * @param partitionId Partition ID.
+ * @param indexId Index ID.
+ * @param rowId Row ID.
+ */
+ public void putIndexLastBuildRowId(int partitionId, UUID indexId,
@Nullable RowId rowId) {
+ try {
+ metaColumnFamily.put(indexMetaKey(partitionId, indexId),
indexLastBuildRowId(rowId));
+ } catch (RocksDBException e) {
+ throw new StorageException(
+ "Failed to save last row ID for which the index was built:
[partitionId={}, indexId={}, rowId={}]",
+ e,
+ partitionId, indexId, rowId
+ );
+ }
+ }
+
+ /**
+ * Reads last row ID for which the index was built, {@code null} means
index building is finished.
+ *
+ * @param partitionId Partition ID.
+ * @param indexId Index ID.
+ * @param ifAbsent Will be returned if last row ID for which the index was
built has never been saved.
+ */
+ public @Nullable RowId readIndexLastBuildRowId(int partitionId, UUID
indexId, RowId ifAbsent) {
+ try {
+ byte[] lastBuildRowIdBytes =
metaColumnFamily.get(indexMetaKey(partitionId, indexId));
+
+ if (lastBuildRowIdBytes == null) {
+ return ifAbsent;
+ }
+
+ if (lastBuildRowIdBytes.length == 0) {
+ return null;
+ }
+
+ return new RowId(partitionId,
readUuid(ByteBuffer.wrap(lastBuildRowIdBytes), 0));
+ } catch (RocksDBException e) {
+ throw new StorageException(
+ "Failed to read last row ID for which the index was built:
[partitionId={}, indexId={}]",
+ e,
+ partitionId, indexId
+ );
+ }
+ }
+
static byte[] partitionIdKey(int partitionId) {
assert partitionId >= 0 && partitionId <= 0xFFFF : partitionId;
- return ByteBuffer.allocate(PARTITION_ID_PREFIX.length + Short.BYTES)
- .order(ByteOrder.BIG_ENDIAN)
+ return ByteBuffer.allocate(PARTITION_ID_PREFIX.length +
PARTITION_ID_SIZE)
+ .order(KEY_BYTE_ORDER)
.put(PARTITION_ID_PREFIX)
.putShort((short) partitionId)
.array();
}
+
+ static byte[] indexMetaKey(int partitionId, UUID indexId) {
Review Comment:
Should be private
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -534,32 +584,159 @@ public BinaryTuple convert(BinaryRow binaryRow) {
}
private class ConfigurationListener implements
ConfigurationNamedListListener<TableIndexView> {
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onCreate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onCreate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexCreate(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onRename(
+ public CompletableFuture<?> onRename(
String oldName,
String newName,
ConfigurationNotificationEvent<TableIndexView> ctx
) {
return failedFuture(new
UnsupportedOperationException("https://issues.apache.org/jira/browse/IGNITE-16196";));
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onDelete(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onDelete(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexDrop(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onUpdate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onUpdate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return failedFuture(new IllegalStateException("Should not be
called"));
}
}
+
+ /**
+ * Initializes the build of the index.
+ */
+ private void initIndexBuildIfNeeded(TableIndexView tableIndexView,
TableImpl table) {
+ for (int partitionId = 0; partitionId <
table.internalTable().partitions(); partitionId++) {
+ buildIndexExecutor.submit(new BuildIndexTask(table,
tableIndexView, partitionId, true));
+ }
+ }
+
+ /**
+ * Task of building a table index for a partition.
+ *
+ * <p>Only the leader of the raft group will manage the building of the
index. Leader sends batches of row IDs via
+ * {@link BuildIndexCommand}, the next batch will only be send after the
previous batch has been processed.
+ *
+ * <p>Index building itself occurs locally on each node of the raft group
(majority) when processing {@link BuildIndexCommand}. This
+ * ensures that the index build in the raft group is consistent and that
the index build is restored after restarting the raft group
Review Comment:
```suggestion
* ensures that the index build process in the raft group is consistent
and that the index build process is restored after restarting the raft group
```
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -534,32 +584,159 @@ public BinaryTuple convert(BinaryRow binaryRow) {
}
private class ConfigurationListener implements
ConfigurationNamedListListener<TableIndexView> {
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onCreate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onCreate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexCreate(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onRename(
+ public CompletableFuture<?> onRename(
String oldName,
String newName,
ConfigurationNotificationEvent<TableIndexView> ctx
) {
return failedFuture(new
UnsupportedOperationException("https://issues.apache.org/jira/browse/IGNITE-16196";));
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onDelete(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onDelete(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexDrop(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onUpdate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onUpdate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return failedFuture(new IllegalStateException("Should not be
called"));
}
}
+
+ /**
+ * Initializes the build of the index.
+ */
+ private void initIndexBuildIfNeeded(TableIndexView tableIndexView,
TableImpl table) {
+ for (int partitionId = 0; partitionId <
table.internalTable().partitions(); partitionId++) {
+ buildIndexExecutor.submit(new BuildIndexTask(table,
tableIndexView, partitionId, true));
+ }
+ }
+
+ /**
+ * Task of building a table index for a partition.
+ *
+ * <p>Only the leader of the raft group will manage the building of the
index. Leader sends batches of row IDs via
+ * {@link BuildIndexCommand}, the next batch will only be send after the
previous batch has been processed.
+ *
+ * <p>Index building itself occurs locally on each node of the raft group
(majority) when processing {@link BuildIndexCommand}. This
+ * ensures that the index build in the raft group is consistent and that
the index build is restored after restarting the raft group
+ * (not from the beginning).
+ */
+ private class BuildIndexTask implements Runnable {
+ private final TableImpl table;
+
+ private final TableIndexView tableIndexView;
+
+ private final int partitionId;
+
+ private final boolean firstBatch;
+
+ private BuildIndexTask(TableImpl table, TableIndexView tableIndexView,
int partitionId, boolean firstBatch) {
+ this.table = table;
+ this.tableIndexView = tableIndexView;
+ this.partitionId = partitionId;
+ this.firstBatch = firstBatch;
+ }
+
+ @Override
+ public void run() {
+ if (!busyLock.enterBusy()) {
+ return;
+ }
+
+ try {
+ InternalTable internalTable = table.internalTable();
+
+ RaftGroupService raftGroupService =
internalTable.partitionRaftGroupService(partitionId);
+
+ if (!isLocalNodeLeader(raftGroupService)) {
+ // TODO: IGNITE-19053 Must handle the change of leader
+ return;
+ }
+
+ RowId lastBuildRowId =
internalTable.storage().getOrCreateIndex(partitionId,
tableIndexView.id()).getLastBuildRowId();
+
+ if (lastBuildRowId == null) {
+ // Index has already been built.
+ return;
+ }
+
+ if (firstBatch) {
+ LOG.info("Start building the index: [{}]",
createCommonTableIndexInfo());
+ }
+
+ List<RowId> batchRowIds = createBatchRowIds(lastBuildRowId,
BUILD_INDEX_ROW_ID_BATCH_SIZE);
+
+ boolean finish = batchRowIds.size() <
BUILD_INDEX_ROW_ID_BATCH_SIZE;
+
+ raftGroupService.run(createBuildIndexCommand(batchRowIds,
finish))
+ .thenAccept(unused -> {
+ if (!finish) {
+ buildIndexExecutor.submit(new
BuildIndexTask(table, tableIndexView, partitionId, false));
+ }
+ });
+ } catch (Throwable t) {
+ LOG.error("Index build error: [{}]", t,
createCommonTableIndexInfo());
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ private boolean isLocalNodeLeader(RaftGroupService raftGroupService) {
+ Peer leader = raftGroupService.leader();
+
+ assert leader != null : "tableId=" + table.tableId() + ",
partitionId=" + partitionId;
+
+ return localNodeConsistentId().equals(leader.consistentId());
+ }
+
+ private List<RowId> createBatchRowIds(RowId lastBuildRowId, int
batchSize) {
+ MvPartitionStorage mvPartition =
table.internalTable().storage().getMvPartition(partitionId);
+
+ assert mvPartition != null : createCommonTableIndexInfo();
+
+ List<RowId> batch = new ArrayList<>(batchSize);
+
+ for (int i = 0; i < batchSize; i++) {
+ lastBuildRowId =
mvPartition.closestRowId(lastBuildRowId.increment());
Review Comment:
Do we guarantee somewhere that `RowId.lowestRowId` is never used as a valid
row ID? Otherwise we will skip this row when starting to build an index
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java:
##########
@@ -354,7 +354,7 @@ private List<IgniteComponent> startPartialNode(
new OutgoingSnapshotsManager(clusterSvc.messagingService())
);
- var indexManager = new IndexManager(tblCfg, schemaManager,
tableManager);
+ var indexManager = new IndexManager("test", tblCfg, schemaManager,
tableManager, clusterSvc);
Review Comment:
```suggestion
var indexManager = new IndexManager(name, tblCfg, schemaManager,
tableManager, clusterSvc);
```
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -534,32 +584,159 @@ public BinaryTuple convert(BinaryRow binaryRow) {
}
private class ConfigurationListener implements
ConfigurationNamedListListener<TableIndexView> {
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onCreate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onCreate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexCreate(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onRename(
+ public CompletableFuture<?> onRename(
String oldName,
String newName,
ConfigurationNotificationEvent<TableIndexView> ctx
) {
return failedFuture(new
UnsupportedOperationException("https://issues.apache.org/jira/browse/IGNITE-16196";));
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onDelete(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onDelete(ConfigurationNotificationEvent<TableIndexView> ctx) {
return onIndexDrop(ctx);
}
- /** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<?> onUpdate(@NotNull
ConfigurationNotificationEvent<TableIndexView> ctx) {
+ public CompletableFuture<?>
onUpdate(ConfigurationNotificationEvent<TableIndexView> ctx) {
return failedFuture(new IllegalStateException("Should not be
called"));
}
}
+
+ /**
+ * Initializes the build of the index.
+ */
+ private void initIndexBuildIfNeeded(TableIndexView tableIndexView,
TableImpl table) {
+ for (int partitionId = 0; partitionId <
table.internalTable().partitions(); partitionId++) {
+ buildIndexExecutor.submit(new BuildIndexTask(table,
tableIndexView, partitionId, true));
+ }
+ }
+
+ /**
+ * Task of building a table index for a partition.
+ *
+ * <p>Only the leader of the raft group will manage the building of the
index. Leader sends batches of row IDs via
+ * {@link BuildIndexCommand}, the next batch will only be send after the
previous batch has been processed.
+ *
+ * <p>Index building itself occurs locally on each node of the raft group
(majority) when processing {@link BuildIndexCommand}. This
+ * ensures that the index build in the raft group is consistent and that
the index build is restored after restarting the raft group
+ * (not from the beginning).
+ */
+ private class BuildIndexTask implements Runnable {
+ private final TableImpl table;
+
+ private final TableIndexView tableIndexView;
+
+ private final int partitionId;
+
+ private final boolean firstBatch;
+
+ private BuildIndexTask(TableImpl table, TableIndexView tableIndexView,
int partitionId, boolean firstBatch) {
+ this.table = table;
+ this.tableIndexView = tableIndexView;
+ this.partitionId = partitionId;
+ this.firstBatch = firstBatch;
+ }
+
+ @Override
+ public void run() {
+ if (!busyLock.enterBusy()) {
+ return;
+ }
+
+ try {
+ InternalTable internalTable = table.internalTable();
+
+ RaftGroupService raftGroupService =
internalTable.partitionRaftGroupService(partitionId);
+
+ if (!isLocalNodeLeader(raftGroupService)) {
+ // TODO: IGNITE-19053 Must handle the change of leader
+ return;
+ }
+
+ RowId lastBuildRowId =
internalTable.storage().getOrCreateIndex(partitionId,
tableIndexView.id()).getLastBuildRowId();
+
+ if (lastBuildRowId == null) {
+ // Index has already been built.
+ return;
+ }
+
+ if (firstBatch) {
+ LOG.info("Start building the index: [{}]",
createCommonTableIndexInfo());
+ }
+
+ List<RowId> batchRowIds = createBatchRowIds(lastBuildRowId,
BUILD_INDEX_ROW_ID_BATCH_SIZE);
+
+ boolean finish = batchRowIds.size() <
BUILD_INDEX_ROW_ID_BATCH_SIZE;
+
+ raftGroupService.run(createBuildIndexCommand(batchRowIds,
finish))
+ .thenAccept(unused -> {
+ if (!finish) {
+ buildIndexExecutor.submit(new
BuildIndexTask(table, tableIndexView, partitionId, false));
+ }
+ });
+ } catch (Throwable t) {
+ LOG.error("Index build error: [{}]", t,
createCommonTableIndexInfo());
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ private boolean isLocalNodeLeader(RaftGroupService raftGroupService) {
+ Peer leader = raftGroupService.leader();
+
+ assert leader != null : "tableId=" + table.tableId() + ",
partitionId=" + partitionId;
+
+ return localNodeConsistentId().equals(leader.consistentId());
+ }
+
+ private List<RowId> createBatchRowIds(RowId lastBuildRowId, int
batchSize) {
+ MvPartitionStorage mvPartition =
table.internalTable().storage().getMvPartition(partitionId);
+
+ assert mvPartition != null : createCommonTableIndexInfo();
+
+ List<RowId> batch = new ArrayList<>(batchSize);
+
+ for (int i = 0; i < batchSize; i++) {
+ lastBuildRowId =
mvPartition.closestRowId(lastBuildRowId.increment());
+
+ if (lastBuildRowId == null) {
+ break;
+ }
+
+ batch.add(lastBuildRowId);
+ }
+
+ return batch;
+ }
+
+ private BuildIndexCommand createBuildIndexCommand(List<RowId> rowIds,
boolean finish) {
+ return TABLE_MESSAGES_FACTORY.buildIndexCommand()
+
.tablePartitionId(TABLE_MESSAGES_FACTORY.tablePartitionIdMessage()
+ .tableId(table.tableId())
+ .partitionId(partitionId)
+ .build()
+ )
+ .indexId(tableIndexView.id())
+ .rowIds(rowIds.stream().map(RowId::uuid).collect(toList()))
+ .finish(finish)
+ .build();
+ }
+
+ private String createCommonTableIndexInfo() {
+ return "table=" + table.name() + ", tableId=" + table.tableId()
+ + ", partitionId=" + partitionId
+ + ", index=" + tableIndexView.name() + ", indexId=" +
tableIndexView.id();
+ }
+ }
+
+ private String localNodeConsistentId() {
Review Comment:
This method is only used by `BuildIndexTask`, I guess it should be moved
there
##########
modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/AbstractMvTableStorageTest.java:
##########
@@ -729,6 +729,82 @@ void testDestroyStartedRebalance() {
assertThat(tableStorage.destroyPartition(PARTITION_ID),
willCompleteSuccessfully());
}
+ @Test
+ void testIndexLastBuildRowId() {
+ MvPartitionStorage mvPartitionStorage =
getOrCreateMvPartition(PARTITION_ID);
+
+ IndexStorage sortedIndexStorage =
tableStorage.getOrCreateIndex(PARTITION_ID, sortedIdx.id());
Review Comment:
I would recommend to split this and the next tests in two (each): one for
hash index and one for the sorted index
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java:
##########
@@ -460,6 +456,48 @@ public void testSafeTime() {
applySafeTimeCommand(SafeTimeSyncCommand.class, testClock.now());
}
+ @Test
+ void testBuildIndexCommand() {
+ UUID indexId = UUID.randomUUID();
+
+ doNothing().when(storageUpdateHandler).buildIndex(eq(indexId),
any(List.class), anyBoolean());
Review Comment:
what's the point of this mocking? Why are we using a spy here?
##########
modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMetaStorage.java:
##########
@@ -103,13 +112,85 @@ void putPartitionId(int partitionId) {
}
}
+ /**
+ * Puts last row ID for which the index was built, {@code null} means
index building is finished.
+ *
+ * @param partitionId Partition ID.
+ * @param indexId Index ID.
+ * @param rowId Row ID.
+ */
+ public void putIndexLastBuildRowId(int partitionId, UUID indexId,
@Nullable RowId rowId) {
+ try {
+ metaColumnFamily.put(indexMetaKey(partitionId, indexId),
indexLastBuildRowId(rowId));
+ } catch (RocksDBException e) {
+ throw new StorageException(
+ "Failed to save last row ID for which the index was built:
[partitionId={}, indexId={}, rowId={}]",
+ e,
+ partitionId, indexId, rowId
+ );
+ }
+ }
+
+ /**
+ * Reads last row ID for which the index was built, {@code null} means
index building is finished.
+ *
+ * @param partitionId Partition ID.
+ * @param indexId Index ID.
+ * @param ifAbsent Will be returned if last row ID for which the index was
built has never been saved.
+ */
+ public @Nullable RowId readIndexLastBuildRowId(int partitionId, UUID
indexId, RowId ifAbsent) {
+ try {
+ byte[] lastBuildRowIdBytes =
metaColumnFamily.get(indexMetaKey(partitionId, indexId));
+
+ if (lastBuildRowIdBytes == null) {
+ return ifAbsent;
+ }
+
+ if (lastBuildRowIdBytes.length == 0) {
+ return null;
+ }
+
+ return new RowId(partitionId,
readUuid(ByteBuffer.wrap(lastBuildRowIdBytes), 0));
+ } catch (RocksDBException e) {
+ throw new StorageException(
+ "Failed to read last row ID for which the index was built:
[partitionId={}, indexId={}]",
+ e,
+ partitionId, indexId
+ );
+ }
+ }
+
static byte[] partitionIdKey(int partitionId) {
assert partitionId >= 0 && partitionId <= 0xFFFF : partitionId;
- return ByteBuffer.allocate(PARTITION_ID_PREFIX.length + Short.BYTES)
- .order(ByteOrder.BIG_ENDIAN)
+ return ByteBuffer.allocate(PARTITION_ID_PREFIX.length +
PARTITION_ID_SIZE)
+ .order(KEY_BYTE_ORDER)
.put(PARTITION_ID_PREFIX)
.putShort((short) partitionId)
.array();
}
+
+ static byte[] indexMetaKey(int partitionId, UUID indexId) {
+ assert partitionId >= 0 && partitionId <= 0xFFFF : partitionId;
+
+ ByteBuffer buffer =
ByteBuffer.allocate(INDEX_META_KEY_SIZE).order(KEY_BYTE_ORDER);
+
+ buffer.put(INDEX_META_KEY_PREFIX).putShort((short) partitionId);
+
+ putUuid(buffer, indexId);
+
+ return buffer.array();
+ }
+
+ static byte[] indexLastBuildRowId(@Nullable RowId rowId) {
Review Comment:
Should be private
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
