changefeedccl: disable unsafe sql updates in nemeses test

Some queries generated by sql smith can hang due to being unsafe (e.g.,
executing many self-joins). This PR enables the session setting
`sql_safe_updates` in order to reject these kinds of queries, then
redisables it for the remainder of the test. It also makes some minor
improvements, such as only populating initial table state via sql smith
if sql smit is enabled, logging to make it easy to reproduce
the table's initial state before the changefeed is created, and only
allowing normal sql mutations with sql smith.

Epic: none
Fixes: #140591

Release note: none

Author: rharding6373

pkg/ccl/changefeedccl/cdctest/nemeses.go

@@ -375,29 +375,36 @@ func RunNemesis(
 	// Initialize table rows by repeatedly running the `openTxn` transition,
 	// then randomly either committing or rolling back transactions. This will
 	// leave some committed rows.
-	for i := 0; i < ns.rowCount*5; i++ {
-		payload, err := newOpenTxnPayload(ns)
-		if err != nil {
-			return nil, err
-		}
-		if err := openTxn(fsm.Args{Ctx: ctx, Extended: ns, Payload: payload}); err != nil {
-			return nil, err
-		}
-		// Randomly commit or rollback, but commit at least one row to the table.
-		if rand.Intn(3) < 2 || i == 0 {
-			if err := commit(fsm.Args{Ctx: ctx, Extended: ns}); err != nil {
+	// If sql smith is enabled, we'll insert rows below instead.
+	if !nOp.EnableSQLSmith {
+		for i := 0; i < ns.rowCount*5; i++ {
+			payload, err := newOpenTxnPayload(ns)
+			if err != nil {
 				return nil, err
 			}
-		} else {
-			if err := rollback(fsm.Args{Ctx: ctx, Extended: ns}); err != nil {
+			if err := openTxn(fsm.Args{Ctx: ctx, Extended: ns, Payload: payload}); err != nil {
 				return nil, err
 			}
+			// Randomly commit or rollback, but commit at least one row to the table.
+			if rand.Intn(3) < 2 || i == 0 {
+				if err := commit(fsm.Args{Ctx: ctx, Extended: ns}); err != nil {
+					return nil, err
+				}
+			} else {
+				if err := rollback(fsm.Args{Ctx: ctx, Extended: ns}); err != nil {
+					return nil, err
+				}
+			}
 		}
 	}
 
 	if nOp.EnableSQLSmith {
+		// Some unsafe queries can hang, so avoid them.
+		if _, err := db.Exec("SET sql_safe_updates=true"); err != nil {
+			return nil, err
+		}
 		queryGen, _ := sqlsmith.NewSmither(db, rng,
-			sqlsmith.MutationsOnly(),
+			sqlsmith.InsUpdDelOnly(),
 			sqlsmith.SetScalarComplexity(0.5),
 			sqlsmith.SetComplexity(0.1),
 			// TODO(#129072): Reenable cross joins when the likelihood of generating
@@ -408,6 +415,7 @@ func RunNemesis(
 			// supported under the serializable isolation.
 			sqlsmith.DisableIsolationChange(),
 		)
+
 		defer queryGen.Close()
 		const numInserts = 100
 		time := timeutil.Now()
@@ -422,6 +430,28 @@ func RunNemesis(
 				continue
 			}
 		}
+		// Reenable unsafe queries, since the test sometimes alters tables.
+		if _, err := db.Exec("SET sql_safe_updates=false"); err != nil {
+			return nil, err
+		}
+	}
+
+	// Print the contents of the table in a way that can be reproduced.
+	rows, err := db.Query("SELECT * FROM foo")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	// Iterate over the rows and print them
+	for rows.Next() {
+		var id int
+		var ts string
+		if err := rows.Scan(&id, &ts); err != nil {
+			log.Infof(ctx, "# skipping row because error: %s", err)
+			continue
+		}
+		log.Infof(ctx, "INSERT INTO foo (id,ts) VALUES (%d, %s);", id, ts)
 	}
 
 	cfo := newChangefeedOption(testName)

pkg/internal/sqlsmith/sqlsmith.go

@@ -470,6 +470,17 @@ var InsUpdOnly = simpleOption("inserts and updates only", func(s *Smither) {
 	}
 })
 
+// InsUpdDelOnly causes the Smither to emit 80% INSERT, 10% UPDATE,
+// 10% DELETE statements.
+var InsUpdDelOnly = simpleOption("inserts updates and deletes only",
+	func(s *Smither) {
+		s.stmtWeights = []statementWeight{
+			{8, makeInsert},
+			{1, makeUpdate},
+			{1, makeDelete},
+		}
+	})
+
 // IgnoreFNs causes the Smither to ignore functions that match the regex.
 func IgnoreFNs(regex string) SmitherOption {
 	r := regexp.MustCompile(regex)