JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite: adjust to latest CRDB version

Browse Source 
We have an issue with latest CRDB. Single query cannot modify
the same table multiple times. Now build is blocked.
This change is unblocking build by:
* adjusting query for inserting into repair queue
* temporary removing code for deletion for server-side copy
* temporary disable backward compatibility tests for CRDB

Change-Id: Idd9744ebd228e5dc05bdaf65cfc8f779472a975d
This commit is contained in:
Michał Niewrzał 2022-03-22 15:15:45 +01:00
parent c49c646d4e
commit ba74cb17a9
4 changed files with 22 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Jenkinsfile.public
View File

@ -302,6 +302,12 @@ pipeline {
}
stage('Cockroach Backwards Compatibility') {
// skip this stage until next release because of
// https://www.cockroachlabs.com/docs/releases/v21.2.html#v21-2-3-sql-language-changes
when {
expression { false }
}
environment {
STORJ_NETWORK_HOST4 = '127.0.0.5'
STORJ_NETWORK_HOST6 = '127.0.0.5'

96
satellite/metabase/delete.go
View File

@ -118,99 +118,6 @@ SELECT
FROM deleted_objects
LEFT JOIN deleted_segments ON deleted_objects.stream_id = deleted_segments.stream_id`
// There are 3 scenarios:
// 1) If the object to be removed is singular, it should simply remove the object and its segments
// 2) If the object is a copy, it should also remove the entry from segment_copies
// 3) If the object has copies of its own, a new ancestor should be promoted,
// removed from segment_copies, and the other copies should point to the new ancestor
var deleteObjectExactVersionWithCopyFeatureSQL = `
WITH deleted_objects AS (
DELETE FROM objects
WHERE
project_id = $1 AND
bucket_name = $2 AND
object_key = $3 AND
version = $4
RETURNING
version, stream_id,
created_at, expires_at,
status, segment_count,
encrypted_metadata_nonce, encrypted_metadata, encrypted_metadata_encrypted_key,
total_plain_size, total_encrypted_size, fixed_segment_size,
encryption
),
deleted_segments AS (
DELETE FROM segments
WHERE segments.stream_id IN (SELECT deleted_objects.stream_id FROM deleted_objects)
RETURNING
segments.stream_id,
segments.root_piece_id,
segments.remote_alias_pieces,
segments.plain_size,
segments.encrypted_size,
segments.position,
segments.inline_data,
segments.repaired_at
),
deleted_copies AS (
DELETE FROM segment_copies
WHERE segment_copies.stream_id IN (SELECT deleted_objects.stream_id FROM deleted_objects)
RETURNING segment_copies.stream_id
),
-- lowest stream_id becomes new ancestor
promoted_ancestor AS (
DELETE FROM segment_copies
WHERE segment_copies.stream_id = (
SELECT stream_id
FROM segment_copies
WHERE segment_copies.ancestor_stream_id = (
SELECT stream_id
FROM deleted_objects
ORDER BY stream_id
LIMIT 1
)
ORDER BY stream_id
LIMIT 1
)
RETURNING
stream_id
),
update_other_copies_with_new_ancestor AS (
UPDATE segment_copies
SET ancestor_stream_id = (SELECT stream_id FROM promoted_ancestor)
WHERE segment_copies.ancestor_stream_id IN (SELECT stream_id FROM deleted_objects)
-- bit weird condition but this seems necessary to prevent it from updating a row which is being deleted
AND segment_copies.stream_id != (SELECT stream_id FROM promoted_ancestor)
RETURNING segment_copies.stream_id
),
update_copy_promoted_to_ancestors AS (
UPDATE segments AS promoted_segments
SET
remote_alias_pieces = deleted_segments.remote_alias_pieces,
inline_data = deleted_segments.inline_data,
plain_size = deleted_segments.plain_size,
encrypted_size = deleted_segments.encrypted_size,
repaired_at = deleted_segments.repaired_at
FROM deleted_segments
-- i suppose this only works when there is 1 stream_id being deleted
WHERE
promoted_segments.stream_id IN (SELECT stream_id FROM promoted_ancestor)
AND promoted_segments.position = deleted_segments.position
RETURNING deleted_segments.stream_id
)
SELECT
deleted_objects.version, deleted_objects.stream_id,
deleted_objects.created_at, deleted_objects.expires_at,
deleted_objects.status, deleted_objects.segment_count,
deleted_objects.encrypted_metadata_nonce, deleted_objects.encrypted_metadata, deleted_objects.encrypted_metadata_encrypted_key,
deleted_objects.total_plain_size, deleted_objects.total_encrypted_size, deleted_objects.fixed_segment_size,
deleted_objects.encryption,
deleted_segments.root_piece_id, deleted_segments.remote_alias_pieces,
(SELECT stream_id FROM promoted_ancestor)
FROM deleted_objects
LEFT JOIN deleted_segments
ON deleted_objects.stream_id = deleted_segments.stream_id`
// DeleteObjectExactVersion deletes an exact object version.
func (db *DB) DeleteObjectExactVersion(ctx context.Context, opts DeleteObjectExactVersion) (result DeleteObjectResult, err error) {
defer mon.Task()(&ctx)(&err)
@ -220,9 +127,6 @@ func (db *DB) DeleteObjectExactVersion(ctx context.Context, opts DeleteObjectExa
}
deleteSQL := deleteObjectExactVersionWithoutCopyFeatureSQL
if db.config.ServerSideCopy {
deleteSQL = deleteObjectExactVersionWithCopyFeatureSQL
}
err = withRows(
db.db.QueryContext(ctx, deleteSQL, opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, opts.Version),

2
satellite/metabase/delete_test.go
View File

@ -955,6 +955,8 @@ func TestDeleteObjectsAllVersions(t *testing.T) {
}
func TestDeleteCopy(t *testing.T) {
t.Skip("skip until deletion query will be fixed for CRDB")
metabasetest.Run(t, func(ctx *testcontext.Context, t *testing.T, db *metabase.DB) {
for _, numberOfSegments := range []int{0, 1, 3} {
t.Run(fmt.Sprintf("%d segments", numberOfSegments), func(t *testing.T) {

21
satellite/satellitedb/repairqueue.go
View File

@ -50,16 +50,23 @@ func (r *repairQueue) Insert(ctx context.Context, seg *queue.InjuredSegment) (al
RETURNING (xmax != 0) AS alreadyInserted
`
case dbutil.Cockroach:
// TODO it's not optimal solution but crdb is not used in prod for repair queue
query = `
WITH updater AS (
UPDATE repair_queue SET segment_health = $3, updated_at = current_timestamp
WITH inserted AS (
SELECT count(*) as alreadyInserted FROM repair_queue
WHERE stream_id = $1 AND position = $2
RETURNING *
)
INSERT INTO repair_queue (stream_id, position, segment_health)
SELECT $1, $2, $3
WHERE NOT EXISTS (SELECT * FROM updater)
RETURNING false
INSERT INTO repair_queue
(
stream_id, position, segment_health
)
VALUES (
$1, $2, $3
)
ON CONFLICT (stream_id, position)
DO UPDATE
SET segment_health=$3, updated_at=current_timestamp
RETURNING (SELECT alreadyInserted FROM inserted)
`
}
rows, err := r.db.QueryContext(ctx, query, seg.StreamID, seg.Position.Encode(), seg.SegmentHealth)