jobs: replace the ie in Registry.createJobsInBatchWithTxn #91679
Conversation
|
This change is |
ZhouXing19
force-pushed
the
create-job-with-txn
branch
7 times, most recently
from
97c175c
to
82f2c6e
Compare
ZhouXing19
requested review from
a team and
rhu713
and removed request for
a team
pkg/jobs/adopt.go
Outdated
| // If the job has failed, and the dump mode is set to anything | ||
| // except noDump, then we should dump the trace. | ||
| // The string comparison is unfortunate but is used to differentiate a job | ||
| // that has failed from a job that has been canceled. | ||
| if jobErr != nil && !HasErrJobCanceled(jobErr) && resumerCtx.Err() == nil { | ||
| r.td.Dump(dumpCtx, strconv.Itoa(int(jobID)), traceID, r.ex) | ||
|
|
pkg/jobs/adopt.go
Outdated
| // NextRunClause calculates the next execution time of a job with exponential backoff delay, calculated | ||
| // using last_run and num_runs values. |
There was a problem hiding this comment.
nit: This comment belongs to the query below I believe.
nit: Add comment for this newly added query?
There was a problem hiding this comment.
Instead of this approach, can we leave the query exactly as it is and change the way the value which gets generated comes to be by not using the default value? Consider:
diff --git a/pkg/jobs/registry.go b/pkg/jobs/registry.go
index 24a96699ad1..55b7dfed541 100644
--- a/pkg/jobs/registry.go
+++ b/pkg/jobs/registry.go
@@ -379,7 +379,7 @@ func (r *Registry) batchJobInsertStmt(
) (string, []interface{}, []jobspb.JobID, error) {
instanceID := r.ID()
const numColumns = 6
- columns := [numColumns]string{`id`, `status`, `payload`, `progress`, `claim_session_id`, `claim_instance_id`}
+ columns := [numColumns]string{`id`, `created`, `status`, `payload`, `progress`, `claim_session_id`, `claim_instance_id`}
marshalPanic := func(m protoutil.Message) []byte {
data, err := protoutil.Marshal(m)
if err != nil {
@@ -387,8 +387,13 @@ func (r *Registry) batchJobInsertStmt(
}
return data
}
+ created, err := tree.MakeDTimestamp(timeutil.FromUnixMicros(modifiedMicros), time.Microsecond)
+ if err != nil {
+ return "", nil, nil, errors.NewAssertionErrorWithWrappedErrf(err, "failed to make timestamp")
+ }
valueFns := map[string]func(*Record) interface{}{
`id`: func(rec *Record) interface{} { return rec.JobID },
+ `created`: func(record *Record) interface{} { return created },
`status`: func(rec *Record) interface{} { return StatusRunning },
`claim_session_id`: func(rec *Record) interface{} { return sessionID.UnsafeBytes() },
`claim_instance_id`: func(rec *Record) interface{} { return instanceID },Reviewable status:
complete! 0 of 0 LGTMs obtained (waiting on @rhu713 and @ZhouXing19)
pkg/jobs/adopt.go line 135 at r1 (raw file):
PossibleDelay = `args.initial_delay * (power(2, least(62, COALESCE(num_runs, 0))) - 1)::FLOAT` Piece1 = `COALESCE(last_run, created)` Piece2 = `least(
nit: don't export these
Code quote:
PossibleDelay = `args.initial_delay * (power(2, least(62, COALESCE(num_runs, 0))) - 1)::FLOAT`
Piece1 = `COALESCE(last_run, created)`
Piece2 = `least(
There was a problem hiding this comment.
change the way the value which gets generated comes to be by not using the default value
Oh I did something similar in the second commit already: 82f2c6e
But I was with r.clock.Now().GoTime().Format(time.RFC3339Nano). Do you think that would work too?
Reviewable status:
ZhouXing19
force-pushed
the
create-job-with-txn
branch
from
82f2c6e
to
7c43111
Compare
There was a problem hiding this comment.
can we leave the query exactly as it is
Removed the first commit.
Reviewable status:
pkg/jobs/adopt.go line 85 at r1 (raw file):
Previously, Xiang-Gu (Xiang Gu) wrote…
nit: extra empty line
Done.
pkg/jobs/adopt.go line 132 at r1 (raw file):
Previously, Xiang-Gu (Xiang Gu) wrote…
nit: This comment belongs to the query below I believe.
nit: Add comment for this newly added query?
Those are queries that I used for debugging and forgot to remove them in that commit 😅 Have them deleted now.
There was a problem hiding this comment.
Reviewable status:
pkg/jobs/registry.go line 397 at r3 (raw file):
valueFns := map[string]func(*Record) interface{}{ `id`: func(rec *Record) interface{} { return rec.JobID }, `created`: func(rec *Record) interface{} { return r.clock.Now().GoTime().Format(time.RFC3339Nano) },
I think this should use modifiedMicros. It'll evaluate to the same value as now() which it relative to the transaction timestamp. I also have a mild preference for constructing the datum as opposed to passing the string literal to the sql layer and making it parse it back. It also alleviates the need to ask the question whether RFC3339Nano is a format that sql likes.
When creating a job, we should use an internal executor bound to the outer txn and related metadata. We also found that in this case, if a user changes the timezone and then triggers a schema change job, the job's creation time (i.e. value createdcolumn in the system.jobs) will automatically switch to the timestamp in the updated timezone, rather than UTC. This is because we have the created using now() in the SQL statement. We now change it to be inserted always with UTC time. Informs: cockroachdb#91004 Informs: cockroachdb#91718 Release note: None
ZhouXing19
force-pushed
the
create-job-with-txn
branch
from
7c43111
to
a7ef8b2
Compare
|
Previously, ajwerner wrote…
That makes sense, thanks for explaining! |
|
TFTR! |
|
Build succeeded: |
When creating a job, we should use an internal executor bound to the outer txn and related metadata.
We also found that in this case, if a user changes the timezone and then triggers a schema change job, the job's creation time (i.e. value
createdcolumn in thesystem.jobs) will automatically switch to the timestamp in the updated timezone, rather than UTC. This is because we have thecreatedusingnow()in the SQL statement. We now change it to be inserted always with UTC time.Informs: #91004
Informs: #91718
Release note: None