IGNITE-28864 Add technical columns to Calcite engin SQL scan

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java

@@ -70,6 +70,7 @@
 import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
 import org.apache.ignite.internal.processors.query.calcite.schema.IgniteCacheTable;
 import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
+import org.apache.ignite.internal.processors.query.calcite.schema.TechnicalColumns;
 import org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlDecimalLiteral;
 import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
 import org.apache.ignite.internal.processors.query.calcite.type.OtherType;
@@ -246,6 +247,29 @@ private void validateTableModify(SqlNode table) {
         super.validateSelect(select, targetRowType);
     }
 
+    /** {@inheritDoc} */
+    @Override protected void validateOrderList(SqlSelect select) {
+        super.validateOrderList(select);
+
+        SqlNodeList orderList = select.getOrderList();
+
+        if (orderList == null)
+            return;
+
+        for (SqlNode orderItem : orderList) {
+            SqlNode node = orderItem;
+
+            // Unwrap DESC / NULLS FIRST / NULLS LAST wrappers to get the actual expression.
+            while (node.getKind() == SqlKind.DESCENDING
+                || node.getKind() == SqlKind.NULLS_FIRST
+                || node.getKind() == SqlKind.NULLS_LAST) {
+                node = ((SqlCall)node).operand(0);
+            }
+
+            validateTechnicalColumnAccess(node);
+        }
+    }
+
     /** {@inheritDoc} */
     @Override protected void validateNamespace(SqlValidatorNamespace namespace, RelDataType targetRowType) {
         SqlValidatorTable table = namespace.getTable();
@@ -411,18 +435,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) {
+            String alias = deriveAlias(exp, 0);
 
-            if (qualified.namespace == null)
+            // Technical columns are never exposed via SELECT *.
+            if (TechnicalColumns.isTechnicalFieldNameIgnoreCase(alias))
                 return;
 
-            if (qualified.namespace.getTable() != null) {
-                // If child is table and has only system fields, expand star to these fields.
-                // Otherwise, expand star to non-system fields only.
-                for (RelDataTypeField fld : qualified.namespace.getRowType().getFieldList()) {
-                    if (!isSystemField(fld))
-                        return;
+            if (isSystemFieldName(alias)) {
+                SqlQualified qualified = scope.fullyQualify((SqlIdentifier)exp);
+
+                if (qualified.namespace == null)
+                    return;
+
+                if (qualified.namespace.getTable() != null) {
+                    // If child is table and has only system fields, expand star to these fields.
+                    // Otherwise, expand star to non-system fields only.
+                    for (RelDataTypeField fld : qualified.namespace.getRowType().getFieldList()) {
+                        if (!isSystemField(fld))
+                            return;
+                    }
                 }
             }
         }
@@ -554,11 +586,14 @@ private IgniteTypeFactory typeFactory() {
     /** */
     private boolean isSystemFieldName(String alias) {
         return QueryUtils.KEY_FIELD_NAME.equalsIgnoreCase(alias)
-            || QueryUtils.VAL_FIELD_NAME.equalsIgnoreCase(alias);
+            || QueryUtils.VAL_FIELD_NAME.equalsIgnoreCase(alias)
+            || TechnicalColumns.isTechnicalFieldNameIgnoreCase(alias);
     }
 
     /** {@inheritDoc} */
     @Override public RelDataType deriveType(SqlValidatorScope scope, SqlNode expr) {
+        validateTechnicalColumnAccess(expr);
+
         if (expr instanceof SqlDynamicParam) {
             RelDataType type = deriveDynamicParameterType((SqlDynamicParam)expr, nullType);
 
@@ -569,6 +604,21 @@ private boolean isSystemFieldName(String alias) {
         return super.deriveType(scope, expr);
     }
 
+    /** */
+    private void validateTechnicalColumnAccess(SqlNode expr) {
+        if (!(expr instanceof SqlIdentifier))
+            return;
+
+        SqlIdentifier id = (SqlIdentifier)expr;
+
+        String fieldName = id.names.get(id.names.size() - 1);
+
+        if (id.isStar() || !TechnicalColumns.isTechnicalFieldNameIgnoreCase(fieldName))
+            return;
+
+        throw newValidationError(id, IgniteResource.INSTANCE.cannotAccessTechnicalColumn(id.toString()));
+    }
+
     /** @return A derived type or {@code null} if unable to determine. */
     @Nullable private RelDataType deriveDynamicParameterType(SqlDynamicParam node, RelDataType nullValType) {
         RelDataType type = getValidatedNodeTypeIfKnown(node);

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/CacheTableDescriptorImpl.java

@@ -53,6 +53,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
 import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
 import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.processors.query.GridQueryProperty;
 import org.apache.ignite.internal.processors.query.GridQueryTypeDescriptor;
 import org.apache.ignite.internal.processors.query.QueryUtils;
@@ -111,6 +112,9 @@ public class CacheTableDescriptorImpl extends NullInitializerExpressionFactory
     /** */
     private final ImmutableBitSet insertFields;
 
+    /** Count of non-technical columns used in the UPDATE source SELECT. */
+    private final int updateRowFieldCnt;
+
     /** */
     private RelDataType tableRowType;
 
@@ -123,7 +127,7 @@ public CacheTableDescriptorImpl(GridCacheContextInfo<?, ?> cacheInfo, GridQueryT
 
         Set<String> fields = this.typeDesc.fields().keySet();
 
-        List<CacheColumnDescriptor> descriptors = new ArrayList<>(fields.size() + 2);
+        List<CacheColumnDescriptor> descriptors = new ArrayList<>(fields.size() + 4);
 
         // A _key/_val field is virtual in case there is an alias or a property(es) mapped to the _key/_val field.
         BitSet virtualFields = new BitSet();
@@ -177,6 +181,28 @@ else if (Objects.equals(field, typeDesc.valueFieldAlias())) {
             }
         }
 
+        virtualFields.set(descriptors.size());
+        descriptors.add(new TechnicalDescriptor(TechnicalColumns.VER_FIELD_NAME, GridCacheVersion.class, descriptors.size()) {
+            @Override public Object value(ExecutionContext<?> ectx, GridCacheContext<?, ?> cctx, CacheDataRow src)
+                throws IgniteCheckedException {
+                GridCacheVersion ver = src.version();
+
+                if (ver != null)
+                    return ver;
+
+                CacheDataRow row = cctx.offheap().read(cctx, src.key());
+
+                return row == null ? null : row.version();
+            }
+        });
+
+        virtualFields.set(descriptors.size());
+        descriptors.add(new TechnicalDescriptor(TechnicalColumns.SRC_FIELD_NAME, Integer.class, descriptors.size()) {
+            @Override public Object value(ExecutionContext<?> ectx, GridCacheContext<?, ?> cctx, CacheDataRow src) {
+                return cctx.cacheId();
+            }
+        });
+
         Map<String, CacheColumnDescriptor> descriptorsMap = U.newHashMap(descriptors.size());
         for (CacheColumnDescriptor descriptor : descriptors)
             descriptorsMap.put(descriptor.name(), descriptor);
@@ -212,6 +238,9 @@ else if (!F.isEmpty(typeDesc.primaryKeyFields())) {
         this.affFields = ImmutableIntList.copyOf(affFields);
         this.descriptors = descriptors.toArray(DUMMY);
         this.descriptorsMap = descriptorsMap;
+        this.updateRowFieldCnt = (int)descriptors.stream()
+            .filter(d -> !TechnicalColumns.isTechnicalFieldNameIgnoreCase(d.name()))
+            .count();
 
         virtualFields.flip(0, descriptors.size());
         insertFields = ImmutableBitSet.fromBitSet(virtualFields);
@@ -222,6 +251,18 @@ else if (!F.isEmpty(typeDesc.primaryKeyFields())) {
         return rowType(factory, insertFields);
     }
 
+    /** {@inheritDoc} */
+    @Override public RelDataType selectForUpdateRowType(IgniteTypeFactory factory) {
+        RelDataTypeFactory.Builder b = new RelDataTypeFactory.Builder(factory);
+
+        for (CacheColumnDescriptor desc : descriptors) {
+            if (!TechnicalColumns.isTechnicalFieldNameIgnoreCase(desc.name()))
+                b.add(desc.name(), desc.logicalType(factory));
+        }
+
+        return b.build();
+    }
+
     /** {@inheritDoc} */
     @Override public GridCacheContext cacheContext() {
         return cacheInfo.cacheContext();
@@ -278,6 +319,9 @@ else if (affFields.isEmpty())
     @Override public boolean isUpdateAllowed(RelOptTable tbl, int colIdx) {
         final CacheColumnDescriptor desc = descriptors[colIdx];
 
+        if (isTechnicalColumn(desc.name()))
+            return false;
+
         return !desc.key() && (desc.field() || QueryUtils.isSqlType(desc.storageType()));
     }
 
@@ -289,6 +333,11 @@ else if (affFields.isEmpty())
         return super.generationStrategy(tbl, colIdx);
     }
 
+    /** */
+    private static boolean isTechnicalColumn(String fieldName) {
+        return TechnicalColumns.isTechnicalFieldName(fieldName);
+    }
+
     /** {@inheritDoc} */
     @Override public RexNode newColumnDefaultValue(RelOptTable tbl, int colIdx, InitializerContext ctx) {
         final ColumnDescriptor desc = descriptors[colIdx];
@@ -412,7 +461,8 @@ private <Row> ModifyTuple updateTuple(Row row, List<String> updateColList, int o
         Object key = Objects.requireNonNull(hnd.get(offset + QueryUtils.KEY_COL, row));
         Object val = clone(Objects.requireNonNull(hnd.get(offset + QueryUtils.VAL_COL, row)));
 
-        offset += descriptorsMap.size();
+        // New values start after the source fields; compute dynamically to handle both UPDATE and MERGE.
+        offset = hnd.columnCount(row) - updateColList.size();
 
         for (int i = 0; i < updateColList.size(); i++) {
             final CacheColumnDescriptor desc = Objects.requireNonNull(descriptorsMap.get(updateColList.get(i)));
@@ -442,14 +492,19 @@ 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())
+        else if (rowColumnsCnt == updateRowFieldCnt + 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;
+            // INSERT section has all fields (insertRowType); UPDATE section excludes technical columns.
+            assert rowColumnsCnt == descriptors.length + updateRowFieldCnt + updateColList.size() :
+                "Unexpected columns count: " + rowColumnsCnt;
 
             int updateOffset = descriptors.length; // Offset of fields for update statement.
 
@@ -805,6 +860,79 @@ private FieldDescriptor(GridQueryProperty desc, int fieldIdx) {
         }
     }
 
+    /** */
+    private abstract static class TechnicalDescriptor implements CacheColumnDescriptor {
+        /** */
+        private final String name;
+
+        /** */
+        private final Class<?> storageType;
+
+        /** */
+        private final int fieldIdx;
+
+        /** */
+        private volatile RelDataType logicalType;
+
+        /** */
+        private TechnicalDescriptor(String name, Class<?> storageType, int fieldIdx) {
+            this.name = name;
+            this.storageType = storageType;
+            this.fieldIdx = fieldIdx;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean field() {
+            return false;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean key() {
+            return false;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean hasDefaultValue() {
+            return false;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Object defaultValue() {
+            throw new AssertionError();
+        }
+
+        /** {@inheritDoc} */
+        @Override public String name() {
+            return name;
+        }
+
+        /** {@inheritDoc} */
+        @Override public int fieldIndex() {
+            return fieldIdx;
+        }
+
+        /** {@inheritDoc} */
+        @Override public RelDataType logicalType(IgniteTypeFactory f) {
+            if (logicalType == null) {
+                logicalType = storageType == GridCacheVersion.class
+                    ? f.createTypeWithNullability(f.createJavaType(storageType), true)
+                    : TypeUtils.sqlType(f, storageType, PRECISION_NOT_SPECIFIED, SCALE_NOT_SPECIFIED, true);
+            }
+
+            return logicalType;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Class<?> storageType() {
+            return storageType;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void set(Object dst, Object val) {
+            throw new AssertionError();
+        }
+    }
+
     /** {@inheritDoc} */
     @Override public GridQueryTypeDescriptor typeDescription() {
         return typeDesc;

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/TechnicalColumns.java

@@ -0,0 +1,42 @@
+/*
+ * 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.schema;
+
+/** Technical cache table columns. */
+public final class TechnicalColumns {
+    /** Field name for row version. */
+    public static final String VER_FIELD_NAME = "_VER";
+
+    /** Field name for row source. */
+    public static final String SRC_FIELD_NAME = "_SRC";
+
+    /** */
+    private TechnicalColumns() {
+        // No-op.
+    }
+
+    /** */
+    public static boolean isTechnicalFieldName(String fieldName) {
+        return VER_FIELD_NAME.equals(fieldName) || SRC_FIELD_NAME.equals(fieldName);
+    }
+
+    /** */
+    public static boolean isTechnicalFieldNameIgnoreCase(String fieldName) {
+        return VER_FIELD_NAME.equalsIgnoreCase(fieldName) || SRC_FIELD_NAME.equalsIgnoreCase(fieldName);
+    }
+}

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/IgniteResource.java

@@ -35,6 +35,10 @@ public interface IgniteResource {
     @Resources.BaseMessage("Cannot update field \"{0}\". You cannot update key, key fields or val field in case the val is a complex type.")
     Resources.ExInst<SqlValidatorException> cannotUpdateField(String field);
 
+    /** */
+    @Resources.BaseMessage("Cannot access technical column \"{0}\".")
+    Resources.ExInst<SqlValidatorException> cannotAccessTechnicalColumn(String field);
+
     /** */
     @Resources.BaseMessage("Illegal aggregate function. {0} is unsupported at the moment.")
     Resources.ExInst<SqlValidatorException> unsupportedAggregationFunction(String a0);