github-actions[bot] commented on code in PR #66802:
URL: https://github.com/apache/doris/pull/66802#discussion_r3793492198
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -442,22 +465,60 @@ Status FileScannerV2::_open_impl(RuntimeState* state) {
SCOPED_TIMER(_open_timer);
RETURN_IF_CANCELLED(state);
RETURN_IF_ERROR(Scanner::_open_impl(state));
+ // The first split may open its physical reader eagerly while refining row
groups. Synchronize
+ // ready filters first so COUNT(*) never treats a future RF carrier as a
disposable placeholder.
+ RETURN_IF_ERROR(try_append_late_arrival_runtime_filter());
Review Comment:
[P1] Keep the pre-open Runtime Filter refresh best-effort
The scheduler deliberately logs and ignores failures from this same late-RF
hook so an optional Runtime Filter cannot fail the scan. Returning the status
from `open()` changes that contract: expression creation or prepare/open errors
now abort the query before any data is read. Moreover,
`RuntimeFilterConsumer::_apply_ready_expr()` marks the consumer APPLIED before
`_get_push_exprs()` can fail, so the scheduler's later best-effort call cannot
recover it. Synchronize ready filters here without making their failure fatal,
consistent with the existing scheduler path.
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -551,6 +574,90 @@ Status ParquetReader::init(RuntimeState* state) {
return Status::OK();
}
+Status ParquetReader::build_physical_splits(const FileScanSplit& source_split,
+ std::vector<FileScanSplit>* splits,
+ bool* was_split) const {
+ DORIS_CHECK(splits != nullptr);
+ DORIS_CHECK(was_split != nullptr);
+ splits->clear();
+ *was_split = false;
+ if (_state == nullptr || _state->file_context.native_metadata == nullptr ||
+ _state->file_context.shared_file_context == nullptr) {
+ return Status::Uninitialized("ParquetReader is not open");
+ }
+ if (!_state->file_context.shared_file_context->has_stable_identity) {
+ // A path and size do not identify a mutable remote object. Keep the
initialized parent
+ // reader instead of publishing children whose shared footer could
become stale.
+ return Status::OK();
+ }
+ if (!_state->file_context.can_refine_physical_splits()) {
+ return Status::OK();
+ }
+
+ ParquetScanRange scan_range {
+ .start_offset =
+ source_split.range.__isset.start_offset ?
source_split.range.start_offset : 0,
+ .size = source_split.range.__isset.size ? source_split.range.size
: -1,
+ .file_size = source_split.range.__isset.file_size ?
source_split.range.file_size
+ :
_file_description->file_size,
+ };
+ std::vector<int> selected_row_groups;
+ RETURN_IF_ERROR(detail::select_native_row_groups_by_scan_range(
+ _state->file_context.native_metadata->to_thrift(), scan_range,
+ _state->file_context.native_metadata->row_group_first_rows(),
&selected_row_groups));
+ const auto& metadata = _state->file_context.native_metadata->to_thrift();
+ const auto compat = native::parquet_reader_compat(
+ metadata.__isset.created_by ? metadata.created_by : std::string
{});
+ const size_t file_size = _state->file_context.native_file->size();
+ auto shared_source_range =
std::make_shared<TFileRangeDesc>(source_split.range);
+ splits->reserve(selected_row_groups.size());
+ for (const int row_group_id : selected_row_groups) {
+ const auto& row_group = metadata.row_groups[row_group_id];
+ if (row_group.num_rows == 0) {
+ // Empty row groups are valid and the ordinary scan planner
ignores them. Refinement
+ // must preserve that behavior before inspecting their potentially
empty chunks.
+ continue;
+ }
+ if (row_group.columns.empty()) {
+ // A root-only schema can still carry rows for metadata COUNT(*),
but it has no byte
+ // envelope that can identify a child. Keep the initialized source
reader so those
+ // rows are not turned into a corruption error or discarded after
tentative children.
+ splits->clear();
+ return Status::OK();
+ }
+ size_t group_start = std::numeric_limits<size_t>::max();
+ size_t group_end = 0;
+ for (size_t column_id = 0; column_id < row_group.columns.size();
++column_id) {
+ const auto& chunk = row_group.columns[column_id];
+ if (!chunk.__isset.meta_data) {
+ return Status::Corruption("Parquet row group {} column {} has
no metadata",
+ row_group_id, column_id);
+ }
+ native::ColumnChunkRange chunk_range;
+ RETURN_IF_ERROR(native::compute_column_chunk_range(
Review Comment:
[P2] Preserve projection isolation when deriving child ranges
This loop validates offsets and extents for every Column Chunk, whereas the
ordinary full-file path computes ranges only for projected leaves. A file with
a valid requested column A and malformed offsets only in unprojected B can
therefore read A with one scanner but fail here with Corruption when two
scanners enable refinement. If an unused chunk cannot yield a safe scheduling
envelope, decline refinement and keep the initialized source reader so the
normal projected scan decides whether that chunk is reachable; add a
one-versus-multiple-scanner differential test.
##########
be/test/exec/scan/file_scanner_v2_test.cpp:
##########
@@ -492,6 +501,384 @@ TEST(FileScannerV2Test,
LegacyCountExemptionRequiresMetadataCountOnEveryRange) {
EXPECT_FALSE(invalid.all_ranges_have_table_level_row_count());
}
+TScanRangeParams scan_range_with_path(std::string path) {
+ TScanRangeParams params;
+ TFileRangeDesc range;
+ range.__set_path(std::move(path));
+
params.scan_range.ext_scan_range.file_scan_range.ranges.push_back(std::move(range));
+ return params;
+}
+
+class RemoteStyleSplitSourceConnector final : public SplitSourceConnector {
+public:
+ explicit RemoteStyleSplitSourceConnector(std::vector<TFileRangeDesc>
ranges)
+ : _ranges(std::move(ranges)) {}
+
+ Status get_next(bool* has_next, TFileRangeDesc* range) override {
+ std::lock_guard lock(_lock);
+ ++_get_next_calls;
+ *has_next = _next < _ranges.size();
+ if (*has_next) {
+ *range = _ranges[_next++];
+ }
+ return Status::OK();
+ }
+
+ int num_scan_ranges() override { return cast_set<int>(_ranges.size()); }
+
+ TFileScanRangeParams* get_params() override { return &_params; }
+
+ int get_next_calls() const { return _get_next_calls.load(); }
+
+private:
+ std::mutex _lock;
+ std::vector<TFileRangeDesc> _ranges;
+ size_t _next = 0;
+ std::atomic<int> _get_next_calls = 0;
+ TFileScanRangeParams _params;
+};
+
+class BlockingRemoteStyleSplitSourceConnector final : public
SplitSourceConnector {
+public:
+ explicit BlockingRemoteStyleSplitSourceConnector(TFileRangeDesc range)
+ : _range(std::move(range)) {}
+
+ Status get_next(bool* has_next, TFileRangeDesc* range) override {
+ std::unique_lock lock(_lock);
+ if (!_source_returned) {
+ _source_returned = true;
+ *has_next = true;
+ *range = _range;
+ return Status::OK();
+ }
+ _fetch_started = true;
+ _fetch_cv.notify_all();
+ _fetch_cv.wait(lock, [&]() { return _release_fetch; });
+ *has_next = false;
+ return Status::OK();
+ }
+
+ void wait_for_fetch() {
+ std::unique_lock lock(_lock);
+ _fetch_cv.wait(lock, [&]() { return _fetch_started; });
+ }
+
+ void release_fetch() {
+ {
+ std::lock_guard lock(_lock);
+ _release_fetch = true;
+ }
+ _fetch_cv.notify_all();
+ }
+
+ int num_scan_ranges() override { return 1; }
+
+ TFileScanRangeParams* get_params() override { return &_params; }
+
+private:
+ std::mutex _lock;
+ std::condition_variable _fetch_cv;
+ TFileRangeDesc _range;
+ bool _source_returned = false;
+ bool _fetch_started = false;
+ bool _release_fetch = false;
+ TFileScanRangeParams _params;
+};
+
+void expect_generated_splits_keep_source_alive(SplitSourceConnector*
connector) {
+ FileScanSplit source;
+ bool has_next = false;
+ ASSERT_TRUE(connector->get_next_split(&has_next, &source).ok());
+ ASSERT_TRUE(has_next);
+ ASSERT_TRUE(source.is_source_split);
+
+ auto waiter = std::async(std::launch::async, [&]() {
+ FileScanSplit split;
+ bool waiter_has_next = false;
+ auto status = connector->get_next_split(&waiter_has_next, &split);
+ return std::make_tuple(status, waiter_has_next, std::move(split));
+ });
+ EXPECT_EQ(waiter.wait_for(std::chrono::milliseconds(50)),
std::future_status::timeout);
+
+ FileScanSplit child;
+ child.range = source.range;
+ child.range.__set_start_offset(101);
+ child.range.__set_size(17);
+ ASSERT_TRUE(connector->finish_source_split(source, {child}).ok());
+
+ ASSERT_EQ(waiter.wait_for(std::chrono::seconds(2)),
std::future_status::ready);
+ auto [status, waiter_has_next, generated] = waiter.get();
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_TRUE(waiter_has_next);
+ EXPECT_FALSE(generated.is_source_split);
+ EXPECT_EQ(generated.range.start_offset, 101);
+ EXPECT_EQ(generated.range.size, 17);
+
+ FileScanSplit end;
+ ASSERT_TRUE(connector->get_next_split(&has_next, &end).ok());
+ EXPECT_FALSE(has_next);
+}
+
+TEST(FileScannerV2Test, LocalSplitSourceWaitsForGeneratedRowGroupSplits) {
+ LocalSplitSourceConnector
connector({scan_range_with_path("local.parquet")}, 4);
+ expect_generated_splits_keep_source_alive(&connector);
+}
+
+TEST(FileScannerV2Test, RemoteStyleSplitSourceWaitsForGeneratedRowGroupSplits)
{
+ TFileRangeDesc range;
+ range.__set_path("remote.parquet");
+ RemoteStyleSplitSourceConnector connector({range});
+ expect_generated_splits_keep_source_alive(&connector);
+ EXPECT_EQ(connector.get_next_calls(), 2);
+}
+
+TEST(FileScannerV2Test, RemoteFetchDoesNotBlockGeneratedSplitPublication) {
+ TFileRangeDesc range;
+ range.__set_path("remote.parquet");
+ BlockingRemoteStyleSplitSourceConnector connector(std::move(range));
+ FileScanSplit source;
+ bool has_next = false;
+ ASSERT_TRUE(connector.get_next_split(&has_next, &source).ok());
+ ASSERT_TRUE(has_next);
+
+ auto remote_fetch = std::async(std::launch::async, [&]() {
+ FileScanSplit split;
+ bool fetch_has_next = false;
+ auto status = connector.get_next_split(&fetch_has_next, &split);
+ return std::make_pair(status, fetch_has_next);
+ });
+ connector.wait_for_fetch();
+ auto child_waiter = std::async(std::launch::async, [&]() {
+ FileScanSplit split;
+ bool child_available = false;
+ auto status = connector.get_next_split(&child_available, &split);
+ return std::make_tuple(status, child_available, std::move(split));
+ });
+ EXPECT_EQ(child_waiter.wait_for(std::chrono::milliseconds(50)),
std::future_status::timeout);
+
+ FileScanSplit child;
+ child.range = source.range;
+ child.range.__set_start_offset(101);
+ auto publish = std::async(std::launch::async,
+ [&]() { return
connector.finish_source_split(source, {child}); });
+ const bool publish_completed_while_fetch_blocked =
+ publish.wait_for(std::chrono::seconds(2)) ==
std::future_status::ready;
+ const bool child_published_while_fetch_blocked =
+ child_waiter.wait_for(std::chrono::seconds(2)) ==
std::future_status::ready;
+ connector.release_fetch();
+
+ const auto publish_status = publish.get();
+ ASSERT_TRUE(publish_status.ok()) << publish_status;
+ ASSERT_TRUE(publish_completed_while_fetch_blocked);
+ ASSERT_TRUE(child_published_while_fetch_blocked);
+ auto [child_status, child_available, generated] = child_waiter.get();
+ ASSERT_TRUE(child_status.ok()) << child_status;
+ ASSERT_TRUE(child_available);
+ EXPECT_EQ(generated.range.start_offset, 101);
+ auto [fetch_status, fetch_has_next] = remote_fetch.get();
+ ASSERT_TRUE(fetch_status.ok()) << fetch_status;
+ EXPECT_FALSE(fetch_has_next);
+}
+
+TEST(FileScannerV2Test, RawSourceClearsGeneratedChildEnvelope) {
+ LocalSplitSourceConnector
connector({scan_range_with_path("next-local.parquet")}, 1);
+ FileScanSplit reused;
+ auto stale_source_range = std::make_shared<TFileRangeDesc>();
+ stale_source_range->__set_path("stale-child.parquet");
+ reused.source_range = std::move(stale_source_range);
+ reused.start_offset = 101;
+ reused.size = 17;
+ reused.clear_table_level_row_count = true;
+ reused.file_context = std::make_shared<const FileContext>();
+ reused.format_split_id = 3;
+
+ bool has_next = false;
+ ASSERT_TRUE(connector.get_next_split(&has_next, &reused).ok());
+ ASSERT_TRUE(has_next);
+ EXPECT_TRUE(reused.is_source_split);
+ EXPECT_EQ(reused.materialize_range().path, "next-local.parquet");
+ EXPECT_EQ(reused.source_range, nullptr);
+ EXPECT_EQ(reused.file_context, nullptr);
+ EXPECT_EQ(reused.format_split_id, -1);
+ ASSERT_TRUE(connector.finish_source_split(reused, {}).ok());
+}
+
+TEST(FileScannerV2Test, EosKeepsLastRangeIdentityForFinalAccounting) {
+ TFileRangeDesc range;
+ range.__set_path("remote.parquet");
+ range.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+ range.__set_file_type(TFileType::FILE_S3);
+ auto split_source =
+
std::make_shared<RemoteStyleSplitSourceConnector>(std::vector<TFileRangeDesc>
{range});
+
+ TFileScanRangeParams params;
+ params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+ MockRuntimeState state;
+ RuntimeProfile profile("last_range_accounting");
+ FileScannerV2 scanner(&state, &profile, nullptr);
+ scanner._split_source = std::move(split_source);
+ scanner._params = ¶ms;
+
+ bool has_next = false;
+ ASSERT_TRUE(scanner._get_next_scan_range(&has_next).ok());
+ ASSERT_TRUE(has_next);
+ ASSERT_TRUE(scanner._get_next_scan_range(&has_next).ok());
Review Comment:
[P1] Retire the source before synchronously fetching EOS
The first `_get_next_scan_range()` registers this raw range in
`_active_source_splits`, but the test immediately fetches again without
completing or retiring it. `RemoteStyleSplitSourceConnector` then reports
source exhaustion and `get_next_split()` waits while that active set is
nonempty; because this is the same test thread, nobody can call
`finish_source_split()` or signal the condition variable, so the test hangs
here (and repeats the hang at line 737). Complete/retire the first split before
the EOS fetch, while retaining the range identity that this test intends to
assert.
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -883,10 +884,23 @@ bool same_physical_scan_layout(const FileScanRequest&
lhs, const FileScanRequest
} // namespace
-Status TableReader::refresh_conjuncts(VExprContextSPtrs conjuncts) {
+Status TableReader::refresh_conjuncts(VExprContextSPtrs conjuncts,
+ std::optional<uint64_t>
condition_cache_digest,
+ bool all_runtime_filters_applied) {
SCOPED_TIMER(_profile.total_timer);
SCOPED_TIMER(_profile.refresh_conjuncts_timer);
_conjuncts = std::move(conjuncts);
+ if (all_runtime_filters_applied) {
+ // A refresh can prove the last pending RF has arrived, but a later
partial refresh must
+ // never make the same split incomplete again.
+ _all_runtime_filters_applied_for_split = true;
Review Comment:
[P1] Keep residual-predicate inputs out of COUNT placeholders
`non_predicate_columns` is not limited to the arbitrary COUNT(*) carrier:
`create_scan_request()` also moves a column here when its table conjunct cannot
be localized (for example, a non-deterministic predicate), leaving the original
conjunct for `Scanner::_filter_output_block()`. Marking every such column as a
placeholder makes Parquet skip the physical read and synthesize defaults, and
`same_physical_scan_layout()` still accepts the refresh. If the final RF
arrives at this boundary, the scanner therefore evaluates its residual
predicate on defaults and can return the wrong count. Only mark columns proven
unused by all residual conjuncts, and cover a late RF on B with a
non-localizable predicate on A.
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -551,6 +574,90 @@ Status ParquetReader::init(RuntimeState* state) {
return Status::OK();
}
+Status ParquetReader::build_physical_splits(const FileScanSplit& source_split,
+ std::vector<FileScanSplit>* splits,
+ bool* was_split) const {
+ DORIS_CHECK(splits != nullptr);
+ DORIS_CHECK(was_split != nullptr);
+ splits->clear();
+ *was_split = false;
+ if (_state == nullptr || _state->file_context.native_metadata == nullptr ||
+ _state->file_context.shared_file_context == nullptr) {
+ return Status::Uninitialized("ParquetReader is not open");
+ }
+ if (!_state->file_context.shared_file_context->has_stable_identity) {
+ // A path and size do not identify a mutable remote object. Keep the
initialized parent
+ // reader instead of publishing children whose shared footer could
become stale.
+ return Status::OK();
+ }
+ if (!_state->file_context.can_refine_physical_splits()) {
+ return Status::OK();
+ }
+
+ ParquetScanRange scan_range {
+ .start_offset =
+ source_split.range.__isset.start_offset ?
source_split.range.start_offset : 0,
+ .size = source_split.range.__isset.size ? source_split.range.size
: -1,
+ .file_size = source_split.range.__isset.file_size ?
source_split.range.file_size
+ :
_file_description->file_size,
+ };
+ std::vector<int> selected_row_groups;
+ RETURN_IF_ERROR(detail::select_native_row_groups_by_scan_range(
+ _state->file_context.native_metadata->to_thrift(), scan_range,
+ _state->file_context.native_metadata->row_group_first_rows(),
&selected_row_groups));
+ const auto& metadata = _state->file_context.native_metadata->to_thrift();
+ const auto compat = native::parquet_reader_compat(
+ metadata.__isset.created_by ? metadata.created_by : std::string
{});
+ const size_t file_size = _state->file_context.native_file->size();
+ auto shared_source_range =
std::make_shared<TFileRangeDesc>(source_split.range);
+ splits->reserve(selected_row_groups.size());
+ for (const int row_group_id : selected_row_groups) {
+ const auto& row_group = metadata.row_groups[row_group_id];
+ if (row_group.num_rows == 0) {
+ // Empty row groups are valid and the ordinary scan planner
ignores them. Refinement
+ // must preserve that behavior before inspecting their potentially
empty chunks.
+ continue;
+ }
+ if (row_group.columns.empty()) {
+ // A root-only schema can still carry rows for metadata COUNT(*),
but it has no byte
+ // envelope that can identify a child. Keep the initialized source
reader so those
+ // rows are not turned into a corruption error or discarded after
tentative children.
+ splits->clear();
+ return Status::OK();
+ }
+ size_t group_start = std::numeric_limits<size_t>::max();
+ size_t group_end = 0;
+ for (size_t column_id = 0; column_id < row_group.columns.size();
++column_id) {
+ const auto& chunk = row_group.columns[column_id];
+ if (!chunk.__isset.meta_data) {
+ return Status::Corruption("Parquet row group {} column {} has
no metadata",
+ row_group_id, column_id);
+ }
+ native::ColumnChunkRange chunk_range;
+ RETURN_IF_ERROR(native::compute_column_chunk_range(
+ chunk.meta_data, file_size, compat.parquet_816_padding,
&chunk_range));
+ group_start = std::min(group_start, chunk_range.offset);
+ group_end = std::max(group_end, chunk_range.offset +
chunk_range.length);
+ }
+ if (group_end <= group_start) {
+ return Status::Corruption("Parquet row group {} has an empty
physical byte range",
+ row_group_id);
+ }
+ FileScanSplit child;
+ child.source_range = shared_source_range;
+ child.start_offset = cast_set<int64_t>(group_start);
Review Comment:
[P2] Keep Condition Cache coordinates stable across refinement
These child offsets also become `ExternalCacheKey` range coordinates. An
identical predicate run with one scanner caches one FE source-range bitmap,
while a run with multiple scanners misses it and inserts one entry per Row
Group envelope; switching back to one scanner misses all of those entries as
well. Files with many small groups also replace one entry with R separately
keyed/rounded allocations, increasing LRU churn. Use a canonical Row Group key
in both execution modes, or merge a shared source-coordinate bitmap only after
the final child reaches EOF, and test serial-to-refined reuse in both
directions.
##########
be/src/exec/scan/split_source_connector.cpp:
##########
@@ -24,6 +24,113 @@ namespace doris {
using apache::thrift::transport::TTransportException;
+Status SplitSourceConnector::get_next_split(bool* has_next, FileScanSplit*
split) {
+ DORIS_CHECK(has_next != nullptr);
+ DORIS_CHECK(split != nullptr);
+ std::unique_lock lock(_split_lock);
+ while (true) {
+ *has_next = false;
+ if (_stopped) {
+ return Status::OK();
+ }
+ if (!_generated_splits.empty()) {
+ *split = std::move(_generated_splits.front());
+ _generated_splits.pop_front();
+ *has_next = true;
+ return Status::OK();
+ }
+ if (_source_exhausted) {
+ if (_active_source_splits.empty()) {
+ return Status::OK();
+ }
+ _split_ready.wait(lock);
+ continue;
+ }
+ if (_source_claim_in_progress) {
+ _split_ready.wait(lock);
+ continue;
+ }
+
+ TFileRangeDesc range;
+ bool has_source = false;
+ // Record the in-flight claim before dropping the queue lock, so
another scanner cannot
+ // report raw EOS before this claim becomes an active source. The
potentially blocking
+ // remote RPC must not hold the queue lock because a footer producer
may publish children
+ // while that fetch is in flight.
+ _source_claim_in_progress = true;
+ lock.unlock();
+ const auto source_status = get_next(&has_source, &range);
+ lock.lock();
+ _source_claim_in_progress = false;
+ if (source_status.ok() && !has_source) {
+ // Both local exhaustion and an empty final remote batch are
terminal. Remember EOS so
+ // parent waiters do not wake each other into repeated empty
source fetches.
+ _source_exhausted = true;
+ }
+ _split_ready.notify_all();
+ RETURN_IF_ERROR(source_status);
+ if (_stopped) {
+ return Status::OK();
+ }
+ if (has_source) {
+ // A scanner reuses its output envelope across files. Reset
child-only shared range and
+ // context state before installing a raw FE range, or
materialization can read the
+ // previous child's file instead of this source.
+ *split = {};
+ split->range = std::move(range);
+ split->is_source_split = true;
+ split->source_split_id = _next_source_split_id++;
+ split->source_progress = std::make_shared<SourceSplitProgress>();
+ _active_source_splits.insert(split->source_split_id);
+ *has_next = true;
+ return Status::OK();
+ }
+ if (_active_source_splits.empty()) {
Review Comment:
[P2] Recheck generated children after the final source fetch
`get_next()` runs without `_split_lock`, so another scanner can finish its
source and enqueue several Row Group children while this final remote fetch is
blocked. When the fetch returns empty, this path sets `_source_exhausted` and
can return `has_next=false` as soon as the producer has retired its active
source, without rechecking `_generated_splits`. The claimant then exits while
the producer drains all remaining children serially, defeating the new
parallelism. Loop back through the queue check after an empty fetch (or check
the queue here), and cover the two-scanner interleaving without a third child
waiter.
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -551,6 +574,90 @@ Status ParquetReader::init(RuntimeState* state) {
return Status::OK();
}
+Status ParquetReader::build_physical_splits(const FileScanSplit& source_split,
+ std::vector<FileScanSplit>* splits,
+ bool* was_split) const {
+ DORIS_CHECK(splits != nullptr);
+ DORIS_CHECK(was_split != nullptr);
+ splits->clear();
+ *was_split = false;
+ if (_state == nullptr || _state->file_context.native_metadata == nullptr ||
+ _state->file_context.shared_file_context == nullptr) {
+ return Status::Uninitialized("ParquetReader is not open");
+ }
+ if (!_state->file_context.shared_file_context->has_stable_identity) {
+ // A path and size do not identify a mutable remote object. Keep the
initialized parent
+ // reader instead of publishing children whose shared footer could
become stale.
+ return Status::OK();
+ }
+ if (!_state->file_context.can_refine_physical_splits()) {
+ return Status::OK();
+ }
+
+ ParquetScanRange scan_range {
+ .start_offset =
+ source_split.range.__isset.start_offset ?
source_split.range.start_offset : 0,
+ .size = source_split.range.__isset.size ? source_split.range.size
: -1,
+ .file_size = source_split.range.__isset.file_size ?
source_split.range.file_size
+ :
_file_description->file_size,
+ };
+ std::vector<int> selected_row_groups;
Review Comment:
[P2] Apply known footer pruning before publishing children
Split planning selects Row Groups only by the source byte range; the current
request's footer-statistics pruning does not run until each generated child
independently reaches `plan_parquet_row_groups()`. For a stable R-group file
whose min/max rejects every group, the old path performs one reader lifecycle
and no page reads, while refinement performs the planning lifecycle plus R
child create/map/open/close cycles with no data work to parallelize. Apply the
initial request's safe footer-only pruning once on the initialized reader
before publishing children (late RFs can still narrow them), and cover
all/most-groups-pruned cases.
--
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]
