ibessonov commented on code in PR #1800: URL: https://github.com/apache/ignite-3/pull/1800#discussion_r1145763549
########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,220 @@ +/* + * 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.index; + +import static java.util.stream.Collectors.toList; +import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.index.TableIndexView; +import org.apache.ignite.internal.storage.MvPartitionStorage; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.internal.table.distributed.TableMessagesFactory; +import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand; +import org.apache.ignite.internal.thread.NamedThreadFactory; +import org.apache.ignite.internal.util.IgniteSpinBusyLock; +import org.apache.ignite.network.ClusterService; + +/** + * Class for managing the index building process. + */ +class IndexBuilder { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuilder.class); + + /** Batch size of row IDs to build the index. */ + private static final int BUILD_INDEX_ROW_ID_BATCH_SIZE = 100; + + /** Message factory to create messages - RAFT commands. */ + private static final TableMessagesFactory TABLE_MESSAGES_FACTORY = new TableMessagesFactory(); + + /** Busy lock to stop synchronously. */ + private final IgniteSpinBusyLock busyLock; + + /** Cluster service. */ + private final ClusterService clusterService; + + /** Index building executor. */ + private final ExecutorService buildIndexExecutor; + + IndexBuilder(String nodeName, IgniteSpinBusyLock busyLock, ClusterService clusterService) { + this.busyLock = busyLock; + this.clusterService = clusterService; + + int cpus = Runtime.getRuntime().availableProcessors(); + + buildIndexExecutor = new ThreadPoolExecutor( + cpus, + cpus, + 30, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + NamedThreadFactory.create(nodeName, "build-index", LOG) + ); + } + + /** + * Stops the index builder. + */ + void stop() { Review Comment: Is this method called after the busyLock is blocked or before? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,220 @@ +/* + * 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.index; + +import static java.util.stream.Collectors.toList; +import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.index.TableIndexView; +import org.apache.ignite.internal.storage.MvPartitionStorage; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.internal.table.distributed.TableMessagesFactory; +import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand; +import org.apache.ignite.internal.thread.NamedThreadFactory; +import org.apache.ignite.internal.util.IgniteSpinBusyLock; +import org.apache.ignite.network.ClusterService; + +/** + * Class for managing the index building process. + */ +class IndexBuilder { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuilder.class); + + /** Batch size of row IDs to build the index. */ + private static final int BUILD_INDEX_ROW_ID_BATCH_SIZE = 100; + + /** Message factory to create messages - RAFT commands. */ + private static final TableMessagesFactory TABLE_MESSAGES_FACTORY = new TableMessagesFactory(); + + /** Busy lock to stop synchronously. */ + private final IgniteSpinBusyLock busyLock; + + /** Cluster service. */ + private final ClusterService clusterService; + + /** Index building executor. */ + private final ExecutorService buildIndexExecutor; + + IndexBuilder(String nodeName, IgniteSpinBusyLock busyLock, ClusterService clusterService) { + this.busyLock = busyLock; + this.clusterService = clusterService; + + int cpus = Runtime.getRuntime().availableProcessors(); + + buildIndexExecutor = new ThreadPoolExecutor( + cpus, + cpus, + 30, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + NamedThreadFactory.create(nodeName, "build-index", LOG) + ); + } + + /** + * Stops the index builder. + */ + void stop() { + shutdownAndAwaitTermination(buildIndexExecutor, 10, TimeUnit.SECONDS); + } + + /** + * Initializes the build of the index. + */ + void startIndexBuild(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 when processing {@link BuildIndexCommand}. This 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 + * (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(); Review Comment: I don't like the usage of "getOrCreateIndex", it should already be created at this point, right? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,220 @@ +/* + * 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.index; + +import static java.util.stream.Collectors.toList; +import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.index.TableIndexView; +import org.apache.ignite.internal.storage.MvPartitionStorage; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.internal.table.distributed.TableMessagesFactory; +import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand; +import org.apache.ignite.internal.thread.NamedThreadFactory; +import org.apache.ignite.internal.util.IgniteSpinBusyLock; +import org.apache.ignite.network.ClusterService; + +/** + * Class for managing the index building process. + */ +class IndexBuilder { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuilder.class); + + /** Batch size of row IDs to build the index. */ + private static final int BUILD_INDEX_ROW_ID_BATCH_SIZE = 100; + + /** Message factory to create messages - RAFT commands. */ + private static final TableMessagesFactory TABLE_MESSAGES_FACTORY = new TableMessagesFactory(); + + /** Busy lock to stop synchronously. */ + private final IgniteSpinBusyLock busyLock; + + /** Cluster service. */ + private final ClusterService clusterService; + + /** Index building executor. */ + private final ExecutorService buildIndexExecutor; + + IndexBuilder(String nodeName, IgniteSpinBusyLock busyLock, ClusterService clusterService) { + this.busyLock = busyLock; + this.clusterService = clusterService; + + int cpus = Runtime.getRuntime().availableProcessors(); + + buildIndexExecutor = new ThreadPoolExecutor( + cpus, + cpus, + 30, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + NamedThreadFactory.create(nodeName, "build-index", LOG) + ); + } + + /** + * Stops the index builder. + */ + void stop() { + shutdownAndAwaitTermination(buildIndexExecutor, 10, TimeUnit.SECONDS); + } + + /** + * Initializes the build of the index. + */ + void startIndexBuild(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 when processing {@link BuildIndexCommand}. This 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 + * (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) { Review Comment: What exactly do you mean by first batch? First batch after restart? Or first batch for the index in general? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,220 @@ +/* + * 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.index; + +import static java.util.stream.Collectors.toList; +import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.index.TableIndexView; +import org.apache.ignite.internal.storage.MvPartitionStorage; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.internal.table.distributed.TableMessagesFactory; +import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand; +import org.apache.ignite.internal.thread.NamedThreadFactory; +import org.apache.ignite.internal.util.IgniteSpinBusyLock; +import org.apache.ignite.network.ClusterService; + +/** + * Class for managing the index building process. + */ +class IndexBuilder { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuilder.class); + + /** Batch size of row IDs to build the index. */ + private static final int BUILD_INDEX_ROW_ID_BATCH_SIZE = 100; + + /** Message factory to create messages - RAFT commands. */ + private static final TableMessagesFactory TABLE_MESSAGES_FACTORY = new TableMessagesFactory(); + + /** Busy lock to stop synchronously. */ + private final IgniteSpinBusyLock busyLock; + + /** Cluster service. */ + private final ClusterService clusterService; + + /** Index building executor. */ + private final ExecutorService buildIndexExecutor; + + IndexBuilder(String nodeName, IgniteSpinBusyLock busyLock, ClusterService clusterService) { + this.busyLock = busyLock; + this.clusterService = clusterService; + + int cpus = Runtime.getRuntime().availableProcessors(); + + buildIndexExecutor = new ThreadPoolExecutor( + cpus, + cpus, + 30, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + NamedThreadFactory.create(nodeName, "build-index", LOG) + ); + } + + /** + * Stops the index builder. + */ + void stop() { + shutdownAndAwaitTermination(buildIndexExecutor, 10, TimeUnit.SECONDS); + } + + /** + * Initializes the build of the index. + */ + void startIndexBuild(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 when processing {@link BuildIndexCommand}. This 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 + * (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; Review Comment: With such check, there's a chance of replicating empty list of row ids. I don't thing that this is a good solution, can you please fix it? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,220 @@ +/* + * 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.index; + +import static java.util.stream.Collectors.toList; +import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.index.TableIndexView; +import org.apache.ignite.internal.storage.MvPartitionStorage; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.internal.table.distributed.TableMessagesFactory; +import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand; +import org.apache.ignite.internal.thread.NamedThreadFactory; +import org.apache.ignite.internal.util.IgniteSpinBusyLock; +import org.apache.ignite.network.ClusterService; + +/** + * Class for managing the index building process. + */ +class IndexBuilder { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuilder.class); + + /** Batch size of row IDs to build the index. */ + private static final int BUILD_INDEX_ROW_ID_BATCH_SIZE = 100; + + /** Message factory to create messages - RAFT commands. */ + private static final TableMessagesFactory TABLE_MESSAGES_FACTORY = new TableMessagesFactory(); + + /** Busy lock to stop synchronously. */ + private final IgniteSpinBusyLock busyLock; + + /** Cluster service. */ + private final ClusterService clusterService; + + /** Index building executor. */ + private final ExecutorService buildIndexExecutor; + + IndexBuilder(String nodeName, IgniteSpinBusyLock busyLock, ClusterService clusterService) { + this.busyLock = busyLock; + this.clusterService = clusterService; + + int cpus = Runtime.getRuntime().availableProcessors(); + + buildIndexExecutor = new ThreadPoolExecutor( + cpus, + cpus, + 30, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + NamedThreadFactory.create(nodeName, "build-index", LOG) + ); + } + + /** + * Stops the index builder. + */ + void stop() { + shutdownAndAwaitTermination(buildIndexExecutor, 10, TimeUnit.SECONDS); + } + + /** + * Initializes the build of the index. + */ + void startIndexBuild(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 when processing {@link BuildIndexCommand}. This 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 + * (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)) + .thenRun(() -> { + 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) { Review Comment: By the way, when you say "last build", you mean "last buil**t**" or something else? Maybe it would make sense to rename it, for clarity ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,220 @@ +/* + * 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.index; + +import static java.util.stream.Collectors.toList; +import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.index.TableIndexView; +import org.apache.ignite.internal.storage.MvPartitionStorage; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.internal.table.distributed.TableMessagesFactory; +import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand; +import org.apache.ignite.internal.thread.NamedThreadFactory; +import org.apache.ignite.internal.util.IgniteSpinBusyLock; +import org.apache.ignite.network.ClusterService; + +/** + * Class for managing the index building process. + */ +class IndexBuilder { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuilder.class); + + /** Batch size of row IDs to build the index. */ + private static final int BUILD_INDEX_ROW_ID_BATCH_SIZE = 100; Review Comment: Should we make it configurable? I think that it's a big problem, that we don't discuss configuration and just hard-code such parameters. ########## modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItBuildIndexTest.java: ########## @@ -0,0 +1,144 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.sql.engine; + +import static java.util.stream.Collectors.joining; +import static java.util.stream.Collectors.toSet; +import static org.apache.ignite.internal.sql.engine.util.QueryChecker.containsIndexScan; +import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition; +import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Stream; +import org.apache.ignite.Ignite; +import org.apache.ignite.internal.app.IgniteImpl; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.TablesConfiguration; +import org.apache.ignite.internal.storage.index.IndexStorage; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.lang.IgniteStringFormatter; +import org.apache.ignite.table.Table; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Integration test of index building. + */ +public class ItBuildIndexTest extends ClusterPerClassIntegrationTest { + private static final String TABLE_NAME = "test_table"; + + private static final String INDEX_NAME = "test_index"; + + @AfterEach + void tearDown() { + sql("DROP TABLE IF EXISTS " + TABLE_NAME); + } + + @ParameterizedTest + @MethodSource("replicas") + @Disabled("https://issues.apache.org/jira/browse/IGNITE-19085";) + void testBuildIndexOnStableTopology(int replicas) throws Exception { + sql(IgniteStringFormatter.format( + "CREATE TABLE {} (i0 INTEGER PRIMARY KEY, i1 INTEGER) WITH replicas={}, partitions={}", + TABLE_NAME, replicas, 2 + )); + + sql(IgniteStringFormatter.format( + "INSERT INTO {} VALUES {}", + TABLE_NAME, toValuesString(List.of(1, 1), List.of(2, 2), List.of(3, 3), List.of(4, 4), List.of(5, 5)) + )); + + sql(IgniteStringFormatter.format("CREATE INDEX {} ON {} (i1)", INDEX_NAME, TABLE_NAME)); + + // FIXME: IGNITE-18733 + waitForIndex(INDEX_NAME); + + waitForIndexBuild(TABLE_NAME, INDEX_NAME); + + assertQuery(IgniteStringFormatter.format("SELECT * FROM {} WHERE i1 > 0", TABLE_NAME)) Review Comment: Wow, that's a nice method, didn't know about its existence ########## modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItBuildIndexTest.java: ########## @@ -0,0 +1,144 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.sql.engine; + +import static java.util.stream.Collectors.joining; +import static java.util.stream.Collectors.toSet; +import static org.apache.ignite.internal.sql.engine.util.QueryChecker.containsIndexScan; +import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition; +import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Stream; +import org.apache.ignite.Ignite; +import org.apache.ignite.internal.app.IgniteImpl; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.TablesConfiguration; +import org.apache.ignite.internal.storage.index.IndexStorage; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.lang.IgniteStringFormatter; +import org.apache.ignite.table.Table; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Integration test of index building. + */ +public class ItBuildIndexTest extends ClusterPerClassIntegrationTest { + private static final String TABLE_NAME = "test_table"; + + private static final String INDEX_NAME = "test_index"; + + @AfterEach + void tearDown() { + sql("DROP TABLE IF EXISTS " + TABLE_NAME); + } + + @ParameterizedTest + @MethodSource("replicas") + @Disabled("https://issues.apache.org/jira/browse/IGNITE-19085";) + void testBuildIndexOnStableTopology(int replicas) throws Exception { + sql(IgniteStringFormatter.format( + "CREATE TABLE {} (i0 INTEGER PRIMARY KEY, i1 INTEGER) WITH replicas={}, partitions={}", + TABLE_NAME, replicas, 2 + )); + + sql(IgniteStringFormatter.format( + "INSERT INTO {} VALUES {}", + TABLE_NAME, toValuesString(List.of(1, 1), List.of(2, 2), List.of(3, 3), List.of(4, 4), List.of(5, 5)) + )); + + sql(IgniteStringFormatter.format("CREATE INDEX {} ON {} (i1)", INDEX_NAME, TABLE_NAME)); + + // FIXME: IGNITE-18733 + waitForIndex(INDEX_NAME); + + waitForIndexBuild(TABLE_NAME, INDEX_NAME); + + assertQuery(IgniteStringFormatter.format("SELECT * FROM {} WHERE i1 > 0", TABLE_NAME)) + .matches(containsIndexScan("PUBLIC", TABLE_NAME.toUpperCase(), INDEX_NAME.toUpperCase())) + .returns(1, 1) + .returns(2, 2) + .returns(3, 3) + .returns(4, 4) + .returns(5, 5) + .check(); + } + + private static int[] replicas() { + // FIXME: IGNITE-19086 Fix NullPointerException on insertALl + // return new int[]{1, 2, 3}; + return new int[]{1}; + } + + private static String toValuesString(List<Object>... values) { + return Stream.of(values) + .peek(Assertions::assertNotNull) + .map(objects -> objects.stream().map(Object::toString).collect(joining(", ", "(", ")"))) + .collect(joining(", ")); + } + + private void waitForIndexBuild(String tableName, String indexName) throws Exception { + for (Ignite clusterNode : CLUSTER_NODES) { + CompletableFuture<Table> tableFuture = clusterNode.tables().tableAsync(tableName); + + assertThat(tableFuture, willCompleteSuccessfully()); + + TableImpl tableImpl = (TableImpl) tableFuture.join(); + + InternalTable internalTable = tableImpl.internalTable(); + + UUID indexId = ((IgniteImpl) clusterNode).clusterConfiguration() + .getConfiguration(TablesConfiguration.KEY) + .indexes() + .get(INDEX_NAME.toUpperCase()) + .id() + .value(); + + assertNotNull(indexId, "table=" + tableName + ", index=" + indexName); + + for (int partitionId = 0; partitionId < internalTable.partitions(); partitionId++) { + RaftGroupService raftGroupService = internalTable.partitionRaftGroupService(partitionId); + + Set<String> allPeers = Stream.concat( + Stream.of(raftGroupService.leader()), + raftGroupService.peers().stream() + ).map(Peer::consistentId).collect(toSet()); + + if (!allPeers.contains(clusterNode.name())) { + continue; + } Review Comment: Oh no, please, never collect anything into a collection just to check it for "contains". Why don't you use `anyMatch`? ########## 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 the last row ID that has been processed by an ongoing index build process or {@code null} if the process has finished. Review Comment: Please document, what happens if the process is not started. ########## modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItBuildIndexTest.java: ########## @@ -0,0 +1,144 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.sql.engine; + +import static java.util.stream.Collectors.joining; +import static java.util.stream.Collectors.toSet; +import static org.apache.ignite.internal.sql.engine.util.QueryChecker.containsIndexScan; +import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition; +import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Stream; +import org.apache.ignite.Ignite; +import org.apache.ignite.internal.app.IgniteImpl; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.TablesConfiguration; +import org.apache.ignite.internal.storage.index.IndexStorage; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.lang.IgniteStringFormatter; +import org.apache.ignite.table.Table; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Integration test of index building. + */ +public class ItBuildIndexTest extends ClusterPerClassIntegrationTest { + private static final String TABLE_NAME = "test_table"; + + private static final String INDEX_NAME = "test_index"; + + @AfterEach + void tearDown() { + sql("DROP TABLE IF EXISTS " + TABLE_NAME); + } + + @ParameterizedTest + @MethodSource("replicas") + @Disabled("https://issues.apache.org/jira/browse/IGNITE-19085";) + void testBuildIndexOnStableTopology(int replicas) throws Exception { + sql(IgniteStringFormatter.format( + "CREATE TABLE {} (i0 INTEGER PRIMARY KEY, i1 INTEGER) WITH replicas={}, partitions={}", + TABLE_NAME, replicas, 2 + )); + + sql(IgniteStringFormatter.format( + "INSERT INTO {} VALUES {}", + TABLE_NAME, toValuesString(List.of(1, 1), List.of(2, 2), List.of(3, 3), List.of(4, 4), List.of(5, 5)) + )); + + sql(IgniteStringFormatter.format("CREATE INDEX {} ON {} (i1)", INDEX_NAME, TABLE_NAME)); + + // FIXME: IGNITE-18733 + waitForIndex(INDEX_NAME); + + waitForIndexBuild(TABLE_NAME, INDEX_NAME); + + assertQuery(IgniteStringFormatter.format("SELECT * FROM {} WHERE i1 > 0", TABLE_NAME)) + .matches(containsIndexScan("PUBLIC", TABLE_NAME.toUpperCase(), INDEX_NAME.toUpperCase())) + .returns(1, 1) + .returns(2, 2) + .returns(3, 3) + .returns(4, 4) + .returns(5, 5) + .check(); + } + + private static int[] replicas() { + // FIXME: IGNITE-19086 Fix NullPointerException on insertALl + // return new int[]{1, 2, 3}; + return new int[]{1}; + } + + private static String toValuesString(List<Object>... values) { + return Stream.of(values) + .peek(Assertions::assertNotNull) + .map(objects -> objects.stream().map(Object::toString).collect(joining(", ", "(", ")"))) + .collect(joining(", ")); + } + + private void waitForIndexBuild(String tableName, String indexName) throws Exception { + for (Ignite clusterNode : CLUSTER_NODES) { + CompletableFuture<Table> tableFuture = clusterNode.tables().tableAsync(tableName); + + assertThat(tableFuture, willCompleteSuccessfully()); + + TableImpl tableImpl = (TableImpl) tableFuture.join(); + + InternalTable internalTable = tableImpl.internalTable(); + + UUID indexId = ((IgniteImpl) clusterNode).clusterConfiguration() + .getConfiguration(TablesConfiguration.KEY) + .indexes() + .get(INDEX_NAME.toUpperCase()) + .id() + .value(); + + assertNotNull(indexId, "table=" + tableName + ", index=" + indexName); + + for (int partitionId = 0; partitionId < internalTable.partitions(); partitionId++) { + RaftGroupService raftGroupService = internalTable.partitionRaftGroupService(partitionId); + + Set<String> allPeers = Stream.concat( + Stream.of(raftGroupService.leader()), + raftGroupService.peers().stream() + ).map(Peer::consistentId).collect(toSet()); + + if (!allPeers.contains(clusterNode.name())) { + continue; + } + + IndexStorage index = internalTable.storage().getOrCreateIndex(partitionId, indexId); Review Comment: Do we even have a `getIndex` method? ########## modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/PageMemoryHashIndexStorage.java: ########## @@ -329,4 +340,34 @@ public void startCleanup() { public void finishCleanup() { state.compareAndSet(StorageState.CLEANUP, StorageState.RUNNABLE); } + + @Override + public @Nullable RowId getLastBuildRowId() { + return busy(() -> { + throwExceptionIfStorageInProgressOfRebalance(state.get(), this::createStorageInfo); + + try { + UUID lastBuildRowIdUuid = indexMetaTree.findOne(new IndexMetaKey(indexDescriptor().id())).lastBuildRowIdUuid(); Review Comment: I don't like that we always read it from the tree, access to cached version would be so much faster. The only time we really must read it is the start ########## modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/UpdateLastBuildRowIdInvokeClosure.java: ########## @@ -0,0 +1,59 @@ +/* + * 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.pagememory.index.meta; + +import org.apache.ignite.internal.pagememory.tree.IgniteTree.InvokeClosure; +import org.apache.ignite.internal.pagememory.tree.IgniteTree.OperationType; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.lang.IgniteInternalCheckedException; +import org.jetbrains.annotations.Nullable; + +/** + * Closure for updating the last row ID for which the index was built. + */ +public class UpdateLastBuildRowIdInvokeClosure implements InvokeClosure<IndexMeta> { + private final @Nullable RowId newRowId; + + private IndexMeta newRow; + + /** + * Constructor. + * + * @param newRowId New last row ID for which the index was built, {@code null} means index building is finished. + */ + public UpdateLastBuildRowIdInvokeClosure(@Nullable RowId newRowId) { + this.newRowId = newRowId; + } + + @Override + public void call(@Nullable IndexMeta oldRow) throws IgniteInternalCheckedException { + assert oldRow != null; + + newRow = new IndexMeta(oldRow.indexId(), oldRow.metaPageId(), newRowId == null ? null : newRowId.uuid()); Review Comment: This is not the best place to write such comment, probably, but can new value be smaller then the old one? On leader reelection, for example? ########## 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 the last row ID that has been processed by an ongoing index build process or {@code null} if the process has finished. + * + * @throws StorageException If failed to get the last row ID. + */ + @Nullable RowId getLastBuildRowId(); + + /** + * Sets last row ID for which the index was built, {@code null} means index building is finished. + * + * @apiNote This method <b>must</b> always be called inside the corresponding partition's + * {@link org.apache.ignite.internal.storage.MvPartitionStorage#runConsistently} closure. + * + * @param rowId Row ID. + * @throws StorageException If failed to set the last row ID. + */ + void setLastBuildRowId(@Nullable RowId rowId); Review Comment: ```suggestion void setLastBuildRowId(@Nullable RowId rowId) throws StorageExeption; ``` ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,220 @@ +/* + * 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.index; + +import static java.util.stream.Collectors.toList; +import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.index.TableIndexView; +import org.apache.ignite.internal.storage.MvPartitionStorage; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.internal.table.distributed.TableMessagesFactory; +import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand; +import org.apache.ignite.internal.thread.NamedThreadFactory; +import org.apache.ignite.internal.util.IgniteSpinBusyLock; +import org.apache.ignite.network.ClusterService; + +/** + * Class for managing the index building process. + */ +class IndexBuilder { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuilder.class); + + /** Batch size of row IDs to build the index. */ + private static final int BUILD_INDEX_ROW_ID_BATCH_SIZE = 100; + + /** Message factory to create messages - RAFT commands. */ + private static final TableMessagesFactory TABLE_MESSAGES_FACTORY = new TableMessagesFactory(); + + /** Busy lock to stop synchronously. */ + private final IgniteSpinBusyLock busyLock; + + /** Cluster service. */ + private final ClusterService clusterService; + + /** Index building executor. */ + private final ExecutorService buildIndexExecutor; + + IndexBuilder(String nodeName, IgniteSpinBusyLock busyLock, ClusterService clusterService) { + this.busyLock = busyLock; + this.clusterService = clusterService; + + int cpus = Runtime.getRuntime().availableProcessors(); + + buildIndexExecutor = new ThreadPoolExecutor( + cpus, + cpus, + 30, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + NamedThreadFactory.create(nodeName, "build-index", LOG) + ); + } + + /** + * Stops the index builder. + */ + void stop() { + shutdownAndAwaitTermination(buildIndexExecutor, 10, TimeUnit.SECONDS); + } + + /** + * Initializes the build of the index. + */ + void startIndexBuild(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 when processing {@link BuildIndexCommand}. This 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 + * (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) { Review Comment: It's weird that you rely on the value in the storage itself, not the last row id that you replicated. What's the reason of such solution? Can you please document, why the value that you're reading is already replicated and processed? Or maybe use locally preserved last row id instead? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,220 @@ +/* + * 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.index; + +import static java.util.stream.Collectors.toList; +import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.index.TableIndexView; +import org.apache.ignite.internal.storage.MvPartitionStorage; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.internal.table.distributed.TableMessagesFactory; +import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand; +import org.apache.ignite.internal.thread.NamedThreadFactory; +import org.apache.ignite.internal.util.IgniteSpinBusyLock; +import org.apache.ignite.network.ClusterService; + +/** + * Class for managing the index building process. + */ +class IndexBuilder { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuilder.class); + + /** Batch size of row IDs to build the index. */ + private static final int BUILD_INDEX_ROW_ID_BATCH_SIZE = 100; + + /** Message factory to create messages - RAFT commands. */ + private static final TableMessagesFactory TABLE_MESSAGES_FACTORY = new TableMessagesFactory(); + + /** Busy lock to stop synchronously. */ + private final IgniteSpinBusyLock busyLock; + + /** Cluster service. */ + private final ClusterService clusterService; + + /** Index building executor. */ + private final ExecutorService buildIndexExecutor; + + IndexBuilder(String nodeName, IgniteSpinBusyLock busyLock, ClusterService clusterService) { + this.busyLock = busyLock; + this.clusterService = clusterService; + + int cpus = Runtime.getRuntime().availableProcessors(); + + buildIndexExecutor = new ThreadPoolExecutor( + cpus, + cpus, + 30, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + NamedThreadFactory.create(nodeName, "build-index", LOG) + ); + } + + /** + * Stops the index builder. + */ + void stop() { + shutdownAndAwaitTermination(buildIndexExecutor, 10, TimeUnit.SECONDS); + } + + /** + * Initializes the build of the index. + */ + void startIndexBuild(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 when processing {@link BuildIndexCommand}. This 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 + * (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); Review Comment: If current node is not an affinity node, will this method return null? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,220 @@ +/* + * 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.index; + +import static java.util.stream.Collectors.toList; +import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.index.TableIndexView; +import org.apache.ignite.internal.storage.MvPartitionStorage; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.internal.table.distributed.TableMessagesFactory; +import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand; +import org.apache.ignite.internal.thread.NamedThreadFactory; +import org.apache.ignite.internal.util.IgniteSpinBusyLock; +import org.apache.ignite.network.ClusterService; + +/** + * Class for managing the index building process. + */ +class IndexBuilder { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuilder.class); + + /** Batch size of row IDs to build the index. */ + private static final int BUILD_INDEX_ROW_ID_BATCH_SIZE = 100; + + /** Message factory to create messages - RAFT commands. */ + private static final TableMessagesFactory TABLE_MESSAGES_FACTORY = new TableMessagesFactory(); + + /** Busy lock to stop synchronously. */ + private final IgniteSpinBusyLock busyLock; + + /** Cluster service. */ + private final ClusterService clusterService; + + /** Index building executor. */ + private final ExecutorService buildIndexExecutor; + + IndexBuilder(String nodeName, IgniteSpinBusyLock busyLock, ClusterService clusterService) { + this.busyLock = busyLock; + this.clusterService = clusterService; + + int cpus = Runtime.getRuntime().availableProcessors(); + + buildIndexExecutor = new ThreadPoolExecutor( + cpus, + cpus, + 30, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + NamedThreadFactory.create(nodeName, "build-index", LOG) + ); + } + + /** + * Stops the index builder. + */ + void stop() { + shutdownAndAwaitTermination(buildIndexExecutor, 10, TimeUnit.SECONDS); + } + + /** + * Initializes the build of the index. + */ + void startIndexBuild(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 when processing {@link BuildIndexCommand}. This 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 + * (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) { Review Comment: What's the initial value for the "lastBuildRowId" if build hasn't started yet? Minimal possible value? ########## modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/RocksDbHashIndexStorage.java: ########## @@ -369,4 +373,26 @@ public void finishCleanup() { busyLock.unblock(); } } + + @Override + public @Nullable RowId getLastBuildRowId() { + return busy(() -> { + throwExceptionIfStorageInProgressOfRebalance(state.get(), this::createStorageInfo); Review Comment: Same comment, it's much more convenient to read local field, rather than the entry from the storage ########## modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java: ########## @@ -1163,7 +1164,7 @@ private CompletableFuture<?> createTableLocally(long causalityToken, String name partitions, clusterNodeResolver, txManager, tableStorage, txStateStorage, replicaSvc, clock); // TODO: IGNITE-16288 directIndexIds should use async configuration API - var table = new TableImpl(internalTable, lockMgr, () -> CompletableFuture.supplyAsync(() -> directIndexIds())); + var table = new TableImpl(internalTable, lockMgr, () -> supplyAsync(this::directIndexIds, ioExecutor)); Review Comment: Nice, thank you! ########## modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java: ########## @@ -416,4 +416,39 @@ public void addToIndexes(@Nullable BinaryRow binaryRow, RowId rowId) { public void waitIndexes() { indexes.get(); } + + /** + * Builds an index for all versions of a row. + * + * <p>Index is expected to exist, skips the tombstones. + * + * @param indexId Index ID. + * @param rowUuids Row uuids. + * @param finish Index build completion flag. + */ + public void buildIndex(UUID indexId, List<UUID> rowUuids, boolean finish) { + TableSchemaAwareIndexStorage index = indexes.get().get(indexId); + + assert index != null : "indexId=" + indexId + ", partitionId=" + partitionId; + + RowId lastRowId = null; + + for (UUID rowUuid : rowUuids) { + lastRowId = new RowId(partitionId, rowUuid); + + try (Cursor<ReadResult> cursor = storage.scanVersions(lastRowId)) { + while (cursor.hasNext()) { + ReadResult next = cursor.next(); + + if (!next.isEmpty()) { + index.put(next.binaryRow(), lastRowId); + } + } + } + } + + assert lastRowId != null || finish : "indexId=" + indexId + ", partitionId=" + partitionId; Review Comment: Oh, so you explicitly assert that last batch can be empty. Why? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java: ########## @@ -91,33 +92,44 @@ public class IndexManager extends Producer<IndexEvent, IndexEventParameters> imp /** Prevents double stopping of the component. */ private final AtomicBoolean stopGuard = new AtomicBoolean(); + /** Index builder. */ + private IndexBuilder indexBuilder; + /** * 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"); Review Comment: You renamed the field, please change the name in the check as well ########## modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItBuildIndexTest.java: ########## @@ -0,0 +1,144 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.sql.engine; + +import static java.util.stream.Collectors.joining; +import static java.util.stream.Collectors.toSet; +import static org.apache.ignite.internal.sql.engine.util.QueryChecker.containsIndexScan; +import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition; +import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Stream; +import org.apache.ignite.Ignite; +import org.apache.ignite.internal.app.IgniteImpl; +import org.apache.ignite.internal.raft.Peer; +import org.apache.ignite.internal.raft.service.RaftGroupService; +import org.apache.ignite.internal.schema.configuration.TablesConfiguration; +import org.apache.ignite.internal.storage.index.IndexStorage; +import org.apache.ignite.internal.table.InternalTable; +import org.apache.ignite.internal.table.TableImpl; +import org.apache.ignite.lang.IgniteStringFormatter; +import org.apache.ignite.table.Table; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Integration test of index building. + */ +public class ItBuildIndexTest extends ClusterPerClassIntegrationTest { + private static final String TABLE_NAME = "test_table"; + + private static final String INDEX_NAME = "test_index"; + + @AfterEach + void tearDown() { + sql("DROP TABLE IF EXISTS " + TABLE_NAME); + } + + @ParameterizedTest + @MethodSource("replicas") + @Disabled("https://issues.apache.org/jira/browse/IGNITE-19085";) + void testBuildIndexOnStableTopology(int replicas) throws Exception { + sql(IgniteStringFormatter.format( + "CREATE TABLE {} (i0 INTEGER PRIMARY KEY, i1 INTEGER) WITH replicas={}, partitions={}", + TABLE_NAME, replicas, 2 + )); + + sql(IgniteStringFormatter.format( + "INSERT INTO {} VALUES {}", + TABLE_NAME, toValuesString(List.of(1, 1), List.of(2, 2), List.of(3, 3), List.of(4, 4), List.of(5, 5)) + )); + + sql(IgniteStringFormatter.format("CREATE INDEX {} ON {} (i1)", INDEX_NAME, TABLE_NAME)); + + // FIXME: IGNITE-18733 + waitForIndex(INDEX_NAME); + + waitForIndexBuild(TABLE_NAME, INDEX_NAME); + + assertQuery(IgniteStringFormatter.format("SELECT * FROM {} WHERE i1 > 0", TABLE_NAME)) + .matches(containsIndexScan("PUBLIC", TABLE_NAME.toUpperCase(), INDEX_NAME.toUpperCase())) + .returns(1, 1) + .returns(2, 2) + .returns(3, 3) + .returns(4, 4) + .returns(5, 5) + .check(); + } + + private static int[] replicas() { + // FIXME: IGNITE-19086 Fix NullPointerException on insertALl Review Comment: ```suggestion // FIXME: IGNITE-19086 Fix NullPointerException on insertAll ``` ########## modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/BuildIndexCommand.java: ########## @@ -0,0 +1,50 @@ +/* + * 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 { + /** + * Return ID of table partition. + */ + TablePartitionIdMessage tablePartitionId(); + + /** + * Returns index ID. + */ + UUID indexId(); + + /** + * Returns row IDs for which to build indexes. + */ + List<UUID> rowIds(); + + /** + * Returns {@code true} if index building for the partition has completed. Review Comment: I'd rephrase that, not "completed", but that this batch is the last ########## 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. Review Comment: I wonder, why do we have this check in every single method, instead of making it once in "onWrite". Any ideas? ########## modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableMessageGroup.java: ########## @@ -154,5 +155,8 @@ interface Commands { /** Message type for {@link TablePartitionIdMessage}. */ short TABLE_PARTITION_ID = 61; + + /** Message type for {@link BuildIndexCommand}. */ + short BUILD_INDEX = 62; Review Comment: I guess you didn't notice the pattern. Here commands are 40 and higher. 60+ is for random stuff, like TablePartitionId pojos. Please make it 44 and move it higher. ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java: ########## @@ -385,6 +399,8 @@ private CompletableFuture<?> createIndexLocally(long causalityToken, UUID tableI index.descriptor().columns().toArray(STRING_EMPTY_ARRAY) ); + indexBuilder.startIndexBuild(tableIndexView, table); Review Comment: Why do you call this method before methods like "table.registerHashIndex"? Are these operations independent? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java: ########## @@ -385,6 +399,8 @@ private CompletableFuture<?> createIndexLocally(long causalityToken, UUID tableI index.descriptor().columns().toArray(STRING_EMPTY_ARRAY) ); + indexBuilder.startIndexBuild(tableIndexView, table); Review Comment: Maybe you should start this operation somewhere else, but I don't know where exactly. There's a `fireEvent(IndexEvent.CREATE...` later in the code, but you start build even before that firing. Seems dangerous ########## 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 agree with Alexander, why writing the same code twice? ########## modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/HashIndex.java: ########## @@ -39,9 +39,12 @@ class HashIndex { private final ConcurrentMap<Integer, RocksDbHashIndexStorage> storages = new ConcurrentHashMap<>(); - HashIndex(ColumnFamily indexCf, HashIndexDescriptor descriptor) { + private final RocksDbMetaStorage metaStorage; + + HashIndex(ColumnFamily indexCf, HashIndexDescriptor descriptor, RocksDbMetaStorage metaStorage) { this.indexCf = indexCf; this.descriptor = descriptor; + this.metaStorage = metaStorage; Review Comment: I would prefer different name, this one looks too similar to the name of other, completely unrelated, component ########## modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/PartitionDataHelper.java: ########## @@ -138,14 +142,24 @@ void putRowId(ByteBuffer keyBuffer, RowId rowId) { assert rowId.partitionId() == partitionId : rowId; assert keyBuffer.order() == KEY_BYTE_ORDER; - keyBuffer.putLong(normalize(rowId.mostSignificantBits())); - keyBuffer.putLong(normalize(rowId.leastSignificantBits())); + putUuid(keyBuffer, rowId.uuid()); + } + + static void putUuid(ByteBuffer buffer, UUID uuid) { Review Comment: Maybe a "rowIdUuid", because there's also "normalize", which may or may not be necessary for general-purpose UUIDs ########## 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)); Review Comment: Batch is encapsulated somewhere inside, right? ########## modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java: ########## @@ -497,4 +511,34 @@ public void startCleanup() { public void finishCleanup() { state.compareAndSet(StorageState.CLEANUP, StorageState.RUNNABLE); } + + @Override + public @Nullable RowId getLastBuildRowId() { + return busy(() -> { + throwExceptionIfStorageInProgressOfRebalance(state.get(), this::createStorageInfo); + + try { + UUID lastBuildRowIdUuid = indexMetaTree.findOne(new IndexMetaKey(indexDescriptor().id())).lastBuildRowIdUuid(); Review Comment: Looks copy-pasted, maybe we should have a common super-class for sorted and hash indexes? -- 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]
