Merge branch '1.10_release_4.0.x' into hotfix_1.10_4.0.x_29910

# Conflicts:
#	hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/RowKeyBuilder.java

Author: FlechazoW

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;
+    }
 }

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

@@ -140,7 +140,7 @@ private void extractPredicateInfo(SqlNode whereNode, List<PredicateInfo> predica
         }
     }
 
-    private void fillPredicateInfoToList(SqlBasicCall whereNode, List<PredicateInfo> predicatesInfoList, String operatorName, SqlKind operatorKind,
+    private void  fillPredicateInfoToList(SqlBasicCall whereNode, List<PredicateInfo> predicatesInfoList, String operatorName, SqlKind operatorKind,
                                          int fieldIndex, int conditionIndex) {
         SqlNode sqlNode = whereNode.getOperands()[fieldIndex];
         if (sqlNode.getKind() == SqlKind.IDENTIFIER) {

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

@@ -409,10 +409,9 @@ private void joinFun(Object pollObj,
 
         RowTypeInfo typeInfo = new RowTypeInfo(targetTable.getSchema().getFieldTypes(), targetTable.getSchema().getFieldNames());
 
-        DataStream adaptStream = tableEnv.toRetractStream(targetTable, Row.class)
+        DataStream adaptStream = tableEnv.toRetractStream(targetTable, typeInfo)
                 .filter(f -> f.f0)
-                .map(f -> f.f1)
-                .returns(Row.class);
+                .map(f -> f.f1);
 
         //join side table before keyby ===> Reducing the size of each dimension table cache of async
         if (sideTableInfo.isPartitionedJoin()) {

core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfo.java

@@ -49,7 +49,7 @@ public abstract class AbstractTableInfo implements Serializable {
     private final List<String> fieldList = Lists.newArrayList();
 
     /**key:别名, value: realField */
-    private Map<String, String> physicalFields = Maps.newHashMap();
+    private Map<String, String> physicalFields = Maps.newLinkedHashMap();
 
     private final List<String> fieldTypeList = Lists.newArrayList();
 

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

@@ -0,0 +1,36 @@
+/*
+ * 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.util;
+
+import org.apache.flink.util.Preconditions;
+
+import java.io.File;
+
+/**
+ * @program: flinkStreamSQL
+ * @author: wuren
+ * @create: 2020/09/21
+ **/
+public class DtFileUtils {
+    public static void checkExists(String path) {
+        File file = new File(path);
+        String errorMsg = "%s file is not exist!";
+        Preconditions.checkState(file.exists(), errorMsg, path);
+    }
+}

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

@@ -0,0 +1,54 @@
+/*
+ * 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.util;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * @program: flinkStreamSQL
+ * @author: wuren
+ * @create: 2020/09/14
+ **/
+public class KrbUtils {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KrbUtils.class);
+
+    public static final String KRB5_CONF_KEY = "java.security.krb5.conf";
+    public static final String HADOOP_AUTH_KEY = "hadoop.security.authentication";
+    public static final String KRB_STR = "Kerberos";
+//    public static final String FALSE_STR = "false";
+//    public static final String SUBJECT_ONLY_KEY = "javax.security.auth.useSubjectCredsOnly";
+
+    public static UserGroupInformation getUgi(String principal, String keytabPath, String krb5confPath) throws IOException {
+        LOG.info("Kerberos login with principal: {} and keytab: {}", principal, keytabPath);
+        System.setProperty(KRB5_CONF_KEY, krb5confPath);
+        // TODO 尚未探索出此选项的意义，以后研究明白方可打开
+//        System.setProperty(SUBJECT_ONLY_KEY, FALSE_STR);
+        Configuration configuration = new Configuration();
+        configuration.set(HADOOP_AUTH_KEY , KRB_STR);
+        UserGroupInformation.setConfiguration(configuration);
+        return UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keytabPath);
+    }
+
+}