# Premium Import Into Async Import SQL Implementation Plan

> Update 2026-04-20: async `IMPORT INTO` submission must use the same PEM-encrypted TiDB credential flow as synchronous `ExecSQL`. Plaintext `user/password` fields in the async task-create API are no longer acceptable and should be removed from the public contract.

## Analysis

### Impact Table

| Area | File | Change | Verification | Success Criteria |
|------|------|--------|--------------|------------------|
| NG agent async import-submit contract | `dataflow-agent-ng/idl/idl/agent/ng_agent/agent.proto` | Tighten `CreateImportIntoJobTaskReq` to the encrypted-cluster contract: add `cluster`, remove plaintext `user/password`, and reserve removed field ids/names | `make generate-idl` | Generated protobuf/gateway artifacts expose only the encrypted-cluster request shape and typed result |
| NG agent async execution | `dataflow-agent-ng/agent/grpc/services.go`, `dataflow-agent-ng/agent/grpc/server.go` | Reuse the existing cluster SQL execution path to run only `IMPORT INTO` statements through `importTaskManager`, with async task creation requiring encrypted `cluster` input and reusing `getClusterPassword` | `go test ./agent/grpc` in `dataflow-agent-ng` | Import job submission can be started once, polled repeatedly, and yields a stable job ID on success without rebuilding a plaintext credential path |
| NG regional client wrapper | `internal/handler/agent/regional-ng.go`, `internal/handler/agent/regional_ng_test.go` | Add `CreateImportIntoJobTask` / `GetImportIntoJobTask` helpers for the worker and reuse the same PEM-encryption helper used by sync paths | `go test ./internal/handler/agent -run '^(TestNGRegionalClientCreateImportIntoJobTask|TestNGRegionalClientGetImportIntoJobTask)$'` | The service client hits the expected endpoints, sends encrypted cluster credentials, and decodes the task result correctly |
| Premium import scaling orchestration | `internal/model/import/import_into.go`, `internal/model/import/import_into_prepare_test.go` | Replace synchronous `executeImportSQL()` submission with deterministic async task create/poll logic while leaving other SQL calls synchronous | `go test ./internal/model/import -run '^(TestBuildImportIntoSubmitTaskID|TestBuildImportIntoGetFilesTaskID|TestBuildImportIntoCreateSchemaTaskID)$'` | `ScalingUp` waits for submission tasks to finish before moving to `Importing`, and only then persists `ImportIntoJobID` |

### Dead Code Cascade

- The synchronous `executeImportSQL()` path that directly calls `executeSQL()` should be removed or reduced to building the SQL plus polling the async task result.
- `ScalingUp` must no longer transition to `Importing` while any pending subtask still lacks a real `ImportIntoJobID`; that old assumption becomes invalid once submission is async.
- Other synchronous SQL helpers such as empty-table checks, `SHOW IMPORT JOB`, and cancel SQL should stay untouched for this task.

### Symbols To Keep / Introduce / Remove

Keep:

- `executeSQL`
- `syncImportIntoProgress`
- `onImportIntoImporting`
- `onImportIntoCanceling`

Introduce:

- `CreateImportIntoJobTask`
- `GetImportIntoJobTask`
- `CreateImportIntoJobTaskStatus`
- `buildImportIntoSubmitTaskID`
- helper that builds the `IMPORT INTO` SQL without executing it synchronously

Remove or replace:

- direct synchronous submission inside `executeImportSQL`

### Scope Notes

- Only the detached `IMPORT INTO ...` submission becomes async.
- Empty-table validation before submission remains synchronous.
- Progress polling through `SHOW IMPORT JOB` remains synchronous.
- Cancellation remains synchronous and keeps using `CANCEL IMPORT JOB` once a job ID exists.

## Execution Steps

1. Tighten the async import-submit agent API to the encrypted-cluster contract.
   - Keep the existing create/get RPCs, request/response messages, and status enum dedicated to import job submission.
   - Add `cluster`, remove plaintext `user/password`, and reserve removed field ids/names.
   - Return a typed result containing the created `job_id`.
   - Verification:
     - `make generate-idl`
     - Expected: generated pb/gateway files update cleanly under `dataflow-agent-ng/idl/pbgen`.

2. Implement the async task in the NG agent.
   - Reuse the existing cluster SQL execution helper rather than inventing a second DB execution stack.
   - Require `req.Cluster` and pass it through directly when building `ExecSQLReq`.
   - Validate that only `IMPORT INTO` statements are accepted by the new task endpoint.
   - Parse `Job_ID` from the SQL response and expose it as the task result.
   - Verification:
     - `go test ./agent/grpc`
     - Expected: tests cover running, succeeded, failed, and invalid-SQL task states.

3. Add service-side regional client wrappers with shared PEM request construction.
   - Extend `internal/handler/agent/regional-ng.go` with create/get methods for the new task endpoint.
   - Reuse the same PEM fetch + password encryption helper used by sync `ExecSQL`-equivalent paths.
   - Keep signatures narrow so only import submission uses the new API.
   - Verification:
     - `go test ./internal/handler/agent -run '^(TestNGRegionalClientCreateImportIntoJobTask|TestNGRegionalClientGetImportIntoJobTask)$'`
     - Expected: tests assert path, method, request body, absence of plaintext password, and decoded job ID.

4. Convert `ScalingUp` job submission to async polling.
   - Add deterministic task ID generation per import subtask.
   - Keep empty-table checks synchronous.
   - For each pending subtask with no `ImportIntoJobID`, create or poll the async submission task.
   - Stay in `ScalingUp` while any submission task is still running.
   - Move to `Importing` only after every surviving pending subtask has a terminal result or a real `ImportIntoJobID`.
   - Verification:
     - `go test ./internal/model/import -run '^(TestBuildImportIntoSubmitTaskID|TestBuildImportIntoGetFilesTaskID|TestBuildImportIntoCreateSchemaTaskID)$'`
     - Expected: deterministic IDs are stable and the package compiles with the new orchestration.

5. Run focused verification and required regression checks.
   - `go test ./agent/grpc` in `dataflow-agent-ng`
   - `go test ./internal/handler/agent -run '^(TestNGRegionalClientCreateImportIntoJobTask|TestNGRegionalClientGetImportIntoJobTask|TestNGRegionalClientCreateSchemasTask|TestNGRegionalClientGetSchemasTask)$'`
   - `go test ./internal/model/import -run '^(TestBuildImportIntoSubmitTaskID|TestBuildImportIntoGetFilesTaskID|TestBuildImportIntoCreateSchemaTaskID|TestImportIntoFileDiscoveryStateHelpers)$'`
   - `make app-unit-test-local`
   - Expected: task-specific tests pass and the quick app regression suite still passes.

## Risks

- Task IDs must be derived from stable subtask identity; otherwise retries may create duplicate detached imports.
- Because detached import submission is now async, `ScalingUp` can span multiple worker ticks; any code assuming one-pass submission is now a bug.
- If the agent accepts non-`IMPORT INTO` SQL on this endpoint, the scope boundary is lost. Validation should be explicit.