tanmayrauth commented on code in PR #1392:
URL: https://github.com/apache/iceberg-go/pull/1392#discussion_r3525696458
##########
table/rewrite_data_files.go:
##########
@@ -300,38 +300,61 @@ func ExecuteCompactionGroup(ctx context.Context, tbl
*Table, group CompactionTas
scanOpts = append(scanOpts,
WitMaxConcurrency(cfg.scanConcurrency))
}
+ if tbl.metadata.Version() >= 3 {
+ lineageTasks, legacyTasks := splitTasksByLineage(group.Tasks)
+ if len(lineageTasks) > 0 && len(legacyTasks) > 0 {
+ slog.Info("compaction group has mixed row-lineage
tasks; splitting to preserve _row_id for lineage-capable files",
+ "partition_key", group.PartitionKey,
+ "lineage_tasks", len(lineageTasks),
+ "legacy_tasks", len(legacyTasks))
+
+ lineageGroup := CompactionTaskGroup{
+ PartitionKey: group.PartitionKey,
+ Tasks: lineageTasks,
+ TotalSizeBytes: sumTaskBytes(lineageTasks),
+ }
+ legacyGroup := CompactionTaskGroup{
+ PartitionKey: group.PartitionKey,
+ Tasks: legacyTasks,
+ TotalSizeBytes: sumTaskBytes(legacyTasks),
+ }
+
+ preserve, err := executeCompactionTaskGroup(ctx, tbl,
scanOpts, lineageGroup, cfg.targetFileSize, true)
+ if err != nil {
+ return CompactionGroupResult{}, err
+ }
+ legacy, err := executeCompactionTaskGroup(ctx, tbl,
scanOpts, legacyGroup, cfg.targetFileSize, false)
+ if err != nil {
+ return CompactionGroupResult{}, err
+ }
+
+ safePosDeletes :=
mergeTaskFiles(preserve.SafePosDeletes, legacy.SafePosDeletes)
+ safeDeletionVectors :=
mergeTaskFiles(preserve.SafeDeletionVectors, legacy.SafeDeletionVectors)
+
+ return CompactionGroupResult{
+ PartitionKey: group.PartitionKey,
+ OldDataFiles:
append(preserve.OldDataFiles, legacy.OldDataFiles...),
+ NewDataFiles:
append(preserve.NewDataFiles, legacy.NewDataFiles...),
+ SafePosDeletes: safePosDeletes,
+ SafeDeletionVectors: safeDeletionVectors,
+ BytesBefore: preserve.BytesBefore +
legacy.BytesBefore,
+ BytesAfter: preserve.BytesAfter +
legacy.BytesAfter,
+ }, nil
+ }
+ }
+
// Preserve row lineage only when every source file in the group carries
Review Comment:
This comment lost its tail when the old degradation block was removed — it
now ends mid-sentence at "carries". Worth completing or trimming it.
##########
table/rewrite_data_files.go:
##########
@@ -300,38 +300,61 @@ func ExecuteCompactionGroup(ctx context.Context, tbl
*Table, group CompactionTas
scanOpts = append(scanOpts,
WitMaxConcurrency(cfg.scanConcurrency))
}
+ if tbl.metadata.Version() >= 3 {
+ lineageTasks, legacyTasks := splitTasksByLineage(group.Tasks)
+ if len(lineageTasks) > 0 && len(legacyTasks) > 0 {
+ slog.Info("compaction group has mixed row-lineage
tasks; splitting to preserve _row_id for lineage-capable files",
+ "partition_key", group.PartitionKey,
+ "lineage_tasks", len(lineageTasks),
+ "legacy_tasks", len(legacyTasks))
+
+ lineageGroup := CompactionTaskGroup{
+ PartitionKey: group.PartitionKey,
+ Tasks: lineageTasks,
+ TotalSizeBytes: sumTaskBytes(lineageTasks),
+ }
+ legacyGroup := CompactionTaskGroup{
+ PartitionKey: group.PartitionKey,
+ Tasks: legacyTasks,
+ TotalSizeBytes: sumTaskBytes(legacyTasks),
+ }
+
+ preserve, err := executeCompactionTaskGroup(ctx, tbl,
scanOpts, lineageGroup, cfg.targetFileSize, true)
+ if err != nil {
+ return CompactionGroupResult{}, err
+ }
+ legacy, err := executeCompactionTaskGroup(ctx, tbl,
scanOpts, legacyGroup, cfg.targetFileSize, false)
+ if err != nil {
+ return CompactionGroupResult{}, err
+ }
+
+ safePosDeletes :=
mergeTaskFiles(preserve.SafePosDeletes, legacy.SafePosDeletes)
+ safeDeletionVectors :=
mergeTaskFiles(preserve.SafeDeletionVectors, legacy.SafeDeletionVectors)
Review Comment:
`mergeTaskFiles` dedups by `df.FilePath()`, which is the Puffin path — but a
single Puffin file can hold DV blobs for multiple data files, so DVs have to be
deduped by `ReferencedDataFile()`, not path. That's the invariant the rest of
the codebase already relies on (see `readAllDeletionVectors` keying by
referenced data file, and `deleteFileRemoved`/`removeDeletionVector` in
snapshot_producers.go using `deletedDVsByRef`).
Since the split sends lineage-capable and legacy data files into different
subgroups, a shared Puffin holding one DV blob for a lineage file and one for a
legacy file produces two entries with the same `FilePath()` but different
`ReferencedDataFile()`. This merge collapses them to one, so the dropped DV
never reaches `ApplyResult` — its data file gets rewritten while its DV is left
orphaned (stale delete referencing a removed file). This bites on tables whose
DVs were written by Java/Spark, which bundle blobs.
Pos-deletes are fine to dedup by path, so I'd suggest a separate DV merge
keyed by referenced data file (mirroring `CollectSafeDeletionVectors`) rather
than reusing the path-based helper for both. Could you also add a mixed-group
case where a single DV Puffin references one lineage and one legacy data file,
and assert both DVs are expunged?
--
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]
