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

@@ -768,7 +768,7 @@ else if (rel instanceof Intersect)
 
         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, rowFactory));
     }
 
     /** {@inheritDoc} */

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

@@ -23,10 +23,15 @@
 import org.apache.calcite.rel.type.RelDataType;
 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;
 
 /** */
 public class TableFunctionScan<Row> implements Iterable<Row> {
+    /** */
+    private final ExecutionContext<Row> ctx;
+
     /** */
     private final RelDataType rowType;
 
@@ -38,10 +43,12 @@ public class TableFunctionScan<Row> implements Iterable<Row> {
 
     /** */
     public TableFunctionScan(
+        ExecutionContext<Row> ctx,
         RelDataType rowType,
         Supplier<Iterable<?>> dataSupplier,
         RowFactory<Row> rowFactory
     ) {
+        this.ctx = ctx;
         this.rowType = rowType;
         this.dataSupplier = dataSupplier;
         this.rowFactory = rowFactory;
@@ -58,14 +65,19 @@ 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++) {
+            if (!(rowType.getFieldList().get(i).getType() instanceof OtherType))
+                rowArr[i] = TypeUtils.toInternal(ctx, rowArr[i]);
+        }
+
         return rowFactory.create(rowArr);
     }
 }

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,7 @@
 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;
 
 /** */
 public class ConverterUtils {
@@ -74,6 +76,9 @@ else if (fromType == java.sql.Timestamp.class) {
             else if (targetType == Long.class)
                 return Expressions.call(BuiltInMethod.TIMESTAMP_TO_LONG_OPTIONAL.method, operand);
         }
+        else if (fromType == byte[].class && targetType == ByteString.class)
+            return Expressions.call(BuiltInMethod.BYTE_ARRAY_TO_BYTE_STRING.method, operand);
+
         return operand;
     }
 
@@ -111,6 +116,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".
@@ -145,6 +153,64 @@ static List<Expression> fromInternal(Class<?>[] targetTypes,
         return list;
     }
 
+    /** */
+    static List<Expression> fromInternal(RexToLixTranslator translator,
+        Class<?>[] targetTypes,
+        List<Expression> expressions
+    ) {
+        final List<Expression> list = new ArrayList<>();
+
+        if (targetTypes.length == expressions.size()) {
+            for (int i = 0; i < expressions.size(); i++)
+                list.add(fromInternal(translator, expressions.get(i), targetTypes[i]));
+        }
+        else {
+            int j = 0;
+
+            for (Expression expression : expressions) {
+                Class<?> targetType;
+
+                if (!targetTypes[j].isArray()) {
+                    targetType = targetTypes[j];
+                    j++;
+                }
+                else
+                    targetType = targetTypes[j].getComponentType();
+
+                list.add(fromInternal(translator, expression, targetType));
+            }
+        }
+
+        return list;
+    }
+
+    /** */
+    private static Expression fromInternal(RexToLixTranslator translator, Expression operand, Type targetType) {
+        if (Types.isAssignableFrom(targetType, operand.getType()))
+            return operand;
+
+        if (!TypeUtils.isConvertableType(targetType))
+            return targetType == BigDecimal.class ? fromInternal(operand, targetType) :
+                convert(operand, operand.getType(), targetType);
+
+        if (Primitive.is(operand.getType()))
+            operand = Expressions.box(operand);
+
+        Expression converted = Expressions.call(
+            TypeUtils.class,
+            "fromInternal",
+            translator.getRoot(),
+            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 +296,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,79 @@
+/*
+ * 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;
+
+    /**
+     * Constructor.
+     *
+     * @param delegate Original function parameter.
+     */
+    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/IgniteScalarFunction.java

@@ -17,9 +17,12 @@
 package org.apache.ignite.internal.processors.query.calcite.exec.exp;
 
 import java.lang.reflect.Method;
+import java.util.List;
 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.FunctionParameter;
 import org.apache.calcite.schema.ScalarFunction;
 
 /**
@@ -29,13 +32,18 @@ public class IgniteScalarFunction extends IgniteReflectiveFunctionBase implement
     /** */
     private final boolean deterministic;
 
+    /** */
+    private final List<FunctionParameter> funcParams;
+
     /**
      * Private constructor.
      */
     private IgniteScalarFunction(Method method, CallImplementor implementor, boolean deterministic) {
         super(method, implementor);
 
         this.deterministic = deterministic;
+
+        funcParams = IgniteFunctionParameter.toSql(super.getParameters());
     }
 
     /**
@@ -54,7 +62,14 @@ 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()));
+    }
+
+    /** {@inheritDoc} */
+    @Override public List<FunctionParameter> getParameters() {
+        return funcParams;
     }
 
     /**

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

@@ -27,6 +27,7 @@
 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.schema.TableFunction;
 import org.apache.ignite.cache.query.annotations.QuerySqlTableFunction;
 import org.apache.ignite.internal.processors.query.IgniteSQLException;
@@ -44,6 +45,9 @@ public class IgniteTableFunction extends IgniteReflectiveFunctionBase implements
     /** Column names of the returned table representation. */
     private final List<String> colNames;
 
+    /** */
+    private final List<FunctionParameter> funcParams;
+
     /**
      * Creates user-defined table function holder.
      *
@@ -59,6 +63,8 @@ private IgniteTableFunction(Method method, Class<?>[] colTypes, String[] colName
 
         this.colTypes = colTypes;
         this.colNames = Arrays.asList(colNames);
+
+        funcParams = IgniteFunctionParameter.toSql(super.getParameters());
     }
 
     /**
@@ -80,7 +86,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 -> tf.toSql(tf.createJavaType(cl))).collect(Collectors.toList());
 
         return typeFactory.createStructType(converted, colNames);
     }
@@ -100,6 +106,11 @@ public static IgniteTableFunction create(Method method, Class<?>[] colTypes, Str
         return Iterable.class;
     }
 
+    /** {@inheritDoc} */
+    @Override public List<FunctionParameter> getParameters() {
+        return funcParams;
+    }
+
     /** 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))

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, 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,17 @@ public ReflectiveCallNotNullImplementor(Method method) {
 
             callExpr = Expressions.call(target, method, translatedOperands);
         }
+
+        if (TypeUtils.isConvertableType(method.getReturnType())) {
+            Type targetType = translator.typeFactory.getJavaClass(call.getType());
+            Expression result = method.getReturnType().isPrimitive() ? Expressions.box(callExpr) : callExpr;
+
+            callExpr = Expressions.convert_(
+                Expressions.call(TypeUtils.class, "toInternal", translator.getRoot(), result),
+                targetType
+            );
+        }
+
         if (!containsCheckedException(method))
             return callExpr;