zeroshade commented on code in PR #2539:
URL: https://github.com/apache/iceberg-go/pull/2539#discussion_r4542965631
##########
table/partition_statistics_write.go:
##########
@@ -620,9 +814,25 @@ func writePartitionStatisticsParquet(
bldr.Field(2).(*array.Int64Builder).Append(row.dataRecordCount)
bldr.Field(3).(*array.Int32Builder).Append(row.dataFileCount)
bldr.Field(4).(*array.Int64Builder).Append(row.dataFileSize)
- // v1 has no delete files, so the exact live record
count equals
- // the sum of data-file record counts.
-
bldr.Field(5).(*array.Int64Builder).Append(row.dataRecordCount)
+ if formatVersion == 1 {
+
bldr.Field(5).(*array.Int64Builder).Append(row.dataRecordCount)
+ } else {
+
bldr.Field(5).(*array.Int64Builder).Append(row.positionDeleteCount)
+
bldr.Field(6).(*array.Int32Builder).Append(row.positionDeleteFiles)
+
bldr.Field(7).(*array.Int64Builder).Append(row.equalityDeleteCount)
+
bldr.Field(8).(*array.Int32Builder).Append(row.equalityDeleteFiles)
+ total := bldr.Field(9).(*array.Int64Builder)
+ if row.totalRecordCount == nil {
+ total.AppendNull()
+ } else {
+ total.Append(*row.totalRecordCount)
+ }
+
bldr.Field(10).(*array.Int64Builder).AppendNull()
Review Comment:
Java fills `last_updated_at`/`last_updated_snapshot_id` here.
[`PartitionStatsHandler.liveEntry`](https://github.com/apache/iceberg/blob/014b3ce5ffbf7c1c51dabb144f4d94e45addcd88/core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java#L355-L397)
calls `updateSnapshotInfo` for every live entry, keeping the newest timestamp
and its snapshot ID per partition. Engines use these fields to work out which
partitions changed since a given point, so files written by iceberg-go would be
less useful to them.
The data is already available here: each live entry has
`entry.SnapshotID()`, and its timestamp comes from the table metadata's
snapshot list. Could you track the max (timestamp, snapshot ID) per row in
`collectPartitionStatistics`, leave it NULL when the adding snapshot has
expired from metadata, and write it in fields 11 and 12?
##########
table/partition_statistics_write.go:
##########
@@ -363,21 +416,157 @@ func collectPartitionStatistics(
rows = append(rows, row)
tree.insert(newPartitionRecord(partition,
partitionType), row)
}
- if row.dataFileCount == math.MaxInt32 ||
- row.dataRecordCount >
math.MaxInt64-file.Count() ||
- row.dataFileSize >
math.MaxInt64-file.FileSizeBytes() {
- return nil, fmt.Errorf("%w: partition
statistics count overflow",
- iceberg.ErrInvalidArgument)
+ if err := accumulatePartitionStatistics(row, file); err
!= nil {
+ return nil, err
+ }
+ switch file.ContentType() {
+ case iceberg.EntryContentData:
+ if formatVersion == 3 {
+ dataFiles[path] =
partitionStatisticsDataFile{row: row, count: file.Count(), sequence:
entry.DataSequenceNumber()}
+ }
+ case iceberg.EntryContentPosDeletes:
+ if vector {
+ target := *file.ReferencedDataFile()
+ if _, duplicate := vectors[target];
duplicate {
+ return nil, fmt.Errorf("%w:
multiple deletion vectors reference data file %q", iceberg.ErrInvalidArgument,
target)
+ }
+ vectors[target] =
partitionStatisticsVector{row: row, count: file.Count(), sequence:
entry.DataSequenceNumber()}
+ } else {
+ exactTotals = false
+ }
+ case iceberg.EntryContentEqDeletes:
+ exactTotals = false
}
- row.dataFileCount++
- row.dataRecordCount += file.Count()
- row.dataFileSize += file.FileSizeBytes()
}
}
+ if err := finishPartitionStatistics(ctx, rows, dataFiles, vectors,
exactTotals); err != nil {
+ return nil, err
+ }
+
return rows, nil
}
+type partitionStatisticsDataFile struct {
+ row *inspectPartitionAggregate
+ count int64
+ sequence int64
+}
+
+type partitionStatisticsVector = partitionStatisticsDataFile
+
+// Keep regular-file identity as a path. Only V3 vectors need blob keys and
+// container sizes, so V1/V2 collection does not retain a second file index.
+type partitionStatisticsFileIndex struct {
+ paths map[string]string
+ blobs map[deletionVectorBlobKey]string
+ containers map[string]int64
+}
+
+func (i *partitionStatisticsFileIndex) add(file iceberg.DataFile, manifest
string, formatVersion int) (bool, error) {
+ path := file.FilePath()
+ vector := IsDeletionVector(file)
+ if vector {
+ if formatVersion < 3 {
+ return false, fmt.Errorf("%w: deletion vector %q
requires format v3", iceberg.ErrInvalidArgument, path)
+ }
+ key, err := snapshotFileDiffKey(file)
+ if err != nil {
+ return false, fmt.Errorf("%w: invalid statistics
deletion vector: %v", iceberg.ErrInvalidArgument, err)
+ }
+ if key.offset > file.FileSizeBytes()-key.length {
+ return false, fmt.Errorf("%w: deletion vector %q is
outside its Puffin file", iceberg.ErrInvalidArgument, path)
+ }
+ if previous, found := i.blobs[key]; found {
+ return false, fmt.Errorf("%w: duplicate live deletion
vector %q in manifests %q and %q", iceberg.ErrInvalidArgument, path, previous,
manifest)
+ }
+ if _, found := i.paths[path]; found {
+ return false, fmt.Errorf("%w: incompatible live
references to file %q", iceberg.ErrInvalidArgument, path)
+ }
+ if size, found := i.containers[path]; found && size !=
file.FileSizeBytes() {
+ return false, fmt.Errorf("%w: inconsistent Puffin
container size for %q", iceberg.ErrInvalidArgument, path)
+ }
+ if i.blobs == nil {
+ i.blobs = make(map[deletionVectorBlobKey]string)
+ i.containers = make(map[string]int64)
+ }
+ i.blobs[key] = manifest
+ i.containers[path] = file.FileSizeBytes()
+ } else {
+ if previous, found := i.paths[path]; found {
+ kind := "delete file"
+ if file.ContentType() == iceberg.EntryContentData {
+ kind = "data file"
+ }
+
+ return false, fmt.Errorf("%w: duplicate live %s %q in
manifests %q and %q", iceberg.ErrInvalidArgument, kind, path, previous,
manifest)
+ }
+ if _, found := i.containers[path]; found {
+ return false, fmt.Errorf("%w: incompatible live
references to file %q", iceberg.ErrInvalidArgument, path)
+ }
+ i.paths[path] = manifest
+ }
+
+ return vector, nil
+}
+
+func accumulatePartitionStatistics(row *inspectPartitionAggregate, file
iceberg.DataFile) error {
+ var records *int64
+ var files *int32
+ switch file.ContentType() {
+ case iceberg.EntryContentData:
+ records, files = &row.dataRecordCount, &row.dataFileCount
+ if row.dataFileSize > math.MaxInt64-file.FileSizeBytes() {
+ return fmt.Errorf("%w: partition statistics size
overflow", iceberg.ErrInvalidArgument)
+ }
+ row.dataFileSize += file.FileSizeBytes()
+ case iceberg.EntryContentPosDeletes:
+ records, files = &row.positionDeleteCount,
&row.positionDeleteFiles
+ if IsDeletionVector(file) {
+ files = &row.deletionVectorCount
+ }
+ case iceberg.EntryContentEqDeletes:
+ records, files = &row.equalityDeleteCount,
&row.equalityDeleteFiles
+ default:
+ return fmt.Errorf("%w: unsupported statistics file content %d",
iceberg.ErrInvalidArgument, file.ContentType())
+ }
+ if *files == math.MaxInt32 || *records > math.MaxInt64-file.Count() {
+ return fmt.Errorf("%w: partition statistics count overflow",
iceberg.ErrInvalidArgument)
+ }
+ *files++
+ *records += file.Count()
+
+ return nil
+}
+
+func finishPartitionStatistics(ctx context.Context, rows
[]*inspectPartitionAggregate, dataFiles map[string]partitionStatisticsDataFile,
vectors map[string]partitionStatisticsVector, exactTotals bool) error {
+ for _, row := range rows {
+ row.totalRecordCount = nil
+ if exactTotals {
+ row.totalRecordCount = new(row.dataRecordCount)
+ }
+ }
+ for target, vector := range vectors {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ data, found := dataFiles[target]
+ if !found || vector.row != data.row {
+ return fmt.Errorf("%w: deletion vector target %q must
be live with the same spec and partition", iceberg.ErrInvalidArgument, target)
Review Comment:
This turns any DV inconsistency into an `ErrInvalidArgument` for the whole
table: no live target, a target in another partition or spec, `count >
data.count`, or a data sequence number newer than the DV's. Partition
statistics are advisory, and Java never fails on this because it doesn't
compute `total_record_count` at all. One stale DV, for example from a writer
that didn't clean up after removing a data file, would block statistics for
every partition.
Could these cases set the affected row's `totalRecordCount` to nil instead,
as you already do for unknown sequence numbers, and still write the file? The
DV still counts toward `dv_count` and `position_delete_record_count`, which
only describe the delete files themselves.
--
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]
