2 csv pass

morningman · morningman · commit 93ff2fca088f · 2026-02-12T22:21:49.000+08:00
diff --git a/be/src/pipeline/exec/operator.cpp b/be/src/pipeline/exec/operator.cpp
@@ -40,6 +40,7 @@
 #include "pipeline/exec/hashjoin_build_sink.h"
 #include "pipeline/exec/hashjoin_probe_operator.h"
 #include "pipeline/exec/hive_table_sink_operator.h"
+#include "pipeline/exec/tvf_table_sink_operator.h"
 #include "pipeline/exec/iceberg_table_sink_operator.h"
 #include "pipeline/exec/jdbc_scan_operator.h"
 #include "pipeline/exec/jdbc_table_sink_operator.h"
@@ -817,6 +818,7 @@ DECLARE_OPERATOR(ResultFileSinkLocalState)
 DECLARE_OPERATOR(OlapTableSinkLocalState)
 DECLARE_OPERATOR(OlapTableSinkV2LocalState)
 DECLARE_OPERATOR(HiveTableSinkLocalState)
+DECLARE_OPERATOR(TVFTableSinkLocalState)
 DECLARE_OPERATOR(IcebergTableSinkLocalState)
 DECLARE_OPERATOR(AnalyticSinkLocalState)
 DECLARE_OPERATOR(BlackholeSinkLocalState)
@@ -936,6 +938,7 @@ template class AsyncWriterSink<doris::vectorized::VTabletWriter, OlapTableSinkOp
 template class AsyncWriterSink<doris::vectorized::VTabletWriterV2, OlapTableSinkV2OperatorX>;
 template class AsyncWriterSink<doris::vectorized::VHiveTableWriter, HiveTableSinkOperatorX>;
 template class AsyncWriterSink<doris::vectorized::VIcebergTableWriter, IcebergTableSinkOperatorX>;
+template class AsyncWriterSink<doris::vectorized::VTVFTableWriter, TVFTableSinkOperatorX>;
 
 #ifdef BE_TEST
 template class OperatorX<DummyOperatorLocalState>;
diff --git a/be/src/pipeline/pipeline_fragment_context.cpp b/be/src/pipeline/pipeline_fragment_context.cpp
@@ -62,7 +62,6 @@
 #include "pipeline/exec/hashjoin_build_sink.h"
 #include "pipeline/exec/hashjoin_probe_operator.h"
 #include "pipeline/exec/hive_table_sink_operator.h"
-#include "pipeline/exec/tvf_table_sink_operator.h"
 #include "pipeline/exec/iceberg_table_sink_operator.h"
 #include "pipeline/exec/jdbc_scan_operator.h"
 #include "pipeline/exec/jdbc_table_sink_operator.h"
@@ -101,6 +100,7 @@
 #include "pipeline/exec/spill_sort_source_operator.h"
 #include "pipeline/exec/streaming_aggregation_operator.h"
 #include "pipeline/exec/table_function_operator.h"
+#include "pipeline/exec/tvf_table_sink_operator.h"
 #include "pipeline/exec/union_sink_operator.h"
 #include "pipeline/exec/union_source_operator.h"
 #include "pipeline/local_exchange/local_exchange_sink_operator.h"
diff --git a/be/src/vec/sink/writer/vtvf_table_writer.cpp b/be/src/vec/sink/writer/vtvf_table_writer.cpp
@@ -53,10 +53,63 @@ Status VTVFTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
     _writer_close_timer = ADD_TIMER(writer_profile, "FileWriterCloseTime");
 
     _file_path = _tvf_sink.file_path;
-    _max_file_size_bytes = _tvf_sink.__isset.max_file_size_bytes ? _tvf_sink.max_file_size_bytes : 0;
+    _max_file_size_bytes =
+            _tvf_sink.__isset.max_file_size_bytes ? _tvf_sink.max_file_size_bytes : 0;
     _delete_existing_files_flag =
             _tvf_sink.__isset.delete_existing_files ? _tvf_sink.delete_existing_files : true;
 
+    // Log all parameters received from FE
+    LOG(INFO) << "TVF table writer open, query_id=" << print_id(_state->query_id())
+              << ", tvf_name=" << _tvf_sink.tvf_name
+              << ", file_path=" << _tvf_sink.file_path
+              << ", file_format=" << _tvf_sink.file_format
+              << ", file_type=" << _tvf_sink.file_type
+              << ", max_file_size_bytes=" << _max_file_size_bytes
+              << ", delete_existing_files=" << _delete_existing_files_flag;
+
+    if (_tvf_sink.__isset.column_separator) {
+        LOG(INFO) << "TVF table writer column_separator=[" << _tvf_sink.column_separator << "]";
+    }
+    if (_tvf_sink.__isset.line_delimiter) {
+        LOG(INFO) << "TVF table writer line_delimiter=[" << _tvf_sink.line_delimiter << "]";
+    }
+    if (_tvf_sink.__isset.compression_type) {
+        LOG(INFO) << "TVF table writer compression_type=" << _tvf_sink.compression_type;
+    }
+    if (_tvf_sink.__isset.backend_id) {
+        LOG(INFO) << "TVF table writer backend_id=" << _tvf_sink.backend_id;
+    }
+
+    // Log columns info
+    if (_tvf_sink.__isset.columns) {
+        LOG(INFO) << "TVF table writer columns count=" << _tvf_sink.columns.size();
+        for (size_t i = 0; i < _tvf_sink.columns.size(); i++) {
+            const auto& col = _tvf_sink.columns[i];
+            LOG(INFO) << "TVF table writer column[" << i << "]: name=" << col.column_name
+                      << ", type=" << col.column_type;
+        }
+    } else {
+        LOG(INFO) << "TVF table writer: no columns info from FE";
+    }
+
+    // Log properties
+    if (_tvf_sink.__isset.properties) {
+        LOG(INFO) << "TVF table writer properties count=" << _tvf_sink.properties.size();
+        for (const auto& kv : _tvf_sink.properties) {
+            LOG(INFO) << "TVF table writer property: " << kv.first << "=" << kv.second;
+        }
+    } else {
+        LOG(INFO) << "TVF table writer: no properties from FE";
+    }
+
+    // Log hadoop config
+    if (_tvf_sink.__isset.hadoop_config) {
+        LOG(INFO) << "TVF table writer hadoop_config count=" << _tvf_sink.hadoop_config.size();
+        for (const auto& kv : _tvf_sink.hadoop_config) {
+            LOG(INFO) << "TVF table writer hadoop_config: " << kv.first << "=" << kv.second;
+        }
+    }
+
     // Delete existing files if requested
     if (_delete_existing_files_flag) {
         RETURN_IF_ERROR(_delete_existing_files());
@@ -66,6 +119,19 @@ Status VTVFTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
 }
 
 Status VTVFTableWriter::write(RuntimeState* state, vectorized::Block& block) {
+    if (block.rows() > 0 && _written_rows_counter->value() == 0) {
+        // Log first block structure for debugging
+        LOG(INFO) << "TVF table writer first block: rows=" << block.rows()
+                  << ", columns=" << block.columns()
+                  << ", query_id=" << print_id(_state->query_id());
+        for (size_t i = 0; i < block.columns(); i++) {
+            const auto& col_with_name = block.get_by_position(i);
+            LOG(INFO) << "TVF table writer block column[" << i << "]: name=" << col_with_name.name
+                      << ", type=" << col_with_name.type->get_name()
+                      << ", rows=" << col_with_name.column->size();
+        }
+    }
+
     COUNTER_UPDATE(_written_rows_counter, block.rows());
 
     {
@@ -99,31 +165,38 @@ Status VTVFTableWriter::_create_file_writer(const std::string& file_name) {
         properties = _tvf_sink.properties;
     }
 
-    _file_writer_impl = DORIS_TRY(FileFactory::create_file_writer(
-            file_type, _state->exec_env(), {}, properties, file_name,
-            {
-                    .write_file_cache = false,
-                    .sync_file_data = false,
-            }));
+    LOG(INFO) << "TVF table writer creating file writer: file_name=" << file_name
+              << ", file_type=" << file_type
+              << ", properties_count=" << properties.size()
+              << ", output_expr_count=" << _vec_output_expr_ctxs.size();
+
+    _file_writer_impl = DORIS_TRY(FileFactory::create_file_writer(file_type, _state->exec_env(), {},
+                                                                  properties, file_name,
+                                                                  {
+                                                                          .write_file_cache = false,
+                                                                          .sync_file_data = false,
+                                                                  }));
 
     TFileFormatType::type format = _tvf_sink.file_format;
     switch (format) {
     case TFileFormatType::FORMAT_CSV_PLAIN: {
-        std::string column_separator =
+        _column_separator =
                 _tvf_sink.__isset.column_separator ? _tvf_sink.column_separator : ",";
-        std::string line_delimiter =
+        _line_delimiter =
                 _tvf_sink.__isset.line_delimiter ? _tvf_sink.line_delimiter : "\n";
         TFileCompressType::type compress_type = TFileCompressType::PLAIN;
         if (_tvf_sink.__isset.compression_type) {
             compress_type = _tvf_sink.compression_type;
         }
-        _vfile_writer.reset(new VCSVTransformer(_state, _file_writer_impl.get(),
-                                                _vec_output_expr_ctxs, false, {}, {}, column_separator,
-                                                line_delimiter, false, compress_type));
+        LOG(INFO) << "TVF table writer CSV config: column_separator=[" << _column_separator
+                  << "], line_delimiter=[" << _line_delimiter
+                  << "], compress_type=" << compress_type;
+        _vfile_writer.reset(new VCSVTransformer(
+                _state, _file_writer_impl.get(), _vec_output_expr_ctxs, false, {}, {},
+                _column_separator, _line_delimiter, false, compress_type));
         break;
     }
     case TFileFormatType::FORMAT_PARQUET: {
-        // Build parquet schemas from columns
         std::vector<TParquetSchema> parquet_schemas;
         if (_tvf_sink.__isset.columns) {
             for (const auto& col : _tvf_sink.columns) {
@@ -132,6 +205,7 @@ Status VTVFTableWriter::_create_file_writer(const std::string& file_name) {
                 parquet_schemas.push_back(schema);
             }
         }
+        LOG(INFO) << "TVF table writer Parquet config: schema_count=" << parquet_schemas.size();
         _vfile_writer.reset(new VParquetTransformer(
                 _state, _file_writer_impl.get(), _vec_output_expr_ctxs, parquet_schemas, false,
                 {TParquetCompressionType::SNAPPY, TParquetVersion::PARQUET_1_0, false, false}));
@@ -142,6 +216,7 @@ Status VTVFTableWriter::_create_file_writer(const std::string& file_name) {
         if (_tvf_sink.__isset.compression_type) {
             compress_type = _tvf_sink.compression_type;
         }
+        LOG(INFO) << "TVF table writer ORC config: compress_type=" << compress_type;
         _vfile_writer.reset(new VOrcTransformer(_state, _file_writer_impl.get(),
                                                 _vec_output_expr_ctxs, "", {}, false,
                                                 compress_type));
@@ -151,8 +226,7 @@ Status VTVFTableWriter::_create_file_writer(const std::string& file_name) {
         return Status::InternalError("Unsupported TVF sink file format: {}", format);
     }
 
-    LOG(INFO) << "TVF table writer created file: " << file_name
-              << ", format: " << format
+    LOG(INFO) << "TVF table writer created file: " << file_name << ", format: " << format
               << ", query_id: " << print_id(_state->query_id());
 
     return _vfile_writer->open();
diff --git a/be/src/vec/sink/writer/vtvf_table_writer.h b/be/src/vec/sink/writer/vtvf_table_writer.h
@@ -77,6 +77,8 @@ class VTVFTableWriter final : public AsyncResultWriter {
     int _file_idx = 0;
     bool _delete_existing_files_flag = true;
     std::string _file_path;
+    std::string _column_separator;
+    std::string _line_delimiter;
 
     // profile counters
     RuntimeProfile::Counter* _written_rows_counter = nullptr;
diff --git a/fe/.idea/vcs.xml b/fe/.idea/vcs.xml
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
@@ -465,6 +465,7 @@ public PlanFragment visitPhysicalTVFTableSink(PhysicalTVFTableSink<? extends Pla
         PlanFragment rootFragment = tvfSink.child().accept(this, context);
         rootFragment.setOutputPartition(DataPartition.UNPARTITIONED);
         TVFTableSink sink = new TVFTableSink(
+                rootFragment.getPlanRoot().getId(),
                 tvfSink.getTvfName(), tvfSink.getProperties(), tvfSink.getCols());
         try {
             sink.bindDataSink();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
@@ -97,6 +97,10 @@
 import org.apache.doris.qe.AutoCloseSessionVariable;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.tablefunction.ExternalFileTableValuedFunction;
+import org.apache.doris.tablefunction.HdfsTableValuedFunction;
+import org.apache.doris.tablefunction.LocalTableValuedFunction;
+import org.apache.doris.tablefunction.S3TableValuedFunction;
 import org.apache.doris.thrift.TPartialUpdateNewRowPolicy;
 
 import com.google.common.base.Preconditions;
@@ -108,6 +112,8 @@
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Table;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -121,6 +127,8 @@
  * bind an unbound logicalTableSink represent the target table of an insert command
  */
 public class BindSink implements AnalysisRuleFactory {
+    private static final Logger LOG = LogManager.getLogger(BindSink.class);
+
     public boolean needTruncateStringWhenInsert;
 
     public BindSink() {
@@ -583,17 +591,79 @@ private Plan bindTVFTableSink(MatchingContext<UnboundTVFTableSink<Plan>> ctx) {
 
         LogicalPlan child = ((LogicalPlan) sink.child());
 
-        // Derive columns from child query output
-        List<Column> cols = child.getOutput().stream()
-                .map(slot -> new Column(slot.getName(), slot.getDataType().toCatalogDataType()))
-                .collect(ImmutableList.toImmutableList());
+        // Determine target schema: if append mode and file exists, use existing file schema;
+        // otherwise derive from child query output.
+        boolean deleteExisting = Boolean.parseBoolean(
+                properties.getOrDefault("delete_existing_files", "true"));
+        List<Column> cols = null;
+        if (!deleteExisting) {
+            cols = tryGetExistingFileSchema(tvfName, properties);
+        }
+        if (cols == null) {
+            cols = child.getOutput().stream()
+                    .map(slot -> new Column(slot.getName(), slot.getDataType().toCatalogDataType()))
+                    .collect(ImmutableList.toImmutableList());
+        }
+
+        // Validate column count
+        if (cols.size() != child.getOutput().size()) {
+            throw new AnalysisException(
+                    "insert into cols should be corresponding to the query output"
+                            + ", target columns: " + cols.size()
+                            + ", query output: " + child.getOutput().size());
+        }
+
+        // Build columnToOutput mapping and reuse getOutputProjectByCoercion for type cast,
+        // same as OlapTable INSERT INTO.
+        Map<String, NamedExpression> columnToOutput = Maps.newLinkedHashMap();
+        for (int i = 0; i < cols.size(); i++) {
+            Column col = cols.get(i);
+            NamedExpression childExpr = (NamedExpression) child.getOutput().get(i);
+            Alias output = new Alias(TypeCoercionUtils.castIfNotSameType(
+                    childExpr, DataType.fromCatalogType(col.getType())), col.getName());
+            columnToOutput.put(col.getName(), output);
+        }
+        LogicalProject<?> projectWithCast = getOutputProjectByCoercion(cols, child, columnToOutput);
 
-        List<NamedExpression> outputExprs = child.getOutput().stream()
+        List<NamedExpression> outputExprs = projectWithCast.getOutput().stream()
                 .map(NamedExpression.class::cast)
                 .collect(ImmutableList.toImmutableList());
 
         return new LogicalTVFTableSink<>(tvfName, properties, cols, outputExprs,
-                Optional.empty(), Optional.empty(), child);
+                Optional.empty(), Optional.empty(), projectWithCast);
+    }
+
+    /**
+     * Try to instantiate the corresponding TVF to read the existing file's schema.
+     * Returns null if the file does not exist or schema inference fails.
+     */
+    private List<Column> tryGetExistingFileSchema(String tvfName, Map<String, String> properties) {
+        try {
+            ExternalFileTableValuedFunction tvf;
+            Map<String, String> propsCopy = new HashMap<>(properties);
+            switch (tvfName) {
+                case "local":
+                    tvf = new LocalTableValuedFunction(propsCopy);
+                    break;
+                case "s3":
+                    tvf = new S3TableValuedFunction(propsCopy);
+                    break;
+                case "hdfs":
+                    tvf = new HdfsTableValuedFunction(propsCopy);
+                    break;
+                default:
+                    return null;
+            }
+            List<Column> columns = tvf.getTableColumns();
+            if (columns != null && !columns.isEmpty()) {
+                return columns;
+            }
+        } catch (Exception e) {
+            // File does not exist or schema inference failed — fall back to child query schema
+            LOG.info("TVF sink: could not read existing file schema for append mode, "
+                    + "will use child query schema. Reason: " + e.getMessage());
+        }
+        return null;
     }
 
     private Plan bindHiveTableSink(MatchingContext<UnboundHiveTableSink<Plan>> ctx) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/CollectRelation.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/CollectRelation.java
@@ -32,6 +32,7 @@
 import org.apache.doris.nereids.analyzer.UnboundDictionarySink;
 import org.apache.doris.nereids.analyzer.UnboundRelation;
 import org.apache.doris.nereids.analyzer.UnboundResultSink;
+import org.apache.doris.nereids.analyzer.UnboundTVFTableSink;
 import org.apache.doris.nereids.analyzer.UnboundTableSink;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.parser.NereidsParser;
@@ -139,6 +140,10 @@ private Plan collectFromAny(MatchingContext<Plan> ctx) {
     }
 
     private Plan collectFromUnboundSink(MatchingContext<UnboundLogicalSink<Plan>> ctx) {
+        // TVF sink (local/s3/hdfs) is not a real table, skip table collection
+        if (ctx.root instanceof UnboundTVFTableSink) {
+            return null;
+        }
         List<String> nameParts = ctx.root.getNameParts();
         switch (nameParts.size()) {
             case 1:
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTVFCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTVFCommand.java
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/TVFTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/TVFTableSink.java
