fix

Signed-off-by: zihe.liu <[email protected]>
StarRocks · Sep 18, 2024 · c258e32 · c258e32
1 parent b3a6939
commit c258e32
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 2 deletions.
diff --git a/be/src/exec/spill/spiller.hpp b/be/src/exec/spill/spiller.hpp
@@ -346,7 +346,7 @@ Status PartitionedSpillerWriter::flush(RuntimeState* state, bool is_final_flush,
         return Status::OK();
     };
     auto yield_func = [&](workgroup::ScanTask&& task) { TaskExecutor::force_submit(std::move(task)); };
-    auto io_task = workgroup::ScanTask(_spiller->options().wg.get(), std::move(task), std::move(yield_func));
+    auto io_task = workgroup::ScanTask(_spiller->options().wg, std::move(task), std::move(yield_func));
     RETURN_IF_ERROR(TaskExecutor::submit(std::move(io_task)));
     COUNTER_UPDATE(_spiller->metrics().flush_io_task_count, 1);
     COUNTER_SET(_spiller->metrics().peak_flush_io_task_count, _running_flush_tasks);

diff --git a/be/src/exec/topn_node.cpp b/be/src/exec/topn_node.cpp
@@ -278,6 +278,9 @@ std::vector<std::shared_ptr<pipeline::OperatorFactory>> TopNNode::_decompose_to_
             ops_sink_with_sort = context->maybe_interpolate_local_passthrough_exchange(
                     runtime_state(), id(), ops_sink_with_sort, context->degree_of_parallelism(), is_partition_skewed);
         }
+    } else if (is_partition_topn) {
+        ops_sink_with_sort = context->maybe_interpolate_local_passthrough_exchange(
+                runtime_state(), id(), ops_sink_with_sort, context->degree_of_parallelism(), is_partition_skewed);
     } else {
         // prepend local shuffle to PartitionSortSinkOperator
         ops_sink_with_sort = context->maybe_interpolate_local_shuffle_exchange(

diff --git a/be/src/udf/java/utils.cpp b/be/src/udf/java/utils.cpp
@@ -51,7 +51,7 @@ PromiseStatusPtr call_hdfs_scan_function_in_pthread(const std::function<Status()
     PromiseStatusPtr ms = std::make_unique<PromiseStatus>();
     if (bthread_self()) {
         ExecEnv::GetInstance()->connector_scan_executor()->submit(workgroup::ScanTask(
-                ExecEnv::GetInstance()->workgroup_manager()->get_default_workgroup().get(),
+                ExecEnv::GetInstance()->workgroup_manager()->get_default_workgroup(),
                 [promise = ms.get(), func](workgroup::YieldContext&) { promise->set_value(func()); }));
     } else {
         ms->set_value(func());