You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2025/03/17 15:01:20 UTC
(iceberg-go) branch main updated: feat(table): Basic Transaction and AddFiles (#330)
This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 66d839e feat(table): Basic Transaction and AddFiles (#330)
66d839e is described below
commit 66d839ed0347561bdafa968a33af39d1c88d8823
Author: Matt Topol <zo...@gmail.com>
AuthorDate: Mon Mar 17 11:01:15 2025 -0400
feat(table): Basic Transaction and AddFiles (#330)
And here we go! Snapshot producers, basic Transaction object, and a
basic implementation of adding files to a table!
Complete with unit tests and a general TableWritingTestSuite.
---
literals.go | 3 +-
literals_test.go | 5 +
table/arrow_utils.go | 601 ++++++++++++++++++++++++++++++++++++-
table/arrow_utils_internal_test.go | 297 ++++++++++++++++--
table/internal/utils.go | 60 ++++
table/metadata.go | 51 ++++
table/properties.go | 18 +-
table/snapshot_producers.go | 384 ++++++++++++++++++++++++
table/snapshots_internal_test.go | 11 +
table/table.go | 62 ++++
table/table_test.go | 243 +++++++++++++++
table/transaction.go | 193 ++++++++++++
types.go | 4 +-
13 files changed, 1905 insertions(+), 27 deletions(-)
diff --git a/literals.go b/literals.go
index 06ac8de..0671e77 100644
--- a/literals.go
+++ b/literals.go
@@ -1215,8 +1215,9 @@ func (d DecimalLiteral) MarshalBinary() (data []byte, err error) {
// stored as unscaled value in two's compliment big-endian values
// using the minimum number of bytes for the values
n := decimal128.Num(d.Val).BigInt()
+ minBytes := (n.BitLen() + 8) / 8
// bytes gives absolute value as big-endian bytes
- data = n.Bytes()
+ data = n.FillBytes(make([]byte, minBytes))
if n.Sign() < 0 {
// convert to 2's complement for negative value
for i, v := range data {
diff --git a/literals_test.go b/literals_test.go
index 96c6606..20e21e6 100644
--- a/literals_test.go
+++ b/literals_test.go
@@ -840,6 +840,11 @@ func TestUnmarshalBinary(t *testing.T) {
[]byte{0xff, 0xed, 0x29, 0x79},
iceberg.DecimalLiteral{Scale: 4, Val: decimal128.FromI64(-1234567)},
},
+ {
+ iceberg.DecimalTypeOf(8, 2),
+ []byte{0x00, 0xbc, 0x61, 0x4e},
+ iceberg.DecimalLiteral{Scale: 2, Val: decimal128.FromI64(12345678)},
+ },
}
for _, tt := range tests {
diff --git a/table/arrow_utils.go b/table/arrow_utils.go
index 752f677..f987bb4 100644
--- a/table/arrow_utils.go
+++ b/table/arrow_utils.go
@@ -19,21 +19,31 @@ package table
import (
"context"
+ "errors"
"fmt"
+ "iter"
"maps"
"regexp"
"slices"
"strconv"
"strings"
+ "unsafe"
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/compute"
+ "github.com/apache/arrow-go/v18/arrow/decimal128"
"github.com/apache/arrow-go/v18/arrow/extensions"
"github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet/file"
"github.com/apache/arrow-go/v18/parquet/metadata"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/internal"
+ iceio "github.com/apache/iceberg-go/io"
+ tblutils "github.com/apache/iceberg-go/table/internal"
+ "github.com/google/uuid"
+ "github.com/pterm/pterm"
)
// constants to look for as Keys in Arrow field metadata
@@ -900,6 +910,139 @@ func ToRequestedSchema(ctx context.Context, requested, fileSchema *iceberg.Schem
return out, nil
}
+type schemaCompatVisitor struct {
+ provided *iceberg.Schema
+
+ errorData pterm.TableData
+}
+
+func checkSchemaCompat(requested, provided *iceberg.Schema) error {
+ sc := &schemaCompatVisitor{
+ provided: provided,
+ errorData: pterm.TableData{{"", "Table Field", "Requested Field"}},
+ }
+
+ _, compat := iceberg.PreOrderVisit(requested, sc)
+
+ return compat
+}
+
+func checkArrowSchemaCompat(requested *iceberg.Schema, provided *arrow.Schema, downcastNanoToMicro bool) error {
+ mapping := requested.NameMapping()
+ providedSchema, err := ArrowSchemaToIceberg(provided, downcastNanoToMicro, mapping)
+ if err != nil {
+ return err
+ }
+
+ return checkSchemaCompat(requested, providedSchema)
+}
+
+func (sc *schemaCompatVisitor) isFieldCompat(lhs iceberg.NestedField) bool {
+ rhs, ok := sc.provided.FindFieldByID(lhs.ID)
+ if !ok {
+ if lhs.Required {
+ sc.errorData = append(sc.errorData,
+ []string{"❌", lhs.String(), "missing"})
+
+ return false
+ }
+ sc.errorData = append(sc.errorData,
+ []string{"✅", lhs.String(), "missing"})
+
+ return true
+ }
+
+ if lhs.Required && !rhs.Required {
+ sc.errorData = append(sc.errorData,
+ []string{"❌", lhs.String(), rhs.String()})
+
+ return false
+ }
+
+ if lhs.Type.Equals(rhs.Type) {
+ sc.errorData = append(sc.errorData,
+ []string{"✅", lhs.String(), rhs.String()})
+
+ return true
+ }
+
+ // we only check that parent node is also of the same type
+ // we check the type of the child nodes as we traverse them later
+ switch lhs.Type.(type) {
+ case *iceberg.StructType:
+ if rhs, ok := rhs.Type.(*iceberg.StructType); ok {
+ sc.errorData = append(sc.errorData,
+ []string{"✅", lhs.String(), rhs.String()})
+
+ return true
+ }
+ case *iceberg.ListType:
+ if rhs, ok := rhs.Type.(*iceberg.ListType); ok {
+ sc.errorData = append(sc.errorData,
+ []string{"✅", lhs.String(), rhs.String()})
+
+ return true
+ }
+ case *iceberg.MapType:
+ if rhs, ok := rhs.Type.(*iceberg.MapType); ok {
+ sc.errorData = append(sc.errorData,
+ []string{"✅", lhs.String(), rhs.String()})
+
+ return true
+ }
+ }
+
+ if _, err := iceberg.PromoteType(rhs.Type, lhs.Type); err != nil {
+ sc.errorData = append(sc.errorData,
+ []string{"❌", lhs.String(), rhs.String()})
+
+ return false
+ }
+
+ sc.errorData = append(sc.errorData,
+ []string{"✅", lhs.String(), rhs.String()})
+
+ return true
+}
+
+func (sc *schemaCompatVisitor) Schema(s *iceberg.Schema, v func() bool) bool {
+ if !v() {
+ pterm.DisableColor()
+ tbl := pterm.DefaultTable.WithHasHeader(true).WithData(sc.errorData)
+ tbl.Render()
+ txt, _ := tbl.Srender()
+ pterm.EnableColor()
+ panic("mismatch in fields:\n" + txt)
+ }
+
+ return true
+}
+
+func (sc *schemaCompatVisitor) Struct(st iceberg.StructType, v []func() bool) bool {
+ out := true
+ for _, res := range v {
+ out = res() && out
+ }
+
+ return out
+}
+
+func (sc *schemaCompatVisitor) Field(n iceberg.NestedField, v func() bool) bool {
+ return sc.isFieldCompat(n) && v()
+}
+
+func (sc *schemaCompatVisitor) List(l iceberg.ListType, v func() bool) bool {
+ return sc.isFieldCompat(l.ElementField()) && v()
+}
+
+func (sc *schemaCompatVisitor) Map(m iceberg.MapType, vk, vv func() bool) bool {
+ return sc.isFieldCompat(m.KeyField()) && sc.isFieldCompat(m.ValueField()) && vk() && vv()
+}
+
+func (sc *schemaCompatVisitor) Primitive(p iceberg.PrimitiveType) bool {
+ return true
+}
+
func must[T any](v T, err error) T {
if err != nil {
panic(err)
@@ -908,6 +1051,344 @@ func must[T any](v T, err error) T {
return v
}
+func primitiveToPhysicalType(typ iceberg.Type) string {
+ switch t := typ.(type) {
+ case iceberg.BooleanType:
+ return "BOOLEAN"
+ case iceberg.Int32Type:
+ return "INT32"
+ case iceberg.Int64Type:
+ return "INT64"
+ case iceberg.Float32Type:
+ return "FLOAT"
+ case iceberg.Float64Type:
+ return "DOUBLE"
+ case iceberg.DateType:
+ return "INT32"
+ case iceberg.TimeType:
+ return "INT64"
+ case iceberg.TimestampType:
+ return "INT64"
+ case iceberg.TimestampTzType:
+ return "INT64"
+ case iceberg.StringType:
+ return "BYTE_ARRAY"
+ case iceberg.UUIDType:
+ return "FIXED_LEN_BYTE_ARRAY"
+ case iceberg.FixedType:
+ return "FIXED_LEN_BYTE_ARRAY"
+ case iceberg.BinaryType:
+ return "BYTE_ARRAY"
+ case iceberg.DecimalType:
+ switch {
+ case t.Precision() <= 9:
+ return "INT32"
+ case t.Precision() <= 18:
+ return "INT64"
+ default:
+ return "FIXED_LEN_BYTE_ARRAY"
+ }
+ default:
+ panic(fmt.Errorf("expected primitive type, got: %s", typ))
+ }
+}
+
+type typedStat[T iceberg.LiteralType] interface {
+ Min() T
+ Max() T
+}
+
+type wrappedBinaryStats struct {
+ *metadata.ByteArrayStatistics
+}
+
+func (w wrappedBinaryStats) Min() []byte {
+ return w.ByteArrayStatistics.Min()
+}
+
+func (w wrappedBinaryStats) Max() []byte {
+ return w.ByteArrayStatistics.Max()
+}
+
+type wrappedStringStats struct {
+ *metadata.ByteArrayStatistics
+}
+
+func (w wrappedStringStats) Min() string {
+ data := w.ByteArrayStatistics.Min()
+
+ return unsafe.String(unsafe.SliceData(data), len(data))
+}
+
+func (w wrappedStringStats) Max() string {
+ data := w.ByteArrayStatistics.Max()
+
+ return unsafe.String(unsafe.SliceData(data), len(data))
+}
+
+type wrappedUUIDStats struct {
+ *metadata.FixedLenByteArrayStatistics
+}
+
+func (w wrappedUUIDStats) Min() uuid.UUID {
+ uid, err := uuid.FromBytes(w.FixedLenByteArrayStatistics.Min())
+ if err != nil {
+ panic(err)
+ }
+
+ return uid
+}
+
+func (w wrappedUUIDStats) Max() uuid.UUID {
+ uid, err := uuid.FromBytes(w.FixedLenByteArrayStatistics.Max())
+ if err != nil {
+ panic(err)
+ }
+
+ return uid
+}
+
+type wrappedFLBAStats struct {
+ *metadata.FixedLenByteArrayStatistics
+}
+
+func (w wrappedFLBAStats) Min() []byte {
+ return w.FixedLenByteArrayStatistics.Min()
+}
+
+func (w wrappedFLBAStats) Max() []byte {
+ return w.FixedLenByteArrayStatistics.Max()
+}
+
+type wrappedDecStats struct {
+ *metadata.FixedLenByteArrayStatistics
+ scale int
+}
+
+func (w wrappedDecStats) Min() iceberg.Decimal {
+ dec, err := tblutils.BigEndianToDecimal(w.FixedLenByteArrayStatistics.Min())
+ if err != nil {
+ panic(err)
+ }
+
+ return iceberg.Decimal{Val: dec, Scale: w.scale}
+}
+
+func (w wrappedDecStats) Max() iceberg.Decimal {
+ dec, err := tblutils.BigEndianToDecimal(w.FixedLenByteArrayStatistics.Max())
+ if err != nil {
+ panic(err)
+ }
+
+ return iceberg.Decimal{Val: dec, Scale: w.scale}
+}
+
+type statsAgg interface {
+ update(stats metadata.TypedStatistics)
+ minAsBytes() ([]byte, error)
+ maxAsBytes() ([]byte, error)
+}
+
+type statsAggregator[T iceberg.LiteralType] struct {
+ curMin iceberg.TypedLiteral[T]
+ curMax iceberg.TypedLiteral[T]
+
+ cmp iceberg.Comparator[T]
+ primitiveType iceberg.PrimitiveType
+ truncLen int
+}
+
+func newStatAgg[T iceberg.LiteralType](typ iceberg.PrimitiveType, trunc int) statsAgg {
+ var z T
+
+ return &statsAggregator[T]{
+ primitiveType: typ,
+ truncLen: trunc,
+ cmp: iceberg.NewLiteral(z).(iceberg.TypedLiteral[T]).Comparator(),
+ }
+}
+
+func createStatsAgg(typ iceberg.PrimitiveType, physicalTypeStr string, truncLen int) (statsAgg, error) {
+ expectedPhysical := primitiveToPhysicalType(typ)
+ if physicalTypeStr != expectedPhysical {
+ switch {
+ case physicalTypeStr == "INT32" && expectedPhysical == "INT64":
+ case physicalTypeStr == "FLOAT" && expectedPhysical == "DOUBLE":
+ default:
+ return nil, fmt.Errorf("unexpected physical type %s for %s, expected %s",
+ physicalTypeStr, typ, expectedPhysical)
+ }
+ }
+
+ switch physicalTypeStr {
+ case "BOOLEAN":
+ return newStatAgg[bool](typ, truncLen), nil
+ case "INT32":
+ switch typ.(type) {
+ case iceberg.DecimalType:
+ return &decAsIntAgg[int32]{
+ newStatAgg[int32](typ, truncLen).(*statsAggregator[int32]),
+ }, nil
+ }
+
+ return newStatAgg[int32](typ, truncLen), nil
+ case "INT64":
+ switch typ.(type) {
+ case iceberg.DecimalType:
+ return &decAsIntAgg[int64]{
+ newStatAgg[int64](typ, truncLen).(*statsAggregator[int64]),
+ }, nil
+ }
+
+ return newStatAgg[int64](typ, truncLen), nil
+ case "FLOAT":
+ return newStatAgg[float32](typ, truncLen), nil
+ case "DOUBLE":
+ return newStatAgg[float64](typ, truncLen), nil
+ case "FIXED_LEN_BYTE_ARRAY":
+ switch typ.(type) {
+ case iceberg.UUIDType:
+ return newStatAgg[uuid.UUID](typ, truncLen), nil
+ case iceberg.DecimalType:
+ return newStatAgg[iceberg.Decimal](typ, truncLen), nil
+ default:
+ return newStatAgg[[]byte](typ, truncLen), nil
+ }
+ case "BYTE_ARRAY":
+ if typ.Equals(iceberg.PrimitiveTypes.String) {
+ return newStatAgg[string](typ, truncLen), nil
+ }
+
+ return newStatAgg[[]byte](typ, truncLen), nil
+ default:
+ return nil, fmt.Errorf("unsupported physical type: %s", physicalTypeStr)
+ }
+}
+
+func (s *statsAggregator[T]) update(stats metadata.TypedStatistics) {
+ st := stats.(typedStat[T])
+ s.updateMin(st.Min())
+ s.updateMax(st.Max())
+}
+
+func (s *statsAggregator[T]) toBytes(val iceberg.Literal) ([]byte, error) {
+ v, err := val.To(s.primitiveType)
+ if err != nil {
+ return nil, err
+ }
+
+ return v.MarshalBinary()
+}
+
+func (s *statsAggregator[T]) updateMin(val T) {
+ if s.curMin == nil {
+ s.curMin = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+ } else {
+ if s.cmp(val, s.curMin.Value()) < 0 {
+ s.curMin = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+ }
+ }
+}
+
+func (s *statsAggregator[T]) updateMax(val T) {
+ if s.curMax == nil {
+ s.curMax = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+ } else {
+ if s.cmp(val, s.curMax.Value()) > 0 {
+ s.curMax = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+ }
+ }
+}
+
+func (s *statsAggregator[T]) minAsBytes() ([]byte, error) {
+ if s.curMin == nil {
+ return nil, nil
+ }
+
+ if s.truncLen > 0 {
+ return s.toBytes((&iceberg.TruncateTransform{Width: s.truncLen}).
+ Apply(iceberg.Optional[iceberg.Literal]{Valid: true, Val: s.curMin}).Val)
+ }
+
+ return s.toBytes(s.curMin)
+}
+
+func (s *statsAggregator[T]) maxAsBytes() ([]byte, error) {
+ if s.curMax == nil {
+ return nil, nil
+ }
+
+ if s.truncLen <= 0 {
+ return s.toBytes(s.curMax)
+ }
+
+ switch s.primitiveType.(type) {
+ case iceberg.StringType:
+ if !s.curMax.Type().Equals(s.primitiveType) {
+ return nil, errors.New("expected current max to be a string")
+ }
+
+ curMax := any(s.curMax.Value()).(string)
+ result := truncateUpperBoundText(curMax, s.truncLen)
+ if result != "" {
+ return s.toBytes(iceberg.StringLiteral(result))
+ }
+
+ return nil, nil
+ case iceberg.BinaryType:
+ if !s.curMax.Type().Equals(s.primitiveType) {
+ return nil, errors.New("expected current max to be a binary")
+ }
+
+ curMax := any(s.curMax.Value()).([]byte)
+ result := truncateUpperBoundBinary(curMax, s.truncLen)
+ if len(result) > 0 {
+ return s.toBytes(iceberg.BinaryLiteral(result))
+ }
+
+ return nil, nil
+ default:
+ return nil, fmt.Errorf("%s cannot be truncated for upper bound", s.primitiveType)
+ }
+}
+
+type decAsIntAgg[T int32 | int64] struct {
+ *statsAggregator[T]
+}
+
+func (s *decAsIntAgg[T]) minAsBytes() ([]byte, error) {
+ if s.curMin == nil {
+ return nil, nil
+ }
+
+ lit := iceberg.DecimalLiteral(iceberg.Decimal{
+ Val: decimal128.FromI64(int64(s.curMin.Value())),
+ Scale: s.primitiveType.(iceberg.DecimalType).Scale(),
+ })
+ if s.truncLen > 0 {
+ return s.toBytes((&iceberg.TruncateTransform{Width: s.truncLen}).
+ Apply(iceberg.Optional[iceberg.Literal]{Valid: true, Val: lit}).Val)
+ }
+
+ return s.toBytes(lit)
+}
+
+func (s *decAsIntAgg[T]) maxAsBytes() ([]byte, error) {
+ if s.curMax == nil {
+ return nil, nil
+ }
+
+ lit := iceberg.DecimalLiteral(iceberg.Decimal{
+ Val: decimal128.FromI64(int64(s.curMax.Value())),
+ Scale: s.primitiveType.(iceberg.DecimalType).Scale(),
+ })
+ if s.truncLen <= 0 {
+ return s.toBytes(lit)
+ }
+
+ return nil, fmt.Errorf("%s cannot be truncated for upper bound", s.primitiveType)
+}
+
type metricModeType string
const (
@@ -1146,17 +1627,45 @@ type dataFileStatistics struct {
valueCounts map[int]int64
nullValueCounts map[int]int64
nanValueCounts map[int]int64
- // column aggregates
- splitOffsets []int64
+ colAggs map[int]statsAgg
+ splitOffsets []int64
}
func (d *dataFileStatistics) toDataFile(spec iceberg.PartitionSpec, path string, format iceberg.FileFormat, filesize int64) (iceberg.DataFile, error) {
bldr, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData,
- path, iceberg.ParquetFile, nil, d.recordCount, filesize)
+ path, format, nil, d.recordCount, filesize)
if err != nil {
return nil, err
}
+ lowerBounds := make(map[int][]byte)
+ upperBounds := make(map[int][]byte)
+
+ for fieldID, agg := range d.colAggs {
+ min, err := agg.minAsBytes()
+ if err != nil {
+ return nil, err
+ }
+
+ max, err := agg.maxAsBytes()
+ if err != nil {
+ return nil, err
+ }
+
+ if len(min) > 0 {
+ lowerBounds[fieldID] = min
+ }
+ if len(max) > 0 {
+ upperBounds[fieldID] = max
+ }
+ }
+
+ if len(lowerBounds) > 0 {
+ bldr.LowerBoundValues(lowerBounds)
+ }
+ if len(upperBounds) > 0 {
+ bldr.UpperBoundValues(upperBounds)
+ }
bldr.ColumnSizes(d.colSizes)
bldr.ValueCounts(d.valueCounts)
bldr.NullValueCounts(d.nullValueCounts)
@@ -1174,6 +1683,7 @@ func dataFileStatsFromParquetMetadata(pqmeta *metadata.FileMetaData, statsCols m
nullValueCounts = make(map[int]int64)
nanValueCounts = make(map[int]int64)
invalidateCol = make(map[int]struct{})
+ colAggs = make(map[int]statsAgg)
)
for rg := range pqmeta.NumRowGroups() {
@@ -1227,11 +1737,36 @@ func dataFileStatsFromParquetMetadata(pqmeta *metadata.FileMetaData, statsCols m
nullValueCounts[fieldID] += stats.NullCount()
}
- if statsCol.mode.typ == metricModeCounts {
+ if statsCol.mode.typ == metricModeCounts || !stats.HasMinMax() {
continue
}
- // TODO stats aggregator!
+ agg, ok := colAggs[fieldID]
+ if !ok {
+ agg, err = createStatsAgg(statsCol.icebergTyp, stats.Type().String(), statsCol.mode.len)
+ if err != nil {
+ panic(err)
+ }
+
+ colAggs[fieldID] = agg
+ }
+
+ switch typ := statsCol.icebergTyp.(type) {
+ case iceberg.BinaryType:
+ stats = wrappedBinaryStats{stats.(*metadata.ByteArrayStatistics)}
+ case iceberg.UUIDType:
+ stats = wrappedUUIDStats{stats.(*metadata.FixedLenByteArrayStatistics)}
+ case iceberg.StringType:
+ stats = wrappedStringStats{stats.(*metadata.ByteArrayStatistics)}
+ case iceberg.FixedType:
+ stats = wrappedFLBAStats{stats.(*metadata.FixedLenByteArrayStatistics)}
+ case iceberg.DecimalType:
+ if typ.Precision() > 18 {
+ stats = wrappedDecStats{stats.(*metadata.FixedLenByteArrayStatistics), typ.Scale()}
+ }
+ }
+
+ agg.update(stats)
}
}
@@ -1242,6 +1777,11 @@ func dataFileStatsFromParquetMetadata(pqmeta *metadata.FileMetaData, statsCols m
return ok
})
+ maps.DeleteFunc(colAggs, func(fieldID int, _ statsAgg) bool {
+ _, ok := invalidateCol[fieldID]
+
+ return ok
+ })
return &dataFileStatistics{
recordCount: pqmeta.GetNumRows(),
@@ -1250,5 +1790,56 @@ func dataFileStatsFromParquetMetadata(pqmeta *metadata.FileMetaData, statsCols m
nullValueCounts: nullValueCounts,
nanValueCounts: nanValueCounts,
splitOffsets: splitOffsets,
+ colAggs: colAggs,
+ }
+}
+
+func parquetFilesToDataFiles(fileIO iceio.IO, meta *MetadataBuilder, paths iter.Seq[string]) iter.Seq2[iceberg.DataFile, error] {
+ return func(yield func(iceberg.DataFile, error) bool) {
+ defer func() {
+ if r := recover(); r != nil {
+ switch e := r.(type) {
+ case string:
+ yield(nil, fmt.Errorf("error encountered during parquet file conversion: %s", e))
+ case error:
+ yield(nil, fmt.Errorf("error encountered during parquet file conversion: %w", e))
+ }
+ }
+ }()
+
+ for filePath := range paths {
+ inputFile := must(fileIO.Open(filePath))
+ defer inputFile.Close()
+
+ rdr := must(file.NewParquetReader(inputFile))
+ defer rdr.Close()
+
+ arrRdr := must(pqarrow.NewFileReader(rdr, pqarrow.ArrowReadProperties{}, memory.DefaultAllocator))
+ arrSchema := must(arrRdr.Schema())
+
+ if hasIDs := must(VisitArrowSchema(arrSchema, hasIDs{})); hasIDs {
+ yield(nil, fmt.Errorf("%w: cannot add file %s because it has field-ids. add-files only supports the addition of files without field_ids",
+ iceberg.ErrNotImplemented, filePath))
+
+ return
+ }
+
+ if err := checkArrowSchemaCompat(meta.CurrentSchema(), arrSchema, false); err != nil {
+ panic(err)
+ }
+
+ statistics := dataFileStatsFromParquetMetadata(rdr.MetaData(),
+ must(computeStatsPlan(meta.CurrentSchema(), meta.props)),
+ must(parquetPathToIDMapping(meta.CurrentSchema())))
+
+ df, err := statistics.toDataFile(meta.CurrentSpec(), filePath, iceberg.ParquetFile, rdr.MetaData().GetSourceFileSize())
+ if err != nil {
+ panic(err)
+ }
+
+ if !yield(df, nil) {
+ return
+ }
+ }
}
}
diff --git a/table/arrow_utils_internal_test.go b/table/arrow_utils_internal_test.go
index 85730d1..5584d50 100644
--- a/table/arrow_utils_internal_test.go
+++ b/table/arrow_utils_internal_test.go
@@ -21,22 +21,27 @@ import (
"bytes"
"cmp"
"math"
+ "math/big"
"slices"
+ "strings"
"testing"
+ "time"
"github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/decimal128"
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/arrow-go/v18/parquet"
"github.com/apache/arrow-go/v18/parquet/file"
"github.com/apache/arrow-go/v18/parquet/metadata"
"github.com/apache/arrow-go/v18/parquet/pqarrow"
"github.com/apache/iceberg-go"
+ "github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/stretchr/testify/suite"
)
-func constructTestTable(t *testing.T, writeStats bool) (*metadata.FileMetaData, Metadata) {
+func constructTestTable(t *testing.T, writeStats []string) (*metadata.FileMetaData, Metadata) {
tableMeta, err := ParseMetadataString(`{
"format-version": 2,
"location": "s3://bucket/test/location",
@@ -141,9 +146,19 @@ func constructTestTable(t *testing.T, writeStats bool) (*metadata.FileMetaData,
rec := bldr.NewRecord()
defer rec.Release()
+ var opts []parquet.WriterProperty
+ if len(writeStats) == 0 {
+ opts = append(opts, parquet.WithStats(true))
+ } else {
+ opts = append(opts, parquet.WithStats(false))
+ for _, stat := range writeStats {
+ opts = append(opts, parquet.WithStatsFor(stat, true))
+ }
+ }
+
var buf bytes.Buffer
wr, err := pqarrow.NewFileWriter(arrowSchema, &buf,
- parquet.NewWriterProperties(parquet.WithStats(writeStats)),
+ parquet.NewWriterProperties(opts...),
pqarrow.DefaultWriterProps())
require.NoError(t, err)
@@ -161,8 +176,8 @@ type FileStatsMetricsSuite struct {
suite.Suite
}
-func (suite *FileStatsMetricsSuite) getDataFile(meta iceberg.Properties) iceberg.DataFile {
- fileMeta, tableMeta := constructTestTable(suite.T(), true)
+func (suite *FileStatsMetricsSuite) getDataFile(meta iceberg.Properties, writeStats []string) iceberg.DataFile {
+ fileMeta, tableMeta := constructTestTable(suite.T(), writeStats)
require.NotNil(suite.T(), tableMeta)
require.NotNil(suite.T(), fileMeta)
@@ -190,12 +205,12 @@ func (suite *FileStatsMetricsSuite) getDataFile(meta iceberg.Properties) iceberg
}
func (suite *FileStatsMetricsSuite) TestRecordCount() {
- df := suite.getDataFile(nil)
+ df := suite.getDataFile(nil, nil)
suite.EqualValues(int64(4), df.Count())
}
func (suite *FileStatsMetricsSuite) TestValueCounts() {
- df := suite.getDataFile(nil)
+ df := suite.getDataFile(nil, nil)
suite.Len(df.ValueCounts(), 7)
suite.EqualValues(4, df.ValueCounts()[1])
suite.EqualValues(4, df.ValueCounts()[2])
@@ -207,7 +222,7 @@ func (suite *FileStatsMetricsSuite) TestValueCounts() {
}
func (suite *FileStatsMetricsSuite) TestColumnSizes() {
- df := suite.getDataFile(nil)
+ df := suite.getDataFile(nil, nil)
suite.Len(df.ColumnSizes(), 7)
suite.Greater(df.ColumnSizes()[1], int64(0))
suite.Greater(df.ColumnSizes()[2], int64(0))
@@ -217,13 +232,13 @@ func (suite *FileStatsMetricsSuite) TestColumnSizes() {
}
func (suite *FileStatsMetricsSuite) TestOffsets() {
- df := suite.getDataFile(nil)
+ df := suite.getDataFile(nil, nil)
suite.Len(df.SplitOffsets(), 1)
suite.EqualValues(4, df.SplitOffsets()[0])
}
func (suite *FileStatsMetricsSuite) TestNullValueCounts() {
- df := suite.getDataFile(nil)
+ df := suite.getDataFile(nil, nil)
suite.Len(df.NullValueCounts(), 7)
suite.EqualValues(1, df.NullValueCounts()[1])
suite.EqualValues(0, df.NullValueCounts()[2])
@@ -236,8 +251,23 @@ func (suite *FileStatsMetricsSuite) TestNullValueCounts() {
// pqarrow doesn't currently write the NaN counts
}
+func (suite *FileStatsMetricsSuite) TestBounds() {
+ df := suite.getDataFile(nil, nil)
+ suite.Len(df.LowerBoundValues(), 2)
+ suite.Equal([]byte("aaaaaaaaaaaaaaaa"), df.LowerBoundValues()[1])
+ lb, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.LowerBoundValues()[2])
+ suite.Require().NoError(err)
+ suite.Equal(float32(1.69), lb.(iceberg.Float32Literal).Value())
+
+ suite.Len(df.UpperBoundValues(), 2)
+ suite.Equal([]byte("zzzzzzzzzzzzzzz{"), df.UpperBoundValues()[1])
+ ub, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.UpperBoundValues()[2])
+ suite.Require().NoError(err)
+ suite.Equal(float32(100), ub.(iceberg.Float32Literal).Value())
+}
+
func (suite *FileStatsMetricsSuite) TestMetricsModeNone() {
- df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "none"})
+ df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "none"}, nil)
suite.Len(df.ValueCounts(), 0)
suite.Len(df.ColumnSizes(), 0)
suite.Len(df.NullValueCounts(), 0)
@@ -247,7 +277,7 @@ func (suite *FileStatsMetricsSuite) TestMetricsModeNone() {
}
func (suite *FileStatsMetricsSuite) TestMetricsModeCounts() {
- df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "counts"})
+ df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "counts"}, nil)
suite.Len(df.ValueCounts(), 7)
suite.Len(df.NullValueCounts(), 7)
suite.Len(df.NaNValueCounts(), 0)
@@ -256,31 +286,73 @@ func (suite *FileStatsMetricsSuite) TestMetricsModeCounts() {
}
func (suite *FileStatsMetricsSuite) TestMetricsModeFull() {
- df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "full"})
+ df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "full"}, nil)
suite.Len(df.ValueCounts(), 7)
suite.Len(df.NullValueCounts(), 7)
suite.Len(df.NaNValueCounts(), 0)
- // bound values not yet implemented
+ suite.Len(df.LowerBoundValues(), 2)
+ suite.Equal([]byte("aaaaaaaaaaaaaaaaaaaa"), df.LowerBoundValues()[1])
+ lb, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.LowerBoundValues()[2])
+ suite.Require().NoError(err)
+ suite.Equal(float32(1.69), lb.(iceberg.Float32Literal).Value())
+
+ suite.Len(df.UpperBoundValues(), 2)
+ suite.Equal([]byte("zzzzzzzzzzzzzzzzzzzz"), df.UpperBoundValues()[1])
+ ub, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.UpperBoundValues()[2])
+ suite.Require().NoError(err)
+ suite.Equal(float32(100), ub.(iceberg.Float32Literal).Value())
}
func (suite *FileStatsMetricsSuite) TestMetricsModeNonDefaultTrunc() {
- df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "truncate(2)"})
+ df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "truncate(2)"}, nil)
suite.Len(df.ValueCounts(), 7)
suite.Len(df.NullValueCounts(), 7)
suite.Len(df.NaNValueCounts(), 0)
- // bound values not yet implemented
+ suite.Len(df.LowerBoundValues(), 2)
+ suite.Equal([]byte("aa"), df.LowerBoundValues()[1])
+ lb, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.LowerBoundValues()[2])
+ suite.Require().NoError(err)
+ suite.Equal(float32(1.69), lb.(iceberg.Float32Literal).Value())
+
+ suite.Len(df.UpperBoundValues(), 2)
+ suite.Equal([]byte("z{"), df.UpperBoundValues()[1])
+ ub, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.UpperBoundValues()[2])
+ suite.Require().NoError(err)
+ suite.Equal(float32(100), ub.(iceberg.Float32Literal).Value())
}
func (suite *FileStatsMetricsSuite) TestColumnMetricsMode() {
df := suite.getDataFile(iceberg.Properties{
"write.metadata.metrics.default": "truncate(2)",
"write.metadata.metrics.column.strings": "none",
- })
+ }, nil)
suite.Len(df.ValueCounts(), 6)
suite.Len(df.NullValueCounts(), 6)
suite.Len(df.NaNValueCounts(), 0)
- // bound values not yet implemented
+
+ suite.Len(df.LowerBoundValues(), 1)
+ lb, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.LowerBoundValues()[2])
+ suite.Require().NoError(err)
+ suite.Equal(float32(1.69), lb.(iceberg.Float32Literal).Value())
+
+ suite.Len(df.UpperBoundValues(), 1)
+ ub, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.UpperBoundValues()[2])
+ suite.Require().NoError(err)
+ suite.Equal(float32(100), ub.(iceberg.Float32Literal).Value())
+}
+
+func (suite *FileStatsMetricsSuite) TestReadMissingStats() {
+ df := suite.getDataFile(nil, []string{"strings"})
+
+ stringColIdx := 1
+ suite.Len(df.LowerBoundValues(), 1)
+ suite.Equal([]byte("aaaaaaaaaaaaaaaa"), df.LowerBoundValues()[stringColIdx])
+ suite.Len(df.UpperBoundValues(), 1)
+ suite.Equal([]byte("zzzzzzzzzzzzzzz{"), df.UpperBoundValues()[stringColIdx])
+
+ suite.Len(df.NullValueCounts(), 1)
+ suite.EqualValues(1, df.NullValueCounts()[stringColIdx])
}
func TestFileMetrics(t *testing.T) {
@@ -408,3 +480,194 @@ func TestStatsTypes(t *testing.T) {
iceberg.PrimitiveTypes.Int32,
}, actual)
}
+
+func constructTestTablePrimitiveTypes(t *testing.T) (*metadata.FileMetaData, Metadata) {
+ tableMeta, err := ParseMetadataString(`{
+ "format-version": 2,
+ "location": "s3://bucket/test/location",
+ "last-column-id": 7,
+ "current-schema-id": 0,
+ "schemas": [
+ {
+ "type": "struct",
+ "schema-id": 0,
+ "fields": [
+ {"id": 1, "name": "booleans", "required": false, "type": "boolean"},
+ {"id": 2, "name": "ints", "required": false, "type": "int"},
+ {"id": 3, "name": "longs", "required": false, "type": "long"},
+ {"id": 4, "name": "floats", "required": false, "type": "float"},
+ {"id": 5, "name": "doubles", "required": false, "type": "double"},
+ {"id": 6, "name": "dates", "required": false, "type": "date"},
+ {"id": 7, "name": "times", "required": false, "type": "time"},
+ {"id": 8, "name": "timestamps", "required": false, "type": "timestamp"},
+ {"id": 9, "name": "timestamptzs", "required": false, "type": "timestamptz"},
+ {"id": 10, "name": "strings", "required": false, "type": "string"},
+ {"id": 11, "name": "uuids", "required": false, "type": "uuid"},
+ {"id": 12, "name": "binaries", "required": false, "type": "binary"},
+ {"id": 13, "name": "small_dec", "required": false, "type": "decimal(8, 2)"},
+ {"id": 14, "name": "med_dec", "required": false, "type": "decimal(16, 2)"},
+ {"id": 15, "name": "large_dec", "required": false, "type": "decimal(24, 2)"}
+ ]
+ }
+ ],
+ "last-updated-ms": -1,
+ "default-spec-id": 0,
+ "partition-specs": [{"spec-id": 0, "fields": []}],
+ "properties": {}
+ }`)
+ require.NoError(t, err)
+
+ arrowSchema, err := SchemaToArrowSchema(tableMeta.Schemas()[0], nil, true, false)
+ require.NoError(t, err)
+
+ rec, _, err := array.RecordFromJSON(memory.DefaultAllocator, arrowSchema, strings.NewReader(`[
+ {
+ "booleans": true,
+ "ints": 23,
+ "longs": 54,
+ "floats": 454.1223,
+ "doubles": 8542.12,
+ "dates": "2022-01-02",
+ "times": "17:30:34",
+ "timestamps": "2022-01-02T17:30:34.399",
+ "timestamptzs": "2022-01-02T17:30:34.399",
+ "strings": "hello",
+ "uuids": "`+uuid.NewMD5(uuid.NameSpaceDNS, []byte("foo")).String()+`",
+ "binaries": "aGVsbG8=",
+ "small_dec": "123456.78",
+ "med_dec": "12345678901234.56",
+ "large_dec": "1234567890123456789012.34"
+ },
+ {
+ "booleans": false,
+ "ints": 89,
+ "longs": 2,
+ "floats": 24342.29,
+ "doubles": -43.9,
+ "dates": "2023-02-04",
+ "times": "13:21:04",
+ "timestamps": "2023-02-04T13:21:04.354",
+ "timestamptzs": "2023-02-04T13:21:04.354",
+ "strings": "world",
+ "uuids": "`+uuid.NewMD5(uuid.NameSpaceDNS, []byte("bar")).String()+`",
+ "binaries": "d29ybGQ=",
+ "small_dec": "876543.21",
+ "med_dec": "65432109876543.21",
+ "large_dec": "4321098765432109876543.21"
+ }
+ ]`))
+ require.NoError(t, err)
+ defer rec.Release()
+
+ var buf bytes.Buffer
+ wr, err := pqarrow.NewFileWriter(arrowSchema, &buf,
+ parquet.NewWriterProperties(parquet.WithStats(true),
+ parquet.WithStoreDecimalAsInteger(true)),
+ pqarrow.DefaultWriterProps())
+ require.NoError(t, err)
+
+ require.NoError(t, wr.Write(rec))
+ require.NoError(t, wr.Close())
+
+ rdr, err := file.NewParquetReader(bytes.NewReader(buf.Bytes()))
+ require.NoError(t, err)
+ defer rdr.Close()
+
+ return rdr.MetaData(), tableMeta
+}
+
+func assertBounds[T iceberg.LiteralType](t *testing.T, bound []byte, typ iceberg.Type, expected T) {
+ lit, err := iceberg.LiteralFromBytes(typ, bound)
+ require.NoError(t, err)
+ assert.Equal(t, expected, lit.(iceberg.TypedLiteral[T]).Value())
+}
+
+func toDate(tm time.Time) iceberg.Date {
+ return iceberg.Date(tm.Truncate(24*time.Hour).Unix() / int64((time.Hour * 24).Seconds()))
+}
+
+func TestMetricsPrimitiveTypes(t *testing.T) {
+ meta, tblMeta := constructTestTablePrimitiveTypes(t)
+ require.NotNil(t, tblMeta)
+ require.NotNil(t, meta)
+
+ tblMeta.Properties()["write.metadata.metrics.default"] = "truncate(2)"
+
+ collector, err := computeStatsPlan(tblMeta.CurrentSchema(), tblMeta.Properties())
+ require.NoError(t, err)
+ mapping, err := parquetPathToIDMapping(tblMeta.CurrentSchema())
+ require.NoError(t, err)
+
+ stats := dataFileStatsFromParquetMetadata(meta, collector, mapping)
+ df, err := stats.toDataFile(tblMeta.PartitionSpec(), "fake-path.parquet",
+ iceberg.ParquetFile, meta.GetSourceFileSize())
+ require.NoError(t, err)
+
+ assert.Len(t, df.ValueCounts(), 15)
+ assert.Len(t, df.NullValueCounts(), 15)
+ assert.Len(t, df.NaNValueCounts(), 0)
+
+ assert.Len(t, df.LowerBoundValues(), 15)
+ assertBounds(t, df.LowerBoundValues()[1], iceberg.PrimitiveTypes.Bool, false)
+ assertBounds(t, df.LowerBoundValues()[2], iceberg.PrimitiveTypes.Int32, int32(23))
+ assertBounds(t, df.LowerBoundValues()[3], iceberg.PrimitiveTypes.Int64, int64(2))
+ assertBounds(t, df.LowerBoundValues()[4], iceberg.PrimitiveTypes.Float32, float32(454.1223))
+ assertBounds(t, df.LowerBoundValues()[5], iceberg.PrimitiveTypes.Float64, -43.9)
+ assertBounds(t, df.LowerBoundValues()[6], iceberg.PrimitiveTypes.Date,
+ toDate(time.Date(2022, time.January, 2, 0, 0, 0, 0, time.UTC)))
+ assertBounds(t, df.LowerBoundValues()[7], iceberg.PrimitiveTypes.Time,
+ iceberg.Time(time.Duration(13*time.Hour+21*time.Minute+4*time.Second).Microseconds()))
+ assertBounds(t, df.LowerBoundValues()[8], iceberg.PrimitiveTypes.Timestamp,
+ iceberg.Timestamp(time.Date(2022, time.January, 2, 17, 30, 34, 399000000, time.UTC).UnixMicro()))
+ assertBounds(t, df.LowerBoundValues()[9], iceberg.PrimitiveTypes.TimestampTz,
+ iceberg.Timestamp(time.Date(2022, time.January, 2, 17, 30, 34, 399000000, time.UTC).UnixMicro()))
+ assertBounds(t, df.LowerBoundValues()[10], iceberg.PrimitiveTypes.String, "he")
+ assertBounds(t, df.LowerBoundValues()[11], iceberg.PrimitiveTypes.UUID,
+ uuid.NewMD5(uuid.NameSpaceDNS, []byte("foo")))
+ assertBounds(t, df.LowerBoundValues()[12], iceberg.PrimitiveTypes.Binary, []byte("he"))
+ assertBounds(t, df.LowerBoundValues()[13], iceberg.DecimalTypeOf(8, 2), iceberg.Decimal{
+ Val: decimal128.FromI64(12345678),
+ Scale: 2,
+ })
+ assertBounds(t, df.LowerBoundValues()[14], iceberg.DecimalTypeOf(16, 2), iceberg.Decimal{
+ Val: decimal128.FromI64(1234567890123456),
+ Scale: 2,
+ })
+ expected, _ := (&big.Int{}).SetString("123456789012345678901234", 10)
+ assertBounds(t, df.LowerBoundValues()[15], iceberg.DecimalTypeOf(24, 2), iceberg.Decimal{
+ Val: decimal128.FromBigInt(expected),
+ Scale: 2,
+ })
+
+ assert.Len(t, df.UpperBoundValues(), 15)
+ assertBounds(t, df.UpperBoundValues()[1], iceberg.PrimitiveTypes.Bool, true)
+ assertBounds(t, df.UpperBoundValues()[2], iceberg.PrimitiveTypes.Int32, int32(89))
+ assertBounds(t, df.UpperBoundValues()[3], iceberg.PrimitiveTypes.Int64, int64(54))
+ assertBounds(t, df.UpperBoundValues()[4], iceberg.PrimitiveTypes.Float32, float32(24342.29))
+ assertBounds(t, df.UpperBoundValues()[5], iceberg.PrimitiveTypes.Float64, 8542.12)
+ assertBounds(t, df.UpperBoundValues()[6], iceberg.PrimitiveTypes.Date,
+ toDate(time.Date(2023, time.February, 4, 0, 0, 0, 0, time.UTC)))
+ assertBounds(t, df.UpperBoundValues()[7], iceberg.PrimitiveTypes.Time,
+ iceberg.Time(time.Duration(17*time.Hour+30*time.Minute+34*time.Second).Microseconds()))
+ assertBounds(t, df.UpperBoundValues()[8], iceberg.PrimitiveTypes.Timestamp,
+ iceberg.Timestamp(time.Date(2023, time.February, 4, 13, 21, 4, 354000000, time.UTC).UnixMicro()))
+ assertBounds(t, df.UpperBoundValues()[9], iceberg.PrimitiveTypes.TimestampTz,
+ iceberg.Timestamp(time.Date(2023, time.February, 4, 13, 21, 4, 354000000, time.UTC).UnixMicro()))
+ assertBounds(t, df.UpperBoundValues()[10], iceberg.PrimitiveTypes.String, "wp")
+ assertBounds(t, df.UpperBoundValues()[11], iceberg.PrimitiveTypes.UUID,
+ uuid.NewMD5(uuid.NameSpaceDNS, []byte("bar")))
+ assertBounds(t, df.UpperBoundValues()[12], iceberg.PrimitiveTypes.Binary, []byte("wp"))
+ assertBounds(t, df.UpperBoundValues()[13], iceberg.DecimalTypeOf(8, 2), iceberg.Decimal{
+ Val: decimal128.FromI64(87654321),
+ Scale: 2,
+ })
+ assertBounds(t, df.UpperBoundValues()[14], iceberg.DecimalTypeOf(16, 2), iceberg.Decimal{
+ Val: decimal128.FromI64(6543210987654321),
+ Scale: 2,
+ })
+ expectedUpper, _ := (&big.Int{}).SetString("432109876543210987654321", 10)
+ assertBounds(t, df.UpperBoundValues()[15], iceberg.DecimalTypeOf(24, 2), iceberg.Decimal{
+ Val: decimal128.FromBigInt(expectedUpper),
+ Scale: 2,
+ })
+}
diff --git a/table/internal/utils.go b/table/internal/utils.go
index 7b5f7d8..cab0d73 100644
--- a/table/internal/utils.go
+++ b/table/internal/utils.go
@@ -19,6 +19,11 @@ package internal
import (
"container/heap"
+ "encoding/binary"
+ "fmt"
+
+ "github.com/apache/arrow-go/v18/arrow/decimal"
+ "github.com/apache/arrow-go/v18/arrow/decimal128"
)
// Enumerated is a quick way to represent a sequenced value that can
@@ -79,3 +84,58 @@ func MakeSequencedChan[T any](bufferSize uint, source <-chan T, comesAfter, isNe
return out
}
+
+func u64FromBigEndianShifted(buf []byte) uint64 {
+ var bytes [8]byte
+ copy(bytes[8-len(buf):], buf)
+
+ return binary.BigEndian.Uint64(bytes[:])
+}
+
+func BigEndianToDecimal(buf []byte) (decimal.Decimal128, error) {
+ const (
+ minDecBytes = 1
+ maxDecBytes = 16
+ )
+
+ if len(buf) < minDecBytes || len(buf) > maxDecBytes {
+ return decimal.Decimal128{},
+ fmt.Errorf("invalid length for conversion to decimal: %d, must be between %d and %d",
+ len(buf), minDecBytes, maxDecBytes)
+ }
+
+ // big endian, so first byte is the MSB and host the sign bit
+ isNeg := int8(buf[0]) < 0
+
+ var hi, lo int64
+ // step 1. extract high bits
+ highBitsOffset := max(0, len(buf)-8)
+ highBits := u64FromBigEndianShifted(buf[:highBitsOffset])
+
+ if highBitsOffset == 8 {
+ hi = int64(highBits)
+ } else {
+ if isNeg && len(buf) < maxDecBytes {
+ hi = -1
+ }
+
+ hi = int64(uint64(hi) << (uint64(highBitsOffset) * 8))
+ hi |= int64(highBits)
+ }
+
+ // step 2. extract low bits
+ lowBitsOffset := min(len(buf), 8)
+ lowBits := u64FromBigEndianShifted(buf[highBitsOffset:])
+ if lowBitsOffset == 8 {
+ lo = int64(lowBits)
+ } else {
+ if isNeg && len(buf) < 8 {
+ lo = -1
+ }
+
+ lo = int64(uint64(lo) << (uint64(lowBitsOffset) * 8))
+ lo |= int64(lowBits)
+ }
+
+ return decimal128.New(hi, uint64(lo)), nil
+}
diff --git a/table/metadata.go b/table/metadata.go
index 6828979..454320e 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -18,6 +18,7 @@
package table
import (
+ "encoding/binary"
"encoding/json"
"errors"
"fmt"
@@ -43,6 +44,25 @@ const (
addSortOrderAction = "add-sort-order"
)
+func generateSnapshotID() int64 {
+ var (
+ rndUUID = uuid.New()
+ out [8]byte
+ )
+
+ for i := range 8 {
+ lhs, rhs := rndUUID[i], rndUUID[i+8]
+ out[i] = lhs ^ rhs
+ }
+
+ snapshotID := int64(binary.LittleEndian.Uint64(out[:]))
+ if snapshotID < 0 {
+ snapshotID = -snapshotID
+ }
+
+ return snapshotID
+}
+
// Metadata for an iceberg table as specified in the Iceberg spec
//
// https://iceberg.apache.org/spec/#iceberg-table-spec
@@ -214,6 +234,37 @@ func (b *MetadataBuilder) CurrentSchema() *iceberg.Schema {
return s
}
+func (b *MetadataBuilder) nextSequenceNumber() int64 {
+ if b.formatVersion > 1 {
+ if b.lastSequenceNumber == nil {
+ return 0
+ }
+
+ return *b.lastSequenceNumber + 1
+ }
+
+ return 0
+}
+
+func (b *MetadataBuilder) newSnapshotID() int64 {
+ snapshotID := generateSnapshotID()
+ for slices.ContainsFunc(b.snapshotList, func(s Snapshot) bool { return s.SnapshotID == snapshotID }) {
+ snapshotID = generateSnapshotID()
+ }
+
+ return snapshotID
+}
+
+func (b *MetadataBuilder) currentSnapshot() *Snapshot {
+ if b.currentSnapshotID == nil {
+ return nil
+ }
+
+ s, _ := b.SnapshotByID(*b.currentSnapshotID)
+
+ return s
+}
+
func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema, newLastColumnID int, initial bool) (*MetadataBuilder, error) {
if newLastColumnID < b.lastColumnId {
return nil, fmt.Errorf("%w: newLastColumnID %d, must be >= %d", iceberg.ErrInvalidArgument, newLastColumnID, b.lastColumnId)
diff --git a/table/properties.go b/table/properties.go
index e2f9e8c..0005049 100644
--- a/table/properties.go
+++ b/table/properties.go
@@ -27,8 +27,22 @@ const (
DefaultNameMappingKey = "schema.name-mapping.default"
- MetricsModeColumnConfPrefix = "write.metadata.metrics.column"
-
+ MetricsModeColumnConfPrefix = "write.metadata.metrics.column"
DefaultWriteMetricsModeKey = "write.metadata.metrics.default"
DefaultWriteMetricsModeDefault = "truncate(16)"
+
+ ManifestMergeEnabledKey = "commit.manifest-merge.enabled"
+ ManifestMergeEnabledDefault = false
+
+ ManifestTargetSizeBytesKey = "commit.manifest.target-size-bytes"
+ ManifestTargetSizeBytesDefault = 8 * 1024 * 1024 // 8 MB
+
+ ManifestMinMergeCountKey = "commit.manifest.min-count-to-merge"
+ ManifestMinMergeCountDefault = 100
+
+ WritePartitionSummaryLimitKey = "write.summary.partition-limit"
+ WritePartitionSummaryLimitDefault = 0
+
+ MetadataDeleteAfterCommitEnabledKey = "write.metadata.delete-after-commit.enabled"
+ MetadataDeleteAfterCommitEnabledDefault = false
)
diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
new file mode 100644
index 0000000..298a2f5
--- /dev/null
+++ b/table/snapshot_producers.go
@@ -0,0 +1,384 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "bytes"
+ "fmt"
+ "io"
+ "maps"
+ "slices"
+ "sync/atomic"
+ "unicode/utf8"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/internal"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/google/uuid"
+ "golang.org/x/sync/errgroup"
+)
+
+type producerImpl interface {
+ processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error)
+ existingManifests() ([]iceberg.ManifestFile, error)
+ deletedEntries() ([]iceberg.ManifestEntry, error)
+}
+
+func newManifestFileName(num int, commit uuid.UUID) string {
+ return fmt.Sprintf("%s-m%d.avro", commit, num)
+}
+
+func newManifestListFileName(snapshotID int64, attempt int, commit uuid.UUID) string {
+ // mimics behavior of java
+ // https://github.com/apache/iceberg/blob/c862b9177af8e2d83122220764a056f3b96fd00c/core/src/main/java/org/apache/iceberg/SnapshotProducer.java#L491
+ return fmt.Sprintf("snap-%d-%d-%s.avro", snapshotID, attempt, commit)
+}
+
+func newFastAppendFilesProducer(op Operation, txn *Transaction, fs iceio.WriteFileIO, commitUUID *uuid.UUID, snapshotProps iceberg.Properties) *snapshotProducer {
+ prod := createSnapshotProducer(op, txn, fs, commitUUID, snapshotProps)
+ prod.producerImpl = &fastAppendFiles{base: prod}
+
+ return prod
+}
+
+type fastAppendFiles struct {
+ base *snapshotProducer
+}
+
+func (fa *fastAppendFiles) processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
+ return manifests, nil
+}
+
+func (fa *fastAppendFiles) existingManifests() ([]iceberg.ManifestFile, error) {
+ existing := make([]iceberg.ManifestFile, 0)
+ if fa.base.parentSnapshotID > 0 {
+ previous, err := fa.base.txn.meta.SnapshotByID(fa.base.parentSnapshotID)
+ if err != nil {
+ return nil, fmt.Errorf("could not find parent snapshot %d", fa.base.parentSnapshotID)
+ }
+
+ manifests, err := previous.Manifests(fa.base.io)
+ if err != nil {
+ return nil, err
+ }
+
+ for _, m := range manifests {
+ if m.HasAddedFiles() || m.HasExistingFiles() || m.SnapshotID() == fa.base.snapshotID {
+ existing = append(existing, m)
+ }
+ }
+ }
+
+ return existing, nil
+}
+
+func (fa *fastAppendFiles) deletedEntries() ([]iceberg.ManifestEntry, error) {
+ return nil, nil
+}
+
+type snapshotProducer struct {
+ producerImpl
+
+ commitUuid uuid.UUID
+ io iceio.WriteFileIO
+ txn *Transaction
+ op Operation
+ snapshotID int64
+ parentSnapshotID int64
+ addedFiles []iceberg.DataFile
+ manifestCount atomic.Int32
+ deletedFiles map[string]iceberg.DataFile
+ snapshotProps iceberg.Properties
+}
+
+func createSnapshotProducer(op Operation, txn *Transaction, fs iceio.WriteFileIO, commitUUID *uuid.UUID, snapshotProps iceberg.Properties) *snapshotProducer {
+ var (
+ commit uuid.UUID
+ parentSnapshot int64 = -1
+ )
+
+ if commitUUID == nil {
+ commit = uuid.New()
+ } else {
+ commit = *commitUUID
+ }
+
+ if snap := txn.meta.currentSnapshot(); snap != nil {
+ parentSnapshot = snap.SnapshotID
+ }
+
+ return &snapshotProducer{
+ commitUuid: commit,
+ io: fs,
+ txn: txn,
+ op: op,
+ snapshotID: txn.meta.newSnapshotID(),
+ parentSnapshotID: parentSnapshot,
+ addedFiles: []iceberg.DataFile{},
+ deletedFiles: make(map[string]iceberg.DataFile),
+ snapshotProps: snapshotProps,
+ }
+}
+
+func (sp *snapshotProducer) spec(id int) iceberg.PartitionSpec {
+ if spec, _ := sp.txn.meta.GetSpecByID(id); spec != nil {
+ return *spec
+ }
+
+ return iceberg.NewPartitionSpec()
+}
+
+func (sp *snapshotProducer) appendDataFile(df iceberg.DataFile) *snapshotProducer {
+ sp.addedFiles = append(sp.addedFiles, df)
+
+ return sp
+}
+
+func (sp *snapshotProducer) newManifestOutput() (io.WriteCloser, string, error) {
+ provider, err := sp.txn.tbl.LocationProvider()
+ if err != nil {
+ return nil, "", err
+ }
+ fname := newManifestFileName(int(sp.manifestCount.Add(1)), sp.commitUuid)
+ filepath := provider.NewMetadataLocation(fname)
+ f, err := sp.io.Create(filepath)
+ if err != nil {
+ return nil, "", fmt.Errorf("could not create manifest file: %w", err)
+ }
+
+ return f, filepath, nil
+}
+
+func (sp *snapshotProducer) manifests() ([]iceberg.ManifestFile, error) {
+ var g errgroup.Group
+
+ results := [...][]iceberg.ManifestFile{nil, nil, nil}
+
+ if len(sp.addedFiles) > 0 {
+ g.Go(func() error {
+ out, path, err := sp.newManifestOutput()
+ if err != nil {
+ return err
+ }
+ defer out.Close()
+
+ counter := &internal.CountingWriter{W: out}
+
+ wr, err := iceberg.NewManifestWriter(sp.txn.meta.formatVersion, counter,
+ sp.txn.meta.CurrentSpec(), sp.txn.meta.CurrentSchema(),
+ sp.snapshotID)
+ if err != nil {
+ return err
+ }
+
+ for _, df := range sp.addedFiles {
+ err := wr.Add(iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &sp.snapshotID,
+ nil, nil, df))
+ if err != nil {
+ return err
+ }
+ }
+
+ mf, err := wr.ToManifestFile(path, counter.Count)
+ if err == nil {
+ results[0] = append(results[0], mf)
+ }
+
+ return err
+ })
+ }
+
+ deleted, err := sp.deletedEntries()
+ if err != nil {
+ return nil, err
+ }
+
+ if len(deleted) > 0 {
+ g.Go(func() error {
+ partitionGroups := map[int][]iceberg.ManifestEntry{}
+ for _, entry := range deleted {
+ specid := int(entry.DataFile().SpecID())
+
+ group := partitionGroups[specid]
+ partitionGroups[specid] = append(group, entry)
+ }
+
+ for specid, entries := range partitionGroups {
+ out, path, err := sp.newManifestOutput()
+ if err != nil {
+ return err
+ }
+ defer out.Close()
+
+ mf, err := iceberg.WriteManifest(path, out, sp.txn.meta.formatVersion,
+ sp.spec(specid), sp.txn.meta.CurrentSchema(), sp.snapshotID, entries)
+ if err != nil {
+ return err
+ }
+ results[1] = append(results[1], mf)
+ }
+
+ return nil
+ })
+ }
+
+ g.Go(func() error {
+ m, err := sp.existingManifests()
+ if err != nil {
+ return err
+ }
+ results[2] = m
+
+ return nil
+ })
+
+ if err := g.Wait(); err != nil {
+ return nil, err
+ }
+
+ manifests := slices.Concat(results[0], results[1], results[2])
+
+ return sp.processManifests(manifests)
+}
+
+func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error) {
+ var ssc SnapshotSummaryCollector
+ partitionSummaryLimit := sp.txn.meta.props.
+ GetInt(WritePartitionSummaryLimitKey, WritePartitionSummaryLimitDefault)
+ ssc.setPartitionSummaryLimit(partitionSummaryLimit)
+
+ currentSchema := sp.txn.meta.CurrentSchema()
+ partitionSpec := sp.txn.meta.CurrentSpec()
+ for _, df := range sp.addedFiles {
+ ssc.addFile(df, currentSchema, partitionSpec)
+ }
+
+ if len(sp.deletedFiles) > 0 {
+ specs := sp.txn.meta.specs
+ for _, df := range sp.deletedFiles {
+ ssc.removeFile(df, currentSchema, specs[df.SpecID()])
+ }
+ }
+
+ var previousSnapshot *Snapshot
+ if sp.parentSnapshotID > 0 {
+ previousSnapshot, _ = sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
+ }
+
+ var previousSummary iceberg.Properties
+ if previousSnapshot != nil {
+ previousSummary = previousSnapshot.Summary.Properties
+ }
+
+ summaryProps := ssc.build()
+ maps.Copy(summaryProps, props)
+
+ return updateSnapshotSummaries(Summary{
+ Operation: sp.op,
+ Properties: summaryProps,
+ }, previousSummary, sp.op == OpOverwrite)
+}
+
+func (sp *snapshotProducer) commit() ([]Update, []Requirement, error) {
+ newManifests, err := sp.manifests()
+ if err != nil {
+ return nil, nil, err
+ }
+
+ nextSequence := sp.txn.meta.nextSequenceNumber()
+ summary, err := sp.summary(sp.snapshotProps)
+ if err != nil {
+ return nil, nil, err
+ }
+
+ fname := newManifestListFileName(sp.snapshotID, 0, sp.commitUuid)
+ locProvider, err := sp.txn.tbl.LocationProvider()
+ if err != nil {
+ return nil, nil, err
+ }
+
+ manifestListFilePath := locProvider.NewMetadataLocation(fname)
+
+ var parentSnapshot *int64
+ if sp.parentSnapshotID > 0 {
+ parentSnapshot = &sp.parentSnapshotID
+ }
+
+ out, err := sp.io.Create(manifestListFilePath)
+ if err != nil {
+ return nil, nil, err
+ }
+ defer out.Close()
+
+ err = iceberg.WriteManifestList(sp.txn.meta.formatVersion, out,
+ sp.snapshotID, parentSnapshot, &nextSequence, newManifests)
+ if err != nil {
+ return nil, nil, err
+ }
+
+ snapshot := Snapshot{
+ SnapshotID: sp.snapshotID,
+ ParentSnapshotID: parentSnapshot,
+ SequenceNumber: nextSequence,
+ ManifestList: manifestListFilePath,
+ Summary: &summary,
+ SchemaID: &sp.txn.meta.currentSchemaID,
+ }
+
+ return []Update{
+ NewAddSnapshotUpdate(&snapshot),
+ NewSetSnapshotRefUpdate("main", sp.snapshotID, BranchRef, -1, -1, -1),
+ }, []Requirement{
+ AssertRefSnapshotID("main", sp.txn.meta.currentSnapshotID),
+ }, nil
+}
+
+func truncateUpperBoundText(s string, trunc int) string {
+ if trunc == utf8.RuneCountInString(s) {
+ return s
+ }
+
+ result := []rune(s)[:trunc]
+ for i := len(result) - 1; i >= 0; i-- {
+ next := result[i] + 1
+ if utf8.ValidRune(next) {
+ result[i] = next
+
+ return string(result)
+ }
+ }
+
+ return ""
+}
+
+func truncateUpperBoundBinary(val []byte, trunc int) []byte {
+ result := val[:trunc]
+ if bytes.Equal(result, val) {
+ return result
+ }
+
+ for i := len(result) - 1; i >= 0; i-- {
+ if result[i] < 255 {
+ result[i]++
+
+ return result
+ }
+ }
+
+ return nil
+}
diff --git a/table/snapshots_internal_test.go b/table/snapshots_internal_test.go
index c374178..5c8afe0 100644
--- a/table/snapshots_internal_test.go
+++ b/table/snapshots_internal_test.go
@@ -194,3 +194,14 @@ func TestInvalidOperation(t *testing.T) {
_, err := updateSnapshotSummaries(Summary{Operation: OpReplace}, nil, false)
assert.ErrorIs(t, err, iceberg.ErrNotImplemented)
}
+
+func TestTruncateUpperBoundString(t *testing.T) {
+ assert.Equal(t, "ab", truncateUpperBoundText("aaaa", 2))
+ // \u10FFFF\u10FFFF\x00
+ assert.Equal(t, "", truncateUpperBoundText("\xf4\x8f\xbf\xbf\xf4\x8f\xbf\xbf\x00", 2))
+}
+
+func TestTruncateUpperBoundBinary(t *testing.T) {
+ assert.Equal(t, []byte{0x01, 0x03}, truncateUpperBoundBinary([]byte{0x01, 0x02, 0x03}, 2))
+ assert.Nil(t, truncateUpperBoundBinary([]byte{0xff, 0xff, 0x00}, 2))
+}
diff --git a/table/table.go b/table/table.go
index b6b7786..2584b72 100644
--- a/table/table.go
+++ b/table/table.go
@@ -19,10 +19,13 @@ package table
import (
"context"
+ "fmt"
+ "iter"
"runtime"
"slices"
"github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/internal"
"github.com/apache/iceberg-go/io"
)
@@ -55,6 +58,7 @@ func (t Table) Schema() *iceberg.Schema { return t.metadata.Current
func (t Table) Spec() iceberg.PartitionSpec { return t.metadata.PartitionSpec() }
func (t Table) SortOrder() SortOrder { return t.metadata.SortOrder() }
func (t Table) Properties() iceberg.Properties { return t.metadata.Properties() }
+func (t Table) NameMapping() iceberg.NameMapping { return t.metadata.NameMapping() }
func (t Table) Location() string { return t.metadata.Location() }
func (t Table) CurrentSnapshot() *Snapshot { return t.metadata.CurrentSnapshot() }
func (t Table) SnapshotByID(id int64) *Snapshot { return t.metadata.SnapshotByID(id) }
@@ -72,6 +76,64 @@ func (t Table) LocationProvider() (LocationProvider, error) {
return LoadLocationProvider(t.metadataLocation, t.metadata.Properties())
}
+func (t Table) NewTransaction() *Transaction {
+ meta, _ := MetadataBuilderFromBase(t.metadata)
+
+ return &Transaction{
+ tbl: &t,
+ meta: meta,
+ reqs: []Requirement{},
+ }
+}
+
+func (t Table) doCommit(ctx context.Context, updates []Update, reqs []Requirement) (*Table, error) {
+ newMeta, newLoc, err := t.cat.CommitTable(ctx, &t, reqs, updates)
+ if err != nil {
+ return nil, err
+ }
+
+ if err := deleteOldMetadata(t.fs, t.metadata, newMeta); err != nil {
+ return nil, err
+ }
+
+ return New(t.identifier, newMeta, newLoc, t.fs, t.cat), nil
+}
+
+func getFiles(it iter.Seq[MetadataLogEntry]) iter.Seq[string] {
+ return func(yield func(string) bool) {
+ next, stop := iter.Pull(it)
+ defer stop()
+ for {
+ entry, ok := next()
+ if !ok {
+ return
+ }
+ if !yield(entry.MetadataFile) {
+ return
+ }
+ }
+ }
+}
+
+func deleteOldMetadata(fs io.IO, baseMeta, newMeta Metadata) error {
+ deleteAfterCommit := newMeta.Properties().GetBool(MetadataDeleteAfterCommitEnabledKey,
+ MetadataDeleteAfterCommitEnabledDefault)
+
+ if deleteAfterCommit {
+ removedPrevious := slices.Collect(getFiles(baseMeta.PreviousFiles()))
+ currentMetadata := slices.Collect(getFiles(newMeta.PreviousFiles()))
+ toRemove := internal.Difference(removedPrevious, currentMetadata)
+
+ for _, file := range toRemove {
+ if err := fs.Remove(file); err != nil {
+ return fmt.Errorf("failed to delete old metadata file %s: %w", file, err)
+ }
+ }
+ }
+
+ return nil
+}
+
type ScanOption func(*Scan)
func noopOption(*Scan) {}
diff --git a/table/table_test.go b/table/table_test.go
index f139ae2..8ae66d7 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -19,11 +19,21 @@ package table_test
import (
"bytes"
+ "context"
+ "fmt"
+ "strconv"
"testing"
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/internal"
+ iceio "github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/table"
+ "github.com/google/uuid"
+ "github.com/pterm/pterm"
"github.com/stretchr/testify/suite"
)
@@ -128,3 +138,236 @@ func (t *TableTestSuite) TestSnapshotByName() {
t.True(testSnapshot.Equals(*t.tbl.SnapshotByName("test")))
}
+
+type TableWritingTestSuite struct {
+ suite.Suite
+
+ tableSchema *iceberg.Schema
+ arrSchema *arrow.Schema
+ arrTbl arrow.Table
+ arrSchemaWithIDs *arrow.Schema
+ arrTblWithIDs arrow.Table
+ arrSchemaUpdated *arrow.Schema
+ arrTblUpdated arrow.Table
+
+ location string
+ formatVersion int
+}
+
+func (t *TableWritingTestSuite) SetupSuite() {
+ mem := memory.DefaultAllocator
+
+ t.tableSchema = iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.Bool},
+ iceberg.NestedField{ID: 2, Name: "bar", Type: iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 4, Name: "baz", Type: iceberg.PrimitiveTypes.Int32},
+ iceberg.NestedField{ID: 10, Name: "qux", Type: iceberg.PrimitiveTypes.Date})
+
+ t.arrSchema = arrow.NewSchema([]arrow.Field{
+ {Name: "foo", Type: arrow.FixedWidthTypes.Boolean, Nullable: true},
+ {Name: "bar", Type: arrow.BinaryTypes.String, Nullable: true},
+ {Name: "baz", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+ {Name: "qux", Type: arrow.PrimitiveTypes.Date32, Nullable: true},
+ }, nil)
+
+ var err error
+ t.arrTbl, err = array.TableFromJSON(mem, t.arrSchema, []string{
+ `[{"foo": true, "bar": "bar_string", "baz": 123, "qux": "2024-03-07"}]`,
+ })
+ t.Require().NoError(err)
+
+ t.arrSchemaWithIDs = arrow.NewSchema([]arrow.Field{
+ {
+ Name: "foo", Type: arrow.FixedWidthTypes.Boolean,
+ Metadata: arrow.MetadataFrom(map[string]string{"PARQUET:field_id": "1"}),
+ },
+ {
+ Name: "bar", Type: arrow.BinaryTypes.String,
+ Metadata: arrow.MetadataFrom(map[string]string{"PARQUET:field_id": "2"}),
+ },
+ {
+ Name: "baz", Type: arrow.PrimitiveTypes.Int32,
+ Metadata: arrow.MetadataFrom(map[string]string{"PARQUET:field_id": "3"}),
+ },
+ {
+ Name: "qux", Type: arrow.PrimitiveTypes.Date32,
+ Metadata: arrow.MetadataFrom(map[string]string{"PARQUET:field_id": "4"}),
+ },
+ }, nil)
+
+ t.arrTblWithIDs, err = array.TableFromJSON(mem, t.arrSchemaWithIDs, []string{
+ `[{"foo": true, "bar": "bar_string", "baz": 123, "qux": "2024-03-07"}]`,
+ })
+ t.Require().NoError(err)
+
+ t.arrSchemaUpdated = arrow.NewSchema([]arrow.Field{
+ {Name: "foo", Type: arrow.FixedWidthTypes.Boolean, Nullable: true},
+ {Name: "baz", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+ {Name: "qux", Type: arrow.PrimitiveTypes.Date32, Nullable: true},
+ {Name: "quux", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+ }, nil)
+
+ t.arrTblUpdated, err = array.TableFromJSON(mem, t.arrSchemaUpdated, []string{
+ `[{"foo": true, "baz": 123, "qux": "2024-03-07", "quux": 234}]`,
+ })
+ t.Require().NoError(err)
+}
+
+func (t *TableWritingTestSuite) SetupTest() {
+ t.location = t.T().TempDir()
+}
+
+func (t *TableWritingTestSuite) TearDownSuite() {
+ t.arrTbl.Release()
+ t.arrTblUpdated.Release()
+ t.arrTblWithIDs.Release()
+}
+
+func (t *TableWritingTestSuite) getMetadataLoc() string {
+ return fmt.Sprintf("%s/metadata/%05d-%s.metadata.json",
+ t.location, 1, uuid.New().String())
+}
+
+func (t *TableWritingTestSuite) writeParquet(fio iceio.WriteFileIO, filePath string, arrTbl arrow.Table) {
+ fo, err := fio.Create(filePath)
+ t.Require().NoError(err)
+
+ t.Require().NoError(pqarrow.WriteTable(arrTbl, fo, arrTbl.NumRows(),
+ nil, pqarrow.DefaultWriterProps()))
+}
+
+func (t *TableWritingTestSuite) createTable(identifier table.Identifier, formatVersion int, spec iceberg.PartitionSpec, sc *iceberg.Schema) *table.Table {
+ meta, err := table.NewMetadata(sc, &spec, table.UnsortedSortOrder,
+ t.getMetadataLoc(), iceberg.Properties{"format-version": strconv.Itoa(formatVersion)})
+ t.Require().NoError(err)
+
+ return table.New(identifier, meta, t.location, iceio.LocalFS{}, nil)
+}
+
+func (t *TableWritingTestSuite) TestAddFilesUnpartitioned() {
+ ident := table.Identifier{"default", "unpartitioned_table_v" + strconv.Itoa(t.formatVersion)}
+ tbl := t.createTable(ident, t.formatVersion,
+ *iceberg.UnpartitionedSpec, t.tableSchema)
+
+ t.NotNil(tbl)
+
+ files := make([]string, 0)
+ for i := range 5 {
+ filePath := fmt.Sprintf("%s/unpartitioned/test-%d.parquet", t.location, i)
+ t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTbl)
+ files = append(files, filePath)
+ }
+
+ tx := tbl.NewTransaction()
+ t.Require().NoError(tx.AddFiles(files, nil, false))
+
+ stagedTbl, err := tx.StagedTable()
+ t.Require().NoError(err)
+ t.NotNil(stagedTbl.NameMapping())
+
+ t.Equal(stagedTbl.CurrentSnapshot().Summary,
+ &table.Summary{
+ Operation: table.OpAppend,
+ Properties: iceberg.Properties{
+ "added-data-files": "5",
+ "added-files-size": "3660",
+ "added-records": "5",
+ "total-data-files": "5",
+ "total-delete-files": "0",
+ "total-equality-deletes": "0",
+ "total-files-size": "3660",
+ "total-position-deletes": "0",
+ "total-records": "5",
+ },
+ })
+
+ contents, err := stagedTbl.Scan().ToArrowTable(context.Background())
+ t.Require().NoError(err)
+ defer contents.Release()
+
+ t.EqualValues(5, contents.NumRows())
+}
+
+func (t *TableWritingTestSuite) TestAddFilesFileNotFound() {
+ ident := table.Identifier{"default", "unpartitioned_table_file_not_found_v" + strconv.Itoa(t.formatVersion)}
+ tbl := t.createTable(ident, t.formatVersion,
+ *iceberg.UnpartitionedSpec, t.tableSchema)
+
+ t.NotNil(tbl)
+
+ files := make([]string, 0)
+ for i := range 5 {
+ filePath := fmt.Sprintf("%s/unpartitioned_file_not_found/test-%d.parquet", t.location, i)
+ t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTbl)
+ files = append(files, filePath)
+ }
+
+ files = append(files, t.location+"/unpartitioned_file_not_found/unknown.parquet")
+ tx := tbl.NewTransaction()
+ err := tx.AddFiles(files, nil, false)
+ t.Error(err)
+ t.ErrorContains(err, "no such file or directory")
+}
+
+func (t *TableWritingTestSuite) TestAddFilesUnpartitionedHasFieldIDs() {
+ ident := table.Identifier{"default", "unpartitioned_table_with_ids_v" + strconv.Itoa(t.formatVersion)}
+ tbl := t.createTable(ident, t.formatVersion,
+ *iceberg.UnpartitionedSpec, t.tableSchema)
+
+ t.NotNil(tbl)
+
+ files := make([]string, 0)
+ for i := range 5 {
+ filePath := fmt.Sprintf("%s/unpartitioned_with_ids/test-%d.parquet", t.location, i)
+ t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTblWithIDs)
+ files = append(files, filePath)
+ }
+
+ tx := tbl.NewTransaction()
+ err := tx.AddFiles(files, nil, false)
+ t.Error(err)
+ t.ErrorIs(err, iceberg.ErrNotImplemented)
+}
+
+func (t *TableWritingTestSuite) TestAddFilesFailsSchemaMismatch() {
+ ident := table.Identifier{"default", "unpartitioned_table_schema_mismatch_v" + strconv.Itoa(t.formatVersion)}
+ tbl := t.createTable(ident, t.formatVersion,
+ *iceberg.UnpartitionedSpec, t.tableSchema)
+ t.Require().NotNil(tbl)
+
+ wrongSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "foo", Type: arrow.FixedWidthTypes.Boolean, Nullable: true},
+ {Name: "bar", Type: arrow.BinaryTypes.String, Nullable: true},
+ {Name: "baz", Type: arrow.BinaryTypes.String, Nullable: true}, // should be int32
+ {Name: "qux", Type: arrow.PrimitiveTypes.Date32, Nullable: true},
+ }, nil)
+
+ filePath := fmt.Sprintf("%s/unpartitioned_schema_mismatch_v%d/test.parquet", t.location, t.formatVersion)
+ mismatchTable, err := array.TableFromJSON(memory.DefaultAllocator, wrongSchema, []string{
+ `[{"foo": true, "bar": "bar_string", "baz": "123", "qux": "2024-03-07"},
+ {"foo": false, "bar": "bar_string", "baz": "456", "qux": "2024-03-07"}]`,
+ })
+ t.Require().NoError(err)
+ t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, mismatchTable)
+
+ files := []string{filePath}
+
+ pterm.DisableOutput() // disable error to console
+ defer pterm.EnableOutput()
+
+ tx := tbl.NewTransaction()
+ err = tx.AddFiles(files, nil, false)
+ t.Error(err)
+ t.EqualError(err, `error encountered during parquet file conversion: error encountered during schema visitor: mismatch in fields:
+ | Table Field | Requested Field
+✅ | 1: foo: optional boolean | 1: foo: optional boolean
+✅ | 2: bar: optional string | 2: bar: optional string
+❌ | 3: baz: optional int | 3: baz: optional string
+✅ | 4: qux: optional date | 4: qux: optional date
+`)
+}
+
+func TestTableWriting(t *testing.T) {
+ suite.Run(t, &TableWritingTestSuite{formatVersion: 1})
+ suite.Run(t, &TableWritingTestSuite{formatVersion: 2})
+}
diff --git a/table/transaction.go b/table/transaction.go
new file mode 100644
index 0000000..15785b5
--- /dev/null
+++ b/table/transaction.go
@@ -0,0 +1,193 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "slices"
+ "sync"
+ "time"
+
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/io"
+)
+
+type snapshotUpdate struct {
+ txn *Transaction
+ io io.WriteFileIO
+ snapshotProps iceberg.Properties
+}
+
+func (s snapshotUpdate) fastAppend() *snapshotProducer {
+ return newFastAppendFilesProducer(OpAppend, s.txn, s.io, nil, s.snapshotProps)
+}
+
+type Transaction struct {
+ tbl *Table
+ meta *MetadataBuilder
+
+ reqs []Requirement
+
+ mx sync.Mutex
+ committed bool
+}
+
+func (t *Transaction) apply(updates []Update, reqs []Requirement) error {
+ t.mx.Lock()
+ defer t.mx.Unlock()
+
+ if t.committed {
+ return errors.New("transaction has already been committed")
+ }
+
+ current, err := t.meta.Build()
+ if err != nil {
+ return err
+ }
+
+ for _, r := range reqs {
+ if err := r.Validate(current); err != nil {
+ return err
+ }
+ }
+
+ existing := map[string]struct{}{}
+ for _, r := range t.reqs {
+ existing[r.GetType()] = struct{}{}
+ }
+
+ for _, r := range reqs {
+ if _, ok := existing[r.GetType()]; !ok {
+ t.reqs = append(t.reqs, r)
+ }
+ }
+
+ prevUpdates, prevLastUpdated := len(t.meta.updates), t.meta.lastUpdatedMS
+ for _, u := range updates {
+ if err := u.Apply(t.meta); err != nil {
+ return err
+ }
+ }
+
+ // u.Apply will add updates to t.meta.updates if they are not no-ops
+ // and actually perform changes. So let's check if we actually had any
+ // changes added and thus need to update the lastupdated value.
+ if prevUpdates < len(t.meta.updates) {
+ if prevLastUpdated == t.meta.lastUpdatedMS {
+ t.meta.lastUpdatedMS = time.Now().UnixMilli()
+ }
+ }
+
+ return nil
+}
+
+func (t *Transaction) updateSnapshot(props iceberg.Properties) snapshotUpdate {
+ return snapshotUpdate{
+ txn: t,
+ io: t.tbl.fs.(io.WriteFileIO),
+ snapshotProps: props,
+ }
+}
+
+func (t *Transaction) SetProperties(props iceberg.Properties) error {
+ if len(props) > 0 {
+ return t.apply([]Update{NewSetPropertiesUpdate(props)}, nil)
+ }
+
+ return nil
+}
+
+func (t *Transaction) Append(rdr array.RecordReader, snapshotProps iceberg.Properties) error {
+ return iceberg.ErrNotImplemented
+}
+
+func (t *Transaction) AddFiles(files []string, snapshotProps iceberg.Properties, checkDuplicates bool) error {
+ set := make(map[string]string)
+ for _, f := range files {
+ set[f] = f
+ }
+
+ if len(set) != len(files) {
+ return errors.New("file paths must be unique for AppendDataFiles")
+ }
+
+ if checkDuplicates {
+ return iceberg.ErrNotImplemented
+ }
+
+ if t.meta.NameMapping() == nil {
+ nameMapping := t.meta.CurrentSchema().NameMapping()
+ mappingJson, err := json.Marshal(nameMapping)
+ if err != nil {
+ return err
+ }
+ err = t.SetProperties(iceberg.Properties{DefaultNameMappingKey: string(mappingJson)})
+ if err != nil {
+ return err
+ }
+ }
+
+ updater := t.updateSnapshot(snapshotProps).fastAppend()
+
+ dataFiles := parquetFilesToDataFiles(t.tbl.fs, t.meta, slices.Values(files))
+ for df, err := range dataFiles {
+ if err != nil {
+ return err
+ }
+ updater.appendDataFile(df)
+ }
+
+ updates, reqs, err := updater.commit()
+ if err != nil {
+ return err
+ }
+
+ return t.apply(updates, reqs)
+}
+
+func (t *Transaction) StagedTable() (*Table, error) {
+ updatedMeta, err := t.meta.Build()
+ if err != nil {
+ return nil, err
+ }
+
+ return New(t.tbl.identifier, updatedMeta,
+ updatedMeta.Location(), t.tbl.fs, t.tbl.cat), nil
+}
+
+func (t *Transaction) Commit(ctx context.Context) (*Table, error) {
+ t.mx.Lock()
+ defer t.mx.Unlock()
+
+ if t.committed {
+ return nil, errors.New("transaction has already been committed")
+ }
+
+ t.committed = true
+
+ if len(t.meta.updates) > 0 {
+ t.reqs = append(t.reqs, AssertTableUUID(t.meta.uuid))
+
+ return t.tbl.doCommit(ctx, t.meta.updates, t.reqs)
+ }
+
+ return t.tbl, nil
+}
diff --git a/types.go b/types.go
index ea93007..e3fea7a 100644
--- a/types.go
+++ b/types.go
@@ -26,7 +26,7 @@ import (
"strings"
"time"
- "github.com/apache/arrow-go/v18/arrow/decimal128"
+ "github.com/apache/arrow-go/v18/arrow/decimal"
)
var (
@@ -484,7 +484,7 @@ func (d DecimalType) Scale() int { return d.scale }
func (DecimalType) primitive() {}
type Decimal struct {
- Val decimal128.Num
+ Val decimal.Decimal128
Scale int
}