sql, stats: use the jobs framework for CREATE STATISTICS

[rytaft]

Prior to this commit, CREATE STATISTICS was a regular SQL statement
and followed the standard logic of AST -> planNode -> DistSQL physical
plan -> execution. This commit changes CREATE STATISTICS to use the
jobs framework, and as a result the createStatsNode has been
changed to execute a function which starts a CreateStats job. The job
is then responsible for performing DistSQL planning and execution.

There are several advantages to using the jobs framework:

• Now CREATE STATISTICS jobs can easily be cancelled, paused and resumed,
  and viewed from the Admin UI.
• Nodes can adopt the job if the original gateway node fails.
• We will be able to use the JobID to lock creation of automatic
  statistics, so that only one automatic statistics job can run at
  a time. Job adoption will ensure that a dead node never prevents progress
  by holding a lock on stats creation (implementation of locking will be
  saved for the next PR).

Release note (sql change): CREATE STATISTICS now runs as a job instead
of as a regular SQL statement.

[cockroach-teamcity]

This change is 

[RaduBerinde]

Cool! It will take some time to unpack this, it involves a lot of infrastructure I'm not familiar with.

The plan node hook stuff is to allow the implementation to live in CCL packages (which is not the case here). Are you sure that part is needed?

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andy-kimball, @danhhz, @dt, @mjibson, @RaduBerinde, and @rytaft)

---

pkg/jobs/jobspb/jobs.proto, line 206 at r1 (raw file):

  ];
  sqlbase.TableDescriptor table = 2 [(gogoproto.nullable) = false];
  repeated Columns columns = 3 [(gogoproto.nullable) = false];

This should be a list of lists. Even though there's no syntax, a single CreateStats can in principle create stats for multiple sets of columns. In fact, that's what it does now when no columns are specified (in the future, we may want to populate this with those auto-determined columns).

[rytaft]

Thanks!

The plan node hook stuff is to allow the implementation to live in CCL packages (which is not the case here). Are you sure that part is needed?

I'm not sure, but the reason I did it this way was because we don't want the normal AST -> planNode -> DistSQL plan -> execution pipeline. The planNode returned by planner.newPlan should just create a new job and stop. hookFnNode is a convenient way to do that since it just executes the function registered with AddPlanHook. I could put the createStats planNode back, but I think it would effectively do the same thing as hookFnNode.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andy-kimball, @danhhz, @dt, @mjibson, and @RaduBerinde)

---

pkg/jobs/jobspb/jobs.proto, line 206 at r1 (raw file):

Previously, RaduBerinde wrote…

This should be a list of lists. Even though there's no syntax, a single CreateStats can in principle create stats for multiple sets of columns. In fact, that's what it does now when no columns are specified (in the future, we may want to populate this with those auto-determined columns).

This is actually a list of lists -- it's a repeated element which itself contains a repeated element (I don't know of another way to represent that with protobufs). But I agree the naming doesn't make that obvious... I've changed the names to hopefully make it more clear.

[RaduBerinde]

Oh, I see, thanks, that makes sense!

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andy-kimball, @danhhz, @dt, @mjibson, and @rytaft)

---

pkg/jobs/jobspb/jobs.proto, line 206 at r1 (raw file):

Previously, rytaft wrote…

This is actually a list of lists -- it's a repeated element which itself contains a repeated element (I don't know of another way to represent that with protobufs). But I agree the naming doesn't make that obvious... I've changed the names to hopefully make it more clear.

Sorry, my bad, missed the inner "repeated". I like the new name though, thanks!

[danhhz]

All the pieces here look good! I didn't do a detailed review, but I can if you need an additional pair of eyes.

From a high level, my two large concerns with any new job are 1) overloading the jobs row with progress updates, 2) zombie jobs and 3) the user UX of the jobs entries.

1. It's really easy to gum things up with txn contention updating the job progress. Changefeeds, for example, funnel progress info to a singleton distsql processor, which then rate limits and does the updates. I don't think it's worth the complexity in your case of a processor just for that, but given that the number of updates here scales with the number of processors, it's worth a few preventative measures to mitigate the risk.

First, you could rate limit the progress updates based on time. No need to update more than every second or so.

Even easier here, if your TODO is going to stick around for a while, since you're not actually updating the progress, you're just checking whether the job is still live, is to use a read-only txn. I don't recall offhand if the jobs framework already has an accessor to grab the state, but if it doesn't, feel free to add one. This would be used instead of the progress update in your worker loop. If it also surfaced the error, you could also use it the error handling bit in your planner code.

2. Dunno the details of the lock you describe, but important to know that you can end up with zombie jobs. The framework does its best to let only one copy of each job run at a time, but there are various conditions around failures where we can end up with two copies of the job running at once: the blessed one and a zombie one. I think you'll be fine, since the worst thing that happens here is probably some wasted work (as opposed to restore/changefeeds/etc where we have correctness issues to worry about), but important to keep in mind.

3. How often do we make these jobs? What does that look like in the jobs admin ui page? I've only been following the statistics stuff at a high level, but the number of these we run scales with traffic right? Maybe with num tables, too? Unless something has changed since I last looked, we don't garbage collect old jobs, so these could clutter things up. On the other hand, that page already has a dropdown for job type, so maybe that's sufficient. I just want to make sure this is considered.

I could put the createStats planNode back, but I think it would effectively do the same thing as hookFnNode.

This is likely what I'd recommend, make the planNode submit the job and then either return the jobID (so the client can poll progress) or have it wait until the job finishes. Most (all?) existing jobs do the former. We tend to avoid planHook when we can (which is any non-ccl code).

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andy-kimball, @danhhz, @dt, @mjibson, and @rytaft)

[RaduBerinde]

@danhhz How does changefeed funnel progress info? Is it through a special ProducerMetadata? We already have a single "final" processor (SampleAggregator), we should be able to use the same mechanism.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andy-kimball, @dt, @mjibson, and @rytaft)

[danhhz]

Changefeed progress is both used for some user facing stuff and super complicated and as a result, ended up needing stronger ordering guarantees than distsql metadata provides.

For this job, you have two "progress" concerns. One is stopping the workers when the job is paused or fails. This can be done with the read only txn I mentioned or (now that I think about it) handled with ctx cancellation. The latter is what changefeeds do. The second concern is actually updating the progress. Distsql metadata and doing it on SampleAggregator sgtm

[rytaft]

Thanks, @danhhz!

1. Regarding the read-only txn: In my first attempt I actually took this approach and stored the transaction info in the CreateStatsDetails. Not sure I understand the suggestion to create an accessor for the jobs framework, but that sounds like it might be better -- can you explain a bit more? Alternatively, ctx cancellation also sounds like a good option. Is there a reason to prefer one over the other?

2. Good to know about the possibility of zombie jobs, but I think you're right that the only consequence would be some wasted work and perhaps temporarily degraded performance. The purpose of the lock is for performance reasons -- there is no correctness issue with running multiple CREATE STATISTICS jobs at once.

3. You're right that there could be a lot of CREATE STATISTICS jobs cluttering the admin UI. The TYPE dropdown does help, though (speaking of which, looks like I need to add CREATE STATISTICS to the dropdown....). Should I create an issue to garbage collect old jobs (at least CreateStats jobs)?

4. Good to know about avoiding planHook -- I'll put the createStats node back.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andy-kimball, @dt, and @mjibson)

[danhhz]

Regarding the read-only txn: In my first attempt I actually took this approach and stored the transaction info in the CreateStatsDetails. Not sure I understand the suggestion to create an accessor for the jobs framework, but that sounds like it might be better -- can you explain a bit more? Alternatively, ctx cancellation also sounds like a good option. Is there a reason to prefer one over the other?

Right now, in (*samplerProcessor).mainLoop you're using job.FractionProgressed (via progFn) to "periodically check that the job has not been paused or canceled." My original suggestion was to add a method to Job that fetches and returns the current status without also doing the write that FractionProgressed does. However, I think that's moot because it's better to use ctx cancellation. This is because the case of a zombie job, where you could end up with both the zombie job and the real job seeing the status of "running", so the zombie job wouldn't shut down. But the context for the zombie job will be cancelled, so it's a better signal. Basically, where you currently have if err := progFn(0); err != nil {, you'd replace it with if err := ctx.Err(); err != nil { and everything should work 🤞

Should I create an issue to garbage collect old jobs (at least CreateStats jobs)?

Jobs is unstaffed at this point, so if this needs to be done as a result of making stats a job, then you might have to pick this one off yourself : - ).

[rytaft]

@danhhz -- It seems like the context is not getting cancelled automatically. I could add a function to cancel the job's context and call it from OnFailOrCancel, but that wouldn't work for PAUSE JOB. Perhaps I should do both of your suggestions: add a function to get the status and add a function to cancel the context? Does that seem reasonable?

Jobs is unstaffed at this point, so if this needs to be done as a result of making stats a job, then you might have to pick this one off yourself : - ).

No problem! I'll create an issue and assign to myself. Doesn't seem urgent (the UI seems to work fine, especially given the TYPE dropdown), but I'll try to get it done before the release...

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andy-kimball, @dt, and @mjibson)

[danhhz]

The context cancellation does work, changefeeds rely on it. What test are you running that cause you to say it doesn't?

Usually, the tricky bit is getting the distsql error handling right, but it's worth the effort to investigate what's going on. Basically, if the job is paused or cancelled, the ctx your Resume function gets will be cancelled. If the job coordinator node is killed, then your SampleAggregator should get an error back from job.FractionProgressed and it should use that error to tear down the processor, which will cause distsql to propagate the ctx cancellation to the rest of your flow.

[danhhz]

jobs stuff lgtm modulo a few last comments. great work!

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andy-kimball, @dt, @mjibson, and @rytaft)

---

pkg/sql/distsqlrun/sample_aggregator.go, line 19 at r2 (raw file):

import (
	"context"
	time "time"

nit: this is odd

---

pkg/sql/distsqlrun/sample_aggregator.go, line 145 at r2 (raw file):

	var job *jobs.Job
	jobID := s.spec.JobID
	if jobID != 0 {

are there times that you won't have a job id? maybe stick a comment here with when that happens

---

pkg/sql/distsqlrun/sample_aggregator.go, line 173 at r2 (raw file):

			// rows have been processed. The purpose of this function call is to
			// periodically check that the job has not been paused or canceled.
			if err := progFn(0); err != nil {

I was imagining doing this in response to getting distsql metadata with progress info filled in (rate limited with a util.Every(sampleAggregatorProgressInterval)). This works too, but my suggestion would save you the goroutine (and what you have here leaks the timer).

---

pkg/sql/distsqlrun/sample_aggregator.go, line 246 at r2 (raw file):

	// Report progress one last time so we don't write results if the job was
	// canceled.
	if err = progFn(1.0); err != nil {

Sorry if i've missed it, but do you have a job.Succeeded somewhere? You want to call it when the job finishes successfully to put the job into the terminal "succeeded" status

[rytaft]

Thanks! Updated.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andy-kimball, @danhhz, @dt, and @mjibson)

---

pkg/sql/distsqlrun/sample_aggregator.go, line 19 at r2 (raw file):

Previously, danhhz (Daniel Harrison) wrote…

nit: this is odd

Done.

---

pkg/sql/distsqlrun/sample_aggregator.go, line 145 at r2 (raw file):

Previously, danhhz (Daniel Harrison) wrote…

are there times that you won't have a job id? maybe stick a comment here with when that happens

Done.

---

pkg/sql/distsqlrun/sample_aggregator.go, line 173 at r2 (raw file):

Previously, danhhz (Daniel Harrison) wrote…

I was imagining doing this in response to getting distsql metadata with progress info filled in (rate limited with a util.Every(sampleAggregatorProgressInterval)). This works too, but my suggestion would save you the goroutine (and what you have here leaks the timer).

Oh nice! Didn't know that function existed. Done.

---

pkg/sql/distsqlrun/sample_aggregator.go, line 246 at r2 (raw file):

Previously, danhhz (Daniel Harrison) wrote…

Sorry if i've missed it, but do you have a job.Succeeded somewhere? You want to call it when the job finishes successfully to put the job into the terminal "succeeded" status

Seems like that's getting called inside Registry.resume (in jobs/registry.go) once createStatsResumer.Resume finishes successfully.

[danhhz]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andy-kimball, @danhhz, @dt, @mjibson, and @rytaft)

---

pkg/sql/distsqlrun/sample_aggregator.go, line 246 at r2 (raw file):

Previously, rytaft wrote…

Seems like that's getting called inside Registry.resume (in jobs/registry.go) once createStatsResumer.Resume finishes successfully.

🤦‍♂️ carry on

[rytaft]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andy-kimball, @danhhz, @dt, and @mjibson)

---

pkg/sql/distsqlrun/sample_aggregator.go, line 246 at r2 (raw file):

Previously, danhhz (Daniel Harrison) wrote…

🤦‍♂️ carry on

👍 No worries!

[RaduBerinde]

Nice work!

Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @andy-kimball, @danhhz, @dt, @mjibson, and @rytaft)

---

pkg/jobs/jobspb/jobs.proto, line 195 at r3 (raw file):

}

message CreateStatsDetails {

[nit] could use a comment mentioning this is related to table stats and CREATE STATS statement.

---

pkg/sql/create_stats.go, line 80 at r3 (raw file):

func (p *planner) CreateStatistics(ctx context.Context, n *tree.CreateStats) (planNode, error) {
	fn := func(ctx context.Context, resultsCh chan<- tree.Datums) error {

Do we need to store this as a function? Can't it just be a separate method?

---

pkg/sql/create_stats.go, line 221 at r3 (raw file):

}

type createStatsResumer struct {

I would add a description here and explain how the object is used. Especially that an instance of this is associated with a single job (because the methods in the interface suggest otherwise)

---

pkg/sql/create_stats.go, line 249 at r3 (raw file):

	dsp := p.DistSQLPlanner()
	err := r.evalCtx.ExecCfg.DB.Txn(ctx, func(ctx context.Context, txn *client.Txn) error {

[nit] can just return so we don't have yet another err instance

---

pkg/sql/create_stats.go, line 276 at r3 (raw file):

				return err
			}
			return err

We have two different errs here which makes this kind of confusing, I'd give the inner one a different name (eg jobErr)

---

pkg/sql/create_stats.go, line 293 at r3 (raw file):

// OnSuccess is part of the jobs.Resumer interface.
func (r *createStatsResumer) OnSuccess(ctx context.Context, txn *client.Txn, job *jobs.Job) error {

nit: maybe use _ instead of txn here to make it clear we're not using that

---

pkg/sql/create_stats.go, line 323 at r3 (raw file):

}

func createStatsResumeHook(typ jobspb.Type, settings *cluster.Settings) jobs.Resumer {

[nit] maybe add a comment, or if not I'd just declare this directly in the AddResumeHook call.

[rytaft]

TFTR! Updated.

Reviewable status: complete! 0 of 0 LGTMs obtained (and 1 stale) (waiting on @andy-kimball, @dt, and @mjibson)

---

pkg/jobs/jobspb/jobs.proto, line 195 at r3 (raw file):

Previously, RaduBerinde wrote…

[nit] could use a comment mentioning this is related to table stats and CREATE STATS statement.

Done.

---

pkg/sql/create_stats.go, line 80 at r3 (raw file):

Previously, RaduBerinde wrote…

Do we need to store this as a function? Can't it just be a separate method?

Done.

---

pkg/sql/create_stats.go, line 221 at r3 (raw file):

Previously, RaduBerinde wrote…

I would add a description here and explain how the object is used. Especially that an instance of this is associated with a single job (because the methods in the interface suggest otherwise)

Done.

---

pkg/sql/create_stats.go, line 249 at r3 (raw file):

Previously, RaduBerinde wrote…

[nit] can just return so we don't have yet another err instance

Done.

---

pkg/sql/create_stats.go, line 276 at r3 (raw file):

Previously, RaduBerinde wrote…

We have two different errs here which makes this kind of confusing, I'd give the inner one a different name (eg jobErr)

Done.

---

pkg/sql/create_stats.go, line 293 at r3 (raw file):

Previously, RaduBerinde wrote…

nit: maybe use _ instead of txn here to make it clear we're not using that

Done.

---

pkg/sql/create_stats.go, line 323 at r3 (raw file):

Previously, RaduBerinde wrote…

[nit] maybe add a comment, or if not I'd just declare this directly in the AddResumeHook call.

Done.

[rytaft]

bors r+

[craig]

Build succeeded

• GitHub CI (Cockroach)