feat(table): WKB encoding + GeoArrow conversion for geometry/geography

go.mod

@@ -47,6 +47,7 @@ require (
 	github.com/testcontainers/testcontainers-go/modules/compose v0.42.0
 	github.com/twmb/avro v1.7.2
 	github.com/twmb/murmur3 v1.1.8
+	github.com/twpayne/go-geom v1.6.1
 	github.com/uptrace/bun v1.2.18
 	github.com/uptrace/bun/dialect/mssqldialect v1.2.18
 	github.com/uptrace/bun/dialect/mysqldialect v1.2.18

go.sum

@@ -56,6 +56,8 @@ github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1 h1:WJ
 github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1/go.mod h1:tCcJZ0uHAmvjsVYzEFivsRTN00oz5BEsRgQHu5JZ9WE=
 github.com/AzureAD/microsoft-authentication-library-for-go v1.7.2 h1:RHK7bS+HQMslb1sZpAokUt+zTVmue0hKSs2C791hhzU=
 github.com/AzureAD/microsoft-authentication-library-for-go v1.7.2/go.mod h1:HKpQxkWaGLJ+D/5H8QRpyQXA1eKjxkFlOMwck5+33Jk=
+github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU=
+github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
 github.com/DefangLabs/secret-detector v0.0.0-20250403165618-22662109213e h1:rd4bOvKmDIx0WeTv9Qz+hghsgyjikFiPrseXHlKepO0=
 github.com/DefangLabs/secret-detector v0.0.0-20250403165618-22662109213e/go.mod h1:blbwPQh4DTlCZEfk1BLU4oMIhLda2U+A840Uag9DsZw=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.31.0 h1:DHa2U07rk8syqvCge0QIGMCE1WxGj9njT44GH7zNJLQ=
@@ -85,6 +87,10 @@ github.com/RoaringBitmap/roaring/v2 v2.18.2 h1:oPq3Cgx//iDuJQVp6xSInAKW34J9CEwE5
 github.com/RoaringBitmap/roaring/v2 v2.18.2/go.mod h1:eq4wdNXxtJIS/oikeCzdX1rBzek7ANzbth041hrU8Q4=
 github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d h1:licZJFw2RwpHMqeKTCYkitsPqHNxTmd4SNR5r94FGM8=
 github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d/go.mod h1:asat636LX7Bqt5lYEZ27JNDcqxfjdBQuJ/MM4CN/Lzo=
+github.com/alecthomas/assert/v2 v2.10.0 h1:jjRCHsj6hBJhkmhznrCzoNpbA3zqy0fYiUcYZP/GkPY=
+github.com/alecthomas/assert/v2 v2.10.0/go.mod h1:Bze95FyfUr7x34QZrjL+XP+0qgp/zg8yS+TtBj1WA3k=
+github.com/alecthomas/repr v0.4.0 h1:GhI2A8MACjfegCPVq9f1FLvIBS+DrQ2KQBFZP1iFzXc=
+github.com/alecthomas/repr v0.4.0/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4=
 github.com/alexflint/go-arg v1.6.1 h1:uZogJ6VDBjcuosydKgvYYRhh9sRCusjOvoOLZopBlnA=
 github.com/alexflint/go-arg v1.6.1/go.mod h1:nQ0LFYftLJ6njcaee0sU+G0iS2+2XJQfA8I062D0LGc=
 github.com/alexflint/go-scalar v1.2.0 h1:WR7JPKkeNpnYIOfHRa7ivM21aWAdHD0gEWHCx+WQBRw=
@@ -392,6 +398,8 @@ github.com/hashicorp/go-version v1.9.0 h1:CeOIz6k+LoN3qX9Z0tyQrPtiB1DFYRPfCIBtaX
 github.com/hashicorp/go-version v1.9.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
 github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
+github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM=
+github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg=
 github.com/in-toto/attestation v1.1.2 h1:MBFn6lsMq6dptQZJBhalXTcWMb/aJy3V+GX3VYj/V1E=
 github.com/in-toto/attestation v1.1.2/go.mod h1:gYFddHMZj3DiQ0b62ltNi1Vj5rC879bTmBbrv9CRHpM=
 github.com/in-toto/in-toto-golang v0.11.0 h1:nfidMYBFx+E0lnmX5KUnN2Pdm8zdNKal1ayjJuzzRoA=
@@ -624,6 +632,8 @@ github.com/twmb/avro v1.7.2 h1:cmrEBRSbELRqsg/dRkQvVWuOaR2EfGifHIt/2iJ9lfI=
 github.com/twmb/avro v1.7.2/go.mod h1:X0fT1dY2xcbV4YuCE4mYro+qljHl4kUF5uA/2z1rgSk=
 github.com/twmb/murmur3 v1.1.8 h1:8Yt9taO/WN3l08xErzjeschgZU2QSrwm1kclYq+0aRg=
 github.com/twmb/murmur3 v1.1.8/go.mod h1:Qq/R7NUyOfr65zD+6Q5IHKsJLwP7exErjN6lyyq3OSQ=
+github.com/twpayne/go-geom v1.6.1 h1:iLE+Opv0Ihm/ABIcvQFGIiFBXd76oBIar9drAwHFhR4=
+github.com/twpayne/go-geom v1.6.1/go.mod h1:Kr+Nly6BswFsKM5sd31YaoWS5PeDDH2NftJTK7Gd028=
 github.com/uptrace/bun v1.2.18 h1:3HnRcMfS6OBPMG1eSOzlbFJ/X/AyMEJb7rMxE6VQvDU=
 github.com/uptrace/bun v1.2.18/go.mod h1:wNltaKJk4JtOt4SG5I5zmA7v0/Mzjh1+/S906Rayd3Y=
 github.com/uptrace/bun/dialect/mssqldialect v1.2.18 h1:nYzHoyJKJlIyl5i95Exi8ZTK8ooKWG+o3z3f404d/yQ=

table/arrow_utils.go

@@ -18,12 +18,15 @@
 package table
 
 import (
+	"bytes"
 	"context"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"iter"
 	"log/slog"
+	"regexp"
 	"slices"
 	"strconv"
 	"strings"
@@ -375,6 +378,17 @@ func (c convertToIceberg) Primitive(dt arrow.DataType) (result iceberg.NestedFie
 			result.Type = iceberg.PrimitiveTypes.UUID
 		case "parquet.variant":
 			result.Type = iceberg.VariantType{}
+		case "geoarrow.wkb":
+			wkb, ok := dt.(*geoarrow.WKBType)
+			if !ok {
+				panic(fmt.Errorf("%w: unsupported arrow type for conversion - %s", iceberg.ErrInvalidSchema, dt))
+			}
+
+			iceType, err := geoArrowMetadataToIcebergType(wkb.Metadata())
+			if err != nil {
+				panic(fmt.Errorf("%w: converting geoarrow metadata: %w", iceberg.ErrInvalidSchema, err))
+			}
+			result.Type = iceType
 		default:
 			panic(fmt.Errorf("%w: unsupported arrow type for conversion - %s", iceberg.ErrInvalidSchema, dt))
 		}
@@ -658,20 +672,26 @@ func (c convertToArrow) VisitVariant() arrow.Field {
 	return arrow.Field{Type: extensions.NewDefaultVariantType()}
 }
 
-func (c convertToArrow) VisitGeometry(iceberg.GeometryType) arrow.Field {
+func (c convertToArrow) VisitGeometry(g iceberg.GeometryType) arrow.Field {
+	meta := icebergCRSToGeoArrowMetadata(g.CRS())
 	if c.useLargeTypes {
-		return arrow.Field{Type: geoarrow.NewWKBType(geoarrow.WKBWithLargeBinaryStorage())}
+		return arrow.Field{Type: geoarrow.NewWKBType(geoarrow.WKBWithLargeBinaryStorage(), geoarrow.WKBWithMetadata(meta))}
 	}
 
-	return arrow.Field{Type: geoarrow.NewWKBType(geoarrow.WKBWithBinaryStorage())}
+	return arrow.Field{Type: geoarrow.NewWKBType(geoarrow.WKBWithBinaryStorage(), geoarrow.WKBWithMetadata(meta))}
 }
 
-func (c convertToArrow) VisitGeography(iceberg.GeographyType) arrow.Field {
+func (c convertToArrow) VisitGeography(g iceberg.GeographyType) arrow.Field {
+	meta := icebergCRSToGeoArrowMetadata(g.CRS())
+	// Always add an edge to differentiate between Geography and Geometry arrow fields.
+	// Note that the edge convention is a best-effort hint and planar geography from other clients won't round-trip through Arrow alone.
+	meta.Edges = geoarrow.EdgeInterpolation(g.Algorithm())
+
 	if c.useLargeTypes {
-		return arrow.Field{Type: geoarrow.NewWKBType(geoarrow.WKBWithLargeBinaryStorage())}
+		return arrow.Field{Type: geoarrow.NewWKBType(geoarrow.WKBWithLargeBinaryStorage(), geoarrow.WKBWithMetadata(meta))}
 	}
 
-	return arrow.Field{Type: geoarrow.NewWKBType(geoarrow.WKBWithBinaryStorage())}
+	return arrow.Field{Type: geoarrow.NewWKBType(geoarrow.WKBWithBinaryStorage(), geoarrow.WKBWithMetadata(meta))}
 }
 
 var _ iceberg.SchemaVisitorPerPrimitiveType[arrow.Field] = convertToArrow{}
@@ -1826,3 +1846,150 @@ func positionDeleteRecordsToDataFilesDV(ctx context.Context, rootLocation string
 		}
 	}
 }
+
+func checkCRSString(rawCrs json.RawMessage) bool {
+	b := bytes.TrimSpace(rawCrs)
+
+	return len(b) > 0 && b[0] == '"'
+}
+
+func checkCRSJSON(rawCrs json.RawMessage) bool {
+	b := bytes.TrimSpace(rawCrs)
+
+	return len(b) > 0 && b[0] == '{'
+}
+
+func geoArrowCRSToIcebergCRS(meta geoarrow.Metadata) (string, error) {
+	if len(meta.CRS) == 0 {
+		return "srid:0", nil
+	}
+
+	switch {
+	case checkCRSString(meta.CRS):
+		var crs string
+
+		if err := json.Unmarshal(meta.CRS, &crs); err != nil {
+			return "", fmt.Errorf("invalid geoarrow CRS metadata: %w", err)
+		}
+
+		if strings.EqualFold(crs, "OGC:CRS84") || strings.EqualFold(crs, "EPSG:4326") {
+			return "OGC:CRS84", nil
+		}
+
+		switch meta.CRSType {
+		case geoarrow.CRSTypeSRID:
+			return "srid:" + crs, nil
+		case geoarrow.CRSTypeWKT22019:
+			return "", errors.New("CRS type wkt2:2019 not supported")
+		default:
+			if len(crs) <= 32 {
+				return crs, nil
+			}
+
+			return "", errors.New("crs length too long")
+		}
+	case checkCRSJSON(meta.CRS):
+		var crs map[string]json.RawMessage
+
+		if err := json.Unmarshal(meta.CRS, &crs); err != nil {
+			return "", fmt.Errorf("invalid geoarrow CRS metadata: %w", err)
+		}
+
+		idRaw, ok := crs["id"]
+		if !ok || len(idRaw) == 0 {
+			return "", errors.New("unsupported CRS")
+		}
+
+		var id map[string]json.RawMessage
+		if err := json.Unmarshal(idRaw, &id); err != nil {
+			return "", errors.New("unsupported CRS")
+		}
+
+		var authority string
+		if err := json.Unmarshal(id["authority"], &authority); err != nil || authority == "" {
+			return "", errors.New("unsupported CRS")
+		}
+
+		codeRaw := id["code"]
+		if len(codeRaw) == 0 {
+			return "", errors.New("unsupported CRS")
+		}
+
+		var code string
+		if err := json.Unmarshal(codeRaw, &code); err != nil {
+			var codeNum json.Number
+			if err := json.Unmarshal(codeRaw, &codeNum); err != nil {
+				return "", errors.New("unsupported CRS")
+			}
+			code = codeNum.String()
+		}
+		if code == "" {
+			return "", errors.New("unsupported CRS")
+		}
+
+		authorityCode := authority + ":" + code
+		if strings.EqualFold(authorityCode, "OGC:CRS84") || strings.EqualFold(authorityCode, "EPSG:4326") {
+			return "OGC:CRS84", nil
+		}
+
+		return authorityCode, nil
+	default:
+		return "", errors.New("unsupported CRS: CRS must either be omitted, a string, or a JSON object")
+	}
+}
+
+func geoArrowMetadataToIcebergType(meta geoarrow.Metadata) (iceberg.Type, error) {
+	crs, err := geoArrowCRSToIcebergCRS(meta)
+	if err != nil {
+		return nil, err
+	}
+
+	switch meta.Edges {
+	case geoarrow.EdgePlanar:
+		return iceberg.GeometryTypeOf(crs)
+	case geoarrow.EdgeVincenty, geoarrow.EdgeKarney, geoarrow.EdgeThomas, geoarrow.EdgeAndoyer, geoarrow.EdgeSpherical:
+		return iceberg.GeographyTypeOf(crs, string(meta.Edges))
+	default:
+		return nil, fmt.Errorf("unsupported geoarrow edges %q", meta.Edges)
+	}
+}
+
+var authorityCodeCRS = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9_-]*:[A-Za-z0-9_.-]+$`)
+
+func icebergCRSToGeoArrowMetadata(crs string) geoarrow.Metadata {
+	lowerCRS := strings.ToLower(crs)
+	if strings.HasPrefix(lowerCRS, "srid:") {
+		id := crs[len("srid:"):]
+
+		if id == "0" {
+			return geoarrow.NewMetadata() // srid:0 maps to omitted GeoArrow CRS
+		}
+
+		raw, _ := json.Marshal(id) //nolint:errcheck // Marshalling a string can't fail
+
+		return geoarrow.Metadata{
+			CRS:     raw,
+			CRSType: geoarrow.CRSTypeSRID,
+		}
+	}
+
+	if strings.HasPrefix(lowerCRS, "projjson:") {
+		panic(fmt.Errorf("%w: projjson CRS not supported yet", iceberg.ErrInvalidSchema))
+	}
+
+	var raw []byte
+
+	if lowerCRS == "epsg:4326" {
+		// collapse EPSG:4326 to OGC:CRS84
+		raw, _ = json.Marshal("OGC:CRS84") //nolint:errcheck // Marshalling a string can't fail
+	} else {
+		raw, _ = json.Marshal(crs) //nolint:errcheck // Marshalling a string can't fail
+	}
+
+	meta := geoarrow.Metadata{CRS: raw}
+	if authorityCodeCRS.MatchString(crs) {
+		meta.CRSType = geoarrow.CRSTypeAuthorityCode
+	}
+
+	return meta
+}