Merge branch 'v1.8.0_dev' of ssh://git.dtstack.cn:10022/dt-insight-engine/flinkStreamSQL into feat_elasticsearch6-sink

Author: dengdejiushini

core/pom.xml

@@ -121,6 +121,11 @@
             <artifactId>junit</artifactId>
             <version>4.12</version>
         </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <version>1.1.7</version>
+        </dependency>
 
     </dependencies>
 

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -77,6 +77,8 @@
 import java.util.Set;
 
 import com.dtstack.flink.sql.option.Options;
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.LoggerContext;
 
 /**
  * Date: 2018/6/26
@@ -105,9 +107,9 @@ public static void main(String[] args) throws Exception {
         String pluginLoadMode = options.getPluginLoadMode();
         String deployMode = options.getMode();
         String confProp = options.getConfProp();
-
         sql = URLDecoder.decode(sql, Charsets.UTF_8.name());
         SqlParser.setLocalSqlPluginRoot(localSqlPluginPath);
+        setLogLevel(options.getLogLevel());
 
         List<String> addJarFileList = Lists.newArrayList();
         if (!Strings.isNullOrEmpty(addJarListStr)) {
@@ -308,4 +310,10 @@ private static StreamExecutionEnvironment getStreamExeEnv(Properties confPropert
         StreamEnvConfigManager.streamExecutionEnvironmentConfig(env, confProperties);
         return env;
     }
+    private static void setLogLevel(String level){
+        LoggerContext loggerContext= (LoggerContext) LoggerFactory.getILoggerFactory();
+        //设置全局日志级别
+        ch.qos.logback.classic.Logger logger = loggerContext.getLogger("root");
+        logger.setLevel(Level.toLevel(level, Level.INFO));
+    }
 }

core/src/main/java/com/dtstack/flink/sql/option/Options.java

@@ -69,6 +69,8 @@ public class Options {
     @OptionRequired(description = "plugin load mode, by classpath or shipfile")
     private String pluginLoadMode = EPluginLoadMode.CLASSPATH.name();
 
+    private String logLevel = "logLevel";
+
     public String getMode() {
         return mode;
     }
@@ -172,4 +174,12 @@ public String getPluginLoadMode() {
     public void setPluginLoadMode(String pluginLoadMode) {
         this.pluginLoadMode = pluginLoadMode;
     }
+
+    public String getLogLevel() {
+        return logLevel;
+    }
+
+    public void setLogLevel(String logLevel) {
+        this.logLevel = logLevel;
+    }
 }

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

@@ -22,6 +22,7 @@
 
 import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.DtStringUtil;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
@@ -45,6 +46,7 @@ public abstract class AbsTableParser {
 
     private static Pattern primaryKeyPattern = Pattern.compile("(?i)PRIMARY\\s+KEY\\s*\\((.*)\\)");
     private static Pattern nestJsonFieldKeyPattern = Pattern.compile("(?i)((@*\\S+\\.)*\\S+)\\s+(\\w+)\\s+AS\\s+(\\w+)(\\s+NOT\\s+NULL)?$");
+    private static Pattern physicalFieldFunPattern = Pattern.compile("\\w+\\((\\w+)\\)$");
 
     private Map<String, Pattern> patternMap = Maps.newHashMap();
 
@@ -131,9 +133,12 @@ public void dealPrimaryKey(Matcher matcher, TableInfo tableInfo){
      */
     protected void dealNestField(Matcher matcher, TableInfo tableInfo) {
         String physicalField = matcher.group(1);
+        Preconditions.checkArgument(!physicalFieldFunPattern.matcher(physicalField).find(),
+                "No need to add data types when using functions, The correct way is : strLen(name) as nameSize, ");
+
         String fieldType = matcher.group(3);
         String mappingField = matcher.group(4);
-        Class fieldClass= dbTypeConvertToJavaType(fieldType);
+        Class fieldClass = dbTypeConvertToJavaType(fieldType);
         boolean notNull = matcher.group(5) != null;
         TableInfo.FieldExtraInfo fieldExtraInfo = new TableInfo.FieldExtraInfo();
         fieldExtraInfo.setNotNull(notNull);