Coverage Report

Created: 2026-10-27 11:57

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format/table/equality_delete.cpp
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
#include "format/table/equality_delete.h"
19
20
#include "core/column/column_const.h"
21
#include "core/column/column_nullable.h"
22
#include "exprs/create_predicate_function.h"
23
#include "util/hash_util.hpp"
24
25
namespace doris {
26
27
namespace {
28
29
// Iceberg BINARY / FIXED keys are read as VARBINARY and string keys as STRING, and the delete file and
30
// the data file may carry the same key in different ones of these classes (a delete block holding
31
// STRING bytes for a VARBINARY data column, say). Keys of these types therefore match by their bytes.
32
208
bool is_equality_delete_byte_type(const DataTypePtr& type) {
33
208
    const auto primitive_type = remove_nullable(type)->get_primitive_type();
34
208
    return is_string_type(primitive_type) || is_varbinary(primitive_type);
35
208
}
36
37
// The rows of a byte-key column. A const column repeats its single value, which can be NULL: when the FE does
38
// not send Iceberg scan semantics v2, a key column that the data file lacks is filled from
39
// Const(Nullable(String) NULL), and its rows must match a NULL delete key rather than an empty string.
40
class ByteKeyColumn {
41
public:
42
18
    explicit ByteKeyColumn(const IColumn& column) : _values(&column) {
43
18
        if (const auto* const_column = check_and_get_column<ColumnConst>(_values)) {
44
2
            _values = &const_column->get_data_column();
45
2
            _is_const = true;
46
2
        }
47
18
        if (const auto* nullable = check_and_get_column<ColumnNullable>(_values)) {
48
10
            _null_map = &nullable->get_null_map_data();
49
10
            _values = &nullable->get_nested_column();
50
10
        }
51
18
    }
52
53
37
    bool is_null(size_t row) const {
54
37
        return _null_map != nullptr && (*_null_map)[_is_const ? 0 : row] != 0;
55
37
    }
56
57
24
    StringRef value(size_t row) const { return _values->get_data_at(_is_const ? 0 : row); }
58
59
private:
60
    const IColumn* _values;
61
    const NullMap* _null_map = nullptr;
62
    bool _is_const = false;
63
};
64
65
3
void insert_byte_values(const IColumn& column, HybridSetBase* byte_set) {
66
3
    const ByteKeyColumn keys(column);
67
9
    for (size_t row = 0; row < column.size(); ++row) {
68
6
        if (keys.is_null(row)) {
69
3
            byte_set->insert(static_cast<const void*>(nullptr));
70
3
            continue;
71
3
        }
72
3
        const StringRef value = keys.value(row);
73
3
        byte_set->insert(&value);
74
3
    }
75
3
}
76
77
void find_byte_values(const IColumn& column, const HybridSetBase& byte_set,
78
3
                      IColumn::Filter* matches) {
79
3
    const ByteKeyColumn keys(column);
80
12
    for (size_t row = 0; row < column.size(); ++row) {
81
9
        if (keys.is_null(row)) {
82
3
            (*matches)[row] = byte_set.contain_null();
83
3
            continue;
84
3
        }
85
6
        const StringRef value = keys.value(row);
86
6
        (*matches)[row] = byte_set.find(&value);
87
6
    }
88
3
}
89
90
void update_equality_delete_hashes(const ColumnWithTypeAndName& column_and_type,
91
58
                                   std::vector<uint64_t>* hashes) {
92
58
    if (!is_equality_delete_byte_type(column_and_type.type)) {
93
52
        column_and_type.column->update_hashes_with_value(hashes->data(), nullptr);
94
52
        return;
95
52
    }
96
    // Hash the bytes rather than call update_hashes_with_value: VARBINARY does not implement it, and a
97
    // STRING delete key must land on the same hash as the equal VARBINARY data key.
98
6
    const ByteKeyColumn keys(*column_and_type.column);
99
21
    for (size_t row = 0; row < column_and_type.column->size(); ++row) {
100
15
        if (keys.is_null(row)) {
101
4
            (*hashes)[row] = HashUtil::xxHash64NullWithSeed((*hashes)[row]);
102
4
            continue;
103
4
        }
104
11
        const StringRef value = keys.value(row);
105
11
        (*hashes)[row] = HashUtil::xxHash64WithSeed(value.data, value.size, (*hashes)[row]);
106
11
    }
107
6
}
108
109
bool equality_delete_values_equal(const ColumnWithTypeAndName& data_column, size_t data_row,
110
30
                                  const ColumnWithTypeAndName& delete_column, size_t delete_row) {
111
30
    if (!is_equality_delete_byte_type(data_column.type) ||
112
30
        !is_equality_delete_byte_type(delete_column.type)) {
113
27
        return data_column.column->compare_at(data_row, delete_row, *delete_column.column, -1) == 0;
114
27
    }
115
3
    const ByteKeyColumn data_keys(*data_column.column);
116
3
    const ByteKeyColumn delete_keys(*delete_column.column);
117
3
    if (data_keys.is_null(data_row) || delete_keys.is_null(delete_row)) {
118
1
        return data_keys.is_null(data_row) && delete_keys.is_null(delete_row);
119
1
    }
120
2
    return data_keys.value(data_row) == delete_keys.value(delete_row);
121
3
}
122
123
} // namespace
124
125
std::unique_ptr<EqualityDeleteBase> EqualityDeleteBase::get_delete_impl(
126
52
        const Block* delete_block, const std::vector<int>& delete_col_ids) {
127
52
    DCHECK_EQ(delete_block->columns(), delete_col_ids.size());
128
52
    if (delete_block->columns() == 1) {
129
41
        return std::make_unique<SimpleEqualityDelete>(delete_block, delete_col_ids);
130
41
    } else {
131
11
        return std::make_unique<MultiEqualityDelete>(delete_block, delete_col_ids);
132
11
    }
133
52
}
134
135
41
Status SimpleEqualityDelete::_build_set() {
136
41
    COUNTER_UPDATE(num_delete_rows, _delete_block->rows());
137
41
    if (_delete_block->columns() != 1) [[unlikely]] {
138
0
        return Status::InternalError("Simple equality delete can be only applied with one column");
139
0
    }
140
41
    const auto& column_and_type = _delete_block->get_by_position(0);
141
41
    auto delete_column_type = remove_nullable(column_and_type.type)->get_primitive_type();
142
41
    size_t non_null_rows = _delete_block->rows();
143
41
    if (const auto* nullable = check_and_get_column<ColumnNullable>(column_and_type.column.get());
144
41
        nullable != nullptr) {
145
41
        non_null_rows = std::ranges::count(nullable->get_null_map_data(), UInt8(0));
146
41
    }
147
41
    if (is_equality_delete_byte_type(column_and_type.type)) {
148
        // create_set has no VARBINARY set (it throws NOT_IMPLEMENTED), so every byte key goes into a
149
        // STRING set as bytes, whichever of STRING and VARBINARY the delete file was read as.
150
3
        _hybrid_set.reset(create_set(TYPE_STRING, non_null_rows, true));
151
3
        insert_byte_values(*column_and_type.column, _hybrid_set.get());
152
38
    } else {
153
38
        _hybrid_set.reset(create_set(delete_column_type, non_null_rows, true));
154
38
        _hybrid_set->insert_fixed_len(column_and_type.column, 0);
155
38
    }
156
41
    return Status::OK();
157
41
}
158
159
Status SimpleEqualityDelete::filter_data_block(
160
        Block* data_block, const std::unordered_map<std::string, uint32_t>* col_name_to_block_idx,
161
        const std::unordered_map<int, std::string>& id_to_block_column_name,
162
41
        IColumn::Filter& filter) {
163
41
    SCOPED_TIMER(equality_delete_time);
164
41
    DCHECK(_delete_col_ids.size() == 1);
165
41
    auto column_field_id = _delete_col_ids[0];
166
167
41
    const auto& block_col_name = id_to_block_column_name.at(column_field_id);
168
41
    auto block_idx = col_name_to_block_idx->at(block_col_name);
169
170
41
    auto column_and_type = data_block->get_by_position(block_idx);
171
41
    const bool delete_is_byte =
172
41
            is_equality_delete_byte_type(_delete_block->get_by_position(0).type);
173
41
    const bool byte_compatible =
174
41
            delete_is_byte && is_equality_delete_byte_type(column_and_type.type);
175
    // The byte set can only be probed with bytes: an INT data column would be read as StringRef values.
176
41
    if (delete_is_byte && !byte_compatible) [[unlikely]] {
177
0
        return Status::InternalError(
178
0
                "Not support type change in column '{}', src type: {}, target type: {}",
179
0
                column_and_type.name, _delete_block->get_by_position(0).type->get_name(),
180
0
                column_and_type.type->get_name());
181
0
    }
182
183
41
    size_t rows = data_block->rows();
184
    //     _filter: 1 => in _hybrid_set; 0 => not in _hybrid_set
185
41
    if (_single_filter == nullptr) {
186
41
        _single_filter = std::make_unique<IColumn::Filter>(rows, 0);
187
41
    } else {
188
        // reset the array capacity and fill all elements using the 0
189
0
        _single_filter->assign(rows, UInt8(0));
190
0
    }
191
41
    auto column = column_and_type.column->convert_to_full_column_if_const();
192
41
    if (byte_compatible) {
193
3
        find_byte_values(*column, *_hybrid_set, _single_filter.get());
194
38
    } else if (const auto* nullable = check_and_get_column<ColumnNullable>(column.get())) {
195
38
        const NullMap& null_map = nullable->get_null_map_data();
196
38
        _hybrid_set->find_batch_nullable(*remove_nullable(column), rows, null_map, *_single_filter);
197
38
        if (_hybrid_set->contain_null()) {
198
4
            auto* filter_data = _single_filter->data();
199
16
            for (size_t i = 0; i < rows; ++i) {
200
12
                filter_data[i] = filter_data[i] || null_map[i];
201
12
            }
202
4
        }
203
38
    } else {
204
0
        _hybrid_set->find_batch(*column, rows, *_single_filter);
205
0
    }
206
    // should reverse _filter
207
41
    auto* filter_data = filter.data();
208
162
    for (size_t i = 0; i < rows; ++i) {
209
121
        filter_data[i] &= !_single_filter->data()[i];
210
121
    }
211
41
    return Status::OK();
212
41
}
213
214
13
Status MultiEqualityDelete::_build_set() {
215
13
    COUNTER_UPDATE(num_delete_rows, _delete_block->rows());
216
13
    size_t rows = _delete_block->rows();
217
13
    _delete_hashes.clear();
218
13
    _delete_hashes.resize(rows, 0);
219
29
    for (const auto& column : _delete_block->get_columns_with_type_and_name()) {
220
29
        update_equality_delete_hashes(column, &_delete_hashes);
221
29
    }
222
29
    for (size_t i = 0; i < rows; ++i) {
223
16
        _delete_hash_map.insert({_delete_hashes[i], i});
224
16
    }
225
13
    _data_column_index.resize(_delete_block->columns());
226
13
    return Status::OK();
227
13
}
228
229
Status MultiEqualityDelete::filter_data_block(
230
        Block* data_block, const std::unordered_map<std::string, uint32_t>* col_name_to_block_idx,
231
        const std::unordered_map<int, std::string>& id_to_block_column_name,
232
13
        IColumn::Filter& filter) {
233
13
    SCOPED_TIMER(equality_delete_time);
234
13
    DCHECK_EQ(_delete_block->get_columns_with_type_and_name().size(), _delete_col_ids.size());
235
13
    size_t column_index = 0;
236
237
42
    for (size_t idx = 0; idx < _delete_block->get_columns_with_type_and_name().size(); ++idx) {
238
29
        auto delete_col = _delete_block->get_columns_with_type_and_name()[idx];
239
29
        auto delete_col_id = _delete_col_ids[idx];
240
241
29
        DCHECK(id_to_block_column_name.contains(delete_col_id));
242
29
        const auto& block_column_name = id_to_block_column_name.at(delete_col_id);
243
29
        if (!col_name_to_block_idx->contains(block_column_name)) [[unlikely]] {
244
0
            return Status::InternalError("Column '{}' not found in data block: {}",
245
0
                                         block_column_name, data_block->dump_structure());
246
0
        }
247
29
        auto column_and_type =
248
29
                data_block->safe_get_by_position(col_name_to_block_idx->at(block_column_name));
249
        // A STRING delete key may meet a VARBINARY data column for the same Iceberg BINARY field: both are
250
        // hashed and compared as bytes, so only a change to or from a non-byte type is rejected.
251
29
        const bool byte_compatible = is_equality_delete_byte_type(delete_col.type) &&
252
29
                                     is_equality_delete_byte_type(column_and_type.type);
253
29
        if (!delete_col.type->equals(*column_and_type.type) && !byte_compatible) [[unlikely]] {
254
0
            return Status::InternalError(
255
0
                    "Not support type change in column '{}', src type: {}, target type: {}",
256
0
                    block_column_name, delete_col.type->get_name(),
257
0
                    column_and_type.type->get_name());
258
0
        }
259
29
        _data_column_index[column_index++] = col_name_to_block_idx->at(block_column_name);
260
29
    }
261
13
    size_t rows = data_block->rows();
262
13
    _data_hashes.clear();
263
13
    _data_hashes.resize(rows, 0);
264
29
    for (size_t index : _data_column_index) {
265
29
        update_equality_delete_hashes(data_block->get_by_position(index), &_data_hashes);
266
29
    }
267
13
    auto* filter_data = filter.data();
268
47
    for (size_t i = 0; i < rows; ++i) {
269
34
        for (auto beg = _delete_hash_map.lower_bound(_data_hashes[i]),
270
34
                  end = _delete_hash_map.upper_bound(_data_hashes[i]);
271
34
             beg != end; ++beg) {
272
14
            if (filter[i] && _equal(data_block, i, beg->second)) {
273
14
                filter_data[i] = 0;
274
14
                break;
275
14
            }
276
14
        }
277
34
    }
278
279
13
    return Status::OK();
280
13
}
281
282
bool MultiEqualityDelete::_equal(Block* data_block, size_t data_row_index,
283
14
                                 size_t delete_row_index) {
284
44
    for (size_t i = 0; i < _delete_block->columns(); ++i) {
285
30
        if (!equality_delete_values_equal(data_block->get_by_position(_data_column_index[i]),
286
30
                                          data_row_index, _delete_block->get_by_position(i),
287
30
                                          delete_row_index)) {
288
0
            return false;
289
0
        }
290
30
    }
291
14
    return true;
292
14
}
293
294
} // namespace doris