[fix](fe) Preserve path partition columns with CSV schema

fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java

@@ -288,7 +288,20 @@ public ScanNode getScanNode(PlanNodeId id, TupleDescriptor desc, SessionVariable
     @Override
     public List<Column> getTableColumns() throws AnalysisException {
         if (!csvSchema.isEmpty()) {
-            return csvSchema;
+            List<Column> schema = Lists.newArrayList(csvSchema);
+            Set<String> columnLowerNames = new HashSet<>();
+            for (Column column : csvSchema) {
+                columnLowerNames.add(column.getName().toLowerCase());
+            }
+            for (String colName : pathPartitionKeys) {
+                if (!columnLowerNames.add(colName.toLowerCase())) {
+                    throw new NotSupportedException(
+                            "Path partition column conflicts with an existing column: " + colName);
+                }
+                schema.add(new Column(colName,
+                        ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH), false));
+            }
+            return schema;
         }
         // if (FeConstants.runningUnitTest) {
         //     Object mockedUtObj = FeConstants.unitTestConstant;

fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java

@@ -17,12 +17,15 @@
 
 package org.apache.doris.tablefunction;
 
+import org.apache.doris.analysis.BrokerDesc;
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.PrimitiveType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.util.FileFormatConstants;
 import org.apache.doris.common.util.FileFormatUtils;
+import org.apache.doris.nereids.exceptions.NotSupportedException;
+import org.apache.doris.thrift.TFileType;
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
@@ -63,6 +66,40 @@ public void testHiveParquetTimeZoneRejectsAmbiguousShortAlias() {
         Assert.assertTrue(exception.getMessage().contains("short timezone aliases are not supported"));
     }
 
+    @Test
+    public void testCsvSchemaIncludesPathPartitionColumns() throws AnalysisException {
+        ExternalFileTableValuedFunction tvf = new TestExternalFileTableValuedFunction();
+        Map<String, String> properties = Maps.newHashMap();
+        properties.put(FileFormatConstants.PROP_FORMAT, FileFormatConstants.FORMAT_CSV);
+        properties.put(FileFormatConstants.PROP_CSV_SCHEMA, "id:int;name:string");
+        properties.put(FileFormatConstants.PROP_PATH_PARTITION_KEYS, "pt,region");
+
+        tvf.parseCommonProperties(properties);
+
+        List<Column> columns = tvf.getTableColumns();
+        Assert.assertEquals(4, columns.size());
+        Assert.assertEquals("id", columns.get(0).getName());
+        Assert.assertEquals("name", columns.get(1).getName());
+        Assert.assertEquals("pt", columns.get(2).getName());
+        Assert.assertEquals("region", columns.get(3).getName());
+    }
+
+    @Test
+    public void testCsvSchemaRejectsConflictingPathPartitionColumn() throws AnalysisException {
+        ExternalFileTableValuedFunction tvf = new TestExternalFileTableValuedFunction();
+        Map<String, String> properties = Maps.newHashMap();
+        properties.put(FileFormatConstants.PROP_FORMAT, FileFormatConstants.FORMAT_CSV);
+        properties.put(FileFormatConstants.PROP_CSV_SCHEMA, "id:int;name:string");
+        properties.put(FileFormatConstants.PROP_PATH_PARTITION_KEYS, "ID");
+
+        tvf.parseCommonProperties(properties);
+
+        NotSupportedException exception = Assert.assertThrows(
+                NotSupportedException.class, tvf::getTableColumns);
+        Assert.assertTrue(exception.getMessage()
+                .contains("Path partition column conflicts with an existing column: ID"));
+    }
+
     @Test
     public void testCsvSchemaParse() {
         Config.enable_date_conversion = true;
@@ -146,4 +183,26 @@ public void testCsvSchemaParse() {
             Assert.fail();
         }
     }
+
+    private static class TestExternalFileTableValuedFunction extends ExternalFileTableValuedFunction {
+        @Override
+        public TFileType getTFileType() {
+            return TFileType.FILE_LOCAL;
+        }
+
+        @Override
+        public String getFilePath() {
+            return "";
+        }
+
+        @Override
+        public BrokerDesc getBrokerDesc() {
+            return null;
+        }
+
+        @Override
+        public String getTableName() {
+            return "test_external_file_tvf";
+        }
+    }
 }

regression-test/data/external_table_p0/tvf/test_csv_schema_with_path_partition_keys.out

@@ -0,0 +1,6 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !csv_schema_with_path_partition_keys --
+1111	mkdir	hello
+11111	8888888	hello
+33333	helloworld	hello
+

regression-test/suites/external_table_p0/tvf/test_csv_schema_with_path_partition_keys.groovy

@@ -0,0 +1,52 @@
+// 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.
+
+suite("test_csv_schema_with_path_partition_keys", "p0,external") {
+    String ak = getS3AK()
+    String sk = getS3SK()
+    String s3Endpoint = getS3Endpoint()
+    String bucket = context.config.otherConfigs.get("s3BucketName")
+
+    order_qt_csv_schema_with_path_partition_keys """
+        select c1, c2, dt1 from
+        s3(
+            "URI" = "https://${bucket}.${s3Endpoint}/regression/tvf/test_path_partition_keys/dt1=hello/c.csv",
+            "s3.access_key" = "${ak}",
+            "s3.secret_key" = "${sk}",
+            "FORMAT" = "csv",
+            "column_separator" = ",",
+            "csv_schema" = "c1:int;c2:string",
+            "use_path_style" = "false", -- aliyun does not support path_style
+            "path_partition_keys" = "dt1") order by c1, c2;
+    """
+
+    test {
+        sql """
+            select * from
+            s3(
+                "URI" = "https://${bucket}.${s3Endpoint}/regression/tvf/test_path_partition_keys/dt1=hello/c.csv",
+                "s3.access_key" = "${ak}",
+                "s3.secret_key" = "${sk}",
+                "FORMAT" = "csv",
+                "column_separator" = ",",
+                "csv_schema" = "c1:int;c2:string",
+                "use_path_style" = "false", -- aliyun does not support path_style
+                "path_partition_keys" = "C1");
+        """
+        exception "Path partition column conflicts with an existing column: C1"
+    }
+}