compaction

Documentation

Index

• Variables
• type Compactor
  • func (c *Compactor) RunOnce(ctx context.Context) error
• type CopierAdapter
  • func (c *CopierAdapter) CopyObject(ctx context.Context, srcKey, dstKey string) error
  • func (c *CopierAdapter) DeleteObject(ctx context.Context, key string) error
• type FileProvider
• type KeyedCopier
• type ManifestProvider
  • func NewFSManifestProvider(cfg cdc.ManifestConfig, rootFS fs.FS) *ManifestProvider
  • func NewManifestProvider(cfg cdc.ManifestConfig, store manifest.Store) *ManifestProvider
  • func NewS3ManifestProvider(cfg cdc.ManifestConfig, s3Client *s3.Client) *ManifestProvider
  • func (p *ManifestProvider) LoadManifest(ctx context.Context, schemaID int16) (*manifest.Manifest, string, error)
  • func (p *ManifestProvider) SaveManifest(ctx context.Context, schemaID int16, m *manifest.Manifest, etag string) (string, error)
• type S3Copier

Variables

var ErrConcurrentModification = errors.New("manifest concurrently modified")

ErrConcurrentModification indicates manifest was modified by another process.

Types

type Compactor

type Compactor struct {
	Logger     *zap.Logger
	Config     cdc.CompactionConfig
	Provider   FileProvider
	Copier     S3Copier
	Bucket     string
	DataPrefix string // root prefix for parquet files
	Resolver   manifest.PathResolver
}

Compactor performs Base/Delta maintenance per schema.

func (*Compactor) RunOnce

func (c *Compactor) RunOnce(ctx context.Context) error

RunOnce executes compaction for a schema (or all if Config.SchemaID==0).

type CopierAdapter

type CopierAdapter struct {
	Impl KeyedCopier
}

CopierAdapter adapts KeyedCopier to S3Copier.

func (*CopierAdapter) CopyObject

func (c *CopierAdapter) CopyObject(ctx context.Context, srcKey, dstKey string) error

func (*CopierAdapter) DeleteObject

func (c *CopierAdapter) DeleteObject(ctx context.Context, key string) error

type FileProvider

type FileProvider interface {
	LoadManifest(ctx context.Context, schemaID int16) (*manifest.Manifest, string, error)
	SaveManifest(ctx context.Context, schemaID int16, m *manifest.Manifest, etag string) (string, error)
}

FileProvider fetches manifest and lists actual files (optionally cross-check).

type KeyedCopier

type KeyedCopier interface {
	Copy(ctx context.Context, srcKey, dstKey string) error
	Delete(ctx context.Context, key string) error
}

KeyedCopier defines minimal copy/delete operations (for S3 or compatible storage).

type ManifestProvider

type ManifestProvider struct {
	Store    manifest.Store
	Resolver manifest.PathResolver
}

ManifestProvider adapts manifest.Store + resolver to FileProvider.

func NewFSManifestProvider

func NewFSManifestProvider(cfg cdc.ManifestConfig, rootFS fs.FS) *ManifestProvider

NewFSManifestProvider creates a ManifestProvider backed by local filesystem. Useful for testing and local development.

func NewManifestProvider

func NewManifestProvider(cfg cdc.ManifestConfig, store manifest.Store) *ManifestProvider

NewManifestProvider creates a ManifestProvider from a ManifestConfig and Store. Use this to wire up the provider from configuration.

func NewS3ManifestProvider

func NewS3ManifestProvider(cfg cdc.ManifestConfig, s3Client *s3.Client) *ManifestProvider

NewS3ManifestProvider creates a ManifestProvider backed by S3.

func (*ManifestProvider) LoadManifest

func (p *ManifestProvider) LoadManifest(ctx context.Context, schemaID int16) (*manifest.Manifest, string, error)

func (*ManifestProvider) SaveManifest

func (p *ManifestProvider) SaveManifest(ctx context.Context, schemaID int16, m *manifest.Manifest, etag string) (string, error)

type S3Copier

type S3Copier interface {
	CopyObject(ctx context.Context, srcKey, dstKey string) error
	DeleteObject(ctx context.Context, key string) error
}