Repository navigation
importinto: integrate async prepare flow for nextgen - #68668
Conversation
ti-chi-bot
Bot
added
the
release-note
|
@D3Hunter I've received your pull request and will start the review. I'll conduct a thorough review covering code quality, potential issues, and implementation details. ⏳ This process typically takes 10-30 minutes depending on the complexity of the changes. ℹ️ Learn more details on Pantheon AI. |
ti-chi-bot
Bot
added
the
size/XXL
|
Hi @D3Hunter. Thanks for your PR. PRs from untrusted users cannot be marked as trusted with I understand the commands that are listed here. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughImplements async prepare for IMPORT INTO: promotes framework prepare hook, gates prepare at submit, persists prepared-chunk metadata externally, scheduler OnPrepare prepares chunks and updates job/task, executor defers init and splits prechecks, adds job DB helpers and tests. ChangesAsync Prepare Phase for IMPORT INTO
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested labels
Suggested reviewers
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
D3Hunter
changed the title
D3Hunter
changed the title
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
tests/realtikvtest/importintotest/job_test.go (1)
658-662: 💤 Low valueConsider: Timeout multiplier may be excessive.
Line 662 uses
3*maxWaitTimewhile other tests in this file usemaxWaitTime. If this is intentional due to the non-prepare-mode path having different timing characteristics, a brief comment explaining why would be helpful. Otherwise, consider using the standardmaxWaitTimefor consistency.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/realtikvtest/importintotest/job_test.go` around lines 658 - 662, The test uses a longer timeout (3*maxWaitTime) in the s.Require().Eventually call for checking the import job status (jobID → importer.JobStatusFinished) which is inconsistent with other tests that use maxWaitTime; either revert this call to use maxWaitTime for consistency or, if the longer timeout is required for the non-prepare-mode path, add a short explanatory comment above the s.Require().Eventually line indicating why 3*maxWaitTime is necessary (e.g., non-prepare-mode tends to take ~3x longer) so future readers understand the intentional deviation.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pkg/dxf/importinto/scheduler_testkit_test.go`:
- Around line 199-201: The test is being skipped because the guard uses
importinto.ShouldUseAsyncPrepare(&logicalPlan.Plan) which returns false for the
local-sort plan built here, so change the guard or the plan: either replace the
skip guard with a check for kernel/runtime support (e.g., the runtime/kernel
feature flag used elsewhere in tests) so the subtest runs when the environment
supports async prepare, or modify the subtest to build a prepare-eligible plan
(make the logical plan non-local-sort or call the helper that produces an
async-prepare-capable plan) so ShouldUseAsyncPrepare(&logicalPlan.Plan) is true
and the preparing→importing assertions execute; update references to
ShouldUseAsyncPrepare and logicalPlan.Plan accordingly.
---
Nitpick comments:
In `@tests/realtikvtest/importintotest/job_test.go`:
- Around line 658-662: The test uses a longer timeout (3*maxWaitTime) in the
s.Require().Eventually call for checking the import job status (jobID →
importer.JobStatusFinished) which is inconsistent with other tests that use
maxWaitTime; either revert this call to use maxWaitTime for consistency or, if
the longer timeout is required for the non-prepare-mode path, add a short
explanatory comment above the s.Require().Eventually line indicating why
3*maxWaitTime is necessary (e.g., non-prepare-mode tends to take ~3x longer) so
future readers understand the intentional deviation.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 3c67e398-2467-451e-a264-956172cf616e
📒 Files selected for processing (22)
pkg/ddl/backfilling_dist_scheduler.gopkg/dxf/example/scheduler.gopkg/dxf/framework/scheduler/scheduler.gopkg/dxf/importinto/BUILD.bazelpkg/dxf/importinto/job.gopkg/dxf/importinto/job_testkit_test.gopkg/dxf/importinto/planner.gopkg/dxf/importinto/planner_test.gopkg/dxf/importinto/proto.gopkg/dxf/importinto/scheduler.gopkg/dxf/importinto/scheduler_test.gopkg/dxf/importinto/scheduler_testkit_test.gopkg/executor/import_into.gopkg/executor/importer/import.gopkg/executor/importer/job.gopkg/executor/importer/job_test.gopkg/executor/importer/precheck.gopkg/executor/importer/precheck_test.gopkg/executor/show_test.gopkg/ingestor/globalsort/util.gopkg/ingestor/globalsort/util_test.gotests/realtikvtest/importintotest/job_test.go
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #68668 +/- ##
================================================
+ Coverage 76.3025% 76.6602% +0.3576%
================================================
Files 2041 2054 +13
Lines 563407 577340 +13933
================================================
+ Hits 429894 442590 +12696
+ Misses 132597 132573 -24
- Partials 916 2177 +1261
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pkg/dxf/importinto/planner.go`:
- Around line 428-435: readPreparedChunkMap currently only decodes the new
PreparedMeta shape and will silently yield an empty ChunkMap for legacy JSON;
update readPreparedChunkMap to attempt backward-compatible decoding by first
trying to ReadJSONInto PreparedMeta (PreparedMeta.ReadJSONFromExternalStorage)
and if that produces an empty ChunkMap, re-read/unmarshal the same external JSON
into the legacy chunk-map-only shape (e.g., the previous ChunkMap type or a
map[string]... matching the old format) and populate
preparedChunkMapMeta.ChunkMap from that legacy structure so old exports still
produce import specs.
In `@pkg/dxf/importinto/scheduler_testkit_test.go`:
- Around line 216-218: The test currently defers server.Stop() before checking
error from fakestorage.NewServerWithOptions which can panic if server is nil;
change the order so you first assert success (e.g., call require.NoError(t, err)
or check err==nil) and only then call defer server.Stop(); locate the
NewServerWithOptions call and the server.Stop defer in scheduler_testkit_test.go
and move the defer to immediately after the successful creation check.
In `@pkg/executor/importer/job.go`:
- Around line 147-157: The IsSourceFileSizeUnknown method incorrectly returns
false once Status != JobStatusRunning which causes jobs that failed or were
cancelled during JobStepPreparing with SourceFileSize==0 to be treated as known
zero; change the predicate in JobInfo.IsSourceFileSizeUnknown to return true
when SourceFileSize == 0 AND (Status == jobStatusPending OR Step ==
JobStepPreparing) so that prepare-time failures/cancels (which preserve Step via
FailJob/CancelJob) still report size as unknown.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: c31477e3-76de-4957-bec9-c21adc45e1a1
📒 Files selected for processing (15)
pkg/dxf/importinto/BUILD.bazelpkg/dxf/importinto/job.gopkg/dxf/importinto/planner.gopkg/dxf/importinto/planner_test.gopkg/dxf/importinto/proto.gopkg/dxf/importinto/scheduler.gopkg/dxf/importinto/scheduler_testkit_test.gopkg/executor/importer/import.gopkg/executor/importer/importer_testkit_test.gopkg/executor/importer/job.gopkg/executor/importer/job_test.gopkg/executor/importer/table_import.gopkg/executor/show.gopkg/executor/show_test.gotests/realtikvtest/importintotest/job_test.go
✅ Files skipped from review due to trivial changes (1)
- pkg/executor/importer/table_import.go
🚧 Files skipped from review as they are similar to previous changes (5)
- pkg/dxf/importinto/BUILD.bazel
- pkg/executor/importer/import.go
- pkg/dxf/importinto/job.go
- pkg/dxf/importinto/scheduler.go
- tests/realtikvtest/importintotest/job_test.go
| func (j *JobInfo) IsSourceFileSizeUnknown() bool { | ||
| // if the job is not using async prepare, source_file_size is determined | ||
| // when creating the job, and it will never be 0 since we check total file | ||
| // size > 0 in precheck. | ||
| if j.SourceFileSize > 0 { | ||
| return false | ||
| } | ||
| if j.Status == jobStatusPending { | ||
| return true | ||
| } | ||
| return j.Status == JobStatusRunning && j.Step == JobStepPreparing |
There was a problem hiding this comment.
Keep source_file_size unknown after prepare-time failure/cancel.
This returns false as soon as the job leaves running, so a job that fails or is cancelled during JobStepPreparing with source_file_size = 0 will be reported as a real 0 instead of “unknown”. In this file, FailJob and CancelJob preserve step, so the safer predicate is “size is zero and the job is still pending or its step is preparing”.
Suggested fix
func (j *JobInfo) IsSourceFileSizeUnknown() bool {
// if the job is not using async prepare, source_file_size is determined
// when creating the job, and it will never be 0 since we check total file
// size > 0 in precheck.
if j.SourceFileSize > 0 {
return false
}
- if j.Status == jobStatusPending {
- return true
- }
- return j.Status == JobStatusRunning && j.Step == JobStepPreparing
+ return j.Status == jobStatusPending || j.Step == JobStepPreparing
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/executor/importer/job.go` around lines 147 - 157, The
IsSourceFileSizeUnknown method incorrectly returns false once Status !=
JobStatusRunning which causes jobs that failed or were cancelled during
JobStepPreparing with SourceFileSize==0 to be treated as known zero; change the
predicate in JobInfo.IsSourceFileSizeUnknown to return true when SourceFileSize
== 0 AND (Status == jobStatusPending OR Step == JobStepPreparing) so that
prepare-time failures/cancels (which preserve Step via FailJob/CancelJob) still
report size as unknown.
D3Hunter
left a comment
D3Hunter
left a comment
There was a problem hiding this comment.
Summary
- Total findings: 3
- Inline comments: 3
- Summary-only findings (no inline anchor): 0
Findings (highest risk first)
⚠️ [Major] (1)
- Async prepare re-discovers server-disk files on scheduler owner instead of using submit-time chunk map (
pkg/dxf/importinto/job.go:51; pkg/dxf/importinto/job.go:71; pkg/executor/import_into.go:109; pkg/dxf/importinto/scheduler.go:336; pkg/dxf/importinto/planner.go:372)
🟡 [Minor] (2)
- Plan keeps two option-presence registries without one canonical source (
pkg/executor/importer/import.go:324; pkg/executor/importer/import.go:325; pkg/executor/importer/import.go:767; pkg/executor/importer/import.go:768; pkg/executor/importer/import.go:1845) - Prepare-mode scheduler test hardcodes port and defers cleanup before setup assertion (
pkg/dxf/importinto/scheduler_testkit_test.go:207; pkg/dxf/importinto/scheduler_testkit_test.go:217)
|
/retest |
|
@D3Hunter: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
|
🔍 Starting code review for this PR... |
|
/retest |
|
@D3Hunter: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
|
/retest |
|
@D3Hunter: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
There was a problem hiding this comment.
🧹 Nitpick comments (1)
pkg/dxf/importinto/job.go (1)
96-107: ⚡ Quick winAvoid mutating caller-owned
planfor temporary async defaults.
plan.ThreadCntandplan.MaxNodeCntare set to1and never restored. Even if intended as temporary values for scheduling, this leaks state outsidedoSubmitTask. Use local variables forplanCtxinstead.Proposed fix
asyncPrepare := ShouldUseAsyncPrepare(plan) + threadCnt := plan.ThreadCnt + maxNodeCnt := plan.MaxNodeCnt if asyncPrepare { logicalPlan.PrepareMode = proto.PrepareModeRequired // below params will be filled later in async prepare, init to 1 temporarily. - plan.ThreadCnt = 1 - plan.MaxNodeCnt = 1 + threadCnt = 1 + maxNodeCnt = 1 } planCtx := planner.PlanCtx{ Ctx: ctx, TaskType: proto.ImportInto, - ThreadCnt: plan.ThreadCnt, - MaxNodeCnt: plan.MaxNodeCnt, + ThreadCnt: threadCnt, + MaxNodeCnt: maxNodeCnt, }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/dxf/importinto/job.go` around lines 96 - 107, The code mutates caller-owned plan by setting plan.ThreadCnt and plan.MaxNodeCnt to 1 for async prepare and never restores them; instead, compute local threadCnt/ maxNodeCnt variables (e.g., threadCnt := plan.ThreadCnt; maxNodeCnt := plan.MaxNodeCnt; if asyncPrepare { threadCnt = 1; maxNodeCnt = 1 }) and use those when constructing planner.PlanCtx (and leave plan.ThreadCnt/plan.MaxNodeCnt unchanged), while still setting logicalPlan.PrepareMode = proto.PrepareModeRequired when ShouldUseAsyncPrepare(...) is true.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@pkg/dxf/importinto/job.go`:
- Around line 96-107: The code mutates caller-owned plan by setting
plan.ThreadCnt and plan.MaxNodeCnt to 1 for async prepare and never restores
them; instead, compute local threadCnt/ maxNodeCnt variables (e.g., threadCnt :=
plan.ThreadCnt; maxNodeCnt := plan.MaxNodeCnt; if asyncPrepare { threadCnt = 1;
maxNodeCnt = 1 }) and use those when constructing planner.PlanCtx (and leave
plan.ThreadCnt/plan.MaxNodeCnt unchanged), while still setting
logicalPlan.PrepareMode = proto.PrepareModeRequired when
ShouldUseAsyncPrepare(...) is true.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: b5e82cc6-8b8b-4f31-a079-f453990af958
📒 Files selected for processing (1)
pkg/dxf/importinto/job.go
|
|
||
| // IsSourceFileSizeUnknown returns whether source_file_size has not been | ||
| // determined yet for async-prepare jobs. | ||
| func (j *JobInfo) IsSourceFileSizeUnknown() bool { |
There was a problem hiding this comment.
Can we just show N/A for zero file size, and show actual size for any positive file size, regardless of the job status?
There was a problem hiding this comment.
since the source file size = 0 precheck is moved to async prepare part, it can be 0 in this case
|
/retest |
|
@D3Hunter: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: GMHDBJD, joechenrh The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
ti-chi-bot
Bot
added
approved
lgtm
and removed
needs-1-more-lgtm
4 participants
What problem does this PR solve?
Issue Number: ref #68580
Problem Summary:
For nextgen
IMPORT INTOwith global sort, prepare work still blocks submit time and some prepare outputs are produced too late for consistent lifecycle tracking. We need an async prepare stage that persists prepared metadata and exposes the preparing period inSHOW IMPORT JOBS.What changed and how does it work?
IMPORT INTOtasks.SubmitTaskenables prepare mode for eligible plans and defers heavy prepare work from submit path to schedulerOnPrepare.OnPreparenow runs file discovery/resource calculation/chunk population, writes prepared chunk map to external storage, updates task meta, and persists prepared job info.prepared_meta_external_pathin task meta and corresponding prepared-meta helpers.JobStepPreparingand prepared-info update path (source_file_size+ detected format) before entering first business phase.CheckRequirementsBeforeInitDataFilesfor async-prepare submit path.SHOW IMPORT JOBS.preparingphase while task is still around init/prepare transition.SourceFileSizeasN/Awhen the size is still unknown during pending/preparing.Check List
Tests
before this PR, a 27.84TiB uncompressed CSV datasets takes 21s before submit
after this pr, job submit immediately, with
N/Asizeenter preparing
prepare finished, we can see the data size now
Unit test commands:
Side effects
Documentation
Release note
Please refer to Release Notes Language Style Guide to write a quality release note.
Summary by CodeRabbit
New Features
Improvements
Tests