database_zone_config.go

// 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 scbuildstmt

import (
	"github.com/cockroachdb/cockroach/pkg/config/zonepb"
	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scpb"
	"github.com/cockroachdb/cockroach/pkg/sql/sem/catid"
	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
	"github.com/cockroachdb/cockroach/pkg/sql/sqlerrors"
	"github.com/cockroachdb/cockroach/pkg/sql/sqltelemetry"
)

// databaseZoneConfigObj is used to represent a table-specific zone configuration
// object.
type databaseZoneConfigObj struct {
	databaseID catid.DescID
	zoneConfig *zonepb.ZoneConfig
	seqNum     uint32
}

var _ zoneConfigObject = &databaseZoneConfigObj{}

func (dzo *databaseZoneConfigObj) addZoneConfigToBuildCtx(b BuildCtx) scpb.Element {
	dzo.seqNum += 1
	elem := &scpb.DatabaseZoneConfig{
		DatabaseID: dzo.databaseID,
		ZoneConfig: dzo.zoneConfig,
		SeqNum:     dzo.seqNum,
	}
	b.Add(elem)
	return elem
}

func (dzo *databaseZoneConfigObj) checkPrivilegeForSetZoneConfig(
	b BuildCtx, zs tree.ZoneSpecifier,
) error {
	// Can configure zone of a database if user has either CREATE or ZONECONFIG
	// privilege on the database.
	dbElem := b.ResolveDatabase(zs.Database, ResolveParams{}).FilterDatabase().MustGetOneElement()
	dbCreatePrivilegeErr := b.CheckPrivilege(dbElem, privilege.CREATE)
	dbZoneConfigPrivilegeErr := b.CheckPrivilege(dbElem, privilege.ZONECONFIG)
	if dbZoneConfigPrivilegeErr == nil || dbCreatePrivilegeErr == nil {
		return nil
	}

	// For the system database, the user must be an admin. Otherwise, we
	// require CREATE or ZONECONFIG privilege on the database in question.
	reqNonAdminPrivs := []privilege.Kind{privilege.ZONECONFIG, privilege.CREATE}
	if zs.Database == "system" {
		return b.CheckGlobalPrivilege(privilege.REPAIRCLUSTER)
	}
	return sqlerrors.NewInsufficientPrivilegeOnDescriptorError(b.CurrentUser(),
		reqNonAdminPrivs, string(catalog.Database),
		mustRetrieveNamespaceElem(b, dbElem.DatabaseID).Name)
}

func (dzo *databaseZoneConfigObj) checkZoneConfigChangePermittedForMultiRegion(
	b BuildCtx, zs tree.ZoneSpecifier, options tree.KVOptions,
) error {
	// If the user has specified that they're overriding, then the world is
	// their oyster.
	if b.SessionData().OverrideMultiRegionZoneConfigEnabled {
		// Note that we increment the telemetry counter unconditionally here.
		// It's possible that this will lead to over-counting as the user may
		// have left the override on and is now updating a zone configuration
		// that is not protected by the multi-region abstractions. To get finer
		// grained counting, however, would be more difficult to code, and may
		// not even prove to be that valuable, so we have decided to live with
		// the potential for over-counting.
		telemetry.Inc(sqltelemetry.OverrideMultiRegionZoneConfigurationUser)
		return nil
	}

	// Check if what we're altering is a multi-region entity.
	dbRegionConfigElem := b.ResolveDatabase(zs.Database,
		ResolveParams{}).FilterDatabaseRegionConfig().MustGetZeroOrOneElement()
	if dbRegionConfigElem == nil {
		// Not a multi-region database, we're done here.
		return nil
	}

	return maybeMultiregionErrorWithHint(options)
}

func (dzo *databaseZoneConfigObj) getTargetID() catid.DescID {
	return dzo.databaseID
}

func (dzo *databaseZoneConfigObj) getSeqNum() uint32 {
	return dzo.seqNum
}

func (dzo *databaseZoneConfigObj) retrievePartialZoneConfig(b BuildCtx) *zonepb.ZoneConfig {
	sameDB := func(e *scpb.DatabaseZoneConfig) bool {
		return e.DatabaseID == dzo.getTargetID()
	}
	mostRecentElem := findMostRecentZoneConfig(dzo, func(id catid.DescID) *scpb.ElementCollection[*scpb.DatabaseZoneConfig] {
		return b.QueryByID(id).FilterDatabaseZoneConfig()
	}, sameDB)

	if mostRecentElem != nil {
		dzo.zoneConfig = mostRecentElem.ZoneConfig
		dzo.seqNum = mostRecentElem.SeqNum
	}

	return dzo.zoneConfig
}

func (dzo *databaseZoneConfigObj) retrieveCompleteZoneConfig(
	b BuildCtx, getInheritedDefault bool,
) (*zonepb.ZoneConfig, *zonepb.Subzone, error) {
	var err error
	zc := &zonepb.ZoneConfig{}
	if getInheritedDefault {
		zc, err = dzo.getInheritedDefaultZoneConfig(b)
	} else {
		zc, _, err = dzo.getZoneConfig(b, false /* inheritDefaultRange */)
	}
	if err != nil {
		return nil, nil, err
	}

	completeZc := *zc
	if err = dzo.completeZoneConfig(b, &completeZc); err != nil {
		return nil, nil, err
	}

	return zc, nil, nil
}

func (dzo *databaseZoneConfigObj) completeZoneConfig(b BuildCtx, zone *zonepb.ZoneConfig) error {
	if zone.IsComplete() {
		return nil
	}
	// Check if zone is complete. If not, inherit from the default zone config
	if zone.IsComplete() {
		return nil
	}
	defaultZone, _, err := dzo.getZoneConfig(b, true /* inheritDefaultRange */)
	if err != nil {
		return err
	}
	zone.InheritFromParent(defaultZone)
	return nil
}

func (dzo *databaseZoneConfigObj) setZoneConfigToWrite(zone *zonepb.ZoneConfig) {
	dzo.zoneConfig = zone
}

func (dzo *databaseZoneConfigObj) getInheritedDefaultZoneConfig(
	b BuildCtx,
) (*zonepb.ZoneConfig, error) {
	// Get the zone config of the DEFAULT RANGE.
	zc, _, err := dzo.getZoneConfig(b, true /* inheritDefaultRange */)
	return zc, err
}

func (dzo *databaseZoneConfigObj) getZoneConfig(
	b BuildCtx, inheritDefaultRange bool,
) (*zonepb.ZoneConfig, *zonepb.ZoneConfig, error) {
	var subzones []zonepb.Subzone
	zc, subzones, err := lookUpSystemZonesTable(b, dzo, inheritDefaultRange)
	if err != nil {
		return nil, nil, err
	}

	// If the zone config exists, we know that it is not a subzone placeholder.
	if zc != nil {
		return zc, nil, err
	}

	zc = zonepb.NewZoneConfig()
	zc.Subzones = subzones
	subzone := zc

	// No zone config for this ID. Retrieve the default zone config, but only as
	// long as that wasn't the ID we were trying to retrieve
	// (to avoid infinite recursion).
	if !inheritDefaultRange {
		zc, _, err := dzo.getZoneConfig(b, true /* inheritDefaultRange */)
		if err != nil {
			return nil, nil, err
		}
		return zc, subzone, nil
	}

	// `targetID == keys.RootNamespaceID` but that zc config is not found
	// in `system.zones` table. Return a special, recognizable error!
	return nil, nil, sqlerrors.ErrNoZoneConfigApplies
}

func (dzo *databaseZoneConfigObj) applyZoneConfig(
	b BuildCtx,
	n *tree.SetZoneConfig,
	copyFromParentList []tree.Name,
	setters []func(c *zonepb.ZoneConfig),
) error {
	partialZone, err := prepareZoneConfig(b, n, copyFromParentList, setters, dzo)
	dzo.setZoneConfigToWrite(partialZone)
	return err
}