sql: migrate ObjectMetadata

Author: ChrisSchinnerl

stores/metadata.go

@@ -1428,39 +1428,12 @@ func (s *SQLStore) objectHydrate(tx *gorm.DB, bucket, path string, obj rawObject
 }
 
 // ObjectMetadata returns an object's metadata
-func (s *SQLStore) ObjectMetadata(ctx context.Context, bucket, path string) (api.Object, error) {
-	var resp api.Object
-	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
-		var obj dbObject
-		err := tx.Model(&dbObject{}).
-			Joins("INNER JOIN buckets b ON objects.db_bucket_id = b.id").
-			Where("b.name", bucket).
-			Where("object_id", path).
-			Take(&obj).
-			Error
-		if errors.Is(err, gorm.ErrRecordNotFound) {
-			return api.ErrObjectNotFound
-		} else if err != nil {
-			return err
-		}
-		oum, err := s.objectMetadata(tx, bucket, path)
-		if err != nil {
-			return err
-		}
-		resp = api.Object{
-			ObjectMetadata: newObjectMetadata(
-				obj.ObjectID,
-				obj.Etag,
-				obj.MimeType,
-				obj.Health,
-				obj.CreatedAt,
-				obj.Size,
-			),
-			Metadata: oum,
-		}
-		return nil
+func (s *SQLStore) ObjectMetadata(ctx context.Context, bucket, path string) (obj api.Object, err error) {
+	err = s.bMain.Transaction(ctx, func(tx sql.DatabaseTx) error {
+		obj, err = tx.ObjectMetadata(ctx, bucket, path)
+		return err
 	})
-	return resp, err
+	return
 }
 
 func (s *SQLStore) objectMetadata(tx *gorm.DB, bucket, path string) (api.ObjectUserMetadata, error) {

stores/sql/database.go

@@ -37,13 +37,13 @@ type (
 	DatabaseTx interface {
 		// AbortMultipartUpload aborts a multipart upload and deletes it from
 		// the database.
-		AbortMultipartUpload(ctx context.Context, bucket, path string, uploadID string) error
+		AbortMultipartUpload(ctx context.Context, bucket, key string, uploadID string) error
 
 		// Accounts returns all accounts from the db.
 		Accounts(ctx context.Context) ([]api.Account, error)
 
 		// AddMultipartPart adds a part to an unfinished multipart upload.
-		AddMultipartPart(ctx context.Context, bucket, path, contractSet, eTag, uploadID string, partNumber int, slices object.SlabSlices) error
+		AddMultipartPart(ctx context.Context, bucket, key, contractSet, eTag, uploadID string, partNumber int, slices object.SlabSlices) error
 
 		// AddPeer adds a peer to the store.
 		AddPeer(ctx context.Context, addr string) error
@@ -197,6 +197,9 @@ type (
 		// MultipartUploads returns a list of all multipart uploads.
 		MultipartUploads(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker string, limit int) (api.MultipartListUploadsResponse, error)
 
+		// ObjectMetadata returns an object's metadata.
+		ObjectMetadata(ctx context.Context, bucket, key string) (api.Object, error)
+
 		// ObjectsStats returns overall stats about stored objects
 		ObjectsStats(ctx context.Context, opts api.ObjectsStatsOpts) (api.ObjectsStatsResponse, error)
 

stores/sql/main.go

@@ -1293,6 +1293,59 @@ func NormalizePeer(peer string) (string, error) {
 	return normalized.String(), nil
 }
 
+func ObjectMetadata(ctx context.Context, tx sql.Tx, bucket, key string) (api.Object, error) {
+	// fetch object id
+	var objID int64
+	if err := tx.QueryRow(ctx, `
+		SELECT o.id
+		FROM objects o
+		INNER JOIN buckets b ON b.id = o.db_bucket_id
+		WHERE o.object_id = ? AND b.name = ?
+	`, key, bucket).Scan(&objID); errors.Is(err, dsql.ErrNoRows) {
+		return api.Object{}, api.ErrObjectNotFound
+	} else if err != nil {
+		return api.Object{}, fmt.Errorf("failed to fetch object id: %w", err)
+	}
+
+	// fetch metadata
+	om, err := scanObjectMetadata(tx.QueryRow(ctx, `
+		SELECT o.object_id, o.size, o.health, o.mime_type, o.created_at, o.etag
+		FROM objects o
+		WHERE o.id = ?
+	`, objID))
+	if err != nil {
+		return api.Object{}, fmt.Errorf("failed to fetch object metadata: %w", err)
+	}
+
+	// fetch user metadata
+	rows, err := tx.Query(ctx, `
+		SELECT oum.key, oum.value
+		FROM object_user_metadata oum
+		WHERE oum.db_object_id = ?
+		ORDER BY oum.id ASC
+	`, objID)
+	if err != nil {
+		return api.Object{}, fmt.Errorf("failed to fetch user metadata: %w", err)
+	}
+	defer rows.Close()
+
+	// build object
+	metadata := make(api.ObjectUserMetadata)
+	for rows.Next() {
+		var key, value string
+		if err := rows.Scan(&key, &value); err != nil {
+			return api.Object{}, fmt.Errorf("failed to scan user metadata: %w", err)
+		}
+		metadata[key] = value
+	}
+
+	return api.Object{
+		Metadata:       metadata,
+		ObjectMetadata: om,
+		Object:         nil, // only return metadata
+	}, nil
+}
+
 func ObjectsStats(ctx context.Context, tx sql.Tx, opts api.ObjectsStatsOpts) (api.ObjectsStatsResponse, error) {
 	var args []any
 	var bucketExpr string

stores/sql/mysql/main.go

@@ -522,6 +522,10 @@ func (tx *MainDatabaseTx) MultipartUploads(ctx context.Context, bucket, prefix,
 	return ssql.MultipartUploads(ctx, tx, bucket, prefix, keyMarker, uploadIDMarker, limit)
 }
 
+func (tx *MainDatabaseTx) ObjectMetadata(ctx context.Context, bucket, path string) (api.Object, error) {
+	return ssql.ObjectMetadata(ctx, tx, bucket, path)
+}
+
 func (tx *MainDatabaseTx) ObjectsStats(ctx context.Context, opts api.ObjectsStatsOpts) (api.ObjectsStatsResponse, error) {
 	return ssql.ObjectsStats(ctx, tx, opts)
 }

stores/sql/sqlite/main.go

@@ -519,6 +519,10 @@ func (tx *MainDatabaseTx) MultipartUploads(ctx context.Context, bucket, prefix,
 	return ssql.MultipartUploads(ctx, tx, bucket, prefix, keyMarker, uploadIDMarker, limit)
 }
 
+func (tx *MainDatabaseTx) ObjectMetadata(ctx context.Context, bucket, path string) (api.Object, error) {
+	return ssql.ObjectMetadata(ctx, tx, bucket, path)
+}
+
 func (tx *MainDatabaseTx) ObjectsStats(ctx context.Context, opts api.ObjectsStatsOpts) (api.ObjectsStatsResponse, error) {
 	return ssql.ObjectsStats(ctx, tx, opts)
 }