github-actions[bot] commented on code in PR #65931:
URL: https://github.com/apache/doris/pull/65931#discussion_r3634830168
##########
be/src/format_v2/parquet/native_schema_node.cpp:
##########
@@ -70,7 +70,14 @@ Status build_native_schema_node(const DataTypePtr&
projected_type,
const auto* struct_type = assert_cast<const
DataTypeStruct*>(type.get());
std::map<std::string, const ParquetColumnSchema*> file_children;
for (const auto& child : file_schema.children) {
- file_children.emplace(to_lower(child->name), child.get());
+ const auto [_, inserted] =
file_children.emplace(to_lower(child->name), child.get());
Review Comment:
[P1] Preserve resolved field identity before rejecting case folds
This rejects every case-folded collision in the physical STRUCT before
looking at the already-pruned `projected_type`. For an Iceberg schema
`s<Value(id=1), value(id=2), other(id=3)>`, a field-ID projection of only
`s.other` is fully resolved by `TableColumnMapper`/`projected_type()`, yet this
loop still returns CORRUPTION because the unrelated siblings collide. It also
prevents selecting either case-distinct child by its field ID. Please use the
resolved local/field identity (or exact writer-name match) first and report
ambiguity only when a requested child actually needs a case-insensitive
fallback; add a partial-projection test.
##########
be/src/format_v2/table/remote_doris_reader.cpp:
##########
@@ -77,19 +85,49 @@ class FlightRemoteDorisStream final : public
RemoteDorisStream {
arrow::flight::Ticket::Deserialize(params.ticket).Value(&ticket));
RETURN_DORIS_STATUS_IF_ERROR(
arrow::flight::FlightClient::Connect(location).Value(&_flight_client));
-
RETURN_DORIS_STATUS_IF_ERROR(_flight_client->DoGet(ticket).Value(&_stream));
+ arrow::flight::FlightCallOptions options;
+ // A Flight deadline covers streaming reads as well as DoGet setup, so
a stalled Next()
+ // cannot outlive the query execution timeout indefinitely.
+ options.timeout = std::chrono::seconds(_timeout_seconds);
+ options.stop_token = _stop_source.token();
+ RETURN_DORIS_STATUS_IF_ERROR(_flight_client->DoGet(options,
ticket).Value(&_stream));
+ _cancellation_watcher = std::jthread([this](std::stop_token
stop_token) {
Review Comment:
[P2] Attach the cancellation thread to the query context
This query-owned `std::jthread` does not inherit the scanner worker's
thread-local context, and the lambda enters without `SCOPED_ATTACH_TASK` (or
even `SCOPED_INIT_THREAD_CONTEXT`). That violates the BE runtime thread-entry
contract and leaves cancellation work outside the query's resource/memory
accounting; any Doris thread-local use on this path can also fail for an
unattached thread. Attach the watcher to the retained
RuntimeState/ResourceContext at lambda entry, or run it on an existing
query-attached executor.
##########
be/src/format_v2/table/remote_doris_reader.cpp:
##########
@@ -77,19 +85,49 @@ class FlightRemoteDorisStream final : public
RemoteDorisStream {
arrow::flight::Ticket::Deserialize(params.ticket).Value(&ticket));
RETURN_DORIS_STATUS_IF_ERROR(
arrow::flight::FlightClient::Connect(location).Value(&_flight_client));
-
RETURN_DORIS_STATUS_IF_ERROR(_flight_client->DoGet(ticket).Value(&_stream));
+ arrow::flight::FlightCallOptions options;
+ // A Flight deadline covers streaming reads as well as DoGet setup, so
a stalled Next()
+ // cannot outlive the query execution timeout indefinitely.
+ options.timeout = std::chrono::seconds(_timeout_seconds);
+ options.stop_token = _stop_source.token();
+ RETURN_DORIS_STATUS_IF_ERROR(_flight_client->DoGet(options,
ticket).Value(&_stream));
+ _cancellation_watcher = std::jthread([this](std::stop_token
stop_token) {
+ while (!stop_token.stop_requested()) {
+ if (_runtime_state != nullptr &&
_runtime_state->is_cancelled()) {
+ // Next() runs in the scanner thread. A separate watcher
is required to signal
+ // Arrow while that thread is blocked inside the streaming
RPC.
+ _stop_source.RequestStop();
+ _stream->Cancel();
+ return;
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(25));
Review Comment:
[P2] Wake the watcher promptly during stream close
`request_stop()` does not interrupt `sleep_for`, so a normally completed
Flight stream can block in `join()` for the remainder of this 25 ms polling
interval before cleanup proceeds. `TableReader::get_block()` synchronously
closes the current reader at every split EOF, so workloads with many small or
fast Remote Doris tickets accumulate roughly 12.5 ms average (25 ms worst-case)
avoidable latency per split. Use a stop-token-aware timed wait or notify a
condition variable before joining, and add a prompt-close test for the
production watcher.
##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.h:
##########
@@ -208,7 +218,9 @@ class ColumnChunkReader {
size_t active_decoder_scratch_bytes() const {
// Only the current encoding is active. Old decoder instances retain
reusable capacity but
// must not make the high-water policy treat their last batch as
current working memory.
- return (_page_decoder == nullptr ? 0 :
_page_decoder->active_scratch_bytes()) +
+ const size_t active_decompress_bytes = _page_uses_decompress_buf ?
_decompress_buf_size : 0;
Review Comment:
[P1] Let retained decompression capacity age at a safe page boundary
This reports the full allocation as active whenever any compressed page uses
it. After a 64 MiB outlier, every later 4 KiB compressed page therefore still
looks 64 MiB active, so `release_batch_scratch()` resets the three-idle counter
at each periodic probe. The production loop normally loads the next page before
that probe, while the test calls release directly in the artificial gap after
`next_page()`; `read_with_plain_filter()` never runs the probe at all. Track
the current payload separately, carry a pending release to a safe page
boundary, and test a large page followed by small compressed pages with
boundaries misaligned to the 16-batch cadence plus the direct PLAIN-filter path.
##########
be/src/format_v2/table/remote_doris_reader.cpp:
##########
@@ -77,19 +85,49 @@ class FlightRemoteDorisStream final : public
RemoteDorisStream {
arrow::flight::Ticket::Deserialize(params.ticket).Value(&ticket));
RETURN_DORIS_STATUS_IF_ERROR(
arrow::flight::FlightClient::Connect(location).Value(&_flight_client));
-
RETURN_DORIS_STATUS_IF_ERROR(_flight_client->DoGet(ticket).Value(&_stream));
+ arrow::flight::FlightCallOptions options;
+ // A Flight deadline covers streaming reads as well as DoGet setup, so
a stalled Next()
+ // cannot outlive the query execution timeout indefinitely.
+ options.timeout = std::chrono::seconds(_timeout_seconds);
+ options.stop_token = _stop_source.token();
+ RETURN_DORIS_STATUS_IF_ERROR(_flight_client->DoGet(options,
ticket).Value(&_stream));
Review Comment:
[P1] Arm cancellation for in-flight setup and normal scanner stop
The only code that signals `_stop_source` is created after synchronous
`DoGet()` returns, so query cancellation while the server withholds initial
stream data cannot interrupt setup and waits for the full Flight deadline.
After open, the watcher polls only `RuntimeState::is_cancelled()`; reaching
LIMIT instead calls `FileScannerV2::try_stop()` and sets only
`IOContext::should_stop`, so a scanner already blocked in `Next()` is likewise
not cancelled. The added fake test sets `should_stop` before entering `Next()`
and misses both paths. Please arm a thread-safe cancellation bridge before
`DoGet()` and have both query cancellation and scanner stop cancel the
in-flight call; cover blocked setup and blocked read.
##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp:
##########
@@ -86,6 +86,50 @@ Status validate_uncompressed_page_sizes(const
tparquet::PageHeader& header,
return Status::OK();
}
+Status validate_fixed_width_page_size(const tparquet::PageHeader& header,
int32_t type_length,
+ level_t max_rep_level, level_t
max_def_level,
+ bool schema_is_required) {
+ if (type_length <= 0 || max_rep_level != 0 || max_def_level != 0 ||
!schema_is_required) {
Review Comment:
[P1] Validate optional V2 fixed-width extents before allocation
This early return skips every optional fixed-width page (`max_def_level !=
0`), but Data Page V2 already provides `num_values`, `num_nulls`, and exact
level byte lengths. A compressed optional INT32 page with one value can
therefore declare `uncompressed_page_size = INT32_MAX`; `PageReader` bounds
only the tiny compressed payload, then `load_page_data()` calls
`_reserve_decompress_buf()` with nearly 2 GiB before decompression can reject
it. Extend the pre-allocation check for V2 PLAIN/BYTE_STREAM_SPLIT using the
non-null count and level extents, and test the real compressed optional-V2 load
path.
##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp:
##########
@@ -86,6 +86,50 @@ Status validate_uncompressed_page_sizes(const
tparquet::PageHeader& header,
return Status::OK();
}
+Status validate_fixed_width_page_size(const tparquet::PageHeader& header,
int32_t type_length,
+ level_t max_rep_level, level_t
max_def_level,
+ bool schema_is_required) {
+ if (type_length <= 0 || max_rep_level != 0 || max_def_level != 0 ||
!schema_is_required) {
+ return Status::OK();
+ }
+ const bool is_v2 = header.__isset.data_page_header_v2;
+ if (!is_v2 && !header.__isset.data_page_header) {
+ return Status::OK();
+ }
+ const auto encoding =
+ is_v2 ? header.data_page_header_v2.encoding :
header.data_page_header.encoding;
+ if (encoding != tparquet::Encoding::PLAIN &&
+ encoding != tparquet::Encoding::BYTE_STREAM_SPLIT) {
+ return Status::OK();
+ }
+ const int32_t num_values =
+ is_v2 ? header.data_page_header_v2.num_values :
header.data_page_header.num_values;
+ if (num_values < 0 || static_cast<uint64_t>(num_values) >
+
static_cast<uint64_t>(std::numeric_limits<int32_t>::max()) /
+ static_cast<uint32_t>(type_length)) {
+ return Status::Corruption("Parquet fixed-width PLAIN page byte size
overflows");
+ }
+ const int64_t expected = static_cast<int64_t>(num_values) * type_length;
+ if (UNLIKELY(header.uncompressed_page_size != expected)) {
+ // Required fixed-width PLAIN/BYTE_STREAM_SPLIT pages have no level
bytes, so their exact
+ // extent is known before decompression and must gate
attacker-controlled allocation.
+ return Status::Corruption("Parquet fixed-width page has {}
uncompressed bytes, expected {}",
+ header.uncompressed_page_size, expected);
+ }
+ return Status::OK();
+}
+
+Status validate_dictionary_page_size(const tparquet::PageHeader& header) {
+ DORIS_CHECK(header.__isset.dictionary_page_header);
+ const int32_t num_values = header.dictionary_page_header.num_values;
+ if (UNLIKELY(num_values < 0 || (num_values == 0 &&
header.uncompressed_page_size != 0))) {
Review Comment:
[P1] Validate nonempty fixed-width dictionaries before allocation
This accepts every positive `num_values`. With a compressed fixed-width
dictionary, a one-entry INT32 page can advertise `uncompressed_page_size =
INT32_MAX` while keeping its compressed payload tiny; `PageReader` bounds only
the compressed bytes, and `_decode_dict_page()` allocates nearly 2 GiB before
decompression and `set_dict()` finally enforce the known four-byte extent. Pass
the physical width into this pre-allocation check and validate the checked
count/width product for fixed-width PLAIN dictionaries, with a compressed
`ColumnChunkReader` regression rather than only the empty helper test.
--
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]
