zonepb: make subzone DiffWithZone more accurate

* Subzones may be defined in a different order. We did not take this
  into account which can cause bugs when e.g. ADD REGION adds a subzone
  in the end rather than in the old "expected" location in the subzones
  array. This has been fixed by comparing subzones using an unordered
  map.
* The ApplyZoneConfig we previously did overwrote subzone fields on the
  original subzone array element, meaning that if there was a mismatch
  it would not be reported through validation. This is now fixed by
  applying the expected zone config to *zonepb.NewZoneConfig() instead.
* Added logic to only check for zone config matches subzones from
  active subzone IDs.
* Improve the error messaging when a subzone config is mismatching -
  namely, add index and partitioning information and differentiate
  between missing fields and missing / extraneous zone configs

Release note (bug fix): Fixed validation bugs during ALTER TABLE ... SET
LOCALITY / crdb_internal.validate_multi_region_zone_config where
validation errors could occur when the database of a REGIONAL BY ROW
table has a new region added. Also fix a validation bug partition zone
mismatches configs were not caught.

pkg/ccl/logictestccl/testdata/logic_test/multi_region_zone_configs

@@ -435,11 +435,47 @@ TABLE regional_by_row  ALTER TABLE regional_by_row CONFIGURE ZONE USING
                        lease_preferences = '[[+region=us-east-1]]'
 
 statement ok
-ALTER TABLE regional_by_row SET LOCALITY REGIONAL BY ROW
+DROP TABLE regional_by_row; CREATE TABLE regional_by_row (
+  pk INT PRIMARY KEY,
+  i INT,
+  INDEX idx(i),
+  FAMILY (pk, i)
+) LOCALITY REGIONAL BY ROW
+
+statement ok
+ALTER PARTITION "ap-southeast-2" OF INDEX regional_by_row@idx CONFIGURE ZONE USING gc.ttlseconds = 10;
+ALTER INDEX regional_by_row@idx CONFIGURE ZONE USING gc.ttlseconds = 10
+
+statement ok
+SELECT crdb_internal.validate_multi_region_zone_configs()
+
+statement ok
+SET override_multi_region_zone_config = true;
+ALTER PARTITION "ap-southeast-2" OF INDEX regional_by_row@idx CONFIGURE ZONE USING num_replicas = 10;
+SET override_multi_region_zone_config = false
+
+statement error zone configuration for partition "ap-southeast-2" of regional_by_row@idx contains incorrectly configured field "num_replicas"
+SELECT crdb_internal.validate_multi_region_zone_configs()
+
+statement error attempting to update zone configuration for partition "ap-southeast-2" of regional_by_row@idx which contains modified field "num_replicas"
+ALTER TABLE regional_by_row SET LOCALITY GLOBAL
 
 statement ok
+SET override_multi_region_zone_config = true;
+ALTER PARTITION "ap-southeast-2" OF INDEX regional_by_row@idx CONFIGURE ZONE DISCARD;
+SET override_multi_region_zone_config = false
+
+statement error missing zone configuration for partition "ap-southeast-2" of regional_by_row@idx
+SELECT crdb_internal.validate_multi_region_zone_configs()
+
+statement error attempting to update zone config which is missing an expected zone configuration for partition "ap-southeast-2" of regional_by_row@idx
 ALTER TABLE regional_by_row SET LOCALITY GLOBAL
 
+statement ok
+SET override_multi_region_zone_config = true;
+ALTER TABLE regional_by_row SET LOCALITY GLOBAL;
+SET override_multi_region_zone_config = false
+
 statement ok
 SELECT crdb_internal.validate_multi_region_zone_configs()
 
@@ -448,13 +484,13 @@ SET override_multi_region_zone_config = true;
 ALTER index regional_by_row@primary CONFIGURE ZONE USING num_replicas = 10;
 SET override_multi_region_zone_config = false
 
-statement ok
+statement error attempting to update zone config which contains an extra zone configuration for index regional_by_row@"primary"
 ALTER TABLE regional_by_row SET LOCALITY GLOBAL
 
-statement error zone configuration for index regional_by_row@"primary" contains incorrectly configured field "num_replicas"
+statement error extraneous zone configuration for index regional_by_row@"primary"
 SELECT crdb_internal.validate_multi_region_zone_configs()
 
-statement error attempting to update zone configuration for index regional_by_row@"primary" which contains modified field "num_replicas"
+statement error attempting to update zone config which contains an extra zone configuration for index regional_by_row@"primary"
 ALTER TABLE regional_by_row SET LOCALITY REGIONAL BY ROW
 
 statement ok
@@ -489,7 +525,7 @@ INDEX regional_by_row_as@primary  ALTER INDEX regional_by_row_as@primary CONFIGU
                                   voter_constraints = '[+region=us-east-1]',
                                   lease_preferences = '[[+region=us-east-1]]'
 
-statement error attempting to update zone configuration for index regional_by_row_as@"primary" which contains modified field "num_replicas"
+statement error attempting to update zone config which contains an extra zone configuration for index regional_by_row_as@"primary"
 ALTER TABLE regional_by_row_as SET LOCALITY REGIONAL BY ROW
 
 statement ok

pkg/ccl/logictestccl/testdata/logic_test/regional_by_row

@@ -1601,6 +1601,9 @@ DATABASE add_regions  ALTER DATABASE add_regions CONFIGURE ZONE USING
 statement ok
 ALTER DATABASE add_regions ADD REGION "us-east-1"
 
+statement ok
+SELECT crdb_internal.validate_multi_region_zone_configs()
+
 query TT
 SHOW CREATE TABLE regional_by_row
 ----

pkg/config/zonepb/zone.go

@@ -78,30 +78,6 @@ var MultiRegionZoneConfigFieldsSet = func() map[tree.Name]struct{} {
 	return ret
 }()
 
-// IsAnyMultiRegionFieldSet returns true if any of the multi-region fields are
-// set on the given zone config, and false if none of them are set.
-func (z *ZoneConfig) IsAnyMultiRegionFieldSet() (bool, string) {
-	if z.GlobalReads != nil {
-		return true, "global_reads"
-	}
-	if z.NumVoters != nil {
-		return true, "num_voters"
-	}
-	if z.NumReplicas != nil {
-		return true, "num_replicas"
-	}
-	if len(z.Constraints) != 0 {
-		return true, "constraints"
-	}
-	if len(z.LeasePreferences) != 0 {
-		return true, "lease_preferences"
-	}
-	if len(z.VoterConstraints) != 0 {
-		return true, "voter_constraints"
-	}
-	return false, ""
-}
-
 // ZoneSpecifierFromID creates a tree.ZoneSpecifier for the zone with the
 // given ID.
 func ZoneSpecifierFromID(
@@ -625,10 +601,33 @@ func (z *ZoneConfig) CopyFromZone(other ZoneConfig, fieldList []tree.Name) {
 	}
 }
 
+// DiffWithZoneMismatch indicates a mismatch between zone configurations.
+type DiffWithZoneMismatch struct {
+	// NOTE: the below fields are only set if there is a subzone in the
+	// zone configuration which is mismatching.
+
+	// IndexID represents a subzone with a mismatching index ID.
+	IndexID uint32
+	// PartitionName represents a subzone with a mismatching partitionName.
+	PartitionName string
+
+	// NOTE: only one of the below fields is set.
+
+	// IsMissingSubzone indicates a subzone is missing.
+	IsMissingSubzone bool
+	// IsExtraSubzone indicates we have an extraneous subzone.
+	IsExtraSubzone bool
+	// Field indicates the field which is wrong.
+	Field string
+}
+
 // DiffWithZone diffs all specified fields of the supplied ZoneConfig, with the
 // receiver ZoneConfig. Returns true if all are equal, and false if there is a
-// difference (along with a string which represents the first difference found).
-func (z *ZoneConfig) DiffWithZone(other ZoneConfig, fieldList []tree.Name) (bool, string, error) {
+// difference (along with a DiffWithZoneMismatch which represents the first
+// difference found).
+func (z *ZoneConfig) DiffWithZone(
+	other ZoneConfig, fieldList []tree.Name,
+) (bool, DiffWithZoneMismatch, error) {
 	for _, fieldName := range fieldList {
 		switch fieldName {
 		case "num_replicas":
@@ -637,60 +636,76 @@ func (z *ZoneConfig) DiffWithZone(other ZoneConfig, fieldList []tree.Name) (bool
 			}
 			if z.NumReplicas == nil || other.NumReplicas == nil ||
 				*z.NumReplicas != *other.NumReplicas {
-				return false, "num_replicas", nil
+				return false, DiffWithZoneMismatch{
+					Field: "num_replicas",
+				}, nil
 			}
 		case "num_voters":
 			if other.NumVoters == nil && z.NumVoters == nil {
 				continue
 			}
 			if z.NumVoters == nil || other.NumVoters == nil ||
 				*z.NumVoters != *other.NumVoters {
-				return false, "num_voters", nil
+				return false, DiffWithZoneMismatch{
+					Field: "num_voters",
+				}, nil
 			}
 		case "range_min_bytes":
 			if other.RangeMinBytes == nil && z.RangeMinBytes == nil {
 				continue
 			}
 			if z.RangeMinBytes == nil || other.RangeMinBytes == nil ||
 				*z.RangeMinBytes != *other.RangeMinBytes {
-				return false, "range_min_bytes", nil
+				return false, DiffWithZoneMismatch{
+					Field: "range_min_bytes",
+				}, nil
 			}
 		case "range_max_bytes":
 			if other.RangeMaxBytes == nil && z.RangeMaxBytes == nil {
 				continue
 			}
 			if z.RangeMaxBytes == nil || other.RangeMaxBytes == nil ||
 				*z.RangeMaxBytes != *other.RangeMaxBytes {
-				return false, "range_max_bytes", nil
+				return false, DiffWithZoneMismatch{
+					Field: "range_max_bytes",
+				}, nil
 			}
 		case "global_reads":
 			if other.GlobalReads == nil && z.GlobalReads == nil {
 				continue
 			}
 			if z.GlobalReads == nil || other.GlobalReads == nil ||
 				*z.GlobalReads != *other.GlobalReads {
-				return false, "global_reads", nil
+				return false, DiffWithZoneMismatch{
+					Field: "global_reads",
+				}, nil
 			}
 		case "gc.ttlseconds":
 			if other.GC == nil && z.GC == nil {
 				continue
 			}
 			if z.GC == nil || other.GC == nil || *z.GC != *other.GC {
-				return false, "gc.ttlseconds", nil
+				return false, DiffWithZoneMismatch{
+					Field: "gc.ttlseconds",
+				}, nil
 			}
 		case "constraints":
 			if other.Constraints == nil && z.Constraints == nil {
 				continue
 			}
 			if z.Constraints == nil || other.Constraints == nil {
-				return false, "constraints", nil
+				return false, DiffWithZoneMismatch{
+					Field: "constraints",
+				}, nil
 			}
 			for i, c := range z.Constraints {
 				for j, constraint := range c.Constraints {
 					if len(other.Constraints) <= i ||
 						len(other.Constraints[i].Constraints) <= j ||
 						constraint != other.Constraints[i].Constraints[j] {
-						return false, "constraints", nil
+						return false, DiffWithZoneMismatch{
+							Field: "constraints",
+						}, nil
 					}
 				}
 			}
@@ -699,14 +714,18 @@ func (z *ZoneConfig) DiffWithZone(other ZoneConfig, fieldList []tree.Name) (bool
 				continue
 			}
 			if z.VoterConstraints == nil || other.VoterConstraints == nil {
-				return false, "voter_constraints", nil
+				return false, DiffWithZoneMismatch{
+					Field: "voter_constraints",
+				}, nil
 			}
 			for i, c := range z.VoterConstraints {
 				for j, constraint := range c.Constraints {
 					if len(other.VoterConstraints) <= i ||
 						len(other.VoterConstraints[i].Constraints) <= j ||
 						constraint != other.VoterConstraints[i].Constraints[j] {
-						return false, "voter_constraints", nil
+						return false, DiffWithZoneMismatch{
+							Field: "voter_constraints",
+						}, nil
 					}
 				}
 			}
@@ -715,40 +734,98 @@ func (z *ZoneConfig) DiffWithZone(other ZoneConfig, fieldList []tree.Name) (bool
 				continue
 			}
 			if z.LeasePreferences == nil || other.LeasePreferences == nil {
-				return false, "voter_constraints", nil
+				return false, DiffWithZoneMismatch{
+					Field: "voter_constraints",
+				}, nil
 			}
 			for i, c := range z.LeasePreferences {
 				for j, constraint := range c.Constraints {
 					if len(other.LeasePreferences) <= i ||
 						len(other.LeasePreferences[i].Constraints) <= j ||
 						constraint != other.LeasePreferences[i].Constraints[j] {
-						return false, "lease_preferences", nil
+						return false, DiffWithZoneMismatch{
+							Field: "lease_preferences",
+						}, nil
 					}
 				}
 			}
 		default:
-			return false, "", errors.AssertionFailedf("unknown zone configuration field %q", fieldName)
+			return false, DiffWithZoneMismatch{}, errors.AssertionFailedf("unknown zone configuration field %q", fieldName)
 		}
 	}
 
 	// Look into all subzones and ensure they're equal across both zone
 	// configs.
-	if len(z.Subzones) != len(other.Subzones) {
-		return false, "subzones", nil
+	// These need to be read in as a map as subzones can be added out-of-order.
+	type subzoneKey struct {
+		indexID       uint32
+		partitionName string
 	}
-	for i, s := range z.Subzones {
-		o := other.Subzones[i]
-		if s.IndexID != o.IndexID {
-			return false, "subzone_index_id", nil
+	otherSubzonesBySubzoneKey := make(map[subzoneKey]Subzone, len(other.Subzones))
+	for _, o := range other.Subzones {
+		k := subzoneKey{indexID: o.IndexID, partitionName: o.PartitionName}
+		otherSubzonesBySubzoneKey[k] = o
+	}
+	for _, s := range z.Subzones {
+		k := subzoneKey{indexID: s.IndexID, partitionName: s.PartitionName}
+		o, found := otherSubzonesBySubzoneKey[k]
+		if !found {
+			// There can be an extra zone config defined so long as
+			// it doesn't have any fields in the fieldList set.
+			if b, subzoneMismatch, err := s.Config.DiffWithZone(
+				*NewZoneConfig(),
+				fieldList,
+			); err != nil {
+				return b, subzoneMismatch, err
+			} else if !b {
+				return false, DiffWithZoneMismatch{
+					IndexID:        s.IndexID,
+					PartitionName:  s.PartitionName,
+					IsExtraSubzone: true,
+				}, nil
+			}
+			continue
 		}
-		if s.PartitionName != o.PartitionName {
-			return false, "subzone_partition_name", nil
+		if b, subzoneMismatch, err := s.Config.DiffWithZone(
+			o.Config,
+			fieldList,
+		); err != nil {
+			return b, subzoneMismatch, err
+		} else if !b {
+			// We should never have subzones nested within subzones.
+			if subzoneMismatch.IndexID > 0 {
+				return false, DiffWithZoneMismatch{}, errors.AssertionFailedf(
+					"unexpected subzone index id %d",
+					subzoneMismatch.IndexID,
+				)
+			}
+			return b, DiffWithZoneMismatch{
+				IndexID:       o.IndexID,
+				PartitionName: o.PartitionName,
+				Field:         subzoneMismatch.Field,
+			}, nil
 		}
-		if b, str, err := s.Config.DiffWithZone(o.Config, fieldList); !b {
-			return b, str, err
+		delete(otherSubzonesBySubzoneKey, k)
+	}
+
+	// Anything remaining in the map can be presumed to be missing.
+	// This is permitted provided that everything in the field list
+	// still matches on an empty zone configuration.
+	for _, o := range otherSubzonesBySubzoneKey {
+		if b, subzoneMismatch, err := NewZoneConfig().DiffWithZone(
+			o.Config,
+			fieldList,
+		); err != nil {
+			return b, subzoneMismatch, err
+		} else if !b {
+			return false, DiffWithZoneMismatch{
+				IndexID:          o.IndexID,
+				PartitionName:    o.PartitionName,
+				IsMissingSubzone: true,
+			}, nil
 		}
 	}
-	return true, "", nil
+	return true, DiffWithZoneMismatch{}, nil
 }
 
 // StoreSatisfiesConstraint checks whether a store satisfies the given constraint.