Gabriel39 commented on code in PR #69723:
URL: https://github.com/apache/doris/pull/69723#discussion_r4430807823
##########
be/src/load/delta_writer/push_handler.cpp:
##########
@@ -380,171 +407,81 @@ Status PushBrokerReader::init() {
fmt::format("PushBrokerReader:dummy_id={}", print_id(dummy_id)));
_runtime_state = RuntimeState::create_unique(params, query_options,
query_globals,
ExecEnv::GetInstance(),
nullptr, tracker);
+ TDescriptorTable projected_descriptors;
+ RETURN_IF_ERROR(project_broker_source_slots(_t_desc_tbl, _params,
&projected_descriptors));
DescriptorTbl* desc_tbl = nullptr;
- Status status = DescriptorTbl::create(_runtime_state->obj_pool(),
_t_desc_tbl, &desc_tbl);
+ Status status =
+ DescriptorTbl::create(_runtime_state->obj_pool(),
projected_descriptors, &desc_tbl);
if (UNLIKELY(!status.ok())) {
return Status::Error<PUSH_INIT_ERROR>("Failed to create descriptor
table, msg: {}", status);
}
_runtime_state->set_desc_tbl(desc_tbl);
_runtime_profile = _runtime_state->runtime_profile();
_runtime_profile->set_name("PushBrokerReader");
- _file_cache_statistics.reset(new io::FileCacheStatistics());
- _file_reader_stats.reset(new io::FileReaderStats());
- _io_ctx.reset(new io::IOContext());
+ _file_cache_statistics = std::make_unique<io::FileCacheStatistics>();
+ _file_reader_stats = std::make_unique<io::FileReaderStats>();
+ _io_ctx = std::make_shared<io::IOContext>();
_io_ctx->file_cache_stats = _file_cache_statistics.get();
_io_ctx->file_reader_stats = _file_reader_stats.get();
_io_ctx->query_id = &_runtime_state->query_id();
- auto slot_descs = desc_tbl->get_tuple_descriptor(0)->slots();
- uint32_t idx = 0;
- for (auto& slot_desc : slot_descs) {
- _all_col_names.push_back(to_lower((slot_desc->col_name())));
- _col_name_to_block_idx.insert({to_lower(slot_desc->col_name()),
idx++});
+ auto* source = desc_tbl->get_tuple_descriptor(_params.src_tuple_id);
+ _dest_tuple_desc = desc_tbl->get_tuple_descriptor(_params.dest_tuple_id);
+ if (source == nullptr || _dest_tuple_desc == nullptr) {
+ return Status::InternalError("Unknown source or destination tuple in
push load");
}
-
Review Comment:
Rechecked this scenario against the merge base. The [legacy standalone Push
reader](https://github.com/apache/doris/blob/339e087b11f760e9e45ce33d94862b4b63f1d9ee/be/src/load/delta_writer/push_handler.cpp#L640)
initializes Parquet without column_descs.
[GenericReader](https://github.com/apache/doris/blob/339e087b11f760e9e45ce33d94862b4b63f1d9ee/be/src/format/generic_reader.h#L187)
therefore skips on_before_init_reader, which is where [Parquet extracts path
values](https://github.com/apache/doris/blob/339e087b11f760e9e45ce33d94862b4b63f1d9ee/be/src/format/parquet/vparquet_reader.cpp#L431).
The V1 path also cannot produce the proposed route constant.
The checked legacy ingestion producer omits the count with physical source
slots only and no columns_from_path. The existing
PushLoadsUseSourceConversionsAcrossParquetRanges test covers that omitted-count
shape and remains passing. This does not establish the reported path-constant
scenario as a regression from V1 retirement; support for that additional
payload combination is outside this PR's migration scope. Broker production
behavior is unchanged in this follow-up.
--
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]
