diff --git a/docs/generated/sql/bnf/show_tenant_stmt.bnf b/docs/generated/sql/bnf/show_tenant_stmt.bnf
index 3865df1085f5..39a85d8defde 100644
--- a/docs/generated/sql/bnf/show_tenant_stmt.bnf
+++ b/docs/generated/sql/bnf/show_tenant_stmt.bnf
@@ -1,2 +1,3 @@
 show_tenant_stmt ::=
-	'SHOW' 'TENANT' name
+	'SHOW' 'TENANT' d_expr
+	| 'SHOW' 'TENANT' d_expr 'WITH' 'REPLICATION' 'STATUS'
diff --git a/docs/generated/sql/bnf/stmt_block.bnf b/docs/generated/sql/bnf/stmt_block.bnf
index 8b51f63ca219..b981ae3e5259 100644
--- a/docs/generated/sql/bnf/stmt_block.bnf
+++ b/docs/generated/sql/bnf/stmt_block.bnf
@@ -922,7 +922,8 @@ show_tables_stmt ::=
 	| 'SHOW' 'TABLES' with_comment
 
 show_tenant_stmt ::=
-	'SHOW' 'TENANT' name
+	'SHOW' 'TENANT' d_expr
+	| 'SHOW' 'TENANT' d_expr 'WITH' 'REPLICATION' 'STATUS'
 
 show_trace_stmt ::=
 	'SHOW' opt_compact 'TRACE' 'FOR' 'SESSION'
diff --git a/pkg/ccl/streamingccl/streamclient/client.go b/pkg/ccl/streamingccl/streamclient/client.go
index 923436ad54d9..6462ed731f07 100644
--- a/pkg/ccl/streamingccl/streamclient/client.go
+++ b/pkg/ccl/streamingccl/streamclient/client.go
@@ -172,6 +172,9 @@ func NewStreamClient(
 // GetFirstActiveClient iterates through each provided stream address
 // and returns the first client it's able to successfully Dial.
 func GetFirstActiveClient(ctx context.Context, streamAddresses []string) (Client, error) {
+	if len(streamAddresses) == 0 {
+		return nil, errors.Newf("failed to connect, no partition addresses")
+	}
 	var combinedError error = nil
 	for _, address := range streamAddresses {
 		streamAddress := streamingccl.StreamAddress(address)
diff --git a/pkg/ccl/streamingccl/streamclient/client_test.go b/pkg/ccl/streamingccl/streamclient/client_test.go
index ebb1abf0ca8f..7f8a1c5db78c 100644
--- a/pkg/ccl/streamingccl/streamclient/client_test.go
+++ b/pkg/ccl/streamingccl/streamclient/client_test.go
@@ -122,6 +122,15 @@ func (t testStreamSubscription) Err() error {
 	return nil
 }
 
+func TestGetFirstActiveClientEmpty(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	var streamAddresses []string
+	activeClient, err := GetFirstActiveClient(context.Background(), streamAddresses)
+	require.ErrorContains(t, err, "failed to connect, no partition addresses")
+	require.Nil(t, activeClient)
+}
+
 func TestGetFirstActiveClient(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
diff --git a/pkg/ccl/streamingccl/streamingest/stream_replication_e2e_test.go b/pkg/ccl/streamingccl/streamingest/stream_replication_e2e_test.go
index 9438c3f2a77b..9026f7927363 100644
--- a/pkg/ccl/streamingccl/streamingest/stream_replication_e2e_test.go
+++ b/pkg/ccl/streamingccl/streamingest/stream_replication_e2e_test.go
@@ -44,6 +44,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
@@ -1244,3 +1245,46 @@ func TestTenantReplicationProtectedTimestampManagement(t *testing.T) {
 		})
 	})
 }
+
+// TODO(lidor): consider rewriting this test as a data driven test when #92609 is merged.
+func TestTenantStreamingShowTenant(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	args := defaultTenantStreamingClustersArgs
+
+	c, cleanup := createTenantStreamingClusters(ctx, t, args)
+	defer cleanup()
+	testStartTime := timeutil.Now()
+	producerJobID, ingestionJobID := c.startStreamReplication()
+
+	jobutils.WaitForJobToRun(c.t, c.srcSysSQL, jobspb.JobID(producerJobID))
+	jobutils.WaitForJobToRun(c.t, c.destSysSQL, jobspb.JobID(ingestionJobID))
+	highWatermark := c.srcCluster.Server(0).Clock().Now()
+	c.waitUntilHighWatermark(highWatermark, jobspb.JobID(ingestionJobID))
+
+	var (
+		id            int
+		dest          string
+		status        string
+		source        string
+		sourceUri     string
+		jobId         int
+		maxReplTime   time.Time
+		protectedTime time.Time
+	)
+	row := c.destSysSQL.QueryRow(t, fmt.Sprintf("SHOW TENANT %s WITH REPLICATION STATUS", args.destTenantName))
+	row.Scan(&id, &dest, &status, &source, &sourceUri, &jobId, &maxReplTime, &protectedTime)
+	require.Equal(t, 2, id)
+	require.Equal(t, "destination", dest)
+	require.Equal(t, "ADD", status)
+	require.Equal(t, "source", source)
+	require.Equal(t, c.srcURL.String(), sourceUri)
+	require.Equal(t, ingestionJobID, jobId)
+	require.Less(t, maxReplTime, timeutil.Now())
+	require.Less(t, protectedTime, timeutil.Now())
+	require.GreaterOrEqual(t, maxReplTime, highWatermark.GoTime())
+	// TODO(lidor): replace this start time with the actual replication start time when we have it.
+	require.GreaterOrEqual(t, protectedTime, testStartTime)
+}
diff --git a/pkg/cli/cliflags/flags.go b/pkg/cli/cliflags/flags.go
index d8582ba88338..0dd5ae31f4ba 100644
--- a/pkg/cli/cliflags/flags.go
+++ b/pkg/cli/cliflags/flags.go
@@ -1642,46 +1642,46 @@ the --log flag.`,
 present in the body of the logging configuration.`,
 	}
 
-	DeprecatedStderrThreshold = FlagInfo{
-		Name:        "logtostderr",
-		Description: `Write log messages beyond the specified severity to stderr.`,
+	StderrThresholdOverride = FlagInfo{
+		Name: "logtostderr",
+		Description: `--logtostderr=XXX is an alias for --log='sinks: {stderr: {filter: XXX}}'.
+If no value is specified, the default value for the command is inferred: INFO for server
+commands, WARNING for client commands.`,
 	}
 
-	DeprecatedFileThreshold = FlagInfo{
+	FileThresholdOverride = FlagInfo{
 		Name:        "log-file-verbosity",
-		Description: `Write log messages beyond the specified severity to files.`,
+		Description: `--log-file-verbosity=XXX is an alias for --log='file-defaults: {filter: XXX}}'.`,
 	}
 
-	DeprecatedStderrNoColor = FlagInfo{
+	StderrNoColorOverride = FlagInfo{
 		Name:        "no-color",
-		Description: `Avoid color in the stderr output.`,
+		Description: `--no-color=XXX is an alias for --log='sinks: {stderr: {no-color: XXX}}'.`,
 	}
 
-	DeprecatedRedactableLogs = FlagInfo{
+	RedactableLogsOverride = FlagInfo{
 		Name:        "redactable-logs",
-		Description: `Request redaction markers.`,
+		Description: `--redactable-logs=XXX is an alias for --log='file-defaults: {redactable: XXX}}'.`,
 	}
 
-	DeprecatedLogFileMaxSize = FlagInfo{
+	LogFileMaxSizeOverride = FlagInfo{
 		Name:        "log-file-max-size",
-		Description: "Maximum size of a log file before switching to a new file.",
+		Description: "--log-file-max-size=XXX is an alias for --log='file-defaults: {max-file-size: XXX}'.",
 	}
 
-	DeprecatedLogGroupMaxSize = FlagInfo{
+	LogGroupMaxSizeOverride = FlagInfo{
 		Name:        "log-group-max-size",
-		Description: `Maximum size of a group of log files before old files are removed.`,
+		Description: `--log-group-max-size=XXX is an alias for --log='file-defaults: {max-group-size: XXX}'.`,
 	}
 
-	DeprecatedLogDir = FlagInfo{
+	LogDirOverride = FlagInfo{
 		Name:        "log-dir",
-		Description: `Override the logging directory.`,
+		Description: `--log-dir=XXX is an alias for --log='file-defaults: {dir: XXX}'.`,
 	}
 
-	DeprecatedSQLAuditLogDir = FlagInfo{
-		Name: "sql-audit-dir",
-		Description: `
-If non-empty, create a SQL audit log in this directory.
-`,
+	SQLAuditLogDirOverride = FlagInfo{
+		Name:        "sql-audit-dir",
+		Description: `--sql-audit-dir=XXX is an alias for --log='sinks: {file-groups: {sql-audit: {channels: SENSITIVE_ACCESS, dir: ...}}}'.`,
 	}
 
 	BuildTag = FlagInfo{
diff --git a/pkg/cli/context.go b/pkg/cli/context.go
index fbf60e921e7b..5f9d974c3040 100644
--- a/pkg/cli/context.go
+++ b/pkg/cli/context.go
@@ -171,10 +171,9 @@ type cliContext struct {
 	// before exiting the process. This may block until all buffered log sinks
 	// are shutdown, if any are enabled, or until a timeout is triggered.
 	logShutdownFn func()
-	// deprecatedLogOverrides is the legacy pre-v21.1 discrete flag
-	// overrides for the logging configuration.
-	// TODO(knz): Deprecated in v21.1. Remove this.
-	deprecatedLogOverrides *logConfigFlags
+	// logOverrides encodes discrete flag overrides for the logging
+	// configuration. They are provided for convenience.
+	logOverrides *logConfigFlags
 	// ambiguousLogDir is populated during setupLogging() to indicate
 	// that no log directory was specified and there were multiple
 	// on-disk stores.
@@ -187,9 +186,8 @@ type cliContext struct {
 // cliCtx captures the command-line parameters common to most CLI utilities.
 // See below for defaults.
 var cliCtx = cliContext{
-	Config: baseCfg,
-	// TODO(knz): Deprecated in v21.1. Remove this.
-	deprecatedLogOverrides: newLogConfigOverrides(),
+	Config:       baseCfg,
+	logOverrides: newLogConfigOverrides(),
 }
 
 // setCliContextDefaults set the default values in cliCtx.  This
@@ -214,8 +212,7 @@ func setCliContextDefaults() {
 	cliCtx.logConfig = logconfig.Config{}
 	cliCtx.logShutdownFn = func() {}
 	cliCtx.ambiguousLogDir = false
-	// TODO(knz): Deprecated in v21.1. Remove this.
-	cliCtx.deprecatedLogOverrides.reset()
+	cliCtx.logOverrides.reset()
 	cliCtx.showVersionUsingOnlyBuildTag = false
 }
 
@@ -610,7 +607,7 @@ func setDemoContextDefaults() {
 	demoCtx.NumNodes = 1
 	demoCtx.SQLPoolMemorySize = 128 << 20 // 128MB, chosen to fit 9 nodes on 2GB machine.
 	demoCtx.CacheSize = 64 << 20          // 64MB, chosen to fit 9 nodes on 2GB machine.
-	demoCtx.NoExampleDatabase = false
+	demoCtx.UseEmptyDatabase = false
 	demoCtx.SimulateLatency = false
 	demoCtx.RunWorkload = false
 	demoCtx.Localities = nil
diff --git a/pkg/cli/demo.go b/pkg/cli/demo.go
index eaf5359010d4..c1a5a53d399a 100644
--- a/pkg/cli/demo.go
+++ b/pkg/cli/demo.go
@@ -115,13 +115,13 @@ func checkDemoConfiguration(
 	cmd *cobra.Command, gen workload.Generator,
 ) (workload.Generator, error) {
 	f := cliflagcfg.FlagSetForCmd(cmd)
-	if gen == nil && !demoCtx.NoExampleDatabase {
+	if gen == nil && !demoCtx.UseEmptyDatabase {
 		// Use a default dataset unless prevented by --no-example-database.
 		gen = defaultGenerator
 	}
 
 	// Make sure that the user didn't request a workload and an empty database.
-	if demoCtx.RunWorkload && demoCtx.NoExampleDatabase {
+	if demoCtx.RunWorkload && demoCtx.UseEmptyDatabase {
 		return nil, errors.New("cannot run a workload when generation of the example database is disabled")
 	}
 
@@ -151,7 +151,7 @@ func checkDemoConfiguration(
 		}
 
 		// Make sure that the user didn't request to have a topology and disable the example database.
-		if demoCtx.NoExampleDatabase {
+		if demoCtx.UseEmptyDatabase {
 			return nil, errors.New("cannot setup geo-partitioned replicas topology without generating an example database")
 		}
 
diff --git a/pkg/cli/democluster/context.go b/pkg/cli/democluster/context.go
index de8eb5dfde2a..373c637051ee 100644
--- a/pkg/cli/democluster/context.go
+++ b/pkg/cli/democluster/context.go
@@ -35,9 +35,9 @@ type Context struct {
 	// CacheSize is the size of the storage cache for each KV server.
 	CacheSize int64
 
-	// NoExampleDatabase prevents the auto-creation of a demo database
+	// UseEmptyDatabase prevents the auto-creation of a demo database
 	// from a workload.
-	NoExampleDatabase bool
+	UseEmptyDatabase bool
 
 	// RunWorkload indicates whether to run a workload in the background
 	// after the demo cluster has been initialized.
diff --git a/pkg/cli/flags.go b/pkg/cli/flags.go
index d8f2d354b369..2442dcab925e 100644
--- a/pkg/cli/flags.go
+++ b/pkg/cli/flags.go
@@ -20,7 +20,6 @@ import (
 	"strings"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
-	"github.com/cockroachdb/cockroach/pkg/build"
 	"github.com/cockroachdb/cockroach/pkg/cli/clientflags"
 	"github.com/cockroachdb/cockroach/pkg/cli/clienturl"
 	"github.com/cockroachdb/cockroach/pkg/cli/cliflagcfg"
@@ -327,41 +326,28 @@ func init() {
 		cliflagcfg.VarFlag(pf, &fileContentsValue{settableString: &cliCtx.logConfigInput, fileName: "<unset>"}, cliflags.LogConfigFile)
 		cliflagcfg.StringSliceFlag(pf, &cliCtx.logConfigVars, cliflags.LogConfigVars)
 
-		// Pre-v21.1 overrides. Deprecated.
-		// TODO(knz): Remove this.
-		cliflagcfg.VarFlag(pf, &cliCtx.deprecatedLogOverrides.stderrThreshold, cliflags.DeprecatedStderrThreshold)
-		_ = pf.MarkDeprecated(cliflags.DeprecatedStderrThreshold.Name,
-			"use --"+cliflags.Log.Name+" instead to specify 'sinks: {stderr: {filter: ...}}'.")
+		// Discrete convenience overrides.
+		cliflagcfg.VarFlag(pf, &cliCtx.logOverrides.stderrThreshold, cliflags.StderrThresholdOverride)
 		// This flag can also be specified without an explicit argument.
-		pf.Lookup(cliflags.DeprecatedStderrThreshold.Name).NoOptDefVal = "DEFAULT"
+		pf.Lookup(cliflags.StderrThresholdOverride.Name).NoOptDefVal = "DEFAULT"
 
-		cliflagcfg.VarFlag(pf, &cliCtx.deprecatedLogOverrides.stderrNoColor, cliflags.DeprecatedStderrNoColor)
-		_ = pf.MarkDeprecated(cliflags.DeprecatedStderrNoColor.Name,
-			"use --"+cliflags.Log.Name+" instead to specify 'sinks: {stderr: {no-color: true}}'")
+		cliflagcfg.VarFlag(pf, &cliCtx.logOverrides.stderrNoColor, cliflags.StderrNoColorOverride)
+		_ = pf.MarkHidden(cliflags.StderrNoColorOverride.Name)
+		cliflagcfg.VarFlag(pf, &stringValue{&cliCtx.logOverrides.logDir}, cliflags.LogDirOverride)
 
-		cliflagcfg.VarFlag(pf, &stringValue{&cliCtx.deprecatedLogOverrides.logDir}, cliflags.DeprecatedLogDir)
-		_ = pf.MarkDeprecated(cliflags.DeprecatedLogDir.Name,
-			"use --"+cliflags.Log.Name+" instead to specify 'file-defaults: {dir: ...}'")
+		cliflagcfg.VarFlag(pf, cliCtx.logOverrides.fileMaxSizeVal, cliflags.LogFileMaxSizeOverride)
+		_ = pf.MarkHidden(cliflags.LogFileMaxSizeOverride.Name)
 
-		cliflagcfg.VarFlag(pf, cliCtx.deprecatedLogOverrides.fileMaxSizeVal, cliflags.DeprecatedLogFileMaxSize)
-		_ = pf.MarkDeprecated(cliflags.DeprecatedLogFileMaxSize.Name,
-			"use --"+cliflags.Log.Name+" instead to specify 'file-defaults: {max-file-size: ...}'")
+		cliflagcfg.VarFlag(pf, cliCtx.logOverrides.maxGroupSizeVal, cliflags.LogGroupMaxSizeOverride)
+		_ = pf.MarkHidden(cliflags.LogGroupMaxSizeOverride.Name)
 
-		cliflagcfg.VarFlag(pf, cliCtx.deprecatedLogOverrides.maxGroupSizeVal, cliflags.DeprecatedLogGroupMaxSize)
-		_ = pf.MarkDeprecated(cliflags.DeprecatedLogGroupMaxSize.Name,
-			"use --"+cliflags.Log.Name+" instead to specify 'file-defaults: {max-group-size: ...}'")
+		cliflagcfg.VarFlag(pf, &cliCtx.logOverrides.fileThreshold, cliflags.FileThresholdOverride)
+		_ = pf.MarkHidden(cliflags.FileThresholdOverride.Name)
 
-		cliflagcfg.VarFlag(pf, &cliCtx.deprecatedLogOverrides.fileThreshold, cliflags.DeprecatedFileThreshold)
-		_ = pf.MarkDeprecated(cliflags.DeprecatedFileThreshold.Name,
-			"use --"+cliflags.Log.Name+" instead to specify 'file-defaults: {filter: ...}'")
+		cliflagcfg.VarFlag(pf, &cliCtx.logOverrides.redactableLogs, cliflags.RedactableLogsOverride)
 
-		cliflagcfg.VarFlag(pf, &cliCtx.deprecatedLogOverrides.redactableLogs, cliflags.DeprecatedRedactableLogs)
-		_ = pf.MarkDeprecated(cliflags.DeprecatedRedactableLogs.Name,
-			"use --"+cliflags.Log.Name+" instead to specify 'file-defaults: {redactable: ...}")
-
-		cliflagcfg.VarFlag(pf, &stringValue{&cliCtx.deprecatedLogOverrides.sqlAuditLogDir}, cliflags.DeprecatedSQLAuditLogDir)
-		_ = pf.MarkDeprecated(cliflags.DeprecatedSQLAuditLogDir.Name,
-			"use --"+cliflags.Log.Name+" instead to specify 'sinks: {file-groups: {sql-audit: {channels: SENSITIVE_ACCESS, dir: ...}}}")
+		cliflagcfg.VarFlag(pf, &stringValue{&cliCtx.logOverrides.sqlAuditLogDir}, cliflags.SQLAuditLogDirOverride)
+		_ = pf.MarkHidden(cliflags.SQLAuditLogDirOverride.Name)
 	}
 
 	// Remember we are starting in the background as the `start` command will
@@ -800,11 +786,8 @@ func init() {
 		cliflagcfg.BoolFlag(f, &demoCtx.GeoPartitionedReplicas, cliflags.DemoGeoPartitionedReplicas)
 		cliflagcfg.VarFlag(f, &demoCtx.demoNodeSQLMemSizeValue, cliflags.DemoNodeSQLMemSize)
 		cliflagcfg.VarFlag(f, &demoCtx.demoNodeCacheSizeValue, cliflags.DemoNodeCacheSize)
-		cliflagcfg.BoolFlag(f, &demoCtx.Insecure, cliflags.ClientInsecure)
 		// NB: Insecure for `cockroach demo` is deprecated. See #53404.
-		_ = f.MarkDeprecated(cliflags.ServerInsecure.Name,
-			"to start a test server without any security, run start-single-node --insecure\n"+
-				"For details, see: "+build.MakeIssueURL(53404))
+		cliflagcfg.BoolFlag(f, &demoCtx.Insecure, cliflags.ClientInsecure)
 
 		cliflagcfg.BoolFlag(f, &demoCtx.disableEnterpriseFeatures, cliflags.DemoNoLicense)
 		cliflagcfg.BoolFlag(f, &demoCtx.DefaultEnableRangefeeds, cliflags.DemoEnableRangefeeds)
@@ -830,9 +813,11 @@ func init() {
 		// The --empty flag is only valid for the top level demo command,
 		// so we use the regular flag set.
 		f := demoCmd.Flags()
-		cliflagcfg.BoolFlag(f, &demoCtx.NoExampleDatabase, cliflags.UseEmptyDatabase)
-		_ = f.MarkDeprecated(cliflags.UseEmptyDatabase.Name, "use --no-example-database")
-		cliflagcfg.BoolFlag(f, &demoCtx.NoExampleDatabase, cliflags.NoExampleDatabase)
+		cliflagcfg.BoolFlag(f, &demoCtx.UseEmptyDatabase, cliflags.UseEmptyDatabase)
+
+		// --no-example-database is an old name for --empty.
+		cliflagcfg.BoolFlag(f, &demoCtx.UseEmptyDatabase, cliflags.NoExampleDatabase)
+		_ = f.MarkHidden(cliflags.NoExampleDatabase.Name)
 	}
 
 	// statement-diag command.
diff --git a/pkg/cli/flags_test.go b/pkg/cli/flags_test.go
index 530737eedab1..62127d079c4c 100644
--- a/pkg/cli/flags_test.go
+++ b/pkg/cli/flags_test.go
@@ -1226,23 +1226,35 @@ Available Commands:
   help              Help about any command
 
 Flags:
-  -h, --help                      help for cockroach
-      --log <string>              
-                                   Logging configuration, expressed using YAML syntax. For example, you can
-                                   change the default logging directory with: --log='file-defaults: {dir: ...}'.
-                                   See the documentation for more options and details.  To preview how the log
-                                   configuration is applied, or preview the default configuration, you can use
-                                   the 'cockroach debug check-log-config' sub-command.
-                                  
-      --log-config-file <file>    
-                                   File name to read the logging configuration from. This has the same effect as
-                                   passing the content of the file via the --log flag.
-                                   (default <unset>)
-      --log-config-vars strings   
-                                   Environment variables that will be expanded if present in the body of the
-                                   logging configuration.
-                                  
-      --version                   version for cockroach
+  -h, --help                               help for cockroach
+      --log <string>                       
+                                            Logging configuration, expressed using YAML syntax. For example, you can
+                                            change the default logging directory with: --log='file-defaults: {dir: ...}'.
+                                            See the documentation for more options and details.  To preview how the log
+                                            configuration is applied, or preview the default configuration, you can use
+                                            the 'cockroach debug check-log-config' sub-command.
+                                           
+      --log-config-file <file>             
+                                            File name to read the logging configuration from. This has the same effect as
+                                            passing the content of the file via the --log flag.
+                                            (default <unset>)
+      --log-config-vars strings            
+                                            Environment variables that will be expanded if present in the body of the
+                                            logging configuration.
+                                           
+      --log-dir <string>                   
+                                            --log-dir=XXX is an alias for --log='file-defaults: {dir: XXX}'.
+                                           
+      --logtostderr <severity>[=DEFAULT]   
+                                            --logtostderr=XXX is an alias for --log='sinks: {stderr: {filter: XXX}}'. If
+                                            no value is specified, the default value for the command is inferred: INFO for
+                                            server commands, WARNING for client commands.
+                                            (default UNKNOWN)
+      --redactable-logs                    
+                                            --redactable-logs=XXX is an alias for --log='file-defaults: {redactable:
+                                            XXX}}'.
+                                           
+      --version                            version for cockroach
 
 Use "cockroach [command] --help" for more information about a command.
 `
@@ -1251,7 +1263,7 @@ Use "cockroach [command] --help" for more information about a command.
 	// (But it does if the test is run a second time.)
 	// Strangely, 'cockroach --help' does, as well as usage error messages.
 	helpExpected1 := commandPrefix + expUsage
-	helpExpected2 := strings.ReplaceAll(helpExpected1, "      --version                   version for cockroach\n", "")
+	helpExpected2 := strings.ReplaceAll(helpExpected1, "      --version                            version for cockroach\n", "")
 	badFlagExpected := fmt.Sprintf("%s\nError: unknown flag: --foo\n", expUsage)
 
 	testCases := []struct {
@@ -1313,8 +1325,8 @@ Use "cockroach [command] --help" for more information about a command.
 			if test.expectHelp {
 				if got != helpExpected1 && got != helpExpected2 {
 					diff, _ := difflib.GetUnifiedDiffString(difflib.UnifiedDiff{
-						A:        difflib.SplitLines(helpExpected1),
-						B:        difflib.SplitLines(got),
+						A:        difflib.SplitLines(strings.ReplaceAll(helpExpected1, "\n", "$\n")),
+						B:        difflib.SplitLines(strings.ReplaceAll(got, "\n", "$\n")),
 						FromFile: "Expected",
 						ToFile:   "Actual",
 						Context:  1,
diff --git a/pkg/cli/log_flags.go b/pkg/cli/log_flags.go
index d22acfaa7c5b..7d5f23a93805 100644
--- a/pkg/cli/log_flags.go
+++ b/pkg/cli/log_flags.go
@@ -41,7 +41,7 @@ import (
 // and non-server commands (e.g. 'node ls').
 func setupLogging(ctx context.Context, cmd *cobra.Command, isServerCmd, applyConfig bool) error {
 	// Compatibility check for command-line usage.
-	if cliCtx.deprecatedLogOverrides.anySet() &&
+	if cliCtx.logOverrides.anySet() &&
 		cliCtx.logConfigInput.isSet {
 		return errors.Newf("--%s is incompatible with legacy discrete logging flags", cliflags.Log.Name)
 	}
@@ -65,17 +65,16 @@ func setupLogging(ctx context.Context, cmd *cobra.Command, isServerCmd, applyCon
 	defaultLogDir := firstStoreDir
 
 	// Legacy log directory override.
-	// TODO(knz): Deprecated in v21.1. Remove in v21.2.
-	forceDisableLogDir := cliCtx.deprecatedLogOverrides.logDir.isSet &&
-		cliCtx.deprecatedLogOverrides.logDir.s == ""
+	forceDisableLogDir := cliCtx.logOverrides.logDir.isSet &&
+		cliCtx.logOverrides.logDir.s == ""
 	if forceDisableLogDir {
 		defaultLogDir = nil
 	}
-	forceSetLogDir := cliCtx.deprecatedLogOverrides.logDir.isSet &&
-		cliCtx.deprecatedLogOverrides.logDir.s != ""
+	forceSetLogDir := cliCtx.logOverrides.logDir.isSet &&
+		cliCtx.logOverrides.logDir.s != ""
 	if forceSetLogDir {
 		ambiguousLogDirs = false
-		defaultLogDir = &cliCtx.deprecatedLogOverrides.logDir.s
+		defaultLogDir = &cliCtx.logOverrides.logDir.s
 	}
 
 	// Set up a base configuration template.
@@ -115,11 +114,7 @@ func setupLogging(ctx context.Context, cmd *cobra.Command, isServerCmd, applyCon
 
 	// If some overrides were specified via the discrete flags,
 	// apply them.
-	//
-	// NB: this is for backward-compatibility and is deprecated in
-	// v21.1.
-	// TODO(knz): Remove this.
-	cliCtx.deprecatedLogOverrides.propagate(&h.Config, commandSpecificDefaultLegacyStderrOverride)
+	cliCtx.logOverrides.propagate(&h.Config, commandSpecificDefaultLegacyStderrOverride)
 
 	// If a configuration was specified via --log, load it.
 	if cliCtx.logConfigInput.isSet {
@@ -143,7 +138,6 @@ func setupLogging(ctx context.Context, cmd *cobra.Command, isServerCmd, applyCon
 
 	// Legacy behavior: if no files were specified by the configuration,
 	// add some pre-defined files in servers.
-	// TODO(knz): Deprecated in v21.1. Remove this.
 	if isServerCmd && len(h.Config.Sinks.FileGroups) == 0 {
 		addPredefinedLogFiles(&h.Config)
 	}
@@ -260,8 +254,6 @@ func getDefaultLogDirFromStores() (dir *string, ambiguousLogDirs bool) {
 // This struct interfaces between the YAML-based configuration
 // mechanism in package 'logconfig', and the logic that initializes
 // the logging system.
-//
-// TODO(knz): Deprecated in v21.1. Remove this.
 type logConfigFlags struct {
 	// Override value of file-defaults:dir.
 	logDir settableString
@@ -292,8 +284,6 @@ type logConfigFlags struct {
 
 // newLogConfigOverrides defines a new logConfigFlags
 // from the default logging configuration.
-//
-// TODO(knz): Deprecated in v21.1. Remove this.
 func newLogConfigOverrides() *logConfigFlags {
 	l := &logConfigFlags{}
 	l.fileMaxSizeVal = humanizeutil.NewBytesValue(&l.fileMaxSize)
@@ -444,14 +434,13 @@ func addPredefinedLogFiles(c *logconfig.Config) {
 		panic(errors.NewAssertionErrorWithWrappedErrf(err, "programming error: incorrect config"))
 	}
 	*c = h.Config
-	if cliCtx.deprecatedLogOverrides.sqlAuditLogDir.isSet {
-		c.Sinks.FileGroups["sql-audit"].Dir = &cliCtx.deprecatedLogOverrides.sqlAuditLogDir.s
+	if cliCtx.logOverrides.sqlAuditLogDir.isSet {
+		c.Sinks.FileGroups["sql-audit"].Dir = &cliCtx.logOverrides.sqlAuditLogDir.s
 	}
 }
 
 // predefinedLogFiles are the files defined when the --log flag
 // does not otherwise override the file sinks.
-// TODO(knz): add the PRIVILEGES channel.
 const predefinedLogFiles = `
 sinks:
  file-groups:
diff --git a/pkg/cli/statement_bundle.go b/pkg/cli/statement_bundle.go
index dc8fca6dc2d5..b0d004ce343f 100644
--- a/pkg/cli/statement_bundle.go
+++ b/pkg/cli/statement_bundle.go
@@ -123,7 +123,7 @@ func runBundleRecreate(cmd *cobra.Command, args []string) (resErr error) {
 		return err
 	}
 
-	demoCtx.NoExampleDatabase = true
+	demoCtx.UseEmptyDatabase = true
 	return runDemoInternal(cmd, nil /* gen */, func(ctx context.Context, conn clisqlclient.Conn) error {
 		// Disable autostats collection, which will override the injected stats.
 		if err := conn.Exec(ctx,
diff --git a/pkg/roachprod/config/config.go b/pkg/roachprod/config/config.go
index 59e7b54fd192..d8bc76903fc4 100644
--- a/pkg/roachprod/config/config.go
+++ b/pkg/roachprod/config/config.go
@@ -86,8 +86,6 @@ func DefaultEnvVars() []string {
 		// when moving snapshots around, though.
 		// (For other perf. related knobs, see https://github.com/cockroachdb/cockroach/issues/17165)
 		"COCKROACH_ENABLE_RPC_COMPRESSION=false",
-		// Get rid of an annoying popup in the UI.
-		"COCKROACH_UI_RELEASE_NOTES_SIGNUP_DISMISSED=true",
 		// Allow upgrading a stable release data-dir to a dev version.
 		// N.B. many roachtests which perform upgrade scenarios require this env. var after changes in [1]; otherwise,
 		// the tests will fail even on release branches when attempting to upgrade previous (stable) release to an alpha.
diff --git a/pkg/sql/BUILD.bazel b/pkg/sql/BUILD.bazel
index 0a812636d458..b65429c58773 100644
--- a/pkg/sql/BUILD.bazel
+++ b/pkg/sql/BUILD.bazel
@@ -307,6 +307,7 @@ go_library(
         "//pkg/obsservice/obspb/opentelemetry-proto/common/v1:common",
         "//pkg/obsservice/obspb/opentelemetry-proto/logs/v1:logs",
         "//pkg/repstream",
+        "//pkg/repstream/streampb",
         "//pkg/roachpb",
         "//pkg/rpc",
         "//pkg/rpc/nodedialer",
diff --git a/pkg/sql/catalog/colinfo/result_columns.go b/pkg/sql/catalog/colinfo/result_columns.go
index 4d8dfe768624..43e563a2dfee 100644
--- a/pkg/sql/catalog/colinfo/result_columns.go
+++ b/pkg/sql/catalog/colinfo/result_columns.go
@@ -278,3 +278,17 @@ var TenantColumns = ResultColumns{
 	{Name: "name", Typ: types.String},
 	{Name: "status", Typ: types.String},
 }
+
+var TenantColumnsWithReplication = ResultColumns{
+	{Name: "id", Typ: types.Int},
+	{Name: "name", Typ: types.String},
+	{Name: "status", Typ: types.String},
+	{Name: "source_tenant_name", Typ: types.String},
+	{Name: "source_cluster_uri", Typ: types.String},
+	{Name: "replication_job_id", Typ: types.Int},
+	// The latest fully replicated time.
+	{Name: "replicated_time", Typ: types.Timestamp},
+	// The protected timestamp on the destination cluster, meaning we cannot
+	// cutover to before this time.
+	{Name: "retained_time", Typ: types.Timestamp},
+}
diff --git a/pkg/sql/faketreeeval/BUILD.bazel b/pkg/sql/faketreeeval/BUILD.bazel
index 56ecd1b2a178..81f115711eb7 100644
--- a/pkg/sql/faketreeeval/BUILD.bazel
+++ b/pkg/sql/faketreeeval/BUILD.bazel
@@ -8,6 +8,8 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/clusterversion",
+        "//pkg/jobs/jobspb",
+        "//pkg/repstream/streampb",
         "//pkg/roachpb",
         "//pkg/security/username",
         "//pkg/sql/catalog/catpb",
diff --git a/pkg/sql/faketreeeval/evalctx.go b/pkg/sql/faketreeeval/evalctx.go
index f70d1bc41cf2..d47e8d4ce800 100644
--- a/pkg/sql/faketreeeval/evalctx.go
+++ b/pkg/sql/faketreeeval/evalctx.go
@@ -16,6 +16,8 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
@@ -594,6 +596,12 @@ func (c *DummyTenantOperator) GetTenantInfo(
 	return nil, errors.WithStack(errEvalTenant)
 }
 
+func (p *DummyTenantOperator) GetTenantReplicationInfo(
+	ctx context.Context, replicationJobId jobspb.JobID,
+) (*streampb.StreamIngestionStats, error) {
+	return nil, errors.WithStack(errEvalTenant)
+}
+
 // DestroyTenant is part of the tree.TenantOperator interface.
 func (c *DummyTenantOperator) DestroyTenant(
 	ctx context.Context, tenantName roachpb.TenantName, synchronous bool,
diff --git a/pkg/sql/logictest/testdata/logic_test/tenant b/pkg/sql/logictest/testdata/logic_test/tenant
index a8e2e5ca961c..de923cc459de 100644
--- a/pkg/sql/logictest/testdata/logic_test/tenant
+++ b/pkg/sql/logictest/testdata/logic_test/tenant
@@ -60,6 +60,12 @@ id  name   status
 statement error tenant "seven" does not exist
 SHOW TENANT seven
 
+statement error tenant "tenant-one" does not have an active replication job
+SHOW TENANT "tenant-one" WITH REPLICATION STATUS
+
+statement error tenant "two" does not have an active replication job
+SHOW TENANT two WITH REPLICATION STATUS
+
 # Test creating a tenant with the same name as an existing tenant, but a unique
 # ID.
 statement error tenant with name "three" already exists
diff --git a/pkg/sql/opt/exec/execbuilder/testdata/udf b/pkg/sql/opt/exec/execbuilder/testdata/udf
index ba6cc4827b5d..69f79da7b371 100644
--- a/pkg/sql/opt/exec/execbuilder/testdata/udf
+++ b/pkg/sql/opt/exec/execbuilder/testdata/udf
@@ -95,3 +95,42 @@ SELECT fetch_a_of_1_strict(NULL::INT)
 query T kvtrace
 SELECT fetch_a_of_2_strict(1, NULL::INT)
 ----
+
+# Regression test for #93210. Do not plan unnecessary assignment casts on the
+# return values of UDFs.
+statement ok
+CREATE TABLE t93210 (
+  a INT PRIMARY KEY
+);
+CREATE FUNCTION fn93210() RETURNS INT STABLE LANGUAGE SQL AS 'SELECT a FROM t93210';
+
+# The body of the UDF should have no assignment cast expressions.
+query T
+EXPLAIN (OPT, TYPES)
+SELECT fn93210()
+----
+values
+ ├── columns: fn93210:4(int)
+ ├── cardinality: [1 - 1]
+ ├── stable
+ ├── stats: [rows=1]
+ ├── cost: 0.02
+ ├── key: ()
+ ├── fd: ()-->(4)
+ ├── distribution: test
+ ├── prune: (4)
+ └── tuple [type=tuple{int}]
+      └── udf: fn93210 [type=int]
+           └── body
+                └── limit
+                     ├── columns: a:1(int!null)
+                     ├── cardinality: [0 - 1]
+                     ├── stats: [rows=1]
+                     ├── key: ()
+                     ├── fd: ()-->(1)
+                     ├── distribution: test
+                     ├── scan t93210
+                     │    ├── columns: a:1(int!null)
+                     │    ├── stats: [rows=1000]
+                     │    └── key: (1)
+                     └── const: 1 [type=int]
diff --git a/pkg/sql/opt/optbuilder/scalar.go b/pkg/sql/opt/optbuilder/scalar.go
index 02547e91d1b0..1ca74aefe3f5 100644
--- a/pkg/sql/opt/optbuilder/scalar.go
+++ b/pkg/sql/opt/optbuilder/scalar.go
@@ -694,7 +694,7 @@ func (b *Builder) buildUDF(
 			// its type matches the function return type.
 			returnCol := physProps.Presentation[0].ID
 			returnColMeta := b.factory.Metadata().ColumnMeta(returnCol)
-			if returnColMeta.Type != f.ResolvedType() {
+			if !returnColMeta.Type.Identical(f.ResolvedType()) {
 				if !cast.ValidCast(returnColMeta.Type, f.ResolvedType(), cast.ContextAssignment) {
 					panic(sqlerrors.NewInvalidAssignmentCastError(
 						returnColMeta.Type, f.ResolvedType(), returnColMeta.Alias))
diff --git a/pkg/sql/parser/help_test.go b/pkg/sql/parser/help_test.go
index 7f35ba4cb9ab..51d32d071040 100644
--- a/pkg/sql/parser/help_test.go
+++ b/pkg/sql/parser/help_test.go
@@ -411,6 +411,7 @@ func TestContextualHelp(t *testing.T) {
 		{`SHOW TABLES FROM blah ??`, `SHOW TABLES`},
 
 		{`SHOW TENANT ??`, `SHOW TENANT`},
+		{`SHOW TENANT ?? WITH REPLICATION STATUS`, `SHOW TENANT`},
 
 		{`SHOW TRANSACTION PRIORITY ??`, `SHOW TRANSACTION`},
 		{`SHOW TRANSACTION STATUS ??`, `SHOW TRANSACTION`},
diff --git a/pkg/sql/parser/sql.y b/pkg/sql/parser/sql.y
index 272e1a303fdf..c130733a02f9 100644
--- a/pkg/sql/parser/sql.y
+++ b/pkg/sql/parser/sql.y
@@ -5420,11 +5420,21 @@ backup_kms:
 
 // %Help: SHOW TENANT - display tenant information
 // %Category: Misc
-// %Text: SHOW TENANT
+// %Text: SHOW TENANT <tenant_name> [WITH REPLICATION STATUS]
 show_tenant_stmt:
-  SHOW TENANT name
+  SHOW TENANT d_expr
   {
-   $$.val = &tree.ShowTenant{Name: tree.Name($3)}
+   $$.val = &tree.ShowTenant{
+     Name: $3.expr(),
+     WithReplication: false,
+   }
+  }
+| SHOW TENANT d_expr WITH REPLICATION STATUS
+  {
+   $$.val = &tree.ShowTenant{
+     Name: $3.expr(),
+     WithReplication: true,
+   }
   }
 | SHOW TENANT error // SHOW HELP: SHOW TENANT
 
diff --git a/pkg/sql/parser/testdata/show b/pkg/sql/parser/testdata/show
index 8690b3598f8b..47ac07e0a6fa 100644
--- a/pkg/sql/parser/testdata/show
+++ b/pkg/sql/parser/testdata/show
@@ -1799,6 +1799,14 @@ parse
 SHOW TENANT foo
 ----
 SHOW TENANT foo
-SHOW TENANT foo -- fully parenthesized
+SHOW TENANT (foo) -- fully parenthesized
 SHOW TENANT foo -- literals removed
 SHOW TENANT _ -- identifiers removed
+
+parse
+SHOW TENANT foo WITH REPLICATION STATUS
+----
+SHOW TENANT foo WITH REPLICATION STATUS
+SHOW TENANT (foo) WITH REPLICATION STATUS -- fully parenthesized
+SHOW TENANT foo WITH REPLICATION STATUS -- literals removed
+SHOW TENANT _ WITH REPLICATION STATUS -- identifiers removed
diff --git a/pkg/sql/schemachanger/scplan/internal/scstage/build.go b/pkg/sql/schemachanger/scplan/internal/scstage/build.go
index e22f16f1f020..cea581e905a8 100644
--- a/pkg/sql/schemachanger/scplan/internal/scstage/build.go
+++ b/pkg/sql/schemachanger/scplan/internal/scstage/build.go
@@ -37,15 +37,17 @@ func BuildStages(
 	phase scop.Phase,
 	g *scgraph.Graph,
 	scJobIDSupplier func() jobspb.JobID,
+	enforcePlannerSanityChecks bool,
 ) []Stage {
 	c := buildContext{
-		rollback:               init.InRollback,
-		g:                      g,
-		isRevertibilityIgnored: true,
-		targetState:            init.TargetState,
-		startingStatuses:       init.Current,
-		startingPhase:          phase,
-		descIDs:                screl.AllTargetDescIDs(init.TargetState),
+		rollback:                   init.InRollback,
+		g:                          g,
+		isRevertibilityIgnored:     true,
+		targetState:                init.TargetState,
+		startingStatuses:           init.Current,
+		startingPhase:              phase,
+		descIDs:                    screl.AllTargetDescIDs(init.TargetState),
+		enforcePlannerSanityChecks: enforcePlannerSanityChecks,
 	}
 	// Try building stages while ignoring revertibility constraints.
 	// This is fine as long as there are no post-commit stages.
@@ -92,14 +94,15 @@ func BuildStages(
 // buildContext contains the global constants for building the stages.
 // Only the BuildStages function mutates it, it's read-only everywhere else.
 type buildContext struct {
-	rollback               bool
-	g                      *scgraph.Graph
-	scJobID                jobspb.JobID
-	isRevertibilityIgnored bool
-	targetState            scpb.TargetState
-	startingStatuses       []scpb.Status
-	startingPhase          scop.Phase
-	descIDs                catalog.DescriptorIDSet
+	rollback                   bool
+	g                          *scgraph.Graph
+	scJobID                    jobspb.JobID
+	isRevertibilityIgnored     bool
+	targetState                scpb.TargetState
+	startingStatuses           []scpb.Status
+	startingPhase              scop.Phase
+	descIDs                    catalog.DescriptorIDSet
+	enforcePlannerSanityChecks bool
 }
 
 func buildStages(bc buildContext) (stages []Stage) {
@@ -427,9 +430,12 @@ func (sb stageBuilder) hasUnmeetableOutboundDeps(n *screl.Node) (ret bool) {
 	sb.visited[n] = sb.visitEpoch
 	// Do some sanity checks.
 	if _, isFulfilled := sb.bs.fulfilled[n]; isFulfilled {
-		// This should never happen.
-		panic(errors.AssertionFailedf("%s should not yet be scheduled for this stage",
-			screl.NodeString(n)))
+		if sb.bc.enforcePlannerSanityChecks {
+			panic(errors.AssertionFailedf("%s should not yet be scheduled for this stage",
+				screl.NodeString(n)))
+		} else {
+			return false
+		}
 	}
 	// Look up the current target state for this node, via the lookup table.
 	if t := sb.lut[n.Target]; t == nil {
diff --git a/pkg/sql/schemachanger/scplan/plan.go b/pkg/sql/schemachanger/scplan/plan.go
index c28eee82c8bf..9904d9b4a22f 100644
--- a/pkg/sql/schemachanger/scplan/plan.go
+++ b/pkg/sql/schemachanger/scplan/plan.go
@@ -46,6 +46,10 @@ type Params struct {
 	// SchemaChangerJobIDSupplier is used to return the JobID for a
 	// job if one should exist.
 	SchemaChangerJobIDSupplier func() jobspb.JobID
+
+	// enforcePlannerSanityCheck, if true, strictly enforces sanity checks in the
+	// declarative schema changer planner.
+	EnforcePlannerSanityCheck bool
 }
 
 // Exported internal types
@@ -109,9 +113,7 @@ func makePlan(ctx context.Context, p *Plan) (err error) {
 	}
 	{
 		start := timeutil.Now()
-		p.Stages = scstage.BuildStages(
-			ctx, p.CurrentState, p.Params.ExecutionPhase, p.Graph, p.Params.SchemaChangerJobIDSupplier,
-		)
+		p.Stages = scstage.BuildStages(ctx, p.CurrentState, p.Params.ExecutionPhase, p.Graph, p.Params.SchemaChangerJobIDSupplier, p.Params.EnforcePlannerSanityCheck)
 		if log.ExpensiveLogEnabled(ctx, 2) {
 			log.Infof(ctx, "stage generation took %v", timeutil.Since(start))
 		}
diff --git a/pkg/sql/schemachanger/scrun/BUILD.bazel b/pkg/sql/schemachanger/scrun/BUILD.bazel
index db75cea6687d..a93d5ea12a3a 100644
--- a/pkg/sql/schemachanger/scrun/BUILD.bazel
+++ b/pkg/sql/schemachanger/scrun/BUILD.bazel
@@ -13,6 +13,7 @@ go_library(
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/roachpb",
+        "//pkg/settings",
         "//pkg/settings/cluster",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/descpb",
@@ -21,6 +22,7 @@ go_library(
         "//pkg/sql/schemachanger/scop",
         "//pkg/sql/schemachanger/scpb",
         "//pkg/sql/schemachanger/scplan",
+        "//pkg/util/buildutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
     ],
diff --git a/pkg/sql/schemachanger/scrun/scrun.go b/pkg/sql/schemachanger/scrun/scrun.go
index 602f48fe01fd..4034b8ff76bd 100644
--- a/pkg/sql/schemachanger/scrun/scrun.go
+++ b/pkg/sql/schemachanger/scrun/scrun.go
@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scerrors"
@@ -24,10 +25,18 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scop"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scplan"
+	"github.com/cockroachdb/cockroach/pkg/util/buildutil"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
 )
 
+var enforcePlannerSanityCheck = settings.RegisterBoolSetting(
+	settings.TenantWritable,
+	"sql.schemachanger.strict_planning_sanity_check.enabled",
+	"enforce strict sanity checks in the declarative schema changer planner",
+	buildutil.CrdbTestBuild,
+)
+
 // RunStatementPhase executes in-transaction schema changes for the targeted
 // state. These are the immediate changes which take place at DDL statement
 // execution time (scop.StatementPhase).
@@ -67,6 +76,7 @@ func runTransactionPhase(
 		ActiveVersion:              deps.ClusterSettings().Version.ActiveVersion(ctx),
 		ExecutionPhase:             phase,
 		SchemaChangerJobIDSupplier: deps.TransactionalJobRegistry().SchemaChangerJobID,
+		EnforcePlannerSanityCheck:  enforcePlannerSanityCheck.Get(&deps.ClusterSettings().SV),
 	})
 	if err != nil {
 		return scpb.CurrentState{}, jobspb.InvalidJobID, err
diff --git a/pkg/sql/sem/eval/deps.go b/pkg/sql/sem/eval/deps.go
index 02c78e20882b..c3b8208c69d1 100644
--- a/pkg/sql/sem/eval/deps.go
+++ b/pkg/sql/sem/eval/deps.go
@@ -573,9 +573,6 @@ type TenantOperator interface {
 	// the gc job will not wait for a GC ttl.
 	DestroyTenant(ctx context.Context, tenantName roachpb.TenantName, synchronous bool) error
 
-	// GetTenantInfo returns information about the specified tenant.
-	GetTenantInfo(ctx context.Context, tenantName roachpb.TenantName) (*descpb.TenantInfo, error)
-
 	// GCTenant attempts to garbage collect a DROP tenant from the system. Upon
 	// success it also removes the tenant record.
 	// It returns an error if the tenant does not exist.
diff --git a/pkg/sql/sem/tree/show.go b/pkg/sql/sem/tree/show.go
index adc28680c1d8..89d4bf9e1aae 100644
--- a/pkg/sql/sem/tree/show.go
+++ b/pkg/sql/sem/tree/show.go
@@ -777,13 +777,18 @@ func (node *ShowTableStats) Format(ctx *FmtCtx) {
 
 // ShowTenant represents a SHOW TENANT statement.
 type ShowTenant struct {
-	Name Name
+	Name            Expr
+	WithReplication bool
 }
 
 // Format implements the NodeFormatter interface.
 func (node *ShowTenant) Format(ctx *FmtCtx) {
 	ctx.WriteString("SHOW TENANT ")
-	ctx.FormatNode(&node.Name)
+	ctx.FormatNode(node.Name)
+
+	if node.WithReplication {
+		ctx.WriteString(" WITH REPLICATION STATUS")
+	}
 }
 
 // ShowHistogram represents a SHOW HISTOGRAM statement.
diff --git a/pkg/sql/show_tenant.go b/pkg/sql/show_tenant.go
index 7bf2c529e0e3..b0bf671b92b1 100644
--- a/pkg/sql/show_tenant.go
+++ b/pkg/sql/show_tenant.go
@@ -12,33 +12,120 @@ package sql
 
 import (
 	"context"
+	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/paramparse"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/types"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/cockroachdb/errors"
 )
 
 type showTenantNode struct {
-	columns colinfo.ResultColumns
-	name    roachpb.TenantName
-	info    *descpb.TenantInfo
-	done    bool
+	name               tree.TypedExpr
+	tenantInfo         *descpb.TenantInfo
+	withReplication    bool
+	replicationInfo    *streampb.StreamIngestionStats
+	protectedTimestamp hlc.Timestamp
+	columns            colinfo.ResultColumns
+	done               bool
 }
 
-func (p *planner) ShowTenant(_ context.Context, n *tree.ShowTenant) (planNode, error) {
-	return &showTenantNode{
-		name:    roachpb.TenantName(n.Name),
-		columns: colinfo.TenantColumns,
-	}, nil
+func (p *planner) ShowTenant(ctx context.Context, n *tree.ShowTenant) (planNode, error) {
+	if err := p.RequireAdminRole(ctx, "show tenant"); err != nil {
+		return nil, err
+	}
+
+	if err := rejectIfCantCoordinateMultiTenancy(p.execCfg.Codec, "show"); err != nil {
+		return nil, err
+	}
+
+	var dummyHelper tree.IndexedVarHelper
+	strName := paramparse.UnresolvedNameToStrVal(n.Name)
+	typedName, err := p.analyzeExpr(
+		ctx, strName, nil, dummyHelper, types.String,
+		true, "SHOW TENANT ... WITH REPLICATION STATUS")
+	if err != nil {
+		return nil, err
+	}
+
+	node := &showTenantNode{
+		name:            typedName,
+		withReplication: n.WithReplication,
+	}
+	if n.WithReplication {
+		node.columns = colinfo.TenantColumnsWithReplication
+	} else {
+		node.columns = colinfo.TenantColumns
+	}
+
+	return node, nil
+}
+
+func (n *showTenantNode) getTenantName(params runParams) (roachpb.TenantName, error) {
+	dName, err := eval.Expr(params.ctx, params.p.EvalContext(), n.name)
+	if err != nil {
+		return "", err
+	}
+	name, ok := dName.(*tree.DString)
+	if !ok || name == nil {
+		return "", errors.Newf("expected a string, got %T", dName)
+	}
+	return roachpb.TenantName(*name), nil
 }
 
 func (n *showTenantNode) startExec(params runParams) error {
-	info, err := params.p.GetTenantInfo(params.ctx, n.name)
+	tenantName, err := n.getTenantName(params)
+	if err != nil {
+		return err
+	}
+	tenantRecord, err := GetTenantRecordByName(params.ctx, params.p.execCfg, params.p.Txn(), tenantName)
 	if err != nil {
 		return err
 	}
-	n.info = info
+
+	n.tenantInfo = tenantRecord
+	if n.withReplication {
+		if n.tenantInfo.TenantReplicationJobID == 0 {
+			return errors.Newf("tenant %q does not have an active replication job", tenantName)
+		}
+		mgr, err := params.p.EvalContext().StreamManagerFactory.GetStreamIngestManager(params.ctx)
+		if err != nil {
+			return err
+		}
+		stats, err := mgr.GetStreamIngestionStats(params.ctx, n.tenantInfo.TenantReplicationJobID)
+		if err != nil {
+			// An error means we don't have stats but we can still present some info,
+			// therefore we don't fail here.
+			// TODO(lidor): we need a better signal from GetStreamIngestionStats(), instead of
+			// ignoring all errors.
+			log.Infof(params.ctx, "stream ingestion stats unavailable for tenant %q and job %d",
+				tenantName, n.tenantInfo.TenantReplicationJobID)
+		} else {
+			n.replicationInfo = stats
+			if stats.IngestionDetails.ProtectedTimestampRecordID == nil {
+				// We don't have the protected timestamp record but we still want to show
+				// the info we do have about tenant replication status, logging an error
+				// and continuing.
+				log.Warningf(params.ctx, "protected timestamp unavailable for tenant %q and job %d",
+					tenantName, n.tenantInfo.TenantReplicationJobID)
+			} else {
+				ptp := params.p.execCfg.ProtectedTimestampProvider
+				record, err := ptp.GetRecord(params.ctx, params.p.Txn(), *stats.IngestionDetails.ProtectedTimestampRecordID)
+				if err != nil {
+					return err
+				}
+				n.protectedTimestamp = record.Timestamp
+			}
+		}
+	}
 	return nil
 }
 
@@ -49,11 +136,49 @@ func (n *showTenantNode) Next(_ runParams) (bool, error) {
 	n.done = true
 	return true, nil
 }
+
 func (n *showTenantNode) Values() tree.Datums {
+	tenantId := tree.NewDInt(tree.DInt(n.tenantInfo.ID))
+	tenantName := tree.NewDString(string(n.tenantInfo.Name))
+	tenantStatus := tree.NewDString(n.tenantInfo.State.String())
+	if !n.withReplication {
+		// This is a simple 'SHOW TENANT name'.
+		return tree.Datums{
+			tenantId,
+			tenantName,
+			tenantStatus,
+		}
+	}
+
+	// This is a 'SHOW TENANT name WITH REPLICATION STATUS' command.
+	sourceTenantName := tree.DNull
+	sourceClusterUri := tree.DNull
+	replicationJobId := tree.NewDInt(tree.DInt(n.tenantInfo.TenantReplicationJobID))
+	replicatedTimestamp := tree.DNull
+	retainedTimestamp := tree.DNull
+
+	if n.replicationInfo != nil {
+		sourceTenantName = tree.NewDString(string(n.replicationInfo.IngestionDetails.SourceTenantName))
+		sourceClusterUri = tree.NewDString(n.replicationInfo.IngestionDetails.StreamAddress)
+		if n.replicationInfo.ReplicationLagInfo != nil {
+			minIngested := n.replicationInfo.ReplicationLagInfo.MinIngestedTimestamp.WallTime
+			// The latest fully replicated time.
+			replicatedTimestamp, _ = tree.MakeDTimestamp(timeutil.Unix(0, minIngested), time.Nanosecond)
+		}
+		// The protected timestamp on the destination cluster.
+		retainedTimestamp, _ = tree.MakeDTimestamp(timeutil.Unix(0, n.protectedTimestamp.WallTime), time.Nanosecond)
+	}
+
 	return tree.Datums{
-		tree.NewDInt(tree.DInt(n.info.ID)),
-		tree.NewDString(string(n.info.Name)),
-		tree.NewDString(n.info.State.String()),
+		tenantId,
+		tenantName,
+		tenantStatus,
+		sourceTenantName,
+		sourceClusterUri,
+		replicationJobId,
+		replicatedTimestamp,
+		retainedTimestamp,
 	}
 }
+
 func (n *showTenantNode) Close(_ context.Context) {}
diff --git a/pkg/sql/tenant.go b/pkg/sql/tenant.go
index a7f2286f216b..06f482c5468e 100644
--- a/pkg/sql/tenant.go
+++ b/pkg/sql/tenant.go
@@ -819,18 +819,3 @@ WHERE id = $1`, tenantID, tenantName); err != nil {
 
 	return nil
 }
-
-// GetTenantInfo implements the tree.TenantOperator interface.
-func (p *planner) GetTenantInfo(
-	ctx context.Context, tenantName roachpb.TenantName,
-) (*descpb.TenantInfo, error) {
-	if err := p.RequireAdminRole(ctx, "show tenant"); err != nil {
-		return nil, err
-	}
-
-	if err := rejectIfCantCoordinateMultiTenancy(p.execCfg.Codec, "show"); err != nil {
-		return nil, err
-	}
-
-	return GetTenantRecordByName(ctx, p.execCfg, p.Txn(), tenantName)
-}
diff --git a/pkg/ui/workspaces/db-console/src/views/app/components/feedbackSurveyLink/feedbackSurveyLink.styl b/pkg/ui/workspaces/db-console/src/views/app/components/feedbackSurveyLink/feedbackSurveyLink.styl
deleted file mode 100644
index b4d5b30e66a1..000000000000
--- a/pkg/ui/workspaces/db-console/src/views/app/components/feedbackSurveyLink/feedbackSurveyLink.styl
+++ /dev/null
@@ -1,29 +0,0 @@
-// Copyright 2019 The Cockroach Authors.
-//
-// Use of this software is governed by the Business Source License
-// included in the file licenses/BSL.txt.
-//
-// As of the Change Date specified in that file, in accordance with
-// the Business Source License, use of this software will be governed
-// by the Apache License, Version 2.0, included in the file
-// licenses/APL.txt.
-
-@require '~src/components/core/index.styl'
-
-.feedback-survey-link
-  display flex
-  flex-direction row
-  font-family $font-family--base
-  align-items center
-  border-right 1px solid #C0C6D9
-  margin-right $spacing-smaller
-  cursor pointer
-  &__title
-    margin-right $spacing-x-small
-  .image-container
-    width 32px
-    height 32px
-    display flex
-    justify-content center
-    align-items center
-    border-radius 16px
\ No newline at end of file
diff --git a/pkg/ui/workspaces/db-console/src/views/app/components/feedbackSurveyLink/feedbackSurveyLink.tsx b/pkg/ui/workspaces/db-console/src/views/app/components/feedbackSurveyLink/feedbackSurveyLink.tsx
deleted file mode 100644
index 44696c7e0a78..000000000000
--- a/pkg/ui/workspaces/db-console/src/views/app/components/feedbackSurveyLink/feedbackSurveyLink.tsx
+++ /dev/null
@@ -1,46 +0,0 @@
-// Copyright 2021 The Cockroach Authors.
-//
-// Use of this software is governed by the Business Source License
-// included in the file licenses/BSL.txt.
-//
-// As of the Change Date specified in that file, in accordance with
-// the Business Source License, use of this software will be governed
-// by the Apache License, Version 2.0, included in the file
-// licenses/APL.txt.
-
-import React from "react";
-import { useSelector } from "react-redux";
-
-import externalLinkIcon from "!!raw-loader!assets/external-link.svg";
-import { trustIcon } from "src/util/trust";
-import {
-  singleVersionSelector,
-  clusterIdSelector,
-} from "../../../../redux/nodes";
-import "./feedbackSurveyLink.styl";
-
-const FeedBackSurveyLink = () => {
-  const singleVersion = useSelector(singleVersionSelector);
-  const clusterId = useSelector(clusterIdSelector);
-  const feedbackLink = new URL("https://www.cockroachlabs.com/survey/");
-  feedbackLink.searchParams.append("clusterId", clusterId);
-  feedbackLink.searchParams.append("clusterVersion", singleVersion);
-  if (!clusterId || !singleVersion) {
-    return <></>;
-  }
-  return (
-    <div
-      className="feedback-survey-link"
-      onClick={() => window.open(feedbackLink.toString())}
-    >
-      <div
-        className="image-container"
-        title="Share Feedback"
-        dangerouslySetInnerHTML={trustIcon(externalLinkIcon)}
-      />
-      <div className="feedback-survey-link__title">Share feedback</div>
-    </div>
-  );
-};
-
-export default FeedBackSurveyLink;
diff --git a/pkg/ui/workspaces/db-console/src/views/app/containers/layout/index.tsx b/pkg/ui/workspaces/db-console/src/views/app/containers/layout/index.tsx
index 105068622728..6976d9112756 100644
--- a/pkg/ui/workspaces/db-console/src/views/app/containers/layout/index.tsx
+++ b/pkg/ui/workspaces/db-console/src/views/app/containers/layout/index.tsx
@@ -25,7 +25,6 @@ import {
 } from "src/redux/nodes";
 import { AdminUIState } from "src/redux/state";
 import LoginIndicator from "src/views/app/components/loginIndicator";
-import FeedbackSurveyLink from "src/views/app/components/feedbackSurveyLink/feedbackSurveyLink";
 import {
   GlobalNavigation,
   CockroachLabsLockupIcon,
@@ -85,7 +84,6 @@ class Layout extends React.Component<LayoutProps & RouteComponentProps> {
                 <CockroachLabsLockupIcon height={26} />
               </Left>
               <Right>
-                <FeedbackSurveyLink />
                 <LoginIndicator />
               </Right>
             </GlobalNavigation>