adding tests and arrow and substrait passthroughs

Signed-off-by: happydave1 <dzhao2004@gmail.com>

Author: happydave1

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.

table/arrow_utils.go

@@ -630,6 +630,22 @@ 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/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)
 

table/update_schema_test.go

@@ -344,6 +344,31 @@ func TestAddColumn(t *testing.T) {
 			}},
 		}, newSchema.Fields())
 	})
+
+	t.Run("test update schema with add geometry and geography columns", func(t *testing.T) {
+		table := New([]string{"id"}, testMetadata, "", nil, nil)
+		txn := table.NewTransaction()
+
+		geog, err := iceberg.GeographyTypeOf("srid:4269", iceberg.EdgeAlgorithmKarney)
+		assert.NoError(t, err)
+
+		newSchema, err := NewUpdateSchema(txn, true, true).
+			AddColumn([]string{"geom"}, iceberg.GeometryType{}, "", false, nil).
+			AddColumn([]string{"geog"}, geog, "", false, nil).
+			Apply()
+		assert.NoError(t, err)
+		assert.NotNil(t, newSchema)
+
+		geomField, ok := newSchema.FindFieldByName("geom")
+		assert.True(t, ok)
+		assert.Equal(t, 12, geomField.ID)
+		assert.Equal(t, iceberg.GeometryType{}, geomField.Type)
+
+		geogField, ok := newSchema.FindFieldByName("geog")
+		assert.True(t, ok)
+		assert.Equal(t, 13, geogField.ID)
+		assert.True(t, geogField.Type.Equals(geog))
+	})
 }
 
 func TestApplyChanges(t *testing.T) {
@@ -861,6 +886,31 @@ func TestErrorHandling(t *testing.T) {
 		assert.Contains(t, err.Error(), "cannot change column nullability from optional to required")
 	})
 
+	t.Run("test update geography CRS and edge algorithm without allowIncompatibleChanges", func(t *testing.T) {
+		currentGeog, err := iceberg.GeographyTypeOf("srid:4269", iceberg.EdgeAlgorithmKarney)
+		assert.NoError(t, err)
+		targetGeog, err := iceberg.GeographyTypeOf("srid:4326", iceberg.EdgeAlgorithmSpherical)
+		assert.NoError(t, err)
+
+		geoSchema := iceberg.NewSchema(1,
+			iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+			iceberg.NestedField{ID: 2, Name: "geog", Type: currentGeog, Required: false},
+		)
+		geoMeta, err := NewMetadata(geoSchema, nil, UnsortedSortOrder, "", iceberg.Properties{
+			PropertyFormatVersion: "3",
+		})
+		assert.NoError(t, err)
+
+		table := New([]string{"geo"}, geoMeta, "", nil, nil)
+		txn := table.NewTransaction()
+
+		_, err = NewUpdateSchema(txn, true, false).UpdateColumn([]string{"geog"}, ColumnUpdate{
+			FieldType: iceberg.Optional[iceberg.Type]{Valid: true, Val: targetGeog},
+		}).Apply()
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "cannot promote geography(srid:4269, karney) to geography(srid:4326, spherical)")
+	})
+
 	t.Run("test add required field without default value", func(t *testing.T) {
 		table := New([]string{"id"}, testMetadata, "", nil, nil)
 		txn := table.NewTransaction()

table/update_spec_test.go

@@ -204,6 +204,49 @@ func TestUpdateSpecAddField(t *testing.T) {
 		assert.NotNil(t, newSpec)
 		assert.Equal(t, "street_void_1001", newSpec.FieldsBySourceID(5)[0].Name)
 	})
+
+	t.Run("reject geometry source for identity partition transform", func(t *testing.T) {
+		geoSchema := iceberg.NewSchema(1,
+			iceberg.NestedField{ID: 1, Name: "id", Required: true, Type: iceberg.PrimitiveTypes.Int64},
+			iceberg.NestedField{ID: 2, Name: "geom", Required: false, Type: iceberg.GeometryType{}},
+		)
+		metadata, err := table.NewMetadata(geoSchema, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, "", nil)
+		assert.NoError(t, err)
+
+		tbl := table.New([]string{"geo_geometry"}, metadata, "", nil, nil)
+		specUpdate := table.NewUpdateSpec(tbl.NewTransaction(), true)
+
+		updates, reqs, err := specUpdate.
+			AddField("geom", iceberg.IdentityTransform{}, "geom_identity").
+			BuildUpdates()
+		assert.Error(t, err)
+		assert.Nil(t, updates)
+		assert.Nil(t, reqs)
+	})
+
+	t.Run("reject geography source for identity partition transform", func(t *testing.T) {
+		geog, err := iceberg.GeographyTypeOf("srid:4269", iceberg.EdgeAlgorithmKarney)
+		assert.NoError(t, err)
+
+		geoSchema := iceberg.NewSchema(1,
+			iceberg.NestedField{ID: 1, Name: "id", Required: true, Type: iceberg.PrimitiveTypes.Int64},
+			iceberg.NestedField{ID: 2, Name: "geog", Required: false, Type: geog},
+		)
+		metadata, err := table.NewMetadata(geoSchema, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, "", nil)
+		assert.NoError(t, err)
+
+		tbl := table.New([]string{"geo_geography"}, metadata, "", nil, nil)
+		specUpdate := table.NewUpdateSpec(tbl.NewTransaction(), true)
+
+		updates, reqs, err := specUpdate.
+			AddField("geog", iceberg.IdentityTransform{}, "geog_identity").
+			BuildUpdates()
+		assert.Error(t, err)
+		assert.ErrorContains(t, err, "cannot transform")
+		assert.ErrorContains(t, err, "geog")
+		assert.Nil(t, updates)
+		assert.Nil(t, reqs)
+	})
 }
 
 func TestUpdateSpecAddIdentityField(t *testing.T) {