englefly commented on code in PR #68314:
URL: https://github.com/apache/doris/pull/68314#discussion_r4069197851
##########
be/test/storage/segment/column_reader_test.cpp:
##########
@@ -16,34 +16,402 @@
// under the License.
#include "storage/segment/column_reader.h"
+#include <gen_cpp/Descriptors_constants.h>
#include <gen_cpp/Descriptors_types.h>
#include <gen_cpp/olap_file.pb.h>
#include <gen_cpp/segment_v2.pb.h>
#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <chrono>
+#include <cstdint>
+#include <iterator>
#include <memory>
+#include <string>
#include <thread>
+#include <utility>
#include <vector>
#include "agent/be_exec_version_manager.h"
#include "common/config.h"
#include "io/fs/file_reader.h"
+#include "io/fs/file_system.h"
+#include "io/fs/file_writer.h"
+#include "io/fs/local_file_system.h"
+#include "storage/olap_common.h"
#include "storage/segment/column_reader_cache.h"
+#include "storage/segment/column_writer.h"
#include "storage/segment/mock/mock_segment.h"
#include "storage/segment/segment.h"
#include "storage/segment/variant/variant_column_reader.h"
#include "storage/tablet/tablet_schema.h"
+#include "storage/types.h"
#include "util/json/path_in_data.h"
namespace doris::segment_v2 {
+namespace {
+class TestColumnIterator final : public ColumnIterator {
+public:
+ Status seek_to_ordinal(ordinal_t /* ord */) override { return
Status::OK(); }
+
+ Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null)
override {
+ dst->insert_many_defaults(*n);
+ if (has_null != nullptr) {
+ *has_null = false;
+ }
+ return Status::OK();
+ }
+
+ Status read_by_rowids(const rowid_t* /* rowids */, const size_t count,
+ MutableColumnPtr& dst) override {
+ dst->insert_many_defaults(count);
+ return Status::OK();
+ }
+
+ ordinal_t get_current_ordinal() const override { return 0; }
+
+ void force_set_read_requirement(ReadRequirement requirement) {
+ _read_requirement = requirement;
+ }
+
+ using ColumnIterator::AccessPathSplit;
+
+ Result<AccessPathSplit> split_access_paths(const TColumnAccessPaths&
access_paths) const {
+ return _split_access_paths(access_paths);
+ }
+
+ Status check_and_set_meta_read_mode(ReadRequirement
requirement_before_access_path,
+ const TColumnAccessPaths&
access_paths) {
+ auto split = DORIS_TRY(_split_access_paths(access_paths));
+ return _check_and_set_meta_read_mode(requirement_before_access_path,
split);
+ }
+
+ void convert_to_place_holder_column(MutableColumnPtr& dst, size_t count) {
+ _convert_to_place_holder_column(dst, count);
+ }
+};
+
+TColumnAccessPath create_data_access_path(std::vector<std::string> path) {
+ TColumnAccessPath access_path;
+
access_path.__set_version(g_Descriptors_constants.TCOLUMN_ACCESS_PATH_VERSION_TYPED);
+ access_path.__set_type(TAccessPathType::DATA);
+ TDataAccessPath data_access_path;
+ data_access_path.__set_path(std::move(path));
+ access_path.__set_data_access_path(std::move(data_access_path));
+ return access_path;
+}
+
+TColumnAccessPath create_legacy_data_access_path(std::vector<std::string>
path) {
+ TColumnAccessPath access_path;
+ access_path.__set_type(TAccessPathType::DATA);
+ TDataAccessPath data_access_path;
+ data_access_path.__set_path(std::move(path));
+ access_path.__set_data_access_path(std::move(data_access_path));
+ return access_path;
+}
+
+TColumnAccessPath create_meta_access_path(std::vector<std::string> path) {
+ TColumnAccessPath access_path;
+
access_path.__set_version(g_Descriptors_constants.TCOLUMN_ACCESS_PATH_VERSION_TYPED);
+ access_path.__set_type(TAccessPathType::META);
+ TMetaAccessPath meta_access_path;
+ meta_access_path.__set_path(std::move(path));
+ access_path.__set_meta_access_path(std::move(meta_access_path));
+ return access_path;
+}
+
+std::shared_ptr<ColumnReader> create_test_reader(
+ bool is_nullable = false, uint64_t num_rows = 0,
+ FieldType field_type = FieldType::OLAP_FIELD_TYPE_INT) {
+ auto reader = std::make_shared<ColumnReader>();
+ reader->_meta_is_nullable = is_nullable;
+ reader->_num_rows = num_rows;
+ reader->_meta_type = field_type;
+ return reader;
+}
+
+class TrackingColumnIterator final : public ColumnIterator {
+public:
+ Status seek_to_ordinal(ordinal_t ord) override {
+ seek_ordinals.emplace_back(ord);
+ _current_ordinal = ord;
+ return Status::OK();
+ }
+
+ Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null)
override {
+ next_batch_sizes.emplace_back(*n);
+ if (!need_to_read()) {
+ _convert_to_place_holder_column(dst, *n);
+ if (has_null != nullptr) {
+ *has_null = false;
+ }
+ return Status::OK();
+ }
+
+ _recovery_from_place_holder_column(dst);
+ dst->insert_many_defaults(*n);
+ _current_ordinal += *n;
+ if (has_null != nullptr) {
+ *has_null = false;
+ }
+ return Status::OK();
+ }
+
+ Status read_by_rowids(const rowid_t* rowids, const size_t count,
+ MutableColumnPtr& dst) override {
+ read_by_rowids_batches.emplace_back(rowids, rowids + count);
+ if (!need_to_read()) {
+ _convert_to_place_holder_column(dst, count);
+ return Status::OK();
+ }
+
+ _recovery_from_place_holder_column(dst);
+ dst->insert_many_defaults(count);
+ return Status::OK();
+ }
+
+ ordinal_t get_current_ordinal() const override { return _current_ordinal; }
+
+ Status set_access_paths(const TColumnAccessPaths& all_access_paths,
+ const TColumnAccessPaths& predicate_access_paths)
override {
+ routed_all_access_paths = all_access_paths;
+ routed_predicate_access_paths = predicate_access_paths;
+ return ColumnIterator::set_access_paths(all_access_paths,
predicate_access_paths);
+ }
+
+ void collect_prefetchers(
+ std::map<PrefetcherInitMethod, std::vector<SegmentPrefetcher*>>&
prefetchers,
+ PrefetcherInitMethod init_method) override {
+ record_collect_method(init_method);
+ prefetchers[init_method].emplace_back(prefetcher());
+ }
+
+ SegmentPrefetcher* prefetcher() const {
+ return
reinterpret_cast<SegmentPrefetcher*>(const_cast<TrackingColumnIterator*>(this));
+ }
+
+ void clear_tracking() {
+ seek_ordinals.clear();
+ next_batch_sizes.clear();
+ read_by_rowids_batches.clear();
+ collect_methods.clear();
+ routed_all_access_paths.clear();
+ routed_predicate_access_paths.clear();
+ }
+
+ std::vector<ordinal_t> seek_ordinals;
+ std::vector<size_t> next_batch_sizes;
+ std::vector<std::vector<rowid_t>> read_by_rowids_batches;
+ std::vector<PrefetcherInitMethod> collect_methods;
+ TColumnAccessPaths routed_all_access_paths;
+ TColumnAccessPaths routed_predicate_access_paths;
+
+private:
+ void record_collect_method(PrefetcherInitMethod init_method) {
+ collect_methods.emplace_back(init_method);
+ }
+
+ ordinal_t _current_ordinal = 0;
+};
+
+class TrackingFileColumnIterator final : public FileColumnIterator {
+public:
+ explicit TrackingFileColumnIterator(std::shared_ptr<ColumnReader> reader)
+ : FileColumnIterator(std::move(reader)) {}
+
+ Status seek_to_ordinal(ordinal_t ord) override {
+ seek_ordinals.emplace_back(ord);
+ _current_ordinal = ord;
+ return Status::OK();
+ }
+
+ Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null)
override {
+ next_batch_sizes.emplace_back(*n);
+ dst->insert_many_defaults(*n);
+ _current_ordinal += *n;
+ if (has_null != nullptr) {
+ *has_null = false;
+ }
+ return Status::OK();
+ }
+
+ Status read_by_rowids(const rowid_t* rowids, const size_t count,
+ MutableColumnPtr& dst) override {
+ read_by_rowids_batches.emplace_back(rowids, rowids + count);
+ dst->insert_many_defaults(count);
+ return Status::OK();
+ }
+
+ ordinal_t get_current_ordinal() const override { return _current_ordinal; }
+
+ void collect_prefetchers(
+ std::map<PrefetcherInitMethod, std::vector<SegmentPrefetcher*>>&
prefetchers,
+ PrefetcherInitMethod init_method) override {
+ record_collect_method(init_method);
+ prefetchers[init_method].emplace_back(prefetcher());
+ }
+
+ SegmentPrefetcher* prefetcher() const {
+ return
reinterpret_cast<SegmentPrefetcher*>(const_cast<TrackingFileColumnIterator*>(this));
+ }
+
+ std::vector<ordinal_t> seek_ordinals;
+ std::vector<size_t> next_batch_sizes;
+ std::vector<std::vector<rowid_t>> read_by_rowids_batches;
+ std::vector<PrefetcherInitMethod> collect_methods;
+
+private:
+ void record_collect_method(PrefetcherInitMethod init_method) {
+ collect_methods.emplace_back(init_method);
+ }
+
+ ordinal_t _current_ordinal = 0;
+};
+
+class NullMapOnlyFileColumnIterator final : public FileColumnIterator {
+public:
+ explicit NullMapOnlyFileColumnIterator(std::shared_ptr<ColumnReader>
reader)
+ : FileColumnIterator(std::move(reader)) {}
+
+ void force_null_map_only() { _meta_read_mode =
MetaReadMode::NULL_MAP_ONLY; }
+};
+
+MutableColumnPtr create_int_struct_column(size_t field_count) {
+ Columns columns;
+ for (size_t i = 0; i < field_count; ++i) {
+ columns.emplace_back(ColumnInt32::create());
+ }
+ return ColumnStruct::create(std::move(columns));
+}
+
+MutableColumnPtr create_nullable_int_struct_column(size_t field_count) {
+ return ColumnNullable::create(create_int_struct_column(field_count),
ColumnUInt8::create());
+}
+
+MutableColumnPtr create_nullable_int_array_column() {
+ return ColumnNullable::create(
+ ColumnArray::create(ColumnInt32::create(),
ColumnArray::ColumnOffsets::create()),
+ ColumnUInt8::create());
+}
+
+MutableColumnPtr create_nullable_int_map_column() {
+ return ColumnNullable::create(ColumnMap::create(ColumnInt32::create(),
ColumnInt32::create(),
+
ColumnArray::ColumnOffsets::create()),
+ ColumnUInt8::create());
+}
+
+struct TrackingOffsetIterator {
+ OffsetFileColumnIteratorUPtr iterator;
+ TrackingFileColumnIterator* tracker = nullptr;
+};
+
+TrackingOffsetIterator create_tracking_offset_iterator() {
+ auto file_iterator =
std::make_unique<TrackingFileColumnIterator>(create_test_reader());
+ auto* tracker = file_iterator.get();
+ return
{std::make_unique<OffsetFileColumnIterator>(std::move(file_iterator)), tracker};
+}
+} // namespace
+
+static const std::string COLUMN_READER_FILE_TEST_DIR =
"./ut_dir/column_reader_test";
+
class ColumnReaderTest : public ::testing::Test {
protected:
- void SetUp() override {}
- void TearDown() override {}
+ void SetUp() override {
+ _old_disable_storage_page_cache = config::disable_storage_page_cache;
+ config::disable_storage_page_cache = true;
+ auto st =
io::global_local_filesystem()->delete_directory(COLUMN_READER_FILE_TEST_DIR);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ st =
io::global_local_filesystem()->create_directory(COLUMN_READER_FILE_TEST_DIR);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ }
+
+ void TearDown() override {
+ EXPECT_TRUE(
+
io::global_local_filesystem()->delete_directory(COLUMN_READER_FILE_TEST_DIR).ok());
+ config::disable_storage_page_cache = _old_disable_storage_page_cache;
+ }
+
+private:
+ bool _old_disable_storage_page_cache = false;
};
+TEST_F(ColumnReaderTest, NullMapOnlyReadBySparseRowidsAcrossPages) {
+ ColumnMetaPB meta;
+ std::string fname = COLUMN_READER_FILE_TEST_DIR +
"/null_map_only_sparse_rowids";
+ auto fs = io::global_local_filesystem();
+
+ {
+ io::FileWriterPtr file_writer;
+ Status st = fs->create_file(fname, &file_writer);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+
+ ColumnWriterOptions writer_opts;
+ writer_opts.meta = &meta;
+ writer_opts.meta->set_column_id(0);
+ writer_opts.meta->set_unique_id(0);
+
writer_opts.meta->set_type(static_cast<int32_t>(FieldType::OLAP_FIELD_TYPE_INT));
+ writer_opts.meta->set_length(0);
+ writer_opts.meta->set_encoding(PLAIN_ENCODING);
+ writer_opts.meta->set_compression(segment_v2::CompressionTypePB::LZ4F);
+ writer_opts.meta->set_is_nullable(true);
+ writer_opts.data_page_size = sizeof(int32_t) * 2;
+ writer_opts.need_zone_map = false;
+
+ TabletColumn
column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE,
+ FieldType::OLAP_FIELD_TYPE_INT);
+ std::unique_ptr<ColumnWriter> writer;
+ st = ColumnWriter::create(writer_opts, &column, file_writer.get(),
&writer);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ st = writer->init();
+ ASSERT_TRUE(st.ok()) << st.to_string();
+
+ for (int32_t i = 0; i < 6; ++i) {
+ st = writer->append(i == 2, &i);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ }
+
+ st = writer->finish();
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ st = writer->write_data();
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ st = writer->write_ordinal_index();
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ st = file_writer->close();
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ }
+
+ io::FileReaderSPtr file_reader;
+ auto st = fs->open_file(fname, &file_reader);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+
+ ColumnReaderOptions reader_opts;
+ std::shared_ptr<ColumnReader> reader;
+ st = ColumnReader::create(reader_opts, meta, 6, file_reader, &reader);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+
+ NullMapOnlyFileColumnIterator iter(reader);
+ ColumnIteratorOptions iter_opts;
+ OlapReaderStatistics stats;
+ iter_opts.stats = &stats;
+ iter_opts.file_reader = file_reader.get();
+ st = iter.init(iter_opts);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ iter.force_null_map_only();
+
+ MutableColumnPtr dst = ColumnNullable::create(ColumnInt32::create(),
ColumnUInt8::create());
+ const rowid_t rowids[] = {0, 2};
+ st = iter.read_by_rowids(rowids, std::size(rowids), dst);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+
+ ASSERT_EQ(2, dst->size());
+ const auto& nullable_col = assert_cast<const ColumnNullable&>(*dst);
+ const auto& null_map = nullable_col.get_null_map_data();
+ ASSERT_EQ(2, null_map.size());
+ EXPECT_EQ(0, null_map[0]);
+ EXPECT_EQ(1, null_map[1]);
+ EXPECT_EQ(2, nullable_col.get_nested_column().size());
Review Comment:
Fixed in the BE source rather than in the test.
`FileColumnIterator::read_by_rowids()` filled `count` placeholders into the
nested column before the rowid loop and then appended `total_read_count` more
at the end, so for `{0, 2}` the null map had 2 entries while the nested column
had 4 — exactly the inconsistency you describe. The pre-loop insertion is
removed so the nested column is appended once, after the real read count is
known; this also matches master's implementation of #65805.
`ColumnReaderTest.NullMapOnlyReadBySparseRowidsAcrossPages` is unchanged and
now passes (`BUILD_TYPE_UT=Release ./run-be-ut.sh --run
--filter=ColumnReaderTest.NullMapOnlyReadBySparseRowidsAcrossPages` → OK).
##########
be/test/storage/segment/segment_iterator_lazy_pruned_test.cpp:
##########
@@ -0,0 +1,186 @@
+// 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 <gtest/gtest.h>
+
+#include <memory>
+#include <vector>
+
+#include "common/cast_set.h"
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
+#include "storage/olap_common.h"
+#include "storage/segment/column_reader.h"
+#include "storage/tablet/tablet_schema.h"
+
+// Use #define private public to access
SegmentIterator::_read_lazy_pruned_columns()
+// and the small amount of state it consumes. This mirrors the existing
+// segment_iterator_* white-box tests.
+#if defined(__clang__)
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wkeyword-macro"
+#endif
+#define private public
+#include "storage/segment/segment_iterator.h"
+#undef private
+#if defined(__clang__)
+#pragma clang diagnostic pop
+#endif
+
+namespace doris::segment_v2 {
+namespace {
+
+class TrackingLazyColumnIterator final : public ColumnIterator {
+public:
+ Status seek_to_ordinal(ordinal_t ord) override {
+ seek_ordinals.push_back(ord);
+ return Status::OK();
+ }
+
+ Status read_by_rowids(const rowid_t* rowids, const size_t count,
+ MutableColumnPtr& dst) override {
+ read_phases.push_back(_read_phase);
+ read_rowids.assign(rowids, rowids + count);
+ ++read_by_rowids_count;
+
+ auto& int_column = assert_cast<ColumnVector<TYPE_INT>&>(*dst);
+ for (size_t i = 0; i < count; ++i) {
+ int_column.insert_value(cast_set<int32_t>(rowids[i]));
+ }
+ return Status::OK();
+ }
+
+ void finalize_lazy_phase(MutableColumnPtr& dst) override {
+ finalize_phases.push_back(_read_phase);
+ ++finalize_count;
+ }
+
+ ordinal_t get_current_ordinal() const override { return 0; }
+
+ ReadPhase phase() const { return _read_phase; }
+
+ std::vector<ordinal_t> seek_ordinals;
+ std::vector<rowid_t> read_rowids;
+ std::vector<ReadPhase> read_phases;
+ std::vector<ReadPhase> finalize_phases;
+ int read_by_rowids_count = 0;
+ int finalize_count = 0;
+};
+
+TabletSchemaSPtr make_tablet_schema() {
+ TabletSchemaPB schema_pb;
+ schema_pb.set_keys_type(KeysType::DUP_KEYS);
+ auto* col = schema_pb.add_column();
+ col->set_unique_id(0);
+ col->set_name("c0");
+ col->set_type("INT");
+ col->set_is_key(true);
+ col->set_is_nullable(false);
+
+ auto tablet_schema = std::make_shared<TabletSchema>();
+ tablet_schema->init_from_pb(schema_pb);
+ return tablet_schema;
+}
+
+SchemaSPtr make_read_schema(const TabletSchemaSPtr& tablet_schema) {
+ std::vector<ColumnId> read_column_ids(tablet_schema->num_columns());
+ for (uint32_t cid = 0; cid < read_column_ids.size(); ++cid) {
+ read_column_ids[cid] = cid;
+ }
+ return std::make_shared<Schema>(tablet_schema->columns(), read_column_ids);
+}
+
+Block make_int_block() {
+ Block block;
+ block.insert({ColumnInt32::create(), std::make_shared<DataTypeInt32>(),
"c0"});
+ return block;
+}
+
+} // namespace
+
+class SegmentIteratorLazyPrunedTest : public ::testing::Test {
+protected:
+ void SetUp() override {
+ _tablet_schema = make_tablet_schema();
+ _read_schema = make_read_schema(_tablet_schema);
+ }
+
+ std::unique_ptr<SegmentIterator> make_iter(TrackingLazyColumnIterator**
tracking_iter) {
+ auto iter = std::make_unique<SegmentIterator>(nullptr, _read_schema);
+ iter->_opts.tablet_schema = _tablet_schema;
+ iter->_opts.stats = &_stats;
+ iter->_support_lazy_read_pruned_columns.insert(0);
Review Comment:
Fixed: the fixture now builds `_schema_block_id_map` the same way production
does in `_vec_init_lazy_materialization()` — resize to the number of columns in
the read schema, then map every read column id to its block position — before
the tests call `_read_lazy_pruned_columns()` directly. Both tests in the suite
pass now (`--filter=SegmentIteratorLazyPrunedTest.*` → 2 tests, 0 failures).
##########
be/test/storage/segment/column_reader_test.cpp:
##########
@@ -285,7 +3039,7 @@ TEST_F(ColumnReaderTest,
MapReadByRowidsSkipReadingResizesDestination) {
MapFileColumnIterator map_iter(map_reader, std::move(null_iter),
std::move(offsets_iter),
std::move(key_iter), std::move(val_iter));
map_iter.set_column_name("map_col");
- map_iter.set_reading_flag(ColumnIterator::ReadingFlag::SKIP_READING);
+ map_iter.set_read_requirement(ColumnIterator::ReadRequirement::SKIP);
Review Comment:
Fixed: the destination now starts genuinely empty (the extra base offset
entry is gone), and the test is renamed to
`MapReadByRowidsSkipReadingAppendsPlaceholders`, since append — not an absolute
resize — is the intended semantics. It asserts `count` after the first
`read_by_rowids()` and `2 * count` after a second one, so a regression back to
`resize(count)` would be caught.
--
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]
