Merge #69202 #69346

69202: sql: add txn rows written/read guardrails r=yuzefovich a=yuzefovich

This commit introduces the guardrails on the number of rows written/read
by a single txn. The limits are enforced after each statement of a txn
has been fully executed (i.e. we don't proactively cancel work in the
middle of the execution if the txn has just reached the limits). This is
done in the connExecutor since it is a very convenient place to enforce
the limits so that they apply only to the desired statements. Notably,
things for which we don't want the application of the limits (things
like BACKUP, IMPORT, CREATE STATISTICS, etc) don't go through the
connExecutor and, thus, aren't affected.

The accumulation of the number of rows read by a txn has already been in
place, and this commit introduces the explicit collection of the number
of rows written via the same mechanism - by propagating "rows written"
metrics during the draining of the execution flow. Initially, we
considered using "rows affected" values, but those have a different
meaning from what we want. This metrics collection required teaching the
planNodeToRowSource adapter to ask the mutation planNodes for the number
of rows written.

Note that in many cases, the internal executor doesn't have the
sessionData properly set (i.e. the default values are used), so we'll
never log anything then. This seems acceptable since the focus of these
guardrails is on the externally initiated queries.

One notable change is because of our "after the fact" enforcement of
these limits, this commit disables the auto commit option for the
mutation statements in the execbuilder for non-internal statements if
`transaction_rows_read_err` guardrail is enabled.

Release note (ops change): New cluster settings
`sql.defaults.transaction_rows_written_log`,
`sql.defaults.transaction_rows_written_err`,
`sql.defaults.transaction_rows_read_log`, and
`sql.defaults.transaction_rows_read_err` (as well as the corresponding
session variables have been introduced. These settings determine the
"size" of the transactions in written and read rows upon reaching of
which the transactions are logged or rejected. The logging will go into
SQL_PERF logging channel. Note that the internal queries (i.e. those
issued by CockroachDB internally) cannot error out but can be logged
instead into SQL_INTERNAL_PERF logging channel. The "written" limits
apply to INSERT, INSERT INTO SELECT FROM, INSERT ON CONFLICT, UPSERT,
UPDATE, and DELETE whereas the "read" limits apply to SELECT statement
in addition to all of these. These limits will not apply to CREATE
TABLE AS SELECT, IMPORT, TRUNCATE, DROP, ALTER TABLE, BACKUP, RESTORE,
or CREATE STATISTICS statements. Note that enabling
`transaction_rows_read_err` guardrail comes at the cost of disabling
the usage of the auto commit optimization for the mutation statements
in implicit transactions.

Release justification: low-risk, high benefit change to existing
functionality.

69346: sql: fix incorrect follower read and testing knob for sql stats compaction job r=ajwerner,maryliag,matthewtodd a=Azhng

Previously, the SQL Stats Compaction job did not check for nil testing 
knobs and constructed its SQL query with incorrect follower read syntax.
This commit address those two bugs.

Release justification: Category 2: Bug fixes and low-risk updates
 to new functionality

Release note: None

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

docs/generated/eventlog.md

@@ -1939,6 +1939,63 @@ set to a non-zero value, AND
 | `FullIndexScan` | Whether the query contains a full secondary index scan. | no |
 | `TxnCounter` | The sequence number of the SQL transaction inside its session. | no |
 
+### `txn_rows_read_limit`
+
+An event of type `txn_rows_read_limit` is recorded when a transaction tries to read more rows than
+cluster setting `sql.defaults.transaction_rows_read_log`. There will only be
+a single record for a single transaction (unless it is retried) even if there
+are more statement within the transaction that haven't been executed yet.
+
+
+
+
+#### Common fields
+
+| Field | Description | Sensitive |
+|--|--|--|
+| `Timestamp` | The timestamp of the event. Expressed as nanoseconds since the Unix epoch. | no |
+| `EventType` | The type of the event. | no |
+| `Statement` | A normalized copy of the SQL statement that triggered the event. The statement string contains a mix of sensitive and non-sensitive details (it is redactable). | partially |
+| `Tag` | The statement tag. This is separate from the statement string, since the statement string can contain sensitive information. The tag is guaranteed not to. | no |
+| `User` | The user account that triggered the event. The special usernames `root` and `node` are not considered sensitive. | depends |
+| `DescriptorID` | The primary object descriptor affected by the operation. Set to zero for operations that don't affect descriptors. | no |
+| `ApplicationName` | The application name for the session where the event was emitted. This is included in the event to ease filtering of logging output by application. Application names starting with a dollar sign (`$`) are not considered sensitive. | depends |
+| `PlaceholderValues` | The mapping of SQL placeholders to their values, for prepared statements. | yes |
+| `TxnID` | TxnID is the ID of the transaction that hit the row count limit. | no |
+| `SessionID` | SessionID is the ID of the session that initiated the transaction. | no |
+| `Limit` | Limit indicates the value of the transaction row count limit that was reached. | no |
+| `ViolatesTxnRowsLimitErr` | ViolatesTxnRowsLimitErr if true indicates that 'transaction_rows_{written|read}_err' limit is violated. | no |
+| `IsRead` | IsRead if true indicates that the "rows read" limit is reached and the "rows written" limit otherwise. | no |
+
+### `txn_rows_written_limit`
+
+An event of type `txn_rows_written_limit` is recorded when a transaction tries to write more rows
+than cluster setting `sql.defaults.transaction_rows_written_log`. There will
+only be a single record for a single transaction (unless it is retried) even
+if there are more mutation statements within the transaction that haven't
+been executed yet.
+
+
+
+
+#### Common fields
+
+| Field | Description | Sensitive |
+|--|--|--|
+| `Timestamp` | The timestamp of the event. Expressed as nanoseconds since the Unix epoch. | no |
+| `EventType` | The type of the event. | no |
+| `Statement` | A normalized copy of the SQL statement that triggered the event. The statement string contains a mix of sensitive and non-sensitive details (it is redactable). | partially |
+| `Tag` | The statement tag. This is separate from the statement string, since the statement string can contain sensitive information. The tag is guaranteed not to. | no |
+| `User` | The user account that triggered the event. The special usernames `root` and `node` are not considered sensitive. | depends |
+| `DescriptorID` | The primary object descriptor affected by the operation. Set to zero for operations that don't affect descriptors. | no |
+| `ApplicationName` | The application name for the session where the event was emitted. This is included in the event to ease filtering of logging output by application. Application names starting with a dollar sign (`$`) are not considered sensitive. | depends |
+| `PlaceholderValues` | The mapping of SQL placeholders to their values, for prepared statements. | yes |
+| `TxnID` | TxnID is the ID of the transaction that hit the row count limit. | no |
+| `SessionID` | SessionID is the ID of the session that initiated the transaction. | no |
+| `Limit` | Limit indicates the value of the transaction row count limit that was reached. | no |
+| `ViolatesTxnRowsLimitErr` | ViolatesTxnRowsLimitErr if true indicates that 'transaction_rows_{written|read}_err' limit is violated. | no |
+| `IsRead` | IsRead if true indicates that the "rows read" limit is reached and the "rows written" limit otherwise. | no |
+
 ## SQL Slow Query Log (Internal)
 
 Events in this category report slow query execution by
@@ -2006,6 +2063,65 @@ the "slow query" condition.
 | `FullIndexScan` | Whether the query contains a full secondary index scan. | no |
 | `TxnCounter` | The sequence number of the SQL transaction inside its session. | no |
 
+### `txn_rows_read_limit_internal`
+
+An event of type `txn_rows_read_limit_internal` is recorded when an internal transaction tries to
+read more rows than cluster setting `sql.defaults.transaction_rows_read_log`
+or `sql.defaults.transaction_rows_read_err`. There will only be a single
+record for a single transaction (unless it is retried) even if there are more
+mutation statements within the transaction that haven't been executed yet.
+
+
+
+
+#### Common fields
+
+| Field | Description | Sensitive |
+|--|--|--|
+| `Timestamp` | The timestamp of the event. Expressed as nanoseconds since the Unix epoch. | no |
+| `EventType` | The type of the event. | no |
+| `Statement` | A normalized copy of the SQL statement that triggered the event. The statement string contains a mix of sensitive and non-sensitive details (it is redactable). | partially |
+| `Tag` | The statement tag. This is separate from the statement string, since the statement string can contain sensitive information. The tag is guaranteed not to. | no |
+| `User` | The user account that triggered the event. The special usernames `root` and `node` are not considered sensitive. | depends |
+| `DescriptorID` | The primary object descriptor affected by the operation. Set to zero for operations that don't affect descriptors. | no |
+| `ApplicationName` | The application name for the session where the event was emitted. This is included in the event to ease filtering of logging output by application. Application names starting with a dollar sign (`$`) are not considered sensitive. | depends |
+| `PlaceholderValues` | The mapping of SQL placeholders to their values, for prepared statements. | yes |
+| `TxnID` | TxnID is the ID of the transaction that hit the row count limit. | no |
+| `SessionID` | SessionID is the ID of the session that initiated the transaction. | no |
+| `Limit` | Limit indicates the value of the transaction row count limit that was reached. | no |
+| `ViolatesTxnRowsLimitErr` | ViolatesTxnRowsLimitErr if true indicates that 'transaction_rows_{written|read}_err' limit is violated. | no |
+| `IsRead` | IsRead if true indicates that the "rows read" limit is reached and the "rows written" limit otherwise. | no |
+
+### `txn_rows_written_limit_internal`
+
+An event of type `txn_rows_written_limit_internal` is recorded when an internal transaction tries to
+write more rows than cluster setting
+`sql.defaults.transaction_rows_written_log` or
+`sql.defaults.transaction_rows_written_err`. There will only be a single
+record for a single transaction (unless it is retried) even if there are more
+mutation statements within the transaction that haven't been executed yet.
+
+
+
+
+#### Common fields
+
+| Field | Description | Sensitive |
+|--|--|--|
+| `Timestamp` | The timestamp of the event. Expressed as nanoseconds since the Unix epoch. | no |
+| `EventType` | The type of the event. | no |
+| `Statement` | A normalized copy of the SQL statement that triggered the event. The statement string contains a mix of sensitive and non-sensitive details (it is redactable). | partially |
+| `Tag` | The statement tag. This is separate from the statement string, since the statement string can contain sensitive information. The tag is guaranteed not to. | no |
+| `User` | The user account that triggered the event. The special usernames `root` and `node` are not considered sensitive. | depends |
+| `DescriptorID` | The primary object descriptor affected by the operation. Set to zero for operations that don't affect descriptors. | no |
+| `ApplicationName` | The application name for the session where the event was emitted. This is included in the event to ease filtering of logging output by application. Application names starting with a dollar sign (`$`) are not considered sensitive. | depends |
+| `PlaceholderValues` | The mapping of SQL placeholders to their values, for prepared statements. | yes |
+| `TxnID` | TxnID is the ID of the transaction that hit the row count limit. | no |
+| `SessionID` | SessionID is the ID of the session that initiated the transaction. | no |
+| `Limit` | Limit indicates the value of the transaction row count limit that was reached. | no |
+| `ViolatesTxnRowsLimitErr` | ViolatesTxnRowsLimitErr if true indicates that 'transaction_rows_{written|read}_err' limit is violated. | no |
+| `IsRead` | IsRead if true indicates that the "rows read" limit is reached and the "rows written" limit otherwise. | no |
+
 ## SQL User and Role operations
 
 Events in this category pertain to SQL statements that modify the

docs/generated/settings/settings-for-tenants.txt

@@ -110,6 +110,10 @@ sql.defaults.results_buffer.size	byte size	16 KiB	default size of the buffer tha
 sql.defaults.serial_normalization	enumeration	rowid	default handling of SERIAL in table definitions [rowid = 0, virtual_sequence = 1, sql_sequence = 2, sql_sequence_cached = 3]
 sql.defaults.statement_timeout	duration	0s	default value for the statement_timeout; default value for the statement_timeout session setting; controls the duration a query is permitted to run before it is canceled; if set to 0, there is no timeout
 sql.defaults.stub_catalog_tables.enabled	boolean	true	default value for stub_catalog_tables session setting
+sql.defaults.transaction_rows_read_err	integer	0	the limit for the number of rows read by a SQL transaction which - once reached - will fail the transaction (or will trigger a logging event to SQL_INTERNAL_PERF for internal transactions); use 0 to disable
+sql.defaults.transaction_rows_read_log	integer	0	the threshold for the number of rows read by a SQL transaction which - once reached - will trigger a logging event to SQL_PERF (or SQL_INTERNAL_PERF for internal transactions); use 0 to disable
+sql.defaults.transaction_rows_written_err	integer	0	the limit for the number of rows written by a SQL transaction which - once reached - will fail the transaction (or will trigger a logging event to SQL_INTERNAL_PERF for internal transactions); use 0 to disable
+sql.defaults.transaction_rows_written_log	integer	0	the threshold for the number of rows written by a SQL transaction which - once reached - will trigger a logging event to SQL_PERF (or SQL_INTERNAL_PERF for internal transactions); use 0 to disable
 sql.defaults.vectorize	enumeration	on	default vectorize mode [on = 0, on = 2, experimental_always = 3, off = 4]
 sql.defaults.zigzag_join.enabled	boolean	true	default value for enable_zigzag_join session setting; allows use of zig-zag join by default
 sql.distsql.max_running_flows	integer	500	maximum number of concurrent flows that can be run on a node

docs/generated/settings/settings.html

@@ -114,6 +114,10 @@
 <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.defaults.statement_timeout</code></td><td>duration</td><td><code>0s</code></td><td>default value for the statement_timeout; default value for the statement_timeout session setting; controls the duration a query is permitted to run before it is canceled; if set to 0, there is no timeout</td></tr>
 <tr><td><code>sql.defaults.stub_catalog_tables.enabled</code></td><td>boolean</td><td><code>true</code></td><td>default value for stub_catalog_tables session setting</td></tr>
+<tr><td><code>sql.defaults.transaction_rows_read_err</code></td><td>integer</td><td><code>0</code></td><td>the limit for the number of rows read by a SQL transaction which - once reached - will fail the transaction (or will trigger a logging event to SQL_INTERNAL_PERF for internal transactions); use 0 to disable</td></tr>
+<tr><td><code>sql.defaults.transaction_rows_read_log</code></td><td>integer</td><td><code>0</code></td><td>the threshold for the number of rows read by a SQL transaction which - once reached - will trigger a logging event to SQL_PERF (or SQL_INTERNAL_PERF for internal transactions); use 0 to disable</td></tr>
+<tr><td><code>sql.defaults.transaction_rows_written_err</code></td><td>integer</td><td><code>0</code></td><td>the limit for the number of rows written by a SQL transaction which - once reached - will fail the transaction (or will trigger a logging event to SQL_INTERNAL_PERF for internal transactions); use 0 to disable</td></tr>
+<tr><td><code>sql.defaults.transaction_rows_written_log</code></td><td>integer</td><td><code>0</code></td><td>the threshold for the number of rows written by a SQL transaction which - once reached - will trigger a logging event to SQL_PERF (or SQL_INTERNAL_PERF for internal transactions); use 0 to disable</td></tr>
 <tr><td><code>sql.defaults.vectorize</code></td><td>enumeration</td><td><code>on</code></td><td>default vectorize mode [on = 0, on = 2, experimental_always = 3, off = 4]</td></tr>
 <tr><td><code>sql.defaults.zigzag_join.enabled</code></td><td>boolean</td><td><code>true</code></td><td>default value for enable_zigzag_join session setting; allows use of zig-zag join by default</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>

pkg/ccl/serverccl/BUILD.bazel

@@ -33,6 +33,8 @@ go_test(
         "//pkg/sql",
         "//pkg/sql/catalog/catconstants",
         "//pkg/sql/pgwire/pgcode",
+        "//pkg/sql/sqlstats",
+        "//pkg/sql/tests",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",

pkg/ccl/serverccl/tenant_grpc_test.go

@@ -18,6 +18,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/rpc"
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
+	"github.com/cockroachdb/cockroach/pkg/sql/tests"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/httputil"
@@ -37,14 +39,22 @@ func TestTenantGRPCServices(t *testing.T) {
 
 	ctx := context.Background()
 
-	testCluster := serverutils.StartNewTestCluster(t, 3, base.TestClusterArgs{})
+	serverParams, _ := tests.CreateTestServerParams()
+	testCluster := serverutils.StartNewTestCluster(t, 3, base.TestClusterArgs{
+		ServerArgs: serverParams,
+	})
 	defer testCluster.Stopper().Stop(ctx)
 
 	server := testCluster.Server(0)
 
 	tenantID := serverutils.TestTenantID()
 	tenant, connTenant := serverutils.StartTenant(t, server, base.TestTenantArgs{
 		TenantID: tenantID,
+		TestingKnobs: base.TestingKnobs{
+			SQLStatsKnobs: &sqlstats.TestingKnobs{
+				AOSTClause: "AS OF SYSTEM TIME '-1us'",
+			},
+		},
 	})
 	defer connTenant.Close()
 

pkg/ccl/serverccl/tenant_status_test.go

@@ -22,6 +22,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catconstants"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
+	"github.com/cockroachdb/cockroach/pkg/sql/tests"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -35,13 +37,21 @@ func TestTenantCannotSeeNonTenantStats(t *testing.T) {
 
 	ctx := context.Background()
 
-	testCluster := serverutils.StartNewTestCluster(t, 3 /* numNodes */, base.TestClusterArgs{})
+	serverParams, _ := tests.CreateTestServerParams()
+	testCluster := serverutils.StartNewTestCluster(t, 3 /* numNodes */, base.TestClusterArgs{
+		ServerArgs: serverParams,
+	})
 	defer testCluster.Stopper().Stop(ctx)
 
 	server := testCluster.Server(0 /* idx */)
 
 	tenant, sqlDB := serverutils.StartTenant(t, server, base.TestTenantArgs{
 		TenantID: roachpb.MakeTenantID(10 /* id */),
+		TestingKnobs: base.TestingKnobs{
+			SQLStatsKnobs: &sqlstats.TestingKnobs{
+				AOSTClause: "AS OF SYSTEM TIME '-1us'",
+			},
+		},
 	})
 
 	nonTenant := testCluster.Server(1 /* idx */)

pkg/cli/BUILD.bazel

@@ -154,6 +154,7 @@ go_library(
         "//pkg/sql/rowenc",
         "//pkg/sql/sem/builtins",
         "//pkg/sql/sem/tree",
+        "//pkg/sql/sqlstats",
         "//pkg/startupmigrations",
         "//pkg/storage",
         "//pkg/storage/enginepb",

pkg/cli/testutils.go

@@ -32,6 +32,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/security/securitytest"
 	"github.com/cockroachdb/cockroach/pkg/server"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/errors"
@@ -123,6 +124,11 @@ func newCLITestWithArgs(params TestCLIParams, argsFn func(args *base.TestServerA
 			StoreSpecs:    params.StoreSpecs,
 			Locality:      params.Locality,
 			ExternalIODir: filepath.Join(certsDir, "extern"),
+			Knobs: base.TestingKnobs{
+				SQLStatsKnobs: &sqlstats.TestingKnobs{
+					AOSTClause: "AS OF SYSTEM TIME '-1us'",
+				},
+			},
 		}
 		if argsFn != nil {
 			argsFn(&args)

pkg/server/BUILD.bazel

@@ -149,7 +149,7 @@ go_library(
         "//pkg/sql/sqlinstance/instanceprovider",
         "//pkg/sql/sqlliveness",
         "//pkg/sql/sqlliveness/slprovider",
-        "//pkg/sql/sqlstats/persistedsqlstats",
+        "//pkg/sql/sqlstats",
         "//pkg/sql/sqlutil",
         "//pkg/sql/stats",
         "//pkg/sql/stmtdiagnostics",

pkg/server/server_sql.go

@@ -79,7 +79,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlinstance/instanceprovider"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness/slprovider"
-	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/persistedsqlstats"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlutil"
 	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/sql/stmtdiagnostics"
@@ -722,7 +722,7 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 		execCfg.IndexUsageStatsTestingKnobs = indexUsageStatsKnobs.(*idxusage.TestingKnobs)
 	}
 	if sqlStatsKnobs := cfg.TestingKnobs.SQLStatsKnobs; sqlStatsKnobs != nil {
-		execCfg.SQLStatsTestingKnobs = sqlStatsKnobs.(*persistedsqlstats.TestingKnobs)
+		execCfg.SQLStatsTestingKnobs = sqlStatsKnobs.(*sqlstats.TestingKnobs)
 	}
 
 	statsRefresher := stats.MakeRefresher(

pkg/server/statements_test.go

@@ -14,9 +14,9 @@ import (
 	"context"
 	"testing"
 
-	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/rpc"
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/tests"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -33,7 +33,8 @@ func TestStatements(t *testing.T) {
 
 	ctx := context.Background()
 
-	testServer, db, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	params, _ := tests.CreateTestServerParams()
+	testServer, db, _ := serverutils.StartServer(t, params)
 	defer testServer.Stopper().Stop(ctx)
 
 	conn, err := testServer.RPCContext().GRPCDialNode(