morningman opened a new pull request, #69142:
URL: https://github.com/apache/doris/pull/69142
### What problem does this PR solve?
Issue Number: None
Related PR: #68921
Problem Summary:
**In short.** `ColumnVarbinary` has no CRC32 / CRC32C hash, so every
exchange that partitions rows by a VARBINARY value fails. Since #68786 reads
Iceberg BINARY and FIXED as VARBINARY, MERGE INTO an Iceberg table with an
`identity` or `truncate` partition on a binary column fails, and #68921 has to
leave INSERT into such a table on random writes. This PR hashes VARBINARY
values by their bytes, exactly as `ColumnString` does, and lets the Iceberg
connector route binary partition values on backends that can hash them.
**Background.** An exchange that partitions rows hashes the partition key of
each row with CRC32 (`update_crcs_with_value`, `update_crc_with_value`) or
CRC32C (`update_crc32c_batch`, `update_crc32c_single`): a shuffle join on its
keys, an Iceberg INSERT routed by the transformed partition value (#68921), and
MERGE INTO an Iceberg table, which routes the rows it writes through the same
Iceberg partition function. Array, map and struct keys hash their elements
through the same methods. `ColumnVarbinary` inherits these methods from
`IColumn`, where they throw `NOT_IMPLEMENTED_ERROR`.
**The problem, and what it cost.** On a local single-BE cluster with an
Iceberg REST catalog (tables created by Spark, `parallel_pipeline_task_num =
8`):
| Statement | master | with this PR |
|---|---|---|
| MERGE INTO a table partitioned by `identity(binary_key)` or `truncate(1,
binary_key)`, with a `numbers()` source | fails: `update_crc32c_batch is not
supported for ColumnVarbinary` | succeeds; the rows match what Spark reads back
|
| INSERT OVERWRITE of 10000 rows into four `identity(binary_key)` partitions
from a `numbers()` source | random writes: 8 data files (measured with this PR
at `be_exec_version` 18, the plan master makes) | routed: 4 data files, one per
partition |
| `SELECT ... JOIN [shuffle] ... ON l.a = r.a` on `ARRAY<VARBINARY>` keys |
fails: `update_crc32c_single is not supported for ColumnVarbinary` | same rows
as `JOIN [broadcast]` |
A scalar VARBINARY comparison is rejected during analysis, so a scalar
VARBINARY never becomes a join key; the shuffle join only reaches the exchange
through a nested type.
**How this PR fixes it.**
- **BE.** `ColumnVarbinary` implements the four CRC methods by hashing each
value's bytes, with and without a null map, as `ColumnString` does, so the same
bytes get the same hash in both columns. SipHash and xxHash
(`update_hash_with_value`, `update_hashes_with_value`,
`update_xxHash_with_value`) stay unsupported, and so does tablet routing of
VARBINARY.
- **Mixed versions.** BE exec version 19 marks the capability and FE
`max_be_exec_version` moves to 19.
`IcebergWritePlanProvider.getWriteDistribution` routes a BINARY or FIXED
transformed partition value only when the statement's exec version is at least
19; below it the INSERT keeps random writes, as on master today. `bucket(n,
binary_key)` yields an INT and is routed from version 18 as before. MERGE needs
no FE change.
**Results.**
- **BE UT.** `ColumnVarbinaryTest.ShuffleHashesMatchStringOfTheSameBytes`
checks the four methods against `ColumnString` on empty, NUL, invalid UTF-8 and
inline / out-of-line values, with and without a null map and on a sub-range;
`ArrayOfVarbinaryHashesLikeArrayOfString` does the same through `ColumnArray`
with a NULL element; `ValueHashesAreNotSupported` keeps SipHash and xxHash
unsupported.
`ExternalTableSinkBinaryHashPartitionerTest.IcebergTransformHashesBinaryPartitionValues`
routes `identity` and `truncate[1]` binary values with CRC32 and CRC32C, so
equal partition values share a channel. On master the first two and the
partitioner case fail with `NOT_IMPLEMENTED_ERROR`, since the methods are
inherited from `IColumn`. The three suites (36 cases) pass under ASAN;
clang-tidy reports nothing on the changed files.
- **FE UT.**
`IcebergWritePlanProviderTest#getWriteDistributionRoutesBinaryPartitionValuesOnlyOnBackendsThatHashThem`:
identity / truncate on BINARY, identity on FIXED and a mixed INT + BINARY spec
keep random writes at exec version 18 and are routed at 19; `bucket(16,
binary_key)` is routed at 18. All 1568 tests of fe-connector-iceberg pass.
- **Regression.** `test_iceberg_write_binary_partitions` now checks the data
files of the multi-writer INSERT and INSERT OVERWRITE (4 for identity, 1 for
truncate(1), which folds the keys into one partition) and adds MERGE INTO
identity / truncate binary partitions, compared with Spark.
`test_varbinary_sql_support` adds a broadcast and a shuffle join on
`ARRAY<VARBINARY>` keys with `parallel_pipeline_task_num = 4`; both return the
same 12 rows, and the shuffle plan hash-partitions both sides by the array.
Both suites pass on a local single-BE cluster against the Iceberg REST docker
environment, where their `.out` files were generated; only the new blocks were
added.
- **Mixed versions.** On the same cluster, with `be_exec_version` set to 18
the INSERT OVERWRITE above writes 8 data files (random writes), and with 19 it
writes 4.
### Release note
MERGE INTO an Iceberg table with an `identity` or `truncate` partition on a
binary column no longer fails, and an INSERT into such a table writes each
partition from one writer. A shuffle join on `ARRAY<VARBINARY>` keys no longer
fails.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- `test_iceberg_write_binary_partitions` and
`test_varbinary_sql_support`, run on a local single-BE cluster against the
Iceberg REST docker environment; the `.out` files were generated there.
- [x] Unit Test
- BE: `ColumnVarbinaryTest`, `ExternalTableSinkHashPartitionerTest`,
`ExternalTableSinkBinaryHashPartitionerTest` (36 cases, ASAN).
- FE: all 1568 tests of fe-connector-iceberg, including
`IcebergWritePlanProviderTest`.
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [x] Yes. <!-- Explain the behavior change -->
- Exchanges can partition rows by VARBINARY values; Iceberg INSERT
routes binary partition values on backends at exec version 19.
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
