You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/04/24 20:09:00 UTC
(doris) branch branch-4.0 updated: branch-4.0:[fix](storage) fix IOContext Use-After-Free (#59947) (#62240)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 4a58efd7527 branch-4.0:[fix](storage) fix IOContext Use-After-Free (#59947) (#62240)
4a58efd7527 is described below
commit 4a58efd7527b540985e582cce7ce5c2f05157f7f
Author: zhengyu <zh...@selectdb.com>
AuthorDate: Sat Apr 25 04:08:53 2026 +0800
branch-4.0:[fix](storage) fix IOContext Use-After-Free (#59947) (#62240)
Backport of #59947 to branch-4.0.
Original PR: https://github.com/apache/doris/pull/59947
No compile/test run for this pick.
---------
Signed-off-by: zhengyu <zh...@selectdb.com>
---
be/src/io/fs/benchmark/s3_benchmark.hpp | 4 +-
be/src/io/fs/buffered_reader.cpp | 30 +++++++++--
be/src/io/fs/buffered_reader.h | 21 ++++++--
be/src/service/internal_service.cpp | 20 ++++----
be/src/vec/exec/format/csv/csv_reader.cpp | 26 +++++++---
be/src/vec/exec/format/csv/csv_reader.h | 3 +-
be/src/vec/exec/format/json/new_json_reader.cpp | 33 +++++++++---
be/src/vec/exec/format/json/new_json_reader.h | 5 +-
be/src/vec/exec/format/orc/vorc_reader.cpp | 60 ++++++++++++++++++++--
be/src/vec/exec/format/orc/vorc_reader.h | 10 ++++
be/src/vec/exec/format/parquet/vparquet_reader.cpp | 48 +++++++++++++++++
be/src/vec/exec/format/parquet/vparquet_reader.h | 10 ++++
12 files changed, 229 insertions(+), 41 deletions(-)
diff --git a/be/src/io/fs/benchmark/s3_benchmark.hpp b/be/src/io/fs/benchmark/s3_benchmark.hpp
index 3ad5d71527b..3a4fb16681f 100644
--- a/be/src/io/fs/benchmark/s3_benchmark.hpp
+++ b/be/src/io/fs/benchmark/s3_benchmark.hpp
@@ -114,10 +114,10 @@ public:
fd.path = get_file_path(state);
fd.file_size = _file_size;
io::FileReaderOptions reader_options;
- IOContext io_ctx;
+ auto io_ctx = std::make_shared<IOContext>();
auto reader = DORIS_TRY(io::DelegateReader::create_file_reader(
nullptr, fs_props, fd, reader_options, io::DelegateReader::AccessMode::SEQUENTIAL,
- &io_ctx));
+ io_ctx));
return read(state, reader);
}
};
diff --git a/be/src/io/fs/buffered_reader.cpp b/be/src/io/fs/buffered_reader.cpp
index 2bf6fe8453a..87a301f1d76 100644
--- a/be/src/io/fs/buffered_reader.cpp
+++ b/be/src/io/fs/buffered_reader.cpp
@@ -637,9 +637,14 @@ void PrefetchBuffer::_collect_profile_before_close() {
// buffered reader
PrefetchBufferedReader::PrefetchBufferedReader(RuntimeProfile* profile, io::FileReaderSPtr reader,
- PrefetchRange file_range, const IOContext* io_ctx,
+ PrefetchRange file_range,
+ std::shared_ptr<const IOContext> io_ctx,
int64_t buffer_size)
- : _reader(std::move(reader)), _file_range(file_range), _io_ctx(io_ctx) {
+ : _reader(std::move(reader)), _file_range(file_range), _io_ctx_holder(std::move(io_ctx)) {
+ if (_io_ctx_holder == nullptr) {
+ _io_ctx_holder = std::make_shared<IOContext>();
+ }
+ _io_ctx = _io_ctx_holder.get();
if (buffer_size == -1L) {
buffer_size = config::remote_storage_read_buffer_mb * 1024 * 1024;
}
@@ -676,8 +681,8 @@ PrefetchBufferedReader::PrefetchBufferedReader(RuntimeProfile* profile, io::File
// to make sure the buffer reader will start to read at right position.
for (int i = 0; i < buffer_num; i++) {
_pre_buffers.emplace_back(std::make_shared<PrefetchBuffer>(
- _file_range, s_max_pre_buffer_size, _whole_pre_buffer_size, _reader.get(), _io_ctx,
- sync_buffer));
+ _file_range, s_max_pre_buffer_size, _whole_pre_buffer_size, _reader.get(),
+ _io_ctx_holder, sync_buffer));
}
}
@@ -847,6 +852,23 @@ Result<io::FileReaderSPtr> DelegateReader::create_file_reader(
RuntimeProfile* profile, const FileSystemProperties& system_properties,
const FileDescription& file_description, const io::FileReaderOptions& reader_options,
AccessMode access_mode, const IOContext* io_ctx, const PrefetchRange file_range) {
+ std::shared_ptr<const IOContext> io_ctx_holder;
+ if (io_ctx != nullptr) {
+ // Old API: best-effort safety by copying the IOContext onto the heap.
+ io_ctx_holder = std::make_shared<IOContext>(*io_ctx);
+ }
+ return create_file_reader(profile, system_properties, file_description, reader_options,
+ access_mode, std::move(io_ctx_holder), file_range);
+}
+
+Result<io::FileReaderSPtr> DelegateReader::create_file_reader(
+ RuntimeProfile* profile, const FileSystemProperties& system_properties,
+ const FileDescription& file_description, const io::FileReaderOptions& reader_options,
+ AccessMode access_mode, std::shared_ptr<const IOContext> io_ctx,
+ const PrefetchRange file_range) {
+ if (io_ctx == nullptr) {
+ io_ctx = std::make_shared<IOContext>();
+ }
return FileFactory::create_file_reader(system_properties, file_description, reader_options,
profile)
.transform([&](auto&& reader) -> io::FileReaderSPtr {
diff --git a/be/src/io/fs/buffered_reader.h b/be/src/io/fs/buffered_reader.h
index 0ac256211be..c4c15010e5e 100644
--- a/be/src/io/fs/buffered_reader.h
+++ b/be/src/io/fs/buffered_reader.h
@@ -34,6 +34,7 @@
#include "io/fs/file_reader.h"
#include "io/fs/path.h"
#include "io/fs/s3_file_reader.h"
+#include "io/io_common.h"
#include "olap/olap_define.h"
#include "util/runtime_profile.h"
#include "util/slice.h"
@@ -47,7 +48,6 @@ namespace doris {
namespace io {
class FileSystem;
-struct IOContext;
struct PrefetchRange {
size_t start_offset;
@@ -412,6 +412,12 @@ public:
const FileDescription& file_description, const io::FileReaderOptions& reader_options,
AccessMode access_mode = SEQUENTIAL, const IOContext* io_ctx = nullptr,
const PrefetchRange file_range = PrefetchRange(0, 0));
+
+ static Result<io::FileReaderSPtr> create_file_reader(
+ RuntimeProfile* profile, const FileSystemProperties& system_properties,
+ const FileDescription& file_description, const io::FileReaderOptions& reader_options,
+ AccessMode access_mode, std::shared_ptr<const IOContext> io_ctx,
+ const PrefetchRange file_range = PrefetchRange(0, 0));
};
class PrefetchBufferedReader;
@@ -419,13 +425,14 @@ struct PrefetchBuffer : std::enable_shared_from_this<PrefetchBuffer>, public Pro
enum class BufferStatus { RESET, PENDING, PREFETCHED, CLOSED };
PrefetchBuffer(const PrefetchRange file_range, size_t buffer_size, size_t whole_buffer_size,
- io::FileReader* reader, const IOContext* io_ctx,
+ io::FileReader* reader, std::shared_ptr<const IOContext> io_ctx,
std::function<void(PrefetchBuffer&)> sync_profile)
: _file_range(file_range),
_size(buffer_size),
_whole_buffer_size(whole_buffer_size),
_reader(reader),
- _io_ctx(io_ctx),
+ _io_ctx_holder(std::move(io_ctx)),
+ _io_ctx(_io_ctx_holder.get()),
_sync_profile(std::move(sync_profile)) {}
PrefetchBuffer(PrefetchBuffer&& other)
@@ -435,7 +442,8 @@ struct PrefetchBuffer : std::enable_shared_from_this<PrefetchBuffer>, public Pro
_size(other._size),
_whole_buffer_size(other._whole_buffer_size),
_reader(other._reader),
- _io_ctx(other._io_ctx),
+ _io_ctx_holder(std::move(other._io_ctx_holder)),
+ _io_ctx(_io_ctx_holder.get()),
_buf(std::move(other._buf)),
_sync_profile(std::move(other._sync_profile)) {}
@@ -451,6 +459,7 @@ struct PrefetchBuffer : std::enable_shared_from_this<PrefetchBuffer>, public Pro
size_t _len {0};
size_t _whole_buffer_size;
io::FileReader* _reader = nullptr;
+ std::shared_ptr<const IOContext> _io_ctx_holder;
const IOContext* _io_ctx = nullptr;
vectorized::PODArray<char> _buf;
BufferStatus _buffer_status {BufferStatus::RESET};
@@ -520,7 +529,8 @@ constexpr int64_t s_max_pre_buffer_size = 4 * 1024 * 1024; // 4MB
class PrefetchBufferedReader final : public io::FileReader {
public:
PrefetchBufferedReader(RuntimeProfile* profile, io::FileReaderSPtr reader,
- PrefetchRange file_range, const IOContext* io_ctx = nullptr,
+ PrefetchRange file_range,
+ std::shared_ptr<const IOContext> io_ctx = nullptr,
int64_t buffer_size = -1L);
~PrefetchBufferedReader() override;
@@ -565,6 +575,7 @@ private:
io::FileReaderSPtr _reader;
PrefetchRange _file_range;
const std::vector<PrefetchRange>* _random_access_ranges = nullptr;
+ std::shared_ptr<const IOContext> _io_ctx_holder;
const IOContext* _io_ctx = nullptr;
std::vector<std::shared_ptr<PrefetchBuffer>> _pre_buffers;
int64_t _whole_pre_buffer_size;
diff --git a/be/src/service/internal_service.cpp b/be/src/service/internal_service.cpp
index bbd86d9c56c..39029e62588 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -823,11 +823,11 @@ void PInternalService::fetch_table_schema(google::protobuf::RpcController* contr
std::unique_ptr<RuntimeProfile> profile =
std::make_unique<RuntimeProfile>("FetchTableSchema");
std::unique_ptr<vectorized::GenericReader> reader(nullptr);
- io::IOContext io_ctx;
- io::FileCacheStatistics file_cache_statis;
- io_ctx.file_cache_stats = &file_cache_statis;
- io::FileReaderStats file_reader_stats;
- io_ctx.file_reader_stats = &file_reader_stats;
+ auto io_ctx = std::make_shared<io::IOContext>();
+ auto file_cache_statis = std::make_shared<io::FileCacheStatistics>();
+ auto file_reader_stats = std::make_shared<io::FileReaderStats>();
+ io_ctx->file_cache_stats = file_cache_statis.get();
+ io_ctx->file_reader_stats = file_reader_stats.get();
// file_slots is no use, but the lifetime should be longer than reader
std::vector<SlotDescriptor*> file_slots;
switch (params.format_type) {
@@ -840,25 +840,25 @@ void PInternalService::fetch_table_schema(google::protobuf::RpcController* contr
case TFileFormatType::FORMAT_CSV_LZOP:
case TFileFormatType::FORMAT_CSV_DEFLATE: {
reader = vectorized::CsvReader::create_unique(nullptr, profile.get(), nullptr, params,
- range, file_slots, &io_ctx);
+ range, file_slots, io_ctx.get(), io_ctx);
break;
}
case TFileFormatType::FORMAT_TEXT: {
reader = vectorized::TextReader::create_unique(nullptr, profile.get(), nullptr, params,
- range, file_slots, &io_ctx);
+ range, file_slots, io_ctx.get());
break;
}
case TFileFormatType::FORMAT_PARQUET: {
- reader = vectorized::ParquetReader::create_unique(params, range, &io_ctx, nullptr);
+ reader = vectorized::ParquetReader::create_unique(params, range, io_ctx, nullptr);
break;
}
case TFileFormatType::FORMAT_ORC: {
- reader = vectorized::OrcReader::create_unique(params, range, "", &io_ctx);
+ reader = vectorized::OrcReader::create_unique(params, range, "", io_ctx);
break;
}
case TFileFormatType::FORMAT_JSON: {
reader = vectorized::NewJsonReader::create_unique(profile.get(), params, range,
- file_slots, &io_ctx);
+ file_slots, io_ctx.get(), io_ctx);
break;
}
case TFileFormatType::FORMAT_AVRO: {
diff --git a/be/src/vec/exec/format/csv/csv_reader.cpp b/be/src/vec/exec/format/csv/csv_reader.cpp
index 04fdf1d938b..9db8c4dc560 100644
--- a/be/src/vec/exec/format/csv/csv_reader.cpp
+++ b/be/src/vec/exec/format/csv/csv_reader.cpp
@@ -173,7 +173,8 @@ void PlainCsvTextFieldSplitter::do_split(const Slice& line, std::vector<Slice>*
CsvReader::CsvReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
const TFileScanRangeParams& params, const TFileRangeDesc& range,
- const std::vector<SlotDescriptor*>& file_slot_descs, io::IOContext* io_ctx)
+ const std::vector<SlotDescriptor*>& file_slot_descs, io::IOContext* io_ctx,
+ std::shared_ptr<io::IOContext> io_ctx_holder)
: _profile(profile),
_params(params),
_file_reader(nullptr),
@@ -185,7 +186,11 @@ CsvReader::CsvReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounte
_file_slot_descs(file_slot_descs),
_line_reader_eof(false),
_skip_lines(0),
- _io_ctx(io_ctx) {
+ _io_ctx(io_ctx),
+ _io_ctx_holder(std::move(io_ctx_holder)) {
+ if (_io_ctx == nullptr && _io_ctx_holder) {
+ _io_ctx = _io_ctx_holder.get();
+ }
_file_format_type = _params.format_type;
_is_proto_format = _file_format_type == TFileFormatType::FORMAT_PROTO;
if (_range.__isset.compress_type) {
@@ -552,10 +557,19 @@ Status CsvReader::_create_file_reader(bool need_schema) {
_file_description.mtime = _range.__isset.modification_time ? _range.modification_time : 0;
io::FileReaderOptions reader_options =
FileFactory::get_reader_options(_state, _file_description);
- auto file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
- _profile, _system_properties, _file_description, reader_options,
- io::DelegateReader::AccessMode::SEQUENTIAL, _io_ctx,
- io::PrefetchRange(_range.start_offset, _range.start_offset + _range.size)));
+ io::FileReaderSPtr file_reader;
+ if (_io_ctx_holder) {
+ file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+ _profile, _system_properties, _file_description, reader_options,
+ io::DelegateReader::AccessMode::SEQUENTIAL,
+ std::static_pointer_cast<const io::IOContext>(_io_ctx_holder),
+ io::PrefetchRange(_range.start_offset, _range.start_offset + _range.size)));
+ } else {
+ file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+ _profile, _system_properties, _file_description, reader_options,
+ io::DelegateReader::AccessMode::SEQUENTIAL, _io_ctx,
+ io::PrefetchRange(_range.start_offset, _range.start_offset + _range.size)));
+ }
_file_reader = _io_ctx ? std::make_shared<io::TracingFileReader>(std::move(file_reader),
_io_ctx->file_reader_stats)
: file_reader;
diff --git a/be/src/vec/exec/format/csv/csv_reader.h b/be/src/vec/exec/format/csv/csv_reader.h
index 1596083d9fb..e5600314319 100644
--- a/be/src/vec/exec/format/csv/csv_reader.h
+++ b/be/src/vec/exec/format/csv/csv_reader.h
@@ -173,7 +173,8 @@ class CsvReader : public GenericReader {
public:
CsvReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
const TFileScanRangeParams& params, const TFileRangeDesc& range,
- const std::vector<SlotDescriptor*>& file_slot_descs, io::IOContext* io_ctx);
+ const std::vector<SlotDescriptor*>& file_slot_descs, io::IOContext* io_ctx,
+ std::shared_ptr<io::IOContext> io_ctx_holder = nullptr);
~CsvReader() override = default;
Status init_reader(bool is_load);
diff --git a/be/src/vec/exec/format/json/new_json_reader.cpp b/be/src/vec/exec/format/json/new_json_reader.cpp
index 3fd33053811..946094511b6 100644
--- a/be/src/vec/exec/format/json/new_json_reader.cpp
+++ b/be/src/vec/exec/format/json/new_json_reader.cpp
@@ -80,7 +80,7 @@ using namespace ErrorCode;
NewJsonReader::NewJsonReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
const TFileScanRangeParams& params, const TFileRangeDesc& range,
const std::vector<SlotDescriptor*>& file_slot_descs, bool* scanner_eof,
- io::IOContext* io_ctx)
+ io::IOContext* io_ctx, std::shared_ptr<io::IOContext> io_ctx_holder)
: _vhandle_json_callback(nullptr),
_state(state),
_profile(profile),
@@ -100,7 +100,11 @@ NewJsonReader::NewJsonReader(RuntimeState* state, RuntimeProfile* profile, Scann
_origin_json_doc(&_value_allocator, sizeof(_parse_buffer), &_parse_allocator),
_scanner_eof(scanner_eof),
_current_offset(0),
- _io_ctx(io_ctx) {
+ _io_ctx(io_ctx),
+ _io_ctx_holder(std::move(io_ctx_holder)) {
+ if (_io_ctx == nullptr && _io_ctx_holder) {
+ _io_ctx = _io_ctx_holder.get();
+ }
_read_timer = ADD_TIMER(_profile, "ReadTime");
if (_range.__isset.compress_type) {
// for compatibility
@@ -115,7 +119,7 @@ NewJsonReader::NewJsonReader(RuntimeState* state, RuntimeProfile* profile, Scann
NewJsonReader::NewJsonReader(RuntimeProfile* profile, const TFileScanRangeParams& params,
const TFileRangeDesc& range,
const std::vector<SlotDescriptor*>& file_slot_descs,
- io::IOContext* io_ctx)
+ io::IOContext* io_ctx, std::shared_ptr<io::IOContext> io_ctx_holder)
: _vhandle_json_callback(nullptr),
_state(nullptr),
_profile(profile),
@@ -131,7 +135,11 @@ NewJsonReader::NewJsonReader(RuntimeProfile* profile, const TFileScanRangeParams
_value_allocator(_value_buffer, sizeof(_value_buffer)),
_parse_allocator(_parse_buffer, sizeof(_parse_buffer)),
_origin_json_doc(&_value_allocator, sizeof(_parse_buffer), &_parse_allocator),
- _io_ctx(io_ctx) {
+ _io_ctx(io_ctx),
+ _io_ctx_holder(std::move(io_ctx_holder)) {
+ if (_io_ctx == nullptr && _io_ctx_holder) {
+ _io_ctx = _io_ctx_holder.get();
+ }
if (_range.__isset.compress_type) {
// for compatibility
_file_compress_type = _range.compress_type;
@@ -411,10 +419,19 @@ Status NewJsonReader::_open_file_reader(bool need_schema) {
_file_description.mtime = _range.__isset.modification_time ? _range.modification_time : 0;
io::FileReaderOptions reader_options =
FileFactory::get_reader_options(_state, _file_description);
- auto file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
- _profile, _system_properties, _file_description, reader_options,
- io::DelegateReader::AccessMode::SEQUENTIAL, _io_ctx,
- io::PrefetchRange(_range.start_offset, _range.size)));
+ io::FileReaderSPtr file_reader;
+ if (_io_ctx_holder) {
+ file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+ _profile, _system_properties, _file_description, reader_options,
+ io::DelegateReader::AccessMode::SEQUENTIAL,
+ std::static_pointer_cast<const io::IOContext>(_io_ctx_holder),
+ io::PrefetchRange(_range.start_offset, _range.size)));
+ } else {
+ file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+ _profile, _system_properties, _file_description, reader_options,
+ io::DelegateReader::AccessMode::SEQUENTIAL, _io_ctx,
+ io::PrefetchRange(_range.start_offset, _range.size)));
+ }
_file_reader = _io_ctx ? std::make_shared<io::TracingFileReader>(std::move(file_reader),
_io_ctx->file_reader_stats)
: file_reader;
diff --git a/be/src/vec/exec/format/json/new_json_reader.h b/be/src/vec/exec/format/json/new_json_reader.h
index 338611ecacf..df46ad321ae 100644
--- a/be/src/vec/exec/format/json/new_json_reader.h
+++ b/be/src/vec/exec/format/json/new_json_reader.h
@@ -72,11 +72,11 @@ public:
NewJsonReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
const TFileScanRangeParams& params, const TFileRangeDesc& range,
const std::vector<SlotDescriptor*>& file_slot_descs, bool* scanner_eof,
- io::IOContext* io_ctx);
+ io::IOContext* io_ctx, std::shared_ptr<io::IOContext> io_ctx_holder = nullptr);
NewJsonReader(RuntimeProfile* profile, const TFileScanRangeParams& params,
const TFileRangeDesc& range, const std::vector<SlotDescriptor*>& file_slot_descs,
- io::IOContext* io_ctx);
+ io::IOContext* io_ctx, std::shared_ptr<io::IOContext> io_ctx_holder = nullptr);
~NewJsonReader() override = default;
Status init_reader(const std::unordered_map<std::string, vectorized::VExprContextSPtr>&
@@ -229,6 +229,7 @@ private:
size_t _current_offset;
io::IOContext* _io_ctx = nullptr;
+ std::shared_ptr<io::IOContext> _io_ctx_holder;
RuntimeProfile::Counter* _read_timer = nullptr;
diff --git a/be/src/vec/exec/format/orc/vorc_reader.cpp b/be/src/vec/exec/format/orc/vorc_reader.cpp
index 4169aacc4ba..454cbf87882 100644
--- a/be/src/vec/exec/format/orc/vorc_reader.cpp
+++ b/be/src/vec/exec/format/orc/vorc_reader.cpp
@@ -199,6 +199,35 @@ OrcReader::OrcReader(RuntimeProfile* profile, RuntimeState* state,
_init_file_description();
}
+OrcReader::OrcReader(RuntimeProfile* profile, RuntimeState* state,
+ const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ size_t batch_size, const std::string& ctz,
+ std::shared_ptr<io::IOContext> io_ctx_holder, FileMetaCache* meta_cache,
+ bool enable_lazy_mat)
+ : _profile(profile),
+ _state(state),
+ _scan_params(params),
+ _scan_range(range),
+ _batch_size(std::max(batch_size, _MIN_BATCH_SIZE)),
+ _range_start_offset(range.start_offset),
+ _range_size(range.size),
+ _ctz(ctz),
+ _io_ctx(io_ctx_holder ? io_ctx_holder.get() : nullptr),
+ _io_ctx_holder(std::move(io_ctx_holder)),
+ _enable_lazy_mat(enable_lazy_mat),
+ _enable_filter_by_min_max(
+ state == nullptr ? true : state->query_options().enable_orc_filter_by_min_max),
+ _dict_cols_has_converted(false) {
+ TimezoneUtils::find_cctz_time_zone(ctz, _time_zone);
+ VecDateTimeValue t;
+ t.from_unixtime(0, ctz);
+ _offset_days = t.day() == 31 ? -1 : 0; // If 1969-12-31, then returns -1.
+ _meta_cache = meta_cache;
+ _init_profile();
+ _init_system_properties();
+ _init_file_description();
+}
+
OrcReader::OrcReader(const TFileScanRangeParams& params, const TFileRangeDesc& range,
const std::string& ctz, io::IOContext* io_ctx, FileMetaCache* meta_cache,
bool enable_lazy_mat)
@@ -216,6 +245,24 @@ OrcReader::OrcReader(const TFileScanRangeParams& params, const TFileRangeDesc& r
_init_file_description();
}
+OrcReader::OrcReader(const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ const std::string& ctz, std::shared_ptr<io::IOContext> io_ctx_holder,
+ FileMetaCache* meta_cache, bool enable_lazy_mat)
+ : _profile(nullptr),
+ _scan_params(params),
+ _scan_range(range),
+ _ctz(ctz),
+ _file_system(nullptr),
+ _io_ctx(io_ctx_holder ? io_ctx_holder.get() : nullptr),
+ _io_ctx_holder(std::move(io_ctx_holder)),
+ _enable_lazy_mat(enable_lazy_mat),
+ _enable_filter_by_min_max(true),
+ _dict_cols_has_converted(false) {
+ _meta_cache = meta_cache;
+ _init_system_properties();
+ _init_file_description();
+}
+
OrcReader::~OrcReader() {
if (_obj_pool && _obj_pool.get()) {
_obj_pool->clear();
@@ -292,9 +339,16 @@ Status OrcReader::_create_file_reader() {
_scan_range.__isset.modification_time ? _scan_range.modification_time : 0;
io::FileReaderOptions reader_options =
FileFactory::get_reader_options(_state, _file_description);
- auto inner_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
- _profile, _system_properties, _file_description, reader_options,
- io::DelegateReader::AccessMode::RANDOM, _io_ctx));
+ io::FileReaderSPtr inner_reader;
+ if (_io_ctx_holder != nullptr) {
+ inner_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+ _profile, _system_properties, _file_description, reader_options,
+ io::DelegateReader::AccessMode::RANDOM, _io_ctx_holder));
+ } else {
+ inner_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+ _profile, _system_properties, _file_description, reader_options,
+ io::DelegateReader::AccessMode::RANDOM, _io_ctx));
+ }
_file_input_stream = std::make_unique<ORCFileInputStream>(
_scan_range.path, std::move(inner_reader), _io_ctx, _profile,
_orc_once_max_read_bytes, _orc_max_merge_distance_bytes);
diff --git a/be/src/vec/exec/format/orc/vorc_reader.h b/be/src/vec/exec/format/orc/vorc_reader.h
index 798cd288e34..5a55f8e3895 100644
--- a/be/src/vec/exec/format/orc/vorc_reader.h
+++ b/be/src/vec/exec/format/orc/vorc_reader.h
@@ -150,10 +150,19 @@ public:
io::IOContext* io_ctx, FileMetaCache* meta_cache = nullptr,
bool enable_lazy_mat = true);
+ OrcReader(RuntimeProfile* profile, RuntimeState* state, const TFileScanRangeParams& params,
+ const TFileRangeDesc& range, size_t batch_size, const std::string& ctz,
+ std::shared_ptr<io::IOContext> io_ctx_holder, FileMetaCache* meta_cache = nullptr,
+ bool enable_lazy_mat = true);
+
OrcReader(const TFileScanRangeParams& params, const TFileRangeDesc& range,
const std::string& ctz, io::IOContext* io_ctx, FileMetaCache* meta_cache = nullptr,
bool enable_lazy_mat = true);
+ OrcReader(const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ const std::string& ctz, std::shared_ptr<io::IOContext> io_ctx_holder,
+ FileMetaCache* meta_cache = nullptr, bool enable_lazy_mat = true);
+
~OrcReader() override;
//If you want to read the file by index instead of column name, set hive_use_column_names to false.
Status init_reader(
@@ -697,6 +706,7 @@ private:
std::shared_ptr<io::FileSystem> _file_system;
io::IOContext* _io_ctx = nullptr;
+ std::shared_ptr<io::IOContext> _io_ctx_holder;
bool _enable_lazy_mat = true;
bool _enable_filter_by_min_max = true;
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
index a84ee2b9d2c..af1958aa7a1 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
@@ -109,6 +109,33 @@ ParquetReader::ParquetReader(RuntimeProfile* profile, const TFileScanRangeParams
_init_file_description();
}
+ParquetReader::ParquetReader(RuntimeProfile* profile, const TFileScanRangeParams& params,
+ const TFileRangeDesc& range, size_t batch_size, cctz::time_zone* ctz,
+ std::shared_ptr<io::IOContext> io_ctx_holder, RuntimeState* state,
+ FileMetaCache* meta_cache, bool enable_lazy_mat)
+ : _profile(profile),
+ _scan_params(params),
+ _scan_range(range),
+ _batch_size(std::max(batch_size, _MIN_BATCH_SIZE)),
+ _range_start_offset(range.start_offset),
+ _range_size(range.size),
+ _ctz(ctz),
+ _io_ctx(io_ctx_holder ? io_ctx_holder.get() : nullptr),
+ _io_ctx_holder(std::move(io_ctx_holder)),
+ _state(state),
+ _enable_lazy_mat(enable_lazy_mat),
+ _enable_filter_by_min_max(
+ state == nullptr ? true
+ : state->query_options().enable_parquet_filter_by_min_max),
+ _enable_filter_by_bloom_filter(
+ state == nullptr ? true
+ : state->query_options().enable_parquet_filter_by_bloom_filter) {
+ _meta_cache = meta_cache;
+ _init_profile();
+ _init_system_properties();
+ _init_file_description();
+}
+
ParquetReader::ParquetReader(const TFileScanRangeParams& params, const TFileRangeDesc& range,
io::IOContext* io_ctx, RuntimeState* state, FileMetaCache* meta_cache,
bool enable_lazy_mat)
@@ -129,6 +156,27 @@ ParquetReader::ParquetReader(const TFileScanRangeParams& params, const TFileRang
_init_file_description();
}
+ParquetReader::ParquetReader(const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ std::shared_ptr<io::IOContext> io_ctx_holder, RuntimeState* state,
+ FileMetaCache* meta_cache, bool enable_lazy_mat)
+ : _profile(nullptr),
+ _scan_params(params),
+ _scan_range(range),
+ _io_ctx(io_ctx_holder ? io_ctx_holder.get() : nullptr),
+ _io_ctx_holder(std::move(io_ctx_holder)),
+ _state(state),
+ _enable_lazy_mat(enable_lazy_mat),
+ _enable_filter_by_min_max(
+ state == nullptr ? true
+ : state->query_options().enable_parquet_filter_by_min_max),
+ _enable_filter_by_bloom_filter(
+ state == nullptr ? true
+ : state->query_options().enable_parquet_filter_by_bloom_filter) {
+ _meta_cache = meta_cache;
+ _init_system_properties();
+ _init_file_description();
+}
+
ParquetReader::~ParquetReader() {
_close_internal();
}
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.h b/be/src/vec/exec/format/parquet/vparquet_reader.h
index b6451ad15d4..c67aab46b69 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.h
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.h
@@ -105,10 +105,19 @@ public:
io::IOContext* io_ctx, RuntimeState* state, FileMetaCache* meta_cache = nullptr,
bool enable_lazy_mat = true);
+ ParquetReader(RuntimeProfile* profile, const TFileScanRangeParams& params,
+ const TFileRangeDesc& range, size_t batch_size, cctz::time_zone* ctz,
+ std::shared_ptr<io::IOContext> io_ctx_holder, RuntimeState* state,
+ FileMetaCache* meta_cache = nullptr, bool enable_lazy_mat = true);
+
ParquetReader(const TFileScanRangeParams& params, const TFileRangeDesc& range,
io::IOContext* io_ctx, RuntimeState* state, FileMetaCache* meta_cache = nullptr,
bool enable_lazy_mat = true);
+ ParquetReader(const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ std::shared_ptr<io::IOContext> io_ctx_holder, RuntimeState* state,
+ FileMetaCache* meta_cache = nullptr, bool enable_lazy_mat = true);
+
~ParquetReader() override;
#ifdef BE_TEST
// for unit test
@@ -322,6 +331,7 @@ private:
ParquetProfile _parquet_profile;
bool _closed = false;
io::IOContext* _io_ctx = nullptr;
+ std::shared_ptr<io::IOContext> _io_ctx_holder;
RuntimeState* _state = nullptr;
bool _enable_lazy_mat = true;
bool _enable_filter_by_min_max = true;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org