Merge branch '1.10_test_4.0.x' into feat_1.10_impalaSinkKerberos_mergedTest4.0

# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/util/KrbUtils.java

Author: a49a

.gitignore

@@ -14,4 +14,6 @@ lib/
 .DS_Store
 bin/nohup.out
 .DS_Store
-bin/sideSql.txt
+bin/sideSql.txt
+*.keytab
+krb5.conf

cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java

@@ -18,7 +18,18 @@
 
 package com.dtstack.flink.sql.side.cassandra;
 
-import com.datastax.driver.core.*;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.HostDistance;
+import com.datastax.driver.core.PoolingOptions;
+import com.datastax.driver.core.QueryOptions;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.SocketOptions;
 import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
@@ -32,10 +43,7 @@
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.table.dataformat.BaseRow;
-import org.apache.flink.types.Row;
-import org.apache.flink.util.Collector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -72,27 +80,6 @@ public CassandraAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<Field
         super(new com.dtstack.flink.sql.side.cassandra.CassandraAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 
-    @Override
-    public Row fillData(Row input, Object sideInput) {
-        Map<String, Object> cacheInfo = (Map<String, Object>) sideInput;
-        Row row = new Row(sideInfo.getOutFieldInfoList().size());
-        for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {
-            Object obj = input.getField(entry.getValue());
-            obj = convertTimeIndictorTypeInfo(entry.getValue(), obj);
-            row.setField(entry.getKey(), obj);
-        }
-
-        for (Map.Entry<Integer, String> entry : sideInfo.getSideFieldNameIndex().entrySet()) {
-            if (cacheInfo == null) {
-                row.setField(entry.getKey(), null);
-            } else {
-                row.setField(entry.getKey(), cacheInfo.get(entry.getValue()));
-            }
-        }
-
-        return row;
-    }
-
     @Override
     protected void initCache() throws SQLException {
         Map<String, List<Map<String, Object>>> newCache = Maps.newConcurrentMap();

cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncSideInfo.java

@@ -72,57 +72,4 @@ public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInf
         LOG.info("---------side_exe_sql-----\n{}" + sqlCondition);
     }
 
-
-    @Override
-    public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
-        if (sqlNode.getKind() != SqlKind.EQUALS) {
-            throw new RuntimeException("not equal operator.");
-        }
-
-        SqlIdentifier left = (SqlIdentifier) ((SqlBasicCall) sqlNode).getOperands()[0];
-        SqlIdentifier right = (SqlIdentifier) ((SqlBasicCall) sqlNode).getOperands()[1];
-
-        String leftTableName = left.getComponent(0).getSimple();
-        String leftField = left.getComponent(1).getSimple();
-
-        String rightTableName = right.getComponent(0).getSimple();
-        String rightField = right.getComponent(1).getSimple();
-
-        if (leftTableName.equalsIgnoreCase(sideTableName)) {
-            equalFieldList.add(leftField);
-            int equalFieldIndex = -1;
-            for (int i = 0; i < getFieldNames().length; i++) {
-                String fieldName = getFieldNames()[i];
-                if (fieldName.equalsIgnoreCase(rightField)) {
-                    equalFieldIndex = i;
-                }
-            }
-            if (equalFieldIndex == -1) {
-                throw new RuntimeException("can't deal equal field: " + sqlNode);
-            }
-
-            equalValIndex.add(equalFieldIndex);
-
-        } else if (rightTableName.equalsIgnoreCase(sideTableName)) {
-
-            equalFieldList.add(rightField);
-            int equalFieldIndex = -1;
-            for (int i = 0; i < getFieldNames().length; i++) {
-                String fieldName = getFieldNames()[i];
-                if (fieldName.equalsIgnoreCase(leftField)) {
-                    equalFieldIndex = i;
-                }
-            }
-            if (equalFieldIndex == -1) {
-                throw new RuntimeException("can't deal equal field: " + sqlNode.toString());
-            }
-
-            equalValIndex.add(equalFieldIndex);
-
-        } else {
-            throw new RuntimeException("resolve equalFieldList error:" + sqlNode.toString());
-        }
-
-    }
-
 }

core/src/main/java/com/dtstack/flink/sql/constant/PluginParamConsts.java

@@ -0,0 +1,30 @@
+/*
+ * 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 com.dtstack.flink.sql.constant;
+
+/**
+ * @program: flinkStreamSQL
+ * @author: wuren
+ * @create: 2020/09/15
+ **/
+public class PluginParamConsts {
+    public static final String PRINCIPAL = "principal";
+    public static final String KEYTAB = "keytab";
+    public static final String KRB5_CONF = "krb5conf";
+}

core/src/main/java/com/dtstack/flink/sql/format/dtnest/DtNestRowDeserializationSchema.java

@@ -42,6 +42,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.regex.Pattern;
 
 /**
  * source data parse to json format
@@ -64,6 +65,9 @@ public class DtNestRowDeserializationSchema extends AbstractDeserializationSchem
     private final List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos;
     private final String charsetName;
 
+    private static final Pattern TIMESTAMP_PATTERN = Pattern.compile("^\\d+$");
+    private static final Pattern TIME_FORMAT_PATTERN = Pattern.compile("\\w+\\d+:\\d+:\\d+");
+
     public DtNestRowDeserializationSchema(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping,
                                           List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos,
                                           String charsetName) {
@@ -146,11 +150,11 @@ private Object convert(JsonNode node, TypeInformation<?> info) {
             return Date.valueOf(node.asText());
         } else if (info.getTypeClass().equals(Types.SQL_TIME.getTypeClass())) {
             // local zone
-            return Time.valueOf(node.asText());
+            return convertToTime(node.asText());
         } else if (info.getTypeClass().equals(Types.SQL_TIMESTAMP.getTypeClass())) {
             // local zone
-            return Timestamp.valueOf(node.asText());
-        }  else if (info instanceof RowTypeInfo) {
+            return convertToTimestamp(node.asText());
+        } else if (info instanceof RowTypeInfo) {
             return convertRow(node, (RowTypeInfo) info);
         } else if (info instanceof ObjectArrayTypeInfo) {
             return convertObjectArray(node, ((ObjectArrayTypeInfo) info).getComponentInfo());
@@ -165,6 +169,29 @@ private Object convert(JsonNode node, TypeInformation<?> info) {
         }
     }
 
+    /**
+     * 将 2020-09-07 14:49:10.0 和 1598446699685 两种格式都转化为 Timestamp
+     */
+    private Timestamp convertToTimestamp(String timestamp) {
+        if (TIMESTAMP_PATTERN.matcher(timestamp).find()) {
+            return new Timestamp(Long.parseLong(timestamp));
+        }
+        if (TIME_FORMAT_PATTERN.matcher(timestamp).find()) {
+            return Timestamp.valueOf(timestamp);
+        }
+        throw new IllegalArgumentException("Incorrect time format of timestamp");
+    }
+
+    private Time convertToTime(String timestamp) {
+        if (TIMESTAMP_PATTERN.matcher(timestamp).find()) {
+            return new Time(Long.parseLong(timestamp));
+        }
+        if (TIME_FORMAT_PATTERN.matcher(timestamp).find()) {
+            return Time.valueOf(timestamp);
+        }
+        throw new IllegalArgumentException("Incorrect time format of time");
+    }
+
     private Row convertTopRow() {
         Row row = new Row(fieldNames.length);
         try {
@@ -175,7 +202,7 @@ private Row convertTopRow() {
                 if (node == null) {
                     if (fieldExtraInfo != null && fieldExtraInfo.getNotNull()) {
                         throw new IllegalStateException("Failed to find field with name '"
-                            + fieldNames[i] + "'.");
+                                + fieldNames[i] + "'.");
                     } else {
                         row.setField(i, null);
                     }
@@ -216,6 +243,7 @@ private Object convertObjectArray(JsonNode node, TypeInformation<?> elementType)
         }
         return array;
     }
+
     @Override
     public TypeInformation<Row> getProducedType() {
         return typeInfo;

core/src/main/java/com/dtstack/flink/sql/krb/KerberosTable.java

@@ -0,0 +1,65 @@
+/*
+ * 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 com.dtstack.flink.sql.krb;
+
+import com.google.common.base.Strings;
+
+/**
+ * @program: flinkStreamSQL
+ * @author: wuren
+ * @create: 2020/09/15
+ **/
+public interface KerberosTable {
+
+    String getPrincipal();
+
+    void setPrincipal(String principal);
+
+    String getKeytab();
+
+    void setKeytab(String keytab);
+
+    String getKrb5conf();
+
+    void setKrb5conf(String krb5conf);
+
+    boolean isEnableKrb();
+
+    void setEnableKrb(boolean enableKrb);
+
+    default void judgeKrbEnable() {
+        boolean allSet =
+            !Strings.isNullOrEmpty(getPrincipal()) &&
+            !Strings.isNullOrEmpty(getKeytab()) &&
+            !Strings.isNullOrEmpty(getKrb5conf());
+
+        boolean allNotSet =
+            Strings.isNullOrEmpty(getPrincipal()) &&
+            Strings.isNullOrEmpty(getKeytab()) &&
+            Strings.isNullOrEmpty(getKrb5conf());
+
+        if (allSet) {
+            setEnableKrb(true);
+        } else if (allNotSet) {
+            setEnableKrb(false);
+        } else {
+            throw new RuntimeException("Missing kerberos parameter! all kerberos params must be set, or all kerberos params are not set");
+        }
+    }
+}

core/src/main/java/com/dtstack/flink/sql/parser/SqlParser.java

@@ -28,8 +28,11 @@
 import com.google.common.collect.Lists;
 import com.google.common.base.Strings;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /**
  * Reason:
@@ -51,6 +54,8 @@ public static void setLocalSqlPluginRoot(String localSqlPluginRoot){
         LOCAL_SQL_PLUGIN_ROOT = localSqlPluginRoot;
     }
 
+    private static final Pattern ADD_FIlE_PATTERN = Pattern.compile("(?i).*add\\s+file\\s+.+");
+
     /**
      * flink support sql syntax
      * CREATE TABLE sls_stream() with ();
@@ -70,6 +75,7 @@ public static SqlTree parseSql(String sql, String pluginLoadMode) throws Excepti
                 .replace("\t", " ").trim();
 
         List<String> sqlArr = DtStringUtil.splitIgnoreQuota(sql, SQL_DELIMITER);
+        sqlArr = removeAddFileStmt(sqlArr);
         SqlTree sqlTree = new SqlTree();
         AbstractTableInfoParser tableInfoParser = new AbstractTableInfoParser();
         for(String childSql : sqlArr){
@@ -150,4 +156,18 @@ public static SqlTree parseSql(String sql, String pluginLoadMode) throws Excepti
 
         return sqlTree;
     }
+
+    /**
+     * remove add file with statment etc. add file /etc/krb5.conf;
+     */
+    private static List<String> removeAddFileStmt(List<String> stmts) {
+        List<String> cleanedStmts = new ArrayList<>();
+        for (String stmt : stmts) {
+            Matcher matcher = ADD_FIlE_PATTERN.matcher(stmt);
+            if(!matcher.matches()) {
+                cleanedStmts.add(stmt);
+            }
+        }
+        return cleanedStmts;
+    }
 }