You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2025/05/13 18:06:32 UTC
(iceberg-go) branch main updated: fix(table) getPartitionRecord to use field ID rather than field name (#427)
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 8a4b029 fix(table) getPartitionRecord to use field ID rather than field name (#427)
8a4b029 is described below
commit 8a4b029a18e74eef999c65267a0ed3688b23c34e
Author: loicalleyne <lo...@gmail.com>
AuthorDate: Tue May 13 14:06:28 2025 -0400
fix(table) getPartitionRecord to use field ID rather than field name (#427)
supersedes [#412](https://github.com/apache/iceberg-go/pull/412)
Changes:
- `getPartitionRecord` uses field ID rather than field Name to look up
partition (addresses issue
[404](https://github.com/apache/iceberg-go/issues/404) where java
implementation is sanitizing name and causing mismatch)
- Breaking change - NewDataFileBuilder signature change
from:
```go
func NewDataFileBuilder(
spec PartitionSpec,
content ManifestEntryContent,
path string,
format FileFormat,
partitionData map[string]any,
recordCount int64,
fileSize int64,
) (*DataFileBuilder, error)
```
to:
```go
func NewDataFileBuilder(
spec PartitionSpec,
content ManifestEntryContent,
path string,
format FileFormat,
fieldIDToPartitionData map[int]any,
recordCount int64,
fileSize int64,
) (*DataFileBuilder, error)
```
---
manifest.go | 106 +++++++++++++++++++++++----------------
manifest_test.go | 4 +-
table/evaluators_test.go | 4 +-
table/internal/utils.go | 8 +--
table/scanner.go | 2 +-
table/snapshots_internal_test.go | 4 +-
table/table_test.go | 4 +-
7 files changed, 75 insertions(+), 57 deletions(-)
diff --git a/manifest.go b/manifest.go
index e741734..eab8acf 100644
--- a/manifest.go
+++ b/manifest.go
@@ -428,7 +428,6 @@ func getFieldIDMap(sc avro.Schema) (map[string]int, map[int]avro.LogicalType) {
avroTyp = t
}
}
-
if ps, ok := avroTyp.(*avro.PrimitiveSchema); ok && ps.Logical() != nil {
logicalTypes[int(fid)] = ps.Logical().Type()
}
@@ -959,7 +958,7 @@ func (p *partitionFieldStats[T]) update(value any) {
}
}
-func constructPartitionSummaries(spec PartitionSpec, schema *Schema, partitions []map[string]any) ([]FieldSummary, error) {
+func constructPartitionSummaries(spec PartitionSpec, schema *Schema, partitions []map[int]any) ([]FieldSummary, error) {
partType := spec.PartitionType(schema)
fieldStats := make([]fieldStats, len(partType.FieldList))
for i, field := range partType.FieldList {
@@ -973,7 +972,7 @@ func constructPartitionSummaries(spec PartitionSpec, schema *Schema, partitions
for _, part := range partitions {
for i, field := range partType.FieldList {
- fieldStats[i].update(part[field.Name])
+ fieldStats[i].update(part[field.ID])
}
}
@@ -1004,7 +1003,7 @@ type ManifestWriter struct {
deletedFiles int32
deletedRows int64
- partitions []map[string]any
+ partitions []map[int]any
minSeqNum int64
reusedEntry manifestEntry
}
@@ -1039,7 +1038,7 @@ func NewManifestWriter(version int, out io.Writer, spec PartitionSpec, schema *S
schema: schema,
snapshotID: snapshotID,
minSeqNum: -1,
- partitions: make([]map[string]any, 0),
+ partitions: make([]map[int]any, 0),
}
md, err := w.meta()
@@ -1451,28 +1450,26 @@ func mapToAvroColMap[K comparable, V any](m map[K]V) *[]colMap[K, V] {
return &out
}
-func avroPartitionData(input map[string]any, nameToID map[string]int, logicalTypes map[int]avro.LogicalType) map[string]any {
- out := make(map[string]any)
+func avroPartitionData(input map[int]any, logicalTypes map[int]avro.LogicalType) map[int]any {
+ out := make(map[int]any)
for k, v := range input {
- if id, ok := nameToID[k]; ok {
- if logical, ok := logicalTypes[id]; ok {
- switch logical {
- case avro.Date:
- out[k] = Date(v.(time.Time).Truncate(24*time.Hour).Unix() / int64((time.Hour * 24).Seconds()))
- case avro.TimeMillis:
- out[k] = Time(v.(time.Duration).Milliseconds())
- case avro.TimeMicros:
- out[k] = Time(v.(time.Duration).Microseconds())
- case avro.TimestampMillis:
- out[k] = Timestamp(v.(time.Time).UTC().UnixMilli())
- case avro.TimestampMicros:
- out[k] = Timestamp(v.(time.Time).UTC().UnixMicro())
- default:
- out[k] = v
- }
-
- continue
+ if logical, ok := logicalTypes[k]; ok {
+ switch logical {
+ case avro.Date:
+ out[k] = Date(v.(time.Time).Truncate(24*time.Hour).Unix() / int64((time.Hour * 24).Seconds()))
+ case avro.TimeMillis:
+ out[k] = Time(v.(time.Duration).Milliseconds())
+ case avro.TimeMicros:
+ out[k] = Time(v.(time.Duration).Microseconds())
+ case avro.TimestampMillis:
+ out[k] = Timestamp(v.(time.Time).UTC().UnixMilli())
+ case avro.TimestampMicros:
+ out[k] = Timestamp(v.(time.Time).UTC().UnixMicro())
+ default:
+ out[k] = v
}
+
+ continue
}
out[k] = v
}
@@ -1508,11 +1505,10 @@ type dataFile struct {
lowerBoundMap map[int][]byte
upperBoundMap map[int][]byte
- // not used for anything yet, but important to maintain the information
- // for future development and updates such as when we get to writes,
- // and scan planning
- fieldNameToID map[string]int
- fieldIDToLogicalType map[int]avro.LogicalType
+ // used for partition retrieval
+ fieldNameToID map[string]int
+ fieldIDToLogicalType map[int]avro.LogicalType
+ fieldIDToPartitionData map[int]any
specID int32
initMaps sync.Once
@@ -1527,7 +1523,16 @@ func (d *dataFile) initializeMapData() {
d.distinctCntMap = avroColMapToMap(d.DistinctCounts)
d.lowerBoundMap = avroColMapToMap(d.LowerBounds)
d.upperBoundMap = avroColMapToMap(d.UpperBounds)
- d.PartitionData = avroPartitionData(d.PartitionData, d.fieldNameToID, d.fieldIDToLogicalType)
+ // Populate fieldIDToPartition map if dataFile read from manifest file
+ if len(d.fieldIDToPartitionData) < len(d.PartitionData) {
+ d.fieldIDToPartitionData = make(map[int]any, len(d.PartitionData))
+ for k, v := range d.PartitionData {
+ if id, ok := d.fieldNameToID[k]; ok {
+ d.fieldIDToPartitionData[id] = v
+ }
+ }
+ }
+ d.fieldIDToPartitionData = avroPartitionData(d.fieldIDToPartitionData, d.fieldIDToLogicalType)
})
}
@@ -1539,10 +1544,12 @@ func (d *dataFile) setFieldIDToLogicalTypeMap(m map[int]avro.LogicalType) {
func (d *dataFile) ContentType() ManifestEntryContent { return d.Content }
func (d *dataFile) FilePath() string { return d.Path }
func (d *dataFile) FileFormat() FileFormat { return d.Format }
-func (d *dataFile) Partition() map[string]any {
+
+// Partition returns the partition data as a map of partition field ID to value.
+func (d *dataFile) Partition() map[int]any {
d.initializeMapData()
- return d.PartitionData
+ return d.fieldIDToPartitionData
}
func (d *dataFile) Count() int64 { return d.RecordCount }
@@ -1747,7 +1754,7 @@ func NewDataFileBuilder(
content ManifestEntryContent,
path string,
format FileFormat,
- partitionData map[string]any,
+ fieldIDToPartitionData map[int]any,
recordCount int64,
fileSize int64,
) (*DataFileBuilder, error) {
@@ -1776,16 +1783,26 @@ func NewDataFileBuilder(
if fileSize <= 0 {
return nil, fmt.Errorf("%w: file size must be greater than 0", ErrInvalidArgument)
}
+ partitionData := make(map[string]any)
+ fieldNameToID := make(map[string]int)
+ for _, p := range spec.fields {
+ if pData, ok := fieldIDToPartitionData[p.FieldID]; ok {
+ partitionData[p.Name] = pData
+ fieldNameToID[p.Name] = p.FieldID
+ }
+ }
return &DataFileBuilder{
d: &dataFile{
- Content: content,
- Path: path,
- Format: format,
- PartitionData: partitionData,
- RecordCount: recordCount,
- FileSize: fileSize,
- specID: int32(spec.id),
+ Content: content,
+ Path: path,
+ Format: format,
+ PartitionData: partitionData,
+ RecordCount: recordCount,
+ FileSize: fileSize,
+ specID: int32(spec.id),
+ fieldIDToPartitionData: fieldIDToPartitionData,
+ fieldNameToID: fieldNameToID,
},
}, nil
}
@@ -1889,10 +1906,11 @@ type DataFile interface {
FilePath() string
// FileFormat is the format of the data file, AVRO, Orc, or Parquet.
FileFormat() FileFormat
- // Partition returns a mapping of field name to partition value for
+ // Partition returns a mapping of field id to partition value for
// each of the partition spec's fields.
- Partition() map[string]any
- // Count returns the number of records in this file.
+ Partition() map[int]any
+ // PartitionFieldData returns a mapping of field id to partition value
+ // for each of the partition spec's fields.
Count() int64
// FileSizeBytes is the total file size in bytes.
FileSizeBytes() int64
diff --git a/manifest_test.go b/manifest_test.go
index 66bc07f..4957877 100644
--- a/manifest_test.go
+++ b/manifest_test.go
@@ -810,7 +810,7 @@ func (m *ManifestTestSuite) TestReadManifestIncompleteSchema() {
EntryContentData,
"s3://bucket/namespace/table/data/abcd-0123.parquet",
ParquetFile,
- map[string]any{},
+ map[int]any{},
100,
100*1000*1000,
)
@@ -1074,7 +1074,7 @@ func (m *ManifestTestSuite) TestManifestEntryBuilder() {
EntryContentData,
"sample.parquet",
ParquetFile,
- map[string]any{"int": int(1), "datetime": time.Unix(1925, 0).UnixMicro()},
+ map[int]any{1001: int(1), 1002: time.Unix(1925, 0).UnixMicro()},
1,
2,
)
diff --git a/table/evaluators_test.go b/table/evaluators_test.go
index 13bbf45..c363f38 100644
--- a/table/evaluators_test.go
+++ b/table/evaluators_test.go
@@ -1072,7 +1072,7 @@ func (p *ProjectionTestSuite) TestPartialProjectedFields() {
type mockDataFile struct {
path string
format iceberg.FileFormat
- partition map[string]any
+ partition map[int]any
count int64
columnSizes map[int]int64
filesize int64
@@ -1088,7 +1088,7 @@ type mockDataFile struct {
func (*mockDataFile) ContentType() iceberg.ManifestEntryContent { return iceberg.EntryContentData }
func (m *mockDataFile) FilePath() string { return m.path }
func (m *mockDataFile) FileFormat() iceberg.FileFormat { return m.format }
-func (m *mockDataFile) Partition() map[string]any { return m.partition }
+func (m *mockDataFile) Partition() map[int]any { return m.partition }
func (m *mockDataFile) Count() int64 { return m.count }
func (m *mockDataFile) FileSizeBytes() int64 { return m.filesize }
func (m *mockDataFile) ColumnSizes() map[int]int64 { return m.columnSizes }
diff --git a/table/internal/utils.go b/table/internal/utils.go
index a5eeb86..eebe40c 100644
--- a/table/internal/utils.go
+++ b/table/internal/utils.go
@@ -231,19 +231,19 @@ func (d *DataFileStatistics) PartitionValue(field iceberg.PartitionField, sc *ic
}
func (d *DataFileStatistics) ToDataFile(schema *iceberg.Schema, spec iceberg.PartitionSpec, path string, format iceberg.FileFormat, filesize int64) iceberg.DataFile {
- var partitionData map[string]any
+ var fieldIDToPartitionData map[int]any
if !spec.Equals(*iceberg.UnpartitionedSpec) {
- partitionData = make(map[string]any)
+ fieldIDToPartitionData = make(map[int]any)
for field := range spec.Fields() {
val := d.PartitionValue(field, schema)
if val != nil {
- partitionData[field.Name] = val
+ fieldIDToPartitionData[field.FieldID] = val
}
}
}
bldr, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData,
- path, format, partitionData, d.RecordCount, filesize)
+ path, format, fieldIDToPartitionData, d.RecordCount, filesize)
if err != nil {
panic(err)
}
diff --git a/table/scanner.go b/table/scanner.go
index c0eaf93..14a3811 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -118,7 +118,7 @@ func getPartitionRecord(dataFile iceberg.DataFile, partitionType *iceberg.Struct
out := make(partitionRecord, len(partitionType.FieldList))
for i, f := range partitionType.FieldList {
- out[i] = partitionData[f.Name]
+ out[i] = partitionData[f.ID]
}
return out
diff --git a/table/snapshots_internal_test.go b/table/snapshots_internal_test.go
index 0838abd..d0e1581 100644
--- a/table/snapshots_internal_test.go
+++ b/table/snapshots_internal_test.go
@@ -64,11 +64,11 @@ func TestSnapshotSummaryCollectorWithPartition(t *testing.T) {
dataFile1 := must(iceberg.NewDataFileBuilder(
spec, iceberg.EntryContentData, "/path/to/file1.parquet",
- iceberg.ParquetFile, map[string]any{"int_field": int32(1)}, 100, 1234)).Build()
+ iceberg.ParquetFile, map[int]any{1001: int32(1)}, 100, 1234)).Build()
dataFile2 := must(iceberg.NewDataFileBuilder(
spec, iceberg.EntryContentData, "/path/to/file2.parquet",
- iceberg.ParquetFile, map[string]any{"int_field": int32(2)}, 200, 4321)).Build()
+ iceberg.ParquetFile, map[int]any{1001: int32(2)}, 200, 4321)).Build()
ssc.addFile(dataFile1, sc, spec)
ssc.removeFile(dataFile1, sc, spec)
diff --git a/table/table_test.go b/table/table_test.go
index 108493b..9bd2f13 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -485,8 +485,8 @@ func (t *TableWritingTestSuite) TestAddFilesPartitionedTable() {
t.Require().NoError(err)
for _, e := range entries {
- t.Equal(map[string]any{
- "baz": 123, "qux_month": 650,
+ t.Equal(map[int]any{
+ 1000: 123, 1001: 650,
}, e.DataFile().Partition())
}
}