catalog

package
v0.0.0-...-bace2a7 Latest Latest
Warning

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

 Go to latest Published: Aug 12, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// PropTableType is the primary format marker, e.g. "ICEBERG" or "DELTA".
	PropTableType = "table_type"
	// PropSparkSQLSourcesProvider is Spark's data-source provider, used as a fallback format marker.
	// Delta tables registered by Spark carry "delta" here and often nothing in table_type.
	PropSparkSQLSourcesProvider = "spark.sql.sources.provider"
	// PropWriteDataLocation and the two below are the Iceberg table properties that relocate data
	// away from <basePath>/data, checked in this order.
	PropWriteDataLocation          = "write.data.path"
	PropWriteFolderStorageLocation = "write.folder-storage.path"
	PropObjectStorePath            = "write.object-storage.path"
)

Table property keys used to describe a table registered in an external catalog. These mirror the keys Java XTable reads and the ones GlueCatalogSyncClient writes, so a table registered by either implementation resolves identically.

View Source 
const DefaultMaxPartitionsPerRequest = 1000

DefaultMaxPartitionsPerRequest bounds how many partitions are sent per catalog call. It mirrors Java's HMSCatalogConfig.maxPartitionsPerRequest. Catalog-specific hard limits are enforced by the implementations themselves, which may batch more finely than this.

Variables

View Source 
var ErrCatalogNotImplemented = errors.New("catalog type recognised but not implemented")

ErrCatalogNotImplemented is returned for a catalog type that is recognised but has no client in this repository. Hive Metastore is the only such type today; see docs/improvement-plan.md T13 for the scope a real implementation would need.

Functions

func DataLocationForFormat 

func DataLocationForFormat(format model.TableFormat, basePath string, properties map[string]string) (string, error)

DataLocationForFormat resolves where a table's data files live. Delta and Hudi keep data under the table location; Iceberg may relocate it, defaulting to <basePath>/data.

func ModelTypeToGlueType 

func ModelTypeToGlueType(s *model.Schema) string

ModelTypeToGlueType converts canonical model.Schema type into AWS Glue Hive data type string.

func SyncPartitions 

func SyncPartitions(ctx context.Context, ops PartitionSyncOperations, id TableIdentifier, desired []Partition, batchSize int) error

SyncPartitions reconciles a table's partitions in the catalog. Every action is attempted even when an earlier one fails, so one bad batch does not hide the rest; the errors are joined.

func TableFormatFromProperties 

func TableFormatFromProperties(properties map[string]string) (model.TableFormat, error)

TableFormatFromProperties resolves a table format from catalog table properties. It prefers table_type and falls back to spark.sql.sources.provider, matching Java's TableFormatUtils.

Types

type CatalogType 

type CatalogType string

CatalogType represents supported external metastores / data catalogs. 

const (
	CatalogTypeGlue        CatalogType = "AWS_GLUE"
	CatalogTypeHMS         CatalogType = "HIVE_METASTORE"
	CatalogTypeIcebergREST CatalogType = "ICEBERG_REST"
)

Catalog type identifiers. Only CatalogTypeGlue and CatalogTypeIcebergREST have client implementations; CatalogTypeHMS is retained for parity with Java XTable's identifier set so configuration written against it round-trips, and any attempt to use it must fail with ErrCatalogNotImplemented rather than be silently ignored.

func (CatalogType) Implemented 

func (c CatalogType) Implemented() bool

Implemented reports whether this catalog type has a client implementation.

type Config 

type Config struct {
	Type         CatalogType       `json:"type" yaml:"type"`
	CatalogID    string            `json:"catalogId,omitempty" yaml:"catalogId,omitempty"`
	DatabaseName string            `json:"databaseName" yaml:"databaseName"`
	URI          string            `json:"uri,omitempty" yaml:"uri,omitempty"`
	Properties   map[string]string `json:"properties,omitempty" yaml:"properties,omitempty"`
	// MaxPartitionsPerRequest caps how many partitions are sent per catalog call. Zero means
	// DefaultMaxPartitionsPerRequest. Catalog service limits are enforced beneath this regardless.
	MaxPartitionsPerRequest int `json:"maxPartitionsPerRequest,omitempty" yaml:"maxPartitionsPerRequest,omitempty"`
}

Config holds configuration parameters for connecting to an external catalog.

func (*Config) Validate 

func (c *Config) Validate() error

Validate validates catalog configuration settings.

type ConversionSource 

type ConversionSource interface {
	// CatalogType returns the catalog type identifier.
	CatalogType() CatalogType

	// GetSourceTable resolves a catalog entry into a SourceTable.
	GetSourceTable(ctx context.Context, id TableIdentifier) (*SourceTable, error)

	// Close releases any network connections.
	Close() error
}

ConversionSource resolves a table registered in an external catalog into a SourceTable. This is the read half of catalog integration; SyncClient is the write half.

func NewConversionSource 

func NewConversionSource(ctx context.Context, cfg *Config) (ConversionSource, error)

NewConversionSource creates a catalog conversion source for the configured catalog type.

type GlueCatalogSyncClient 

type GlueCatalogSyncClient struct {

	// Embedded so the four PartitionSyncOperations methods promote onto this client: Glue is a
	// Hive-style catalog that tracks partitions separately from the table definition.
	*GluePartitionSyncOperations
	// contains filtered or unexported fields
}

GlueCatalogSyncClient manages table and partition metadata synchronization to AWS Glue Data Catalog.

func NewGlueCatalogSyncClient 

func NewGlueCatalogSyncClient(ctx context.Context, cfg *Config) (*GlueCatalogSyncClient, error)

NewGlueCatalogSyncClient creates a new AWS Glue Catalog sync client.

func NewGlueCatalogSyncClientWithClient 

func NewGlueCatalogSyncClientWithClient(client *glue.Client, databaseName string, catalogID *string) *GlueCatalogSyncClient

NewGlueCatalogSyncClientWithClient creates a client with an existing glue.Client.

func (*GlueCatalogSyncClient) CatalogType 

func (g *GlueCatalogSyncClient) CatalogType() CatalogType

CatalogType returns AWS_GLUE.

func (*GlueCatalogSyncClient) Close 

func (g *GlueCatalogSyncClient) Close() error

Close is a no-op for GlueCatalogSyncClient.

func (*GlueCatalogSyncClient) CreateOrUpdateTable 

func (g *GlueCatalogSyncClient) CreateOrUpdateTable(ctx context.Context, table *model.Table, snapshot *model.Snapshot) error

CreateOrUpdateTable registers or updates the table definition in AWS Glue Data Catalog.

func (*GlueCatalogSyncClient) DropTable 

func (g *GlueCatalogSyncClient) DropTable(ctx context.Context, databaseName, tableName string) error

DropTable removes the table from AWS Glue.

type GlueConversionSource 

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

GlueConversionSource resolves tables registered in the AWS Glue Data Catalog.

func NewGlueConversionSource 

func NewGlueConversionSource(ctx context.Context, cfg *Config) (*GlueConversionSource, error)

NewGlueConversionSource creates a Glue-backed catalog conversion source.

func NewGlueConversionSourceWithClient 

func NewGlueConversionSourceWithClient(client glueTableReader, catalogID *string) *GlueConversionSource

NewGlueConversionSourceWithClient creates a source over an existing Glue API client.

func (*GlueConversionSource) CatalogType 

func (g *GlueConversionSource) CatalogType() CatalogType

CatalogType returns AWS_GLUE.

func (*GlueConversionSource) Close 

func (g *GlueConversionSource) Close() error

Close releases resources. The Glue SDK client needs no explicit teardown.

func (*GlueConversionSource) GetSourceTable 

func (g *GlueConversionSource) GetSourceTable(ctx context.Context, id TableIdentifier) (*SourceTable, error)

GetSourceTable resolves a Glue table into a SourceTable.

type GluePartitionSyncOperations 

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

GluePartitionSyncOperations implements PartitionSyncOperations against the AWS Glue Data Catalog.

func NewGluePartitionSyncOperations 

func NewGluePartitionSyncOperations(client gluePartitionAPI, catalogID *string) *GluePartitionSyncOperations

NewGluePartitionSyncOperations creates Glue-backed partition operations over an existing client.

func (*GluePartitionSyncOperations) AddPartitions 

func (g *GluePartitionSyncOperations) AddPartitions(ctx context.Context, id TableIdentifier, partitions []Partition) error

AddPartitions registers new partitions, chunked to Glue's per-call limit.

func (*GluePartitionSyncOperations) DropPartitions 

func (g *GluePartitionSyncOperations) DropPartitions(ctx context.Context, id TableIdentifier, partitions []Partition) error

DropPartitions removes partitions, chunked to Glue's per-call delete limit.

func (*GluePartitionSyncOperations) GetAllPartitions 

func (g *GluePartitionSyncOperations) GetAllPartitions(ctx context.Context, id TableIdentifier) ([]Partition, error)

GetAllPartitions lists every partition Glue records for the table, following pagination.

func (*GluePartitionSyncOperations) UpdatePartitions 

func (g *GluePartitionSyncOperations) UpdatePartitions(ctx context.Context, id TableIdentifier, partitions []Partition) error

UpdatePartitions rewrites partitions one at a time; Glue exposes no batch update.

type IcebergRESTCatalogClient 

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

IcebergRESTCatalogClient synchronizes Iceberg table metadata with standard Iceberg REST Catalogs (Polaris, Unity, Tabular, Nessie).

func NewIcebergRESTCatalogClient 

func NewIcebergRESTCatalogClient(cfg *Config) (*IcebergRESTCatalogClient, error)

NewIcebergRESTCatalogClient creates a new client for an Iceberg REST Catalog.

func (*IcebergRESTCatalogClient) CatalogType 

func (c *IcebergRESTCatalogClient) CatalogType() CatalogType

CatalogType returns ICEBERG_REST.

func (*IcebergRESTCatalogClient) Close 

func (c *IcebergRESTCatalogClient) Close() error

Close is a no-op.

func (*IcebergRESTCatalogClient) CreateOrUpdateTable 

func (c *IcebergRESTCatalogClient) CreateOrUpdateTable(ctx context.Context, table *model.Table, _ *model.Snapshot) error

CreateOrUpdateTable registers or commits the Iceberg table metadata to the REST catalog.

func (*IcebergRESTCatalogClient) DropTable 

func (c *IcebergRESTCatalogClient) DropTable(ctx context.Context, databaseName, tableName string) error

DropTable removes the table registration from the REST catalog.

type IcebergRESTConversionSource 

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

IcebergRESTConversionSource resolves tables registered in an Iceberg REST Catalog (Polaris, Unity, Tabular, Nessie).

func NewIcebergRESTConversionSource 

func NewIcebergRESTConversionSource(cfg *Config) (*IcebergRESTConversionSource, error)

NewIcebergRESTConversionSource creates a conversion source for an Iceberg REST Catalog.

func NewIcebergRESTConversionSourceWithClient 

func NewIcebergRESTConversionSourceWithClient(client *http.Client, baseURI, namespace, authToken string) *IcebergRESTConversionSource

NewIcebergRESTConversionSourceWithClient creates a source using a caller-supplied HTTP client.

func (*IcebergRESTConversionSource) CatalogType 

func (c *IcebergRESTConversionSource) CatalogType() CatalogType

CatalogType returns ICEBERG_REST.

func (*IcebergRESTConversionSource) Close 

func (c *IcebergRESTConversionSource) Close() error

Close releases any network connections.

func (*IcebergRESTConversionSource) GetSourceTable 

func (c *IcebergRESTConversionSource) GetSourceTable(ctx context.Context, id TableIdentifier) (*SourceTable, error)

GetSourceTable loads a table from the REST catalog and resolves it to a SourceTable. An Iceberg REST catalog only ever serves Iceberg tables, so the format is not inferred from properties.

type Partition 

type Partition struct {
	// Values are the partition key values in partition-spec order.
	Values []string
	// StorageLocation is the directory holding this partition's data files.
	StorageLocation string
}

Partition is a single catalog partition: the ordered partition-key values and where its data lives.

func PartitionsFromSnapshot 

func PartitionsFromSnapshot(snapshot *model.Snapshot) []Partition

PartitionsFromSnapshot derives the partitions a table currently has from its data files. Files carrying no partition values yield no partitions, which is the correct answer for an unpartitioned table. The storage location is the directory containing the file.

func (Partition) Key 

func (p Partition) Key() string

Key renders the partition values as a stable identity for diffing. Location is deliberately excluded so a relocated partition is reported as changed rather than as an unrelated pair.

type PartitionEvent 

type PartitionEvent struct {
	Type      PartitionEventType
	Partition Partition
}

PartitionEvent pairs a reconciliation action with the partition it applies to.

func DiffPartitions 

func DiffPartitions(existing, desired []Partition) []PartitionEvent

DiffPartitions computes the events needed to bring `existing` (what the catalog has) in line with `desired` (what the table has). Results are sorted by partition key so the output is deterministic regardless of input ordering — the same guarantee model.DiffFiles fails to make.

type PartitionEventType 

type PartitionEventType string

PartitionEventType describes what must happen to a partition to reconcile the catalog. 

const (
	PartitionEventAdd    PartitionEventType = "ADD"
	PartitionEventUpdate PartitionEventType = "UPDATE"
	PartitionEventDrop   PartitionEventType = "DROP"
)

Partition reconciliation actions.

type PartitionSyncOperations 

type PartitionSyncOperations interface {
	// GetAllPartitions lists every partition the catalog currently records for the table.
	GetAllPartitions(ctx context.Context, id TableIdentifier) ([]Partition, error)

	// AddPartitions registers new partitions.
	AddPartitions(ctx context.Context, id TableIdentifier, partitions []Partition) error

	// UpdatePartitions rewrites existing partitions, typically because their location moved.
	UpdatePartitions(ctx context.Context, id TableIdentifier, partitions []Partition) error

	// DropPartitions removes partitions that no longer exist in the table.
	DropPartitions(ctx context.Context, id TableIdentifier, partitions []Partition) error
}

PartitionSyncOperations is the catalog-side partition API. It is separate from SyncClient because not every catalog tracks partitions: Iceberg and Delta carry partition data in their own metadata, so only Hive-style catalogs such as Glue need this.

type SourceTable 

type SourceTable struct {
	// Name is the table name as the catalog records it.
	Name string
	// BasePath is the table's root location in storage.
	BasePath string
	// DataPath is where the data files live. It equals BasePath for Delta and Hudi; Iceberg tables
	// may relocate it via write.data.path and friends.
	DataPath string
	// Format is the table format resolved from the catalog's table properties.
	Format model.TableFormat
	// Properties is the full property map the catalog returned, for callers needing more.
	Properties map[string]string
}

SourceTable describes a table resolved out of an external catalog, carrying everything needed to construct a ConversionSource for it without the caller knowing a base path up front.

type SyncClient 

type SyncClient interface {
	// CatalogType returns the catalog type identifier.
	CatalogType() CatalogType

	// CreateOrUpdateTable registers or updates the table metadata and schema in the external catalog.
	CreateOrUpdateTable(ctx context.Context, table *model.Table, snapshot *model.Snapshot) error

	// DropTable removes the table registration from the external catalog.
	DropTable(ctx context.Context, databaseName, tableName string) error

	// Close releases any network connections.
	Close() error
}

SyncClient defines the standard contract for synchronizing table metadata into external data catalogs.

func NewSyncClient 

func NewSyncClient(ctx context.Context, cfg *Config) (SyncClient, error)

NewSyncClient creates a catalog sync client for the specified type. For HMS (Hive Metastore), this returns ErrCatalogNotImplemented since no client implementation exists yet; see docs/improvement-plan.md T13 for the full scope a real HMS implementation would require.

type TableIdentifier 

type TableIdentifier struct {
	// Database is the catalog database or namespace holding the table.
	Database string
	// Table is the table name within that database.
	Table string
}

TableIdentifier addresses a table inside an external catalog. It is the catalog-side equivalent of a base path: the whole point of a ConversionSource is to accept one of these instead.

func (TableIdentifier) String 

func (t TableIdentifier) String() string

String renders the identifier as "database.table".

func (TableIdentifier) Validate 

func (t TableIdentifier) Validate() error

Validate reports whether the identifier is usable.

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.