systemschema: switch session_uuid back to session_id

The SessionID encodes a region prefix and a UUID. The initial
implementation of RBR sqlliveness stored the crdb_region and the
session_uuid. Now, sqlliveness stores the crdb_region and value of the
SessionID.

```
SELECT * FROM system.sqlliveness;
                 session_id                |           expiration           | crdb_region
-------------------------------------------+--------------------------------+--------------
  \x01018028b91818a7584268992bb246f9755b37 | 1676502803528637593.0000000000 | \x80
(1 row)
```

Storing the complete SessionID takes an extra 3 bytes of storage but it
makes it simpler to inspect the state of the system and join with tables
that refer to the session_id. For example, it's possible to write this
query:

```
SELECT *
FROM system.sqlliveness AS s
LEFT JOIN ON system.jobs AS j
WHERE s.session_id = j.session_id
AND s.crdb_region = 'us-central-1';
```

Part of cockroachdb#94843

Release note: None

pkg/ccl/multiregionccl/multiregion_system_table_test.go

@@ -78,11 +78,11 @@ func TestMrSystemDatabase(t *testing.T) {
 	tDB.CheckQueryResults(t, `SELECT * FROM crdb_internal.invalid_objects`, [][]string{})
 
 	t.Run("Sqlliveness", func(t *testing.T) {
-		row := tDB.QueryRow(t, `SELECT crdb_region, session_uuid, expiration FROM system.sqlliveness LIMIT 1`)
-		var sessionUUID string
+		row := tDB.QueryRow(t, `SELECT crdb_region, session_id, expiration FROM system.sqlliveness LIMIT 1`)
+		var sessionID string
 		var crdbRegion string
 		var rawExpiration apd.Decimal
-		row.Scan(&crdbRegion, &sessionUUID, &rawExpiration)
+		row.Scan(&crdbRegion, &sessionID, &rawExpiration)
 		require.Equal(t, "us-east1", crdbRegion)
 	})
 
@@ -239,14 +239,14 @@ func TestMrSystemDatabase(t *testing.T) {
 		// optimizer, because the stats will have the wrong type for the
 		// crdb_column.
 		row := tDB.QueryRow(t, `
-			SELECT crdb_region, session_uuid, expiration 
+			SELECT crdb_region, session_id, expiration 
 			FROM system.sqlliveness 
 			WHERE crdb_region = 'us-east1'
 			LIMIT 1;`)
-		var sessionUUID string
+		var sessionID string
 		var crdbRegion string
 		var rawExpiration apd.Decimal
-		row.Scan(&crdbRegion, &sessionUUID, &rawExpiration)
+		row.Scan(&crdbRegion, &sessionID, &rawExpiration)
 		require.Equal(t, "us-east1", crdbRegion)
 	})
 }
@@ -301,30 +301,28 @@ func TestTenantStartupWithMultiRegionEnum(t *testing.T) {
 	var sessionID string
 	tenSQLDB2.QueryRow(t, `SELECT session_id FROM system.sql_instances WHERE id = $1`,
 		ten.SQLInstanceID()).Scan(&sessionID)
-	region, id, err := slstorage.UnsafeDecodeSessionID(sqlliveness.SessionID(sessionID))
+	region, _, err := slstorage.UnsafeDecodeSessionID(sqlliveness.SessionID(sessionID))
 	require.NoError(t, err)
-	require.NotNil(t, id)
 	require.Equal(t, enum.One, region)
 
 	// Ensure that the sqlliveness entry created by the second SQL server has
 	// the right region and session UUID.
 	tenSQLDB2.QueryRow(t, `SELECT session_id FROM system.sql_instances WHERE id = $1`,
 		ten2.SQLInstanceID()).Scan(&sessionID)
-	region, id, err = slstorage.UnsafeDecodeSessionID(sqlliveness.SessionID(sessionID))
+	region, _, err = slstorage.UnsafeDecodeSessionID(sqlliveness.SessionID(sessionID))
 	require.NoError(t, err)
-	require.NotNil(t, id)
 	require.NotEqual(t, enum.One, region)
 
-	rows := tenSQLDB2.Query(t, `SELECT crdb_region, session_uuid FROM system.sqlliveness`)
+	rows := tenSQLDB2.Query(t, `SELECT crdb_region, session_id FROM system.sqlliveness`)
 	defer rows.Close()
 	livenessMap := map[string][]byte{}
 	for rows.Next() {
-		var region, sessionUUID string
-		require.NoError(t, rows.Scan(&region, &sessionUUID))
-		livenessMap[sessionUUID] = []byte(region)
+		var region, ID string
+		require.NoError(t, rows.Scan(&region, &ID))
+		livenessMap[ID] = []byte(region)
 	}
 	require.NoError(t, rows.Err())
-	r, ok := livenessMap[string(id)]
+	r, ok := livenessMap[sessionID]
 	require.True(t, ok)
 	require.Equal(t, r, region)
 }

pkg/sql/catalog/systemschema/system.go

@@ -492,11 +492,11 @@ CREATE TABLE system.sqlliveness (
 
 	MrSqllivenessTableSchema = `
 CREATE TABLE system.sqlliveness (
-    session_uuid         BYTES NOT NULL,
+    session_id           BYTES NOT NULL,
     expiration           DECIMAL NOT NULL,
     crdb_region          BYTES NOT NULL,
-    CONSTRAINT "primary" PRIMARY KEY (crdb_region, session_uuid),
-    FAMILY "primary" (crdb_region, session_uuid, expiration)
+    CONSTRAINT "primary" PRIMARY KEY (crdb_region, session_id),
+    FAMILY "primary" (crdb_region, session_id, expiration)
 )`
 
 	// system.migrations stores completion records for upgrades performed by the
@@ -2309,26 +2309,26 @@ var (
 					catconstants.SqllivenessTableName,
 					keys.SqllivenessID,
 					[]descpb.ColumnDescriptor{
-						{Name: "crdb_region", ID: 4, Type: types.Bytes, Nullable: false},
-						{Name: "session_uuid", ID: 3, Type: types.Bytes, Nullable: false},
+						{Name: "session_id", ID: 1, Type: types.Bytes, Nullable: false},
 						{Name: "expiration", ID: 2, Type: types.Decimal, Nullable: false},
+						{Name: "crdb_region", ID: 3, Type: types.Bytes, Nullable: false},
 					},
 					[]descpb.ColumnFamilyDescriptor{
 						{
 							Name:            "primary",
 							ID:              0,
-							ColumnNames:     []string{"crdb_region", "session_uuid", "expiration"},
-							ColumnIDs:       []descpb.ColumnID{4, 3, 2},
+							ColumnNames:     []string{"session_id", "expiration", "crdb_region"},
+							ColumnIDs:       []descpb.ColumnID{1, 2, 3},
 							DefaultColumnID: 2,
 						},
 					},
 					descpb.IndexDescriptor{
 						Name:                "primary",
 						ID:                  2,
 						Unique:              true,
-						KeyColumnNames:      []string{"crdb_region", "session_uuid"},
+						KeyColumnNames:      []string{"crdb_region", "session_id"},
 						KeyColumnDirections: []catenumpb.IndexColumn_Direction{catenumpb.IndexColumn_ASC, catenumpb.IndexColumn_ASC},
-						KeyColumnIDs:        []descpb.ColumnID{4, 3},
+						KeyColumnIDs:        []descpb.ColumnID{3, 1},
 					},
 				))
 		}

pkg/sql/sqlliveness/slstorage/key_encoder.go

@@ -19,7 +19,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/catid"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness"
 	"github.com/cockroachdb/cockroach/pkg/util/encoding"
-	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/cockroachdb/errors"
 )
 
@@ -52,7 +51,7 @@ type rbrEncoder struct {
 }
 
 func (e *rbrEncoder) encode(session sqlliveness.SessionID) (roachpb.Key, error) {
-	region, uuid, err := UnsafeDecodeSessionID(session)
+	region, _, err := UnsafeDecodeSessionID(session)
 	if err != nil {
 		return nil, err
 	}
@@ -64,7 +63,7 @@ func (e *rbrEncoder) encode(session sqlliveness.SessionID) (roachpb.Key, error)
 
 	key := e.indexPrefix()
 	key = encoding.EncodeBytesAscending(key, region)
-	key = encoding.EncodeBytesAscending(key, uuid)
+	key = encoding.EncodeBytesAscending(key, session.UnsafeBytes())
 	return keys.MakeFamilyKey(key, columnFamilyID), nil
 }
 
@@ -74,22 +73,17 @@ func (e *rbrEncoder) decode(key roachpb.Key) (sqlliveness.SessionID, error) {
 	}
 	rem := key[len(e.rbrIndex):]
 
-	rem, region, err := encoding.DecodeBytesAscending(rem, nil)
+	rem, _, err := encoding.DecodeBytesAscending(rem, nil)
 	if err != nil {
 		return "", errors.Wrap(err, "failed to decode region from session key")
 	}
 
-	rem, rawUUID, err := encoding.DecodeBytesAscending(rem, nil)
+	rem, id, err := encoding.DecodeBytesAscending(rem, nil)
 	if err != nil {
 		return "", errors.Wrap(err, "failed to decode uuid from session key")
 	}
 
-	id, err := uuid.FromBytes(rawUUID)
-	if err != nil {
-		return "", errors.Wrap(err, "failed to convert uuid bytes to uuid id for session key")
-	}
-
-	return MakeSessionID(region, id)
+	return sqlliveness.SessionID(id), nil
 }
 
 func (e *rbrEncoder) indexPrefix() roachpb.Key {