ibessonov commented on code in PR #1800: URL: https://github.com/apache/ignite-3/pull/1800#discussion_r1153358132
########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,273 @@ +/* + * 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.concurrent.CompletableFuture.completedFuture; +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.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; +import org.jetbrains.annotations.Nullable; + +/** + * 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, null)); Review Comment: Please pass the real row id value instead of null. ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,273 @@ +/* + * 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.concurrent.CompletableFuture.completedFuture; +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.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; +import org.jetbrains.annotations.Nullable; + +/** + * 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, null)); + } + } + + /** + * 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; + + /** + * ID of the next row to build the index from the previous batch, {@code null} if it is the first row after the index was crated + * (both on a live node and after a restore). + */ + private final @Nullable RowId nextRowIdToBuiltFromPreviousBatch; Review Comment: to buil**d** Please check your grammar when it comes to the word "build", it's the third time I correct you on it ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,273 @@ +/* + * 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.concurrent.CompletableFuture.completedFuture; +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.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; +import org.jetbrains.annotations.Nullable; + +/** + * 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(); Review Comment: Do we want to configure the pool size? Probably ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,273 @@ +/* + * 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.concurrent.CompletableFuture.completedFuture; +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.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; +import org.jetbrains.annotations.Nullable; + +/** + * 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, null)); + } + } + + /** + * 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; + + /** + * ID of the next row to build the index from the previous batch, {@code null} if it is the first row after the index was crated Review Comment: ```suggestion * ID of the next row to build the index from the previous batch, {@code null} if it is the first row after the index was created ``` ########## modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableIndexStoragesSupplier.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; + +import java.util.Map; +import java.util.UUID; + +/** + * Supplier table index storages. + */ +public interface TableIndexStoragesSupplier { Review Comment: What's the reason of calling it a supplier? Design of this interface look peculiar, I guess it's temporary ########## modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItBuildIndexTest.java: ########## @@ -0,0 +1,143 @@ +/* + * 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 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.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.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Integration test of index building. + */ +public class ItBuildIndexTest extends ClusterPerClassIntegrationTest { + private static final String ZONE_NAME = "zone_table"; + + 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") + void testBuildIndexOnStableTopology(int replicas) throws Exception { + sql(IgniteStringFormatter.format("CREATE ZONE IF NOT EXISTS {} WITH REPLICAS={}, PARTITIONS={}", + ZONE_NAME, replicas, 2 + )); + + sql(IgniteStringFormatter.format( + "CREATE TABLE {} (i0 INTEGER PRIMARY KEY, i1 INTEGER) WITH PRIMARY_ZONE='{}'", + TABLE_NAME, ZONE_NAME.toUpperCase() + )); + + 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 Review Comment: Usually we use TODO ########## modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItBuildIndexTest.java: ########## @@ -0,0 +1,143 @@ +/* + * 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 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.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.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Integration test of index building. + */ +public class ItBuildIndexTest extends ClusterPerClassIntegrationTest { + private static final String ZONE_NAME = "zone_table"; + + 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") + void testBuildIndexOnStableTopology(int replicas) throws Exception { + sql(IgniteStringFormatter.format("CREATE ZONE IF NOT EXISTS {} WITH REPLICAS={}, PARTITIONS={}", + ZONE_NAME, replicas, 2 + )); + + sql(IgniteStringFormatter.format( + "CREATE TABLE {} (i0 INTEGER PRIMARY KEY, i1 INTEGER) WITH PRIMARY_ZONE='{}'", + TABLE_NAME, ZONE_NAME.toUpperCase() + )); + + 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: Same here. There are automated tools, targeted to TODO comments specifically ########## modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItBuildIndexTest.java: ########## @@ -0,0 +1,143 @@ +/* + * 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 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.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.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Integration test of index building. + */ +public class ItBuildIndexTest extends ClusterPerClassIntegrationTest { + private static final String ZONE_NAME = "zone_table"; + + 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") + void testBuildIndexOnStableTopology(int replicas) throws Exception { + sql(IgniteStringFormatter.format("CREATE ZONE IF NOT EXISTS {} WITH REPLICAS={}, PARTITIONS={}", Review Comment: Do we have a default zone? Maybe there's a chance of making this test simpler ########## modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java: ########## @@ -462,6 +464,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()); + + List<UUID> rowUuids0 = List.of(UUID.randomUUID()); + List<UUID> rowUuids1 = List.of(UUID.randomUUID()); + List<UUID> rowUuids2 = List.of(UUID.randomUUID()); + + InOrder inOrder = inOrder(partitionDataStorage, storageUpdateHandler); + + commandListener.handleBuildIndexCommand(createBuildIndexCommand(indexId, rowUuids0, false), 10, 1); + + inOrder.verify(partitionDataStorage).lastApplied(10, 1); + inOrder.verify(storageUpdateHandler).buildIndex(indexId, rowUuids0, false); + + commandListener.handleBuildIndexCommand(createBuildIndexCommand(indexId, rowUuids1, true), 20, 2); + + inOrder.verify(partitionDataStorage).lastApplied(20, 2); + inOrder.verify(storageUpdateHandler).buildIndex(indexId, rowUuids1, true); + + commandListener.handleBuildIndexCommand(createBuildIndexCommand(indexId, rowUuids2, false), 5, 1); Review Comment: Why to you pass smaller values here? Shouldn't such command be ignored? ########## modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java: ########## @@ -462,6 +464,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()); + + List<UUID> rowUuids0 = List.of(UUID.randomUUID()); Review Comment: Same here, order here is random. I guess it may not be important in this particular test, right? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,273 @@ +/* + * 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.concurrent.CompletableFuture.completedFuture; +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.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; +import org.jetbrains.annotations.Nullable; + +/** + * 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, null)); + } + } + + /** + * 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; + + /** + * ID of the next row to build the index from the previous batch, {@code null} if it is the first row after the index was crated + * (both on a live node and after a restore). + */ + private final @Nullable RowId nextRowIdToBuiltFromPreviousBatch; + + private BuildIndexTask( + TableImpl table, + TableIndexView tableIndexView, + int partitionId, + @Nullable RowId nextRowIdToBuiltFromPreviousBatch + ) { + this.table = table; + this.tableIndexView = tableIndexView; + this.partitionId = partitionId; + this.nextRowIdToBuiltFromPreviousBatch = nextRowIdToBuiltFromPreviousBatch; + } + + @Override + public void run() { + if (!busyLock.enterBusy()) { + return; + } + + try { + // At the time of creating the index, we should have already waited for the table to be created and its raft of clients + // (services) to start for all partitions, so there should be no errors. + RaftGroupService raftGroupService = table.internalTable().partitionRaftGroupService(partitionId); + + raftGroupService + // We do not check the presence of nodes in the topology on purpose, so as not to get into races on + // rebalancing, it will be more convenient and reliable for us to wait for a stable topology with a chosen + // leader. + .refreshAndGetLeaderWithTerm() + .thenComposeAsync(leaderWithTerm -> { + if (!busyLock.enterBusy()) { + return completedFuture(null); + } + + try { + // At this point, we have a stable topology, each node of which has already applied all local updates. + if (!localNodeConsistentId().equals(leaderWithTerm.leader().consistentId())) { + // TODO: IGNITE-19053 Must handle the change of leader + // TODO: IGNITE-19053 Add a test to change the leader even at the start of the task + return completedFuture(null); + } + + List<RowId> batchRowIds = collectRowIdBatch(); + + RowId nextRowId = getNextRowIdForNextBatch(batchRowIds); + + boolean finish = batchRowIds.size() < BUILD_INDEX_ROW_ID_BATCH_SIZE || nextRowId == null; + + // TODO: IGNITE-19053 Must handle the change of leader + return raftGroupService.run(createBuildIndexCommand(batchRowIds, finish)) + .thenRun(() -> { + if (!finish) { + assert nextRowId != null : createCommonTableIndexInfo(); + + buildIndexExecutor.submit( + new BuildIndexTask(table, tableIndexView, partitionId, nextRowId) Review Comment: Can we pass `this` as the argument? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,273 @@ +/* + * 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.concurrent.CompletableFuture.completedFuture; +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.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; +import org.jetbrains.annotations.Nullable; + +/** + * 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++) { Review Comment: We should only start rebuild process for indexes that exist ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,273 @@ +/* + * 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.concurrent.CompletableFuture.completedFuture; +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.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; +import org.jetbrains.annotations.Nullable; + +/** + * 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, null)); + } + } + + /** + * 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; + + /** + * ID of the next row to build the index from the previous batch, {@code null} if it is the first row after the index was crated + * (both on a live node and after a restore). + */ + private final @Nullable RowId nextRowIdToBuiltFromPreviousBatch; + + private BuildIndexTask( + TableImpl table, + TableIndexView tableIndexView, + int partitionId, + @Nullable RowId nextRowIdToBuiltFromPreviousBatch + ) { + this.table = table; + this.tableIndexView = tableIndexView; + this.partitionId = partitionId; + this.nextRowIdToBuiltFromPreviousBatch = nextRowIdToBuiltFromPreviousBatch; + } + + @Override + public void run() { + if (!busyLock.enterBusy()) { + return; + } + + try { + // At the time of creating the index, we should have already waited for the table to be created and its raft of clients + // (services) to start for all partitions, so there should be no errors. + RaftGroupService raftGroupService = table.internalTable().partitionRaftGroupService(partitionId); + + raftGroupService + // We do not check the presence of nodes in the topology on purpose, so as not to get into races on + // rebalancing, it will be more convenient and reliable for us to wait for a stable topology with a chosen + // leader. + .refreshAndGetLeaderWithTerm() + .thenComposeAsync(leaderWithTerm -> { + if (!busyLock.enterBusy()) { + return completedFuture(null); + } + + try { + // At this point, we have a stable topology, each node of which has already applied all local updates. + if (!localNodeConsistentId().equals(leaderWithTerm.leader().consistentId())) { + // TODO: IGNITE-19053 Must handle the change of leader + // TODO: IGNITE-19053 Add a test to change the leader even at the start of the task + return completedFuture(null); + } + + List<RowId> batchRowIds = collectRowIdBatch(); + + RowId nextRowId = getNextRowIdForNextBatch(batchRowIds); + + boolean finish = batchRowIds.size() < BUILD_INDEX_ROW_ID_BATCH_SIZE || nextRowId == null; + + // TODO: IGNITE-19053 Must handle the change of leader + return raftGroupService.run(createBuildIndexCommand(batchRowIds, finish)) + .thenRun(() -> { + if (!finish) { + assert nextRowId != null : createCommonTableIndexInfo(); + + buildIndexExecutor.submit( + new BuildIndexTask(table, tableIndexView, partitionId, nextRowId) + ); + } + }); + } finally { + busyLock.leaveBusy(); + } + }, buildIndexExecutor) + .whenComplete((unused, throwable) -> { + if (throwable != null) { + LOG.error("Index build error: [{}]", throwable, createCommonTableIndexInfo()); + } + }); + } 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 lastBuiltRowId, int batchSize) { + MvPartitionStorage mvPartition = table.internalTable().storage().getMvPartition(partitionId); Review Comment: Why can't we store partition instance in the task, instead of retrieving it every time? ########## modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableIndexStoragesSupplier.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; + +import java.util.Map; +import java.util.UUID; + +/** + * Supplier table index storages. + */ +public interface TableIndexStoragesSupplier { + /** + * Returns indexes by their ID. + * + * <p>Waits for the primary key index and all other registered indexes to be created. + */ + Map<UUID, TableSchemaAwareIndexStorage> get(); + + /** + * Adds a wait to create an index, if not already created, on a subsequent call to {@link #get()}. Review Comment: This comment is hard to understand, please rephrase it ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,273 @@ +/* + * 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.concurrent.CompletableFuture.completedFuture; +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.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; +import org.jetbrains.annotations.Nullable; + +/** + * 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( Review Comment: Can you use some method of `java.util.concurrent.Executors`? What's "30 seconds" here? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java: ########## @@ -0,0 +1,273 @@ +/* + * 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.concurrent.CompletableFuture.completedFuture; +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.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; +import org.jetbrains.annotations.Nullable; + +/** + * 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, null)); + } + } + + /** + * 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; + + /** + * ID of the next row to build the index from the previous batch, {@code null} if it is the first row after the index was crated + * (both on a live node and after a restore). + */ + private final @Nullable RowId nextRowIdToBuiltFromPreviousBatch; + + private BuildIndexTask( + TableImpl table, + TableIndexView tableIndexView, + int partitionId, + @Nullable RowId nextRowIdToBuiltFromPreviousBatch + ) { + this.table = table; + this.tableIndexView = tableIndexView; + this.partitionId = partitionId; + this.nextRowIdToBuiltFromPreviousBatch = nextRowIdToBuiltFromPreviousBatch; + } + + @Override + public void run() { + if (!busyLock.enterBusy()) { + return; + } + + try { + // At the time of creating the index, we should have already waited for the table to be created and its raft of clients + // (services) to start for all partitions, so there should be no errors. + RaftGroupService raftGroupService = table.internalTable().partitionRaftGroupService(partitionId); + + raftGroupService + // We do not check the presence of nodes in the topology on purpose, so as not to get into races on + // rebalancing, it will be more convenient and reliable for us to wait for a stable topology with a chosen + // leader. + .refreshAndGetLeaderWithTerm() + .thenComposeAsync(leaderWithTerm -> { + if (!busyLock.enterBusy()) { + return completedFuture(null); + } + + try { + // At this point, we have a stable topology, each node of which has already applied all local updates. + if (!localNodeConsistentId().equals(leaderWithTerm.leader().consistentId())) { + // TODO: IGNITE-19053 Must handle the change of leader + // TODO: IGNITE-19053 Add a test to change the leader even at the start of the task + return completedFuture(null); + } + + List<RowId> batchRowIds = collectRowIdBatch(); + + RowId nextRowId = getNextRowIdForNextBatch(batchRowIds); + + boolean finish = batchRowIds.size() < BUILD_INDEX_ROW_ID_BATCH_SIZE || nextRowId == null; + + // TODO: IGNITE-19053 Must handle the change of leader + return raftGroupService.run(createBuildIndexCommand(batchRowIds, finish)) + .thenRun(() -> { + if (!finish) { + assert nextRowId != null : createCommonTableIndexInfo(); + + buildIndexExecutor.submit( + new BuildIndexTask(table, tableIndexView, partitionId, nextRowId) + ); + } + }); + } finally { + busyLock.leaveBusy(); + } + }, buildIndexExecutor) + .whenComplete((unused, throwable) -> { + if (throwable != null) { + LOG.error("Index build error: [{}]", throwable, createCommonTableIndexInfo()); + } + }); + } 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 lastBuiltRowId, 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 && lastBuiltRowId != null; i++) { + lastBuiltRowId = mvPartition.closestRowId(lastBuiltRowId); + + if (lastBuiltRowId == null) { + break; + } + + batch.add(lastBuiltRowId); + + lastBuiltRowId = lastBuiltRowId.increment(); + } + + 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() { + return clusterService.topologyService().localMember().name(); + } + + private @Nullable RowId getNextRowIdForNextBatch(List<RowId> batch) { + return batch.isEmpty() ? null : batch.get(batch.size() - 1).increment(); + } + + private @Nullable List<RowId> collectRowIdBatch() { + RowId nextRowIdToBuilt; + + if (nextRowIdToBuiltFromPreviousBatch == null) { Review Comment: As I already mentioned, I don't want to see this null-check, I'd prefer this value to be non-null ########## modules/table/src/test/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandlerTest.java: ########## @@ -0,0 +1,132 @@ +/* + * 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; + +import static java.util.Arrays.asList; +import static java.util.Collections.singletonList; +import static java.util.stream.Collectors.toList; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.UUID; +import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension; +import org.apache.ignite.internal.configuration.testframework.InjectConfiguration; +import org.apache.ignite.internal.hlc.HybridClock; +import org.apache.ignite.internal.hlc.HybridClockImpl; +import org.apache.ignite.internal.schema.BinaryRow; +import org.apache.ignite.internal.schema.configuration.storage.DataStorageConfiguration; +import org.apache.ignite.internal.storage.ReadResult; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.internal.storage.index.IndexStorage; +import org.apache.ignite.internal.table.distributed.raft.PartitionDataStorage; +import org.apache.ignite.internal.util.Cursor; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +/** + * For {@link StorageUpdateHandler} testing. + */ +@ExtendWith(ConfigurationExtension.class) +public class StorageUpdateHandlerTest { + private static final int PARTITION_ID = 0; + + @InjectConfiguration + private DataStorageConfiguration dataStorageConfig; + + private final HybridClock clock = new HybridClockImpl(); + + @Test + void testBuildIndex() { + PartitionDataStorage partitionStorage = mock(PartitionDataStorage.class); + + TableSchemaAwareIndexStorage indexStorage = createIndexStorage(); + + UUID indexId = UUID.randomUUID(); + + TableIndexStoragesSupplier indexes = mock(TableIndexStoragesSupplier.class); + + when(indexes.get()).thenReturn(Map.of(indexId, indexStorage)); + + StorageUpdateHandler storageUpdateHandler = createStorageUpdateHandler(partitionStorage, indexes); + + RowId rowId0 = new RowId(PARTITION_ID, UUID.randomUUID()); + RowId rowId1 = new RowId(PARTITION_ID, UUID.randomUUID()); + + List<BinaryRow> rowVersions0 = asList(mock(BinaryRow.class), null); + List<BinaryRow> rowVersions1 = asList(mock(BinaryRow.class), null); + + setRowVersions(partitionStorage, Map.of(rowId0.uuid(), rowVersions0, rowId1.uuid(), rowVersions1)); + + storageUpdateHandler.buildIndex(indexId, List.of(rowId0.uuid(), rowId1.uuid()), false); Review Comment: It's interesting that you pass rowIds in potentially incorrect order. You shouldn't do it, because it's not realistic and it may spontaneously stop working in the future if we add some assertion somewhere. ########## modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java: ########## @@ -422,4 +425,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); Review Comment: Why is this not the last line in the closure, but the first one instead? It should represent the end of the operation. Order of things matter in the code, it helps understanding what's going on. In the future, we may want to split the "buildIndex" loop into several "runConssitently" calls, for example. In that case, wrong operations order could lead to actual bugs. ########## modules/table/src/test/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandlerTest.java: ########## @@ -0,0 +1,132 @@ +/* + * 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; + +import static java.util.Arrays.asList; +import static java.util.Collections.singletonList; +import static java.util.stream.Collectors.toList; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.UUID; +import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension; +import org.apache.ignite.internal.configuration.testframework.InjectConfiguration; +import org.apache.ignite.internal.hlc.HybridClock; +import org.apache.ignite.internal.hlc.HybridClockImpl; +import org.apache.ignite.internal.schema.BinaryRow; +import org.apache.ignite.internal.schema.configuration.storage.DataStorageConfiguration; +import org.apache.ignite.internal.storage.ReadResult; +import org.apache.ignite.internal.storage.RowId; +import org.apache.ignite.internal.storage.index.IndexStorage; +import org.apache.ignite.internal.table.distributed.raft.PartitionDataStorage; +import org.apache.ignite.internal.util.Cursor; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +/** + * For {@link StorageUpdateHandler} testing. + */ +@ExtendWith(ConfigurationExtension.class) +public class StorageUpdateHandlerTest { + private static final int PARTITION_ID = 0; + + @InjectConfiguration + private DataStorageConfiguration dataStorageConfig; + + private final HybridClock clock = new HybridClockImpl(); + + @Test + void testBuildIndex() { + PartitionDataStorage partitionStorage = mock(PartitionDataStorage.class); + + TableSchemaAwareIndexStorage indexStorage = createIndexStorage(); + + UUID indexId = UUID.randomUUID(); + + TableIndexStoragesSupplier indexes = mock(TableIndexStoragesSupplier.class); + + when(indexes.get()).thenReturn(Map.of(indexId, indexStorage)); + + StorageUpdateHandler storageUpdateHandler = createStorageUpdateHandler(partitionStorage, indexes); + + RowId rowId0 = new RowId(PARTITION_ID, UUID.randomUUID()); Review Comment: There's a constructor that doesn't require explicit UUID as a parameter. Maybe you should use it ########## modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItBuildIndexTest.java: ########## @@ -0,0 +1,143 @@ +/* + * 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 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.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.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Integration test of index building. + */ +public class ItBuildIndexTest extends ClusterPerClassIntegrationTest { + private static final String ZONE_NAME = "zone_table"; + + 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") + void testBuildIndexOnStableTopology(int replicas) throws Exception { + sql(IgniteStringFormatter.format("CREATE ZONE IF NOT EXISTS {} WITH REPLICAS={}, PARTITIONS={}", + ZONE_NAME, replicas, 2 + )); + + sql(IgniteStringFormatter.format( + "CREATE TABLE {} (i0 INTEGER PRIMARY KEY, i1 INTEGER) WITH PRIMARY_ZONE='{}'", + TABLE_NAME, ZONE_NAME.toUpperCase() + )); + + 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: By the way, what's the reason for the NPE? Can it be fixed now? -- 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]
