[Improve] File souce refactor (#10758)

Author: liunaijie

seatunnel-connectors-v2/connector-file/connector-file-base-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/BaseHdfsFileSource.java

@@ -23,6 +23,8 @@
 
 import java.util.List;
 
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY;
+
 public class FileBaseOptions extends ConnectorCommonOptions {
 
     public static final Option<String> FILENAME_EXTENSION =
@@ -165,4 +167,10 @@ public class FileBaseOptions extends ConnectorCommonOptions {
                             "File split size in bytes when enable_file_split=true. Must be greater than 0. "
                                     + "For text-like formats, the split end will be aligned to the next row_delimiter. "
                                     + "Default is 128MB (128*1024*1024).");
+
+    public static final Option<String> DEFAULT_FS =
+            Options.key(FS_DEFAULT_NAME_KEY)
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("HDFS namenode host");
 }

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileBaseOptions.java

@@ -17,40 +17,125 @@
 
 package org.apache.seatunnel.connectors.seatunnel.file.source;
 
+import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.options.ConnectorCommonOptions;
 import org.apache.seatunnel.api.source.Boundedness;
 import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.source.SourceReader;
 import org.apache.seatunnel.api.source.SourceSplitEnumerator;
 import org.apache.seatunnel.api.source.SupportColumnProjection;
 import org.apache.seatunnel.api.source.SupportParallelism;
-import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
+import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
+import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
 import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
+import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode;
+import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
 import org.apache.seatunnel.connectors.seatunnel.file.source.reader.ReadStrategy;
 import org.apache.seatunnel.connectors.seatunnel.file.source.split.FileSourceSplit;
 import org.apache.seatunnel.connectors.seatunnel.file.source.split.FileSourceSplitEnumerator;
 import org.apache.seatunnel.connectors.seatunnel.file.source.state.FileSourceState;
 
+import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
 
 public abstract class BaseFileSource
         implements SeaTunnelSource<SeaTunnelRow, FileSourceSplit, FileSourceState>,
                 SupportParallelism,
                 SupportColumnProjection {
-    protected SeaTunnelRowType rowType;
-    protected ReadStrategy readStrategy;
-    protected HadoopConf hadoopConf;
-    protected List<String> filePaths;
+    protected ReadonlyConfig pluginConfig;
+    private final CatalogTable catalogTable;
+    private final List<String> filePaths;
+    private final ReadStrategy readStrategy;
+
+    /** shouldn't use this construct method. just for testing */
+    @VisibleForTesting
+    public BaseFileSource() {
+        this.catalogTable = null;
+        this.filePaths = null;
+        this.readStrategy = null;
+    }
+
+    public BaseFileSource(ReadonlyConfig pluginConfig) {
+        this.pluginConfig = pluginConfig;
+        HadoopConf hadoopConf = initHadoopConf();
+        this.readStrategy =
+                pluginConfig.get(FileBaseSourceOptions.FILE_FORMAT_TYPE).getReadStrategy();
+        this.readStrategy.setPluginConfig(pluginConfig.toConfig());
+        this.readStrategy.init(hadoopConf);
+        String path = pluginConfig.get(FileBaseSourceOptions.FILE_PATH);
+        try {
+            filePaths = readStrategy.getFileNamesByPath(path);
+        } catch (IOException e) {
+            String errorMsg = String.format("Get file list from this path [%s] failed", path);
+            throw new FileConnectorException(
+                    FileConnectorErrorCode.FILE_LIST_GET_FAILED, errorMsg, e);
+        }
+
+        // support user-defined schema
+        CatalogTable userDefinedCatalogTable;
+        FileFormat fileFormat = pluginConfig.get(FileBaseSourceOptions.FILE_FORMAT_TYPE);
+        // only json text csv type support user-defined schema now
+        if (pluginConfig.getOptional(ConnectorCommonOptions.SCHEMA).isPresent()) {
+            switch (fileFormat) {
+                case CSV:
+                case TEXT:
+                case JSON:
+                case EXCEL:
+                case XML:
+                    userDefinedCatalogTable = CatalogTableUtil.buildWithConfig(pluginConfig);
+                    readStrategy.setCatalogTable(userDefinedCatalogTable);
+                    break;
+                case ORC:
+                case PARQUET:
+                case BINARY:
+                    throw new FileConnectorException(
+                            CommonErrorCodeDeprecated.UNSUPPORTED_OPERATION,
+                            "SeaTunnel does not support user-defined schema for [parquet, orc, binary] files");
+                default:
+                    // never got in there
+                    throw new FileConnectorException(
+                            CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
+                            "SeaTunnel does not supported this file format");
+            }
+        } else {
+            if (filePaths.isEmpty() && fileFormat != FileFormat.BINARY) {
+                // When the directory is empty, distribute default behavior schema
+                userDefinedCatalogTable = CatalogTableUtil.buildSimpleTextTable();
+            } else {
+                try {
+                    SeaTunnelRowType rowType =
+                            readStrategy.getSeaTunnelRowTypeInfo(filePaths.get(0));
+                    userDefinedCatalogTable = CatalogTableUtil.getCatalogTable("default", rowType);
+                } catch (FileConnectorException e) {
+                    String errorMsg =
+                            String.format(
+                                    "Get table schema from file [%s] failed", filePaths.get(0));
+                    throw new FileConnectorException(
+                            CommonErrorCodeDeprecated.TABLE_SCHEMA_GET_FAILED, errorMsg, e);
+                }
+            }
+        }
+        this.catalogTable = userDefinedCatalogTable;
+    }
+
+    protected abstract HadoopConf initHadoopConf();
 
     @Override
     public Boundedness getBoundedness() {
         return Boundedness.BOUNDED;
     }
 
     @Override
-    public SeaTunnelDataType<SeaTunnelRow> getProducedType() {
-        return rowType;
+    public List<CatalogTable> getProducedCatalogTables() {
+        return Collections.singletonList(catalogTable);
     }
 
     @Override