You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by at...@apache.org on 2026/09/23 17:48:49 UTC
(iceberg-go) branch main updated: perf(codec): avoid rebuilding partition schema on schema-cache hits (#2029)
This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 b46a92756 perf(codec): avoid rebuilding partition schema on schema-cache hits (#2029)
b46a92756 is described below
commit b46a9275634b2b8cb7384709c05ea498ef495711
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Wed Sep 23 19:48:43 2026 +0200
perf(codec): avoid rebuilding partition schema on schema-cache hits (#2029)
* perf(codec): avoid rebuilding partition schema on cache hits
Signed-off-by: Minh Vu <vu...@gmail.com>
* perf(codec): avoid cloning partition source types
Signed-off-by: Minh Vu <vu...@gmail.com>
* perf(codec): compact partition schema fingerprint types
Signed-off-by: Minh Vu <vu...@gmail.com>
* test(codec): split schema cache benchmarks
Signed-off-by: Minh Vu <vu...@gmail.com>
* test(codec): guard partition fingerprint compatibility
Cross-check fingerprint type support against Avro conversion, cover
multi-field parameterized schemas and name boundaries, and share the
integer encoder.
Signed-off-by: Minh Vu <vu...@gmail.com>
---------
Signed-off-by: Minh Vu <vu...@gmail.com>
---
data_file_codec.go | 100 +++++++++++++++++++---
data_file_codec_bench_test.go | 101 ++++++++++++++++++++++
data_file_codec_fingerprint_test.go | 141 ++++++++++++++++++++++++++++++
data_file_codec_test.go | 165 ++++++++++++++++++++++++++++++++++++
4 files changed, 497 insertions(+), 10 deletions(-)
diff --git a/data_file_codec.go b/data_file_codec.go
index 894996f74..41445834f 100644
--- a/data_file_codec.go
+++ b/data_file_codec.go
@@ -20,6 +20,8 @@ package iceberg
import (
"fmt"
"reflect"
+ "strconv"
+ "strings"
"github.com/apache/iceberg-go/internal"
"github.com/apache/iceberg-go/internal/datafileavro"
@@ -261,13 +263,9 @@ type dataFileFieldMaps struct {
unknownFieldIDs map[int]struct{}
}
-// dataFileSchemaCacheKey identifies a cached avro schema by the
-// structural fingerprint of the partition Avro shape and the format
-// version. The fingerprint is taken from the avro schema produced by
-// [partitionTypeToAvroSchema] rather than [StructType.String]: the
-// avro shape ignores doc strings and other metadata that don't change
-// the wire format, so structurally identical specs that differ only
-// in documentation share a single cache entry.
+// dataFileSchemaCacheKey identifies a cached Avro schema by the ordered
+// partition field IDs, names, result types, and format version. These determine
+// the partition Avro shape without including docs or table-local schema/spec IDs.
type dataFileSchemaCacheKey struct {
partAvroFingerprint string
version int
@@ -311,15 +309,18 @@ func SetSchemaCacheSize(size int) error {
// The cache key fingerprints the partition Avro shape, so specs that
// differ only in field documentation share a single entry.
func manifestEntrySchemaFor(spec PartitionSpec, schema *Schema, version int) (*avro.Schema, dataFileFieldMaps, error) {
- partType := spec.PartitionType(schema)
- partSchema, err := partitionTypeToAvroSchema(partType)
+ fingerprint, err := partitionSchemaFingerprint(spec, schema)
if err != nil {
return nil, dataFileFieldMaps{}, err
}
- key := dataFileSchemaCacheKey{partAvroFingerprint: partSchema.String(), version: version}
+ key := dataFileSchemaCacheKey{partAvroFingerprint: fingerprint, version: version}
if cached, ok := dataFileSchemaCache.Get(key); ok {
return cached.schema, cached.maps, nil
}
+ partSchema, err := partitionTypeToAvroSchema(spec.PartitionType(schema))
+ if err != nil {
+ return nil, dataFileFieldMaps{}, err
+ }
fullSchema, err := internal.NewManifestEntrySchema(partSchema, version)
if err != nil {
return nil, dataFileFieldMaps{}, err
@@ -333,3 +334,82 @@ func manifestEntrySchemaFor(spec PartitionSpec, schema *Schema, version int) (*a
return entry.schema, entry.maps, nil
}
+
+func partitionSchemaFingerprint(spec PartitionSpec, schema *Schema) (string, error) {
+ var key strings.Builder
+ for _, field := range spec.fields {
+ sourceType := Type(UnknownType{})
+ // ResultType only inspects the source type here; borrow it to avoid
+ // cloning nested source types on every schema-cache hit.
+ if sourceField, ok := schema.FindFieldByIDRef(field.SourceID(), internal.SchemaRef{}); ok {
+ sourceType = sourceField.Type
+ }
+ resultType := field.Transform.ResultType(sourceType)
+ // Reserve room for the ID, name length, type tag, and typical parameters.
+ key.Grow(len(field.Name) + 32)
+
+ // Length prefixes keep names unambiguous. The compact type tag also
+ // includes fixed lengths and decimal precision/scale without formatting
+ // a temporary type string on every cache hit.
+ writePartitionFingerprintInt(&key, field.FieldID)
+ writePartitionFingerprintInt(&key, len(field.Name))
+ key.WriteString(field.Name)
+ if err := writePartitionTypeFingerprint(&key, resultType); err != nil {
+ return "", err
+ }
+ }
+
+ return key.String(), nil
+}
+
+// Keep the supported types and parameters in sync with partitionTypeToAvroSchema.
+// TestPartitionTypeFingerprintMatchesAvroConversion checks both paths together.
+func writePartitionTypeFingerprint(key *strings.Builder, typ Type) error {
+ // These tags are internal to the cache key. Parameterized types append their
+ // parameters so equal Avro partition shapes still produce equal keys.
+ switch t := typ.(type) {
+ case Int32Type:
+ key.WriteByte('i')
+ case Int64Type:
+ key.WriteByte('j')
+ case Float32Type:
+ key.WriteByte('k')
+ case Float64Type:
+ key.WriteByte('d')
+ case StringType:
+ key.WriteByte('S')
+ case DateType:
+ key.WriteByte('D')
+ case TimeType:
+ key.WriteByte('T')
+ case TimestampType:
+ key.WriteByte('t')
+ case TimestampTzType:
+ key.WriteByte('z')
+ case UUIDType:
+ key.WriteByte('u')
+ case BooleanType:
+ key.WriteByte('b')
+ case BinaryType:
+ key.WriteByte('B')
+ case FixedType:
+ key.WriteByte('f')
+ writePartitionFingerprintInt(key, t.Len())
+ case DecimalType:
+ key.WriteByte('q')
+ writePartitionFingerprintInt(key, t.Precision())
+ writePartitionFingerprintInt(key, t.Scale())
+ case UnknownType:
+ key.WriteByte('U')
+ default:
+ return fmt.Errorf("unsupported partition type: %s", typ.String())
+ }
+
+ return nil
+}
+
+func writePartitionFingerprintInt(key *strings.Builder, value int) {
+ var buf [20]byte
+ key.Write(strconv.AppendInt(buf[:0], int64(value), 10))
+ key.WriteByte(':')
+}
diff --git a/data_file_codec_bench_test.go b/data_file_codec_bench_test.go
new file mode 100644
index 000000000..c812c48e0
--- /dev/null
+++ b/data_file_codec_bench_test.go
@@ -0,0 +1,101 @@
+// 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 iceberg
+
+import (
+ "strconv"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/decimal128"
+ "github.com/stretchr/testify/require"
+)
+
+func BenchmarkDataFileSchemaCache(b *testing.B) {
+ schema := NewSchema(1,
+ NestedField{ID: 1, Name: "id", Type: Int64Type{}},
+ NestedField{ID: 2, Name: "category", Type: StringType{}},
+ NestedField{ID: 3, Name: "ts", Type: TimestampType{}},
+ NestedField{ID: 4, Name: "price", Type: DecimalTypeOf(10, 2)},
+ NestedField{ID: 5, Name: "nested", Type: &StructType{FieldList: []NestedField{
+ {ID: 6, Name: "value", Type: StringType{}},
+ }}},
+ )
+ for _, tc := range []struct {
+ name string
+ spec PartitionSpec
+ partition map[int]any
+ }{
+ {"unpartitioned", NewPartitionSpec(), nil},
+ {"identity", NewPartitionSpec(PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform: IdentityTransform{},
+ }), map[int]any{1000: int64(42)}},
+ {"mixed_4", NewPartitionSpec(
+ PartitionField{SourceIDs: []int{1}, FieldID: 1000, Name: "bucket", Transform: BucketTransform{NumBuckets: 16}},
+ PartitionField{SourceIDs: []int{2}, FieldID: 1001, Name: "category", Transform: TruncateTransform{Width: 4}},
+ PartitionField{SourceIDs: []int{3}, FieldID: 1002, Name: "day", Transform: DayTransform{}},
+ PartitionField{SourceIDs: []int{4}, FieldID: 1003, Name: "price", Transform: IdentityTransform{}},
+ ), map[int]any{1000: int32(3), 1001: "east", 1002: Date(123), 1003: Decimal{Val: decimal128.FromI64(4212), Scale: 2}}},
+ {"nested_source", NewPartitionSpec(PartitionField{
+ SourceIDs: []int{5}, FieldID: 1000, Name: "bucket", Transform: BucketTransform{NumBuckets: 16},
+ }), map[int]any{1000: int32(3)}},
+ } {
+ for _, version := range []int{1, 2, 3} {
+ b.Run(tc.name+"/v"+strconv.Itoa(version), func(b *testing.B) {
+ builder, err := NewDataFileBuilder(tc.spec, EntryContentData,
+ "s3://bucket/table/data.parquet", ParquetFile, tc.partition, nil, nil, 1024, 1024*1024)
+ require.NoError(b, err)
+ builder.ColumnSizes(map[int]int64{1: 512, 2: 256}).
+ ValueCounts(map[int]int64{1: 1024, 2: 1024}).
+ NullValueCounts(map[int]int64{1: 0, 2: 4}).
+ LowerBoundValues(map[int][]byte{1: {0x01}, 2: []byte("a")}).
+ UpperBoundValues(map[int][]byte{1: {0xff}, 2: []byte("z")}).
+ SplitOffsets([]int64{0, 4096})
+ df := builder.Build().(*dataFile)
+ encoded, err := df.MarshalAvroEntry(tc.spec, schema, version)
+ require.NoError(b, err)
+ _, err = unmarshalAvroDataFileEntry(encoded, tc.spec, schema, version)
+ require.NoError(b, err)
+
+ b.Run("lookup", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ if _, _, err := manifestEntrySchemaFor(tc.spec, schema, version); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+ b.Run("marshal", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ if _, err := df.MarshalAvroEntry(tc.spec, schema, version); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+ b.Run("unmarshal", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ if _, err := unmarshalAvroDataFileEntry(encoded, tc.spec, schema, version); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+ })
+ }
+ }
+}
diff --git a/data_file_codec_fingerprint_test.go b/data_file_codec_fingerprint_test.go
new file mode 100644
index 000000000..204b4a956
--- /dev/null
+++ b/data_file_codec_fingerprint_test.go
@@ -0,0 +1,141 @@
+// 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 iceberg
+
+import (
+ "strconv"
+ "strings"
+ "testing"
+
+ "github.com/apache/iceberg-go/internal"
+ "github.com/stretchr/testify/require"
+)
+
+func TestPartitionTypeFingerprintMatchesAvroConversion(t *testing.T) {
+ // Include currently unsupported types as well: adding support in the Avro
+ // converter must also make the fingerprint path accept the same type.
+ types := []Type{
+ BooleanType{},
+ Int32Type{},
+ Int64Type{},
+ Float32Type{},
+ Float64Type{},
+ DateType{},
+ TimeType{},
+ TimestampType{},
+ TimestampTzType{},
+ TimestampNsType{},
+ TimestampTzNsType{},
+ StringType{},
+ UUIDType{},
+ BinaryType{},
+ UnknownType{},
+ VariantType{},
+ GeometryType{},
+ GeographyType{},
+ FixedTypeOf(8),
+ FixedTypeOf(16),
+ DecimalTypeOf(10, 2),
+ DecimalTypeOf(11, 2),
+ DecimalTypeOf(10, 3),
+ &StructType{FieldList: []NestedField{{ID: 2, Name: "child", Type: StringType{}}}},
+ &ListType{ElementID: 2, Element: StringType{}},
+ &MapType{KeyID: 2, KeyType: StringType{}, ValueID: 3, ValueType: Int64Type{}},
+ }
+ shapes := make(map[string]string, len(types))
+ for _, typ := range types {
+ t.Run(typ.String(), func(t *testing.T) {
+ var key strings.Builder
+ fingerprintErr := writePartitionTypeFingerprint(&key, typ)
+ partition, avroErr := partitionTypeToAvroSchema(&StructType{FieldList: []NestedField{
+ {ID: 1000, Name: "partition", Type: typ},
+ }})
+ if avroErr != nil {
+ require.EqualError(t, fingerprintErr, avroErr.Error())
+
+ return
+ }
+ require.NoError(t, fingerprintErr)
+ if previous, ok := shapes[key.String()]; ok {
+ require.Equal(t, previous, partition.String(), "fingerprint collision for %s", typ)
+ } else {
+ shapes[key.String()] = partition.String()
+ }
+ })
+ }
+}
+
+func TestManifestEntrySchemaForMultipleParameterizedFields(t *testing.T) {
+ price := PartitionField{SourceIDs: []int{1}, FieldID: 1000, Name: "price", Transform: IdentityTransform{}}
+ code := PartitionField{SourceIDs: []int{2}, FieldID: 1001, Name: "code", Transform: IdentityTransform{}}
+ for _, version := range []int{1, 2, 3} {
+ for _, tc := range []struct {
+ name string
+ spec PartitionSpec
+ decimal Type
+ fixed Type
+ }{
+ {"ordered", NewPartitionSpec(price, code), DecimalTypeOf(10, 2), FixedTypeOf(16)},
+ {"reversed", NewPartitionSpec(code, price), DecimalTypeOf(10, 2), FixedTypeOf(16)},
+ {"precision", NewPartitionSpec(price, code), DecimalTypeOf(11, 2), FixedTypeOf(16)},
+ {"scale", NewPartitionSpec(price, code), DecimalTypeOf(10, 3), FixedTypeOf(16)},
+ {"fixed_size", NewPartitionSpec(price, code), DecimalTypeOf(10, 2), FixedTypeOf(8)},
+ } {
+ t.Run(tc.name+"/v"+strconv.Itoa(version), func(t *testing.T) {
+ t.Parallel()
+
+ schema := NewSchema(1,
+ NestedField{ID: 1, Name: "source_price", Type: tc.decimal},
+ NestedField{ID: 2, Name: "source_code", Type: tc.fixed},
+ )
+ partition, err := partitionTypeToAvroSchema(tc.spec.PartitionType(schema))
+ require.NoError(t, err)
+ want, err := internal.NewManifestEntrySchema(partition, version)
+ require.NoError(t, err)
+ for range 2 {
+ got, maps, err := manifestEntrySchemaFor(tc.spec, schema, version)
+ require.NoError(t, err)
+ require.Equal(t, want.String(), got.String())
+ require.Equal(t, getFieldIDMap(want), maps)
+ }
+ })
+ }
+ }
+}
+
+func TestPartitionSchemaFingerprintNameBoundaries(t *testing.T) {
+ schema := NewSchema(1,
+ NestedField{ID: 1, Name: "source_price", Type: DecimalTypeOf(10, 2)},
+ NestedField{ID: 2, Name: "source_code", Type: FixedTypeOf(16)},
+ )
+ // Exercise the key's length prefixes independently of Avro name validation.
+ names := []string{"a", "ab", "1:a", "a:q10:2:", "f16:", "λ"}
+ keys := make(map[string][2]string, len(names)*len(names))
+ for _, first := range names {
+ for _, second := range names {
+ spec := NewPartitionSpec(
+ PartitionField{SourceIDs: []int{1}, FieldID: 1000, Name: first, Transform: IdentityTransform{}},
+ PartitionField{SourceIDs: []int{2}, FieldID: 1001, Name: second, Transform: IdentityTransform{}},
+ )
+ key, err := partitionSchemaFingerprint(spec, schema)
+ require.NoError(t, err)
+ require.NotContains(t, keys, key, "names %q/%q collide with %q", first, second, keys[key])
+ keys[key] = [2]string{first, second}
+ }
+ }
+}
diff --git a/data_file_codec_test.go b/data_file_codec_test.go
index 1c7606ca0..fc42558ee 100644
--- a/data_file_codec_test.go
+++ b/data_file_codec_test.go
@@ -20,9 +20,11 @@ package iceberg
import (
"reflect"
"strconv"
+ "sync"
"testing"
"github.com/apache/arrow-go/v18/arrow/decimal128"
+ "github.com/apache/iceberg-go/internal"
"github.com/stretchr/testify/require"
)
@@ -343,3 +345,166 @@ func BenchmarkMarshalAvroEntry(b *testing.B) {
})
}
}
+
+func TestManifestEntrySchemaForMatchesPartitionAvroShape(t *testing.T) {
+ types := []Type{
+ Int32Type{},
+ Int64Type{},
+ Float32Type{},
+ Float64Type{},
+ StringType{},
+ DateType{},
+ TimeType{},
+ TimestampType{},
+ TimestampTzType{},
+ UUIDType{},
+ BooleanType{},
+ BinaryType{},
+ FixedTypeOf(8), FixedTypeOf(16),
+ DecimalTypeOf(10, 2), DecimalTypeOf(11, 2), DecimalTypeOf(10, 3),
+ UnknownType{},
+ }
+ for _, version := range []int{1, 2, 3} {
+ for _, typ := range types {
+ t.Run("v"+strconv.Itoa(version)+"/"+typ.String(), func(t *testing.T) {
+ schema := NewSchema(1, NestedField{ID: 1, Name: "source", Type: typ})
+ spec := NewPartitionSpec(PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "partition", Transform: IdentityTransform{},
+ })
+ partition, err := partitionTypeToAvroSchema(spec.PartitionType(schema))
+ require.NoError(t, err)
+ want, err := internal.NewManifestEntrySchema(partition, version)
+ require.NoError(t, err)
+ for range 2 {
+ got, maps, err := manifestEntrySchemaFor(spec, schema, version)
+ require.NoError(t, err)
+ require.Equal(t, want.String(), got.String())
+ require.Equal(t, getFieldIDMap(want), maps)
+ }
+ })
+ }
+ }
+}
+
+func TestManifestEntrySchemaForPartitionShapeEquivalence(t *testing.T) {
+ schema := NewSchema(1,
+ NestedField{ID: 1, Name: "source", Type: Int32Type{}},
+ NestedField{ID: 2, Name: "ts", Type: TimestampType{}},
+ )
+ field := PartitionField{SourceIDs: []int{1}, FieldID: 1000, Name: "partition", Transform: IdentityTransform{}}
+ spec := NewPartitionSpec(field)
+ original, _, err := manifestEntrySchemaFor(spec, schema, 2)
+ require.NoError(t, err)
+
+ for _, tc := range []struct {
+ name string
+ spec PartitionSpec
+ schema *Schema
+ }{
+ {"spec_id", NewPartitionSpecID(99, field), schema},
+ {"schema_metadata", spec, NewSchema(99,
+ NestedField{
+ ID: 1, Name: "renamed_source", Type: Int32Type{}, Required: true,
+ Doc: "documentation", InitialDefault: int32(1), WriteDefault: int32(2),
+ },
+ NestedField{ID: 3, Name: "unrelated", Type: StringType{}},
+ )},
+ {"bucket", NewPartitionSpec(PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "partition", Transform: BucketTransform{NumBuckets: 8},
+ }), schema},
+ {"year", NewPartitionSpec(PartitionField{
+ SourceIDs: []int{2}, FieldID: 1000, Name: "partition", Transform: YearTransform{},
+ }), schema},
+ {"dropped_bucket_source", NewPartitionSpec(PartitionField{
+ SourceIDs: []int{3}, FieldID: 1000, Name: "partition", Transform: BucketTransform{NumBuckets: 16},
+ }), schema},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ got, _, err := manifestEntrySchemaFor(tc.spec, tc.schema, 2)
+ require.NoError(t, err)
+ require.Same(t, original, got)
+ })
+ }
+
+ for _, tc := range []struct {
+ name string
+ fields []PartitionField
+ version int
+ }{
+ {"field_id", []PartitionField{{SourceIDs: []int{1}, FieldID: 1001, Name: "partition", Transform: IdentityTransform{}}}, 2},
+ {"field_name", []PartitionField{{SourceIDs: []int{1}, FieldID: 1000, Name: "renamed", Transform: IdentityTransform{}}}, 2},
+ {"field_count", []PartitionField{field, {SourceIDs: []int{2}, FieldID: 1001, Name: "year", Transform: YearTransform{}}}, 2},
+ {"empty", nil, 2},
+ {"v1", []PartitionField{field}, 1},
+ {"v3", []PartitionField{field}, 3},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ got, _, err := manifestEntrySchemaFor(NewPartitionSpec(tc.fields...), schema, tc.version)
+ require.NoError(t, err)
+ require.NotSame(t, original, got)
+ })
+ }
+
+ other := PartitionField{SourceIDs: []int{2}, FieldID: 1001, Name: "year", Transform: YearTransform{}}
+ ordered, _, err := manifestEntrySchemaFor(NewPartitionSpec(field, other), schema, 2)
+ require.NoError(t, err)
+ reversed, _, err := manifestEntrySchemaFor(NewPartitionSpec(other, field), schema, 2)
+ require.NoError(t, err)
+ require.NotSame(t, ordered, reversed)
+}
+
+type unsupportedCodecPartitionType struct{ Int64Type }
+
+func TestManifestEntrySchemaForRejectsInvalidTypesAfterCacheHit(t *testing.T) {
+ spec := NewPartitionSpec(PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "partition", Transform: IdentityTransform{},
+ })
+ _, _, err := manifestEntrySchemaFor(spec, NewSchema(1, NestedField{ID: 1, Name: "source", Type: Int64Type{}}), 2)
+ require.NoError(t, err)
+
+ for _, typ := range []Type{
+ unsupportedCodecPartitionType{},
+ TimestampNsType{},
+ TimestampTzNsType{},
+ VariantType{},
+ &StructType{}, &ListType{ElementID: 2, Element: Int64Type{}},
+ &MapType{KeyID: 2, KeyType: StringType{}, ValueID: 3, ValueType: Int64Type{}},
+ } {
+ schema := NewSchema(1, NestedField{ID: 1, Name: "source", Type: typ})
+ _, wantErr := partitionTypeToAvroSchema(spec.PartitionType(schema))
+ require.Error(t, wantErr)
+ _, _, err := manifestEntrySchemaFor(spec, schema, 2)
+ require.EqualError(t, err, wantErr.Error())
+ }
+}
+
+func TestManifestEntrySchemaForConcurrentTableLocalIDs(t *testing.T) {
+ spec := NewPartitionSpec(PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "partition", Transform: IdentityTransform{},
+ })
+ schemas := []*Schema{
+ NewSchema(1, NestedField{ID: 1, Name: "source", Type: Int64Type{}}),
+ NewSchema(1, NestedField{ID: 1, Name: "source", Type: StringType{}}),
+ }
+ const workers = 32
+ results := make([]struct {
+ schema string
+ err error
+ }, workers)
+ var wg sync.WaitGroup
+ for i := range results {
+ wg.Go(func() {
+ got, _, err := manifestEntrySchemaFor(spec, schemas[i%len(schemas)], 2)
+ results[i].err = err
+ if err == nil {
+ results[i].schema = got.String()
+ }
+ })
+ }
+ wg.Wait()
+ for i, result := range results {
+ require.NoError(t, result.err)
+ require.Equal(t, results[i%len(schemas)].schema, result.schema)
+ }
+ require.NotEqual(t, results[0].schema, results[1].schema)
+}