[feat](tvf) support write data using JNI writer

.licenserc.yaml

@@ -47,6 +47,7 @@ header:
     - "fe/fe-core/src/main/java/software/amazon/awssdk/core/client/builder/SdkDefaultClientBuilder.java"
     - "fe/fe-core/src/main/antlr4/org/apache/doris/nereids/JavaLexer.g4"
     - "fe/fe-core/src/main/antlr4/org/apache/doris/nereids/JavaParser.g4"
+    - "fe/.idea/vcs.xml"
     - "be/dict/ik/*"
     - "be/dict/pinyin/*"
     - "be/src/common/signal_handler.h"

be/src/pipeline/exec/operator.cpp

@@ -80,6 +80,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"
@@ -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>;

be/src/pipeline/exec/tvf_table_sink_operator.cpp

@@ -0,0 +1,32 @@
+// 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.
+
+#include "tvf_table_sink_operator.h"
+
+#include "common/status.h"
+
+namespace doris::pipeline {
+#include "common/compile_check_begin.h"
+
+Status TVFTableSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) {
+    RETURN_IF_ERROR(Base::init(state, info));
+    SCOPED_TIMER(exec_time_counter());
+    SCOPED_TIMER(_init_timer);
+    return Status::OK();
+}
+
+} // namespace doris::pipeline

be/src/pipeline/exec/tvf_table_sink_operator.h

@@ -0,0 +1,87 @@
+// 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.
+
+#pragma once
+
+#include "operator.h"
+#include "vec/sink/writer/vtvf_table_writer.h"
+
+namespace doris::pipeline {
+#include "common/compile_check_begin.h"
+
+class TVFTableSinkOperatorX;
+
+class TVFTableSinkLocalState final
+        : public AsyncWriterSink<vectorized::VTVFTableWriter, TVFTableSinkOperatorX> {
+public:
+    using Base = AsyncWriterSink<vectorized::VTVFTableWriter, TVFTableSinkOperatorX>;
+    using Parent = TVFTableSinkOperatorX;
+    ENABLE_FACTORY_CREATOR(TVFTableSinkLocalState);
+    TVFTableSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state)
+            : Base(parent, state) {};
+    Status init(RuntimeState* state, LocalSinkStateInfo& info) override;
+    Status open(RuntimeState* state) override {
+        SCOPED_TIMER(exec_time_counter());
+        SCOPED_TIMER(_open_timer);
+        return Base::open(state);
+    }
+
+    friend class TVFTableSinkOperatorX;
+};
+
+class TVFTableSinkOperatorX final : public DataSinkOperatorX<TVFTableSinkLocalState> {
+public:
+    using Base = DataSinkOperatorX<TVFTableSinkLocalState>;
+    TVFTableSinkOperatorX(ObjectPool* pool, int operator_id, const RowDescriptor& row_desc,
+                          const std::vector<TExpr>& t_output_expr)
+            : Base(operator_id, 0, 0),
+              _row_desc(row_desc),
+              _t_output_expr(t_output_expr),
+              _pool(pool) {};
+
+    Status init(const TDataSink& thrift_sink) override {
+        RETURN_IF_ERROR(Base::init(thrift_sink));
+        RETURN_IF_ERROR(vectorized::VExpr::create_expr_trees(_t_output_expr, _output_vexpr_ctxs));
+        return Status::OK();
+    }
+
+    Status prepare(RuntimeState* state) override {
+        RETURN_IF_ERROR(Base::prepare(state));
+        RETURN_IF_ERROR(vectorized::VExpr::prepare(_output_vexpr_ctxs, state, _row_desc));
+        return vectorized::VExpr::open(_output_vexpr_ctxs, state);
+    }
+
+    Status sink(RuntimeState* state, vectorized::Block* in_block, bool eos) override {
+        auto& local_state = get_local_state(state);
+        SCOPED_TIMER(local_state.exec_time_counter());
+        COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)in_block->rows());
+        return local_state.sink(state, in_block, eos);
+    }
+
+private:
+    friend class TVFTableSinkLocalState;
+    template <typename Writer, typename Parent>
+        requires(std::is_base_of_v<vectorized::AsyncResultWriter, Writer>)
+    friend class AsyncWriterSink;
+    const RowDescriptor& _row_desc;
+    vectorized::VExprContextSPtrs _output_vexpr_ctxs;
+    const std::vector<TExpr>& _t_output_expr;
+    ObjectPool* _pool = nullptr;
+};
+
+#include "common/compile_check_end.h"
+} // namespace doris::pipeline

be/src/pipeline/pipeline_fragment_context.cpp

@@ -100,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"
@@ -1186,6 +1187,14 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS
         _sink.reset(new BlackholeSinkOperatorX(next_sink_operator_id()));
         break;
     }
+    case TDataSinkType::TVF_TABLE_SINK: {
+        if (!thrift_sink.__isset.tvf_table_sink) {
+            return Status::InternalError("Missing TVF table sink.");
+        }
+        _sink = std::make_shared<TVFTableSinkOperatorX>(pool, next_sink_operator_id(), row_desc,
+                                                        output_exprs);
+        break;
+    }
     default:
         return Status::InternalError("Unsuported sink type in pipeline: {}", thrift_sink.type);
     }

be/src/vec/runtime/vcsv_transformer.h

@@ -65,8 +65,8 @@ class VCSVTransformer final : public VFileFormatTransformer {
     std::string _gen_csv_header_types();
 
     std::string _csv_header;
-    std::string_view _column_separator;
-    std::string_view _line_delimiter;
+    std::string _column_separator;
+    std::string _line_delimiter;
 
     doris::io::FileWriter* _file_writer = nullptr;
     // Used to buffer the export data of plain text

be/src/vec/runtime/vfile_format_transformer_factory.cpp

@@ -0,0 +1,108 @@
+// 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.
+
+#include "vec/runtime/vfile_format_transformer_factory.h"
+
+#include <gen_cpp/PlanNodes_types.h>
+
+#include <string>
+#include <vector>
+
+#include "vec/runtime/vcsv_transformer.h"
+#include "vec/runtime/vjni_format_transformer.h"
+#include "vec/runtime/vorc_transformer.h"
+#include "vec/runtime/vparquet_transformer.h"
+
+namespace doris::vectorized {
+
+Status create_tvf_format_transformer(const TTVFTableSink& tvf_sink, RuntimeState* state,
+                                     io::FileWriter* file_writer,
+                                     const VExprContextSPtrs& output_vexpr_ctxs,
+                                     std::unique_ptr<VFileFormatTransformer>* result) {
+    // JNI writer path: delegate to Java-side writer
+    if (tvf_sink.__isset.writer_type && tvf_sink.writer_type == TTVFWriterType::JNI) {
+        if (!tvf_sink.__isset.writer_class) {
+            return Status::InternalError("writer_class is required when writer_type is JNI");
+        }
+        std::map<std::string, std::string> writer_params;
+        if (tvf_sink.__isset.properties) {
+            writer_params = tvf_sink.properties;
+        }
+        writer_params["file_path"] = tvf_sink.file_path;
+        if (tvf_sink.__isset.column_separator) {
+            writer_params["column_separator"] = tvf_sink.column_separator;
+        }
+        if (tvf_sink.__isset.line_delimiter) {
+            writer_params["line_delimiter"] = tvf_sink.line_delimiter;
+        }
+        result->reset(new VJniFormatTransformer(state, output_vexpr_ctxs, tvf_sink.writer_class,
+                                                std::move(writer_params)));
+        return Status::OK();
+    }
+
+    // Native writer path
+    TFileFormatType::type format = tvf_sink.file_format;
+    switch (format) {
+    case TFileFormatType::FORMAT_CSV_PLAIN: {
+        std::string column_separator =
+                tvf_sink.__isset.column_separator ? tvf_sink.column_separator : ",";
+        std::string 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;
+        }
+        result->reset(new VCSVTransformer(state, file_writer, output_vexpr_ctxs, false, {}, {},
+                                          column_separator, line_delimiter, false, compress_type));
+        break;
+    }
+    case TFileFormatType::FORMAT_PARQUET: {
+        std::vector<TParquetSchema> parquet_schemas;
+        if (tvf_sink.__isset.columns) {
+            for (const auto& col : tvf_sink.columns) {
+                TParquetSchema schema;
+                schema.__set_schema_column_name(col.column_name);
+                parquet_schemas.push_back(schema);
+            }
+        }
+        result->reset(new VParquetTransformer(
+                state, file_writer, output_vexpr_ctxs, parquet_schemas, false,
+                {TParquetCompressionType::SNAPPY, TParquetVersion::PARQUET_1_0, false, false}));
+        break;
+    }
+    case TFileFormatType::FORMAT_ORC: {
+        TFileCompressType::type compress_type = TFileCompressType::PLAIN;
+        if (tvf_sink.__isset.compression_type) {
+            compress_type = tvf_sink.compression_type;
+        }
+        std::vector<std::string> orc_column_names;
+        if (tvf_sink.__isset.columns) {
+            for (const auto& col : tvf_sink.columns) {
+                orc_column_names.push_back(col.column_name);
+            }
+        }
+        result->reset(new VOrcTransformer(state, file_writer, output_vexpr_ctxs, "",
+                                          orc_column_names, false, compress_type));
+        break;
+    }
+    default:
+        return Status::InternalError("Unsupported TVF sink file format: {}", format);
+    }
+    return Status::OK();
+}
+
+} // namespace doris::vectorized

be/src/vec/runtime/vfile_format_transformer_factory.h

@@ -0,0 +1,43 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/DataSinks_types.h>
+
+#include <memory>
+
+#include "common/status.h"
+#include "vec/exprs/vexpr_fwd.h"
+
+namespace doris {
+class RuntimeState;
+namespace io {
+class FileWriter;
+} // namespace io
+} // namespace doris
+
+namespace doris::vectorized {
+
+class VFileFormatTransformer;
+
+Status create_tvf_format_transformer(const TTVFTableSink& tvf_sink, RuntimeState* state,
+                                     io::FileWriter* file_writer,
+                                     const VExprContextSPtrs& output_vexpr_ctxs,
+                                     std::unique_ptr<VFileFormatTransformer>* result);
+
+} // namespace doris::vectorized