Gabriel39 commented on code in PR #66399:
URL: https://github.com/apache/doris/pull/66399#discussion_r3708894135
##########
be/src/service/CMakeLists.txt:
##########
@@ -49,6 +49,24 @@ if (${MAKE_TEST} STREQUAL "OFF" AND ${BUILD_BENCHMARK}
STREQUAL "OFF")
# This permits libraries loaded by dlopen to link to the symbols in the
program.
set_target_properties(doris_be PROPERTIES ENABLE_EXPORTS 1)
+ # ...but not the symbols of the RocksDB we link statically. Exporting
those makes this
+ # executable the definition every later-loaded library binds to, and a JNI
library that
+ # carries its own RocksDB then runs half on ours: the fluss scanner
bundles frocksdbjni,
+ # whose librocksdbjni.so defines 2576 rocksdb symbols under names
identical to ours but
+ # was built against the pre-C++11 libstdc++ string ABI. Objects laid out
by one and used
+ # by the other yield a garbage length, an std::bad_alloc that escapes the
JNI frame, and
+ # an aborted BE. Hiding this archive lets that library bind to its own
copy.
+ #
+ # Scoped to the archive rather than dropping ENABLE_EXPORTS: what needs
the exports is
+ # native UDFs (runtime/user_function_cache.cpp dlopens them), and those
use the Doris UDF
+ # ABI, which has nothing to do with RocksDB. Crash stacks do not need it
either -- they are
+ # symbolized from debug info, which is why they name even
anonymous-namespace functions.
+ #
+ # The same library also duplicates zstd, lz4, snappy, bzip2 and zlib
symbols. Those are C
+ # ABIs, stable across versions and layout-free, so they are left alone
until something
+ # shows otherwise -- unlike RocksDB, whose C++ objects are what actually
corrupt.
+ target_link_options(doris_be PRIVATE "-Wl,--exclude-libs,librocksdb.a")
Review Comment:
[P1] Please make this linker option platform-specific. `--exclude-libs` is a
GNU/ELF linker option; Apple ld rejects it with `ld: unknown options:
--exclude-libs`, which is exactly why the current **BE UT (macOS)** check fails
while linking `doris_be`. Gate this option to the supported ELF platforms (and
use a Darwin-specific solution only if the symbol-hiding behavior is needed
there) so macOS can still build.
##########
fe/fe-connector/fe-connector-fluss/src/main/resources/META-INF/services/org.apache.doris.connector.spi.ConnectorProvider:
##########
@@ -0,0 +1 @@
+org.apache.doris.connector.fluss.FlussConnectorProvider
Review Comment:
[P1] Please add the ASF license header using `#` comments, as the existing
connector service descriptors do. This new one-line descriptor currently makes
License Check fail with an invalid/missing license header, so the PR cannot
merge as-is.
##########
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:
[P1] Please bound or release these cached tail-key blocks at the scan level.
`fluss.union_read.max_tail_rows` limits one partition/bucket tail, but
`ShardedKVCache` has no eviction and lives for the entire `FileScanLocalState`,
so every tail touched by every partition and bucket remains resident until the
scan finishes. A partitioned table can therefore retain `partition_count *
bucket_count * max_tail_rows` key records (the default is two million per
tail), even after all lake splits for earlier tails have completed, and hit the
query/BE memory limit although the input can otherwise be streamed. Consider a
scan-wide byte/row budget plus eviction/ref-counting after the last split using
a tail, and deduplicate touched keys before retaining them.
--
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]
