924060929 commented on code in PR #68283:
URL: https://github.com/apache/doris/pull/68283#discussion_r4059738871
##########
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:
Fixed by removing the INSERT INTO TVF changes from this PR. TVF replacement
is a separate non-transactional path: the FE recursively deletes existing
external files before the write, and there is no rollback boundary that
coordinator publication can make atomic. A safe fix requires staged or atomic
replacement (or an explicitly non-cancellable destructive section), which is
broader than DORIS-28658. The PR now limits coordinator publication to
transactional AbstractInsertExecutor paths, and the description no longer
claims TVF coverage.
##########
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:
Fixed in ab37671102b. Both coordinator implementations now validate and
publish the terminal status and invoke internal cancellation before queue/scan
cleanup. Legacy Coordinator does that under its coordinator lock; Nereids does
it through CoordinatorContext.updateStatusIfOk under the existing monitor, with
a finally block preserving the final cancel resend. Added tests for both
implementations where ScanNode.stop() throws and verified the TIMEOUT status
and internal cancellation were already visible before the cleanup failure.
--
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]
