This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 20ee7a3ec7a [fix](be) Match Iceberg BINARY and FIXED equality delete
keys by bytes (#69503)
20ee7a3ec7a is described below
commit 20ee7a3ec7a011f01622110698b1297651fbbcf9
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Sun Oct 25 08:07:01 2026 +0000
[fix](be) Match Iceberg BINARY and FIXED equality delete keys by bytes
(#69503)
### What problem does this PR solve?
Issue Number: None
Related PR: #66538, #68786
Problem Summary:
**In short.** With the file scanner V1, an Iceberg equality delete whose
key is a BINARY or FIXED column fails the read, and in one case aborts
the BE. This PR matches string and VARBINARY delete keys by their bytes,
as branch-4.1 and branch-4.2 do (#66538).
**Background.** Since #68786 the BE reads Iceberg BINARY and FIXED
columns as VARBINARY. V1 (`enable_file_scanner_v2 = false`) applies
equality deletes in `be/src/format/table/equality_delete.cpp`. For one
key column, `SimpleEqualityDelete` builds a `HybridSet` with
`create_set` on the key type and probes it with the data column. For
several key columns, `MultiEqualityDelete` hashes every row with
`update_hashes_with_value` and confirms a hash hit with `compare_at`.
Neither handles VARBINARY: `create_set` has no VARBINARY set, and
`ColumnVarbinary` does not implement `update_hashes_with_value`. The
delete file and the data file can also carry the same key as STRING on
one side and VARBINARY on the other. `MultiEqualityDelete` rejects that
as a type change, and `SimpleEqualityDelete` probes the set with a
column of the wrong class. branch-4.1 and branch-4.2 (#66538) store,
hash and compare these keys as bytes.
**The problem, and what it cost.** From the new unit tests, run against
master's `equality_delete.cpp`:
| Equality delete key | Master today (V1) | After this PR |
|---|---|---|
| one VARBINARY key | `VARBINARY predicates are not supported` | the
rows with matching bytes are deleted |
| one key, VARBINARY in the delete file and STRING in the data file |
`VARBINARY predicates are not supported` | deleted by bytes |
| one key, STRING in the delete file and VARBINARY in the data file |
the BE aborts: `Bad cast from type:doris::ColumnVarbinary to
doris::ColumnStr<unsigned int>` | deleted by bytes |
| VARBINARY key with an INT key | `Method update_hashes_with_value is
not supported for ColumnVarbinary` | deleted by bytes |
| composite key, STRING in the delete file and VARBINARY in the data
file | `Not support type change` | deleted by bytes |
| composite key whose data column is a const NULL STRING | `Method
update_hashes_with_value is not supported for Const(Nullable(String))` |
the row matching the NULL delete key is deleted |
The const column comes from
`IcebergReaderMixin::_create_missing_equality_delete_value`: when the FE
does not send Iceberg scan semantics v2 and the data file lacks the key
column, the column is filled from `create_column_const(1, Field())`.
**How this PR fixes it.**
- Keys of a string type or VARBINARY are byte keys. A single byte key
goes into a STRING `HybridSet` as bytes, and data rows probe it with
their bytes. A byte delete key against a non-byte data column is still
rejected as a type change.
- For composite keys, a byte key is hashed with xxhash over its bytes
(`HashUtil::xxHash64WithSeed`, and `xxHash64NullWithSeed` for NULL) on
both the delete side and the data side, and a hash hit compares the
bytes. STRING and VARBINARY keys count as the same type.
- Unlike branch-4.2, the byte path also reads const key columns, so a
const NULL key matches a NULL delete key instead of being read as an
empty string.
- Non-byte keys keep the current code path.
**Results.**
- BE unit tests: the new `be/test/format/table/equality_delete_test.cpp`
has the 5 tests of branch-4.2 and one for a const NULL key. With
master's `equality_delete.cpp` all 6 fail (one aborts the test process);
with this PR all 6 pass. The other BE tests around V1 Iceberg deletes
pass too: `IcebergReaderTest`, `ParquetReadLinesTest`,
`IcebergPositionDeleteSysTableReaderTest`,
`IcebergPositionDeleteSysTableV2ReaderTest` and
`VIcebergDeleteSinkTest`.
- No regression test: Spark SQL cannot write equality delete files, and
no docker fixture has an equality delete with a BINARY or FIXED key.
`test_iceberg_uuid_binary_compatibility` covers UUID keys, on the V2
scanner only.
---
be/src/format/table/equality_delete.cpp | 147 ++++++++++++++++++--
be/test/format/table/equality_delete_test.cpp | 193 ++++++++++++++++++++++++++
2 files changed, 328 insertions(+), 12 deletions(-)
diff --git a/be/src/format/table/equality_delete.cpp
b/be/src/format/table/equality_delete.cpp
index 2386e8c1bc1..21071a89114 100644
--- a/be/src/format/table/equality_delete.cpp
+++ b/be/src/format/table/equality_delete.cpp
@@ -17,11 +17,111 @@
#include "format/table/equality_delete.h"
+#include "core/column/column_const.h"
#include "core/column/column_nullable.h"
#include "exprs/create_predicate_function.h"
+#include "util/hash_util.hpp"
namespace doris {
+namespace {
+
+// Iceberg BINARY / FIXED keys are read as VARBINARY and string keys as
STRING, and the delete file and
+// the data file may carry the same key in different ones of these classes (a
delete block holding
+// STRING bytes for a VARBINARY data column, say). Keys of these types
therefore match by their bytes.
+bool is_equality_delete_byte_type(const DataTypePtr& type) {
+ const auto primitive_type = remove_nullable(type)->get_primitive_type();
+ return is_string_type(primitive_type) || is_varbinary(primitive_type);
+}
+
+// The rows of a byte-key column. A const column repeats its single value,
which can be NULL: when the FE does
+// not send Iceberg scan semantics v2, a key column that the data file lacks
is filled from
+// Const(Nullable(String) NULL), and its rows must match a NULL delete key
rather than an empty string.
+class ByteKeyColumn {
+public:
+ explicit ByteKeyColumn(const IColumn& column) : _values(&column) {
+ if (const auto* const_column =
check_and_get_column<ColumnConst>(_values)) {
+ _values = &const_column->get_data_column();
+ _is_const = true;
+ }
+ if (const auto* nullable =
check_and_get_column<ColumnNullable>(_values)) {
+ _null_map = &nullable->get_null_map_data();
+ _values = &nullable->get_nested_column();
+ }
+ }
+
+ bool is_null(size_t row) const {
+ return _null_map != nullptr && (*_null_map)[_is_const ? 0 : row] != 0;
+ }
+
+ StringRef value(size_t row) const { return _values->get_data_at(_is_const
? 0 : row); }
+
+private:
+ const IColumn* _values;
+ const NullMap* _null_map = nullptr;
+ bool _is_const = false;
+};
+
+void insert_byte_values(const IColumn& column, HybridSetBase* byte_set) {
+ const ByteKeyColumn keys(column);
+ for (size_t row = 0; row < column.size(); ++row) {
+ if (keys.is_null(row)) {
+ byte_set->insert(static_cast<const void*>(nullptr));
+ continue;
+ }
+ const StringRef value = keys.value(row);
+ byte_set->insert(&value);
+ }
+}
+
+void find_byte_values(const IColumn& column, const HybridSetBase& byte_set,
+ IColumn::Filter* matches) {
+ const ByteKeyColumn keys(column);
+ for (size_t row = 0; row < column.size(); ++row) {
+ if (keys.is_null(row)) {
+ (*matches)[row] = byte_set.contain_null();
+ continue;
+ }
+ const StringRef value = keys.value(row);
+ (*matches)[row] = byte_set.find(&value);
+ }
+}
+
+void update_equality_delete_hashes(const ColumnWithTypeAndName&
column_and_type,
+ std::vector<uint64_t>* hashes) {
+ if (!is_equality_delete_byte_type(column_and_type.type)) {
+ column_and_type.column->update_hashes_with_value(hashes->data(),
nullptr);
+ return;
+ }
+ // Hash the bytes rather than call update_hashes_with_value: VARBINARY
does not implement it, and a
+ // STRING delete key must land on the same hash as the equal VARBINARY
data key.
+ const ByteKeyColumn keys(*column_and_type.column);
+ for (size_t row = 0; row < column_and_type.column->size(); ++row) {
+ if (keys.is_null(row)) {
+ (*hashes)[row] = HashUtil::xxHash64NullWithSeed((*hashes)[row]);
+ continue;
+ }
+ const StringRef value = keys.value(row);
+ (*hashes)[row] = HashUtil::xxHash64WithSeed(value.data, value.size,
(*hashes)[row]);
+ }
+}
+
+bool equality_delete_values_equal(const ColumnWithTypeAndName& data_column,
size_t data_row,
+ const ColumnWithTypeAndName& delete_column,
size_t delete_row) {
+ if (!is_equality_delete_byte_type(data_column.type) ||
+ !is_equality_delete_byte_type(delete_column.type)) {
+ return data_column.column->compare_at(data_row, delete_row,
*delete_column.column, -1) == 0;
+ }
+ const ByteKeyColumn data_keys(*data_column.column);
+ const ByteKeyColumn delete_keys(*delete_column.column);
+ if (data_keys.is_null(data_row) || delete_keys.is_null(delete_row)) {
+ return data_keys.is_null(data_row) && delete_keys.is_null(delete_row);
+ }
+ return data_keys.value(data_row) == delete_keys.value(delete_row);
+}
+
+} // namespace
+
std::unique_ptr<EqualityDeleteBase> EqualityDeleteBase::get_delete_impl(
const Block* delete_block, const std::vector<int>& delete_col_ids) {
DCHECK_EQ(delete_block->columns(), delete_col_ids.size());
@@ -37,15 +137,22 @@ Status SimpleEqualityDelete::_build_set() {
if (_delete_block->columns() != 1) [[unlikely]] {
return Status::InternalError("Simple equality delete can be only
applied with one column");
}
- auto& column_and_type = _delete_block->get_by_position(0);
+ const auto& column_and_type = _delete_block->get_by_position(0);
auto delete_column_type =
remove_nullable(column_and_type.type)->get_primitive_type();
size_t non_null_rows = _delete_block->rows();
if (const auto* nullable =
check_and_get_column<ColumnNullable>(column_and_type.column.get());
nullable != nullptr) {
non_null_rows = std::ranges::count(nullable->get_null_map_data(),
UInt8(0));
}
- _hybrid_set.reset(create_set(delete_column_type, non_null_rows, true));
- _hybrid_set->insert_fixed_len(column_and_type.column, 0);
+ if (is_equality_delete_byte_type(column_and_type.type)) {
+ // create_set has no VARBINARY set (it throws NOT_IMPLEMENTED), so
every byte key goes into a
+ // STRING set as bytes, whichever of STRING and VARBINARY the delete
file was read as.
+ _hybrid_set.reset(create_set(TYPE_STRING, non_null_rows, true));
+ insert_byte_values(*column_and_type.column, _hybrid_set.get());
+ } else {
+ _hybrid_set.reset(create_set(delete_column_type, non_null_rows, true));
+ _hybrid_set->insert_fixed_len(column_and_type.column, 0);
+ }
return Status::OK();
}
@@ -61,6 +168,17 @@ Status SimpleEqualityDelete::filter_data_block(
auto block_idx = col_name_to_block_idx->at(block_col_name);
auto column_and_type = data_block->get_by_position(block_idx);
+ const bool delete_is_byte =
+
is_equality_delete_byte_type(_delete_block->get_by_position(0).type);
+ const bool byte_compatible =
+ delete_is_byte &&
is_equality_delete_byte_type(column_and_type.type);
+ // The byte set can only be probed with bytes: an INT data column would be
read as StringRef values.
+ if (delete_is_byte && !byte_compatible) [[unlikely]] {
+ return Status::InternalError(
+ "Not support type change in column '{}', src type: {}, target
type: {}",
+ column_and_type.name,
_delete_block->get_by_position(0).type->get_name(),
+ column_and_type.type->get_name());
+ }
size_t rows = data_block->rows();
// _filter: 1 => in _hybrid_set; 0 => not in _hybrid_set
@@ -71,7 +189,9 @@ Status SimpleEqualityDelete::filter_data_block(
_single_filter->assign(rows, UInt8(0));
}
auto column = column_and_type.column->convert_to_full_column_if_const();
- if (const auto* nullable =
check_and_get_column<ColumnNullable>(column.get())) {
+ if (byte_compatible) {
+ find_byte_values(*column, *_hybrid_set, _single_filter.get());
+ } else if (const auto* nullable =
check_and_get_column<ColumnNullable>(column.get())) {
const NullMap& null_map = nullable->get_null_map_data();
_hybrid_set->find_batch_nullable(*remove_nullable(column), rows,
null_map, *_single_filter);
if (_hybrid_set->contain_null()) {
@@ -96,8 +216,8 @@ Status MultiEqualityDelete::_build_set() {
size_t rows = _delete_block->rows();
_delete_hashes.clear();
_delete_hashes.resize(rows, 0);
- for (ColumnPtr column : _delete_block->get_columns()) {
- column->update_hashes_with_value(_delete_hashes.data(), nullptr);
+ for (const auto& column : _delete_block->get_columns_with_type_and_name())
{
+ update_equality_delete_hashes(column, &_delete_hashes);
}
for (size_t i = 0; i < rows; ++i) {
_delete_hash_map.insert({_delete_hashes[i], i});
@@ -126,7 +246,11 @@ Status MultiEqualityDelete::filter_data_block(
}
auto column_and_type =
data_block->safe_get_by_position(col_name_to_block_idx->at(block_column_name));
- if (!delete_col.type->equals(*column_and_type.type)) [[unlikely]] {
+ // A STRING delete key may meet a VARBINARY data column for the same
Iceberg BINARY field: both are
+ // hashed and compared as bytes, so only a change to or from a
non-byte type is rejected.
+ const bool byte_compatible =
is_equality_delete_byte_type(delete_col.type) &&
+
is_equality_delete_byte_type(column_and_type.type);
+ if (!delete_col.type->equals(*column_and_type.type) &&
!byte_compatible) [[unlikely]] {
return Status::InternalError(
"Not support type change in column '{}', src type: {},
target type: {}",
block_column_name, delete_col.type->get_name(),
@@ -138,8 +262,7 @@ Status MultiEqualityDelete::filter_data_block(
_data_hashes.clear();
_data_hashes.resize(rows, 0);
for (size_t index : _data_column_index) {
-
data_block->get_by_position(index).column->update_hashes_with_value(_data_hashes.data(),
-
nullptr);
+ update_equality_delete_hashes(data_block->get_by_position(index),
&_data_hashes);
}
auto* filter_data = filter.data();
for (size_t i = 0; i < rows; ++i) {
@@ -159,9 +282,9 @@ Status MultiEqualityDelete::filter_data_block(
bool MultiEqualityDelete::_equal(Block* data_block, size_t data_row_index,
size_t delete_row_index) {
for (size_t i = 0; i < _delete_block->columns(); ++i) {
- ColumnPtr data_col =
data_block->get_by_position(_data_column_index[i]).column;
- ColumnPtr delete_col = _delete_block->get_by_position(i).column;
- if (data_col->compare_at(data_row_index, delete_row_index,
*delete_col, -1) != 0) {
+ if
(!equality_delete_values_equal(data_block->get_by_position(_data_column_index[i]),
+ data_row_index,
_delete_block->get_by_position(i),
+ delete_row_index)) {
return false;
}
}
diff --git a/be/test/format/table/equality_delete_test.cpp
b/be/test/format/table/equality_delete_test.cpp
new file mode 100644
index 00000000000..8e4233cb98e
--- /dev/null
+++ b/be/test/format/table/equality_delete_test.cpp
@@ -0,0 +1,193 @@
+// 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 "format/table/equality_delete.h"
+
+#include <gtest/gtest.h>
+
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "core/column/column_const.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_varbinary.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_varbinary.h"
+
+namespace doris {
+
+namespace {
+
+ColumnWithTypeAndName nullable_varbinary_column(
+ const std::string& name, const
std::vector<std::optional<std::string>>& values) {
+ auto data = ColumnVarbinary::create();
+ auto null_map = ColumnUInt8::create();
+ for (const auto& value : values) {
+ const std::string bytes = value.value_or("");
+ data->insert_data(bytes.data(), bytes.size());
+ null_map->insert_value(!value.has_value());
+ }
+ return {ColumnNullable::create(std::move(data), std::move(null_map)),
+ make_nullable(std::make_shared<DataTypeVarbinary>()), name};
+}
+
+ColumnWithTypeAndName varbinary_column(const std::string& name,
+ const std::vector<std::string>& values)
{
+ auto data = ColumnVarbinary::create();
+ for (const auto& value : values) {
+ data->insert_data(value.data(), value.size());
+ }
+ return {std::move(data), std::make_shared<DataTypeVarbinary>(), name};
+}
+
+ColumnWithTypeAndName nullable_string_column(
+ const std::string& name, const
std::vector<std::optional<std::string>>& values) {
+ auto data = ColumnString::create();
+ auto null_map = ColumnUInt8::create();
+ for (const auto& value : values) {
+ const std::string bytes = value.value_or("");
+ data->insert_data(bytes.data(), bytes.size());
+ null_map->insert_value(!value.has_value());
+ }
+ return {ColumnNullable::create(std::move(data), std::move(null_map)),
+ make_nullable(std::make_shared<DataTypeString>()), name};
+}
+
+ColumnWithTypeAndName string_column(const std::string& name,
+ const std::vector<std::string>& values) {
+ auto data = ColumnString::create();
+ for (const auto& value : values) {
+ data->insert_data(value.data(), value.size());
+ }
+ return {std::move(data), std::make_shared<DataTypeString>(), name};
+}
+
+ColumnWithTypeAndName int_column(const std::string& name, const
std::vector<int32_t>& values) {
+ auto data = ColumnInt32::create();
+ data->get_data().assign(values.begin(), values.end());
+ return {std::move(data), std::make_shared<DataTypeInt32>(), name};
+}
+
+std::vector<UInt8> apply_equality_delete(
+ const Block& delete_block, const std::vector<int>& field_ids, Block*
data_block,
+ const std::unordered_map<std::string, uint32_t>& column_indexes,
+ const std::unordered_map<int, std::string>& field_names) {
+ RuntimeProfile profile("equality_delete_varbinary");
+ auto equality_delete = EqualityDeleteBase::get_delete_impl(&delete_block,
field_ids);
+ EXPECT_TRUE(equality_delete->init(&profile).ok());
+ IColumn::Filter filter(data_block->rows(), 1);
+ EXPECT_TRUE(equality_delete->filter_data_block(data_block,
&column_indexes, field_names, filter)
+ .ok());
+ return {filter.begin(), filter.end()};
+}
+
+} // namespace
+
+TEST(EqualityDeleteTest, NullableSingleVarbinaryKeyUsesByteHashing) {
+ Block delete_block;
+ delete_block.insert(nullable_varbinary_column(
+ "binary_key", {std::nullopt, std::string("delete\0value", 12)}));
+ Block data_block;
+ data_block.insert(nullable_varbinary_column(
+ "binary_key",
+ {std::nullopt, std::string("keep\0value", 10),
std::string("delete\0value", 12)}));
+
+ EXPECT_EQ((std::vector<UInt8> {0, 1, 0}),
+ apply_equality_delete(delete_block, {7}, &data_block,
{{"binary_key", 0}},
+ {{7, "binary_key"}}));
+}
+
+TEST(EqualityDeleteTest, NullableStringDeleteMatchesVarbinaryData) {
+ Block delete_block;
+ delete_block.insert(
+ nullable_string_column("binary_key", {std::nullopt,
std::string("delete\0value", 12)}));
+ Block data_block;
+ data_block.insert(nullable_varbinary_column(
+ "binary_key",
+ {std::nullopt, std::string("keep\0value", 10),
std::string("delete\0value", 12)}));
+
+ EXPECT_EQ((std::vector<UInt8> {0, 1, 0}),
+ apply_equality_delete(delete_block, {7}, &data_block,
{{"binary_key", 0}},
+ {{7, "binary_key"}}));
+}
+
+TEST(EqualityDeleteTest, NullableVarbinaryDeleteMatchesStringData) {
+ Block delete_block;
+ delete_block.insert(nullable_varbinary_column(
+ "binary_key", {std::nullopt, std::string("delete\0value", 12)}));
+ Block data_block;
+ data_block.insert(nullable_string_column(
+ "binary_key",
+ {std::nullopt, std::string("keep\0value", 10),
std::string("delete\0value", 12)}));
+
+ EXPECT_EQ((std::vector<UInt8> {0, 1, 0}),
+ apply_equality_delete(delete_block, {7}, &data_block,
{{"binary_key", 0}},
+ {{7, "binary_key"}}));
+}
+
+TEST(EqualityDeleteTest, CompositeVarbinaryKeyUsesByteHashing) {
+ Block delete_block;
+ delete_block.insert(varbinary_column("binary_key", {"same", "other"}));
+ delete_block.insert(int_column("version", {1, 2}));
+ Block data_block;
+ data_block.insert(varbinary_column("binary_key", {"same", "other",
"same"}));
+ data_block.insert(int_column("version", {2, 2, 3}));
+
+ EXPECT_EQ((std::vector<UInt8> {1, 0, 1}),
+ apply_equality_delete(delete_block, {7, 8}, &data_block,
+ {{"binary_key", 0}, {"version", 1}},
+ {{7, "binary_key"}, {8, "version"}}));
+}
+
+TEST(EqualityDeleteTest, CompositeStringDeleteMatchesVarbinaryData) {
+ Block delete_block;
+ delete_block.insert(string_column("binary_key",
{std::string("same\0bytes", 10), "other"}));
+ delete_block.insert(int_column("version", {1, 2}));
+ Block data_block;
+ data_block.insert(
+ varbinary_column("binary_key", {std::string("same\0bytes", 10),
"other", "other"}));
+ data_block.insert(int_column("version", {2, 2, 3}));
+
+ EXPECT_EQ((std::vector<UInt8> {1, 0, 1}),
+ apply_equality_delete(delete_block, {7, 8}, &data_block,
+ {{"binary_key", 0}, {"version", 1}},
+ {{7, "binary_key"}, {8, "version"}}));
+}
+
+TEST(EqualityDeleteTest, CompositeConstNullStringKeyMatchesNullDeleteKey) {
+ Block delete_block;
+ delete_block.insert(nullable_string_column("string_key", {std::nullopt,
std::string()}));
+ delete_block.insert(int_column("version", {2, 3}));
+ // A key column that the data file lacks can arrive as one const NULL for
all rows.
+ auto const_null = nullable_string_column("string_key", {std::nullopt});
+ Block data_block;
+ data_block.insert({ColumnConst::create(const_null.column, 3),
const_null.type, "string_key"});
+ data_block.insert(int_column("version", {1, 2, 3}));
+
+ EXPECT_EQ((std::vector<UInt8> {1, 0, 1}),
+ apply_equality_delete(delete_block, {7, 8}, &data_block,
+ {{"string_key", 0}, {"version", 1}},
+ {{7, "string_key"}, {8, "version"}}));
+}
+
+} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
