class DeletePlanner {

    public static Plan planDelete(AnalyzedDeleteStatement delete,
                                  SubqueryPlanner subqueryPlanner,
                                  PlannerContext context) {
        Plan plan = planDelete(delete, context);
        return MultiPhasePlan.createIfNeeded(plan, subqueryPlanner.planSubQueries(delete).uncorrelated());
    }

    private static Plan planDelete(AnalyzedDeleteStatement delete, PlannerContext context) {<FILL_FUNCTION_BODY>}

    @VisibleForTesting
    public static class Delete implements Plan {

        private final DocTableRelation table;
        private final WhereClauseOptimizer.DetailedQuery detailedQuery;

        public Delete(DocTableRelation table, WhereClauseOptimizer.DetailedQuery detailedQuery) {
            this.table = table;
            this.detailedQuery = detailedQuery;
        }

        @Override
        public StatementType type() {
            return StatementType.DELETE;
        }

        @Override
        public void executeOrFail(DependencyCarrier executor,
            PlannerContext plannerContext,
            RowConsumer consumer,
            Row params,
            SubQueryResults subQueryResults) {

            WhereClause where = detailedQuery.toBoundWhereClause(
                table.tableInfo(),
                params,
                subQueryResults,
                plannerContext.transactionContext(),
                executor.nodeContext());
            if (!where.partitions().isEmpty()
                && (!where.hasQuery() || Literal.BOOLEAN_TRUE.equals(where.query()))) {
                DeleteIndexRequest request = new DeleteIndexRequest(where.partitions().toArray(new String[0]));
                request.indicesOptions(IndicesOptions.lenientExpandOpen());
                executor.client().execute(DeleteIndexAction.INSTANCE, request)
                    .whenComplete(new OneRowActionListener<>(consumer, o -> new Row1(-1L)));
                return;
            }

            ExecutionPlan executionPlan = deleteByQuery(table, plannerContext, where);
            NodeOperationTree nodeOpTree = NodeOperationTreeGenerator.fromPlan(executionPlan, executor.localNodeId());
            executor.phasesTaskFactory()
                .create(plannerContext.jobId(), Collections.singletonList(nodeOpTree))
                .execute(consumer, plannerContext.transactionContext());
        }

        @Override
        public List<CompletableFuture<Long>> executeBulk(DependencyCarrier executor,
                                                         PlannerContext plannerContext,
                                                         List<Row> bulkParams,
                                                         SubQueryResults subQueryResults) {
            ArrayList<NodeOperationTree> nodeOperationTreeList = new ArrayList<>(bulkParams.size());
            for (Row params : bulkParams) {
                WhereClause where = detailedQuery.toBoundWhereClause(
                    table.tableInfo(),
                    params,
                    subQueryResults,
                    plannerContext.transactionContext(),
                    executor.nodeContext());
                ExecutionPlan executionPlan = deleteByQuery(table, plannerContext, where);
                nodeOperationTreeList.add(NodeOperationTreeGenerator.fromPlan(executionPlan, executor.localNodeId()));
            }
            return executor.phasesTaskFactory()
                .create(plannerContext.jobId(), nodeOperationTreeList)
                .executeBulk(plannerContext.transactionContext());
        }
    }

    private static ExecutionPlan deleteByQuery(DocTableRelation table, PlannerContext context, WhereClause where) {
        DocTableInfo tableInfo = table.tableInfo();
        Reference idReference = requireNonNull(tableInfo.getReference(DocSysColumns.ID), "Table has to have a _id reference");
        DeleteProjection deleteProjection = new DeleteProjection(new InputColumn(0, idReference.valueType()));
        var sessionSettings = context.transactionContext().sessionSettings();
        Routing routing = context.allocateRouting(
            tableInfo, where, RoutingProvider.ShardSelection.PRIMARIES, sessionSettings);
        RoutedCollectPhase collectPhase = new RoutedCollectPhase(
            context.jobId(),
            context.nextExecutionPhaseId(),
            "collect",
            routing,
            tableInfo.rowGranularity(),
            List.of(idReference),
            List.of(deleteProjection),
            Optimizer.optimizeCasts(where.queryOrFallback(), context),
            DistributionInfo.DEFAULT_BROADCAST
        );
        Collect collect = new Collect(collectPhase, LimitAndOffset.NO_LIMIT, 0, 1, 1, null);
        return Merge.ensureOnHandler(collect, context, Collections.singletonList(MergeCountProjection.INSTANCE));
    }
}

        DocTableRelation tableRel = delete.relation();
        DocTableInfo table = tableRel.tableInfo();
        EvaluatingNormalizer normalizer = EvaluatingNormalizer.functionOnlyNormalizer(context.nodeContext());
        WhereClauseOptimizer.DetailedQuery detailedQuery = WhereClauseOptimizer.optimize(
            normalizer, delete.query(), table, context.transactionContext(), context.nodeContext());
        Symbol query = detailedQuery.query();
        if (!detailedQuery.partitions().isEmpty()) {
            // deleting whole partitions is only valid if the query only contains filters based on partition-by cols
            var hasNonPartitionReferences = SymbolVisitors.any(
                s -> s instanceof Reference && table.partitionedByColumns().contains(s) == false,
                query
            );
            if (hasNonPartitionReferences == false) {
                return new DeletePartitions(table.ident(), detailedQuery.partitions());
            }
        }

        if (detailedQuery.docKeys().isPresent() && detailedQuery.queryHasPkSymbolsOnly()) {
            return new DeleteById(tableRel.tableInfo(), detailedQuery.docKeys().get());
        }
        if (table.isPartitioned() && query instanceof Input<?> input && DataTypes.BOOLEAN.sanitizeValue(input.value())) {
            return new DeleteAllPartitions(Lists.map(table.partitions(), IndexParts::toIndexName));
        }

        return new Delete(tableRel, detailedQuery);
    
<no_super_class>