laskoviymishka commented on code in PR #2029:
URL: https://github.com/apache/iceberg-go/pull/2029#discussion_r4064199682
##########
data_file_codec.go:
##########
@@ -333,3 +334,84 @@ 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
+ var buf [20]byte
+ writeInt := func(value int) {
Review Comment:
This closure does exactly what `writePartitionFingerprintInt` does a few
lines down, just with a reused buffer instead of a fresh one. Both are
stack-local and the standalone one inlines, so there's no perf difference.
I'd drop the closure and call `writePartitionFingerprintInt` here too, so
the format lives in one place if the separator or base ever changes.
##########
data_file_codec.go:
##########
@@ -333,3 +334,84 @@ 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
+ var buf [20]byte
+ writeInt := func(value int) {
+ key.Write(strconv.AppendInt(buf[:0], int64(value), 10))
+ key.WriteByte(':')
+ }
+ 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)
+ 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.
+ writeInt(field.FieldID)
+ writeInt(len(field.Name))
+ key.WriteString(field.Name)
+ if err := writePartitionTypeFingerprint(&key, resultType); err
!= nil {
+ return "", err
+ }
+ }
+
+ return key.String(), nil
+}
+
+func writePartitionTypeFingerprint(key *strings.Builder, typ Type) error {
Review Comment:
I think there's a maintenance trap here worth closing before we merge. This
switch and `partitionTypeToAvroSchema` now have to enumerate the exact same
type set, but neither references the other and nothing fails if they drift.
The concrete failure mode: someone adds `timestamp_ns` support to
`partitionTypeToAvroSchema` (those are valid V3 primitives, and the reject test
already lists `TimestampNsType`/`TimestampTzNsType` as the ones we don't handle
yet). If they don't also touch this switch, the `default` fires and
`manifestEntrySchemaFor` starts returning "unsupported partition type" for
every encode/decode on those tables. That's a regression the old single-path
design couldn't produce.
I'd add a test that feeds each type `partitionTypeToAvroSchema` accepts
through `writePartitionTypeFingerprint` and asserts no error, so the two
switches fail loudly at test time the moment they fall out of sync. A short
comment on each function naming the other as its counterpart would help too.
wdyt?
##########
data_file_codec.go:
##########
@@ -333,3 +334,84 @@ 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
+ var buf [20]byte
+ writeInt := func(value int) {
+ key.Write(strconv.AppendInt(buf[:0], int64(value), 10))
+ key.WriteByte(':')
+ }
+ 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)
+ key.Grow(len(field.Name) + 32)
Review Comment:
The 32 here is a tight-but-fine budget (fieldID up to 7, name-length prefix
up to 4, tag 1, decimal params up to 14 all fit, and Builder grows anyway if it
doesn't). A named const or a one-line comment on what the 32 covers would save
the next reader the arithmetic.
##########
data_file_codec.go:
##########
@@ -333,3 +334,84 @@ 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
+ var buf [20]byte
+ writeInt := func(value int) {
+ key.Write(strconv.AppendInt(buf[:0], int64(value), 10))
+ key.WriteByte(':')
+ }
+ 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)
+ 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.
+ writeInt(field.FieldID)
+ writeInt(len(field.Name))
+ key.WriteString(field.Name)
+ if err := writePartitionTypeFingerprint(&key, resultType); err
!= nil {
+ return "", err
+ }
+ }
+
+ return key.String(), nil
+}
+
+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())
Review Comment:
`%s` on `typ.String()` with no `%w` is a pure format call, which perfsprint
will flag if it's in the lint config. `errors.New("unsupported partition type:
" + typ.String())` sidesteps it.
One catch: `partitionTypeToAvroSchema` uses the same `fmt.Errorf` shape and
`TestManifestEntrySchemaForRejectsInvalidTypesAfterCacheHit` asserts the two
error strings are `EqualError`, so if you change this one you'll want to change
that one too or the test breaks. Worth checking whether perfsprint is actually
on before bothering.
##########
data_file_codec_test.go:
##########
@@ -343,3 +345,166 @@ func BenchmarkMarshalAvroEntry(b *testing.B) {
})
}
}
+
+func TestManifestEntrySchemaForMatchesPartitionAvroShape(t *testing.T) {
Review Comment:
Every case here is a single-field spec, so this proves per-type correctness
but never exercises multi-field ordering or the delimiter logic that keeps
adjacent fields from bleeding into each other.
Could we add one case pairing two parameterized types (say Decimal(10,2)
then Fixed(16)) and assert the returned schema matches the directly computed
one? That doubles as a regression test for the length and colon delimiters,
which is exactly where a subtle off-by-one would hide.
##########
data_file_codec_test.go:
##########
@@ -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)
Review Comment:
For i<2 this compares `results[i]` against itself, so the equality check is
a tautology for the first two workers. The real signal is the NotEqual below
plus the even/odd grouping for i>=2.
I'd compare against the schema string computed directly from
`schemas[i%len(schemas)]` (a known-good value) so every index gets a
non-vacuous check.
##########
data_file_codec_test.go:
##########
@@ -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) {
Review Comment:
With the new code the fingerprint returns the error before the cache is ever
queried, so "AfterCacheHit" doesn't describe what this checks anymore.
Something like `TestManifestEntrySchemaForRejectsUnsupportedPartitionTypes`
would read truer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
