sql: add sql.mutations.max_row_size.warn guardrail

Addresses: cockroachdb#67400

Add sql.mutations.max_row_size.warn, a new cluster setting which
controls large row logging. Rows larger than this size will have their
keys logged to the SQL_PERF or SQL_INTERNAL_PERF channels whenever the
SQL layer puts them into the KV layer.

This logging takes place in rowHelper, which is used by both
row.Inserter and row.Updater. Most of the work is plumbing
settings.Values and SessionData into rowHelper, and adding a new
structured event type.

Release note (ops change): A new cluster setting,
sql.mutations.max_row_size.warn, was added, which controls large row
logging. Whenever a row larger than this size is added (or a simgle
column family if multiple column families are in use) a LargeRow event
is logged to the SQL_PERF channel (or a LargeRowInternal event is logged
to SQL_INTERNAL_PERF if the row was added by an internal query). This
could occur for INSERT, UPSERT, UPDATE, CREATE TABLE AS, CREATE INDEX,
ALTER TABLE, ALTER INDEX, IMPORT, or RESTORE statements.

docs/generated/eventlog.md

@@ -1827,6 +1827,28 @@ are only emitted via external logging.
 Events in this category are logged to the `SQL_PERF` channel.
 
 
+### `large_row`
+
+An event of type `large_row` is recorded when a row larger than cluster setting
+`sql.mutations.max_row_size.warn` is added to the database. A single
+statement could cause multiple LargeRow events to be emitted.
+
+
+
+
+#### 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 |
+| `RowSize` | | no |
+| `MaxRowSize` | | no |
+| `TableID` | | no |
+| `IndexID` | | no |
+| `FamilyID` | | no |
+| `Key` | | yes |
+
 ### `slow_query`
 
 An event of type `slow_query` is recorded when a query triggers the "slow query" condition.
@@ -1877,6 +1899,29 @@ are only emitted via external logging.
 Events in this category are logged to the `SQL_INTERNAL_PERF` channel.
 
 
+### `large_row_internal`
+
+An event of type `large_row_internal` is recorded when a row larger than cluster setting
+`sql.mutations.max_row_size.warn` is added to the database by an internal
+query. A single internal query could cause multiple LargeRowInternal events
+to be emitted.
+
+
+
+
+#### 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 |
+| `RowSize` | | no |
+| `MaxRowSize` | | no |
+| `TableID` | | no |
+| `IndexID` | | no |
+| `FamilyID` | | no |
+| `Key` | | yes |
+
 ### `slow_query_internal`
 
 An event of type `slow_query_internal` is recorded when a query triggers the "slow query" condition,

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

@@ -128,6 +128,7 @@ sql.metrics.statement_details.plan_collection.period duration 5m0s the time unti
 sql.metrics.statement_details.threshold duration 0s minimum execution time to cause statement statistics to be collected. If configured, no transaction stats are collected.
 sql.metrics.transaction_details.enabled boolean true collect per-application transaction statistics
 sql.multiregion.drop_primary_region.enabled boolean true allows dropping the PRIMARY REGION of a database if it is the last region
+sql.mutations.max_row_size.warn byte size 64 MiB maximum size of row (or column family if multiple column families are in use) that SQL can write to the KV store, above which an event is logged to SQL_PERF or SQL_INTERNAL_PERF (setting to 0 disables large row logging)
 sql.notices.enabled boolean true enable notices in the server/client protocol being sent
 sql.optimizer.uniqueness_checks_for_gen_random_uuid.enabled boolean false if enabled, uniqueness checks may be planned for mutations of UUID columns updated with gen_random_uuid(); otherwise, uniqueness is assumed due to near-zero collision probability
 sql.spatial.experimental_box2d_comparison_operators.enabled boolean false enables the use of certain experimental box2d comparison operators

docs/generated/settings/settings.html

@@ -132,6 +132,7 @@
 <tr><td><code>sql.metrics.statement_details.threshold</code></td><td>duration</td><td><code>0s</code></td><td>minimum execution time to cause statement statistics to be collected. If configured, no transaction stats are collected.</td></tr>
 <tr><td><code>sql.metrics.transaction_details.enabled</code></td><td>boolean</td><td><code>true</code></td><td>collect per-application transaction statistics</td></tr>
 <tr><td><code>sql.multiregion.drop_primary_region.enabled</code></td><td>boolean</td><td><code>true</code></td><td>allows dropping the PRIMARY REGION of a database if it is the last region</td></tr>
+<tr><td><code>sql.mutations.max_row_size.warn</code></td><td>byte size</td><td><code>64 MiB</code></td><td>maximum size of row (or column family if multiple column families are in use) that SQL can write to the KV store, above which an event is logged to SQL_PERF or SQL_INTERNAL_PERF (setting to 0 disables large row logging)</td></tr>
 <tr><td><code>sql.notices.enabled</code></td><td>boolean</td><td><code>true</code></td><td>enable notices in the server/client protocol being sent</td></tr>
 <tr><td><code>sql.optimizer.uniqueness_checks_for_gen_random_uuid.enabled</code></td><td>boolean</td><td><code>false</code></td><td>if enabled, uniqueness checks may be planned for mutations of UUID columns updated with gen_random_uuid(); otherwise, uniqueness is assumed due to near-zero collision probability</td></tr>
 <tr><td><code>sql.spatial.experimental_box2d_comparison_operators.enabled</code></td><td>boolean</td><td><code>false</code></td><td>enables the use of certain experimental box2d comparison operators</td></tr>

pkg/ccl/importccl/import_stmt.go

@@ -2512,7 +2512,9 @@ func (r *importResumer) dropTables(
  // older-format (v1.1) descriptor. This enables ClearTableData to use a
  // RangeClear for faster data removal, rather than removing by chunks.
  empty[i].TableDesc().DropTime = dropTime
- if err := gcjob.ClearTableData(ctx, execCfg.DB, execCfg.DistSender, execCfg.Codec, empty[i]); err != nil {
+ if err := gcjob.ClearTableData(
+ ctx, execCfg.DB, execCfg.DistSender, execCfg.Codec, &execCfg.Settings.SV, empty[i],
+ ); err != nil {
  return errors.Wrapf(err, "clearing data for table %d", empty[i].GetID())
  }
  }

pkg/kv/kvserver/replica.go

@@ -90,15 +90,20 @@ var disableSyncRaftLog = settings.RegisterBoolSetting(
  false,
 )
 
-// MaxCommandSizeFloor is the minimum allowed value for the MaxCommandSize
-// cluster setting.
-const MaxCommandSizeFloor = 4 << 20 // 4MB
+const (
+ // MaxCommandSizeFloor is the minimum allowed value for the
+ // kv.raft.command.max_size cluster setting.
+ MaxCommandSizeFloor = 4 << 20 // 4MB
+ // MaxCommandSizeDefault is the default for the kv.raft.command.max_size
+ // cluster setting.
+ MaxCommandSizeDefault = 64 << 20
+)
 
 // MaxCommandSize wraps "kv.raft.command.max_size".
 var MaxCommandSize = settings.RegisterByteSizeSetting(
  "kv.raft.command.max_size",
  "maximum size of a raft command",
- 64<<20,
+ MaxCommandSizeDefault,
  func(size int64) error {
  if size < MaxCommandSizeFloor {
  return fmt.Errorf("max_size must be greater than %s", humanizeutil.IBytes(MaxCommandSizeFloor))

pkg/sql/backfill.go

@@ -799,7 +799,9 @@ func TruncateInterleavedIndexes(
  resumeAt := resume
  // Make a new txn just to drop this chunk.
  if err := db.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
- rd := row.MakeDeleter(codec, table, nil /* requestedCols */)
+ rd := row.MakeDeleter(
+ codec, table, nil /* requestedCols */, &execCfg.Settings.SV, true, /* internal */
+ )
  td := tableDeleter{rd: rd, alloc: alloc}
  if err := td.init(ctx, txn, nil /* *tree.EvalContext */); err != nil {
  return err
@@ -878,7 +880,10 @@ func (sc *SchemaChanger) truncateIndexes(
  if err != nil {
  return err
  }
- rd := row.MakeDeleter(sc.execCfg.Codec, tableDesc, nil /* requestedCols */)
+ rd := row.MakeDeleter(
+ sc.execCfg.Codec, tableDesc, nil /* requestedCols */, &sc.settings.SV,
+ true, /* internal */
+ )
  td := tableDeleter{rd: rd, alloc: alloc}
  if err := td.init(ctx, txn, nil /* *tree.EvalContext */); err != nil {
  return err
@@ -2473,7 +2478,10 @@ func indexTruncateInTxn(
  alloc := &rowenc.DatumAlloc{}
  var sp roachpb.Span
  for done := false; !done; done = sp.Key == nil {
- rd := row.MakeDeleter(execCfg.Codec, tableDesc, nil /* requestedCols */)
+ rd := row.MakeDeleter(
+ execCfg.Codec, tableDesc, nil /* requestedCols */, &execCfg.Settings.SV,
+ evalCtx.SessionData.Internal,
+ )
  td := tableDeleter{rd: rd, alloc: alloc}
  if err := td.init(ctx, txn, evalCtx); err != nil {
  return err

pkg/sql/backfill/backfill.go

@@ -266,6 +266,8 @@ func (cb *ColumnBackfiller) RunColumnBackfillChunk(
  requestedCols,
  row.UpdaterOnlyColumns,
  &cb.alloc,
+ &cb.evalCtx.Settings.SV,
+ cb.evalCtx.SessionData.Internal,
  )
  if err != nil {
  return roachpb.Key{}, err

pkg/sql/create_table.go

@@ -455,7 +455,10 @@ func (n *createTableNode) startExec(params runParams) error {
  params.ExecCfg().Codec,
  desc.ImmutableCopy().(catalog.TableDescriptor),
  desc.PublicColumns(),
- params.p.alloc)
+ params.p.alloc,
+ &params.ExecCfg().Settings.SV,
+ params.p.SessionData().Internal,
+ )
  if err != nil {
  return err
  }

pkg/sql/gcjob/BUILD.bazel

@@ -24,6 +24,7 @@ go_library(
  "//pkg/kv/kvserver/protectedts",
  "//pkg/kv/kvserver/protectedts/ptpb:ptpb_go_proto",
  "//pkg/roachpb:with-mocks",
+ "//pkg/settings",
  "//pkg/settings/cluster",
  "//pkg/sql",
  "//pkg/sql/catalog",

pkg/sql/gcjob/table_garbage_collection.go

@@ -19,6 +19,7 @@ import (
  "github.com/cockroachdb/cockroach/pkg/kv"
  "github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvcoord"
  "github.com/cockroachdb/cockroach/pkg/roachpb"
+ "github.com/cockroachdb/cockroach/pkg/settings"
  "github.com/cockroachdb/cockroach/pkg/sql"
  "github.com/cockroachdb/cockroach/pkg/sql/catalog"
  "github.com/cockroachdb/cockroach/pkg/sql/catalog/catalogkv"
@@ -65,7 +66,9 @@ func gcTables(
  }
 
  // First, delete all the table data.
- if err := ClearTableData(ctx, execCfg.DB, execCfg.DistSender, execCfg.Codec, table); err != nil {
+ if err := ClearTableData(
+ ctx, execCfg.DB, execCfg.DistSender, execCfg.Codec, &execCfg.Settings.SV, table,
+ ); err != nil {
  return errors.Wrapf(err, "clearing data for table %d", table.GetID())
  }
 
@@ -86,6 +89,7 @@ func ClearTableData(
  db *kv.DB,
  distSender *kvcoord.DistSender,
  codec keys.SQLCodec,
+ sv *settings.Values,
  table catalog.TableDescriptor,
 ) error {
  // If DropTime isn't set, assume this drop request is from a version
@@ -95,7 +99,7 @@ func ClearTableData(
  // cleaned up.
  if table.GetDropTime() == 0 || table.IsInterleaved() {
  log.Infof(ctx, "clearing data in chunks for table %d", table.GetID())
- return sql.ClearTableDataInChunks(ctx, db, codec, table, false /* traceKV */)
+ return sql.ClearTableDataInChunks(ctx, db, codec, sv, table, false /* traceKV */)
  }
  log.Infof(ctx, "clearing data for table %d", table.GetID())
 

pkg/sql/opt_exec_factory.go

@@ -1279,7 +1279,14 @@ func (ef *execFactory) ConstructInsert(
 
  // Create the table inserter, which does the bulk of the work.
  ri, err := row.MakeInserter(
- ctx, ef.planner.txn, ef.planner.ExecCfg().Codec, tabDesc, cols, ef.planner.alloc,
+ ctx,
+ ef.planner.txn,
+ ef.planner.ExecCfg().Codec,
+ tabDesc,
+ cols,
+ ef.planner.alloc,
+ &ef.planner.ExecCfg().Settings.SV,
+ ef.planner.SessionData().Internal,
  )
  if err != nil {
  return nil, err
@@ -1345,7 +1352,14 @@ func (ef *execFactory) ConstructInsertFastPath(
 
  // Create the table inserter, which does the bulk of the work.
  ri, err := row.MakeInserter(
- ctx, ef.planner.txn, ef.planner.ExecCfg().Codec, tabDesc, cols, ef.planner.alloc,
+ ctx,
+ ef.planner.txn,
+ ef.planner.ExecCfg().Codec,
+ tabDesc,
+ cols,
+ ef.planner.alloc,
+ &ef.planner.ExecCfg().Settings.SV,
+ ef.planner.SessionData().Internal,
  )
  if err != nil {
  return nil, err
@@ -1450,6 +1464,8 @@ func (ef *execFactory) ConstructUpdate(
  fetchCols,
  row.UpdaterDefault,
  ef.planner.alloc,
+ &ef.planner.ExecCfg().Settings.SV,
+ ef.planner.SessionData().Internal,
  )
  if err != nil {
  return nil, err
@@ -1550,6 +1566,8 @@ func (ef *execFactory) ConstructUpsert(
  tabDesc,
  insertCols,
  ef.planner.alloc,
+ &ef.planner.ExecCfg().Settings.SV,
+ ef.planner.SessionData().Internal,
  )
  if err != nil {
  return nil, err
@@ -1565,6 +1583,8 @@ func (ef *execFactory) ConstructUpsert(
  fetchCols,
  row.UpdaterDefault,
  ef.planner.alloc,
+ &ef.planner.ExecCfg().Settings.SV,
+ ef.planner.SessionData().Internal,
  )
  if err != nil {
  return nil, err
@@ -1636,7 +1656,13 @@ func (ef *execFactory) ConstructDelete(
  // the deleter derives the columns that need to be fetched. By contrast, the
  // CBO will have already determined the set of fetch columns, and passes
  // those sets into the deleter (which will basically be a no-op).
- rd := row.MakeDeleter(ef.planner.ExecCfg().Codec, tabDesc, fetchCols)
+ rd := row.MakeDeleter(
+ ef.planner.ExecCfg().Codec,
+ tabDesc,
+ fetchCols,
+ &ef.planner.ExecCfg().Settings.SV,
+ ef.planner.SessionData().Internal,
+ )
 
  // Now make a delete node. We use a pool.
  del := deleteNodePool.Get().(*deleteNode)

pkg/sql/row/BUILD.bazel

@@ -25,8 +25,10 @@ go_library(
  "//pkg/jobs/jobspb",
  "//pkg/keys",
  "//pkg/kv",
+ "//pkg/kv/kvserver",
  "//pkg/kv/kvserver/concurrency/lock",
  "//pkg/roachpb:with-mocks",
+ "//pkg/settings",
  "//pkg/sql/catalog",
  "//pkg/sql/catalog/catalogkeys",
  "//pkg/sql/catalog/catalogkv",
@@ -52,6 +54,7 @@ go_library(
  "//pkg/util/encoding",
  "//pkg/util/hlc",
  "//pkg/util/log",
+ "//pkg/util/log/eventpb",
  "//pkg/util/mon",
  "//pkg/util/sequence",
  "//pkg/util/syncutil",

pkg/sql/row/deleter.go

@@ -16,6 +16,7 @@ import (
  "github.com/cockroachdb/cockroach/pkg/keys"
  "github.com/cockroachdb/cockroach/pkg/kv"
  "github.com/cockroachdb/cockroach/pkg/roachpb"
+ "github.com/cockroachdb/cockroach/pkg/settings"
  "github.com/cockroachdb/cockroach/pkg/sql/catalog"
  "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
  "github.com/cockroachdb/cockroach/pkg/sql/rowenc"
@@ -41,7 +42,11 @@ type Deleter struct {
 // FetchCols; otherwise, all columns that are part of the key of any index
 // (either primary or secondary) are included in FetchCols.
 func MakeDeleter(
- codec keys.SQLCodec, tableDesc catalog.TableDescriptor, requestedCols []catalog.Column,
+ codec keys.SQLCodec,
+ tableDesc catalog.TableDescriptor,
+ requestedCols []catalog.Column,
+ sv *settings.Values,
+ internal bool,
 ) Deleter {
  indexes := tableDesc.DeletableNonPrimaryIndexes()
 
@@ -86,7 +91,7 @@ func MakeDeleter(
  }
 
  rd := Deleter{
- Helper: newRowHelper(codec, tableDesc, indexes),
+ Helper: newRowHelper(codec, tableDesc, indexes, sv, internal),
  FetchCols: fetchCols,
  FetchColIDtoRowIndex: fetchColIDtoRowIndex,
  }