IGNITE-29031 SQL Calcite: Support byte[] 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 TypeUtils.toInternal(ctx, 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

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.UUID;
 import org.apache.calcite.adapter.enumerable.RexImpTable;
+import org.apache.calcite.avatica.util.ByteString;
 import org.apache.calcite.linq4j.tree.ConstantExpression;
 import org.apache.calcite.linq4j.tree.ConstantUntypedNull;
 import org.apache.calcite.linq4j.tree.Expression;
@@ -38,6 +39,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 {
@@ -80,11 +83,6 @@ else if (targetType == Long.class)
     /** Converts from internal representation to JDBC representation used by
      * arguments of user-defined functions. For example, converts date values from
      * {@code int} to {@link java.sql.Date}. */
-    private static Expression fromInternal(Expression operand, Type targetType) {
-        return fromInternal(operand, operand.getType(), targetType);
-    }
-
-    /** */
     private static Expression fromInternal(Expression operand,
         Type fromType, Type targetType) {
         if (operand == ConstantUntypedNull.INSTANCE)
@@ -111,6 +109,9 @@ else if (targetType == java.sql.Timestamp.class) {
             if (isA(fromType, Primitive.LONG))
                 return Expressions.call(BuiltInMethod.INTERNAL_TO_TIMESTAMP.method, operand);
         }
+        else if (targetType == byte[].class && fromType == ByteString.class)
+            return Expressions.call(BuiltInMethod.BYTE_STRING_TO_BYTE_ARRAY.method, operand);
+
         if (Primitive.is(operand.type)
             && Primitive.isBox(targetType)) {
             // E.g. operand is "int", target is "Long", generate "(long) operand".
@@ -123,28 +124,73 @@ 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);
+    }
+
+    /** */
+    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;
-            for (int i = 0; i < expressions.size(); i++) {
-                Class<?> type;
+
+            for (Expression expression : expressions) {
+                Class<?> targetType;
+
                 if (!targetTypes[j].isArray()) {
-                    type = targetTypes[j];
+                    targetType = targetTypes[j];
                     j++;
                 }
                 else
-                    type = targetTypes[j].getComponentType();
+                    targetType = targetTypes[j].getComponentType();
 
-                list.add(fromInternal(expressions.get(i), type));
+                list.add(fromInternal(root, expression, targetType));
             }
         }
+
         return list;
     }
 
+    /** */
+    private static Expression fromInternal(@Nullable Expression root, Expression operand, Type targetType) {
+        // Preserve Calcite conversions when no execution context is available, including temporal conversions.
+        if (root == null)
+            return fromInternal(operand, operand.getType(), targetType);
+
+        // Let the Java method call box compatible primitives instead of generating a reference cast.
+        if (Types.isAssignableFrom(targetType, operand.getType())
+            || Types.isAssignableFrom(targetType, Primitive.box(operand.getType())))
+            return operand;
+
+        if (!TypeUtils.isConvertableType(targetType))
+            return targetType == BigDecimal.class ? fromInternal(operand, operand.getType(), targetType) :
+                convert(operand, operand.getType(), targetType);
+
+        if (Primitive.is(operand.getType()))
+            operand = Expressions.box(operand);
+
+        Expression converted = Expressions.call(
+            TypeUtils.class,
+            "fromInternal",
+            root,
+            operand,
+            Expressions.constant(targetType)
+        );
+
+        Primitive primitive = Primitive.of(targetType);
+
+        return primitive == null
+            ? Expressions.convert_(converted, targetType)
+            : Expressions.unbox(Expressions.convert_(converted, primitive.boxClass), primitive);
+    }
+
     /** */
     private static Type toInternal(RelDataType type) {
         return toInternal(type, false);
@@ -230,6 +276,12 @@ public static Expression convert(Expression operand, Type fromType, Type toType)
         if (toType == BigDecimal.class)
             throw new AssertionError("For conversion to decimal, ConverterUtils#convertToDecimal method should be used instead.");
 
+        if (fromType == byte[].class && toType == ByteString.class)
+            return Expressions.call(BuiltInMethod.BYTE_ARRAY_TO_BYTE_STRING.method, operand);
+
+        if (fromType == ByteString.class && toType == byte[].class)
+            return Expressions.call(BuiltInMethod.BYTE_STRING_TO_BYTE_ARRAY.method, operand);
+
         // E.g. from "Short" to "int".
         // Generate "x.intValue()".
         final Primitive toPrimitive = Primitive.of(toType);

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

@@ -0,0 +1,75 @@
+/*
+ * 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 java.util.List;
+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;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.ignite.internal.processors.query.calcite.type.OtherType;
+
+/**
+ * Reflective Java function parameter represented with a SQL type.
+ *
+ * <p>The SQL representation is required to validate user-defined function arguments and to convert literal arguments
+ * while deriving a table function row type.
+ */
+final class IgniteFunctionParameter implements FunctionParameter {
+    /** Original function parameter. */
+    private final FunctionParameter delegate;
+
+    /** */
+    private IgniteFunctionParameter(FunctionParameter delegate) {
+        this.delegate = delegate;
+    }
+
+    /** Returns function parameters represented with SQL types. */
+    static List<FunctionParameter> toSql(List<FunctionParameter> parameters) {
+        return parameters.stream().map(IgniteFunctionParameter::toSql).toList();
+    }
+
+    /** Returns a function parameter represented with a SQL type. */
+    static FunctionParameter toSql(FunctionParameter parameter) {
+        return new IgniteFunctionParameter(parameter);
+    }
+
+    /** {@inheritDoc} */
+    @Override public int getOrdinal() {
+        return delegate.getOrdinal();
+    }
+
+    /** {@inheritDoc} */
+    @Override public String getName() {
+        return delegate.getName();
+    }
+
+    /** {@inheritDoc} */
+    @Override public RelDataType getType(RelDataTypeFactory typeFactory) {
+        JavaTypeFactory tf = (JavaTypeFactory)typeFactory;
+        RelDataType type = tf.toSql(delegate.getType(typeFactory));
+
+        // Prevent the validator from replacing OTHER with a structured type derived from a dynamic parameter value.
+        return type.getSqlTypeName() == SqlTypeName.OTHER ? new OtherType(type.isNullable()) : type;
+    }
+
+    /** {@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,30 @@
 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;
 
 /** A base for outer java-method functions. */
 abstract class IgniteReflectiveFunctionBase extends ReflectiveFunctionBase implements ImplementableFunction {
     /** */
     protected final CallImplementor implementor;
 
+    /** */
+    private final List<FunctionParameter> funcParams;
+
     /** */
     protected IgniteReflectiveFunctionBase(Method method, CallImplementor implementor) {
         super(method);
 
         this.implementor = implementor;
+
+        funcParams = IgniteFunctionParameter.toSql(super.getParameters());
+    }
+
+    /** {@inheritDoc} */
+    @Override public List<FunctionParameter> getParameters() {
+        return funcParams;
     }
 
     /** {@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/IgniteTableFunction.java

@@ -21,6 +21,7 @@
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.calcite.adapter.enumerable.NullPolicy;
@@ -80,7 +81,7 @@ public static IgniteTableFunction create(Method method, Class<?>[] colTypes, Str
     @Override public RelDataType getRowType(RelDataTypeFactory typeFactory, List<?> arguments) {
         JavaTypeFactory tf = (JavaTypeFactory)typeFactory;
 
-        List<RelDataType> converted = Stream.of(colTypes).map(cl -> tf.toSql(tf.createType(cl))).collect(Collectors.toList());
+        List<RelDataType> converted = Stream.of(colTypes).map(cl -> columnType(tf, cl)).collect(Collectors.toList());
 
         return typeFactory.createStructType(converted, colNames);
     }
@@ -100,6 +101,15 @@ public static IgniteTableFunction create(Method method, Class<?>[] colTypes, Str
         return Iterable.class;
     }
 
+    /** Resolves collection types without treating user-defined classes as records. */
+    private static RelDataType columnType(JavaTypeFactory tf, Class<?> cls) {
+        RelDataType type = cls.isArray() || List.class.isAssignableFrom(cls) || Map.class.isAssignableFrom(cls)
+            ? tf.createType(cls)
+            : tf.createJavaType(cls);
+
+        return tf.toSql(type);
+    }
+
     /** Validates the parameters and throws an exception if it finds an incorrect parameter. */
     private static void validate(Method mtd, Class<?>[] colTypes, String[] colNames) {
         if (F.isEmpty(colTypes))