storage

Documentation

Index

• Constants
• func GetS3Config(ctx context.Context) (aws.Config, error)
• type ConsistencyChecker
  • func NewConsistencyChecker(client client.Client, s3Client *s3.Client) *ConsistencyChecker
  • func (c *ConsistencyChecker) ListParquetFiles(ctx context.Context, table *ducklakev1alpha1.DuckLakeTable) ([]types.Object, error)
  • func (c *ConsistencyChecker) VerifyParquetConsistency(ctx context.Context, table *ducklakev1alpha1.DuckLakeTable) error
  • func (c *ConsistencyChecker) WaitForParquetFile(ctx context.Context, table *ducklakev1alpha1.DuckLakeTable, filename string) error
  • func (c *ConsistencyChecker) WaitForParquetFiles(ctx context.Context, table *ducklakev1alpha1.DuckLakeTable, filenames []string) error
  • func (c *ConsistencyChecker) WithInterval(interval time.Duration) *ConsistencyChecker
  • func (c *ConsistencyChecker) WithRetries(retries int) *ConsistencyChecker
  • func (c *ConsistencyChecker) WithTimeout(timeout time.Duration) *ConsistencyChecker
• type ObjectStore
  • func NewObjectStore(s3Client *s3.Client) ObjectStore
• type S3Client
  • func NewS3Client(ctx context.Context, cfg S3Config) (S3Client, error)
• type S3Config

Constants

const (
	// DefaultTimeout is the default timeout for consistency checks
	DefaultTimeout = 5 * time.Minute

	// DefaultInterval is the default interval between consistency checks
	DefaultInterval = 5 * time.Second

	// DefaultRetries is the default number of retries for consistency checks
	DefaultRetries = 3
)

Functions

func GetS3Config

func GetS3Config(ctx context.Context) (aws.Config, error)

GetS3Config returns the AWS configuration for S3 operations

Types

type ConsistencyChecker

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

ConsistencyChecker handles S3 consistency checks for Parquet files

func NewConsistencyChecker

func NewConsistencyChecker(client client.Client, s3Client *s3.Client) *ConsistencyChecker

NewConsistencyChecker creates a new consistency checker

func (*ConsistencyChecker) ListParquetFiles

func (c *ConsistencyChecker) ListParquetFiles(ctx context.Context, table *ducklakev1alpha1.DuckLakeTable) ([]types.Object, error)

ListParquetFiles lists all Parquet files in a table's location

func (*ConsistencyChecker) VerifyParquetConsistency

func (c *ConsistencyChecker) VerifyParquetConsistency(ctx context.Context, table *ducklakev1alpha1.DuckLakeTable) error

VerifyParquetConsistency verifies the consistency of Parquet files

func (*ConsistencyChecker) WaitForParquetFile

func (c *ConsistencyChecker) WaitForParquetFile(ctx context.Context, table *ducklakev1alpha1.DuckLakeTable, filename string) error

WaitForParquetFile waits for a Parquet file to be available in S3

func (*ConsistencyChecker) WaitForParquetFiles

func (c *ConsistencyChecker) WaitForParquetFiles(ctx context.Context, table *ducklakev1alpha1.DuckLakeTable, filenames []string) error

WaitForParquetFiles waits for multiple Parquet files to be available in S3

func (*ConsistencyChecker) WithInterval

func (c *ConsistencyChecker) WithInterval(interval time.Duration) *ConsistencyChecker

WithInterval sets the interval between consistency checks

func (*ConsistencyChecker) WithRetries

func (c *ConsistencyChecker) WithRetries(retries int) *ConsistencyChecker

WithRetries sets the number of retries for consistency checks

func (*ConsistencyChecker) WithTimeout

func (c *ConsistencyChecker) WithTimeout(timeout time.Duration) *ConsistencyChecker

WithTimeout sets the timeout for consistency checks

type ObjectStore

type ObjectStore interface {
	// ListBuckets lists all buckets
	ListBuckets(ctx context.Context) ([]string, error)
	// CreateBucket creates a new bucket
	CreateBucket(ctx context.Context, bucket string) error
	// DeleteBucket deletes a bucket
	DeleteBucket(ctx context.Context, bucket string) error
	// ListObjects lists objects in a bucket with optional prefix
	ListObjects(ctx context.Context, bucket, prefix string) ([]string, error)
	// GetObject gets an object from a bucket
	GetObject(ctx context.Context, bucket, key string) ([]byte, error)
	// PutObject puts an object in a bucket
	PutObject(ctx context.Context, bucket, key string, data []byte) error
	// DeleteObject deletes an object from a bucket
	DeleteObject(ctx context.Context, bucket, key string) error
}

ObjectStore defines the interface for object storage operations

func NewObjectStore

func NewObjectStore(s3Client *s3.Client) ObjectStore

NewObjectStore creates a new ObjectStore implementation from an S3 client

type S3Client

type S3Client interface {
	// UploadParquet uploads a Parquet file to S3
	UploadParquet(ctx context.Context, key string, reader io.Reader) error
	// DownloadParquet downloads a Parquet file from S3
	DownloadParquet(ctx context.Context, key string) (io.ReadCloser, error)
	// ListParquetFiles lists Parquet files under a prefix
	ListParquetFiles(ctx context.Context, prefix string) ([]string, error)
	// DeleteParquetFiles deletes Parquet files under a prefix
	DeleteParquetFiles(ctx context.Context, prefix string) error
	// HeadParquet checks if a Parquet file exists and gets its metadata
	HeadParquet(ctx context.Context, key string) (*s3.HeadObjectOutput, error)
}

S3Client provides operations for S3-compatible storage

func NewS3Client

func NewS3Client(ctx context.Context, cfg S3Config) (S3Client, error)

NewS3Client creates a new S3Client instance

type S3Config

type S3Config struct {
	// Endpoint is the S3-compatible endpoint URL
	Endpoint string
	// Region is the S3 region
	Region string
	// Bucket is the S3 bucket name
	Bucket string
	// AccessKeyID for authentication
	AccessKeyID string
	// SecretAccessKey for authentication
	SecretAccessKey string
}

S3Config represents the configuration for S3 storage