fix: use hash to compare user metadata value (#10749)

# Which Problems Are Solved

Depending on the metadata values (already existing), the newly created
index (#10415) cannot be created or error in the future.

# How the Problems Are Solved

- Create the index using `sha256` and change the query to use sha256 as
well when comparing bytes values such as user_metadata.
- Added a setup step to cleanup potentially created index on
`projections.user_metadata5`

# Additional Changes

None

# Additional Context

- relates to #10415
- requires backport to v4.x

(cherry picked from commit 57e8033b6e)

internal/query/projection/user_metadata.go

@@ -49,8 +49,8 @@ func (*userMetadataProjection) Init() *old_handler.Check {
 		},
 			handler.NewPrimaryKey(UserMetadataColumnInstanceID, UserMetadataColumnUserID, UserMetadataColumnKey),
 			handler.WithIndex(handler.NewIndex("resource_owner", []string{UserGrantResourceOwner})),
-			handler.WithIndex(handler.NewIndex("metadata_key", []string{UserMetadataColumnKey})),
-			handler.WithIndex(handler.NewIndex("metadata_value", []string{UserMetadataColumnValue})),
+			handler.WithIndex(handler.NewIndex("key", []string{UserMetadataColumnKey})),
+			handler.WithIndex(handler.NewIndex("value", []string{"sha256(" + UserMetadataColumnValue + ")"})),
 		),
 	)
 }

internal/query/search_query.go

@@ -648,13 +648,12 @@ func (q *BytesQuery) toQuery(query sq.SelectBuilder) sq.SelectBuilder {
 func (q *BytesQuery) comp() sq.Sqlizer {
 	switch q.Compare {
 	case BytesEquals:
-		return sq.Eq{q.Column.identifier(): q.Value}
+		return sq.Expr("sha256("+q.Column.identifier()+") = sha256(?)", q.Value)
 	case BytesNotEquals:
-		return sq.NotEq{q.Column.identifier(): q.Value}
+		return sq.Expr("sha256("+q.Column.identifier()+") <> sha256(?)", q.Value)
 	case bytesCompareMax:
 		return nil
 	}
-
 	return nil
 }
 

internal/query/search_query_test.go

@@ -2287,7 +2287,7 @@ func TestBytesQuery_comp(t *testing.T) {
 				Compare: BytesEquals,
 			},
 			want: want{
-				query: sq.Eq{"test_table.test_col": []byte("foo")},
+				query: sq.Expr("sha256(test_table.test_col) = sha256(?)", []byte("foo")),
 			},
 		},
 		{
@@ -2298,7 +2298,7 @@ func TestBytesQuery_comp(t *testing.T) {
 				Compare: BytesNotEquals,
 			},
 			want: want{
-				query: sq.NotEq{"test_table.test_col": []byte("foo")},
+				query: sq.Expr("sha256(test_table.test_col) <> sha256(?)", []byte("foo")),
 			},
 		},
 		{
@@ -2322,7 +2322,7 @@ func TestBytesQuery_comp(t *testing.T) {
 			},
 			want: want{
 				err:   true,
-				query: sq.Eq{"": []byte("foo")},
+				query: sq.Expr("sha256() = sha256(?)", []byte("foo")),
 			},
 		},
 	}