laskoviymishka commented on code in PR #1975: URL: https://github.com/apache/iceberg-go/pull/1975#discussion_r3903427948
########## table/delete_file_index.go: ########## @@ -0,0 +1,217 @@ +// 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 table + +import ( + "cmp" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + iceberginternal "github.com/apache/iceberg-go/internal" +) + +// deleteFileIndexEntry is the part of a manifest entry used after delete +// manifests have been read. Keeping the sequence number separate lets the +// index release the full ManifestEntry, including its unused metadata maps. +type deleteFileIndexEntry struct { + file iceberg.DataFile + sequenceNum int64 +} + +// compactDeleteFileForIndex copies a delete file without retaining the wide +// metadata maps that are not needed by scan planning or delete-file readers. +// The returned file still carries all non-statistics metadata exposed by the +// DataFile interface, plus only the selected statistics fields. +// +// Positional-delete indexes select file_pathFieldID because partition-scoped +// position deletes use those bounds for candidate pruning. Equality-delete +// indexes select their equality field IDs. Deletion vectors select no stats. +func compactDeleteFileForIndex( Review Comment: Still the always-nil error return, and the split into compactDeleteFileForIndex / compactDeleteFileForIndexWithReference doubled it: both declare (iceberg.DataFile, error), the NewDataFileBuilder failure path returns (file, nil) as the intended fallback, and builder.Build() has no error. So the dead `if err != nil` branch is now in four call sites (equality_delete_index.go:481, positional_delete_index.go:62 and its by-path sibling, scanner.go:883). I'd drop the error from both signatures and simplify the callers, keeping the fallback comment so the "hold onto the original file for external/malformed metadata" intent stays documented. wdyt? ########## table/scanner.go: ########## @@ -872,14 +872,21 @@ func partitionsMatch(a, b map[int]any) bool { // buildDVIndex indexes deletion vectors by the data file path they reference. // The spec requires at most one DV per data file; a second entry for the same // path is rejected with an error. -func buildDVIndex(dvEntries []iceberg.ManifestEntry) (map[string]iceberg.ManifestEntry, error) { - dvIndex := make(map[string]iceberg.ManifestEntry, len(dvEntries)) +func buildDVIndex(dvEntries []iceberg.ManifestEntry) (map[string]deleteFileIndexEntry, error) { Review Comment: This is the same base-staleness issue from last round, and I was able to pin it down concretely this time. The branch sits on 8778910, one commit before the streaming scanner (#1974) that's now main's head, and both sides rewrote planFilesLocal. buildDVIndex here returns map[string]deleteFileIndexEntry and matchDVToData takes it, but on current main planDataManifestTasks and fileScanTaskForDataEntry still thread dvIndex map[string]iceberg.ManifestEntry into matchDVToData. A merge-tree of this branch into origin/main conflicts in scanner.go, and once that's resolved the DV type won't line up with the streaming path. I'd rebase onto current main, thread deleteFileIndexEntry through planDataManifestTasks and fileScanTaskForDataEntry, and land the entries.*Entries = nil release just before the planDataManifestTasks call. Worth building the merge result locally, since the branch checks stay green regardless. ########## table/compaction/analyze_test.go: ########## @@ -167,6 +167,56 @@ func TestAnalyze_AllOptimal(t *testing.T) { assert.Empty(t, plan.Groups) } +func TestPlanCompaction_BoundsScopedPositionalDeleteRatioFromScan(t *testing.T) { Review Comment: This new test is the right shape, it proves a bounds-scoped positional delete keeps its ReferencedDataFile all the way through scan and into compaction grouping, which is exactly the kind of "compaction dropped a needed field" regression I was worried about last round. The gap that's left is the general one: nothing asserts a compact index returns the same delete-file matches as a full-stat index across the equality and DV paths too. This test covers the positional referenced-file case specifically. A small case that builds both a full-stat and a compact index from the same entries and asserts forDataFile returns the same set would lock the whole trimming contract, not just this slice. Not blocking, and I'd be fine taking it as a follow-up given the merge rework this needs anyway. ########## table/delete_file_index.go: ########## @@ -0,0 +1,217 @@ +// 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 table + +import ( + "cmp" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + iceberginternal "github.com/apache/iceberg-go/internal" +) + +// deleteFileIndexEntry is the part of a manifest entry used after delete +// manifests have been read. Keeping the sequence number separate lets the +// index release the full ManifestEntry, including its unused metadata maps. +type deleteFileIndexEntry struct { + file iceberg.DataFile + sequenceNum int64 +} + +// compactDeleteFileForIndex copies a delete file without retaining the wide +// metadata maps that are not needed by scan planning or delete-file readers. +// The returned file still carries all non-statistics metadata exposed by the +// DataFile interface, plus only the selected statistics fields. +// +// Positional-delete indexes select file_pathFieldID because partition-scoped +// position deletes use those bounds for candidate pruning. Equality-delete +// indexes select their equality field IDs. Deletion vectors select no stats. +func compactDeleteFileForIndex( + file iceberg.DataFile, + partition map[int]any, + statFieldIDs []int, +) (iceberg.DataFile, error) { + return compactDeleteFileForIndexWithReference(file, partition, statFieldIDs, nil) +} + +func compactDeleteFileForIndexWithReference( + file iceberg.DataFile, + partition map[int]any, + statFieldIDs []int, + referencedDataFile *string, +) (iceberg.DataFile, error) { + partitionSpec := syntheticPartitionSpec(file.SpecID(), partition) + builder, err := iceberg.NewDataFileBuilder( + partitionSpec, + file.ContentType(), + file.FilePath(), + file.FileFormat(), + partition, + nil, + nil, + file.Count(), + file.FileSizeBytes(), + ) + if err != nil { + // Keep the existing file for malformed or external metadata. Index + // construction historically accepted such DataFile implementations; + // valid manifest files take the compact path above without changing the + // error behavior of the surrounding planner. + return file, nil + } + + valueCounts, nullCounts, nanCounts, lowerBounds, upperBounds := dataFileStatsForFields(file, statFieldIDs) + if valueCounts != nil { + builder.ValueCounts(valueCounts) + } + if nullCounts != nil { + builder.NullValueCounts(nullCounts) + } + if nanCounts != nil { + builder.NaNValueCounts(nanCounts) + } + if lowerBounds != nil { + builder.LowerBoundValues(lowerBounds) + } + if upperBounds != nil { + builder.UpperBoundValues(upperBounds) + } + + // Read these small non-statistics fields directly. The borrowed collection + // helper also exposes column sizes and the built-in implementation lazily + // initializes every statistics map while preparing that view. + keyMetadata := file.KeyMetadata() + splitOffsets := file.SplitOffsets() + equalityFieldIDs := file.EqualityFieldIDs() + if keyMetadata != nil { + builder.KeyMetadata(keyMetadata) + } + if splitOffsets != nil { + builder.SplitOffsets(splitOffsets) + } + if equalityFieldIDs != nil { + builder.EqualityFieldIDs(equalityFieldIDs) + } + + sortOrderID, firstRowID, manifestReferencedDataFile, contentOffset, contentSize := iceberginternal.BorrowedDataFilePointers(file) + if sortOrderID != nil { + builder.SortOrderID(*sortOrderID) + } + if firstRowID != nil { + builder.FirstRowID(*firstRowID) + } + if referencedDataFile != nil { + builder.ReferencedDataFile(*referencedDataFile) + } else if manifestReferencedDataFile != nil { + builder.ReferencedDataFile(*manifestReferencedDataFile) + } + if contentOffset != nil { + builder.ContentOffset(*contentOffset) + } + if contentSize != nil { + builder.ContentSizeInBytes(*contentSize) + } + + return builder.Build(), nil +} + +// syntheticPartitionSpec gives the built-in DataFile implementation enough +// field metadata to expose the copied partition map and to remain usable by +// the DataFile codec. Partition values are already transformed values, so the +// identity transforms here are only a local storage description. +func syntheticPartitionSpec(specID int32, partition map[int]any) iceberg.PartitionSpec { + fields := make([]iceberg.PartitionField, 0, len(partition)) + for fieldID := range partition { + fields = append(fields, iceberg.PartitionField{ + SourceIDs: []int{fieldID}, Review Comment: Unchanged from last round. partition is keyed by partition-field IDs, so SourceIDs: []int{fieldID} and FieldID: fieldID end up equal, which claims the source column is field 1000 rather than the real schema column. No impact on the current lookups, since they take (specID, partition map) directly and never interpret this synthetic spec. But the compacted DataFile rides along in the FileScanTask, and now that 11d180e is deliberately making these compact copies survive into compaction planning, a consumer that resolves compacted.SpecID() to reason about partitioning would read a source-id that doesn't match the schema. I'd rather preserve the original spec than hand out one with wrong source IDs, or at least drop a comment saying the source IDs are synthetic. -- 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]
