[feature](catalog) support trino connector catalog sink

be/src/pipeline/exec/operator.cpp

@@ -74,6 +74,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/trino_connector_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"
@@ -746,6 +747,7 @@ DECLARE_OPERATOR(OlapTableSinkLocalState)
 DECLARE_OPERATOR(OlapTableSinkV2LocalState)
 DECLARE_OPERATOR(HiveTableSinkLocalState)
 DECLARE_OPERATOR(IcebergTableSinkLocalState)
+DECLARE_OPERATOR(TrinoConnectorTableSinkLocalState)
 DECLARE_OPERATOR(AnalyticSinkLocalState)
 DECLARE_OPERATOR(SortSinkLocalState)
 DECLARE_OPERATOR(SpillSortSinkLocalState)
@@ -856,6 +858,8 @@ 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::VTrinoConnectorTableWriter,
+                               TrinoConnectorTableSinkOperatorX>;
 
 #ifdef BE_TEST
 template class OperatorX<DummyOperatorLocalState>;

be/src/pipeline/exec/trino_connector_table_sink_operator.cpp

@@ -0,0 +1,57 @@
+// 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 "trino_connector_table_sink_operator.h"
+
+#include <memory>
+
+#include "common/object_pool.h"
+#include "pipeline/exec/operator.h"
+#include "vec/exprs/vexpr.h"
+#include "vec/exprs/vexpr_context.h"
+
+namespace doris::pipeline {
+#include "common/compile_check_begin.h"
+TrinoConnectorTableSinkOperatorX::TrinoConnectorTableSinkOperatorX(
+        const RowDescriptor& row_desc, int operator_id, const std::vector<TExpr>& t_output_expr)
+        : DataSinkOperatorX(operator_id, 0, 0),
+          _row_desc(row_desc),
+          _t_output_expr(t_output_expr) {}
+
+Status TrinoConnectorTableSinkOperatorX::init(const TDataSink& thrift_sink) {
+    RETURN_IF_ERROR(DataSinkOperatorX<TrinoConnectorTableSinkLocalState>::init(thrift_sink));
+    RETURN_IF_ERROR(vectorized::VExpr::create_expr_trees(_t_output_expr, _output_vexpr_ctxs));
+    return Status::OK();
+}
+
+Status TrinoConnectorTableSinkOperatorX::prepare(RuntimeState* state) {
+    RETURN_IF_ERROR(DataSinkOperatorX<TrinoConnectorTableSinkLocalState>::prepare(state));
+    RETURN_IF_ERROR(vectorized::VExpr::prepare(_output_vexpr_ctxs, state, _row_desc));
+    RETURN_IF_ERROR(vectorized::VExpr::open(_output_vexpr_ctxs, state));
+    return Status::OK();
+}
+
+Status TrinoConnectorTableSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block,
+                                              bool eos) {
+    auto& local_state = get_local_state(state);
+    SCOPED_TIMER(local_state.exec_time_counter());
+    COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)block->rows());
+    RETURN_IF_ERROR(local_state.sink(state, block, eos));
+    return Status::OK();
+}
+
+} // namespace doris::pipeline

be/src/pipeline/exec/trino_connector_table_sink_operator.h

@@ -0,0 +1,65 @@
+// 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 <stdint.h>
+
+#include "operator.h"
+#include "vec/sink/writer/jni/trino/vtrino_connector_table_writer.h"
+
+namespace doris::pipeline {
+#include "common/compile_check_begin.h"
+
+class TrinoConnectorTableSinkOperatorX;
+class TrinoConnectorTableSinkLocalState final
+        : public AsyncWriterSink<vectorized::VTrinoConnectorTableWriter,
+                                 TrinoConnectorTableSinkOperatorX> {
+    ENABLE_FACTORY_CREATOR(TrinoConnectorTableSinkLocalState);
+
+public:
+    using Base = AsyncWriterSink<vectorized::VTrinoConnectorTableWriter,
+                                 TrinoConnectorTableSinkOperatorX>;
+    TrinoConnectorTableSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state)
+            : Base(parent, state) {}
+
+private:
+    friend class TrinoConnectorTableSinkOperator;
+};
+
+class TrinoConnectorTableSinkOperatorX final
+        : public DataSinkOperatorX<TrinoConnectorTableSinkLocalState> {
+public:
+    TrinoConnectorTableSinkOperatorX(const RowDescriptor& row_desc, int operator_id,
+                                     const std::vector<TExpr>& select_exprs);
+    Status init(const TDataSink& thrift_sink) override;
+    Status prepare(RuntimeState* state) override;
+    Status sink(RuntimeState* state, vectorized::Block* in_block, bool eos) override;
+
+private:
+    friend class TrinoConnectorTableSinkLocalState;
+    template <typename Writer, typename Parent>
+        requires(std::is_base_of_v<vectorized::AsyncResultWriter, Writer>)
+    friend class AsyncWriterSink;
+
+    const RowDescriptor& _row_desc;
+    const std::vector<TExpr>& _t_output_expr;
+    vectorized::VExprContextSPtrs _output_vexpr_ctxs;
+};
+
+#include "common/compile_check_end.h"
+} // namespace doris::pipeline

be/src/pipeline/pipeline_fragment_context.cpp

@@ -94,6 +94,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/trino_connector_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"
@@ -1063,6 +1064,14 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS
                                                   output_exprs));
         break;
     }
+    case TDataSinkType::TRINO_CONNECTOR_TABLE_SINK: {
+        if (!thrift_sink.__isset.trino_connector_table_sink) {
+            return Status::InternalError("Missing trino connector table sink.");
+        }
+        _sink.reset(new TrinoConnectorTableSinkOperatorX(row_desc, next_sink_operator_id(),
+                                                         output_exprs));
+        break;
+    }
     case TDataSinkType::JDBC_TABLE_SINK: {
         if (!thrift_sink.__isset.jdbc_table_sink) {
             return Status::InternalError("Missing data jdbc sink.");