tkalkirill commented on code in PR #13366:
URL: https://github.com/apache/ignite/pull/13366#discussion_r3713264856
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java:
##########
@@ -581,6 +598,228 @@ private FieldsQueryCursor<List<?>>
executeDdl(RootQuery<Row> qry, DdlPlan plan)
}
}
+ /**
+ * Executes a {@code SELECT ... FOR UPDATE} plan.
+ *
+ * <ol>
+ * <li>Validates that the current transaction is PESSIMISTIC.</li>
+ * <li>Runs the inner SELECT with hidden key, value, and version columns
and materialises all rows.</li>
+ * <li>Builds cache entries from the hidden columns.</li>
+ * <li>Creates a savepoint, acquires pessimistic locks via {@code
lockTxEntries()},
+ * and releases the savepoint on success (or rolls back on
failure).</li>
+ * <li>Repeats the SELECT and lock attempt after a concurrent version
change while the deadline permits.</li>
+ * <li>Returns a cursor with only the user-visible columns (the appended
_KEY is stripped).</li>
+ * </ol>
+ */
+ private FieldsQueryCursor<List<?>> executeForUpdate(RootQuery<Row> qry,
SelectForUpdatePlan plan) {
+ GridNearTxLocal userTx = Commons.queryTransaction(qry.context(),
ctx.cache().context());
+
+ if (userTx == null || !userTx.pessimistic())
+ throw new IgniteSQLException(
+
IgniteResource.INSTANCE.selectForUpdateRequiresPessimisticTx().str(),
+ IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
+
+ // waitSeconds: null = use tx remaining time (0), 0 = NOWAIT (-1),
positive = ms.
+ Long waitSeconds = plan.waitSeconds();
+ long waitMs;
+
+ if (waitSeconds == null)
+ waitMs = 0L;
+ else if (waitSeconds == 0L)
+ waitMs = -1L;
+ else
+ waitMs = waitSeconds * 1000L;
+
+ // Zero means that retries are limited only by the transaction or
query timeout.
+ long deadline = waitMs > 0
+ ? U.currentTimeMillis() + waitMs
+ : waitMs < 0 ? U.currentTimeMillis() : 0L;
+
+ RootQuery<Row> selectQry = qry;
+
+ while (true) {
+ FieldsQueryCursor<List<?>> cursor = tryExecuteForUpdate(selectQry,
plan, userTx, waitMs, deadline);
+
+ if (cursor != null)
+ return cursor;
+
+ if (deadline != 0 && U.currentTimeMillis() >= deadline) {
+ throw new IgniteSQLException(
+ IgniteResource.INSTANCE.selectForUpdateLockFailed().str(),
+ IgniteQueryErrorCode.CONCURRENT_UPDATE);
+ }
+
+ // The previous query has already been closed after
materialisation, so retry with a fresh root query.
+ selectQry = qry.retryQuery();
+ qryReg.register(selectQry);
+ }
+ }
+
+ /**
+ * Executes the inner SELECT once and tries to lock the selected row
versions.
+ *
+ * @return Result cursor when locking succeeds, or {@code null} when the
SELECT must be executed again.
+ */
+ @Nullable private FieldsQueryCursor<List<?>> tryExecuteForUpdate(
Review Comment:
The method is a bit hard to read; could you split the different parts into
separate methods?
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java:
##########
@@ -584,6 +613,14 @@ private boolean isSystemFieldName(String alias) {
return super.deriveType(scope, expr);
}
+ /** */
+ private void validateVersionColumnDmlTarget(SqlIdentifier id) {
Review Comment:
I don't think other system fields should be modifiable during an update
either. Could you check that? If so, we need to fix it for all of them
preferably in a separate ticket, ideally one addressed before this one, to
avoid bloating the changes here.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/CacheTableDescriptorImpl.java:
##########
@@ -278,6 +294,9 @@ else if (affFields.isEmpty())
@Override public boolean isUpdateAllowed(RelOptTable tbl, int colIdx) {
final CacheColumnDescriptor desc = descriptors[colIdx];
+ if (QueryUtils.VER_FIELD_NAME.equals(desc.name()))
Review Comment:
Maybe we could introduce a boolean field or something similar to avoid
checking strings?
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java:
##########
@@ -247,6 +272,7 @@ private void validateTableModify(SqlNode table) {
super.validateSelect(select, targetRowType);
}
+
Review Comment:
```suggestion
```
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java:
##########
@@ -136,16 +136,32 @@ public IgniteSqlValidator(
nullType = typeFactory.createSqlType(SqlTypeName.NULL);
}
+
Review Comment:
```suggestion
```
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java:
##########
@@ -581,6 +598,228 @@ private FieldsQueryCursor<List<?>>
executeDdl(RootQuery<Row> qry, DdlPlan plan)
}
}
+ /**
+ * Executes a {@code SELECT ... FOR UPDATE} plan.
+ *
+ * <ol>
+ * <li>Validates that the current transaction is PESSIMISTIC.</li>
+ * <li>Runs the inner SELECT with hidden key, value, and version columns
and materialises all rows.</li>
+ * <li>Builds cache entries from the hidden columns.</li>
+ * <li>Creates a savepoint, acquires pessimistic locks via {@code
lockTxEntries()},
+ * and releases the savepoint on success (or rolls back on
failure).</li>
+ * <li>Repeats the SELECT and lock attempt after a concurrent version
change while the deadline permits.</li>
+ * <li>Returns a cursor with only the user-visible columns (the appended
_KEY is stripped).</li>
+ * </ol>
+ */
+ private FieldsQueryCursor<List<?>> executeForUpdate(RootQuery<Row> qry,
SelectForUpdatePlan plan) {
+ GridNearTxLocal userTx = Commons.queryTransaction(qry.context(),
ctx.cache().context());
+
+ if (userTx == null || !userTx.pessimistic())
+ throw new IgniteSQLException(
+
IgniteResource.INSTANCE.selectForUpdateRequiresPessimisticTx().str(),
+ IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
+
+ // waitSeconds: null = use tx remaining time (0), 0 = NOWAIT (-1),
positive = ms.
+ Long waitSeconds = plan.waitSeconds();
+ long waitMs;
Review Comment:
Let's move this code into a separate method.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java:
##########
@@ -198,10 +214,19 @@ private void validateTableModify(SqlNode table) {
SqlIdentifier alias = call.getAlias() != null ? call.getAlias() :
new SqlIdentifier(deriveAlias(targetTable, 0), SqlParserPos.ZERO);
-
table.unwrap(IgniteTable.class).descriptor().selectForUpdateRowType((IgniteTypeFactory)typeFactory)
- .getFieldNames().stream()
- .map(name -> alias.plus(name, SqlParserPos.ZERO))
- .forEach(selectList::add);
+ RelDataType updateRowType =
table.unwrap(IgniteTable.class).descriptor()
+ .selectForUpdateRowType(typeFactory());
+
+ for (RelDataTypeField field : updateRowType.getFieldList()) {
+ if (QueryUtils.VER_FIELD_NAME.equals(field.getName())) {
+ selectList.add(SqlValidatorUtil.addAlias(
Review Comment:
And what is this for?
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java:
##########
@@ -419,18 +445,26 @@ private SqlNode rewriteTableToQuery(SqlNode from) {
SelectScope scope,
boolean includeSysVars
) {
- if (!includeSysVars && exp.getKind() == SqlKind.IDENTIFIER &&
isSystemFieldName(deriveAlias(exp, 0))) {
- SqlQualified qualified = scope.fullyQualify((SqlIdentifier)exp);
+ if (!includeSysVars && exp.getKind() == SqlKind.IDENTIFIER) {
Review Comment:
I’m not quite sure I understand - what exactly has changed here?
In theory, things should remain as they are: system fields shouldn't be
included in `SELECT *`. However, if they are explicitly specified, everything
should be fine.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/QueryPlan.java:
##########
@@ -22,7 +22,7 @@
*/
public interface QueryPlan {
/** Query type */
- enum Type { QUERY, FRAGMENT, DML, DDL, EXPLAIN }
+ enum Type { QUERY, FRAGMENT, DML, DDL, EXPLAIN, FOR_UPDATE }
Review Comment:
Isn't SELECT FOR UPDATE a DML statement? Why does it appear as a separate
query type?
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SystemColumnsScanTest.java:
##########
@@ -0,0 +1,365 @@
+/*
+ * 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.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import javax.cache.CacheException;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.SqlConfiguration;
+import org.apache.ignite.configuration.TransactionConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import
org.apache.ignite.internal.processors.query.calcite.CalciteQueryProcessor;
+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.tracker.NoOpIoTracker;
+import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.NoOpMemoryTracker;
+import
org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
+import
org.apache.ignite.internal.processors.query.calcite.metadata.FragmentDescription;
+import
org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMapping;
+import
org.apache.ignite.internal.processors.query.calcite.prepare.BaseQueryContext;
+import
org.apache.ignite.internal.processors.query.calcite.prepare.MappingQueryContext;
+import
org.apache.ignite.internal.processors.query.calcite.schema.ColumnDescriptor;
+import
org.apache.ignite.internal.processors.query.calcite.schema.IgniteCacheTable;
+import org.apache.ignite.internal.processors.query.calcite.schema.IgniteIndex;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.junit.Test;
+
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+
+/** Tests system columns returned by direct table and index scans. */
+public class SystemColumnsScanTest extends GridCommonAbstractTest {
Review Comment:
It looks like you can inherit from
'org.apache.ignite.internal.processors.query.calcite.integration.AbstractBasicIntegrationTest'.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/CacheTableDescriptorImpl.java:
##########
@@ -442,14 +464,18 @@ private <Row> ModifyTuple mergeTuple(Row row,
List<String> updateColList, Execut
int rowColumnsCnt = hnd.columnCount(row);
- if (rowColumnsCnt == descriptors.length)
+ // An empty update column list unambiguously means there is no WHEN
MATCHED clause at all (a MERGE
+ // statement always has at least one WHEN clause), so the row can only
originate from the INSERT
+ // section. Note: the row width alone can't be used to detect this
case, since, depending on the
+ // number of updated columns, it may coincide with the width of a WHEN
MATCHED-only row.
+ if (updateColList.isEmpty())
return insertTuple(row, ectx); // Only WHEN NOT MATCHED clause in
MERGE.
else if (rowColumnsCnt == descriptors.length + updateColList.size())
return updateTuple(row, updateColList, 0, ectx); // Only WHEN
MATCHED clause in MERGE.
else {
// Both WHEN MATCHED and WHEN NOT MATCHED clauses in MERGE.
- assert rowColumnsCnt == descriptors.length * 2 +
updateColList.size() : "Unexpected columns count: " +
- rowColumnsCnt;
+ assert rowColumnsCnt == 2 * descriptors.length +
updateColList.size() : "Unexpected columns count: "
Review Comment:
It looks like you haven't changed anything here other than repositioning the
multiplication operator.
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SystemColumnsScanTest.java:
##########
@@ -0,0 +1,365 @@
+/*
+ * 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.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import javax.cache.CacheException;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.SqlConfiguration;
+import org.apache.ignite.configuration.TransactionConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import
org.apache.ignite.internal.processors.query.calcite.CalciteQueryProcessor;
+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.tracker.NoOpIoTracker;
+import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.NoOpMemoryTracker;
+import
org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
+import
org.apache.ignite.internal.processors.query.calcite.metadata.FragmentDescription;
+import
org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMapping;
+import
org.apache.ignite.internal.processors.query.calcite.prepare.BaseQueryContext;
+import
org.apache.ignite.internal.processors.query.calcite.prepare.MappingQueryContext;
+import
org.apache.ignite.internal.processors.query.calcite.schema.ColumnDescriptor;
+import
org.apache.ignite.internal.processors.query.calcite.schema.IgniteCacheTable;
+import org.apache.ignite.internal.processors.query.calcite.schema.IgniteIndex;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.junit.Test;
+
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+
+/** Tests system columns returned by direct table and index scans. */
+public class SystemColumnsScanTest extends GridCommonAbstractTest {
+ /** */
+ private IgniteEx node;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setSqlConfiguration(new
SqlConfiguration().setQueryEnginesConfiguration(
+ new CalciteQueryEngineConfiguration().setDefault(true)))
+ .setTransactionConfiguration(new
TransactionConfiguration().setTxAwareQueriesEnabled(true));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ node = startGrid(0);
+
+ awaitPartitionMapExchange();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ super.afterTest();
+ }
+
+ /** */
+ @Test
+ public void testTableScanReturnsSystemColumns() throws Exception {
+ createAndPopulatePersonTable();
+
+ IgniteCacheTable tbl = personTable();
+ ScanContext scanCtx = scanContext(tbl);
+
+ assertSystemColumns(tbl.scan(scanCtx.ectx, scanCtx.grp,
requiredSystemColumns(tbl)));
+ }
+
+ /** */
+ @Test
+ public void testIndexScanReturnsSystemColumns() throws Exception {
+ createAndPopulatePersonTable();
+
+ IgniteCacheTable tbl = personTable();
+ IgniteIndex idx = tbl.getIndex("AGE_IDX");
+
+ assertNotNull(idx);
+
+ ScanContext scanCtx = scanContext(tbl);
+
+ assertSystemColumns(idx.scan(scanCtx.ectx, scanCtx.grp, null,
requiredSystemColumns(tbl)));
+ }
+
+ /** Verifies that an explicit SQL query returns system columns. */
+ @Test
+ public void testExplicitSelectReturnsSystemColumns() throws Exception {
+ createAndPopulatePersonTable();
+
+ List<List<?>> rows = sql("SELECT _key, _val, _ver FROM Person");
+
+ assertEquals(30, rows.size());
+
+ for (List<?> row : rows) {
+ assertEquals(3, row.size());
+ assertTrue(row.get(0) instanceof Integer);
+ assertNotNull(row.get(1));
+ assertTrue("Unexpected _VER value: " + row.get(2), row.get(2)
instanceof GridCacheVersion);
+ }
+ }
+
+ /** */
+ @Test
+ public void testCannotCreateTableWithSystemColumnName() {
+ assertSystemColumnCreateForbidden("CREATE TABLE PersonVer (id INT
PRIMARY KEY, _ver INT)",
+ QueryUtils.VER_FIELD_NAME);
+ }
+
+ /** */
+ @Test
+ public void testCannotAddSystemColumnName() throws Exception {
+ createAndPopulatePersonTable();
+
+ assertSystemColumnAddForbidden("ALTER TABLE Person ADD COLUMN _ver
INT",
+ QueryUtils.VER_FIELD_NAME);
+ }
+
+ /** */
+ @Test
+ public void testSystemColumnsAreHiddenFromSelectStar() throws Exception {
Review Comment:
Column name validation is missing.
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AbstractBasicIntegrationTest.java:
##########
@@ -269,20 +271,34 @@ protected List<List<?>> sqlAsRoot(IgniteEx ignite, String
sql) throws Exception
/** */
protected List<List<?>> sql(IgniteEx ignite, String sql, Object... params)
{
- // {@code sql} can contain more than one query.
- List<FieldsQueryCursor<List<?>>> allCurs =
queryProcessor(ignite).query(queryContext(), "PUBLIC", sql, params);
+ Transaction tx = ignite.transactions().tx();
Review Comment:
I suggest not keeping these changes in this class but moving them to the
appropriate ones, or creating a subclass of `AbstractBasicIntegrationTest` and
defining this logic there.
--
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]
