@@ -45,7 +45,7 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
4545 const std::shared_ptr<FileSystem>& fs, uint32_t prefetch_max_parallel_num, int32_t batch_size,
4646 uint32_t prefetch_batch_count, bool enable_adaptive_prefetch_strategy,
4747 const std::shared_ptr<Executor>& executor, bool initialize_read_ranges,
48- bool enable_prefetch_cache , const CacheConfig& cache_config,
48+ PrefetchCacheMode prefetch_cache_mode , const CacheConfig& cache_config,
4949 const std::shared_ptr<MemoryPool>& pool) {
5050 if (prefetch_max_parallel_num == 0 ) {
5151 return Status::Invalid (" prefetch max parallel num should be greater than 0." );
@@ -67,7 +67,7 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
6767 }
6868
6969 std::shared_ptr<ReadAheadCache> cache;
70- if (enable_prefetch_cache ) {
70+ if (prefetch_cache_mode != PrefetchCacheMode:: NEVER ) {
7171 PAIMON_ASSIGN_OR_RAISE (std::shared_ptr<InputStream> input_stream, fs->Open (data_file_path));
7272 cache = std::make_shared<ReadAheadCache>(input_stream, cache_config, pool);
7373 }
@@ -102,9 +102,9 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
102102 }
103103 uint32_t prefetch_queue_capacity = prefetch_batch_count / readers.size ();
104104
105- auto reader = std::unique_ptr<PrefetchFileBatchReaderImpl>(
106- new PrefetchFileBatchReaderImpl ( readers, batch_size, prefetch_queue_capacity,
107- enable_adaptive_prefetch_strategy, executor, cache ));
105+ auto reader = std::unique_ptr<PrefetchFileBatchReaderImpl>(new PrefetchFileBatchReaderImpl (
106+ readers, batch_size, prefetch_queue_capacity, enable_adaptive_prefetch_strategy, executor ,
107+ cache, prefetch_cache_mode ));
108108 if (initialize_read_ranges) {
109109 // normally initialize read ranges should be false, as set read schema will refresh read
110110 // ranges, and set read schema will always be called before read.
@@ -116,11 +116,13 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
116116PrefetchFileBatchReaderImpl::PrefetchFileBatchReaderImpl (
117117 const std::vector<std::shared_ptr<PrefetchFileBatchReader>>& readers, int32_t batch_size,
118118 uint32_t prefetch_queue_capacity, bool enable_adaptive_prefetch_strategy,
119- const std::shared_ptr<Executor>& executor, const std::shared_ptr<ReadAheadCache>& cache)
119+ const std::shared_ptr<Executor>& executor, const std::shared_ptr<ReadAheadCache>& cache,
120+ PrefetchCacheMode cache_mode)
120121 : readers_(std::move(readers)),
121122 batch_size_ (batch_size),
122123 executor_(executor),
123124 cache_(cache),
125+ cache_mode_(cache_mode),
124126 prefetch_queue_capacity_(prefetch_queue_capacity),
125127 enable_adaptive_prefetch_strategy_(enable_adaptive_prefetch_strategy) {
126128 for (size_t i = 0 ; i < readers_.size (); i++) {
@@ -146,6 +148,7 @@ Status PrefetchFileBatchReaderImpl::SetReadSchema(
146148 PAIMON_RETURN_NOT_OK (reader->SetReadSchema (c_schema.get (), predicate, selection_bitmap));
147149 }
148150 selection_bitmap_ = selection_bitmap;
151+ predicate_ = predicate;
149152 return RefreshReadRanges ();
150153}
151154
@@ -273,10 +276,27 @@ Status PrefetchFileBatchReaderImpl::CleanUp() {
273276 return Status::OK ();
274277}
275278
279+ bool PrefetchFileBatchReaderImpl::NeedInitCache () const {
280+ switch (cache_mode_) {
281+ case PrefetchCacheMode::NEVER :
282+ return false ;
283+ case PrefetchCacheMode::EXCLUDE_PREDICATE :
284+ return predicate_ == nullptr ;
285+ case PrefetchCacheMode::EXCLUDE_BITMAP :
286+ return selection_bitmap_ == std::nullopt ;
287+ case PrefetchCacheMode::EXCLUDE_BITMAP_OR_PREDICATE :
288+ return predicate_ == nullptr && selection_bitmap_ == std::nullopt ;
289+ case PrefetchCacheMode::ALWAYS :
290+ return true ;
291+ default :
292+ return true ;
293+ }
294+ }
295+
276296void PrefetchFileBatchReaderImpl::Workloop () {
277297 std::vector<std::future<void >> futures;
278298 futures.resize (readers_.size ());
279- if (cache_) {
299+ if (cache_ && NeedInitCache () ) {
280300 auto read_ranges = readers_[0 ]->PreBufferRange ();
281301 if (read_ranges.ok ()) {
282302 std::vector<ByteRange> ranges;
@@ -286,11 +306,9 @@ void PrefetchFileBatchReaderImpl::Workloop() {
286306 auto s = cache_->Init (std::move (ranges));
287307 if (!s.ok ()) {
288308 SetReadStatus (s);
289- return ;
290309 }
291310 } else {
292311 SetReadStatus (read_ranges.status ());
293- return ;
294312 }
295313 }
296314
0 commit comments