internal

package
v0.0.0-...-2094565 Latest Latest
Warning

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

 Go to latest Published: Aug 19, 2026 License: Apache-2.0 Imports: 45 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 

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 

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 

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

func BuildPartitionMatchPredicate 

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/1215): 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 

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 

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 MapExec 

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 

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 

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

func ShredRecordVariants 

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 

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 

func TablePropertiesFromContext(ctx context.Context) iceberg.Properties

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

func TruncateUpperBoundBinary 

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

func TruncateUpperBoundText 

func TruncateUpperBoundText(s string, trunc int) string

func ValidateParquetWriteProperties 

func ValidateParquetWriteProperties(props iceberg.Properties) error

func VariantFieldIDsFromSchema 

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

func WithTableProperties 

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 

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 

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 

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

func (*DataFileStatistics) ToDataFile 

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 

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 

func FormatFromFileName(fileName string) FileFormat

func GetFileFormat 

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 

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 

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 

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 

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 

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 

type Metadata any

type MetricModeType 

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

type MetricsMode 

type MetricsMode struct {
	Typ MetricModeType
	Len int
}

func MatchMetricsMode 

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 

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 

func (w *ParquetFileWriter) Abort() error

func (*ParquetFileWriter) BytesWritten 

func (w *ParquetFileWriter) BytesWritten() int64

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

func (*ParquetFileWriter) Close 

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 

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

Write appends a record batch to the Parquet file.

type ParquetRowGroupTester 

type ParquetRowGroupTester struct {
	StatsFn    func(*metadata.RowGroupMetaData, []int) (bool, error)
	BloomPreds []RowGroupBloomPred // nil = no bloom filter pass
	// 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 and bloom filter row group pruning. Pass it as the tester argument to wrapPqArrowReader.GetRecords.

type RowGroupBloomPred 

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 RowGroupSpan 

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 

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

type StatsAgg 

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

type TypedStats 

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

type WriteFileInfo 

type WriteFileInfo struct {
	FileSchema       *iceberg.Schema
	Spec             iceberg.PartitionSpec
	FileName         string
	StatsCols        map[int]StatisticsCollector
	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.