Merge #120805 #122674

120805: opt/memo: improve zigzag join cost and selectivity estimation with multi-column stats r=rytaft a=rytaft

**opt: update seek and distribution cost of zigzag join to match scan**

Prior to this commit, the optimizer could prefer a zigzag join over a
scan even if they produced the same number of rows. This was because scans
always included the cost of at least one seek (involving random I/O) and
some distribution cost, while zigzag joins did not. This commit updates
the cost of zigzag joins to include seek and distribution costs so they
will never be chosen over scans unless they produce fewer rows.

This change is behind the setting `optimizer_use_improved_zigzag_join_costing`.

Release note (performance improvement): Added a new setting
`optimizer_use_improved_zigzag_join_costing`. When enabled, the cost of zigzag
joins is updated so they will be never be chosen over scans unless they
produce fewer rows. This change only matters if the setting `enable_zigzag_join`
is also true.

**opt/memo: improve selectivity estimation with multi-column stats**

This commit updates `correlationFromMultiColDistinctCounts` in `statisticsBuilder`
to use a tighter lower bound for the multi-column selectivity. This avoids
cases where we significantly over-estimate the selectivity of a multi-column
predicate.

Fixes #121397

Release note (performance improvement): Improved the selectivity estimation of
multi-column filters when the multi-column distinct count is high. This avoids
cases where we significantly over-estimate the selectivity of a multi-column
predicate and as a result can prevent the optimizer from choosing a bad query
plan.

**sql: add setting optimizer_use_improved_multi_column_selectivity_estimate**

Informs #121397

Release note (sql change): Added a setting
`optimizer_use_improved_multi_column_selectivity_estimate`, which if enabled,
causes the optimizer to use an improved selectivity estimate for multi-column
predicates. This setting will default to true on versions 24.2+, and false
on prior versions.

**opt: improve variable names in selectivityFromMultiColDistinctCounts**

This commit improves the variable names in
`selectivityFromMultiColDistinctCounts` in `statisticsBuilder` to be more
self-documenting.

Release note: None

122674: jobs: avoid errors related to fraction completed r=dt a=stevendanna

See individual commits for details.



Co-authored-by: Rebecca Taft <[email protected]>
Co-authored-by: Steven Danna <[email protected]>

pkg/ccl/backupccl/backup_job.go

@@ -250,7 +250,7 @@ func backup(
 		numTotalSpans += len(spec.IntroducedSpans) + len(spec.Spans)
 	}
 
-	progressLogger := jobs.NewChunkProgressLogger(job, numTotalSpans, job.FractionCompleted(), jobs.ProgressUpdateOnly)
+	progressLogger := jobs.NewChunkProgressLoggerForJob(job, numTotalSpans, job.FractionCompleted(), jobs.ProgressUpdateOnly)
 
 	requestFinishedCh := make(chan struct{}, numTotalSpans) // enough buffer to never block
 	var jobProgressLoop func(ctx context.Context) error

pkg/ccl/backupccl/restore_job.go

@@ -410,7 +410,7 @@ func restore(
 		// Only update the job progress on the main data bundle. This should account
 		// for the bulk of the data to restore. Other data (e.g. zone configs in
 		// cluster restores) may be restored first.
-		progressLogger := jobs.NewChunkProgressLogger(job, numImportSpans, job.FractionCompleted(), progressTracker.updateJobCallback)
+		progressLogger := jobs.NewChunkProgressLoggerForJob(job, numImportSpans, job.FractionCompleted(), progressTracker.updateJobCallback)
 
 		jobProgressLoop := func(ctx context.Context) error {
 			ctx, progressSpan := tracing.ChildSpan(ctx, "progress-loop")

pkg/ccl/backupccl/testdata/backup-restore/in-progress-restores

@@ -155,7 +155,10 @@ SELECT * FROM b2.me.baz;
 
 
 query-sql
-SELECT table_name,index_name,non_unique,seq_in_index,column_name FROM [SHOW INDEX FROM b2.me.baz] WHERE index_name = 'greeting_idx';
+SELECT table_name,index_name,non_unique,seq_in_index,column_name
+FROM [SHOW INDEX FROM b2.me.baz]
+WHERE index_name = 'greeting_idx'
+ORDER BY seq_in_index;
 ----
 baz greeting_idx true 1 y
 baz greeting_idx true 2 rowid

pkg/jobs/BUILD.bazel

@@ -32,6 +32,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/base",
+        "//pkg/build",
         "//pkg/clusterversion",
         "//pkg/jobs/jobspb",
         "//pkg/jobs/jobsprofiler/profilerconstants",
@@ -102,6 +103,7 @@ go_test(
         "jobs_test.go",
         "lease_test.go",
         "main_test.go",
+        "progress_test.go",
         "registry_external_test.go",
         "registry_test.go",
         "scheduled_job_executor_test.go",

pkg/jobs/jobs.go

@@ -19,6 +19,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/build"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
@@ -32,6 +33,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/json"
+	"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"
@@ -328,16 +330,30 @@ func (u Updater) FractionProgressed(ctx context.Context, progressedFn FractionPr
 			return err
 		}
 		fractionCompleted := progressedFn(ctx, md.Progress.Details)
-		// allow for slight floating-point rounding inaccuracies
-		if fractionCompleted > 1.0 && fractionCompleted < 1.01 {
-			fractionCompleted = 1.0
+
+		if !build.IsRelease() {
+			// We allow for slight floating-point rounding
+			// inaccuracies. We only want to error in non-release
+			// builds because in large production installations the
+			// method at least one job uses to calculate process can
+			// result in substantial floating point inaccuracy.
+			if fractionCompleted < 0.0 || fractionCompleted > 1.01 {
+				return errors.Errorf(
+					"fraction completed %f is outside allowable range [0.0, 1.01]",
+					fractionCompleted,
+				)
+			}
 		}
-		if fractionCompleted < 0.0 || fractionCompleted > 1.0 {
-			return errors.Errorf(
-				"job %d: fractionCompleted %f is outside allowable range [0.0, 1.0]",
-				u.j.ID(), fractionCompleted,
-			)
+
+		// Clamp to [0.0, 1.0].
+		if fractionCompleted > 1.0 {
+			log.VInfof(ctx, 1, "clamping fraction completed %f to [0.0, 1.0]", fractionCompleted)
+			fractionCompleted = 1.0
+		} else if fractionCompleted < 0.0 {
+			log.VInfof(ctx, 1, "clamping fraction completed %f to [0.0, 1.0]", fractionCompleted)
+			fractionCompleted = 0
 		}
+
 		md.Progress.Progress = &jobspb.Progress_FractionCompleted{
 			FractionCompleted: fractionCompleted,
 		}

pkg/jobs/progress.go

@@ -52,9 +52,8 @@ func TestingSetProgressThresholds() func() {
 // It then updates the actual job periodically using a ProgressUpdateBatcher.
 type ChunkProgressLogger struct {
 	// These fields must be externally initialized.
-	expectedChunks       int
-	completedChunks      int
-	perChunkContribution float32
+	expectedChunks  int
+	completedChunks int
 
 	batcher ProgressUpdateBatcher
 }
@@ -63,27 +62,34 @@ type ChunkProgressLogger struct {
 // progress fraction (ie. when a custom func with side-effects is not needed).
 var ProgressUpdateOnly func(context.Context, jobspb.ProgressDetails)
 
-// NewChunkProgressLogger returns a ChunkProgressLogger.
-func NewChunkProgressLogger(
+func NewChunkProgressLoggerForJob(
 	j *Job,
 	expectedChunks int,
 	startFraction float32,
 	progressedFn func(context.Context, jobspb.ProgressDetails),
+) *ChunkProgressLogger {
+	return NewChunkProgressLogger(
+		func(ctx context.Context, pct float32) error {
+			return j.NoTxn().FractionProgressed(ctx, func(ctx context.Context, details jobspb.ProgressDetails) float32 {
+				if progressedFn != nil {
+					progressedFn(ctx, details)
+				}
+				return pct
+			})
+		}, expectedChunks, startFraction)
+}
+
+// NewChunkProgressLogger returns a ChunkProgressLogger.
+func NewChunkProgressLogger(
+	report func(ctx context.Context, pct float32) error, expectedChunks int, startFraction float32,
 ) *ChunkProgressLogger {
 	return &ChunkProgressLogger{
-		expectedChunks:       expectedChunks,
-		perChunkContribution: (1.0 - startFraction) * 1.0 / float32(expectedChunks),
+		expectedChunks: expectedChunks,
 		batcher: ProgressUpdateBatcher{
-			completed: startFraction,
-			reported:  startFraction,
-			Report: func(ctx context.Context, pct float32) error {
-				return j.NoTxn().FractionProgressed(ctx, func(ctx context.Context, details jobspb.ProgressDetails) float32 {
-					if progressedFn != nil {
-						progressedFn(ctx, details)
-					}
-					return pct
-				})
-			},
+			perChunkContribution: (1.0 - startFraction) * 1.0 / float32(expectedChunks),
+			start:                startFraction,
+			reported:             startFraction,
+			Report:               report,
 		},
 	}
 }
@@ -93,7 +99,7 @@ func NewChunkProgressLogger(
 // system.jobs.
 func (jpl *ChunkProgressLogger) chunkFinished(ctx context.Context) error {
 	jpl.completedChunks++
-	return jpl.batcher.Add(ctx, jpl.perChunkContribution)
+	return jpl.batcher.Add(ctx)
 }
 
 // Loop calls chunkFinished for every message received over chunkCh. It exits
@@ -127,31 +133,37 @@ type ProgressUpdateBatcher struct {
 	// Report is the function called to record progress
 	Report func(context.Context, float32) error
 
+	// The following are set at initialization time.
+	// start is the starting percentage complete.
+	start                float32
+	perChunkContribution float32
+
 	syncutil.Mutex
-	// completed is the fraction of a proc's work completed
-	completed float32
+
 	// reported is the most recently reported value of completed
-	reported float32
+	reported  float32
+	completed int
 	// lastReported is when we last called report
 	lastReported time.Time
 }
 
 // Add records some additional progress made and checks there has been enough
 // change in the completed progress (and enough time has passed) to report the
 // new progress amount.
-func (p *ProgressUpdateBatcher) Add(ctx context.Context, delta float32) error {
+func (p *ProgressUpdateBatcher) Add(ctx context.Context) error {
 	shouldReport, completed := func() (bool, float32) {
 		p.Lock()
 		defer p.Unlock()
-		p.completed += delta
-		c := p.completed
-		sReport := p.completed-p.reported > progressFractionThreshold
-		sReport = sReport || (p.completed > p.reported && p.lastReported.Add(progressTimeThreshold).Before(timeutil.Now()))
+		p.completed += 1
+
+		next := p.start + (float32(p.completed) * p.perChunkContribution)
+		sReport := next-p.reported > progressFractionThreshold
+		sReport = sReport || (next > p.reported && p.lastReported.Add(progressTimeThreshold).Before(timeutil.Now()))
 		if sReport {
-			p.reported = p.completed
+			p.reported = next
 			p.lastReported = timeutil.Now()
 		}
-		return sReport, c
+		return sReport, next
 	}()
 	if shouldReport {
 		return p.Report(ctx, completed)
@@ -163,11 +175,11 @@ func (p *ProgressUpdateBatcher) Add(ctx context.Context, delta float32) error {
 // worrying about update frequency now that it is done.
 func (p *ProgressUpdateBatcher) Done(ctx context.Context) error {
 	p.Lock()
-	completed := p.completed
-	shouldReport := completed-p.reported > progressFractionThreshold
+	next := p.start + (float32(p.completed) * p.perChunkContribution)
+	shouldReport := next-p.reported > progressFractionThreshold
 	p.Unlock()
 	if shouldReport {
-		return p.Report(ctx, completed)
+		return p.Report(ctx, next)
 	}
 	return nil
 }

pkg/jobs/progress_test.go

@@ -0,0 +1,41 @@
+// Copyright 2024 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.
+
+package jobs
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/stretchr/testify/require"
+)
+
+func TestChunkProgressLoggerLimitsFloatingPointError(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+
+	defer TestingSetProgressThresholds()()
+
+	rangeCount := 1240725
+
+	var lastReported float32
+	l := NewChunkProgressLogger(func(_ context.Context, pct float32) error {
+		require.Less(t, pct, float32(1.01))
+		lastReported = pct
+		return nil
+	}, rangeCount, 0)
+	for i := 0; i < rangeCount; i++ {
+		require.NoError(t, l.chunkFinished(ctx), "failed at update %d", i)
+	}
+	require.Greater(t, lastReported, float32(0.99))
+}

pkg/sql/exec_util.go

@@ -3761,6 +3761,14 @@ func (m *sessionDataMutator) SetOptimizerUseImprovedTrigramSimilaritySelectivity
 	m.data.OptimizerUseImprovedTrigramSimilaritySelectivity = val
 }
 
+func (m *sessionDataMutator) SetOptimizerUseImprovedZigzagJoinCosting(val bool) {
+	m.data.OptimizerUseImprovedZigzagJoinCosting = val
+}
+
+func (m *sessionDataMutator) SetOptimizerUseImprovedMultiColumnSelectivityEstimate(val bool) {
+	m.data.OptimizerUseImprovedMultiColumnSelectivityEstimate = val
+}
+
 // Utility functions related to scrubbing sensitive information on SQL Stats.
 
 // quantizeCounts ensures that the Count field in the

pkg/sql/logictest/testdata/logic_test/distsql_stats

@@ -2814,7 +2814,7 @@ vectorized: true
 • render
 │
 └── • filter
-    │ estimated row count: 1
+    │ estimated row count: 0
     │ filter: n = 1
     │
     └── • scan
@@ -2831,7 +2831,7 @@ vectorized: true
 • render
 │
 └── • filter
-    │ estimated row count: 1
+    │ estimated row count: 0
     │ filter: sqrt(m::FLOAT8)::INT8 = 11
     │
     └── • scan

pkg/sql/logictest/testdata/logic_test/information_schema

@@ -6165,8 +6165,10 @@ optimizer_use_improved_computed_column_filters_derivation  on
 optimizer_use_improved_disjunction_stats                   on
 optimizer_use_improved_distinct_on_limit_hint_costing      on
 optimizer_use_improved_join_elimination                    on
+optimizer_use_improved_multi_column_selectivity_estimate   on
 optimizer_use_improved_split_disjunction_for_joins         on
 optimizer_use_improved_trigram_similarity_selectivity      on
+optimizer_use_improved_zigzag_join_costing                 on
 optimizer_use_limit_ordering_for_streaming_group_by        on
 optimizer_use_lock_op_for_serializable                     off
 optimizer_use_multicol_stats                               on

pkg/sql/logictest/testdata/logic_test/pg_catalog

@@ -2894,8 +2894,10 @@ optimizer_use_improved_computed_column_filters_derivation  on                  N
 optimizer_use_improved_disjunction_stats                   on                  NULL      NULL        NULL        string
 optimizer_use_improved_distinct_on_limit_hint_costing      on                  NULL      NULL        NULL        string
 optimizer_use_improved_join_elimination                    on                  NULL      NULL        NULL        string
+optimizer_use_improved_multi_column_selectivity_estimate   on                  NULL      NULL        NULL        string
 optimizer_use_improved_split_disjunction_for_joins         on                  NULL      NULL        NULL        string
 optimizer_use_improved_trigram_similarity_selectivity      on                  NULL      NULL        NULL        string
+optimizer_use_improved_zigzag_join_costing                 on                  NULL      NULL        NULL        string
 optimizer_use_limit_ordering_for_streaming_group_by        on                  NULL      NULL        NULL        string
 optimizer_use_lock_op_for_serializable                     off                 NULL      NULL        NULL        string
 optimizer_use_multicol_stats                               on                  NULL      NULL        NULL        string
@@ -3075,8 +3077,10 @@ optimizer_use_improved_computed_column_filters_derivation  on                  N
 optimizer_use_improved_disjunction_stats                   on                  NULL  user     NULL      on                  on
 optimizer_use_improved_distinct_on_limit_hint_costing      on                  NULL  user     NULL      on                  on
 optimizer_use_improved_join_elimination                    on                  NULL  user     NULL      on                  on
+optimizer_use_improved_multi_column_selectivity_estimate   on                  NULL  user     NULL      on                  on
 optimizer_use_improved_split_disjunction_for_joins         on                  NULL  user     NULL      on                  on
 optimizer_use_improved_trigram_similarity_selectivity      on                  NULL  user     NULL      on                  on
+optimizer_use_improved_zigzag_join_costing                 on                  NULL  user     NULL      on                  on
 optimizer_use_limit_ordering_for_streaming_group_by        on                  NULL  user     NULL      on                  on
 optimizer_use_lock_op_for_serializable                     off                 NULL  user     NULL      off                 off
 optimizer_use_multicol_stats                               on                  NULL  user     NULL      on                  on
@@ -3255,8 +3259,10 @@ optimizer_use_improved_computed_column_filters_derivation  NULL    NULL     NULL
 optimizer_use_improved_disjunction_stats                   NULL    NULL     NULL     NULL        NULL
 optimizer_use_improved_distinct_on_limit_hint_costing      NULL    NULL     NULL     NULL        NULL
 optimizer_use_improved_join_elimination                    NULL    NULL     NULL     NULL        NULL
+optimizer_use_improved_multi_column_selectivity_estimate   NULL    NULL     NULL     NULL        NULL
 optimizer_use_improved_split_disjunction_for_joins         NULL    NULL     NULL     NULL        NULL
 optimizer_use_improved_trigram_similarity_selectivity      NULL    NULL     NULL     NULL        NULL
+optimizer_use_improved_zigzag_join_costing                 NULL    NULL     NULL     NULL        NULL
 optimizer_use_limit_ordering_for_streaming_group_by        NULL    NULL     NULL     NULL        NULL
 optimizer_use_lock_op_for_serializable                     NULL    NULL     NULL     NULL        NULL
 optimizer_use_multicol_stats                               NULL    NULL     NULL     NULL        NULL

pkg/sql/logictest/testdata/logic_test/show_source

@@ -131,8 +131,10 @@ optimizer_use_improved_computed_column_filters_derivation  on
 optimizer_use_improved_disjunction_stats                   on
 optimizer_use_improved_distinct_on_limit_hint_costing      on
 optimizer_use_improved_join_elimination                    on
+optimizer_use_improved_multi_column_selectivity_estimate   on
 optimizer_use_improved_split_disjunction_for_joins         on
 optimizer_use_improved_trigram_similarity_selectivity      on
+optimizer_use_improved_zigzag_join_costing                 on
 optimizer_use_limit_ordering_for_streaming_group_by        on
 optimizer_use_lock_op_for_serializable                     off
 optimizer_use_multicol_stats                               on

pkg/sql/opt/bench/bench_test.go

@@ -734,6 +734,8 @@ func newHarness(tb testing.TB, query benchQuery, schemas []string) *harness {
 	h.evalCtx.SessionData().OptimizerUseTrigramSimilarityOptimization = true
 	h.evalCtx.SessionData().OptimizerUseImprovedDistinctOnLimitHintCosting = true
 	h.evalCtx.SessionData().OptimizerUseImprovedTrigramSimilaritySelectivity = true
+	h.evalCtx.SessionData().OptimizerUseImprovedZigzagJoinCosting = true
+	h.evalCtx.SessionData().OptimizerUseImprovedMultiColumnSelectivityEstimate = true
 
 	// Set up the test catalog.
 	h.testCat = testcat.New()

pkg/sql/opt/exec/execbuilder/testdata/forecast

@@ -2355,7 +2355,7 @@ vectorized: true
 └── • render
     │
     └── • filter
-        │ estimated row count: 9
+        │ estimated row count: 1
         │ filter: t IS NULL
         │
         └── • index join