You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: flush.go
    path of file: ./repos/cockroach/pkg/sql/sqlstats/persistedsqlstats
    
    the code of the file until where you have to start completion: // 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.

package persistedsqlstats

import (
	"context"
	"fmt"
	"math/rand"
	"time"

	"github.com/cockroachdb/cockroach/pkg/sql/appstatspb"
	"github.com/cockroachdb/cockroach/pkg/sql/catalog/systemschema"
	"github.com/cockroachdb/cockroach/pkg/sql/isql"
	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil"
	"github.com/cockroachdb/cockroach/pkg/sql/types"
	"github.com/cockroachdb/cockroach/pkg/util/log"
	"github.com/cockroachdb/cockroach/pkg/util/stop"
	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
	"github.com/cockroachdb/errors"
)

// Flush flushes in-memory sql stats into a system table. Any errors encountered
// during the flush will be logged as warning.
func (s *PersistedSQLStats) Flush(ctx context.Context, stopper *stop.Stopper) {
	now := s.getTimeNow()

	allowDiscardWhenDisabled := DiscardInMemoryStatsWhenFlushDisabled.Get(&s.cfg.Settings.SV)
	minimumFlushInterval := MinimumInterval.Get(&s.cfg.Settings.SV)

	enabled := SQLStatsFlushEnabled.Get(&s.cfg.Settings.SV)
	flushingTooSoon := now.Before(s.lastFlushStarted.Add(minimumFlushInterval))

	// Reset stats is performed individually for statement and transaction stats
	// within SQLStats.ConsumeStats function. Here, we reset stats only when
	// sql stats flush is disabled.
	if !enabled && allowDiscardWhenDisabled {
		defer func() {
			if err := s.SQLStats.Reset(ctx); err != nil {
				log.Warningf(ctx, "fail to reset in-memory SQL Stats: %s", err)
			}
		}()
	}

	// Handle early abortion of the flush.
	if !enabled {
		return
	}

	if flushingTooSoon {
		log.Infof(ctx, "flush aborted due to high flush frequency. "+
			"The minimum interval between flushes is %s", minimumFlushInterval.String())
		return
	}

	log.Infof(ctx, "flushing %d stmt/txn fingerprints (%d bytes) after %s",
		s.SQLStats.GetTotalFingerprintCount(), s.SQLStats.GetTotalFingerprintBytes(), timeutil.Since(s.lastFlushStarted
    
func (s *PersistedSQLStats) Flush(ctx context.Context, stopper *stop.Stopper) {
	now := s.getTimeNow()

	allowDiscardWhenDisabled := DiscardInMemoryStatsWhenFlushDisabled.Get(&s.cfg.Settings.SV)
	minimumFlushInterval := MinimumInterval.Get(&s.cfg.Settings.SV)

	enabled := SQLStatsFlushEnabled.Get(&s.cfg.Settings.SV)
	flushingTooSoon := now.Before(s.lastFlushStarted.Add(minimumFlushInterval))

	// Reset stats is performed individually for statement and transaction stats
	// within SQLStats.ConsumeStats function. Here, we reset stats only when
	// sql stats flush is disabled.
	if !enabled && allowDiscardWhenDisabled {
		defer func() {
			if err := s.SQLStats.Reset(ctx); err != nil {
				log.Warningf(ctx, "fail to reset in-memory SQL Stats: %s", err)
			}
		}()
	}

	// Handle early abortion of the flush.
	if !enabled {
		return
	}

	if flushingTooSoon {
		log.Infof(ctx, "flush aborted due to high flush frequency. "+
			"The minimum interval between flushes is %s", minimumFlushInterval.String())
		return
	}

	log.Infof(ctx, "flushing %d stmt/txn fingerprints (%d bytes) after %s",
		s.SQLStats.GetTotalFingerprintCount(), s.SQLStats.GetTotalFingerprintBytes(), timeutil.Since(s.lastFlushStarted))
	s.lastFlushStarted = now

	aggregatedTs := s.ComputeAggregatedTs()

	// We only check the statement count as there should always be at least as many statements as transactions.
	limitReached := false

	var err error
	if sqlStatsLimitTableSizeEnabled.Get(&s.SQLStats.GetClusterSettings().SV) {
		limitReached, err = s.StmtsLimitSizeReached(ctx)
	}

	if err != nil {
		log.Errorf(ctx, "encountered an error at flush, checking for statement statistics size limit: %v", err)
	}
	if limitReached {
		log.Infof(ctx, "unable to flush fingerprints because table limit was reached.")
	} else {
		s.SQLStats.ConsumeStats(ctx, stopper,
			func(ctx context.Context, statistics *appstatspb.CollectedStatementStatistics) error {
				s.doFlush(ctx, func() error {
					return s.doFlushSingleStmtStats(ctx, statistics, aggregatedTs)
				}, "failed to flush statement statistics" /* errMsg */)

				return nil
			},
			func(ctx context.Context, statistics *appstatspb.CollectedTransactionStatistics) error {
				s.doFlush(ctx, func() error {
					return s.doFlushSingleTxnStats(ctx, statistics, aggregatedTs)
				}, "failed to flush transaction statistics" /* errMsg */)

				return nil
			})

		if s.cfg.Knobs != nil && s.cfg.Knobs.OnStmtStatsFlushFinished != nil {
			s.cfg.Knobs.OnStmtStatsFlushFinished()
		}

		if s.cfg.Knobs != nil && s.cfg.Knobs.OnTxnStatsFlushFinished != nil {
			s.cfg.Knobs.OnTxnStatsFlushFinished()
		}
	}
}