Merge #61187

61187: bulkio: Correctly check stream replication feature gate. r=miretskiy a=miretskiy Verify experimental stream replication feature is enabled only after making sure that the statement we're attempting to execute is a replication stream statement. Release Justification: minor bug fix. Release Notes: None Co-authored-by: Yevgeniy Miretskiy <[email protected]>
cockroachdb · Feb 26, 2021 · 7df2237 · 7df2237
2 parents ed15112 + 496b012
commit 7df2237
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 9 deletions.
diff --git a/pkg/ccl/streamingccl/streamingest/stream_ingestion_planning.go b/pkg/ccl/streamingccl/streamingest/stream_ingestion_planning.go
@@ -40,6 +40,11 @@ func streamIngestionJobDescription(
 func ingestionPlanHook(
 	ctx context.Context, stmt tree.Statement, p sql.PlanHookState,
 ) (sql.PlanHookRowFn, colinfo.ResultColumns, []sql.PlanNode, bool, error) {
+	ingestionStmt, ok := stmt.(*tree.StreamIngestion)
+	if !ok {
+		return nil, nil, nil, false, nil
+	}
+
 	// Check if the experimental feature is enabled.
 	if !p.SessionData().EnableStreamReplication {
 		return nil, nil, nil, false, errors.WithTelemetry(
@@ -54,11 +59,6 @@ func ingestionPlanHook(
 		)
 	}
 
-	ingestionStmt, ok := stmt.(*tree.StreamIngestion)
-	if !ok {
-		return nil, nil, nil, false, nil
-	}
-
 	fromFn, err := p.TypeAsStringArray(ctx, tree.Exprs(ingestionStmt.From), "INGESTION")
 	if err != nil {
 		return nil, nil, nil, false, err

diff --git a/pkg/ccl/streamingccl/streamproducer/replication_stream_planning.go b/pkg/ccl/streamingccl/streamproducer/replication_stream_planning.go
@@ -167,6 +167,10 @@ var replicationStreamHeader = colinfo.ResultColumns{
 func createReplicationStreamHook(
 	ctx context.Context, stmt tree.Statement, p sql.PlanHookState,
 ) (sql.PlanHookRowFn, colinfo.ResultColumns, []sql.PlanNode, bool, error) {
+	stream, ok := stmt.(*tree.ReplicationStream)
+	if !ok {
+		return nil, nil, nil, false, nil
+	}
 	if !p.SessionData().EnableStreamReplication {
 		return nil, nil, nil, false, errors.WithTelemetry(
 			pgerror.WithCandidateCode(
@@ -180,10 +184,6 @@ func createReplicationStreamHook(
 		)
 	}
 
-	stream, ok := stmt.(*tree.ReplicationStream)
-	if !ok {
-		return nil, nil, nil, false, nil
-	}
 	eval, err := makeReplicationStreamEval(ctx, p, stream)
 	if err != nil {
 		return nil, nil, nil, false, err