sql: don't test error identities using reference equality

Prior to this patch, the SQL code was expecting to identify
errors by addess, ie. something like:

```go
// at global scope
var errSomething = errors.New(...)

// somewhere in the code
   if err == errSomething { ... }
```

This approach was flawed in that errors can be wrapped (`errors.Wrap`,
`errors.WithMessage`) or flattened (`pgerror.Wrap`) or sent over the
wire.

To make the code more robust, this patch introduces a new API
`pgerror.WithMarker()` which instantiates errors with a unique
marker. The marker is guaranteed to be preserved through `errors.Wrap`
and `pgerror.Wrap` calls. It can be tested with
`pgerror.IsMarkedError()`.

The patch then updates the various points in the SQL code where errors
were tested in that way, to use the new API.

(In some places some attempt is made to "work around" with a string
comparison on the message instead, but that's arguably even worse for
reasons left as an exercise to the reader. This should also be fixed,
but that fix is out of scope here.)

Release note: None

pkg/sql/crdb_internal.go

@@ -1844,7 +1844,7 @@ CREATE TABLE crdb_internal.zones (
 				for _, s := range subzones {
 					index, err := table.FindIndexByID(sqlbase.IndexID(s.IndexID))
 					if err != nil {
-						if err == sqlbase.ErrIndexGCMutationsList {
+						if pgerror.IsMarkedError(err, sqlbase.ErrIndexGCMutationsList) {
 							continue
 						}
 						return err

pkg/sql/exec_util.go

@@ -976,7 +976,7 @@ func (scc *schemaChangerCollection) execSchemaChanges(
 					log.Warningf(ctx, "error executing schema change: %s", err)
 				}
 
-				if err == sqlbase.ErrDescriptorNotFound || err == ctx.Err() {
+				if pgerror.IsMarkedError(err, sqlbase.ErrDescriptorNotFound) || err == ctx.Err() {
 					// 1. If the descriptor is dropped while the schema change
 					// is executing, the schema change is considered completed.
 					// 2. If the context is canceled the schema changer quits here

pkg/sql/lease.go

@@ -46,8 +46,8 @@ import (
 	"github.com/pkg/errors"
 )
 
-var errRenewLease = errors.New("renew lease on id")
-var errReadOlderTableVersion = errors.New("read older table version from store")
+var errRenewLease = pgerror.WithMarker(errors.New("renew lease on id"))
+var errReadOlderTableVersion = pgerror.WithMarker(errors.New("read older table version from store"))
 
 // A lease stored in system.lease.
 type storedTableLease struct {
@@ -337,7 +337,9 @@ func maybeIncrementVersion(
 	return nil
 }
 
-var errDidntUpdateDescriptor = errors.New("didn't update the table descriptor")
+var errDidntUpdateDescriptor = pgerror.WithMarker(errors.New("didn't update the table descriptor"))
+
+var errLeaseVersionChanged = pgerror.WithMarker(errors.New("lease version changed"))
 
 // Publish updates a table descriptor. It also maintains the invariant that
 // there are at most two versions of the descriptor out in the wild at any time
@@ -355,7 +357,6 @@ func (s LeaseStore) Publish(
 	update func(*sqlbase.MutableTableDescriptor) error,
 	logEvent func(*client.Txn) error,
 ) (*sqlbase.ImmutableTableDescriptor, error) {
-	errLeaseVersionChanged := errors.New("lease version changed")
 	// Retry while getting errLeaseVersionChanged.
 	for r := retry.Start(base.DefaultRetryOptions()); r.Next(); {
 		// Wait until there are no unexpired leases on the previous version
@@ -428,10 +429,10 @@ func (s LeaseStore) Publish(
 			return txn.CommitInBatch(ctx, b)
 		})
 
-		switch err {
-		case nil, errDidntUpdateDescriptor:
+		switch {
+		case err == nil, pgerror.IsMarkedError(err, errDidntUpdateDescriptor):
 			return sqlbase.NewImmutableTableDescriptor(tableDesc.TableDescriptor), nil
-		case errLeaseVersionChanged:
+		case pgerror.IsMarkedError(err, errLeaseVersionChanged):
 			// will loop around to retry
 		default:
 			return nil, err
@@ -1061,7 +1062,7 @@ func purgeOldVersions(
 	// active lease, so that it doesn't get released when removeInactives()
 	// is called below. Release this lease after calling removeInactives().
 	table, _, err := t.findForTimestamp(ctx, m.execCfg.Clock.Now())
-	if dropped := err == errTableDropped; dropped || err == nil {
+	if dropped := pgerror.IsMarkedError(err, errTableDropped); dropped || err == nil {
 		removeInactives(dropped)
 		if table != nil {
 			s, err := t.release(&table.ImmutableTableDescriptor, m.removeOnceDereferenced())
@@ -1531,8 +1532,9 @@ func (m *LeaseManager) Acquire(
 			}
 			return &table.ImmutableTableDescriptor, table.expiration, nil
 		}
-		switch err {
-		case errRenewLease:
+
+		switch {
+		case pgerror.IsMarkedError(err, errRenewLease):
 			// Renew lease and retry. This will block until the lease is acquired.
 			if _, errLease := acquireNodeLease(ctx, m, tableID); errLease != nil {
 				return nil, hlc.Timestamp{}, errLease
@@ -1541,7 +1543,7 @@ func (m *LeaseManager) Acquire(
 				m.testingKnobs.LeaseStoreTestingKnobs.LeaseAcquireResultBlockEvent(LeaseAcquireBlock)
 			}
 
-		case errReadOlderTableVersion:
+		case pgerror.IsMarkedError(err, errReadOlderTableVersion):
 			// Read old table versions from the store. This can block while reading
 			// old table versions from the store.
 			versions, errRead := m.readOlderVersionForTimestamp(ctx, tableID, timestamp)

pkg/sql/opt_catalog.go

@@ -155,7 +155,7 @@ func (oc *optCatalog) ResolveDataSourceByID(
 	tableLookup, err := oc.planner.LookupTableByID(ctx, sqlbase.ID(dataSourceID))
 
 	if err != nil || tableLookup.IsAdding {
-		if err == sqlbase.ErrDescriptorNotFound || tableLookup.IsAdding {
+		if pgerror.IsMarkedError(err, sqlbase.ErrDescriptorNotFound) || tableLookup.IsAdding {
 			return nil, sqlbase.NewUndefinedRelationError(&tree.TableRef{TableID: int64(dataSourceID)})
 		}
 		return nil, err

pkg/sql/pgwire/pgerror/errors.go

@@ -64,11 +64,13 @@ func formatMsgHintDetail(prefix, msg, hint, detail string) string {
 // information at the specified depth level.
 func NewErrorWithDepthf(depth int, code string, format string, args ...interface{}) *Error {
 	srcCtx := makeSrcCtx(depth + 1)
-	return &Error{
+	err := &Error{
 		Message: fmt.Sprintf(format, args...),
 		Code:    code,
 		Source:  &srcCtx,
 	}
+	err.OriginMarker = getMarker(err)
+	return err
 }
 
 // NewError creates an Error.
@@ -221,7 +223,7 @@ var _ fmt.Formatter = &Error{}
 
 // Format implements the fmt.Formatter interface.
 //
-// %v/%s prints the rror as usual.
+// %v/%s prints the error as usual.
 // %#v adds the pg error code at the beginning.
 // %+v prints all the details, including the embedded stack traces.
 func (pg *Error) Format(s fmt.State, verb rune) {

pkg/sql/pgwire/pgerror/errors.proto

@@ -50,4 +50,10 @@ message Error {
   // complement to the detail field that can be reported
   // in sentry reports. This is scrubbed of PII.
   repeated SafeDetail safe_detail = 7;
+
+  // the origin marker is guaranteed to propagate as-is
+  // through error wraps, and can be used to recognize
+  // the original cause of a pgerror.
+  // This may contain PII and should not be reported.
+  string origin_marker = 8;
 };