fix: Writing Avro for Spark (#435)

Turns out that the Java Iceberg implementation *requires* the Avro field
names to match exactly instead of only relying on the
field-ids/element-ids.

In this case, the issue is the Partition Field Summary element records.
It appears that all "struct" types in iceberg java are expected to be
named ["r" +
fieldID](https://github.com/apache/iceberg/blob/a7f3dc79a2f42a4875ac35eec2137ecff15204fc/core/src/main/java/org/apache/iceberg/avro/TypeToSchema.java#L100-L105)
and if you don't name it appropriately, Spark throws a very unhelpful
exception (`ArrayStoreException`).

This was causing anything best on Java Iceberg to fail when reading Avro
written by Iceberg-Go.

fixes #434

Author: zeroshade

go.mod

@@ -49,7 +49,6 @@ require (
 	github.com/uptrace/bun/extra/bundebug v1.2.11
 	gocloud.dev v0.41.0
 	golang.org/x/sync v0.14.0
-	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da
 	google.golang.org/api v0.233.0
 	gopkg.in/yaml.v3 v3.0.1
 )
@@ -306,6 +305,7 @@ require (
 	golang.org/x/text v0.25.0 // indirect
 	golang.org/x/time v0.11.0 // indirect
 	golang.org/x/tools v0.32.0 // indirect
+	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
 	google.golang.org/genproto v0.0.0-20250324211829-b45e905df463 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250324211829-b45e905df463 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250505200425-f936aa4a68b2 // indirect

internal/avro_schemas.go

@@ -92,7 +92,7 @@ func WithElementID(id int) avro.SchemaOption {
 }
 
 func init() {
-	AvroSchemaCache.Add("field_summary", Must(avro.NewRecordSchema("field_summary", "", []*avro.Field{
+	AvroSchemaCache.Add("field_summary", Must(avro.NewRecordSchema("r508", "", []*avro.Field{
 		Must(avro.NewField("contains_null",
 			BoolSchema,
 			avro.WithDoc("true if the field contains null values"),

internal/recipe/Dockerfile

@@ -20,6 +20,7 @@ RUN pip3 install pyiceberg[s3fs,hive]
 RUN pip3 install pyarrow
 
 COPY provision.py .
+COPY run_spark_count_sql.py .
 
 ENTRYPOINT ["./entrypoint.sh"]
 CMD ["notebook"]

internal/recipe/local_spark.go

@@ -25,42 +25,40 @@ import (
 	"os"
 	"testing"
 
-	"golang.org/x/xerrors"
-
 	"github.com/testcontainers/testcontainers-go/modules/compose"
 )
 
 //go:embed docker-compose.yml
 var composeFile []byte
 
-func Start(t *testing.T) error {
+func Start(t *testing.T) (*compose.DockerCompose, error) {
 	if _, ok := os.LookupEnv("AWS_S3_ENDPOINT"); ok {
-		return nil
+		return nil, nil
 	}
 	stack, err := compose.NewDockerComposeWith(
 		compose.WithStackReaders(bytes.NewBuffer(composeFile)),
 	)
 	if err != nil {
-		return xerrors.Errorf("fail to start compose: %w", err)
+		return nil, fmt.Errorf("fail to start compose: %w", err)
 	}
 	if err := stack.Up(t.Context()); err != nil {
-		return xerrors.Errorf("fail to up compose: %w", err)
+		return stack, fmt.Errorf("fail to up compose: %w", err)
 	}
 	spark, err := stack.ServiceContainer(t.Context(), "spark-iceberg")
 	if err != nil {
-		return xerrors.Errorf("fail to find spark-iceberg: %w", err)
+		return stack, fmt.Errorf("fail to find spark-iceberg: %w", err)
 	}
 	_, stdout, err := spark.Exec(t.Context(), []string{"ipython", "./provision.py"})
 	if err != nil {
-		return xerrors.Errorf("fail to seed provision.py: %w", err)
+		return stack, fmt.Errorf("fail to seed provision.py: %w", err)
 	}
 	data, err := io.ReadAll(stdout)
 	if err != nil {
-		return xerrors.Errorf("fail to read stdout: %w", err)
+		return stack, fmt.Errorf("fail to read stdout: %w", err)
 	}
 	fmt.Println(string(data))
 	t.Setenv("AWS_S3_ENDPOINT", "http://localhost:9000")
 	t.Setenv("AWS_REGION", "us-east-1")
 
-	return nil
+	return stack, nil
 }

internal/recipe/run_spark_count_sql.py

@@ -0,0 +1,22 @@
+# 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.
+
+from pyspark.sql import SparkSession
+
+spark = SparkSession.builder.getOrCreate()
+
+spark.sql("SELECT COUNT(*) FROM default.test_partitioned_by_days").show()

io/s3_integration_test.go

@@ -21,9 +21,10 @@ package io_test
 
 import (
 	"context"
-	"github.com/apache/iceberg-go/internal/recipe"
 	"testing"
 
+	"github.com/apache/iceberg-go/internal/recipe"
+
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/catalog"
 	sqlcat "github.com/apache/iceberg-go/catalog/sql"
@@ -33,7 +34,8 @@ import (
 )
 
 func TestMinioWarehouse(t *testing.T) {
-	require.NoError(t, recipe.Start(t))
+	_, err := recipe.Start(t)
+	require.NoError(t, err)
 
 	cat, err := catalog.Load(context.Background(), "default", iceberg.Properties{
 		"uri":                ":memory:",
@@ -64,7 +66,8 @@ func TestMinioWarehouse(t *testing.T) {
 }
 
 func TestMinioWarehouseNoLocation(t *testing.T) {
-	require.NoError(t, recipe.Start(t))
+	_, err := recipe.Start(t)
+	require.NoError(t, err)
 
 	cat, err := catalog.Load(context.Background(), "default", iceberg.Properties{
 		"uri":                ":memory:",

table/scanner_test.go

@@ -52,7 +52,8 @@ type ScannerSuite struct {
 }
 
 func (s *ScannerSuite) SetupSuite() {
-	require.NoError(s.T(), recipe.Start(s.T()))
+	_, err := recipe.Start(s.T())
+	require.NoError(s.T(), err)
 }
 
 func (s *ScannerSuite) SetupTest() {

table/transaction_test.go

@@ -0,0 +1,136 @@
+// 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.
+
+//go:build integration
+
+package table_test
+
+import (
+	"context"
+	"io"
+	"strings"
+	"testing"
+	"time"
+
+	"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"
+	"github.com/apache/arrow-go/v18/parquet/pqarrow"
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/catalog"
+	"github.com/apache/iceberg-go/catalog/rest"
+	"github.com/apache/iceberg-go/internal/recipe"
+	iceio "github.com/apache/iceberg-go/io"
+	"github.com/apache/iceberg-go/table"
+	"github.com/stretchr/testify/suite"
+	"github.com/testcontainers/testcontainers-go/modules/compose"
+)
+
+type SparkIntegrationTestSuite struct {
+	suite.Suite
+
+	ctx   context.Context
+	cat   catalog.Catalog
+	props iceberg.Properties
+	stack *compose.DockerCompose
+}
+
+func (s *SparkIntegrationTestSuite) SetupSuite() {
+	var err error
+	s.stack, err = recipe.Start(s.T())
+	s.Require().NoError(err)
+	if s.stack == nil {
+		s.T().Skip("skipping test, AWS_S3_ENDPOINT is set")
+	}
+}
+
+func (s *SparkIntegrationTestSuite) SetupTest() {
+	s.ctx = context.Background()
+	s.props = iceberg.Properties{
+		iceio.S3Region:          "us-east-1",
+		iceio.S3EndpointURL:     "http://localhost:9000",
+		iceio.S3AccessKeyID:     "admin",
+		iceio.S3SecretAccessKey: "password",
+	}
+
+	cat, err := rest.NewCatalog(s.ctx, "rest", "http://localhost:8181", rest.WithAdditionalProps(s.props))
+	s.Require().NoError(err)
+	s.cat = cat
+}
+
+func (s *SparkIntegrationTestSuite) TestAddFile() {
+	const filename = "s3://warehouse/default/test_partitioned_by_days/data/ts_day=2023-03-13/supertest.parquet"
+
+	tbl, err := s.cat.LoadTable(s.ctx, catalog.ToIdentifier("default", "test_partitioned_by_days"), nil)
+	s.Require().NoError(err)
+
+	sc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false)
+	if err != nil {
+		panic(err)
+	}
+
+	bldr := array.NewRecordBuilder(memory.DefaultAllocator, sc)
+	defer bldr.Release()
+
+	tm := time.Date(2023, 03, 13, 13, 22, 0, 0, time.UTC)
+	ts, _ := arrow.TimestampFromTime(tm, arrow.Microsecond)
+	bldr.Field(0).(*array.Date32Builder).Append(arrow.Date32FromTime(tm))
+	bldr.Field(1).(*array.TimestampBuilder).Append(ts)
+	bldr.Field(2).(*array.Int32Builder).Append(13)
+	bldr.Field(3).(*array.StringBuilder).Append("m")
+
+	rec := bldr.NewRecord()
+	defer rec.Release()
+
+	fw, err := tbl.FS().(iceio.WriteFileIO).Create(filename)
+	if err != nil {
+		panic(err)
+	}
+	defer fw.Close()
+
+	if err := pqarrow.WriteTable(array.NewTableFromRecords(sc, []arrow.Record{rec}), fw, rec.NumRows(), parquet.NewWriterProperties(), pqarrow.DefaultWriterProps()); err != nil {
+		panic(err)
+	}
+
+	tx := tbl.NewTransaction()
+	err = tx.AddFiles(s.ctx, []string{filename}, nil, false)
+	s.Require().NoError(err)
+
+	tbl, err = tx.Commit(s.ctx)
+	s.Require().NoError(err)
+
+	spark, err := s.stack.ServiceContainer(s.T().Context(), "spark-iceberg")
+	s.Require().NoError(err)
+
+	_, stdout, err := spark.Exec(s.ctx, []string{"ipython", "./run_spark_count_sql.py"})
+	s.Require().NoError(err)
+
+	output, err := io.ReadAll(stdout)
+	s.Require().NoError(err)
+	strings.HasSuffix(string(output), `
++--------+
+|count(1)|
++--------+
+|      13|
++--------+
+`)
+}
+
+func TestSparkIntegration(t *testing.T) {
+	suite.Run(t, new(SparkIntegrationTestSuite))
+}