sql: CREATE TABLE AS in an explicit txn does not flush batched rows

Fixes: #81554

Previously, when running a CREATE TABLE AS in explicit
transactions, a single batch inserted all rows. This can
lead to errors running into the KV command size limits. To
address this, this patch will intermittently flush batches
KV, so that a single gigantic batch isn't used.

Release note (bug fix): CREATE TABLE AS in explicit transaction
would fail with an error if the size of the source table exceeded
the raft command size limit.

pkg/sql/create_table.go

@@ -578,6 +578,14 @@ func (n *createTableNode) startExec(params runParams) error {
 					break
 				}
 
+				// Periodically flush out the batches, so that we don't issue gigantic,
+				// raft commands.
+				if ti.currentBatchSize >= ti.maxBatchSize {
+					if err := tw.flushAndStartNewBatch(params.ctx); err != nil {
+						return err
+					}
+				}
+
 				// Populate the buffer.
 				copy(rowBuffer, n.sourcePlan.Values())
 

pkg/sql/logictest/testdata/logic_test/create_as

@@ -382,3 +382,17 @@ query I
 SELECT * FROM tab_from_seq
 ----
 2
+
+# Regression test for #81554, where tried to do gigantic batches for CTAS in
+# explicit transactions. Use a fixed command size, so that an error is decoupled
+# fom the default size.
+statement ok
+SET CLUSTER SETTING kv.raft.command.max_size='5m'
+
+statement ok
+BEGIN;
+CREATE TABLE source_tbl_huge AS SELECT 1::CHAR(256) FROM generate_series(1, 500000);
+COMMIT;
+
+statement ok
+SET CLUSTER SETTING kv.raft.command.max_size to default