sql: add serial normalization setting for cached sequences

Closes: #51259

Release note (sql change): The `serial_normalization` session
variable can now be set to the value `sql_sequence_cached`.
Cached sequences will allow nodes to cache 256 sequence numbers
locally. The underlying sequence will only be incremened once (by
256 increments) when the cache is empty. Using `sql_sequence_cached`
will result in better performance than `sql_sequence` because the
former will perform fewer distributed calls to increment sequences.
However, cached seqences may contribute to large gaps between
sequence numbers if cached values are lost due to errors or
node outages.

docs/generated/settings/settings.html

@@ -70,7 +70,7 @@
 <tr><td><code>sql.defaults.default_int_size</code></td><td>integer</td><td><code>8</code></td><td>the size, in bytes, of an INT type</td></tr>
 <tr><td><code>sql.defaults.disallow_full_table_scans.enabled</code></td><td>boolean</td><td><code>false</code></td><td>setting to true rejects queries that have planned a full table scan</td></tr>
 <tr><td><code>sql.defaults.results_buffer.size</code></td><td>byte size</td><td><code>16 KiB</code></td><td>default size of the buffer that accumulates results for a statement or a batch of statements before they are sent to the client. This can be overridden on an individual connection with the 'results_buffer_size' parameter. Note that auto-retries generally only happen while no results have been delivered to the client, so reducing this size can increase the number of retriable errors a client receives. On the other hand, increasing the buffer size can increase the delay until the client receives the first result row. Updating the setting only affects new connections. Setting to 0 disables any buffering.</td></tr>
-<tr><td><code>sql.defaults.serial_normalization</code></td><td>enumeration</td><td><code>rowid</code></td><td>default handling of SERIAL in table definitions [rowid = 0, virtual_sequence = 1, sql_sequence = 2]</td></tr>
+<tr><td><code>sql.defaults.serial_normalization</code></td><td>enumeration</td><td><code>rowid</code></td><td>default handling of SERIAL in table definitions [rowid = 0, virtual_sequence = 1, sql_sequence = 2, sql_sequence_cached = 3]</td></tr>
 <tr><td><code>sql.distsql.max_running_flows</code></td><td>integer</td><td><code>500</code></td><td>maximum number of concurrent flows that can be run on a node</td></tr>
 <tr><td><code>sql.log.slow_query.experimental_full_table_scans.enabled</code></td><td>boolean</td><td><code>false</code></td><td>when set to true, statements that perform a full table/index scan will be logged to the slow query log even if they do not meet the latency threshold. Must have the slow query log enabled for this setting to have any effect.</td></tr>
 <tr><td><code>sql.log.slow_query.internal_queries.enabled</code></td><td>boolean</td><td><code>false</code></td><td>when set to true, internal queries which exceed the slow query log threshold are logged to a separate log. Must have the slow query log enabled for this setting to have any effect.</td></tr>

pkg/sql/exec_util.go

@@ -401,9 +401,10 @@ var SerialNormalizationMode = settings.RegisterEnumSetting(
 	"default handling of SERIAL in table definitions",
 	"rowid",
 	map[int64]string{
-		int64(sessiondata.SerialUsesRowID):            "rowid",
-		int64(sessiondata.SerialUsesVirtualSequences): "virtual_sequence",
-		int64(sessiondata.SerialUsesSQLSequences):     "sql_sequence",
+		int64(sessiondata.SerialUsesRowID):              "rowid",
+		int64(sessiondata.SerialUsesVirtualSequences):   "virtual_sequence",
+		int64(sessiondata.SerialUsesSQLSequences):       "sql_sequence",
+		int64(sessiondata.SerialUsesCachedSQLSequences): "sql_sequence_cached",
 	},
 ).WithPublic()
 

pkg/sql/logictest/testdata/logic_test/serial

@@ -335,3 +335,47 @@ SELECT count(DISTINCT a), count(DISTINCT b), count(DISTINCT c) FROM serials
 
 statement ok
 DROP TABLE serials, smallbig, serial
+
+# Verify that the sequence gets incremented to the default cache
+# size of 256 and that serial values increase by 1 each time.
+subtest serial_cached_sequence
+
+statement ok
+SET serial_normalization = sql_sequence_cached
+
+statement ok
+CREATE TABLE serial (
+  cached SERIAL
+);
+
+statement ok
+INSERT INTO serial (cached) VALUES (DEFAULT);
+
+query I
+SELECT cached from serial;
+----
+1
+
+query T
+SELECT pg_get_serial_sequence('serial', 'cached')
+----
+public.serial_cached_seq
+
+query I
+SELECT last_value from public.serial_cached_seq;
+----
+256
+
+statement ok
+INSERT INTO serial (cached) VALUES (DEFAULT);
+
+query I
+SELECT cached from serial ORDER BY cached;
+----
+1
+2
+
+query I
+SELECT last_value from public.serial_cached_seq;
+----
+256

pkg/sql/serial.go

@@ -40,6 +40,16 @@ var virtualSequenceOpts = tree.SequenceOptions{
 	tree.SequenceOption{Name: tree.SeqOptVirtual},
 }
 
+// cachedSequencesCacheSize is the default cache size used when
+// SessionNormalizationMode is SerialUsesCachedSQLSequences.
+var cachedSequencesCacheSize int64 = 256
+
+// cachedSequenceOpts is used when SessionNormalizationMode is
+// SerialUsesCachedSQLSequences.
+var cachedSequenceOpts = tree.SequenceOptions{
+	tree.SequenceOption{Name: tree.SeqOptCache, IntVal: &cachedSequencesCacheSize},
+}
+
 // processSerialInColumnDef analyzes a column definition and determines
 // whether to use a sequence if the requested type is SERIAL-like.
 // If a sequence must be created, it returns an TableName to use
@@ -84,7 +94,7 @@ func (p *planner) processSerialInColumnDef(
 		// switch this behavior around.
 		newSpec.Type = types.Int
 
-	case sessiondata.SerialUsesSQLSequences:
+	case sessiondata.SerialUsesSQLSequences, sessiondata.SerialUsesCachedSQLSequences:
 		// With real sequences we can use the requested type as-is.
 
 	default:
@@ -154,6 +164,9 @@ func (p *planner) processSerialInColumnDef(
 	if serialNormalizationMode == sessiondata.SerialUsesVirtualSequences {
 		seqType = "virtual "
 		seqOpts = virtualSequenceOpts
+	} else if serialNormalizationMode == sessiondata.SerialUsesCachedSQLSequences {
+		seqType = "cached "
+		seqOpts = cachedSequenceOpts
 	}
 	log.VEventf(ctx, 2, "new column %q of %q will have %s sequence name %q and default %q",
 		d, tableName, seqType, seqName, defaultExpr)

pkg/sql/sessiondata/session_data.go

@@ -373,8 +373,18 @@ const (
 	// use INT NOT NULL DEFAULT nextval(...).
 	SerialUsesVirtualSequences
 	// SerialUsesSQLSequences means create a regular SQL sequence and
-	// use INT NOT NULL DEFAULT nextval(...).
+	// use INT NOT NULL DEFAULT nextval(...). Each call to nextval()
+	// is a distributed call to kv. This minimizes the size of gaps
+	// between successive sequence numbers (which occur due to
+	// node failures or errors), but the multiple kv calls
+	// can impact performance negatively.
 	SerialUsesSQLSequences
+	// SerialUsesCachedSQLSequences is identical to SerialUsesSQLSequences with
+	// the exception that nodes can cache sequence values. This significantly
+	// reduces contention and distributed calls to kv, which results in better
+	// performance. Gaps between sequences may be larger as a result of cached
+	// values being lost to errors and/or node failures.
+	SerialUsesCachedSQLSequences
 )
 
 func (m SerialNormalizationMode) String() string {
@@ -385,6 +395,8 @@ func (m SerialNormalizationMode) String() string {
 		return "virtual_sequence"
 	case SerialUsesSQLSequences:
 		return "sql_sequence"
+	case SerialUsesCachedSQLSequences:
+		return "sql_sequence_cached"
 	default:
 		return fmt.Sprintf("invalid (%d)", m)
 	}
@@ -399,6 +411,8 @@ func SerialNormalizationModeFromString(val string) (_ SerialNormalizationMode, o
 		return SerialUsesVirtualSequences, true
 	case "SQL_SEQUENCE":
 		return SerialUsesSQLSequences, true
+	case "SQL_SEQUENCE_CACHED":
+		return SerialUsesCachedSQLSequences, true
 	default:
 		return 0, false
 	}