Coverage Report

Created: 2026-10-25 20:03

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
110
bool is_equality_delete_byte_type(const DataTypePtr& type) {
33
110
    const auto primitive_type = remove_nullable(type)->get_primitive_type();
34
110
    return is_string_type(primitive_type) || is_varbinary(primitive_type);
35
110
}
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
18
                                   std::vector<uint64_t>* hashes) {
92
18
    if (!is_equality_delete_byte_type(column_and_type.type)) {
93
12
        column_and_type.column->update_hashes_with_value(hashes->data(), nullptr);
94
12
        return;
95
12
    }
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
10
                                  const ColumnWithTypeAndName& delete_column, size_t delete_row) {
111
10
    if (!is_equality_delete_byte_type(data_column.type) ||
112
10
        !is_equality_delete_byte_type(delete_column.type)) {
113
7
        return data_column.column->compare_at(data_row, delete_row, *delete_column.column, -1) == 0;
114
7
    }
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
35
        const Block* delete_block, const std::vector<int>& delete_col_ids) {
127
35
    DCHECK_EQ(delete_block->columns(), delete_col_ids.size());
128
35
    if (delete_block->columns() == 1) {
129
32
        return std::make_unique<SimpleEqualityDelete>(delete_block, delete_col_ids);
130
32
    } else {
131
3
        return std::make_unique<MultiEqualityDelete>(delete_block, delete_col_ids);
132
3
    }
133
35
}
134
135
32
Status SimpleEqualityDelete::_build_set() {
136
32
    COUNTER_UPDATE(num_delete_rows, _delete_block->rows());
137
32
    if (_delete_block->columns() != 1) [[unlikely]] {
138
0
        return Status::InternalError("Simple equality delete can be only applied with one column");
139
0
    }
140
32
    const auto& column_and_type = _delete_block->get_by_position(0);
141
32
    auto delete_column_type = remove_nullable(column_and_type.type)->get_primitive_type();
142
32
    size_t non_null_rows = _delete_block->rows();
143
32
    if (const auto* nullable = check_and_get_column<ColumnNullable>(column_and_type.column.get());
144
32
        nullable != nullptr) {
145
32
        non_null_rows = std::ranges::count(nullable->get_null_map_data(), UInt8(0));
146
32
    }
147
32
    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
29
    } else {
153
29
        _hybrid_set.reset(create_set(delete_column_type, non_null_rows, true));
154
29
        _hybrid_set->insert_fixed_len(column_and_type.column, 0);
155
29
    }
156
32
    return Status::OK();
157
32
}
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
32
        IColumn::Filter& filter) {
163
32
    SCOPED_TIMER(equality_delete_time);
164
32
    DCHECK(_delete_col_ids.size() == 1);
165
32
    auto column_field_id = _delete_col_ids[0];
166
167
32
    const auto& block_col_name = id_to_block_column_name.at(column_field_id);
168
32
    auto block_idx = col_name_to_block_idx->at(block_col_name);
169
170
32
    auto column_and_type = data_block->get_by_position(block_idx);
171
32
    const bool delete_is_byte =
172
32
            is_equality_delete_byte_type(_delete_block->get_by_position(0).type);
173
32
    const bool byte_compatible =
174
32
            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
32
    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
32
    size_t rows = data_block->rows();
184
    //     _filter: 1 => in _hybrid_set; 0 => not in _hybrid_set
185
32
    if (_single_filter == nullptr) {
186
32
        _single_filter = std::make_unique<IColumn::Filter>(rows, 0);
187
32
    } else {
188
        // reset the array capacity and fill all elements using the 0
189
0
        _single_filter->assign(rows, UInt8(0));
190
0
    }
191
32
    auto column = column_and_type.column->convert_to_full_column_if_const();
192
32
    if (byte_compatible) {
193
3
        find_byte_values(*column, *_hybrid_set, _single_filter.get());
194
29
    } else if (const auto* nullable = check_and_get_column<ColumnNullable>(column.get())) {
195
29
        const NullMap& null_map = nullable->get_null_map_data();
196
29
        _hybrid_set->find_batch_nullable(*remove_nullable(column), rows, null_map, *_single_filter);
197
29
        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
29
    } else {
204
0
        _hybrid_set->find_batch(*column, rows, *_single_filter);
205
0
    }
206
    // should reverse _filter
207
32
    auto* filter_data = filter.data();
208
128
    for (size_t i = 0; i < rows; ++i) {
209
96
        filter_data[i] &= !_single_filter->data()[i];
210
96
    }
211
32
    return Status::OK();
212
32
}
213
214
5
Status MultiEqualityDelete::_build_set() {
215
5
    COUNTER_UPDATE(num_delete_rows, _delete_block->rows());
216
5
    size_t rows = _delete_block->rows();
217
5
    _delete_hashes.clear();
218
5
    _delete_hashes.resize(rows, 0);
219
9
    for (const auto& column : _delete_block->get_columns_with_type_and_name()) {
220
9
        update_equality_delete_hashes(column, &_delete_hashes);
221
9
    }
222
13
    for (size_t i = 0; i < rows; ++i) {
223
8
        _delete_hash_map.insert({_delete_hashes[i], i});
224
8
    }
225
5
    _data_column_index.resize(_delete_block->columns());
226
5
    return Status::OK();
227
5
}
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
5
        IColumn::Filter& filter) {
233
5
    SCOPED_TIMER(equality_delete_time);
234
5
    DCHECK_EQ(_delete_block->get_columns_with_type_and_name().size(), _delete_col_ids.size());
235
5
    size_t column_index = 0;
236
237
14
    for (size_t idx = 0; idx < _delete_block->get_columns_with_type_and_name().size(); ++idx) {
238
9
        auto delete_col = _delete_block->get_columns_with_type_and_name()[idx];
239
9
        auto delete_col_id = _delete_col_ids[idx];
240
241
9
        DCHECK(id_to_block_column_name.contains(delete_col_id));
242
9
        const auto& block_column_name = id_to_block_column_name.at(delete_col_id);
243
9
        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
9
        auto column_and_type =
248
9
                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
9
        const bool byte_compatible = is_equality_delete_byte_type(delete_col.type) &&
252
9
                                     is_equality_delete_byte_type(column_and_type.type);
253
9
        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
9
        _data_column_index[column_index++] = col_name_to_block_idx->at(block_column_name);
260
9
    }
261
5
    size_t rows = data_block->rows();
262
5
    _data_hashes.clear();
263
5
    _data_hashes.resize(rows, 0);
264
9
    for (size_t index : _data_column_index) {
265
9
        update_equality_delete_hashes(data_block->get_by_position(index), &_data_hashes);
266
9
    }
267
5
    auto* filter_data = filter.data();
268
19
    for (size_t i = 0; i < rows; ++i) {
269
14
        for (auto beg = _delete_hash_map.lower_bound(_data_hashes[i]),
270
14
                  end = _delete_hash_map.upper_bound(_data_hashes[i]);
271
14
             beg != end; ++beg) {
272
6
            if (filter[i] && _equal(data_block, i, beg->second)) {
273
6
                filter_data[i] = 0;
274
6
                break;
275
6
            }
276
6
        }
277
14
    }
278
279
5
    return Status::OK();
280
5
}
281
282
bool MultiEqualityDelete::_equal(Block* data_block, size_t data_row_index,
283
6
                                 size_t delete_row_index) {
284
16
    for (size_t i = 0; i < _delete_block->columns(); ++i) {
285
10
        if (!equality_delete_values_equal(data_block->get_by_position(_data_column_index[i]),
286
10
                                          data_row_index, _delete_block->get_by_position(i),
287
10
                                          delete_row_index)) {
288
0
            return false;
289
0
        }
290
10
    }
291
6
    return true;
292
6
}
293
294
} // namespace doris