Extend native Calc function coverage with released Flink semantics

docs/connectors/paimon.md

@@ -273,8 +273,11 @@ or `table.optimizer.reuse-sub-plan-enabled` restores independent native readers.
 
 `PaimonSourceSharingTest` checks three-sink projection union, identical projections, repeated
 compilation, disabled sharing, distinct scans, mixed native/Flink branches, and snapshot-to-tail
-results including deletes and nulls for both formats, with chained and network edges. It also
-checks window closure on every shared branch after a new commit.
+results including deletes and nulls for both formats, with chained and network edges. The admitted
+ABS projection participates in three-consumer sharing; a STRING-to-BOOLEAN TRY_CAST projection
+retains a separate host branch and two native consumers. Both shapes execute snapshot/tail
+comparisons against Flink. The suite also checks window closure on every shared branch after a
+new commit.
 Paimon's unchanged `ContinuousFileStoreITCase.testSourceReuseWithScanPushDown` passes its `Reused`
 assertion and its filter/limit separation assertions. Cross-sink sharing requires the deployed
 planner hook; installing a program into an already constructed stock planner still optimizes each

docs/flink-compatibility.md

@@ -73,6 +73,10 @@ The following are host-line differences, not missing native substitutions:
   session windows remain a separate supported planner construct.
 - Several newer scalar functions and `TO_TIMESTAMP_LTZ` string/default-precision overloads are
   absent from 1.18. Tests mark only those specific host constructs N/A.
+  The function-coverage regressions specifically exclude BTRIM, binary STARTSWITH/ENDSWITH/ELT,
+  PRINTF, REGEXP_COUNT/INSTR/SUBSTR and REGEXP_EXTRACT_ALL on that released line. STR_TO_MAP,
+  Base64 decoding, exact numerics, extrema and the other available forms still run against 1.18's
+  own implementation rather than borrowing 2.2 behavior.
 - The 1.18 host planner cannot consume update/delete streams in window TVF aggregation. Those
   SQL parity cases are N/A; native retraction handling still has operator-level tests.
 - 1.18 has released host code-generation defects for nullable `TINYINT`/`SMALLINT` array lookup,
@@ -84,6 +88,15 @@ The following are host-line differences, not missing native substitutions:
 on Flink because the native boundary cannot treat arbitrary bytes as a one-byte fixed vector.
 The fallback parity tests check the complete bytes, including UTF-16 encodings and nulls.
 
+Scalar SQL regression tests assert native routing, resolved output schemas and collected results
+for admitted exact-numeric ABS, Java-backed string extrema and dynamic trims, BOOLEAN IF,
+PARSE_URL, dynamic SHA2, FROM_BASE64 and the released binary prefix/suffix overloads.
+The remaining floating-extrema, binary-backed string, oversized SHA2/ELT and fixed BINARY-result
+restrictions retain explicit fallback checks. Generic Calc/filter rejection fixtures use
+STRING-to-BOOLEAN TRY_CAST, which still falls back, rather than functions already admitted by
+the planner. A fallback assertion failure does not establish a result mismatch: routing and
+host/native result parity both need to pass.
+
 ## JSON and formats
 
 The shared nested ARRAY/ROW JSON parity fixtures use each release line's collection-source

src/main/java/tech/streamfusion/planner/CalcOutputTypeCheck.java

@@ -61,18 +61,26 @@ static String mismatch(RexExpression encoded, RelDataType inputType, RowType dec
       if (notInferable != null) {
         return notInferable;
       }
-      return mismatch(Data.importSchema(allocator, outputSchema, null).getFields(), declared);
+      return mismatch(Data.importSchema(allocator, outputSchema, null).getFields(), declared, encoded);
     } catch (NativeException compileFailure) {
       return "expression does not compile natively: " + compileFailure.getMessage();
     }
   }
 
-  private static String mismatch(List<Field> inferred, RowType declared) {
+  private static String mismatch(List<Field> inferred, RowType declared, RexExpression encoded) {
     if (inferred.size() != declared.getFieldCount()) {
       return inferred.size() + " projections for " + declared.getFieldCount() + " declared columns";
     }
     for (int i = 0; i < inferred.size(); i++) {
       Field actual = inferred.get(i);
+      if (encoded.isBinaryStringProjection(i)
+          && actual.getType() instanceof org.apache.arrow.vector.types.pojo.ArrowType.Binary
+          && (declared.getTypeAt(i).getTypeRoot()
+                  == org.apache.flink.table.types.logical.LogicalTypeRoot.VARCHAR
+              || declared.getTypeAt(i).getTypeRoot()
+                  == org.apache.flink.table.types.logical.LogicalTypeRoot.CHAR)) {
+        continue;
+      }
       if (!ArrowConversion.readsAs(actual, declared.getTypeAt(i))) {
         return String.format(
             "projection `%s` evaluates natively as %s but the plan declares %s",

src/main/java/tech/streamfusion/planner/FlinkExpressionFunction.java

@@ -45,8 +45,17 @@ public interface Evaluator extends Function {
       LogicalType[] argumentTypes,
       ReadableConfig config,
       ClassLoader classLoader) {
+    this(expression, argumentTypes, config, classLoader, false);
+  }
+
+  FlinkExpressionFunction(
+      RexNode expression,
+      LogicalType[] argumentTypes,
+      ReadableConfig config,
+      ClassLoader classLoader,
+      boolean binaryStringResult) {
     this(
-        scalarBody(expression, argumentTypes, config, classLoader),
+        scalarBody(expression, argumentTypes, config, classLoader, binaryStringResult),
         argumentTypes,
         config,
         classLoader);
@@ -72,7 +81,8 @@ private static Body scalarBody(
       RexNode expression,
       LogicalType[] argumentTypes,
       ReadableConfig config,
-      ClassLoader classLoader) {
+      ClassLoader classLoader,
+      boolean binaryStringResult) {
     var context = new Context(config, classLoader);
     var generator = new ExprCodeGenerator(context, false);
     generator.bindInput(RowType.of(argumentTypes), "input", scala.Option.empty());
@@ -85,6 +95,7 @@ private static Body scalarBody(
             + result.nullTerm()
             + ") { return null; }\nreturn "
             + result.resultTerm()
+            + (binaryStringResult ? ".toBytes()" : "")
             + ";\n",
         null);
   }

src/main/java/tech/streamfusion/planner/HostStringInputs.java

@@ -0,0 +1,45 @@
+package tech.streamfusion.planner;
+
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDataStreamScan;
+import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+
+/** Proof for functions whose Flink behavior depends on Java-backed versus binary-backed strings. */
+final class HostStringInputs {
+  private HostStringInputs() {}
+
+  static boolean areJavaBacked(RelNode input) {
+    if (input instanceof StreamPhysicalDataStreamScan scan) {
+      return hasExternalStrings(scan.dataStreamTable().dataType());
+    }
+    if (!(input instanceof TableScan scan)) return false;
+    TableSourceTable table = scan.getTable().unwrap(TableSourceTable.class);
+    if (table == null) return false;
+    Object source = table.tableSource();
+    if (!source.getClass().getName().equals(
+        "org.apache.flink.table.planner.connectors.ExternalDynamicSource")) return false;
+    try {
+      // ExternalDynamicSource is package-private and exposes no physical-conversion accessor.
+      var field = source.getClass().getDeclaredField("physicalDataType");
+      if (!field.trySetAccessible()) return false;
+      return hasExternalStrings((DataType) field.get(source));
+    } catch (ReflectiveOperationException | RuntimeException unavailable) {
+      return false;
+    }
+  }
+
+  private static boolean hasExternalStrings(DataType type) {
+    if (org.apache.flink.table.data.RowData.class.isAssignableFrom(type.getConversionClass())
+        || org.apache.flink.table.data.ArrayData.class.isAssignableFrom(type.getConversionClass())
+        || org.apache.flink.table.data.MapData.class.isAssignableFrom(type.getConversionClass())) {
+      return false;
+    }
+    if (type.getLogicalType().is(LogicalTypeFamily.CHARACTER_STRING)) {
+      return type.getConversionClass() == String.class;
+    }
+    return type.getChildren().stream().allMatch(HostStringInputs::hasExternalStrings);
+  }
+}

src/main/java/tech/streamfusion/planner/JsonStringIdentity.java

@@ -10,7 +10,7 @@
 import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc;
 import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalSink;
 
-/** Keeps Java UTF-16 JSON results inside a generated expression or at the final output boundary. */
+/** Keeps sensitive Java UTF-16 results inside a generated expression or at the final output boundary. */
 final class JsonStringIdentity {
   private JsonStringIdentity() {}
 
@@ -24,10 +24,14 @@ static boolean containsSensitiveString(RexNode expression) {
       return containsSensitiveString(access.getReferenceExpr());
     }
     if (!(expression instanceof RexCall call)) return false;
+    if (call.getOperator().getName().equals("REGEXP_EXTRACT_ALL")
+        || call.getOperator().getName().equals("STR_TO_MAP")) return true;
     if (SqlTypeFamily.CHARACTER.contains(call.getType())
         && (call.getOperator().getName().equals("JSON_VALUE")
             || call.getOperator().getName().equals("JSON_QUERY")
-            || call.getOperator().getName().equals("JSON_UNQUOTE"))) {
+            || call.getOperator().getName().equals("JSON_UNQUOTE")
+            || call.getOperator().getName().equals("PRINTF")
+            || call.getOperator().getName().equals("FROM_BASE64"))) {
       return true;
     }
     return call.getOperands().stream().anyMatch(JsonStringIdentity::containsSensitiveString);
@@ -67,7 +71,27 @@ private static boolean crossesOperatorBoundary(
     return false;
   }
 
-  private static boolean containsCharacter(RelDataType type) {
+  static boolean containsBinaryString(RexNode expression) {
+    if (expression instanceof RexFieldAccess access) {
+      return containsBinaryString(access.getReferenceExpr());
+    }
+    return expression instanceof RexCall call
+        && (call.getOperator().getName().equals("FROM_BASE64")
+            || call.getOperands().stream().anyMatch(JsonStringIdentity::containsBinaryString));
+  }
+
+  static boolean projectsBinaryString(RelNode node) {
+    if (node instanceof StreamPhysicalCalc calc) {
+      var program = calc.getProgram();
+      for (var projection : program.getProjectList()) {
+        RexNode expression = program.expandLocalRef(projection);
+        if (containsCharacter(expression.getType()) && containsBinaryString(expression)) return true;
+      }
+    }
+    return node.getInputs().stream().anyMatch(JsonStringIdentity::projectsBinaryString);
+  }
+
+  static boolean containsCharacter(RelDataType type) {
     if (SqlTypeFamily.CHARACTER.contains(type)) return true;
     if (type.getComponentType() != null && containsCharacter(type.getComponentType())) return true;
     if (type.getKeyType() != null && containsCharacter(type.getKeyType())) return true;

src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java

@@ -160,6 +160,13 @@ private RelNode substitute(RelNode root, Set<String> repeatedSources, boolean fi
     // Pass 1 substitutes native (columnar) operators.
     int previousSubstitutions = substitutions;
     RelNode substituted = rewrite(root, new PlanContext(this, repeatedSources));
+    if (root instanceof org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalSink
+        && substituted instanceof ColumnarInput
+        && JsonStringIdentity.projectsBinaryString(root)) {
+      substitutions = previousSubstitutions;
+      recordFallback("binary-backed STRING requires a row sink");
+      return root;
+    }
     // Whole-query all-or-nothing: every native operator but a source/sink is Arrow → Arrow.
     // If any operator other than a source (a leaf) or the sink (the plan root) is still row-wise,
     // the