hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

An exception occurred in the task, notify other tasks to exit

Browse Source
This commit is contained in:
obdev
2024-02-27 04:21:10 +00:00
committed by ob-robot
parent bc414092c3
commit f91e753574
3 changed files with 11 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/sql/engine/px/exchange/ob_px_transmit_op.cpp
View File

@ -1372,6 +1372,7 @@ int ObPxTransmitOp::link_ch_sets(ObPxTaskChSet &ch_set,
return ret;
}
ERRSIM_POINT_DEF(ERRSIM_DYNAMIC_SAMPLE_FAIL)
int ObPxTransmitOp::do_datahub_dynamic_sample(int64_t op_id, ObDynamicSamplePieceMsg &piece_msg)
{
int ret = OB_SUCCESS;
@ -1381,6 +1382,9 @@ int ObPxTransmitOp::do_datahub_dynamic_sample(int64_t op_id, ObDynamicSamplePiec
ret = OB_NOT_SUPPORTED;
LOG_WARN("dynimic sample only supported in parallel execution mode", K(ret));
LOG_USER_ERROR(OB_NOT_SUPPORTED, "dynimic sample in non-px mode");
} else if (OB_SUCCESS != ERRSIM_DYNAMIC_SAMPLE_FAIL &&
ctx_.get_px_task_id() == 0) {
ret = ERRSIM_DYNAMIC_SAMPLE_FAIL;
} else {
const ObDynamicSampleWholeMsg *temp_whole_msg = NULL;
ObPxSQCProxy &proxy = handler->get_sqc_proxy();

3
src/sql/engine/px/ob_px_sqc_proxy.h
View File

@ -307,7 +307,8 @@ int ObPxSQCProxy::inner_get_dh_msg(
// wait 50us
ob_usleep(50);
}
} while (common::OB_EAGAIN == ret);
} while (common::OB_EAGAIN == ret &&
OB_SUCC(THIS_WORKER.check_status()));
}
}
return ret;

4
src/sql/engine/px/ob_px_task_process.cpp
View File

@ -333,6 +333,7 @@ int ObPxTaskProcess::execute(ObOpSpec &root_spec)
return ret;
}
ERRSIM_POINT_DEF(ERRSIM_INTERRUPT_QC_FAILED)
int ObPxTaskProcess::do_process()
{
LOG_TRACE("[CMD] run task", "task", arg_.task_);
@ -484,8 +485,11 @@ int ObPxTaskProcess::do_process()
&& ObVirtualTableErrorWhitelist::should_ignore_vtable_error(ret)) {
// 忽略虚拟表错误
} else {
if (OB_SUCCESS == ERRSIM_INTERRUPT_QC_FAILED) {
(void) ObInterruptUtil::interrupt_qc(arg_.task_, ret, arg_.exec_ctx_);
}
(void) ObInterruptUtil::interrupt_tasks(arg_.get_sqc_handler()->get_sqc_init_arg().sqc_, ret);
}
}
LOG_TRACE("notify SQC task exit", K(dfo_id), K(sqc_id), K(task_id), K(ret));