diff --git a/docs/migrate.md b/docs/migrate.md index 81dc29a5..43fb4819 100644 --- a/docs/migrate.md +++ b/docs/migrate.md @@ -669,10 +669,10 @@ Note that the whole report is a single log record containing newlines. Spirit's | Field | Meaning | | --- | --- | -| `%` | Rows copied out of the estimated total. The total comes from table statistics, so the percentage can drift slightly and is not a row count you should reconcile against. | -| `n/m` | The figures the percentage is derived from. | +| `%` | Rows settled out of the estimated total. The total comes from table statistics, so the percentage is approximate and not a row count you should reconcile against. Rows the binlog applier wrote before the copy reached them are not counted (the copy inserts with `INSERT IGNORE`), so on a busy table the percentage finishes short of 100%. A resume from checkpoint continues the count from where the previous run left it. | +| `n/m` | The figures the percentage is derived from: rows settled so far and the estimated table cardinality, the same numbers the status API reports per table and as `Copy`. | | `chunk-size` | Rows in the most recently claimed chunk. The chunker sizes chunks dynamically to hit the [`--target-chunk-size`](#target-chunk-size) byte budget, so this number moving is normal and healthy — it is how Spirit adapts to row width. A chunk size that has collapsed to its floor and stayed there means the rows are too wide to fit the budget even at the floor, i.e. lower `--target-chunk-size` than the data wants, not a struggling server. | -| `eta` | Remaining rows divided by the recently measured copy rate. `TBD` for the first minute (no rate measured yet) and `DUE` past 99.99%. It is computed from a single 10-second sample, so early on it swings a lot; treat a large jump as noise unless it persists. | +| `eta` | Remaining copy divided by the recently measured copy rate. On a table with an auto_increment key the remaining copy is distance through the key range, so `DUE` (past 99.99% of the range) need not coincide with the `%` column when the ids are sparse; on other keys it is the same rows-against-estimate ratio as `%`, and the two move together. `TBD` for the first minute (no rate measured yet). It is computed from a single 10-second sample, so early on it swings a lot; treat a large jump as noise unless it persists. | | `throttled` | Whether the copy is currently paused by a throttler (replica lag, commit latency, or load). A migration that is throttled is behaving as designed — it is protecting the server, not stalling. | The `checksum` row that replaces this one during the checksum phase has the same shape — including its own `chunk-size`, since the checksum sizes chunks dynamically as well, though against a fixed 5s time budget rather than the byte budget — plus `threads=` and `throttled=` for the checksum's own pacing. diff --git a/pkg/copier/README.md b/pkg/copier/README.md index 25c12197..108f3201 100644 --- a/pkg/copier/README.md +++ b/pkg/copier/README.md @@ -36,11 +36,14 @@ The trade-offs are higher network transfer and CPU for serialization, which the type Copier interface { Run(ctx context.Context) error GetETA() string + GetETAState() status.ETA GetChunker() table.Chunker SetThrottler(throttler throttler.Throttler) GetThrottler() throttler.Throttler StartTime() time.Time GetProgress() string + CopyProgress() status.CopyProgress + ChunkSize() uint64 } ``` @@ -58,7 +61,9 @@ type ChunkCopier interface { - **`Run(ctx)`**: Starts the copy process and blocks until completion or error. Spawns multiple worker goroutines based on the configured concurrency level. - **`GetETA()`**: Returns estimated time to completion as a human-readable string. Returns "TBD" during the initial warmup period (1 minute), "DUE" when >99.99% complete, or a duration like "2h30m15s". -- **`GetProgress()`**: Returns progress as "copied/total percentage%" (e.g., "1000000/5000000 20.00%"). +- **`GetETAState()`**: The same estimate as a `status.ETA{State, Duration}`, for callers that branch on whether an estimate exists yet. `GetETA()` is its `String()`. +- **`CopyProgress()`**: Returns the copier's own progress as `status.CopyProgress{RowsCopied, RowsTotal}`. This is the measure the copier paces on: for the optimistic chunker it is keyspace distance against the auto_increment max, not a row count, so the runners report settled rows from the chunker instead (see `status.CopyFromTables`). `GetProgress()` is its rendered form, kept for interface compatibility; Spirit itself no longer calls it. +- **`ChunkSize()`**: Rows in the most recently claimed chunk, the `chunk-size=` field of the status log block. The chunker sizes chunks dynamically to hit the target byte budget, so this moves during a copy. - **`GetChunker()`**: Returns the underlying chunker for accessing detailed progress information. - **`SetThrottler(throttler)`**: Updates the throttler used to control copy rate. - **`GetThrottler()`**: Returns the current throttler. @@ -167,8 +172,10 @@ for { case <-ctx.Done(): return case <-ticker.C: - progress := copier.GetProgress() - eta := copier.GetETA() + // Settled rows against the row estimates, summed over the tables. + // CopyProgress() is the copier's own pacing measure, not a row count. + progress := status.CopyFromTables(status.TablesFromChunker(copier.GetChunker())) + eta := copier.GetETAState() fmt.Printf("Progress: %s, ETA: %s\n", progress, eta) } } diff --git a/pkg/copier/buffered.go b/pkg/copier/buffered.go index 535c03e3..c6444921 100644 --- a/pkg/copier/buffered.go +++ b/pkg/copier/buffered.go @@ -643,20 +643,9 @@ func (c *buffered) ChunkSize() uint64 { return c.chunkSize.Load() } +// GetETA renders GetETAState for the status block. func (c *buffered) GetETA() string { - c.Lock() - defer c.Unlock() - copiedRows, totalRows, pct := c.getCopyStats() - estimate, st := etaEstimate(copiedRows, totalRows, pct, c.rowsPerSecond.Load(), c.startTime) - switch st { - case status.ETADue: - return "DUE" - case status.ETAMeasuring: - return "TBD" - case status.ETAReady, status.ETANone: - // A ready estimate is formatted below; ETANone cannot occur during copy. - } - return estimate.String() + return c.GetETAState().String() } func (c *buffered) GetETAState() status.ETA { diff --git a/pkg/copier/copier.go b/pkg/copier/copier.go index b82fcd14..f957f9b7 100644 --- a/pkg/copier/copier.go +++ b/pkg/copier/copier.go @@ -65,9 +65,13 @@ type Copier interface { GetThrottler() throttler.Throttler StartTime() time.Time GetProgress() string - // CopyProgress returns the same progress as GetProgress in numeric form, - // which the status block needs in order to lay the percentage and the - // row counts out as separate fields. + // CopyProgress returns the copier's own measure of the copy in numeric + // form: the chunker's Progress, which for the optimistic chunker is + // keyspace distance against the auto_increment max rather than rows. It + // is the measure the ETA is paced on. Callers reporting rows to a human + // or a wrapper should sum the chunker's per-table settled counts instead + // (status.CopyFromTables), which is what the runners do. GetProgress is + // this value rendered. CopyProgress() status.CopyProgress // ChunkSize returns the row count of the most recently claimed chunk, or // 0 before the first one. This is the dynamic chunker's current sizing diff --git a/pkg/copier/copiertest/stub.go b/pkg/copier/copiertest/stub.go new file mode 100644 index 00000000..fd2ab3ec --- /dev/null +++ b/pkg/copier/copiertest/stub.go @@ -0,0 +1,29 @@ +// Package copiertest provides a Copier stub for tests of the runners that +// report copier state, so every runner package shares one definition. It +// lives beside the copier rather than in testutils because the copier's own +// tests import testutils, which a stub there would turn into an import cycle. +package copiertest + +import ( + "github.com/block/spirit/pkg/copier" + "github.com/block/spirit/pkg/status" + "github.com/block/spirit/pkg/throttler" +) + +// Stub answers the read-only status methods of copier.Copier from its +// fields, and reports a throttler that never throttles. Every other method is +// inherited from the embedded nil interface and panics if reached, which is +// the point: a test that needs it is exercising more than status reporting. +type Stub struct { + copier.Copier + ETA status.ETA + Copy status.CopyProgress + Chunk uint64 +} + +func (s Stub) GetETA() string { return s.ETA.String() } +func (s Stub) GetETAState() status.ETA { return s.ETA } +func (s Stub) GetProgress() string { return s.Copy.String() } +func (s Stub) CopyProgress() status.CopyProgress { return s.Copy } +func (s Stub) ChunkSize() uint64 { return s.Chunk } +func (s Stub) GetThrottler() throttler.Throttler { return &throttler.Noop{} } diff --git a/pkg/datasync/progress_test.go b/pkg/datasync/progress_test.go index 073b443b..55aa40e3 100644 --- a/pkg/datasync/progress_test.go +++ b/pkg/datasync/progress_test.go @@ -5,24 +5,12 @@ import ( "time" "github.com/block/spirit/pkg/applier" - "github.com/block/spirit/pkg/copier" + "github.com/block/spirit/pkg/copier/copiertest" "github.com/block/spirit/pkg/status" "github.com/block/spirit/pkg/table" "github.com/stretchr/testify/require" ) -type progressCopier struct{ copier.Copier } - -func (progressCopier) GetProgress() string { return "50%" } -func (progressCopier) GetETA() string { return "1m" } -func (progressCopier) GetETAState() status.ETA { - return status.ETA{State: status.ETAReady, Duration: time.Minute} -} -func (progressCopier) CopyProgress() status.CopyProgress { - return status.CopyProgress{RowsCopied: 50, RowsTotal: 100} -} -func (progressCopier) ChunkSize() uint64 { return 25 } - type progressApplier struct{ applier.Applier } func (progressApplier) Stats() applier.Stats { return applier.Stats{ActiveWorkers: 4} } @@ -31,21 +19,44 @@ func TestSyncProgressAndLogFormat(t *testing.T) { r, err := NewRunner(&Sync{}) require.NoError(t, err) require.Empty(t, r.Progress().ETA) - r.copyChunker = table.NewMultiChunker(table.NewMockChunker("b", 100), table.NewMockChunker("a", 200)) - r.copier = progressCopier{} - r.applier = progressApplier{} + b := table.NewMockChunker("b", 100) + a := table.NewMockChunker("a", 200) + b.Feedback(nil, 0, 30) // rows settled by the applier + a.Feedback(nil, 0, 40) + r.copyChunker = table.NewMultiChunker(b, a) + // With a chunker but no copier, the API and the log block agree: settled + // rows from the chunker, and an ETA that is not yet measured. r.status.Set(status.CopyRows) p := r.Progress() + require.Equal(t, status.CopyProgress{RowsCopied: 70, RowsTotal: 300}, p.Copy) + require.Equal(t, status.ETA{State: status.ETAMeasuring}, p.ETA) + require.Equal(t, "70/300 23.33% copyRows ETA TBD", p.Summary) + require.Contains(t, r.Status(), " 23.33% 70/300 chunk-size=0 eta=TBD") + r.copier = copiertest.Stub{ + ETA: status.ETA{State: status.ETAReady, Duration: time.Minute}, + // The copier's own measure, which neither Progress nor Status may report. + Copy: status.CopyProgress{RowsCopied: 7, RowsTotal: 9}, + Chunk: 25, + } + r.applier = progressApplier{} + r.status.Set(status.CopyRows) + p = r.Progress() require.Equal(t, status.ETA{State: status.ETAReady, Duration: time.Minute}, p.ETA) + require.Equal(t, status.CopyProgress{RowsCopied: 70, RowsTotal: 300}, p.Copy) // Both counters summed across Tables. + require.Equal(t, "70/300 23.33% copyRows ETA 1m0s", p.Summary) require.Len(t, p.Tables, 2) require.Less(t, p.Tables[0].TableName, p.Tables[1].TableName) block := r.Status() for _, text := range []string{"copier-time=", "\n copier", "\n applier", "\n binlog", "\n ckpt"} { require.Contains(t, block, text) } + // The log block reports the same copy measure as the API, on the same tick. + require.Contains(t, block, " 23.33% 70/300 chunk-size=25 eta=1m0s") + require.NotContains(t, block, "7/9") r.status.Set(status.ApplyChangeset) require.Empty(t, r.Progress().ETA) - require.Empty(t, r.Progress().Checksum) // The continuous verifier has no finite initial-checksum phase. + require.Equal(t, status.CopyProgress{RowsCopied: 70, RowsTotal: 300}, r.Progress().Copy) // The copy reading outlives the copy phase. + require.Empty(t, r.Progress().Checksum) // The continuous verifier has no finite initial-checksum phase. r.status.Set(status.RestoreSecondaryIndexes) block = r.Status() require.Contains(t, block, "state-time=") diff --git a/pkg/datasync/runner.go b/pkg/datasync/runner.go index edc4daf7..220c8075 100644 --- a/pkg/datasync/runner.go +++ b/pkg/datasync/runner.go @@ -70,10 +70,11 @@ type Runner struct { sourceTables []*table.TableInfo - applier applier.Applier - replClient change.Source - copyChunker table.Chunker - copier copier.Copier + applier applier.Applier + replClient change.Source + copyChunker table.Chunker + copyRowsAtResume uint64 // settled rows restored from the checkpoint, excluded from this invocation's copy aggregate + copier copier.Copier // resuming is set when a checkpoint was found on the target: the // initial copy is skipped and the change feed is opened from the @@ -183,16 +184,17 @@ func NewRunner(s *Sync) (*Runner, error) { } // recordCopyCompleted reports the copy aggregate settled during this -// Runner.Run invocation. The optimistic chunker does not persist its -// actual-row counter in a checkpoint, so a resumed invocation reports only -// work settled after it resumed. +// Runner.Run invocation. The chunker restores its settled count from the +// checkpoint so that progress continues across a resume; that restored count +// is subtracted here, so a resumed invocation reports only the rows settled +// after it resumed, alongside the chunks it copied. func (r *Runner) recordCopyCompleted() { chunker := r.copier.GetChunker() if chunker == nil { return } _, chunks, _ := chunker.Progress() - r.status.RecordCopyCompleted(chunker.RowsCopied(), chunks) + r.status.RecordCopyCompleted(chunker.RowsCopied()-r.copyRowsAtResume, chunks) } func (r *Runner) runCopy(ctx context.Context) error { @@ -1228,6 +1230,7 @@ func (r *Runner) startResume(ctx context.Context, watermark, pos string) error { if err := r.copyChunker.OpenAtWatermark(watermark); err != nil { return fmt.Errorf("failed to open copier at checkpoint watermark: %w", err) } + r.copyRowsAtResume = r.copyChunker.RowsCopied() } else { if err := r.copyChunker.Open(); err != nil { return err @@ -1599,16 +1602,29 @@ func (r *Runner) Progress() status.Progress { repl := r.replClient r.progMu.RUnlock() + tables := status.TablesFromChunker(chunker) + // The runner-wide copy is the sum of the per-table rows, so it reconciles + // with Tables and keeps its final reading once the copy has finished. + // Status derives its copier row the same way, so the API and the log + // block report one measure. The copier's own progress is not used for + // either: on an auto_increment key that measures keyspace distance, not + // rows. + copyProgress := status.CopyFromTables(tables) + var summary string var eta status.ETA switch state { //nolint:exhaustive // sync does not reach the cutover/checksum states case status.CopyRows: + // The copy phase is entered only after the pipeline is built, so the + // copier is normally present; without one the estimate is not yet + // measured, the same reading Status gives. + eta = status.ETA{State: status.ETAMeasuring} if cp != nil { - summary = fmt.Sprintf("%s copyRows ETA %s", cp.GetProgress(), cp.GetETA()) + // One copier read, so the ETA in Summary and the ETA field + // describe the same instant. eta = cp.GetETAState() - } else { - summary = "copyRows" } + summary = fmt.Sprintf("%s copyRows ETA %s", copyProgress.String(), eta.String()) case status.ApplyChangeset: if repl != nil { summary = fmt.Sprintf("continuous sync position=%s pending-changes=%d", repl.Position(), repl.GetDeltaLen()) @@ -1619,14 +1635,13 @@ func (r *Runner) Progress() status.Progress { summary = state.String() } - tables := status.TablesFromChunker(chunker) - return status.Progress{ CurrentState: state, Summary: summary, Resume: r.resuming.Load(), Tables: tables, ETA: eta, + Copy: copyProgress, // Throttle is deliberately left zero: a sync copies through a Noop // throttler, so there is nothing to report yet. } @@ -1640,6 +1655,7 @@ func (r *Runner) Status() string { r.progMu.RLock() cp := r.copier + chunker := r.copyChunker repl := r.replClient appl := r.applier r.progMu.RUnlock() @@ -1652,10 +1668,21 @@ func (r *Runner) Status() string { switch state { //nolint:exhaustive // sync does not reach the cutover/checksum states case status.CopyRows: b := status.NewBlock("sync status: state=%s total-time=%s copier-time=%s", state.String(), elapsed, r.status.Elapsed().Round(time.Second)) - // The copy pipeline is built asynchronously, so a status tick can land - // before there is a copier to report on. - if cp != nil { - progress := cp.CopyProgress() + // The chunker and the copier are published separately while the + // pipeline is built, and the copy phase is entered only once both + // exist, so these guards are defensive. The figures are settled rows + // from the chunker, the same measure Progress reports rather than the + // copier's own keyspace position; chunk-size and the ETA come from + // the copier and read as nothing claimed and nothing measured without + // one, as Progress does. + if chunker != nil { + progress := status.CopyFromTables(status.TablesFromChunker(chunker)) + var chunkSize uint64 + eta := status.ETA{State: status.ETAMeasuring} + if cp != nil { + chunkSize = cp.ChunkSize() + eta = cp.GetETAState() + } // No throttled= here, unlike migrate and move: a sync copies // through a Noop throttler, so the field would be a constant // false. @@ -1663,8 +1690,8 @@ func (r *Runner) Status() string { progress.Fraction()*100, progress.RowsCopied, progress.RowsTotal, - cp.ChunkSize(), - cp.GetETA(), + chunkSize, + eta.String(), ) } b.Row("applier", "%s", applier.StatusRow(appl)) diff --git a/pkg/migration/binlog_test.go b/pkg/migration/binlog_test.go index 155b213a..eed75678 100644 --- a/pkg/migration/binlog_test.go +++ b/pkg/migration/binlog_test.go @@ -162,7 +162,7 @@ func TestE2EBinlogSubscribingCompositeKey(t *testing.T) { require.NotNil(t, chunk) require.Equal(t, "((`id1` < 1001)\n OR (`id1` = 1001 AND `id2` < 1))", chunk.String()) require.NoError(t, ccopier.CopyChunk(t.Context(), chunk)) - require.Equal(t, status.Progress{CurrentState: status.CopyRows, Summary: "1000/1200 83.33% copyRows ETA TBD", ETA: status.ETA{State: status.ETAMeasuring}, Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1000, RowsTotal: 1200, IsComplete: false}}}, m.Progress()) + require.Equal(t, status.Progress{CurrentState: status.CopyRows, Summary: "1000/1200 83.33% copyRows ETA TBD", ETA: status.ETA{State: status.ETAMeasuring}, Copy: status.CopyProgress{RowsCopied: 1000, RowsTotal: 1200}, Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1000, RowsTotal: 1200, IsComplete: false}}}, m.Progress()) // Now insert some data. testutils.RunSQL(t, `insert into e2et1 (id1, id2) values (1002, 2)`) @@ -177,7 +177,7 @@ func TestE2EBinlogSubscribingCompositeKey(t *testing.T) { require.NoError(t, err) require.Equal(t, "((`id1` > 1001)\n OR (`id1` = 1001 AND `id2` >= 1))", chunk.String()) require.NoError(t, ccopier.CopyChunk(t.Context(), chunk)) - require.Equal(t, status.Progress{CurrentState: status.CopyRows, Summary: "1201/1200 100.08% copyRows ETA DUE", ETA: status.ETA{State: status.ETADue}, Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1201, RowsTotal: 1200, IsComplete: true}}}, m.Progress()) + require.Equal(t, status.Progress{CurrentState: status.CopyRows, Summary: "1201/1200 100.08% copyRows ETA DUE", ETA: status.ETA{State: status.ETADue}, Copy: status.CopyProgress{RowsCopied: 1201, RowsTotal: 1200}, Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1201, RowsTotal: 1200, IsComplete: true}}}, m.Progress()) // Now insert some data. // This should be picked up by the binlog subscription @@ -207,7 +207,8 @@ func TestE2EBinlogSubscribingCompositeKey(t *testing.T) { m.dbConfig = dbconn.NewDBConfig() require.NoError(t, m.checksum(t.Context())) require.Equal(t, "postChecksum", m.status.Get().String()) - require.Equal(t, status.Progress{CurrentState: status.PostChecksum, Summary: "Applying Changeset Deltas=0", Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1201, RowsTotal: 1200, IsComplete: true}}}, m.Progress()) + // The copy reading outlives the copy phase. + require.Equal(t, status.Progress{CurrentState: status.PostChecksum, Summary: "Applying Changeset Deltas=0", Copy: status.CopyProgress{RowsCopied: 1201, RowsTotal: 1200}, Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1201, RowsTotal: 1200, IsComplete: true}}}, m.Progress()) // All done! require.Equal(t, 0, m.db.Stats().InUse) // all connections are returned. diff --git a/pkg/migration/progress_copy_test.go b/pkg/migration/progress_copy_test.go new file mode 100644 index 00000000..3416ee3f --- /dev/null +++ b/pkg/migration/progress_copy_test.go @@ -0,0 +1,84 @@ +package migration + +import ( + "fmt" + "testing" + + "github.com/block/spirit/pkg/copier" + "github.com/block/spirit/pkg/dbconn" + "github.com/block/spirit/pkg/status" + "github.com/block/spirit/pkg/table" + "github.com/block/spirit/pkg/testutils" + "github.com/block/spirit/pkg/utils" + "github.com/stretchr/testify/require" +) + +// TestProgressCopyReconcilesWithTablesOnAutoIncrementKey exercises Progress.Copy +// on the chunker Spirit selects for a single auto_increment key. That chunker +// paces itself on keyspace distance against the auto_increment max, which is +// what the copier's own progress reports, and a table whose ids are sparse +// makes that measure differ from the row count by orders of magnitude. Copy +// must follow the row-count path Tables uses, so a caller reading both in one +// snapshot sees one story, and the reading must survive leaving the copy phase. +func TestProgressCopyReconcilesWithTablesOnAutoIncrementKey(t *testing.T) { + testutils.NewTestTable(t, "copyprog", `CREATE TABLE copyprog ( + id INT NOT NULL AUTO_INCREMENT PRIMARY KEY, + pad INT NOT NULL DEFAULT 0)`) + // 500 contiguous ids, then one row far out so the auto_increment max + // dwarfs the row count. + testutils.RunSQL(t, `INSERT INTO copyprog (id) + WITH RECURSIVE seq (n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 500) + SELECT n FROM seq`) + testutils.RunSQL(t, `INSERT INTO copyprog (id) VALUES (1000000)`) + + m := NewTestRunner(t, "copyprog", "ENGINE=InnoDB") + defer utils.CloseAndLog(m) + m.status.Begin() + m.dbConfig = dbconn.NewDBConfig() + var err error + m.db, err = dbconn.New(testutils.DSN(), m.dbConfig) + require.NoError(t, err) + defer utils.CloseAndLog(m.db) + m.changes[0].table = table.NewTableInfo(m.db, m.migration.Database, m.changes[0].stmt.Table) + require.NoError(t, m.changes[0].table.SetInfo(t.Context())) + require.NoError(t, m.setup(t.Context())) + disableDynamicChunking(t, m.copyChunker) + m.status.Set(status.CopyRows) + + // The first chunk is the open lower bound below the minimum id and copies + // nothing; the second covers ids 1..1000 and so every contiguous row. + ccopier, ok := m.copier.(copier.ChunkCopier) + require.True(t, ok) + for range 2 { + chunk, nextErr := m.copyChunker.Next() + require.NoError(t, nextErr) + require.NoError(t, ccopier.CopyChunk(t.Context(), chunk)) + } + + p := m.Progress() + require.Len(t, p.Tables, 1) + require.EqualValues(t, 500, p.Copy.RowsCopied) + require.Equal(t, p.Tables[0].RowsCopied, p.Copy.RowsCopied) + require.Equal(t, p.Tables[0].RowsTotal, p.Copy.RowsTotal) + require.Less(t, p.Copy.RowsTotal, uint64(1000000), "the total is the row estimate, not the auto_increment max") + require.Equal(t, p.Copy.String()+" copyRows ETA TBD", p.Summary) + + // The copier's own measure is the one Copy must not be: keyspace distance + // against the auto_increment max. Its total is the highest id, and its + // numerator counts ids the table never had, so it overshoots the rows + // settled. + own := m.copier.CopyProgress() + require.EqualValues(t, 1000000, own.RowsTotal) + require.Equal(t, 2*m.copier.ChunkSize(), own.RowsCopied, "two chunks of the pinned size, counted as ids rather than rows") + require.Greater(t, own.RowsCopied, p.Copy.RowsCopied) + require.NotEqual(t, p.Copy, own) + + // The log block reports the same measure as the API on the same tick, + // percentage included. + block := m.Status() + require.Contains(t, block, fmt.Sprintf("%6.2f%% %d/%d", p.Copy.Fraction()*100, p.Copy.RowsCopied, p.Copy.RowsTotal)) + require.NotContains(t, block, fmt.Sprintf("%d/%d", own.RowsCopied, own.RowsTotal)) + + m.status.Set(status.WaitingOnSentinelTable) + require.Equal(t, p.Copy, m.Progress().Copy) +} diff --git a/pkg/migration/progress_test.go b/pkg/migration/progress_test.go index 4e9dd6c2..d563cb9a 100644 --- a/pkg/migration/progress_test.go +++ b/pkg/migration/progress_test.go @@ -3,18 +3,51 @@ package migration import ( "sync/atomic" "testing" + "time" + "github.com/block/spirit/pkg/copier/copiertest" "github.com/block/spirit/pkg/status" + "github.com/block/spirit/pkg/table" "github.com/block/spirit/pkg/testutils" "github.com/block/spirit/pkg/throttler" "github.com/stretchr/testify/require" ) -// The tests here use a minimal hand-constructed Runner. Progress() in the -// Initial state reads neither the copier nor the chunkers, and throttleStatus +// The tests here use a minimal hand-constructed Runner. Progress() reads the +// copier only during the copy, tolerates a missing chunker, and throttleStatus // reads nothing but the throttler, so the fields under test can be exercised // without a live migration. +// TestProgressReportsCopyAlongsideTables pins that the runner-wide copy is the +// sum of Tables: present as soon as the copy chunker exists, counting settled +// rows rather than the copier's own measure, kept through the later phases, +// and rendered into Summary from the same reading together with a single ETA +// read. +func TestProgressReportsCopyAlongsideTables(t *testing.T) { + r := &Runner{copier: copiertest.Stub{ + ETA: status.ETA{State: status.ETAReady, Duration: time.Minute}, + // The copier's own measure, which Copy must never report. + Copy: status.CopyProgress{RowsCopied: 7, RowsTotal: 9}, + }} + require.Empty(t, r.Progress().Copy) + + c := table.NewMockChunker("t1", 100) + r.copyChunker = c + require.Equal(t, status.CopyProgress{RowsTotal: 100}, r.Progress().Copy) + + c.Feedback(nil, 0, 40) // rows settled by the applier + r.status.Set(status.CopyRows) + p := r.Progress() + require.Equal(t, status.CopyProgress{RowsCopied: 40, RowsTotal: 100}, p.Copy) + require.Equal(t, status.ETA{State: status.ETAReady, Duration: time.Minute}, p.ETA) + require.Equal(t, "40/100 40.00% copyRows ETA 1m0s", p.Summary) + + r.status.Set(status.WaitingOnSentinelTable) + p = r.Progress() + require.Equal(t, status.CopyProgress{RowsCopied: 40, RowsTotal: 100}, p.Copy) + require.Empty(t, p.ETA) +} + func TestProgressReportsResume(t *testing.T) { // Resume exists so a wrapper can tell a recovering run from one that is // starting over — a resumed run walks the whole state machine again, so diff --git a/pkg/migration/resume_test.go b/pkg/migration/resume_test.go index 74ded65e..92d281aa 100644 --- a/pkg/migration/resume_test.go +++ b/pkg/migration/resume_test.go @@ -15,6 +15,7 @@ import ( "fmt" "log/slog" "strings" + "sync/atomic" "testing" "time" @@ -23,6 +24,7 @@ import ( "github.com/block/spirit/pkg/checkpoint" "github.com/block/spirit/pkg/copier" "github.com/block/spirit/pkg/dbconn" + "github.com/block/spirit/pkg/metrics" "github.com/block/spirit/pkg/status" "github.com/block/spirit/pkg/table" "github.com/block/spirit/pkg/testutils" @@ -92,7 +94,8 @@ func TestChangeIntToBigIntPKResumeFromChkPt(t *testing.T) { func TestCheckpoint(t *testing.T) { // This test manually steps through the migration process to verify // watermark, checkpoint dump, and restore behavior. - // It uses specific INSERT patterns that produce exactly 11040 rows. + // It seeds about eleven thousand rows with bulk INSERT ... SELECT, which + // leaves auto_increment gaps, so ids are not contiguous. // // It drives the copier's synchronous CopyChunk API (copier.ChunkCopier) // to complete chunks in a controlled order (2, 1, 3) and assert the @@ -159,9 +162,15 @@ func TestCheckpoint(t *testing.T) { require.Equal(t, "copyRows", r.status.Get().String()) // The status block: a header line, then one row per subsystem. chunk is 0 - // until the first chunk is claimed, and the bar is empty at 0%. + // until the first chunk is claimed, and the bar is empty at 0%. The copier + // row counts settled rows against the table's row estimate, which comes + // from table statistics, so it is read from the table rather than pinned. + estimatedRows := atomic.LoadUint64(&r.changes[0].table.EstimatedRows) + var actualRows uint64 + require.NoError(t, r.db.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM cpt1").Scan(&actualRows)) + require.InEpsilon(t, actualRows, estimatedRows, 0.2, "the row estimate must be in the neighbourhood of the true count") require.Contains(t, r.Status(), "migration status: state=copyRows total-time=") - require.Contains(t, r.Status(), "\n copier 0.00% 0/11040 chunk-size=0 eta=") + require.Contains(t, r.Status(), fmt.Sprintf("\n copier 0.00%% 0/%d chunk-size=0 eta=", estimatedRows)) // The rows the change feed and the checkpoint dumper used to log for // themselves, plus the applier pipeline snapshot. // No write worker has started yet, so the applier row is the idle one. Every @@ -200,17 +209,28 @@ func TestCheckpoint(t *testing.T) { require.NoError(t, ccopier.CopyChunk(t.Context(), chunk1)) require.NoError(t, ccopier.CopyChunk(t.Context(), chunk3)) + // The copier row counts the rows the three chunks settled. That is not + // three chunks' worth of ids: the first chunk is the open lower bound + // below the minimum id and copies nothing, and the bulk INSERT ... SELECT + // seed leaves auto_increment gaps, so the count is read from the new + // table rather than pinned. + var settled uint64 + require.NoError(t, r.db.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM _cpt1_new").Scan(&settled)) + require.Positive(t, settled) + wantCopier := fmt.Sprintf("\n copier %6.2f%% %d/%d chunk-size=1000 eta=", float64(settled)/float64(estimatedRows)*100, settled, estimatedRows) // The status update is asynchronous (the applier phones home after each // chunk completes), so poll until it reflects all three copied chunks. require.Eventually(t, func() bool { - return strings.Contains(r.Status(), "\n copier 27.17% 3000/11040 chunk-size=1000 eta=") - }, 10*time.Second, 50*time.Millisecond, "status never reached expected copy progress; last status: %s", r.Status()) + return strings.Contains(r.Status(), wantCopier) + }, 10*time.Second, 50*time.Millisecond, "status never reached expected copy progress; want %q in: %s", wantCopier, r.Status()) // The watermark should exist now, because migrateChunk() // gives feedback back to table. watermark, err := r.copyChunker.GetLowWatermark() require.NoError(t, err) - require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"2001\"],\"Inclusive\":false}}", watermark) + chunkJSON, checkpointed := copierWatermark(t, watermark) + require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"2001\"],\"Inclusive\":false}}", chunkJSON) + require.Equal(t, settled, checkpointed, "the checkpoint carries the settled row count") // Dump a checkpoint require.NoError(t, r.DumpCheckpoint(t.Context())) // Which the status block now reports in place of the checkpoint's own log @@ -234,6 +254,12 @@ func TestCheckpoint(t *testing.T) { // the watermark to this point so new watermarks "align" correctly. // So lets now call NextChunk to verify. + // Before the resumed run copies anything, the API and the log block + // report the copy where the checkpoint left it, not from zero. + r.status.Set(status.CopyRows) + require.Equal(t, settled, r.Progress().Copy.RowsCopied) + require.Contains(t, r.Status(), fmt.Sprintf(" %d/%d chunk-size=", settled, atomic.LoadUint64(&r.changes[0].table.EstimatedRows))) + ccopier, ok = r.copier.(copier.ChunkCopier) require.True(t, ok) @@ -245,9 +271,14 @@ func TestCheckpoint(t *testing.T) { // It's ideally not typical but you can still dump checkpoint from // a restored checkpoint state. We won't have advanced anywhere from // the last checkpoint because on restore, the LowerBound is taken. + // In a migration the new table keeps the rows a re-copied chunk carries, + // so re-copying it settles nothing new and the count does not double up + // across the resume. watermark, err = r.copyChunker.GetLowWatermark() require.NoError(t, err) - require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"2001\"],\"Inclusive\":false}}", watermark) + chunkJSON, checkpointed = copierWatermark(t, watermark) + require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"2001\"],\"Inclusive\":false}}", chunkJSON) + require.Equal(t, settled, checkpointed) // Dump a checkpoint require.NoError(t, r.DumpCheckpoint(t.Context())) @@ -260,7 +291,48 @@ func TestCheckpoint(t *testing.T) { watermark, err = r.copyChunker.GetLowWatermark() require.NoError(t, err) - require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"11001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"12001\"],\"Inclusive\":false}}", watermark) + chunkJSON, checkpointed = copierWatermark(t, watermark) + require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"11001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"12001\"],\"Inclusive\":false}}", chunkJSON) + require.Greater(t, checkpointed, settled, "rows copied after the resume add to the restored count") + + // The copy aggregate reported to the metrics sink is per invocation: the + // count restored from the checkpoint is excluded, and the chunks are the + // eleven this runner copied. + sink := ©AggregateSink{} + r.status.SetMetricsSink(sink, r.logger) + r.recordCopyCompleted() + require.Equal(t, r.copyChunker.RowsCopied()-settled, sink.rows) + require.Equal(t, uint64(11), sink.chunks) +} + +// copyAggregateSink records the copy aggregate the runner reports when the +// copy completes. +type copyAggregateSink struct { + rows, chunks uint64 +} + +func (s *copyAggregateSink) Send(_ context.Context, m *metrics.Metrics) error { + for _, v := range m.Values { + switch v.Name { + case metrics.CopyRowsCompletedMetricName: + s.rows = uint64(v.Value) + case metrics.CopyChunksCompletedMetricName: + s.chunks = uint64(v.Value) + } + } + return nil +} + +// copierWatermark decodes the copy chunker's checkpoint into the chunk +// position and the settled row count it carries. +func copierWatermark(t *testing.T, watermark string) (string, uint64) { + var envelope struct { + ChunkJSON string + RowsCopied uint64 + } + require.NoError(t, json.Unmarshal([]byte(watermark), &envelope)) + require.NotEmpty(t, envelope.ChunkJSON) + return envelope.ChunkJSON, envelope.RowsCopied } func TestCheckpointRestore(t *testing.T) { diff --git a/pkg/migration/runner.go b/pkg/migration/runner.go index 70ed7ebc..1265fd16 100644 --- a/pkg/migration/runner.go +++ b/pkg/migration/runner.go @@ -80,8 +80,9 @@ type Runner struct { // the throttler to report whether the migration is currently paused. throttlerMu sync.RWMutex - copier copier.Copier - copyChunker table.Chunker // the chunker for copying + copier copier.Copier + copyChunker table.Chunker // the chunker for copying + copyRowsAtResume uint64 // settled rows restored from the checkpoint, excluded from this invocation's copy aggregate // applier is the shared write layer used by both the copier (buffered // copy) and the replication client (binlog deltas). Kept on the runner @@ -274,16 +275,17 @@ func (r *Runner) attemptMySQLDDL(ctx context.Context) error { } // recordCopyCompleted reports the copy aggregate settled during this -// Runner.Run invocation. The optimistic chunker does not persist its -// actual-row counter in a checkpoint, so a resumed invocation reports only -// work settled after it resumed. +// Runner.Run invocation. The chunker restores its settled count from the +// checkpoint so that progress continues across a resume; that restored count +// is subtracted here, so a resumed invocation reports only the rows settled +// after it resumed, alongside the chunks it copied. func (r *Runner) recordCopyCompleted() { chunker := r.copier.GetChunker() if chunker == nil { return } _, chunks, _ := chunker.Progress() - r.status.RecordCopyCompleted(chunker.RowsCopied(), chunks) + r.status.RecordCopyCompleted(chunker.RowsCopied()-r.copyRowsAtResume, chunks) } func (r *Runner) runCopy(ctx context.Context) error { @@ -1447,22 +1449,39 @@ func (r *Runner) Result() status.WorkflowResult { } } +// copyTables snapshots the copy chunker and returns the per-table progress. +// Progress and Status both derive their copy figures from it, so the API and +// the log block report one measure: settled rows against the tables' +// cardinality estimates, kept past the end of the copy. The copier's own +// progress is not used for either, because on an auto_increment key it +// measures keyspace distance, not rows. The chunker is read under chunkerMu +// to synchronize with initChunkers(), which may be assigning it concurrently +// during setup. +func (r *Runner) copyTables() []status.TableProgress { + r.chunkerMu.RLock() + copyChunker := r.copyChunker + r.chunkerMu.RUnlock() + return status.TablesFromChunker(copyChunker) +} + func (r *Runner) Progress() status.Progress { // Read the state once: the phase-specific fields below (summary, ETA, // checksum, throttle) must all describe the same state, not whichever state // each happened to observe. state := r.status.Get() + + tables := r.copyTables() + copyProgress := status.CopyFromTables(tables) + var summary string var eta status.ETA var checksum status.ChecksumProgress switch state { //nolint: exhaustive case status.CopyRows: - summary = fmt.Sprintf("%v %s ETA %v", - r.copier.GetProgress(), - state.String(), - r.copier.GetETA(), - ) + // One copier read, so the ETA in Summary and the ETA field describe + // the same instant. eta = r.copier.GetETAState() + summary = fmt.Sprintf("%s %s ETA %s", copyProgress.String(), state.String(), eta.String()) case status.WaitingOnSentinelTable: summary = "Waiting on Sentinel Table" case status.ApplyChangeset, status.PostChecksum: @@ -1471,20 +1490,13 @@ func (r *Runner) Progress() status.Progress { checksum = r.checker.GetProgress() summary = "Checksum Progress=" + checksum.String() } - - // Get per-table progress if available (multi-table migrations). - // We hold chunkerMu to synchronize with initChunkers(), which - // may be assigning r.copyChunker concurrently during setup. - r.chunkerMu.RLock() - copyChunker := r.copyChunker - r.chunkerMu.RUnlock() - tables := status.TablesFromChunker(copyChunker) return status.Progress{ CurrentState: state, Summary: summary, Resume: r.usedResumeFromCheckpoint.Load(), Throttle: r.throttleStatus(state), ETA: eta, + Copy: copyProgress, Checksum: checksum, Tables: tables, } @@ -1676,6 +1688,7 @@ func (r *Runner) resumeFromCheckpoint(ctx context.Context) error { if err := r.copyChunker.OpenAtWatermark(copierWatermark); err != nil { return err } + r.copyRowsAtResume = r.copyChunker.RowsCopied() if checksumWatermark != "" { if err := r.checksumChunker.OpenAtWatermark(checksumWatermark); err != nil { @@ -1981,7 +1994,7 @@ func (r *Runner) Status() string { } switch state { //nolint: exhaustive case status.CopyRows: - progress := r.copier.CopyProgress() + progress := status.CopyFromTables(r.copyTables()) b := status.NewBlock("migration status: state=%s total-time=%s copier-time=%s", state.String(), r.status.TotalElapsed().Round(time.Second), diff --git a/pkg/move/progress_test.go b/pkg/move/progress_test.go index 8f5b95a5..c3636c10 100644 --- a/pkg/move/progress_test.go +++ b/pkg/move/progress_test.go @@ -8,21 +8,13 @@ import ( "github.com/block/mysql" "github.com/block/spirit/pkg/checksum" - "github.com/block/spirit/pkg/copier" + "github.com/block/spirit/pkg/copier/copiertest" "github.com/block/spirit/pkg/status" "github.com/block/spirit/pkg/table" "github.com/block/spirit/pkg/testutils" "github.com/stretchr/testify/require" ) -type progressCopier struct{ copier.Copier } - -func (progressCopier) GetProgress() string { return "50%" } -func (progressCopier) GetETA() string { return "1m" } -func (progressCopier) GetETAState() status.ETA { - return status.ETA{State: status.ETAReady, Duration: time.Minute} -} - type progressChecker struct{ checksum.Checker } func (progressChecker) GetProgress() status.ChecksumProgress { @@ -32,8 +24,11 @@ func (progressChecker) GetProgress() status.ChecksumProgress { func TestMoveProgress(t *testing.T) { r := &Runner{} require.Empty(t, r.Progress().Tables) + require.Empty(t, r.Progress().Copy) a := table.NewMockChunker("a", 100) b := table.NewMockChunker("b", 200) + a.Feedback(nil, 0, 50) // rows settled by the applier + b.Feedback(nil, 0, 20) r.copyChunker = table.NewMultiChunker(a, b) p := r.Progress() require.Len(t, p.Tables, 2) @@ -43,24 +38,38 @@ func TestMoveProgress(t *testing.T) { require.False(t, row.IsComplete) } require.EqualValues(t, 300, total) + require.Equal(t, status.CopyProgress{RowsCopied: 70, RowsTotal: 300}, p.Copy) // Both counters summed across Tables, before any state is set. r.copyChunker = a - require.Equal(t, []status.TableProgress{{TableName: "a", RowsTotal: 100}}, r.Progress().Tables) - r.copier = progressCopier{} + require.Equal(t, []status.TableProgress{{TableName: "a", RowsCopied: 50, RowsTotal: 100}}, r.Progress().Tables) + r.copier = copiertest.Stub{ + ETA: status.ETA{State: status.ETAReady, Duration: time.Minute}, + // The copier's own measure, which Copy must never report. + Copy: status.CopyProgress{RowsCopied: 7, RowsTotal: 9}, + } r.status.Set(status.CopyRows) p = r.Progress() require.Equal(t, status.ETA{State: status.ETAReady, Duration: time.Minute}, p.ETA) + require.Equal(t, status.CopyProgress{RowsCopied: 50, RowsTotal: 100}, p.Copy) + require.Equal(t, "50/100 50.00% copyRows ETA 1m0s", p.Summary) + // The log block reports the same copy measure as the API, percentage + // included, on the same tick. + block := r.Status() + require.Contains(t, block, " 50.00% 50/100 chunk-size=0 eta=1m0s throttled=false") + require.NotContains(t, block, "7/9") r.checker = progressChecker{} r.status.Set(status.Checksum) p = r.Progress() require.Equal(t, status.ChecksumProgress{RowsChecked: 25, RowsTotal: 100}, p.Checksum) require.Equal(t, "Checksum Progress="+p.Checksum.String(), p.Summary) require.Empty(t, p.ETA) + require.Equal(t, status.CopyProgress{RowsCopied: 50, RowsTotal: 100}, p.Copy) // The copy reading outlives the copy phase. r.usedResumeFromCheckpoint.Store(true) r.status.Set(status.WaitingOnSentinelTable) p = r.Progress() require.True(t, p.Resume) require.Equal(t, "Waiting on Sentinel Table", p.Summary) // No logging or target access. require.Empty(t, p.ETA) + require.Equal(t, status.CopyProgress{RowsCopied: 50, RowsTotal: 100}, p.Copy) require.Empty(t, p.Checksum) } diff --git a/pkg/move/runner.go b/pkg/move/runner.go index 93701ed3..9da5a7ec 100644 --- a/pkg/move/runner.go +++ b/pkg/move/runner.go @@ -111,6 +111,7 @@ type Runner struct { applier applier.Applier chunkerMu sync.RWMutex // Publishes copyChunker to concurrent Progress callers. copyChunker table.Chunker + copyRowsAtResume uint64 // settled rows restored from the checkpoint, excluded from this invocation's copy aggregate checksumChunker table.Chunker copier copier.Copier checker checksum.Checker @@ -236,16 +237,17 @@ func NewRunner(m *Move) (*Runner, error) { } // recordCopyCompleted reports the copy aggregate settled during this -// Runner.Run invocation. The optimistic chunker does not persist its -// actual-row counter in a checkpoint, so a resumed invocation reports only -// work settled after it resumed. +// Runner.Run invocation. The chunker restores its settled count from the +// checkpoint so that progress continues across a resume; that restored count +// is subtracted here, so a resumed invocation reports only the rows settled +// after it resumed, alongside the chunks it copied. func (r *Runner) recordCopyCompleted() { chunker := r.copier.GetChunker() if chunker == nil { return } _, chunks, _ := chunker.Progress() - r.status.RecordCopyCompleted(chunker.RowsCopied(), chunks) + r.status.RecordCopyCompleted(chunker.RowsCopied()-r.copyRowsAtResume, chunks) } func (r *Runner) runCopy(ctx context.Context) error { @@ -590,6 +592,7 @@ func (r *Runner) resumeFromCheckpoint(ctx context.Context) error { if err := r.copyChunker.OpenAtWatermark(copierWatermark); err != nil { return err } + r.copyRowsAtResume = r.copyChunker.RowsCopied() // Open each source's change feed at its checkpointed position. // OpenFromPosition primes the position and starts streaming in one call. @@ -1536,7 +1539,7 @@ func (r *Runner) Status() string { } switch state { //nolint:exhaustive case status.CopyRows: - progress := r.copier.CopyProgress() + progress := status.CopyFromTables(r.copyTables()) b := status.NewBlock("migration status: state=%s total-time=%s copier-time=%s", state.String(), r.status.TotalElapsed().Round(time.Second), @@ -1976,22 +1979,38 @@ func (r *Runner) SetReverseCutoverWithResult(fn CutoverResultCallback) { r.reverseCutoverResultFunc = fn } +// copyTables snapshots the published copy chunker and returns the per-table +// progress. Progress and Status both derive their copy figures from it, so +// the API and the log block report one measure: settled rows against the +// tables' cardinality estimates, kept past the end of the copy. The copier's +// own progress is not used for either, because on an auto_increment key it +// measures keyspace distance, not rows. The chunker is read under chunkerMu +// because setup and checkpoint resume may publish it while a caller polls. +func (r *Runner) copyTables() []status.TableProgress { + r.chunkerMu.RLock() + copyChunker := r.copyChunker + r.chunkerMu.RUnlock() + return status.TablesFromChunker(copyChunker) +} + func (r *Runner) Progress() status.Progress { // Read the state once: the phase-specific fields below (summary, ETA, // checksum, throttle) must all describe the same state, not whichever state // each happened to observe. state := r.status.Get() + + tables := r.copyTables() + copyProgress := status.CopyFromTables(tables) + var summary string var eta status.ETA var checksum status.ChecksumProgress switch state { //nolint: exhaustive case status.CopyRows: - summary = fmt.Sprintf("%v %s ETA %v", - r.copier.GetProgress(), - state.String(), - r.copier.GetETA(), - ) + // One copier read, so the ETA in Summary and the ETA field describe + // the same instant. eta = r.copier.GetETAState() + summary = fmt.Sprintf("%s %s ETA %s", copyProgress.String(), state.String(), eta.String()) case status.WaitingOnSentinelTable: summary = "Waiting on Sentinel Table" case status.ApplyChangeset, status.PostChecksum: @@ -2000,18 +2019,12 @@ func (r *Runner) Progress() status.Progress { checksum = r.checker.GetProgress() summary = "Checksum Progress=" + checksum.String() } - - // Get per-table progress from the published copy chunker. Setup and - // checkpoint resume may publish it while an API caller polls Progress. - r.chunkerMu.RLock() - copyChunker := r.copyChunker - r.chunkerMu.RUnlock() - tables := status.TablesFromChunker(copyChunker) return status.Progress{ CurrentState: state, Summary: summary, Resume: r.usedResumeFromCheckpoint.Load(), ETA: eta, + Copy: copyProgress, Checksum: checksum, Tables: tables, // Throttle is deliberately zero: move currently uses a Noop throttler. @@ -2371,8 +2384,8 @@ func (r *Runner) flushAllReplClients(ctx context.Context) error { func (r *Runner) deleteRecopyRange(ctx context.Context, copierWatermark string) error { // The checkpoint watermark format depends on how many chunkers the copy // chunker wraps: a single (source, table) pair stores that chunker's own - // watermark (raw chunk JSON for auto-inc PKs, or the composite chunker's - // envelope), while multiple pairs store a JSON map keyed by + // watermark (the chunk envelope, or a bare chunk from an older + // checkpoint), while multiple pairs store a JSON map keyed by // table.QualifiedName(). WatermarkPerTable normalizes every format into // a per-table map of raw chunk JSON. allTables := make([]*table.TableInfo, 0, len(r.sources)*len(r.sourceTables)) diff --git a/pkg/move/runner_test.go b/pkg/move/runner_test.go index 0c9be326..120a2642 100644 --- a/pkg/move/runner_test.go +++ b/pkg/move/runner_test.go @@ -471,16 +471,18 @@ func TestMoveResumeDeletesRecopyRange(t *testing.T) { checkpointAndStop(t, move) // Read back the copier watermark the checkpoint recorded. A single-table - // auto-inc move uses the optimistic chunker, whose watermark is the raw - // chunk JSON of the last contiguously-completed bounded chunk. + // auto-inc move uses the optimistic chunker, whose watermark carries the + // last contiguously-completed bounded chunk beside the settled row count. targetDB, err := sql.Open("block-mysql", targetDSN) require.NoError(t, err) defer utils.CloseAndLog(targetDB) var watermark string require.NoError(t, targetDB.QueryRowContext(t.Context(), "SELECT copier_watermark FROM "+checkpointTableName+" ORDER BY id DESC LIMIT 1").Scan(&watermark)) + var envelope struct{ ChunkJSON string } + require.NoError(t, json.Unmarshal([]byte(watermark), &envelope)) var chunk table.JSONChunk - require.NoError(t, json.Unmarshal([]byte(watermark), &chunk)) + require.NoError(t, json.Unmarshal([]byte(envelope.ChunkJSON), &chunk)) require.Len(t, chunk.LowerBound.Value, 1) lower, err := strconv.Atoi(chunk.LowerBound.Value[0]) require.NoError(t, err) diff --git a/pkg/status/README.md b/pkg/status/README.md index a9f51ae2..da649e84 100644 --- a/pkg/status/README.md +++ b/pkg/status/README.md @@ -36,7 +36,7 @@ Every runner passes its existing `metrics.Sink` to `Tracker`. Generic sinks rece The typed capability deliberately extends `metrics.Sink` rather than creating another observer mechanism. A runner with the default `metrics.NoopSink` disables transition delivery entirely and adds no transition allocations. Sink calls happen outside the tracker's timing mutex, their latency is excluded from phase duration, and a panic in a typed callback is recovered so telemetry cannot change migration behavior. -Copy totals count work settled during the current `Run` invocation and are emitted even when the copy attempt fails or is cancelled. The optimistic chunker does not persist its actual-row counter, so a resumed invocation reports only rows and chunks settled after resume. +Copy totals count work settled during the current `Run` invocation and are emitted even when the copy attempt fails or is cancelled. The chunker restores its settled count from the checkpoint so that `Progress` continues across a resume, and the runner subtracts that restored count from the aggregate, so a resumed invocation reports only the rows and chunks it settled itself. Durable mutation and physical ownership are correctness facts, not metrics. `migration.Runner.Result` and `move.Runner.Result` return `status.WorkflowResult` after `Run`; failures also preserve machine-checkable `status.ErrDurableMutation` and `status.ErrOwnershipAmbiguous` markers through `errors.Is`. Result-bearing forward and reverse cutover callbacks carry the same two independent facts, so a caller can report a confirmed partial write without inventing ownership ambiguity. @@ -69,7 +69,7 @@ migration status: state=copyRows total-time=2m6s copier-time=2m0s | Row | Source | Contents | | --- | --- | --- | -| `copier` | `copier.Copier` | Percentage and counts from `CopyProgress()`, then `chunk-size=` (rows in the most recently claimed chunk — the dynamic chunker's current sizing decision, previously visible only inside the checkpoint line's watermark JSON), the ETA, and whether a throttler is pausing the copy. | +| `copier` | chunker + `copier.Copier` | Percentage and counts of settled rows against the tables' cardinality estimates — the same figures `Progress().Copy` reports, not the copier's own keyspace measure — then `chunk-size=` (rows in the most recently claimed chunk — the dynamic chunker's current sizing decision, previously visible only inside the checkpoint line's watermark JSON), the ETA, and whether a throttler is pausing the copy. | | `applier` | `applier.Stats` | `queue=` is occupancy, not progress: at capacity is the healthy steady state for a copy, and a queue that empties means the pipeline has gone read-limited. See `pkg/applier/README.md` for which fields render and which appear only when they carry a diagnosis. | | `binlog` | runner + `change.FeedStats` | `deltas=` is the runner's unapplied-change count; the rest is the feed. `rotations=` replaces go-mysql's per-rotation `rotate to next binlog` line, which spirit now demotes to DEBUG, and `(n forced)` is the subset spirit caused itself by issuing `FLUSH BINARY LOGS` from `BlockWait` when the buffered position stalled. | | `ckpt` | `status.LastCheckpoint` | How long ago the checkpoint was persisted and the change-feed coordinate it saved — where a resumed run would restart reading. The pair is what answers whether that point is still within the source's binlog retention. `never` before the first checkpoint; a multi-source move renders `key=position` per source. | @@ -86,7 +86,13 @@ Two things the block gives up, deliberately: the whole report is one log record `Progress` is a struct (not just a string) containing the current state and a summary. It is designed as a struct specifically to allow future expansion for GUI wrappers and external tooling. -Alongside the summary it carries structured fields for the things a wrapper would otherwise have to parse out of prose or scrape from the logs: `ETA`, per-table `Tables` progress, `Checksum` progress, and — from [#844](https://github.com/block/spirit/issues/844) — `Resume` and `Throttle`. +Alongside the summary it carries structured fields for the things a wrapper would otherwise have to parse out of prose or scrape from the logs: `ETA`, the runner-wide `Copy` counts, per-table `Tables` progress, `Checksum` progress, and — from [#844](https://github.com/block/spirit/issues/844) — `Resume` and `Throttle`. + +### `Copy` + +`Copy` is the runner-wide row copy as a `CopyProgress{RowsCopied, RowsTotal}`. It is the sum of `Tables`, so the two always reconcile: both count settled rows against the tables' cardinality estimates, and neither is the optimistic chunker's keyspace position, which is what the copier paces on and what the ETA is derived from. It is populated as soon as the copy chunker exists and keeps its final reading through the later phases, so "how much did this run copy" stays answerable after the copy ends. + +Two caveats carry over from the per-table counts. `RowsCopied` counts rows the copy settled, so a row the binlog applier wrote before the copy reached it is not counted (the copy inserts with `INSERT IGNORE`, which reports it as unaffected), and on a busy table the copy finishes short of `RowsTotal`; a resume restores the count from the checkpoint and continues it, though a move deletes and re-copies the rows at or above the resume position, so the rows among them settled before the checkpoint are counted again. And on an auto_increment key the ETA's `DUE` is paced on the keyspace, so over a sparse key range `Copy` can read close to complete while the ETA is still counting down, or the reverse; `Summary` carries both halves. ### `Resume` @@ -128,7 +134,7 @@ A `move` reports no throttling at all — it copies through a `Noop` throttler f ### Structured runner progress -Migration, move and datasync use `TablesFromChunker` to return table progress in a stable order. Multi-source identifiers retain the source qualifier so equally named tables remain distinct. Copy ETA is populated during `CopyRows` and cleared afterwards. Migration and move also expose the finite initial checksum counts; datasync’s continuous verifier has no corresponding finite phase. +Migration, move and datasync use `TablesFromChunker` to return table progress in a stable order. Multi-source identifiers retain the source qualifier so equally named tables remain distinct. Copy ETA is populated during `CopyRows` and cleared afterwards; the `Copy` counts persist past it. Migration and move also expose the finite initial checksum counts; datasync’s continuous verifier has no corresponding finite phase. All three use multiline status blocks. Datasync includes `copier-time` while copying and `state-time` while restoring indexes, alongside its existing binlog and checkpoint rows. Sentinel progress polling returns a summary without emitting logs; periodic logging remains the responsibility of `WatchTask`. diff --git a/pkg/status/progress.go b/pkg/status/progress.go index 2e694fbb..ca381694 100644 --- a/pkg/status/progress.go +++ b/pkg/status/progress.go @@ -37,6 +37,21 @@ type ETA struct { Duration time.Duration } +// String renders the ETA the way the copy summary shows it: "TBD" while the +// rate is still being measured, "DUE" once the copy has reached its estimated +// end, and otherwise the remaining duration. +func (e ETA) String() string { + switch e.State { + case ETADue: + return "DUE" + case ETAMeasuring: + return "TBD" + case ETAReady, ETANone: + return e.Duration.String() + } + return e.Duration.String() +} + // ThrottleStatus reports whether the current phase is paused by a throttler, // and why. Before this, throttling was only visible in the logs, so a wrapper // polling status saw a migration that had gone quiet with no way to say why @@ -109,6 +124,28 @@ type Progress struct { // ETA is the structured remaining row-copy estimate and its availability. ETA ETA + // Copy is the runner-wide row copy: the sum of Tables, so the two reconcile + // by construction. Both count settled rows against the tables' cardinality + // estimates, never the chunker's keyspace position, which is what the + // copier itself paces on for an auto_increment key and what the ETA is + // derived from. Copy is zero until the copy chunker exists and keeps its + // final reading through the later phases, so a caller can read how much + // the run copied at any point. RowsTotal is an estimate, so RowsCopied can + // exceed it. + // + // Two consequences of that split follow. RowsCopied counts rows the copy + // settled, so a row the binlog applier wrote before the copy reached it + // is not counted (the copy inserts with INSERT IGNORE, which reports it + // as unaffected), and on a busy table the copy finishes short of + // RowsTotal. A resume restores the count from the checkpoint and + // continues it; a move deletes and re-copies the rows at or above the + // resume position, so the rows among them settled before the checkpoint + // are counted again. And the ETA, including its DUE state, is paced on the + // keyspace for an auto_increment key, so over a sparse key range the copy + // can read close to complete here while the ETA is still counting down, + // or the reverse. Summary carries both halves. + Copy CopyProgress + // Checksum is the structured progress of the post-copy checksum phase, // populated while CurrentState is Checksum and zero otherwise. It is the // structured form of the checksum progress embedded in Summary. diff --git a/pkg/status/progress_test.go b/pkg/status/progress_test.go index 52fc4458..2ce4f8f1 100644 --- a/pkg/status/progress_test.go +++ b/pkg/status/progress_test.go @@ -2,10 +2,33 @@ package status import ( "testing" + "time" "github.com/stretchr/testify/assert" ) +// ETA.String renders the estimate the way the copy summary and the copier's +// GetETA show it: the availability states as fixed words, otherwise the +// remaining duration. +func TestETAString(t *testing.T) { + tests := []struct { + name string + eta ETA + want string + }{ + {"none", ETA{}, "0s"}, + {"measuring", ETA{State: ETAMeasuring}, "TBD"}, + {"due", ETA{State: ETADue}, "DUE"}, + {"due ignores a leftover duration", ETA{State: ETADue, Duration: time.Minute}, "DUE"}, + {"ready", ETA{State: ETAReady, Duration: 90 * time.Second}, "1m30s"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, tt.eta.String()) + }) + } +} + // ChecksumProgress.String renders the verified/total rows and percentage shown // in the checksum summary line, e.g. "71436/221193 32.30%". func TestChecksumProgressString(t *testing.T) { diff --git a/pkg/status/tables.go b/pkg/status/tables.go index 234749c9..9ed59c8d 100644 --- a/pkg/status/tables.go +++ b/pkg/status/tables.go @@ -26,3 +26,16 @@ func TablesFromChunker(chunker table.Chunker) []TableProgress { slices.SortFunc(rows, func(a, b TableProgress) int { return strings.Compare(a.TableName, b.TableName) }) return rows } + +// CopyFromTables sums a per-table snapshot into the runner-wide copy progress. +// Deriving it from the same snapshot is what lets Progress.Copy and +// Progress.Tables reconcile: both count settled rows against the tables' +// cardinality estimates, whichever chunker is doing the copying. +func CopyFromTables(tables []TableProgress) CopyProgress { + var c CopyProgress + for _, t := range tables { + c.RowsCopied += t.RowsCopied + c.RowsTotal += t.RowsTotal + } + return c +} diff --git a/pkg/status/tables_test.go b/pkg/status/tables_test.go index 98e4757d..1d41a879 100644 --- a/pkg/status/tables_test.go +++ b/pkg/status/tables_test.go @@ -8,6 +8,19 @@ import ( "github.com/stretchr/testify/require" ) +// CopyFromTables sums both counters across every table, so a multi-table +// aggregate cannot quietly follow a single table. +func TestCopyFromTables(t *testing.T) { + require.Equal(t, CopyProgress{}, CopyFromTables(nil)) + require.Equal(t, CopyProgress{RowsCopied: 50, RowsTotal: 100}, + CopyFromTables([]TableProgress{{TableName: "a", RowsCopied: 50, RowsTotal: 100}})) + require.Equal(t, CopyProgress{RowsCopied: 271, RowsTotal: 350}, CopyFromTables([]TableProgress{ + {TableName: "a", RowsCopied: 50, RowsTotal: 100}, + {TableName: "b", RowsCopied: 20, RowsTotal: 200}, + {TableName: "c", RowsCopied: 201, RowsTotal: 50, IsComplete: true}, // settled rows may exceed the estimate + })) +} + func TestTablesFromChunker(t *testing.T) { require.Empty(t, TablesFromChunker(nil)) a := table.NewMockChunker("items", 100) diff --git a/pkg/table/chunk.go b/pkg/table/chunk.go index fa1fd7f0..1c7fe057 100644 --- a/pkg/table/chunk.go +++ b/pkg/table/chunk.go @@ -269,7 +269,7 @@ func newChunkFromJSON(ti *TableInfo, jsonStr string) (*Chunk, error) { } // Validate the shape before converting. encoding/json silently ignores // unknown keys, so JSON in a foreign format (such as the multi-chunker's - // per-table map or the composite chunker's {"ChunkJSON":...} envelope) + // per-table map or the {"ChunkJSON":...} watermark envelope) // would otherwise decode into a zero-value chunk and produce a nonsense // WHERE clause downstream. Fail loudly instead. Watermark chunks always // carry both bounds with one value per key column (see GetLowWatermark). @@ -340,8 +340,10 @@ func WatermarkRecopyClause(ti *TableInfo, watermarkJSON string) (string, error) // // - multiChunker (used for two or more chunkers): a JSON map keyed by // QualifiedName, where each value is the child chunker's own watermark. -// - chunkerComposite: an envelope {"ChunkJSON": "...", "RowsCopied": N}. -// - chunkerOptimistic: the raw chunk JSON itself. +// - chunkerComposite and chunkerOptimistic: an envelope +// {"ChunkJSON": "...", "RowsCopied": N}. Optimistic checkpoints written +// before the row count was recorded hold the raw chunk JSON itself, and +// are still accepted. // // The tables argument is required to attribute single-chunker watermarks // (which carry no table name) to their table: those formats are only produced @@ -360,7 +362,8 @@ func WatermarkPerTable(watermark string, tables ...*TableInfo) (map[string]strin if err := json.Unmarshal([]byte(watermark), &multi); err == nil { out := make(map[string]string, len(multi)) for key, wm := range multi { - out[key] = unwrapCompositeWatermark(wm) + chunkJSON, _ := unwrapWatermark(wm) + out[key] = chunkJSON } return out, nil } @@ -369,19 +372,30 @@ func WatermarkPerTable(watermark string, tables ...*TableInfo) (map[string]strin if len(tables) != 1 { return nil, fmt.Errorf("watermark is in single-table format but %d tables were supplied: %s", len(tables), truncateForError(watermark)) } + chunkJSON, _ := unwrapWatermark(watermark) return map[string]string{ - tables[0].QualifiedName(): unwrapCompositeWatermark(watermark), + tables[0].QualifiedName(): chunkJSON, }, nil } -// unwrapCompositeWatermark unwraps the composite chunker's watermark envelope -// ({"ChunkJSON": "...", "RowsCopied": N}) into the raw chunk JSON it carries. -// A watermark in any other format (e.g. the optimistic chunker's raw chunk -// JSON, which has no ChunkJSON field) is returned unchanged. -func unwrapCompositeWatermark(watermark string) string { - var envelope compositeWatermark +// watermarkEnvelope is the single-chunker watermark format: the chunk the +// copy resumes from and the rows settled when the checkpoint was written. The +// position alone cannot recover the settled count, on either chunker, so it +// travels with the position and the resumed run reports the copy where it +// left off. +type watermarkEnvelope struct { + ChunkJSON string + RowsCopied uint64 +} + +// unwrapWatermark splits a single-chunker watermark into the chunk JSON it +// carries and the settled row count recorded with it. A bare chunk JSON, +// which is what optimistic chunker checkpoints held before the count was +// recorded, is returned unchanged with a zero count. +func unwrapWatermark(watermark string) (string, uint64) { + var envelope watermarkEnvelope if err := json.Unmarshal([]byte(watermark), &envelope); err == nil && envelope.ChunkJSON != "" { - return envelope.ChunkJSON + return envelope.ChunkJSON, envelope.RowsCopied } - return watermark + return watermark, 0 } diff --git a/pkg/table/chunk_test.go b/pkg/table/chunk_test.go index c4674985..8d36ec8c 100644 --- a/pkg/table/chunk_test.go +++ b/pkg/table/chunk_test.go @@ -302,7 +302,8 @@ func TestWatermarkPerTable(t *testing.T) { rawChunk := `{"Key":["id"],"ChunkSize":1000,"LowerBound":{"Value":["50"],"Inclusive":true},"UpperBound":{"Value":["100"],"Inclusive":false}}` compositeEnvelope := `{"ChunkJSON":"{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\":[\"50\"],\"Inclusive\":true},\"UpperBound\":{\"Value\":[\"100\"],\"Inclusive\":false}}","RowsCopied":50}` - // Optimistic chunker raw chunk format (single table). + // Bare chunk format (single table), as optimistic chunker checkpoints held + // before the settled row count was recorded. wms, err := WatermarkPerTable(rawChunk, t1) require.NoError(t, err) require.Equal(t, map[string]string{"localhost:3306.test.t1": rawChunk}, wms) diff --git a/pkg/table/chunker.go b/pkg/table/chunker.go index 2e68e877..506678f1 100644 --- a/pkg/table/chunker.go +++ b/pkg/table/chunker.go @@ -71,9 +71,10 @@ type Chunker interface { // nothing like a row count. Use Progress to render a percentage, and this // to report how much data was copied. // - // A resumed run reports only what the chunker itself has seen unless the - // watermark carried an earlier count forward (the composite chunker's - // does; the optimistic chunker's watermark stores key positions only). + // The watermark carries the count beside the position, so a resumed run + // continues from the rows the previous run settled rather than from zero. + // A bare chunk watermark from before the count was recorded resumes at + // zero. RowsCopied() uint64 OpenAtWatermark(watermark string) error GetLowWatermark() (watermark string, err error) diff --git a/pkg/table/chunker_composite.go b/pkg/table/chunker_composite.go index 62e0070f..02201422 100644 --- a/pkg/table/chunker_composite.go +++ b/pkg/table/chunker_composite.go @@ -42,11 +42,6 @@ type chunkerComposite struct { logger *slog.Logger } -type compositeWatermark struct { - ChunkJSON string - RowsCopied uint64 -} - var _ MappedChunker = &chunkerComposite{} func (t *chunkerComposite) additionalConditionsSQL(whereSent bool) string { @@ -243,7 +238,7 @@ func (t *chunkerComposite) OpenAtWatermark(checkpnt string) error { t.Lock() defer t.Unlock() - var watermark compositeWatermark + var watermark watermarkEnvelope if err := json.Unmarshal([]byte(checkpnt), &watermark); err != nil { return fmt.Errorf("could not parse composite watermark: %w", err) } @@ -383,7 +378,7 @@ func (t *chunkerComposite) GetLowWatermark() (string, error) { if err != nil { return "", fmt.Errorf("could not serialize chunk watermark: %w", err) } - watermark := compositeWatermark{ + watermark := watermarkEnvelope{ ChunkJSON: chunkJSON, RowsCopied: atomic.LoadUint64(&t.rowsCopied), } diff --git a/pkg/table/chunker_composite_test.go b/pkg/table/chunker_composite_test.go index cdda1257..d248d509 100644 --- a/pkg/table/chunker_composite_test.go +++ b/pkg/table/chunker_composite_test.go @@ -260,7 +260,7 @@ func TestCompositeChunkerBinaryHexStringWatermark(t *testing.T) { require.NoError(t, err) // The boundary value must be serialized as a hex literal of the ASCII // string "0xAB" (0x30784142), not as the plain string "0xAB". - var compositeWM compositeWatermark + var compositeWM watermarkEnvelope require.NoError(t, json.Unmarshal([]byte(watermark), &compositeWM)) require.Contains(t, compositeWM.ChunkJSON, `0x30784142`) require.NotContains(t, compositeWM.ChunkJSON, `"0xAB"`) @@ -424,7 +424,7 @@ func TestCompositeLowWatermark(t *testing.T) { watermark, err := chunker.GetLowWatermark() require.NoError(t, err) // The watermark can be divided into the chunkJSON and the rows. - var compositeWM compositeWatermark + var compositeWM watermarkEnvelope require.NoError(t, json.Unmarshal([]byte(watermark), &compositeWM)) require.JSONEq(t, "{\"Key\":[\"pk\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1008\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"2032\"],\"Inclusive\":false}}", compositeWM.ChunkJSON) diff --git a/pkg/table/chunker_optimistic.go b/pkg/table/chunker_optimistic.go index 44200e55..9eebc67b 100644 --- a/pkg/table/chunker_optimistic.go +++ b/pkg/table/chunker_optimistic.go @@ -2,6 +2,7 @@ package table import ( "context" + "encoding/json" "errors" "fmt" "log/slog" @@ -337,7 +338,8 @@ func (t *chunkerOptimistic) OpenAtWatermark(cp string) error { } t.checkpointHighPtr = checkpointHighPtr } - chunk, err := newChunkFromJSON(t.Ti, cp) + chunkJSON, settled := unwrapWatermark(cp) + chunk, err := newChunkFromJSON(t.Ti, chunkJSON) if err != nil { return err } @@ -369,6 +371,10 @@ func (t *chunkerOptimistic) OpenAtWatermark(cp string) error { ptrVal -= minVal } t.rowsCopied = ptrVal + // Settled rows cannot be derived from the key space, so they are + // restored from the checkpoint. A watermark written before the count + // was recorded carries none, and the count restarts at zero. + t.actualRowsCopied.Store(settled) return nil } @@ -651,11 +657,21 @@ func (t *chunkerOptimistic) GetLowWatermark() (string, error) { return "", ErrWatermarkNotReady } - watermark, err := t.watermark.marshalJSON() + chunkJSON, err := t.watermark.marshalJSON() if err != nil { return "", fmt.Errorf("could not serialize watermark: %w", err) } - return watermark, nil + // The settled row count travels with the position so a resumed run + // reports the copy where it left off. The keyspace position needs no + // such record: OpenAtWatermark re-derives it from the chunk pointer. + watermark, err := json.Marshal(watermarkEnvelope{ + ChunkJSON: chunkJSON, + RowsCopied: t.actualRowsCopied.Load(), + }) + if err != nil { + return "", fmt.Errorf("could not serialize watermark envelope: %w", err) + } + return string(watermark), nil } func (t *chunkerOptimistic) open() (err error) { diff --git a/pkg/table/chunker_optimistic_test.go b/pkg/table/chunker_optimistic_test.go index 3042332f..fafb881a 100644 --- a/pkg/table/chunker_optimistic_test.go +++ b/pkg/table/chunker_optimistic_test.go @@ -118,7 +118,7 @@ func TestLowWatermark(t *testing.T) { chunker.Feedback(chunk, time.Second, 1) watermark, err := chunker.GetLowWatermark() require.NoError(t, err) - require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"1001\"],\"Inclusive\":false}}", watermark) + require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"1001\"],\"Inclusive\":false}}", watermarkChunkJSON(watermark)) // Check key w.r.t. watermark require.False(t, chunker.KeyAboveHighWatermark(1000)) @@ -135,7 +135,7 @@ func TestLowWatermark(t *testing.T) { require.True(t, chunker.KeyBelowLowWatermark(1001)) watermark, err = chunker.GetLowWatermark() require.NoError(t, err) - require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"2001\"],\"Inclusive\":false}}", watermark) + require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"2001\"],\"Inclusive\":false}}", watermarkChunkJSON(watermark)) chunkAsync1, err := chunker.Next() require.NoError(t, err) @@ -155,18 +155,18 @@ func TestLowWatermark(t *testing.T) { chunker.Feedback(chunkAsync2, time.Second, 1) watermark, err = chunker.GetLowWatermark() require.NoError(t, err) - require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"2001\"],\"Inclusive\":false}}", watermark) + require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"2001\"],\"Inclusive\":false}}", watermarkChunkJSON(watermark)) chunker.Feedback(chunkAsync3, time.Second, 1) watermark, err = chunker.GetLowWatermark() require.NoError(t, err) - require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"2001\"],\"Inclusive\":false}}", watermark) + require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"1001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"2001\"],\"Inclusive\":false}}", watermarkChunkJSON(watermark)) require.False(t, chunker.KeyBelowLowWatermark(2001)) chunker.Feedback(chunkAsync1, time.Second, 1) watermark, err = chunker.GetLowWatermark() require.NoError(t, err) - require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"4001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"5001\"],\"Inclusive\":false}}", watermark) + require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"4001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"5001\"],\"Inclusive\":false}}", watermarkChunkJSON(watermark)) require.True(t, chunker.KeyBelowLowWatermark(2001)) require.True(t, chunker.KeyBelowLowWatermark(5000)) @@ -175,12 +175,12 @@ func TestLowWatermark(t *testing.T) { require.Equal(t, "`id` >= 5001 AND `id` < 6001", chunk.String()) // should bump immediately watermark, err = chunker.GetLowWatermark() require.NoError(t, err) - require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"4001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"5001\"],\"Inclusive\":false}}", watermark) + require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"4001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"5001\"],\"Inclusive\":false}}", watermarkChunkJSON(watermark)) chunker.Feedback(chunk, time.Second, 1) watermark, err = chunker.GetLowWatermark() require.NoError(t, err) - require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"5001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"6001\"],\"Inclusive\":false}}", watermark) + require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":1000,\"LowerBound\":{\"Value\": [\"5001\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"6001\"],\"Inclusive\":false}}", watermarkChunkJSON(watermark)) // Test that we have applied all stored chunks and the map is empty, // as we gave Feedback for all chunks. @@ -333,7 +333,7 @@ func TestOptimisticDynamicChunking(t *testing.T) { watermark, err := chunker.GetLowWatermark() require.NoError(t, err) - require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":22,\"LowerBound\":{\"Value\": [\"584\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"606\"],\"Inclusive\":false}}", watermark) + require.JSONEq(t, "{\"Key\":[\"id\"],\"ChunkSize\":22,\"LowerBound\":{\"Value\": [\"584\"],\"Inclusive\":true},\"UpperBound\":{\"Value\": [\"606\"],\"Inclusive\":false}}", watermarkChunkJSON(watermark)) // Start everything over again as t2. t2 := newTableInfo4Test("test", "t1") @@ -351,6 +351,9 @@ func TestOptimisticDynamicChunking(t *testing.T) { chunker2, err := NewChunker(t2, ChunkerConfig{NewTable: t2, TargetChunkTime: 100}) require.NoError(t, err) require.NoError(t, chunker2.OpenAtWatermark(watermark)) + // The settled row count resumes from the checkpoint rather than from zero. + require.Positive(t, chunker.RowsCopied()) + require.Equal(t, chunker.RowsCopied(), chunker2.RowsCopied()) // The pointer goes to the lowerbound.value. // It could equally go to the upperbound.value but then @@ -394,6 +397,9 @@ func TestOptimisticResumeProgressAccounting(t *testing.T) { // the bogus ~53% before the fix). require.Equal(t, uint64(713192535-682769913), rowsCopied) require.Equal(t, uint64(1341021280), total) + // A bare chunk watermark predates the settled row count, so that count + // starts over. + require.Zero(t, chunker.RowsCopied()) // The reported percentage should be a few percent, nowhere near the ~53% // the bug produced. @@ -1010,3 +1016,10 @@ func TestOptimisticPrefetchDensityUsesSourceRows(t *testing.T) { } require.True(t, affectedOnly.sparse()) } + +// watermarkChunkJSON returns the chunk position an optimistic watermark +// carries, without the settled row count recorded beside it. +func watermarkChunkJSON(watermark string) string { + chunkJSON, _ := unwrapWatermark(watermark) + return chunkJSON +}