You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by zh...@apache.org on 2026/04/17 08:23:38 UTC
(auron) branch master updated: [AURON #1859] Convert math operators to Auron native operators (#2167)
This is an automated email from the ASF dual-hosted git repository.
zhangmang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/auron.git
The following commit(s) were added to refs/heads/master by this push:
new 0be0e799 [AURON #1859] Convert math operators to Auron native operators (#2167)
0be0e799 is described below
commit 0be0e7994738df08ddd87554ca023452f03c6341
Author: Weiqing Yang <wi...@linkedin.com>
AuthorDate: Fri Apr 17 01:23:33 2026 -0700
[AURON #1859] Convert math operators to Auron native operators (#2167)
# Which issue does this PR close?
Closes #1859
# Rationale for this change
Next step in the Flink integration track (#1264). After #1856
established the converter framework (`FlinkRexNodeConverter`,
`FlinkNodeConverterFactory`, `ConverterContext`), this PR adds the first
concrete converter implementations that translate Flink/Calcite
`RexNode` expressions into Auron native `PhysicalExprNode` protobuf
representations. Required before #1857 (`FlinkAuronCalcOperator`) can
wire them into the execution pipeline.
# What changes are included in this PR?
Three new `FlinkRexNodeConverter` implementations (3 commits, 6 new
files, 1 modified):
**Commit 1 — `RexInputRefConverter`**
- Converts `RexInputRef` column references to `PhysicalColumn` (name +
index)
- Resolves column names from the input schema via `ConverterContext`
**Commit 2 — `RexLiteralConverter`**
- Converts `RexLiteral` scalar values to `ScalarValue` with Arrow IPC
bytes
- Supports: TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DECIMAL,
BOOLEAN, CHAR, VARCHAR, NULL
- Serializes via single-element Arrow vector → `ArrowStreamWriter` →
`ipc_bytes` field
**Commit 3 — `RexCallConverter`**
- Converts `RexCall` operator expressions dispatched by `SqlKind`
- Arithmetic: `+`, `-`, `*`, `/`, `%` → `PhysicalBinaryExprNode` with
explicit type promotion
- Unary: `-` → `PhysicalNegativeNode`, `+` → identity passthrough
- CAST → `PhysicalTryCastNode`
- Type promotion via `getCommonTypeForComparison()` +
`castIfNecessary()` (per reviewer PoC)
- Recursive operand conversion via `FlinkNodeConverterFactory`
**pom.xml** — Added `**/*Test.java` to surefire includes for the planner
module
**Key design decisions** (per @Tartarus0zm PoC in
https://github.com/apache/auron/issues/1859#issuecomment-4181565860):
- `PhysicalTryCastNode` (not `PhysicalCastNode`) for all type casts
- Explicit type promotion — does not rely solely on Calcite upstream
CASTs
- Output type cast when compatible type differs from declared return
type
- Division by zero handling deferred to #1857 (requires Rust-side
`Flink_NullIfZero`)
# Are there any user-facing changes?
No.
# How was this patch tested?
33 unit tests across 3 test classes:
- `RexInputRefConverterTest` — 4 tests (node class, isSupported,
first/second column)
- `RexLiteralConverterTest` — 12 tests (all supported types, null
literal, unsupported type rejection)
- `RexCallConverterTest` — 17 tests (all 5 arithmetic ops, unary
minus/plus, cast, mixed-type promotion, output-type cast, nested
expressions, isSupported checks, `getCommonTypeForComparison` direct
tests)
---
auron-flink-extension/auron-flink-planner/pom.xml | 1 +
.../planner/converter/FlinkNodeConverterUtils.java | 126 +++++++++++
.../table/planner/converter/RexCallConverter.java | 197 +++++++++++++++++
.../planner/converter/RexInputRefConverter.java | 75 +++++++
.../planner/converter/RexLiteralConverter.java | 201 +++++++++++++++++
.../converter/FlinkNodeConverterUtilsTest.java | 123 +++++++++++
.../planner/converter/RexCallConverterTest.java | 238 +++++++++++++++++++++
.../converter/RexInputRefConverterTest.java | 93 ++++++++
.../planner/converter/RexLiteralConverterTest.java | 171 +++++++++++++++
9 files changed, 1225 insertions(+)
diff --git a/auron-flink-extension/auron-flink-planner/pom.xml b/auron-flink-extension/auron-flink-planner/pom.xml
index bce3ae82..4f2a9b18 100644
--- a/auron-flink-extension/auron-flink-planner/pom.xml
+++ b/auron-flink-extension/auron-flink-planner/pom.xml
@@ -303,6 +303,7 @@
<configuration>
<includes>
<include>**/*ITCase.java</include>
+ <include>**/*Test.java</include>
</includes>
</configuration>
</plugin>
diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterUtils.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterUtils.java
new file mode 100644
index 00000000..5c989cdd
--- /dev/null
+++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterUtils.java
@@ -0,0 +1,126 @@
+/*
+ * 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.auron.flink.table.planner.converter;
+
+import org.apache.auron.flink.utils.SchemaConverters;
+import org.apache.auron.protobuf.PhysicalExprNode;
+import org.apache.auron.protobuf.PhysicalTryCastNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeSystem;
+import org.apache.calcite.sql.type.SqlTypeFactoryImpl;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.type.SqlTypeUtil;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.types.logical.LogicalType;
+
+/**
+ * Shared utility helpers for Flink {@code RelNode} / {@code RexNode} converters.
+ *
+ * <p>Currently hosts type-promotion and cast-wrapping helpers used during
+ * {@link RexCallConverter} expression conversion. New helpers shared across
+ * Flink node converters should be added here.
+ */
+public final class FlinkNodeConverterUtils {
+
+ /** Shared type factory used for creating common SQL types. */
+ public static final RelDataTypeFactory TYPE_FACTORY = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
+
+ private FlinkNodeConverterUtils() {
+ // utility class
+ }
+
+ /**
+ * Computes the common type for two operand types during arithmetic
+ * promotion.
+ *
+ * <p>Rules:
+ * <ul>
+ * <li>Same type → return as-is
+ * <li>Both numeric: DECIMAL wins; BIGINT wins over smaller integers
+ * (when neither is approximate); otherwise DOUBLE
+ * <li>Both character → VARCHAR
+ * <li>Otherwise → {@code null} (incompatible)
+ * </ul>
+ *
+ * @param type1 left operand type
+ * @param type2 right operand type
+ * @param typeFactory factory for creating result types
+ * @return the common type, or {@code null} if incompatible
+ */
+ public static RelDataType getCommonTypeForComparison(
+ RelDataType type1, RelDataType type2, RelDataTypeFactory typeFactory) {
+ if (type1.getSqlTypeName().equals(type2.getSqlTypeName())) {
+ return type1;
+ }
+ if (SqlTypeUtil.isNumeric(type1) && SqlTypeUtil.isNumeric(type2)) {
+ SqlTypeName t1 = type1.getSqlTypeName();
+ SqlTypeName t2 = type2.getSqlTypeName();
+ if (t1 == SqlTypeName.DECIMAL || t2 == SqlTypeName.DECIMAL) {
+ return typeFactory.createSqlType(SqlTypeName.DECIMAL);
+ }
+ if (notApproxType(t1) && notApproxType(t2)) {
+ return typeFactory.createSqlType(SqlTypeName.BIGINT);
+ }
+ return typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ }
+ if (SqlTypeUtil.inCharFamily(type1) && SqlTypeUtil.inCharFamily(type2)) {
+ return typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ }
+ return null;
+ }
+
+ /**
+ * Wraps the given expression in a {@link PhysicalTryCastNode} targeting
+ * the specified type if the source type differs from the target type.
+ *
+ * <p>If {@code sourceType} and {@code targetType} share the same
+ * {@link SqlTypeName}, the original expression is returned unchanged.
+ *
+ * @param converted the already-converted native expression
+ * @param sourceType the current SQL type of the expression
+ * @param targetType the desired SQL type
+ * @return {@code converted} unchanged, or wrapped in a TryCast
+ */
+ public static PhysicalExprNode castIfNecessary(
+ PhysicalExprNode converted, RelDataType sourceType, RelDataType targetType) {
+ if (sourceType.getSqlTypeName().equals(targetType.getSqlTypeName())) {
+ return converted;
+ }
+ return wrapInTryCast(converted, targetType);
+ }
+
+ /**
+ * Wraps the given native expression in a {@link PhysicalTryCastNode}
+ * targeting the specified Calcite type.
+ *
+ * @param expr the native expression to wrap
+ * @param targetType the desired output type
+ * @return a new {@link PhysicalExprNode} containing a {@code TryCast}
+ */
+ public static PhysicalExprNode wrapInTryCast(PhysicalExprNode expr, RelDataType targetType) {
+ LogicalType logicalType = FlinkTypeFactory.toLogicalType(targetType);
+ org.apache.auron.protobuf.ArrowType arrowType = SchemaConverters.convertToAuronArrowType(logicalType);
+ return PhysicalExprNode.newBuilder()
+ .setTryCast(PhysicalTryCastNode.newBuilder().setExpr(expr).setArrowType(arrowType))
+ .build();
+ }
+
+ private static boolean notApproxType(SqlTypeName typeName) {
+ return typeName != SqlTypeName.FLOAT && typeName != SqlTypeName.DOUBLE;
+ }
+}
diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java
new file mode 100644
index 00000000..ecdada75
--- /dev/null
+++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java
@@ -0,0 +1,197 @@
+/*
+ * 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.auron.flink.table.planner.converter;
+
+import java.util.EnumSet;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.auron.protobuf.PhysicalBinaryExprNode;
+import org.apache.auron.protobuf.PhysicalExprNode;
+import org.apache.auron.protobuf.PhysicalNegativeNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.SqlTypeUtil;
+
+/**
+ * Converts a Calcite {@link RexCall} (operator expression) to an Auron native
+ * {@link PhysicalExprNode}.
+ *
+ * <p>Handles arithmetic operators ({@code +}, {@code -}, {@code *}, {@code /},
+ * {@code %}), unary minus/plus, and {@code CAST}. Binary arithmetic operands
+ * are promoted to a common type before conversion, and the result is cast to
+ * the output type if it differs from the common type.
+ */
+public class RexCallConverter implements FlinkRexNodeConverter {
+
+ /** Binary arithmetic kinds that require numeric result type. */
+ private static final Set<SqlKind> BINARY_ARITHMETIC_KINDS =
+ EnumSet.of(SqlKind.PLUS, SqlKind.MINUS, SqlKind.TIMES, SqlKind.DIVIDE, SqlKind.MOD);
+
+ /** All supported SqlKinds including unary and cast. */
+ private static final Set<SqlKind> SUPPORTED_KINDS = EnumSet.of(
+ SqlKind.PLUS,
+ SqlKind.MINUS,
+ SqlKind.TIMES,
+ SqlKind.DIVIDE,
+ SqlKind.MOD,
+ SqlKind.MINUS_PREFIX,
+ SqlKind.PLUS_PREFIX,
+ SqlKind.CAST);
+
+ private final FlinkNodeConverterFactory factory;
+
+ /**
+ * Creates a new converter that delegates operand conversion to the given
+ * factory.
+ *
+ * @param factory the factory used for recursive operand conversion
+ */
+ public RexCallConverter(FlinkNodeConverterFactory factory) {
+ this.factory = factory;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public Class<? extends RexNode> getNodeClass() {
+ return RexCall.class;
+ }
+
+ /**
+ * Returns {@code true} if the call's {@link SqlKind} is supported.
+ *
+ * <p>For binary arithmetic kinds, the call's result type must also be
+ * numeric to reject non-arithmetic uses (e.g., TIMESTAMP + INTERVAL).
+ */
+ @Override
+ public boolean isSupported(RexNode node, ConverterContext context) {
+ RexCall call = (RexCall) node;
+ SqlKind kind = call.getKind();
+ if (!SUPPORTED_KINDS.contains(kind)) {
+ return false;
+ }
+ if (BINARY_ARITHMETIC_KINDS.contains(kind)) {
+ return SqlTypeUtil.isNumeric(call.getType());
+ }
+ return true;
+ }
+
+ /**
+ * Converts the given {@link RexCall} to a native {@link PhysicalExprNode}.
+ *
+ * <p>Dispatches by {@link SqlKind}:
+ * <ul>
+ * <li>Binary arithmetic → {@link PhysicalBinaryExprNode} with type
+ * promotion
+ * <li>{@code MINUS_PREFIX} → {@link PhysicalNegativeNode}
+ * <li>{@code PLUS_PREFIX} → identity (passthrough to operand)
+ * <li>{@code CAST} → {@link org.apache.auron.protobuf.PhysicalTryCastNode}
+ * </ul>
+ *
+ * @throws IllegalArgumentException if the SqlKind is not supported
+ */
+ @Override
+ public PhysicalExprNode convert(RexNode node, ConverterContext context) {
+ RexCall call = (RexCall) node;
+ SqlKind kind = call.getKind();
+ switch (kind) {
+ case PLUS:
+ return buildBinaryExpr(call, "Plus", context);
+ case MINUS:
+ return buildBinaryExpr(call, "Minus", context);
+ case TIMES:
+ return buildBinaryExpr(call, "Multiply", context);
+ case DIVIDE:
+ return buildBinaryExpr(call, "Divide", context);
+ case MOD:
+ return buildBinaryExpr(call, "Modulo", context);
+ case MINUS_PREFIX:
+ return buildNegative(call, context);
+ case PLUS_PREFIX:
+ return convertOperand(call.getOperands().get(0), context);
+ case CAST:
+ return buildTryCast(call, context);
+ default:
+ throw new IllegalArgumentException("Unsupported SqlKind: " + kind);
+ }
+ }
+
+ /**
+ * Builds a binary expression with type promotion between operands.
+ *
+ * <p>Operands are promoted to a common type. If the call's output type
+ * differs from the common type, the result is wrapped in a TryCast.
+ */
+ private PhysicalExprNode buildBinaryExpr(RexCall call, String op, ConverterContext context) {
+ RexNode left = call.getOperands().get(0);
+ RexNode right = call.getOperands().get(1);
+ RelDataType outputType = call.getType();
+
+ RelDataType compatibleType = FlinkNodeConverterUtils.getCommonTypeForComparison(
+ left.getType(), right.getType(), FlinkNodeConverterUtils.TYPE_FACTORY);
+ if (compatibleType == null) {
+ throw new IllegalStateException("Incompatible types: "
+ + left.getType().getSqlTypeName()
+ + " and "
+ + right.getType().getSqlTypeName());
+ }
+
+ PhysicalExprNode leftExpr =
+ FlinkNodeConverterUtils.castIfNecessary(convertOperand(left, context), left.getType(), compatibleType);
+ PhysicalExprNode rightExpr = FlinkNodeConverterUtils.castIfNecessary(
+ convertOperand(right, context), right.getType(), compatibleType);
+
+ PhysicalExprNode binaryExpr = PhysicalExprNode.newBuilder()
+ .setBinaryExpr(PhysicalBinaryExprNode.newBuilder()
+ .setL(leftExpr)
+ .setR(rightExpr)
+ .setOp(op))
+ .build();
+
+ if (!outputType.getSqlTypeName().equals(compatibleType.getSqlTypeName())) {
+ return FlinkNodeConverterUtils.wrapInTryCast(binaryExpr, outputType);
+ }
+ return binaryExpr;
+ }
+
+ /**
+ * Delegates operand conversion to the factory.
+ *
+ * @throws IllegalStateException if no converter is registered for
+ * the operand
+ */
+ private PhysicalExprNode convertOperand(RexNode operand, ConverterContext context) {
+ Optional<PhysicalExprNode> result = factory.convertRexNode(operand, context);
+ if (!result.isPresent()) {
+ throw new IllegalStateException("Failed to convert operand: " + operand + " (no converter registered)");
+ }
+ return result.get();
+ }
+
+ private PhysicalExprNode buildNegative(RexCall call, ConverterContext context) {
+ PhysicalExprNode operand = convertOperand(call.getOperands().get(0), context);
+ return PhysicalExprNode.newBuilder()
+ .setNegative(PhysicalNegativeNode.newBuilder().setExpr(operand))
+ .build();
+ }
+
+ private PhysicalExprNode buildTryCast(RexCall call, ConverterContext context) {
+ PhysicalExprNode operand = convertOperand(call.getOperands().get(0), context);
+ return FlinkNodeConverterUtils.wrapInTryCast(operand, call.getType());
+ }
+}
diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexInputRefConverter.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexInputRefConverter.java
new file mode 100644
index 00000000..b65a8b4d
--- /dev/null
+++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexInputRefConverter.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.auron.flink.table.planner.converter;
+
+import org.apache.auron.protobuf.PhysicalColumn;
+import org.apache.auron.protobuf.PhysicalExprNode;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+
+/**
+ * Converts a Calcite {@link RexInputRef} (column reference) to an Auron native {@link
+ * PhysicalExprNode} containing a {@link PhysicalColumn}.
+ *
+ * <p>Column references are supported when the index is within the input schema bounds. Every valid
+ * {@code RexInputRef} maps directly to a named, indexed column in the input schema provided by the
+ * {@link ConverterContext}.
+ */
+public class RexInputRefConverter implements FlinkRexNodeConverter {
+
+ /** {@inheritDoc} */
+ @Override
+ public Class<? extends RexNode> getNodeClass() {
+ return RexInputRef.class;
+ }
+
+ /**
+ * Returns {@code true} if the column index is within the input schema bounds.
+ *
+ * @param node the RexNode to check (must be a {@link RexInputRef})
+ * @param context shared conversion state
+ * @return {@code true} if the index is valid
+ */
+ @Override
+ public boolean isSupported(RexNode node, ConverterContext context) {
+ RexInputRef inputRef = (RexInputRef) node;
+ return inputRef.getIndex() < context.getInputType().getFieldCount();
+ }
+
+ /**
+ * Converts the given {@link RexInputRef} to a {@link PhysicalExprNode} with a {@link
+ * PhysicalColumn}.
+ *
+ * <p>Resolves the column name from the input schema via {@link
+ * ConverterContext#getInputType()}.
+ *
+ * @param node the RexNode to convert (must be a {@link RexInputRef})
+ * @param context shared conversion state containing the input schema
+ * @return a {@link PhysicalExprNode} wrapping a {@link PhysicalColumn} with name and index
+ * @throws IllegalArgumentException if the node is not a {@link RexInputRef}
+ */
+ @Override
+ public PhysicalExprNode convert(RexNode node, ConverterContext context) {
+ RexInputRef inputRef = (RexInputRef) node;
+ int index = inputRef.getIndex();
+ String name = context.getInputType().getFieldNames().get(index);
+
+ return PhysicalExprNode.newBuilder()
+ .setColumn(PhysicalColumn.newBuilder().setName(name).setIndex(index))
+ .build();
+ }
+}
diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverter.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverter.java
new file mode 100644
index 00000000..90621a03
--- /dev/null
+++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverter.java
@@ -0,0 +1,201 @@
+/*
+ * 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.auron.flink.table.planner.converter;
+
+import com.google.protobuf.ByteString;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.util.EnumSet;
+import java.util.Set;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowStreamWriter;
+import org.apache.auron.flink.arrow.FlinkArrowUtils;
+import org.apache.auron.flink.arrow.FlinkArrowWriter;
+import org.apache.auron.protobuf.PhysicalExprNode;
+import org.apache.auron.protobuf.ScalarValue;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.BooleanType;
+import org.apache.flink.table.types.logical.CharType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.DoubleType;
+import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.SmallIntType;
+import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+/**
+ * Converts a Calcite {@link RexLiteral} to an Auron native {@link PhysicalExprNode}
+ * containing a {@link ScalarValue} with Arrow IPC bytes.
+ *
+ * <p>The literal value is serialized as a single-element Arrow record batch in IPC stream
+ * format, using Flink's {@link GenericRowData} and {@link FlinkArrowWriter} to perform
+ * type-aware conversion via the project's shared {@link FlinkArrowUtils} infrastructure.
+ *
+ * <p>Supported types: {@code TINYINT}, {@code SMALLINT}, {@code INTEGER}, {@code BIGINT},
+ * {@code FLOAT}, {@code DOUBLE}, {@code DECIMAL}, {@code BOOLEAN}, {@code CHAR},
+ * {@code VARCHAR}, and {@code NULL} (of a supported type).
+ */
+public class RexLiteralConverter implements FlinkRexNodeConverter {
+
+ private static final Set<SqlTypeName> SUPPORTED_TYPES = EnumSet.of(
+ SqlTypeName.TINYINT,
+ SqlTypeName.SMALLINT,
+ SqlTypeName.INTEGER,
+ SqlTypeName.BIGINT,
+ SqlTypeName.FLOAT,
+ SqlTypeName.DOUBLE,
+ SqlTypeName.DECIMAL,
+ SqlTypeName.BOOLEAN,
+ SqlTypeName.CHAR,
+ SqlTypeName.VARCHAR);
+
+ /** {@inheritDoc} */
+ @Override
+ public Class<? extends RexNode> getNodeClass() {
+ return RexLiteral.class;
+ }
+
+ /**
+ * Returns {@code true} if the literal's SQL type is supported for native conversion.
+ *
+ * <p>For null literals, the underlying type is still checked — a null of an unsupported
+ * type (e.g., TIMESTAMP) returns {@code false}.
+ */
+ @Override
+ public boolean isSupported(RexNode node, ConverterContext context) {
+ RexLiteral literal = (RexLiteral) node;
+ SqlTypeName typeName = literal.getType().getSqlTypeName();
+ return isSupportedType(typeName);
+ }
+
+ /**
+ * Converts the given {@link RexLiteral} to a {@link PhysicalExprNode} with Arrow IPC bytes.
+ *
+ * @throws IllegalArgumentException if the literal type is not supported
+ */
+ @Override
+ public PhysicalExprNode convert(RexNode node, ConverterContext context) {
+ RexLiteral literal = (RexLiteral) node;
+ byte[] ipcBytes = serializeToIpc(literal);
+ return PhysicalExprNode.newBuilder()
+ .setLiteral(ScalarValue.newBuilder().setIpcBytes(ByteString.copyFrom(ipcBytes)))
+ .build();
+ }
+
+ private static boolean isSupportedType(SqlTypeName typeName) {
+ return SUPPORTED_TYPES.contains(typeName);
+ }
+
+ /**
+ * Serializes the literal value as a single-element Arrow record batch in IPC stream format.
+ *
+ * <p>Uses Flink's {@link GenericRowData} and {@link FlinkArrowWriter} for type-aware
+ * value conversion, delegating all type-to-Arrow mapping to {@link FlinkArrowUtils}.
+ */
+ private static byte[] serializeToIpc(RexLiteral literal) {
+ LogicalType logicalType = FlinkTypeFactory.toLogicalType(literal.getType());
+ RowType rowType = RowType.of(logicalType);
+
+ try (BufferAllocator allocator =
+ FlinkArrowUtils.getRootAllocator().newChildAllocator("literal", 0, Long.MAX_VALUE);
+ VectorSchemaRoot root = VectorSchemaRoot.create(FlinkArrowUtils.toArrowSchema(rowType), allocator)) {
+
+ GenericRowData rowData = new GenericRowData(1);
+ rowData.setField(0, toFlinkInternalValue(literal, logicalType));
+
+ FlinkArrowWriter writer = FlinkArrowWriter.create(root, rowType);
+ writer.write(rowData);
+ writer.finish();
+
+ return writeIpcBytes(root);
+ } catch (IOException e) {
+ throw new IllegalStateException("Failed to serialize literal to Arrow IPC", e);
+ }
+ }
+
+ /**
+ * Converts a {@link RexLiteral} value to its Flink internal RowData representation.
+ *
+ * <p>Returns {@code null} for null literals. For DECIMAL, returns {@link DecimalData};
+ * for CHAR/VARCHAR, returns {@link StringData}. For numeric and boolean types, returns
+ * the correctly-typed boxed Java value that {@link FlinkArrowWriter} expects.
+ *
+ * @param literal the Calcite literal to convert
+ * @param logicalType the Flink logical type of the literal
+ * @return the Flink internal value, or {@code null} if the literal is null
+ * @throws IllegalArgumentException if the logical type is not supported
+ */
+ private static Object toFlinkInternalValue(RexLiteral literal, LogicalType logicalType) {
+ if (literal.isNull()) {
+ return null;
+ }
+ if (logicalType instanceof TinyIntType) {
+ return literal.getValueAs(Byte.class);
+ }
+ if (logicalType instanceof SmallIntType) {
+ return literal.getValueAs(Short.class);
+ }
+ if (logicalType instanceof IntType) {
+ return literal.getValueAs(Integer.class);
+ }
+ if (logicalType instanceof BigIntType) {
+ return literal.getValueAs(Long.class);
+ }
+ if (logicalType instanceof FloatType) {
+ return literal.getValueAs(Float.class);
+ }
+ if (logicalType instanceof DoubleType) {
+ return literal.getValueAs(Double.class);
+ }
+ if (logicalType instanceof BooleanType) {
+ return literal.getValueAs(Boolean.class);
+ }
+ if (logicalType instanceof DecimalType) {
+ DecimalType dt = (DecimalType) logicalType;
+ return DecimalData.fromBigDecimal(literal.getValueAs(BigDecimal.class), dt.getPrecision(), dt.getScale());
+ }
+ if (logicalType instanceof CharType || logicalType instanceof VarCharType) {
+ return StringData.fromString(literal.getValueAs(String.class));
+ }
+ throw new IllegalArgumentException("Unsupported logical type: " + logicalType.asSummaryString());
+ }
+
+ /**
+ * Writes the given {@link VectorSchemaRoot} to Arrow IPC stream format.
+ */
+ private static byte[] writeIpcBytes(VectorSchemaRoot root) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ try (ArrowStreamWriter writer = new ArrowStreamWriter(root, null, out)) {
+ writer.start();
+ writer.writeBatch();
+ writer.end();
+ }
+ return out.toByteArray();
+ }
+}
diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterUtilsTest.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterUtilsTest.java
new file mode 100644
index 00000000..db35d032
--- /dev/null
+++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterUtilsTest.java
@@ -0,0 +1,123 @@
+/*
+ * 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.auron.flink.table.planner.converter;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.auron.protobuf.PhysicalExprNode;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.junit.jupiter.api.Test;
+
+/** Tests for {@link FlinkNodeConverterUtils}. */
+class FlinkNodeConverterUtilsTest {
+
+ private static final RelDataTypeFactory TYPE_FACTORY = new JavaTypeFactoryImpl();
+
+ @Test
+ void testCommonTypeDecimalWins() {
+ RelDataType intType = TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER);
+ RelDataType decType = TYPE_FACTORY.createSqlType(SqlTypeName.DECIMAL, 10, 2);
+ RelDataType result = FlinkNodeConverterUtils.getCommonTypeForComparison(intType, decType, TYPE_FACTORY);
+
+ assertEquals(SqlTypeName.DECIMAL, result.getSqlTypeName());
+ }
+
+ @Test
+ void testCommonTypeExactIntegerPromotesToBigint() {
+ // TINYINT + INTEGER → BIGINT (both exact, promoted to widest exact type)
+ RelDataType tinyintType = TYPE_FACTORY.createSqlType(SqlTypeName.TINYINT);
+ RelDataType integerType = TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER);
+ RelDataType result = FlinkNodeConverterUtils.getCommonTypeForComparison(tinyintType, integerType, TYPE_FACTORY);
+
+ assertEquals(SqlTypeName.BIGINT, result.getSqlTypeName());
+ }
+
+ @Test
+ void testCommonTypeApproxFallbackToDouble() {
+ // INT + FLOAT → DOUBLE (FLOAT is approx, so exact integer rule skipped)
+ RelDataType intType = TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER);
+ RelDataType floatType = TYPE_FACTORY.createSqlType(SqlTypeName.FLOAT);
+ RelDataType result = FlinkNodeConverterUtils.getCommonTypeForComparison(intType, floatType, TYPE_FACTORY);
+
+ assertEquals(SqlTypeName.DOUBLE, result.getSqlTypeName());
+ }
+
+ @Test
+ void testCommonTypeIncompatible() {
+ // BOOLEAN + INTEGER → null (incompatible)
+ RelDataType boolType = TYPE_FACTORY.createSqlType(SqlTypeName.BOOLEAN);
+ RelDataType intType = TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER);
+ RelDataType result = FlinkNodeConverterUtils.getCommonTypeForComparison(boolType, intType, TYPE_FACTORY);
+
+ assertNull(result, "Incompatible types should return null");
+ }
+
+ @Test
+ void testCommonTypeSameTypeReturnsAsIs() {
+ RelDataType intType = TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER);
+ RelDataType result = FlinkNodeConverterUtils.getCommonTypeForComparison(intType, intType, TYPE_FACTORY);
+
+ assertSame(intType, result, "Same type should return the first operand's type");
+ }
+
+ @Test
+ void testCommonTypeCharFamilyProducesVarchar() {
+ RelDataType charType = TYPE_FACTORY.createSqlType(SqlTypeName.CHAR, 10);
+ RelDataType varcharType = TYPE_FACTORY.createSqlType(SqlTypeName.VARCHAR, 20);
+ RelDataType result = FlinkNodeConverterUtils.getCommonTypeForComparison(charType, varcharType, TYPE_FACTORY);
+
+ assertEquals(SqlTypeName.VARCHAR, result.getSqlTypeName());
+ }
+
+ @Test
+ void testCastIfNecessarySameTypePassthrough() {
+ PhysicalExprNode expr = PhysicalExprNode.getDefaultInstance();
+ RelDataType intType = TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER);
+
+ PhysicalExprNode result = FlinkNodeConverterUtils.castIfNecessary(expr, intType, intType);
+
+ assertSame(expr, result, "Same type should return the original expression unchanged");
+ }
+
+ @Test
+ void testCastIfNecessaryDifferentTypeWrapsTryCast() {
+ PhysicalExprNode expr = PhysicalExprNode.getDefaultInstance();
+ RelDataType intType = TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER);
+ RelDataType bigintType = TYPE_FACTORY.createSqlType(SqlTypeName.BIGINT);
+
+ PhysicalExprNode result = FlinkNodeConverterUtils.castIfNecessary(expr, intType, bigintType);
+
+ assertTrue(result.hasTryCast(), "Different types should wrap in TryCast");
+ }
+
+ @Test
+ void testWrapInTryCast() {
+ PhysicalExprNode expr = PhysicalExprNode.getDefaultInstance();
+ RelDataType bigintType = TYPE_FACTORY.createSqlType(SqlTypeName.BIGINT);
+
+ PhysicalExprNode result = FlinkNodeConverterUtils.wrapInTryCast(expr, bigintType);
+
+ assertTrue(result.hasTryCast(), "Should produce a TryCast node");
+ assertTrue(result.getTryCast().hasArrowType(), "TryCast should have an ArrowType");
+ }
+}
diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java
new file mode 100644
index 00000000..04fb1617
--- /dev/null
+++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java
@@ -0,0 +1,238 @@
+/*
+ * 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.auron.flink.table.planner.converter;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.math.BigDecimal;
+import java.util.Arrays;
+import org.apache.auron.protobuf.PhysicalExprNode;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+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.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Tests for {@link RexCallConverter}. */
+class RexCallConverterTest {
+
+ private static final RelDataTypeFactory TYPE_FACTORY = new JavaTypeFactoryImpl();
+ private static final RexBuilder REX_BUILDER = new RexBuilder(TYPE_FACTORY);
+
+ private FlinkNodeConverterFactory factory;
+ private RexCallConverter converter;
+ private ConverterContext context;
+
+ @BeforeEach
+ void setUp() {
+ factory = new FlinkNodeConverterFactory();
+ converter = new RexCallConverter(factory);
+ factory.registerRexConverter(new RexInputRefConverter());
+ factory.registerRexConverter(new RexLiteralConverter());
+ factory.registerRexConverter(converter);
+
+ RowType inputType = RowType.of(new LogicalType[] {new IntType(), new BigIntType()}, new String[] {"f0", "f1"});
+ context = new ConverterContext(new Configuration(), null, getClass().getClassLoader(), inputType);
+ }
+
+ @Test
+ void testGetNodeClass() {
+ assertEquals(RexCall.class, converter.getNodeClass());
+ }
+
+ @Test
+ void testConvertPlus() {
+ RexNode plus = makeCall(intType(), SqlStdOperatorTable.PLUS, makeIntRef(0), makeIntRef(0));
+
+ PhysicalExprNode result = converter.convert(plus, context);
+
+ assertTrue(result.hasBinaryExpr());
+ assertEquals("Plus", result.getBinaryExpr().getOp());
+ }
+
+ @Test
+ void testConvertMinus() {
+ RexNode minus = makeCall(intType(), SqlStdOperatorTable.MINUS, makeIntRef(0), makeIntRef(0));
+
+ PhysicalExprNode result = converter.convert(minus, context);
+
+ assertTrue(result.hasBinaryExpr());
+ assertEquals("Minus", result.getBinaryExpr().getOp());
+ }
+
+ @Test
+ void testConvertTimes() {
+ RexNode times = makeCall(intType(), SqlStdOperatorTable.MULTIPLY, makeIntRef(0), makeIntRef(0));
+
+ PhysicalExprNode result = converter.convert(times, context);
+
+ assertTrue(result.hasBinaryExpr());
+ assertEquals("Multiply", result.getBinaryExpr().getOp());
+ }
+
+ @Test
+ void testConvertDivide() {
+ RexNode divide = makeCall(intType(), SqlStdOperatorTable.DIVIDE, makeIntRef(0), makeIntRef(0));
+
+ PhysicalExprNode result = converter.convert(divide, context);
+
+ assertTrue(result.hasBinaryExpr());
+ assertEquals("Divide", result.getBinaryExpr().getOp());
+ }
+
+ @Test
+ void testConvertMod() {
+ RexNode mod = makeCall(intType(), SqlStdOperatorTable.MOD, makeIntRef(0), makeIntRef(0));
+
+ PhysicalExprNode result = converter.convert(mod, context);
+
+ assertTrue(result.hasBinaryExpr());
+ assertEquals("Modulo", result.getBinaryExpr().getOp());
+ }
+
+ @Test
+ void testConvertUnaryMinus() {
+ RexNode neg = REX_BUILDER.makeCall(SqlStdOperatorTable.UNARY_MINUS, makeIntRef(0));
+
+ PhysicalExprNode result = converter.convert(neg, context);
+
+ assertTrue(result.hasNegative());
+ assertTrue(result.getNegative().getExpr().hasColumn());
+ }
+
+ @Test
+ void testConvertUnaryPlus() {
+ RexNode pos = REX_BUILDER.makeCall(SqlStdOperatorTable.UNARY_PLUS, makeIntRef(0));
+
+ PhysicalExprNode result = converter.convert(pos, context);
+
+ // Unary plus is identity — passthrough to operand
+ assertTrue(result.hasColumn());
+ assertEquals("f0", result.getColumn().getName());
+ }
+
+ @Test
+ void testConvertCast() {
+ RexNode cast = makeCall(bigintType(), SqlStdOperatorTable.CAST, makeIntRef(0));
+
+ PhysicalExprNode result = converter.convert(cast, context);
+
+ assertTrue(result.hasTryCast());
+ assertTrue(result.getTryCast().getExpr().hasColumn());
+ assertTrue(result.getTryCast().hasArrowType());
+ }
+
+ @Test
+ void testConvertMixedTypePromotion() {
+ // INT (f0) + BIGINT (f1) — left operand should be promoted
+ RexNode intRef = makeIntRef(0);
+ RexNode bigintRef = REX_BUILDER.makeInputRef(bigintType(), 1);
+ RexNode mixedPlus = makeCall(bigintType(), SqlStdOperatorTable.PLUS, intRef, bigintRef);
+
+ PhysicalExprNode result = converter.convert(mixedPlus, context);
+
+ assertTrue(result.hasBinaryExpr());
+ assertEquals("Plus", result.getBinaryExpr().getOp());
+ // Left operand (INT) should be wrapped in TryCast to BIGINT
+ PhysicalExprNode left = result.getBinaryExpr().getL();
+ assertTrue(left.hasTryCast(), "Left operand should be cast from INT to BIGINT");
+ // Right operand (BIGINT) should be plain column
+ PhysicalExprNode right = result.getBinaryExpr().getR();
+ assertTrue(right.hasColumn(), "Right operand should be plain column (already BIGINT)");
+ }
+
+ @Test
+ void testConvertOutputTypeCast() {
+ // INT + INT where output type is BIGINT → result wrapped in TryCast
+ RexNode plus = makeCall(bigintType(), SqlStdOperatorTable.PLUS, makeIntRef(0), makeIntRef(0));
+
+ PhysicalExprNode result = converter.convert(plus, context);
+
+ // Both operands are INT, compatible type is INT,
+ // but output type is BIGINT → outer TryCast
+ assertTrue(result.hasTryCast(), "Result should be wrapped in TryCast when output " + "!= compatible type");
+ assertTrue(result.getTryCast().getExpr().hasBinaryExpr());
+ }
+
+ @Test
+ void testConvertNestedExpr() {
+ // (f0 + 1) * f0
+ RexNode f0 = makeIntRef(0);
+ RexNode one = REX_BUILDER.makeExactLiteral(BigDecimal.ONE, intType());
+ RexNode innerPlus = makeCall(intType(), SqlStdOperatorTable.PLUS, f0, one);
+ RexNode outer = makeCall(intType(), SqlStdOperatorTable.MULTIPLY, innerPlus, makeIntRef(0));
+
+ PhysicalExprNode result = converter.convert(outer, context);
+
+ assertTrue(result.hasBinaryExpr());
+ assertEquals("Multiply", result.getBinaryExpr().getOp());
+ // Left child is the inner (f0 + 1)
+ PhysicalExprNode leftChild = result.getBinaryExpr().getL();
+ assertTrue(leftChild.hasBinaryExpr());
+ assertEquals("Plus", leftChild.getBinaryExpr().getOp());
+ }
+
+ @Test
+ void testIsSupportedNumericArithmetic() {
+ RexNode plus = makeCall(intType(), SqlStdOperatorTable.PLUS, makeIntRef(0), makeIntRef(0));
+
+ assertTrue(converter.isSupported(plus, context));
+ }
+
+ @Test
+ void testIsNotSupportedNonNumericKind() {
+ // EQUALS is not in the supported set
+ RexNode eq = REX_BUILDER.makeCall(SqlStdOperatorTable.EQUALS, makeIntRef(0), makeIntRef(0));
+
+ assertFalse(converter.isSupported(eq, context));
+ }
+
+ // ---- Helpers ----
+
+ private static RexNode makeIntRef(int index) {
+ return REX_BUILDER.makeInputRef(intType(), index);
+ }
+
+ private static RelDataType intType() {
+ return TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER);
+ }
+
+ private static RelDataType bigintType() {
+ return TYPE_FACTORY.createSqlType(SqlTypeName.BIGINT);
+ }
+
+ /**
+ * Creates a {@link org.apache.calcite.rex.RexCall} with an explicit
+ * return type using the List-based {@code makeCall} overload.
+ */
+ private static RexNode makeCall(
+ RelDataType returnType, org.apache.calcite.sql.SqlOperator op, RexNode... operands) {
+ return REX_BUILDER.makeCall(returnType, op, Arrays.asList(operands));
+ }
+}
diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexInputRefConverterTest.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexInputRefConverterTest.java
new file mode 100644
index 00000000..db22cf5c
--- /dev/null
+++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexInputRefConverterTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.auron.flink.table.planner.converter;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.auron.protobuf.PhysicalExprNode;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Tests for {@link RexInputRefConverter}. */
+class RexInputRefConverterTest {
+
+ private static final RelDataTypeFactory TYPE_FACTORY = new JavaTypeFactoryImpl();
+ private static final RexBuilder REX_BUILDER = new RexBuilder(TYPE_FACTORY);
+
+ private RexInputRefConverter converter;
+ private ConverterContext context;
+
+ @BeforeEach
+ void setUp() {
+ converter = new RexInputRefConverter();
+ RowType inputType = RowType.of(new LogicalType[] {new IntType(), new BigIntType()}, new String[] {"f0", "f1"});
+ context = new ConverterContext(new Configuration(), null, getClass().getClassLoader(), inputType);
+ }
+
+ @Test
+ void testGetNodeClass() {
+ assertEquals(RexInputRef.class, converter.getNodeClass());
+ }
+
+ @Test
+ void testIsSupportedValidIndex() {
+ RexNode inputRef = REX_BUILDER.makeInputRef(TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER), 0);
+ assertTrue(converter.isSupported(inputRef, context));
+ }
+
+ @Test
+ void testIsNotSupportedOutOfRangeIndex() {
+ // Schema has 2 fields (f0, f1) — index 5 is out of range
+ RexNode inputRef = REX_BUILDER.makeInputRef(TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER), 5);
+ assertFalse(converter.isSupported(inputRef, context));
+ }
+
+ @Test
+ void testConvertFirstColumn() {
+ RexNode inputRef = REX_BUILDER.makeInputRef(TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER), 0);
+
+ PhysicalExprNode result = converter.convert(inputRef, context);
+
+ assertTrue(result.hasColumn());
+ assertEquals("f0", result.getColumn().getName());
+ assertEquals(0, result.getColumn().getIndex());
+ }
+
+ @Test
+ void testConvertSecondColumn() {
+ RexNode inputRef = REX_BUILDER.makeInputRef(TYPE_FACTORY.createSqlType(SqlTypeName.BIGINT), 1);
+
+ PhysicalExprNode result = converter.convert(inputRef, context);
+
+ assertTrue(result.hasColumn());
+ assertEquals("f1", result.getColumn().getName());
+ assertEquals(1, result.getColumn().getIndex());
+ }
+}
diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverterTest.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverterTest.java
new file mode 100644
index 00000000..b408272b
--- /dev/null
+++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverterTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.auron.flink.table.planner.converter;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.math.BigDecimal;
+import org.apache.auron.protobuf.PhysicalExprNode;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Tests for {@link RexLiteralConverter}. */
+class RexLiteralConverterTest {
+
+ private static final RelDataTypeFactory TYPE_FACTORY = new JavaTypeFactoryImpl();
+ private static final RexBuilder REX_BUILDER = new RexBuilder(TYPE_FACTORY);
+
+ private RexLiteralConverter converter;
+ private ConverterContext context;
+
+ @BeforeEach
+ void setUp() {
+ converter = new RexLiteralConverter();
+ context =
+ new ConverterContext(new Configuration(), null, getClass().getClassLoader(), RowType.of(new IntType()));
+ }
+
+ @Test
+ void testGetNodeClass() {
+ assertEquals(RexLiteral.class, converter.getNodeClass());
+ }
+
+ @Test
+ void testConvertTinyIntLiteral() {
+ RexLiteral lit = (RexLiteral)
+ REX_BUILDER.makeExactLiteral(BigDecimal.valueOf(7), TYPE_FACTORY.createSqlType(SqlTypeName.TINYINT));
+
+ assertTrue(converter.isSupported(lit, context));
+ PhysicalExprNode result = converter.convert(lit, context);
+ assertTrue(result.hasLiteral());
+ assertTrue(result.getLiteral().getIpcBytes().size() > 0);
+ }
+
+ @Test
+ void testConvertSmallIntLiteral() {
+ RexLiteral lit = (RexLiteral)
+ REX_BUILDER.makeExactLiteral(BigDecimal.valueOf(256), TYPE_FACTORY.createSqlType(SqlTypeName.SMALLINT));
+
+ assertTrue(converter.isSupported(lit, context));
+ PhysicalExprNode result = converter.convert(lit, context);
+ assertTrue(result.hasLiteral());
+ assertTrue(result.getLiteral().getIpcBytes().size() > 0);
+ }
+
+ @Test
+ void testConvertIntLiteral() {
+ RexLiteral intLit = (RexLiteral)
+ REX_BUILDER.makeExactLiteral(BigDecimal.valueOf(42), TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER));
+
+ assertTrue(converter.isSupported(intLit, context));
+ PhysicalExprNode result = converter.convert(intLit, context);
+ assertTrue(result.hasLiteral());
+ assertTrue(result.getLiteral().getIpcBytes().size() > 0);
+ }
+
+ @Test
+ void testConvertLongLiteral() {
+ RexLiteral longLit = (RexLiteral) REX_BUILDER.makeExactLiteral(
+ BigDecimal.valueOf(123456789L), TYPE_FACTORY.createSqlType(SqlTypeName.BIGINT));
+
+ assertTrue(converter.isSupported(longLit, context));
+ PhysicalExprNode result = converter.convert(longLit, context);
+ assertTrue(result.hasLiteral());
+ assertTrue(result.getLiteral().getIpcBytes().size() > 0);
+ }
+
+ @Test
+ void testConvertFloatLiteral() {
+ RexLiteral lit = (RexLiteral)
+ REX_BUILDER.makeApproxLiteral(BigDecimal.valueOf(2.5f), TYPE_FACTORY.createSqlType(SqlTypeName.FLOAT));
+
+ assertTrue(converter.isSupported(lit, context));
+ PhysicalExprNode result = converter.convert(lit, context);
+ assertTrue(result.hasLiteral());
+ assertTrue(result.getLiteral().getIpcBytes().size() > 0);
+ }
+
+ @Test
+ void testConvertDoubleLiteral() {
+ RexLiteral doubleLit = (RexLiteral)
+ REX_BUILDER.makeApproxLiteral(BigDecimal.valueOf(3.14), TYPE_FACTORY.createSqlType(SqlTypeName.DOUBLE));
+
+ assertTrue(converter.isSupported(doubleLit, context));
+ PhysicalExprNode result = converter.convert(doubleLit, context);
+ assertTrue(result.hasLiteral());
+ assertTrue(result.getLiteral().getIpcBytes().size() > 0);
+ }
+
+ @Test
+ void testConvertBooleanTrue() {
+ RexLiteral boolLit = (RexLiteral) REX_BUILDER.makeLiteral(true);
+
+ assertTrue(converter.isSupported(boolLit, context));
+ PhysicalExprNode result = converter.convert(boolLit, context);
+ assertTrue(result.hasLiteral());
+ assertTrue(result.getLiteral().getIpcBytes().size() > 0);
+ }
+
+ @Test
+ void testConvertStringLiteral() {
+ RexLiteral strLit = (RexLiteral) REX_BUILDER.makeLiteral("hello");
+
+ assertTrue(converter.isSupported(strLit, context));
+ PhysicalExprNode result = converter.convert(strLit, context);
+ assertTrue(result.hasLiteral());
+ assertTrue(result.getLiteral().getIpcBytes().size() > 0);
+ }
+
+ @Test
+ void testConvertDecimalLiteral() {
+ RexLiteral decLit = (RexLiteral) REX_BUILDER.makeExactLiteral(
+ new BigDecimal("123.45"), TYPE_FACTORY.createSqlType(SqlTypeName.DECIMAL, 10, 2));
+
+ assertTrue(converter.isSupported(decLit, context));
+ PhysicalExprNode result = converter.convert(decLit, context);
+ assertTrue(result.hasLiteral());
+ assertTrue(result.getLiteral().getIpcBytes().size() > 0);
+ }
+
+ @Test
+ void testConvertNullLiteral() {
+ RexLiteral nullLit = (RexLiteral) REX_BUILDER.makeNullLiteral(TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER));
+
+ assertTrue(converter.isSupported(nullLit, context));
+ PhysicalExprNode result = converter.convert(nullLit, context);
+ assertTrue(result.hasLiteral());
+ assertTrue(result.getLiteral().getIpcBytes().size() > 0);
+ }
+
+ @Test
+ void testUnsupportedTypeNotSupported() {
+ RexNode tsLit = REX_BUILDER.makeNullLiteral(TYPE_FACTORY.createSqlType(SqlTypeName.TIMESTAMP));
+
+ assertFalse(converter.isSupported(tsLit, context));
+ }
+}