Refine assert in Pipeline Model

dbms/src/Flash/Pipeline/Exec/PipelineExecBuilder.cpp

@@ -18,27 +18,27 @@ namespace DB
 {
 void PipelineExecBuilder::setSourceOp(SourceOpPtr && source_op_)
 {
-    assert(!source_op && source_op_);
+    RUNTIME_CHECK(!source_op && source_op_);
     source_op = std::move(source_op_);
 }
 void PipelineExecBuilder::appendTransformOp(TransformOpPtr && transform_op)
 {
-    assert(source_op && transform_op);
+    RUNTIME_CHECK(source_op && transform_op);
     Block header = getCurrentHeader();
     transform_op->transformHeader(header);
     transform_ops.push_back(std::move(transform_op));
 }
 void PipelineExecBuilder::setSinkOp(SinkOpPtr && sink_op_)
 {
-    assert(!sink_op && sink_op_);
+    RUNTIME_CHECK(!sink_op && sink_op_);
     Block header = getCurrentHeader();
     sink_op_->setHeader(header);
     sink_op = std::move(sink_op_);
 }
 
 PipelineExecPtr PipelineExecBuilder::build()
 {
-    assert(source_op && sink_op);
+    RUNTIME_CHECK(source_op && sink_op);
     return std::make_unique<PipelineExec>(
         std::move(source_op),
         std::move(transform_ops),
@@ -53,22 +53,22 @@ Block PipelineExecBuilder::getCurrentHeader() const
         return transform_ops.back()->getHeader();
     else
     {
-        assert(source_op);
+        RUNTIME_CHECK(source_op);
         return source_op->getHeader();
     }
 }
 
 void PipelineExecGroupBuilder::init(size_t init_concurrency)
 {
-    assert(concurrency == 0);
-    assert(init_concurrency > 0);
+    RUNTIME_CHECK(concurrency == 0);
+    RUNTIME_CHECK(init_concurrency > 0);
     concurrency = init_concurrency;
     group.resize(concurrency);
 }
 
 PipelineExecGroup PipelineExecGroupBuilder::build()
 {
-    assert(concurrency > 0);
+    RUNTIME_CHECK(concurrency > 0);
     PipelineExecGroup pipeline_exec_group;
     for (auto & builder : group)
         pipeline_exec_group.push_back(builder.build());
@@ -77,7 +77,7 @@ PipelineExecGroup PipelineExecGroupBuilder::build()
 
 Block PipelineExecGroupBuilder::getCurrentHeader()
 {
-    assert(!group.empty());
+    RUNTIME_CHECK(!group.empty());
     return group.back().getCurrentHeader();
 }
 } // namespace DB

dbms/src/Flash/Pipeline/Pipeline.cpp

@@ -121,13 +121,13 @@ void Pipeline::addPlanNode(const PhysicalPlanNodePtr & plan_node)
 
 void Pipeline::addChild(const PipelinePtr & child)
 {
-    assert(child);
+    RUNTIME_CHECK(child);
     children.push_back(child);
 }
 
 Block Pipeline::getSampleBlock() const
 {
-    assert(!plan_nodes.empty());
+    RUNTIME_CHECK(!plan_nodes.empty());
     return plan_nodes.back()->getSampleBlock();
 }
 
@@ -155,14 +155,14 @@ void Pipeline::toTreeString(FmtBuffer & buffer, size_t level) const
 
 void Pipeline::addGetResultSink(const ResultQueuePtr & result_queue)
 {
-    assert(!plan_nodes.empty());
+    RUNTIME_CHECK(!plan_nodes.empty());
     auto get_result_sink = PhysicalGetResultSink::build(result_queue, log, plan_nodes.back());
     addPlanNode(get_result_sink);
 }
 
 PipelineExecGroup Pipeline::buildExecGroup(PipelineExecutorStatus & exec_status, Context & context, size_t concurrency)
 {
-    assert(!plan_nodes.empty());
+    RUNTIME_CHECK(!plan_nodes.empty());
     PipelineExecGroupBuilder builder;
     for (const auto & plan_node : plan_nodes)
     {
@@ -188,7 +188,7 @@ PipelineExecGroup Pipeline::buildExecGroup(PipelineExecutorStatus & exec_status,
  */
 bool Pipeline::isFineGrainedMode() const
 {
-    assert(!plan_nodes.empty());
+    RUNTIME_CHECK(!plan_nodes.empty());
     // The source plan node determines whether the execution mode is fine grained or non-fine grained.
     return plan_nodes.front()->getFineGrainedShuffle().enable();
 }
@@ -197,15 +197,15 @@ Events Pipeline::toEvents(PipelineExecutorStatus & status, Context & context, si
 {
     Events all_events;
     doToEvents(status, context, concurrency, all_events);
-    assert(!all_events.empty());
+    RUNTIME_CHECK(!all_events.empty());
     return all_events;
 }
 
 PipelineEvents Pipeline::toSelfEvents(PipelineExecutorStatus & status, Context & context, size_t concurrency)
 {
     auto memory_tracker = current_memory_tracker ? current_memory_tracker->shared_from_this() : nullptr;
     Events self_events;
-    assert(!plan_nodes.empty());
+    RUNTIME_CHECK(!plan_nodes.empty());
     if (isFineGrainedMode())
     {
         auto fine_grained_exec_group = buildExecGroup(status, context, concurrency);

dbms/src/Flash/Pipeline/PipelineBuilder.h

@@ -53,8 +53,8 @@ class PipelineBuilder
             : pipeline(pipeline_)
             , breaker_node(breaker_node_)
         {
-            assert(pipeline);
-            assert(breaker_node);
+            RUNTIME_CHECK(pipeline);
+            RUNTIME_CHECK(breaker_node);
         }
 
         // the broken pipeline.
@@ -87,7 +87,7 @@ class PipelineBuilder
 
     PipelinePtr build()
     {
-        assert(pipeline);
+        RUNTIME_CHECK(pipeline);
         if (pipeline_breaker)
         {
             // First add the breaker node as the last node in this pipeline.

dbms/src/Flash/Pipeline/Schedule/Events/Event.cpp

@@ -40,8 +40,8 @@ extern const char random_pipeline_model_event_finish_failpoint[];
 
 void Event::addInput(const EventPtr & input)
 {
-    assert(status == EventStatus::INIT);
-    assert(input.get() != this);
+    RUNTIME_CHECK(status == EventStatus::INIT);
+    RUNTIME_CHECK(input.get() != this);
     input->addOutput(shared_from_this());
     ++unfinished_inputs;
     is_source = false;
@@ -50,8 +50,8 @@ void Event::addInput(const EventPtr & input)
 void Event::addOutput(const EventPtr & output)
 {
     /// Output will also be added in the Finished state, as can be seen in the `insertEvent`.
-    assert(status == EventStatus::INIT || status == EventStatus::FINISHED);
-    assert(output.get() != this);
+    RUNTIME_CHECK(status == EventStatus::INIT || status == EventStatus::FINISHED);
+    RUNTIME_CHECK(output.get() != this);
     outputs.push_back(output);
 }
 
@@ -81,7 +81,7 @@ void Event::onInputFinish() noexcept
 
 bool Event::prepare()
 {
-    assert(status == EventStatus::INIT);
+    RUNTIME_CHECK(status == EventStatus::INIT);
     if (is_source)
     {
         // For source event, `exec_status.onEventSchedule()` needs to be called before schedule.

dbms/src/Flash/Pipeline/Schedule/Events/FineGrainedPipelineEvent.h

@@ -30,7 +30,7 @@ class FineGrainedPipelineEvent : public Event
         : Event(exec_status_, std::move(mem_tracker_), req_id)
         , pipeline_exec(std::move(pipeline_exec_))
     {
-        assert(pipeline_exec);
+        RUNTIME_CHECK(pipeline_exec);
     }
 
 protected:

dbms/src/Flash/Pipeline/Schedule/Events/PlainPipelineEvent.cpp

@@ -20,7 +20,7 @@ namespace DB
 {
 std::vector<TaskPtr> PlainPipelineEvent::scheduleImpl()
 {
-    assert(pipeline);
+    RUNTIME_CHECK(pipeline);
     auto pipeline_exec_group = pipeline->buildExecGroup(exec_status, context, concurrency);
     RUNTIME_CHECK(!pipeline_exec_group.empty());
     std::vector<TaskPtr> tasks;

dbms/src/Flash/Pipeline/Schedule/Tasks/EventTask.cpp

@@ -28,7 +28,7 @@ EventTask::EventTask(
     : exec_status(exec_status_)
     , event(event_)
 {
-    assert(event);
+    RUNTIME_CHECK(event);
 }
 
 EventTask::EventTask(
@@ -40,7 +40,7 @@ EventTask::EventTask(
     , exec_status(exec_status_)
     , event(event_)
 {
-    assert(event);
+    RUNTIME_CHECK(event);
 }
 
 EventTask::~EventTask()

dbms/src/Flash/Pipeline/Schedule/Tasks/EventTask.h

@@ -34,7 +34,7 @@ class EventTask : public Task
         PipelineExecutorStatus & exec_status_,
         const EventPtr & event_);
 
-    ~EventTask();
+    ~EventTask() override;
 
 protected:
     ExecTaskStatus executeImpl() noexcept override;

dbms/src/Flash/Pipeline/Schedule/Tasks/PipelineTask.cpp

@@ -28,13 +28,13 @@ PipelineTask::PipelineTask(
     : EventTask(std::move(mem_tracker_), req_id, exec_status_, event_)
     , pipeline_exec(std::move(pipeline_exec_))
 {
-    assert(pipeline_exec);
+    RUNTIME_CHECK(pipeline_exec);
     pipeline_exec->executePrefix();
 }
 
 void PipelineTask::finalizeImpl()
 {
-    assert(pipeline_exec);
+    RUNTIME_CHECK(pipeline_exec);
     pipeline_exec->executeSuffix();
     pipeline_exec.reset();
 }
@@ -62,7 +62,7 @@ void PipelineTask::finalizeImpl()
 
 ExecTaskStatus PipelineTask::doExecuteImpl()
 {
-    assert(pipeline_exec);
+    RUNTIME_CHECK(pipeline_exec);
     auto op_status = pipeline_exec->execute();
     switch (op_status)
     {
@@ -78,7 +78,7 @@ ExecTaskStatus PipelineTask::doExecuteImpl()
 
 ExecTaskStatus PipelineTask::doExecuteIOImpl()
 {
-    assert(pipeline_exec);
+    RUNTIME_CHECK(pipeline_exec);
     auto op_status = pipeline_exec->executeIO();
     switch (op_status)
     {
@@ -97,7 +97,7 @@ ExecTaskStatus PipelineTask::doExecuteIOImpl()
 
 ExecTaskStatus PipelineTask::doAwaitImpl()
 {
-    assert(pipeline_exec);
+    RUNTIME_CHECK(pipeline_exec);
     auto op_status = pipeline_exec->await();
     switch (op_status)
     {

dbms/src/Flash/Planner/FinalizeHelper.cpp

@@ -27,7 +27,7 @@ namespace DB::FinalizeHelper
 {
 void prependProjectInputIfNeed(ExpressionActionsPtr & actions, size_t columns_from_previous)
 {
-    assert(columns_from_previous >= actions->getRequiredColumnsWithTypes().size());
+    RUNTIME_CHECK(columns_from_previous >= actions->getRequiredColumnsWithTypes().size());
     if (!actions->getRequiredColumnsWithTypes().empty()
         && columns_from_previous > actions->getRequiredColumnsWithTypes().size())
     {

dbms/src/Flash/Planner/PhysicalPlan.cpp

@@ -64,7 +64,7 @@ bool pushDownSelection(Context & context, const PhysicalPlanNodePtr & plan, cons
 
 void PhysicalPlan::build(const tipb::DAGRequest * dag_request)
 {
-    assert(dag_request);
+    RUNTIME_CHECK(dag_request);
     traverseExecutorsReverse(
         dag_request,
         [&](const tipb::Executor & executor) {
@@ -85,8 +85,8 @@ void PhysicalPlan::buildTableScan(const String & executor_id, const tipb::Execut
 
 void PhysicalPlan::build(const tipb::Executor * executor)
 {
-    assert(executor);
-    assert(executor->has_executor_id());
+    RUNTIME_CHECK(executor);
+    RUNTIME_CHECK(executor->has_executor_id());
     const auto & executor_id = executor->executor_id();
     switch (executor->tp())
     {
@@ -214,15 +214,15 @@ DAGContext & PhysicalPlan::dagContext() const
 
 void PhysicalPlan::pushBack(const PhysicalPlanNodePtr & plan_node)
 {
-    assert(plan_node);
+    RUNTIME_CHECK(plan_node);
     cur_plan_nodes.push_back(plan_node);
 }
 
 PhysicalPlanNodePtr PhysicalPlan::popBack()
 {
     RUNTIME_CHECK(!cur_plan_nodes.empty());
     PhysicalPlanNodePtr back = cur_plan_nodes.back();
-    assert(back);
+    RUNTIME_CHECK(back);
     cur_plan_nodes.pop_back();
     return back;
 }
@@ -231,7 +231,7 @@ PhysicalPlanNodePtr PhysicalPlan::popBack()
 /// For batchcop/cop that without PhysicalExchangeSender or PhysicalMockExchangeSender, We need to add root final projection.
 void PhysicalPlan::addRootFinalProjectionIfNeed()
 {
-    assert(root_node);
+    RUNTIME_CHECK(root_node);
     if (root_node->tp() != PlanType::ExchangeSender && root_node->tp() != PlanType::MockExchangeSender)
     {
         pushBack(root_node);
@@ -265,19 +265,19 @@ PhysicalPlanNodePtr PhysicalPlan::outputAndOptimize()
 
 String PhysicalPlan::toString() const
 {
-    assert(root_node);
+    RUNTIME_CHECK(root_node);
     return PhysicalPlanVisitor::visitToString(root_node);
 }
 
 void PhysicalPlan::buildBlockInputStream(DAGPipeline & pipeline, Context & context, size_t max_streams)
 {
-    assert(root_node);
+    RUNTIME_CHECK(root_node);
     root_node->buildBlockInputStream(pipeline, context, max_streams);
 }
 
 PipelinePtr PhysicalPlan::toPipeline(PipelineExecutorStatus & exec_status, Context & context)
 {
-    assert(root_node);
+    RUNTIME_CHECK(root_node);
     PipelineBuilder builder{log->identifier()};
     root_node->buildPipeline(builder, context, exec_status);
     root_node.reset();

dbms/src/Flash/Planner/PhysicalPlanHelper.cpp

@@ -42,8 +42,8 @@ NamesAndTypes addSchemaProjectAction(
     const NamesAndTypes & before_schema,
     const String & column_prefix)
 {
-    assert(expr_actions);
-    assert(!before_schema.empty());
+    RUNTIME_CHECK(expr_actions);
+    RUNTIME_CHECK(!before_schema.empty());
 
     NamesAndTypes after_schema = before_schema;
     NamesWithAliases project_aliases;
@@ -68,7 +68,7 @@ void addParentRequireProjectAction(
     const ExpressionActionsPtr & expr_actions,
     const Names & parent_require)
 {
-    assert(expr_actions);
+    RUNTIME_CHECK(expr_actions);
     NamesWithAliases project_aliases;
     {
         std::unordered_set<String> column_name_set;

dbms/src/Flash/Planner/PhysicalPlanNode.cpp

@@ -101,7 +101,7 @@ void PhysicalPlanNode::buildPipelineExecGroup(
 
 void PhysicalPlanNode::buildPipeline(PipelineBuilder & builder, Context & context, PipelineExecutorStatus & exec_status)
 {
-    assert(childrenSize() <= 1);
+    RUNTIME_CHECK(childrenSize() <= 1);
     if (childrenSize() == 1)
         children(0)->buildPipeline(builder, context, exec_status);
     builder.addPlanNode(shared_from_this());

dbms/src/Flash/Planner/PhysicalPlanVisitor.cpp

@@ -27,7 +27,7 @@ void addPrefix(FmtBuffer & buffer, size_t level)
 void doVisitToString(FmtBuffer & buffer, const PhysicalPlanNodePtr & physical_plan, size_t level)
 {
     visit(physical_plan, [&buffer, &level](const PhysicalPlanNodePtr & plan) {
-        assert(plan);
+        RUNTIME_CHECK(plan);
         addPrefix(buffer, level);
         buffer.fmtAppend("{}\n", plan->toString());
         ++level;