[AURON #1853] Convert Flink Calc operators to Native Calc operators

auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterFactory.java

@@ -37,6 +37,10 @@
  *       (keyed by {@code AggregateCall} class)
  * </ul>
  *
+ * <p>The singleton returned by {@link #getInstance()} is initialized with the built-in
+ * {@link FlinkRexNodeConverter} implementations ({@link RexInputRefConverter},
+ * {@link RexLiteralConverter}, {@link RexCallConverter}) registered.
+ *
  * <p>Usage:
  * <pre>
  *   FlinkNodeConverterFactory factory = FlinkNodeConverterFactory.getInstance();
@@ -52,6 +56,14 @@ public class FlinkNodeConverterFactory {
 
     private static final FlinkNodeConverterFactory INSTANCE = new FlinkNodeConverterFactory();
 
+    static {
+        // Production callers reach RexNode converters through the singleton; register the
+        // built-ins eagerly so getInstance() is usable without setup.
+        INSTANCE.registerRexConverter(new RexInputRefConverter());
+        INSTANCE.registerRexConverter(new RexLiteralConverter());
+        INSTANCE.registerRexConverter(new RexCallConverter(INSTANCE));
+    }
+
     private final Map<Class<? extends RexNode>, FlinkRexNodeConverter> rexConverterMap;
     private final Map<Class<? extends AggregateCall>, FlinkAggCallConverter> aggConverterMap;
 

auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalcWarnState.java

@@ -0,0 +1,69 @@
+/*
+ * 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.flink.table.planner.plan.nodes.exec.stream;
+
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.calcite.rex.RexNode;
+import org.apache.flink.annotation.VisibleForTesting;
+
+/**
+ * Holds the per-fallback WARN dedup state for the shadowed {@link StreamExecCalc}. Lives in a
+ * non-shadowed sibling class so unit tests can call the {@link VisibleForTesting} seams directly
+ * — the shadow shares its FQCN with Flink's stock {@code StreamExecCalc}, which can confuse javac
+ * when test sources reference symbols that exist only on the shadow.
+ *
+ * <p>Dedup is JVM-wide: a given unsupported {@link RexNode} class is logged at most once per JVM,
+ * bounded by the small finite set of {@link RexNode} subclasses Flink can generate.
+ */
+final class StreamExecCalcWarnState {
+
+    private static final Set<Class<? extends RexNode>> WARN_DEDUP = ConcurrentHashMap.newKeySet();
+    private static final AtomicInteger WARN_EMIT_COUNT = new AtomicInteger();
+
+    private StreamExecCalcWarnState() {}
+
+    /**
+     * Marks an unsupported {@link RexNode} class as seen and increments the emit counter. Returns
+     * {@code true} on the first occurrence of the class (caller should emit the WARN line);
+     * {@code false} on subsequent occurrences (caller should be silent).
+     */
+    static boolean recordFallback(Class<? extends RexNode> unsupportedRexClass) {
+        if (WARN_DEDUP.add(unsupportedRexClass)) {
+            WARN_EMIT_COUNT.incrementAndGet();
+            return true;
+        }
+        return false;
+    }
+
+    /** Increments the emit counter for plan-composition failures, which always log. */
+    static void recordCompositionFailure() {
+        WARN_EMIT_COUNT.incrementAndGet();
+    }
+
+    @VisibleForTesting
+    static void resetForTest() {
+        WARN_DEDUP.clear();
+        WARN_EMIT_COUNT.set(0);
+    }
+
+    @VisibleForTesting
+    static int peekEmitCount() {
+        return WARN_EMIT_COUNT.get();
+    }
+}

auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterFactoryTest.java

@@ -27,6 +27,8 @@
 import org.apache.calcite.rel.core.AggregateCall;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
@@ -143,6 +145,22 @@ void testGetConverterAbsent() {
         assertFalse(found.isPresent());
     }
 
+    @Test
+    void testSingletonHasBuiltInRexConvertersRegistered() {
+        FlinkNodeConverterFactory singleton = FlinkNodeConverterFactory.getInstance();
+
+        Optional<FlinkNodeConverter<?>> inputRefConverter = singleton.getConverter(RexInputRef.class);
+        Optional<FlinkNodeConverter<?>> literalConverter = singleton.getConverter(RexLiteral.class);
+        Optional<FlinkNodeConverter<?>> callConverter = singleton.getConverter(RexCall.class);
+
+        assertTrue(inputRefConverter.isPresent());
+        assertTrue(literalConverter.isPresent());
+        assertTrue(callConverter.isPresent());
+        assertTrue(inputRefConverter.get() instanceof RexInputRefConverter);
+        assertTrue(literalConverter.get() instanceof RexLiteralConverter);
+        assertTrue(callConverter.get() instanceof RexCallConverter);
+    }
+
     // ---- Test stubs ----
 
     /** Stub FlinkRexNodeConverter for testing. */

auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronCalcRewriteITCase.java

@@ -0,0 +1,83 @@
+/*
+ * 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.runtime;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import org.apache.auron.flink.table.AuronFlinkTableTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+import org.junit.jupiter.api.Test;
+
+/**
+ * End-to-end IT cases for the shadowed {@code StreamExecCalc}. Each test submits a real SQL job
+ * through {@link org.apache.flink.table.api.bridge.java.StreamTableEnvironment} over the {@code T1}
+ * table registered in {@link AuronFlinkTableTestBase} and asserts the final row set is correct
+ * regardless of whether the Calc executed natively or fell back to Flink's codegen.
+ */
+public class AuronCalcRewriteITCase extends AuronFlinkTableTestBase {
+
+    /** Multi-column arithmetic projection exercises the projection loop with more than one
+     * convertible expression. */
+    @Test
+    public void testMultiColumnArithmeticProjection() {
+        List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
+                .executeSql("select `int` + 1, `int` * 2 from T1")
+                .collect());
+        rows.sort(Comparator.comparingInt(o -> (int) o.getField(0)));
+        assertThat(rows).isEqualTo(Arrays.asList(Row.of(2, 2), Row.of(3, 4), Row.of(3, 4)));
+    }
+
+    /** A filter-plus-projection Calc whose condition uses a not-yet-supported comparison operator
+     * falls back to Flink's codegen path. Asserts the job still produces correct results; the
+     * Auron-side {@code Filter[FFIReader]} plan-shape coverage will be added when a
+     * predicate-returning converter lands. */
+    @Test
+    public void testFilterAndProjectEndToEnd() {
+        List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
+                .executeSql("select `int` * 2 from T1 where `int` > 1")
+                .collect());
+        rows.sort(Comparator.comparingInt(o -> (int) o.getField(0)));
+        assertThat(rows).isEqualTo(Arrays.asList(Row.of(4), Row.of(4)));
+    }
+
+    /** Unsupported expression (a string function not in the converter set) triggers silent
+     * fallback. The job must still complete and emit the correct rows. */
+    @Test
+    public void testFallbackOnUnsupportedExprStillExecutes() {
+        List<Row> rows = CollectionUtil.iteratorToList(
+                tableEnvironment.executeSql("select UPPER(`string`) from T1").collect());
+        rows.sort(Comparator.comparing(o -> (String) o.getField(0)));
+        assertThat(rows).isEqualTo(Arrays.asList(Row.of("COMMENT#1"), Row.of("COMMENT#1"), Row.of("HI")));
+    }
+
+    /** A job containing two Calcs — one whose expressions are all converter-supported and one
+     * that uses an unsupported function — must run end-to-end and emit the correct union of rows.
+     * This asserts the job-level correctness contract; observability of which Calc fell back is
+     * surfaced through the per-fallback WARN log rather than the test's value assertion. */
+    @Test
+    public void testMixedSupportedAndUnsupportedCalcs() {
+        List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
+                .executeSql("select `int` + 1 from T1 union all select CHAR_LENGTH(`string`) from T1")
+                .collect());
+        rows.sort(Comparator.comparingInt(o -> (int) o.getField(0)));
+        assertThat(rows).isEqualTo(Arrays.asList(Row.of(2), Row.of(2), Row.of(3), Row.of(3), Row.of(9), Row.of(9)));
+    }
+}