github-actions[bot] commented on code in PR #65931:
URL: https://github.com/apache/doris/pull/65931#discussion_r3635445788
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -625,9 +621,15 @@ Status NativeFieldDescriptor::parse_list_field(
if (num_children > 0) {
const bool structural_wrapper = is_struct_list_node(second_level,
first_level.name);
const auto& only_child = t_schemas[curr_pos + 2];
- if (num_children == 1 && !structural_wrapper &&
has_logical_annotation(second_level)) {
+ const bool nested_collection_annotation =
+ is_list_node(second_level) ||
+ (is_map_node(second_level) &&
Review Comment:
[P1] Preserve legacy MAP_KEY_VALUE elements nested in LIST
This exclusion treats every repeated `MAP_KEY_VALUE` node inside a `LIST` as
the one-key SET-wrapper case. However, Parquet's compatibility contract says a
`MAP_KEY_VALUE` group that is not contained by a `MAP` group must be
interpreted as a map. For a valid legacy shape such as `LIST -> repeated
element (MAP_KEY_VALUE) -> repeated map -> key/value`, the previous branch
called `parse_map_field(..., true)` and produced `ARRAY<MAP<...>>`; this branch
now falls through to `parse_struct_field()` and exposes a different nested
shape (or fails table mapping). Please distinguish the direct one-key wrapper
of an enclosing SET from a legacy `MAP_KEY_VALUE` list element, and add
coverage for the latter.
##########
be/src/format_v2/table/remote_doris_reader.cpp:
##########
@@ -75,21 +87,93 @@ class FlightRemoteDorisStream final : public
RemoteDorisStream {
arrow::flight::Ticket ticket;
RETURN_DORIS_STATUS_IF_ERROR(
arrow::flight::Ticket::Deserialize(params.ticket).Value(&ticket));
+ struct PendingOpen {
+ std::mutex mutex;
+ std::condition_variable cv;
+ bool done = false;
+ arrow::Status status = arrow::Status::OK();
+ std::unique_ptr<arrow::flight::FlightClient> client;
+ std::unique_ptr<arrow::flight::FlightStreamReader> stream;
+ };
+ auto pending = std::make_shared<PendingOpen>();
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::FlightClient::Connect(location).Value(&pending->client));
+ 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);
+ // Start before DoGet because endpoint setup is itself a blocking RPC
covered by the same
+ // query/scanner cancellation contract as streaming Next().
+ _cancellation_watcher = std::jthread(
+ [this](std::stop_token stop_token) {
_watch_cancellation(stop_token); });
Review Comment:
[P2] Attach the new DoGet worker to the query task
The watcher now follows the task-thread contract, but this newly introduced
worker uses `SCOPED_INIT_THREAD_CONTEXT()` even though it performs the query's
`DoGet` and owns its client/stream cleanup. On cancellation it is detached and
can remain alive until the full Flight deadline, after the reader and query
have gone away, so its work runs in the orphan context rather than the query's
resource context. This is a distinct new thread from the watcher covered by the
earlier review. Prefer removing the worker by using Flight's call-establishment
stop token; if it remains, capture a shared `ResourceContext`, attach it at
entry, and ensure its Flight objects are destroyed before that attachment ends.
--
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]
