sql: fix a couple of edge case bugs with CHECK EXTERNAL CONNECTION

This commit fixes a couple of edge case bugs with CHECK EXTERNAL
CONNECTION implementation. Namely:
- previously, the execution goroutine (that is created in `startExec`)
would use the tracing span that is finished, which is not allowed. This
is now fixed by deriving a new tracing span.
- also previously we would call `MoveToDraining` from an auxiliary
goroutine of the cloud check processor when the context cancellation is
observed. This would race with `MoveToDraining` call from the main
goroutine in `Next`. The implicit contract of this helper method is that
it's only called from the main goroutine, and otherwise it can lead to
"MoveToDraining called in state × with err" errors. This is now fixed by
removing the call from the auxiliary goroutine since it's actually not
needed.

Additionally, this commit simplifies dealing with the `rows` channel
a bit.

I decided to omit the release note given we recently merged a fix that
contained one.

Release note: None

Author: yuzefovich

pkg/sql/BUILD.bazel

@@ -655,6 +655,7 @@ go_test(
         "backfill_test.go",
         "builtin_mem_usage_test.go",
         "builtin_test.go",
+        "check_external_connection_test.go",
         "check_test.go",
         "closed_session_cache_test.go",
         "comment_on_column_test.go",

pkg/sql/check_external_connection.go

@@ -50,7 +50,7 @@ func (n *checkExternalConnectionNode) startExec(params runParams) error {
 		return err
 	}
 
-	ctx, span := tracing.ChildSpan(params.ctx, "CheckExternalConnection")
+	ctx, span := tracing.ChildSpan(params.ctx, "CheckExternalConnection-planning")
 	defer span.Finish()
 
 	store, err := params.ExecCfg().DistSQLSrv.ExternalStorageFromURI(ctx, n.loc, params.p.User())
@@ -103,9 +103,14 @@ func (n *checkExternalConnectionNode) startExec(params runParams) error {
 		return nil
 	})
 
-	grp := ctxgroup.WithContext(ctx)
+	grp := ctxgroup.WithContext(params.ctx)
 	n.execGrp = grp
 	grp.GoCtx(func(ctx context.Context) error {
+		// Derive a separate tracing span since the planning one will be
+		// finished when the main goroutine exits from startExec.
+		ctx, span := tracing.ChildSpan(ctx, "CheckExternalConnection-execution")
+		defer span.Finish()
+
 		recv := MakeDistSQLReceiver(
 			ctx,
 			rowWriter,
@@ -128,9 +133,6 @@ func (n *checkExternalConnectionNode) startExec(params runParams) error {
 }
 
 func (n *checkExternalConnectionNode) Next(params runParams) (bool, error) {
-	if n.rows == nil {
-		return false, nil
-	}
 	select {
 	case <-params.ctx.Done():
 		return false, params.ctx.Err()
@@ -149,7 +151,6 @@ func (n *checkExternalConnectionNode) Values() tree.Datums {
 
 func (n *checkExternalConnectionNode) Close(_ context.Context) {
 	_ = n.execGrp.Wait()
-	n.rows = nil
 }
 
 func (n *checkExternalConnectionNode) parseParams(params runParams) error {

pkg/sql/check_external_connection_test.go

@@ -0,0 +1,45 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package sql_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/testutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/randutil"
+)
+
+func TestCheckExternalConnection(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	rng, _ := randutil.NewTestRand()
+	dir, dirCleanupFn := testutils.TempDir(t)
+	defer dirCleanupFn()
+	s, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{
+		ExternalIODir: dir,
+	})
+	defer s.Stopper().Stop(ctx)
+
+	runner := sqlutils.MakeSQLRunner(sqlDB)
+	runner.Exec(t, "CREATE EXTERNAL CONNECTION foo_conn AS 'nodelocal://1/foo';")
+	query := "CHECK EXTERNAL CONNECTION 'nodelocal://1/foo';"
+	// Should execute successfully without a statement timeout.
+	runner.Exec(t, query)
+	// Run with a random statement timeout which will likely make the query
+	// fail. We don't care whether it does nor which error is returned as long
+	// as the process doesn't crash.
+	runner.Exec(t, fmt.Sprintf("SET statement_timeout='%dms'", rng.Intn(100)+1))
+	_, _ = sqlDB.Exec(query)
+}

pkg/sql/cloud_check_processor.go

@@ -216,7 +216,10 @@ func (p *proc) Start(ctx context.Context) {
 
 	if err := p.FlowCtx.Stopper().RunAsyncTask(p.Ctx(), "cloudcheck.proc", func(ctx context.Context) {
 		defer close(p.results)
-		if err := ctxgroup.GroupWorkers(ctx, concurrency, func(ctx context.Context, _ int) error {
+		// We're ignoring the context cancellation error (which is the only one
+		// possible) because the main goroutine will observe it on its own
+		// anyway in Next.
+		_ = ctxgroup.GroupWorkers(ctx, concurrency, func(ctx context.Context, _ int) error {
 			select {
 			case p.results <- checkURI(
 				ctx,
@@ -229,9 +232,7 @@ func (p *proc) Start(ctx context.Context) {
 			case <-ctx.Done():
 				return ctx.Err()
 			}
-		}); err != nil {
-			p.MoveToDraining(err)
-		}
+		})
 	}); err != nil {
 		p.MoveToDraining(err)
 	}

pkg/sql/execinfra/processorsbase.go

@@ -523,6 +523,9 @@ const (
 //
 // An error can be optionally passed. It will be the first piece of metadata
 // returned by DrainHelper().
+//
+// MoveToDraining should only be called from the main goroutine of the
+// processor.
 func (pb *ProcessorBaseNoHelper) MoveToDraining(err error) {
 	if pb.State != StateRunning {
 		// Calling MoveToDraining in any state is allowed in order to facilitate the