Skip to content

[Refactor] Consolidate HdfsScannerParams into HdfsScannerContext, pass by pointer, and eliminate HdfsScannerState #74643

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dirtysalt merged 12 commits into from
Jun 11, 2026
Merged

[Refactor] Consolidate HdfsScannerParams into HdfsScannerContext, pass by pointer, and eliminate HdfsScannerState #74643

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
62ddc90
update files
dirtysalt Jun 10, 2026
3f699de
style: auto-fix clang-format issues
wanpengfei-git Jun 10, 2026
f58bf80
update files
dirtysalt Jun 11, 2026
0d35ef1
update files
dirtysalt Jun 10, 2026
ec7967c
update files
dirtysalt Jun 11, 2026
064960a
style: auto-fix clang-format issues
wanpengfei-git Jun 11, 2026
07ca5b1
update files
dirtysalt Jun 11, 2026
0dd73f7
refactor
dirtysalt Jun 11, 2026
c11c435
style: auto-fix clang-format issues
wanpengfei-git Jun 11, 2026
480ecec
refactor
dirtysalt Jun 11, 2026
360630c
update files
dirtysalt Jun 11, 2026
394ae4e
update files
dirtysalt Jun 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 13 additions & 13 deletions be/src/connector/deletion_vector/deletion_vector.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,15 +33,15 @@ Status DeletionVector::fill_row_indexes(const SkipRowsContextPtr& skip_rows_ctx)
} else {
std::shared_ptr<SharedBufferedInputStream> shared_buffered_input_stream = nullptr;
std::shared_ptr<CacheInputStream> cache_input_stream = nullptr;
HdfsScanStats app_scan_stats;
HdfsScanStats fs_scan_stats;
HdfsScannerStats app_stats;
HdfsScannerStats fs_stats;

ASSIGN_OR_RETURN(auto path, get_absolute_path(_params.table_location));
ASSIGN_OR_RETURN(auto path, get_absolute_path(_ctx.table_location));
int64_t offset = _deletion_vector_descriptor->offset;
int64_t length = _deletion_vector_descriptor->sizeInBytes;

ASSIGN_OR_RETURN(auto dv_file, open_random_access_file(path, fs_scan_stats, app_scan_stats,
shared_buffered_input_stream, cache_input_stream));
ASSIGN_OR_RETURN(auto dv_file, open_random_access_file(path, fs_stats, app_stats, shared_buffered_input_stream,
cache_input_stream));
// Check the dv size
uint32_t size_from_deletion_vector_file;
RETURN_IF_ERROR(dv_file->read_at_fully(offset, &size_from_deletion_vector_file, DV_SIZE_LENGTH));
Expand Down Expand Up @@ -70,22 +70,22 @@ Status DeletionVector::fill_row_indexes(const SkipRowsContextPtr& skip_rows_ctx)
std::vector<char> deletion_vector(serialized_bitmap_length);
RETURN_IF_ERROR(dv_file->read_at_fully(offset, deletion_vector.data(), serialized_bitmap_length));

update_dv_file_io_counter(_params.profile.runtime_profile, app_scan_stats, fs_scan_stats, cache_input_stream,
update_dv_file_io_counter(_ctx.profile.runtime_profile, app_stats, fs_stats, cache_input_stream,
shared_buffered_input_stream);
return deserialized_deletion_vector(magic_number_from_deletion_vector_file, deletion_vector,
serialized_bitmap_length, skip_rows_ctx);
}
}

StatusOr<std::unique_ptr<RandomAccessFile>> DeletionVector::open_random_access_file(
const std::string& file_path, HdfsScanStats& fs_scan_stats, HdfsScanStats& app_scan_stats,
const std::string& file_path, HdfsScannerStats& fs_stats, HdfsScannerStats& app_stats,
std::shared_ptr<SharedBufferedInputStream>& shared_buffered_input_stream,
std::shared_ptr<CacheInputStream>& cache_input_stream) const {
const OpenFileOptions options{.fs = _params.fs,
const OpenFileOptions options{.fs = _ctx.fs,
.file_path = file_path,
.fs_stats = &fs_scan_stats,
.app_stats = &app_scan_stats,
.datacache_options = _params.datacache_options};
.fs_stats = &fs_stats,
.app_stats = &app_stats,
.datacache_options = _ctx.datacache_options};
ASSIGN_OR_RETURN(auto file,
HdfsScanner::create_random_access_file(shared_buffered_input_stream, cache_input_stream, options));
std::vector<SharedBufferedInputStream::IORange> io_ranges{};
Expand Down Expand Up @@ -134,7 +134,7 @@ Status DeletionVector::deserialized_deletion_vector(uint32_t magic_number, std::
skip_rows_ctx->deletion_bitmap = std::make_shared<DeletionBitmap>(bitmap);
}
#ifndef BE_TEST
update_dv_build_counter(_params.profile.runtime_profile, _build_stats);
update_dv_build_counter(_ctx.profile.runtime_profile, _build_stats);
#endif
return Status::OK();
}
Expand Down Expand Up @@ -172,7 +172,7 @@ std::string DeletionVector::assemble_deletion_vector_path(const string& table_lo
}

void DeletionVector::update_dv_file_io_counter(
RuntimeProfile* parent_profile, const HdfsScanStats& app_stats, const HdfsScanStats& fs_stats,
RuntimeProfile* parent_profile, const HdfsScannerStats& app_stats, const HdfsScannerStats& fs_stats,
const std::shared_ptr<CacheInputStream>& cache_input_stream,
const std::shared_ptr<SharedBufferedInputStream>& shared_buffered_input_stream) {
const std::string DV_TIMER = DeletionVector::DELETION_VECTOR;
Expand Down
13 changes: 6 additions & 7 deletions be/src/connector/deletion_vector/deletion_vector.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,9 +26,8 @@ struct DeletionVectorBuildStats {

class DeletionVector {
public:
DeletionVector(const HdfsScannerParams& scanner_params)
: _deletion_vector_descriptor(scanner_params.table_specific.deletion_vector_descriptor),
_params(scanner_params) {}
DeletionVector(const HdfsScannerContext& ctx)
: _deletion_vector_descriptor(ctx.table_specific.deletion_vector_descriptor), _ctx(ctx) {}

Status fill_row_indexes(const SkipRowsContextPtr& skip_rows_ctx);
Status deserialized_inline_dv(std::string& encoded_bitmap_data, const SkipRowsContextPtr& skip_rows_ctx);
Expand All @@ -47,7 +46,7 @@ class DeletionVector {

private:
StatusOr<std::unique_ptr<RandomAccessFile>> open_random_access_file(
const std::string& file_path, HdfsScanStats& fs_scan_stats, HdfsScanStats& app_scan_stats,
const std::string& file_path, HdfsScannerStats& fs_stats, HdfsScannerStats& app_stats,
std::shared_ptr<SharedBufferedInputStream>& shared_buffered_input_stream,
std::shared_ptr<CacheInputStream>& cache_input_stream) const;

Expand All @@ -57,15 +56,15 @@ class DeletionVector {
std::string assemble_deletion_vector_path(const std::string& table_location, std::string&& uuid,
std::string& prefix) const;

void update_dv_file_io_counter(RuntimeProfile* parent_profile, const HdfsScanStats& app_stats,
const HdfsScanStats& fs_stats,
void update_dv_file_io_counter(RuntimeProfile* parent_profile, const HdfsScannerStats& app_stats,
const HdfsScannerStats& fs_stats,
const std::shared_ptr<CacheInputStream>& cache_input_stream,
const std::shared_ptr<SharedBufferedInputStream>& shared_buffered_input_stream);

void update_dv_build_counter(RuntimeProfile* parent_profile, const DeletionVectorBuildStats& build_stats);

const std::shared_ptr<TDeletionVectorDescriptor> _deletion_vector_descriptor;
const HdfsScannerParams& _params;
const HdfsScannerContext& _ctx;
DeletionVectorBuildStats _build_stats;
};
} // namespace starrocks
Loading
Loading