laskoviymishka commented on code in PR #2288: URL: https://github.com/apache/iceberg-go/pull/2288#discussion_r4362317604
########## table/compaction/compaction_file_grain_test.go: ########## @@ -0,0 +1,90 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package compaction_test + +import ( + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/apache/iceberg-go/table/compaction" + "github.com/stretchr/testify/require" +) + +func TestPlanCompactionCountsWholeFiles(t *testing.T) { + file := newDataFile("split.parquet", 10) + firstRowID, sequence := int64(20), int64(3) + first := makeTask(file, 1, 1) + first.DeletionVectorFiles = []iceberg.DataFile{newScopedDeleteFile("delete.puffin", file.path, 1)} + first.FirstRowID, first.DataSequenceNumber = &firstRowID, &sequence + first.Residual = iceberg.AlwaysTrue{} + first.Length = file.size / 2 + second := first Review Comment: Could the two same-path tasks differ in more than the range? `second := first` copies the DV, `FirstRowID` and sequence number, and `require.Contains` compares with `reflect.DeepEqual`, so this passes whether the planner keeps the earlier task or the later one. Since `second` sits first in `tasks`, giving `first` its own `FirstRowID` after the copy would pin the [first-task rule](https://github.com/apache/iceberg-go/blob/06b53e07fd1b4e0d758c25cafcfb089d0f0f0ce2/table/compaction/compaction.go#L191-L192) the doc comment describes. ########## table/compaction/compaction.go: ########## @@ -188,16 +188,38 @@ type Plan struct { // PlanCompaction analyzes the given scan tasks and produces a Plan. // Tasks are grouped by partition, classified as candidates or skipped, // and candidates are bin-packed into groups targeting TargetFileSizeBytes. +// Tasks should come from one coherent scan plan; split tasks for the same file +// are collapsed, retaining the first task's per-file delete and lineage data. func (cfg Config) PlanCompaction(tasks []table.FileScanTask) (Plan, error) { if err := cfg.Validate(); err != nil { return Plan{}, fmt.Errorf("invalid compaction config: %w", err) } + // Compaction operates on whole physical files. A scan planner may split a + // file into multiple tasks, so collapse tasks by path before counting, + // selecting, or packing them. Keep the first task's delete and lineage + // metadata while replacing its range with the complete file range. + wholeFileTasks := make([]table.FileScanTask, 0, len(tasks)) + seenFiles := make(map[string]struct{}, len(tasks)) + for _, task := range tasks { + if task.File == nil { + return Plan{}, fmt.Errorf("%w: compaction task has no data file", table.ErrInvalidOperation) + } + path := task.File.FilePath() + if _, ok := seenFiles[path]; ok { + continue Review Comment: Should `PlanCompaction` return `ErrInvalidOperation` when a later task for the same path carries different delete files or `FirstRowID`? Today it keeps the first and skips the rest without comparing, while `ExecuteCompactionGroup` [rejects a duplicate path](https://github.com/apache/iceberg-go/blob/06b53e07fd1b4e0d758c25cafcfb089d0f0f0ce2/table/rewrite_data_files.go#L437-L446) outright. Splits from one scan are [copies of a single task](https://github.com/apache/iceberg-go/blob/06b53e07fd1b4e0d758c25cafcfb089d0f0f0ce2/table/scan_splits.go#L66-L94), so nothing in-repo hits this. The check would enforce the "one coherent scan plan" precondition instead of only documenting it. ########## table/compaction/compaction.go: ########## @@ -188,16 +188,38 @@ type Plan struct { // PlanCompaction analyzes the given scan tasks and produces a Plan. // Tasks are grouped by partition, classified as candidates or skipped, // and candidates are bin-packed into groups targeting TargetFileSizeBytes. +// Tasks should come from one coherent scan plan; split tasks for the same file +// are collapsed, retaining the first task's per-file delete and lineage data. func (cfg Config) PlanCompaction(tasks []table.FileScanTask) (Plan, error) { if err := cfg.Validate(); err != nil { return Plan{}, fmt.Errorf("invalid compaction config: %w", err) } + // Compaction operates on whole physical files. A scan planner may split a + // file into multiple tasks, so collapse tasks by path before counting, + // selecting, or packing them. Keep the first task's delete and lineage + // metadata while replacing its range with the complete file range. + wholeFileTasks := make([]table.FileScanTask, 0, len(tasks)) + seenFiles := make(map[string]struct{}, len(tasks)) + for _, task := range tasks { + if task.File == nil { + return Plan{}, fmt.Errorf("%w: compaction task has no data file", table.ErrInvalidOperation) + } + path := task.File.FilePath() + if _, ok := seenFiles[path]; ok { + continue + } + seenFiles[path] = struct{}{} + task.Start = 0 + task.Length = task.File.FileSizeBytes() Review Comment: Could this also set `task.Residual = nil`, to match Java's planner, which scans with [`ignoreResiduals()`](https://github.com/apache/iceberg/blob/2114bf631e49af532d66e2ce148ee49dd1dd1f1f/core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java#L286-L294)? As it stands, a task planned from `Scan(WithRowFilter(...))` with a data-column predicate keeps its residual through planning and is only rejected later by the [new check in `ExecuteCompactionGroup`](https://github.com/apache/iceberg-go/blob/06b53e07fd1b4e0d758c25cafcfb089d0f0f0ce2/table/rewrite_data_files.go#L452-L454). `Analyze` scans unfiltered, so this isn't a blocker. It's more a question of where we want that contract to live. -- 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]
