Gabriel39 commented on code in PR #66360:
URL: https://github.com/apache/doris/pull/66360#discussion_r3702054584
##########
be/src/format_v2/parquet/selection_vector.h:
##########
@@ -173,19 +197,65 @@ class SelectionVector {
}
previous = current;
}
+ if (!_mutable_data_exposed) {
+ _verified_generation = _generation;
+ _verified_count = count;
+ _verified_batch_rows = batch_rows;
+ }
return Status::OK();
}
private:
+ void _materialize_identity() {
+ if (_data != nullptr) {
+ return;
+ }
+ _owned.resize(_size);
+ _data = _owned.data();
+ for (size_t idx = 0; idx < _size; ++idx) {
+ _data[idx] = static_cast<Index>(idx);
+ }
+ }
+
+ size_t _compact(const uint8_t* filter, size_t count, bool
filter_uses_row_index) {
+ DORIS_CHECK(filter != nullptr);
+ DORIS_CHECK(count <= _size);
+ Index* source = _data;
+ if (_data == nullptr) {
+ _owned.resize(_size);
Review Comment:
Fixed in fb1c5d1a06. SelectionVector::resize retains the owned high-water
scratch and grows it only when a later batch exceeds that size. The first
implicit-identity compaction also has a specialized path without per-row
source/coordinate branches. A batch-reset regression test covers retained
scratch, and the final paired microbenchmark removes all previously reported
dense/successive-filter regressions.
##########
be/benchmark/parquet/AGENTS.md:
##########
@@ -46,7 +48,10 @@ be/output/lib/benchmark_test --benchmark_list_tests \
| grep -c '^ParquetKernel/' # currently 80
be/output/lib/benchmark_test --benchmark_list_tests \
- | grep -c '^ParquetReader/' # currently 152
+ | grep -c '^ParquetSelection/' # currently 25
+
+be/output/lib/benchmark_test --benchmark_list_tests \
+ | grep -c '^ParquetReader/' # currently 167
Review Comment:
Fixed in fb1c5d1a06. The current validation record now lists 228 decoder, 92
kernel, 25 selection, 167 reader, and 8 expression-lifecycle cases, matching
the verified Release benchmark registrations.
##########
be/test/format_v2/parquet/parquet_scan_test.cpp:
##########
@@ -3163,17 +3179,82 @@ TEST_F(ParquetScanTest,
PredicateOnlyDictionaryRangeSkipsTypedValueMaterializati
conjunct->close();
}
+TEST_F(ParquetScanTest, DictionaryFiltersAreBuiltFromEachReaderSnapshot) {
+ struct ScanResult {
+ std::vector<int32_t> scores;
+ int64_t typed_compare_columns = 0;
+ };
+
+ auto scan = [&](int32_t lower_bound) {
+ RuntimeProfile profile("profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto reader = create_reader(0, -1, &profile);
+ EXPECT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ EXPECT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
EXPECT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
EXPECT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ request->predicate_only_columns.push_back(format::LocalColumnId(0));
+ auto conjunct =
+ create_int32_function_conjunct(0, "gt", TExprOpcode::GT,
lower_bound, false);
+ EXPECT_TRUE(conjunct->prepare(&state, RowDescriptor()).ok());
+ EXPECT_TRUE(conjunct->open(&state).ok());
+ request->conjuncts.push_back(conjunct);
+ EXPECT_TRUE(reader->open(request).ok());
+
+ ScanResult result;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ EXPECT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ const auto& score_column =
int32_data_column(*block.get_by_position(1).column);
+ for (size_t row = 0; row < rows; ++row) {
+ result.scores.push_back(score_column.get_element(row));
+ }
+ }
+ result.typed_compare_columns = counter_value(profile,
"DictFilterTypedCompareColumns");
+ conjunct->close();
+ EXPECT_TRUE(reader->close().ok());
+ return result;
+ };
+
+ write_dictionary_int_pair_parquet_file(_file_path);
+ const auto first = scan(2);
+ EXPECT_EQ(first.scores, std::vector<int32_t>({30, 40, 50, 60}));
+ EXPECT_EQ(first.typed_compare_columns, 1);
+
+ const auto repeated = scan(2);
+ EXPECT_EQ(repeated.scores, first.scores);
+ EXPECT_EQ(repeated.typed_compare_columns, 1);
+
+ const auto changed_predicate = scan(3);
+ EXPECT_EQ(changed_predicate.scores, std::vector<int32_t>({40, 50, 60}));
+ EXPECT_EQ(changed_predicate.typed_compare_columns, 1);
+
+ write_dictionary_int_pair_parquet_file(_file_path, {1, 2, 7, 8, 9, 10});
Review Comment:
Fixed in fb1c5d1a06. The replacement dictionary is now {7, 1, 8, 2, 9, 3};
the expected score positions change to {10, 30, 50, 60}, so reusing the first
snapshot bitmap can no longer pass.
##########
be/benchmark/parquet/benchmark_parquet_selection.hpp:
##########
@@ -0,0 +1,149 @@
+// 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.
+
+#pragma once
+
+#include <benchmark/benchmark.h>
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <string>
+#include <vector>
+
+#include "format_v2/parquet/selection_vector.h"
+#include "parquet_benchmark_scenarios.h"
+
+namespace doris::parquet_benchmark::selection_detail {
+
+constexpr size_t SELECTION_ROWS = 1UL << 12;
+constexpr int CASCADE_FIRST_SELECTIVITY = 90;
+
+inline std::vector<uint8_t> make_filter(size_t rows, int selectivity_percent,
Pattern pattern) {
+ std::vector<uint8_t> filter(rows, 0);
+ const size_t selected_rows =
+ rows * static_cast<size_t>(std::clamp(selectivity_percent, 0,
100)) / 100;
+ if (pattern == Pattern::CLUSTERED) {
+ std::fill_n(filter.begin(), selected_rows, uint8_t {1});
+ } else if (selected_rows != 0) {
+ for (size_t selected = 0; selected < selected_rows; ++selected) {
+ filter[selected * rows / selected_rows] = 1;
+ }
+ }
+ return filter;
+}
+
+// Keep the benchmark source buildable on revisions before the bulk compaction
helpers. The
+// fallback mirrors the former Parquet scan loops so one named matrix can
compare both revisions.
+template <typename Selection>
+size_t compact_with_row_filter(Selection* selection, const uint8_t* filter,
size_t rows) {
+ if constexpr (requires { selection->compact_with_row_filter(filter, rows);
}) {
+ return selection->compact_with_row_filter(filter, rows);
+ } else {
+ size_t output = 0;
+ for (size_t position = 0; position < rows; ++position) {
+ const auto row = selection->get_index(position);
+ if (filter[row] != 0) {
+ selection->set_index(output++, row);
+ }
+ }
+ return output;
+ }
+}
+
+template <typename Selection>
+size_t compact_with_selection_filter(Selection* selection, const uint8_t*
filter, size_t rows) {
+ if constexpr (requires { selection->compact_with_selection_filter(filter,
rows); }) {
+ return selection->compact_with_selection_filter(filter, rows);
+ } else {
+ size_t output = 0;
+ for (size_t position = 0; position < rows; ++position) {
+ if (filter[position] != 0) {
+ selection->set_index(output++, selection->get_index(position));
+ }
+ }
+ return output;
+ }
+}
+
+inline void run_selection(benchmark::State& state, const SelectionScenario&
scenario) {
+ format::parquet::SelectionVector selection;
+ const auto row_filter =
+ make_filter(SELECTION_ROWS, scenario.selectivity_percent,
scenario.pattern);
+ const auto first_filter =
+ make_filter(SELECTION_ROWS, CASCADE_FIRST_SELECTIVITY,
Pattern::ALTERNATING);
+ const size_t first_selected =
+ static_cast<size_t>(std::count(first_filter.begin(),
first_filter.end(), uint8_t {1}));
+ const auto selection_filter =
+ make_filter(first_selected, scenario.selectivity_percent,
scenario.pattern);
+ const auto& final_filter =
+ scenario.operation == SelectionOperation::ROW_FILTER ? row_filter
: selection_filter;
+ const size_t expected_selected =
+ scenario.operation == SelectionOperation::RESIZE_IDENTITY
+ ? SELECTION_ROWS
+ : static_cast<size_t>(
+ std::count(final_filter.begin(),
final_filter.end(), uint8_t {1}));
+
+ size_t selected_rows = 0;
+ for (auto _ : state) {
+ selection.resize(SELECTION_ROWS);
+ switch (scenario.operation) {
+ case SelectionOperation::RESIZE_IDENTITY:
+ selected_rows = SELECTION_ROWS;
+ break;
+ case SelectionOperation::ROW_FILTER:
+ selected_rows = compact_with_row_filter(&selection,
row_filter.data(), SELECTION_ROWS);
+ break;
+ case SelectionOperation::CASCADE_FILTER:
+ selected_rows =
+ compact_with_row_filter(&selection, first_filter.data(),
SELECTION_ROWS);
+ selected_rows = compact_with_selection_filter(&selection,
selection_filter.data(),
+ selected_rows);
+ break;
+ }
+ benchmark::DoNotOptimize(selected_rows);
+ benchmark::ClobberMemory();
+ }
+ if (selected_rows != expected_selected) {
Review Comment:
Fixed in fb1c5d1a06. The selection benchmark now constructs an independent
expected original-row-ID sequence before timing and compares every final
selection index afterward. This validation runs for row-filter, cascade, and
compatibility-fallback implementations; the final 25-case Release smoke run
completed without benchmark errors.
--
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]
