changefeedccl: refactor: move sinkURL to its own file in changefeedbase

Epic: none
Release note: None

Author: asg0451

pkg/ccl/changefeedccl/changefeedbase/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "errors.go",
         "options.go",
         "settings.go",
+        "sink_url.go",
         "target.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase",
@@ -20,6 +21,7 @@ go_library(
         "//pkg/sql/catalog/descpb",
         "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/pgwire/pgerror",
+        "//pkg/util",
         "//pkg/util/iterutil",
         "//pkg/util/metamorphic",
         "@com_github_cockroachdb_errors//:errors",

pkg/ccl/changefeedccl/changefeedbase/sink_url.go

@@ -0,0 +1,128 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package changefeedbase
+
+import (
+	"encoding/base64"
+	"net/url"
+	"strconv"
+
+	"github.com/cockroachdb/cockroach/pkg/util"
+	"github.com/cockroachdb/errors"
+)
+
+// SinkURL is a helper struct which for "consuming" URL query
+// parameters from the underlying URL.
+type SinkURL struct {
+	_ util.NoCopy
+	*url.URL
+	q url.Values
+}
+
+func (u *SinkURL) PeekParam(p string) string {
+	if u.q == nil {
+		u.q = u.Query()
+	}
+	v := u.q.Get(p)
+	return v
+}
+
+func (u *SinkURL) ConsumeParam(p string) string {
+	v := u.PeekParam(p)
+	u.q.Del(p)
+	return v
+}
+
+func (u *SinkURL) ConsumeParams(p string) []string {
+	if u.q == nil {
+		u.q = u.Query()
+	}
+	v := u.q[p]
+	u.q.Del(p)
+	return v
+}
+
+func (u *SinkURL) AddParam(p string, value string) {
+	if u.q == nil {
+		u.q = u.Query()
+	}
+	u.q.Add(p, value)
+}
+
+func (u *SinkURL) SetParam(p string, value string) {
+	if u.q == nil {
+		u.q = u.Query()
+	}
+	u.q.Set(p, value)
+}
+
+func (u *SinkURL) ConsumeBool(param string, dest *bool) (wasSet bool, err error) {
+	if paramVal := u.ConsumeParam(param); paramVal != "" {
+		wasSet, err := strToBool(paramVal, dest)
+		if err != nil {
+			return false, errors.Wrapf(err, "param %s must be a bool", param)
+		}
+		return wasSet, err
+	}
+	return false, nil
+}
+
+func (u *SinkURL) ConsumeBoolParam(param string) (bool, error) {
+	var b bool
+	_, err := u.ConsumeBool(param, &b)
+	return b, err
+}
+
+func (u *SinkURL) DecodeBase64(param string, dest *[]byte) error {
+	// TODO(dan): There's a straightforward and unambiguous transformation
+	//  between the base 64 encoding defined in RFC 4648 and the URL variant
+	//  defined in the same RFC: simply replace all `+` with `-` and `/` with
+	//  `_`. Consider always doing this for the user and accepting either
+	//  variant.
+	val := u.ConsumeParam(param)
+	err := DecodeBase64FromString(val, dest)
+	if err != nil {
+		return errors.Wrapf(err, `param %s must be base 64 encoded`, param)
+	}
+	return nil
+}
+
+func (u *SinkURL) RemainingQueryParams() (res []string) {
+	for p := range u.q {
+		res = append(res, p)
+	}
+	return
+}
+
+func (u *SinkURL) String() string {
+	if u.q != nil {
+		// If we changed query params, re-encode them.
+		u.URL.RawQuery = u.q.Encode()
+		u.q = nil
+	}
+	return u.URL.String()
+}
+
+func strToBool(src string, dest *bool) (wasSet bool, err error) {
+	b, err := strconv.ParseBool(src)
+	if err != nil {
+		return false, err
+	}
+	*dest = b
+	return true, nil
+}
+
+func DecodeBase64FromString(src string, dest *[]byte) error {
+	if src == `` {
+		return nil
+	}
+	decoded, err := base64.StdEncoding.DecodeString(src)
+	if err != nil {
+		return err
+	}
+	*dest = decoded
+	return nil
+}

pkg/ccl/changefeedccl/schema_registry.go

@@ -89,7 +89,7 @@ func getAndDeleteParams(u *url.URL) (*schemaRegistryParams, error) {
 		changefeedbase.RegistryParamClientKey} {
 		if stringParam := query.Get(k); stringParam != "" {
 			var decoded []byte
-			err := decodeBase64FromString(stringParam, &decoded)
+			err := changefeedbase.DecodeBase64FromString(stringParam, &decoded)
 			if err != nil {
 				return nil, errors.Wrapf(err, "param %s must be base 64 encoded", k)
 			}

pkg/ccl/changefeedccl/sink.go

@@ -246,23 +246,23 @@ func getSink(
 			if knobs, ok := serverCfg.TestingKnobs.Changefeed.(*TestingKnobs); ok {
 				nullIsAccounted = knobs.NullSinkIsExternalIOAccounted
 			}
-			return makeNullSink(sinkURL{URL: u}, metricsBuilder(nullIsAccounted))
+			return makeNullSink(&changefeedbase.SinkURL{URL: u}, metricsBuilder(nullIsAccounted))
 		case isKafkaSink(u):
 			return validateOptionsAndMakeSink(changefeedbase.KafkaValidOptions, func() (Sink, error) {
 				if KafkaV2Enabled.Get(&serverCfg.Settings.SV) {
-					return makeKafkaSinkV2(ctx, sinkURL{URL: u}, AllTargets(feedCfg), opts.GetKafkaConfigJSON(),
+					return makeKafkaSinkV2(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), opts.GetKafkaConfigJSON(),
 						numSinkIOWorkers(serverCfg), newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
 						serverCfg.Settings, metricsBuilder, kafkaSinkV2Knobs{})
 				} else {
-					return makeKafkaSink(ctx, sinkURL{URL: u}, AllTargets(feedCfg), opts.GetKafkaConfigJSON(), serverCfg.Settings, metricsBuilder)
+					return makeKafkaSink(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), opts.GetKafkaConfigJSON(), serverCfg.Settings, metricsBuilder)
 				}
 			})
 		case isPulsarSink(u):
 			var testingKnobs *TestingKnobs
 			if knobs, ok := serverCfg.TestingKnobs.Changefeed.(*TestingKnobs); ok {
 				testingKnobs = knobs
 			}
-			return makePulsarSink(ctx, sinkURL{URL: u}, encodingOpts, AllTargets(feedCfg), opts.GetKafkaConfigJSON(),
+			return makePulsarSink(ctx, &changefeedbase.SinkURL{URL: u}, encodingOpts, AllTargets(feedCfg), opts.GetKafkaConfigJSON(),
 				serverCfg.Settings, metricsBuilder, testingKnobs)
 		case isWebhookSink(u):
 			webhookOpts, err := opts.GetWebhookSinkOptions()
@@ -271,13 +271,13 @@ func getSink(
 			}
 			if WebhookV2Enabled.Get(&serverCfg.Settings.SV) {
 				return validateOptionsAndMakeSink(changefeedbase.WebhookValidOptions, func() (Sink, error) {
-					return makeWebhookSink(ctx, sinkURL{URL: u}, encodingOpts, webhookOpts,
+					return makeWebhookSink(ctx, &changefeedbase.SinkURL{URL: u}, encodingOpts, webhookOpts,
 						numSinkIOWorkers(serverCfg), newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
 						metricsBuilder, serverCfg.Settings)
 				})
 			} else {
 				return validateOptionsAndMakeSink(changefeedbase.WebhookValidOptions, func() (Sink, error) {
-					return makeDeprecatedWebhookSink(ctx, sinkURL{URL: u}, encodingOpts, webhookOpts,
+					return makeDeprecatedWebhookSink(ctx, &changefeedbase.SinkURL{URL: u}, encodingOpts, webhookOpts,
 						defaultWorkerCount(), timeutil.DefaultTimeSource{}, metricsBuilder)
 				})
 			}
@@ -307,18 +307,18 @@ func getSink(
 					nodeID = serverCfg.NodeID.SQLInstanceID()
 				}
 				return makeCloudStorageSink(
-					ctx, sinkURL{URL: u}, nodeID, serverCfg.Settings, encodingOpts,
+					ctx, &changefeedbase.SinkURL{URL: u}, nodeID, serverCfg.Settings, encodingOpts,
 					timestampOracle, serverCfg.ExternalStorageFromURI, user, metricsBuilder, testingKnobs,
 				)
 			})
 		case u.Scheme == changefeedbase.SinkSchemeExperimentalSQL:
 			return validateOptionsAndMakeSink(changefeedbase.SQLValidOptions, func() (Sink, error) {
-				return makeSQLSink(sinkURL{URL: u}, sqlSinkTableName, AllTargets(feedCfg), metricsBuilder)
+				return makeSQLSink(&changefeedbase.SinkURL{URL: u}, sqlSinkTableName, AllTargets(feedCfg), metricsBuilder)
 			})
 		case u.Scheme == changefeedbase.SinkSchemeExternalConnection:
 			return validateOptionsAndMakeSink(changefeedbase.ExternalConnectionValidOptions, func() (Sink, error) {
 				return makeExternalConnectionSink(
-					ctx, sinkURL{URL: u}, user, makeExternalConnectionProvider(ctx, serverCfg.DB),
+					ctx, &changefeedbase.SinkURL{URL: u}, user, makeExternalConnectionProvider(ctx, serverCfg.DB),
 					serverCfg, feedCfg, timestampOracle, jobID, m,
 				)
 			})
@@ -362,70 +362,6 @@ func validateSinkOptions(opts map[string]string, sinkSpecificOpts map[string]str
 	return nil
 }
 
-// sinkURL is a helper struct which for "consuming" URL query
-// parameters from the underlying URL.
-type sinkURL struct {
-	*url.URL
-	q url.Values
-}
-
-func (u *sinkURL) consumeParam(p string) string {
-	if u.q == nil {
-		u.q = u.Query()
-	}
-	v := u.q.Get(p)
-	u.q.Del(p)
-	return v
-}
-
-func (u *sinkURL) addParam(p string, value string) {
-	if u.q == nil {
-		u.q = u.Query()
-	}
-	u.q.Add(p, value)
-}
-
-func (u *sinkURL) consumeBool(param string, dest *bool) (wasSet bool, err error) {
-	if paramVal := u.consumeParam(param); paramVal != "" {
-		wasSet, err := strToBool(paramVal, dest)
-		if err != nil {
-			return false, errors.Wrapf(err, "param %s must be a bool", param)
-		}
-		return wasSet, err
-	}
-	return false, nil
-}
-
-func (u *sinkURL) decodeBase64(param string, dest *[]byte) error {
-	// TODO(dan): There's a straightforward and unambiguous transformation
-	//  between the base 64 encoding defined in RFC 4648 and the URL variant
-	//  defined in the same RFC: simply replace all `+` with `-` and `/` with
-	//  `_`. Consider always doing this for the user and accepting either
-	//  variant.
-	val := u.consumeParam(param)
-	err := decodeBase64FromString(val, dest)
-	if err != nil {
-		return errors.Wrapf(err, `param %s must be base 64 encoded`, param)
-	}
-	return nil
-}
-
-func (u *sinkURL) remainingQueryParams() (res []string) {
-	for p := range u.q {
-		res = append(res, p)
-	}
-	return
-}
-
-func (u *sinkURL) String() string {
-	if u.q != nil {
-		// If we changed query params, re-encode them.
-		u.URL.RawQuery = u.q.Encode()
-		u.q = nil
-	}
-	return u.URL.String()
-}
-
 // errorWrapperSink delegates to another sink and marks all returned errors as
 // retryable. During changefeed setup, we use the sink once without this to
 // verify configuration, but in the steady state, no sink error should be
@@ -616,9 +552,9 @@ func (n *nullSink) getConcreteType() sinkType {
 
 var _ Sink = (*nullSink)(nil)
 
-func makeNullSink(u sinkURL, m metricsRecorder) (Sink, error) {
+func makeNullSink(u *changefeedbase.SinkURL, m metricsRecorder) (Sink, error) {
 	var pacer *time.Ticker
-	if delay := u.consumeParam(`delay`); delay != "" {
+	if delay := u.ConsumeParam(`delay`); delay != "" {
 		pace, err := time.ParseDuration(delay)
 		if err != nil {
 			return nil, err

pkg/ccl/changefeedccl/sink_cloudstorage.go

@@ -377,7 +377,7 @@ const flushQueueDepth = 256
 
 func makeCloudStorageSink(
 	ctx context.Context,
-	u sinkURL,
+	u *changefeedbase.SinkURL,
 	srcID base.SQLInstanceID,
 	settings *cluster.Settings,
 	encodingOpts changefeedbase.EncodingOptions,
@@ -388,7 +388,7 @@ func makeCloudStorageSink(
 	testingKnobs *TestingKnobs,
 ) (Sink, error) {
 	var targetMaxFileSize int64 = 16 << 20 // 16MB
-	if fileSizeParam := u.consumeParam(changefeedbase.SinkParamFileSize); fileSizeParam != `` {
+	if fileSizeParam := u.ConsumeParam(changefeedbase.SinkParamFileSize); fileSizeParam != `` {
 		var err error
 		if targetMaxFileSize, err = humanizeutil.ParseBytes(fileSizeParam); err != nil {
 			return nil, pgerror.Wrapf(err, pgcode.Syntax, `parsing %s`, fileSizeParam)
@@ -431,7 +431,7 @@ func makeCloudStorageSink(
 	}
 	s.flushGroup.GoCtx(s.asyncFlusher)
 
-	if partitionFormat := u.consumeParam(changefeedbase.SinkParamPartitionFormat); partitionFormat != "" {
+	if partitionFormat := u.ConsumeParam(changefeedbase.SinkParamPartitionFormat); partitionFormat != "" {
 		dateFormat, ok := partitionDateFormats[partitionFormat]
 		if !ok {
 			return nil, errors.Errorf("invalid partition_format of %s", partitionFormat)

pkg/ccl/changefeedccl/sink_cloudstorage_test.go

@@ -605,7 +605,7 @@ func TestCloudStorageSink(t *testing.T) {
 					timestampOracle := &changeAggregatorLowerBoundOracle{sf: sf}
 
 					sinkURIWithParam := sinkURI(t, targetMaxFileSize)
-					sinkURIWithParam.addParam(changefeedbase.SinkParamPartitionFormat, tc.format)
+					sinkURIWithParam.AddParam(changefeedbase.SinkParamPartitionFormat, tc.format)
 					t.Logf("format=%s sinkgWithParam: %s", tc.format, sinkURIWithParam.String())
 					s, err := makeCloudStorageSink(
 						ctx, sinkURIWithParam, 1, settings, opts,
@@ -944,12 +944,12 @@ func testDir(t *testing.T) string {
 	return strings.ReplaceAll(t.Name(), "/", ";")
 }
 
-func sinkURI(t *testing.T, maxFileSize int64) sinkURL {
+func sinkURI(t *testing.T, maxFileSize int64) *changefeedbase.SinkURL {
 	u, err := url.Parse(fmt.Sprintf("nodelocal://1/%s", testDir(t)))
 	require.NoError(t, err)
-	sink := sinkURL{URL: u}
+	sink := &changefeedbase.SinkURL{URL: u}
 	if maxFileSize != unlimitedFileSize {
-		sink.addParam(changefeedbase.SinkParamFileSize, strconv.FormatInt(maxFileSize, 10))
+		sink.AddParam(changefeedbase.SinkParamFileSize, strconv.FormatInt(maxFileSize, 10))
 	}
 	return sink
 }

pkg/ccl/changefeedccl/sink_external_connection.go

@@ -21,7 +21,7 @@ import (
 
 func makeExternalConnectionSink(
 	ctx context.Context,
-	u sinkURL,
+	u *changefeedbase.SinkURL,
 	user username.SQLUsername,
 	p externalConnectionProvider,
 	serverCfg *execinfra.ServerConfig,