objectstore

package
v1.10.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 20, 2026 License: Apache-2.0 Imports: 27 Imported by: 2

Documentation

Overview

Package objectstore provides an S3-compatible object storage abstraction. It can be used by any part of the system that needs to store or retrieve objects from S3, GCS (via S3 interop), MinIO, R2, or other S3-compatible stores.

Index

Constants

View Source 
const (
	FormatJSON    = "json"
	FormatParquet = "parquet"
)

Object formats accepted by Config.Format.

View Source 
const (
	CompressionNone   = "none"
	CompressionGzip   = "gzip"
	CompressionSnappy = "snappy"
	CompressionZstd   = "zstd"
)

Compression codecs accepted by Config.Compression.

View Source 
const (
	ContentTypeJSON    = "application/json"
	ContentTypeParquet = "application/vnd.apache.parquet"
)

Content types written for each object format.

Variables

View Source 
var ErrNotFound = errors.New("objectstore: object not found")

ErrNotFound is wrapped by Get when the key definitely has no object, so callers can tell "nothing stored" from a read that failed for another reason (and so must not assume the object is absent).

Functions

func DecodeParquetToJSONObject added in v1.10.0 

func DecodeParquetToJSONObject(data []byte) ([]byte, error)

DecodeParquetToJSONObject reverses EncodeJSONObjectToParquet and returns the row as a JSON object with keys in sorted order.

func EncodeJSONObjectToParquet added in v1.10.0 

func EncodeJSONObjectToParquet(data []byte, compression string) ([]byte, error)

EncodeJSONObjectToParquet converts a top-level JSON object into a parquet file holding one row. Every key becomes an optional UTF8 column (columns are sorted by name). String values are stored unquoted so the columns are directly queryable; any other value (object, array, number, bool) is stored as its raw JSON text and its column is listed in the bifrost.json_columns metadata so DecodeParquetToJSONObject can restore it exactly. JSON null is stored as a parquet null.

func IsParquet added in v1.10.0 

func IsParquet(data []byte) bool

IsParquet reports whether data starts with the parquet file magic.

func PutWith added in v1.10.0 

func PutWith(ctx context.Context, store ObjectStore, key string, data []byte, tags map[string]string, opts PutOptions) error

PutWith writes through PutWithOptions when the store supports it and falls back to Put otherwise. Callers that need the options honoured (for example a parquet body that must not be gzipped again) should only pass stores that implement OptionsPutter; NewObjectStore and the in-memory store all do.

Types

type Config 

type Config struct {
	Type   StoreType         `json:"type"` // "s3" or "gcs"
	Bucket schemas.SecretVar `json:"bucket"`

	// Common fields (apply to all store types)
	Prefix   string `json:"prefix,omitempty"`   // Key prefix for all stored objects. Default: "bifrost".
	Compress bool   `json:"compress,omitempty"` // Legacy: gzip stored objects; same as compression "gzip".

	Format      string `json:"format,omitempty"`      // "json" (default) or "parquet".
	Compression string `json:"compression,omitempty"` // "none", "gzip", "snappy" or "zstd".

	// S3 fields (used when Type == "s3")
	Region          *schemas.SecretVar `json:"region,omitempty"`
	Endpoint        *schemas.SecretVar `json:"endpoint,omitempty"`
	AccessKeyID     *schemas.SecretVar `json:"access_key_id,omitempty"`
	SecretAccessKey *schemas.SecretVar `json:"secret_access_key,omitempty"`
	SessionToken    *schemas.SecretVar `json:"session_token,omitempty"`
	RoleARN         *schemas.SecretVar `json:"role_arn,omitempty"`
	ForcePathStyle  bool               `json:"force_path_style,omitempty"`

	// GCS fields (used when Type == "gcs")
	Credentials     *schemas.SecretVar `json:"credentials,omitempty"`      // Deprecated: use credentials_json
	CredentialsJSON *schemas.SecretVar `json:"credentials_json,omitempty"` // Service account JSON or path
	ProjectID       *schemas.SecretVar `json:"project_id,omitempty"`       // GCP project ID override
}

Config holds the configuration for an object store.

func (*Config) GetPrefix 

func (c *Config) GetPrefix() string

GetPrefix returns the configured prefix or "bifrost" as default.

func (*Config) ResolvedCompression added in v1.10.0 

func (c *Config) ResolvedCompression() string

ResolvedCompression returns the effective codec. An explicit compression wins; otherwise the legacy compress flag means gzip; otherwise json objects are stored uncompressed and parquet objects use snappy pages. For json the codec is applied to the whole object (Content-Encoding); for parquet it is the page codec inside the file.

func (*Config) ResolvedFormat added in v1.10.0 

func (c *Config) ResolvedFormat() string

ResolvedFormat returns the configured object format, defaulting to json.

func (*Config) Validate added in v1.10.0 

func (c *Config) Validate() error

Validate rejects unknown formats and codecs, and combinations the backend cannot serve.

type GCSObjectStore 

type GCSObjectStore struct {
	// contains filtered or unexported fields
}

GCSObjectStore implements ObjectStore using Google Cloud Storage.

func NewGCSObjectStore 

func NewGCSObjectStore(ctx context.Context, cfg *Config, logger schemas.Logger) (*GCSObjectStore, error)

NewGCSObjectStore creates a new GCS object store from the given config.

func (*GCSObjectStore) Close 

func (g *GCSObjectStore) Close() error

Close releases the GCS client resources.

func (*GCSObjectStore) Delete 

func (g *GCSObjectStore) Delete(ctx context.Context, key string) error

Delete removes a single object by key.

func (*GCSObjectStore) DeleteBatch 

func (g *GCSObjectStore) DeleteBatch(ctx context.Context, keys []string) error

DeleteBatch removes multiple objects.

func (*GCSObjectStore) Get 

func (g *GCSObjectStore) Get(ctx context.Context, key string) ([]byte, error)

Get retrieves and decompresses an object by key.

func (*GCSObjectStore) ListByPrefix added in v1.3.20 

func (g *GCSObjectStore) ListByPrefix(ctx context.Context, prefix string) ([]ObjectInfo, error)

ListByPrefix returns object keys matching the given prefix.

func (*GCSObjectStore) Ping 

func (g *GCSObjectStore) Ping(ctx context.Context) error

Ping checks connectivity by writing and deleting a small object, proving that the credentials have upload access (not just read). This is important because HybridLogStore strips DB payloads before async upload — a read-only principal would pass a read-based ping but silently fail all Put calls.

func (*GCSObjectStore) Put 

func (g *GCSObjectStore) Put(ctx context.Context, key string, data []byte, tags map[string]string) error

Put uploads data with optional custom metadata. When compression is enabled, data is gzip-compressed before upload.

func (*GCSObjectStore) PutWithOptions added in v1.10.0 

func (g *GCSObjectStore) PutWithOptions(ctx context.Context, key string, data []byte, tags map[string]string, opts PutOptions) error

PutWithOptions uploads data with a caller-chosen content type, optionally bypassing the store's gzip.

type InMemoryObjectStore 

type InMemoryObjectStore struct {

	// PutErr, if set, is returned by Put for simulating failures.
	PutErr error
	// GetErr, if set, is returned by Get for simulating failures.
	GetErr error
	// contains filtered or unexported fields
}

InMemoryObjectStore is an in-memory ObjectStore implementation for testing.

func NewInMemoryObjectStore 

func NewInMemoryObjectStore() *InMemoryObjectStore

NewInMemoryObjectStore creates a new in-memory object store.

func (*InMemoryObjectStore) Close 

func (m *InMemoryObjectStore) Close() error

func (*InMemoryObjectStore) ContentType added in v1.10.0 

func (m *InMemoryObjectStore) ContentType(key string) string

ContentType returns the content type recorded for key (test helper).

func (*InMemoryObjectStore) Delete 

func (m *InMemoryObjectStore) Delete(_ context.Context, key string) error

func (*InMemoryObjectStore) DeleteBatch 

func (m *InMemoryObjectStore) DeleteBatch(_ context.Context, keys []string) error

func (*InMemoryObjectStore) Get 

func (m *InMemoryObjectStore) Get(_ context.Context, key string) ([]byte, error)

func (*InMemoryObjectStore) GetTags 

func (m *InMemoryObjectStore) GetTags(key string) map[string]string

GetTags returns the tags stored for a given key. For testing assertions.

func (*InMemoryObjectStore) Keys 

func (m *InMemoryObjectStore) Keys() []string

Keys returns all stored keys. For testing assertions.

func (*InMemoryObjectStore) Len 

func (m *InMemoryObjectStore) Len() int

Len returns the number of stored objects. For testing assertions.

func (*InMemoryObjectStore) ListByPrefix added in v1.3.20 

func (m *InMemoryObjectStore) ListByPrefix(_ context.Context, prefix string) ([]ObjectInfo, error)

func (*InMemoryObjectStore) Ping 

func (m *InMemoryObjectStore) Ping(_ context.Context) error

func (*InMemoryObjectStore) Put 

func (m *InMemoryObjectStore) Put(ctx context.Context, key string, data []byte, tags map[string]string) error

func (*InMemoryObjectStore) PutWithOptions added in v1.10.0 

func (m *InMemoryObjectStore) PutWithOptions(_ context.Context, key string, data []byte, tags map[string]string, opts PutOptions) error

PutWithOptions stores data as is (the in-memory store never compresses) and records the content type for ContentType.

func (*InMemoryObjectStore) SetCreatedAt added in v1.3.20 

func (m *InMemoryObjectStore) SetCreatedAt(key string, t time.Time)

SetCreatedAt overrides the creation time for the given key. For testing assertions.

type ObjectInfo added in v1.3.20 

type ObjectInfo struct {
	Key          string
	LastModified time.Time
}

ObjectInfo describes a stored object returned by listing operations.

type ObjectStore 

type ObjectStore interface {
	// Put uploads data to the given key with optional tags.
	// The implementation handles compression (e.g., gzip) internally.
	Put(ctx context.Context, key string, data []byte, tags map[string]string) error

	// Get retrieves and decompresses data for the given key.
	Get(ctx context.Context, key string) ([]byte, error)

	// ListByPrefix returns objects matching the given prefix with metadata.
	ListByPrefix(ctx context.Context, prefix string) ([]ObjectInfo, error)

	// Delete removes an object by key.
	Delete(ctx context.Context, key string) error

	// DeleteBatch removes multiple objects by key.
	DeleteBatch(ctx context.Context, keys []string) error

	// Ping checks connectivity to the storage backend.
	Ping(ctx context.Context) error

	// Close releases resources held by the store.
	Close() error
}

ObjectStore abstracts S3-compatible blob storage operations.

func NewObjectStore 

func NewObjectStore(ctx context.Context, cfg *Config, logger schemas.Logger) (ObjectStore, error)

NewObjectStore creates the appropriate ObjectStore implementation based on config type.

type OptionsPutter added in v1.10.0 

type OptionsPutter interface {
	PutWithOptions(ctx context.Context, key string, data []byte, tags map[string]string, opts PutOptions) error
}

OptionsPutter is implemented by stores that accept per-object write options. It is separate from ObjectStore so existing implementations keep compiling.

type PutOptions added in v1.10.0 

type PutOptions struct {
	ContentType     string // Defaults to application/json.
	SkipCompression bool   // Store the body as is, with no gzip Content-Encoding.
}

PutOptions overrides how a single object is written.

type S3ObjectStore 

type S3ObjectStore struct {
	// contains filtered or unexported fields
}

S3ObjectStore implements ObjectStore using an S3-compatible backend.

func NewS3ObjectStore 

func NewS3ObjectStore(ctx context.Context, cfg *Config, logger schemas.Logger) (*S3ObjectStore, error)

NewS3ObjectStore creates a new S3-compatible object store from the given config.

func (*S3ObjectStore) Close 

func (s *S3ObjectStore) Close() error

Close is a no-op for S3 (no persistent connections to release).

func (*S3ObjectStore) Delete 

func (s *S3ObjectStore) Delete(ctx context.Context, key string) error

Delete removes a single object by key.

func (*S3ObjectStore) DeleteBatch 

func (s *S3ObjectStore) DeleteBatch(ctx context.Context, keys []string) error

DeleteBatch removes multiple objects. It uses the S3 DeleteObjects API which supports up to 1000 keys per call.

func (*S3ObjectStore) Get 

func (s *S3ObjectStore) Get(ctx context.Context, key string) ([]byte, error)

Get retrieves and decompresses an object by key.

func (*S3ObjectStore) ListByPrefix added in v1.3.20 

func (s *S3ObjectStore) ListByPrefix(ctx context.Context, prefix string) ([]ObjectInfo, error)

ListByPrefix returns all non-empty object keys matching the given prefix.

func (*S3ObjectStore) Ping 

func (s *S3ObjectStore) Ping(ctx context.Context) error

Ping checks connectivity by performing a HeadBucket call. Note: HeadBucket requires the s3:ListBucket IAM permission on the bucket resource.

func (*S3ObjectStore) Put 

func (s *S3ObjectStore) Put(ctx context.Context, key string, data []byte, tags map[string]string) error

Put uploads data with optional S3 object tags. When compression is enabled, data is gzip-compressed before upload.

func (*S3ObjectStore) PutWithOptions added in v1.10.0 

func (s *S3ObjectStore) PutWithOptions(ctx context.Context, key string, data []byte, tags map[string]string, opts PutOptions) error

PutWithOptions uploads data with a caller-chosen content type, optionally bypassing the store's gzip (used for parquet bodies, which compress pages internally and must stay readable by parquet tooling).

type StoreType 

type StoreType string

StoreType identifies the object storage backend. 

const (
	StoreTypeS3  StoreType = "s3"
	StoreTypeGCS StoreType = "gcs"
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.