fix iscp

[jiangxinmeng1]

User description

What type of PR is this?

• API-change
• BUG
• Improvement
• Documentation
• Feature
• Test and CI
• Code Refactoring

Which issue(s) this PR fixes:

issue #22437

What this PR does / why we need it:

get mo_iscp_log by table name
add UnregisterJobsByDBName

---

PR Type

Bug fix

---

Description

• Remove hardcoded table ID lookup for ISCP log table

• Replace with dynamic table name-based relation retrieval

• Fix transaction handling in ISCP log operations

• Improve error handling with table ID validation

---

Diagram Walkthrough

flowchart LR
  A["Cached Table ID"] -- "Remove" --> B["Dynamic Table Lookup"]
  B -- "Use" --> C["Table Name Resolution"]
  C -- "Create" --> D["Direct Relation Access"]

Loading

File Walkthrough

Relevant files

Bug fix

executor.go Remove cached table ID approach                                                    pkg/iscp/executor.go Remove setISCPLogTableID method and iscpLogTableID field Replace cached table ID with direct table name lookup in applyISCPLog Update transaction creation to use inline approach +21/-25  iteration.go Refactor transaction and table access logic                            pkg/iscp/iteration.go Move transaction creation logic inline in ExecuteIteration Get table name from job specs instead of relation Add table ID validation check for consistency Simplify getRelation function by removing transaction creation +27/-23  types.go Remove cached table ID field                                                          pkg/iscp/types.go Remove iscpLogTableID field from ISCPTaskExecutor struct +0/-1

---

[qodo-merge-pro]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🧪 No relevant tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Transaction Scope The new inline txn in applyISCPLog opens engine/txn before relation lookup and defers Commit, but there is no rollback on error paths; verify engine semantics ensure no partial state or leaked ops when errors occur between txn creation and commit. nowTs := exec.txnEngine.LatestLogtailAppliedTime() createByOpt := client.WithTxnCreateBy( 0, "", "iscp iteration", 0) txnOp, err := exec.cnTxnClient.New(ctx, nowTs, createByOpt) if err != nil { return } err = exec.txnEngine.New(ctx, txnOp) if err != nil { return } db, err := exec.txnEngine.Database(ctx, catalog.MO_CATALOG, txnOp) if err != nil { return } rel, err := db.Relation(ctx, MOISCPLogTableName, nil) if msg, injected := objectio.ISCPExecutorInjected(); injected && msg == "applyISCPLog" { err = moerr.NewInternalErrorNoCtx(msg) } if err != nil { return } defer txnOp.Commit(ctx) changes, err := CollectChanges(ctx, rel, from, to, exec.mp) Nil Access Risk Code reads jobSpecs[0] without guarding for empty slice from GetJobSpecs; add validation to avoid potential panic when no specs are returned. dbName := jobSpecs[0].ConsumerInfo.SrcTable.DBName tableName := jobSpecs[0].ConsumerInfo.SrcTable.TableName db, err := cnEngine.Database(ctx, dbName, txnOp) if err != nil { return } rel, err := db.Relation(ctx, tableName, nil) if err != nil { Error Context On table ID mismatch, a generic internal error is returned; consider adding db/table names and expected vs actual IDs to aid diagnostics. tableDef := rel.CopyTableDef(ctx) if rel.GetTableID(ctx) != iterCtx.tableID { return moerr.NewInternalErrorNoCtx("table id mismatch") }

[qodo-merge-pro]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
High-level	Avoid name-based, tenant-agnostic lookup ExecuteIteration now relies on jobSpecs[0] and name-based db/rel resolution under System_Account, which can panic on empty jobSpecs and fail to open user tables for non-system tenants. Prefer resolving the relation by table ID (cdc.GetRelationById) to derive tableDef/names, or set the txn/context tenant to iterCtx.accountID before name-based lookup, and validate jobSpecs is non-empty before indexing. This restores correctness across tenants and removes a crash risk. Examples: pkg/iscp/iteration.go [58-105] ctx = context.WithValue(ctx, defines.TenantIDKey{}, catalog.System_Account) ctx, cancel := context.WithTimeout(ctx, time.Hour) defer cancel() nowTs := cnEngine.LatestLogtailAppliedTime() createByOpt := client.WithTxnCreateBy( 0, "", "iscp iteration", 0) ... (clipped 38 lines) Solution Walkthrough: Before: // pkg/iscp/iteration.go func ExecuteIteration( ctx context.Context, ... iterCtx *IterationContext, ... ) (err error) { ctx = context.WithValue(ctx, defines.TenantIDKey{}, catalog.System_Account) // ... jobSpecs, err := GetJobSpecs(...) if err != nil { return } dbName := jobSpecs[0].ConsumerInfo.SrcTable.DBName tableName := jobSpecs[0].ConsumerInfo.SrcTable.TableName db, err := cnEngine.Database(ctx, dbName, txnOp) rel, err := db.Relation(ctx, tableName, nil) // ... } After: // pkg/iscp/iteration.go func ExecuteIteration( ctx context.Context, ... iterCtx *IterationContext, ... ) (err error) { ctx = context.WithValue(ctx, defines.TenantIDKey{}, iterCtx.accountID) // ... jobSpecs, err := GetJobSpecs(...) if err != nil { return } if len(jobSpecs) == 0 { return moerr.NewInternalError(ctx, "no job specs found for table %d", iterCtx.tableID) } // ... (use jobSpecs[0] to get names) // ... or preferably, use table ID to get relation directly _, _, rel, err = cdc.GetRelationById(ctx, cnEngine, txnOp, iterCtx.tableID) // ... } Suggestion importance[1-10]: 9 __ Why: The suggestion correctly identifies a critical regression where ExecuteIteration is hardcoded to the system tenant, breaking functionality for all other tenants, and also points out a potential panic from indexing an empty slice.	High
Possible issue	Add empty slice validation Add validation to ensure jobSpecs is not empty before accessing the first element. This prevents potential panic if no job specifications are found. pkg/iscp/iteration.go [92-93] +if len(jobSpecs) == 0 { + return moerr.NewInternalErrorNoCtx("no job specs found") +} dbName := jobSpecs[0].ConsumerInfo.SrcTable.DBName tableName := jobSpecs[0].ConsumerInfo.SrcTable.TableName Apply / Chat Suggestion importance[1-10]: 8 __ Why: This suggestion correctly identifies a potential panic if jobSpecs is an empty slice, and proposes adding a check to prevent an "index out of range" error, which enhances the robustness of the code.	Medium
General	Fix error handling order The injection check should occur after the relation retrieval error check to avoid overriding legitimate database errors with test injection errors. pkg/iscp/executor.go [453-459] rel, err := db.Relation(ctx, MOISCPLogTableName, nil) -if msg, injected := objectio.ISCPExecutorInjected(); injected && msg == "applyISCPLog" { - err = moerr.NewInternalErrorNoCtx(msg) -} if err != nil { return } +if msg, injected := objectio.ISCPExecutorInjected(); injected && msg == "applyISCPLog" { + err = moerr.NewInternalErrorNoCtx(msg) + return +} Apply / Chat Suggestion importance[1-10]: 6 __ Why: The suggestion correctly points out that a legitimate error from db.Relation could be masked by the test injection logic. Reordering the checks ensures that database errors are prioritized, improving error handling and debuggability.	Low
Update