Coverage Report

Created: 2026-08-18 15:11

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/partial_update_info.h
Line
Count
Source
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
#include <gen_cpp/olap_file.pb.h>
20
21
#include <cstdint>
22
#include <functional>
23
#include <map>
24
#include <memory>
25
#include <set>
26
#include <string>
27
#include <vector>
28
29
#include "common/status.h"
30
#include "core/column/column.h"
31
#include "core/data_type/primitive_type.h"
32
#include "storage/rowset/rowset_fwd.h"
33
#include "storage/tablet/tablet_fwd.h"
34
35
namespace doris {
36
class TabletSchema;
37
class PartialUpdateInfoPB;
38
class BitmapValue;
39
struct RowLocation;
40
class Block;
41
class MutableBlock;
42
class IOlapColumnDataAccessor;
43
namespace segment_v2 {
44
struct HistoricalRowRetrieverContext;
45
}
46
47
struct RowsetWriterContext;
48
struct RowsetId;
49
class BitmapValue;
50
class HistoricalRowFetcher;
51
class OlapBlockDataConvertor;
52
class RowKeyEncoder;
53
struct MowContext;
54
namespace segment_v2 {
55
class MowKeyProbe;
56
}
57
58
class SegmentCacheHandle;
59
60
struct PartialUpdateInfo {
61
    Status init(int64_t tablet_id, int64_t txn_id, const TabletSchema& tablet_schema,
62
                UniqueKeyUpdateModePB unique_key_update_mode, PartialUpdateNewRowPolicyPB policy,
63
                const std::set<std::string>& partial_update_cols, bool is_strict_mode,
64
                int64_t timestamp_ms, int32_t nano_seconds, const std::string& timezone,
65
                const std::string& auto_increment_column, int32_t sequence_map_col_uid = -1,
66
                int64_t cur_max_version = -1);
67
    void to_pb(PartialUpdateInfoPB* partial_update_info) const;
68
    void from_pb(PartialUpdateInfoPB* partial_update_info);
69
    Status handle_new_key(const TabletSchema& tablet_schema,
70
                          const std::function<std::string()>& line,
71
                          BitmapValue* skip_bitmap = nullptr);
72
    std::string summary() const;
73
74
0
    std::string partial_update_mode_str() const {
75
0
        switch (partial_update_mode) {
76
0
        case UniqueKeyUpdateModePB::UPSERT:
77
0
            return "upsert";
78
0
        case UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS:
79
0
            return "partial update";
80
0
        case UniqueKeyUpdateModePB::UPDATE_FLEXIBLE_COLUMNS:
81
0
            return "flexible partial update";
82
0
        }
83
0
        return "";
84
0
    }
85
325
    bool is_partial_update() const { return partial_update_mode != UniqueKeyUpdateModePB::UPSERT; }
86
189
    bool is_fixed_partial_update() const {
87
189
        return partial_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS;
88
189
    }
89
57
    bool is_flexible_partial_update() const {
90
57
        return partial_update_mode == UniqueKeyUpdateModePB::UPDATE_FLEXIBLE_COLUMNS;
91
57
    }
92
16
    UniqueKeyUpdateModePB update_mode() const { return partial_update_mode; }
93
0
    int32_t sequence_map_col_uid() const { return sequence_map_col_unqiue_id; }
94
95
private:
96
    void _generate_default_values_for_missing_cids(const TabletSchema& tablet_schema);
97
98
public:
99
    UniqueKeyUpdateModePB partial_update_mode {UniqueKeyUpdateModePB::UPSERT};
100
    PartialUpdateNewRowPolicyPB partial_update_new_key_policy {PartialUpdateNewRowPolicyPB::APPEND};
101
    int64_t max_version_in_flush_phase {-1};
102
    std::set<std::string> partial_update_input_columns;
103
    std::vector<uint32_t> missing_cids;
104
    std::vector<uint32_t> update_cids;
105
    // if key not exist in old rowset, use default value or null value for the unmentioned cols
106
    // to generate a new row, only available in non-strict mode
107
    bool can_insert_new_rows_in_partial_update {true};
108
    bool is_strict_mode {false};
109
    int64_t timestamp_ms {0};
110
    int32_t nano_seconds {0};
111
    std::string timezone;
112
    bool is_input_columns_contains_auto_inc_column = false;
113
    bool is_schema_contains_auto_inc_column = false;
114
115
    // default values for missing cids
116
    std::vector<std::string> default_values;
117
118
    int32_t sequence_map_col_unqiue_id {-1};
119
};
120
121
// used in mow partial update
122
struct RidAndPos {
123
    uint32_t rid;
124
    // pos in block
125
    size_t pos;
126
};
127
128
class FixedReadPlan {
129
public:
130
    bool empty() const;
131
0
    void clear() { plan.clear(); }
132
    void prepare_to_read(const RowLocation& row_location, size_t pos);
133
    Status read_columns_by_plan(const TabletSchema& tablet_schema,
134
                                std::vector<uint32_t> cids_to_read,
135
                                const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
136
                                Block& block, std::map<uint32_t, uint32_t>* read_index,
137
                                bool force_read_old_delete_signs,
138
                                const signed char* __restrict cur_delete_signs = nullptr) const;
139
    Status fill_missing_columns(const segment_v2::HistoricalRowRetrieverContext& historical_context,
140
                                const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
141
                                const TabletSchema& tablet_schema, Block& full_block,
142
                                const std::vector<bool>& use_default_or_null_flag,
143
                                bool has_default_or_nullable, uint32_t segment_start_pos,
144
                                const Block* block,
145
                                std::vector<signed char>* old_delete_signs = nullptr) const;
146
    Status fill_old_delete_signs(const Block& old_value_block,
147
                                 const std::map<uint32_t, uint32_t>& read_index, size_t num_rows,
148
                                 std::vector<signed char>* old_delete_signs) const;
149
150
private:
151
    std::map<RowsetId, std::map<uint32_t /* segment_id */, std::vector<RidAndPos>>> plan;
152
};
153
154
class FlexibleReadPlan {
155
public:
156
142
    FlexibleReadPlan(bool has_row_store_for_column) : use_row_store(has_row_store_for_column) {}
157
    void prepare_to_read(const RowLocation& row_location, size_t pos,
158
                         const BitmapValue& skip_bitmap);
159
    // for column store
160
    Status read_columns_by_plan(const TabletSchema& tablet_schema,
161
                                const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
162
                                Block& old_value_block,
163
                                std::map<uint32_t, std::map<uint32_t, uint32_t>>* read_index) const;
164
165
    // for row_store
166
    Status read_columns_by_plan(const TabletSchema& tablet_schema,
167
                                const std::vector<uint32_t>& cids_to_read,
168
                                const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
169
                                Block& old_value_block,
170
                                std::map<uint32_t, uint32_t>* read_index) const;
171
    Status fill_non_primary_key_columns(
172
            const segment_v2::HistoricalRowRetrieverContext& historical_context,
173
            const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
174
            const TabletSchema& tablet_schema, Block& full_block,
175
            const std::vector<bool>& use_default_or_null_flag, bool has_default_or_nullable,
176
            uint32_t segment_start_pos, uint32_t block_start_pos, const Block* block,
177
            std::vector<BitmapValue>* skip_bitmaps) const;
178
179
    Status fill_non_primary_key_columns_for_column_store(
180
            const segment_v2::HistoricalRowRetrieverContext& historical_context,
181
            const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
182
            const TabletSchema& tablet_schema, const std::vector<uint32_t>& non_sort_key_cids,
183
            Block& old_value_block, MutableColumns& mutable_full_columns,
184
            const std::vector<bool>& use_default_or_null_flag, bool has_default_or_nullable,
185
            uint32_t segment_start_pos, uint32_t block_start_pos, const Block* block,
186
            std::vector<BitmapValue>* skip_bitmaps) const;
187
    Status fill_non_primary_key_columns_for_row_store(
188
            const segment_v2::HistoricalRowRetrieverContext& historical_context,
189
            const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
190
            const TabletSchema& tablet_schema, const std::vector<uint32_t>& non_sort_key_cids,
191
            Block& old_value_block, MutableColumns& mutable_full_columns,
192
            const std::vector<bool>& use_default_or_null_flag, bool has_default_or_nullable,
193
            uint32_t segment_start_pos, uint32_t block_start_pos, const Block* block,
194
            std::vector<BitmapValue>* skip_bitmaps) const;
195
196
private:
197
    bool use_row_store {false};
198
    // rowset_id -> segment_id -> column unique id -> mappings
199
    std::map<RowsetId, std::map<uint32_t, std::map<uint32_t, std::vector<RidAndPos>>>> plan;
200
    std::map<RowsetId, std::map<uint32_t /* segment_id */, std::vector<RidAndPos>>> row_store_plan;
201
};
202
203
ColumnBitmap* get_mutable_skip_bitmap_column(Block* block, size_t skip_bitmap_col_idx);
204
205
class BlockAggregator {
206
public:
207
    ~BlockAggregator();
208
    // All references must live longer than the aggregator; the flexible fill
209
    // stage builds everything as locals in one apply() scope. The aggregator
210
    // owns its block convertor (key + sequence column slots).
211
    BlockAggregator(TabletSchema& tablet_schema, BaseTabletSPtr tablet,
212
                    std::shared_ptr<MowContext> mow_context,
213
                    const PartialUpdateInfo& partial_update_info, const RowKeyEncoder& key_encoder,
214
                    const segment_v2::MowKeyProbe& probe, HistoricalRowFetcher& fetcher);
215
216
    Status convert_pk_columns(Block* block, size_t row_pos, size_t num_rows,
217
                              std::vector<IOlapColumnDataAccessor*>& key_columns);
218
    Status convert_seq_column(Block* block, size_t row_pos, size_t num_rows,
219
                              IOlapColumnDataAccessor*& seq_column);
220
    Status aggregate_for_flexible_partial_update(
221
            Block* block, size_t num_rows, const std::vector<RowsetSharedPtr>& specified_rowsets,
222
            std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches);
223
224
private:
225
    Status aggregate_for_sequence_column(
226
            Block* block, int num_rows, const std::vector<IOlapColumnDataAccessor*>& key_columns,
227
            IOlapColumnDataAccessor* seq_column,
228
            const std::vector<RowsetSharedPtr>& specified_rowsets,
229
            std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches);
230
    Status aggregate_for_insert_after_delete(
231
            Block* block, size_t num_rows, const std::vector<IOlapColumnDataAccessor*>& key_columns,
232
            const std::vector<RowsetSharedPtr>& specified_rowsets,
233
            std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches);
234
    Status filter_block(Block* block, size_t num_rows, MutableColumnPtr filter_column,
235
                        int duplicate_rows, std::string col_name);
236
237
    Status fill_sequence_column(Block* block, size_t num_rows, const FixedReadPlan& read_plan,
238
                                std::vector<BitmapValue>& skip_bitmaps);
239
240
    void append_or_merge_row(MutableBlock& dst_block, Block* src_block, int rid,
241
                             BitmapValue& skip_bitmap, bool have_delete_sign);
242
    void merge_one_row(MutableBlock& dst_block, Block* src_block, int rid,
243
                       BitmapValue& skip_bitmap);
244
    void append_one_row(MutableBlock& dst_block, Block* src_block, int rid);
245
    void remove_last_n_rows(MutableBlock& dst_block, int n);
246
247
    // aggregate rows with same keys in range [start, end) from block to output_block
248
    Status aggregate_rows(MutableBlock& output_block, Block* block, int start, int end,
249
                          std::string key, std::vector<BitmapValue>* skip_bitmaps,
250
                          const signed char* delete_signs, IOlapColumnDataAccessor* seq_column,
251
                          const std::vector<RowsetSharedPtr>& specified_rowsets,
252
                          std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches);
253
254
    Status _generate_encoded_default_seq_value(std::string* encoded_value);
255
256
    TabletSchema& _tablet_schema;
257
    BaseTabletSPtr _tablet;
258
    std::shared_ptr<MowContext> _mow_context;
259
    const PartialUpdateInfo& _partial_update_info;
260
    const RowKeyEncoder& _key_encoder;
261
    std::unique_ptr<OlapBlockDataConvertor> _convertor;
262
    const segment_v2::MowKeyProbe& _probe;
263
    HistoricalRowFetcher& _fetcher;
264
265
    // used to store state when aggregating rows in block
266
    struct AggregateState {
267
        int rows {0};
268
        bool has_row_with_delete_sign {false};
269
270
27
        bool should_merge() const {
271
27
            return ((rows == 1 && !has_row_with_delete_sign) || rows == 2);
272
27
        }
273
274
21
        void reset() {
275
21
            rows = 0;
276
21
            has_row_with_delete_sign = false;
277
21
        }
278
279
0
        std::string to_string() const {
280
0
            return fmt::format("rows={}, have_delete_row={}", rows, has_row_with_delete_sign);
281
0
        }
282
    } _state {};
283
};
284
285
struct PartialUpdateStats {
286
    int64_t num_rows_updated {0};
287
    int64_t num_rows_new_added {0};
288
    int64_t num_rows_deleted {0};
289
    int64_t num_rows_filtered {0};
290
};
291
} // namespace doris