more wip

Author: aerfrei

pkg/ccl/changefeedccl/changefeed_test.go

@@ -122,8 +122,6 @@ func TestChangefeedBasics(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	skip.WithIssue(t, 148858) // initial scan?
-
 	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
 		sqlDB := sqlutils.MakeSQLRunner(s.DB)
 		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
@@ -138,7 +136,7 @@ func TestChangefeedBasics(t *testing.T) {
 			`foo: [0]->{"after": {"a": 0, "b": "updated"}}`,
 		})
 
-		sqlDB.Exec(t, `INSERT INTO foo VALUES (1, 'a')`) //, (2, 'b')`)
+		sqlDB.Exec(t, `INSERT INTO foo VALUES (1, 'a'), (2, 'b')`)
 		assertPayloads(t, foo, []string{
 			`foo: [1]->{"after": {"a": 1, "b": "a"}}`,
 			`foo: [2]->{"after": {"a": 2, "b": "b"}}`,
@@ -3350,7 +3348,6 @@ WITH resolved='100ms', min_checkpoint_frequency='1ns', no_initial_scan`)
 // TODO: remove this test when the legacy schema changer is  deprecated.
 func TestChangefeedSchemaChangeAllowBackfill_Legacy(t *testing.T) {
 	defer leaktest.AfterTest(t)()
-	skip.WithIssue(t, 148858) // Also uses initial scan.
 	defer log.Scope(t).Close(t)
 
 	testutils.SetVModule(t, "kv_feed=2,changefeed_processors=2")
@@ -3414,7 +3411,8 @@ func TestChangefeedSchemaChangeAllowBackfill_Legacy(t *testing.T) {
 			sqlDB.Exec(t, `CREATE TABLE add_col_comp (a INT PRIMARY KEY, b INT AS (a + 5) STORED)`)
 			sqlDB.Exec(t, `INSERT INTO add_col_comp VALUES (1)`)
 			sqlDB.Exec(t, `INSERT INTO add_col_comp (a) VALUES (2)`)
-			addColComp := feed(t, f, `CREATE CHANGEFEED FOR add_col_comp WITH updated`)
+			addColComp := feed(t, f, `CREATE CHANGEFEED FOR add_col_comp WITH updated`,
+				optOutOfDBLevelChangefeedUnwatchedTables)
 			defer closeFeed(t, addColComp)
 			assertPayloadsStripTs(t, addColComp, []string{
 				`add_col_comp: [1]->{"after": {"a": 1, "b": 6}}`,
@@ -3440,7 +3438,8 @@ func TestChangefeedSchemaChangeAllowBackfill_Legacy(t *testing.T) {
 			sqlDB.Exec(t, `CREATE TABLE drop_column (a INT PRIMARY KEY, b STRING)`)
 			sqlDB.Exec(t, `INSERT INTO drop_column VALUES (1, '1')`)
 			sqlDB.Exec(t, `INSERT INTO drop_column VALUES (2, '2')`)
-			dropColumn := feed(t, f, `CREATE CHANGEFEED FOR drop_column WITH updated`)
+			dropColumn := feed(t, f, `CREATE CHANGEFEED FOR drop_column WITH updated`,
+				optOutOfDBLevelChangefeedUnwatchedTables)
 			defer closeFeed(t, dropColumn)
 			assertPayloadsStripTs(t, dropColumn, []string{
 				`drop_column: [1]->{"after": {"a": 1, "b": "1"}}`,
@@ -3482,7 +3481,8 @@ func TestChangefeedSchemaChangeAllowBackfill_Legacy(t *testing.T) {
 				Changefeed.(*TestingKnobs)
 			knobs.BeforeEmitRow = waitSinkHook
 
-			multipleAlters := feed(t, f, `CREATE CHANGEFEED FOR multiple_alters WITH updated`)
+			multipleAlters := feed(t, f, `CREATE CHANGEFEED FOR multiple_alters WITH updated`,
+				optOutOfDBLevelChangefeedUnwatchedTables)
 			defer closeFeed(t, multipleAlters)
 			assertPayloadsStripTs(t, multipleAlters, []string{
 				`multiple_alters: [1]->{"after": {"a": 1, "b": "1"}}`,
@@ -3551,7 +3551,6 @@ func TestChangefeedSchemaChangeAllowBackfill(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	skip.WithIssue(t, 148858) // Also uses initial scan.
 	testutils.SetVModule(t, "kv_feed=2,changefeed_processors=2")
 
 	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
@@ -3598,7 +3597,8 @@ func TestChangefeedSchemaChangeAllowBackfill(t *testing.T) {
 			sqlDB.Exec(t, `CREATE TABLE add_col_comp (a INT PRIMARY KEY, b INT AS (a + 5) STORED)`)
 			sqlDB.Exec(t, `INSERT INTO add_col_comp VALUES (1)`)
 			sqlDB.Exec(t, `INSERT INTO add_col_comp (a) VALUES (2)`)
-			addColComp := feed(t, f, `CREATE CHANGEFEED FOR add_col_comp WITH updated`)
+			addColComp := feed(t, f, `CREATE CHANGEFEED FOR add_col_comp WITH updated`,
+				optOutOfDBLevelChangefeedUnwatchedTables)
 			defer closeFeed(t, addColComp)
 			assertPayloadsStripTs(t, addColComp, []string{
 				`add_col_comp: [1]->{"after": {"a": 1, "b": 6}}`,
@@ -3618,7 +3618,8 @@ func TestChangefeedSchemaChangeAllowBackfill(t *testing.T) {
 			sqlDB.Exec(t, `CREATE TABLE drop_column (a INT PRIMARY KEY, b STRING)`)
 			sqlDB.Exec(t, `INSERT INTO drop_column VALUES (1, '1')`)
 			sqlDB.Exec(t, `INSERT INTO drop_column VALUES (2, '2')`)
-			dropColumn := feed(t, f, `CREATE CHANGEFEED FOR drop_column WITH updated`)
+			dropColumn := feed(t, f, `CREATE CHANGEFEED FOR drop_column WITH updated`,
+				optOutOfDBLevelChangefeedUnwatchedTables)
 			defer closeFeed(t, dropColumn)
 			assertPayloadsStripTs(t, dropColumn, []string{
 				`drop_column: [1]->{"after": {"a": 1, "b": "1"}}`,
@@ -3658,7 +3659,8 @@ func TestChangefeedSchemaChangeAllowBackfill(t *testing.T) {
 				Changefeed.(*TestingKnobs)
 			knobs.BeforeEmitRow = waitSinkHook
 
-			multipleAlters := feed(t, f, `CREATE CHANGEFEED FOR multiple_alters WITH updated`)
+			multipleAlters := feed(t, f, `CREATE CHANGEFEED FOR multiple_alters WITH updated`,
+				optOutOfDBLevelChangefeedUnwatchedTables)
 			defer closeFeed(t, multipleAlters)
 			assertPayloadsStripTs(t, multipleAlters, []string{
 				`multiple_alters: [1]->{"after": {"a": 1, "b": "1"}}`,
@@ -4252,13 +4254,12 @@ func TestChangefeedJobControl(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	skip.WithIssue(t, 148858) // ?
-
 	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
 		ChangefeedJobPermissionsTestSetup(t, s)
 
 		createFeed := func(stmt string) (cdctest.EnterpriseTestFeed, func()) {
-			successfulFeed := feed(t, f, stmt)
+			successfulFeed := feed(t, f, stmt, optOutOfMetamorphicDBLevelChangefeed{
+				reason: "tests table level changefeed permissions"})
 			closeCf := func() {
 				closeFeed(t, successfulFeed)
 			}
@@ -5668,7 +5669,7 @@ func TestChangefeedResolvedNotice(t *testing.T) {
 
 func TestChangefeedLowFrequencyNotices(t *testing.T) {
 	defer leaktest.AfterTest(t)()
-	skip.WithIssue(t, 148858) // I think these notices have an issue
+	skip.WithIssue(t, 148858) // REAL ISSUE 2: These notices have an issue
 	defer log.Scope(t).Close(t)
 
 	cluster, _, cleanup := startTestCluster(t)
@@ -5724,16 +5725,9 @@ func TestChangefeedLowFrequencyNotices(t *testing.T) {
 	})
 }
 
-// Fails: this could be something real
-// test logs left over in: /tmp/cockroach/_tmp/f5cb02fa45891c8256821a187ced6fe9/logTestChangefeedOutputTopics3499146109
-// --- FAIL: TestChangefeedOutputTopics (7.67s)
-//
-//	--- FAIL: TestChangefeedOutputTopics/kafka (0.25s)
-//	--- FAIL: TestChangefeedOutputTopics/pubsub_v2 (0.22s)
-//	--- PASS: TestChangefeedOutputTopics/webhooks_does_not_emit_anything (0.24s)
 func TestChangefeedOutputTopics(t *testing.T) {
 	defer leaktest.AfterTest(t)()
-	skip.WithIssue(t, 148858) // I don't know exactly what this is, could be real.
+	skip.WithIssue(t, 148858) // REAL ISSUE 2: Notices have an issue
 	defer log.Scope(t).Close(t)
 
 	cluster, _, cleanup := startTestCluster(t)
@@ -9125,7 +9119,6 @@ func TestChangefeedCheckpointSchemaChange(t *testing.T) {
 	defer log.Scope(t).Close(t)
 
 	skip.UnderRace(t)
-	skip.WithIssue(t, 148858) // has unwatched tables? complicated, look
 	skip.UnderShort(t)
 
 	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
@@ -12187,7 +12180,6 @@ func TestChangefeedAvroDecimalColumnWithDiff(t *testing.T) {
 func TestChangefeedProtectedTimestampUpdate(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
-	skip.WithIssue(t, 148858) // ?
 
 	verifyFunc := func() {}
 	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
@@ -12320,13 +12312,11 @@ WITH resolved='10ms', min_checkpoint_frequency='10ms', no_initial_scan`
 	cdcTest(t, testFn, feedTestForceSink("kafka"), withTxnRetries)
 }
 
-// Fails: this could be something real
 // TestChangefeedProtectedTimestampUpdateError tests that a changefeed that
 // errors while managing its protected timestamp records will increment the
 // manage PTS error counter.
 func TestChangefeedProtectedTimestampUpdateError(t *testing.T) {
 	defer leaktest.AfterTest(t)()
-	skip.WithIssue(t, 148858) // ?
 	defer log.Scope(t).Close(t)
 
 	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {

pkg/ccl/changefeedccl/protected_timestamps_test.go

@@ -45,7 +45,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
-	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -370,21 +369,19 @@ func TestChangefeedProtectedTimestamps(t *testing.T) {
 	}))
 }
 
-// Fails for DB level changefeeds: this could be real.
 // TestChangefeedAlterPTS is a regression test for (#103855).
 // It verifies that we do not lose track of existing PTS records nor create
 // extraneous PTS records when altering a changefeed by adding a table.
 func TestChangefeedAlterPTS(t *testing.T) {
 	defer leaktest.AfterTest(t)()
-	skip.WithIssue(t, 148858) // This is an issue with having too many PTS records.
 	defer log.Scope(t).Close(t)
 
 	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
 		sqlDB := sqlutils.MakeSQLRunner(s.DB)
 		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
 		sqlDB.Exec(t, `CREATE TABLE foo2 (a INT PRIMARY KEY, b STRING)`)
 		f2 := feed(t, f, `CREATE CHANGEFEED FOR table foo with protect_data_from_gc_on_pause,
-			resolved='1s', min_checkpoint_frequency='1s'`)
+			resolved='1s', min_checkpoint_frequency='1s'`, optOutOfDBLevelChangefeedUnwatchedTables)
 		defer closeFeed(t, f2)
 
 		getNumPTSRecords := func() int {