Fixed branch control insertion bug

go/libraries/doltcore/branch_control/access.go

@@ -69,7 +69,8 @@ func newAccess() *Access {
 }
 
 // Match returns whether any entries match the given database, branch, user, and host, along with their permissions.
-// Requires external synchronization handling, therefore manually manage the RWMutex.
+// This will match subsets against their superset as well. Requires external synchronization handling, therefore
+// manually manage the RWMutex.
 func (tbl *Access) Match(database string, branch string, user string, host string) (bool, Permissions) {
 	results := tbl.Root.Match(database, branch, user, host)
 	// We use the result(s) with the longest length
@@ -86,6 +87,12 @@ func (tbl *Access) Match(database string, branch string, user string, host strin
 	return len(results) > 0, perms
 }
 
+// ExactMatch returns whether any entries exactly match the given database, branch, user, and host. Requires external
+// synchronization handling, therefore manually manage the RWMutex.
+func (tbl *Access) ExactMatch(database string, branch string, user string, host string) bool {
+	return tbl.Root.ExactMatch(database, branch, user, host)
+}
+
 // GetBinlog returns the table's binlog.
 func (tbl *Access) GetBinlog() *Binlog {
 	return tbl.binlog

go/libraries/doltcore/branch_control/expr_parser_node.go

@@ -143,6 +143,52 @@ func (mn *MatchNode) Match(database, branch, user, host string) []MatchResult {
 	return results
 }
 
+// ExactMatch will return whether the expressions already exist in the exact form given. This differs from match, as
+// this is not checking for subset matches. Assumes that the given expressions have already been folded.
+func (mn *MatchNode) ExactMatch(databaseExpr, branchExpr, userExpr, hostExpr string) bool {
+	root := mn
+	allSortOrders := mn.parseExpression(databaseExpr, branchExpr, userExpr, hostExpr)
+	defer func() {
+		concatenatedSortOrderPool.Put(allSortOrders)
+	}()
+
+	remainingRootSortOrders := root.SortOrders
+	allSortOrdersMaxIndex := len(allSortOrders) - 1
+ParentLoop:
+	for i, sortOrder := range allSortOrders {
+		if remainingRootSortOrders[0] == sortOrder {
+			if len(remainingRootSortOrders) > 1 && i < allSortOrdersMaxIndex {
+				// There are more sort orders on both sides, so we simply continue
+				remainingRootSortOrders = remainingRootSortOrders[1:]
+				continue
+			} else if len(remainingRootSortOrders) > 1 && i == allSortOrdersMaxIndex {
+				// We have more sort orders on the root, but no more in our expressions, so this is a partial match and
+				// not an exact match
+				return false
+			} else if len(remainingRootSortOrders) == 1 && i < allSortOrdersMaxIndex {
+				// We've run out of sort orders on the root, but still have more from children, so check if there's a
+				// matching child
+				nextSortOrder := allSortOrders[i+1]
+				if child, ok := root.Children[nextSortOrder]; ok {
+					remainingRootSortOrders = child.SortOrders
+					root = root.Children[nextSortOrder]
+					continue ParentLoop
+				}
+				// None of the children matched, so this does not have an exact match
+				return false
+			} else {
+				// We have no more sort orders on either side so this is an exact match
+				return true
+			}
+		} else {
+			// The sort orders do not match, so this does not have an exact match
+			return false
+		}
+	}
+	// Shouldn't be possible to get here, so we can assume that this isn't an exact match as well
+	return false
+}
+
 // processMatch handles the behavior of how to process a sort order against a node. Returns a new slice with any newly
 // appended nodes (which should overwrite the first parameter in the calling function).
 func processMatch(matches []matchNodeCounted, node matchNodeCounted, sortOrder int32) []matchNodeCounted {

go/libraries/doltcore/sqle/dtables/branch_control_table.go

@@ -202,11 +202,13 @@ func (tbl BranchControlTable) Insert(ctx *sql.Context, row sql.Row) error {
 		}
 	}
 
-	// We check if we're inserting a subset of an already-existing row. We only consider this a subset if the
+	// We check if we're inserting a subset (or exact match) of an already-existing row. We only consider this a subset if the
 	// permissions are as permissible as the existing ones, or are more restrictive (i.e. write is a "subset permission"
 	// of admin). If we are, we deny the insertion as the existing row will already match against ALL possible values for this row.
-	if ok, modPerms := tbl.Match(database, branch, user, host); ok && perms.Consolidate() >= modPerms.Consolidate() {
-		permBits := uint64(modPerms)
+	// If we have an exact match, then we ignore permission restrictions altogether, as we can only associate one set of
+	// permissions to a given set of expressions.
+	if ok, modPerms := tbl.Match(database, branch, user, host); (ok && perms.Consolidate() >= modPerms.Consolidate()) || tbl.ExactMatch(database, branch, user, host) {
+		permBits := uint64(perms)
 		permStr, _ := accessSchema[4].Type.(sql.SetType).BitsToString(permBits)
 		return sql.NewUniqueKeyErr(
 			fmt.Sprintf(`[%q, %q, %q, %q, %q]`, database, branch, user, host, permStr),
@@ -239,7 +241,7 @@ func (tbl BranchControlTable) Update(ctx *sql.Context, old sql.Row, new sql.Row)
 		return branch_control.ErrExpressionsTooLong.New(newDatabase, newBranch, newUser, newHost)
 	}
 
-	// If we're not updating the same row, then we check for a row violation
+	// If we're not updating the exact same row, then we check for a row violation
 	if oldDatabase != newDatabase || oldBranch != newBranch || oldUser != newUser || oldHost != newHost {
 		if ok, modPerms := tbl.Match(newDatabase, newBranch, newUser, newHost); ok {
 			permBits := uint64(modPerms)

go/libraries/doltcore/sqle/enginetest/branch_control_test.go

@@ -1361,6 +1361,25 @@ var BranchControlTests = []BranchControlTest{
 			},
 		},
 	},
+	{
+		Name: "Cannot insert an exact match even with elevated permissions",
+		Assertions: []BranchControlTestAssertion{
+			{
+				User:        "root",
+				Host:        "localhost",
+				Query:       "INSERT INTO dolt_branch_control VALUES ('%', '%', '%', '%', 'admin');",
+				ExpectedErr: sql.ErrPrimaryKeyViolation,
+			},
+			{ // This will fail if the above succeeded, so kind of a redundant check but ultimately harmless
+				User:  "root",
+				Host:  "localhost",
+				Query: "DELETE FROM dolt_branch_control WHERE permissions='admin';",
+				Expected: []sql.Row{
+					{types.NewOkResult(0)},
+				},
+			},
+		},
+	},
 }
 
 func TestBranchControl(t *testing.T) {