This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 30a44d61a8a [improvement](be) Optimize filtered Parquet scans and
condition cache (#65602)
30a44d61a8a is described below
commit 30a44d61a8a1b7d66dff0d136850314d9ec337e6
Author: Gabriel <[email protected]>
AuthorDate: Wed Jul 15 09:52:39 2026 +0800
[improvement](be) Optimize filtered Parquet scans and condition cache
(#65602)
[improvement](be) Optimize filtered Parquet scans and condition cache
Problem Summary: Parquet v2 repeatedly advanced lazy non-predicate readers
for every fully filtered batch, and external condition cache could not safely
reuse entries when runtime filters were present. Accumulate skipped rows for
lazy readers and flush once before the next surviving batch, discard the lag at
row-group boundaries, expose Arrow skip time, and key scanner-driven condition
cache entries with the exact predicate and runtime-filter snapshot. Runtime
filters without a reliab [...]
Condition cache + runtime filter behavior
Before this change, the presence of any runtime filter disabled condition
cache, even when the filter was ready and had a stable payload. Runtime-filter
pushdown still worked, but repeated scans could neither look up nor publish a
granule survivor bitmap.
After this change, scanner-driven splits rebuild the condition-cache digest
from the exact conjunct snapshot used by that split. Ready IN, Bloom, and
Min/Max runtime filters encode their probe expression, filter semantics,
payload, and null behavior in the digest.
For example, for static predicate P: price > 100:
P AND customer_id IN {7, 9} uses a payload-specific key digest(P, RF{7,9}).
After a cache miss reaches reader EOF, its survivor bitmap is published under
that key.
P AND customer_id IN {9, 7} has the same IN-set semantics and therefore
reuses the same key.
P AND customer_id IN {8, 10} produces a different key and cannot reuse the
{7,9} bitmap.
If split 0 starts with only P and RF {7,9} arrives before split 1, split 0
uses digest(P) while split 1 rebuilds and uses digest(P, RF{7,9}); the late RF
is never stored under the stale P key.
If any current conjunct returns digest 0, condition cache is disabled for
that split while runtime-filter pushdown remains enabled. This is the
correctness fallback for expressions that cannot represent their complete
semantics in a stable digest—for example, an RF probe expression containing a
digest-unsupported lambda, or a future/custom stateful RF whose payload is not
digestible. Standalone TableReader callers with an RF also remain conservative
unless they provide an explicit dig [...]
Problem Summary: Parquet v2 repeatedly advanced lazy non-predicate
readers for every fully filtered batch, and external condition cache
could not safely reuse entries when runtime filters were present.
Accumulate skipped rows for lazy readers and flush once before the next
surviving batch, discard the lag at row-group boundaries, expose Arrow
skip time, and key scanner-driven condition cache entries with the exact
predicate and runtime-filter snapshot. Runtime filters without a
reliable digest and standalone callers without an explicit split digest
remain uncached.
#### Condition cache + runtime filter behavior
Before this change, the presence of any runtime filter disabled
condition cache, even when the filter was ready and had a stable
payload. Runtime-filter pushdown still worked, but repeated scans could
neither look up nor publish a granule survivor bitmap.
After this change, scanner-driven splits rebuild the condition-cache
digest from the exact conjunct snapshot used by that split. Ready IN,
Bloom, and Min/Max runtime filters encode their probe expression, filter
semantics, payload, and null behavior in the digest.
For example, for static predicate `P: price > 100`:
- `P AND customer_id IN {7, 9}` uses a payload-specific key `digest(P,
RF{7,9})`. After a cache miss reaches reader EOF, its survivor bitmap is
published under that key.
- `P AND customer_id IN {9, 7}` has the same IN-set semantics and
therefore reuses the same key.
- `P AND customer_id IN {8, 10}` produces a different key and cannot
reuse the `{7,9}` bitmap.
- If split 0 starts with only `P` and RF `{7,9}` arrives before split 1,
split 0 uses `digest(P)` while split 1 rebuilds and uses `digest(P,
RF{7,9})`; the late RF is never stored under the stale `P` key.
If any current conjunct returns digest `0`, condition cache is disabled
for that split while runtime-filter pushdown remains enabled. This is
the correctness fallback for expressions that cannot represent their
complete semantics in a stable digest—for example, an RF probe
expression containing a digest-unsupported lambda, or a future/custom
stateful RF whose payload is not digestible. Standalone `TableReader`
callers with an RF also remain conservative unless they provide an
explicit digest for the exact split snapshot.
### Release note
Improve Parquet external scan performance for highly selective
predicates and allow payload-safe condition cache reuse with ready
runtime filters.
### Check List (For Author)
- Test: Unit Test
- `./run-be-ut.sh -j 64 --run
--filter='ParquetScanTest.*:NewParquetReaderTest.PageIndexFilteredGapFlushesPendingOutputSkipOnce:TableReaderTest.ConditionCache*:FileScannerV2Test.ConditionCacheDigestIncludesRuntimeFilterPayload'`
- `./run-be-ut.sh -j 64 --run
--filter='TableReaderTest.ConditionCache*'`
- `build-support/check-format.sh`
- clang-tidy attempted but blocked by pre-existing remote toolchain
header and NOLINT errors
- Behavior changed: Yes. Lazy Parquet output readers defer skips across
empty selections, and condition cache supports runtime filters only with
an exact split digest.
- Does this need documentation: No
---
be/src/exec/scan/file_scanner.cpp | 26 ++--
be/src/exec/scan/file_scanner_v2.cpp | 1 +
be/src/exec/scan/scanner.cpp | 34 ++++++
be/src/exec/scan/scanner.h | 12 ++
be/src/format_v2/parquet/parquet_profile.cpp | 3 +
be/src/format_v2/parquet/parquet_profile.h | 2 +
be/src/format_v2/parquet/parquet_scan.cpp | 38 +++++-
be/src/format_v2/parquet/parquet_scan.h | 5 +
.../parquet/reader/scalar_column_reader.cpp | 1 +
be/src/format_v2/table_reader.cpp | 23 +++-
be/src/format_v2/table_reader.h | 13 +-
be/test/exec/scan/file_scanner_v2_test.cpp | 98 +++++++++++++++
be/test/format_v2/jni/jni_table_reader_test.cpp | 1 +
be/test/format_v2/parquet/parquet_reader_test.cpp | 7 +-
be/test/format_v2/parquet/parquet_scan_test.cpp | 133 +++++++++++++++++++++
be/test/format_v2/table_reader_test.cpp | 61 +++++++++-
16 files changed, 427 insertions(+), 31 deletions(-)
diff --git a/be/src/exec/scan/file_scanner.cpp
b/be/src/exec/scan/file_scanner.cpp
index 343c87edc51..8851d9b2c05 100644
--- a/be/src/exec/scan/file_scanner.cpp
+++ b/be/src/exec/scan/file_scanner.cpp
@@ -1942,25 +1942,7 @@ bool FileScanner::_should_enable_condition_cache() {
return false;
}
- // Runtime filters are query-local dynamic predicates. Some ready RF
implementations can hash
- // their payload into get_digest(), but FileScanner cannot rely on that
for all RFs reaching the
- // native reader. In particular, ScanLocalState computes
_condition_cache_digest during open(),
- // while FileScanner may append late-arrival RFs in
_process_late_arrival_conjuncts()
- // immediately before initializing Parquet/ORC readers.
- //
- // Reading a weaker cache entry would be safe by itself: if a cached
bitmap only represented
- // static predicate P, false granules for P are also false for P AND RF.
The unsafe part is
- // writing. On cache miss, native readers mark survivor granules using all
pushed-down
- // predicates, including late RFs. Without a read-only cache mode, this
would insert a bitmap for
- // P AND RF under a digest that only represents P.
- //
- // Example:
- // Q1 static predicate: k = 1, late RF payload: partition_key IN
('2024-02-01')
- // Q2 static predicate: k = 1, late RF payload: partition_key IN
('2024-03-01')
- // If both scans share the same file/range/digest, reusing Q1's bitmap for
Q2 can skip row
- // ranges according to the wrong RF payload. Keep RF predicate pushdown
enabled for reader-side
- // filtering, but do not persist its result in condition cache.
- return !_contains_runtime_filter(_conjuncts) &&
!_contains_runtime_filter(_push_down_conjuncts);
+ return true;
}
bool FileScanner::_should_enable_condition_cache_for_load() const {
@@ -1990,6 +1972,12 @@ void FileScanner::_init_reader_condition_cache() {
_condition_cache = nullptr;
_condition_cache_ctx = nullptr;
+ // _process_late_arrival_conjuncts() runs before the Parquet/ORC reader is
initialized. Rebuild
+ // the key here so a newly arrived cache-safe RF is represented by both
its semantics and its
+ // payload. If any current conjunct cannot produce a reliable digest, this
becomes zero and the
+ // existing safety gate below disables condition cache.
+ _condition_cache_digest = _current_condition_cache_digest();
+
if (!_should_enable_condition_cache() || !_cur_reader) {
return;
}
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index ef0896867c1..d1c0edd54b1 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -531,6 +531,7 @@ Status FileScannerV2::_prepare_table_reader_split(const
TFileRangeDesc& range,
// A metadata COUNT split may span scheduler turns. Do not enter
that irreversible
// synthetic-row path while a runtime filter can still arrive
between batches.
.all_runtime_filters_applied = _applied_rf_num == _total_rf_num,
+ .condition_cache_digest = _current_condition_cache_digest(),
.cache = _kv_cache,
.current_range = range,
.current_split_format = current_split_format,
diff --git a/be/src/exec/scan/scanner.cpp b/be/src/exec/scan/scanner.cpp
index 8788331c245..3eaf5ee77eb 100644
--- a/be/src/exec/scan/scanner.cpp
+++ b/be/src/exec/scan/scanner.cpp
@@ -270,6 +270,40 @@ Status Scanner::try_append_late_arrival_runtime_filter() {
return Status::OK();
}
+uint64_t Scanner::_current_condition_cache_digest() const {
+ DORIS_CHECK(_state != nullptr);
+ DORIS_CHECK(_local_state != nullptr);
+ if (_local_state->get_condition_cache_digest() == 0) {
+ return 0;
+ }
+
+ // ScanLocalState computed its digest after collecting the RFs that were
ready during open(). A
+ // scanner may later clone more RF conjuncts between file splits, so
rebuild from its current
+ // snapshot instead of reusing that stale value. For example, split 0 may
use P, while split 1
+ // starts after an IN RF with payload {7, 9} arrives and must use digest(P
AND RF{7, 9}). A
+ // different payload {8, 10} consequently receives a different key.
get_digest() returning zero
+ // is the correctness fallback for an RF whose complete semantics cannot
be represented.
+ return
_build_condition_cache_digest(_state->query_options().condition_cache_digest,
+ _conjuncts);
+}
+
+uint64_t Scanner::_build_condition_cache_digest(uint64_t seed, const
VExprContextSPtrs& conjuncts) {
+ for (const auto& conjunct : conjuncts) {
+ seed = conjunct->get_digest(seed);
+ if (seed == 0) {
+ return 0;
+ }
+ }
+ return seed;
+}
+
+#ifdef BE_TEST
+uint64_t Scanner::TEST_build_condition_cache_digest(uint64_t seed,
+ const VExprContextSPtrs&
conjuncts) {
+ return _build_condition_cache_digest(seed, conjuncts);
+}
+#endif
+
Status Scanner::close(RuntimeState* state) {
#ifndef BE_TEST
COUNTER_UPDATE(_local_state->_scanner_wait_worker_timer,
_scanner_wait_worker_timer);
diff --git a/be/src/exec/scan/scanner.h b/be/src/exec/scan/scanner.h
index a91395b911e..27db928a500 100644
--- a/be/src/exec/scan/scanner.h
+++ b/be/src/exec/scan/scanner.h
@@ -97,7 +97,19 @@ public:
// eg, for file scanner, return the current file path.
virtual std::string get_current_scan_range_name() { return "not
implemented"; }
+#ifdef BE_TEST
+ static uint64_t TEST_build_condition_cache_digest(uint64_t seed,
+ const VExprContextSPtrs&
conjuncts);
+#endif
+
protected:
+ // Rebuild the condition-cache digest from the scanner's current conjunct
snapshot. The local
+ // state's digest is used only as a safety gate: zero means condition
cache was disabled during
+ // scan-node open (for example by TopN or an expression without a reliable
digest).
+ uint64_t _current_condition_cache_digest() const;
+ static uint64_t _build_condition_cache_digest(uint64_t seed,
+ const VExprContextSPtrs&
conjuncts);
+
virtual Status _prepare_impl() {
_has_prepared = true;
return Status::OK();
diff --git a/be/src/format_v2/parquet/parquet_profile.cpp
b/be/src/format_v2/parquet/parquet_profile.cpp
index d41ff295ec4..29783e14a05 100644
--- a/be/src/format_v2/parquet/parquet_profile.cpp
+++ b/be/src/format_v2/parquet/parquet_profile.cpp
@@ -71,6 +71,8 @@ void ParquetProfile::init(RuntimeProfile* profile) {
parquet_profile, 1);
arrow_read_records_time =
ADD_CHILD_TIMER_WITH_LEVEL(profile, "ArrowReadRecordsTime",
parquet_profile, 1);
+ arrow_skip_records_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(profile, "ArrowSkipRecordsTime",
parquet_profile, 1);
materialization_time =
ADD_CHILD_TIMER_WITH_LEVEL(profile, "MaterializationTime",
parquet_profile, 1);
lazy_read_filtered_rows = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"FilteredRowsByLazyRead",
@@ -198,6 +200,7 @@ ParquetColumnReaderProfile
ParquetProfile::column_reader_profile() const {
.reader_skip_rows = reader_skip_rows,
.reader_select_rows = reader_select_rows,
.arrow_read_records_time = arrow_read_records_time,
+ .arrow_skip_records_time = arrow_skip_records_time,
.materialization_time = materialization_time,
};
}
diff --git a/be/src/format_v2/parquet/parquet_profile.h
b/be/src/format_v2/parquet/parquet_profile.h
index 27f9818f4a0..827cac45d94 100644
--- a/be/src/format_v2/parquet/parquet_profile.h
+++ b/be/src/format_v2/parquet/parquet_profile.h
@@ -35,6 +35,7 @@ struct ParquetColumnReaderProfile {
RuntimeProfile::Counter* reader_skip_rows = nullptr; // rows
skipped by skip()
RuntimeProfile::Counter* reader_select_rows = nullptr; // rows
selected by select()
RuntimeProfile::Counter* arrow_read_records_time = nullptr; // Arrow
RecordReader time (ns)
+ RuntimeProfile::Counter* arrow_skip_records_time = nullptr; // Arrow
SkipRecords time (ns)
RuntimeProfile::Counter* materialization_time = nullptr; // value
materialization time (ns)
};
@@ -103,6 +104,7 @@ struct ParquetProfile {
RuntimeProfile::Counter* reader_skip_rows = nullptr;
RuntimeProfile::Counter* reader_select_rows = nullptr;
RuntimeProfile::Counter* arrow_read_records_time = nullptr;
+ RuntimeProfile::Counter* arrow_skip_records_time = nullptr;
RuntimeProfile::Counter* materialization_time = nullptr;
RuntimeProfile::Counter* lazy_read_filtered_rows = nullptr;
diff --git a/be/src/format_v2/parquet/parquet_scan.cpp
b/be/src/format_v2/parquet/parquet_scan.cpp
index a7602cd11b2..1abddc777aa 100644
--- a/be/src/format_v2/parquet/parquet_scan.cpp
+++ b/be/src/format_v2/parquet/parquet_scan.cpp
@@ -655,6 +655,12 @@ void ParquetScanScheduler::reset_current_row_group() {
_current_selected_ranges.clear();
_current_range_idx = 0;
_current_range_rows_read = 0;
+ // Readers are row-group scoped. If every remaining row was filtered, no
future output can
+ // observe the non-predicate readers' position, so dropping them together
with their pending lag
+ // avoids a useless end-of-row-group SkipRecords call. Example: predicate
readers advance from 0
+ // to 10,000 while lazy readers stay at 0; clearing both readers here is
sufficient because the
+ // next row group constructs a new set starting at its own row 0.
+ _pending_non_predicate_skip_rows = 0;
_current_predicate_prefetched = false;
_current_non_predicate_prefetched = false;
_current_merge_range_active = false;
@@ -787,10 +793,23 @@ Status
ParquetScanScheduler::skip_current_row_group_rows(int64_t rows) {
for (const auto& column_reader : _current_predicate_columns |
std::views::values) {
RETURN_IF_ERROR(column_reader->skip(rows));
}
+ // Keep page-index/condition-cache gaps pending for lazy columns as well.
For example, after a
+ // fully filtered [0, 32) batch and a pruned [32, 96) gap, predicate
readers are at 96 while lazy
+ // readers remain at 0; one later skip(96) is cheaper than skip(32)
followed by skip(64).
+ DORIS_CHECK(_pending_non_predicate_skip_rows <=
std::numeric_limits<int64_t>::max() - rows);
+ _pending_non_predicate_skip_rows += rows;
+ _current_row_group_rows_read += rows;
+ return Status::OK();
+}
+
+Status ParquetScanScheduler::flush_pending_non_predicate_skip_rows() {
+ if (_pending_non_predicate_skip_rows == 0) {
+ return Status::OK();
+ }
for (const auto& column_reader : _current_non_predicate_columns |
std::views::values) {
- RETURN_IF_ERROR(column_reader->skip(rows));
+ RETURN_IF_ERROR(column_reader->skip(_pending_non_predicate_skip_rows));
}
- _current_row_group_rows_read += rows;
+ _pending_non_predicate_skip_rows = 0;
return Status::OK();
}
@@ -1395,6 +1414,18 @@ Status
ParquetScanScheduler::read_current_row_group_batch(
.column->filter(output_filter,
selected_rows)));
}
}
+ if (selected_rows == 0) {
+ // Predicate readers have consumed this physical batch, but touching
every lazy column here
+ // turns a long rejected prefix into `empty_batches * lazy_columns`
Arrow calls. Record only
+ // the positional lag. If [0, 32), [32, 64), and [64, 96) are empty,
the first surviving
+ // batch performs one skip(96) per lazy column. If the row group ends
instead, reset drops the
+ // lazy readers without flushing because no value from them can be
observed.
+ DORIS_CHECK(_pending_non_predicate_skip_rows <=
+ std::numeric_limits<int64_t>::max() - batch_rows);
+ _pending_non_predicate_skip_rows += batch_rows;
+ *rows = 0;
+ return Status::OK();
+ }
if (!_current_merge_range_active && selected_rows > 0 &&
!_current_non_predicate_columns.empty()) {
// Do not prefetch lazy output columns until at least one row survives
filtering. This is
@@ -1406,6 +1437,9 @@ Status ParquetScanScheduler::read_current_row_group_batch(
{
SCOPED_TIMER(_scan_profile.column_read_time);
+ // Bring lazy readers to the first row of the current physical batch
before interpreting its
+ // selection vector. This also merges pending range gaps with fully
filtered batches.
+ RETURN_IF_ERROR(flush_pending_non_predicate_skip_rows());
for (const auto& [fid, column_reader] :
_current_non_predicate_columns) {
auto position_it =
request.local_positions.find(format::LocalColumnId(fid));
DORIS_CHECK(position_it != request.local_positions.end());
diff --git a/be/src/format_v2/parquet/parquet_scan.h
b/be/src/format_v2/parquet/parquet_scan.h
index 3fa7586fb1b..c656b146cef 100644
--- a/be/src/format_v2/parquet/parquet_scan.h
+++ b/be/src/format_v2/parquet/parquet_scan.h
@@ -152,6 +152,7 @@ private:
const format::FileScanRequest& request, bool*
has_row_group);
Status skip_current_row_group_rows(int64_t rows);
+ Status flush_pending_non_predicate_skip_rows();
Status read_filter_columns(int64_t batch_rows, const
format::FileScanRequest& request,
Block* file_block, SelectionVector* selection,
@@ -203,6 +204,10 @@ private:
_current_selected_ranges; // selected ranges for the current row
group after page-index pruning
size_t _current_range_idx = 0; // current selected_range index
int64_t _current_range_rows_read = 0; // rows read in the current range
+ // Predicate readers move immediately because they decide which rows
survive. Non-predicate
+ // readers can lag behind across fully filtered batches and range gaps;
the lag is flushed once
+ // before the next surviving batch is materialized, or discarded with the
row group.
+ int64_t _pending_non_predicate_skip_rows = 0;
bool _current_predicate_prefetched = false;
bool _current_non_predicate_prefetched = false;
diff --git a/be/src/format_v2/parquet/reader/scalar_column_reader.cpp
b/be/src/format_v2/parquet/reader/scalar_column_reader.cpp
index 784e4cdc900..bdbd9ac8779 100644
--- a/be/src/format_v2/parquet/reader/scalar_column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/scalar_column_reader.cpp
@@ -183,6 +183,7 @@ Status ScalarColumnReader::skip_records(int64_t rows) {
return Status::OK();
}
int64_t skipped_rows = 0;
+ SCOPED_TIMER(_profile.arrow_skip_records_time);
try {
_record_reader->Reset();
while (skipped_rows < rows) {
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index f30a1d567cf..5f37538c1f2 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -527,7 +527,8 @@ Status TableReader::init(TableReadOptions&& options) {
_scanner_profile = options.scanner_profile;
_file_slot_descs = options.file_slot_descs;
_push_down_agg_type = options.push_down_agg_type;
- _condition_cache_digest = options.condition_cache_digest;
+ _initial_condition_cache_digest = options.condition_cache_digest;
+ _condition_cache_digest = _initial_condition_cache_digest;
_projected_columns = std::move(options.projected_columns);
_system_properties = create_system_properties(_scan_params);
_mapper_options.mode = TableColumnMappingMode::BY_NAME;
@@ -631,10 +632,11 @@ bool TableReader::_should_enable_condition_cache(const
FileScanRequest& file_req
!file_request.delete_conjuncts.empty()) {
return false;
}
- // Runtime filters can arrive late and their payload is not guaranteed to
be represented by the
- // scan-local digest. Without a read-only mode, a MISS could insert a
bitmap for P AND RF under
- // the digest for only P. This mirrors the old FileScanner guard.
- return !contains_runtime_filter(file_request.conjuncts);
+ // Only scanner-driven splits provide a digest rebuilt from the exact RF
snapshot. Keep the
+ // conservative behavior for standalone TableReader callers: their initial
digest may describe
+ // only static predicate P and must not store P AND RF under that key.
+ return _condition_cache_digest_covers_current_split ||
+ !contains_runtime_filter(file_request.conjuncts);
}
Status TableReader::_init_reader_condition_cache(const FileScanRequest&
file_request) {
@@ -828,6 +830,17 @@ Status TableReader::prepare_split(const SplitReadOptions&
options) {
SCOPED_TIMER(_profile.prepare_split_timer);
_current_split_pruned = false;
_all_runtime_filters_applied_for_split =
options.all_runtime_filters_applied;
+ _condition_cache_digest_covers_current_split =
options.condition_cache_digest.has_value();
+ if (options.condition_cache_digest.has_value()) {
+ // The split snapshot may include RFs that arrived after
TableReader::init(). Use the digest
+ // computed from that exact snapshot. Example: an initial P digest
must not be used to store
+ // the bitmap for P AND late RF{7, 9}; the scanner supplies digest(P
AND RF{7, 9}) here.
+ _condition_cache_digest = *options.condition_cache_digest;
+ } else {
+ // An explicit scanner digest is split-scoped. Restore the init-time
digest when a later
+ // standalone split omits it instead of leaking the previous split's
RF payload into its key.
+ _condition_cache_digest = _initial_condition_cache_digest;
+ }
if (options.conjuncts.has_value()) {
_conjuncts = *options.conjuncts;
}
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index 8d99195c21d..73bade81bc4 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -135,7 +135,9 @@ struct TableReadOptions {
const std::vector<SlotDescriptor*>* file_slot_descs = nullptr;
// Push-down aggregate type.
const TPushAggOp::type push_down_agg_type = TPushAggOp::type::NONE;
- // Digest of stable pushed-down predicates. A zero digest disables
condition cache.
+ // Initial digest of predicates available during scanner open.
Scanner-driven splits override it
+ // with SplitReadOptions::condition_cache_digest after collecting
late-arrival runtime filters.
+ // A zero digest disables condition cache.
uint64_t condition_cache_digest = 0;
};
@@ -154,6 +156,10 @@ struct SplitReadOptions {
// filter could arrive after synthetic rows have already been returned and
those rows cannot be
// retracted. Standalone TableReader callers have no scanner
runtime-filter lifecycle.
bool all_runtime_filters_applied = true;
+ // Digest for the exact scanner conjunct snapshot attached to this split.
FileScannerV2 rebuilds
+ // it after collecting late-arrival RFs, so different RF payloads cannot
share a cache entry. A
+ // zero value explicitly disables condition cache for this split.
+ std::optional<uint64_t> condition_cache_digest;
ShardedKVCache* cache = nullptr;
TFileRangeDesc current_range;
FileFormat current_split_format = FileFormat::PARQUET;
@@ -1595,7 +1601,12 @@ protected:
FileFormat _format;
TPushAggOp::type _push_down_agg_type = TPushAggOp::type::NONE;
size_t _batch_size = 0;
+ uint64_t _initial_condition_cache_digest = 0;
uint64_t _condition_cache_digest = 0;
+ // True only when prepare_split() received a digest for the exact conjunct
snapshot used by
+ // this split. Standalone callers that only supplied
TableReadOptions::condition_cache_digest
+ // keep the conservative runtime-filter guard.
+ bool _condition_cache_digest_covers_current_split = false;
segment_v2::ConditionCache::ExternalCacheKey _condition_cache_key;
std::shared_ptr<std::vector<bool>> _condition_cache;
std::shared_ptr<ConditionCacheContext> _condition_cache_ctx;
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index 2dc1d7f72c5..ea24b4b21eb 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -36,11 +36,15 @@
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
#include "exec/operator/file_scan_operator.h"
+#include "exec/runtime_filter/runtime_filter_definitions.h"
#include "exec/scan/file_scan_io_context.h"
#include "exec/scan/file_scanner.h"
#include "exec/scan/split_source_connector.h"
+#include "exprs/create_predicate_function.h"
#include "exprs/runtime_filter_expr.h"
+#include "exprs/vbloom_predicate.h"
#include "exprs/vdirect_in_predicate.h"
+#include "exprs/vliteral.h"
#include "exprs/vslot_ref.h"
#include "format_v2/expr/cast.h"
#include "testutil/mock/mock_runtime_state.h"
@@ -141,6 +145,23 @@ private:
const std::string _expr_name = "UnsafePartitionPredicate";
};
+class UndigestibleRuntimePredicate final : public VExpr {
+public:
+ UndigestibleRuntimePredicate() : VExpr(std::make_shared<DataTypeUInt8>(),
false) {}
+
+ Status execute_column_impl(VExprContext*, const Block*, const Selector*,
size_t count,
+ ColumnPtr& result_column) const override {
+ result_column = ColumnUInt8::create(count, 1);
+ return Status::OK();
+ }
+
+ const std::string& expr_name() const override { return _expr_name; }
+ uint64_t get_digest(uint64_t) const override { return 0; }
+
+private:
+ const std::string _expr_name = "undigestible_runtime_predicate";
+};
+
VExprContextSPtr runtime_filter_context(VExprSPtr impl, int filter_id) {
const auto node = bool_in_pred_node();
return VExprContext::create_shared(
@@ -195,6 +216,54 @@ TExprNode bool_in_pred_node() {
return node;
}
+VExprContextSPtr int_in_runtime_filter(const std::vector<int32_t>& values, int
filter_id) {
+ std::shared_ptr<HybridSetBase> filter(create_set(PrimitiveType::TYPE_INT,
false));
+ for (const auto value : values) {
+ filter->insert(&value);
+ }
+ const auto node = bool_in_pred_node();
+ auto impl = VDirectInPredicate::create_shared(node, std::move(filter),
true);
+ impl->add_child(slot_ref(1, 0, std::make_shared<DataTypeInt32>(),
"rf_key"));
+ return runtime_filter_context(std::move(impl), filter_id);
+}
+
+VExprContextSPtr int_bloom_runtime_filter(const std::vector<int32_t>& values,
int filter_id) {
+ std::shared_ptr<BloomFilterFuncBase> filter(
+ create_bloom_filter(PrimitiveType::TYPE_INT, false));
+ RuntimeFilterParams params;
+ params.filter_type = RuntimeFilterType::BLOOM_FILTER;
+ params.column_return_type = PrimitiveType::TYPE_INT;
+ params.bloom_filter_size = 1024;
+ filter->init_params(¶ms);
+ EXPECT_TRUE(filter->init_with_fixed_length(1024).ok());
+ auto value_column = ColumnInt32::create();
+ for (const auto value : values) {
+ value_column->insert_value(value);
+ }
+ ColumnPtr values_column_ptr = std::move(value_column);
+ filter->insert_fixed_len(values_column_ptr, 0);
+
+ TExprNode node = bool_in_pred_node();
+ node.__set_node_type(TExprNodeType::BLOOM_PRED);
+ node.__set_opcode(TExprOpcode::RT_FILTER);
+ auto impl = VBloomPredicate::create_shared(node);
+ impl->set_filter(std::move(filter));
+ impl->add_child(slot_ref(1, 0, std::make_shared<DataTypeInt32>(),
"rf_key"));
+ return runtime_filter_context(std::move(impl), filter_id);
+}
+
+VExprContextSPtr int_minmax_runtime_filter(int32_t upper_bound, int filter_id)
{
+ const auto int_type = std::make_shared<DataTypeInt32>();
+ VExprSPtr impl;
+ TExprNode node;
+ EXPECT_TRUE(create_vbin_predicate(int_type, TExprOpcode::LE, impl, &node,
false).ok());
+ impl->add_child(slot_ref(1, 0, int_type, "rf_key"));
+ VExprSPtr literal;
+ EXPECT_TRUE(create_literal(int_type, &upper_bound, literal).ok());
+ impl->add_child(std::move(literal));
+ return runtime_filter_context(std::move(impl), filter_id);
+}
+
} // namespace
// Scenario: FileScannerV2::is_supported should honor table format, scan
params format, and the
@@ -285,6 +354,35 @@ TEST(FileScannerV2Test,
IcebergPositionDeletesSupportNativeFormats) {
EXPECT_FALSE(FileScannerV2::is_supported(params, avro_position_delete));
}
+// Ready IN/Bloom/MinMax runtime filters all expose their payload through
get_digest(). Rebuilding
+// the scanner digest must therefore be stable for the same payload and
isolated for a different
+// payload. An RF without a complete digest remains the zero-digest safety
fallback.
+TEST(FileScannerV2Test, ConditionCacheDigestIncludesRuntimeFilterPayload) {
+ constexpr uint64_t seed = 12345;
+ const auto digest = [](uint64_t initial_seed, const VExprContextSPtr&
conjunct) {
+ return Scanner::TEST_build_condition_cache_digest(initial_seed,
{conjunct});
+ };
+
+ EXPECT_EQ(digest(seed, int_in_runtime_filter({7, 9}, 1)),
+ digest(seed, int_in_runtime_filter({9, 7}, 1)));
+ EXPECT_NE(digest(seed, int_in_runtime_filter({7, 9}, 1)),
+ digest(seed, int_in_runtime_filter({8, 10}, 1)));
+
+ EXPECT_EQ(digest(seed, int_bloom_runtime_filter({7, 9}, 2)),
+ digest(seed, int_bloom_runtime_filter({7, 9}, 2)));
+ EXPECT_NE(digest(seed, int_bloom_runtime_filter({7, 9}, 2)),
+ digest(seed, int_bloom_runtime_filter({8, 10}, 2)));
+
+ EXPECT_EQ(digest(seed, int_minmax_runtime_filter(9, 3)),
+ digest(seed, int_minmax_runtime_filter(9, 3)));
+ EXPECT_NE(digest(seed, int_minmax_runtime_filter(9, 3)),
+ digest(seed, int_minmax_runtime_filter(10, 3)));
+
+ EXPECT_EQ(digest(seed,
+
runtime_filter_context(std::make_shared<UndigestibleRuntimePredicate>(), 4)),
+ 0);
+}
+
TEST(FileScannerV2Test, FileScanLocalStateSelectsV2ForSupportedQueriesOnly) {
TQueryOptions query_options;
TFileScanRangeParams params;
diff --git a/be/test/format_v2/jni/jni_table_reader_test.cpp
b/be/test/format_v2/jni/jni_table_reader_test.cpp
index db3ec3c7e02..e3e5e69b825 100644
--- a/be/test/format_v2/jni/jni_table_reader_test.cpp
+++ b/be/test/format_v2/jni/jni_table_reader_test.cpp
@@ -179,6 +179,7 @@ TEST(JniTableReaderTest,
AdaptiveProbeSetBeforePrepareControlsFirstJniOpen) {
.conjuncts = std::nullopt,
.partition_prune_conjuncts = {},
.all_runtime_filters_applied =
true,
+ .condition_cache_digest =
std::nullopt,
.cache = nullptr,
.current_range = {},
.current_split_format =
FileFormat::JNI,
diff --git a/be/test/format_v2/parquet/parquet_reader_test.cpp
b/be/test/format_v2/parquet/parquet_reader_test.cpp
index 796123832c6..c5bf251c155 100644
--- a/be/test/format_v2/parquet/parquet_reader_test.cpp
+++ b/be/test/format_v2/parquet/parquet_reader_test.cpp
@@ -2545,7 +2545,10 @@ TEST_F(NewParquetReaderTest,
NestedStructPredicateDoesNotNarrowRowRangesByPageIn
EXPECT_EQ(plan.pruning_stats.selected_row_ranges,
plan.row_groups[0].selected_ranges.size());
}
-TEST_F(NewParquetReaderTest,
PageIndexFilteredPagesDoNotDoubleSkipOutputColumns) {
+// Scenario: the selected range starts after page-index-pruned rows. The
scheduler defers that range
+// gap for the non-predicate payload reader, then flushes it exactly once
before materialization. The
+// page skip plan advances the reader without calling Arrow SkipRecords or
double-skipping row 64.
+TEST_F(NewParquetReaderTest, PageIndexFilteredGapFlushesPendingOutputSkipOnce)
{
write_page_index_filter_pair_parquet_file(_file_path);
RuntimeProfile profile("new_parquet_reader_page_skip");
auto reader = create_reader(0, -1, &profile);
@@ -2586,6 +2589,7 @@ TEST_F(NewParquetReaderTest,
PageIndexFilteredPagesDoNotDoubleSkipOutputColumns)
ASSERT_NE(profile.get_counter("SelectedRows"), nullptr);
ASSERT_NE(profile.get_counter("RangeGapSkippedRows"), nullptr);
ASSERT_NE(profile.get_counter("ReaderSkipRows"), nullptr);
+ ASSERT_NE(profile.get_counter("ArrowSkipRecordsTime"), nullptr);
ASSERT_NE(profile.get_counter("RowGroupFilterTime"), nullptr);
ASSERT_NE(profile.get_counter("PageIndexFilterTime"), nullptr);
ASSERT_NE(profile.get_counter("PageIndexReadTime"), nullptr);
@@ -2595,6 +2599,7 @@ TEST_F(NewParquetReaderTest,
PageIndexFilteredPagesDoNotDoubleSkipOutputColumns)
EXPECT_EQ(profile.get_counter("SelectedRows")->value(), 64);
EXPECT_GT(profile.get_counter("RangeGapSkippedRows")->value(), 0);
EXPECT_EQ(profile.get_counter("ReaderSkipRows")->value(), 0);
+ EXPECT_EQ(profile.get_counter("ArrowSkipRecordsTime")->value(), 0);
EXPECT_GT(profile.get_counter("RowGroupFilterTime")->value(), 0);
EXPECT_GT(profile.get_counter("PageIndexFilterTime")->value(), 0);
EXPECT_GT(profile.get_counter("PageIndexReadTime")->value(), 0);
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp
b/be/test/format_v2/parquet/parquet_scan_test.cpp
index 04d46097c98..ec02b8700d1 100644
--- a/be/test/format_v2/parquet/parquet_scan_test.cpp
+++ b/be/test/format_v2/parquet/parquet_scan_test.cpp
@@ -36,6 +36,7 @@
#include "common/config.h"
#include "core/assert_cast.h"
#include "core/block/block.h"
+#include "core/column/column_array.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_vector.h"
@@ -343,6 +344,15 @@ void write_list_parquet_file(const std::string& file_path)
{
write_table(file_path, table, 2);
}
+void write_int_list_parquet_file(const std::string& file_path) {
+ auto schema = arrow::schema({
+ arrow::field("id", arrow::int32(), false),
+ arrow::field("xs", arrow::list(arrow::int32()), false),
+ });
+ auto table = arrow::Table::Make(schema, {build_int32_array({1, 2, 3}),
build_list_array()});
+ write_table(file_path, table, 3);
+}
+
void write_page_index_parquet_file(const std::string& file_path) {
std::vector<int32_t> ids(128);
std::iota(ids.begin(), ids.end(), 0);
@@ -1068,6 +1078,129 @@ TEST_F(ParquetScanTest,
PredicateColumnsSkipUnreadColumnsWhenFirstPredicateFilte
EXPECT_EQ(counter_value(profile, "ReaderSkipRows"), 6);
}
+// Scenario: every physical batch in every row group is rejected. Predicate
readers reach each row
+// group boundary, while the lazy score reader remains at row 0. The boundary
reset must discard that
+// reader and its pending lag instead of issuing SkipRecords for values that
can never be observed.
+TEST_F(ParquetScanTest, FullyFilteredRowGroupsDropPendingLazyReaders) {
+ write_int_pair_parquet_file(_file_path, 2, false);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ reader->set_batch_size(1);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ request->conjuncts.push_back(create_int32_zonemap_conjunct(0,
Int32ZoneMapExpr::Op::GT, 100));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ size_t total_rows = 0;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ total_rows += rows;
+ }
+
+ EXPECT_EQ(total_rows, 0);
+ EXPECT_EQ(counter_value(profile, "EmptySelectionBatches"), 6);
+ EXPECT_EQ(counter_value(profile, "ReaderSkipRows"), 0);
+ ASSERT_NE(profile.get_counter("ArrowSkipRecordsTime"), nullptr);
+ EXPECT_EQ(profile.get_counter("ArrowSkipRecordsTime")->value(), 0);
+}
+
+// Scenario: row group 0 is fully filtered and leaves two pending lazy rows.
Reset must discard that
+// lag before row group 1 creates fresh readers at its own row 0; otherwise
score 30 would be skipped
+// as if it belonged to the rejected prefix from the previous row group.
+TEST_F(ParquetScanTest, PendingLazySkipDoesNotCrossRowGroupReset) {
+ write_int_pair_parquet_file(_file_path, 2, false);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ reader->set_batch_size(1);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ request->conjuncts.push_back(create_int32_zonemap_conjunct(0,
Int32ZoneMapExpr::Op::GT, 2));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> scores;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& score_column =
int32_data_column(*block.get_by_position(1).column);
+ for (size_t row = 0; row < rows; ++row) {
+ scores.push_back(score_column.get_element(row));
+ }
+ }
+
+ EXPECT_EQ(scores, std::vector<int32_t>({30, 40, 50, 60}));
+ EXPECT_EQ(counter_value(profile, "ReaderSkipRows"), 0);
+}
+
+// Scenario: a nested lazy column stays behind while id=1 is rejected.
Flushing skip(1) must consume
+// the complete repetition/definition-level span for the first list, then
materialize the remaining
+// two parent rows without corrupting their child boundaries.
+TEST_F(ParquetScanTest, PendingLazySkipPreservesNestedRowBoundaries) {
+ write_int_list_parquet_file(_file_path);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ reader->set_batch_size(1);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ request->conjuncts.push_back(create_int32_zonemap_conjunct(0,
Int32ZoneMapExpr::Op::GT, 1));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<size_t> list_lengths;
+ std::vector<int32_t> list_values;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ const IColumn* list_column = block.get_by_position(1).column.get();
+ if (const auto* nullable =
check_and_get_column<ColumnNullable>(list_column)) {
+ list_column = &nullable->get_nested_column();
+ }
+ const auto& array_column = assert_cast<const
ColumnArray&>(*list_column);
+ const auto& value_column = int32_data_column(array_column.get_data());
+ for (size_t row = 0; row < rows; ++row) {
+ const size_t list_start = array_column.offset_at(row);
+ const size_t list_length = array_column.size_at(row);
+ list_lengths.push_back(list_length);
+ for (size_t element = 0; element < list_length; ++element) {
+ list_values.push_back(value_column.get_element(list_start +
element));
+ }
+ }
+ }
+
+ EXPECT_EQ(list_lengths, std::vector<size_t>({1, 0}));
+ EXPECT_EQ(list_values, std::vector<int32_t>({3}));
+ EXPECT_EQ(counter_value(profile, "ReaderSkipRows"), 1);
+}
+
TEST_F(ParquetScanTest, MultiColumnPredicateWaitsForAllPredicateColumns) {
write_int_pair_parquet_file(_file_path, 6, false);
RuntimeProfile profile("profile");
diff --git a/be/test/format_v2/table_reader_test.cpp
b/be/test/format_v2/table_reader_test.cpp
index 970e4b21535..b5ec1dfc765 100644
--- a/be/test/format_v2/table_reader_test.cpp
+++ b/be/test/format_v2/table_reader_test.cpp
@@ -2146,9 +2146,10 @@ TEST(TableReaderTest,
ConditionCacheSkipsRequestWithoutFileLocalConjuncts) {
ASSERT_TRUE(reader.close().ok());
}
-// Scenario: runtime filters can arrive late and are not represented by the
stable predicate digest.
-// A MISS must not insert a bitmap for `stable predicate AND runtime filter`
under the stable digest.
-TEST(TableReaderTest, ConditionCacheSkipsRuntimeFilterConjunct) {
+// Scenario: a standalone caller has only the initial digest for stable
predicate P, while its
+// current conjunct snapshot also contains an RF. Without an explicit split
digest, TableReader must
+// not store P AND RF under P's stale key.
+TEST(TableReaderTest, ConditionCacheSkipsRuntimeFilterWithoutSplitDigest) {
std::vector<ColumnDefinition> file_schema;
file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
@@ -2186,6 +2187,60 @@ TEST(TableReaderTest,
ConditionCacheSkipsRuntimeFilterConjunct) {
ASSERT_TRUE(reader.close().ok());
}
+// Scenario: FileScannerV2 supplies a non-zero digest computed from the exact
Ready RF payload in
+// this split. The RF wrapper is no longer a reason to disable condition
cache; a MISS context is
+// created and can be published under that payload-specific key after EOF.
+TEST(TableReaderTest, ConditionCacheAllowsRuntimeFilterCoveredBySplitDigest) {
+ ScopedConditionCacheForTest cache;
+ std::vector<ColumnDefinition> file_schema;
+ file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ set_name_identifiers(&projected_columns);
+
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ fake_state->total_rows = ConditionCacheContext::GRANULE_SIZE;
+ FakeTableReader reader(file_schema, fake_state);
+ ASSERT_TRUE(
+ reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {prepared_conjunct(
+ &state, runtime_filter_wrapper_expr(
+
table_int32_greater_than_expr(0, 0, 0)))},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ .condition_cache_digest = 7,
+ })
+ .ok());
+
+ SplitReadOptions split_options;
+ split_options.current_range.__set_path("fake-table-reader-input");
+ split_options.condition_cache_digest = 11;
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_NE(fake_state->condition_cache_ctx, nullptr);
+ EXPECT_FALSE(fake_state->condition_cache_ctx->is_hit);
+
+ segment_v2::ConditionCacheHandle handle;
+ segment_v2::ConditionCache::ExternalCacheKey initial_digest_key(
+ "fake-table-reader-input", 0, -1, 7, 0, -1,
+
segment_v2::ConditionCache::ExternalCacheKey::BASE_GRANULE_AWARE_VERSION);
+ EXPECT_FALSE(cache.get()->lookup(initial_digest_key, &handle));
+ segment_v2::ConditionCache::ExternalCacheKey split_digest_key(
+ "fake-table-reader-input", 0, -1, 11, 0, -1,
+
segment_v2::ConditionCache::ExternalCacheKey::BASE_GRANULE_AWARE_VERSION);
+ EXPECT_TRUE(cache.get()->lookup(split_digest_key, &handle));
+ ASSERT_TRUE(reader.close().ok());
+}
+
// Scenario: table-format delete files/deletion vectors are outside the
data-file cache key. When
// TableReader injects delete conjuncts into the file scan request, condition
cache must be disabled
// for that split.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
