Txsql

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java

@@ -114,6 +114,8 @@
 import org.apache.ignite.internal.util.typedef.F;
 
 import static org.apache.calcite.rel.RelDistribution.Type.HASH_DISTRIBUTED;
+import static org.apache.calcite.sql.SqlKind.IS_DISTINCT_FROM;
+import static org.apache.calcite.sql.SqlKind.IS_NOT_DISTINCT_FROM;
 import static org.apache.ignite.internal.processors.query.calcite.util.TypeUtils.combinedRowType;
 
 /**
@@ -298,7 +300,8 @@ public LogicalRelImplementor(
 
         Comparator<Row> comp = expressionFactory.comparator(
             rel.leftCollation().getFieldCollations().subList(0, pairsCnt),
-            rel.rightCollation().getFieldCollations().subList(0, pairsCnt)
+            rel.rightCollation().getFieldCollations().subList(0, pairsCnt),
+            rel.getCondition().getKind() == IS_NOT_DISTINCT_FROM || rel.getCondition().getKind() == IS_DISTINCT_FROM
         );
 
         Node<Row> node = MergeJoinNode.create(ctx, outType, leftType, rightType, joinType, comp, hasExchange(rel));

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactory.java

@@ -60,9 +60,11 @@ Supplier<List<AccumulatorWrapper<Row>>> accumulatorsFactory(
      *
      * @param left Collations of left row.
      * @param right Collations of right row.
+     * @param nullsEqual If {@code true}, nulls are considered equal. Usually, NULL <> NULL in SQL. So, the value should
+     *                   be {@code false}. Except cases with IS DISTINCT / IS NOT DISTINCT.
      * @return Rows comparator.
      */
-    Comparator<Row> comparator(List<RelFieldCollation> left, List<RelFieldCollation> right);
+    Comparator<Row> comparator(List<RelFieldCollation> left, List<RelFieldCollation> right, boolean nullsEqual);
 
     /**
      * Creates a Filter predicate.

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactoryImpl.java

@@ -193,7 +193,7 @@ else if (c2 != HIGHEST_VALUE)
     }
 
     /** {@inheritDoc} */
-    @Override public Comparator<Row> comparator(List<RelFieldCollation> left, List<RelFieldCollation> right) {
+    @Override public Comparator<Row> comparator(List<RelFieldCollation> left, List<RelFieldCollation> right, boolean nullsEqual) {
         if (F.isEmpty(left) || F.isEmpty(right) || left.size() != right.size())
             throw new IllegalArgumentException("Both inputs should be non-empty and have the same size: left="
                 + (left != null ? left.size() : "null") + ", right=" + (right != null ? right.size() : "null"));
@@ -238,7 +238,8 @@ else if (c2 != HIGHEST_VALUE)
                 }
 
                 // If compared rows contain NULLs, they shouldn't be treated as equals, since NULL <> NULL in SQL.
-                return hasNulls ? 1 : 0;
+                // Except cases with IS DISTINCT / IS NOT DISTINCT.
+                return hasNulls && !nullsEqual ? 1 : 0;
             }
         };
     }

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/CorrelatedNestedLoopJoinNode.java

@@ -127,7 +127,7 @@ public CorrelatedNestedLoopJoinNode(ExecutionContext<Row> ctx, RelDataType rowTy
     /** {@inheritDoc} */
     @Override public void request(int rowsCnt) throws Exception {
         assert !F.isEmpty(sources()) && sources().size() == 2;
-        assert rowsCnt > 0 && requested == 0;
+        assert rowsCnt > 0 && requested == 0 : "rowsCnt=" + rowsCnt + ", requested=" + requested;
 
         checkState();
 
@@ -430,14 +430,21 @@ private void join() throws Exception {
 
                 int notMatchedIdx = leftMatched.nextClearBit(0);
 
-                while (requested > 0 && notMatchedIdx < leftInBuf.size()) {
-                    downstream().push(handler.concat(leftInBuf.get(notMatchedIdx), rightEmptyRow));
+                state = State.IN_LOOP;
 
-                    requested--;
+                try {
+                    while (requested > 0 && notMatchedIdx < leftInBuf.size()) {
+                        requested--;
+
+                        downstream().push(handler.concat(leftInBuf.get(notMatchedIdx), rightEmptyRow));
 
-                    leftMatched.set(notMatchedIdx);
+                        leftMatched.set(notMatchedIdx);
 
-                    notMatchedIdx = leftMatched.nextClearBit(notMatchedIdx + 1);
+                        notMatchedIdx = leftMatched.nextClearBit(notMatchedIdx + 1);
+                    }
+                }
+                finally {
+                    state = State.IDLE;
                 }
 
                 if (requested == 0 && notMatchedIdx < leftInBuf.size())

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

@@ -18,12 +18,14 @@
 package org.apache.ignite.internal.processors.query.calcite.prepare;
 
 import java.nio.charset.Charset;
+import java.util.Arrays;
 import org.apache.calcite.adapter.java.JavaTypeFactory;
 import org.apache.calcite.rel.type.DynamicRecordType;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rel.type.RelDataTypeFactoryImpl;
 import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
 import org.apache.calcite.sql.SqlCollation;
 import org.apache.calcite.sql.SqlDataTypeSpec;
 import org.apache.calcite.sql.SqlIdentifier;
@@ -33,6 +35,7 @@
 import org.apache.calcite.sql.SqlUserDefinedTypeNameSpec;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.type.SqlTypeFamily;
 import org.apache.calcite.sql.type.SqlTypeUtil;
 import org.apache.calcite.sql.validate.SqlValidator;
 import org.apache.calcite.sql.validate.SqlValidatorScope;
@@ -55,6 +58,41 @@ public IgniteTypeCoercion(RelDataTypeFactory typeFactory, SqlValidator validator
         super(typeFactory, validator);
     }
 
+    /** {@inheritDoc} **/
+    @Override public boolean binaryComparisonCoercion(SqlCallBinding binding) {
+        // Although it is not reflected in the docs, this method is also invoked for MAX, MIN (and other similar operators)
+        // by ComparableOperandTypeChecker. When that is the case, fallback to default rules.
+        SqlCall call = binding.getCall();
+
+        if (binding.getOperandCount() != 2 || !SqlKind.BINARY_COMPARISON.contains(call.getKind()))
+            return super.binaryComparisonCoercion(binding);
+
+        SqlValidatorScope scope = binding.getScope();
+
+        RelDataType leftType = validator.deriveType(scope, call.operand(0));
+        RelDataType rightType = validator.deriveType(scope, call.operand(1));
+
+        if (leftType.equals(rightType))
+            return super.binaryComparisonCoercion(binding);
+        else {
+            // Find the least restrictive type among the operand types
+            // and coerce the operands to that type if such type exists.
+            //
+            // An example of a least restrictive type from the javadoc for RelDataTypeFactory::leastRestrictive:
+            // leastRestrictive(INT, NUMERIC(3, 2)) could be NUMERIC(12, 2)
+            //
+            // A least restrictive type between types of different type families does not exist -
+            // the method returns null (See SqlTypeFactoryImpl::leastRestrictive).
+            //
+            RelDataType targetType = factory.leastRestrictive(Arrays.asList(leftType, rightType));
+
+            if (targetType == null || targetType.getFamily() == SqlTypeFamily.ANY)
+                return super.binaryComparisonCoercion(binding);
+            else
+                return coerceOperandsType(scope, call, targetType);
+        }
+    }
+
     /** {@inheritDoc} */
     @Override protected boolean coerceOperandType(
         SqlValidatorScope scope,

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/SetOpConverterRule.java

@@ -38,6 +38,7 @@
 import org.apache.ignite.internal.processors.query.calcite.rel.set.IgniteReduceIntersect;
 import org.apache.ignite.internal.processors.query.calcite.rel.set.IgniteReduceMinus;
 import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
 
 /**
  * Set op (MINUS, INTERSECT) converter rule.
@@ -77,6 +78,8 @@ private abstract static class ColocatedSetOpConverterRule<T extends SetOp> exten
             RelTraitSet outTrait = cluster.traitSetOf(IgniteConvention.INSTANCE).replace(IgniteDistributions.single());
             List<RelNode> inputs = Util.transform(setOp.getInputs(), rel -> convert(rel, inTrait));
 
+            inputs = Commons.castToLeastRestrictiveIfRequired(inputs, cluster, inTrait);
+
             return createNode(cluster, outTrait, inputs, setOp.all);
         }
     }
@@ -131,6 +134,8 @@ abstract PhysicalNode createReduceNode(RelOptCluster cluster, RelTraitSet traits
             RelTraitSet outTrait = cluster.traitSetOf(IgniteConvention.INSTANCE);
             List<RelNode> inputs = Util.transform(setOp.getInputs(), rel -> convert(rel, inTrait));
 
+            inputs = Commons.castToLeastRestrictiveIfRequired(inputs, cluster, inTrait);
+
             RelNode map = createMapNode(cluster, outTrait, inputs, setOp.all);
 
             return createReduceNode(

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/UnionConverterRule.java

@@ -47,12 +47,14 @@ public UnionConverterRule(Config cfg) {
 
     /** {@inheritDoc} */
     @Override public void onMatch(RelOptRuleCall call) {
-        final LogicalUnion union = call.rel(0);
+        LogicalUnion union = call.rel(0);
 
         RelOptCluster cluster = union.getCluster();
         RelTraitSet traits = cluster.traitSetOf(IgniteConvention.INSTANCE);
         List<RelNode> inputs = Commons.transform(union.getInputs(), input -> convert(input, traits));
 
+        inputs = Commons.castToLeastRestrictiveIfRequired(inputs, cluster, traits);
+
         RelNode res = new IgniteUnionAll(cluster, traits, inputs);
 
         if (!union.all) {

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/type/IgniteTypeFactory.java

@@ -27,6 +27,7 @@
 import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.Period;
+import java.util.EnumMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -42,12 +43,23 @@
 import org.apache.calcite.sql.type.BasicSqlType;
 import org.apache.calcite.sql.type.IntervalSqlType;
 import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.ignite.IgniteException;
 import org.apache.ignite.internal.util.typedef.F;
 
 /**
  * Ignite type factory.
  */
 public class IgniteTypeFactory extends JavaTypeFactoryImpl {
+    /** */
+    private static final EnumMap<SqlTypeName, String> UNSUPPORTED_TYPES = new EnumMap<>(SqlTypeName.class);
+
+    static {
+        UNSUPPORTED_TYPES.put(SqlTypeName.TIME_TZ, "TIME WITH TIME ZONE");
+        UNSUPPORTED_TYPES.put(SqlTypeName.TIMESTAMP_TZ, "TIMESTAMP WITH TIME ZONE");
+        UNSUPPORTED_TYPES.put(SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE, "TIMESTAMP WITH LOCAL TIME ZONE");
+        UNSUPPORTED_TYPES.put(SqlTypeName.TIME_WITH_LOCAL_TIME_ZONE, "TIME WITH LOCAL TIME ZONE");
+    }
+
     /** Interval qualifier to create year-month interval types. */
     private static final SqlIntervalQualifier INTERVAL_QUALIFIER_YEAR_MONTH = new SqlIntervalQualifier(TimeUnit.YEAR,
         TimeUnit.MONTH, SqlParserPos.ZERO);
@@ -133,9 +145,8 @@ else if (type instanceof BasicSqlType || type instanceof IntervalSqlType) {
                     return Enum.class;
                 case ANY:
                 case OTHER:
-                    return Object.class;
                 case NULL:
-                    return Void.class;
+                    return Object.class;
                 default:
                     break;
             }
@@ -345,6 +356,35 @@ private boolean allEquals(List<RelDataType> types) {
         return true;
     }
 
+    /** {@inheritDoc} */
+    @Override public RelDataType createSqlType(SqlTypeName typeName) {
+        checkUnsupportedType(typeName);
+
+        return super.createSqlType(typeName);
+    }
+
+    /** {@inheritDoc} */
+    @Override public RelDataType createSqlType(SqlTypeName typeName, int precision) {
+        checkUnsupportedType(typeName);
+
+        return super.createSqlType(typeName, precision);
+    }
+
+    /** {@inheritDoc} */
+    @Override public RelDataType createSqlType(SqlTypeName typeName, int precision, int scale) {
+        checkUnsupportedType(typeName);
+
+        return super.createSqlType(typeName, precision, scale);
+    }
+
+    /** */
+    private static void checkUnsupportedType(SqlTypeName typeName) {
+        String unsupportedTypeName = UNSUPPORTED_TYPES.get(typeName);
+
+        if (unsupportedTypeName != null)
+            throw new IgniteException("Type '" + unsupportedTypeName + "' is not supported.");
+    }
+
     /** {@inheritDoc} */
     @Override public RelDataType createUnknownType() {
         // TODO workaround for https://issues.apache.org/jira/browse/CALCITE-5297

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

@@ -38,10 +38,16 @@
 import org.apache.calcite.plan.Context;
 import org.apache.calcite.plan.Contexts;
 import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
 import org.apache.calcite.rel.RelNode;
+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.RexNode;
 import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
+import org.apache.calcite.sql.type.SqlTypeUtil;
 import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.calcite.util.ImmutableIntList;
 import org.apache.calcite.util.SourceStringReader;
@@ -64,6 +70,7 @@
 import org.apache.ignite.internal.processors.query.calcite.exec.exp.ExpressionFactoryImpl;
 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.rel.IgniteProject;
 import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.A;
@@ -143,6 +150,61 @@ public static <T> List<T> intersect(Set<T> set, List<T> list) {
             .collect(Collectors.toList());
     }
 
+    /**
+     * Finds the least restrictive type of the inputs and adds a cast projection if required.
+     *
+     * @param inputs Inputs to try to cast.
+     * @param cluster Cluster.
+     * @param traits Traits.
+     * @return Converted inputs.
+     */
+    public static List<RelNode> castToLeastRestrictiveIfRequired(List<RelNode> inputs, RelOptCluster cluster, RelTraitSet traits) {
+        List<RelDataType> inputRowTypes = inputs.stream().map(RelNode::getRowType).collect(Collectors.toList());
+
+        // Output type of a set operator is equal to leastRestrictive(inputTypes) (see SetOp::deriveRowType).
+        RelDataTypeFactory typeFactory = cluster.getTypeFactory();
+
+        RelDataType leastRestrictive = typeFactory.leastRestrictive(inputRowTypes);
+
+        if (leastRestrictive == null)
+            throw new IllegalStateException("Cannot find least restrictive type for arguments to set op: " + inputRowTypes);
+
+        // If input's type does not match the result type, then add a cast projection for non-matching fields.
+        RexBuilder rexBuilder = cluster.getRexBuilder();
+        List<RelNode> newInputs = new ArrayList<>(inputs.size());
+
+        for (RelNode input : inputs) {
+            RelDataType inputRowType = input.getRowType();
+
+            // It is always safe to convert from [T1 nullable, T2 not nullable] to [T1 nullable, T2 nullable] and
+            // the least restrictive type does exactly that.
+            if (SqlTypeUtil.equalAsStructSansNullability(typeFactory, leastRestrictive, inputRowType, null)) {
+                newInputs.add(input);
+
+                continue;
+            }
+
+            List<RexNode> expressions = new ArrayList<>(inputRowType.getFieldCount());
+
+            for (int i = 0; i < leastRestrictive.getFieldCount(); i++) {
+                RelDataType fieldType = inputRowType.getFieldList().get(i).getType();
+
+                RelDataType outFieldType = leastRestrictive.getFieldList().get(i).getType();
+
+                RexNode ref = rexBuilder.makeInputRef(input, i);
+
+                if (fieldType.equals(outFieldType))
+                    expressions.add(ref);
+                else
+                    expressions.add(rexBuilder.makeCast(outFieldType, ref, true, false));
+            }
+
+            newInputs.add(new IgniteProject(cluster, traits, input, expressions, leastRestrictive));
+        }
+
+        return newInputs;
+    }
+
     /**
      * Returns a given list as a typed list.
      */

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

@@ -145,6 +145,11 @@ public static boolean needCast(RelDataTypeFactory factory, RelDataType fromType,
             return false;
         }
 
+        // Currently, RelDataTypeFactoryImpl#CLASS_FAMILIES doesn't consider the byte type as an integer.
+        if ((fromType.getSqlTypeName() == SqlTypeName.TINYINT && SqlTypeUtil.isIntType(toType))
+            || (toType.getSqlTypeName() == SqlTypeName.TINYINT && SqlTypeUtil.isIntType(fromType)))
+            return false;
+
         // Implicit type coercion does not handle nullability.
         if (SqlTypeUtil.equalSansNullability(factory, fromType, toType))
             return false;

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ExecutionTest.java

@@ -515,8 +515,11 @@ public void testCorrelatedNestedLoopJoin() {
 
                         join.register(Arrays.asList(left, right));
 
+                        FilterNode<Object[]> filter = new FilterNode<>(ctx, joinRowType, r -> true);
+                        filter.register(join);
+
                         RootNode<Object[]> root = new RootNode<>(ctx, joinRowType);
-                        root.register(join);
+                        root.register(filter);
 
                         int cnt = 0;
                         while (root.hasNext()) {

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AbstractBasicIntegrationTest.java

@@ -145,10 +145,7 @@ protected QueryChecker assertQuery(Ignite ignite, String qry) {
         };
     }
 
-    /**
-     * @deprecated Use {@link #sql(String, Object...)} instead.
-     */
-    @Deprecated
+    /** */
     protected List<List<?>> executeSql(String sql, Object... args) {
         return executeSql(client, sql, args);
     }