stage: updated the MarkApplied so that table updates happen in batches #1033
Conversation
| logical[idx] = mut.Time.Logical() | ||
| } | ||
|
|
||
| tag, err := db.Exec(ctx, s.sql.markApplied, keys, nanos, logical) |
There was a problem hiding this comment.
From my read of the code db can be a tx if we want extraSanityChecks == true. This is why I don't do any rollback or TX committing here.
However, in the ticket, @bobvawter did mention that you wanted the caller of this to pass in a TX and create it if it doesn't exist. To clarify, did you want to change this logic so that it's unconditionally a transaction instead of potentially separate db.Exec calls?
I understand that it is good to keep this a transaction now that we update in batches because we want to update these atomically, just like we previously were technically. If that's the case, I can just make it unconditionally a transaction and make the extraSanityChecks merely check consistency, if you don't see issues with that.
There was a problem hiding this comment.
Reviewable status: 0 of 1 files reviewed, 4 unresolved discussions (waiting on @Jeremyyang920 and @ryanluu12345)
internal/staging/stage/stage.go line 554 at r1 (raw file):
Previously, ryanluu12345 wrote…
Open questions here:
- What is a good sane default? I seem to understand that millions to couple of millions is when we can start hitting the limits of the SQL cluster setting. But not sure if this is too low.
- Should this be configurable? I defaulted not since this is a property of the staging logic which shouldn't be managed by the user.
That seems like a reasonable default. It's far larger than any interactive OLTP workload would tend to generate. This change is to address bulk transactions.
internal/staging/stage/stage.go line 592 at r1 (raw file):
} // Applies the mutations in batches to avoid exceeding the `sql.conn.max_read_buffer_message_size`
Reflow comments to 72 chars. The GoLand "Wrap to Column" plugin is my preferred tool.
internal/staging/stage/stage.go line 595 at r1 (raw file):
// This also reduces the memory being used during this step in the case there are millions or more rows. if err := batches.Window(markAppliedBatchSize, len(muts), func(begin, end int) error { keys := make([]json.RawMessage, len(muts))
These slices are going to be full of empty values if windowing happens. The length should be end-begin.
internal/staging/stage/stage.go line 604 at r1 (raw file):
Previously, ryanluu12345 wrote…
From my read of the code
dbcan be atxif we wantextraSanityChecks == true. This is why I don't do any rollback or TX committing here.However, in the ticket, @bobvawter did mention that you wanted the caller of this to pass in a TX and create it if it doesn't exist. To clarify, did you want to change this logic so that it's unconditionally a transaction instead of potentially separate db.Exec calls?
I understand that it is good to keep this a transaction now that we update in batches because we want to update these atomically, just like we previously were technically. If that's the case, I can just make it unconditionally a transaction and make the
extraSanityChecksmerely check consistency, if you don't see issues with that.
There's also
https://github.com/cockroachdb/replicator/blob/master/internal/sequencer/decorators/marker.go
and a handful of other places where a staging transaction may be created when multiple tables are in play. The thing to do here is to ensure there's a transaction if the number of marks exceeds the window size. We want to be able to use implicit transactions, unless there's a need to do something else.
ryanluu12345
force-pushed
the
user/rluu/batch-mark-apply
branch
from
2bde3b8
to
866b0a1
Compare
There was a problem hiding this comment.
Reviewable status: 0 of 1 files reviewed, 4 unresolved discussions (waiting on @bobvawter and @Jeremyyang920)
internal/staging/stage/stage.go line 554 at r1 (raw file):
Previously, bobvawter (Bob Vawter) wrote…
That seems like a reasonable default. It's far larger than any interactive OLTP workload would tend to generate. This change is to address bulk transactions.
Ok great. Then I'll go ahead and leave it at this default. If configuration is needed later, we can drill it in, but unnecessary for now.
internal/staging/stage/stage.go line 592 at r1 (raw file):
Previously, bobvawter (Bob Vawter) wrote…
Reflow comments to 72 chars. The GoLand "Wrap to Column" plugin is my preferred tool.
Noted, just installed Rewrap and will use that for my comment formatting. Note that I'll use crlfmt for my coding needs.
internal/staging/stage/stage.go line 595 at r1 (raw file):
Previously, bobvawter (Bob Vawter) wrote…
These slices are going to be full of empty values if windowing happens. The length should be
end-begin.
Great catch!
ryanluu12345
force-pushed
the
user/rluu/batch-mark-apply
branch
from
866b0a1
to
3eead5d
Compare
There was a problem hiding this comment.
Reviewable status: 0 of 1 files reviewed, 4 unresolved discussions (waiting on @bobvawter and @Jeremyyang920)
internal/staging/stage/stage.go line 604 at r1 (raw file):
Previously, bobvawter (Bob Vawter) wrote…
There's also
https://github.com/cockroachdb/replicator/blob/master/internal/sequencer/decorators/marker.go
and a handful of other places where a staging transaction may be created when multiple tables are in play. The thing to do here is to ensure there's a transaction if the number of marks exceeds the window size. We want to be able to use implicit transactions, unless there's a need to do something else.
Ok ended up fixing this! Please let me know your thoughts. I bundled this logic with the block above that creates a transaction for the extraSanityChecks. The logic here is that we need a new transaction in the two cases where we want to do consistency checks or we want to do multiple batches of applies but do so atomically.
There was a problem hiding this comment.
The implementation looks good, but it needs a test before merging.
Reviewed all commit messages.
Reviewable status: 0 of 1 files reviewed, 1 unresolved discussion (waiting on @Jeremyyang920 and @ryanluu12345)
internal/staging/stage/stage.go line 604 at r1 (raw file):
Previously, ryanluu12345 wrote…
Ok ended up fixing this! Please let me know your thoughts. I bundled this logic with the block above that creates a transaction for the extraSanityChecks. The logic here is that we need a new transaction in the two cases where we want to do consistency checks or we want to do multiple batches of applies but do so atomically.
That makes sense.
internal/staging/stage/stage.go line 619 at r3 (raw file):
} log.Debugf("MarkApplied: %s marked %d mutations", s.stage, tag.RowsAffected())
Drop this back to trace; it's going to be very spammy in production.
ryanluu12345
force-pushed
the
user/rluu/batch-mark-apply
branch
from
3eead5d
to
f9682f0
Compare
There was a problem hiding this comment.
My thought is to put another test case for func TestPutAndDrain(t \*testing.T) { where the batch size is set to a very small number (like 1) and verify that it properly marks applied for all the data. Does this sound reasonable or were you thinking a more integration or e2e test?
Reviewable status: 0 of 1 files reviewed, 1 unresolved discussion (waiting on @bobvawter and @Jeremyyang920)
internal/staging/stage/stage.go line 619 at r3 (raw file):
Previously, bobvawter (Bob Vawter) wrote…
Drop this back to trace; it's going to be very spammy in production.
Done.
ryanluu12345
force-pushed
the
user/rluu/batch-mark-apply
branch
from
f9682f0
to
b75fc01
Compare
There was a problem hiding this comment.
Finished the tests. PR should be ready!
Reviewable status: 0 of 2 files reviewed, 1 unresolved discussion (waiting on @bobvawter and @Jeremyyang920)
There was a problem hiding this comment.
Reviewed 2 of 2 files at r4, all commit messages.
Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @Jeremyyang920 and @ryanluu12345)
internal/staging/stage/stage.go line 560 at r4 (raw file):
} // These implementations are used for testing and help to set the mark applied
SetMarkAppliedBatchSize is used for testing ....
internal/staging/stage/stage.go line 566 at r4 (raw file):
} func (s *stage) GetMarkAppliedBatchSize() int {
This method is unused.
internal/staging/stage/stage.go line 583 at r4 (raw file):
ctx context.Context, db types.StagingQuerier, muts []types.Mutation, ) error { fmt.Println(s.markAppliedBatchSize)
Dangling log statement.
Integrate the batches.Window functionality from `util.Batches` to split up the work for MarkApplied in order to address concerns with sending payloads that exceed the sql.conn.max_read_buffer_message_size. The new logic handles the processing of the mutations inside of batches.Window and sets the default window size to 100,000. Resolves: #995 Release Note: None
ryanluu12345
force-pushed
the
user/rluu/batch-mark-apply
branch
from
b75fc01
to
e5937a3
Compare
There was a problem hiding this comment.
Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @bobvawter and @Jeremyyang920)
internal/staging/stage/stage.go line 560 at r4 (raw file):
Previously, bobvawter (Bob Vawter) wrote…
SetMarkAppliedBatchSize is used for testing ....
done
internal/staging/stage/stage.go line 566 at r4 (raw file):
Previously, bobvawter (Bob Vawter) wrote…
This method is unused.
done
internal/staging/stage/stage.go line 583 at r4 (raw file):
Previously, bobvawter (Bob Vawter) wrote…
Dangling log statement.
done
Integrate the batches.Window functionality from
util.Batchesto split up the work for MarkApplied in order to address concerns with sending payloads that exceed the sql.conn.max_read_buffer_message_size.The new logic handles the processing of the mutations inside of batches.Window and sets the default window size to 100,000.
Resolves: #995
Release Note: None
Testing
This change is