1+ // Licensed to the Apache Software Foundation (ASF) under one
2+ // or more contributor license agreements. See the NOTICE file
3+ // distributed with this work for additional information
4+ // regarding copyright ownership. The ASF licenses this file
5+ // to you under the Apache License, Version 2.0 (the
6+ // "License"); you may not use this file except in compliance
7+ // with the License. You may obtain a copy of the License at
8+ //
9+ // http://www.apache.org/licenses/LICENSE-2.0
10+ //
11+ // Unless required by applicable law or agreed to in writing,
12+ // software distributed under the License is distributed on an
13+ // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+ // KIND, either express or implied. See the License for the
15+ // specific language governing permissions and limitations
16+ // under the License.
17+
18+ #pragma once
19+
20+ #include < memory>
21+
22+ #include " common/status.h"
23+ #include " operator.h"
24+ #include " pipeline/exec/union_sink_operator.h"
25+ #include " pipeline/rec_cte_shared_state.h"
26+ #include " vec/core/block.h"
27+
28+ namespace doris {
29+ #include " common/compile_check_begin.h"
30+ class RuntimeState ;
31+
32+ namespace pipeline {
33+ class DataQueue ;
34+
35+ class RecCTEAnchorSinkOperatorX ;
36+ class RecCTEAnchorSinkLocalState final : public PipelineXSinkLocalState<RecCTESharedState> {
37+ public:
38+ ENABLE_FACTORY_CREATOR (RecCTEAnchorSinkLocalState);
39+ RecCTEAnchorSinkLocalState (DataSinkOperatorXBase* parent, RuntimeState* state)
40+ : Base(parent, state) {}
41+ Status open (RuntimeState* state) override ;
42+
43+ bool is_blockable () const override { return true ; }
44+
45+ private:
46+ friend class RecCTEAnchorSinkOperatorX ;
47+ using Base = PipelineXSinkLocalState<RecCTESharedState>;
48+ using Parent = RecCTEAnchorSinkOperatorX;
49+
50+ vectorized::VExprContextSPtrs _child_expr;
51+ };
52+
53+ class RecCTEAnchorSinkOperatorX MOCK_REMOVE (final )
54+ : public DataSinkOperatorX<RecCTEAnchorSinkLocalState> {
55+ public:
56+ using Base = DataSinkOperatorX<RecCTEAnchorSinkLocalState>;
57+
58+ friend class RecCTEAnchorSinkLocalState ;
59+ RecCTEAnchorSinkOperatorX (int sink_id, int dest_id, const TPlanNode& tnode,
60+ const DescriptorTbl& descs)
61+ : Base (sink_id, tnode.node_id , dest_id),
62+ _row_descriptor (descs, tnode.row_tuples , tnode.nullable_tuples ) {}
63+
64+ ~RecCTEAnchorSinkOperatorX () override = default ;
65+
66+ Status init (const TPlanNode& tnode, RuntimeState* state) override ;
67+ Status prepare (RuntimeState* state) override ;
68+
69+ bool is_serial_operator () const override { return true ; }
70+
71+ DataDistribution required_data_distribution (RuntimeState* /* state*/ const override {
72+ return {ExchangeType::NOOP};
73+ }
74+
75+ Status sink (RuntimeState* state, vectorized::Block* input_block, bool eos) override {
76+ auto & local_state = get_local_state (state);
77+
78+ if (_need_notify_rec_side_ready) {
79+ RETURN_IF_ERROR (get_local_state (state)._shared_state ->send_data_to_targets (state, 0 ));
80+ _need_notify_rec_side_ready = false ;
81+ }
82+
83+ COUNTER_UPDATE (local_state.rows_input_counter (), (int64_t )input_block->rows ());
84+ if (input_block->rows () != 0 ) {
85+ vectorized::Block block;
86+ RETURN_IF_ERROR (materialize_block (local_state._child_expr , input_block, &block, true ));
87+ RETURN_IF_ERROR (local_state._shared_state ->emplace_block (state, std::move (block)));
88+ }
89+
90+ if (eos) {
91+ local_state._shared_state ->anchor_dep ->set_ready ();
92+ }
93+ return Status::OK ();
94+ }
95+
96+ std::shared_ptr<BasicSharedState> create_shared_state () const override {
97+ std::shared_ptr<BasicSharedState> ss = std::make_shared<RecCTESharedState>();
98+ ss->id = operator_id ();
99+ for (const auto & dest : dests_id ()) {
100+ ss->related_op_ids .insert (dest);
101+ }
102+ return ss;
103+ }
104+
105+ private:
106+ const RowDescriptor _row_descriptor;
107+ vectorized::VExprContextSPtrs _child_expr;
108+
109+ bool _need_notify_rec_side_ready = true ;
110+ };
111+
112+ } // namespace pipeline
113+ #include " common/compile_check_end.h"
114+ } // namespace doris
0 commit comments