iceberg

Documentation

Index

• func IcebergToSchema(icebergSchema *TableSchema) (*model.Schema, error)
• type ManifestDataFile
• type ManifestEntry
• type ManifestListEntry
• type NestedField
• type PartitionFieldDef
• type PartitionSpec
• type SnapshotSummary
• type Source
  • func NewSource(storage io.Storage, basePath string) *Source
  • func (s *Source) Close() error
  • func (s *Source) Format() model.TableFormat
  • func (s *Source) GetChangesSince(ctx context.Context, fromInstant int64) (*model.IncrementalTableChanges, error)
  • func (s *Source) GetCurrentSnapshot(ctx context.Context) (*model.Snapshot, error)
  • func (s *Source) GetCurrentTable(ctx context.Context) (*model.Table, error)
  • func (s *Source) GetTable(ctx context.Context, commitID string) (*model.Table, error)
  • func (s *Source) GetTableChangeForCommit(ctx context.Context, commitID string) (*model.TableChange, error)
  • func (s *Source) IsIncrementalSyncSafeFrom(ctx context.Context, earliestInstant int64) (bool, error)
• type TableMetadata
• type TableSchema
  • func SchemaToIceberg(schema *model.Schema, schemaID int) (*TableSchema, int, error)
• type TableSnapshot
• type Target
  • func NewTarget(storage io.Storage) *Target
  • func (t *Target) Close() error
  • func (t *Target) CommitChanges(ctx context.Context, changes *model.IncrementalTableChanges) error
  • func (t *Target) CommitSnapshot(ctx context.Context, snapshot *model.Snapshot) error
  • func (t *Target) Format() model.TableFormat
  • func (t *Target) GetTableMetadata(ctx context.Context) (*model.TableSyncMetadata, error)
  • func (t *Target) Init(_ context.Context, targetTable *model.Table) error

Functions

func IcebergToSchema

func IcebergToSchema(icebergSchema *TableSchema) (*model.Schema, error)

IcebergToSchema converts an Iceberg TableSchema into a canonical model.Schema.

Types

type ManifestDataFile

type ManifestDataFile struct {
	FilePath        string         `json:"file_path"`
	FileFormat      string         `json:"file_format"`
	Partition       map[string]any `json:"partition"`
	RecordCount     int64          `json:"record_count"`
	FileSizeInBytes int64          `json:"file_size_in_bytes"`
	ColumnSizes     map[int]int64  `json:"column_sizes,omitempty"`
	ValueCounts     map[int]int64  `json:"value_counts,omitempty"`
	NullValueCounts map[int]int64  `json:"null_value_counts,omitempty"`
	NanValueCounts  map[int]int64  `json:"nan_value_counts,omitempty"`
	LowerBounds     map[int]string `json:"lower_bounds,omitempty"`
	UpperBounds     map[int]string `json:"upper_bounds,omitempty"`
}

ManifestDataFile represents the metadata of a data file inside a manifest.

type ManifestEntry

type ManifestEntry struct {
	Status     int               `json:"status"` // 0: EXISTING, 1: ADDED, 2: DELETED
	SnapshotID int64             `json:"snapshot_id"`
	DataFile   *ManifestDataFile `json:"data_file"`
}

ManifestEntry represents a data file entry inside an Iceberg manifest.

type ManifestListEntry

type ManifestListEntry struct {
	ManifestPath       string `json:"manifest_path"`
	ManifestLength     int64  `json:"manifest_length"`
	PartitionSpecID    int    `json:"partition_spec_id"`
	AddedSnapshotID    int64  `json:"added_snapshot_id"`
	AddedFilesCount    int    `json:"added_data_files_count"`
	ExistingFilesCount int    `json:"existing_data_files_count"`
	DeletedFilesCount  int    `json:"deleted_data_files_count"`
}

ManifestListEntry represents an entry inside a manifest list file.

type NestedField

type NestedField struct {
	ID             int    `json:"id"`
	Name           string `json:"name"`
	Type           any    `json:"type"` // string (e.g. "int", "string") or nested map/struct
	Required       bool   `json:"required"`
	Doc            string `json:"doc,omitempty"`
	InitialDefault any    `json:"initial-default,omitempty"`
	WriteDefault   any    `json:"write-default,omitempty"`
}

NestedField represents a field inside an Iceberg schema.

type PartitionFieldDef

type PartitionFieldDef struct {
	SourceID  int    `json:"source-id"`
	FieldID   int    `json:"field-id"`
	Name      string `json:"name"`
	Transform string `json:"transform"` // identity, year, month, day, hour, bucket[N], truncate[W]
}

PartitionFieldDef represents a partition field in an Iceberg partition specification.

type PartitionSpec

type PartitionSpec struct {
	SpecID int                  `json:"spec-id"`
	Fields []*PartitionFieldDef `json:"fields"`
}

PartitionSpec represents an Iceberg partition specification.

type SnapshotSummary

type SnapshotSummary struct {
	Operation       string            `json:"operation"` // append, replace, overwrite, delete
	AddedDataFiles  string            `json:"added-data-files,omitempty"`
	AddedRecords    string            `json:"added-records,omitempty"`
	TotalDataFiles  string            `json:"total-data-files,omitempty"`
	TotalRecords    string            `json:"total-records,omitempty"`
	ExtraProperties map[string]string `json:"extra-properties,omitempty"`
}

SnapshotSummary holds metadata describing the commit operation and row counts.

type Source

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

Source implements spi.ConversionSource for Apache Iceberg tables.

func NewSource

func NewSource(storage io.Storage, basePath string) *Source

NewSource creates a new Iceberg ConversionSource instance.

func (*Source) Close

func (s *Source) Close() error

Close is a no-op for Iceberg source.

func (*Source) Format

func (s *Source) Format() model.TableFormat

Format returns the format identifier.

func (*Source) GetChangesSince

func (s *Source) GetChangesSince(ctx context.Context, fromInstant int64) (*model.IncrementalTableChanges, error)

GetChangesSince returns incremental changes since a timestamp.

func (*Source) GetCurrentSnapshot

func (s *Source) GetCurrentSnapshot(ctx context.Context) (*model.Snapshot, error)

GetCurrentSnapshot constructs the complete Snapshot from Iceberg manifests.

func (*Source) GetCurrentTable

func (s *Source) GetCurrentTable(ctx context.Context) (*model.Table, error)

GetCurrentTable returns the Table descriptor at the latest Iceberg metadata version.

func (*Source) GetTable

func (s *Source) GetTable(ctx context.Context, commitID string) (*model.Table, error)

GetTable returns the Table descriptor at a specific metadata version.

func (*Source) GetTableChangeForCommit

func (s *Source) GetTableChangeForCommit(ctx context.Context, commitID string) (*model.TableChange, error)

GetTableChangeForCommit returns the diff of added and removed files in a snapshot.

func (*Source) IsIncrementalSyncSafeFrom

func (s *Source) IsIncrementalSyncSafeFrom(ctx context.Context, earliestInstant int64) (bool, error)

IsIncrementalSyncSafeFrom checks if snapshot history is available.

type TableMetadata

type TableMetadata struct {
	FormatVersion      int               `json:"format-version"`
	TableUUID          string            `json:"table-uuid"`
	Location           string            `json:"location"`
	LastSequenceNumber int64             `json:"last-sequence-number"`
	LastUpdatedMs      int64             `json:"last-updated-ms"`
	LastColumnID       int               `json:"last-column-id"`
	CurrentSchemaID    int               `json:"current-schema-id"`
	Schemas            []*TableSchema    `json:"schemas"`
	DefaultSpecID      int               `json:"default-spec-id"`
	PartitionSpecs     []*PartitionSpec  `json:"partition-specs"`
	LastPartitionID    int               `json:"last-partition-id"`
	DefaultSortOrderID int               `json:"default-sort-order-id"`
	Properties         map[string]string `json:"properties,omitempty"`
	CurrentSnapshotID  *int64            `json:"current-snapshot-id,omitempty"`
	Snapshots          []*TableSnapshot  `json:"snapshots,omitempty"`
}

TableMetadata matches the Apache Iceberg v2/v3 metadata.json specification.

type TableSchema

type TableSchema struct {
	Type               string         `json:"type"`
	SchemaID           int            `json:"schema-id"`
	IdentifierFieldIDs []int          `json:"identifier-field-ids,omitempty"`
	Fields             []*NestedField `json:"fields"`
}

TableSchema represents an Iceberg schema definition.

func SchemaToIceberg

func SchemaToIceberg(schema *model.Schema, schemaID int) (*TableSchema, int, error)

SchemaToIceberg converts a canonical model.Schema to an Iceberg TableSchema, assigning field IDs.

type TableSnapshot

type TableSnapshot struct {
	SnapshotID       int64             `json:"snapshot-id"`
	ParentSnapshotID *int64            `json:"parent-snapshot-id,omitempty"`
	SequenceNumber   int64             `json:"sequence-number"`
	TimestampMs      int64             `json:"timestamp-ms"`
	ManifestList     string            `json:"manifest-list"`
	Summary          map[string]string `json:"summary"`
	SchemaID         *int              `json:"schema-id,omitempty"`
}

TableSnapshot represents a snapshot entry in Iceberg table metadata.

type Target

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

Target implements spi.ConversionTarget for Apache Iceberg tables.

func NewTarget

func NewTarget(storage io.Storage) *Target

NewTarget creates a new Iceberg ConversionTarget instance.

func (*Target) Close

func (t *Target) Close() error

Close is a no-op for Iceberg target.

func (*Target) CommitChanges

func (t *Target) CommitChanges(ctx context.Context, changes *model.IncrementalTableChanges) error

CommitChanges writes incremental changes.

func (*Target) CommitSnapshot

func (t *Target) CommitSnapshot(ctx context.Context, snapshot *model.Snapshot) error

CommitSnapshot writes a full snapshot into Apache Iceberg metadata and manifests.

func (*Target) Format

func (t *Target) Format() model.TableFormat

Format returns the format identifier.

func (*Target) GetTableMetadata

func (t *Target) GetTableMetadata(ctx context.Context) (*model.TableSyncMetadata, error)

GetTableMetadata retrieves previously recorded TableSyncMetadata from Iceberg table properties.

func (*Target) Init

func (t *Target) Init(_ context.Context, targetTable *model.Table) error

Init initializes the target with table configuration.