# Batch Import Into Operations

## Analysis

This document consolidates Import Into batch work. The progress-sync batch query is already represented by the existing `SHOW IMPORT JOBS WHERE group_key = ...` plan; the next batch work extends the same direction to submit/get/cancel operations so one import task avoids per-subtask agent or TiDB round trips.

| Area | Impact | Verification |
| --- | --- | --- |
| Importing progress sync | Replace per-subtask `SHOW IMPORT JOB <jobID>` polling with one `SHOW IMPORT JOBS WHERE group_key = ...` query per import task and reuse rows via in-memory lookup. | `go test ./internal/model/import -run 'TestRunImportIntoImportingProgressSyncWithBatchLookup|TestFindImportIntoJobStatusesFromResp'` |
| Canceling progress sync | Reuse the same batch query after cancel execution while preserving the existing `jobID == 0 => canceled` shortcut. | `go test ./internal/model/import -run 'TestRunImportIntoCancelingProgressSyncWithBatchLookup'` |
| Async submit create/get | Replace per-subtask `CreateImportIntoJobTask` / `GetImportIntoJobTask` calls with batch agent APIs while preserving deterministic task IDs and per-subtask result handling. | `go test ./internal/model/import -run 'Test(RunImportIntoScalingUp.*Batch|RecoverImportIntoJobIDWithExec|FindRecoverableImportIntoJobIDFailsClosed)' -count=1` |
| Cancel import jobs | Replace per-subtask `CANCEL IMPORT JOB <jobID>` execution with one batch cancel agent call that returns per-job errors. | `go test ./internal/model/import -run 'TestRunImportIntoCanceling.*Batch' -count=1` |
| Submit failure recovery | Reuse one `SHOW IMPORT JOBS WHERE group_key = ...` lookup for all recoverable submit failures in one worker tick. | `go test ./internal/model/import -run 'Test(RecoverImportIntoJobIDWithExec|FindRecoverableImportIntoJobIDFailsClosed)' -count=1` |
| Logging and resilience | Keep per-subtask sync, submit, and cancel failures non-fatal so later subtasks still progress in the same worker tick. | `make app-unit-test-local` |
| Review follow-up resilience | Keep malformed batch progress fields and zero-job cancel ticks local to one subtask, and treat omitted submit-task batch items as wait states instead of triggering same-loop single-task polling. | `go test ./internal/model/import -run 'Test(FindImportIntoJobStatusesFromRespFallsBackOnInvalidProgressPercent|ShouldSyncImportIntoCancelingProgressBatch|RunImportIntoScalingUpBatchWaitsOnOmittedPendingSubmitTask|RunImportIntoCancelingBatchWaitsOnOmittedPendingSubmitTask|ResolveImportIntoJobTaskStateAndApplyResult)' -count=1` and `go test ./dataflow-agent-ng/agent/grpc -run 'Test(BatchGetImportIntoJobTasks|GetImportIntoJobTaskFailed)' -count=1` |

Symbols to keep:
- `recoverImportIntoJobIDWithExec`
- `findRecoverableImportIntoJobID`
- `applyImportIntoProgressStatus`
- `resolveImportIntoImportingDecision`
- `resolveImportIntoCancelingDecision`
- existing single `CreateImportIntoJobTask` / `GetImportIntoJobTask` agent APIs for compatibility

Symbols to add:
- batch `SHOW IMPORT JOBS` query helper
- parsed batch row status struct / lookup
- lookup-based importing and canceling sync hooks
- `BatchCreateImportIntoJobTasks`
- `BatchGetImportIntoJobTasks`
- `BatchCancelImportIntoJobs`
- batch submit/cancel orchestration helpers in the Import Into model

Symbols to reduce or replace:
- per-subtask `SHOW IMPORT JOB <jobID>` polling during importing/canceling
- per-subtask async submit task create/get calls during scaling up
- per-subtask `CANCEL IMPORT JOB <jobID>` execution during canceling
- repeated recovery `SHOW IMPORT JOBS WHERE group_key = ...` queries in the same worker tick

## Already Implemented / Progress Sync

1. Add a batch query helper in `internal/model/import/import_into.go`.
   - Execute `SHOW IMPORT JOBS WHERE group_key = ...` once per import task.
   - Parse each row into a reusable status object with `Job_ID`, `Target_Table`, phase, status, imported rows, timestamps, and current-step progress fields.
   - Build lookup indexes keyed by `jobID` and `target_table`.
   - Verification:
     - `go test ./internal/model/import -run 'TestFindImportIntoJobStatusesFromResp'`
   - Success criteria:
     - Batch rows can be parsed into `TaskStatus`-compatible data and located by `jobID` / table.

2. Refactor importing progress sync to use the batch lookup.
   - Query once in `onImportIntoImporting()`.
   - Keep terminal subtasks skipped.
   - Keep `jobID == 0` as defensive skip only.
   - Treat missing batch rows for non-zero `jobID` as subtask-local sync errors, not whole-task failure.
   - Update the top-level importing log message to describe batched `SHOW IMPORT JOBS` sync.
   - Verification:
     - `go test ./internal/model/import -run 'TestRunImportIntoImportingProgressSyncWithBatchLookup'`
   - Success criteria:
     - Multiple active subtasks share one SQL fetch and later subtasks still run if one lookup fails.

3. Refactor canceling progress sync to use the same batch lookup.
   - Query the batch result once after cancel execution.
   - Preserve `jobID == 0 => canceled` for subtasks that have no pending async submit task.
   - Verification:
     - `go test ./internal/model/import -run 'TestRunImportIntoCancelingProgressSyncWithBatchLookup'`
   - Success criteria:
     - Canceling keeps existing zero-job semantics and uses one shared batch status fetch.

## Next Changes

1. Add batch submit/get/cancel agent APIs.
   - Add `BatchCreateImportIntoJobTasks`, `BatchGetImportIntoJobTasks`, and `BatchCancelImportIntoJobs` to `dataflow-agent-ng/idl/idl/agent/ng_agent/agent.proto`.
   - Use one encrypted `Cluster` per batch create/cancel request.
   - Keep batch submit items narrow: `task_id`, `sql`, and `setup_sqls`.
   - Return per-task submit/get responses and per-job cancel results.
   - Verification:
     - `make generate-idl`
   - Success criteria:
     - Generated protobuf/gateway artifacts expose the batch APIs without removing the compatible single-task APIs.

2. Implement batch APIs in the NG agent.
   - Validate all batch-create items before creating any async task.
   - Reuse the existing import-submit execution behavior: only `IMPORT INTO` SQL, async task manager, `Job_ID` parsing, and `bizerr` surfaced through polling `error_msg`.
   - Batch get should omit missing task IDs so dataflow-service can decide which deterministic tasks need creation.
   - Batch get should also omit task-local non-`bizerr` failures; dataflow-service keeps omitted pending-submit tasks waiting for the next batch loop instead of single-polling them immediately.
   - Batch cancel should execute `CANCEL IMPORT JOB <id>` for each requested job and return per-job `error_msg` instead of failing the whole batch for one TiDB business error.
   - Verification:
     - `go test ./dataflow-agent-ng/agent/grpc -run 'Test(BatchCreateImportIntoJobTasks|BatchGetImportIntoJobTasks|BatchCancelImportIntoJobs|CreateImportIntoJobTask|GetImportIntoJobTaskFailed)' -count=1`
   - Success criteria:
     - Batch submit, get, and cancel preserve existing single-operation semantics and expose per-item outcomes.
     - `BatchGetImportIntoJobTasks` omission may mean either not-found or task-local internal failure, and the caller handles that ambiguity without same-loop single-task polling.

3. Add regional client wrappers.
   - Add `BatchCreateImportIntoJobTasks`, `BatchGetImportIntoJobTasks`, and `BatchCancelImportIntoJobs` to `internal/handler/agent/regional-ng.go`.
   - Reuse existing PEM encryption and cluster request construction.
   - Assert request bodies do not contain plaintext passwords.
   - Verification:
     - `go test ./internal/handler/agent -run 'TestNGRegionalClientBatch(Create|Get|Cancel)ImportInto' -count=1`
   - Success criteria:
     - The service client hits expected paths, sends one encrypted cluster for batch create/cancel, and decodes per-item responses.

4. Refactor scaling-up submit orchestration.
   - Collect pending subtasks with `ImportIntoJobID == 0`.
   - Batch-get all subtasks marked `submit-task-pending`.
   - Leave omitted pending-submit tasks unresolved for the current loop.
   - Build SQL/setup SQLs and batch-create only deterministic submit tasks that have not yet entered the pending-submit phase.
   - Mark newly created subtasks pending and synthesize running responses for the current tick.
   - Apply existing per-subtask decisions unchanged: wait, success applies `job_id`, failure may run 8173 recovery, unrecoverable failure marks only that subtask failed.
   - Verification:
     - `go test ./internal/model/import -run 'Test(RunImportIntoScalingUp.*Batch|RecoverImportIntoJobIDWithExec|FindRecoverableImportIntoJobIDFailsClosed)' -count=1`
   - Success criteria:
     - Scaling up can submit and poll multiple subtasks with batch calls while preserving deterministic idempotency and recovery behavior.
     - `ScalingUp` does not issue same-loop recreate attempts for already-pending omitted submit tasks.

5. Refactor canceling orchestration.
   - Batch-get pending submit tasks before deciding that a zero-job subtask can be canceled.
   - If a pending submit task succeeded, apply its `job_id` and include it in batch cancel.
   - If a pending submit task is still running, keep the import in `Canceling` so a remote submit cannot create an uncanceled job after the service has marked the subtask canceled.
   - If a pending submit task is omitted from batch get, keep it pending for the next loop instead of rewriting it to `Canceled`.
   - Batch-cancel all non-terminal subtasks with non-zero `ImportIntoJobID`.
   - Reuse the existing batch progress lookup after cancel execution.
   - Verification:
     - `go test ./internal/model/import -run 'TestRunImportIntoCanceling.*Batch' -count=1`
   - Success criteria:
     - Canceling avoids per-job cancel SQL calls and does not prematurely cancel local state while remote submit tasks are still unresolved.
     - `Canceling` does not rewrite omitted pending submit tasks into `Canceled`.

6. Run required regression verification.
   - Verification:
     - `make app-unit-test-local`
   - Success criteria:
     - Quick app regression suite passes after the batch Import Into operation refactor.

7. Address review follow-up resilience regressions without widening API scope.
   - Restore batch progress parsing so an invalid `CurStepProgressPct` field falls back to `0` for that row while keeping the other status fields usable.
   - Skip the canceling batch progress query when all subtasks are already terminal or zero-job canceled after the first local canceling sync pass.
   - Keep batch polling batch-only: omitted submit-task results wait for the next loop, and same-loop fallback RPCs are not used.
   - Verification:
     - `go test ./internal/model/import -run 'Test(FindImportIntoJobStatusesFromRespFallsBackOnInvalidProgressPercent|ShouldSyncImportIntoCancelingProgressBatch|RunImportIntoScalingUpBatchWaitsOnOmittedPendingSubmitTask|RunImportIntoCancelingBatchWaitsOnOmittedPendingSubmitTask|ResolveImportIntoJobTaskStateAndApplyResult)' -count=1`
     - `go test ./dataflow-agent-ng/agent/grpc -run 'Test(BatchGetImportIntoJobTasks|GetImportIntoJobTaskFailed)' -count=1`
   - Success criteria:
     - One malformed progress field or one zero-job cancel state no longer blocks unrelated subtasks in the same import worker tick.
     - An omitted pending submit task is never recreated or canceled in the same loop solely because it was missing from the batch response.

## Risk Assessment

- Query-shape risk: `SHOW IMPORT JOBS WHERE group_key = ...` must expose the same progress columns needed by the existing `TaskStatus` parser. If not, keep batch discovery and fall back to per-job detail fetch.
- Matching risk: batch lookup must prefer `jobID` for active sync to avoid ambiguity when a table has historical jobs under the same `group_key`.
- Submit idempotency risk: batch create must preserve deterministic task IDs and `CreateIfAbsent` behavior to avoid duplicate detached imports.
- Cancel race risk: a subtask with `submit-task-pending` and no `job_id` cannot be marked canceled until the remote submit task resolves or a cancellable job ID is known.
- API scope risk: prefer domain-specific `BatchCancelImportIntoJobs` over a generic arbitrary-SQL batch API to avoid broadening the agent SQL surface.
- Review-fix risk: relaxing batch progress parsing must stay narrow to the known optional/derived `CurStepProgressPct` field so structural row corruption still fails closed.