Merge #148615 #148871 #148906 #148964

148615: sql: add validation for CTAS / materialized views r=fqazi a=fqazi

Previously, no validation existed when backfilling queries for CTAS statements or creating materialized view. This could lead to scenarios with unnoticed data loss if anything went wrong. To address this, this patch adds validation operations that backfill queries into tables to confirm that row counts match the source.

Fixes: #144957 
Release note: None

148871: cli: fix printing of the store spec r=yuzefovich a=yuzefovich

This was recently broken in 4744a80.

Epic: None
Release note: None

148906: crdb_internal: replace crdb_internal.jobs vtable with view r=dt a=dt

Release note: none.
Epic: https://cockroachlabs.atlassian.net/browse/CRDB-48791

148964: nightly-stress: move engflow stress jobs to run on mesolite r=rickystewart a=jlinder

Part of: DEVINF-1489
Release note: None

Co-authored-by: Faizan Qazi <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>
Co-authored-by: David Taylor <[email protected]>
Co-authored-by: James H. Linder <[email protected]>

Author: 5 people

build/teamcity/cockroach/nightlies/stress_engflow_impl.sh

@@ -17,9 +17,10 @@ mkdir -p artifacts
 
 # NB: The certs are set up and cleaned up in unconditional build steps before
 # and after this build step.
-ENGFLOW_FLAGS="--config engflow --config crosslinux \
---jobs 400 --tls_client_certificate=/home/agent/engflow/engflow.crt \
---tls_client_key=/home/agent/engflow/engflow.key"
+ENGFLOW_FLAGS="--config engflowpublic --config crosslinux \
+--jobs 400 --tls_client_certificate=/home/agent/engflow/engflowpublic.crt \
+--tls_client_key=/home/agent/engflow/engflowpublic.key \
+--remote_execution_priority=-2"
 
 BES_KEYWORDS_ARGS=
 if [ ! -z "${EXTRA_ISSUE_PARAMS:+$EXTRA_ISSUE_PARAMS}" ]
@@ -41,7 +42,7 @@ bazel test //pkg:all_tests $ENGFLOW_FLAGS --remote_download_minimal \
 bazel build //pkg/cmd/bazci/process-bep-file $ENGFLOW_FLAGS --bes_keywords helper-binary
 _bazel/bin/pkg/cmd/bazci/process-bep-file/process-bep-file_/process-bep-file \
     -eventsfile artifacts/eventstream \
-    -cert /home/agent/engflow/engflow.crt -key /home/agent/engflow/engflow.key \
+    -cert /home/agent/engflow/engflowpublic.crt -key /home/agent/engflow/engflowpublic.key \
     -extra "${EXTRA_ISSUE_PARAMS:+$EXTRA_ISSUE_PARAMS}" \
     -jsonoutfile test-results.json
 

pkg/bench/rttanalysis/testdata/benchmark_expectations

@@ -68,7 +68,7 @@ exp,benchmark
 3,Jobs/jobs_page_type_filtered
 1-3,Jobs/jobs_page_type_filtered_no_matches
 4,Jobs/non_admin_crdb_internal.system_jobs
-4,Jobs/non_admin_show_jobs
+2,Jobs/non_admin_show_jobs
 12-15,Jobs/pause_job
 12-15,Jobs/resume_job
 3,Jobs/show_job

pkg/cli/start.go

@@ -1240,7 +1240,7 @@ func reportServerInfo(
 		buf.Printf("external I/O path: \t<disabled>\n")
 	}
 	for i, spec := range serverCfg.Stores.Specs {
-		buf.Printf("store[%d]:\t%s\n", i, log.SafeManaged(spec))
+		buf.Printf("store[%d]:\t%s\n", i, log.SafeManaged(base.StoreSpecCmdLineString(spec)))
 	}
 
 	// Print the commong server identifiers.

pkg/crosscluster/physical/alter_replication_job_test.go

@@ -533,6 +533,10 @@ func TestTenantReplicationStatus(t *testing.T) {
 	_, status, err = getReplicationStatsAndStatus(ctx, registry, nil, jobspb.JobID(producerJobID))
 	require.ErrorContains(t, err, "is not a stream ingestion job")
 	require.Equal(t, "replication error", status)
+	c.DestSysSQL.CheckQueryResults(t, "SELECT count(*) > 0 FROM crdb_internal.cluster_replication_spans", [][]string{{"true"}})
+	c.DestSysSQL.Exec(t, fmt.Sprintf("CREATE USER %s", username.TestUser))
+	noPrivs := sqlutils.MakeSQLRunner(c.DestSysServer.SQLConn(t, serverutils.User(username.TestUser)))
+	noPrivs.CheckQueryResults(t, "SELECT count(*) > 0 FROM crdb_internal.cluster_replication_spans", [][]string{{"false"}})
 }
 
 // TestAlterTenantHandleFutureProtectedTimestamp verifies that cutting over "TO

pkg/server/application_api/jobs_test.go

@@ -355,8 +355,8 @@ func TestAdminAPIJobsDetails(t *testing.T) {
 			t.Fatal(err)
 		}
 		sqlDB.Exec(t,
-			`INSERT INTO system.jobs (id, status, num_runs, last_run, job_type) VALUES ($1, $2, $3, $4, $5)`,
-			job.id, job.status, job.numRuns, job.lastRun, payload.Type().String(),
+			`INSERT INTO system.jobs (id, status, num_runs, last_run, job_type, owner) VALUES ($1, $2, $3, $4, $5, $6)`,
+			job.id, job.status, job.numRuns, job.lastRun, payload.Type().String(), job.username.Normalized(),
 		)
 		sqlDB.Exec(t,
 			`INSERT INTO system.job_info (job_id, info_key, value) VALUES ($1, $2, $3)`,

pkg/sql/crdb_internal.go

@@ -162,7 +162,7 @@ var crdbInternal = virtualSchema{
 		catconstants.CrdbInternalIndexSpansTableID:                  crdbInternalIndexSpansTable,
 		catconstants.CrdbInternalIndexUsageStatisticsTableID:        crdbInternalIndexUsageStatistics,
 		catconstants.CrdbInternalInflightTraceSpanTableID:           crdbInternalInflightTraceSpanTable,
-		catconstants.CrdbInternalJobsTableID:                        crdbInternalJobsTable,
+		catconstants.CrdbInternalJobsTableID:                        crdbInternalJobsView,
 		catconstants.CrdbInternalSystemJobsTableID:                  crdbInternalSystemJobsTable,
 		catconstants.CrdbInternalKVNodeStatusTableID:                crdbInternalKVNodeStatusTable,
 		catconstants.CrdbInternalKVStoreStatusTableID:               crdbInternalKVStoreStatusTable,
@@ -1134,142 +1134,82 @@ func wrapPayloadUnMarshalError(err error, jobID tree.Datum) error {
 		" consider deleting this job from system.jobs", jobID)
 }
 
-const (
-	jobIDFilter      = ` WHERE j.id = $1`
-	jobsStatusFilter = ` WHERE j.status = $1`
-	jobsTypeFilter   = ` WHERE j.job_type = $1`
-)
-
-// TODO(tbg): prefix with kv_.
-var crdbInternalJobsTable = virtualSchemaTable{
-	schema: `
-CREATE TABLE crdb_internal.jobs (
-  job_id                INT,
-  job_type              STRING,
-  description           STRING,
-  statement             STRING,
-  user_name             STRING,
-  status                STRING,
-  running_status        STRING,
-  created               TIMESTAMPTZ,
-  finished              TIMESTAMPTZ,
-  modified              TIMESTAMPTZ,
-  fraction_completed    FLOAT,
-  high_water_timestamp  DECIMAL,
-  error                 STRING,
-  coordinator_id        INT,
-  INDEX(job_id),
-  INDEX(status),
-  INDEX(job_type)
+var crdbInternalJobsView = virtualSchemaView{
+	// TODO(dt): the left-outer joins here in theory mean that if there are more
+	// than one row per job in status or progress the job row would need to be
+	// repeated for each. While this is never the case in practice, it does mean
+	// the optimizer cannot elide the joins even if the progress or status is not
+	// actually being read from a query on the view. We could change the joins to
+	// be `LATERAL (... WHERE job_id = id LiMIT 1) ON true` which would promise to
+	// the optimizer that the join cannot change the number of rows, allowing it
+	// to elide it entirely if the joined columns are not used, however SHOW JOBS
+	// and other callers typically _are_ looking for progress and status so we
+	// likely will need the joins anyway, queries experts expressed some concern
+	// that it could be possible "lateral join prevents some other optimizations".
+	// Given we typically expect to need the joined columns anyway, these are left
+	// as left joins for now. NB: the `union`, needs to be `union all` to avoid
+	// materializing all of the columns even when they're not used, since a
+	// non-`all` union has to be prepared to compare them all to eliminate
+	// duplicates, even when in the vast majority of calls we expect the session
+	// jobs generator to be empty.
+	schema: `CREATE VIEW crdb_internal.jobs (
+  job_id,
+  job_type,
+  description,
+  statement,
+  user_name,
+  status,
+  running_status,
+  created,
+  finished,
+  modified,
+  fraction_completed,
+  high_water_timestamp,
+  error,
+  coordinator_id
+) AS (
+	SELECT
+		j.id,
+		j.job_type,
+		coalesce(j.description, '') as description,
+		coalesce(j.description, '') as statement,
+		coalesce(j.owner, '') as user_name,
+		j.status as status,
+		s.status as running_status,
+		j.created::timestamptz,
+		j.finished,
+		greatest(j.created, j.finished, p.written, s.written)::timestamptz AS modified,
+		p.fraction as fraction_completed,
+		p.resolved as high_water_timestamp,
+		coalesce(j.error_msg, '') as error,
+		j.claim_instance_id as coordinator_id
+	FROM system.public.jobs AS j
+	LEFT OUTER JOIN system.public.job_progress AS p ON j.id = p.job_id
+	LEFT OUTER JOIN system.public.job_status AS s ON j.id = s.job_id
+	WHERE crdb_internal.can_view_job(j.owner)
+	UNION ALL
+		(SELECT job_id, job_type, description, description, user_name, 'pending',
+					NULL, now(), NULL, now(), NULL, NULL, NULL, NULL
+			FROM crdb_internal.session_pending_jobs()
+		)
 )`,
-	comment: `decoded job metadata from crdb_internal.system_jobs (KV scan)`,
-	indexes: []virtualIndex{{
-		populate: func(ctx context.Context, unwrappedConstraint tree.Datum, p *planner, _ catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) (matched bool, err error) {
-			targetID := tree.MustBeDInt(unwrappedConstraint)
-			return makeJobsTableRows(ctx, p, addRow, jobIDFilter, targetID)
-		},
-	}, {
-		populate: func(ctx context.Context, unwrappedConstraint tree.Datum, p *planner, _ catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) (matched bool, err error) {
-			targetStatus := tree.MustBeDString(unwrappedConstraint)
-			return makeJobsTableRows(ctx, p, addRow, jobsStatusFilter, targetStatus)
-		},
-	}, {
-		populate: func(ctx context.Context, unwrappedConstraint tree.Datum, p *planner, _ catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) (matched bool, err error) {
-			targetType := tree.MustBeDString(unwrappedConstraint)
-			return makeJobsTableRows(ctx, p, addRow, jobsTypeFilter, targetType)
-		},
-	}},
-	populate: func(ctx context.Context, p *planner, db catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) error {
-		_, err := makeJobsTableRows(ctx, p, addRow, "")
-		return err
+	resultColumns: colinfo.ResultColumns{
+		{Name: "job_id", Typ: types.Int},
+		{Name: "job_type", Typ: types.String},
+		{Name: "description", Typ: types.String},
+		{Name: "statement", Typ: types.String},
+		{Name: "user_name", Typ: types.String},
+		{Name: "status", Typ: types.String},
+		{Name: "running_status", Typ: types.String},
+		{Name: "created", Typ: types.TimestampTZ},
+		{Name: "finished", Typ: types.TimestampTZ},
+		{Name: "modified", Typ: types.TimestampTZ},
+		{Name: "fraction_completed", Typ: types.Float},
+		{Name: "high_water_timestamp", Typ: types.Decimal},
+		{Name: "error", Typ: types.String},
+		{Name: "coordinator_id", Typ: types.Int},
 	},
-}
-
-func makeJobsTableRows(
-	ctx context.Context,
-	p *planner,
-	addRow func(...tree.Datum) error,
-	whereClause string,
-	params ...interface{},
-) (emitted bool, retErr error) {
-	globalPrivileges, err := jobsauth.GetGlobalJobPrivileges(ctx, p)
-	if err != nil {
-		return false, err
-	}
-
-	query := `SELECT 
-j.id, j.job_type, coalesce(j.description, ''), coalesce(j.owner, ''), j.status as state,
-s.status, j.created::timestamptz, j.finished, greatest(j.created, j.finished, p.written, s.written)::timestamptz AS last_modified,
-p.fraction,
-p.resolved,
-j.error_msg,
-j.claim_instance_id
-FROM system.public.jobs AS j
-LEFT OUTER JOIN system.public.job_progress AS p ON j.id = p.job_id
-LEFT OUTER JOIN system.public.job_status AS s ON j.id = s.job_id
-	` + whereClause + `UNION
-	(SELECT job_id, job_type, description, user_name, 'pending',
-					NULL, now(), NULL, now(), NULL, NULL, NULL, NULL
-	FROM crdb_internal.session_pending_jobs())`
-
-	it, err := p.InternalSQLTxn().QueryIteratorEx(
-		ctx, "system-jobs-join", p.txn, sessiondata.NodeUserSessionDataOverride, query, params...)
-	if err != nil {
-		return emitted, err
-	}
-	defer func() {
-		if err := it.Close(); err != nil {
-			retErr = errors.CombineErrors(retErr, err)
-		}
-	}()
-
-	// Loop while we need to skip a row.
-	for {
-		ok, err := it.Next(ctx)
-		if err != nil || !ok {
-			return emitted, err
-		}
-		r := it.Cur()
-		id, typStr, desc, ownerStr, state, status, created, finished, modified, fraction, resolved, errorMsg, instanceID :=
-			r[0], r[1], r[2], r[3], r[4], r[5], r[6], r[7], r[8], r[9], r[10], r[11], r[12]
-
-		owner := username.MakeSQLUsernameFromPreNormalizedString(string(tree.MustBeDString(ownerStr)))
-		jobID := jobspb.JobID(tree.MustBeDInt(id))
-
-		if errorMsg == tree.DNull {
-			errorMsg = emptyString
-		}
-
-		if err := jobsauth.Authorize(
-			ctx, p, jobID, owner, jobsauth.ViewAccess, globalPrivileges,
-		); err != nil {
-			// Filter out jobs which the user is not allowed to see.
-			if IsInsufficientPrivilegeError(err) {
-				continue
-			}
-			return emitted, err
-		}
-
-		if err = addRow(
-			id,
-			typStr,
-			desc,
-			desc,
-			ownerStr,
-			state,
-			status,
-			created,
-			finished,
-			modified,
-			fraction,
-			resolved,
-			errorMsg,
-			instanceID,
-		); err != nil {
-			return emitted, err
-		}
-		emitted = true
-	}
+	comment: `decoded job metadata from various jobs tables`,
 }
 
 const crdbInternalKVProtectedTSTableQuery = `
@@ -9355,7 +9295,7 @@ var crdbInternalClusterReplicationResolvedView = virtualSchemaView{
 			SELECT
 				j.id AS job_id, jsonb_array_elements(crdb_internal.pb_to_json('progress', i.value)->'streamIngest'->'checkpoint'->'resolvedSpans') AS s
 			FROM system.jobs j LEFT JOIN system.job_info i ON j.id = i.job_id AND i.info_key = 'legacy_progress'
-			WHERE j.job_type = 'REPLICATION STREAM INGESTION'
+			WHERE j.job_type = 'REPLICATION STREAM INGESTION' AND pg_has_role(current_user, 'admin', 'member')
 			) SELECT
 				job_id,
 				crdb_internal.pretty_key(decode(s->'span'->>'key', 'base64'), 0) AS start_key,
@@ -9378,7 +9318,7 @@ var crdbInternalLogicalReplicationResolvedView = virtualSchemaView{
 			SELECT
 				j.id AS job_id, jsonb_array_elements(crdb_internal.pb_to_json('progress', i.value)->'LogicalReplication'->'checkpoint'->'resolvedSpans') AS s
 			FROM system.jobs j LEFT JOIN system.job_info i ON j.id = i.job_id AND i.info_key = 'legacy_progress'
-			WHERE j.job_type = 'LOGICAL REPLICATION'
+			WHERE j.job_type = 'LOGICAL REPLICATION' AND pg_has_role(current_user, 'admin', 'member')
 			) SELECT
 				job_id,
 				crdb_internal.pretty_key(decode(s->'span'->>'key', 'base64'), 0) AS start_key,

pkg/sql/crdb_internal_test.go

@@ -1454,13 +1454,13 @@ func TestInternalSystemJobsAccess(t *testing.T) {
 // worker goroutine returns "query canceled error".
 //
 // In particular, the following setup is used:
-//   - issue SHOW JOBS query which internally issues a query against
-//     crdb_internal.system_jobs virtual table
-//   - that virtual table is generated by issuing "system-jobs-scan" internal
+//   - issue SHOW ZONE CONFIGURATIONS query which internally issues a query against
+//     crdb_internal.zones virtual table
+//   - that virtual table is generated by issuing "crdb-internal-zones-table" internal
 //     query
-//   - during that "system-jobs-scan" query we're injecting the query canceled
+//   - during that "crdb-internal-zones-table" query we're injecting the query canceled
 //     error (in other words, the error is injected during the generation of
-//     crdb_internal.system_jobs virtual table).
+//     crdb_internal.zones virtual table).
 //
 // The injection is achieved by adding a callback to DistSQLReceiver.Push which
 // replaces the first piece of metadata it sees with the error.
@@ -1482,7 +1482,7 @@ func TestVirtualTableDoesntHangOnQueryCanceledError(t *testing.T) {
 							return nil
 						}
 						opName, ok := sql.GetInternalOpName(ctx)
-						if !ok || !(opName == "system-jobs-scan" || opName == "system-jobs-join") {
+						if !ok || !(opName == "crdb-internal-zones-table") {
 							return nil
 						}
 						numCallbacksAdded.Add(1)
@@ -1504,7 +1504,7 @@ func TestVirtualTableDoesntHangOnQueryCanceledError(t *testing.T) {
 	sqlDB := sqlutils.MakeSQLRunner(db)
 
 	addCallback.Store(true)
-	sqlDB.ExpectErr(t, err.Error(), "SHOW JOBS")
+	sqlDB.ExpectErr(t, err.Error(), "SHOW ZONE CONFIGURATIONS")
 	addCallback.Store(false)
 
 	// Sanity check that the callback was added at least once.

pkg/sql/faketreeeval/evalctx.go

@@ -647,6 +647,14 @@ func (ep *DummySessionAccessor) HasViewActivityOrViewActivityRedactedRole(
 	return false, false, errors.WithStack(errEvalSessionVar)
 }
 
+// HasViewAccessToJob implements SessionAccessor.
+func (ep *DummySessionAccessor) HasViewAccessToJob(
+	ctx context.Context, owner username.SQLUsername,
+) bool {
+	// This is a no-op in the dummy implementation.
+	return false
+}
+
 func (ep *DummySessionAccessor) ForEachSessionPendingJob(
 	_ func(job jobspb.PendingJob) error,
 ) error {

pkg/sql/jobs_collection.go

@@ -6,8 +6,12 @@
 package sql
 
 import (
+	"context"
+
 	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobsauth"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 )
 
@@ -100,3 +104,11 @@ func (p *planner) ForEachSessionPendingJob(fn func(job jobspb.PendingJob) error)
 		})
 	})
 }
+
+func (p *planner) HasViewAccessToJob(ctx context.Context, owner username.SQLUsername) bool {
+	privs, err := jobsauth.GetGlobalJobPrivileges(ctx, p)
+	if err != nil {
+		return false
+	}
+	return jobsauth.Authorize(ctx, p, jobspb.InvalidJobID, owner, jobsauth.ViewAccess, privs) == nil
+}

pkg/sql/logictest/testdata/logic_test/select

@@ -831,7 +831,7 @@ statement ok
 SELECT * FROM pg_class
 
 statement ok
-SELECT * FROM crdb_internal.jobs
+SELECT * FROM crdb_internal.node_build_info
 
 # Tests for 'large_full_scan_rows'.
 statement ok