morningman commented on code in PR #66399: URL: https://github.com/apache/doris/pull/66399#discussion_r3733150356
########## be/src/format_v2/table/fluss_union_lake_reader.cpp: ########## @@ -0,0 +1,524 @@ +// 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_v2/table/fluss_union_lake_reader.h" + +#include <algorithm> +#include <charconv> +#include <memory> +#include <string> +#include <string_view> +#include <utility> +#include <vector> + +#include "common/cast_set.h" +#include "core/assert_cast.h" +#include "core/column/column_vector.h" +#include "exprs/vexpr_context.h" +#include "exprs/vslot_ref.h" +#include "format_v2/column_mapper.h" +#include "format_v2/expr/equality_delete_predicate.h" +#include "format_v2/jni/fluss_jni_reader.h" +#include "format_v2/table/paimon_reader.h" +#include "runtime/descriptors.h" +#include "runtime/file_scan_profile.h" +#include "runtime/runtime_state.h" + +namespace doris::format::fluss { +namespace { + +// The scan-node properties this reader reads. The fluss connector states that `fluss.union.*` is the +// whole of what BE's C++ side knows about fluss; anything added here has to be added there too. +constexpr const char* PROP_PK_NAMES = "fluss.union.pk_names"; +constexpr const char* PROP_MAX_TAIL_ROWS = "fluss.union.max_tail_rows"; + +// The per-range payload of a wrapped lake split. +constexpr const char* PROP_RANGE_TYPE = "fluss.range_type"; +constexpr const char* PROP_TAIL = "fluss.union.tail"; +constexpr const char* RANGE_TYPE_LAKE_SUPPRESS = "LAKE_SUPPRESS"; + +// The range this reader synthesizes to read the tail. A plain bounded log read, which is what the +// suppression set is: every key the tail touched, whatever it ended up saying about it. +constexpr const char* RANGE_TYPE_LOG = "LOG"; +constexpr const char* PROP_PARTITION_ID = "fluss.partition_id"; +constexpr const char* PROP_BUCKET_ID = "fluss.bucket_id"; +constexpr const char* PROP_LOG_START_OFFSET = "fluss.log_start_offset"; +constexpr const char* PROP_LOG_STOP_OFFSET = "fluss.log_stop_offset"; + +constexpr size_t TAIL_BATCH_ROWS = 4096; + +std::vector<std::string_view> split_on(std::string_view value, char separator) { + std::vector<std::string_view> parts; + size_t start = 0; + while (true) { + const auto end = value.find(separator, start); + if (end == std::string_view::npos) { + parts.push_back(value.substr(start)); + return parts; + } + parts.push_back(value.substr(start, end - start)); + start = end + 1; + } +} + +void update_counter(RuntimeProfile::Counter* counter, int64_t value) { + if (counter != nullptr) { + COUNTER_UPDATE(counter, value); + } +} + +template <typename T> +bool parse_integer(std::string_view text, T* value) { + if (text.empty()) { + return false; + } + const auto result = std::from_chars(text.data(), text.data() + text.size(), *value); + return result.ec == std::errc() && result.ptr == text.data() + text.size(); +} + +} // namespace + +Status FlussUnionLakeReader::parse_tail(const std::string& spec, Tail* tail) { + DORIS_CHECK(tail != nullptr); + const auto parts = split_on(spec, ':'); + if (parts.size() != 4) { + return Status::InternalError( + "fluss union read: '{}' is not a log tail of the form " + "partitionId:bucket:start:stop", + spec); + } + // An unpartitioned table leaves the partition segment empty rather than writing a sentinel, so + // that its bucket 0 and a partitioned table's bucket 0 cannot become the same cache entry. + if (!parts[0].empty()) { + int64_t partition_id = 0; + if (!parse_integer(parts[0], &partition_id)) { + return Status::InternalError( + "fluss union read: '{}' has a partition id that is not a " + "number in log tail '{}'", + parts[0], spec); + } + } + Tail parsed; + parsed.partition_id = std::string(parts[0]); + if (!parse_integer(parts[1], &parsed.bucket_id) || + !parse_integer(parts[2], &parsed.start_offset) || + !parse_integer(parts[3], &parsed.stop_offset)) { + return Status::InternalError( + "fluss union read: log tail '{}' has a bucket or offset that is not a number", + spec); + } + if (parsed.start_offset >= parsed.stop_offset) { + // Planning never wraps a lake split whose bucket has nothing left in its log. One arriving + // here means the two halves of this read were bounded by different offsets, and that is a + // duplicated or a missing row either way. + return Status::InternalError( + "fluss union read: a suppressing log tail must contain something, but bucket {} " + "was " + "given [{}, {})", + parsed.bucket_id, parsed.start_offset, parsed.stop_offset); + } + parsed.spec = spec; + *tail = std::move(parsed); + return Status::OK(); +} + +TFileRangeDesc FlussUnionLakeReader::tail_scan_range(const Tail& tail) { + std::map<std::string, std::string> params { + {PROP_RANGE_TYPE, RANGE_TYPE_LOG}, + {PROP_BUCKET_ID, std::to_string(tail.bucket_id)}, + {PROP_LOG_START_OFFSET, std::to_string(tail.start_offset)}, + {PROP_LOG_STOP_OFFSET, std::to_string(tail.stop_offset)}, + }; + if (!tail.partition_id.empty()) { + // Absent rather than -1 on an unpartitioned table: that is how the scanner tells the two + // apart, and fluss subscribes to a bucket of each by a different call. + params.emplace(PROP_PARTITION_ID, tail.partition_id); + } + TTableFormatFileDesc table_format_params; + table_format_params.__set_table_format_type("fluss"); + table_format_params.__set_fluss_params(std::move(params)); + TFileRangeDesc range; + range.__set_table_format_params(std::move(table_format_params)); + range.__set_format_type(TFileFormatType::FORMAT_JNI); + return range; +} + +Status FlussUnionLakeReader::init(format::TableReadOptions&& options) { + RETURN_IF_ERROR(format::TableReader::init(std::move(options))); + RETURN_IF_ERROR(_resolve_union_properties()); + _init_union_profile(); + + VExprContextSPtrs conjuncts; + conjuncts.reserve(_conjuncts.size()); + for (const auto& conjunct : _conjuncts) { + VExprSPtr root; + RETURN_IF_ERROR(format::clone_table_expr_tree(conjunct->root(), &root)); + conjuncts.push_back(VExprContext::create_shared(std::move(root))); + } + _lake_reader = std::make_unique<format::paimon::PaimonHybridReader>(); + RETURN_IF_ERROR(_lake_reader->init({ + .projected_columns = _projected_columns, + .conjuncts = std::move(conjuncts), + .format = _format, + .scan_params = _scan_params, + .io_ctx = _io_ctx, + .runtime_state = _runtime_state, + .scanner_profile = _scanner_profile, + .file_slot_descs = _file_slot_descs, + // Aggregate pushdown is withheld from the lake half on purpose. A COUNT answered from + // paimon's own file metadata would count the very rows this reader is about to suppress, + // and it would do so without ever producing a block to suppress them from. + .push_down_agg_type = TPushAggOp::type::NONE, + .condition_cache_digest = _condition_cache_digest, + })); + if (_batch_size > 0) { + _lake_reader->set_batch_size(_batch_size); + } + return Status::OK(); +} + +Status FlussUnionLakeReader::_resolve_union_properties() { + if (_scan_params == nullptr || !_scan_params->__isset.fluss_properties) { + return Status::InternalError( + "missing fluss_properties for a fluss union read, possibly caused by FE/BE " + "protocol " + "mismatch"); + } + const auto& properties = _scan_params->fluss_properties; + const auto names_it = properties.find(PROP_PK_NAMES); + if (names_it == properties.end() || names_it->second.empty()) { + return Status::InternalError( + "missing '{}' for a fluss union read: without the key columns the lake rows the " + "log " + "tail supersedes cannot be identified", + PROP_PK_NAMES); + } + for (const auto name : split_on(names_it->second, ',')) { + const auto column = std::ranges::find_if( + _projected_columns, + [&](const format::ColumnDefinition& candidate) { return candidate.name == name; }); + if (column == _projected_columns.end()) { + // FE keeps the key columns in the scan's tuple whenever it plans a union read, so this + // means its planning-time decision and its split-planning decision disagreed. Suppressing + // nothing would return every superseded lake row a second time, silently. + return Status::InternalError( + "fluss union read: key column '{}' is not among the columns this scan " + "projects. " + "The lake rows its log tail supersedes cannot be identified without it", + name); + } + _key_column_indexes.push_back( + cast_set<size_t>(std::distance(_projected_columns.begin(), column))); + _key_columns.push_back(*column); + } + + const auto rows_it = properties.find(PROP_MAX_TAIL_ROWS); + if (rows_it == properties.end() || + !parse_integer(std::string_view(rows_it->second), &_max_tail_rows) || _max_tail_rows <= 0) { + return Status::InternalError( + "fluss union read: '{}' must be a positive number of rows, but was '{}'", + PROP_MAX_TAIL_ROWS, + rows_it == properties.end() ? std::string("missing") : rows_it->second); + } + return Status::OK(); +} + +void FlussUnionLakeReader::_init_union_profile() { + if (_scanner_profile == nullptr) { + return; + } + static const char* table_profile = file_scan_profile::TABLE_READER; + _suppressed_rows_counter = ADD_CHILD_COUNTER_WITH_LEVEL( + _scanner_profile, "FlussUnionSuppressedRows", TUnit::UNIT, table_profile, 1); + _tail_keys_read_counter = ADD_CHILD_COUNTER_WITH_LEVEL( + _scanner_profile, "FlussUnionTailKeysRead", TUnit::UNIT, table_profile, 1); + _tail_cache_hit_counter = ADD_CHILD_COUNTER_WITH_LEVEL( + _scanner_profile, "FlussUnionTailCacheHitCount", TUnit::UNIT, table_profile, 1); +} + +Status FlussUnionLakeReader::prepare_split(const format::SplitReadOptions& options) { + DORIS_CHECK(_lake_reader != nullptr); + RETURN_IF_ERROR(_lake_reader->prepare_split(options)); + if (_lake_reader->current_split_pruned()) { + // A pruned split returns no rows, so there is nothing to suppress and no reason to spend a + // read of the tail on it. + return Status::OK(); + } + return _prepare_suppression(options); +} + +Status FlussUnionLakeReader::_prepare_suppression(const format::SplitReadOptions& options) { + const auto& range = options.current_range; + if (!range.__isset.table_format_params || !range.table_format_params.__isset.fluss_params) { + return Status::InternalError( + "missing fluss_params on a fluss union lake split, possibly caused by FE/BE " + "protocol " + "mismatch"); + } + const auto& params = range.table_format_params.fluss_params; + const auto type_it = params.find(PROP_RANGE_TYPE); + if (type_it == params.end() || type_it->second != RANGE_TYPE_LAKE_SUPPRESS) { + return Status::InternalError( + "a fluss union lake split must carry '{}={}', but carries '{}'", PROP_RANGE_TYPE, + RANGE_TYPE_LAKE_SUPPRESS, + type_it == params.end() ? std::string("nothing") : type_it->second); + } + const auto tail_it = params.find(PROP_TAIL); + if (tail_it == params.end()) { + return Status::InternalError("missing '{}' on a fluss union lake split", PROP_TAIL); + } + if (_suppression != nullptr && _suppression_tail_spec == tail_it->second) { + // Consecutive splits of one bucket are common; their suppression is the same one. + return Status::OK(); + } + Tail tail; + RETURN_IF_ERROR(parse_tail(tail_it->second, &tail)); + RETURN_IF_ERROR(_load_suppression_keys(options, tail)); + _suppression_tail_spec = tail_it->second; + return Status::OK(); +} + +Status FlussUnionLakeReader::_load_suppression_keys(const format::SplitReadOptions& options, + const Tail& tail) { + if (options.cache == nullptr) { + return Status::InternalError( + "fluss union read: no split cache to hold the keys of log tail '{}'", tail.spec); + } + // Length-prefixed so that no boundary between the fixed prefix and the tail can be reinterpreted + // as part of the tail itself. One scan node reads one table, so the tail alone identifies it. + const auto cache_key = fmt::format("fluss_union_tail:{}:{}", tail.spec.size(), tail.spec); + Status read_status = Status::OK(); + bool cache_hit = false; + auto* cached = options.cache->get<SuppressionKeys>( Review Comment: Real, and fixed — 0ff6ed493c6 and a92b68c45e8. Two corrections to the scale first, because they change which fix is the right one. **What is actually retained.** Only the key `Block` goes in the cache. The expensive part per row — `EqualityDeletePredicate`’s `std::multimap<uint64_t, size_t>`, ~56 bytes a row — is *not* cached: `_suppression` holds one predicate at a time and drops the previous one when the tail changes. And 2,000,000 is the abort threshold, not a typical tail; a tail is by design the data written within one `table.datalake.freshness` period. So the pathological case is the one you identified — `partitions * buckets` tails resident at once — rather than the per-tail size. **This is a property of `ShardedKVCache`, not something fluss introduced**: `KVCache` is an `unordered_map` freed at destruction, and iceberg’s equality-delete filters and position-delete files use the same cache with the same lifetime. That is not a defence, though: iceberg’s entry count is bounded by the number of delete files, fluss’s grows with partitions × buckets, which is a different axis. **The fix, in planning rather than in eviction.** FE already knows the exact size of every tail and was not looking: `stop - lakeEnd` per bucket, straight from the offsets fluss reports while planning. So: - `fluss.union_read.max_total_tail_rows` bounds the sum across the scan (default 10× the per-bucket ceiling). A value below the per-bucket ceiling is refused at CREATE rather than silently taken as a smaller per-bucket limit, which would make every later error quote a number nobody configured. - The per-bucket ceiling is now checked at planning too. It used to be left to the readers, so the answer came back only after one of them had read two million rows. Over either ceiling, `auto` gives up the lake half exactly as a truncated tail already does — and the fluss-only read that replaces it caches no tail at all, which is what makes this a real bound rather than a smaller one; `required` fails with the number that tripped it. **On deduplication — you were right, and it was worth more than I expected.** A tail names each key once per change-log *record*, and an update writes two, so the repeats were being retained and indexed for nothing. `EqualityDeletePredicate::distinct_rows` deduplicates before the block is cached. I put it on the predicate rather than beside the caller deliberately: that class owns the definition of when two key rows are equal, and a caller that decided for itself could call two keys equal where the matching would not — the row it then dropped would be a suppression that silently stops happening. The test pins that directly: the predicate built from the deduplicated block answers exactly what the one built from the raw block answers, across composite keys, NULLs and hash-colliding neighbours. Deduplication happens only after the tail is read in full, so the limit keeps counting records — the same yardstick the Java `PK_TAIL` replay applies to the same range. `FlussUnionTailKe ysRead` against the new `FlussUnionTailKeysRetained` shows the ratio in the profile. **What I did not do:** ref-counted release after the last split using a tail. `KVCache::erase` exists, but FE does not decide which BE gets which split, so it would mean either a new FE→BE contract for the per-BE reference count or teaching the shared `ShardedKVCache` about ref-counted entries — a cross-connector facility, for a bound that the planning-time budget already provides. Happy to do it as a follow-up if you would rather have the release than the budget. -- 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]
