sql,stats: use the jobs framework for CREATE STATISTICS

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 createStats planNode 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.

pkg/jobs/jobspb/jobs.proto

@@ -192,6 +192,29 @@ message ChangefeedProgress {
   repeated ResolvedSpan resolved_spans = 2 [(gogoproto.nullable) = false];
 }
 
+// CreateStatsDetails are used for the CreateStats job, which is triggered
+// whenever the `CREATE STATISTICS` SQL statement is run. The CreateStats job
+// collects table statistics, which contain info such as the number of rows in
+// the table or the number of distinct values in a column.
+message CreateStatsDetails {
+  message ColList {
+    repeated uint32 ids = 1 [
+      (gogoproto.customname) = "IDs",
+      (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/sqlbase.ColumnID"
+    ];
+  }
+  string name = 1 [
+    (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/sem/tree.Name"
+  ];
+  sqlbase.TableDescriptor table = 2 [(gogoproto.nullable) = false];
+  repeated ColList column_lists = 3 [(gogoproto.nullable) = false];
+  string Statement = 4;
+}
+
+message CreateStatsProgress {
+
+}
+
 message Payload {
   string description = 1;
   string username = 2;
@@ -215,6 +238,7 @@ message Payload {
     SchemaChangeDetails schemaChange = 12;
     ImportDetails import = 13;
     ChangefeedDetails changefeed = 14;
+    CreateStatsDetails createStats = 15;
   }
 }
 
@@ -232,6 +256,7 @@ message Progress {
     SchemaChangeProgress schemaChange = 12;
     ImportProgress import = 13;
     ChangefeedProgress changefeed = 14;
+    CreateStatsProgress createStats = 15;
   }
 }
 
@@ -245,4 +270,5 @@ enum Type {
   SCHEMA_CHANGE = 3 [(gogoproto.enumvalue_customname) = "TypeSchemaChange"];
   IMPORT = 4 [(gogoproto.enumvalue_customname) = "TypeImport"];
   CHANGEFEED = 5 [(gogoproto.enumvalue_customname) = "TypeChangefeed"];
+  CREATE_STATS = 6 [(gogoproto.enumvalue_customname) = "TypeCreateStats"];
 }

pkg/jobs/jobspb/wrap.go

@@ -28,6 +28,7 @@ var _ Details = BackupDetails{}
 var _ Details = RestoreDetails{}
 var _ Details = SchemaChangeDetails{}
 var _ Details = ChangefeedDetails{}
+var _ Details = CreateStatsDetails{}
 
 // ProgressDetails is a marker interface for job progress details proto structs.
 type ProgressDetails interface{}
@@ -36,6 +37,7 @@ var _ ProgressDetails = BackupProgress{}
 var _ ProgressDetails = RestoreProgress{}
 var _ ProgressDetails = SchemaChangeProgress{}
 var _ ProgressDetails = ChangefeedProgress{}
+var _ ProgressDetails = CreateStatsProgress{}
 
 // Type returns the payload's job type.
 func (p *Payload) Type() Type {
@@ -55,6 +57,8 @@ func DetailsType(d isPayload_Details) Type {
 		return TypeImport
 	case *Payload_Changefeed:
 		return TypeChangefeed
+	case *Payload_CreateStats:
+		return TypeCreateStats
 	default:
 		panic(fmt.Sprintf("Payload.Type called on a payload with an unknown details type: %T", d))
 	}
@@ -79,6 +83,8 @@ func WrapProgressDetails(details ProgressDetails) interface {
 		return &Progress_Import{Import: &d}
 	case ChangefeedProgress:
 		return &Progress_Changefeed{Changefeed: &d}
+	case CreateStatsProgress:
+		return &Progress_CreateStats{CreateStats: &d}
 	default:
 		panic(fmt.Sprintf("WrapProgressDetails: unknown details type %T", d))
 	}
@@ -98,6 +104,8 @@ func (p *Payload) UnwrapDetails() Details {
 		return *d.Import
 	case *Payload_Changefeed:
 		return *d.Changefeed
+	case *Payload_CreateStats:
+		return *d.CreateStats
 	default:
 		return nil
 	}
@@ -117,6 +125,8 @@ func (p *Progress) UnwrapDetails() ProgressDetails {
 		return *d.Import
 	case *Progress_Changefeed:
 		return *d.Changefeed
+	case *Progress_CreateStats:
+		return *d.CreateStats
 	default:
 		return nil
 	}
@@ -149,6 +159,8 @@ func WrapPayloadDetails(details Details) interface {
 		return &Payload_Import{Import: &d}
 	case ChangefeedDetails:
 		return &Payload_Changefeed{Changefeed: &d}
+	case CreateStatsDetails:
+		return &Payload_CreateStats{CreateStats: &d}
 	default:
 		panic(fmt.Sprintf("jobs.WrapPayloadDetails: unknown details type %T", d))
 	}