Merge #85847 #85874 #85889 #86035

85847: externalconn: write,list,read from ExternalStorage on `CREATE EXTERNAL CONNECTION` r=benbardin a=adityamaru

Previously, `CREATE EXTERNAL CONNECTION` for an `ExternalStorage` would
only parse the URI and perform basic validation on the URI construct.
This change teaches the factory to actually check the ExternalStorage is
reachable, by writing a sentinel file, listing it, and reading the file back
from it. By doing this any configuration errors will be surfaced to the user
at the time the external connection is being created rather than when it
is going to be used in a backup or changefeed etc. This offers a better UX
and stronger guarantees that the underlying resource is valid.

This change adds a testing knob to skip this validation step for certain
datadriven tests that are only concerned with the creation and dropping of
the external connection object, and do not have the correct credentials to
read,write and list from the external storage. The provider specific unit tests
in `pkg/ccl/cloudccl` however will perfrom this validation.

Release note (sql change): Creating an external connection to represent
and ExternalStorage will now write a sentinel file, list it, and read the contents
back to validate the underlying resource.

85874: sql: grant/revoke privileges for udfs r=chengxiong-ruan a=chengxiong-ruan

There are 4 commits:
(1) add syntax support for grant/revoke privileges of function.
(2) implement grant/revoke privilege for functions (mostly just reading correct descriptors)
(3) fix a minor issue when hydrating a mutable function which use table implicit type (mutable table implicit type cannot be fetched)
(4) populate `information_schema.role_routine_grants`, and we can test privileges on that.

85889: rangefeed: correctly handle intent outside of time-bound r=erikgrinaker a=tbg

Release note (bug fix): Changefeed jobs undergoing a catch-up scans
could fail with an error "expected provisional value for intent with ts
X, found Y". The problem would eitehr spontaneously resolve or be
rectified after a high-priority scan of the affected index. This bug is
now fixed.

86035: kvserver: use static tracing span name for AddSSTable r=[dt,erikgrinaker] a=stevendanna

Including the span bounds in the name of the tracing span means that
we can't aggregate information across all AddSSTable calls.

Release note: None

Co-authored-by: Aditya Maru <adityamaru@gmail.com>
Co-authored-by: Chengxiong Ruan <chengxiongruan@gmail.com>
Co-authored-by: Tobias Grieger <tobias.b.grieger@gmail.com>
Co-authored-by: Steven Danna <danna@cockroachlabs.com>

docs/generated/eventlog.md

@@ -1757,6 +1757,31 @@ added to / removed from a user for a database object.
 | `DatabaseName` | The name of the affected database. | yes |
 
 
+#### Common fields
+
+| Field | Description | Sensitive |
+|--|--|--|
+| `Timestamp` | The timestamp of the event. Expressed as nanoseconds since the Unix epoch. | no |
+| `EventType` | The type of the event. | no |
+| `Statement` | A normalized copy of the SQL statement that triggered the event. The statement string contains a mix of sensitive and non-sensitive details (it is redactable). | partially |
+| `Tag` | The statement tag. This is separate from the statement string, since the statement string can contain sensitive information. The tag is guaranteed not to. | no |
+| `User` | The user account that triggered the event. The special usernames `root` and `node` are not considered sensitive. | depends |
+| `DescriptorID` | The primary object descriptor affected by the operation. Set to zero for operations that don't affect descriptors. | no |
+| `ApplicationName` | The application name for the session where the event was emitted. This is included in the event to ease filtering of logging output by application. Application names starting with a dollar sign (`$`) are not considered sensitive. | no |
+| `PlaceholderValues` | The mapping of SQL placeholders to their values, for prepared statements. | yes |
+| `Grantee` | The user/role affected by the grant or revoke operation. | yes |
+| `GrantedPrivileges` | The privileges being granted to the grantee. | no |
+| `RevokedPrivileges` | The privileges being revoked from the grantee. | no |
+
+### `change_function_privilege`
+
+
+
+| Field | Description | Sensitive |
+|--|--|--|
+| `FuncName` | The name of the affected function. | yes |
+
+
 #### Common fields
 
 | Field | Description | Sensitive |

docs/generated/sql/bnf/alter_default_privileges_stmt.bnf

@@ -1,5 +1,5 @@
 alter_default_privileges_stmt ::=
-	'ALTER' 'DEFAULT' 'PRIVILEGES' ( 'FOR' ( 'ROLE' | 'USER' ) role_spec_list |  ) ( 'IN' 'SCHEMA' ( ( qualifiable_schema_name ) ( ( ',' qualifiable_schema_name ) )* ) |  ) ( 'GRANT' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' ) 'TO' role_spec_list ( 'WITH' 'GRANT' 'OPTION' |  ) )
-	| 'ALTER' 'DEFAULT' 'PRIVILEGES' ( 'FOR' ( 'ROLE' | 'USER' ) role_spec_list |  ) ( 'IN' 'SCHEMA' ( ( qualifiable_schema_name ) ( ( ',' qualifiable_schema_name ) )* ) |  ) ( 'REVOKE' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' ) 'FROM' role_spec_list ( 'CASCADE' | 'RESTRICT' |  ) | 'REVOKE' 'GRANT' 'OPTION' 'FOR' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' ) 'FROM' role_spec_list ( 'CASCADE' | 'RESTRICT' |  ) )
-	| 'ALTER' 'DEFAULT' 'PRIVILEGES' 'FOR' 'ALL' 'ROLES' ( 'IN' 'SCHEMA' ( ( qualifiable_schema_name ) ( ( ',' qualifiable_schema_name ) )* ) |  ) ( 'GRANT' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' ) 'TO' role_spec_list ( 'WITH' 'GRANT' 'OPTION' |  ) )
-	| 'ALTER' 'DEFAULT' 'PRIVILEGES' 'FOR' 'ALL' 'ROLES' ( 'IN' 'SCHEMA' ( ( qualifiable_schema_name ) ( ( ',' qualifiable_schema_name ) )* ) |  ) ( 'REVOKE' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' ) 'FROM' role_spec_list ( 'CASCADE' | 'RESTRICT' |  ) | 'REVOKE' 'GRANT' 'OPTION' 'FOR' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' ) 'FROM' role_spec_list ( 'CASCADE' | 'RESTRICT' |  ) )
+	'ALTER' 'DEFAULT' 'PRIVILEGES' ( 'FOR' ( 'ROLE' | 'USER' ) role_spec_list |  ) ( 'IN' 'SCHEMA' ( ( qualifiable_schema_name ) ( ( ',' qualifiable_schema_name ) )* ) |  ) ( 'GRANT' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' | 'FUNCTIONS' ) 'TO' role_spec_list ( 'WITH' 'GRANT' 'OPTION' |  ) )
+	| 'ALTER' 'DEFAULT' 'PRIVILEGES' ( 'FOR' ( 'ROLE' | 'USER' ) role_spec_list |  ) ( 'IN' 'SCHEMA' ( ( qualifiable_schema_name ) ( ( ',' qualifiable_schema_name ) )* ) |  ) ( 'REVOKE' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' | 'FUNCTIONS' ) 'FROM' role_spec_list ( 'CASCADE' | 'RESTRICT' |  ) | 'REVOKE' 'GRANT' 'OPTION' 'FOR' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' | 'FUNCTIONS' ) 'FROM' role_spec_list ( 'CASCADE' | 'RESTRICT' |  ) )
+	| 'ALTER' 'DEFAULT' 'PRIVILEGES' 'FOR' 'ALL' 'ROLES' ( 'IN' 'SCHEMA' ( ( qualifiable_schema_name ) ( ( ',' qualifiable_schema_name ) )* ) |  ) ( 'GRANT' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' | 'FUNCTIONS' ) 'TO' role_spec_list ( 'WITH' 'GRANT' 'OPTION' |  ) )
+	| 'ALTER' 'DEFAULT' 'PRIVILEGES' 'FOR' 'ALL' 'ROLES' ( 'IN' 'SCHEMA' ( ( qualifiable_schema_name ) ( ( ',' qualifiable_schema_name ) )* ) |  ) ( 'REVOKE' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' | 'FUNCTIONS' ) 'FROM' role_spec_list ( 'CASCADE' | 'RESTRICT' |  ) | 'REVOKE' 'GRANT' 'OPTION' 'FOR' privileges 'ON' ( 'TABLES' | 'SEQUENCES' | 'TYPES' | 'SCHEMAS' | 'FUNCTIONS' ) 'FROM' role_spec_list ( 'CASCADE' | 'RESTRICT' |  ) )

docs/generated/sql/bnf/grant_stmt.bnf

@@ -31,6 +31,12 @@ grant_stmt ::=
 	| 'GRANT' 'ALL'  'ON' 'ALL' 'TABLES' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list 
 	| 'GRANT' privilege_list 'ON' 'ALL' 'TABLES' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list 'WITH' 'GRANT' 'OPTION'
 	| 'GRANT' privilege_list 'ON' 'ALL' 'TABLES' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list 
+	| 'GRANT' 'ALL' 'PRIVILEGES' 'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list 'WITH' 'GRANT' 'OPTION'
+	| 'GRANT' 'ALL' 'PRIVILEGES' 'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list 
+	| 'GRANT' 'ALL'  'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list 'WITH' 'GRANT' 'OPTION'
+	| 'GRANT' 'ALL'  'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list 
+	| 'GRANT' privilege_list 'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list 'WITH' 'GRANT' 'OPTION'
+	| 'GRANT' privilege_list 'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list 
 	| 'GRANT' 'SYSTEM' 'ALL' 'PRIVILEGES' 'TO' role_spec_list 'WITH' 'GRANT' 'OPTION'
 	| 'GRANT' 'SYSTEM' 'ALL' 'PRIVILEGES' 'TO' role_spec_list 
 	| 'GRANT' 'SYSTEM' 'ALL'  'TO' role_spec_list 'WITH' 'GRANT' 'OPTION'

docs/generated/sql/bnf/revoke_stmt.bnf

@@ -28,6 +28,12 @@ revoke_stmt ::=
 	| 'REVOKE' 'GRANT' 'OPTION' 'FOR' 'ALL' 'PRIVILEGES' 'ON' 'ALL' 'TABLES' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
 	| 'REVOKE' 'GRANT' 'OPTION' 'FOR' 'ALL'  'ON' 'ALL' 'TABLES' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
 	| 'REVOKE' 'GRANT' 'OPTION' 'FOR' privilege_list 'ON' 'ALL' 'TABLES' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
+	| 'REVOKE' 'ALL' 'PRIVILEGES' 'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
+	| 'REVOKE' 'ALL'  'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
+	| 'REVOKE' privilege_list 'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
+	| 'REVOKE' 'GRANT' 'OPTION' 'FOR' 'ALL' 'PRIVILEGES' 'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
+	| 'REVOKE' 'GRANT' 'OPTION' 'FOR' 'ALL'  'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
+	| 'REVOKE' 'GRANT' 'OPTION' 'FOR' privilege_list 'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
 	| 'REVOKE' 'SYSTEM' 'ALL' 'PRIVILEGES' 'FROM' role_spec_list
 	| 'REVOKE' 'SYSTEM' 'ALL'  'FROM' role_spec_list
 	| 'REVOKE' 'SYSTEM' privilege_list 'FROM' role_spec_list

docs/generated/sql/bnf/stmt_block.bnf

@@ -92,6 +92,7 @@ grant_stmt ::=
 	| 'GRANT' privileges 'ON' 'SCHEMA' schema_name_list 'TO' role_spec_list opt_with_grant_option
 	| 'GRANT' privileges 'ON' 'ALL' 'SEQUENCES' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list opt_with_grant_option
 	| 'GRANT' privileges 'ON' 'ALL' 'TABLES' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list opt_with_grant_option
+	| 'GRANT' privileges 'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'TO' role_spec_list opt_with_grant_option
 	| 'GRANT' 'SYSTEM' privileges 'TO' role_spec_list opt_with_grant_option
 
 prepare_stmt ::=
@@ -109,6 +110,8 @@ revoke_stmt ::=
 	| 'REVOKE' privileges 'ON' 'ALL' 'TABLES' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
 	| 'REVOKE' privileges 'ON' 'ALL' 'SEQUENCES' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
 	| 'REVOKE' 'GRANT' 'OPTION' 'FOR' privileges 'ON' 'ALL' 'TABLES' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
+	| 'REVOKE' privileges 'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
+	| 'REVOKE' 'GRANT' 'OPTION' 'FOR' privileges 'ON' 'ALL' 'FUNCTIONS' 'IN' 'SCHEMA' schema_name_list 'FROM' role_spec_list
 	| 'REVOKE' 'SYSTEM' privileges 'FROM' role_spec_list
 	| 'REVOKE' 'GRANT' 'OPTION' 'FOR' 'SYSTEM' privileges 'FROM' role_spec_list
 
@@ -360,6 +363,7 @@ grant_targets ::=
 	| 'TABLE' table_pattern_list
 	| 'DATABASE' name_list
 	| 'EXTERNAL' 'CONNECTION' name_list
+	| 'FUNCTION' function_with_argtypes_list
 
 role_spec_list ::=
 	( role_spec ) ( ( ',' role_spec ) )*
@@ -1447,6 +1451,9 @@ table_pattern ::=
 	simple_db_object_name
 	| complex_table_pattern
 
+function_with_argtypes_list ::=
+	( function_with_argtypes ) ( ( ',' function_with_argtypes ) )*
+
 privilege ::=
 	name
 	| 'CREATE'
@@ -1963,6 +1970,10 @@ unrestricted_name ::=
 	| type_func_name_keyword
 	| reserved_keyword
 
+function_with_argtypes ::=
+	db_object_name func_args
+	| db_object_name
+
 type_name ::=
 	db_object_name
 
@@ -2465,9 +2476,6 @@ table_index_name_list ::=
 table_name_list ::=
 	( table_name ) ( ( ',' table_name ) )*
 
-function_with_argtypes_list ::=
-	( function_with_argtypes ) ( ( ',' function_with_argtypes ) )*
-
 non_reserved_word ::=
 	'identifier'
 	| unreserved_keyword
@@ -2657,6 +2665,10 @@ reserved_keyword ::=
 	| 'WITH'
 	| cockroachdb_extra_reserved_keyword
 
+func_args ::=
+	'(' func_args_list ')'
+	| '(' ')'
+
 simple_typename ::=
 	general_type_name
 	| '@' iconst32
@@ -2737,6 +2749,7 @@ target_object_type ::=
 	| 'SEQUENCES'
 	| 'TYPES'
 	| 'SCHEMAS'
+	| 'FUNCTIONS'
 
 alter_changefeed_cmd ::=
 	'ADD' changefeed_targets opt_with_options
@@ -2747,10 +2760,6 @@ alter_changefeed_cmd ::=
 alter_backup_cmd ::=
 	'ADD' backup_kms
 
-function_with_argtypes ::=
-	db_object_name func_args
-	| db_object_name
-
 alter_func_opt_list ::=
 	( common_func_opt_item ) ( ( common_func_opt_item ) )*
 
@@ -3116,6 +3125,9 @@ type_func_name_no_crdb_extra_keyword ::=
 	| 'RIGHT'
 	| 'SIMILAR'
 
+func_args_list ::=
+	( func_arg ) ( ( ',' func_arg ) )*
+
 general_type_name ::=
 	type_function_name_no_crdb_extra
 
@@ -3195,10 +3207,6 @@ sequence_option_elem ::=
 backup_kms ::=
 	'NEW_KMS' '=' string_or_placeholder_opt_list 'WITH' 'OLD_KMS' '=' string_or_placeholder_opt_list
 
-func_args ::=
-	'(' func_args_list ')'
-	| '(' ')'
-
 common_func_opt_item ::=
 	'CALLED' 'ON' 'NULL' 'INPUT'
 	| 'RETURNS' 'NULL' 'ON' 'NULL' 'INPUT'
@@ -3474,6 +3482,13 @@ join_qual ::=
 rowsfrom_list ::=
 	( rowsfrom_item ) ( ( ',' rowsfrom_item ) )*
 
+func_arg ::=
+	func_arg_class param_name func_arg_type
+	| param_name func_arg_class func_arg_type
+	| param_name func_arg_type
+	| func_arg_class func_arg_type
+	| func_arg_type
+
 opt_varying ::=
 	'VARYING'
 	| 
@@ -3539,9 +3554,6 @@ storage_parameter_key_list ::=
 partition_by_index ::=
 	partition_by
 
-func_args_list ::=
-	( func_arg ) ( ( ',' func_arg ) )*
-
 opt_float ::=
 	'(' 'ICONST' ')'
 	| 
@@ -3677,13 +3689,6 @@ opt_family_name ::=
 create_as_constraint_elem ::=
 	'PRIMARY' 'KEY' '(' create_as_params ')' opt_with_storage_parameter_list
 
-func_arg ::=
-	func_arg_class param_name func_arg_type
-	| param_name func_arg_class func_arg_type
-	| param_name func_arg_type
-	| func_arg_class func_arg_type
-	| func_arg_type
-
 func_as ::=
 	'SCONST'
 
@@ -3700,6 +3705,12 @@ join_outer ::=
 rowsfrom_item ::=
 	func_expr_windowless
 
+func_arg_class ::=
+	'IN'
+
+param_name ::=
+	type_function_name
+
 char_aliases ::=
 	'CHAR'
 	| 'CHARACTER'
@@ -3768,12 +3779,6 @@ create_as_col_qualification_elem ::=
 create_as_params ::=
 	( create_as_param ) ( ( ',' create_as_param ) )*
 
-func_arg_class ::=
-	'IN'
-
-param_name ::=
-	type_function_name
-
 col_qualification ::=
 	'CONSTRAINT' constraint_name col_qualification_elem
 	| col_qualification_elem

pkg/BUILD.bazel

@@ -800,6 +800,7 @@ GO_TARGETS = [
     "//pkg/cloud/cloudtestutils:cloudtestutils",
     "//pkg/cloud/externalconn/connectionpb:connectionpb",
     "//pkg/cloud/externalconn/providers:providers",
+    "//pkg/cloud/externalconn/testutils:testutils",
     "//pkg/cloud/externalconn/utils:utils",
     "//pkg/cloud/externalconn:externalconn",
     "//pkg/cloud/gcp:gcp",
@@ -2183,6 +2184,7 @@ GET_X_DATA_TARGETS = [
     "//pkg/cloud/externalconn:get_x_data",
     "//pkg/cloud/externalconn/connectionpb:get_x_data",
     "//pkg/cloud/externalconn/providers:get_x_data",
+    "//pkg/cloud/externalconn/testutils:get_x_data",
     "//pkg/cloud/externalconn/utils:get_x_data",
     "//pkg/cloud/gcp:get_x_data",
     "//pkg/cloud/httpsink:get_x_data",

pkg/base/testing_knobs.go

@@ -51,4 +51,5 @@ type TestingKnobs struct {
 	CapturedIndexUsageStatsKnobs ModuleTestingKnobs
 	AdmissionControl             ModuleTestingKnobs
 	UnusedIndexRecommendKnobs    ModuleTestingKnobs
+	ExternalConnection           ModuleTestingKnobs
 }

pkg/ccl/changefeedccl/sink_kafka_connection.go

@@ -15,11 +15,12 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/cloud/externalconn"
 	"github.com/cockroachdb/cockroach/pkg/cloud/externalconn/connectionpb"
+	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/errors"
 )
 
 func parseAndValidateKafkaSinkURI(
-	ctx context.Context, uri *url.URL,
+	ctx context.Context, _ interface{}, _ username.SQLUsername, uri *url.URL,
 ) (externalconn.ExternalConnection, error) {
 	// Validate the kafka URI by creating a kafka sink and throwing it away.
 	//

pkg/ccl/cloudccl/externalconn/BUILD.bazel

@@ -12,8 +12,9 @@ go_test(
         "//pkg/base",
         "//pkg/ccl/changefeedccl",
         "//pkg/ccl/kvccl/kvtenantccl",
+        "//pkg/cloud/externalconn",
         "//pkg/cloud/externalconn/providers",
-        "//pkg/cloud/externalconn/utils",
+        "//pkg/cloud/externalconn/testutils",
         "//pkg/jobs",
         "//pkg/roachpb",
         "//pkg/security/securityassets",

pkg/ccl/cloudccl/externalconn/datadriven_test.go

@@ -15,9 +15,10 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
-	_ "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl"            // register the sink External Connection implementations
+	_ "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl"
+	"github.com/cockroachdb/cockroach/pkg/cloud/externalconn"
 	_ "github.com/cockroachdb/cockroach/pkg/cloud/externalconn/providers" // register all the concrete External Connection implementations
-	"github.com/cockroachdb/cockroach/pkg/cloud/externalconn/utils"
+	ectestutils "github.com/cockroachdb/cockroach/pkg/cloud/externalconn/testutils"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
@@ -33,16 +34,28 @@ func TestDataDriven(t *testing.T) {
 
 	ctx := context.Background()
 	datadriven.Walk(t, testutils.TestDataPath(t), func(t *testing.T, path string) {
+		dir, dirCleanupFn := testutils.TempDir(t)
+		defer dirCleanupFn()
+
+		var skipCheckExternalStorageConnection bool
+		ecTestingKnobs := &externalconn.TestingKnobs{
+			SkipCheckingExternalStorageConnection: func() bool {
+				return skipCheckExternalStorageConnection
+			},
+		}
 		tc := testcluster.StartTestCluster(t, 1, base.TestClusterArgs{
 			ServerArgs: base.TestServerArgs{
 				Knobs: base.TestingKnobs{
-					JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(), // speeds up test
+					JobsTestingKnobs:   jobs.NewTestingKnobsWithShortIntervals(), // speeds up test
+					ExternalConnection: ecTestingKnobs,
 				},
+				ExternalIODirConfig: base.ExternalIODirConfig{},
+				ExternalIODir:       dir,
 			},
 		})
 		defer tc.Stopper().Stop(ctx)
 
-		externalConnTestCluster := utils.NewHandle(t, tc)
+		externalConnTestCluster := ectestutils.NewHandle(t, tc)
 		defer externalConnTestCluster.Cleanup()
 
 		externalConnTestCluster.InitializeTenant(ctx, roachpb.SystemTenantID)
@@ -63,6 +76,12 @@ func TestDataDriven(t *testing.T) {
 			case "initialize":
 				externalConnTestCluster.InitializeTenant(ctx, tenantID)
 
+			case "enable-check-external-storage":
+				skipCheckExternalStorageConnection = false
+
+			case "disable-check-external-storage":
+				skipCheckExternalStorageConnection = true
+
 			case "exec-sql":
 				if d.HasArg("user") {
 					var user string