Merge #149505 #151355

149505: cli: "no such host" error when starting CRDB on Mac r=darrylwong a=srosenberg

When CockroachDB is started without `--advertise-addr`, it fallbacks to `--listen-addr`. If the latter is also unspecified, it defaults to the `hostname`.

Subsequently, `validateAdvertiseAddr` attempts to
resolve the hostname. If the resolution fails, the server aborts the startup sequence. This is a safe default because node discovery is compromised without a functioning DNS.

On Mac, `hostname` is resolved via mDNS / Bonjour. Unless it's suffixed with `.local`, it doesn't resolve. An obvious workaround is to add it to /etc/hosts, or to change it via `sudo hostname $USER.local`.

To make the (DEV) experience on Mac more seamless, this PR replaces DNS resolution error with a warning. Since we can't unambiguously determine if the user is running in DEV-mode, we assume that `--insecure` on Mac implies exactly that.

We also update `roachprod` to explicitly use `--advertised-addr` with `127.0.0.1`. This also fixes `roachtest` in local mode.

Fixes: #66426
Fixes: #149469
Release note: None

151355: changefeedccl/resolvedspan: integrate span.MultiFrontier into frontiers r=aerfrei,asg0451 a=andyyang890

Fixes #148110

---

**changefeedccl: fix ALTER CHANGEFEED span selection** 

This patch updates the ALTER CHANGEFEED code to use primary index spans
instead of entire table spans to be consistent with the rest of the
changefeed code and to help maintain the assumption that we will
never merge spans from different tables in our frontier data structures.

Release note: None

---

**changefeedccl/resolvedspan: integrate span.MultiFrontier into frontiers** 

This patch integrates `span.MultiFrontier`s into the resolved span frontiers
to give changefeed processors easy access to per-table resolved timestamps.

Release note: None

---

**changefeedccl/resolvedspan: add feature flag for per-table tracking** 

This patch adds a new cluster setting named
`changefeed.frontier.per_table_tracking.enabled`
to feature flag enabling per-table tracking in
the resolved span frontiers.

Release note: None

Co-authored-by: Stan Rosenberg <[email protected]>
Co-authored-by: Andy Yang <[email protected]>

Author: 3 people

pkg/base/BUILD.bazel

@@ -30,6 +30,8 @@ go_library(
         "//pkg/util",
         "//pkg/util/envutil",
         "//pkg/util/humanizeutil",
+        "//pkg/util/log",
+        "//pkg/util/log/severity",
         "//pkg/util/metric",
         "//pkg/util/mon",
         "//pkg/util/netutil/addr",

pkg/base/addr_validation.go

@@ -10,9 +10,12 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"runtime"
 	"strconv"
 
 	"github.com/cockroachdb/cockroach/pkg/cli/cliflags"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/log/severity"
 	"github.com/cockroachdb/errors"
 )
 
@@ -32,7 +35,7 @@ import (
 func (cfg *Config) ValidateAddrs(ctx context.Context) error {
 	// Validate the advertise address.
 	advHost, advPort, err := validateAdvertiseAddr(ctx,
-		cfg.AdvertiseAddr, cfg.Addr, "", cliflags.ListenAddr)
+		cfg.AdvertiseAddr, cfg.Addr, "", cliflags.ListenAddr, cfg.Insecure)
 	if err != nil {
 		return invalidFlagErr(err, cliflags.AdvertiseAddr)
 	}
@@ -48,7 +51,7 @@ func (cfg *Config) ValidateAddrs(ctx context.Context) error {
 	// Validate the SQL advertise address. Use the provided advertise
 	// addr as default.
 	advSQLHost, advSQLPort, err := validateAdvertiseAddr(ctx,
-		cfg.SQLAdvertiseAddr, cfg.SQLAddr, advHost, cliflags.ListenSQLAddr)
+		cfg.SQLAdvertiseAddr, cfg.SQLAddr, advHost, cliflags.ListenSQLAddr, cfg.Insecure)
 	if err != nil {
 		return invalidFlagErr(err, cliflags.SQLAdvertiseAddr)
 	}
@@ -64,7 +67,7 @@ func (cfg *Config) ValidateAddrs(ctx context.Context) error {
 	// Validate the HTTP advertise address. Use the provided advertise
 	// addr as default.
 	advHTTPHost, advHTTPPort, err := validateAdvertiseAddr(ctx,
-		cfg.HTTPAdvertiseAddr, cfg.HTTPAddr, advHost, cliflags.ListenHTTPAddr)
+		cfg.HTTPAdvertiseAddr, cfg.HTTPAddr, advHost, cliflags.ListenHTTPAddr, cfg.Insecure)
 	if err != nil {
 		return errors.Wrap(err, "cannot compute public HTTP address")
 	}
@@ -86,7 +89,10 @@ func (cfg *Config) ValidateAddrs(ctx context.Context) error {
 // that if the "host" part is empty, it gets filled in with
 // the configured listen address if any, or the canonical host name.
 func validateAdvertiseAddr(
-	ctx context.Context, advAddr, listenAddr, defaultHost string, listenFlag cliflags.FlagInfo,
+	ctx context.Context,
+	advAddr, listenAddr, defaultHost string,
+	listenFlag cliflags.FlagInfo,
+	insecure bool,
 ) (string, string, error) {
 	listenHost, listenPort, err := getListenAddr(listenAddr, defaultHost)
 	if err != nil {
@@ -132,7 +138,12 @@ func validateAdvertiseAddr(
 			// locally but not elsewhere) but at least it prevents typos.
 			_, err = net.DefaultResolver.LookupIPAddr(ctx, advHost)
 			if err != nil {
-				return "", "", err
+				// Host resolution failed. Don't error out when running on Mac with `--insecure`.
+				if runtime.GOOS == "darwin" && insecure {
+					log.Ops.Shoutf(ctx, severity.WARNING, "Unable to resolve `hostname` due to %v\n", err)
+				} else {
+					return "", "", err
+				}
 			}
 		}
 	}

pkg/ccl/changefeedccl/BUILD.bazel

@@ -196,7 +196,6 @@ go_test(
     srcs = [
         "alter_changefeed_test.go",
         "changefeed_dist_test.go",
-        "changefeed_processors_test.go",
         "changefeed_stmt_test.go",
         "changefeed_test.go",
         "csv_test.go",

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -541,12 +541,19 @@ func generateAndValidateNewTargets(
 				(initialScanType == `only` && originalInitialScanOnlyOption)
 
 			// TODO(#142376): Audit whether this list is generated correctly.
-			var existingTargetIDs []descpb.ID
+			var existingTargetSpanIDs []spanID
 			for _, targetDesc := range newTableDescs {
-				existingTargetIDs = append(existingTargetIDs, targetDesc.GetID())
+				tableDesc, ok := targetDesc.(catalog.TableDescriptor)
+				if !ok {
+					return nil, nil, hlc.Timestamp{}, nil, errors.AssertionFailedf("expected table descriptor")
+				}
+				existingTargetSpanIDs = append(existingTargetSpanIDs, spanID{
+					tableID: tableDesc.GetID(),
+					indexID: tableDesc.GetPrimaryIndexID(),
+				})
 			}
-			existingTargetSpans := fetchSpansForDescs(p, existingTargetIDs)
-			var newTargetIDs []descpb.ID
+			existingTargetSpans := fetchSpansForDescs(p, existingTargetSpanIDs)
+			var addedTargetSpanIDs []spanID
 			for _, target := range v.Targets {
 				desc, found, err := getTargetDesc(ctx, p, descResolver, target.TableName)
 				if err != nil {
@@ -563,10 +570,17 @@ func generateAndValidateNewTargets(
 				k := targetKey{TableID: desc.GetID(), FamilyName: target.FamilyName}
 				newTargets[k] = target
 				newTableDescs[desc.GetID()] = desc
-				newTargetIDs = append(newTargetIDs, k.TableID)
+				tableDesc, ok := desc.(catalog.TableDescriptor)
+				if !ok {
+					return nil, nil, hlc.Timestamp{}, nil, errors.AssertionFailedf("expected table descriptor")
+				}
+				addedTargetSpanIDs = append(addedTargetSpanIDs, spanID{
+					tableID: tableDesc.GetID(),
+					indexID: tableDesc.GetPrimaryIndexID(),
+				})
 			}
 
-			addedTargetSpans := fetchSpansForDescs(p, newTargetIDs)
+			addedTargetSpans := fetchSpansForDescs(p, addedTargetSpanIDs)
 
 			// By default, we will not perform an initial scan on newly added
 			// targets. Hence, the user must explicitly state that they want an
@@ -632,13 +646,18 @@ func generateAndValidateNewTargets(
 		for k := range newTargets {
 			addedTargets[k.TableID] = struct{}{}
 		}
-		droppedIDs := make([]descpb.ID, 0, len(droppedTargets))
+		droppedSpanIDs := make([]spanID, 0, len(droppedTargets))
 		for k := range droppedTargets {
 			if _, wasAdded := addedTargets[k.TableID]; !wasAdded {
-				droppedIDs = append(droppedIDs, k.TableID)
+				// For dropped tables, we might not have the desc anymore so
+				// we can't get the index ID. In any case, it's safe to wipe
+				// out the entire table span.
+				droppedSpanIDs = append(droppedSpanIDs, spanID{
+					tableID: k.TableID,
+				})
 			}
 		}
-		droppedTargetSpans := fetchSpansForDescs(p, droppedIDs)
+		droppedTargetSpans := fetchSpansForDescs(p, droppedSpanIDs)
 		if err := removeSpansFromProgress(newJobProgress, droppedTargetSpans); err != nil {
 			return nil, nil, hlc.Timestamp{}, nil, err
 		}
@@ -874,19 +893,33 @@ func getSpanLevelCheckpointFromProgress(
 	return changefeedProgress.SpanLevelCheckpoint, nil
 }
 
-func fetchSpansForDescs(p sql.PlanHookState, descIDs []descpb.ID) (primarySpans []roachpb.Span) {
-	seen := make(map[descpb.ID]struct{})
+type spanID struct {
+	tableID descpb.ID
+	indexID descpb.IndexID
+}
+
+func fetchSpansForDescs(p sql.PlanHookState, spanIDs []spanID) (primarySpans []roachpb.Span) {
+	seen := make(map[spanID]struct{})
 	codec := p.ExtendedEvalContext().Codec
-	for _, id := range descIDs {
+	for _, id := range spanIDs {
 		if _, isDup := seen[id]; isDup {
 			continue
 		}
 		seen[id] = struct{}{}
-		tablePrefix := codec.TablePrefix(uint32(id))
-		primarySpan := roachpb.Span{
-			Key:    tablePrefix,
-			EndKey: tablePrefix.PrefixEnd(),
-		}
+		primarySpan := func() roachpb.Span {
+			if id.indexID == 0 {
+				tablePrefix := codec.TablePrefix(uint32(id.tableID))
+				return roachpb.Span{
+					Key:    tablePrefix,
+					EndKey: tablePrefix.PrefixEnd(),
+				}
+			}
+			indexPrefix := codec.IndexPrefix(uint32(id.tableID), uint32(id.indexID))
+			return roachpb.Span{
+				Key:    indexPrefix,
+				EndKey: indexPrefix.PrefixEnd(),
+			}
+		}()
 		primarySpans = append(primarySpans, primarySpan)
 	}
 	return primarySpans

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -615,7 +615,10 @@ func (ca *changeAggregator) setupSpansAndFrontier() (spans []roachpb.Span, err e
 	if err != nil {
 		return nil, err
 	}
-	ca.frontier, err = resolvedspan.NewAggregatorFrontier(ca.spec.Feed.StatementTime, initialHighWater, spans...)
+	ca.frontier, err = resolvedspan.NewAggregatorFrontier(
+		ca.spec.Feed.StatementTime, initialHighWater, ca.FlowCtx.Codec(),
+		&ca.FlowCtx.Cfg.Settings.SV,
+		spans...)
 	if err != nil {
 		return nil, err
 	}
@@ -1419,7 +1422,10 @@ func (cf *changeFrontier) Start(ctx context.Context) {
 	}
 
 	// Set up the resolved span frontier.
-	cf.frontier, err = resolvedspan.NewCoordinatorFrontier(cf.spec.Feed.StatementTime, initialHighwater, cf.spec.TrackedSpans...)
+	cf.frontier, err = resolvedspan.NewCoordinatorFrontier(
+		cf.spec.Feed.StatementTime, initialHighwater, cf.FlowCtx.Codec(),
+		&cf.FlowCtx.Cfg.Settings.SV,
+		cf.spec.TrackedSpans...)
 	if err != nil {
 		log.Infof(cf.Ctx(), "change frontier moving to draining due to error setting up frontier: %v", err)
 		cf.MoveToDraining(err)

pkg/ccl/changefeedccl/changefeed_processors_test.go

@@ -8936,7 +8936,13 @@ func TestChangefeedBackfillCheckpoint(t *testing.T) {
 
 		// Verify that the resumed job has restored the progress from the checkpoint
 		// to the change frontier.
-		expectedFrontier, err := span.MakeFrontier(tableSpan)
+		expectedFrontier, err := resolvedspan.NewCoordinatorFrontier(
+			hlc.Timestamp{},
+			hlc.Timestamp{},
+			s.Codec,
+			&s.Server.ClusterSettings().SV,
+			tableSpan,
+		)
 		if err != nil {
 			t.Fatal(err)
 		}

pkg/ccl/changefeedccl/changefeed_test.go

@@ -14,7 +14,9 @@ go_library(
         "//pkg/jobs/jobspb",
         "//pkg/roachpb",
         "//pkg/settings",
+        "//pkg/sql/catalog/descpb",
         "//pkg/util/hlc",
+        "//pkg/util/metamorphic",
         "//pkg/util/span",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
@@ -27,10 +29,15 @@ go_test(
     deps = [
         ":resolvedspan",
         "//pkg/jobs/jobspb",
+        "//pkg/keys",
         "//pkg/roachpb",
+        "//pkg/settings/cluster",
+        "//pkg/sql/catalog/descpb",
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",
+        "//pkg/util/randutil",
+        "//pkg/util/span",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",
     ],