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 1f10685c0c5 [fix](be) Keep external scanner schema in projection order
(#68231)
1f10685c0c5 is described below
commit 1f10685c0c504a89859e36399442798f20236eb5
Author: Gabriel <[email protected]>
AuthorDate: Sun Sep 20 14:29:36 2026 +0800
[fix](be) Keep external scanner schema in projection order (#68231)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
External scanner metadata was returned in tuple-slot order while Arrow
batches from the memory scratch sink followed fragment output expression
order. Reordered projections could therefore bind values to the wrong
columns.
Build the selected-column schema from the fragment output slot
references so metadata and data use the same order. Add a BE unit test
that projects two slots in reverse descriptor order and verifies the
returned names and types follow the projection.
### Release note
Fix external scanner reads with reordered projected columns.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
-
`FragmentMgrExternalScanTest.SelectedColumnsFollowOutputExpressionOrder`
- [ ] Manual test
- [ ] No need to test or manual test.
- Behavior changed:
- [ ] No.
- [x] Yes. External scanner schema now follows projection order.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/runtime/fragment_mgr.cpp | 48 +++++++++++++------
be/src/runtime/fragment_mgr.h | 5 ++
.../fragment_mgr_cross_cluster_cancel_test.cpp | 54 ++++++++++++++++++++++
3 files changed, 92 insertions(+), 15 deletions(-)
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index 72dd4a6d7c7..a26e2c157d8 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -1012,6 +1012,37 @@ void FragmentMgr::_check_brpc_available(const
std::shared_ptr<PBackendService_St
}
void FragmentMgr::debug(std::stringstream& ss) {}
+
+Status FragmentMgr::_build_external_scan_selected_columns(
+ const TPlanFragment& plan_fragment, const DescriptorTbl& desc_tbl,
+ std::vector<TScanColumnDesc>* selected_columns) {
+ // The memory scratch sink emits Arrow columns in output expression order,
so the returned
+ // schema must use the same order to prevent positional column misbinding.
+ for (const auto& expr : plan_fragment.output_exprs) {
+ if (expr.nodes.empty() || expr.nodes[0].node_type !=
TExprNodeType::SLOT_REF) {
+ LOG(WARNING) << "output expr is not slot ref";
+ return Status::InvalidArgument("output expr is not slot ref");
+ }
+
+ const auto& slot_ref = expr.nodes[0].slot_ref;
+ if (desc_tbl.get_tuple_descriptor(slot_ref.tuple_id) == nullptr) {
+ LOG(WARNING) << "tuple descriptor is null. id: " <<
slot_ref.tuple_id;
+ return Status::InvalidArgument("tuple descriptor is null");
+ }
+ const auto* slot_desc = desc_tbl.get_slot_descriptor(slot_ref.slot_id);
+ if (slot_desc == nullptr) {
+ LOG(WARNING) << "slot descriptor is null. id: " <<
slot_ref.slot_id;
+ return Status::InvalidArgument("slot descriptor is null");
+ }
+
+ TScanColumnDesc column;
+ column.__set_name(slot_desc->col_name());
+ column.__set_type(to_thrift(slot_desc->type()->get_primitive_type()));
+ selected_columns->emplace_back(std::move(column));
+ }
+ return Status::OK();
+}
+
/*
* 1. resolve opaqued_query_plan to thrift structure
* 2. build TPipelineFragmentParams
@@ -1032,21 +1063,8 @@ Status FragmentMgr::exec_external_plan_fragment(const
TScanOpenParams& params,
"processed";
return Status::InvalidArgument(msg.str());
}
- TupleDescriptor* tuple_desc = desc_tbl->get_tuple_descriptor(0);
- if (tuple_desc == nullptr) {
- LOG(WARNING) << "open context error: extract TupleDescriptor failure";
- std::stringstream msg;
- msg << " get TupleDescriptor error, should not be modified after
returned Doris FE "
- "processed";
- return Status::InvalidArgument(msg.str());
- }
- // process selected columns form slots
- for (const SlotDescriptor* slot : tuple_desc->slots()) {
- TScanColumnDesc col;
- col.__set_name(slot->col_name());
- col.__set_type(to_thrift(slot->type()->get_primitive_type()));
- selected_columns->emplace_back(std::move(col));
- }
+
RETURN_IF_ERROR(_build_external_scan_selected_columns(t_query_plan_info.plan_fragment,
+ *desc_tbl,
selected_columns));
VLOG_QUERY << "BackendService execute open() TQueryPlanInfo: "
<< apache::thrift::ThriftDebugString(t_query_plan_info);
diff --git a/be/src/runtime/fragment_mgr.h b/be/src/runtime/fragment_mgr.h
index 426e395bc32..5a5bf64fd8b 100644
--- a/be/src/runtime/fragment_mgr.h
+++ b/be/src/runtime/fragment_mgr.h
@@ -53,6 +53,7 @@ extern bvar::Status<uint64_t> g_fragment_last_active_time;
class PipelineFragmentContext;
class QueryContext;
+class DescriptorTbl;
class ExecEnv;
struct FrontendInfo;
class ThreadPool;
@@ -219,6 +220,10 @@ private:
void _check_brpc_available(const std::shared_ptr<PBackendService_Stub>&
brpc_stub,
const BrpcItem& brpc_item);
+ static Status _build_external_scan_selected_columns(
+ const TPlanFragment& plan_fragment, const DescriptorTbl& desc_tbl,
+ std::vector<TScanColumnDesc>* selected_columns);
+
// This is input params
ExecEnv* _exec_env = nullptr;
diff --git a/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
b/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
index c178728712d..dd705635b6e 100644
--- a/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
+++ b/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+#include <gen_cpp/DorisExternalService_types.h>
#include <gen_cpp/PaloInternalService_types.h>
#include <gtest/gtest.h>
@@ -225,4 +226,57 @@ TEST(FragmentMgrRerunnableParamsTest,
StopReleasesLastQueryContextRefOutsideLock
delete fragment_mgr;
}
+TEST(FragmentMgrExternalScanTest, SelectedColumnsFollowOutputExpressionOrder) {
+ TDescriptorTableBuilder desc_tbl_builder;
+ TTupleDescriptorBuilder tuple_builder;
+ tuple_builder
+ .add_slot(TSlotDescriptorBuilder()
+ .type(TYPE_INT)
+ .nullable(true)
+ .column_name("k1")
+ .column_pos(0)
+ .build())
+ .add_slot(TSlotDescriptorBuilder()
+ .type(TYPE_BIGINT)
+ .nullable(true)
+ .column_name("v2")
+ .column_pos(1)
+ .build());
+ tuple_builder.build(&desc_tbl_builder);
+
+ ObjectPool pool;
+ DescriptorTbl* desc_tbl = nullptr;
+ ASSERT_TRUE(DescriptorTbl::create(&pool, desc_tbl_builder.desc_tbl(),
&desc_tbl).ok());
+
+ auto make_slot_ref_expr = [](TSlotId slot_id, TPrimitiveType::type type) {
+ TSlotRef slot_ref;
+ slot_ref.__set_slot_id(slot_id);
+ slot_ref.__set_tuple_id(0);
+
+ TExprNode node;
+ node.__set_node_type(TExprNodeType::SLOT_REF);
+ node.__set_type(TSlotDescriptorBuilder().get_common_type(type));
+ node.__set_num_children(0);
+ node.__set_slot_ref(slot_ref);
+
+ TExpr expr;
+ expr.nodes.emplace_back(std::move(node));
+ return expr;
+ };
+
+ TPlanFragment plan_fragment;
+ plan_fragment.__set_output_exprs({make_slot_ref_expr(1,
TPrimitiveType::BIGINT),
+ make_slot_ref_expr(0,
TPrimitiveType::INT)});
+
+ std::vector<TScanColumnDesc> selected_columns;
+
ASSERT_TRUE(FragmentMgr::_build_external_scan_selected_columns(plan_fragment,
*desc_tbl,
+
&selected_columns)
+ .ok());
+ ASSERT_EQ(selected_columns.size(), 2);
+ EXPECT_EQ(selected_columns[0].name, "v2");
+ EXPECT_EQ(selected_columns[0].type, TPrimitiveType::BIGINT);
+ EXPECT_EQ(selected_columns[1].name, "k1");
+ EXPECT_EQ(selected_columns[1].type, TPrimitiveType::INT);
+}
+
} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
