crdb_internal.is_constraint_active(table_name: string, constraint_name: string) → boolThis function is used to determine if a given constraint is currently. active for the current transaction.
crdb_internal.job_execution_details(job_id: int) → jsonbOutput a JSONB version of the specified job’s execution details. The execution details are collectedand persisted during the lifetime of the job and provide more observability into the job’s execution
+crdb_internal.lease_holder(key: bytes) → intThis function is used to fetch the leaseholder corresponding to a request key
crdb_internal.list_sql_keys_in_range(range_id: int) → tuple{string AS key, string AS value, string AS ts}Returns all SQL K/V pairs within the requested range.
diff --git a/pkg/BUILD.bazel b/pkg/BUILD.bazel index dd9db07b448f..d4c5ab1235b5 100644 --- a/pkg/BUILD.bazel +++ b/pkg/BUILD.bazel @@ -1174,6 +1174,7 @@ GO_TARGETS = [ "//pkg/jobs/jobsauth:jobsauth_test", "//pkg/jobs/jobspb:jobspb", "//pkg/jobs/jobspb:jobspb_test", + "//pkg/jobs/jobsprofiler/profilerconstants:profilerconstants", "//pkg/jobs/jobsprofiler:jobsprofiler", "//pkg/jobs/jobsprofiler:jobsprofiler_test", "//pkg/jobs/jobsprotectedts:jobsprotectedts", @@ -2678,6 +2679,7 @@ GET_X_DATA_TARGETS = [ "//pkg/jobs/jobsauth:get_x_data", "//pkg/jobs/jobspb:get_x_data", "//pkg/jobs/jobsprofiler:get_x_data", + "//pkg/jobs/jobsprofiler/profilerconstants:get_x_data", "//pkg/jobs/jobsprotectedts:get_x_data", "//pkg/jobs/jobstest:get_x_data", "//pkg/jobs/metricspoller:get_x_data", diff --git a/pkg/jobs/jobsprofiler/BUILD.bazel b/pkg/jobs/jobsprofiler/BUILD.bazel index 0c0625cf24bd..31240a88f9a6 100644 --- a/pkg/jobs/jobsprofiler/BUILD.bazel +++ b/pkg/jobs/jobsprofiler/BUILD.bazel @@ -9,6 +9,7 @@ go_library( deps = [ "//pkg/jobs", "//pkg/jobs/jobspb", + "//pkg/jobs/jobsprofiler/profilerconstants", "//pkg/sql", "//pkg/sql/execinfrapb", "//pkg/sql/isql", diff --git a/pkg/jobs/jobsprofiler/profiler.go b/pkg/jobs/jobsprofiler/profiler.go index a0145961736f..5ea2bab647bf 100644 --- a/pkg/jobs/jobsprofiler/profiler.go +++ b/pkg/jobs/jobsprofiler/profiler.go @@ -16,6 +16,7 @@ import ( "github.com/cockroachdb/cockroach/pkg/jobs" "github.com/cockroachdb/cockroach/pkg/jobs/jobspb" + "github.com/cockroachdb/cockroach/pkg/jobs/jobsprofiler/profilerconstants" "github.com/cockroachdb/cockroach/pkg/sql" "github.com/cockroachdb/cockroach/pkg/sql/execinfrapb" "github.com/cockroachdb/cockroach/pkg/sql/isql" @@ -37,15 +38,15 @@ func StorePlanDiagram( err := db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error { flowSpecs := p.GenerateFlowSpecs() - _, diagURL, err := execinfrapb.GeneratePlanDiagramURL(fmt.Sprintf("job:%d", jobID), flowSpecs, execinfrapb.DiagramFlags{}) + _, diagURL, err := execinfrapb.GeneratePlanDiagramURL(fmt.Sprintf("job:%d", jobID), flowSpecs, + execinfrapb.DiagramFlags{}) if err != nil { return err } - const infoKey = "dsp-diag-url-%d" infoStorage := jobs.InfoStorageForJob(txn, jobID) - return infoStorage.Write(ctx, fmt.Sprintf(infoKey, timeutil.Now().UnixNano()), - []byte(diagURL.String())) + return infoStorage.Write(ctx, fmt.Sprintf(profilerconstants.DSPDiagramInfoKeyPrefix+"%d", + timeutil.Now().UnixNano()), []byte(diagURL.String())) }) if err != nil { log.Warningf(ctx, "failed to generate and write DistSQL diagram for job %d: %v", diff --git a/pkg/jobs/jobsprofiler/profilerconstants/BUILD.bazel b/pkg/jobs/jobsprofiler/profilerconstants/BUILD.bazel new file mode 100644 index 000000000000..291fbfb895e2 --- /dev/null +++ b/pkg/jobs/jobsprofiler/profilerconstants/BUILD.bazel @@ -0,0 +1,11 @@ +load("//build/bazelutil/unused_checker:unused.bzl", "get_x_data") +load("@io_bazel_rules_go//go:def.bzl", "go_library") + +go_library( + name = "profilerconstants", + srcs = ["constants.go"], + importpath = "github.com/cockroachdb/cockroach/pkg/jobs/jobsprofiler/profilerconstants", + visibility = ["//visibility:public"], +) + +get_x_data(name = "get_x_data") diff --git a/pkg/jobs/jobsprofiler/profilerconstants/constants.go b/pkg/jobs/jobsprofiler/profilerconstants/constants.go new file mode 100644 index 000000000000..5be37466b682 --- /dev/null +++ b/pkg/jobs/jobsprofiler/profilerconstants/constants.go @@ -0,0 +1,15 @@ +// Copyright 2023 The Cockroach Authors. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +package profilerconstants + +// DSPDiagramInfoKeyPrefix is the prefix of the info key used for rows that +// store the DistSQL plan diagram for a job. +const DSPDiagramInfoKeyPrefix = "dsp-diag-url-" diff --git a/pkg/sql/BUILD.bazel b/pkg/sql/BUILD.bazel index 165f7b551f85..4b092f6db594 100644 --- a/pkg/sql/BUILD.bazel +++ b/pkg/sql/BUILD.bazel @@ -141,6 +141,7 @@ go_library( "job_exec_context.go", "job_exec_context_test_util.go", "jobs_collection.go", + "jobs_execution_details.go", "join.go", "join_predicate.go", "join_token.go", @@ -308,6 +309,7 @@ go_library( "//pkg/jobs", "//pkg/jobs/jobsauth", "//pkg/jobs/jobspb", + "//pkg/jobs/jobsprofiler/profilerconstants", "//pkg/keys", "//pkg/keyvisualizer", "//pkg/kv", @@ -635,6 +637,7 @@ go_test( "indexbackfiller_test.go", "instrumentation_test.go", "internal_test.go", + "jobs_execution_details_test.go", "join_token_test.go", "main_test.go", "materialized_view_test.go", @@ -712,6 +715,7 @@ go_test( deps = [ "//pkg/base", "//pkg/build/bazel", + "//pkg/cloud/impl:cloudimpl", "//pkg/clusterversion", "//pkg/col/coldata", "//pkg/config", @@ -720,6 +724,8 @@ go_test( "//pkg/internal/sqlsmith", "//pkg/jobs", "//pkg/jobs/jobspb", + "//pkg/jobs/jobsprofiler", + "//pkg/jobs/jobsprofiler/profilerconstants", "//pkg/jobs/jobstest", "//pkg/keys", "//pkg/keyvisualizer", diff --git a/pkg/sql/conn_executor.go b/pkg/sql/conn_executor.go index 80a85dbb5240..e466aad140b0 100644 --- a/pkg/sql/conn_executor.go +++ b/pkg/sql/conn_executor.go @@ -3398,6 +3398,7 @@ func (ex *connExecutor) initEvalCtx(ctx context.Context, evalCtx *extendedEvalCo QueryCancelKey: ex.queryCancelKey, DescIDGenerator: ex.getDescIDGenerator(), RangeStatsFetcher: p.execCfg.RangeStatsFetcher, + JobsProfiler: p, }, Tracing: &ex.sessionTracing, MemMetrics: &ex.memMetrics, diff --git a/pkg/sql/delegate/show_jobs.go b/pkg/sql/delegate/show_jobs.go index 950f8047b8bc..16002c063031 100644 --- a/pkg/sql/delegate/show_jobs.go +++ b/pkg/sql/delegate/show_jobs.go @@ -20,25 +20,25 @@ import ( "github.com/cockroachdb/cockroach/pkg/sql/sqltelemetry" ) -func (d *delegator) delegateShowJobs(n *tree.ShowJobs) (tree.Statement, error) { - if n.Schedules != nil { - // Limit the jobs displayed to the ones started by specified schedules. - return d.parse(fmt.Sprintf(` -SHOW JOBS SELECT id FROM system.jobs WHERE created_by_type='%s' and created_by_id IN (%s) -`, jobs.CreatedByScheduledJobs, n.Schedules.String()), - ) - } - - sqltelemetry.IncrementShowCounter(sqltelemetry.Jobs) - - const ( - selectClause = ` +func constructSelectQuery(n *tree.ShowJobs) string { + var baseQuery strings.Builder + baseQuery.WriteString(` SELECT job_id, job_type, description, statement, user_name, status, running_status, created, started, finished, modified, fraction_completed, error, coordinator_id, trace_id, last_run, next_run, num_runs, execution_errors - FROM crdb_internal.jobs` - ) +`) + + // Check if there are any SHOW JOBS options that we need to add columns for. + if n.Options != nil { + if n.Options.ExecutionDetails { + baseQuery.WriteString(`, NULLIF(crdb_internal.job_execution_details(job_id)->>'plan_diagram'::STRING, '') AS plan_diagram`) + } + } + + baseQuery.WriteString("\nFROM crdb_internal.jobs") + + // Now add the predicates and ORDER BY clauses. var typePredicate, whereClause, orderbyClause string if n.Jobs == nil { // Display all [only automatic] jobs without selecting specific jobs. @@ -74,10 +74,23 @@ SELECT job_id, job_type, description, statement, user_name, status, // Limit the jobs displayed to the select statement in n.Jobs. whereClause = fmt.Sprintf(`WHERE job_id in (%s)`, n.Jobs.String()) } + return fmt.Sprintf("%s %s %s", baseQuery.String(), whereClause, orderbyClause) +} + +func (d *delegator) delegateShowJobs(n *tree.ShowJobs) (tree.Statement, error) { + if n.Schedules != nil { + // Limit the jobs displayed to the ones started by specified schedules. + return d.parse(fmt.Sprintf(` +SHOW JOBS SELECT id FROM system.jobs WHERE created_by_type='%s' and created_by_id IN (%s) +`, jobs.CreatedByScheduledJobs, n.Schedules.String()), + ) + } + + sqltelemetry.IncrementShowCounter(sqltelemetry.Jobs) - sqlStmt := fmt.Sprintf("%s %s %s", selectClause, whereClause, orderbyClause) + stmt := constructSelectQuery(n) if n.Block { - sqlStmt = fmt.Sprintf( + stmt = fmt.Sprintf( ` WITH jobs AS (SELECT * FROM [%s]), sleep_and_restart_if_unfinished AS ( @@ -92,7 +105,7 @@ SELECT job_id, job_type, description, statement, user_name, status, ) SELECT * FROM jobs - WHERE NOT EXISTS(SELECT * FROM fail_if_slept_too_long)`, sqlStmt) + WHERE NOT EXISTS(SELECT * FROM fail_if_slept_too_long)`, stmt) } - return d.parse(sqlStmt) + return d.parse(stmt) } diff --git a/pkg/sql/jobs_execution_details.go b/pkg/sql/jobs_execution_details.go new file mode 100644 index 000000000000..43428458708d --- /dev/null +++ b/pkg/sql/jobs_execution_details.go @@ -0,0 +1,75 @@ +// Copyright 2023 The Cockroach Authors. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +package sql + +import ( + "context" + gojson "encoding/json" + + "github.com/cockroachdb/cockroach/pkg/jobs" + "github.com/cockroachdb/cockroach/pkg/jobs/jobspb" + "github.com/cockroachdb/cockroach/pkg/jobs/jobsprofiler/profilerconstants" + "github.com/cockroachdb/cockroach/pkg/sql/isql" + "github.com/cockroachdb/cockroach/pkg/sql/sem/eval" +) + +// GenerateExecutionDetailsJSON implements the Profiler interface. +func (p *planner) GenerateExecutionDetailsJSON( + ctx context.Context, evalCtx *eval.Context, jobID jobspb.JobID, +) ([]byte, error) { + execCfg := evalCtx.Planner.ExecutorConfig().(*ExecutorConfig) + j, err := execCfg.JobRegistry.LoadJob(ctx, jobID) + if err != nil { + return nil, err + } + + var executionDetailsJSON []byte + payload := j.Payload() + switch payload.Type() { + // TODO(adityamaru): This allows different job types to implement custom + // execution detail aggregation. + default: + executionDetailsJSON, err = constructDefaultExecutionDetails(ctx, jobID, execCfg.InternalDB) + } + + return executionDetailsJSON, err +} + +// defaultExecutionDetails is a JSON serializable struct that captures the +// execution details that are not specific to a particular job type. +type defaultExecutionDetails struct { + // PlanDiagram is the URL to render the latest DistSQL plan that is being + // executed by the job. + PlanDiagram string `json:"plan_diagram"` +} + +func constructDefaultExecutionDetails( + ctx context.Context, jobID jobspb.JobID, db isql.DB, +) ([]byte, error) { + executionDetails := &defaultExecutionDetails{} + err := db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error { + // Populate the latest DSP diagram URL. + infoStorage := jobs.InfoStorageForJob(txn, jobID) + err := infoStorage.GetLast(ctx, profilerconstants.DSPDiagramInfoKeyPrefix, func(infoKey string, value []byte) error { + executionDetails.PlanDiagram = string(value) + return nil + }) + if err != nil { + return err + } + return nil + }) + if err != nil { + return nil, err + } + j, err := gojson.Marshal(executionDetails) + return j, err +} diff --git a/pkg/sql/jobs_execution_details_test.go b/pkg/sql/jobs_execution_details_test.go new file mode 100644 index 000000000000..5034e3c95e7b --- /dev/null +++ b/pkg/sql/jobs_execution_details_test.go @@ -0,0 +1,126 @@ +// Copyright 2023 The Cockroach Authors. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +package sql_test + +import ( + "context" + "testing" + "time" + + "github.com/cockroachdb/cockroach/pkg/base" + _ "github.com/cockroachdb/cockroach/pkg/cloud/impl" // register cloud storage providers + "github.com/cockroachdb/cockroach/pkg/jobs" + "github.com/cockroachdb/cockroach/pkg/jobs/jobspb" + "github.com/cockroachdb/cockroach/pkg/jobs/jobsprofiler" + "github.com/cockroachdb/cockroach/pkg/jobs/jobsprofiler/profilerconstants" + "github.com/cockroachdb/cockroach/pkg/settings/cluster" + "github.com/cockroachdb/cockroach/pkg/sql" + "github.com/cockroachdb/cockroach/pkg/sql/isql" + "github.com/cockroachdb/cockroach/pkg/sql/physicalplan" + "github.com/cockroachdb/cockroach/pkg/sql/tests" + "github.com/cockroachdb/cockroach/pkg/testutils" + "github.com/cockroachdb/cockroach/pkg/testutils/jobutils" + "github.com/cockroachdb/cockroach/pkg/testutils/serverutils" + "github.com/cockroachdb/cockroach/pkg/testutils/sqlutils" + "github.com/cockroachdb/cockroach/pkg/util/leaktest" + "github.com/cockroachdb/cockroach/pkg/util/log" + "github.com/cockroachdb/cockroach/pkg/util/uuid" + "github.com/cockroachdb/errors" + "github.com/stretchr/testify/require" +) + +// fakeExecResumer calls optional callbacks during the job lifecycle. +type fakeExecResumer struct { + OnResume func(context.Context) error + FailOrCancel func(context.Context) error +} + +var _ jobs.Resumer = fakeExecResumer{} + +func (d fakeExecResumer) Resume(ctx context.Context, execCtx interface{}) error { + if d.OnResume != nil { + if err := d.OnResume(ctx); err != nil { + return err + } + } + return nil +} + +func (d fakeExecResumer) OnFailOrCancel(ctx context.Context, _ interface{}, _ error) error { + if d.FailOrCancel != nil { + return d.FailOrCancel(ctx) + } + return nil +} + +// checkForPlanDiagram is a method used in tests to wait for the existence of a +// DSP diagram for the provided jobID. +func checkForPlanDiagram(ctx context.Context, t *testing.T, db isql.DB, jobID jobspb.JobID) { + testutils.SucceedsSoon(t, func() error { + return db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error { + infoStorage := jobs.InfoStorageForJob(txn, jobID) + var found bool + err := infoStorage.GetLast(ctx, profilerconstants.DSPDiagramInfoKeyPrefix, + func(infoKey string, value []byte) error { + found = true + return nil + }) + if err != nil || !found { + return errors.New("not found") + } + return nil + }) + }) +} + +// TestJobsExecutionDetails tests that a job's execution details are retrieved +// and rendered correctly. +func TestJobsExecutionDetails(t *testing.T) { + defer leaktest.AfterTest(t)() + defer log.Scope(t).Close(t) + + // Timeout the test in a few minutes if it hasn't succeeded. + ctx := context.Background() + ctx, cancel := context.WithTimeout(ctx, time.Minute*2) + defer cancel() + + params, _ := tests.CreateTestServerParams() + params.Knobs.JobsTestingKnobs = jobs.NewTestingKnobsWithShortIntervals() + defer jobs.ResetConstructors()() + s, sqlDB, _ := serverutils.StartServer(t, params) + defer s.Stopper().Stop(ctx) + + runner := sqlutils.MakeSQLRunner(sqlDB) + + jobs.RegisterConstructor(jobspb.TypeImport, func(j *jobs.Job, _ *cluster.Settings) jobs.Resumer { + return fakeExecResumer{ + OnResume: func(ctx context.Context) error { + p := sql.PhysicalPlan{} + infra := physicalplan.NewPhysicalInfrastructure(uuid.FastMakeV4(), base.SQLInstanceID(1)) + p.PhysicalInfrastructure = infra + jobsprofiler.StorePlanDiagram(ctx, s.Stopper(), &p, s.InternalDB().(isql.DB), j.ID()) + checkForPlanDiagram(ctx, t, s.InternalDB().(isql.DB), j.ID()) + return nil + }, + } + }, jobs.UsesTenantCostControl) + + runner.Exec(t, `CREATE TABLE t (id INT)`) + runner.Exec(t, `INSERT INTO t SELECT generate_series(1, 100)`) + var importJobID int + runner.QueryRow(t, `IMPORT INTO t CSV DATA ('nodelocal://1/foo') WITH DETACHED`).Scan(&importJobID) + jobutils.WaitForJobToSucceed(t, runner, jobspb.JobID(importJobID)) + + var count int + runner.QueryRow(t, `SELECT count(*) FROM [SHOW JOB $1 WITH EXECUTION DETAILS] WHERE plan_diagram IS NOT NULL`, importJobID).Scan(&count) + require.NotZero(t, count) + runner.CheckQueryResults(t, `SELECT count(*) FROM [SHOW JOBS WITH EXECUTION DETAILS] WHERE plan_diagram IS NOT NULL`, [][]string{{"1"}}) +} diff --git a/pkg/sql/parser/BUILD.bazel b/pkg/sql/parser/BUILD.bazel index 07c9f1518fd8..9e1ae39ca90f 100644 --- a/pkg/sql/parser/BUILD.bazel +++ b/pkg/sql/parser/BUILD.bazel @@ -71,7 +71,6 @@ go_test( "//pkg/testutils/datapathutils", "//pkg/testutils/sqlutils", "//pkg/util/leaktest", - "//pkg/util/log", "//pkg/util/randutil", "@com_github_cockroachdb_datadriven//:datadriven", "@com_github_cockroachdb_errors//:errors", diff --git a/pkg/sql/parser/parse_test.go b/pkg/sql/parser/parse_test.go index d1814617d06a..79d5de1c6de6 100644 --- a/pkg/sql/parser/parse_test.go +++ b/pkg/sql/parser/parse_test.go @@ -30,7 +30,6 @@ import ( "github.com/cockroachdb/cockroach/pkg/testutils" "github.com/cockroachdb/cockroach/pkg/testutils/datapathutils" "github.com/cockroachdb/cockroach/pkg/testutils/sqlutils" - _ "github.com/cockroachdb/cockroach/pkg/util/log" // for flags "github.com/cockroachdb/cockroach/pkg/util/randutil" "github.com/cockroachdb/datadriven" "github.com/cockroachdb/errors" diff --git a/pkg/sql/parser/sql.y b/pkg/sql/parser/sql.y index 542f91eeb62c..d0264d99325f 100644 --- a/pkg/sql/parser/sql.y +++ b/pkg/sql/parser/sql.y @@ -701,6 +701,9 @@ func (u *sqlSymUnion) backupOptions() *tree.BackupOptions { func (u *sqlSymUnion) copyOptions() *tree.CopyOptions { return u.val.(*tree.CopyOptions) } +func (u *sqlSymUnion) showJobOptions() *tree.ShowJobOptions { + return u.val.(*tree.ShowJobOptions) +} func (u *sqlSymUnion) showBackupDetails() tree.ShowBackupDetails { return u.val.(tree.ShowBackupDetails) } @@ -1323,6 +1326,7 @@ func (u *sqlSymUnion) showCreateFormatOption() tree.ShowCreateFormatOption { %type <*tree.RestoreOptions> opt_with_restore_options restore_options restore_options_list %type <*tree.TenantReplicationOptions> opt_with_tenant_replication_options tenant_replication_options tenant_replication_options_list %type