Merge cockroachdb#140952

140952: Revert "jobs: parse internal jobs queries at init time" r=yuzefovich a=msbutler

This reverts commit 28315da.

Epic: None

Co-authored-by: Michael Butler <[email protected]>

Author: craig[bot]

pkg/jobs/BUILD.bazel

@@ -52,8 +52,6 @@ go_library(
         "//pkg/sql/catalog/descpb",
         "//pkg/sql/catalog/descs",
         "//pkg/sql/isql",
-        "//pkg/sql/parser",
-        "//pkg/sql/parser/statements",
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/protoreflect",
         "//pkg/sql/sem/catconstants",

pkg/jobs/job_info_storage.go

@@ -15,8 +15,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
-	"github.com/cockroachdb/cockroach/pkg/sql/parser"
-	"github.com/cockroachdb/cockroach/pkg/sql/parser/statements"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness"
@@ -35,16 +33,6 @@ var (
 	)
 )
 
-// Parse all job storage SQL statements at init time to avoid parsing them on
-// every execution.
-func mustParseOne(s string) statements.Statement[tree.Statement] {
-	stmt, err := parser.ParseOne(s)
-	if err != nil {
-		panic(err)
-	}
-	return stmt
-}
-
 // ProgressStorage reads and writes progress rows.
 type ProgressStorage jobspb.JobID
 
@@ -53,10 +41,6 @@ func (j *Job) ProgressStorage() ProgressStorage {
 	return ProgressStorage(j.id)
 }
 
-var getJobProgressQuery = mustParseOne(
-	"SELECT written, fraction, resolved FROM system.job_progress WHERE job_id = $1",
-)
-
 // Get returns the latest progress report for the job along with when it was
 // written. If the fraction is null it is returned as NaN, and if the resolved
 // ts is null, it is empty.
@@ -66,10 +50,10 @@ func (i ProgressStorage) Get(
 	ctx, sp := tracing.ChildSpan(ctx, "get-job-progress")
 	defer sp.Finish()
 
-	row, err := txn.QueryRowExParsed(
+	row, err := txn.QueryRowEx(
 		ctx, "job-progress-get", txn.KV(),
 		sessiondata.NodeUserSessionDataOverride,
-		getJobProgressQuery, i,
+		"SELECT written, fraction, resolved FROM system.job_progress WHERE job_id = $1", i,
 	)
 
 	if err != nil || row == nil {
@@ -107,22 +91,6 @@ func (i ProgressStorage) Get(
 	return fraction, ts, written.Time, nil
 }
 
-var (
-	deleteJobProgressStmt = mustParseOne(
-		`DELETE FROM system.job_progress WHERE job_id = $1`,
-	)
-	insertJobProgressStmt = mustParseOne(
-		`INSERT INTO system.job_progress (job_id, written, fraction, resolved) VALUES ($1, now(), $2, $3)`,
-	)
-	insertJobProgressHistoryStmt = mustParseOne(
-		`UPSERT INTO system.job_progress_history (job_id, written, fraction, resolved) VALUES ($1, now(), $2, $3)`,
-	)
-	pruneJobProgressHistoryStmt = mustParseOne(
-		`DELETE FROM system.job_progress_history WHERE job_id = $1 AND written IN 
-			(SELECT written FROM system.job_progress_history WHERE job_id = $1 ORDER BY written DESC OFFSET $2)`,
-	)
-)
-
 // Set records a progress update. If fraction is NaN or resolved is empty, that
 // field is left null. The time at which the progress was reported is recorded.
 func (i ProgressStorage) Set(
@@ -131,9 +99,9 @@ func (i ProgressStorage) Set(
 	ctx, sp := tracing.ChildSpan(ctx, "write-job-progress")
 	defer sp.Finish()
 
-	if _, err := txn.ExecParsed(
+	if _, err := txn.ExecEx(
 		ctx, "write-job-progress-delete", txn.KV(), sessiondata.NodeUserSessionDataOverride,
-		deleteJobProgressStmt, i,
+		`DELETE FROM system.job_progress WHERE job_id = $1`, i,
 	); err != nil {
 		return err
 	}
@@ -146,25 +114,26 @@ func (i ProgressStorage) Set(
 		ts = resolved.AsOfSystemTime()
 	}
 
-	if _, err := txn.ExecParsed(
+	if _, err := txn.ExecEx(
 		ctx, "write-job-progress-insert", txn.KV(), sessiondata.NodeUserSessionDataOverride,
-		insertJobProgressStmt,
+		`INSERT INTO system.job_progress (job_id, written, fraction, resolved) VALUES ($1, now(), $2, $3)`,
 		i, frac, ts,
 	); err != nil {
 		return err
 	}
 
-	if _, err := txn.ExecParsed(
+	if _, err := txn.ExecEx(
 		ctx, "write-job-progress-history-insert", txn.KV(), sessiondata.NodeUserSessionDataOverride,
-		insertJobProgressHistoryStmt,
+		`UPSERT INTO system.job_progress_history (job_id, written, fraction, resolved) VALUES ($1, now(), $2, $3)`,
 		i, frac, ts,
 	); err != nil {
 		return err
 	}
 
-	if _, err := txn.ExecParsed(
+	if _, err := txn.ExecEx(
 		ctx, "write-job-progress-history-prune", txn.KV(), sessiondata.NodeUserSessionDataOverride,
-		pruneJobProgressHistoryStmt,
+		`DELETE FROM system.job_progress_history WHERE job_id = $1 AND written IN 
+			(SELECT written FROM system.job_progress_history WHERE job_id = $1 ORDER BY written DESC OFFSET $2)`,
 		i, retainedProgressHistory.Get(txn.KV().DB().SettingsValues()),
 	); err != nil {
 		return err
@@ -196,23 +165,15 @@ func (j *Job) StatusStorage() StatusStorage {
 	return StatusStorage(j.id)
 }
 
-var deleteJobStatusStmt = mustParseOne(
-	`DELETE FROM system.job_status WHERE job_id = $1`,
-)
-
 // Clear clears the status message row for the job, if it exists.
 func (i StatusStorage) Clear(ctx context.Context, txn isql.Txn) error {
-	_, err := txn.ExecParsed(
+	_, err := txn.ExecEx(
 		ctx, "clear-job-status-delete", txn.KV(), sessiondata.NodeUserSessionDataOverride,
-		deleteJobStatusStmt, i,
+		`DELETE FROM system.job_status WHERE job_id = $1`, i,
 	)
 	return err
 }
 
-var insertJobStatusStmt = mustParseOne(
-	`INSERT INTO system.job_status (job_id, written, status) VALUES ($1, now(), $2)`,
-)
-
 // Sets writes the current status, replacing the current one if it exists.
 // Setting an empty status is the same as calling Clear().
 func (i StatusStorage) Set(ctx context.Context, txn isql.Txn, status string) error {
@@ -225,9 +186,9 @@ func (i StatusStorage) Set(ctx context.Context, txn isql.Txn, status string) err
 		return err
 	}
 
-	if _, err := txn.ExecParsed(
+	if _, err := txn.ExecEx(
 		ctx, "write-job-status-insert", txn.KV(), sessiondata.NodeUserSessionDataOverride,
-		insertJobStatusStmt,
+		`INSERT INTO system.job_status (job_id, written, status) VALUES ($1, now(), $2)`,
 		i, status,
 	); err != nil {
 		return err
@@ -240,18 +201,14 @@ func (i StatusStorage) Set(ctx context.Context, txn isql.Txn, status string) err
 	return nil
 }
 
-var getJobStatusQuery = mustParseOne(
-	"SELECT written, status FROM system.job_status WHERE job_id = $1",
-)
-
 // Get gets the current status mesasge for a job, if any.
 func (i StatusStorage) Get(ctx context.Context, txn isql.Txn) (string, time.Time, error) {
 	ctx, sp := tracing.ChildSpan(ctx, "get-job-status")
 	defer sp.Finish()
 
-	row, err := txn.QueryRowExParsed(
+	row, err := txn.QueryRowEx(
 		ctx, "job-status-get", txn.KV(), sessiondata.NodeUserSessionDataOverride,
-		getJobStatusQuery,
+		"SELECT written, status FROM system.job_status WHERE job_id = $1",
 		i,
 	)
 
@@ -292,17 +249,6 @@ func (j *Job) Messages() MessageStorage {
 	return MessageStorage(j.id)
 }
 
-var (
-	upsertJobMessageStmt = mustParseOne(
-		`UPSERT INTO system.job_message (job_id, written, kind, message) VALUES ($1, now(),$2, $3)`,
-	)
-	pruneJobMessageStmt = mustParseOne(
-		`DELETE FROM system.job_message WHERE job_id = $1 AND kind = $2 AND written IN (
-			SELECT written FROM system.job_message WHERE job_id = $1 AND kind = $2 ORDER BY written DESC OFFSET $3
-		)`,
-	)
-)
-
 // Record writes a human readable message of the specified kind to the message
 // log for this job, and prunes retained messages of the same kind based on the
 // configured limit to keep the total number of retained messages bounded.
@@ -311,18 +257,20 @@ func (i MessageStorage) Record(ctx context.Context, txn isql.Txn, kind, message
 	defer sp.Finish()
 
 	// Insert the new message.
-	if _, err := txn.ExecParsed(
+	if _, err := txn.ExecEx(
 		ctx, "write-job-message-insert", txn.KV(), sessiondata.NodeUserSessionDataOverride,
-		upsertJobMessageStmt,
+		`UPSERT INTO system.job_message (job_id, written, kind, message) VALUES ($1, now(),$2, $3)`,
 		i, kind, message,
 	); err != nil {
 		return err
 	}
 
 	// Prune old messages of the same kind to bound historical data.
-	if _, err := txn.ExecParsed(
+	if _, err := txn.ExecEx(
 		ctx, "write-job-message-prune", txn.KV(), sessiondata.NodeUserSessionDataOverride,
-		pruneJobMessageStmt,
+		`DELETE FROM system.job_message WHERE job_id = $1 AND kind = $2 AND written IN (
+			SELECT written FROM system.job_message WHERE job_id = $1 AND kind = $2 ORDER BY written DESC OFFSET $3
+		)`,
 		i, kind, retainedMessageHistory.Get(txn.KV().DB().SettingsValues()),
 	); err != nil {
 		return err
@@ -404,18 +352,14 @@ func InfoStorageForJob(txn isql.Txn, jobID jobspb.JobID) InfoStorage {
 	return InfoStorage{j: &Job{id: jobID}, txn: txn}
 }
 
-var checkClaimSessionQuery = mustParseOne(
-	`SELECT claim_session_id FROM system.jobs WHERE id = $1`,
-)
-
 func (i *InfoStorage) checkClaimSession(ctx context.Context) error {
 	if i.claimChecked {
 		return nil
 	}
 
-	row, err := i.txn.QueryRowExParsed(ctx, "check-claim-session", i.txn.KV(),
+	row, err := i.txn.QueryRowEx(ctx, "check-claim-session", i.txn.KV(),
 		sessiondata.NodeUserSessionDataOverride,
-		checkClaimSessionQuery, i.j.ID())
+		`SELECT claim_session_id FROM system.jobs WHERE id = $1`, i.j.ID())
 	if err != nil {
 		return err
 	}
@@ -435,10 +379,6 @@ func (i *InfoStorage) checkClaimSession(ctx context.Context) error {
 	return nil
 }
 
-var getJobInfoQuery = mustParseOne(
-	"SELECT value FROM system.job_info WHERE job_id = $1 AND info_key::string = $2 ORDER BY written DESC LIMIT 1",
-)
-
 func (i InfoStorage) get(ctx context.Context, opName, infoKey string) ([]byte, bool, error) {
 	if i.txn == nil {
 		return nil, false, errors.New("cannot access the job info table without an associated txn")
@@ -452,10 +392,10 @@ func (i InfoStorage) get(ctx context.Context, opName, infoKey string) ([]byte, b
 	// We expect there to be only a single row for a given <job_id, info_key>.
 	// This is because all older revisions are deleted before a new one is
 	// inserted in `InfoStorage.Write`.
-	row, err := i.txn.QueryRowExParsed(
+	row, err := i.txn.QueryRowEx(
 		ctx, "job-info-get", i.txn.KV(),
 		sessiondata.NodeUserSessionDataOverride,
-		getJobInfoQuery,
+		"SELECT value FROM system.job_info WHERE job_id = $1 AND info_key::string = $2 ORDER BY written DESC LIMIT 1",
 		j.ID(), infoKey,
 	)
 
@@ -475,22 +415,13 @@ func (i InfoStorage) get(ctx context.Context, opName, infoKey string) ([]byte, b
 	return []byte(*value), true, nil
 }
 
-var (
-	deleteJobInfoStmt = mustParseOne(
-		`DELETE FROM system.job_info WHERE job_id = $1 AND info_key::string = $2`,
-	)
-	insertJobInfoStmt = mustParseOne(
-		`INSERT INTO system.job_info (job_id, info_key, written, value) VALUES ($1, $2, now(), $3)`,
-	)
-)
-
 func (i InfoStorage) write(ctx context.Context, infoKey string, value []byte) error {
 	return i.doWrite(ctx, func(ctx context.Context, j *Job, txn isql.Txn) error {
 		// First clear out any older revisions of this info.
-		_, err := txn.ExecParsed(
+		_, err := txn.ExecEx(
 			ctx, "write-job-info-delete", txn.KV(),
 			sessiondata.NodeUserSessionDataOverride,
-			deleteJobInfoStmt,
+			"DELETE FROM system.job_info WHERE job_id = $1 AND info_key::string = $2",
 			j.ID(), infoKey,
 		)
 		if err != nil {
@@ -502,10 +433,10 @@ func (i InfoStorage) write(ctx context.Context, infoKey string, value []byte) er
 			return nil
 		}
 		// Write the new info, using the same transaction.
-		_, err = txn.ExecParsed(
+		_, err = txn.ExecEx(
 			ctx, "write-job-info-insert", txn.KV(),
 			sessiondata.NodeUserSessionDataOverride,
-			insertJobInfoStmt,
+			`INSERT INTO system.job_info (job_id, info_key, written, value) VALUES ($1, $2, now(), $3)`,
 			j.ID(), infoKey, value,
 		)
 		return err
@@ -625,46 +556,33 @@ func (i InfoStorage) Delete(ctx context.Context, infoKey string) error {
 	return i.write(ctx, infoKey, nil /* value */)
 }
 
-var (
-	deleteJobInfoRangeLimitStmt = mustParseOne(
-		`DELETE FROM system.job_info WHERE job_id = $1 AND info_key >= $2 AND info_key < $3
-			ORDER BY info_key ASC LIMIT $4`,
-	)
-	deleteJobInfoRangeStmt = mustParseOne(
-		"DELETE FROM system.job_info WHERE job_id = $1 AND info_key >= $2 AND info_key < $3",
-	)
-)
-
 // DeleteRange removes the info records between the provided
 // start key (inclusive) and end key (exclusive).
 func (i InfoStorage) DeleteRange(
 	ctx context.Context, startInfoKey, endInfoKey string, limit int,
 ) error {
 	return i.doWrite(ctx, func(ctx context.Context, j *Job, txn isql.Txn) error {
 		if limit > 0 {
-			_, err := txn.ExecParsed(
-				ctx, "write-job-info-delete-limit", txn.KV(),
+			_, err := txn.ExecEx(
+				ctx, "write-job-info-delete", txn.KV(),
 				sessiondata.NodeUserSessionDataOverride,
-				deleteJobInfoRangeLimitStmt,
+				"DELETE FROM system.job_info WHERE job_id = $1 AND info_key >= $2 AND info_key < $3 "+
+					"ORDER BY info_key ASC LIMIT $4",
 				j.ID(), startInfoKey, endInfoKey, limit,
 			)
 			return err
 		} else {
-			_, err := txn.ExecParsed(
+			_, err := txn.ExecEx(
 				ctx, "write-job-info-delete", txn.KV(),
 				sessiondata.NodeUserSessionDataOverride,
-				deleteJobInfoRangeStmt,
+				"DELETE FROM system.job_info WHERE job_id = $1 AND info_key >= $2 AND info_key < $3",
 				j.ID(), startInfoKey, endInfoKey,
 			)
 			return err
 		}
 	})
 }
 
-var getJobInfoCountQuery = mustParseOne(
-	"SELECT count(*) FROM system.job_info WHERE job_id = $1 AND info_key >= $2 AND info_key < $3",
-)
-
 // Count counts the info records in the range [start, end).
 func (i InfoStorage) Count(ctx context.Context, startInfoKey, endInfoKey string) (int, error) {
 	if i.txn == nil {
@@ -674,10 +592,10 @@ func (i InfoStorage) Count(ctx context.Context, startInfoKey, endInfoKey string)
 	ctx, sp := tracing.ChildSpan(ctx, "count-job-info")
 	defer sp.Finish()
 
-	row, err := i.txn.QueryRowExParsed(
+	row, err := i.txn.QueryRowEx(
 		ctx, "job-info-count", i.txn.KV(),
 		sessiondata.NodeUserSessionDataOverride,
-		getJobInfoCountQuery,
+		"SELECT count(*) FROM system.job_info WHERE job_id = $1 AND info_key >= $2 AND info_key < $3",
 		i.j.ID(), startInfoKey, endInfoKey,
 	)