github-actions[bot] commented on code in PR #68283:
URL: https://github.com/apache/doris/pull/68283#discussion_r4059584495
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTVFCommand.java:
##########
@@ -98,29 +98,34 @@ public void run(ConnectContext ctx, StmtExecutor executor)
throws Exception {
executor.setPlanner(planner);
executor.checkBlockRules();
- // FE-side deletion of existing files (before BE execution)
- PhysicalPlan physicalPlan = planner.getPhysicalPlan();
- if (physicalPlan instanceof PhysicalTVFTableSink) {
- PhysicalTVFTableSink<?> tvfSink = (PhysicalTVFTableSink<?>)
physicalPlan;
- String sinkTvfName = tvfSink.getTvfName();
- Map<String, String> sinkProps = tvfSink.getProperties();
- boolean deleteExisting = Boolean.parseBoolean(
- sinkProps.getOrDefault("delete_existing_files", "false"));
-
- if (deleteExisting && !"local".equals(sinkTvfName)) {
- deleteExistingFilesInFE(sinkTvfName, sinkProps);
- }
- }
-
- // 3. Create coordinator
Coordinator coordinator = EnvFactory.getInstance().createCoordinator(
ctx, planner, ctx.getStatsErrorEstimator());
-
+ executor.setCoord(coordinator);
TUniqueId queryId = ctx.queryId();
- QeProcessorImpl.INSTANCE.registerQuery(queryId,
- new QueryInfo(ctx, "INSERT INTO TVF", coordinator));
try {
+ // FE-side deletion of existing files (before BE execution)
+ PhysicalPlan physicalPlan = planner.getPhysicalPlan();
+ if (physicalPlan instanceof PhysicalTVFTableSink) {
+ PhysicalTVFTableSink<?> tvfSink = (PhysicalTVFTableSink<?>)
physicalPlan;
+ String sinkTvfName = tvfSink.getTvfName();
+ Map<String, String> sinkProps = tvfSink.getProperties();
+ boolean deleteExisting = Boolean.parseBoolean(
+ sinkProps.getOrDefault("delete_existing_files",
"false"));
+
+ if (deleteExisting && !"local".equals(sinkTvfName)) {
+ Status status = coordinator.getExecStatus();
Review Comment:
[P1] Make cancellation atomic with the destructive TVF replace phase. A
cancellation can complete after this OK snapshot and before or during
`deleteExistingFilesInFE()`; the statement then recursively deletes the
existing destination, while `coordinator.exec()` rejects dispatch and writes no
replacement. Serializing only entry to deletion is insufficient because
cancellation can still win before replacement dispatch. Stage and atomically
replace the output, or defer cancellation across delete plus replacement, and
cover the crossing with a latch-controlled test.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java:
##########
@@ -919,6 +923,9 @@ protected void execInternal() throws Exception {
protected void sendPipelineCtx() throws Exception {
lock();
try {
+ if (!queryStatus.ok()) {
Review Comment:
[P1] Publish terminal status before fallible scan cleanup. Both coordinator
`cancel()` implementations call `ScanNode.stop()` before updating their status;
`FileQueryScanNode` reaches `SplitAssignment.stop()`, which can throw, and
cleanup may also block. In that window this new fence still sees OK and can
dispatch, while a thrown cleanup prevents both status publication and backend
cancellation. Store the non-OK status first under the coordinator lock/monitor,
then guarantee `cancelInternal()` despite cleanup failures, and test
blocked/throwing scan stop in both coordinators.
--
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]
