korlov42 commented on a change in pull request #9671:
URL: https://github.com/apache/ignite/pull/9671#discussion_r803023650
##########
File path:
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteIndexScan.java
##########
@@ -87,9 +78,10 @@ public IgniteIndexScan(
@Nullable List<RexNode> proj,
@Nullable RexNode cond,
@Nullable IndexConditions idxCond,
- @Nullable ImmutableBitSet requiredCols
+ @Nullable ImmutableBitSet requiredCols,
+ @Nullable RelCollation collation
Review comment:
I'm not sure if null should be allowed here.
##########
File path:
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
##########
@@ -298,13 +301,106 @@ public LogicalRelImplementor(
Supplier<Row> upper = upperCond == null ? null :
expressionFactory.rowSource(upperCond);
Function<Row, Row> prj = projects == null ? null :
expressionFactory.project(projects, rowType);
+ ColocationGroup grp = ctx.group(rel.sourceId());
+
IgniteIndex idx = tbl.getIndex(rel.indexName());
- ColocationGroup group = ctx.group(rel.sourceId());
+ if (idx != null && !tbl.isIndexRebuildInProgress()) {
+ Iterable<Row> rowsIter = idx.scan(ctx, grp, filters, lower, upper,
prj, requiredColumns);
+
+ return new ScanNode<>(ctx, rowType, rowsIter);
+ }
+ else {
+ // Index was invalidated after planning, workaround through
table-scan -> sort -> index spool.
+ // If there are correlates in filter or project, spool node is
required to provide ability to rewind input.
+ // Sort node is required if output should be sorted or if spool
node required (to provide search by
+ // index conditions).
+ // Additionally, project node is required in case of spool
inserted, since spool requires unmodified
+ // original input for filtering by index conditions.
+ boolean filterHasCorrelation = condition != null &&
RexUtils.hasCorrelation(condition);
+ boolean projectHasCorrelation = projects != null &&
RexUtils.hasCorrelation(projects);
+ boolean spoolNodeRequired = projectHasCorrelation ||
filterHasCorrelation;
+ boolean projNodeRequired = projects != null && spoolNodeRequired;
+
+ Iterable<Row> rowsIter = tbl.scan(
+ ctx,
+ grp,
+ filterHasCorrelation ? null : filters,
+ projNodeRequired ? null : prj,
+ requiredColumns
+ );
- Iterable<Row> rowsIter = idx.scan(ctx, group, filters, lower, upper,
prj, requiredColumns);
+ Node<Row> node = new ScanNode<>(ctx, rowType, rowsIter);
- return new ScanNode<>(ctx, rowType, rowsIter);
+ RelCollation collation = rel.collation();
+
+ if ((!spoolNodeRequired && projects != null) || requiredColumns !=
null) {
+ collation =
collation.apply(LogicalScanConverterRule.createMapping(
+ spoolNodeRequired ? null : projects,
+ requiredColumns,
+ tbl.getRowType(typeFactory).getFieldCount()
+ ));
+ }
+
+ boolean sortNodeRequired =
!collation.getFieldCollations().isEmpty();
+
+ if (sortNodeRequired) {
+ if (!spoolNodeRequired && projects != null)
+ rowType = rel.getRowType();
Review comment:
SortNode should have the same row type as its input
##########
File path:
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/IndexRebuildPlannerTest.java
##########
@@ -0,0 +1,367 @@
+/*
+ * 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.processors.query.calcite.planner;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.RelCollations;
+import org.apache.calcite.rel.core.CorrelationId;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.calcite.util.mapping.Mappings;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import
org.apache.ignite.internal.processors.query.calcite.exec.ArrayRowHandler;
+import
org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import
org.apache.ignite.internal.processors.query.calcite.exec.LogicalRelImplementor;
+import
org.apache.ignite.internal.processors.query.calcite.exec.rel.IndexSpoolNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.rel.Node;
+import
org.apache.ignite.internal.processors.query.calcite.exec.rel.ProjectNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.rel.SortNode;
+import
org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
+import org.apache.ignite.internal.processors.query.calcite.schema.IgniteSchema;
+import
org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
+import
org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
+import org.apache.ignite.internal.processors.query.calcite.trait.TraitUtils;
+import org.apache.ignite.internal.processors.query.calcite.util.RexUtils;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.junit.Test;
+
+/**
+ * Planner test for index rebuild.
+ */
+public class IndexRebuildPlannerTest extends AbstractPlannerTest {
+ /** */
+ private IgniteSchema publicSchema;
+
+ /** */
+ private IndexRebuildAwareTable tbl;
+
+ /** {@inheritDoc} */
+ @Override public void setup() {
+ super.setup();
+
+ RelDataTypeFactory.Builder b = new
RelDataTypeFactory.Builder(TYPE_FACTORY);
+
+ b.add("_KEY", TYPE_FACTORY.createJavaType(Object.class));
+ b.add("_VAL", TYPE_FACTORY.createJavaType(Object.class));
+ b.add("ID", TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER));
+ b.add("VAL", TYPE_FACTORY.createSqlType(SqlTypeName.VARCHAR));
+
+ tbl = new IndexRebuildAwareTable("TBL", b.build(), 100d);
+
+ tbl.addIndex(RelCollations.of(ImmutableIntList.of(2)), "IDX");
+
+ publicSchema = createSchema(tbl);
+ }
+
+ /** */
+ @Test
+ public void testIndexRebuild() throws Exception {
+ String sql = "SELECT * FROM TBL WHERE id = 0";
+
+ assertPlan(sql, publicSchema, isInstanceOf(IgniteIndexScan.class));
+
+ tbl.markIndexRebuildInProgress(true);
+
+ assertPlan(sql, publicSchema, isInstanceOf(IgniteTableScan.class));
+
+ tbl.markIndexRebuildInProgress(false);
+
+ assertPlan(sql, publicSchema, isInstanceOf(IgniteIndexScan.class));
+ }
+
+ /** */
+ @Test
+ public void testConcurrentIndexRebuildStateChange() throws Exception {
+ String sql = "SELECT * FROM TBL WHERE id = 0";
+
+ AtomicBoolean stop = new AtomicBoolean();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runAsync(() -> {
+ while (!stop.get()) {
+ tbl.markIndexRebuildInProgress(true);
+ tbl.markIndexRebuildInProgress(false);
+ }
+ });
+
+ try {
+ for (int i = 0; i < 1000; i++) {
+ IgniteRel rel = physicalPlan(sql, publicSchema);
+
+ assertTrue(rel instanceof IgniteTableScan || rel instanceof
IgniteIndexScan);
+ }
+ }
+ finally {
+ stop.set(true);
+ }
+
+ fut.get();
+ }
+
+ /** */
+ @Test
+ public void testIndexScanRewriter() throws Exception {
Review comment:
I would prefer to move this test to a separate test class
(LogicalRelImplementorSelfTest?)
##########
File path:
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/IndexRebuildIntegrationTest.java
##########
@@ -0,0 +1,316 @@
+/*
+ * 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.processors.query.calcite.integration;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.cluster.ClusterState;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.cache.query.index.IndexProcessor;
+import org.apache.ignite.internal.managers.indexing.IndexesRebuildTask;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.query.calcite.QueryChecker;
+import
org.apache.ignite.internal.processors.query.schema.IndexRebuildCancelToken;
+import
org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.hamcrest.CoreMatchers;
+import org.junit.Test;
+
+/**
+ * Test index rebuild.
+ */
+public class IndexRebuildIntegrationTest extends AbstractBasicIntegrationTest {
+ /** Index rebuild init latch. */
+ private static CountDownLatch initLatch;
+
+ /** Index rebuild start latch. */
+ private static CountDownLatch startLatch;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IndexProcessor.idxRebuildCls = BlockingIndexesRebuildTask.class;
+
+ return
super.getConfiguration(igniteInstanceName).setDataStorageConfiguration(
+ new DataStorageConfiguration().setDefaultDataRegionConfiguration(
+ new DataRegionConfiguration().setPersistenceEnabled(true)));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int nodeCount() {
+ return 2;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ initLatch = null;
+ startLatch = null;
+
+ super.beforeTest();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() {
+ // Override super method to skip caches destroy.
+ cleanQueryPlanCache();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ cleanPersistenceDir();
+
+ super.beforeTestsStarted();
+
+ client.cluster().state(ClusterState.ACTIVE);
+
+ executeSql("CREATE TABLE tbl (id INT PRIMARY KEY, val VARCHAR, val2
VARCHAR) WITH CACHE_NAME=\"test\"");
+ executeSql("CREATE INDEX idx_id_val ON tbl (id DESC, val)");
+ executeSql("CREATE INDEX idx_id_val2 ON tbl (id, val2 DESC)");
+ executeSql("CREATE INDEX idx_val ON tbl (val DESC)");
+
+ for (int i = 0; i < 100; i++)
+ executeSql("INSERT INTO tbl VALUES (?, ?, ?)", i, "val" + i, "val"
+ i);
+
+ executeSql("CREATE TABLE tbl2 (id INT PRIMARY KEY, val VARCHAR)");
+
+ for (int i = 0; i < 100; i++)
+ executeSql("INSERT INTO tbl2 VALUES (?, ?)", i, "val" + i);
+ }
+
+ /** */
+ @Test
+ public void testRebuildOnInitiatorNode() throws Exception {
+ String sql = "SELECT * FROM tbl WHERE id = 0 AND val='val0'";
+
+ QueryChecker validChecker = assertQuery(grid(0), sql)
+ .matches(QueryChecker.containsIndexScan("PUBLIC", "TBL",
"IDX_ID_VAL"))
Review comment:
We need to stop checking query plans in integration tests. There is
already a test IndexRebuildPlannerTest which verifies a planner behaviour
during index rebuilding. For integration tests the only thing that matters is
the result set.
--
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]
