Merge #148400 #148420 #148441

148400: importer: purge all PGDUMP and MYSQLDUMP related code r=yuzefovich a=yuzefovich

All this code is no longer used. Note that there are still some more things to clean up (like we can assume that no tables are new, and we don't need to support creating new schemas in IMPORT either), but all those things will be addressed separately. This commit focuses only on removing the related workload readers as well as the roachpb formats.

Epic: None
Release note: None

148420: execinfrapb: minor cleanup of Expression r=yuzefovich a=yuzefovich

Removed unused version field and fix up a comment.

Epic: None
Release note: None

148441: jobs: delete dead code r=dt a=dt



Co-authored-by: Yahor Yuzefovich <[email protected]>
Co-authored-by: David Taylor <[email protected]>

Author: 3 people

DEPS.bzl

@@ -262,7 +262,6 @@ require (
 	gopkg.in/yaml.v2 v2.4.0
 	gopkg.in/yaml.v3 v3.0.1
 	honnef.co/go/tools v0.5.1
-	vitess.io/vitess v0.0.0-00010101000000-000000000000
 )
 
 require (
@@ -494,8 +493,6 @@ replace github.com/olekukonko/tablewriter => github.com/cockroachdb/tablewriter
 
 replace github.com/abourget/teamcity => github.com/cockroachdb/teamcity v0.0.0-20180905144921-8ca25c33eb11
 
-replace vitess.io/vitess => github.com/cockroachdb/vitess v0.0.0-20210218160543-54524729cc82
-
 replace gopkg.in/yaml.v2 => github.com/cockroachdb/yaml v0.0.0-20210825132133-2d6955c8edbc
 
 replace github.com/docker/docker => github.com/moby/moby v24.0.6+incompatible

build/bazelutil/distdir_files.bzl

@@ -985,9 +985,8 @@ func createImportingDescriptors(
 	for _, desc := range sqlDescs {
 		// Decide which offline tables to include in the restore:
 		//
-		// - An offline table created by RESTORE or IMPORT PGDUMP is
-		//   fully discarded.  The table will not exist in the restoring
-		//   cluster.
+		// - An offline table created by RESTORE is fully discarded. The table
+		//   will not exist in the restoring cluster.
 		//
 		// - An offline table undergoing an IMPORT INTO in traditional
 		//   restore has all importing data elided in the restore

go.mod

@@ -15,14 +15,6 @@ func TestGetExistingMirrorsFromDepsBzl(t *testing.T) {
 	depsbzl := `# leading comment
 load("@bazel_gazelle//:deps.bzl", "go_repository")
 def go_deps():
-    go_repository(
-        name = "io_vitess_vitess",
-        build_file_proto_mode = "disable_global",
-        importpath = "vitess.io/vitess",
-        sha256 = "FAKESHA256",
-        strip_prefix = "github.com/cockroachdb/[email protected]",
-        urls = ["https://example.com/fakeurl"],
-    )
     go_repository(
         name = "com_github_akavel_rsrc",
         build_file_proto_mode = "disable_global",
@@ -41,11 +33,8 @@ def go_deps():
 `
 	mirrors, err := downloadableArtifactsFromDepsBzl(depsbzl)
 	require.NoError(t, err)
-	require.Equal(t, len(mirrors), 2)
-	mirror := mirrors["io_vitess_vitess"]
-	require.Equal(t, mirror.URL, "https://example.com/fakeurl")
-	require.Equal(t, mirror.Sha256, "FAKESHA256")
-	mirror = mirrors["com_github_alecthomas_units"]
+	require.Equal(t, len(mirrors), 1)
+	mirror := mirrors["com_github_alecthomas_units"]
 	require.Equal(t, mirror.URL, "https://foo/bar.zip")
 	require.Equal(t, mirror.Sha256, "abcdefghij")
 }

go.sum

@@ -187,9 +187,6 @@ const (
 
 	TODO_Delete_V25_1_Start
 
-	// TODO_Delete_V25_1_AddJobsTables added new jobs tables.
-	TODO_Delete_V25_1_AddJobsTables
-
 	// TODO_Delete_V25_1_AddRangeForceFlushKey adds the RangeForceFlushKey, a replicated
 	// range-ID local key, which is written below raft.
 	TODO_Delete_V25_1_AddRangeForceFlushKey
@@ -203,19 +200,6 @@ const (
 	// that are part of the XA two-phase commit protocol.
 	TODO_Delete_V25_1_PreparedTransactionsTable
 
-	// TODO_Delete_V25_1_AddJobsColumns added new columns to system.jobs.
-	TODO_Delete_V25_1_AddJobsColumns
-
-	// TODO_Delete_V25_1_JobsWritesFence is an empty version that is used to add a "fence"
-	// between the column addition version and the backfill version. This allows
-	// the backfill version's upgrade step to make the assumption that all nodes
-	// will be writing to the new columns, since moving from fence to backfill can
-	// only start once no nodes are still on add-columnns.
-	TODO_Delete_V25_1_JobsWritesFence
-
-	// TODO_Delete_V25_1_JobsBackfill backfills the new jobs tables and columns.
-	TODO_Delete_V25_1_JobsBackfill
-
 	// V25_1 is CockroachDB v25.1. It's used for all v25.1.x patch releases.
 	V25_1
 
@@ -271,13 +255,9 @@ var versionTable = [numKeys]roachpb.Version{
 	// v25.1 versions. Internal versions must be even.
 	TODO_Delete_V25_1_Start: {Major: 24, Minor: 3, Internal: 2},
 
-	TODO_Delete_V25_1_AddJobsTables:             {Major: 24, Minor: 3, Internal: 4},
 	TODO_Delete_V25_1_AddRangeForceFlushKey:     {Major: 24, Minor: 3, Internal: 8},
 	TODO_Delete_V25_1_BatchStreamRPC:            {Major: 24, Minor: 3, Internal: 10},
 	TODO_Delete_V25_1_PreparedTransactionsTable: {Major: 24, Minor: 3, Internal: 12},
-	TODO_Delete_V25_1_AddJobsColumns:            {Major: 24, Minor: 3, Internal: 14},
-	TODO_Delete_V25_1_JobsWritesFence:           {Major: 24, Minor: 3, Internal: 16},
-	TODO_Delete_V25_1_JobsBackfill:              {Major: 24, Minor: 3, Internal: 18},
 
 	V25_1: {Major: 25, Minor: 1, Internal: 0},
 

pkg/backup/restore_job.go

@@ -17,8 +17,6 @@ import (
 	"github.com/cockroachdb/errors"
 )
 
-// Modified from importer/read_import_pgdump.go.
-
 // Stream streams an io.Reader into tree.Statements.
 type Stream struct {
 	scan *bufio.Scanner

pkg/build/starlarkutil/starlarkutil_test.go

@@ -483,7 +483,7 @@ func batchJobInsertStmt(
 		return "", nil, nil, errors.NewAssertionErrorWithWrappedErrf(err, "failed to make timestamp for creation of job")
 	}
 	instanceID := r.ID()
-	columns := []string{`id`, `created`, `status`, `claim_session_id`, `claim_instance_id`, `job_type`}
+	columns := []string{`id`, `created`, `status`, `claim_session_id`, `claim_instance_id`, `job_type`, `owner`, `description`}
 	valueFns := map[string]func(*Job) (interface{}, error){
 		`id`:                func(job *Job) (interface{}, error) { return job.ID(), nil },
 		`created`:           func(job *Job) (interface{}, error) { return created, nil },
@@ -494,12 +494,8 @@ func batchJobInsertStmt(
 			payload := job.Payload()
 			return payload.Type().String(), nil
 		},
-	}
-
-	if schemaVersion.AtLeast(clusterversion.TODO_Delete_V25_1_AddJobsColumns.Version()) {
-		columns = append(columns, `owner`, `description`)
-		valueFns[`owner`] = func(job *Job) (interface{}, error) { return job.Payload().UsernameProto.Decode().Normalized(), nil }
-		valueFns[`description`] = func(job *Job) (interface{}, error) { return job.Payload().Description, nil }
+		`owner`:       func(job *Job) (interface{}, error) { return job.Payload().UsernameProto.Decode().Normalized(), nil },
+		`description`: func(job *Job) (interface{}, error) { return job.Payload().Description, nil },
 	}
 
 	appendValues := func(job *Job, vals *[]interface{}) (err error) {
@@ -590,15 +586,9 @@ func (r *Registry) CreateJobWithTxn(
 			return errors.NewAssertionErrorWithWrappedErrf(err, "failed to construct job created timestamp")
 		}
 
-		cols := []string{"id", "created", "status", "claim_session_id", "claim_instance_id", "job_type"}
-		vals := []interface{}{jobID, created, StateRunning, s.ID().UnsafeBytes(), r.ID(), jobType.String()}
-		v, err := txn.GetSystemSchemaVersion(ctx)
-		if err != nil {
-			return err
-		}
-		if v.AtLeast(clusterversion.TODO_Delete_V25_1_AddJobsColumns.Version()) {
-			cols = append(cols, "owner", "description")
-			vals = append(vals, j.mu.payload.UsernameProto.Decode().Normalized(), j.mu.payload.Description)
+		cols := []string{"id", "created", "status", "claim_session_id", "claim_instance_id", "job_type", "owner", "description"}
+		vals := []interface{}{
+			jobID, created, StateRunning, s.ID().UnsafeBytes(), r.ID(), jobType.String(), j.mu.payload.UsernameProto.Decode().Normalized(), j.mu.payload.Description,
 		}
 
 		totalNumCols := len(cols)
@@ -730,18 +720,9 @@ func (r *Registry) CreateAdoptableJobWithTxn(
 		}
 		typ := j.mu.payload.Type().String()
 
-		cols := []string{"id", "created", "status", "created_by_type", "created_by_id", "job_type"}
-		placeholders := []string{"$1", "now() at time zone 'utc'", "$2", "$3", "$4", "$5"}
-		vals := []interface{}{jobID, StateRunning, createdByType, createdByID, typ}
-		v, err := txn.GetSystemSchemaVersion(ctx)
-		if err != nil {
-			return err
-		}
-		if v.AtLeast(clusterversion.TODO_Delete_V25_1_AddJobsColumns.Version()) {
-			cols = append(cols, "owner", "description")
-			placeholders = append(placeholders, "$6", "$7")
-			vals = append(vals, j.mu.payload.UsernameProto.Decode().Normalized(), j.mu.payload.Description)
-		}
+		cols := []string{"id", "created", "status", "created_by_type", "created_by_id", "job_type", "owner", "description"}
+		placeholders := []string{"$1", "now() at time zone 'utc'", "$2", "$3", "$4", "$5", "$6", "$7"}
+		vals := []interface{}{jobID, StateRunning, createdByType, createdByID, typ, j.mu.payload.UsernameProto.Decode().Normalized(), j.mu.payload.Description}
 
 		// Insert the job row, but do not set a `claim_session_id`. By not
 		// setting the claim, the job can be adopted by any node and will
@@ -1275,20 +1256,11 @@ func (r *Registry) cleanupOldJobsPage(
 		}
 
 		counts := make(map[string]int)
-		for i, tbl := range jobMetadataTables {
+		for _, tbl := range jobMetadataTables {
 			var deleted int
 			if err := r.db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 				// Tables other than job_info -- the 0th -- are only present if the txn is
 				// running at a version that includes them.
-				if i > 0 {
-					v, err := txn.GetSystemSchemaVersion(ctx)
-					if err != nil {
-						return err
-					}
-					if v.Less(clusterversion.TODO_Delete_V25_1_AddJobsTables.Version()) {
-						return nil
-					}
-				}
 				deleted, err = txn.Exec(ctx, redact.RedactableString("gc-job-"+tbl), txn.KV(),
 					"DELETE FROM system."+tbl+" WHERE job_id = ANY($1)", toDelete,
 				)
@@ -1337,17 +1309,7 @@ func (r *Registry) DeleteTerminalJobByID(ctx context.Context, id jobspb.JobID) e
 			if err != nil {
 				return err
 			}
-			for i, tbl := range jobMetadataTables {
-				if i > 0 {
-					v, err := txn.GetSystemSchemaVersion(ctx)
-					if err != nil {
-						return err
-					}
-					if v.Less(clusterversion.TODO_Delete_V25_1_AddJobsTables.Version()) {
-						break
-					}
-				}
-
+			for _, tbl := range jobMetadataTables {
 				_, err = txn.Exec(
 					ctx, redact.RedactableString("delete-job-"+tbl), txn.KV(),
 					"DELETE FROM system."+tbl+" WHERE job_id = $1", id,

pkg/clusterversion/cockroach_versions.go

@@ -12,7 +12,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -281,96 +280,88 @@ WHERE id = $1
 		}
 	}
 
-	v, err := u.txn.GetSystemSchemaVersion(ctx)
-	if err != nil {
-		return err
-	}
-	if v.AtLeast(clusterversion.TODO_Delete_V25_1_AddJobsTables.Version()) {
-		if ju.md.State != "" && ju.md.State != state {
-			if err := j.Messages().Record(ctx, u.txn, "state", string(ju.md.State)); err != nil {
+	if ju.md.State != "" && ju.md.State != state {
+		if err := j.Messages().Record(ctx, u.txn, "state", string(ju.md.State)); err != nil {
+			return err
+		}
+		// If we are changing state, we should clear out the status, unless
+		// we are about to set it to something instead.
+		if progress == nil || progress.StatusMessage == "" {
+			if err := j.StatusStorage().Clear(ctx, u.txn); err != nil {
 				return err
 			}
-			// If we are changing state, we should clear out the status, unless
-			// we are about to set it to something instead.
-			if progress == nil || progress.StatusMessage == "" {
-				if err := j.StatusStorage().Clear(ctx, u.txn); err != nil {
-					return err
-				}
-			}
 		}
+	}
 
-		if progress != nil {
-			var ts hlc.Timestamp
-			if hwm := progress.GetHighWater(); hwm != nil {
-				ts = *hwm
-			}
+	if progress != nil {
+		var ts hlc.Timestamp
+		if hwm := progress.GetHighWater(); hwm != nil {
+			ts = *hwm
+		}
 
-			if err := j.ProgressStorage().Set(ctx, u.txn, float64(progress.GetFractionCompleted()), ts); err != nil {
-				return err
-			}
+		if err := j.ProgressStorage().Set(ctx, u.txn, float64(progress.GetFractionCompleted()), ts); err != nil {
+			return err
+		}
 
-			if progress.StatusMessage != beforeProgress.StatusMessage {
-				if err := j.StatusStorage().Set(ctx, u.txn, progress.StatusMessage); err != nil {
-					return err
-				}
+		if progress.StatusMessage != beforeProgress.StatusMessage {
+			if err := j.StatusStorage().Set(ctx, u.txn, progress.StatusMessage); err != nil {
+				return err
 			}
+		}
 
-			if progress.TraceID != beforeProgress.TraceID {
-				if err := j.Messages().Record(ctx, u.txn, "trace-id", fmt.Sprintf("%d", progress.TraceID)); err != nil {
-					return err
-				}
+		if progress.TraceID != beforeProgress.TraceID {
+			if err := j.Messages().Record(ctx, u.txn, "trace-id", fmt.Sprintf("%d", progress.TraceID)); err != nil {
+				return err
 			}
 		}
 	}
-	if v.AtLeast(clusterversion.TODO_Delete_V25_1_AddJobsColumns.Version()) {
 
-		vals := []interface{}{j.ID()}
+	vals := []interface{}{j.ID()}
 
-		var update strings.Builder
+	var update strings.Builder
 
-		if payloadBytes != nil {
-			if beforePayload.Description != payload.Description {
-				if update.Len() > 0 {
-					update.WriteString(", ")
-				}
-				vals = append(vals, payload.Description)
-				fmt.Fprintf(&update, "description = $%d", len(vals))
+	if payloadBytes != nil {
+		if beforePayload.Description != payload.Description {
+			if update.Len() > 0 {
+				update.WriteString(", ")
 			}
+			vals = append(vals, payload.Description)
+			fmt.Fprintf(&update, "description = $%d", len(vals))
+		}
 
-			if beforePayload.UsernameProto.Decode() != payload.UsernameProto.Decode() {
-				if update.Len() > 0 {
-					update.WriteString(", ")
-				}
-				vals = append(vals, payload.UsernameProto.Decode().Normalized())
-				fmt.Fprintf(&update, "owner = $%d", len(vals))
+		if beforePayload.UsernameProto.Decode() != payload.UsernameProto.Decode() {
+			if update.Len() > 0 {
+				update.WriteString(", ")
 			}
+			vals = append(vals, payload.UsernameProto.Decode().Normalized())
+			fmt.Fprintf(&update, "owner = $%d", len(vals))
+		}
 
-			if beforePayload.Error != payload.Error {
-				if update.Len() > 0 {
-					update.WriteString(", ")
-				}
-				vals = append(vals, payload.Error)
-				fmt.Fprintf(&update, "error_msg = $%d", len(vals))
+		if beforePayload.Error != payload.Error {
+			if update.Len() > 0 {
+				update.WriteString(", ")
 			}
+			vals = append(vals, payload.Error)
+			fmt.Fprintf(&update, "error_msg = $%d", len(vals))
+		}
 
-			if beforePayload.FinishedMicros != payload.FinishedMicros {
-				if update.Len() > 0 {
-					update.WriteString(", ")
-				}
-				vals = append(vals, time.UnixMicro(payload.FinishedMicros))
-				fmt.Fprintf(&update, "finished = $%d", len(vals))
+		if beforePayload.FinishedMicros != payload.FinishedMicros {
+			if update.Len() > 0 {
+				update.WriteString(", ")
 			}
-
+			vals = append(vals, time.UnixMicro(payload.FinishedMicros))
+			fmt.Fprintf(&update, "finished = $%d", len(vals))
 		}
-		if len(vals) > 1 {
-			stmt := fmt.Sprintf("UPDATE system.jobs SET %s WHERE id = $1", update.String())
-			if _, err := u.txn.ExecEx(
-				ctx, "job-update-row", u.txn.KV(),
-				sessiondata.NodeUserSessionDataOverride,
-				stmt, vals...,
-			); err != nil {
-				return err
-			}
+
+	}
+	if len(vals) > 1 {
+		stmt := fmt.Sprintf("UPDATE system.jobs SET %s WHERE id = $1", update.String())
+		if _, err := u.txn.ExecEx(
+			ctx, "job-update-row", u.txn.KV(),
+			sessiondata.NodeUserSessionDataOverride,
+			stmt, vals...,
+		); err != nil {
+			return err
 		}
 	}