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 2026/07/27 19:50:07 UTC
(iceberg-go) branch main updated: feat(table): projjson read/write support (#1426)
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 b559159e7 feat(table): projjson read/write support (#1426)
b559159e7 is described below
commit b559159e7159770f7874ba461b58edd24861fd8f
Author: Colton Loftus <70...@users.noreply.github.com>
AuthorDate: Mon Jul 27 15:49:58 2026 -0400
feat(table): projjson read/write support (#1426)
Closes #1230
This PR threads through projjson definition metadata by using a `ctx`
variable in accordance to the guidance in #1230 . Note that some of the
writer call stack did not previously take a `ctx` variable. As such,
anything public has an additional new private helper that takes in a
`ctx` to resolve it.
I modified `icebergCRSToGeoArrowMetadata` to take in a `ctx` as well so
it can resolve either normal inline crs or projjson.
This ultimately ends up giving us a VisitGeometry API like the following
```go
func (c convertToArrow) VisitGeometry(g iceberg.GeometryType) arrow.Field {
meta, err := icebergCRSToGeoArrowMetadata(c.ctx, g.CRS())
/// rest removed for brevity ....
}
```
Happy to refactor as maintainers see fit. I wasn't entirely sure on a
few of the refactors and whether or not there is a better way to thread
the context through / if it is reasonable to change the public methods
for things like `SchemaToArrowSchema`
---
table/arrow_scanner.go | 62 +++---
table/arrow_utils.go | 305 ++++++++++++++++++++++++++++--
table/arrow_utils_internal_test.go | 20 +-
table/arrow_utils_test.go | 63 +++++-
table/equality_delete_writer.go | 5 +-
table/equality_delete_writer_test.go | 72 +++++++
table/position_delta_writer.go | 5 +-
table/projjson_roundtrip_internal_test.go | 104 ++++++++++
table/rewrite_data_files.go | 5 +-
table/rolling_data_writer.go | 11 +-
table/transaction.go | 11 +-
table/write_records.go | 4 +-
table/write_records_test.go | 91 +++++++++
table/writer.go | 15 +-
14 files changed, 699 insertions(+), 74 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index 807629606..adfe4b8f9 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -35,7 +35,7 @@ import (
iceinternal "github.com/apache/iceberg-go/internal"
iceio "github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/table/dv"
- "github.com/apache/iceberg-go/table/internal"
+ tblutils "github.com/apache/iceberg-go/table/internal"
"github.com/apache/iceberg-go/table/substrait"
"github.com/substrait-io/substrait-go/v8/expr"
"golang.org/x/sync/errgroup"
@@ -426,7 +426,7 @@ func releasePosDeletes(deletes map[string]*arrow.Chunked) {
}
func readDeletes(ctx context.Context, fs iceio.IO, dataFile iceberg.DataFile) (_ map[string]*arrow.Chunked, err error) {
- src, err := internal.GetFile(ctx, fs, dataFile, true)
+ src, err := tblutils.GetFile(ctx, fs, dataFile, true)
if err != nil {
return nil, err
}
@@ -687,18 +687,18 @@ func (as *arrowScan) projectedFieldIDs() (set[int], error) {
}
type enumeratedRecord struct {
- Record internal.Enumerated[arrow.RecordBatch]
- Task internal.Enumerated[FileScanTask]
+ Record tblutils.Enumerated[arrow.RecordBatch]
+ Task tblutils.Enumerated[FileScanTask]
Err error
}
-func (as *arrowScan) prepareToRead(ctx context.Context, file iceberg.DataFile) (*iceberg.Schema, []int, internal.FileReader, error) {
+func (as *arrowScan) prepareToRead(ctx context.Context, file iceberg.DataFile) (*iceberg.Schema, []int, tblutils.FileReader, error) {
ids, err := as.projectedFieldIDs()
if err != nil {
return nil, nil, nil, err
}
- src, err := internal.GetFile(ctx, as.fs, file, false)
+ src, err := tblutils.GetFile(ctx, as.fs, file, false)
if err != nil {
return nil, nil, nil, err
}
@@ -715,7 +715,11 @@ func (as *arrowScan) prepareToRead(ctx context.Context, file iceberg.DataFile) (
return nil, nil, nil, err
}
- iceSchema, err := ArrowSchemaToIceberg(fileSchema, false, as.nameMapping)
+ iceSchema, err := ArrowSchemaToIcebergWithOptions(fileSchema, ArrowToIcebergOptions{
+ NameMapping: as.nameMapping,
+ TableSchema: as.metadata.CurrentSchema(),
+ TableProperties: as.metadata.Properties(),
+ })
if err != nil {
rdr.Close()
@@ -782,7 +786,7 @@ func fieldIndexByID(schema *arrow.Schema, fieldID int) int {
// With no spans recorded (non-Parquet files, or a read with no pruning pass) the
// cursors fall back to a contiguous counter from zero.
type rowPositionSource struct {
- spans []internal.RowGroupSpan
+ spans []tblutils.RowGroupSpan
}
func (s *rowPositionSource) cursor() *rowPositionCursor {
@@ -934,9 +938,9 @@ func synthesizeRowLineageColumns(
func (as *arrowScan) processRecords(
ctx context.Context,
- task internal.Enumerated[FileScanTask],
+ task tblutils.Enumerated[FileScanTask],
fileSchema *iceberg.Schema,
- rdr internal.FileReader,
+ rdr tblutils.FileReader,
columns []int,
pipeline []recProcessFn,
posSource *rowPositionSource,
@@ -965,7 +969,7 @@ func (as *arrowScan) processRecords(
return err
}
- tester := &internal.ParquetRowGroupTester{
+ tester := &tblutils.ParquetRowGroupTester{
StatsFn: statsFn,
BloomPreds: bloomPreds,
}
@@ -988,7 +992,7 @@ func (as *arrowScan) processRecords(
for recRdr.Next() {
if prev != nil {
- out <- enumeratedRecord{Record: internal.Enumerated[arrow.RecordBatch]{
+ out <- enumeratedRecord{Record: tblutils.Enumerated[arrow.RecordBatch]{
Value: prev, Index: idx, Last: false,
}, Task: task}
idx++
@@ -1006,7 +1010,7 @@ func (as *arrowScan) processRecords(
}
if prev != nil {
- out <- enumeratedRecord{Record: internal.Enumerated[arrow.RecordBatch]{
+ out <- enumeratedRecord{Record: tblutils.Enumerated[arrow.RecordBatch]{
Value: prev, Index: idx, Last: true,
}, Task: task}
}
@@ -1018,7 +1022,7 @@ func (as *arrowScan) processRecords(
return err
}
-func (as *arrowScan) recordsFromTask(ctx context.Context, task internal.Enumerated[FileScanTask], out chan<- enumeratedRecord, positionalDeletes positionDeletes, dvBitmap *dv.RoaringPositionBitmap, eqDeleteSets []*equalityDeleteSet) (err error) {
+func (as *arrowScan) recordsFromTask(ctx context.Context, task tblutils.Enumerated[FileScanTask], out chan<- enumeratedRecord, positionalDeletes positionDeletes, dvBitmap *dv.RoaringPositionBitmap, eqDeleteSets []*equalityDeleteSet) (err error) {
defer func() {
if err != nil {
out <- enumeratedRecord{Task: task, Err: err}
@@ -1026,7 +1030,7 @@ func (as *arrowScan) recordsFromTask(ctx context.Context, task internal.Enumerat
}()
var (
- rdr internal.FileReader
+ rdr tblutils.FileReader
iceSchema *iceberg.Schema
colIndices []int
filterFunc recProcessFn
@@ -1121,7 +1125,7 @@ func (as *arrowScan) recordsFromTask(ctx context.Context, task internal.Enumerat
if err != nil {
return err
}
- out <- enumeratedRecord{Task: task, Record: internal.Enumerated[arrow.RecordBatch]{
+ out <- enumeratedRecord{Task: task, Record: tblutils.Enumerated[arrow.RecordBatch]{
Value: array.NewRecordBatch(emptySchema, nil, 0), Index: 0, Last: true,
}}
@@ -1135,7 +1139,10 @@ func (as *arrowScan) recordsFromTask(ctx context.Context, task internal.Enumerat
pipeline = append(pipeline, func(r arrow.RecordBatch) (arrow.RecordBatch, error) {
defer r.Release()
- return ToRequestedSchema(ctx, as.projectedSchema, readSchema, r, SchemaOptions{UseLargeTypes: as.useLargeTypes})
+ return ToRequestedSchema(ctx, as.projectedSchema, readSchema, r, SchemaOptions{
+ UseLargeTypes: as.useLargeTypes,
+ TableProperties: as.metadata.Properties(),
+ })
})
err = as.processRecords(ctx, task, iceSchema, rdr, colIndices, pipeline, posSource, out)
@@ -1143,7 +1150,7 @@ func (as *arrowScan) recordsFromTask(ctx context.Context, task internal.Enumerat
return err
}
-func (as *arrowScan) producePosDeletesFromTask(ctx context.Context, task internal.Enumerated[FileScanTask], positionalDeletes positionDeletes, out chan<- enumeratedRecord) (err error) {
+func (as *arrowScan) producePosDeletesFromTask(ctx context.Context, task tblutils.Enumerated[FileScanTask], positionalDeletes positionDeletes, out chan<- enumeratedRecord) (err error) {
defer func() {
if err != nil {
out <- enumeratedRecord{Task: task, Err: err}
@@ -1151,7 +1158,7 @@ func (as *arrowScan) producePosDeletesFromTask(ctx context.Context, task interna
}()
var (
- rdr internal.FileReader
+ rdr tblutils.FileReader
iceSchema *iceberg.Schema
colIndices []int
filterFunc recProcessFn
@@ -1196,7 +1203,7 @@ func (as *arrowScan) producePosDeletesFromTask(ctx context.Context, task interna
if err != nil {
return err
}
- out <- enumeratedRecord{Task: task, Record: internal.Enumerated[arrow.RecordBatch]{
+ out <- enumeratedRecord{Task: task, Record: tblutils.Enumerated[arrow.RecordBatch]{
Value: array.NewRecordBatch(emptySchema, nil, 0), Index: 0, Last: true,
}}
@@ -1222,7 +1229,7 @@ func createIterator(ctx context.Context, numWorkers uint, records <-chan enumera
return batch.Task.Index < 0
}
- sequenced := internal.MakeSequencedChan(uint(numWorkers), records,
+ sequenced := tblutils.MakeSequencedChan(uint(numWorkers), records,
func(left, right *enumeratedRecord) bool {
switch {
case isBeforeAny(*left):
@@ -1248,7 +1255,7 @@ func createIterator(ctx context.Context, numWorkers uint, records <-chan enumera
return next.Task.Index == prev.Task.Index+1 &&
prev.Record.Last && next.Record.Index == 0
}
- }, enumeratedRecord{Task: internal.Enumerated[FileScanTask]{Index: -1}})
+ }, enumeratedRecord{Task: tblutils.Enumerated[FileScanTask]{Index: -1}})
totalRowCount := int64(0)
@@ -1320,7 +1327,7 @@ func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context, tasks
as.nameMapping = as.metadata.NameMapping()
ctx, cancel := context.WithCancelCause(exprs.WithExtensionIDSet(ctx, extSet))
- taskChan := make(chan internal.Enumerated[FileScanTask], len(tasks))
+ taskChan := make(chan tblutils.Enumerated[FileScanTask], len(tasks))
// numWorkers := 1
numWorkers := min(as.concurrency, len(tasks))
@@ -1356,7 +1363,7 @@ func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context, tasks
go func() {
for i, t := range tasks {
- taskChan <- internal.Enumerated[FileScanTask]{
+ taskChan <- tblutils.Enumerated[FileScanTask]{
Value: t, Index: i, Last: i == len(tasks)-1,
}
}
@@ -1377,9 +1384,12 @@ func (as *arrowScan) GetRecords(ctx context.Context, tasks []FileScanTask) (*arr
as.useLargeTypes = false
}
- ctx = internal.WithTableProperties(ctx, as.metadata.Properties())
+ ctx = tblutils.WithTableProperties(ctx, as.metadata.Properties())
- resultSchema, err := SchemaToArrowSchema(as.projectedSchema, nil, false, as.useLargeTypes)
+ resultSchema, err := SchemaToArrowSchemaWithOptions(as.projectedSchema, ArrowSchemaOptions{
+ UseLargeTypes: as.useLargeTypes,
+ TableProperties: as.metadata.Properties(),
+ })
if err != nil {
return nil, nil, err
}
diff --git a/table/arrow_utils.go b/table/arrow_utils.go
index 617502cac..6e9e1e38c 100644
--- a/table/arrow_utils.go
+++ b/table/arrow_utils.go
@@ -246,8 +246,16 @@ func (hasIDs) Primitive(arrow.DataType) bool { return true }
type convertToIceberg struct {
downcastTimestamp bool
+ // tableSchema and tableProperties provide table context that Arrow schema
+ // metadata does not carry. They are used to recover things like projjson:<property>
+ // CRS references from embedded PROJJSON metadata.
+ tableSchema *iceberg.Schema
+ tableProperties iceberg.Properties
fieldID func(arrow.Field) int
+ // fields tracks the current Arrow traversal path. The top field's ID is used
+ // to match GeoArrow PROJJSON metadata back to the corresponding table field.
+ fields []arrow.Field
}
func (convertToIceberg) Schema(_ *arrow.Schema, result iceberg.NestedField) iceberg.NestedField {
@@ -302,6 +310,59 @@ func (c convertToIceberg) Map(m *arrow.MapType, keyResult, valueResult iceberg.N
}
}
+func (c *convertToIceberg) pushField(field arrow.Field) {
+ c.fields = append(c.fields, field)
+}
+
+func (c *convertToIceberg) popField() {
+ c.fields = c.fields[:len(c.fields)-1]
+}
+
+func (c *convertToIceberg) BeforeField(field arrow.Field) {
+ c.pushField(field)
+}
+
+func (c *convertToIceberg) AfterField(arrow.Field) {
+ c.popField()
+}
+
+func (c *convertToIceberg) BeforeListElement(field arrow.Field) {
+ c.pushField(field)
+}
+
+func (c *convertToIceberg) AfterListElement(arrow.Field) {
+ c.popField()
+}
+
+func (c *convertToIceberg) BeforeMapKey(field arrow.Field) {
+ c.pushField(field)
+}
+
+func (c *convertToIceberg) AfterMapKey(arrow.Field) {
+ c.popField()
+}
+
+func (c *convertToIceberg) BeforeMapValue(field arrow.Field) {
+ c.pushField(field)
+}
+
+func (c *convertToIceberg) AfterMapValue(arrow.Field) {
+ c.popField()
+}
+
+func (c convertToIceberg) currentFieldID() (int, bool) {
+ if len(c.fields) == 0 {
+ return 0, false
+ }
+
+ id := getFieldID(c.fields[len(c.fields)-1])
+ if id == nil {
+ return 0, false
+ }
+
+ return *id, true
+}
+
var utcAliases = []string{"UTC", "+00:00", "Etc/UTC", "Z"}
func (c convertToIceberg) Primitive(dt arrow.DataType) (result iceberg.NestedField) {
@@ -384,7 +445,7 @@ func (c convertToIceberg) Primitive(dt arrow.DataType) (result iceberg.NestedFie
panic(fmt.Errorf("%w: unsupported arrow type for conversion - %s", iceberg.ErrInvalidSchema, dt))
}
- iceType, err := geoArrowMetadataToIcebergType(wkb.Metadata())
+ iceType, err := c.geoArrowMetadataToIcebergType(wkb.Metadata())
if err != nil {
panic(fmt.Errorf("%w: converting geoarrow metadata: %w", iceberg.ErrInvalidSchema, err))
}
@@ -405,7 +466,7 @@ func ArrowTypeToIceberg(dt arrow.DataType, downcastNsTimestamp bool) (iceberg.Ty
Metadata: arrow.NewMetadata([]string{ArrowParquetFieldIDKey}, []string{"1"}),
}}, nil)
- out, err := VisitArrowSchema(sc, convertToIceberg{
+ out, err := VisitArrowSchema(sc, &convertToIceberg{
downcastTimestamp: downcastNsTimestamp,
fieldID: func(field arrow.Field) int {
if id := getFieldID(field); id != nil {
@@ -424,12 +485,34 @@ func ArrowTypeToIceberg(dt arrow.DataType, downcastNsTimestamp bool) (iceberg.Ty
}
func ArrowSchemaToIceberg(sc *arrow.Schema, downcastNsTimestamp bool, nameMapping iceberg.NameMapping) (*iceberg.Schema, error) {
+ return ArrowSchemaToIcebergWithOptions(sc, ArrowToIcebergOptions{
+ DowncastNsTimestamp: downcastNsTimestamp,
+ NameMapping: nameMapping,
+ })
+}
+
+// ArrowToIcebergOptions controls conversion from an Arrow schema to an Iceberg schema.
+type ArrowToIcebergOptions struct {
+ DowncastNsTimestamp bool
+ NameMapping iceberg.NameMapping
+ TableSchema *iceberg.Schema
+ TableProperties iceberg.Properties
+}
+
+// ArrowSchemaToIcebergWithOptions converts an Arrow schema to an Iceberg schema.
+// Options carry conversion inputs that are not encoded in the Arrow schema
+// itself, such as table metadata needed for exact CRS recovery.
+// When TableSchema and TableProperties are provided, GeoArrow PROJJSON metadata
+// can be matched back to the table schema's projjson:<property> CRS references.
+func ArrowSchemaToIcebergWithOptions(sc *arrow.Schema, opts ArrowToIcebergOptions) (*iceberg.Schema, error) {
hasIDs, _ := VisitArrowSchema(sc, hasIDs{})
switch {
case hasIDs:
- out, err := VisitArrowSchema(sc, convertToIceberg{
- downcastTimestamp: downcastNsTimestamp,
+ out, err := VisitArrowSchema(sc, &convertToIceberg{
+ downcastTimestamp: opts.DowncastNsTimestamp,
+ tableSchema: opts.TableSchema,
+ tableProperties: opts.TableProperties,
fieldID: func(field arrow.Field) int {
if id := getFieldID(field); id != nil {
return *id
@@ -444,13 +527,13 @@ func ArrowSchemaToIceberg(sc *arrow.Schema, downcastNsTimestamp bool, nameMappin
}
return iceberg.NewSchema(0, out.Type.(*iceberg.StructType).FieldList...), nil
- case nameMapping != nil:
- schemaWithoutIDs, err := arrowToSchemaWithoutIDs(sc, downcastNsTimestamp)
+ case opts.NameMapping != nil:
+ schemaWithoutIDs, err := arrowToSchemaWithoutIDs(sc, opts.DowncastNsTimestamp)
if err != nil {
return nil, err
}
- return iceberg.ApplyNameMapping(schemaWithoutIDs, nameMapping)
+ return iceberg.ApplyNameMapping(schemaWithoutIDs, opts.NameMapping)
default:
return nil, fmt.Errorf("%w: arrow schema does not have field-ids and no name mapping provided",
iceberg.ErrInvalidSchema)
@@ -467,7 +550,7 @@ func ArrowSchemaToIcebergWithFreshIDs(sc *arrow.Schema, downcastNsTimestamp bool
}
func arrowToSchemaWithoutIDs(sc *arrow.Schema, downcastNsTimestamp bool) (*iceberg.Schema, error) {
- withoutIDs, err := VisitArrowSchema(sc, convertToIceberg{
+ withoutIDs, err := VisitArrowSchema(sc, &convertToIceberg{
downcastTimestamp: downcastNsTimestamp,
fieldID: func(_ arrow.Field) int { return -1 },
})
@@ -531,6 +614,7 @@ type convertToArrow struct {
metadata map[string]string
includeFieldIDs bool
useLargeTypes bool
+ tableProperties iceberg.Properties
}
func (c convertToArrow) Schema(_ *iceberg.Schema, result arrow.Field) arrow.Field {
@@ -673,7 +757,7 @@ func (c convertToArrow) VisitVariant() arrow.Field {
}
func (c convertToArrow) VisitGeometry(g iceberg.GeometryType) arrow.Field {
- meta, err := icebergCRSToGeoArrowMetadata(g.CRS())
+ meta, err := icebergCRSToGeoArrowMetadata(g.CRS(), c.tableProperties)
if err != nil {
// Panic to thread the error through iceberg.Visit's recover, matching the
// convention used by the other visitor methods.
@@ -688,7 +772,7 @@ func (c convertToArrow) VisitGeometry(g iceberg.GeometryType) arrow.Field {
}
func (c convertToArrow) VisitGeography(g iceberg.GeographyType) arrow.Field {
- meta, err := icebergCRSToGeoArrowMetadata(g.CRS())
+ meta, err := icebergCRSToGeoArrowMetadata(g.CRS(), c.tableProperties)
if err != nil {
// Panic to thread the error through iceberg.Visit's recover, matching the
// convention used by the other visitor methods.
@@ -709,14 +793,35 @@ func (c convertToArrow) VisitGeography(g iceberg.GeographyType) arrow.Field {
var _ iceberg.SchemaVisitorPerPrimitiveType[arrow.Field] = convertToArrow{}
+// ArrowSchemaOptions controls conversion from an Iceberg schema or type to Arrow.
+type ArrowSchemaOptions struct {
+ Metadata map[string]string
+ IncludeFieldIDs bool
+ UseLargeTypes bool
+ TableProperties iceberg.Properties
+}
+
// SchemaToArrowSchema converts an Iceberg schema to an Arrow schema. If the metadata parameter
// is non-nil, it will be included as the top-level metadata in the schema. If includeFieldIDs
// is true, then each field of the schema will contain a metadata key PARQUET:field_id set to
// the field id from the iceberg schema.
func SchemaToArrowSchema(sc *iceberg.Schema, metadata map[string]string, includeFieldIDs, useLargeTypes bool) (*arrow.Schema, error) {
+ return SchemaToArrowSchemaWithOptions(sc, ArrowSchemaOptions{
+ Metadata: metadata,
+ IncludeFieldIDs: includeFieldIDs,
+ UseLargeTypes: useLargeTypes,
+ })
+}
+
+// SchemaToArrowSchemaWithOptions converts an Iceberg schema to an Arrow schema.
+// TableProperties are used to resolve references like projjson:<property> CRS which point to
+// metadata in the table properties.
+func SchemaToArrowSchemaWithOptions(sc *iceberg.Schema, opts ArrowSchemaOptions) (*arrow.Schema, error) {
top, err := iceberg.Visit(sc, convertToArrow{
- metadata: metadata,
- includeFieldIDs: includeFieldIDs, useLargeTypes: useLargeTypes,
+ metadata: opts.Metadata,
+ includeFieldIDs: opts.IncludeFieldIDs,
+ useLargeTypes: opts.UseLargeTypes,
+ tableProperties: opts.TableProperties,
})
if err != nil {
return nil, err
@@ -729,8 +834,22 @@ func SchemaToArrowSchema(sc *iceberg.Schema, metadata map[string]string, include
// For dealing with nested fields (List, Struct, Map) if includeFieldIDs is true, then
// the child fields will contain a metadata key PARQUET:field_id set to the field id.
func TypeToArrowType(t iceberg.Type, includeFieldIDs bool, useLargeTypes bool) (arrow.DataType, error) {
+ return TypeToArrowTypeWithOptions(t, ArrowSchemaOptions{
+ IncludeFieldIDs: includeFieldIDs,
+ UseLargeTypes: useLargeTypes,
+ })
+}
+
+// TypeToArrowTypeWithOptions converts a given Iceberg type into the equivalent Arrow data type.
+// Options control representation details and provide table properties used to
+// resolve metadata references such as CRS projjson:<property> which are stored as table properties.
+func TypeToArrowTypeWithOptions(t iceberg.Type, opts ArrowSchemaOptions) (arrow.DataType, error) {
top, err := iceberg.Visit(iceberg.NewSchema(0, iceberg.NestedField{Type: t}),
- convertToArrow{includeFieldIDs: includeFieldIDs, useLargeTypes: useLargeTypes})
+ convertToArrow{
+ includeFieldIDs: opts.IncludeFieldIDs,
+ useLargeTypes: opts.UseLargeTypes,
+ tableProperties: opts.TableProperties,
+ })
if err != nil {
return nil, err
}
@@ -927,12 +1046,21 @@ func defaultToArray(v any, t iceberg.Type, dt arrow.DataType, n int, alloc memor
type arrowProjectionVisitor struct {
ctx context.Context
fileSchema *iceberg.Schema
+ tableProperties iceberg.Properties
includeFieldIDs bool
downcastNsTimestamp bool
useLargeTypes bool
useWriteDefault bool
}
+func (a *arrowProjectionVisitor) typeToArrowType(t iceberg.Type) arrow.DataType {
+ return retOrPanic(TypeToArrowTypeWithOptions(t, ArrowSchemaOptions{
+ IncludeFieldIDs: a.includeFieldIDs,
+ UseLargeTypes: a.useLargeTypes,
+ TableProperties: a.tableProperties,
+ }))
+}
+
func (a *arrowProjectionVisitor) castIfNeeded(field iceberg.NestedField, vals arrow.Array) arrow.Array {
fileField, ok := a.fileSchema.FindFieldByID(field.ID)
if !ok {
@@ -949,7 +1077,7 @@ func (a *arrowProjectionVisitor) castIfNeeded(field iceberg.NestedField, vals ar
if !field.Type.Equals(typ) {
if !canDowncastTimestampPrecision(fileField.Type, field.Type, a.downcastNsTimestamp) {
promoted := retOrPanic(iceberg.PromoteType(fileField.Type, field.Type))
- targetType := retOrPanic(TypeToArrowType(promoted, a.includeFieldIDs, a.useLargeTypes))
+ targetType := a.typeToArrowType(promoted)
if !a.useLargeTypes {
targetType = retOrPanic(ensureSmallArrowTypes(targetType))
}
@@ -959,7 +1087,7 @@ func (a *arrowProjectionVisitor) castIfNeeded(field iceberg.NestedField, vals ar
}
}
- targetType := retOrPanic(TypeToArrowType(field.Type, a.includeFieldIDs, a.useLargeTypes))
+ targetType := a.typeToArrowType(field.Type)
if !arrow.TypeEqual(targetType, vals.DataType()) {
switch field.Type.(type) {
case iceberg.TimestampType:
@@ -1084,7 +1212,7 @@ func (a *arrowProjectionVisitor) Struct(st iceberg.StructType, structArr arrow.A
fieldArrs[i] = arr
fields[i] = a.constructField(field, arr.DataType())
} else {
- dt := retOrPanic(TypeToArrowType(field.Type, a.includeFieldIDs, a.useLargeTypes))
+ dt := a.typeToArrowType(field.Type)
alloc := compute.GetAllocator(a.ctx)
switch {
@@ -1238,6 +1366,7 @@ type SchemaOptions struct {
IncludeFieldIDs bool
UseLargeTypes bool
UseWriteDefault bool
+ TableProperties iceberg.Properties
}
// ToRequestedSchema will construct a new record batch matching the requested iceberg schema
@@ -1250,6 +1379,7 @@ func ToRequestedSchema(ctx context.Context, requested, fileSchema *iceberg.Schem
&arrowProjectionVisitor{
ctx: ctx,
fileSchema: fileSchema,
+ tableProperties: opts.TableProperties,
includeFieldIDs: opts.IncludeFieldIDs,
downcastNsTimestamp: opts.DowncastTimestamp,
useLargeTypes: opts.UseLargeTypes,
@@ -1291,6 +1421,20 @@ func checkArrowSchemaCompat(requested *iceberg.Schema, provided *arrow.Schema, d
return checkSchemaCompat(requested, providedSchema)
}
+func checkArrowSchemaCompatWithProperties(requested *iceberg.Schema, provided *arrow.Schema, downcastNanoToMicro bool, props iceberg.Properties) error {
+ providedSchema, err := ArrowSchemaToIcebergWithOptions(provided, ArrowToIcebergOptions{
+ DowncastNsTimestamp: downcastNanoToMicro,
+ NameMapping: requested.NameMapping(),
+ TableSchema: requested,
+ TableProperties: props,
+ })
+ 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 {
@@ -1574,13 +1718,16 @@ func fileToDataFile(ctx context.Context, fileIO iceio.IO, filePath string, curre
defer rdr.Close()
arrSchema := must(rdr.Schema())
- if err := checkArrowSchemaCompat(currentSchema, arrSchema, false); err != nil {
+ if err := checkArrowSchemaCompatWithProperties(currentSchema, arrSchema, false, props); err != nil {
panic(err)
}
pathToIDSchema := currentSchema
if fileHasIDs := must(VisitArrowSchema(arrSchema, hasIDs{})); fileHasIDs {
- pathToIDSchema = must(ArrowSchemaToIceberg(arrSchema, false, nil))
+ pathToIDSchema = must(ArrowSchemaToIcebergWithOptions(arrSchema, ArrowToIcebergOptions{
+ TableSchema: currentSchema,
+ TableProperties: props,
+ }))
}
statistics := format.DataFileStatsFromMeta(
rdr.Metadata(),
@@ -1683,7 +1830,11 @@ func recordsToDataFiles(ctx context.Context, rootLocation string, meta *Metadata
WriteTargetFileSizeBytesDefault))
nameMapping := meta.CurrentSchema().NameMapping()
- taskSchema, err := ArrowSchemaToIceberg(args.sc, false, nameMapping)
+ taskSchema, err := ArrowSchemaToIcebergWithOptions(args.sc, ArrowToIcebergOptions{
+ NameMapping: nameMapping,
+ TableSchema: meta.CurrentSchema(),
+ TableProperties: meta.props,
+ })
if err != nil {
return func(yield func(iceberg.DataFile, error) bool) {
yield(nil, err)
@@ -2034,6 +2185,10 @@ func geoArrowCRSToIcebergCRS(meta geoarrow.Metadata) (string, error) {
// Iceberg type contract; preserve them for read/write compatibility.
return crs, nil
case checkCRSJSON(meta.CRS):
+ if meta.CRSType == geoarrow.CRSTypePROJJSON {
+ return "", errors.New("unsupported CRS: projjson metadata requires table schema and properties to recover the Iceberg CRS reference")
+ }
+
var crs map[string]json.RawMessage
if err := json.Unmarshal(meta.CRS, &crs); err != nil {
@@ -2090,12 +2245,38 @@ func geoArrowCRSToIcebergCRS(meta geoarrow.Metadata) (string, error) {
}
}
+func (c convertToIceberg) geoArrowMetadataToIcebergType(meta geoarrow.Metadata) (iceberg.Type, error) {
+ crs, err := c.geoArrowCRSToIcebergCRS(meta)
+ if err != nil {
+ return nil, err
+ }
+
+ return geoArrowMetadataToIcebergTypeWithCRS(meta, crs)
+}
+
+func (c convertToIceberg) geoArrowCRSToIcebergCRS(meta geoarrow.Metadata) (string, error) {
+ if meta.CRSType != geoarrow.CRSTypePROJJSON || !checkCRSJSON(meta.CRS) {
+ return geoArrowCRSToIcebergCRS(meta)
+ }
+
+ fieldID, ok := c.currentFieldID()
+ if !ok {
+ return "", errors.New("unsupported CRS: projjson metadata requires field IDs to recover the Iceberg CRS reference")
+ }
+
+ return projJSONCRSFromTable(fieldID, meta.CRS, c.tableSchema, c.tableProperties)
+}
+
func geoArrowMetadataToIcebergType(meta geoarrow.Metadata) (iceberg.Type, error) {
crs, err := geoArrowCRSToIcebergCRS(meta)
if err != nil {
return nil, err
}
+ return geoArrowMetadataToIcebergTypeWithCRS(meta, crs)
+}
+
+func geoArrowMetadataToIcebergTypeWithCRS(meta geoarrow.Metadata, crs string) (iceberg.Type, error) {
// Missing GeoArrow edges use the planar default, matching arrow-rs
// geoarrow.wkb reads. PyIceberg-authored geography files may omit edges,
// which is an interop gap: Arrow metadata alone cannot distinguish those
@@ -2112,7 +2293,10 @@ func geoArrowMetadataToIcebergType(meta geoarrow.Metadata) (iceberg.Type, error)
var authorityCodeCRS = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9_-]*:[A-Za-z0-9_.-]+$`)
-func icebergCRSToGeoArrowMetadata(crs string) (geoarrow.Metadata, error) {
+// icebergCRSToGeoArrowMetadata converts an Iceberg CRS string to GeoArrow
+// metadata. props is used only for projjson:<property> references, where the
+// property value supplies the PROJJSON document embedded into GeoArrow metadata.
+func icebergCRSToGeoArrowMetadata(crs string, props iceberg.Properties) (geoarrow.Metadata, error) {
lowerCRS := strings.ToLower(crs)
if strings.HasPrefix(lowerCRS, "srid:") {
id := crs[len("srid:"):]
@@ -2130,7 +2314,29 @@ func icebergCRSToGeoArrowMetadata(crs string) (geoarrow.Metadata, error) {
}
if strings.HasPrefix(lowerCRS, "projjson:") {
- return geoarrow.Metadata{}, fmt.Errorf("%w: projjson CRS not supported yet", iceberg.ErrInvalidSchema)
+ // The scheme is case-insensitive, but the table property key is not.
+ propKey := strings.TrimSpace(crs[len("projjson:"):])
+ if propKey == "" {
+ return geoarrow.Metadata{}, fmt.Errorf("%w: projjson CRS missing table property key", iceberg.ErrInvalidSchema)
+ }
+ if props == nil {
+ return geoarrow.Metadata{}, fmt.Errorf("%w: projjson CRS %q could not be resolved from table properties", iceberg.ErrInvalidSchema, crs)
+ }
+
+ projjson, ok := props[propKey]
+ if !ok || strings.TrimSpace(projjson) == "" {
+ return geoarrow.Metadata{}, fmt.Errorf("%w: projjson CRS %q references missing table property %q", iceberg.ErrInvalidSchema, crs, propKey)
+ }
+
+ raw := json.RawMessage(projjson)
+ if !json.Valid(raw) {
+ return geoarrow.Metadata{}, fmt.Errorf("%w: projjson CRS table property %q is not valid JSON", iceberg.ErrInvalidSchema, propKey)
+ }
+
+ return geoarrow.Metadata{
+ CRS: raw,
+ CRSType: geoarrow.CRSTypePROJJSON,
+ }, nil
}
var raw []byte
@@ -2149,3 +2355,60 @@ func icebergCRSToGeoArrowMetadata(crs string) (geoarrow.Metadata, error) {
return meta, nil
}
+
+func projJSONCRSFromTable(fieldID int, rawCRS json.RawMessage, tableSchema *iceberg.Schema, props iceberg.Properties) (string, error) {
+ if tableSchema == nil || props == nil {
+ return "", errors.New("unsupported CRS: projjson metadata requires table schema and properties to recover the Iceberg CRS reference")
+ }
+
+ field, ok := tableSchema.FindFieldByID(fieldID)
+ if !ok {
+ return "", fmt.Errorf("unsupported CRS: projjson field id %d not found in table schema", fieldID)
+ }
+
+ var crs string
+ switch typ := field.Type.(type) {
+ case iceberg.GeometryType:
+ crs = typ.CRS()
+ case iceberg.GeographyType:
+ crs = typ.CRS()
+ default:
+ return "", fmt.Errorf("unsupported CRS: table field id %d is %s, not a geo type", fieldID, field.Type)
+ }
+
+ if !strings.HasPrefix(strings.ToLower(crs), "projjson:") {
+ return "", fmt.Errorf("unsupported CRS: table field id %d has CRS %q, not a projjson reference", fieldID, crs)
+ }
+
+ propKey := strings.TrimSpace(crs[len("projjson:"):])
+ if propKey == "" {
+ return "", fmt.Errorf("%w: projjson CRS missing table property key", iceberg.ErrInvalidSchema)
+ }
+
+ projjson, ok := props[propKey]
+ if !ok || strings.TrimSpace(projjson) == "" {
+ return "", fmt.Errorf("%w: projjson CRS %q references missing table property %q", iceberg.ErrInvalidSchema, crs, propKey)
+ }
+
+ if !json.Valid([]byte(projjson)) {
+ return "", fmt.Errorf("%w: projjson CRS table property %q is not valid JSON", iceberg.ErrInvalidSchema, propKey)
+ }
+
+ if !jsonEqual(rawCRS, []byte(projjson)) {
+ return "", fmt.Errorf("unsupported CRS: projjson metadata for field id %d does not match table property %q", fieldID, propKey)
+ }
+
+ return crs, nil
+}
+
+func jsonEqual(lhs, rhs []byte) bool {
+ var lhsCompact, rhsCompact bytes.Buffer
+ if err := json.Compact(&lhsCompact, lhs); err != nil {
+ return false
+ }
+ if err := json.Compact(&rhsCompact, rhs); err != nil {
+ return false
+ }
+
+ return bytes.Equal(lhsCompact.Bytes(), rhsCompact.Bytes())
+}
diff --git a/table/arrow_utils_internal_test.go b/table/arrow_utils_internal_test.go
index 89dd27390..809155794 100644
--- a/table/arrow_utils_internal_test.go
+++ b/table/arrow_utils_internal_test.go
@@ -459,11 +459,21 @@ func TestStatsTypes(t *testing.T) {
func TestIcebergCRSToGeoArrowMetadata(t *testing.T) {
// Calling the converter directly (outside the schema visitor) must not panic:
// an unsupported CRS comes back as an error, symmetric with the read path.
- t.Run("projjson returns an error instead of panicking", func(t *testing.T) {
- _, err := icebergCRSToGeoArrowMetadata("projjson:my-custom-crs")
+ t.Run("projjson without properties returns an error instead of panicking", func(t *testing.T) {
+ _, err := icebergCRSToGeoArrowMetadata("projjson:my-custom-crs", nil)
require.Error(t, err)
require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
- require.ErrorContains(t, err, "projjson CRS not supported yet")
+ require.ErrorContains(t, err, "could not be resolved from table properties")
+ })
+
+ t.Run("projjson resolves from table properties", func(t *testing.T) {
+ const projjson = `{"type":"GeographicCRS","name":"WGS 84"}`
+ meta, err := icebergCRSToGeoArrowMetadata("projjson:my-custom-crs", iceberg.Properties{
+ "my-custom-crs": projjson,
+ })
+ require.NoError(t, err)
+ assert.Equal(t, geoarrow.CRSTypePROJJSON, meta.CRSType)
+ assert.JSONEq(t, projjson, string(meta.CRS))
})
t.Run("supported CRS values map to the expected CRSType", func(t *testing.T) {
@@ -478,7 +488,7 @@ func TestIcebergCRSToGeoArrowMetadata(t *testing.T) {
}
for _, tc := range cases {
t.Run(tc.crs, func(t *testing.T) {
- meta, err := icebergCRSToGeoArrowMetadata(tc.crs)
+ meta, err := icebergCRSToGeoArrowMetadata(tc.crs, nil)
require.NoError(t, err)
assert.Equal(t, tc.wantCRSType, meta.CRSType)
})
@@ -486,7 +496,7 @@ func TestIcebergCRSToGeoArrowMetadata(t *testing.T) {
})
t.Run("srid:0 maps to an omitted CRS", func(t *testing.T) {
- meta, err := icebergCRSToGeoArrowMetadata("srid:0")
+ meta, err := icebergCRSToGeoArrowMetadata("srid:0", nil)
require.NoError(t, err)
assert.Empty(t, meta.CRS)
assert.Empty(t, meta.CRSType)
diff --git a/table/arrow_utils_test.go b/table/arrow_utils_test.go
index dea43e623..efc8367eb 100644
--- a/table/arrow_utils_test.go
+++ b/table/arrow_utils_test.go
@@ -812,13 +812,24 @@ func TestIcebergGeoTypesToArrowSchema(t *testing.T) {
})
t.Run("projjson_error_behavior", func(t *testing.T) {
+ const projjson = `{"type":"GeographicCRS","name":"Custom WGS 84"}`
geom, err := iceberg.GeometryTypeOf("projjson:my-custom-crs")
require.NoError(t, err)
_, err = table.TypeToArrowType(geom, true, false)
require.Error(t, err)
require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
- require.ErrorContains(t, err, "projjson CRS not supported yet")
+ require.ErrorContains(t, err, "could not be resolved from table properties")
+
+ geomArrow, err := table.TypeToArrowTypeWithOptions(geom, table.ArrowSchemaOptions{
+ IncludeFieldIDs: true,
+ TableProperties: iceberg.Properties{"my-custom-crs": projjson},
+ })
+ require.NoError(t, err)
+ geomWKB, ok := geomArrow.(*geoarrow.WKBType)
+ require.True(t, ok)
+ assert.Equal(t, geoarrow.CRSTypePROJJSON, geomWKB.Metadata().CRSType)
+ assert.JSONEq(t, projjson, string(geomWKB.Metadata().CRS))
// Geography goes through VisitGeography; assert the projjson rejection
// surfaces symmetrically through the visitor recover.
@@ -828,7 +839,7 @@ func TestIcebergGeoTypesToArrowSchema(t *testing.T) {
_, err = table.TypeToArrowType(geog, true, false)
require.Error(t, err)
require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
- require.ErrorContains(t, err, "projjson CRS not supported yet")
+ require.ErrorContains(t, err, "could not be resolved from table properties")
stringArrowType, err := geoarrow.NewWKBType().Deserialize(arrow.BinaryTypes.Binary,
`{"crs_type":"projjson","crs":"EPSG:3857"}`)
@@ -842,9 +853,53 @@ func TestIcebergGeoTypesToArrowSchema(t *testing.T) {
`{"crs_type":"projjson","crs":{"id":{"authority":"OGC", "code":"CRS84"}}}`)
require.NoError(t, err)
- g, err := table.ArrowTypeToIceberg(arrowType, false)
+ _, err = table.ArrowTypeToIceberg(arrowType, false)
+ require.Error(t, err)
+ require.ErrorContains(t, err, "projjson metadata requires table schema and properties")
+
+ tableSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "geom", Type: geom, Required: false},
+ )
+ arrowSchema := arrow.NewSchema([]arrow.Field{{
+ Name: "geom",
+ Type: arrowType,
+ Nullable: true,
+ Metadata: arrow.NewMetadata([]string{table.ArrowParquetFieldIDKey}, []string{"1"}),
+ }}, nil)
+
+ g, err := table.ArrowSchemaToIcebergWithOptions(arrowSchema, table.ArrowToIcebergOptions{
+ TableSchema: tableSchema,
+ TableProperties: iceberg.Properties{"my-custom-crs": `{"id":{"authority":"OGC", "code":"CRS84"}}`},
+ })
require.NoError(t, err)
- assert.Equal(t, g, defaultGeometry)
+ assert.True(t, tableSchema.Equals(g), "expected %s, got %s", tableSchema, g)
+
+ t.Run("table_property_json_mismatch", func(t *testing.T) {
+ mismatched, err := table.ArrowSchemaToIcebergWithOptions(arrowSchema, table.ArrowToIcebergOptions{
+ TableSchema: tableSchema,
+ TableProperties: iceberg.Properties{"my-custom-crs": `{"id":{"authority":"EPSG", "code":3857}}`},
+ })
+ require.Error(t, err)
+ require.Nil(t, mismatched)
+ require.ErrorContains(t, err, "does not match table property")
+ })
+
+ t.Run("missing_field_id", func(t *testing.T) {
+ noIDSchema := arrow.NewSchema([]arrow.Field{{
+ Name: "geom",
+ Type: arrowType,
+ Nullable: true,
+ }}, nil)
+
+ noID, err := table.ArrowSchemaToIcebergWithOptions(noIDSchema, table.ArrowToIcebergOptions{
+ NameMapping: tableSchema.NameMapping(),
+ TableSchema: tableSchema,
+ TableProperties: iceberg.Properties{"my-custom-crs": `{"id":{"authority":"OGC", "code":"CRS84"}}`},
+ })
+ require.Error(t, err)
+ require.Nil(t, noID)
+ require.ErrorContains(t, err, "requires field IDs")
+ })
})
t.Run("wkt2:2019_error_behavior", func(t *testing.T) {
diff --git a/table/equality_delete_writer.go b/table/equality_delete_writer.go
index ec973ac5b..2f5a3ced8 100644
--- a/table/equality_delete_writer.go
+++ b/table/equality_delete_writer.go
@@ -130,7 +130,10 @@ func (t *Transaction) WriteEqualityDeletes(ctx context.Context, equalityFieldIDs
return nil, err
}
- arrowSc, err := SchemaToArrowSchema(deleteSchema, nil, true, false)
+ arrowSc, err := SchemaToArrowSchemaWithOptions(deleteSchema, ArrowSchemaOptions{
+ IncludeFieldIDs: true,
+ TableProperties: meta.props,
+ })
if err != nil {
return nil, err
}
diff --git a/table/equality_delete_writer_test.go b/table/equality_delete_writer_test.go
index d993e56e9..5a9799c06 100644
--- a/table/equality_delete_writer_test.go
+++ b/table/equality_delete_writer_test.go
@@ -32,6 +32,7 @@ import (
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/table"
+ "github.com/geoarrow/geoarrow-go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -150,6 +151,77 @@ func TestWriteEqualityDeleteFilesParquetContent(t *testing.T) {
assert.Equal(t, []int64{10, 20, 30}, vals)
}
+func TestWriteEqualityDeleteFilesResolvesProjJSONCRSFromTableProperties(t *testing.T) {
+ const projjson = `{"type":"GeographicCRS","name":"Custom WGS 84"}`
+
+ // Exercise the public equality-delete writer path to ensure it threads
+ // table properties into Arrow schema conversion for projjson CRS fields.
+ location := filepath.ToSlash(t.TempDir())
+ geom, err := iceberg.GeometryTypeOf("projjson:geo.crs")
+ require.NoError(t, err)
+
+ iceSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "geom", Type: geom, Required: false},
+ )
+ meta, err := table.NewMetadata(iceSchema, iceberg.UnpartitionedSpec,
+ table.UnsortedSortOrder, location,
+ iceberg.Properties{
+ table.PropertyFormatVersion: "3",
+ "geo.crs": projjson,
+ })
+ require.NoError(t, err)
+
+ tbl := table.New(
+ table.Identifier{"db", "eq_del_projjson_test"},
+ meta,
+ location+"/metadata/v1.metadata.json",
+ func(ctx context.Context) (iceio.IO, error) {
+ return iceio.LocalFS{}, nil
+ },
+ &rowDeltaCatalog{metadata: meta},
+ )
+
+ delArrowSc, err := table.SchemaToArrowSchemaWithOptions(iceSchema, table.ArrowSchemaOptions{
+ IncludeFieldIDs: true,
+ TableProperties: meta.Properties(),
+ })
+ require.NoError(t, err)
+
+ records, release := makeEqDeleteRecords(t, delArrowSc, `[
+ {"geom": "01010000000000000000003e400000000000002440"}
+ ]`)
+ defer release()
+
+ tx := tbl.NewTransaction()
+ deleteFiles, err := tx.WriteEqualityDeletes(t.Context(), []int{1}, records)
+ require.NoError(t, err)
+ require.Len(t, deleteFiles, 1)
+
+ df := deleteFiles[0]
+ assert.Equal(t, iceberg.EntryContentEqDeletes, df.ContentType())
+ assert.Equal(t, []int{1}, df.EqualityFieldIDs())
+ assert.EqualValues(t, 1, df.Count())
+
+ rdr, err := file.OpenParquetFile(df.FilePath(), false)
+ require.NoError(t, err)
+ defer rdr.Close()
+
+ arrowRdr, err := pqarrow.NewFileReader(rdr, pqarrow.ArrowReadProperties{}, memory.DefaultAllocator)
+ require.NoError(t, err)
+
+ tblReader, err := arrowRdr.ReadTable(t.Context())
+ require.NoError(t, err)
+ defer tblReader.Release()
+
+ require.Equal(t, int64(1), tblReader.NumCols())
+ wkb, ok := tblReader.Schema().Field(0).Type.(*geoarrow.WKBType)
+ require.True(t, ok)
+
+ geoMeta := wkb.Metadata()
+ assert.Equal(t, geoarrow.CRSTypePROJJSON, geoMeta.CRSType)
+ assert.JSONEq(t, projjson, string(geoMeta.CRS))
+}
+
func TestWriteEqualityDeleteFilesMultiColumnKey(t *testing.T) {
location := filepath.ToSlash(t.TempDir())
diff --git a/table/position_delta_writer.go b/table/position_delta_writer.go
index a27074c45..e7c97541f 100644
--- a/table/position_delta_writer.go
+++ b/table/position_delta_writer.go
@@ -180,7 +180,10 @@ func (w *PositionDeltaWriter) Close(ctx context.Context) ([]iceberg.DataFile, er
}
fileSchema := iceberg.SchemaWithRowID(w.tbl.Schema())
- arrowSc, err := SchemaToArrowSchema(fileSchema, nil, true, false)
+ arrowSc, err := SchemaToArrowSchemaWithOptions(fileSchema, ArrowSchemaOptions{
+ IncludeFieldIDs: true,
+ TableProperties: w.tbl.Metadata().Properties(),
+ })
if err != nil {
return nil, fmt.Errorf("PositionDeltaWriter: build arrow schema: %w", err)
}
diff --git a/table/projjson_roundtrip_internal_test.go b/table/projjson_roundtrip_internal_test.go
new file mode 100644
index 000000000..c7f3cb71e
--- /dev/null
+++ b/table/projjson_roundtrip_internal_test.go
@@ -0,0 +1,104 @@
+// 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"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "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/memory"
+ "github.com/apache/iceberg-go"
+ icebergio "github.com/apache/iceberg-go/io"
+ "github.com/stretchr/testify/require"
+)
+
+func TestWriteRecordsRecoversExactProjJSONCRSOnRead(t *testing.T) {
+ const projjson = `{"type":"GeographicCRS","name":"Custom WGS 84"}`
+
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(context.Background(), mem)
+
+ geom, err := iceberg.GeometryTypeOf("projjson:geo.crs")
+ require.NoError(t, err)
+
+ iceSchema := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "geom", Type: geom, Required: false},
+ )
+ spec := iceberg.NewPartitionSpec()
+ loc := filepath.ToSlash(t.TempDir())
+ meta, err := NewMetadata(iceSchema, &spec, UnsortedSortOrder, loc, iceberg.Properties{
+ PropertyFormatVersion: "3",
+ "geo.crs": projjson,
+ })
+ require.NoError(t, err)
+
+ tbl := New(
+ Identifier{"test", "projjson_exact_round_trip"},
+ meta,
+ filepath.Join(loc, "metadata", "v1.metadata.json"),
+ func(context.Context) (icebergio.IO, error) { return icebergio.LocalFS{}, nil },
+ nil,
+ )
+
+ arrowSchema, err := SchemaToArrowSchemaWithOptions(iceSchema, ArrowSchemaOptions{
+ IncludeFieldIDs: true,
+ TableProperties: meta.Properties(),
+ })
+ require.NoError(t, err)
+
+ records := func(yield func(arrow.RecordBatch, error) bool) {
+ rec, _, err := array.RecordFromJSON(mem, arrowSchema, strings.NewReader(`[
+ {"geom": "01010000000000000000003e400000000000002440"}
+ ]`))
+ require.NoError(t, err)
+ yield(rec, nil)
+ }
+
+ var dataFiles []iceberg.DataFile
+ for df, err := range WriteRecords(ctx, tbl, arrowSchema, records) {
+ require.NoError(t, err)
+ dataFiles = append(dataFiles, df)
+ }
+ require.Len(t, dataFiles, 1)
+
+ fs, err := tbl.fsF(ctx)
+ require.NoError(t, err)
+
+ readSchema, _, rdr, err := (&arrowScan{
+ metadata: tbl.Metadata(),
+ fs: fs,
+ projectedSchema: tbl.Schema(),
+ concurrency: 1,
+ nameMapping: tbl.Metadata().NameMapping(),
+ }).prepareToRead(ctx, dataFiles[0])
+ require.NoError(t, err)
+ defer rdr.Close()
+
+ readField, ok := readSchema.FindFieldByID(1)
+ require.True(t, ok)
+ readGeom, ok := readField.Type.(iceberg.GeometryType)
+ require.True(t, ok)
+ require.Equal(t, "projjson:geo.crs", readGeom.CRS())
+ require.True(t, iceSchema.Equals(readSchema), "expected %s, got %s", iceSchema, readSchema)
+}
diff --git a/table/rewrite_data_files.go b/table/rewrite_data_files.go
index da7df508b..c9c5b5514 100644
--- a/table/rewrite_data_files.go
+++ b/table/rewrite_data_files.go
@@ -378,7 +378,10 @@ func ExecuteCompactionGroup(ctx context.Context, tbl *Table, group CompactionTas
// table's name mapping — which doesn't (and cannot) contain the
// reserved metadata column names, so the fallback path panics.
projectedSchema := iceberg.SchemaWithRowLineage(tbl.Schema())
- arrowSchema, err = SchemaToArrowSchema(projectedSchema, nil, true, false)
+ arrowSchema, err = SchemaToArrowSchemaWithOptions(projectedSchema, ArrowSchemaOptions{
+ IncludeFieldIDs: true,
+ TableProperties: tbl.Metadata().Properties(),
+ })
if err != nil {
return CompactionGroupResult{}, fmt.Errorf("build arrow schema for lineage write in group %q: %w", group.PartitionKey, err)
}
diff --git a/table/rolling_data_writer.go b/table/rolling_data_writer.go
index 6fc2ad1ed..2fee71e02 100644
--- a/table/rolling_data_writer.go
+++ b/table/rolling_data_writer.go
@@ -102,7 +102,10 @@ func withFactoryEqualityFieldIDs(ids []int) writerFactoryOption {
func withFactoryFileSchema(schema *iceberg.Schema) writerFactoryOption {
return func(w *writerFactory) error {
w.fileSchema = schema
- arrowSc, err := SchemaToArrowSchema(schema, nil, true, false)
+ arrowSc, err := SchemaToArrowSchemaWithOptions(schema, ArrowSchemaOptions{
+ IncludeFieldIDs: true,
+ TableProperties: w.tableProps,
+ })
if err != nil {
return fmt.Errorf("withFactoryFileSchema: failed to convert schema: %w", err)
}
@@ -159,7 +162,10 @@ func newWriterFactory(rootLocation string, args recordWritingArgs, meta *Metadat
return nil, err
}
- arrowSchema, err := SchemaToArrowSchema(fileSchema, nil, true, false)
+ arrowSchema, err := SchemaToArrowSchemaWithOptions(fileSchema, ArrowSchemaOptions{
+ IncludeFieldIDs: true,
+ TableProperties: meta.props,
+ })
if err != nil {
stopCount()
@@ -522,6 +528,7 @@ func (r *RollingDataWriter) stream(outputDataFilesCh chan<- iceberg.DataFile) {
DowncastTimestamp: true,
IncludeFieldIDs: true,
UseWriteDefault: true,
+ TableProperties: r.factory.tableProps,
})
record.Release()
if err != nil {
diff --git a/table/transaction.go b/table/transaction.go
index 8eb13b8a8..527d42443 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -36,7 +36,7 @@ import (
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/table/dv"
- "github.com/apache/iceberg-go/table/internal"
+ tblutils "github.com/apache/iceberg-go/table/internal"
"github.com/apache/iceberg-go/table/substrait"
"github.com/google/uuid"
"golang.org/x/sync/errgroup"
@@ -2010,7 +2010,10 @@ func (t *Transaction) rewriteSingleFile(ctx context.Context, args rewriteSingleF
// recordsToDataFiles can resolve _row_id via field ID rather than the name
// mapping (which doesn't include metadata columns).
if preserveRowLineage {
- arrowSchema, err = SchemaToArrowSchema(projectedSchema, nil, true, false)
+ arrowSchema, err = SchemaToArrowSchemaWithOptions(projectedSchema, ArrowSchemaOptions{
+ IncludeFieldIDs: true,
+ TableProperties: builtMeta.Properties(),
+ })
if err != nil {
return nil, fmt.Errorf("failed to build arrow schema with field IDs: %w", err)
}
@@ -2254,7 +2257,7 @@ func (t *Transaction) makePositionDeleteRecordsForFilter(ctx context.Context, fs
extSet := substrait.NewExtensionSet()
ctx, cancel := context.WithCancelCause(exprs.WithExtensionIDSet(ctx, extSet))
- taskChan := make(chan internal.Enumerated[FileScanTask], len(tasks))
+ taskChan := make(chan tblutils.Enumerated[FileScanTask], len(tasks))
numWorkers := min(concurrency, len(tasks))
records := make(chan enumeratedRecord, numWorkers)
@@ -2285,7 +2288,7 @@ func (t *Transaction) makePositionDeleteRecordsForFilter(ctx context.Context, fs
go func() {
for i, t := range tasks {
- taskChan <- internal.Enumerated[FileScanTask]{
+ taskChan <- tblutils.Enumerated[FileScanTask]{
Value: t, Index: i, Last: i == len(tasks)-1,
}
}
diff --git a/table/write_records.go b/table/write_records.go
index 3f6192002..83c2936c7 100644
--- a/table/write_records.go
+++ b/table/write_records.go
@@ -161,9 +161,9 @@ func WriteRecords(ctx context.Context, tbl *Table,
if cfg.fileSchema != nil {
checkSchema = cfg.fileSchema
}
- err := checkArrowSchemaCompat(checkSchema, schema, false)
+ err := checkArrowSchemaCompatWithProperties(checkSchema, schema, false, tbl.Metadata().Properties())
if err != nil {
- if downcastErr := checkArrowSchemaCompat(checkSchema, schema, true); downcastErr == nil {
+ if downcastErr := checkArrowSchemaCompatWithProperties(checkSchema, schema, true, tbl.Metadata().Properties()); downcastErr == nil {
err = nil
}
}
diff --git a/table/write_records_test.go b/table/write_records_test.go
index 474e2f88f..bfe5048e2 100644
--- a/table/write_records_test.go
+++ b/table/write_records_test.go
@@ -20,7 +20,9 @@ package table_test
import (
"context"
"errors"
+ "iter"
"path/filepath"
+ "strings"
"testing"
"github.com/apache/arrow-go/v18/arrow"
@@ -32,6 +34,7 @@ import (
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/table"
+ "github.com/geoarrow/geoarrow-go"
"github.com/google/uuid"
"github.com/stretchr/testify/suite"
)
@@ -116,6 +119,94 @@ func (s *WriteRecordsTestSuite) TestBasicWrite() {
s.Contains(dataFiles[0].FilePath(), loc)
}
+func (s *WriteRecordsTestSuite) TestWriteRecordsResolvesProjJSONCRSFromTableProperties() {
+ const projjson = `{"type":"GeographicCRS","name":"Custom WGS 84"}`
+ loc := filepath.ToSlash(s.T().TempDir())
+ geom, err := iceberg.GeometryTypeOf("projjson:geo.crs")
+ s.Require().NoError(err)
+
+ iceSch := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "geom", Type: geom, Required: false},
+ )
+ spec := iceberg.NewPartitionSpec()
+ meta, err := table.NewMetadata(iceSch, &spec, table.UnsortedSortOrder, loc, iceberg.Properties{
+ table.PropertyFormatVersion: "3",
+ "geo.crs": projjson,
+ })
+ s.Require().NoError(err)
+
+ tbl := table.New(
+ table.Identifier{"test", "write_records_projjson"},
+ meta,
+ filepath.Join(loc, "metadata", "v1.metadata.json"),
+ func(ctx context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil },
+ nil,
+ )
+
+ arrowSchema, err := table.SchemaToArrowSchemaWithOptions(iceSch, table.ArrowSchemaOptions{
+ IncludeFieldIDs: true,
+ TableProperties: meta.Properties(),
+ })
+ s.Require().NoError(err)
+
+ records := func(yield func(arrow.RecordBatch, error) bool) {
+ rec, _, err := array.RecordFromJSON(s.mem, arrowSchema, strings.NewReader(`[
+ {"geom": "01010000000000000000003e400000000000002440"}
+ ]`))
+ s.Require().NoError(err)
+ yield(rec, nil)
+ }
+
+ var dataFiles []iceberg.DataFile
+ for df, err := range table.WriteRecords(s.ctx, tbl, arrowSchema, records) {
+ s.Require().NoError(err)
+ dataFiles = append(dataFiles, df)
+ }
+ s.Require().Len(dataFiles, 1)
+
+ rdr, err := file.OpenParquetFile(dataFiles[0].FilePath(), false)
+ s.Require().NoError(err)
+ defer rdr.Close()
+
+ arrowRdr, err := pqarrow.NewFileReader(rdr, pqarrow.ArrowReadProperties{}, memory.DefaultAllocator)
+ s.Require().NoError(err)
+
+ written, err := arrowRdr.ReadTable(s.ctx)
+ s.Require().NoError(err)
+ defer written.Release()
+
+ wkb, ok := written.Schema().Field(0).Type.(*geoarrow.WKBType)
+ s.Require().True(ok)
+ geoMeta := wkb.Metadata()
+ s.Equal(geoarrow.CRSTypePROJJSON, geoMeta.CRSType)
+ s.JSONEq(projjson, string(geoMeta.CRS))
+
+ // Read the written file through the table-aware scan path. This verifies
+ // that embedded GeoArrow PROJJSON metadata remains usable when the reader
+ // projects records using the table schema and properties.
+ scan := tbl.Scan()
+ readSchema, readRecords, err := scan.ReadTasks(s.ctx, []table.FileScanTask{{File: dataFiles[0]}})
+ s.Require().NoError(err)
+ readWKB, ok := readSchema.Field(0).Type.(*geoarrow.WKBType)
+ s.Require().True(ok)
+ readMeta := readWKB.Metadata()
+ s.Equal(geoarrow.CRSTypePROJJSON, readMeta.CRSType)
+ s.JSONEq(projjson, string(readMeta.CRS))
+
+ next, stop := iter.Pull2(readRecords)
+ defer stop()
+
+ readRecord, err, ok := next()
+ s.Require().True(ok)
+ s.Require().NoError(err)
+ defer readRecord.Release()
+ s.Equal(int64(1), readRecord.NumRows())
+
+ readGeom, ok := tbl.Schema().Field(0).Type.(iceberg.GeometryType)
+ s.Require().True(ok)
+ s.Equal("projjson:geo.crs", readGeom.CRS())
+}
+
func (s *WriteRecordsTestSuite) TestSmallTargetFileSizeProducesMultipleFiles() {
loc := filepath.ToSlash(s.T().TempDir())
tbl := s.newTable(loc)
diff --git a/table/writer.go b/table/writer.go
index bd52ddddc..91382ab76 100644
--- a/table/writer.go
+++ b/table/writer.go
@@ -27,7 +27,7 @@ import (
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/config"
"github.com/apache/iceberg-go/io"
- "github.com/apache/iceberg-go/table/internal"
+ tblutils "github.com/apache/iceberg-go/table/internal"
"github.com/google/uuid"
)
@@ -61,7 +61,7 @@ type defaultDataFileWriter struct {
fs io.WriteFileIO
fileSchema *iceberg.Schema
fileFormat iceberg.FileFormat
- format internal.FileFormat
+ format tblutils.FileFormat
props iceberg.Properties
content iceberg.ManifestEntryContent
meta *MetadataBuilder
@@ -70,7 +70,7 @@ type defaultDataFileWriter struct {
type dataFileWriterOption func(writer *defaultDataFileWriter)
-func withFormat(format internal.FileFormat) dataFileWriterOption {
+func withFormat(format tblutils.FileFormat) dataFileWriterOption {
return func(writer *defaultDataFileWriter) {
writer.format = format
}
@@ -111,7 +111,7 @@ func newDataFileWriter(rootLocation string, fs io.WriteFileIO, meta *MetadataBui
fs: fs,
fileSchema: meta.CurrentSchema(),
fileFormat: fileFormat,
- format: internal.GetFileFormat(fileFormat),
+ format: tblutils.GetFileFormat(fileFormat),
content: iceberg.EntryContentData,
props: props,
meta: meta,
@@ -143,6 +143,7 @@ func (w *defaultDataFileWriter) writeFile(ctx context.Context, partitionValues m
DowncastTimestamp: true,
IncludeFieldIDs: true,
UseWriteDefault: true,
+ TableProperties: w.meta.props,
})
if err != nil {
return nil, err
@@ -164,12 +165,12 @@ func (w *defaultDataFileWriter) writeFile(ctx context.Context, partitionValues m
return nil, err
}
- rowGroupTargetSizeBytes, err := internal.ParquetRowGroupTargetSizeBytes(w.props)
+ rowGroupTargetSizeBytes, err := tblutils.ParquetRowGroupTargetSizeBytes(w.props)
if err != nil {
return nil, err
}
- return w.format.WriteDataFile(ctx, w.fs, partitionValues, internal.WriteFileInfo{
+ return w.format.WriteDataFile(ctx, w.fs, partitionValues, tblutils.WriteFileInfo{
FileSchema: w.fileSchema,
Content: w.content,
FileName: filePath,
@@ -255,7 +256,7 @@ func (w *concurrentDataFileWriter) writeFiles(ctx context.Context, rootLocation
}
}
- return internal.MapExec(ctx, config.EnvConfig.MaxWorkers, tasks, func(t WriteTask) (iceberg.DataFile, error) {
+ return tblutils.MapExec(ctx, config.EnvConfig.MaxWorkers, tasks, func(t WriteTask) (iceberg.DataFile, error) {
return fw.writeFile(ctx, partitionValues, t)
})
}