IGNITE-29045 SQL Calcite: Support Java temporal types in UDF and UDTF parameters and results

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

@@ -784,9 +784,7 @@ else if (rel instanceof Intersect)
 
         RelDataType rowType = rel.getRowType();
 
-        RowFactory<Row> rowFactory = ctx.rowHandler().factory(ctx.getTypeFactory(), rowType);
-
-        return new ScanNode<>(ctx, rowType, new TableFunctionScan<>(rowType, dataSupplier, rowFactory));
+        return new ScanNode<>(ctx, rowType, new TableFunctionScan<>(ctx, rowType, dataSupplier));
     }
 
     /** {@inheritDoc} */

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

@@ -17,16 +17,26 @@
 
 package org.apache.ignite.internal.processors.query.calcite.exec;
 
+import java.lang.reflect.Type;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.function.Supplier;
+import org.apache.calcite.linq4j.tree.Primitive;
+import org.apache.calcite.linq4j.tree.Types;
 import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.ignite.internal.processors.query.IgniteSQLException;
 import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler.RowFactory;
+import org.apache.ignite.internal.processors.query.calcite.type.OtherType;
+import org.apache.ignite.internal.processors.query.calcite.util.TypeUtils;
 import org.apache.ignite.internal.util.typedef.F;
+import org.jetbrains.annotations.Nullable;
 
 /** */
 public class TableFunctionScan<Row> implements Iterable<Row> {
+    /** */
+    private final ExecutionContext<Row> ctx;
+
     /** */
     private final RelDataType rowType;
 
@@ -38,13 +48,15 @@ public class TableFunctionScan<Row> implements Iterable<Row> {
 
     /** */
     public TableFunctionScan(
+        ExecutionContext<Row> ctx,
         RelDataType rowType,
-        Supplier<Iterable<?>> dataSupplier,
-        RowFactory<Row> rowFactory
+        Supplier<Iterable<?>> dataSupplier
     ) {
+        this.ctx = ctx;
         this.rowType = rowType;
         this.dataSupplier = dataSupplier;
-        this.rowFactory = rowFactory;
+
+        rowFactory = ctx.rowHandler().factory(ctx.getTypeFactory(), rowType);
     }
 
     /** {@inheritDoc} */
@@ -58,14 +70,35 @@ private Row convertToRow(Object rowContainer) {
             throw new IgniteSQLException("Unable to process table function data: row type is neither Collection or Object[].");
 
         Object[] rowArr = rowContainer.getClass() == Object[].class
-            ? (Object[])rowContainer
+            ? ((Object[])rowContainer).clone()
             : ((Collection<?>)rowContainer).toArray();
 
         if (rowArr.length != rowType.getFieldCount()) {
             throw new IgniteSQLException("Unable to process table function data: row length [" + rowArr.length
                 + "] doesn't match defined columns number [" + rowType.getFieldCount() + "].");
         }
 
+        for (int i = 0; i < rowArr.length; i++)
+            rowArr[i] = convertToInternal(rowArr[i], rowType.getFieldList().get(i).getType());
+
         return rowFactory.create(rowArr);
     }
+
+    /** */
+    private @Nullable Object convertToInternal(@Nullable Object val, RelDataType type) {
+        // Preserve objects for both Ignite's custom OTHER type and Calcite's SQL OTHER type.
+        if (val == null || type instanceof OtherType || type.getSqlTypeName() == SqlTypeName.OTHER)
+            return val;
+
+        Type storageType = ctx.getTypeFactory().getResultClass(type);
+
+        if (!TypeUtils.isConvertableType(storageType))
+            return val;
+
+        // SQL table functions can already return values in the internal representation.
+        if (Types.isAssignableFrom(Primitive.box(ctx.getTypeFactory().getJavaClass(type)), val.getClass()))
+            return val;
+
+        return TypeUtils.toInternal(ctx, val, storageType);
+    }
 }

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

@@ -38,6 +38,8 @@
 import org.apache.calcite.util.BuiltInMethod;
 import org.apache.calcite.util.Util;
 import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+import org.apache.ignite.internal.processors.query.calcite.util.TypeUtils;
+import org.jetbrains.annotations.Nullable;
 
 /** */
 public class ConverterUtils {
@@ -123,10 +125,18 @@ else if (targetType == java.sql.Timestamp.class) {
     /** */
     static List<Expression> fromInternal(Class<?>[] targetTypes,
         List<Expression> expressions) {
-        final List<Expression> list = new ArrayList<>();
+        return fromInternal(null, targetTypes, expressions);
+    }
+
+    /** Converts user-defined function arguments using the execution context when available. */
+    static List<Expression> fromInternal(@Nullable Expression root,
+        Class<?>[] targetTypes,
+        List<Expression> expressions
+    ) {
+        final List<Expression> list = new ArrayList<>(expressions.size());
         if (targetTypes.length == expressions.size()) {
             for (int i = 0; i < expressions.size(); i++)
-                list.add(fromInternal(expressions.get(i), targetTypes[i]));
+                list.add(fromInternal(root, expressions.get(i), targetTypes[i]));
         }
         else {
             int j = 0;
@@ -139,12 +149,32 @@ static List<Expression> fromInternal(Class<?>[] targetTypes,
                 else
                     type = targetTypes[j].getComponentType();
 
-                list.add(fromInternal(expressions.get(i), type));
+                list.add(fromInternal(root, expressions.get(i), type));
             }
         }
         return list;
     }
 
+    /** */
+    private static Expression fromInternal(@Nullable Expression root, Expression operand, Type targetType) {
+        // Preserve Calcite's calendar conversion for JDBC dates and timestamps.
+        Expression converted = fromInternal(operand, targetType);
+
+        if (root == null || converted != operand || !TypeUtils.isConvertableType(targetType))
+            return converted;
+
+        if (Types.isAssignableFrom(targetType, operand.getType()))
+            return operand;
+
+        if (Primitive.is(operand.getType()))
+            operand = Expressions.box(operand);
+
+        return Expressions.convert_(
+            Expressions.call(TypeUtils.class, "fromInternal", root, operand, Expressions.constant(targetType)),
+            targetType
+        );
+    }
+
     /** */
     private static Type toInternal(RelDataType type) {
         return toInternal(type, false);

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

@@ -0,0 +1,54 @@
+/*
+ * 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.exec.exp;
+
+import org.apache.calcite.adapter.java.JavaTypeFactory;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.schema.FunctionParameter;
+
+/** Function parameter that exposes its SQL type to validation. */
+class IgniteFunctionParameter implements FunctionParameter {
+    /** */
+    private final FunctionParameter delegate;
+
+    /** */
+    IgniteFunctionParameter(FunctionParameter delegate) {
+        this.delegate = delegate;
+    }
+
+    /** {@inheritDoc} */
+    @Override public int getOrdinal() {
+        return delegate.getOrdinal();
+    }
+
+    /** {@inheritDoc} */
+    @Override public String getName() {
+        return delegate.getName();
+    }
+
+    /** {@inheritDoc} */
+    @Override public RelDataType getType(RelDataTypeFactory typeFactory) {
+        // Normalize UDF metadata without losing Java types used to convert query results.
+        return ((JavaTypeFactory)typeFactory).toSql(delegate.getType(typeFactory));
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean isOptional() {
+        return delegate.isOptional();
+    }
+}

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

@@ -17,18 +17,32 @@
 package org.apache.ignite.internal.processors.query.calcite.exec.exp;
 
 import java.lang.reflect.Method;
+import java.util.List;
+import org.apache.calcite.schema.FunctionParameter;
 import org.apache.calcite.schema.impl.ReflectiveFunctionBase;
 
+import static java.util.stream.Collectors.toUnmodifiableList;
+
 /** A base for outer java-method functions. */
 abstract class IgniteReflectiveFunctionBase extends ReflectiveFunctionBase implements ImplementableFunction {
     /** */
     protected final CallImplementor implementor;
 
+    /** */
+    private final List<FunctionParameter> params;
+
     /** */
     protected IgniteReflectiveFunctionBase(Method method, CallImplementor implementor) {
         super(method);
 
         this.implementor = implementor;
+
+        params = super.getParameters().stream().map(IgniteFunctionParameter::new).collect(toUnmodifiableList());
+    }
+
+    /** {@inheritDoc} */
+    @Override public List<FunctionParameter> getParameters() {
+        return params;
     }
 
     /** {@inheritDoc} */

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

@@ -18,6 +18,7 @@
 
 import java.lang.reflect.Method;
 import org.apache.calcite.adapter.enumerable.NullPolicy;
+import org.apache.calcite.adapter.java.JavaTypeFactory;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.schema.ScalarFunction;
@@ -54,7 +55,9 @@ public static ScalarFunction create(Method method, boolean deterministic) {
 
     /** {@inheritDoc} */
     @Override public RelDataType getReturnType(RelDataTypeFactory typeFactory) {
-        return typeFactory.createJavaType(method.getReturnType());
+        JavaTypeFactory tf = (JavaTypeFactory)typeFactory;
+
+        return tf.toSql(tf.createJavaType(method.getReturnType()));
     }
 
     /**

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

@@ -18,11 +18,13 @@
 
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
+import java.lang.reflect.Type;
 import java.util.List;
 
 import org.apache.calcite.linq4j.tree.Expression;
 import org.apache.calcite.linq4j.tree.Expressions;
 import org.apache.calcite.rex.RexCall;
+import org.apache.ignite.internal.processors.query.calcite.util.TypeUtils;
 
 import static org.apache.ignite.internal.processors.query.calcite.util.IgniteMethod.UDF_INSTANCE;
 
@@ -49,10 +51,10 @@ public ReflectiveCallNotNullImplementor(Method method) {
     @Override public Expression implement(RexToLixTranslator translator,
         RexCall call, List<Expression> translatedOperands) {
         translatedOperands =
-            ConverterUtils.fromInternal(method.getParameterTypes(), translatedOperands);
+            ConverterUtils.fromInternal(translator.getRoot(), method.getParameterTypes(), translatedOperands);
         translatedOperands =
             ConverterUtils.convertAssignableTypes(method.getParameterTypes(), translatedOperands);
-        final Expression callExpr;
+        Expression callExpr;
         if ((method.getModifiers() & Modifier.STATIC) != 0)
             callExpr = Expressions.call(method, translatedOperands);
 
@@ -66,6 +68,24 @@ public ReflectiveCallNotNullImplementor(Method method) {
 
             callExpr = Expressions.call(target, method, translatedOperands);
         }
+
+        if (TypeUtils.isConvertableType(method.getReturnType())) {
+            Type targetType = translator.typeFactory.getJavaClass(call.getType());
+            Expression converted = ConverterUtils.toInternal(callExpr, targetType);
+
+            if (converted != callExpr)
+                callExpr = converted;
+            else {
+                Expression result = method.getReturnType().isPrimitive() ? Expressions.box(callExpr) : callExpr;
+
+                callExpr = Expressions.convert_(
+                    Expressions.call(TypeUtils.class, "toInternal", translator.getRoot(), result,
+                        Expressions.constant(method.getReturnType())),
+                    targetType
+                );
+            }
+        }
+
         if (!containsCheckedException(method))
             return callExpr;
 

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

@@ -19,6 +19,7 @@
 
 import java.nio.charset.Charset;
 import java.util.Arrays;
+import java.util.List;
 import org.apache.calcite.adapter.java.JavaTypeFactory;
 import org.apache.calcite.rel.type.DynamicRecordType;
 import org.apache.calcite.rel.type.RelDataType;
@@ -29,6 +30,7 @@
 import org.apache.calcite.sql.SqlCallBinding;
 import org.apache.calcite.sql.SqlCollation;
 import org.apache.calcite.sql.SqlDataTypeSpec;
+import org.apache.calcite.sql.SqlFunction;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
@@ -37,6 +39,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.SqlOperandMetadata;
 import org.apache.calcite.sql.type.SqlTypeFamily;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.sql.type.SqlTypeUtil;
@@ -59,6 +62,34 @@ public IgniteTypeCoercion(RelDataTypeFactory typeFactory, SqlValidator validator
         super(typeFactory, validator);
     }
 
+    /** {@inheritDoc} */
+    @Override public boolean userDefinedFunctionCoercion(SqlValidatorScope scope, SqlCall call, SqlFunction function) {
+        SqlOperandMetadata metadata = (SqlOperandMetadata)function.getOperandTypeChecker();
+        List<RelDataType> paramTypes = metadata.paramTypes(factory);
+
+        for (int i = 0; i < call.operandCount(); i++) {
+            SqlNode operand = call.operand(i);
+            int paramIdx = i;
+
+            if (operand.getKind() == SqlKind.ARGUMENT_ASSIGNMENT) {
+                SqlCall assignment = (SqlCall)operand;
+
+                paramIdx = metadata.paramNames().indexOf(((SqlIdentifier)assignment.operand(1)).getSimple());
+                operand = assignment.operand(0);
+
+                if (paramIdx < 0)
+                    return false;
+            }
+
+            // Numeric-to-timestamp casts are supported explicitly, but are not temporal UDF arguments.
+            if (SqlTypeUtil.isDatetime(paramTypes.get(paramIdx))
+                && SqlTypeUtil.isNumeric(validator.deriveType(scope, operand)))
+                return false;
+        }
+
+        return super.userDefinedFunctionCoercion(scope, call, function);
+    }
+
     /** {@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)