internal

package
v0.7.0 Latest Latest
Warning

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

 Go to latest Published: Sep 28, 2026 License: Apache-2.0 Imports: 48 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ParquetRowGroupSizeBytesKey       = "write.parquet.row-group-size-bytes"
	ParquetRowGroupSizeBytesDefault   = 128 * 1024 * 1024 // 128 MB
	ParquetRowGroupLimitKey           = "write.parquet.row-group-limit"
	ParquetRowGroupLimitDefault       = 1048576
	ParquetPageSizeBytesKey           = "write.parquet.page-size-bytes"
	ParquetPageSizeBytesDefault       = 1024 * 1024 // 1 MB
	ParquetPageRowLimitKey            = "write.parquet.page-row-limit"
	ParquetPageRowLimitDefault        = 20000
	ParquetDictSizeBytesKey           = "write.parquet.dict-size-bytes"
	ParquetDictSizeBytesDefault       = 2 * 1024 * 1024 // 2 MB
	ParquetPageVersionKey             = "write.parquet.page-version"
	ParquetPageVersionDefault         = "2"
	ParquetCompressionKey             = "write.parquet.compression-codec"
	ParquetCompressionDefault         = "zstd"
	ParquetCompressionLevelKey        = "write.parquet.compression-level"
	ParquetCompressionLevelDefault    = -1
	ParquetBloomFilterMaxBytesKey     = "write.parquet.bloom-filter-max-bytes"
	ParquetBloomFilterMaxBytesDefault = 1024 * 1024

	ParquetBloomFilterColumnEnabledKeyPrefix = "write.parquet.bloom-filter-enabled.column"

	// Deliberately not namespaced under write.parquet: this is the parquet-mr key
	// Iceberg Java reads straight from the table properties, so a table carrying
	// it gets the same dictionary behavior from either implementation.
	ParquetDictEnabledKey                     = "parquet.enable.dictionary"
	ParquetDictEnabledDefault                 = true
	ParquetDictEncodingColumnEnabledKeyPrefix = "write.parquet.dict-encoding-enabled.column"

	ParquetShredVariantsKey     = "write.parquet.shred-variants"
	ParquetShredVariantsDefault = false
	// Rows buffered per file to infer shredding (held per open partition writer).
	ParquetVariantBufferSizeKey     = "write.parquet.variant-inference-buffer-size"
	ParquetVariantBufferSizeDefault = 100

	ParquetBatchSizeKey     = "read.parquet.batch-size"
	ParquetBatchSizeDefault = 1 << 17 // 131072 rows
)

Variables

This section is empty.

Functions

func AnalyzeVariantShredding added in v0.7.0 

func AnalyzeVariantShredding(sample []variant.Value) (arrow.DataType, bool)

AnalyzeVariantShredding infers the inner Arrow type to shred the sample by, or ok=false to not shred. The result is the INNER type for NewShreddedVariantType.

func BBoxIntersectsXY added in v0.7.0 

func BBoxIntersectsXY(aMinX, aMinY, aMaxX, aMaxY, bMinX, bMinY, bMaxX, bMaxY float64) bool

BBoxIntersectsXY reports whether two planar (XY) bounding boxes intersect. Boxes touching only at an edge or corner count as intersecting (closed intervals), matching Iceberg's inclusive bbox pruning: a box that might contain a matching value must not be pruned. This is the geometry (planar) rule; geography columns emit no bounds (see Bounds), so the antimeridian wrap-around case never reaches pruning.

func BigEndianToDecimal added in v0.2.0 

func BigEndianToDecimal(buf []byte) (decimal.Decimal128, error)

func BuildPartitionMatchPredicate added in v0.7.0 

func BuildPartitionMatchPredicate(spec iceberg.PartitionSpec, schema *iceberg.Schema, partitions []map[int]any) (iceberg.BooleanExpression, error)

BuildPartitionMatchPredicate constructs a BooleanExpression matching all rows that belong to any of the given partitions. It is the core of dynamic partition overwrite (see https://github.com/apache/iceberg-go/issues/1216): the returned expression selects the existing data to delete before the new data is appended.

partitions holds partition tuples as returned by DataFile.Partition(), each keyed by partition-field ID and carrying the field's (transformed) value. The result is an OR across distinct partitions, each clause an AND across the spec's fields: 

source == value   when the partition value is present
IsNaN(source)      when the value is a floating-point NaN (x == NaN is never true)
IsNull(source)     when the partition value is absent or nil

Duplicate tuples collapse to a single clause, and an empty input yields AlwaysFalse (matching nothing). Callers are expected to pass a partitioned spec; dynamic partition overwrite rejects unpartitioned tables upstream.

Because only identity transforms are accepted (see below), the partition value equals the source-column value, so "source == value" selects exactly the rows in that partition. Non-identity transforms (bucket, truncate, the time transforms) cannot be matched by a source-column predicate and need partition-level matching instead; they are rejected here and tracked as a follow-up under issue #1215.

func GeoBoundsXY added in v0.7.0 

func GeoBoundsXY(lower, upper []byte) (minX, minY, maxX, maxY float64, ok bool)

GeoBoundsXY decodes a geometry column's lower and upper geo bounds (Iceberg single-value serialization; see encodeGeoBound) into their planar XY extents. ok is false when either bound is missing, malformed, or carries a NaN X/Y coordinate - all cases where the bound is unusable for pruning.

func LiteralForPartitionValue added in v0.7.0 

func LiteralForPartitionValue(v any) (iceberg.Literal, error)

LiteralForPartitionValue converts a partition value (as stored on a DataFile) into a typed Literal so the resulting predicate binds against the source field with the correct type, rather than relying on a string rendering.

DataFile.Partition() yields either a Literal (e.g. DecimalLiteral for decimal fields, decoded in manifest.go) or a raw Go value; both are handled.

func MakeSequencedChan 

func MakeSequencedChan[T any](bufferSize uint, source <-chan T, comesAfter, isNext func(a, b *T) bool, initial T) <-chan T

MakeSequencedChan creates a channel that outputs values in a given order based on the comesAfter and isNext functions. The values are read in from the provided source and then re-ordered before being sent to the output.

func MakeSequencedChanWithDiscard added in v0.7.0 

func MakeSequencedChanWithDiscard[T any](bufferSize uint, source <-chan T, comesAfter, isNext func(a, b *T) bool, initial T, discard func(T)) <-chan T

MakeSequencedChanWithDiscard is MakeSequencedChan with a callback for values that remain in the reorder queue when the source is closed. Values already sent to the output are not passed to discard.

func MapExec added in v0.3.0 

func MapExec[T, S any](ctx context.Context, nWorkers int, slice iter.Seq[T], fn func(T) (S, error)) iter.Seq2[S, error]

func ParquetRowGroupTargetSizeBytes added in v0.7.0 

func ParquetRowGroupTargetSizeBytes(props iceberg.Properties) (int64, error)

ParquetRowGroupTargetSizeBytes returns the configured uncompressed row-group size target. Iceberg Java rejects invalid and non-positive values rather than silently falling back to the default.

func PartitionRecordValue added in v0.3.0 

func PartitionRecordValue(field iceberg.PartitionField, val iceberg.Literal, schema *iceberg.Schema) (iceberg.Optional[iceberg.Literal], error)

func ShredRecordVariants added in v0.7.0 

func ShredRecordVariants(rec arrow.RecordBatch, shreddedSchema *arrow.Schema, mem memory.Allocator) (out arrow.RecordBatch, err error)

ShredRecordVariants rebuilds the shredded-variant columns of rec to match shreddedSchema (the inverse of UnshredVariant); other columns pass through.

func ShreddedArrowSchema added in v0.7.0 

func ShreddedArrowSchema(base *arrow.Schema, inferred map[int]arrow.DataType) *arrow.Schema

ShreddedArrowSchema swaps each top-level variant field in inferred (that passes canShredVariant) for a shredded type, preserving name/nullability/field-id.

func TablePropertiesFromContext added in v0.6.0 

func TablePropertiesFromContext(ctx context.Context) iceberg.Properties

TablePropertiesFromContext retrieves table properties from context. Returns nil if not set.

func TruncateUpperBoundBinary added in v0.3.0 

func TruncateUpperBoundBinary(val []byte, trunc int) []byte

func TruncateUpperBoundText added in v0.3.0 

func TruncateUpperBoundText(s string, trunc int) string

func ValidateParquetWriteProperties added in v0.7.0 

func ValidateParquetWriteProperties(props iceberg.Properties) error

func VariantBoundLiteral added in v0.7.0 

func VariantBoundLiteral(raw []byte, path string, typ iceberg.PrimitiveType) (lit iceberg.Literal, ok bool, err error)

VariantBoundLiteral decodes the stored variant bound object at the normalized path and casts it to typ.

func VariantFieldIDsFromSchema added in v0.6.0 

func VariantFieldIDsFromSchema(sc *iceberg.Schema) map[int]struct{}

func WithParquetMetadataCache added in v0.7.0 

func WithParquetMetadataCache(ctx context.Context) context.Context

WithParquetMetadataCache adds a per-operation cache for immutable Parquet file metadata. Readers and input handles are still created per task, so parallel split tasks do not share mutable reader state.

func WithTableProperties added in v0.6.0 

func WithTableProperties(ctx context.Context, props iceberg.Properties) context.Context

WithTableProperties returns a new context with the given table properties attached. These properties are used by readers to configure read behavior (e.g. batch size).

Types

type DataFileOpts added in v0.6.0 

type DataFileOpts struct {
	Schema          *iceberg.Schema
	Spec            iceberg.PartitionSpec
	Path            string
	Format          iceberg.FileFormat
	Content         iceberg.ManifestEntryContent
	FileSize        int64
	PartitionValues map[int]any
	// SortOrderID claims the file's rows are fully sorted by that order; zero
	// makes no claim and leaves the field absent.
	SortOrderID int
	// FirstRowID is honored only for EntryContentData; the spec requires it to
	// be null for delete files.
	FirstRowID         *int64
	ReferencedDataFile *string
}

DataFileOpts groups the fields needed to finalize a DataFile from DataFileStatistics. Collapsing the former positional arguments into a struct keeps call sites readable and makes future additions source- compatible.

type DataFileStatistics added in v0.3.0 

type DataFileStatistics struct {
	RecordCount      int64
	ColSizes         map[int]int64
	ValueCounts      map[int]int64
	NullValueCounts  map[int]int64
	NanValueCounts   map[int]int64
	ColAggs          map[int]StatsAgg
	SplitOffsets     []int64
	EqualityFieldIDs []int
	// Variant bounds: serialized variant objects keyed by parent variant field id.
	VariantLowerBounds map[int][]byte
	VariantUpperBounds map[int][]byte
}

func (*DataFileStatistics) PartitionValue added in v0.3.0 

func (d *DataFileStatistics) PartitionValue(field iceberg.PartitionField, sc *iceberg.Schema) any

func (*DataFileStatistics) ToDataFile added in v0.3.0 

func (d *DataFileStatistics) ToDataFile(opts DataFileOpts) iceberg.DataFile

type Enumerated 

type Enumerated[T any] struct {
	Value T
	Index int
	Last  bool
}

Enumerated is a quick way to represent a sequenced value that can be processed in parallel and then needs to be reordered.

type FileFormat added in v0.3.0 

type FileFormat interface {
	Open(context.Context, iceio.IO, string) (FileReader, error)
	PathToIDMapping(*iceberg.Schema) (map[string]int, error)
	DataFileStatsFromMeta(rdr Metadata, statsCols map[int]StatisticsCollector, colMapping map[string]int, variantFieldIDs map[int]struct{}, arrowSchema *arrow.Schema) *DataFileStatistics
	GetWriteProperties(iceberg.Properties) any
	WriteDataFile(ctx context.Context, fs iceio.WriteFileIO, partitionValues map[int]any, info WriteFileInfo, batches []arrow.RecordBatch) (iceberg.DataFile, error)
	NewFileWriter(ctx context.Context, fs iceio.WriteFileIO, partitionValues map[int]any, info WriteFileInfo, arrowSchema *arrow.Schema) (FileWriter, error)
}

func FormatFromFileName added in v0.3.0 

func FormatFromFileName(fileName string) FileFormat

func GetFileFormat added in v0.3.0 

func GetFileFormat(format iceberg.FileFormat) FileFormat

type FileReader 

type FileReader interface {
	io.Closer

	Metadata() Metadata
	SourceFileSize() int64
	Schema() (*arrow.Schema, error)
	// PrunedSchema takes in the list of projected field IDs and returns the arrow schema
	// that represents the underlying file schema with only the projected fields. It also
	// returns the indexes of the projected columns to allow reading *only* the needed
	// columns.
	PrunedSchema(projectedIDs map[int]struct{}, mapping iceberg.NameMapping) (*arrow.Schema, []int, error)
	// GetRecords returns a record reader for only the provided columns (using nil will read
	// all of the columns of the underlying file.) The `tester` is a function that can be used,
	// if non-nil, to filter aspects of the file such as skipping row groups in a parquet file.
	GetRecords(ctx context.Context, cols []int, tester any) (array.RecordReader, error)
	// ReadTable reads the entire file and returns it as an arrow table.
	ReadTable(context.Context) (arrow.Table, error)
}

type FileSource 

type FileSource interface {
	GetReader(context.Context) (FileReader, error)
}

func GetFile 

func GetFile(ctx context.Context, fs iceio.IO, dataFile iceberg.DataFile, isPosDeletes bool) (FileSource, error)

GetFile opens the given file using the provided file system.

The FileSource interface allows abstracting away the underlying file format while providing utilties to read the file as Arrow record batches.

type FileWriter added in v0.6.0 

type FileWriter interface {
	Write(arrow.RecordBatch) error
	BytesWritten() int64
	Close() (iceberg.DataFile, error)
	// Abort closes the underlying file handle without finalizing the
	// file format (e.g. Parquet footer). It is safe to call regardless
	// of how many rows have been written and should be used on error
	// paths where Close() may panic or produce an invalid file.
	Abort() error
}

FileWriter is an incremental single-file writer with open/write/close lifecycle. It writes Arrow record batches and tracks bytes written for rolling file decisions.

type GeoBoundsAggregator added in v0.7.0 

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

GeoBoundsAggregator combines the per-data-file geospatial bounds emitted by geoBoundsAccumulator (the single-value serialization written into a DataFile's lower/upper bounds; see Bounds) across multiple files into one bounding box. It is the manifest-level analogue of the primitive min/max aggregation the manifest writer already does per partition field, but geo bounds are coordinate tuples with no total order, so they must never be folded with a scalar byte comparison: each bound is decoded to its coordinates and merged dimension by dimension. The combined box is returned in the same serialization, so it round-trips through a manifest bound exactly like a per-file one.

The aggregator is geometry-only: it folds each dimension with a scalar min/max, which is correct for geometry's planar bounds but wrong for geography. A geography box may cross the antimeridian, encoded as lower_x > upper_x (spec Appendix D); scalar min/max would silently unwrap it (merging a wrapped [170, -170] with [10, 20] yields [10, 20], dropping the wrapped range), producing a box that prunes rows it should keep. Because the bound bytes carry no type, callers must declare geography via NewGeoBoundsAggregator so Add can reject it (see Add). iceberg-go itself emits no per-file bounds for geography (see Bounds), so the common path — passing empty geography bounds through — is a harmless no-op; only non-empty geography bounds (e.g. from files written by another engine) are refused, until geodesic/antimeridian-aware aggregation is added.

func NewGeoBoundsAggregator added in v0.7.0 

func NewGeoBoundsAggregator(isGeography bool) *GeoBoundsAggregator

NewGeoBoundsAggregator returns an aggregator for a single geo column. Pass isGeography=true for geography columns so Add refuses their bounds rather than mis-merging antimeridian-crossing boxes (see GeoBoundsAggregator). The zero-value GeoBoundsAggregator is a valid geometry aggregator.

func (*GeoBoundsAggregator) Add added in v0.7.0 

func (g *GeoBoundsAggregator) Add(lower, upper []byte) error

Add merges one data file's lower and upper geo bounds into the aggregate. The two bounds must share a layout, which they always do: a file's lower and upper are written together by geoBoundsAccumulator. An empty pair (nil/empty lower and upper) contributes nothing, so files without geo bounds - including every geography file written by iceberg-go - can be passed through harmlessly. It errors when a bound is a non-empty but invalid length, when lower and upper disagree on layout, or when a non-empty bound is added to a geography aggregator (scalar folding would mis-merge antimeridian-crossing boxes; see GeoBoundsAggregator).

func (*GeoBoundsAggregator) Bounds added in v0.7.0 

func (g *GeoBoundsAggregator) Bounds() (lower, upper []byte)

Bounds returns the combined lower and upper bound points in the Iceberg geospatial single-value serialization, or nil when no file contributed a bounding box. An optional Z/M dimension is dropped unless every added file carried it (see the field docs).

type Metadata added in v0.3.0 

type Metadata any

type MetricModeType added in v0.3.0 

type MetricModeType string
const (
	MetricModeTruncate MetricModeType = "truncate"
	MetricModeNone     MetricModeType = "none"
	MetricModeCounts   MetricModeType = "counts"
	MetricModeFull     MetricModeType = "full"
)

type MetricsMode added in v0.3.0 

type MetricsMode struct {
	Typ MetricModeType
	Len int
}

func MatchMetricsMode added in v0.3.0 

func MatchMetricsMode(mode string) (MetricsMode, error)

type ParquetFileSource 

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

func (*ParquetFileSource) GetReader 

func (pfs *ParquetFileSource) GetReader(ctx context.Context) (result FileReader, err error)

type ParquetFileWriter added in v0.6.0 

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

ParquetFileWriter is an incremental single-file writer with open/write/close lifecycle. It writes Arrow record batches to a Parquet file and tracks bytes written for rolling file decisions.

func (*ParquetFileWriter) Abort added in v0.6.0 

func (w *ParquetFileWriter) Abort() error

func (*ParquetFileWriter) BytesWritten added in v0.6.0 

func (w *ParquetFileWriter) BytesWritten() int64

BytesWritten returns the number of bytes flushed to the output so far.

func (*ParquetFileWriter) Close added in v0.6.0 

func (w *ParquetFileWriter) Close() (_ iceberg.DataFile, err error)

Close finalizes the Parquet file and returns the resulting DataFile with accurate file statistics and size.

func (*ParquetFileWriter) Write added in v0.6.0 

func (w *ParquetFileWriter) Write(batch arrow.RecordBatch) error

Write appends a record batch to the Parquet file.

type ParquetRowGroupTester added in v0.6.0 

type ParquetRowGroupTester struct {
	StatsFn         func(*metadata.RowGroupMetaData, []int) (bool, error)
	BloomPreds      []RowGroupBloomPred      // nil = no bloom filter pass
	DictionaryPreds []RowGroupDictionaryPred // nil = no dictionary filter pass
	// RangeSet indicates that Start and Length came from an explicit scan task.
	// Without it, the zero value keeps the historical full-file behavior for
	// callers that construct a tester only for row-group pruning.
	RangeSet bool
	// When RangeSet is true, Start and Length select row groups whose absolute
	// Parquet row-group offset falls in [Start, Start+Length).
	Start, Length int64
	// PlanningSplitOffsets contains the split offsets used when the scan task
	// was planned. When it contains one valid offset per row group, range
	// selection uses these offsets instead of re-deriving them from the opened
	// Parquet footer. Sparse split-offset lists fall back to the footer offsets.
	PlanningSplitOffsets []int64
	// Survivors, if non-nil, is reset and then filled with one span per row group
	// that survives pruning, in file order, with positions relative to the full
	// file. It lets callers reconstruct each emitted row's original position even
	// when pruning skips groups. Set to nil when no group is skipped.
	Survivors *[]RowGroupSpan
}

ParquetRowGroupTester combines stats-based, dictionary, and bloom filter row group pruning. Pass it as the tester argument to wrapPqArrowReader.GetRecords.

type RowGroupBloomPred added in v0.6.0 

type RowGroupBloomPred struct {
	FieldID   int
	PhysBytes [][]byte // one entry for EqualTo; one per value for In
}

RowGroupBloomPred holds the physical-encoded bytes for each literal in a bloom-filterable predicate on one field. A row group can be skipped when NONE of the bytes appear in the column's bloom filter.

type RowGroupDictionaryPred added in v0.7.0 

type RowGroupDictionaryPred struct {
	FieldID   int
	PhysBytes [][]byte // one entry for EqualTo; one per value for In
}

RowGroupDictionaryPred holds the physical-encoded bytes for each literal in a dictionary-prunable predicate on one field. A row group can be skipped when NONE of the bytes occur in its complete dictionary. It intentionally mirrors RowGroupBloomPred because both predicates use the same physical literal encoding.

type RowGroupSpan added in v0.7.0 

type RowGroupSpan struct {
	FirstRowPos int64
	NumRows     int64
}

RowGroupSpan locates a surviving row group within the data file: the absolute position of its first row and the number of rows it holds.

type StatisticsCollector added in v0.3.0 

type StatisticsCollector struct {
	FieldID    int
	IcebergTyp iceberg.PrimitiveType
	Mode       MetricsMode
	ColName    string
}

type StatsAgg added in v0.3.0 

type StatsAgg interface {
	Min() iceberg.Literal
	Max() iceberg.Literal
	Update(stats interface{ HasMinMax() bool })
	MinAsBytes() ([]byte, error)
	MaxAsBytes() ([]byte, error)
}

type TypedStats added in v0.3.0 

type TypedStats[T iceberg.LiteralType] interface {
	Min() T
	Max() T
}

type WriteFileInfo added in v0.3.0 

type WriteFileInfo struct {
	FileSchema       *iceberg.Schema
	Spec             iceberg.PartitionSpec
	FileName         string
	StatsCols        map[int]StatisticsCollector
	ColMapping       map[string]int
	VariantFieldIDs  map[int]struct{}
	WriteProps       any
	RowGroupBytes    int64
	Content          iceberg.ManifestEntryContent
	EqualityFieldIDs []int
	SortOrderID      int
}

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.