This is an automated email from the ASF dual-hosted git repository.
zeroshade 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 2fc51d108 fix(manifest): normalize file_format casing when reading
manifest entries (#1984)
2fc51d108 is described below
commit 2fc51d108dd845ce79867b94c73acc7caa3a0e60
Author: Will Roever <[email protected]>
AuthorDate: Wed Oct 14 06:21:26 2026 -0700
fix(manifest): normalize file_format casing when reading manifest entries
(#1984)
* fix(manifest): normalize file_format casing when reading entries
Route the decoded file_format value through FileFormatFromString in
ReadEntry so spec-conformant lowercase spellings match the FileFormat
constants, and reject unknown formats there with the offending file path.
Signed-off-by: Will Roever <[email protected]>
* fix(manifest): normalize file_format in the data file codec
Signed-off-by: Will Roever <[email protected]>
* fix(manifest): drop redundant wrapping in the file format error
Signed-off-by: Will Roever <[email protected]>
* test(manifest): pin the file format error message and widen the codec
table
Signed-off-by: Will Roever <[email protected]>
---------
Signed-off-by: Will Roever <[email protected]>
---
data_file_codec.go | 3 ++
data_file_codec_test.go | 56 ++++++++++++++++++++++++++++++++
manifest.go | 20 +++++++++++-
manifest_test.go | 86 +++++++++++++++++++++++++++++++++++++++++++++++++
4 files changed, 164 insertions(+), 1 deletion(-)
diff --git a/data_file_codec.go b/data_file_codec.go
index 15c6c84ba..ea0b96ba6 100644
--- a/data_file_codec.go
+++ b/data_file_codec.go
@@ -146,6 +146,9 @@ func unmarshalAvroDataFileEntry(data []byte, spec
PartitionSpec, schema *Schema,
if _, err := s.Decode(data, entry); err != nil {
return nil, fmt.Errorf("iceberg: unmarshalAvroDataFileEntry:
%w", err)
}
+ if err := df.normalizeFormat(); err != nil {
+ return nil, fmt.Errorf("iceberg: unmarshalAvroDataFileEntry:
%w", err)
+ }
df.specID = int32(spec.ID())
df.fieldNameToID = maps.nameToID
df.fieldIDToLogicalType = maps.idToType
diff --git a/data_file_codec_test.go b/data_file_codec_test.go
index 6126e6975..0e35bcce7 100644
--- a/data_file_codec_test.go
+++ b/data_file_codec_test.go
@@ -18,6 +18,7 @@
package iceberg
import (
+ "fmt"
"reflect"
"strconv"
"sync"
@@ -124,6 +125,61 @@ func TestMarshalAvroEntryOmitsDistinctCounts(t *testing.T)
{
}
}
+func TestUnmarshalAvroDataFileEntryNormalizesFileFormat(t *testing.T) {
+ spec := NewPartitionSpec()
+ schema := NewSchema(0)
+
+ const path = "s3://bucket/table/data.parquet"
+
+ tests := []struct {
+ name string
+ content ManifestEntryContent
+ written FileFormat
+ expected FileFormat
+ expectErr bool
+ }{
+ {name: "spec lowercase", written: "parquet", expected:
ParquetFile},
+ {name: "mixed case", written: "Parquet", expected: ParquetFile},
+ {name: "uppercase", written: ParquetFile, expected:
ParquetFile},
+ {name: "lowercase avro", written: "avro", expected: AvroFile},
+ {name: "lowercase orc", written: "orc", expected: OrcFile},
+ {
+ name: "lowercase puffin in delete entry", content:
EntryContentPosDeletes,
+ written: "puffin", expected: PuffinFile,
+ },
+ {name: "unknown format", written: "csv", expectErr: true},
+ {name: "empty format", written: "", expectErr: true},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ content := tt.content
+ if content == 0 {
+ content = EntryContentData
+ }
+ builder, err := NewDataFileBuilder(spec, content,
+ path, ParquetFile, nil, nil, nil, 1, 1024)
+ require.NoError(t, err)
+ source := builder.Build().(*dataFile)
+ source.Format = tt.written
+
+ encoded, err := source.MarshalAvroEntry(spec, schema, 2)
+ require.NoError(t, err)
+
+ decoded, err := unmarshalAvroDataFileEntry(encoded,
spec, schema, 2)
+ if tt.expectErr {
+ require.EqualError(t, err, fmt.Sprintf(
+ "iceberg: unmarshalAvroDataFileEntry:
data file %q: unknown file format: %s",
+ path, tt.written))
+
+ return
+ }
+ require.NoError(t, err)
+ require.Equal(t, tt.expected, decoded.FileFormat())
+ })
+ }
+}
+
// TestManifestEntrySchemaForDistinguishesSameSpecIDDifferentPartitionTypes
// guards against the cache-key collision that would occur if the cache
// were keyed only by spec.ID(). Two tables can both use
diff --git a/manifest.go b/manifest.go
index e344b14a9..08cb21011 100644
--- a/manifest.go
+++ b/manifest.go
@@ -918,7 +918,11 @@ func (c *ManifestReader) ReadEntry() (ManifestEntry,
error) {
if c.isFallback {
tmp = tmp.(*fallbackManifestEntry).toEntry()
}
- tmp.DataFile().(*dataFile).projected = c.projected
+ df := tmp.DataFile().(*dataFile)
+ df.projected = c.projected
+ if err := df.normalizeFormat(); err != nil {
+ return nil, err
+ }
switch tmp.Status() {
case EntryStatusEXISTING, EntryStatusADDED, EntryStatusDELETED:
default:
@@ -2797,6 +2801,20 @@ func (d *dataFile) setFieldIDToDecimalScaleMap(m
map[int]int) {
d.fieldIDToDecimalScale = m
}
+// normalizeFormat sets d.Format to the FileFormat constant matching its
+// decoded spelling, or returns an error if it names no known format.
+// Every Avro decode path must call it before the value is compared or
+// exposed.
+func (d *dataFile) normalizeFormat() error {
+ format, err := FileFormatFromString(string(d.Format))
+ if err != nil {
+ return fmt.Errorf("data file %q: %w", d.FilePath(), err)
+ }
+ d.Format = format
+
+ return nil
+}
+
func (d *dataFile) ContentType() ManifestEntryContent { return d.Content }
func (d *dataFile) FilePath() string { return d.Path }
func (d *dataFile) FileFormat() FileFormat { return d.Format }
diff --git a/manifest_test.go b/manifest_test.go
index 12b4ad495..e139e5ef4 100644
--- a/manifest_test.go
+++ b/manifest_test.go
@@ -2427,6 +2427,92 @@ func (m *ManifestTestSuite)
TestManifestReaderRejectsInvalidEntries() {
}
}
+func (m *ManifestTestSuite) TestManifestReaderNormalizesFileFormat() {
+ partitionSpec := NewPartitionSpecID(1,
+ PartitionField{FieldID: 1000, SourceIDs: []int{1}, Name:
"VendorID", Transform: IdentityTransform{}},
+ PartitionField{FieldID: 1001, SourceIDs: []int{2}, Name:
"tpep_pickup_datetime", Transform: IdentityTransform{}})
+ partitionSchema, err :=
partitionTypeToAvroSchema(partitionSpec.PartitionType(testSchema))
+ m.Require().NoError(err)
+
+ tests := []struct {
+ name string
+ version int
+ content ManifestContent
+ entryContent ManifestEntryContent
+ written FileFormat
+ expected FileFormat
+ expectErr bool
+ }{
+ {name: "spec lowercase", written: "parquet", expected:
ParquetFile},
+ {name: "mixed case", written: "Parquet", expected: ParquetFile},
+ {name: "uppercase", written: ParquetFile, expected:
ParquetFile},
+ {name: "lowercase avro", written: "avro", expected: AvroFile},
+ {name: "lowercase orc", written: "orc", expected: OrcFile},
+ {
+ name: "lowercase puffin in delete manifest", content:
ManifestContentDeletes,
+ entryContent: EntryContentPosDeletes, written:
"puffin", expected: PuffinFile,
+ },
+ {name: "v1 fallback entry", version: 1, written: "parquet",
expected: ParquetFile},
+ {name: "unknown format", written: "csv", expectErr: true},
+ {name: "empty format", written: "", expectErr: true},
+ }
+
+ for _, tt := range tests {
+ m.Run(tt.name, func() {
+ version := tt.version
+ if version == 0 {
+ version = 2
+ }
+ content := tt.content
+ if content == 0 {
+ content = ManifestContentData
+ }
+ entrySchema, err :=
internal.NewManifestEntrySchema(partitionSchema, version)
+ m.Require().NoError(err)
+
+ records := manifestEntryV2Records
+ if version == 1 {
+ records = manifestEntryV1Records
+ }
+ entry := *records[0]
+ file := cloneDataFileAvroFields(entry.Data.(*dataFile))
+ file.Format = tt.written
+ file.Content = tt.entryContent
+ entry.Data = file
+
+ mw := ManifestWriter{version: version, spec:
partitionSpec, schema: testSchema, content: content}
+ metadata, err := mw.meta()
+ m.Require().NoError(err)
+
+ var buf bytes.Buffer
+ writer, err := ocf.NewWriter(&buf, entrySchema,
+ ocf.WithSchema(entrySchema.String()),
ocf.WithMetadata(metadata))
+ m.Require().NoError(err)
+ var toEncode any = &entry
+ if version == 1 {
+ toEncode =
&fallbackManifestEntry{manifestEntry: entry}
+ }
+ m.Require().NoError(writer.Encode(toEncode))
+ m.Require().NoError(writer.Close())
+
+ manifest := &manifestFile{version: version, SpecID: 1,
Content: content}
+ reader, err := NewManifestReader(manifest,
bytes.NewReader(buf.Bytes()))
+ m.Require().NoError(err)
+ defer reader.Close()
+
+ got, err := reader.ReadEntry()
+ if tt.expectErr {
+ m.Require().EqualError(err,
+ fmt.Sprintf("data file %q: unknown file
format: %s", file.Path, tt.written))
+
+ return
+ }
+ m.Require().NoError(err)
+ m.Equal(tt.expected, got.DataFile().FileFormat())
+ })
+ }
+}
+
func (m *ManifestTestSuite) TestManifestEntryBuilder() {
dataFileBuilder, err := NewDataFileBuilder(
NewPartitionSpec(),
