lakehouse

package module
v0.0.0-...-11037b0 Latest Latest
Warning

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

 Go to latest Published: Jul 27, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrFieldNotFound = fmt.Errorf("field not found")
View Source 
var ErrSnapshotNotFound = fmt.Errorf("snapshot not found")

Functions

This section is empty.

Types

type BackupConfig 

type BackupConfig struct {
	Name    string
	Include []string // directories to include, empty = all
	Exclude []string // directories to exclude
}

type BackupManager 

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

func NewBackupManager 

func NewBackupManager(dir string, logger *slog.Logger) *BackupManager

func (*BackupManager) Backup 

func (b *BackupManager) Backup(ctx context.Context, config BackupConfig) (*BackupResult, error)

func (*BackupManager) DeleteBackup 

func (b *BackupManager) DeleteBackup(ctx context.Context, path string) error

func (*BackupManager) ListBackups 

func (b *BackupManager) ListBackups(ctx context.Context) ([]string, error)

func (*BackupManager) Restore 

func (b *BackupManager) Restore(ctx context.Context, backupPath string) (*RestoreResult, error)

type BackupResult 

type BackupResult struct {
	Path     string
	Size     int64
	Duration time.Duration
	Files    int
}

type CacheEntry 

type CacheEntry struct {
	Key       string
	Result    *query.Result
	CreatedAt time.Time
	Size      int64
}

type CatalogManager 

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

func NewCatalogManager 

func NewCatalogManager(cat catalog.Catalog, logger *slog.Logger) *CatalogManager

func (*CatalogManager) CreateNamespace 

func (m *CatalogManager) CreateNamespace(ctx context.Context, namespace string) error

func (*CatalogManager) DropNamespace 

func (m *CatalogManager) DropNamespace(ctx context.Context, namespace string) error

func (*CatalogManager) ListNamespaces 

func (m *CatalogManager) ListNamespaces(ctx context.Context) ([]string, error)

func (*CatalogManager) Underlying 

func (m *CatalogManager) Underlying() catalog.Catalog

type DataService 

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

func NewDataService 

func NewDataService(ic catalog.DataCatalog, q query.QueryEngine) *DataService

func (*DataService) Delete 

func (s *DataService) Delete(ctx context.Context, namespace, table string, filter map[string]interface{}) error

Delete removes rows matching the filter

func (*DataService) Ingest 

func (s *DataService) Ingest(ctx context.Context, namespace, table string, schema *catalog.Schema, data []map[string]interface{}) error

Ingest inserts data into an Iceberg table via iceberg-go Append

func (*DataService) Query 

func (s *DataService) Query(ctx context.Context, namespace, table, sql string, where, orderBy string, columns []string, limit, offset int) (*query.Result, error)

Query executes SQL via DuckDB iceberg_scan

func (*DataService) RefreshAllViews 

func (s *DataService) RefreshAllViews(ctx context.Context) error

RefreshAllViews creates DuckDB views for all existing tables in the catalog.

func (*DataService) RefreshView 

func (s *DataService) RefreshView(ctx context.Context, namespace, table string) error

RefreshView creates/updates DuckDB view for SQL compatibility

func (*DataService) Update 

func (s *DataService) Update(ctx context.Context, namespace, table string, data []map[string]interface{}, keys []string) error

Update performs a merge: delete matching rows, append new rows

type ExpireConfig 

type ExpireConfig struct {
	MaxAge   time.Duration // expire snapshots older than this
	MaxCount int           // keep only N most recent snapshots
}

type ExpireResult 

type ExpireResult struct {
	Expired   int
	Remaining int
}

type Lakehouse 

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

func New 

func New(opts ...Option) (*Lakehouse, error)

func (*Lakehouse) Catalog 

func (l *Lakehouse) Catalog() *CatalogManager

func (*Lakehouse) Close 

func (l *Lakehouse) Close() error

func (*Lakehouse) Data 

func (l *Lakehouse) Data() *DataService

func (*Lakehouse) IsEncrypted 

func (l *Lakehouse) IsEncrypted() bool

func (*Lakehouse) Lifecycle 

func (l *Lakehouse) Lifecycle() *lifecycle.Handler

func (*Lakehouse) Ping 

func (l *Lakehouse) Ping(ctx context.Context) error

func (*Lakehouse) Schema 

func (l *Lakehouse) Schema() *SchemaManager

func (*Lakehouse) ServeHTTP 

func (l *Lakehouse) ServeHTTP(w http.ResponseWriter, r *http.Request)

func (*Lakehouse) Snapshot 

func (l *Lakehouse) Snapshot() *SnapshotManager

func (*Lakehouse) Storage 

func (l *Lakehouse) Storage() *StorageManager

func (*Lakehouse) Table 

func (l *Lakehouse) Table() *TableManager

func (*Lakehouse) Transaction 

func (l *Lakehouse) Transaction() *TransactionManager

func (*Lakehouse) Warehouse 

func (l *Lakehouse) Warehouse() *warehouse.Warehouse

type Option 

type Option func(*Lakehouse)

func WithCatalog 

func WithCatalog(cat catalog.Catalog) Option

func WithDataCatalog 

func WithDataCatalog(dc catalog.DataCatalog) Option

func WithDir 

func WithDir(dir string) Option

func WithEncryption 

func WithEncryption(masterKey string) Option

func WithLogger 

func WithLogger(logger *slog.Logger) Option

func WithPrefix 

func WithPrefix(prefix string) Option

func WithQueryEngine 

func WithQueryEngine(q query.QueryEngine) Option

func WithRouter 

func WithRouter(r *http.ServeMux) Option

func WithStorage 

func WithStorage(st storage.Storage) Option

func WithWarehouse 

func WithWarehouse(wh *warehouse.Warehouse) Option

type QueryCache 

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

func NewQueryCache 

func NewQueryCache(maxSize int64, logger *slog.Logger) *QueryCache

func (*QueryCache) Count 

func (c *QueryCache) Count() int

func (*QueryCache) Get 

func (c *QueryCache) Get(sql string) (*query.Result, bool)

func (*QueryCache) Invalidate 

func (c *QueryCache) Invalidate(sql string)

func (*QueryCache) InvalidateAll 

func (c *QueryCache) InvalidateAll()

func (*QueryCache) Set 

func (c *QueryCache) Set(sql string, result *query.Result)

func (*QueryCache) Size 

func (c *QueryCache) Size() int64

type RestoreResult 

type RestoreResult struct {
	Path     string
	Duration time.Duration
	Files    int
}

type SchemaManager 

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

func NewSchemaManager 

func NewSchemaManager(cat catalog.Catalog, logger *slog.Logger) *SchemaManager

func (*SchemaManager) AddColumn 

func (m *SchemaManager) AddColumn(ctx context.Context, namespace, name string, field catalog.Field) error

func (*SchemaManager) AddColumns 

func (m *SchemaManager) AddColumns(ctx context.Context, namespace, name string, fields []catalog.Field) error

func (*SchemaManager) DropColumn 

func (m *SchemaManager) DropColumn(ctx context.Context, namespace, name, column string) error

func (*SchemaManager) FieldExists 

func (m *SchemaManager) FieldExists(ctx context.Context, namespace, name, field string) bool

func (*SchemaManager) Get 

func (m *SchemaManager) Get(ctx context.Context, namespace, name string) (*catalog.Schema, error)

func (*SchemaManager) GetField 

func (m *SchemaManager) GetField(ctx context.Context, namespace, name, field string) (*catalog.Field, error)

func (*SchemaManager) ListFields 

func (m *SchemaManager) ListFields(ctx context.Context, namespace, name string) ([]catalog.Field, error)

func (*SchemaManager) RenameColumn 

func (m *SchemaManager) RenameColumn(ctx context.Context, namespace, name, oldName, newName string) error

type SnapshotManager 

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

func NewSnapshotManager 

func NewSnapshotManager(cat catalog.Catalog, logger *slog.Logger) *SnapshotManager

func (*SnapshotManager) Count 

func (m *SnapshotManager) Count(ctx context.Context, namespace, name string) (int, error)

func (*SnapshotManager) Expire 

func (m *SnapshotManager) Expire(ctx context.Context, namespace, name string, config ExpireConfig) (*ExpireResult, error)

func (*SnapshotManager) GetByVersion 

func (m *SnapshotManager) GetByVersion(ctx context.Context, namespace, name string, version int64) (*catalog.Snapshot, error)

func (*SnapshotManager) GetLatest 

func (m *SnapshotManager) GetLatest(ctx context.Context, namespace, name string) (*catalog.Snapshot, error)

func (*SnapshotManager) List 

func (m *SnapshotManager) List(ctx context.Context, namespace, name string) ([]catalog.Snapshot, error)

type StorageManager 

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

func NewStorageManager 

func NewStorageManager(st storage.Storage, logger *slog.Logger) *StorageManager

func (*StorageManager) Delete 

func (m *StorageManager) Delete(ctx context.Context, path string) error

func (*StorageManager) List 

func (m *StorageManager) List(ctx context.Context, prefix string) ([]storage.Entry, error)

func (*StorageManager) Read 

func (m *StorageManager) Read(ctx context.Context, path string) ([]byte, error)

func (*StorageManager) Stat 

func (m *StorageManager) Stat(ctx context.Context, path string) (*storage.Entry, error)

func (*StorageManager) Write 

func (m *StorageManager) Write(ctx context.Context, path string, data []byte) error

type TableManager 

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

func NewTableManager 

func NewTableManager(cat catalog.Catalog, wh *warehouse.Warehouse, logger *slog.Logger) *TableManager

func (*TableManager) Commit 

func (m *TableManager) Commit(ctx context.Context, namespace, name string, updates []catalog.Update) error

func (*TableManager) Create 

func (m *TableManager) Create(ctx context.Context, namespace, name string, schema *catalog.Schema) (*catalog.Table, error)

func (*TableManager) Drop 

func (m *TableManager) Drop(ctx context.Context, namespace, name string) error

func (*TableManager) Exists 

func (m *TableManager) Exists(ctx context.Context, namespace, name string) bool

func (*TableManager) GetDataFiles 

func (m *TableManager) GetDataFiles(ctx context.Context, namespace, name string) ([]catalog.DataFile, error)

func (*TableManager) GetDataFilesAtVersion 

func (m *TableManager) GetDataFilesAtVersion(ctx context.Context, namespace, name string, version int64) ([]catalog.DataFile, error)

func (*TableManager) GetSchema 

func (m *TableManager) GetSchema(ctx context.Context, namespace, name string) (*catalog.Schema, error)

func (*TableManager) List 

func (m *TableManager) List(ctx context.Context, namespace string) ([]string, error)

func (*TableManager) Load 

func (m *TableManager) Load(ctx context.Context, namespace, name string) (*catalog.Table, error)

func (*TableManager) LoadAtVersion 

func (m *TableManager) LoadAtVersion(ctx context.Context, namespace, name string, version int64) (*catalog.Table, error)

func (*TableManager) WriteData 

func (m *TableManager) WriteData(ctx context.Context, namespace, name string, sql string) (*catalog.DataFile, error)

type TransactionManager 

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

func NewTransactionManager 

func NewTransactionManager(cat catalog.Catalog, logger *slog.Logger) *TransactionManager

func (*TransactionManager) Begin 

func (m *TransactionManager) Begin() *Tx

func (*TransactionManager) CommitTables 

func (m *TransactionManager) CommitTables(ctx context.Context, ops []catalog.TableOp) error

type Tx 

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

func (*Tx) Commit 

func (tx *Tx) Commit(ctx context.Context) error

func (*Tx) CreateTable 

func (tx *Tx) CreateTable(namespace, name string, schema *catalog.Schema)

func (*Tx) DropTable 

func (tx *Tx) DropTable(namespace, name string)

func (*Tx) Operations 

func (tx *Tx) Operations() []catalog.TableOp

func (*Tx) Rollback 

func (tx *Tx) Rollback()

Source Files

View all Source files

Directories

Path Synopsis
iceberg
arrowutil
duckdb
filesystem

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.
⌂ 换网址