[opt-36803][file] optimize fileSource code and move parameter "fromLine" to <FileSourceTableInfo>

Author: FlechazoW

file/file-source/src/main/java/com/dtstack/flink/sql/source/file/DTCsvRowDeserializationSchema.java

@@ -68,8 +68,6 @@ public class DTCsvRowDeserializationSchema implements DeserializationSchema<Row>
      */
     private ObjectReader objectReader;
 
-    private int fromLine;
-
     /**
      * 字段值的分割符，默认为','
      */
@@ -154,10 +152,6 @@ private CsvSchema initCsvSchema(TypeInformation<Row> typeInfo) {
     // Setter
     // --------------------------------------------------------------------------------------------
 
-    public void setFromLine(int fromLine) {
-        this.fromLine = fromLine;
-    }
-
     public void setFieldDelimiter(Character fieldDelimiter) {
         this.fieldDelimiter = fieldDelimiter;
     }
@@ -186,10 +180,6 @@ public void setTypeInfo(TypeInformation<Row> typeInfo) {
         this.typeInfo = typeInfo;
     }
 
-    public int getFromLine() {
-        return fromLine;
-    }
-
     // --------------------------------------------------------------------------------------------
     // Builder
     // --------------------------------------------------------------------------------------------
@@ -232,11 +222,6 @@ public Builder setEscapeCharacter(Character escapeCharacter) {
             return this;
         }
 
-        public Builder setFromLine(int fromLine) {
-            deserializationSchema.setFromLine(fromLine);
-            return this;
-        }
-
         public Builder setTypeInfo(TypeInformation<Row> typeInfo) {
             deserializationSchema.setTypeInfo(typeInfo);
             return this;

file/file-source/src/main/java/com/dtstack/flink/sql/source/file/FileSource.java

@@ -79,6 +79,8 @@ public class FileSource extends AbstractRichFunction implements IStreamSourceGen
 
     private String charset;
 
+    private int fromLine;
+
     protected transient Counter errorCounter;
 
     /**
@@ -98,9 +100,12 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo,
         FileSource fileSource = new FileSource();
         FileSourceTableInfo tableInfo = (FileSourceTableInfo) sourceTableInfo;
 
-        DataStreamSource<?> source = fileSource.initDataStream(tableInfo, env);
+        fileSource.initSource(tableInfo);
+
         String fields = StringUtils.join(tableInfo.getFields(), ",");
 
+        DataStreamSource<Row> source = env.addSource(fileSource, tableInfo.getOperatorName(), tableInfo.getTypeInformation());
+
         return tableEnv.fromDataStream(source, fields);
     }
 
@@ -115,15 +120,15 @@ public void initMetric() {
         numInResolveRecord = runtimeContext.getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_RESOVED_IN_COUNTER);
     }
 
-    public DataStreamSource<?> initDataStream(FileSourceTableInfo tableInfo,
-                                              StreamExecutionEnvironment env) {
+    public void initSource(FileSourceTableInfo tableInfo) {
         deserializationSchema = tableInfo.getDeserializationSchema();
         fileUri = URI.create(tableInfo.getFilePath() + SP + tableInfo.getFileName());
+
         charset = tableInfo.getCharsetName();
-        return env.addSource(
-            this,
-            tableInfo.getOperatorName(),
-            tableInfo.buildRowTypeInfo());
+        LOG.info("File charset: " + charset);
+
+        fromLine = tableInfo.getFromLine();
+        LOG.info("Read from line: " + fromLine);
     }
 
     /**
@@ -189,17 +194,11 @@ private InputStream fromLocalFile(URI fileUri) throws FileNotFoundException {
 
     @Override
     public void run(SourceContext<Row> ctx) throws Exception {
-        int fromLine = 1;
         String line;
         initMetric();
         inputStream = getInputStream(fileUri);
         bufferedReader = new BufferedReader(new InputStreamReader(inputStream, charset));
 
-        if (deserializationSchema instanceof DTCsvRowDeserializationSchema) {
-            fromLine = ((DTCsvRowDeserializationSchema) deserializationSchema).getFromLine();
-            LOG.info("Read from line: " + fromLine);
-        }
-
         while (running.get()) {
             line = bufferedReader.readLine();
             if (line == null) {

file/file-source/src/main/java/com/dtstack/flink/sql/source/file/table/CsvSourceTableInfo.java

@@ -51,8 +51,6 @@ public class CsvSourceTableInfo extends FileSourceTableInfo {
 
     private Character escapeCharacter;
 
-    private int fromLine;
-
     public Character getFieldDelimiter() {
         return fieldDelimiter;
     }
@@ -101,14 +99,6 @@ public void setEscapeCharacter(Character escapeCharacter) {
         this.escapeCharacter = escapeCharacter;
     }
 
-    public int getFromLine() {
-        return fromLine;
-    }
-
-    public void setFromLine(int fromLine) {
-        this.fromLine = fromLine;
-    }
-
     public void buildDeserializationSchema() {
         DTCsvRowDeserializationSchema dtCsvRowDeserializationSchema = new DTCsvRowDeserializationSchema
             .Builder()
@@ -119,7 +109,6 @@ public void buildDeserializationSchema() {
             .setArrayElementDelimiter(this.getArrayElementDelimiter())
             .setEscapeCharacter(this.getEscapeCharacter())
             .setQuoteCharacter(this.getQuoteCharacter())
-            .setFromLine(this.getFromLine())
             .build();
         this.setDeserializationSchema(dtCsvRowDeserializationSchema);
     }

file/file-source/src/main/java/com/dtstack/flink/sql/source/file/table/FileSourceParser.java

@@ -42,6 +42,7 @@ public AbstractTableInfo getTableInfo(String tableName,
         tableInfo.setFileName(MathUtil.getString(props.get(FileSourceConstant.FILE_NAME_KEY.toLowerCase())));
         tableInfo.setFilePath(MathUtil.getString(props.getOrDefault(FileSourceConstant.FILE_PATH_KEY.toLowerCase(), FileSourceConstant.DEFAULT_PATH)));
         tableInfo.setCharsetName(MathUtil.getString(props.getOrDefault(FileSourceConstant.CHARSET_NAME_KEY.toLowerCase(), FileSourceConstant.DEFAULT_CHARSET)));
+        tableInfo.setFromLine(MathUtil.getIntegerVal(props.getOrDefault(FileSourceConstant.FROM_LINE_KEY.toLowerCase(), 1)));
 
         tableInfo.check();
 
@@ -73,7 +74,6 @@ private FileSourceTableInfo switchTableType(String tableName,
                     tableInfo.setEscapeCharacter(escapeChar);
                 }
 
-                tableInfo.setFromLine(MathUtil.getIntegerVal(props.getOrDefault(FileSourceConstant.FROM_LINE_KEY.toLowerCase(), 1)));
                 tableInfo.buildDeserializationSchema();
                 return tableInfo;
             }

file/file-source/src/main/java/com/dtstack/flink/sql/source/file/table/FileSourceTableInfo.java

@@ -58,6 +58,8 @@ public class FileSourceTableInfo extends AbstractSourceTableInfo {
 
     private TypeInformation<Row> typeInformation;
 
+    private int fromLine;
+
     /*-------------------------------------------------------------*/
 
     public String getFileName() {
@@ -100,6 +102,14 @@ public void setDeserializationSchema(DeserializationSchema<Row> deserializationS
         this.deserializationSchema = deserializationSchema;
     }
 
+    public int getFromLine() {
+        return fromLine;
+    }
+
+    public void setFromLine(int fromLine) {
+        this.fromLine = fromLine;
+    }
+
     @SuppressWarnings("unchecked")
     public TypeInformation<Row> buildRowTypeInfo() {
         String[] types = getFieldTypes();