From b895d8a813e29dc78ef97615d66e669fbae50c77 Mon Sep 17 00:00:00 2001 From: Refrain Date: Wed, 2 Sep 2026 07:52:01 +0800 Subject: [PATCH] [fix](fe) Preserve path partition columns with CSV schema ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: File table-valued functions returned only explicitly declared CSV columns when csv_schema was set, so columns derived from path_partition_keys disappeared from the table schema. Append non-conflicting path partition columns to the explicit CSV schema in their declared order and reject case-insensitive conflicts before they can create a duplicate schema. ### Release note CSV file table-valued functions now expose path partition columns when csv_schema is specified and reject conflicting column names. ### Check List (For Author) - Test: FE unit test and regression test - FE unit test: ExternalFileTableValuedFunctionTest (5 tests) - Regression test: test_csv_schema_with_path_partition_keys, including conflict rejection - Behavior changed: Yes. Path partition columns remain available with an explicit CSV schema. - Does this need documentation: No --- .../ExternalFileTableValuedFunction.java | 15 ++++- .../ExternalFileTableValuedFunctionTest.java | 59 +++++++++++++++++++ ...st_csv_schema_with_path_partition_keys.out | 6 ++ ...csv_schema_with_path_partition_keys.groovy | 52 ++++++++++++++++ 4 files changed, 131 insertions(+), 1 deletion(-) create mode 100644 regression-test/data/external_table_p0/tvf/test_csv_schema_with_path_partition_keys.out create mode 100644 regression-test/suites/external_table_p0/tvf/test_csv_schema_with_path_partition_keys.groovy diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java index 01abd8b482123e..5bd4ed0a2aaaf0 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java +++ b/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 getTableColumns() throws AnalysisException { if (!csvSchema.isEmpty()) { - return csvSchema; + List schema = Lists.newArrayList(csvSchema); + Set 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; diff --git a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java index 34df496439589c..65e07eaacccb81 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java +++ b/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 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 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 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"; + } + } } diff --git a/regression-test/data/external_table_p0/tvf/test_csv_schema_with_path_partition_keys.out b/regression-test/data/external_table_p0/tvf/test_csv_schema_with_path_partition_keys.out new file mode 100644 index 00000000000000..949be1259b775b --- /dev/null +++ b/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 + diff --git a/regression-test/suites/external_table_p0/tvf/test_csv_schema_with_path_partition_keys.groovy b/regression-test/suites/external_table_p0/tvf/test_csv_schema_with_path_partition_keys.groovy new file mode 100644 index 00000000000000..14f57aefb3f073 --- /dev/null +++ b/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" + } +}