feat(table): Adding geometry and geography type + schema plumbing

exprs.go

@@ -450,6 +450,8 @@ func createBoundRef(field NestedField, acc accessor) BoundReference {
 		return &boundRef[Decimal]{field: field, acc: acc}
 	case UUIDType:
 		return &boundRef[uuid.UUID]{field: field, acc: acc}
+	case GeographyType, GeometryType:
+		return &boundRef[[]byte]{field: field, acc: acc}
 	}
 	panic("unhandled bound reference type: " + field.Type.String())
 }

schema.go

@@ -567,6 +567,8 @@ type SchemaVisitorPerPrimitiveType[T any] interface {
 	VisitBinary() T
 	VisitUUID() T
 	VisitUnknown() T
+	VisitGeometry(GeometryType) T
+	VisitGeography(GeographyType) T
 }
 
 // Visit accepts a visitor and performs a post-order traversal of the given schema.

schema_test.go

@@ -1035,3 +1035,177 @@ func TestSanitizeColumnNamesEmptyFieldName(t *testing.T) {
 	assert.ErrorIs(t, err, iceberg.ErrInvalidSchema)
 	assert.ErrorContains(t, err, "field name cannot be empty")
 }
+
+func TestSchemaWithGeometryGeographyTypes(t *testing.T) {
+	geom, err := iceberg.GeometryTypeOf("srid:4326")
+	require.NoError(t, err)
+	geog, err := iceberg.GeographyTypeOf("srid:4269", iceberg.EdgeAlgorithmKarney)
+	require.NoError(t, err)
+
+	schema := iceberg.NewSchema(1,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+		iceberg.NestedField{ID: 2, Name: "simple_point", Type: iceberg.GeometryType{}, Required: false},
+		iceberg.NestedField{ID: 3, Name: "location", Type: geom, Required: false},
+		iceberg.NestedField{ID: 4, Name: "service_area", Type: geog, Required: false},
+	)
+
+	data, err := json.Marshal(schema)
+	require.NoError(t, err)
+
+	assert.JSONEq(t, `{
+		"type": "struct",
+		"schema-id": 1,
+		"identifier-field-ids": [],
+		"fields": [
+			{"id": 1, "name": "id", "type": "long", "required": true},
+			{"id": 2, "name": "simple_point", "type": "geometry", "required": false},
+			{"id": 3, "name": "location", "type": "geometry(srid:4326)", "required": false},
+			{"id": 4, "name": "service_area", "type": "geography(srid:4269, karney)", "required": false}
+		]
+	}`, string(data))
+
+	var unmarshaledSchema iceberg.Schema
+	require.NoError(t, json.Unmarshal(data, &unmarshaledSchema))
+	assert.True(t, schema.Equals(&unmarshaledSchema))
+}
+
+func TestNestedFieldToStringGeographyGeometry(t *testing.T) {
+	geom, err := iceberg.GeometryTypeOf("srid:3857")
+	require.NoError(t, err)
+	geog, err := iceberg.GeographyTypeOf("srid:4269", iceberg.EdgeAlgorithmKarney)
+	require.NoError(t, err)
+
+	tests := []struct {
+		field    iceberg.NestedField
+		expected string
+	}{
+		{
+			iceberg.NestedField{ID: 1, Name: "point", Type: iceberg.GeometryType{}, Required: false},
+			"1: point: optional geometry",
+		},
+		{
+			iceberg.NestedField{ID: 2, Name: "location", Type: geom, Required: true},
+			"2: location: required geometry(srid:3857)",
+		},
+		{
+			iceberg.NestedField{ID: 3, Name: "area", Type: iceberg.GeographyType{}, Required: false},
+			"3: area: optional geography",
+		},
+		{
+			iceberg.NestedField{ID: 4, Name: "region", Type: geog, Required: false},
+			"4: region: optional geography(srid:4269, karney)",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.field.Name, func(t *testing.T) {
+			assert.Equal(t, tt.expected, tt.field.String())
+		})
+	}
+}
+
+func TestSchemaWithGeometryInNestedStructures(t *testing.T) {
+	geom, err := iceberg.GeometryTypeOf("srid:4326")
+	require.NoError(t, err)
+	geog, err := iceberg.GeographyTypeOf("srid:4269", iceberg.EdgeAlgorithmSpherical)
+	require.NoError(t, err)
+
+	schema := iceberg.NewSchema(1,
+		iceberg.NestedField{
+			ID:   1,
+			Name: "locations",
+			Type: &iceberg.ListType{
+				ElementID:       2,
+				Element:         geom,
+				ElementRequired: true,
+			},
+			Required: true,
+		},
+		iceberg.NestedField{
+			ID:   3,
+			Name: "region_data",
+			Type: &iceberg.MapType{
+				KeyID:         4,
+				KeyType:       iceberg.PrimitiveTypes.String,
+				ValueID:       5,
+				ValueType:     geog,
+				ValueRequired: false,
+			},
+			Required: false,
+		},
+		iceberg.NestedField{
+			ID:   6,
+			Name: "place",
+			Type: &iceberg.StructType{
+				FieldList: []iceberg.NestedField{
+					{ID: 7, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: true},
+					{ID: 8, Name: "coords", Type: iceberg.GeometryType{}, Required: false},
+				},
+			},
+			Required: false,
+		},
+	)
+
+	data, err := json.Marshal(schema)
+	require.NoError(t, err)
+
+	var unmarshaledSchema iceberg.Schema
+	require.NoError(t, json.Unmarshal(data, &unmarshaledSchema))
+	assert.True(t, schema.Equals(&unmarshaledSchema))
+
+	assert.Equal(t, "1: locations: required list<geometry(srid:4326)>", schema.Field(0).String())
+	assert.Equal(t, "3: region_data: optional map<string, geography(srid:4269, spherical)>", schema.Field(1).String())
+}
+
+func TestPruneColumnsWithGeometry(t *testing.T) {
+	geom, err := iceberg.GeometryTypeOf("srid:4326")
+	require.NoError(t, err)
+
+	schema := iceberg.NewSchema(1,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+		iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false},
+		iceberg.NestedField{ID: 3, Name: "location", Type: geom, Required: false},
+	)
+
+	pruned, err := iceberg.PruneColumns(schema, map[int]iceberg.Void{1: {}, 3: {}}, false)
+	require.NoError(t, err)
+
+	expected := iceberg.NewSchema(1,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+		iceberg.NestedField{ID: 3, Name: "location", Type: geom, Required: false},
+	)
+
+	assert.True(t, pruned.Equals(expected))
+}
+
+func TestSchemaIndexByIDWithGeography(t *testing.T) {
+	geog, err := iceberg.GeographyTypeOf("srid:4269", iceberg.EdgeAlgorithmKarney)
+	require.NoError(t, err)
+
+	schema := iceberg.NewSchema(1,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+		iceberg.NestedField{ID: 2, Name: "area", Type: geog, Required: false},
+	)
+
+	index, err := iceberg.IndexByID(schema)
+	require.NoError(t, err)
+
+	assert.Len(t, index, 2)
+	assert.Equal(t, geog, index[2].Type)
+	assert.Equal(t, "area", index[2].Name)
+}
+
+func TestSchemaFindColumnNameWithGeometryGeography(t *testing.T) {
+	schema := iceberg.NewSchema(1,
+		iceberg.NestedField{ID: 1, Name: "point", Type: iceberg.GeometryType{}, Required: false},
+		iceberg.NestedField{ID: 2, Name: "region", Type: iceberg.GeographyType{}, Required: false},
+	)
+
+	name, ok := schema.FindColumnName(1)
+	assert.True(t, ok)
+	assert.Equal(t, "point", name)
+
+	name, ok = schema.FindColumnName(2)
+	assert.True(t, ok)
+	assert.Equal(t, "region", name)
+}

table/arrow_utils.go

@@ -630,6 +630,24 @@ func (c convertToArrow) VisitUnknown() arrow.Field {
 	}
 }
 
+func (c convertToArrow) VisitGeometry(iceberg.GeometryType) arrow.Field {
+	// Passthrough binary for now, adding geoarrow-go support later
+	if c.useLargeTypes {
+		return arrow.Field{Type: arrow.BinaryTypes.LargeBinary}
+	}
+
+	return arrow.Field{Type: arrow.BinaryTypes.Binary}
+}
+
+func (c convertToArrow) VisitGeography(iceberg.GeographyType) arrow.Field {
+	// Passthrough binary for now, adding geoarrow-go support later
+	if c.useLargeTypes {
+		return arrow.Field{Type: arrow.BinaryTypes.LargeBinary}
+	}
+
+	return arrow.Field{Type: arrow.BinaryTypes.Binary}
+}
+
 var _ iceberg.SchemaVisitorPerPrimitiveType[arrow.Field] = convertToArrow{}
 
 // SchemaToArrowSchema converts an Iceberg schema to an Arrow schema. If the metadata parameter

table/metadata_builder_internal_test.go

@@ -1410,6 +1410,8 @@ func TestUnsupportedTypes(t *testing.T) {
 	TestTypes := []iceberg.Type{
 		iceberg.TimestampNsType{},
 		iceberg.TimestampTzNsType{},
+		iceberg.GeometryType{},
+		iceberg.GeographyType{},
 	}
 	for _, typ := range TestTypes {
 		for unsupportedVersion := 1; unsupportedVersion < minFormatVersionForType(typ); unsupportedVersion++ {
@@ -1838,6 +1840,86 @@ func TestUnknownTypeValidation(t *testing.T) {
 	})
 }
 
+func TestGeometryGeographyNullOnlyDefaults(t *testing.T) {
+	testTypes := []struct {
+		name string
+		typ  iceberg.Type
+	}{
+		{"geometry", iceberg.GeometryType{}},
+		{"geography", iceberg.GeographyType{}},
+	}
+
+	for _, tt := range testTypes {
+		t.Run(tt.name+" with non-null initial default", func(t *testing.T) {
+			defaultValue := "POINT(0 0)"
+			sc := iceberg.NewSchema(0,
+				iceberg.NestedField{
+					Type:           tt.typ,
+					ID:             1,
+					Name:           "location",
+					Required:       false,
+					InitialDefault: &defaultValue,
+				},
+			)
+
+			err := checkSchemaCompatibility(sc, 3)
+			require.Error(t, err)
+			require.ErrorContains(t, err, "columns must default to null")
+			require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+		})
+
+		t.Run(tt.name+" with non-null write default", func(t *testing.T) {
+			defaultValue := "POINT(0 0)"
+			sc := iceberg.NewSchema(0,
+				iceberg.NestedField{
+					Type:         tt.typ,
+					ID:           1,
+					Name:         "location",
+					Required:     false,
+					WriteDefault: &defaultValue,
+				},
+			)
+
+			err := checkSchemaCompatibility(sc, 3)
+			require.Error(t, err)
+			require.ErrorContains(t, err, "columns must default to null")
+			require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+		})
+
+		t.Run(tt.name+" with null defaults", func(t *testing.T) {
+			sc := iceberg.NewSchema(0,
+				iceberg.NestedField{
+					Type:     tt.typ,
+					ID:       1,
+					Name:     "location",
+					Required: false,
+				},
+			)
+
+			err := checkSchemaCompatibility(sc, 3)
+			require.NoError(t, err)
+		})
+
+		t.Run(tt.name+" in v2 with non-null initial default", func(t *testing.T) {
+			defaultValue := "POINT(0 0)"
+			sc := iceberg.NewSchema(0,
+				iceberg.NestedField{
+					Type:           tt.typ,
+					ID:             1,
+					Name:           "location",
+					Required:       false,
+					InitialDefault: &defaultValue,
+				},
+			)
+
+			err := checkSchemaCompatibility(sc, 2)
+			require.Error(t, err)
+			require.ErrorContains(t, err, "is not supported until v3")
+			require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+		})
+	}
+}
+
 func TestComplexTypeDefaultValidation(t *testing.T) {
 	t.Run("InvalidStructInitialDefault", func(t *testing.T) {
 		schema := iceberg.NewSchema(1,

table/metadata_schema_compatibility.go

@@ -39,7 +39,17 @@ func (e ErrIncompatibleSchema) Error() string {
 			fmt.Fprintf(&problems, "\n- invalid type for %s: %s is not supported until v%d", f.ColName, f.Field.Type, f.UnsupportedType.MinFormatVersion)
 		}
 		if f.InvalidDefault != nil {
-			fmt.Fprintf(&problems, "\n- invalid initial default for %s: non-null default (%v) is not supported until v%d", f.ColName, f.Field.InitialDefault, f.InvalidDefault.MinFormatVersion)
+			switch f.Field.Type.(type) {
+			case iceberg.GeometryType, iceberg.GeographyType:
+				if f.Field.InitialDefault != nil {
+					fmt.Fprintf(&problems, "\n- invalid initial default for %s: %s columns must default to null", f.ColName, f.Field.Type)
+				}
+				if f.Field.WriteDefault != nil {
+					fmt.Fprintf(&problems, "\n- invalid write default for %s: %s columns must default to null", f.ColName, f.Field.Type)
+				}
+			default:
+				fmt.Fprintf(&problems, "\n- invalid initial default for %s: non-null default (%v) is not supported until v%d", f.ColName, f.Field.InitialDefault, f.InvalidDefault.MinFormatVersion)
+			}
 		}
 	}
 
@@ -113,12 +123,30 @@ func checkSchemaCompatibility(sc *iceberg.Schema, formatVersion int) error {
 			})
 		}
 
-		if field.InitialDefault != nil && formatVersion < defaultValuesMinFormatVersion {
-			problems = append(problems, IncompatibleField{
-				Field:          field,
-				ColName:        colName,
-				InvalidDefault: &InvalidDefault{MinFormatVersion: defaultValuesMinFormatVersion, WriteDefault: field.InitialDefault},
-			})
+		switch field.Type.(type) {
+		case iceberg.GeometryType, iceberg.GeographyType:
+			if field.InitialDefault != nil {
+				problems = append(problems, IncompatibleField{
+					Field:          field,
+					ColName:        colName,
+					InvalidDefault: &InvalidDefault{MinFormatVersion: formatVersion, WriteDefault: field.InitialDefault},
+				})
+			}
+			if field.WriteDefault != nil {
+				problems = append(problems, IncompatibleField{
+					Field:          field,
+					ColName:        colName,
+					InvalidDefault: &InvalidDefault{MinFormatVersion: formatVersion, WriteDefault: field.WriteDefault},
+				})
+			}
+		default:
+			if field.InitialDefault != nil && formatVersion < defaultValuesMinFormatVersion {
+				problems = append(problems, IncompatibleField{
+					Field:          field,
+					ColName:        colName,
+					InvalidDefault: &InvalidDefault{MinFormatVersion: defaultValuesMinFormatVersion, WriteDefault: field.InitialDefault},
+				})
+			}
 		}
 	}
 
@@ -134,7 +162,7 @@ func checkSchemaCompatibility(sc *iceberg.Schema, formatVersion int) error {
 // version number for types that require newer format versions.
 func minFormatVersionForType(t iceberg.Type) int {
 	switch t.(type) {
-	case iceberg.TimestampNsType, iceberg.TimestampTzNsType, iceberg.UnknownType:
+	case iceberg.TimestampNsType, iceberg.TimestampTzNsType, iceberg.UnknownType, iceberg.GeometryType, iceberg.GeographyType:
 		return 3
 	default:
 		// All other types supported in v1+

table/substrait/substrait.go

@@ -169,6 +169,10 @@ func (convertToSubstrait) VisitUnknown() types.Type {
 	// Returning nil indicates this type cannot be converted to Substrait
 	return nil
 }
+func (convertToSubstrait) VisitGeometry(iceberg.GeometryType) types.Type { return &types.BinaryType{} }
+func (convertToSubstrait) VisitGeography(iceberg.GeographyType) types.Type {
+	return &types.BinaryType{}
+}
 
 var _ iceberg.SchemaVisitorPerPrimitiveType[types.Type] = (*convertToSubstrait)(nil)