Merge #65209

65209: sql: use of constraint columns in FK can differ from order in PK definition r=otan a=samxsmith

sql: use of constraint columns in FK can differ from order in PK definition

ColumnIDs: Implement PermutationOf
- This methods returns true if the content of the two sets of ColumnIDs
is the same, even if they are not in the same order.
Contrary to the existing Equal() method, which also requires order
equality.

descpb: IndexDescriptor uses PermutationOf for comparing columns
- Previously when referring to an existing constraint, with a set of columns,
the Equals() method was used. That method requires order equality to match.
- Postgres recognises a constraint by reference even when the
provided fields are not in the order they were given when creating it.
- PermutationOf compares the column elements regardless of order.
This means that the behaviour of a constraint reference matches that of
Postgres.

Fixes #63324

Release note (sql change): The use order of columns in a foreign key no longer needs
to match the order the columns were defined for the reference table's unique constraint.

Co-authored-by: Sam X Smith <[email protected]>

pkg/ccl/logictestccl/testdata/logic_test/partitioning_implicit

@@ -227,6 +227,28 @@ CREATE TABLE fk_using_implicit_columns_against_t (
   ref_t_c INT NOT NULL REFERENCES t(c)
 )
 
+statement ok
+CREATE TABLE fk_using_implicit_columns_against_t_one_family (
+  pk INT NOT NULL PRIMARY KEY,
+  ref_t_pk INT NOT NULL REFERENCES t,
+  ref_t_c INT NOT NULL REFERENCES t(c),
+  FAMILY (pk, ref_t_pk, ref_t_c)
+)
+
+query T
+SELECT create_statement FROM [SHOW CREATE TABLE fk_using_implicit_columns_against_t_one_family]
+----
+CREATE TABLE public.fk_using_implicit_columns_against_t_one_family (
+   pk INT8 NOT NULL,
+   ref_t_pk INT8 NOT NULL,
+   ref_t_c INT8 NOT NULL,
+   CONSTRAINT "primary" PRIMARY KEY (pk ASC),
+   CONSTRAINT fk_ref_t_pk_ref_t FOREIGN KEY (ref_t_pk) REFERENCES public.t(pk),
+   CONSTRAINT fk_ref_t_c_ref_t FOREIGN KEY (ref_t_c) REFERENCES public.t(c),
+   FAMILY fam_0_pk_ref_t_pk_ref_t_c (pk, ref_t_pk, ref_t_c)
+)
+
+
 query T
 EXPLAIN INSERT INTO fk_using_implicit_columns_against_t VALUES (1, 1, 4)
 ----

pkg/sql/catalog/descpb/BUILD.bazel

@@ -29,6 +29,7 @@ go_library(
         "//pkg/sql/privilege",
         "//pkg/sql/sem/tree",
         "//pkg/sql/types",
+        "//pkg/util",
         "//pkg/util/encoding",
         "//pkg/util/errorutil/unimplemented",
         "//pkg/util/hlc",
@@ -41,7 +42,10 @@ go_library(
 go_test(
     name = "descpb_test",
     size = "small",
-    srcs = ["privilege_test.go"],
+    srcs = [
+        "privilege_test.go",
+        "structured_test.go",
+    ],
     embed = [":descpb"],
     deps = [
         "//pkg/keys",

pkg/sql/catalog/descpb/index.go

@@ -107,7 +107,7 @@ func (desc *IndexDescriptor) IsValidOriginIndex(originColIDs ColumnIDs) bool {
 func (desc *IndexDescriptor) IsValidReferencedUniqueConstraint(referencedColIDs ColumnIDs) bool {
 	return desc.Unique &&
 		!desc.IsPartial() &&
-		ColumnIDs(desc.KeyColumnIDs[desc.Partitioning.NumImplicitColumns:]).Equals(referencedColIDs)
+		ColumnIDs(desc.KeyColumnIDs[desc.Partitioning.NumImplicitColumns:]).PermutationOf(referencedColIDs)
 }
 
 // GetName is part of the UniqueConstraint interface.

pkg/sql/catalog/descpb/structured.go

@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catconstants"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/encoding"
 	"github.com/cockroachdb/errors"
 )
@@ -151,6 +152,22 @@ func (c ColumnIDs) Equals(input ColumnIDs) bool {
 	return true
 }
 
+// PermutationOf returns true if this list and the input list contain the same
+// set of column IDs in any order. Duplicate ColumnIDs have no effect.
+func (c ColumnIDs) PermutationOf(input ColumnIDs) bool {
+	ourColsSet := util.MakeFastIntSet()
+	for _, col := range c {
+		ourColsSet.Add(int(col))
+	}
+
+	inputColsSet := util.MakeFastIntSet()
+	for _, inputCol := range input {
+		inputColsSet.Add(int(inputCol))
+	}
+
+	return inputColsSet.Equals(ourColsSet)
+}
+
 // Contains returns whether this list contains the input ID.
 func (c ColumnIDs) Contains(i ColumnID) bool {
 	for _, id := range c {

pkg/sql/catalog/descpb/structured_test.go

@@ -0,0 +1,81 @@
+// 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 descpb
+
+import "testing"
+
+func TestColumnIDsPermutationOf(t *testing.T) {
+	type testCase struct {
+		name                       string
+		columnIDsOne, columnIDsTwo ColumnIDs
+		expectedResult             bool
+	}
+
+	testCases := []testCase{
+		{
+			name:           "different IDs",
+			columnIDsOne:   ColumnIDs{ColumnID(1), ColumnID(2)},
+			columnIDsTwo:   ColumnIDs{ColumnID(3), ColumnID(4)},
+			expectedResult: false,
+		},
+		{
+			name:           "one element extra",
+			columnIDsOne:   ColumnIDs{ColumnID(1), ColumnID(2)},
+			columnIDsTwo:   ColumnIDs{ColumnID(1), ColumnID(2), ColumnID(3)},
+			expectedResult: false,
+		},
+		{
+			name:           "one element less",
+			columnIDsOne:   ColumnIDs{ColumnID(1), ColumnID(2), ColumnID(3)},
+			columnIDsTwo:   ColumnIDs{ColumnID(1), ColumnID(2)},
+			expectedResult: false,
+		},
+		{
+			name:           "same elements, in different order",
+			columnIDsOne:   ColumnIDs{ColumnID(1), ColumnID(2), ColumnID(3)},
+			columnIDsTwo:   ColumnIDs{ColumnID(3), ColumnID(2), ColumnID(1)},
+			expectedResult: true,
+		},
+		{
+			name:           "when duplicate is in both, returns true",
+			columnIDsOne:   ColumnIDs{ColumnID(1), ColumnID(2), ColumnID(1), ColumnID(3)},
+			columnIDsTwo:   ColumnIDs{ColumnID(3), ColumnID(1), ColumnID(2), ColumnID(1)},
+			expectedResult: true,
+		},
+		{
+			name:           "when duplicate in first, returns true",
+			columnIDsOne:   ColumnIDs{ColumnID(1), ColumnID(2), ColumnID(1), ColumnID(3)},
+			columnIDsTwo:   ColumnIDs{ColumnID(3), ColumnID(2), ColumnID(1)},
+			expectedResult: true,
+		},
+		{
+			name:           "when duplicate in second, returns true",
+			columnIDsOne:   ColumnIDs{ColumnID(2), ColumnID(1), ColumnID(3)},
+			columnIDsTwo:   ColumnIDs{ColumnID(3), ColumnID(1), ColumnID(2), ColumnID(1)},
+			expectedResult: true,
+		},
+		{
+			name:           "when each list has a different duplicate, same length, returns true",
+			columnIDsOne:   ColumnIDs{ColumnID(2), ColumnID(1), ColumnID(2)},
+			columnIDsTwo:   ColumnIDs{ColumnID(1), ColumnID(2), ColumnID(1)},
+			expectedResult: true,
+		},
+	}
+
+	for _, tt := range testCases {
+		t.Run(tt.name, func(t *testing.T) {
+			result := tt.columnIDsOne.PermutationOf(tt.columnIDsTwo)
+			if result != tt.expectedResult {
+				t.Errorf("PermuationOf() %s: got %v, want %v", tt.name, result, tt.expectedResult)
+			}
+		})
+	}
+}

pkg/sql/logictest/testdata/logic_test/fk

@@ -3732,3 +3732,53 @@ INSERT INTO t65890_c SELECT 1, 2, 2
 
 statement ok
 SET enable_insert_fast_path = $enable_insert_fast_path
+
+subtest reference_constraint_different_order
+
+# https://github.com/cockroachdb/cockroach/issues/63324
+# Referencing an existing constraint by fields in a different order
+# should still find existing constraint.
+
+statement ok
+CREATE TABLE reference_constraint_different_order_parent (
+  a INT8 NOT NULL,
+  b INT8 NOT NULL,
+  c INT8 NOT NULL,
+  p_str VARCHAR(10),
+  PRIMARY KEY (a, b, c)
+);
+CREATE TABLE reference_constraint_different_order_child (
+  id INT8,
+  x INT8,
+  y INT8,
+  z INT8,
+  c_str VARCHAR(10)
+)
+
+statement ok
+ALTER TABLE reference_constraint_different_order_child
+  ADD CONSTRAINT p_fk FOREIGN KEY (x, y, z) REFERENCES reference_constraint_different_order_parent (a, c, b)
+
+statement ok
+INSERT
+INTO
+	reference_constraint_different_order_parent
+VALUES (1, 2, 3, 'par_val');
+INSERT
+INTO reference_constraint_different_order_child
+VALUES (99, 1, 3, 2, 'child_val')
+
+query IIIITIIIT colnames
+SELECT
+	*
+FROM
+	reference_constraint_different_order_child
+	NATURAL JOIN reference_constraint_different_order_parent
+----
+id  x  y  z  c_str      a  b  c  p_str
+99  1  3  2  child_val  1  2  3  par_val
+
+statement error insert on table "reference_constraint_different_order_child" violates foreign key constraint "p_fk"
+INSERT
+INTO reference_constraint_different_order_child
+VALUES (99, 1, 2, 3, 'child_val')