[Improve][Connector-V2] Support read orc with schema config to cast type (#6531)

Author: Hisoka-X

Diff for: seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/TypeUtil.java

@@ -0,0 +1,44 @@
+/*
+ * 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 org.apache.seatunnel.api.table.type;
+
+public class TypeUtil {
+
+    /** Check if the data type can be converted to another data type. */
+    public static boolean canConvert(SeaTunnelDataType<?> from, SeaTunnelDataType<?> to) {
+        // any type can be converted to string
+        if (from == to || to.getSqlType() == SqlType.STRING) {
+            return true;
+        }
+        if (from.getSqlType() == SqlType.TINYINT) {
+            return to.getSqlType() == SqlType.SMALLINT
+                    || to.getSqlType() == SqlType.INT
+                    || to.getSqlType() == SqlType.BIGINT;
+        }
+        if (from.getSqlType() == SqlType.SMALLINT) {
+            return to.getSqlType() == SqlType.INT || to.getSqlType() == SqlType.BIGINT;
+        }
+        if (from.getSqlType() == SqlType.INT) {
+            return to.getSqlType() == SqlType.BIGINT;
+        }
+        if (from.getSqlType() == SqlType.FLOAT) {
+            return to.getSqlType() == SqlType.DOUBLE;
+        }
+        return false;
+    }
+}

Diff for: seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseFileSourceConfig.java

@@ -80,7 +80,8 @@ private List<String> parseFilePaths(ReadonlyConfig readonlyConfig) {
 
     private CatalogTable parseCatalogTable(ReadonlyConfig readonlyConfig) {
         final CatalogTable catalogTable;
-        if (readonlyConfig.getOptional(TableSchemaOptions.SCHEMA).isPresent()) {
+        boolean configSchema = readonlyConfig.getOptional(TableSchemaOptions.SCHEMA).isPresent();
+        if (configSchema) {
             catalogTable = CatalogTableUtil.buildWithConfig(getPluginName(), readonlyConfig);
         } else {
             catalogTable = CatalogTableUtil.buildSimpleTextTable();
@@ -99,7 +100,10 @@ private CatalogTable parseCatalogTable(ReadonlyConfig readonlyConfig) {
             case ORC:
             case PARQUET:
                 return newCatalogTable(
-                        catalogTable, readStrategy.getSeaTunnelRowTypeInfo(filePaths.get(0)));
+                        catalogTable,
+                        readStrategy.getSeaTunnelRowTypeInfoWithUserConfigRowType(
+                                filePaths.get(0),
+                                configSchema ? catalogTable.getSeaTunnelRowType() : null));
             default:
                 throw new FileConnectorException(
                         FileConnectorErrorCode.FORMAT_NOT_SUPPORT,