changefeedccl: prepend file- to cloudstorage http(s) schemes

Changefeeds previously used http as their URI to align with backups.
Backups no longer support http, and for changefeeds http is ambiguous,
leading users to create cloudstorage sinks when they meant webhook sinks.
This commit changes the scheme to file-http. Webhooks still use webhook-http.

Informs: cockroachdb#98719

Release note (enterprise change): Changefeeds that create files over an http connection may now be specified via `INTO 'file-https://'` to disambiguate with `webhook-https`.

Author: HonoreDB

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -909,8 +909,18 @@ func validateSink(
 		return err
 	}
 
-	if u.Scheme == changefeedbase.SinkSchemeCloudStorageHTTP || u.Scheme == changefeedbase.SinkSchemeCloudStorageHTTPS {
-		p.BufferClientNotice(ctx, pgnotice.Newf(`%s sinks will emit using cloud storage semantics. For a webhook sink, prepend webhook- to the sink URI.`))
+	ambiguousSchemes := map[string][2]string{
+		changefeedbase.DeprecatedSinkSchemeHTTP:  {changefeedbase.SinkSchemeCloudStorageHTTP, changefeedbase.SinkSchemeWebhookHTTP},
+		changefeedbase.DeprecatedSinkSchemeHTTPS: {changefeedbase.SinkSchemeCloudStorageHTTPS, changefeedbase.SinkSchemeWebhookHTTPS},
+	}
+
+	if disambiguations, isAmbiguous := ambiguousSchemes[u.Scheme]; isAmbiguous {
+		p.BufferClientNotice(ctx, pgnotice.Newf(
+			`Interpreting deprecated URI scheme %s as %s. For webhook semantics, use %s.`,
+			u.Scheme,
+			disambiguations[0],
+			disambiguations[1],
+		))
 	}
 
 	var nilOracle timestampLowerBoundOracle

pkg/ccl/changefeedccl/changefeed_test.go

@@ -4821,7 +4821,7 @@ func TestChangefeedErrors(t *testing.T) {
 		`CREATE CHANGEFEED FOR foo INTO $1`, `webhook-https://fake-host?ca_cert=Zm9v`,
 	)
 	sqlDB.ExpectErr(
-		t, `sink requires https`,
+		t, `sink requires webhook-https`,
 		`CREATE CHANGEFEED FOR foo INTO $1`, `webhook-http://fake-host`,
 	)
 	sqlDB.ExpectErr(

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -162,6 +162,11 @@ const (
 	DeprecatedSinkSchemeCloudStorageNodelocal = `experimental-nodelocal`
 	DeprecatedSinkSchemeCloudStorageS3        = `experimental-s3`
 
+	// DeprecatedSinkSchemeHTTP is interpreted as cloudstorage over HTTP PUT.
+	DeprecatedSinkSchemeHTTP = `http`
+	// DeprecatedSinkSchemeHTTPS is interpreted as cloudstorage over HTTPS PUT.
+	DeprecatedSinkSchemeHTTPS = `https`
+
 	// OptKafkaSinkConfig is a JSON configuration for kafka sink (kafkaSinkConfig).
 	OptKafkaSinkConfig   = `kafka_sink_config`
 	OptPubsubSinkConfig  = `pubsub_sink_config`
@@ -186,13 +191,11 @@ const (
 	SinkParamTopicName              = `topic_name`
 	SinkSchemeCloudStorageAzure     = `azure`
 	SinkSchemeCloudStorageGCS       = `gs`
-	SinkSchemeCloudStorageHTTP      = `http`
-	SinkSchemeCloudStorageHTTPS     = `https`
+	SinkSchemeCloudStorageHTTP      = `file-http`
+	SinkSchemeCloudStorageHTTPS     = `file-https`
 	SinkSchemeCloudStorageNodelocal = `nodelocal`
 	SinkSchemeCloudStorageS3        = `s3`
 	SinkSchemeExperimentalSQL       = `experimental-sql`
-	SinkSchemeHTTP                  = `http`
-	SinkSchemeHTTPS                 = `https`
 	SinkSchemeKafka                 = `kafka`
 	SinkSchemeNull                  = `null`
 	SinkSchemeWebhookHTTP           = `webhook-http`

pkg/ccl/changefeedccl/sink_cloudstorage.go

@@ -49,6 +49,10 @@ func isCloudStorageSink(u *url.URL) bool {
 		changefeedbase.SinkSchemeCloudStorageNodelocal, changefeedbase.SinkSchemeCloudStorageHTTP,
 		changefeedbase.SinkSchemeCloudStorageHTTPS, changefeedbase.SinkSchemeCloudStorageAzure:
 		return true
+	// During the deprecation period, we need to keep parsing these as cloudstorage for backwards
+	// compatibility. Afterwards we'll either remove them or move them to webhook.
+	case changefeedbase.DeprecatedSinkSchemeHTTP, changefeedbase.DeprecatedSinkSchemeHTTPS:
+		return true
 	default:
 		return false
 	}
@@ -375,6 +379,7 @@ func makeCloudStorageSink(
 		}
 	}
 	u.Scheme = strings.TrimPrefix(u.Scheme, `experimental-`)
+	u.Scheme = strings.TrimPrefix(u.Scheme, `file-`)
 
 	sinkID := atomic.AddInt64(&cloudStorageSinkIDAtomic, 1)
 	sessID, err := generateChangefeedSessionID()

pkg/ccl/changefeedccl/sink_external_connection.go

@@ -92,6 +92,8 @@ var supportedExternalConnectionTypes = map[string]connectionpb.ConnectionProvide
 	GcpScheme:                                      connectionpb.ConnectionProvider_gcpubsub,
 	changefeedbase.SinkSchemeCloudStorageHTTP:      connectionpb.ConnectionProvider_http,
 	changefeedbase.SinkSchemeCloudStorageHTTPS:     connectionpb.ConnectionProvider_https,
+	changefeedbase.DeprecatedSinkSchemeHTTP:        connectionpb.ConnectionProvider_http,
+	changefeedbase.DeprecatedSinkSchemeHTTPS:       connectionpb.ConnectionProvider_https,
 	changefeedbase.SinkSchemeCloudStorageNodelocal: connectionpb.ConnectionProvider_nodelocal,
 	changefeedbase.SinkSchemeCloudStorageS3:        connectionpb.ConnectionProvider_s3,
 	changefeedbase.SinkSchemeKafka:                 connectionpb.ConnectionProvider_kafka,

pkg/ccl/changefeedccl/sink_webhook.go

@@ -233,7 +233,7 @@ func makeDeprecatedWebhookSink(
 	mb metricsRecorderBuilder,
 ) (Sink, error) {
 	if u.Scheme != changefeedbase.SinkSchemeWebhookHTTPS {
-		return nil, errors.Errorf(`this sink requires %s`, changefeedbase.SinkSchemeHTTPS)
+		return nil, errors.Errorf(`this sink requires %s`, changefeedbase.SinkSchemeWebhookHTTPS)
 	}
 	u.Scheme = strings.TrimPrefix(u.Scheme, `webhook-`)
 

pkg/ccl/changefeedccl/sink_webhook_v2.go

@@ -229,7 +229,7 @@ func validateWebhookOpts(
 	u sinkURL, encodingOpts changefeedbase.EncodingOptions, opts changefeedbase.WebhookSinkOptions,
 ) error {
 	if u.Scheme != changefeedbase.SinkSchemeWebhookHTTPS {
-		return errors.Errorf(`this sink requires %s`, changefeedbase.SinkSchemeHTTPS)
+		return errors.Errorf(`this sink requires %s`, changefeedbase.SinkSchemeWebhookHTTPS)
 	}
 
 	switch encodingOpts.Format {