From 452714e74bc8a87dbc78e201c1a3f3cef81eb206 Mon Sep 17 00:00:00 2001
From: chengjiechengjie <chengjiechengjie@didiglobal.com>
Date: Sat, 27 Sep 2025 23:28:09 +0800
Subject: [PATCH 1/2] =?UTF-8?q?[flink-table-planner]=20Optimize=20RemoveUn?=
 =?UTF-8?q?reachableCoalesceArgumentsRule=20rule=EF=BC=8Cwhen=20including?=
 =?UTF-8?q?=20functions,=20it=20will=20fall=20back=20to=20case=20when?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 ...emoveUnreachableCoalesceArgumentsRule.java | 60 +++++++++++++++++++
 1 file changed, 60 insertions(+)

diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/RemoveUnreachableCoalesceArgumentsRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/RemoveUnreachableCoalesceArgumentsRule.java
index 47cde317256e9..218174e30681d 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/RemoveUnreachableCoalesceArgumentsRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/RemoveUnreachableCoalesceArgumentsRule.java
@@ -23,6 +23,8 @@
 import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
 import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
 
+import org.apache.flink.shaded.guava33.com.google.common.collect.ImmutableList;
+
 import org.apache.calcite.plan.RelOptRuleCall;
 import org.apache.calcite.plan.RelRule;
 import org.apache.calcite.rel.RelNode;
@@ -32,9 +34,12 @@
 import org.apache.calcite.rel.core.Project;
 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.rex.RexShuttle;
 import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.immutables.value.Value;
 
 import java.util.List;
@@ -100,6 +105,10 @@ public RexNode visitCall(RexCall call) {
                 }
             }
 
+            if (containsFunctionCalls(call)) {
+                return convertToCaseWhen(call);
+            }
+
             // If it's the last argument, or no non-null argument was found, return the original
             // call
             if (firstNonNullableArgIndex == call.operands.size() - 1
@@ -121,6 +130,57 @@ private int getFirstNonNullableArgumentIndex(RexCall call) {
             }
             return -1;
         }
+
+        private RexNode convertToCaseWhen(RexCall call) {
+            List<RexNode> operands = call.operands;
+
+            if (operands.size() == 1) {
+                return operands.get(0);
+            }
+
+            RexBuilder rexBuilder = this.rexBuilder;
+            ImmutableList.Builder<RexNode> caseArgs = ImmutableList.builder();
+
+            for (int i = 0; i < operands.size() - 1; i++) {
+                RexNode operand = operands.get(i);
+                RexNode isNotNullCheck =
+                        rexBuilder.makeCall(SqlStdOperatorTable.IS_NOT_NULL, operand);
+                caseArgs.add(isNotNullCheck);
+                caseArgs.add(operand);
+            }
+
+            caseArgs.add(operands.get(operands.size() - 1));
+
+            return rexBuilder.makeCall(SqlStdOperatorTable.CASE, caseArgs.build());
+        }
+
+        private boolean containsFunctionCalls(RexCall call) {
+            for (RexNode operand : call.operands) {
+                if (isFunctionCall(operand)) {
+                    return true;
+                }
+            }
+            return false;
+        }
+
+        private boolean isFunctionCall(RexNode node) {
+            return node instanceof RexCall
+                    && !isSimpleCast((RexCall) node)
+                    && !isFieldReference(node)
+                    && !isLiteral(node);
+        }
+
+        private boolean isSimpleCast(RexCall call) {
+            return call.getOperator().getName().equalsIgnoreCase("CAST");
+        }
+
+        private boolean isFieldReference(RexNode node) {
+            return node instanceof RexInputRef;
+        }
+
+        private boolean isLiteral(RexNode node) {
+            return node instanceof RexLiteral;
+        }
     }
 
     private static boolean hasCoalesceInvocation(RexNode node) {

From 9aa2b3dc80b62fa0c71e38d8369709a27db18d68 Mon Sep 17 00:00:00 2001
From: chengjiechengjie <chengjiechengjie@didiglobal.com>
Date: Wed, 1 Oct 2025 22:47:52 +0800
Subject: [PATCH 2/2] [FLINK-38468][flink-table-planner] When the coalesce
 parameter contains a function, convert it to case when

---
 .../logical/RemoveUnreachableCoalesceArgumentsRuleTest.java  | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/RemoveUnreachableCoalesceArgumentsRuleTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/RemoveUnreachableCoalesceArgumentsRuleTest.java
index 3610b20d6f3a5..0ebbbbc0ad0d5 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/RemoveUnreachableCoalesceArgumentsRuleTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/RemoveUnreachableCoalesceArgumentsRuleTest.java
@@ -84,6 +84,11 @@ void testJoinCoalesce() {
                 "SELECT * FROM T t1 LEFT JOIN T t2 ON COALESCE(t1.f0, '-', t1.f2) = t2.f0");
     }
 
+    @Test
+    void testFunctionCoalesce() {
+        util.verifyRelPlan("SELECT COALESCE(f0, JSON_VALUE('{\"a\": true}', '$.a')) FROM T");
+    }
+
     @Test
     void testMultipleCoalesces() {
         util.verifyRelPlan(