Apache Calcite Catalog 拾遗之 UDF 函数实现和扩展

public interface Function {
    List<FunctionParameter> getParameters();
}

/**
 * Function that returns a scalar result.
 */
public interface ScalarFunction extends Function {
    /**
     * Returns the return type of this function, constructed using the given
     * type factory.
     *
     * @param typeFactory Type factory
     */
    RelDataType getReturnType(RelDataTypeFactory typeFactory);
}

/**
 * Function that can be translated to java code.
 *
 * @see ScalarFunction
 * @see TableFunction
 */
public interface ImplementableFunction extends Function {
    /**
     * Returns implementor that translates the function to linq4j expression.
     *
     * @return implementor that translates the function to linq4j expression.
     */
    CallImplementor getImplementor();
}

public interface CallImplementor {
    /**
     * Implements a call.
     *
     * @param translator Translator for the call
     * @param call Call that should be implemented
     * @param nullAs The desired mode of {@code null} translation
     * @return Translated call
     */
    Expression implement(RexToLixTranslator translator, RexCall call, RexImpTable.NullAs nullAs);
}

/**
 * Creates a ReflectiveFunctionBase.
 *
 * @param method Method that is used to get type information from
 */
protected ReflectiveFunctionBase(Method method) {
    this.method = method;
    this.parameters = builder().addMethodParameters(method).build();
}

public ParameterListBuilder addMethodParameters(Method method) {
    final Class<?>[] types = method.getParameterTypes();
    for (int i = 0; i < types.length; i++) {
        add(types[i], ReflectUtil.getParameterName(method, i), ReflectUtil.isParameterOptional(method, i));
    }
    return this;
}

public ParameterListBuilder add(final Class<?> type, final String name, final boolean optional) {
    final int ordinal = builder.size();
    builder.add(new FunctionParameter() {
        @Override
        public String toString() {
            return ordinal + ": " + name + " " + type.getSimpleName() + (optional ? "?" : "");
        }
        
        // 基于 0 的参数序号
        @Override
        public int getOrdinal() {
            return ordinal;
        }
        
        // 参数名称
        @Override
        public String getName() {
            return name;
        }
        
        // 参数类型
        @Override
        public RelDataType getType(RelDataTypeFactory typeFactory) {
            return typeFactory.createJavaType(type);
        }
        
        // 参数是否可选，可选参数可以在函数调用时省略
        @Override
        public boolean isOptional() {
            return optional;
        }
    });
    return this;
}

/**
 * Private constructor.
 */
private ScalarFunctionImpl(Method method, CallImplementor implementor) {
    super(method);
    this.implementor = implementor;
}

public static ScalarFunction create(Method method) {
    if (!isStatic(method)) {
        Class<?> clazz = method.getDeclaringClass();
        if (!classHasPublicZeroArgsConstructor(clazz) && !classHasPublicFunctionContextConstructor(clazz)) {
            throw RESOURCE.requireDefaultConstructor(clazz.getName()).ex();
        }
    }
    CallImplementor implementor = createImplementor(method);
    return new ScalarFunctionImpl(method, implementor);
}

private static CallImplementor createImplementor(final Method method) {
    final NullPolicy nullPolicy = getNullPolicy(method);
    return RexImpTable.createImplementor(new ReflectiveCallNotNullImplementor(method), nullPolicy, false);
}

public enum NullPolicy {
    /**
     * Returns null if and only if all of the arguments are null;
     * If all of the arguments are false return false otherwise true.
     */
    ALL,
    /**
     * Returns null if and only if one of the arguments are null.
     */
    STRICT,
    /**
     * Returns null if one of the arguments is null, and possibly other times.
     */
    SEMI_STRICT,
    /**
     * If any of the arguments are null, return null.
     */
    ANY,
    /**
     * If the first argument is null, return null.
     */
    ARG0, NONE
}

// 聚合函数累加器
struct Accumulator {
    final int sum;
}

// 聚合函数初始化方法
Accumulator init() {
    return new Accumulator(0);
}

// 聚合函数累加方法
Accumulator add(Accumulator a, int x) {
    return new Accumulator(a.sum + x);
}

// 聚合函数合并方法
Accumulator merge(Accumulator a, Accumulator a2) {
    return new Accumulator(a.sum + a2.sum);
}

// 聚合函数获取结果方法
int result(Accumulator a) {
    return a.sum;
}

/**
 * Function that combines several values into a scalar result.
 */
public interface AggregateFunction extends Function {
    /**
     * Returns the return type of this function, constructed using the given
     * type factory.
     *
     * @param typeFactory Type factory
     */
    RelDataType getReturnType(RelDataTypeFactory typeFactory);
}

public interface ImplementableAggFunction extends AggregateFunction {
    /**
     * Returns implementor that translates the function to linq4j expression.
     *
     * @param windowContext true when aggregate is used in window context
     * @return implementor that translates the function to linq4j expression.
     */
    AggImplementor getImplementor(boolean windowContext);
}

public interface AggImplementor {
    
    // 返回聚合函数实现时，中间变量的类型
    // 例如：字符串连接函数，它的中间变量类型可以是 StringBuilder
    List<Type> getStateType(AggContext info);
    
    // 将中间变量重置为初识状态
    // 应使用 AggResetContext.accumulator() 来引用状态变量
    void implementReset(AggContext info, AggResetContext reset);
    
    // 将新增加的当前值，累加到中间变量
    void implementAdd(AggContext info, AggAddContext add);
    
    // 根据中间变量计算结果值
    Expression implementResult(AggContext info, AggResultContext result);
}

private AggregateFunctionImpl(Class<?> declaringClass, List<FunctionParameter> params, List<Class<?>> valueTypes, Class<?> accumulatorType, Class<?> resultType, Method initMethod, Method addMethod, @Nullable Method mergeMethod, @Nullable Method resultMethod) {
    this.declaringClass = declaringClass;
    this.valueTypes = ImmutableList.copyOf(valueTypes);
    this.parameters = params;
    this.accumulatorType = accumulatorType;
    this.resultType = resultType;
    this.initMethod = requireNonNull(initMethod, "initMethod");
    this.addMethod = requireNonNull(addMethod, "addMethod");
    this.mergeMethod = mergeMethod;
    this.resultMethod = resultMethod;
    this.isStatic = isStatic(initMethod);
    assert resultMethod != null || accumulatorType == resultType;
}

/**
 * Creates an aggregate function, or returns null.
 */
public static @Nullable AggregateFunctionImpl create(Class<?> clazz) {
    // 获取函数类中的 init 方法
    final Method initMethod = ReflectiveFunctionBase.findMethod(clazz, "init");
    // 获取函数类中的 add 方法
    final Method addMethod = ReflectiveFunctionBase.findMethod(clazz, "add");
    // merge 方法暂未实现
    final Method mergeMethod = null; // TODO:
    // 获取函数类中的 result 方法
    final Method resultMethod = ReflectiveFunctionBase.findMethod(clazz, "result");
    if (initMethod != null && addMethod != null) {
        // A is return type of init by definition
        // A 表示 init 方法的返回类型
        final Class<?> accumulatorType = initMethod.getReturnType();
        // R is return type of result by definition
        // R 表示 result 方法的返回类型
        final Class<?> resultType = resultMethod != null ? resultMethod.getReturnType() : accumulatorType;
        // V is remaining args of add by definition
        // V 表示 add 方法第一个参数之外的类型
        final List<Class> addParamTypes = ImmutableList.copyOf(addMethod.getParameterTypes());
        // 检查 add 方法第一个参数和累加器类型是否一致
        if (addParamTypes.isEmpty() || addParamTypes.get(0) != accumulatorType) {
            throw RESOURCE.firstParameterOfAdd(clazz.getName()).ex();
        }
        final ReflectiveFunctionBase.ParameterListBuilder params = ReflectiveFunctionBase.builder();
        final ImmutableList.Builder<Class<?>> valueTypes = ImmutableList.builder();
        // 跳过第一个构造器参数，从第二个参数开始遍历
        for (int i = 1; i < addParamTypes.size(); i++) {
            final Class type = addParamTypes.get(i);
            // 调用 ReflectUtil 获取参数名称和参数是否可选标记，优先选择 Parameter 注解声明的信息
            final String name = ReflectUtil.getParameterName(addMethod, i);
            final boolean optional = ReflectUtil.isParameterOptional(addMethod, i);
            params.add(type, name, optional);
            valueTypes.add(type);
        }
        // 如下展示了聚合函数中的方法和类型
        // A init()
        // A add(A, V)
        // A merge(A, A)
        // R result(A)        
        return new AggregateFunctionImpl(clazz, params.build(), valueTypes.build(), accumulatorType, resultType, initMethod, addMethod, mergeMethod, resultMethod);
    }
    return null;
}

@Override
public AggImplementor getImplementor(boolean windowContext) {
    return new RexImpTable.UserDefinedAggReflectiveImplementor(this);
}

public interface TableFunction extends Function {
    // 使用指定参数产生的表对应的数据行类型
    RelDataType getRowType(RelDataTypeFactory typeFactory, List<? extends @Nullable Object> arguments);
    
    // 获取表对应的数据行类型（Java 类型）
    Type getElementType(List<? extends @Nullable Object> arguments);
}

private TableFunctionImpl(Method method, CallImplementor implementor) {
    super(method);
    this.implementor = implementor;
}

/**
 * Creates a {@link TableFunctionImpl} from a class, looking for a method
 * with a given name. Returns null if there is no such method.
 */
public static @Nullable TableFunction create(Class<?> clazz, String methodName) {
    final Method method = findMethod(clazz, methodName);
    if (method == null) {
        return null;
    }
    return create(method);
}

/**
 * Creates a {@link TableFunctionImpl} from a method.
 */
public static @Nullable TableFunction create(final Method method) {
    // 判断是否为非静态方法
    if (!isStatic(method)) {
        Class clazz = method.getDeclaringClass();
        // 非静态方法，如果没有公有的无参构造方法，则抛出异常
        if (!classHasPublicZeroArgsConstructor(clazz)) {
            throw RESOURCE.requireDefaultConstructor(clazz.getName()).ex();
        }
    }
    final Class<?> returnType = method.getReturnType();
    // 如果表函数返回值不为 QueryableTable 或者 ScannableTable，则直接返回 null
    if (!QueryableTable.class.isAssignableFrom(returnType) && !ScannableTable.class.isAssignableFrom(returnType)) {
        return null;
    }
    // 创建调用实现器
    CallImplementor implementor = createImplementor(method);
    // 创建 TableFunctionImpl 对象
    return new TableFunctionImpl(method, implementor);
}

public interface TableMacro extends Function {
    /**
     * Applies arguments to yield a table.
     *
     * @param arguments Arguments
     * @return Table
     */
    TranslatableTable apply(List<? extends @Nullable Object> arguments);
}

/**
 * Private constructor; use {@link #create}.
 */
private TableMacroImpl(Method method) {
    super(method);
}

/**
 * Creates a {@code TableMacro} from a method.
 */
public static @Nullable TableMacro create(final Method method) {
    Class clazz = method.getDeclaringClass();
    // 非静态方法，必须提供公有无参构造方法
    if (!isStatic(method)) {
        if (!classHasPublicZeroArgsConstructor(clazz)) {
            throw RESOURCE.requireDefaultConstructor(clazz.getName()).ex();
        }
    }
    final Class<?> returnType = method.getReturnType();
    // 检查表宏返回值是否为 TranslatableTable
    if (!TranslatableTable.class.isAssignableFrom(returnType)) {
        return null;
    }
    return new TableMacroImpl(method);
}

@Override
public TranslatableTable apply(List<? extends @Nullable Object> arguments) {
    try {
        Object o = null;
        if (!isStatic(method)) {
            final Constructor<?> constructor = method.getDeclaringClass().getConstructor();
            o = constructor.newInstance();
        }
        return (TranslatableTable) requireNonNull(method.invoke(o, arguments.toArray()), () -> "got null from " + method + " with arguments " + arguments);
    } catch (IllegalArgumentException e) {
        throw new RuntimeException("Expected " + Arrays.toString(method.getParameterTypes()) + " actual " + arguments, e);
    } catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException | InstantiationException e) {
        throw new RuntimeException(e);
    }
}

@Override
public <T> @PolyNull T fun(Class<T> operatorTableClass, @PolyNull T defaultOperatorTable) {
    final String fun = CalciteConnectionProperty.FUN.wrap(properties).getString();
    if (fun == null || fun.equals("") || fun.equals("standard")) {
        return defaultOperatorTable;
    }
    // Parse the libraries
    final List<SqlLibrary> libraryList = SqlLibrary.parse(fun);
    // Load standard plus the specified libraries. If 'all' is among the
    // specified libraries, it is expanded to all libraries (except standard,
    // spatial, all).
    final List<SqlLibrary> libraryList1 = SqlLibrary.expand(ConsList.of(SqlLibrary.STANDARD, libraryList));
    // 根据 libraries 获取对应的 SqlLibraryOperatorTable
    final SqlOperatorTable operatorTable = SqlLibraryOperatorTableFactory.INSTANCE.getOperatorTable(libraryList1);
    return operatorTableClass.cast(operatorTable);
}

for (Class aClass : classes) {
    for (Field field : aClass.getFields()) {
        try {
            if (SqlOperator.class.isAssignableFrom(field.getType())) {
                final SqlOperator op = (SqlOperator) requireNonNull(field.get(this), () -> "null value of " + field + " for " + this);
                // 判断 Library 是否一致
                if (operatorIsInLibrary(op.getName(), field, librarySet)) {
                    list.add(op);
                }
            }
        } catch (IllegalArgumentException | IllegalAccessException e) {
            throw Util.throwAsRuntime(Util.causeOrSelf(e));
        }
    }
}

private static SqlValidator createSqlValidator(Context context, CalciteCatalogReader catalogReader, UnaryOperator<SqlValidator.Config> configTransform) {
    // 根据 context.config() 中配置的 fun 属性，获取对应方言中的函数或运算符
    // 未配置或配置为 standard，则使用 SqlStdOperatorTable.instance() 创建的标准函数、运算符表对象
    final SqlOperatorTable opTab0 = context.config().fun(SqlOperatorTable.class, SqlStdOperatorTable.instance());
    final List<SqlOperatorTable> list = new ArrayList<>();
    list.add(opTab0);
    // 添加 catalogReader
    list.add(catalogReader);
    // 组装为调用链 Chain
    final SqlOperatorTable opTab = SqlOperatorTables.chain(list);
    final JavaTypeFactory typeFactory = context.getTypeFactory();
    final CalciteConnectionConfig connectionConfig = context.config();
    final SqlValidator.Config config = configTransform.apply(SqlValidator.Config.DEFAULT.withLenientOperatorLookup(connectionConfig.lenientOperatorLookup()).withConformance(connectionConfig.conformance()).withDefaultNullCollation(connectionConfig.defaultNullCollation()).withIdentifierExpansion(true));
    return new CalciteSqlValidator(opTab, catalogReader, typeFactory, config);
}

public final SqlOperatorTable init() {
    // Use reflection to register the expressions stored in public fields.
    final List<SqlOperator> list = new ArrayList<>();
    for (Field field : getClass().getFields()) {
        try {
            final Object o = field.get(this);
            if (o instanceof SqlOperator) {
                // Fields do not need the LibraryOperator tag, but if they have it,
                // we index them only if they contain STANDARD library.
                // 获取 LibraryOperator 注解，有注解但是没有 STANDARD 类型的函数为方言函数，直接跳过
                LibraryOperator libraryOperator = field.getAnnotation(LibraryOperator.class);
                if (libraryOperator != null) {
                    if (Arrays.stream(libraryOperator.libraries()).noneMatch(library -> library == SqlLibrary.STANDARD)) {
                        continue;
                    }
                }
                // 标准函数大多不用 LibraryOperator 注解标记，直接添加到集合中
                list.add((SqlOperator) o);
            }
        } catch (IllegalArgumentException | IllegalAccessException e) {
            throw Util.throwAsRuntime(Util.causeOrSelf(e));
        }
    }
    // 将 SqlOperator 构建为 Map 结果，key 为 SqlOperator 大写名称，value 为 SqlOperator 对象
    // 存储到 SqlStdOperatorTable 中的 operators 对象中
    setOperators(buildIndex(list));
    return this;
}

// 判断当前 SqlNode 是否为 SqlCall
if (node instanceof SqlCall) {
    ...
    SqlCall call = (SqlCall) node;
    // 获取 SqlKind 类型
    final SqlKind kind = call.getKind();
    // 获取 SqlNode 中包含的运算符
    final List<SqlNode> operands = call.getOperandList();
    for (int i = 0; i < operands.size(); i++) {
        SqlNode operand = operands.get(i);
        ...
        // 每一个运算法调用 performUnconditionalRewrites 并设置到 SqlCall 中
        SqlNode newOperand = performUnconditionalRewrites(operand, childUnderFrom);
        if (newOperand != null && newOperand != operand) {
            call.setOperand(i, newOperand);
        }
    }

    // 当前运算符为未解析函数 SqlUnresolvedFunction
    if (call.getOperator() instanceof SqlUnresolvedFunction) {
        final SqlUnresolvedFunction function = (SqlUnresolvedFunction) call.getOperator();
        final List<SqlOperator> overloads = new ArrayList<>();
        // 从 SqlOperatorTable 中查找函数，查找的范围包括内置函数以及元数据中注册的函数
        opTab.lookupOperatorOverloads(function.getNameAsId(), function.getFunctionType(), SqlSyntax.FUNCTION, overloads, catalogReader.nameMatcher());
        if (overloads.size() == 1) {
          	// 查找到函数则设置新的运算符
            ((SqlBasicCall) call).setOperator(overloads.get(0));
        }
    }
}

public abstract class SqlOperator {
    // 运算符/函数名称
    private final String name;
    // SqlNode 类型，例如：EQUALS，可以使用 exp.isA(EQUALS) 判断
    public final SqlKind kind;
    // 运算符绑定到左侧表达式的优先级，如果运算符是左结合，则此优先级低于右侧优先级
    private final int leftPrec;
    // 运算符绑定到右侧表达式的优先级，如果运算符是左结合，则此优先级高于右侧优先级
    private final int rightPrec;
    // 推断运算符调用后的返回类型
    private final @Nullable SqlReturnTypeInference returnTypeInference;
    // 推断操作数的类型
    private final @Nullable SqlOperandTypeInference operandTypeInference;
    // 用于校验操作数类型
    private final @Nullable SqlOperandTypeChecker operandTypeChecker;
    // 返回运算符的语法类型
    public abstract SqlSyntax getSyntax();
    // 推断运算符调用后的返回类型，仅在操作数的数量和类型校验后调用
    public RelDataType inferReturnType(SqlOperatorBinding opBinding) {...}
}

public class SqlFunction extends SqlOperator {
    // SQL 函数分类
    private final SqlFunctionCategory category;
    // 函数的全限定名称，内置函数为 null
    private final @Nullable SqlIdentifier sqlIdentifier;
    
    protected SqlFunction(String name, @Nullable SqlIdentifier sqlIdentifier, SqlKind kind, @Nullable SqlReturnTypeInference returnTypeInference, @Nullable SqlOperandTypeInference operandTypeInference, @Nullable SqlOperandTypeChecker operandTypeChecker, SqlFunctionCategory category) {
        // SqlFunction 默认传递给父类 SqlOperator 的 leftPrec 和 rightPrec 为 100
        super(name, kind, 100, 100, returnTypeInference, operandTypeInference, operandTypeChecker);
        this.sqlIdentifier = sqlIdentifier;
        this.category = requireNonNull(category, "category");
    }
    
    // 获取语法类型，SqlSyntax.FUNCTION 表示 Foo(x, y) 格式的函数
    @Override
    public SqlSyntax getSyntax() {
        return SqlSyntax.FUNCTION;
    }
}

public class SqlBasicFunction extends SqlFunction {
    // 语法类型，可以使用 withSyntax 修改
    private final SqlSyntax syntax;
    // 是否是确定函数，即：是否传入相同操作数时返回相同结果，默认为 true
    private final boolean deterministic;
    // 操作数处理器
    private final SqlOperandHandler operandHandler;
    // 校验调用的策略，目前没有发现使用场景
    private final int callValidator;
    // 推断函数单调性的策略
    // 入参是 SqlOperatorBinding，表示 SqlOperator 与操作数之间的绑定关系
    // 出参是 SqlMonotonicity，表示 SQL 中值的单调性
    private final Function<SqlOperatorBinding, SqlMonotonicity> monotonicityInference;
    // 是否是动态函数，即：是否传入相同操作数时返回不同结果，默认为 false，动态函数不能被缓存
    private final boolean dynamic;
}

public class SqlCastFunction extends SqlFunction {
    
    // 所有不保持单调性的类型转换映射，getMonotonicity 方法根据 map 判断，包含在其中的，则返回 NOT_MONOTONIC
    private final SetMultimap<SqlTypeFamily, SqlTypeFamily> nonMonotonicCasts = ImmutableSetMultimap.<SqlTypeFamily, SqlTypeFamily>builder().put(SqlTypeFamily.EXACT_NUMERIC, SqlTypeFamily.CHARACTER).put(SqlTypeFamily.NUMERIC, SqlTypeFamily.CHARACTER).put(SqlTypeFamily.APPROXIMATE_NUMERIC, SqlTypeFamily.CHARACTER).put(SqlTypeFamily.DATETIME_INTERVAL, SqlTypeFamily.CHARACTER).put(SqlTypeFamily.CHARACTER, SqlTypeFamily.EXACT_NUMERIC).put(SqlTypeFamily.CHARACTER, SqlTypeFamily.NUMERIC).put(SqlTypeFamily.CHARACTER, SqlTypeFamily.APPROXIMATE_NUMERIC).put(SqlTypeFamily.CHARACTER, SqlTypeFamily.DATETIME_INTERVAL).put(SqlTypeFamily.DATETIME, SqlTypeFamily.TIME).put(SqlTypeFamily.TIMESTAMP, SqlTypeFamily.TIME).put(SqlTypeFamily.TIME, SqlTypeFamily.DATETIME).put(SqlTypeFamily.TIME, SqlTypeFamily.TIMESTAMP).build();
    
    // 构造方法传入 name 和 kind，SAFE_CAST 和 CAST 类似，它转换失败时会返回 NULL，而不是异常
    public SqlCastFunction(String name, SqlKind kind) {
        super(name, kind, returnTypeInference(kind == SqlKind.SAFE_CAST), InferTypes.FIRST_KNOWN, null, SqlFunctionCategory.SYSTEM);
        checkArgument(kind == SqlKind.CAST || kind == SqlKind.SAFE_CAST, kind);
    }
    
    // 返回一个模板，描述如何构建运算符签名，以 CAST 函数为例，{0} 表示运算符 CAST，{1}、{2}、{3} 是操作数
    @Override
    public String getSignatureTemplate(final int operandsCount) {
        assert operandsCount <= 3;
        return "{0}({1} AS {2} [FORMAT {3}])";
    }
    
    // 获取语法类型，SqlSyntax.SPECIAL 表示特殊语法，例如：CASE、CAST 运算符
    @Override
    public SqlSyntax getSyntax() {
        return SqlSyntax.SPECIAL;
    }
    
    @Override
    public SqlMonotonicity getMonotonicity(SqlOperatorBinding call) {
        // 获取第一个操作数类型，CAST 原始类型
        final RelDataType castFromType = call.getOperandType(0);
        // 获取第一个操作数类型族
        final RelDataTypeFamily castFromFamily = castFromType.getFamily();
        // 获取第一个操作数排序规则对应的排序器
        final Collator castFromCollator = castFromType.getCollation() == null ? null : castFromType.getCollation().getCollator();
        // 获取第二个操作数类型，CAST 目标类型
        final RelDataType castToType = call.getOperandType(1);
        // 获取第二个操作数类型族
        final RelDataTypeFamily castToFamily = castToType.getFamily();
        // 获取第二个操作数排序规则对应的排序器
        final Collator castToCollator = castToType.getCollation() == null ? null : castToType.getCollation().getCollator();
        // 如果两个操作数的排序器不同，则返回 NOT_MONOTONIC
        if (!Objects.equals(castFromCollator, castToCollator)) {
            // Cast between types compared with different collators: not monotonic.
            return SqlMonotonicity.NOT_MONOTONIC;
            // 如果两个操作数的类型族包含在 nonMonotonicCasts 中，则返回 NOT_MONOTONIC
        } else if (castFromFamily instanceof SqlTypeFamily && castToFamily instanceof SqlTypeFamily && nonMonotonicCasts.containsEntry(castFromFamily, castToFamily)) {
            return SqlMonotonicity.NOT_MONOTONIC;
        } else {
            // 否则返回第一个操作数的单调性策略
            return call.getOperandMonotonicity(0);
        }
    }
}

// Project select clause.
int i = -1;
for (SqlNode expr : selectList) {
    ++i;
    final SqlNode measure = SqlValidatorUtil.getMeasure(expr);
    final RexNode e;
    if (measure != null) {
        ...
    } else {
        // 调用 Blackboard 对象 convertExpression 方法，该类实现了 SqlVisitor，用于遍历 AST 处理对应 SqlCall 转换
        e = bb.convertExpression(expr);
    }
    exprs.add(e);
    fieldNames.add(deriveAlias(expr, aliases, i));
}

public RexNode visit (SqlCall call){
    if (agg != null) {
        final SqlOperator op = call.getOperator();
        if (window == null && (op.isAggregator() || op.getKind() == SqlKind.FILTER || op.getKind() == SqlKind.WITHIN_DISTINCT || op.getKind() == SqlKind.WITHIN_GROUP)) {
            return requireNonNull(agg.lookupAggregates(call), () -> "agg.lookupAggregates for call " + call);
        }
    }
    // 调用 SqlNodeToRexConverter#convertCall 方法，SqlNodeToRexConverter 用于将 SqlNode 转换为 RexNode 行表达式
    return exprConverter.convertCall(this, new SqlCallBinding(validator(), scope, call).permutedCall());
}

public RexNode convertFunction(SqlRexContext cx, SqlFunction fun, SqlCall call) {
    // 转换操作数为 RexNode
    final List<RexNode> exprs = convertOperands(cx, call, SqlOperandTypeChecker.Consistency.NONE);
    if (fun.getFunctionType() == SqlFunctionCategory.USER_DEFINED_CONSTRUCTOR) {
        return makeConstructorCall(cx, fun, exprs);
    }
    RelDataType returnType = cx.getValidator().getValidatedNodeTypeIfKnown(call);
    if (returnType == null) {
        returnType = cx.getRexBuilder().deriveReturnType(fun, exprs);
    }
    // 调用 RexBuilder#makeCall 转换
    return cx.getRexBuilder().makeCall(call.getParserPosition(), returnType, fun, exprs);
}

public static Bindable toBindable(Map<String, Object> parameters, CalcitePrepare.@Nullable SparkHandler spark, EnumerableRel rel, EnumerableRel.Prefer prefer) {
    EnumerableRelImplementor relImplementor = new EnumerableRelImplementor(rel.getCluster().getRexBuilder(), parameters);
    // 调用 implementRoot 生成执行代码
    final ClassDeclaration expr = relImplementor.implementRoot(rel, prefer);
    String s = Expressions.toString(expr.memberDeclarations, "\n", false);
    try {
        if (spark != null && spark.enabled()) {
            return spark.compile(expr, s);
        } else {
            // 使用 Janio 编译代码为 Bindable 实现类
            return getBindable(expr, s, rel.getRowType().getFieldCount());
        }
    } catch (Exception e) {
        throw Helper.INSTANCE.wrap("Error while compiling generated Java code:\n" + s, e);
    }
}

{
    return org.apache.calcite.linq4j.Linq4j.asEnumerable(new Integer[] {
        0
    });
}

// EnumerableValues#implement() 方法
List<Expression> expressions = RexToLixTranslator.translateProjects(program, typeFactory, conformance, builder3, null, physType, DataContext.ROOT, new RexToLixTranslator.InputGetterImpl(input, result.physType), implementor.allCorrelateVariables);
builder3.add(Expressions.return_(null, physType.record(expressions)));

public List<Expression> translateList(List<? extends RexNode> operandList, @Nullable List<? extends @Nullable Type> storageTypes) {
    final List<Expression> list = new ArrayList<>(operandList.size());
    // 遍历投影列操作数
    for (int i = 0; i < operandList.size(); i++) {
        // 返回 RexLocalRef 类型，内容为 $t5，表示引用 expr#5=[CONCAT_WS($t1, $t2, $t3, $t4)]
        RexNode rex = operandList.get(i);
        Type desiredType = null;
        if (storageTypes != null) {
            desiredType = storageTypes.get(i);
        }
        // 继续调用 translate 方法转换函数
        final Expression translate = translate(rex, desiredType);
        list.add(translate);
        ...
    }
    return list;
}

Expression translate(RexNode expr, RexImpTable.NullAs nullAs, @Nullable Type storageType) {
    currentStorageType = storageType;
    // 使用 RexToLixTranslator 遍历 RexNode，该方法会访问 RexToLixTranslator#visitLocalRef 方法
    final Result result = expr.accept(this);
    final Expression translated = requireNonNull(EnumUtils.toInternal(result.valueVariable, storageType));
    // When we asked for not null input that would be stored as box, avoid unboxing
    if (RexImpTable.NullAs.NOT_POSSIBLE == nullAs && translated.type.equals(storageType)) {
        return translated;
    }
    return nullAs.handle(translated);
}

/**
 * Visit {@code RexCall}. For most {@code SqlOperator}s, we can get the implementor
 * from {@code RexImpTable}. Several operators (e.g., CaseWhen) with special semantics
 * need to be implemented separately.
 */
@Override
public Result visitCall(RexCall call) {
    ...
    // 从 RexImpTable 中获取运算符实现器
    final RexImpTable.RexCallImplementor implementor = RexImpTable.INSTANCE.get(operator);
    if (implementor == null) {
        throw new RuntimeException("cannot translate call " + call);
    }
    // 获取操作数
    final List<RexNode> operandList = call.getOperands();
    final List<@Nullable Type> storageTypes = EnumUtils.internalTypes(operandList);
    final List<Result> operandResults = new ArrayList<>();
    // 遍历操作数，调用各自的 implement 方法生成代码
    for (int i = 0; i < operandList.size(); i++) {
        final Result operandResult = implementCallOperand(operandList.get(i), storageTypes.get(i), this);
        operandResults.add(operandResult);
    }
    callOperandResultMap.put(call, operandResults);
    final Result result = implementor.implement(this, call, operandResults);
    rexResultMap.put(call, result);
    return result;
}

public org.apache.calcite.linq4j.Enumerable bind(final org.apache.calcite.DataContext root) {
    final org.apache.calcite.linq4j.Enumerable _inputEnumerable = org.apache.calcite.linq4j.Linq4j.asEnumerable(new Integer[] {
        0
    });
    return new org.apache.calcite.linq4j.AbstractEnumerable() {
        public org.apache.calcite.linq4j.Enumerator enumerator() {
            return new org.apache.calcite.linq4j.Enumerator() {
                public final org.apache.calcite.linq4j.Enumerator inputEnumerator = _inputEnumerable.enumerator();
                public void reset() {
                    inputEnumerator.reset();
                }

                public boolean moveNext() {
                    return inputEnumerator.moveNext();
                }

                public void close() {
                    inputEnumerator.close();
                }

                public Object current() {
                    return org.apache.calcite.runtime.SqlFunctions.concatMultiWithSeparator(new String[] {
                        ",",
                        "a", (String) null,
                        "b"
                    });
                }
            };
        }
    };
}

public Class getElementType() {
    return java.lang.String.class;
}

@SuppressWarnings("unused")
public class UDFRegistry {
    // 从 content 中查找 target 字符串，并返回其位置
    public static int indexOf(String content, String target) {
        return content.indexOf(target);
    }
}

@Slf4j
public final class UDFExample {
    
    public static void main(String[] args) throws Exception {
        Class.forName("org.apache.calcite.jdbc.Driver");
        try (Connection connection = DriverManager.getConnection("jdbc:calcite:", initProps())) {
            CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class);
            SchemaPlus rootSchema = calciteConnection.getRootSchema();
            Schema schema = createSchema(rootSchema);
            rootSchema.add("calcite_function", schema);
            rootSchema.add("INDEX_OF", Objects.requireNonNull(ScalarFunctionImpl.create(UDFRegistry.class, "indexOf")));
            executeQuery(calciteConnection, "SELECT INDEX_OF(user_name, 'san') FROM calcite_function.t_user");
        }
    }
}

if (call.getOperator() instanceof SqlUnresolvedFunction) {
    assert call instanceof SqlBasicCall;
    final SqlUnresolvedFunction function = (SqlUnresolvedFunction) call.getOperator();
    // This function hasn't been resolved yet.  Perform
    // a half-hearted resolution now in case it's a
    // builtin function requiring special casing.  If it's
    // not, we'll handle it later during overload resolution.
    final List<SqlOperator> overloads = new ArrayList<>();
    opTab.lookupOperatorOverloads(function.getNameAsId(), function.getFunctionType(), SqlSyntax.FUNCTION, overloads, catalogReader.nameMatcher());
    if (overloads.size() == 1) {
        ((SqlBasicCall) call).setOperator(overloads.get(0));
    }
}

if (function instanceof ScalarFunction) {
    final SqlReturnTypeInference returnTypeInference = infer((ScalarFunction) function);
    // 标量函数转换为 SqlUserDefinedFunction
    return new SqlUserDefinedFunction(name, kind, returnTypeInference, operandTypeInference, operandMetadata, function);
} else if (function instanceof AggregateFunction) {
    final SqlReturnTypeInference returnTypeInference = infer((AggregateFunction) function);
    // 聚合函数转换为 SqlUserDefinedAggFunction
    return new SqlUserDefinedAggFunction(name, kind, returnTypeInference, operandTypeInference, operandMetadata, (AggregateFunction) function, false, false, Optionality.FORBIDDEN);
} else if (function instanceof TableMacro) {
    // 表宏转换为 SqlUserDefinedTableMacro
    return new SqlUserDefinedTableMacro(name, kind, ReturnTypes.CURSOR, operandTypeInference, operandMetadata, (TableMacro) function);
} else if (function instanceof TableFunction) {
    // 表函数转换为 SqlUserDefinedTableFunction
    return new SqlUserDefinedTableFunction(name, kind, ReturnTypes.CURSOR, operandTypeInference, operandMetadata, (TableFunction) function);
} else {
    throw new AssertionError("unknown function type " + function);
}

{
    final Object[] current = (Object[]) inputEnumerator.current();
    return com.strongduanmu.udf.UDFRegistry.indexOf(current[1] == null ? null : current[1].toString(), "san");
}

public static LocalDate sysDate() {
    return LocalDate.now();
}

// SqlValidatorImpl#Expander 类 visit 方法
@Override
public @Nullable SqlNode visit(SqlIdentifier id) {
    // First check for builtin functions which don't have
    // parentheses, like "LOCALTIME".
    final SqlCall call = validator.makeNullaryCall(id);
    if (call != null) {
        return call.accept(this);
    }
    final SqlIdentifier fqId = getScope().fullyQualify(id).identifier;
    SqlNode expandedExpr = expandDynamicStar(id, fqId);
    validator.setOriginal(expandedExpr, id);
    return expandedExpr;
}

// SqlValidatorImpl 类 makeNullaryCall 方法
@Override
public @Nullable SqlCall makeNullaryCall(SqlIdentifier id) {
    if (id.names.size() == 1 && !id.isComponentQuoted(0)) {
        final List<SqlOperator> list = new ArrayList<>();
        opTab.lookupOperatorOverloads(id, null, SqlSyntax.FUNCTION, list, catalogReader.nameMatcher());
        for (SqlOperator operator : list) {
            if (operator.getSyntax() == SqlSyntax.FUNCTION_ID) {
                // Even though this looks like an identifier, it is a
                // actually a call to a function. Construct a fake
                // call to this function, so we can use the regular
                // operator validation.
                return new SqlBasicCall(operator, ImmutableList.of(), id.getParserPosition(), null).withExpanded(true);
            }
        }
    }
    return null;
}

if (function instanceof ScalarFunction) {
    final SqlReturnTypeInference returnTypeInference = infer((ScalarFunction) function);
    SqlSyntax syntax = function.getParameters().isEmpty() && config.conformance().allowNiladicParentheses() ? SqlSyntax.FUNCTION : SqlSyntax.FUNCTION_ID;
    return new SqlUserDefinedFunction(name, kind, returnTypeInference, operandTypeInference, operandMetadata, function, syntax);
}

@SuppressWarnings("unused")
public class UDAFRegistry {
    
    public Collection<Object> init() {
        return new LinkedList<>();
    }
    
    public Collection<Object> add(final Collection<Object> accumulator, final Object newElement) {
        accumulator.add(newElement);
        return accumulator;
    }
    
    public Collection<Object> result(final Collection<Object> accumulator) {
        return accumulator;
    }
}

rootSchema.add("CONCAT_TO_LIST", Objects.requireNonNull(AggregateFunctionImpl.create(UDAFRegistry.class)));
CalciteJDBCUtils.executeQuery(calciteConnection, "SELECT CONCAT_TO_LIST(user_name) FROM calcite_function.t_user");

java.util.List accumulatorAdders = new java.util.LinkedList();
accumulatorAdders.add(new org.apache.calcite.linq4j.function.Function2() {
    public Record3_0 apply(Record3_0 acc, Object[] in ) {
        if (!( in [1] == null || in [1].toString() == null)) {
            acc.f2 = true;
            // 调用 add 方法，增加新元数据
            acc.f0 = acc.f1.add(acc.f0, in [1] == null ? null : in [1].toString());
        }
        return acc;
    }
    public Record3_0 apply(Object acc, Object in ) {
        return apply(
            (Record3_0) acc, (Object[]) in );
    }
});
org.apache.calcite.adapter.enumerable.AggregateLambdaFactory lambdaFactory = new org.apache.calcite.adapter.enumerable.BasicAggregateLambdaFactory(
    new org.apache.calcite.linq4j.function.Function0() {
        public Object apply() {
            java.util.Collection a0s0;
            com.strongduanmu.udaf.UDAFRegistry a0s1;
            boolean a0s2;
            a0s2 = false;
            // 创建 UDAFRegistry 对象
            a0s1 = new com.strongduanmu.udaf.UDAFRegistry();
            // 调用 init 方法初始化
            a0s0 = a0s1.init();
            Record3_0 record0;
            // 将聚合函数的累加器、UDF 对象、是否计算完成标记记录到 Record3_0 的 f0、f1 和 f2 中
            record0 = new Record3_0();
            record0.f0 = a0s0;
            record0.f1 = a0s1;
            record0.f2 = a0s2;
            return record0;
        }
    },
    accumulatorAdders);
// 先调用 apply() 初始化，然后调用 accumulatorAdder() 执行聚合逻辑，最后调用 singleGroupResultSelector 获取结果
return org.apache.calcite.linq4j.Linq4j.singletonEnumerable(enumerable.aggregate(lambdaFactory.accumulatorInitializer().apply(), lambdaFactory.accumulatorAdder(), lambdaFactory.singleGroupResultSelector(new org.apache.calcite.linq4j.function.Function1() {
                    public Object apply(Record3_0 acc) {
                        // 判断是否计算完成，完成则调用 result 方法
                        return acc.f2 ? acc.f1.result(acc.f0) : null;
                    }
                    public Object apply(Object acc) {
                        return apply(
                            (Record3_0) acc);
                    }
                }

@SuppressWarnings({"unused", "unchecked"})
public class UDTFRegistry {
    
    public static final Method UDTF_METHOD = Types.lookupMethod(UDTFRegistry.class, "eval", String.class, int.class);
    
    public QueryableTable eval(final String content, final int columnSize) {
        return new AbstractQueryableTable(Object[].class) {
            @Override
            public <T> Queryable<T> asQueryable(final QueryProvider queryProvider, final SchemaPlus schema, final String tableName) {
                return (Queryable<T>) new BaseQueryable<Object[]>(queryProvider, String.class, null) {
                    @Override
                    public Enumerator<Object[]> enumerator() {
                        return new Enumerator<Object[]>() {
                            String[] rows = content.split(",");
                            int index = -1;
                            
                            @Override
                            public Object[] current() {
                                String row = rows[index];
                                Object[] result = new Object[columnSize];
                                String[] columns = row.split("=");
                                Preconditions.checkArgument(columns.length == 2, String.format("Invalid row: %s, row must constains %d columns.", row, columnSize));
                                return columns;
                            }
                            
                            @Override
                            public boolean moveNext() {
                                if (index < rows.length - 1) {
                                    index++;
                                    return true;
                                }
                                return false;
                            }
                            
                            @Override
                            public void reset() {
                            }
                            
                            @Override
                            public void close() {
                            }
                        };
                    }
                };
            }
            
            @Override
            public RelDataType getRowType(final RelDataTypeFactory typeFactory) {
                return typeFactory.createStructType(Arrays.asList(typeFactory.createJavaType(String.class), typeFactory.createJavaType(String.class)), Arrays.asList("key", "value"));
            }
        };
    }
}

rootSchema.add("EXTRACT_FROM_CSV", Objects.requireNonNull(TableFunctionImpl.create(UDTFRegistry.UDTF_METHOD)));
CalciteJDBCUtils.executeQuery(calciteConnection, "SELECT * FROM EXTRACT_FROM_CSV('java=1,go=2,scala=3', 2)");

public org.apache.calcite.linq4j.Enumerable bind(final org.apache.calcite.DataContext root) {
    final com.strongduanmu.udtf.UDTFRegistry f = new com.strongduanmu.udtf.UDTFRegistry();
    return f.eval("java=1,go=2,scala=3", 2).asQueryable(root.getQueryProvider(), (org.apache.calcite.schema.SchemaPlus) null, "EXTRACT_FROM_CSV").asEnumerable();
}

public Class getElementType() {
    return java.lang.Object[].class;
}