[cdc] Support computed columns when sync_database (#5816)

Author: JackeyLee007

docs/content/cdc-ingestion/kafka-cdc.md

@@ -214,6 +214,7 @@ To use this feature through `flink run`, run the following shell command.
     [--type_mapping to-string] \
     [--partition_keys <partition_keys>] \
     [--primary_keys <primary-keys>] \
+    [--computed_column <'column-name=expr-name(args[, ...])'> [--computed_column ...]] \
     [--kafka_conf <kafka-source-conf> [--kafka_conf <kafka-source-conf> ...]] \
     [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
     [--table_conf <paimon-table-sink-conf> [--table_conf <paimon-table-sink-conf> ...]]
@@ -244,7 +245,8 @@ Synchronization from one Kafka topic to Paimon database.
     --catalog_conf uri=thrift://hive-metastore:9083 \
     --table_conf bucket=4 \
     --table_conf changelog-producer=input \
-    --table_conf sink.parallelism=4
+    --table_conf sink.parallelism=4 \
+    --computed_column 'pt=date_format(event_tm, yyyyMMdd)'
 ```
 
 Synchronization from multiple Kafka topics to Paimon database.

docs/layouts/shortcodes/generated/kafka_sync_database.html

@@ -92,7 +92,7 @@
     </tr>
     <tr>
         <td><h5>--computed_column</h5></td>
-        <td>The definitions of computed columns. The argument field is from Kafka topic's table field name. See <a href="../overview/#computed-functions">here</a> for a complete list of configurations. </td>
+        <td>The definitions of computed columns. The argument field is from Kafka topic's table field name. See <a href="../overview/#computed-functions">here</a> for a complete list of configurations. NOTICE: It returns null if the referenced column does not exist in the source table.</td>
     </tr>
     <tr>
         <td><h5>--eager_init</h5></td>

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumn.java

@@ -53,6 +53,11 @@ public String fieldReference() {
         return expression.fieldReference();
     }
 
+    @Nullable
+    public DataType fieldReferenceType() {
+        return expression.fieldReferenceType();
+    }
+
     /** Compute column's value from given argument. Return null if input is null. */
     @Nullable
     public String eval(@Nullable String input) {
@@ -61,4 +66,13 @@ public String eval(@Nullable String input) {
         }
         return expression.eval(input);
     }
+
+    /** Compute column's value from given argument. Return null if input is null. */
+    @Nullable
+    public String eval(@Nullable String input, DataType inputType) {
+        if (fieldReference() != null && input == null) {
+            return null;
+        }
+        return expression.eval(input, inputType);
+    }
 }

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumnUtils.java

@@ -18,13 +18,18 @@
 
 package org.apache.paimon.flink.action.cdc;
 
+import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.utils.Preconditions;
 
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -75,6 +80,90 @@ public static List<ComputedColumn> buildComputedColumns(
                             Expression.create(typeMapping, caseSensitive, exprName, args)));
         }
 
-        return computedColumns;
+        return sortComputedColumns(computedColumns);
+    }
+
+    @VisibleForTesting
+    public static List<ComputedColumn> sortComputedColumns(List<ComputedColumn> columns) {
+        Set<String> columnNames = new HashSet<>();
+        for (ComputedColumn col : columns) {
+            columnNames.add(col.columnName());
+        }
+
+        // For simple processing, no reference or referring to another computed column, means
+        // independent
+        List<ComputedColumn> independent = new ArrayList<>();
+        List<ComputedColumn> dependent = new ArrayList<>();
+
+        for (ComputedColumn col : columns) {
+            if (col.fieldReference() == null || !columnNames.contains(col.fieldReference())) {
+                independent.add(col);
+            } else {
+                dependent.add(col);
+            }
+        }
+
+        // Sort dependent columns with topological sort
+        Map<String, ComputedColumn> columnMap = new HashMap<>();
+        Map<String, Set<String>> reverseDependencies = new HashMap<>();
+
+        for (ComputedColumn col : dependent) {
+            columnMap.put(col.columnName(), col);
+            reverseDependencies
+                    .computeIfAbsent(col.fieldReference(), k -> new HashSet<>())
+                    .add(col.columnName());
+        }
+
+        List<ComputedColumn> sortedDependent = new ArrayList<>();
+        Set<String> visited = new HashSet<>();
+        Set<String> tempMark = new HashSet<>(); // For cycle detection
+
+        for (ComputedColumn col : dependent) {
+            if (!visited.contains(col.columnName())) {
+                dfs(
+                        col.columnName(),
+                        reverseDependencies,
+                        columnMap,
+                        sortedDependent,
+                        visited,
+                        tempMark);
+            }
+        }
+
+        Collections.reverse(sortedDependent);
+
+        // Independent should precede dependent
+        List<ComputedColumn> result = new ArrayList<>();
+        result.addAll(independent);
+        result.addAll(sortedDependent);
+
+        return result;
+    }
+
+    private static void dfs(
+            String node,
+            Map<String, Set<String>> reverseDependencies,
+            Map<String, ComputedColumn> columnMap,
+            List<ComputedColumn> sorted,
+            Set<String> visited,
+            Set<String> tempMark) {
+        if (tempMark.contains(node)) {
+            throw new IllegalArgumentException("Cycle detected: " + node);
+        }
+        if (visited.contains(node)) {
+            return;
+        }
+
+        tempMark.add(node);
+        ComputedColumn current = columnMap.get(node);
+
+        // Process the dependencies
+        for (String dependent : reverseDependencies.getOrDefault(node, Collections.emptySet())) {
+            dfs(dependent, reverseDependencies, columnMap, sorted, visited, tempMark);
+        }
+
+        tempMark.remove(node);
+        visited.add(node);
+        sorted.add(current);
     }
 }