[Feature-3189][client] Optimize cdcsource and lineage (DataLinkDC#3190)

Co-authored-by: wenmo <[email protected]>

dinky-admin/src/main/java/org/dinky/job/handler/JobRefreshHandler.java

@@ -228,7 +228,7 @@ public static JobDataDto getJobData(Integer id, String jobManagerHost, String jo
                     .config(jobConfigInfo)
                     .build();
         } catch (Exception e) {
-            log.error("Connect {} failed,{}", jobManagerHost, e.getMessage());
+            log.warn("Connect {} failed,{}", jobManagerHost, e.getMessage());
             return builder.id(id).error(true).errorMsg(e.getMessage()).build();
         }
     }

dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/sql/SQLSinkBuilder.java

@@ -38,13 +38,12 @@
 import java.io.Serializable;
 import java.time.Instant;
 import java.time.LocalDate;
+import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
-import cn.hutool.core.collection.CollUtil;
-
 public class SQLSinkBuilder extends AbstractSqlSinkBuilder implements Serializable {
 
     public static final String KEY_WORD = "sql";
@@ -58,7 +57,7 @@ private SQLSinkBuilder(FlinkCDCConfig config) {
 
     @Override
     protected void initTypeConverterList() {
-        typeConverterList = CollUtil.newArrayList(
+        typeConverterList = Arrays.asList(
                 this::convertDateType,
                 this::convertTimestampType,
                 this::convertFloatType,

dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/sql/catalog/SQLCatalogSinkBuilder.java

@@ -34,12 +34,11 @@
 
 import java.io.Serializable;
 import java.time.Instant;
+import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
 
-import com.google.common.collect.Lists;
-
 public class SQLCatalogSinkBuilder extends AbstractSqlSinkBuilder implements Serializable {
 
     public static final String KEY_WORD = "sql-catalog";
@@ -52,7 +51,7 @@ private SQLCatalogSinkBuilder(FlinkCDCConfig config) {
 
     @Override
     protected void initTypeConverterList() {
-        typeConverterList = Lists.newArrayList(
+        typeConverterList = Arrays.asList(
                 this::convertDateType,
                 this::convertTimestampType,
                 this::convertDecimalType,

dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/utils/FlinkStatementUtil.java

@@ -130,6 +130,9 @@ private static String convertSinkColumnType(String type, FlinkCDCConfig config)
         if (config.getSink().get("connector").equals("hudi") && (type.equals("TIMESTAMP"))) {
             return "TIMESTAMP(3)";
         }
+        if (config.getSink().get("connector").equals("doris") && (type.equals("TIME"))) {
+            return "STRING";
+        }
         return type;
     }
 

dinky-client/dinky-client-1.18/src/main/java/org/dinky/executor/AbstractCustomTableEnvironment.java

@@ -31,7 +31,6 @@
 import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
-import org.apache.flink.table.api.internal.TableEnvironmentImpl;
 import org.apache.flink.table.delegation.Planner;
 import org.apache.flink.table.operations.ExplainOperation;
 import org.apache.flink.table.operations.ModifyOperation;
@@ -118,7 +117,7 @@ public SqlExplainResult explainSqlRecord(String statement, ExplainDetail... extr
 
     @Override
     public List<LineageRel> getLineage(String statement) {
-        LineageContext lineageContext = new LineageContext((TableEnvironmentImpl) streamTableEnvironment);
-        return lineageContext.getLineage(statement);
+        LineageContext lineageContext = new LineageContext(this);
+        return lineageContext.analyzeLineage(statement);
     }
 }

dinky-client/dinky-client-1.18/src/main/java/org/dinky/utils/FunctionVisitor.java

@@ -0,0 +1,50 @@
+/*
+ *
+ *  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.dinky.utils;
+
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.util.SqlBasicVisitor;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class FunctionVisitor extends SqlBasicVisitor<Void> {
+
+    private final List<UnresolvedIdentifier> functionList = new ArrayList<>();
+
+    @Override
+    public Void visit(SqlCall call) {
+        if (call instanceof SqlBasicCall && call.getOperator() instanceof SqlFunction) {
+            SqlFunction function = (SqlFunction) call.getOperator();
+            SqlIdentifier opName = function.getNameAsId();
+
+            functionList.add(UnresolvedIdentifier.of(opName.names));
+        }
+        return super.visit(call);
+    }
+
+    public List<UnresolvedIdentifier> getFunctionList() {
+        return functionList;
+    }
+}