sql: support making sequences read only

Previously, the only physical table types that were read only were
materialized viewed. As a part of the PCR work to support reading from a
tenant we added support for replicating catalogs and inserting them w
ith external data rows. We apply the same idea to sequences, however we
are lacking runtime support. To address this, this patch makes sequence
opertions read only if external data rows exists.

Release note: None

pkg/sql/catalog/descpb/structured.go

@@ -230,6 +230,11 @@ func (desc *TableDescriptor) MaterializedView() bool {
 	return desc.IsMaterializedView
 }
 
+// IsReadOnly implements the TableDescriptor interface.
+func (desc *TableDescriptor) IsReadOnly() bool {
+	return desc.IsMaterializedView || desc.GetExternal() != nil
+}
+
 // IsPhysicalTable implements the TableDescriptor interface.
 func (desc *TableDescriptor) IsPhysicalTable() bool {
 	return desc.IsSequence() || (desc.IsTable() && !desc.IsVirtualTable()) || desc.MaterializedView()

pkg/sql/catalog/descriptor.go

@@ -337,6 +337,9 @@ type TableDescriptor interface {
 	IsPhysicalTable() bool
 	// MaterializedView returns whether this TableDescriptor is a MaterializedView.
 	MaterializedView() bool
+	// IsReadOnly returns if this table descriptor has external data, and cannot
+	// be written to.
+	IsReadOnly() bool
 	// IsAs returns true if the TableDescriptor describes a Table that was created
 	// with a CREATE TABLE AS command.
 	IsAs() bool

pkg/sql/catalog/replication/reader_catalog_test.go

@@ -90,6 +90,12 @@ func TestReaderCatalog(t *testing.T) {
 	compareConn("SELECT * FROM system.role_options")
 	compareConn("SELECT * FROM system.database_role_settings")
 
+	// Validate that sequences can be selected.
+	compareConn("SELECT * FROM sq1")
+
+	// Validate that sequence operations are blocked.
+	destRunner.ExpectErr(t, "cannot execute nextval\\(\\) in a read-only transaction", "SELECT nextval('sq1')")
+	destRunner.ExpectErr(t, "cannot execute setval\\(\\) in a read-only transaction", "SELECT setval('sq1', 32)")
 }
 func TestMain(m *testing.M) {
 	securityassets.SetLoader(securitytest.EmbeddedAssets)

pkg/sql/sequence.go

@@ -112,6 +112,11 @@ func incrementSequenceHelper(
 			string(descriptor.DescriptorType()), descriptor.GetName())
 	}
 
+	// If the descriptor is read only block any write operations.
+	if descriptor.IsReadOnly() {
+		return 0, readOnlyError("nextval()")
+	}
+
 	var err error
 	seqOpts := descriptor.GetSequenceOpts()
 
@@ -294,6 +299,10 @@ func (p *planner) SetSequenceValueByID(
 	if err != nil {
 		return err
 	}
+	// If the descriptor is read only block any write operations.
+	if descriptor.IsReadOnly() {
+		return readOnlyError("setval()")
+	}
 	seqName, err := p.getQualifiedTableName(ctx, descriptor)
 	if err != nil {
 		return err
@@ -410,7 +419,14 @@ func (p *planner) GetSequenceValue(
 func getSequenceValueFromDesc(
 	ctx context.Context, txn *kv.Txn, codec keys.SQLCodec, desc catalog.TableDescriptor,
 ) (int64, error) {
-	keyValue, err := txn.Get(ctx, codec.SequenceKey(uint32(desc.GetID())))
+	targetID := desc.GetID()
+	// For external row data, adjust the key that we will
+	// scan.
+	if ext := desc.ExternalRowData(); ext != nil {
+		codec = keys.MakeSQLCodec(ext.TenantID)
+		targetID = desc.ExternalRowData().TableID
+	}
+	keyValue, err := txn.Get(ctx, codec.SequenceKey(uint32(targetID)))
 	if err != nil {
 		return 0, err
 	}