github-actions[bot] commented on code in PR #66758:
URL: https://github.com/apache/doris/pull/66758#discussion_r3794658406
##########
be/test/format_v2/column_mapper_test.cpp:
##########
@@ -4145,6 +4198,10 @@ TEST(ColumnMapperTest,
PredicateAccessPathsCreateDeferredStructOutputProjection)
EXPECT_EQ(request.local_positions.at(LocalColumnId(0)), LocalIndex(0));
EXPECT_EQ(request.non_predicate_position(LocalColumnId(0)), LocalIndex(1));
EXPECT_TRUE(request.is_predicate_only(LocalColumnId(0)));
+ EXPECT_EQ(request.variant_access_paths.at(LocalColumnId(0)),
Review Comment:
[P1] Move these Variant assertions into the following Variant test
This test only maps `STRUCT<a BIGINT, b BIGINT>` and never supplies any
Variant access path. `register_root_variant_access_paths()` explicitly skips
non-Variant roots, so both maps are empty and this first
`.at(LocalColumnId(0))` deterministically throws, making the STRUCT test fail.
These `typed_col` assertions belong in
`PredicateAccessPathsCreateDeferredVariantRootProjection` (after its request
assertions), where that path is configured.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -1018,6 +1623,158 @@ class ParquetVariantShreddedState final : public
VariantShreddedState {
}
private:
+ size_t direct_seek_result_bytes(bool allocated) const {
+ size_t bytes = 0;
+ for (const auto& [path, column] : _direct_seek_multi_path_results) {
+ bytes += allocated ? path.capacity() : path.size();
+ bytes += allocated ? column->allocated_bytes() :
column->byte_size();
+ }
+ bytes += _direct_seek_multi_path_selected_bytes.size() *
+
sizeof(decltype(_direct_seek_multi_path_selected_bytes)::value_type);
+ for (const auto& [path, selected] : _direct_seek_multi_path_values) {
+ bytes += allocated ? path.capacity() : path.size();
+ bytes += allocated ? selected.allocated_bytes() :
selected.byte_size();
+ }
+ return bytes;
+ }
+
+ std::optional<ColumnPtr> find_unshredded_normalized_value(
+ std::span<const VariantShreddedPathSegment> path) const {
+ if (!_complete) {
+ return std::nullopt;
+ }
+ const auto indices = unshredded_child_indices(*_schema);
+ if (!indices.has_value()) {
+ return std::nullopt;
+ }
+
+ std::optional<std::string_view> planned_key;
+ if (path.size() == 1 && path.front().kind ==
VariantShreddedPathSegment::Kind::OBJECT_KEY &&
+ !_planned_direct_seek_paths.empty() &&
+ _physical->size() <=
+ MAX_DIRECT_SEEK_MULTI_PATH_CELLS /
_planned_direct_seek_paths.size()) {
+ planned_key.emplace(path.front().key.data == nullptr ? "" :
path.front().key.data,
+ path.front().key.size);
+ const auto planned = std::lower_bound(
+ _planned_direct_seek_paths.begin(),
_planned_direct_seek_paths.end(),
+ *planned_key,
+ [](std::string_view left, std::string_view right) { return
left < right; });
+ if (planned == _planned_direct_seek_paths.end() || *planned !=
*planned_key) {
+ planned_key.reset();
+ }
+ }
+
+ if (planned_key.has_value()) {
+ DirectResidualMultiSeekResult root_result;
+ DirectResidualSeekResult path_result;
+ ColumnPtr column;
+ int64_t selected_value_bytes = 0;
+ bool built_root = false;
+ bool built_path = false;
+ {
+ std::lock_guard lock(_materialization_lock);
+ if (!_direct_seek_multi_path_values_ready) {
+ // Root/container validation and exact child-interval
lookup run once for the
+ // complete planned set. Child payloads remain opaque
until requested below.
+
SCOPED_TIMER(_profile.variant_direct_residual_seek_time.get());
+ if (!_unshredded_metadata_cache) {
+ _unshredded_metadata_cache =
build_unshredded_metadata_cache(
+ *_schema, *_physical, indices->first);
+ }
+ root_result = seek_unshredded_variant_paths(
+ *_schema, *_physical, indices->second,
*_unshredded_metadata_cache,
+ _direct_seek_container_cache,
+ std::span<const
std::string_view>(_planned_direct_seek_paths),
+ *planned_key);
+ path_result = std::move(root_result.requested_path);
+ _direct_seek_multi_path_values =
std::move(root_result.paths);
+ const auto requested_values =
_direct_seek_multi_path_values.find(*planned_key);
+ DORIS_CHECK(requested_values !=
_direct_seek_multi_path_values.end());
+ _direct_seek_multi_path_values.erase(requested_values);
+ const auto [unused, inserted] =
_direct_seek_multi_path_results.emplace(
+ std::string(*planned_key),
std::move(path_result.column));
+ static_cast<void>(unused);
+ DORIS_CHECK(inserted);
+ const auto [unused_bytes, inserted_bytes] =
+ _direct_seek_multi_path_selected_bytes.emplace(
+ std::string(*planned_key),
path_result.selected_value_bytes);
+ static_cast<void>(unused_bytes);
+ DORIS_CHECK(inserted_bytes);
+ _direct_seek_multi_path_values_ready = true;
+ built_root = true;
+ built_path = true;
+ }
+ auto found =
_direct_seek_multi_path_results.find(*planned_key);
+ if (found == _direct_seek_multi_path_results.end()) {
+ // Preserve selected-branch corruption semantics: only the
path that reached an
+ // executing expression is recursively validated and
rebuilt as an owning value.
+
SCOPED_TIMER(_profile.variant_direct_residual_seek_time.get());
+ const auto selected =
_direct_seek_multi_path_values.find(*planned_key);
+ DORIS_CHECK(selected !=
_direct_seek_multi_path_values.end());
+ const std::string selected_key = selected->first;
+ path_result =
materialize_unshredded_variant_path(selected->second,
+
*_unshredded_metadata_cache);
+ _direct_seek_multi_path_values.erase(selected);
+ const auto [inserted, was_inserted] =
_direct_seek_multi_path_results.emplace(
+ selected_key, std::move(path_result.column));
+ DORIS_CHECK(was_inserted);
+ const auto [unused_bytes, inserted_bytes] =
+ _direct_seek_multi_path_selected_bytes.emplace(
+ selected_key,
path_result.selected_value_bytes);
+ static_cast<void>(unused_bytes);
+ DORIS_CHECK(inserted_bytes);
+ found = inserted;
+ built_path = true;
+ }
+ column = found->second;
+ const auto bytes =
_direct_seek_multi_path_selected_bytes.find(*planned_key);
+ DORIS_CHECK(bytes !=
_direct_seek_multi_path_selected_bytes.end());
+ selected_value_bytes = bytes->second;
+ }
+ if (built_root) {
+
update_counter(_profile.variant_direct_residual_seek_container_index_builds,
+ root_result.container_index_builds);
+
update_counter(_profile.variant_direct_residual_seek_container_index_hits,
+ root_result.container_index_hits);
+
update_counter(_profile.variant_direct_residual_seek_multi_path_batches, 1);
+
update_counter(_profile.variant_direct_residual_seek_multi_path_root_rows,
+ root_result.root_rows);
+ }
+ DORIS_CHECK_LE(_physical->size(),
+
static_cast<size_t>(std::numeric_limits<int64_t>::max()));
+ update_counter(_profile.variant_direct_residual_seek_rows,
Review Comment:
[P2] Do not charge copied bytes again for a cached path result
When this path already exists in `_direct_seek_multi_path_results`, the
branch only returns the cached `ColumnPtr`: `built_path` stays false and
neither seek/materialization timer runs. These unconditional updates
nevertheless add the full stored byte count (and rows) again, although
`VariantDirectResidualSeekBytes` is documented as bytes copied to the result.
Repeated evaluation can therefore inflate reported work arbitrarily while seek
time and `MultiPathPathRows` stay flat—the new repeated-`b` test already
codifies 9 general rows for only 6 built path rows. Gate the work counters on
`built_path` (and use a separate request/cache-hit counter if served-result
frequency is useful).
--
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]
