objstore

package
v0.18.39 Latest Latest
Warning

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

 Go to latest Published: Sep 5, 2026 License: AGPL-3.0 Imports: 23 Imported by: 0

Documentation

Overview

Package objstore provides an abstraction over S3-compatible object storage.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrNotFound is returned when a requested object does not exist.
	ErrNotFound = errors.New("object not found")

	// ErrPreconditionFailed is returned when a conditional write fails due to ETag mismatch.
	ErrPreconditionFailed = errors.New("precondition failed: etag mismatch")

	// ErrBucketNotFound is returned when a referenced bucket does not exist.
	ErrBucketNotFound = errors.New("bucket not found")
)
View Source 
var ErrCircuitOpen = errors.New("circuit breaker open: S3 unavailable")

ErrCircuitOpen is returned when the circuit breaker is open.

Functions

func StoreID 

func StoreID(s Store) string

StoreID returns s's process-unique instance identity, or "" when s does not provide one. Callers keying a cache on object identity MUST treat "" as "not cacheable" — failing closed is the only safe default, because a store that declines to identify itself is exactly the store whose (bucket, key) namespace we cannot reason about.

Types

type BaseTableCache 

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

BaseTableCache is a read-through, disk-backed, whole-file cache for immutable base-table parquet objects, decorating an inner Store (docs/design/base-table-nvme-cache.md). Eligible keys (*.parquet outside the queries/ scratch prefix) are served from the cache directory on hit — never consulting the inner store or, when stacked above CircuitStore, the breaker — and teed into the cache on miss. Non-eligible keys pass through untouched.

A (bucket, key) pair is content-stable for its lifetime: ingest writes UUID-named chunks and compaction/GC swap in new keys via the manifest rather than overwriting (see the design memo §2), so entries need no ETag validation. Population is strictly best-effort — any tee failure, short read, or size mismatch discards the temp file and the caller's stream is unaffected.

func FindBaseTableCache 

func FindBaseTableCache(s Store) *BaseTableCache

FindBaseTableCache walks a decorator chain (via Unwrap) to the BaseTableCache layer, or nil when the store has none. Lets the worker wire the peer tier without threading the concrete cache through every construction path.

func NewBaseTableCache 

func NewBaseTableCache(inner Store, dir string, budget int64, logger *slog.Logger) (*BaseTableCache, error)

NewBaseTableCache creates the cache rooted at dir with an LRU byte budget. The directory layout mirrors <dir>/<bucket>/<key>; existing entries are adopted at startup (recency seeded by mtime) so the cache survives process restarts. budget must be > 0 — a zero budget means the feature is off and the decorator should not be constructed.

func (*BaseTableCache) BucketExists 

func (c *BaseTableCache) BucketExists(ctx context.Context, bucket string) (bool, error)

BucketExists implements Store.

func (*BaseTableCache) CachedLocalPath 

func (c *BaseTableCache) CachedLocalPath(bucket, key string) (string, bool)

CachedLocalPath implements LocalPathStore.

func (*BaseTableCache) Delete 

func (c *BaseTableCache) Delete(ctx context.Context, bucket, key string) error

Delete implements Store. Dropping the entry eagerly reclaims disk that LRU pressure would otherwise take time to find (compaction/GC swaps).

func (*BaseTableCache) Get 

func (c *BaseTableCache) Get(ctx context.Context, bucket, key string) (io.ReadCloser, ObjectInfo, error)

Get implements Store. Hits stream from the cache file; misses try the peer tier (the file's rendezvous owner's cache, when wired), then stream from the inner store while teeing into the cache.

func (*BaseTableCache) GetReaderAt 

func (c *BaseTableCache) GetReaderAt(ctx context.Context, bucket, key string) (ReaderAtCloser, int64, error)

GetReaderAt implements ReaderAtStore. Hits serve column-chunk range reads as local preads; misses pass through WITHOUT populating (ranged misses are footer-sized — the whole-file Get on every scan path is the populator).

func (*BaseTableCache) HasCachedPath 

func (c *BaseTableCache) HasCachedPath(bucket, key string) bool

HasCachedPath implements LocalPathStore.

func (*BaseTableCache) Head 

func (c *BaseTableCache) Head(ctx context.Context, bucket, key string) (ObjectInfo, error)

Head implements Store.

func (*BaseTableCache) List 

func (c *BaseTableCache) List(ctx context.Context, bucket string, opts ListOptions) ([]ObjectInfo, error)

List implements Store.

func (*BaseTableCache) LogStats 

func (c *BaseTableCache) LogStats()

LogStats emits the greppable stats marker (design memo §8).

func (*BaseTableCache) MakeBucket 

func (c *BaseTableCache) MakeBucket(ctx context.Context, bucket string) error

MakeBucket implements Store.

func (*BaseTableCache) PeerLocalPath 

func (c *BaseTableCache) PeerLocalPath(bucket, key string) (string, bool)

PeerLocalPath resolves a resident entry for a peer's fetch, counting it as a peer serve (never a local hit — served bytes leave on the wire and must not inflate the hit ledger). The worker's ShuffleFileResolver base-table branch is the only caller.

func (*BaseTableCache) Put 

func (c *BaseTableCache) Put(ctx context.Context, bucket, key string, r io.Reader, size int64, contentType string) (string, error)

Put implements Store. Eligible keys invalidate defensively — ingest never rewrites a live key (memo §2), but a stale entry after any out-of-contract overwrite would silently serve old bytes.

func (*BaseTableCache) PutIfMatch 

func (c *BaseTableCache) PutIfMatch(ctx context.Context, bucket, key string, r io.Reader, size int64, contentType string, expectedETag string) (string, error)

PutIfMatch implements Store.

func (*BaseTableCache) ReadThrough 

func (c *BaseTableCache) ReadThrough(ctx context.Context, bucket, key string) error

ReadThrough ensures bucket/key is resident, fetching it from the inner store if needed (docs/design/scan-affinity.md §first-touch single-flight). The worker's peer resolver is the only caller: a peer asked this worker — the file's rendezvous owner — for a copy it doesn't hold yet, and fetching it HERE (once) instead of bouncing the peer to S3 is what keeps cluster-wide first-touch at one S3 read per file.

Concurrent callers for one key coalesce onto a single fetch. The fetch runs detached with its own timeout — a canceled waiter neither strands other waiters nor aborts a populate whose bytes this node will want anyway. An error return means the caller should fall back to NotFound-style behavior (the peer goes durable); the cache is unchanged.

func (*BaseTableCache) SetPeerFetcher 

func (c *BaseTableCache) SetPeerFetcher(f BaseTablePeerFetcher)

SetPeerFetcher wires the peer tier into the miss path. Must be called before the cache serves traffic (worker wiring, pre-Start) — the field is read without synchronization on every Get.

func (*BaseTableCache) Stats 

func (c *BaseTableCache) Stats() BaseTableCacheStats

Stats returns a snapshot of the cache counters.

func (*BaseTableCache) StoreID 

func (c *BaseTableCache) StoreID() string

StoreID implements IdentifiedStore by delegating: this cache is a local materialization of inner's objects under the SAME (bucket, key) names, so it must share inner's namespace rather than mint its own.

func (*BaseTableCache) Unwrap 

func (c *BaseTableCache) Unwrap() Store

Unwrap returns the underlying store.

type BaseTableCacheStats 

type BaseTableCacheStats struct {
	Hits      int64
	Misses    int64
	Evictions int64
	HitBytes  int64
	MissBytes int64
	Entries   int
	Bytes     int64

	// Peer tier (docs/design/scan-affinity.md §peer tier). Consumer side:
	// misses served from the file's rendezvous owner instead of S3.
	// Server side: fetches this cache served to non-owner peers.
	// Misses/MissBytes above stay S3-only so the first-touch ledger keeps
	// measuring exactly the reads that left the cluster.
	PeerHits         int64
	PeerBytes        int64
	PeerFallthroughs int64
	PeerServes       int64
	PeerServeBytes   int64
	// PeerFetchNanos is the wall spent inside successful peer fetches
	// (owner stream + local spool + admit), so the per-minute ledger
	// yields an effective peer-tier MB/s — the number that tells a peer
	// transfer apart from an S3 miss when both show up as src_ms.
	PeerFetchNanos int64

	// Owner read-through (docs/design/scan-affinity.md §first-touch
	// single-flight): populates performed on behalf of a peer's fetch for
	// a not-yet-resident owned file. ReadThroughBytes land in the cache
	// and are S3 reads, but are counted separately from Misses/MissBytes —
	// those keep measuring local demand that left the cluster; these
	// measure demand a peer redirected here.
	ReadThroughs     int64
	ReadThroughBytes int64
	ReadThroughFails int64
}

BaseTableCacheStats is a point-in-time snapshot of cache counters.

type BaseTablePeerFetcher 

type BaseTablePeerFetcher interface {
	FetchBaseTable(ctx context.Context, bucket, key string) (io.ReadCloser, bool)
}

BaseTablePeerFetcher is the seam between the cache's miss path and the worker's peer tier. FetchBaseTable returns a whole-object stream from the file's rendezvous owner, or ok=false when the tier cannot help (this worker owns the file, no live owner is known, or the tier is disabled) — the miss then goes to the inner store exactly as before. A returned stream may still fail mid-read; the cache treats any error as a fallthrough to the inner store.

type CircuitConfig 

type CircuitConfig struct {
	FailureThreshold int           // consecutive failures before opening (default: 5)
	ResetTimeout     time.Duration // time in open state before trying half-open (default: 30s)
	HalfOpenMax      int           // max requests in half-open state (default: 1)
	RequestTimeout   time.Duration // per-request timeout (default: 10s)
}

CircuitConfig configures the circuit breaker behavior.

func DefaultCircuitConfig 

func DefaultCircuitConfig() CircuitConfig

DefaultCircuitConfig returns sensible defaults.

type CircuitState 

type CircuitState int

CircuitState represents the current state of the circuit breaker. 

const (
	CircuitClosed   CircuitState = iota // normal operation
	CircuitOpen                         // fast-fail, no requests sent
	CircuitHalfOpen                     // testing with limited requests
)

func (CircuitState) String 

func (s CircuitState) String() string

type CircuitStore 

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

CircuitStore wraps a Store with circuit breaker protection. When consecutive failures on an operation class exceed the threshold that class's circuit opens and its requests immediately return ErrCircuitOpen. After a reset timeout the class moves to half-open and allows a probe request through. The other classes are unaffected.

func FindCircuitStore added in v0.18.17 

func FindCircuitStore(s Store) *CircuitStore

FindCircuitStore walks a Store's Unwrap() chain and returns the first CircuitStore in it, so a caller holding the outermost wrapper (the base-table NVMe cache sits ABOVE the breaker) can still reach the breaker to attach metrics. Returns nil when the chain holds none.

func NewCircuitStore 

func NewCircuitStore(inner Store, cfg CircuitConfig, logger *slog.Logger) *CircuitStore

NewCircuitStore wraps an existing store with circuit breaker protection.

func (*CircuitStore) BucketExists 

func (cs *CircuitStore) BucketExists(ctx context.Context, bucket string) (bool, error)

BucketExists implements Store.

func (*CircuitStore) Config added in v0.18.17 

func (cs *CircuitStore) Config() CircuitConfig

Config returns the breaker's effective configuration (after defaulting).

func (*CircuitStore) Delete 

func (cs *CircuitStore) Delete(ctx context.Context, bucket, key string) error

Delete implements Store. Deletes are off the critical path (scratch reclamation, compaction's post-grace sweep, DROP TABLE reclaim) and count into their own breaker: a slow delete burst must never fast-fail a read.

func (*CircuitStore) Get 

func (cs *CircuitStore) Get(ctx context.Context, bucket, key string) (io.ReadCloser, ObjectInfo, error)

Get implements Store. Get cannot use the standard do() wrapper because do() defers cancel() on the timeout context. Since Get returns a streaming ReadCloser, canceling the context before the caller reads the body kills the HTTP connection, causing every io.ReadAll(rc) to fail with "context canceled".

func (*CircuitStore) GetReaderAt 

func (cs *CircuitStore) GetReaderAt(ctx context.Context, bucket, key string) (ReaderAtCloser, int64, error)

GetReaderAt implements ReaderAtStore if the underlying store supports it. Like Get, this cannot use do() because it returns a streaming handle.

func (*CircuitStore) Head 

func (cs *CircuitStore) Head(ctx context.Context, bucket, key string) (ObjectInfo, error)

Head implements Store.

func (*CircuitStore) List 

func (cs *CircuitStore) List(ctx context.Context, bucket string, opts ListOptions) ([]ObjectInfo, error)

List implements Store.

func (*CircuitStore) MakeBucket 

func (cs *CircuitStore) MakeBucket(ctx context.Context, bucket string) error

MakeBucket implements Store.

func (*CircuitStore) OpenedTotal added in v0.18.17 

func (cs *CircuitStore) OpenedTotal(class OpClass) uint64

OpenedTotal returns how many times a class's breaker has opened.

func (*CircuitStore) Put 

func (cs *CircuitStore) Put(ctx context.Context, bucket, key string, r io.Reader, size int64, contentType string) (string, error)

Put implements Store. PUT operations are not wrapped in a per-call timeout — the inner store's transport-level ResponseHeaderTimeout (30 min) plus the caller's ctx already bound upload time. A short per-call timeout caused multi-process clusters to fail healthy uploads when contended bandwidth dropped per-connection throughput below the previous 20 MB/s sizing assumption.

The circuit breaker's failure counting still applies — repeated PUT errors trip the WRITE breaker — but a slow-yet-progressing upload no longer triggers a context deadline mid-stream, and a tripped write breaker never fast-fails a read.

func (*CircuitStore) PutIfMatch 

func (cs *CircuitStore) PutIfMatch(ctx context.Context, bucket, key string, r io.Reader, size int64, contentType string, expectedETag string) (string, error)

PutIfMatch implements Store. See Put for why we don't wrap in a per-call timeout.

func (*CircuitStore) SetOnOpen added in v0.18.17 

func (cs *CircuitStore) SetOnOpen(fn func(OpClass))

SetOnOpen registers a callback invoked (with the breaker's lock released) each time an operation class transitions into the open state. It drives the wadjet_circuit_breaker_opened_total{class} counter.

func (*CircuitStore) State 

func (cs *CircuitStore) State() CircuitState

State returns the worst state across the operation classes: open if any class is open, half-open if any is half-open, otherwise closed. Callers that care about one class must use StateFor — "is the read path being fast-failed" is a question only StateFor(OpRead) answers.

func (*CircuitStore) StateFor added in v0.18.17 

func (cs *CircuitStore) StateFor(class OpClass) CircuitState

StateFor returns the current state of one operation class.

func (*CircuitStore) StoreID 

func (cs *CircuitStore) StoreID() string

StoreID implements IdentifiedStore by delegating: the breaker changes availability, never object identity.

func (*CircuitStore) Unwrap 

func (cs *CircuitStore) Unwrap() Store

Unwrap returns the underlying store.

type FileStore 

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

FileStore is a local filesystem implementation of Store. Designed for edge deployments where S3 is unavailable.

Layout: 

<rootDir>/<bucket>/<key>

Keys may contain "/" which maps to subdirectories. ETags are computed as MD5 hex of file contents.

func NewFileStore 

func NewFileStore(rootDir string) (*FileStore, error)

NewFileStore creates a FileStore rooted at the given directory. The directory is created if it does not exist.

func (*FileStore) BucketExists 

func (f *FileStore) BucketExists(_ context.Context, bucket string) (bool, error)

func (*FileStore) Delete 

func (f *FileStore) Delete(_ context.Context, bucket, key string) error

func (*FileStore) Get 

func (f *FileStore) Get(_ context.Context, bucket, key string) (io.ReadCloser, ObjectInfo, error)

func (*FileStore) GetReaderAt 

func (f *FileStore) GetReaderAt(_ context.Context, bucket, key string) (ReaderAtCloser, int64, error)

func (*FileStore) Head 

func (f *FileStore) Head(_ context.Context, bucket, key string) (ObjectInfo, error)

func (*FileStore) List 

func (f *FileStore) List(_ context.Context, bucket string, opts ListOptions) ([]ObjectInfo, error)

func (*FileStore) MakeBucket 

func (f *FileStore) MakeBucket(_ context.Context, bucket string) error

func (*FileStore) Put 

func (f *FileStore) Put(_ context.Context, bucket, key string, r io.Reader, _ int64, contentType string) (string, error)

func (*FileStore) PutIfMatch 

func (f *FileStore) PutIfMatch(_ context.Context, bucket, key string, r io.Reader, _ int64, contentType string, expectedETag string) (string, error)

func (*FileStore) StoreID 

func (f *FileStore) StoreID() string

StoreID implements IdentifiedStore.

type HTTPConfig 

type HTTPConfig struct {
	Headers map[string]string // auth headers, API keys, etc.
	Timeout time.Duration
}

HTTPConfig holds configuration for the HTTP object store.

type HTTPStore 

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

HTTPStore implements a read-only Store backed by HTTP/HTTPS URLs. The "bucket" parameter is the base URL (scheme + host, e.g. "https://example.com") and "key" is the path component appended after a slash.

func NewHTTPStore 

func NewHTTPStore(cfg HTTPConfig) *HTTPStore

NewHTTPStore creates a read-only HTTP-backed object store with connection pooling.

func (*HTTPStore) BucketExists 

func (s *HTTPStore) BucketExists(ctx context.Context, bucket string) (bool, error)

BucketExists checks whether the base URL is reachable with a HEAD request.

func (*HTTPStore) Delete 

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

Delete is not supported on a read-only HTTP store.

func (*HTTPStore) Get 

func (s *HTTPStore) Get(ctx context.Context, bucket, key string) (io.ReadCloser, ObjectInfo, error)

Get retrieves the object body and metadata via HTTP GET.

func (*HTTPStore) GetReaderAt 

func (s *HTTPStore) GetReaderAt(ctx context.Context, bucket, key string) (ReaderAtCloser, int64, error)

GetReaderAt returns a ReaderAtCloser that uses HTTP Range requests for random-access reads. This enables efficient Parquet column pruning over HTTP.

func (*HTTPStore) Head 

func (s *HTTPStore) Head(ctx context.Context, bucket, key string) (ObjectInfo, error)

Head retrieves object metadata via HTTP HEAD without downloading the body.

func (*HTTPStore) List 

func (s *HTTPStore) List(ctx context.Context, bucket string, opts ListOptions) ([]ObjectInfo, error)

List is not meaningfully supported over plain HTTP. It returns an empty result if the base URL is reachable, or an error otherwise.

func (*HTTPStore) MakeBucket 

func (s *HTTPStore) MakeBucket(ctx context.Context, bucket string) error

MakeBucket is not supported on a read-only HTTP store.

func (*HTTPStore) Put 

func (s *HTTPStore) Put(ctx context.Context, bucket, key string, r io.Reader, size int64, contentType string) (string, error)

Put is not supported on a read-only HTTP store.

func (*HTTPStore) PutIfMatch 

func (s *HTTPStore) PutIfMatch(ctx context.Context, bucket, key string, r io.Reader, size int64, contentType string, expectedETag string) (string, error)

PutIfMatch is not supported on a read-only HTTP store.

type IdentifiedStore 

type IdentifiedStore interface {
	StoreID() string
}

IdentifiedStore is the optional interface a Store implements when it can name its backing INSTANCE uniquely within this process. It exists for process-lifetime caches keyed by object identity (the parquet footer cache): a cache key must not collide across two unrelated stores that happen to use the same bucket and key names.

The contract a store accepts by implementing this:

  • StoreID is stable for the store's lifetime and never reused by a different backing store within the process.
  • Two Store values that address the SAME backing store (e.g. two FileStores rooted at one directory, or a wrapper and its inner store) may — and should — return the same ID.
  • Under a given StoreID, a (bucket, key) pair names one immutable object: writers give new content a new key rather than rewriting a key in place. That is Wadjet's manifest discipline — ingest names chunks with a fresh UUID (ingest/ingest.go:300); compaction and delete-marker GC write new nanosecond-stamped keys and swap the manifest (compaction/compactor.go:50, catalog.SwapFileForGC) — and it is the same premise the base-table NVMe cache already relies on.

A store that cannot honour those (arbitrary remote URLs, mutable third-party buckets) simply does not implement the interface, and identity-keyed caches fail closed on it.

type ListOptions 

type ListOptions struct {
	Prefix    string
	Delimiter string
	MaxKeys   int
}

ListOptions configures a list operation.

type LocalPathStore 

type LocalPathStore interface {
	// CachedLocalPath returns the local file for a resident object. It
	// counts as a cache hit — call it only to serve.
	CachedLocalPath(bucket, key string) (string, bool)
	// HasCachedPath is the counting-free membership probe, for callers
	// deciding whether to skip work (e.g. a prefetcher electing not to
	// download) without inflating hit statistics.
	HasCachedPath(bucket, key string) bool
}

LocalPathStore is an optional interface for stores that hold whole objects on local disk. Callers that would otherwise stream a copy to their own scratch file (e.g. the worker's parquet mmap path) can open the store's file in place. The path is best-effort: eviction may unlink it between the call and the open — POSIX keeps the inode alive for already-open descriptors, so callers open first, then treat an open failure as a miss and fall back to Get.

type MemStore 

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

MemStore is an in-memory implementation of Store for testing.

func NewMemStore 

func NewMemStore() *MemStore

NewMemStore creates a new in-memory object store.

func (*MemStore) BucketExists 

func (m *MemStore) BucketExists(_ context.Context, bucket string) (bool, error)

func (*MemStore) Delete 

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

func (*MemStore) Get 

func (m *MemStore) Get(_ context.Context, bucket, key string) (io.ReadCloser, ObjectInfo, error)

func (*MemStore) GetReaderAt 

func (m *MemStore) GetReaderAt(_ context.Context, bucket, key string) (ReaderAtCloser, int64, error)

func (*MemStore) Head 

func (m *MemStore) Head(_ context.Context, bucket, key string) (ObjectInfo, error)

func (*MemStore) List 

func (m *MemStore) List(_ context.Context, bucket string, opts ListOptions) ([]ObjectInfo, error)

func (*MemStore) MakeBucket 

func (m *MemStore) MakeBucket(_ context.Context, bucket string) error

func (*MemStore) Put 

func (m *MemStore) Put(_ context.Context, bucket, key string, r io.Reader, _ int64, contentType string) (string, error)

func (*MemStore) PutIfMatch 

func (m *MemStore) PutIfMatch(_ context.Context, bucket, key string, r io.Reader, _ int64, contentType string, expectedETag string) (string, error)

func (*MemStore) StoreID 

func (m *MemStore) StoreID() string

StoreID implements IdentifiedStore. Every MemStore gets its own namespace: unlike a durable store there is no external name to key on, and separate tests in one binary reuse bucket/object names freely for different content.

type MinIOConfig 

type MinIOConfig struct {
	Endpoint  string
	AccessKey string
	SecretKey string
	UseSSL    bool
	Region    string

	// MaxConcurrentUploads bounds the number of in-flight PUT operations
	// originating from a single MinIOStore instance. Defaults to 4 when
	// zero. Each PUT for a >16MB object triggers minio-go's multipart
	// uploader which itself opens multiple TCP connections per object
	// (UploadThreads, default 2 here). With multiple worker processes
	// on the same host, the aggregate connection count needs a ceiling
	// so individual uploads aren't starved of bandwidth.
	MaxConcurrentUploads int

	// UploadThreads is the per-PUT multipart concurrency for objects
	// large enough to trigger multipart upload (>16MB). Defaults to 2.
	// Lower values reduce per-upload connection count; higher values
	// speed up individual uploads when bandwidth is plentiful. With 4
	// processes per node and MaxConcurrentUploads=4, this caps host
	// connection count at 4×4×2 = 32.
	UploadThreads int
}

MinIOConfig holds configuration for connecting to a MinIO/S3 endpoint.

type MinIOStore 

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

MinIOStore implements Store using minio-go. Includes a per-instance upload semaphore that bounds aggregate PUT concurrency so simultaneous uploads from a single process don't fragment available bandwidth.

func NewMinIOStore 

func NewMinIOStore(cfg MinIOConfig) (*MinIOStore, error)

NewMinIOStore creates a new MinIO-backed object store with connection pooling. When AccessKey and SecretKey are both empty, credentials are auto-detected from environment variables (AWS_ACCESS_KEY_ID) and IAM instance profiles.

func (*MinIOStore) BucketExists 

func (s *MinIOStore) BucketExists(ctx context.Context, bucket string) (bool, error)

func (*MinIOStore) Delete 

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

func (*MinIOStore) Get 

func (s *MinIOStore) Get(ctx context.Context, bucket, key string) (io.ReadCloser, ObjectInfo, error)

func (*MinIOStore) GetReaderAt 

func (s *MinIOStore) GetReaderAt(ctx context.Context, bucket, key string) (ReaderAtCloser, int64, error)

func (*MinIOStore) Head 

func (s *MinIOStore) Head(ctx context.Context, bucket, key string) (ObjectInfo, error)

func (*MinIOStore) List 

func (s *MinIOStore) List(ctx context.Context, bucket string, opts ListOptions) ([]ObjectInfo, error)

func (*MinIOStore) MakeBucket 

func (s *MinIOStore) MakeBucket(ctx context.Context, bucket string) error

func (*MinIOStore) Put 

func (s *MinIOStore) Put(ctx context.Context, bucket, key string, r io.Reader, size int64, contentType string) (string, error)

func (*MinIOStore) PutIfMatch 

func (s *MinIOStore) PutIfMatch(ctx context.Context, bucket, key string, r io.Reader, size int64, contentType string, expectedETag string) (string, error)

func (*MinIOStore) StoreID 

func (s *MinIOStore) StoreID() string

StoreID implements IdentifiedStore: the endpoint (plus region) names the backing S3 service, so two MinIOStore values pointed at one endpoint share a namespace, as they should — the objects are the same objects.

type ObjectInfo 

type ObjectInfo struct {
	Key          string
	Size         int64
	ETag         string
	LastModified time.Time
	ContentType  string
}

ObjectInfo contains metadata about a stored object.

type OpClass added in v0.18.17 

type OpClass int

OpClass is the operation class a breaker counts failures for.

The breaker is scoped BY CLASS because the failures it must react to and the requests it must protect are not the same traffic. A query READS to answer; it WRITES stage output and DELETES scratch off the critical path. Three times now (compaction's post-grace deletes, SF100 upload cancels, streaming-fallback 404s) a burst of failures on an off-critical-path operation opened one process-wide breaker and fast-failed healthy base-table reads; each fix excluded one more error class and the defect came back in the next one. The invariant, stated once instead of enumerated per error class: a failure on a non-read, off-critical-path operation never fast-fails the read path (ADR-0028). 

const (
	OpRead   OpClass = iota // Get, GetReaderAt, Head, List, BucketExists
	OpWrite                 // Put, PutIfMatch, MakeBucket
	OpDelete                // Delete

)

func (OpClass) String added in v0.18.17 

func (c OpClass) String() string

type ReaderAtCloser 

type ReaderAtCloser interface {
	io.ReaderAt
	io.Closer
}

ReaderAtCloser combines io.ReaderAt with io.Closer for random-access reads.

type ReaderAtStore 

type ReaderAtStore interface {
	GetReaderAt(ctx context.Context, bucket, key string) (ReaderAtCloser, int64, error)
}

ReaderAtStore is an optional interface that Store implementations can provide to support random-access reads. This enables Parquet column pruning by reading only needed column chunks instead of downloading entire files.

type Store 

type Store interface {
	// Put writes an object to storage. Returns the ETag of the written object.
	Put(ctx context.Context, bucket, key string, r io.Reader, size int64, contentType string) (etag string, err error)

	// PutIfMatch writes an object only if the current ETag matches expectedETag.
	// If expectedETag is empty, the object must not exist (create-only).
	// Returns the new ETag or ErrPreconditionFailed.
	PutIfMatch(ctx context.Context, bucket, key string, r io.Reader, size int64, contentType string, expectedETag string) (etag string, err error)

	// Get retrieves an object from storage.
	Get(ctx context.Context, bucket, key string) (io.ReadCloser, ObjectInfo, error)

	// Head retrieves metadata without downloading the object body.
	Head(ctx context.Context, bucket, key string) (ObjectInfo, error)

	// List returns objects matching the given options.
	List(ctx context.Context, bucket string, opts ListOptions) ([]ObjectInfo, error)

	// Delete removes an object from storage.
	Delete(ctx context.Context, bucket, key string) error

	// BucketExists checks whether a bucket exists.
	BucketExists(ctx context.Context, bucket string) (bool, error)

	// MakeBucket creates a bucket if it does not exist.
	MakeBucket(ctx context.Context, bucket string) error
}

Store defines the interface for S3-compatible object storage operations.

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.