Merge branch 'hotfix_1.8_3.10.x_26255' into 1.8_release_3.10.x

Author: gituser

core/src/main/java/com/dtstack/flink/sql/side/JoinNodeDealer.java

@@ -522,10 +522,53 @@ private void extractJoinField(SqlNode condition, Set<Tuple2<String, String>> joi
         }
 
         SqlKind joinKind = condition.getKind();
-        if( joinKind == AND || joinKind == EQUALS ){
-            extractJoinField(((SqlBasicCall)condition).operands[0], joinFieldSet);
-            extractJoinField(((SqlBasicCall)condition).operands[1], joinFieldSet);
-        }else{
+        if (  AGGREGATE.contains(condition.getKind())
+                || AVG_AGG_FUNCTIONS.contains(joinKind)
+                || COMPARISON.contains(joinKind)
+                || joinKind == OTHER_FUNCTION
+                || joinKind == DIVIDE
+                || joinKind == CAST
+                || joinKind == TRIM
+                || joinKind == TIMES
+                || joinKind == PLUS
+                || joinKind == NOT_IN
+                || joinKind == OR
+                || joinKind == AND
+                || joinKind == MINUS
+                || joinKind == TUMBLE
+                || joinKind == TUMBLE_START
+                || joinKind == TUMBLE_END
+                || joinKind == SESSION
+                || joinKind == SESSION_START
+                || joinKind == SESSION_END
+                || joinKind == HOP
+                || joinKind == HOP_START
+                || joinKind == HOP_END
+                || joinKind == BETWEEN
+                || joinKind == IS_NULL
+                || joinKind == IS_NOT_NULL
+                || joinKind == CONTAINS
+                || joinKind == TIMESTAMP_ADD
+                || joinKind == TIMESTAMP_DIFF
+                || joinKind == LIKE
+                || joinKind == COALESCE
+                || joinKind == EQUALS ){
+
+            SqlBasicCall sqlBasicCall = (SqlBasicCall) condition;
+            for(int i=0; i<sqlBasicCall.getOperands().length; i++){
+                SqlNode sqlNode = sqlBasicCall.getOperands()[i];
+                if(sqlNode instanceof SqlLiteral){
+                    continue;
+                }
+
+                if(sqlNode instanceof SqlDataTypeSpec){
+                    continue;
+                }
+
+                extractJoinField(sqlNode, joinFieldSet);
+            }
+
+        } else if (condition.getKind() == IDENTIFIER){
             Preconditions.checkState(((SqlIdentifier)condition).names.size() == 2, "join condition must be format table.field");
             Tuple2<String, String> tuple2 = Tuple2.of(((SqlIdentifier)condition).names.get(0), ((SqlIdentifier)condition).names.get(1));
             joinFieldSet.add(tuple2);
@@ -833,20 +876,57 @@ public SqlBasicCall buildEmptyCondition(){
     private SqlIdentifier checkAndReplaceJoinCondition(SqlNode node, Map<String, String> tableMap){
 
         SqlKind joinKind = node.getKind();
-        if( joinKind == AND || joinKind == EQUALS ){
-            SqlIdentifier leftNode = checkAndReplaceJoinCondition(((SqlBasicCall)node).operands[0], tableMap);
-            SqlIdentifier rightNode = checkAndReplaceJoinCondition(((SqlBasicCall)node).operands[1], tableMap);
+        if(   AGGREGATE.contains(joinKind)
+                || AVG_AGG_FUNCTIONS.contains(joinKind)
+                || COMPARISON.contains(joinKind)
+                || joinKind == OTHER_FUNCTION
+                || joinKind == DIVIDE
+                || joinKind == CAST
+                || joinKind == TRIM
+                || joinKind == TIMES
+                || joinKind == PLUS
+                || joinKind == NOT_IN
+                || joinKind == OR
+                || joinKind == AND
+                || joinKind == MINUS
+                || joinKind == TUMBLE
+                || joinKind == TUMBLE_START
+                || joinKind == TUMBLE_END
+                || joinKind == SESSION
+                || joinKind == SESSION_START
+                || joinKind == SESSION_END
+                || joinKind == HOP
+                || joinKind == HOP_START
+                || joinKind == HOP_END
+                || joinKind == BETWEEN
+                || joinKind == IS_NULL
+                || joinKind == IS_NOT_NULL
+                || joinKind == CONTAINS
+                || joinKind == TIMESTAMP_ADD
+                || joinKind == TIMESTAMP_DIFF
+                || joinKind == LIKE
+                || joinKind == COALESCE
+                || joinKind == EQUALS ){
+            SqlBasicCall sqlBasicCall = (SqlBasicCall) node;
+            for(int i=0; i<sqlBasicCall.getOperands().length; i++){
+                SqlNode sqlNode = sqlBasicCall.getOperands()[i];
+                if(sqlNode instanceof SqlLiteral){
+                    continue;
+                }
 
-            if(leftNode != null){
-                ((SqlBasicCall)node).setOperand(0, leftNode);
-            }
+                if(sqlNode instanceof SqlDataTypeSpec){
+                    continue;
+                }
+
+                SqlIdentifier replaceNode = checkAndReplaceJoinCondition(sqlNode, tableMap);
+                if(replaceNode != null){
+                    ((SqlBasicCall)node).setOperand(i, replaceNode);
+                }
 
-            if(rightNode != null){
-                ((SqlBasicCall)node).setOperand(1, leftNode);
             }
 
             return null;
-        } else {
+        } else if (node.getKind() == IDENTIFIER) {
             //replace table
             Preconditions.checkState(((SqlIdentifier)node).names.size() == 2, "join condition must be format table.field");
             String tbName = ((SqlIdentifier) node).names.get(0);
@@ -857,6 +937,8 @@ private SqlIdentifier checkAndReplaceJoinCondition(SqlNode node, Map<String, Str
 
             return null;
         }
+
+        return null;
     }
 
     /**

core/src/main/java/com/dtstack/flink/sql/util/TableUtils.java

@@ -439,10 +439,53 @@ public static void replaceJoinFieldRefTableName(SqlNode condition, Map<String, S
             return;
         }
         SqlKind joinKind = condition.getKind();
-        if( joinKind == AND || joinKind == EQUALS ){
-            replaceJoinFieldRefTableName(((SqlBasicCall)condition).operands[0], oldTabFieldRefNew);
-            replaceJoinFieldRefTableName(((SqlBasicCall)condition).operands[1], oldTabFieldRefNew);
-        }else{
+        if( AGGREGATE.contains(joinKind)
+                || AVG_AGG_FUNCTIONS.contains(joinKind)
+                || COMPARISON.contains(joinKind)
+                || joinKind == OTHER_FUNCTION
+                || joinKind == DIVIDE
+                || joinKind == CAST
+                || joinKind == TRIM
+                || joinKind == TIMES
+                || joinKind == PLUS
+                || joinKind == NOT_IN
+                || joinKind == OR
+                || joinKind == AND
+                || joinKind == MINUS
+                || joinKind == TUMBLE
+                || joinKind == TUMBLE_START
+                || joinKind == TUMBLE_END
+                || joinKind == SESSION
+                || joinKind == SESSION_START
+                || joinKind == SESSION_END
+                || joinKind == HOP
+                || joinKind == HOP_START
+                || joinKind == HOP_END
+                || joinKind == BETWEEN
+                || joinKind == IS_NULL
+                || joinKind == IS_NOT_NULL
+                || joinKind == CONTAINS
+                || joinKind == TIMESTAMP_ADD
+                || joinKind == TIMESTAMP_DIFF
+                || joinKind == LIKE
+                || joinKind == COALESCE
+                || joinKind == EQUALS ){
+
+            SqlBasicCall sqlBasicCall = (SqlBasicCall) condition;
+            for(int i=0; i<sqlBasicCall.getOperands().length; i++){
+                SqlNode sqlNode = sqlBasicCall.getOperands()[i];
+                if(sqlNode instanceof SqlLiteral){
+                    continue;
+                }
+
+                if(sqlNode instanceof SqlDataTypeSpec){
+                    continue;
+                }
+
+                replaceJoinFieldRefTableName(sqlNode, oldTabFieldRefNew);
+            }
+
+        } else if (condition.getKind() == IDENTIFIER) {
             Preconditions.checkState(((SqlIdentifier)condition).names.size() == 2, "join condition must be format table.field");
             String fieldRefTable = ((SqlIdentifier)condition).names.get(0);
 

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/AbstractKafkaProducerFactory.java

@@ -26,9 +26,6 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.formats.avro.AvroRowSerializationSchema;
-import org.apache.flink.formats.csv.CsvRowSerializationSchema;
-import org.apache.flink.formats.json.DTJsonRowSerializationSchema;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.runtime.types.CRow;

…s/json/DTJsonRowSerializationSchema.java …zation/DTJsonRowSerializationSchema.javakafka-base/kafka-base-sink/src/main/java/org/apache/flink/formats/json/DTJsonRowSerializationSchema.java renamed to kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/DTJsonRowSerializationSchema.java kafka-base/kafka-base-sink/src/main/java/org/apache/flink/formats/json/DTJsonRowSerializationSchema.java renamed to kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/DTJsonRowSerializationSchema.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.formats.json;
+package com.dtstack.flink.sql.sink.kafka.serialization;
 
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.serialization.SerializationSchema;
@@ -25,6 +25,7 @@
 import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.formats.json.JsonRowSchemaConverter;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/JsonCRowSerializationSchema.java

@@ -159,34 +159,34 @@ private ObjectNode convertRow(ObjectNode reuse, RowTypeInfo info, Row row) {
     }
 
     private JsonNode convert(ContainerNode<?> container, JsonNode reuse, TypeInformation<?> info, Object object) {
-        if (info == Types.VOID || object == null) {
+        if (info.equals(Types.VOID) || object == null) {
             return container.nullNode();
-        } else if (info == Types.BOOLEAN) {
+        } else if (info.equals(Types.BOOLEAN)) {
             return container.booleanNode((Boolean) object);
-        } else if (info == Types.STRING) {
+        } else if (info.equals(Types.STRING)) {
             return container.textNode((String) object);
-        } else if (info == Types.BIG_DEC) {
+        } else if (info.equals(Types.BIG_DEC)) {
             // convert decimal if necessary
             if (object instanceof BigDecimal) {
                 return container.numberNode((BigDecimal) object);
             }
             return container.numberNode(BigDecimal.valueOf(((Number) object).doubleValue()));
-        } else if (info == Types.BIG_INT) {
+        } else if (info.equals(Types.BIG_INT)) {
             // convert integer if necessary
             if (object instanceof BigInteger) {
                 return container.numberNode((BigInteger) object);
             }
             return container.numberNode(BigInteger.valueOf(((Number) object).longValue()));
-        } else if (info == Types.SQL_DATE) {
+        } else if (info.equals(Types.SQL_DATE)) {
             return container.textNode(object.toString());
-        } else if (info == Types.SQL_TIME) {
+        } else if (info.equals(Types.SQL_TIME)) {
             final Time time = (Time) object;
             // strip milliseconds if possible
             if (time.getTime() % 1000 > 0) {
                 return container.textNode(timeFormatWithMillis.format(time));
             }
             return container.textNode(timeFormat.format(time));
-        } else if (info == Types.SQL_TIMESTAMP) {
+        } else if (info.equals(Types.SQL_TIMESTAMP)) {
             return container.textNode(timestampFormat.format((Timestamp) object));
         } else if (info instanceof RowTypeInfo) {
             if (reuse != null && reuse instanceof ObjectNode) {

launcher/pom.xml

@@ -29,6 +29,12 @@
             <version>1.2.17</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+            <version>1.6.1</version>
+        </dependency>
+
         <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>fastjson</artifactId>

launcher/src/main/java/com/dtstack/flink/sql/launcher/LauncherMain.java

@@ -41,6 +41,8 @@
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import org.apache.flink.util.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.IOException;
@@ -57,6 +59,8 @@
  */
 
 public class LauncherMain {
+
+    private static final Logger LOG = LoggerFactory.getLogger(LauncherMain.class);
     private static final String CORE_JAR = "core";
 
     private static String SP = File.separator;
@@ -68,9 +72,13 @@ private static String getLocalCoreJarPath(String localSqlRootJar) throws Excepti
     }
 
     public static void main(String[] args) throws Exception {
+
+        LOG.info("----start----");
+
         if (args.length == 1 && args[0].endsWith(".json")){
             args = parseJson(args);
         }
+
         OptionParser optionParser = new OptionParser(args);
         Options launcherOptions = optionParser.getOptions();
         String mode = launcherOptions.getMode();

launcher/src/main/resource/log4j.properties

@@ -0,0 +1,7 @@
+#控制台输出：
+log4j.rootLogger = INFO,consoleAppender
+log4j.appender.consoleAppender = org.apache.log4j.ConsoleAppender
+log4j.appender.console.Target = System.out
+log4j.appender.consoleAppender.layout = org.apache.log4j.PatternLayout
+log4j.appender.consoleAppender.layout.ConversionPattern =%d %-5p %m %n
+log4j.appender.consoleAppender.ImmediateFlush = true