changefeedccl: check that rangefeeds are enabled earlier

Pushing this check earlier means that for core changefeeds, the check
happens before the backfill. Previously an error would only be seen
after the backfill when we attempted to start the rangefeed, which
some users found confusing.

Release note: None

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -84,14 +84,6 @@ func changefeedPlanHook(
 		return nil, nil, nil, false, nil
 	}
 
-	if err := featureflag.CheckEnabled(
-		ctx,
-		p.ExecCfg(),
-		featureChangefeedEnabled,
-		"CHANGEFEED",
-	); err != nil {
-		return nil, nil, nil, false, err
-	}
 	var sinkURIFn func() (string, error)
 	var header colinfo.ResultColumns
 	unspecifiedSink := changefeedStmt.SinkURI == nil
@@ -132,6 +124,22 @@ func changefeedPlanHook(
 		ctx, span := tracing.ChildSpan(ctx, stmt.StatementTag())
 		defer span.Finish()
 
+		if err := featureflag.CheckEnabled(
+			ctx,
+			p.ExecCfg(),
+			featureChangefeedEnabled,
+			"CHANGEFEED",
+		); err != nil {
+			return err
+		}
+
+		// Changefeeds are based on the Rangefeed abstraction, which
+		// requires the `kv.rangefeed.enabled` setting to be true.
+		if !kvserver.RangefeedEnabled.Get(&p.ExecCfg().Settings.SV) {
+			return errors.Errorf("rangefeeds require the kv.rangefeed.enabled setting. See %s",
+				docs.URL(`change-data-capture.html#enable-rangefeeds-to-reduce-latency`))
+		}
+
 		ok, err := p.HasRoleOption(ctx, roleoption.CONTROLCHANGEFEED)
 		if err != nil {
 			return err
@@ -335,12 +343,6 @@ func changefeedPlanHook(
 			return changefeedbase.MaybeStripRetryableErrorMarker(err)
 		}
 
-		// Changefeeds are based on the Rangefeed abstraction, which requires the
-		// `kv.rangefeed.enabled` setting to be true.
-		if !kvserver.RangefeedEnabled.Get(&p.ExecCfg().Settings.SV) {
-			return errors.Errorf("rangefeeds require the kv.rangefeed.enabled setting. See %s",
-				docs.URL(`change-data-capture.html#enable-rangefeeds-to-reduce-latency`))
-		}
 		if err := utilccl.CheckEnterpriseEnabled(
 			p.ExecCfg().Settings, p.ExecCfg().ClusterID(), p.ExecCfg().Organization(), "CHANGEFEED",
 		); err != nil {

pkg/ccl/changefeedccl/changefeed_test.go

@@ -730,6 +730,7 @@ func TestChangefeedExternalIODisabled(t *testing.T) {
 		})
 		defer s.Stopper().Stop(ctx)
 		sqlDB := sqlutils.MakeSQLRunner(db)
+		sqlDB.Exec(t, serverSetupStatements)
 		sqlDB.Exec(t, "CREATE TABLE target_table (pk INT PRIMARY KEY)")
 		for _, proto := range disallowedSinkProtos {
 			sqlDB.ExpectErr(t, "Outbound IO is disabled by configuration, cannot create changefeed",

pkg/ccl/logictestccl/testdata/logic_test/changefeed

@@ -3,6 +3,9 @@
 statement ok
 CREATE TABLE t ()
 
+statement ok
+SET CLUSTER SETTING kv.rangefeed.enabled = true
+
 user testuser
 
 statement error permission denied to create changefeed