diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go index eb99e68ca..bdf548eaa 100644 --- a/weed/worker/tasks/iceberg/compact.go +++ b/weed/worker/tasks/iceberg/compact.go @@ -113,6 +113,7 @@ func (h *Handler) compactDataFiles( var positionDeletes map[string][]int64 var eqDeleteGroups []equalityDeleteGroup var allDeleteEntries []iceberg.ManifestEntry + var dvPaths map[string]bool if config.ApplyDeletes && len(deleteManifests) > 0 { for _, mf := range deleteManifests { manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath()) @@ -126,10 +127,19 @@ func (h *Handler) compactDataFiles( allDeleteEntries = append(allDeleteEntries, entries...) } + var err error + dvPaths, err = deletionVectorPaths(allDeleteEntries, bucketName, dataPath) + if err != nil { + return "", nil, err + } + // Separate position and equality deletes, filtering by partition // predicate so out-of-scope deletes don't affect the merge. var posDeleteEntries, eqDeleteEntries []iceberg.ManifestEntry for _, entry := range allDeleteEntries { + if entry.DataFile().FileFormat() == iceberg.PuffinFile { + continue + } if predicate != nil { spec, ok := specsByID[int(entry.DataFile().SpecID())] if !ok { @@ -166,10 +176,13 @@ func (h *Handler) compactDataFiles( } } - candidateEntries := allEntries + candidateEntries, err := excludeDeletionVectorFiles(allEntries, dvPaths, bucketName, dataPath) + if err != nil { + return "", nil, err + } if predicate != nil { - candidateEntries = make([]iceberg.ManifestEntry, 0, len(allEntries)) - for _, entry := range allEntries { + filtered := make([]iceberg.ManifestEntry, 0, len(candidateEntries)) + for _, entry := range candidateEntries { spec, ok := specsByID[int(entry.DataFile().SpecID())] if !ok { continue @@ -179,9 +192,10 @@ func (h *Handler) compactDataFiles( return "", nil, err } if match { - candidateEntries = append(candidateEntries, entry) + filtered = append(filtered, entry) } } + candidateEntries = filtered } minInputFiles, err := compactionMinInputFiles(config.MinInputFiles) @@ -205,6 +219,11 @@ func (h *Handler) compactDataFiles( version := meta.Version() snapshotID := currentSnap.SnapshotID + // v3 row lineage: the snapshot starts at the table's next-row-id, and the + // merged files carry every row's _row_id and _last_updated_sequence_number + // (row_lineage.go). + lineage := hasRowLineage(version) + firstRowID := rowLineageFirstRowID(meta) // Compute the snapshot ID for the commit up front so all manifest entries // reference the same snapshot that will actually be committed. @@ -273,9 +292,9 @@ func (h *Handler) compactDataFiles( var recordCount int64 rowGroupRows := rowsPerRowGroup(bin, config) if rewritePlan != nil && rewritePlan.strategy == "sort" { - mergedData, recordCount, err = mergeParquetFilesSorted(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rewritePlan, rowGroupRows) + mergedData, recordCount, err = mergeParquetFilesSorted(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rewritePlan, rowGroupRows, lineage) } else { - mergedData, recordCount, err = mergeParquetFiles(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rowGroupRows) + mergedData, recordCount, err = mergeParquetFiles(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rowGroupRows, lineage) } if err != nil { glog.Warningf("iceberg compact: failed to merge bin %d (%d files): %v", binIdx, len(bin.Entries), err) @@ -325,6 +344,20 @@ func (h *Handler) compactDataFiles( stats.applyTo(dfBuilder) } + // A merged file keeps its rows' row IDs, so its entry names the first + // materialized one instead of inheriting a fresh range at commit. + if lineage { + firstID, err := mergedFileFirstRowID(mergedData) + if err != nil { + glog.Warningf("iceberg compact: failed to read merged file lineage for bin %d: %v", binIdx, err) + _ = deleteFilerFile(ctx, filerClient, dataDir, mergedFileName) + goto binDone + } + if firstID != nil { + dfBuilder.FirstRowID(*firstID) + } + } + mergedDataFile := dfBuilder.Build() summary.addFile(mergedDataFile) newEntry := iceberg.NewManifestEntry( @@ -471,7 +504,7 @@ func (h *Handler) compactDataFiles( // Write new manifest list var manifestListBuf bytes.Buffer seqNum := currentSnap.SequenceNumber + 1 - err = iceberg.WriteManifestList(version, &manifestListBuf, newSnapID, &snapshotID, &seqNum, 0, allManifests) + addedRows, err := writeManifestList(version, &manifestListBuf, newSnapID, &snapshotID, &seqNum, firstRowID, allManifests) if err != nil { return "", nil, fmt.Errorf("write compact manifest list: %w", err) } @@ -486,9 +519,9 @@ func (h *Handler) compactDataFiles( manifestListLocation := absoluteIcebergPath(bucketName, dataPath, "metadata", manifestListFileName) err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, state.MetadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error { // Guard: verify table head hasn't advanced since we planned. - cs := currentMeta.CurrentSnapshot() - if cs == nil || cs.SnapshotID != snapshotID { - return errStalePlan + cs, err := checkCommitPlan(currentMeta, snapshotID, version, firstRowID) + if err != nil { + return err } newSnapshot := &table.Snapshot{ @@ -512,6 +545,7 @@ func (h *Handler) compactDataFiles( if rewritePlan != nil && rewritePlan.strategy == "sort" { newSnapshot.Summary.Properties["sort-fields"] = rewritePlan.summaryLabel() } + setRowLineage(newSnapshot, version, firstRowID, addedRows) if err := builder.AddSnapshot(newSnapshot); err != nil { return err } @@ -722,6 +756,48 @@ func partitionKey(partition map[int]any) string { return strings.Join(parts, "\x00") } +// deletionVectorPaths names the data files a Puffin deletion-vector entry +// covers, normalized to merge-time keys. Merging a covered file needs a +// Puffin reader to apply the vector; without one the merge would resurrect +// deleted rows, so the file must stay out of compaction bins. +func deletionVectorPaths(entries []iceberg.ManifestEntry, bucketName, dataPath string) (map[string]bool, error) { + paths := make(map[string]bool) + for _, entry := range entries { + df := entry.DataFile() + if df.ContentType() != iceberg.EntryContentPosDeletes || df.FileFormat() != iceberg.PuffinFile { + continue + } + ref := df.ReferencedDataFile() + if ref == nil || *ref == "" { + return nil, fmt.Errorf("deletion vector %s has no referenced data file", df.FilePath()) + } + normalized, err := normalizeIcebergPath(*ref, bucketName, dataPath) + if err != nil { + return nil, fmt.Errorf("resolve deletion vector source %s: %w", *ref, err) + } + paths[normalized] = true + } + return paths, nil +} + +// excludeDeletionVectorFiles drops entries a deletion vector covers. +func excludeDeletionVectorFiles(entries []iceberg.ManifestEntry, dvPaths map[string]bool, bucketName, dataPath string) ([]iceberg.ManifestEntry, error) { + if len(dvPaths) == 0 { + return entries, nil + } + kept := make([]iceberg.ManifestEntry, 0, len(entries)) + for _, entry := range entries { + normalized, err := normalizeIcebergPath(entry.DataFile().FilePath(), bucketName, dataPath) + if err != nil { + return nil, fmt.Errorf("resolve data file %s: %w", entry.DataFile().FilePath(), err) + } + if !dvPaths[normalized] { + kept = append(kept, entry) + } + } + return kept, nil +} + // collectPositionDeletes reads position delete Parquet files and returns a map // from normalized data file path to sorted row positions that should be deleted. // Paths are normalized so that absolute S3 URLs and relative paths match. @@ -733,8 +809,9 @@ func collectPositionDeletes( ) (map[string][]int64, error) { result := make(map[string][]int64) for _, entry := range deleteEntries { - if entry.DataFile().ContentType() != iceberg.EntryContentPosDeletes { - continue + df := entry.DataFile() + if df.ContentType() != iceberg.EntryContentPosDeletes || df.FileFormat() == iceberg.PuffinFile { + continue // Puffin is a deletion vector, not a row-level delete file } fileDeletes, err := readPositionDeleteFile(ctx, filerClient, bucketName, dataPath, entry.DataFile().FilePath()) if err != nil { @@ -1011,6 +1088,7 @@ func mergeParquetFiles( eqDeleteGroups []equalityDeleteGroup, icebergSchema *iceberg.Schema, rowGroupRows int64, + lineage bool, ) ([]byte, int64, error) { if len(entries) == 0 { return nil, 0, fmt.Errorf("no entries to merge") @@ -1034,11 +1112,24 @@ func mergeParquetFiles( return nil, 0, fmt.Errorf("resolve equality columns: %w", err) } - var outputBuf bytes.Buffer - writer := parquet.NewWriter(&outputBuf, parquetSchema, parquet.MaxRowsPerRowGroup(rowGroupRows)) + // On a v3 table the output is the first file's columns plus the lineage + // columns (row_lineage.go). + outputSchema := parquetSchema + var layout *lineageLayout + if lineage { + if layout, err = newLineageLayout(parquetSchema); err != nil { + firstReader.Close() + return nil, 0, fmt.Errorf("schema of %s: %w", entries[0].DataFile().FilePath(), err) + } + outputSchema = layout.schema + } - drainReader := func(reader *parquet.Reader, source string) (int64, error) { - return visitFilteredParquetRows(ctx, reader, source, bucketName, dataPath, positionDeletes, resolvedEqGroups, func(filtered []parquet.Row) error { + var outputBuf bytes.Buffer + writer := parquet.NewWriter(&outputBuf, outputSchema, parquet.MaxRowsPerRowGroup(rowGroupRows)) + + drainReader := func(reader *parquet.Reader, entry iceberg.ManifestEntry) (int64, error) { + source := entry.DataFile().FilePath() + return copyFileRows(ctx, reader, entry, layout, bucketName, dataPath, positionDeletes, resolvedEqGroups, eqDeleteGroups, icebergSchema, func(filtered []parquet.Row) error { if _, err := writer.WriteRows(filtered); err != nil { return fmt.Errorf("write rows from %s: %w", source, err) } @@ -1047,8 +1138,7 @@ func mergeParquetFiles( } // Drain the first file. - firstSource := entries[0].DataFile().FilePath() - totalRows, err := drainReader(firstReader, firstSource) + totalRows, err := drainReader(firstReader, entries[0]) if err != nil { writer.Close() return nil, 0, err @@ -1071,13 +1161,14 @@ func mergeParquetFiles( } reader := parquet.NewReader(bytes.NewReader(data)) - if !schemasEqual(parquetSchema, reader.Schema()) { + // A v3 merge matches files with their lineage columns set aside. + if layout == nil && !schemasEqual(parquetSchema, reader.Schema()) { reader.Close() writer.Close() return nil, 0, fmt.Errorf("schema mismatch in %s: cannot merge files with different schemas", entry.DataFile().FilePath()) } - rowsWritten, err := drainReader(reader, entry.DataFile().FilePath()) + rowsWritten, err := drainReader(reader, entry) if err != nil { writer.Close() return nil, 0, err @@ -1124,12 +1215,15 @@ func visitFilteredParquetRows( source, bucketName, dataPath string, positionDeletes map[string][]int64, resolvedEqGroups []resolvedEqDeleteGroup, - onRows func([]parquet.Row) error, + onRows func(rows []parquet.Row, positions []int64) error, ) (int64, error) { defer reader.Close() rows := make([]parquet.Row, 256) filteredRows := make([]parquet.Row, 0, len(rows)) + // each kept row's position in the file before any delete: what a v3 + // _row_id is inherited from + filteredPositions := make([]int64, 0, len(rows)) normalizedSource, err := normalizeIcebergPath(source, bucketName, dataPath) if err != nil { return 0, err @@ -1149,6 +1243,7 @@ func visitFilteredParquetRows( n, readErr := reader.ReadRows(rows) if n > 0 { filteredRows = filteredRows[:0] + filteredPositions = filteredPositions[:0] for i := 0; i < n; i++ { rowPos := absolutePos + int64(i) for posDeleteIdx < len(posDeletes) && posDeletes[posDeleteIdx] < rowPos { @@ -1172,10 +1267,11 @@ func visitFilteredParquetRows( } filteredRows = append(filteredRows, rows[i]) + filteredPositions = append(filteredPositions, rowPos) } absolutePos += int64(n) if len(filteredRows) > 0 { - if err := onRows(filteredRows); err != nil { + if err := onRows(filteredRows, filteredPositions); err != nil { return totalRows, err } totalRows += int64(len(filteredRows)) @@ -1201,6 +1297,7 @@ func mergeParquetFilesSorted( icebergSchema *iceberg.Schema, rewritePlan *compactionRewritePlan, rowGroupRows int64, + lineage bool, ) ([]byte, int64, error) { if len(entries) == 0 { return nil, 0, fmt.Errorf("no entries to merge") @@ -1220,7 +1317,17 @@ func mergeParquetFilesSorted( return nil, 0, fmt.Errorf("no parquet schema found in %s", entries[0].DataFile().FilePath()) } - sortingColumns, err := rewritePlan.parquetSortingColumns(parquetSchema) + outputSchema := parquetSchema + var layout *lineageLayout + if lineage { + if layout, err = newLineageLayout(parquetSchema); err != nil { + firstReader.Close() + return nil, 0, fmt.Errorf("schema of %s: %w", entries[0].DataFile().FilePath(), err) + } + outputSchema = layout.schema + } + + sortingColumns, err := rewritePlan.parquetSortingColumns(outputSchema) if err != nil { firstReader.Close() return nil, 0, fmt.Errorf("resolve parquet sorting columns: %w", err) @@ -1249,7 +1356,7 @@ func mergeParquetFilesSorted( } var outputBuf bytes.Buffer - writer := parquet.NewSortingWriter[any](&outputBuf, sortBufferRows, parquetSchema, + writer := parquet.NewSortingWriter[any](&outputBuf, sortBufferRows, outputSchema, parquet.MaxRowsPerRowGroup(rowGroupRows), parquet.SortingWriterConfig( parquet.SortingColumns(sortingColumns...), @@ -1265,8 +1372,8 @@ func mergeParquetFilesSorted( } }() - writeRows := func(reader *parquet.Reader, source string) (int64, error) { - return visitFilteredParquetRows(ctx, reader, source, bucketName, dataPath, positionDeletes, resolvedEqGroups, func(filtered []parquet.Row) error { + writeRows := func(reader *parquet.Reader, entry iceberg.ManifestEntry) (int64, error) { + return copyFileRows(ctx, reader, entry, layout, bucketName, dataPath, positionDeletes, resolvedEqGroups, eqDeleteGroups, icebergSchema, func(filtered []parquet.Row) error { // The writer's row buffer copies the values it is handed, so these // rows need no cloning out of the reader's own buffer first. _, writeErr := writer.WriteRows(filtered) @@ -1274,7 +1381,7 @@ func mergeParquetFilesSorted( }) } - totalRows, err := writeRows(firstReader, entries[0].DataFile().FilePath()) + totalRows, err := writeRows(firstReader, entries[0]) if err != nil { return nil, 0, err } @@ -1292,12 +1399,12 @@ func mergeParquetFilesSorted( } reader := parquet.NewReader(bytes.NewReader(data)) - if !schemasEqual(parquetSchema, reader.Schema()) { + if layout == nil && !schemasEqual(parquetSchema, reader.Schema()) { reader.Close() return nil, 0, fmt.Errorf("schema mismatch in %s: cannot merge files with different schemas", entry.DataFile().FilePath()) } - rowsWritten, err := writeRows(reader, entry.DataFile().FilePath()) + rowsWritten, err := writeRows(reader, entry) if err != nil { return nil, 0, err } diff --git a/weed/worker/tasks/iceberg/delete_rewrite.go b/weed/worker/tasks/iceberg/delete_rewrite.go index 064dab166..3dd22ec14 100644 --- a/weed/worker/tasks/iceberg/delete_rewrite.go +++ b/weed/worker/tasks/iceberg/delete_rewrite.go @@ -107,6 +107,9 @@ func collectDeleteRewriteGroups( } allPositionEntries = append(allPositionEntries, entry) + if entry.DataFile().FileFormat() == iceberg.PuffinFile { + continue + } fileDeletes, err := readPositionDeleteFile(ctx, filerClient, bucketName, dataPath, entry.DataFile().FilePath()) if err != nil { @@ -258,6 +261,8 @@ func writePositionDeleteFile(rows []positionDeleteRow) ([]byte, error) { return buf.Bytes(), nil } +const skipDeleteRewriteV3 = "rewrite_position_delete_files skipped: position-delete files must not be added to a format-version 3 table" + func (h *Handler) rewritePositionDeleteFiles( ctx context.Context, filerClient filer_pb.SeaweedFilerClient, @@ -275,6 +280,13 @@ func (h *Handler) rewritePositionDeleteFiles( if currentSnap == nil || currentSnap.ManifestList == "" { return "no current snapshot", nil, nil } + // The spec forbids adding position-delete files to a v3 table ("Position + // delete files must not be added to v3 tables"; v3 writes deletion + // vectors), and this rewrite adds new ones. Compaction still applies and + // removes existing position deletes. + if hasRowLineage(meta.Version()) { + return skipDeleteRewriteV3, nil, nil + } manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, currentSnap.ManifestList) if err != nil { @@ -541,8 +553,10 @@ func (h *Handler) rewritePositionDeleteFiles( allManifests = append(allManifests, mf) } + firstRowID := rowLineageFirstRowID(meta) var manifestListBuf bytes.Buffer - if err := iceberg.WriteManifestList(version, &manifestListBuf, newSnapID, &snapshotID, &seqNum, 0, allManifests); err != nil { + addedRows, err := writeManifestList(version, &manifestListBuf, newSnapID, &snapshotID, &seqNum, firstRowID, allManifests) + if err != nil { return "", nil, fmt.Errorf("write delete manifest list: %w", err) } manifestListName := fmt.Sprintf("snap-%d-%s.avro", newSnapID, artifactSuffix) @@ -553,9 +567,9 @@ func (h *Handler) rewritePositionDeleteFiles( manifestListLocation := absoluteIcebergPath(bucketName, dataPath, "metadata", manifestListName) err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, state.MetadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error { - cs := currentMeta.CurrentSnapshot() - if cs == nil || cs.SnapshotID != snapshotID { - return errStalePlan + cs, err := checkCommitPlan(currentMeta, snapshotID, version, firstRowID) + if err != nil { + return err } newSnapshot := &table.Snapshot{ SnapshotID: newSnapID, @@ -574,6 +588,7 @@ func (h *Handler) rewritePositionDeleteFiles( return &id }(), } + setRowLineage(newSnapshot, version, firstRowID, addedRows) if err := builder.AddSnapshot(newSnapshot); err != nil { return err } diff --git a/weed/worker/tasks/iceberg/deletion_vector_test.go b/weed/worker/tasks/iceberg/deletion_vector_test.go new file mode 100644 index 000000000..028bcaa8e --- /dev/null +++ b/weed/worker/tasks/iceberg/deletion_vector_test.go @@ -0,0 +1,375 @@ +package iceberg + +import ( + "bytes" + "context" + "encoding/json" + "path" + "strings" + "testing" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/parquet-go/parquet-go" + + filer_pb "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +// A deletion-vector entry names its data file in referenced_data_file; the +// match against candidates is on the normalized path either side takes. +func TestDeletionVectorPathMatching(t *testing.T) { + spec := *iceberg.UnpartitionedSpec + snapID := int64(1) + + newEntry := func(path, ref string) iceberg.ManifestEntry { + dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentPosDeletes, path, + iceberg.PuffinFile, map[int]any{}, nil, nil, 0, 8) + if err != nil { + t.Fatalf("build dv file: %v", err) + } + if ref != "" { + dfb.ReferencedDataFile(ref) + } + return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()) + } + + paths, err := deletionVectorPaths([]iceberg.ManifestEntry{ + newEntry("metadata/dv.puffin", "s3://test-bucket/ns/tbl/data/f1.parquet"), + }, "test-bucket", "ns/tbl") + if err != nil { + t.Fatalf("deletionVectorPaths: %v", err) + } + if !paths["ns/tbl/data/f1.parquet"] { + t.Fatalf("dv paths = %v, want the normalized f1 path", paths) + } + + dataEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, + mustDataFile(t, spec, "data/f1.parquet", 3)) + otherEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, + mustDataFile(t, spec, "data/f2.parquet", 2)) + kept, err := excludeDeletionVectorFiles([]iceberg.ManifestEntry{dataEntry, otherEntry}, paths, "test-bucket", "ns/tbl") + if err != nil { + t.Fatalf("excludeDeletionVectorFiles: %v", err) + } + if len(kept) != 1 || kept[0].DataFile().FilePath() != "data/f2.parquet" { + t.Fatalf("kept %d entries, want only f2", len(kept)) + } +} + +// A Puffin deletion vector is not a Parquet position-delete file: it must not +// reach readPositionDeleteFile, and the data file it covers must stay +// uncompacted until something applies the vector. Other files still merge and +// ordinary Parquet position deletes still apply. +func TestCompactDataFilesDeletionVector(t *testing.T) { + fs, client := startFakeFiler(t) + ctx := context.Background() + + setup := tableSetup{BucketName: "test-bucket", Namespace: "ns", TableName: "tbl"} + spec := *iceberg.UnpartitionedSpec + snapID := int64(1) + + putDataFileRows(t, fs, "f1.parquet", []lineageDataRow{{1, "a"}, {2, "b"}}) + putDataFileRows(t, fs, "f2.parquet", []lineageDataRow{{3, "c"}, {4, "d"}}) + putDataFileRows(t, fs, "f3.parquet", []lineageDataRow{{5, "e"}}) + + dataEntries := []iceberg.ManifestEntry{ + iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, + mustDataFile(t, spec, "data/f1.parquet", 2)), + iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, + mustDataFile(t, spec, "data/f2.parquet", 2)), + iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, + mustDataFile(t, spec, "data/f3.parquet", 1)), + } + + // An ordinary Parquet position delete against f2's first row. + type posRow struct { + FilePath string `parquet:"file_path"` + Pos int64 `parquet:"pos"` + } + var pdBuf bytes.Buffer + w := parquet.NewWriter(&pdBuf, parquet.SchemaOf(new(posRow))) + if err := w.Write(&posRow{"data/f2.parquet", 0}); err != nil { + t.Fatalf("write pos delete: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("close pos delete: %v", err) + } + dataDir := "/buckets/test-bucket/ns/tbl/data" + fs.putEntry(dataDir, "pd1.parquet", &filer_pb.Entry{Name: "pd1.parquet", Content: pdBuf.Bytes()}) + // The vector itself is never opened, but the file should exist like a real + // engine's write would leave it. + fs.putEntry(dataDir, "dv.puffin", &filer_pb.Entry{Name: "dv.puffin", Content: []byte("puffin-bytes")}) + + posDF, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentPosDeletes, + setup.fileRef("data", "pd1.parquet"), iceberg.ParquetFile, map[int]any{}, nil, nil, 1, int64(pdBuf.Len())) + if err != nil { + t.Fatalf("build pos delete file: %v", err) + } + dvDF, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentPosDeletes, + setup.fileRef("data", "dv.puffin"), iceberg.PuffinFile, map[int]any{}, nil, nil, 0, 12) + if err != nil { + t.Fatalf("build dv file: %v", err) + } + dvDF.ReferencedDataFile("s3://test-bucket/ns/tbl/data/f1.parquet") + deleteEntries := []iceberg.ManifestEntry{ + iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, posDF.Build()), + iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dvDF.Build()), + } + + writeV3Table(t, fs, setup, dataEntries, deleteEntries, 0, 5) + + handler := NewHandler(nil) + config := Config{ + TargetFileSizeBytes: 4096, + MinInputFiles: 2, + ApplyDeletes: true, + } + result, _, err := handler.compactDataFiles(ctx, client, setup.BucketName, setup.tablePath(), config, nil) + if err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + if !strings.Contains(result, "compacted 2 files into 1") { + t.Fatalf("unexpected result: %q", result) + } + + state, err := loadCurrentMetadata(ctx, client, setup.BucketName, setup.tablePath()) + if err != nil { + t.Fatalf("loadCurrentMetadata: %v", err) + } + manifests, err := loadCurrentManifests(ctx, client, setup.BucketName, state.DataPath, state.Metadata) + if err != nil { + t.Fatalf("loadCurrentManifests: %v", err) + } + + var dataMfs, deleteMfs []iceberg.ManifestFile + for _, mf := range manifests { + if mf.ManifestContent() == iceberg.ManifestContentData { + dataMfs = append(dataMfs, mf) + } else { + deleteMfs = append(deleteMfs, mf) + } + } + if len(deleteMfs) != 1 { + t.Fatalf("delete manifests carried = %d, want 1 (the vector still applies to f1)", len(deleteMfs)) + } + + var merged iceberg.DataFile + var f1Live bool + for _, mf := range dataMfs { + manifestData, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, mf.FilePath()) + if err != nil { + t.Fatalf("read manifest: %v", err) + } + entries, err := s3tables.ReadManifest(mf, manifestData, false, specByID(state.Metadata), state.Metadata.CurrentSchema()) + if err != nil { + t.Fatalf("parse manifest: %v", err) + } + for _, e := range entries { + df := e.DataFile() + switch { + case e.Status() == iceberg.EntryStatusADDED: + merged = df + case e.Status() == iceberg.EntryStatusEXISTING && df.FilePath() == "data/f1.parquet": + f1Live = true + } + } + } + if !f1Live { + t.Fatal("f1 is covered by a deletion vector and must stay live") + } + if merged == nil { + t.Fatal("no merged file added") + } + + mergedData, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, merged.FilePath()) + if err != nil { + t.Fatalf("read merged file: %v", err) + } + ids, _, rowIDs, _ := mergedColumns(t, mergedData) + if len(ids) != 2 || ids[0] != int64(4) || ids[1] != int64(5) { + t.Fatalf("merged ids = %v, want [4 5] (f2 minus its deleted row, f3)", ids) + } + if rowIDs[0] != int64(3) || rowIDs[1] != int64(4) { + t.Fatalf("merged _row_id = %v, want [3 4]", rowIDs) + } +} + +// Detection applies the same exclusion compaction does: a table whose only +// binable files are all covered by deletion vectors is not a candidate, while +// unaffected files still qualify. +func TestDetectSchedulesDeletionVector(t *testing.T) { + spec := *iceberg.UnpartitionedSpec + snapID := int64(1) + + dataEntry := func(name string, count int64) iceberg.ManifestEntry { + return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, + mustDataFile(t, spec, name, count)) + } + dvEntry := func(t *testing.T, ref string) iceberg.ManifestEntry { + dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentPosDeletes, + "data/dv.puffin", iceberg.PuffinFile, map[int]any{}, nil, nil, 0, 12) + if err != nil { + t.Fatalf("build dv file: %v", err) + } + dfb.ReferencedDataFile(ref) + return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()) + } + + config := Config{ + SnapshotRetentionMs: hoursToMs(24 * 365), + MaxSnapshotsToKeep: 10, + TargetFileSizeBytes: 4096, + MinInputFiles: 2, + Operations: "compact", + ApplyDeletes: true, + } + + t.Run("unaffected files stay eligible", func(t *testing.T) { + fs, client := startFakeFiler(t) + setup := tableSetup{BucketName: "test-bucket", Namespace: "ns", TableName: "tbl"} + writeV3Table(t, fs, setup, + []iceberg.ManifestEntry{dataEntry("data/f1.parquet", 1), dataEntry("data/f2.parquet", 1), dataEntry("data/f3.parquet", 1)}, + []iceberg.ManifestEntry{dvEntry(t, "data/f1.parquet")}, + 0, 3) + + tables, err := NewHandler(nil).scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0) + if err != nil { + t.Fatalf("scanTablesForMaintenance: %v", err) + } + if len(tables) != 1 { + t.Fatalf("expected 1 compaction candidate (f2+f3 still binable), got %d", len(tables)) + } + }) + + t.Run("all binable files covered", func(t *testing.T) { + fs, client := startFakeFiler(t) + setup := tableSetup{BucketName: "test-bucket", Namespace: "ns", TableName: "tbl"} + writeV3Table(t, fs, setup, + []iceberg.ManifestEntry{dataEntry("data/f1.parquet", 1), dataEntry("data/f2.parquet", 1)}, + []iceberg.ManifestEntry{dvEntry(t, "data/f1.parquet")}, + 0, 2) + + tables, err := NewHandler(nil).scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0) + if err != nil { + t.Fatalf("scanTablesForMaintenance: %v", err) + } + if len(tables) != 0 { + t.Fatalf("expected no candidates (only the covered f1 and the lone f2 remain), got %d", len(tables)) + } + }) +} + +// writeV3Table stores a format-version-3 table in the fake filer: a data +// manifest whose entries carry no first_row_id (so readers exercise manifest +// first_row_id inheritance), an optional delete manifest, and the manifest +// list + snapshot that commit them at sequence number 1. +func writeV3Table(t *testing.T, fs *fakeFilerServer, setup tableSetup, + dataManifestEntries, deleteManifestEntries []iceberg.ManifestEntry, + listFirstRowID, addedRows int64) table.Metadata { + t.Helper() + + schema := newTestSchema() + spec := *iceberg.UnpartitionedSpec + location := "s3://" + setup.BucketName + "/" + setup.dataPath() + + base, err := table.NewMetadata(schema, &spec, table.UnsortedSortOrder, location, nil) + if err != nil { + t.Fatalf("new metadata: %v", err) + } + builder, err := table.MetadataBuilderFromBase(base, location) + if err != nil { + t.Fatalf("metadata builder: %v", err) + } + if err := builder.SetFormatVersion(3); err != nil { + t.Fatalf("set format version: %v", err) + } + if _, err := builder.Build(); err != nil { + t.Fatalf("build v3 metadata: %v", err) + } + + snapID := int64(1) + var manifestBuf bytes.Buffer + dataMf, err := iceberg.WriteManifest(setup.fileRef("metadata", "m0.avro"), &manifestBuf, 3, spec, schema, snapID, dataManifestEntries) + if err != nil { + t.Fatalf("write manifest: %v", err) + } + metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.dataPath(), "metadata") + fs.putEntry(metaDir, "m0.avro", &filer_pb.Entry{ + Name: "m0.avro", + Content: manifestBuf.Bytes(), + }) + + manifests := []iceberg.ManifestFile{dataMf} + if len(deleteManifestEntries) > 0 { + delMf, delBytes := writeDeleteManifestForTest( + t, setup.fileRef("metadata", "del.avro"), 3, spec, schema, snapID, deleteManifestEntries) + fs.putEntry(metaDir, "del.avro", &filer_pb.Entry{ + Name: "del.avro", + Content: delBytes, + }) + manifests = append(manifests, delMf) + } + + var listBuf bytes.Buffer + seqNum := int64(1) + if err := iceberg.WriteManifestList(3, &listBuf, snapID, nil, &seqNum, listFirstRowID, manifests); err != nil { + t.Fatalf("write manifest list: %v", err) + } + fs.putEntry(metaDir, "snap-1.avro", &filer_pb.Entry{ + Name: "snap-1.avro", + Content: listBuf.Bytes(), + }) + + snap := table.Snapshot{ + SnapshotID: snapID, + SequenceNumber: seqNum, + TimestampMs: time.Now().UnixMilli(), + ManifestList: setup.fileRef("metadata", "snap-1.avro"), + FirstRowID: ptrInt64(listFirstRowID), + AddedRows: &addedRows, + } + if err := builder.AddSnapshot(&snap); err != nil { + t.Fatalf("add snapshot: %v", err) + } + if err := builder.SetSnapshotRef(table.MainBranch, snapID, table.BranchRef); err != nil { + t.Fatalf("set snapshot ref: %v", err) + } + meta, err := builder.Build() + if err != nil { + t.Fatalf("build metadata with snapshot: %v", err) + } + + fullMetadataJSON, err := json.Marshal(meta) + if err != nil { + t.Fatalf("marshal metadata: %v", err) + } + internalMeta := map[string]interface{}{ + "metadataVersion": 1, + "metadataLocation": setup.fileRef("metadata", "v1.metadata.json"), + "metadata": map[string]interface{}{"fullMetadata": json.RawMessage(fullMetadataJSON)}, + } + xattr, err := json.Marshal(internalMeta) + if err != nil { + t.Fatalf("marshal xattr: %v", err) + } + bucketPath := path.Join(s3tables.TablesPath, setup.BucketName) + fs.putEntry(s3tables.TablesPath, setup.BucketName, &filer_pb.Entry{ + Name: setup.BucketName, + IsDirectory: true, + Extended: map[string][]byte{s3tables.ExtendedKeyTableBucket: []byte("true")}, + }) + fs.putEntry(bucketPath, setup.Namespace, &filer_pb.Entry{Name: setup.Namespace, IsDirectory: true}) + fs.putEntry(path.Join(bucketPath, setup.Namespace), setup.TableName, &filer_pb.Entry{ + Name: setup.TableName, + IsDirectory: true, + Extended: map[string][]byte{ + s3tables.ExtendedKeyMetadata: xattr, + s3tables.ExtendedKeyMetadataVersion: metadataVersionXattr(1), + }, + }) + return meta +} + +func ptrInt64(v int64) *int64 { return &v } diff --git a/weed/worker/tasks/iceberg/detection.go b/weed/worker/tasks/iceberg/detection.go index 4bfc693bf..74288613e 100644 --- a/weed/worker/tasks/iceberg/detection.go +++ b/weed/worker/tasks/iceberg/detection.go @@ -308,6 +308,9 @@ func (h *Handler) tableNeedsMaintenance( return true, nil } case "rewrite_position_delete_files": + if hasRowLineage(meta.Version()) { + continue // the rewrite would add position-delete files, which a v3 table must not get (skipDeleteRewriteV3) + } manifests, err := getCurrentManifests() if err != nil { opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err)) @@ -441,6 +444,7 @@ func hasEligibleCompaction( schema := meta.CurrentSchema() var allEntries []iceberg.ManifestEntry + var dvPaths map[string]bool for _, mf := range dataManifests { manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath()) if err != nil { @@ -453,10 +457,35 @@ func hasEligibleCompaction( allEntries = append(allEntries, entries...) } - candidateEntries := allEntries + // Deletion vectors exist only from format version 3. + if hasRowLineage(meta.Version()) { + var deleteEntries []iceberg.ManifestEntry + for _, mf := range manifests { + if mf.ManifestContent() != iceberg.ManifestContentDeletes { + continue + } + manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath()) + if err != nil { + return false, fmt.Errorf("read delete manifest %s: %w", mf.FilePath(), err) + } + entries, err := s3tables.ReadManifest(mf, manifestData, true, specsByID, schema) + if err != nil { + return false, fmt.Errorf("parse delete manifest %s: %w", mf.FilePath(), err) + } + deleteEntries = append(deleteEntries, entries...) + } + if dvPaths, err = deletionVectorPaths(deleteEntries, bucketName, dataPath); err != nil { + return false, err + } + } + + candidateEntries, err := excludeDeletionVectorFiles(allEntries, dvPaths, bucketName, dataPath) + if err != nil { + return false, err + } if predicate != nil { - candidateEntries = make([]iceberg.ManifestEntry, 0, len(allEntries)) - for _, entry := range allEntries { + filtered := make([]iceberg.ManifestEntry, 0, len(candidateEntries)) + for _, entry := range candidateEntries { spec, ok := specsByID[int(entry.DataFile().SpecID())] if !ok { continue @@ -466,9 +495,10 @@ func hasEligibleCompaction( return false, err } if match { - candidateEntries = append(candidateEntries, entry) + filtered = append(filtered, entry) } } + candidateEntries = filtered } rewritePlan, err := resolveCompactionRewritePlan(config, meta) diff --git a/weed/worker/tasks/iceberg/handler_test.go b/weed/worker/tasks/iceberg/handler_test.go index be5de8b2e..3d88044ab 100644 --- a/weed/worker/tasks/iceberg/handler_test.go +++ b/weed/worker/tasks/iceberg/handler_test.go @@ -1303,7 +1303,7 @@ func TestMergeParquetFilesWithPositionDeletes(t *testing.T) { merged, count, err := mergeParquetFiles( context.Background(), client, "test-bucket", "ns/tbl", - entries, posDeletes, nil, nil, 0, + entries, posDeletes, nil, nil, 0, false, ) if err != nil { t.Fatalf("mergeParquetFiles: %v", err) @@ -1419,7 +1419,7 @@ func TestMergeParquetFilesWithEqualityDeletes(t *testing.T) { merged, count, err := mergeParquetFiles( context.Background(), client, "test-bucket", "ns/tbl", - entries, nil, eqGroups, schema, 0, + entries, nil, eqGroups, schema, 0, false, ) if err != nil { t.Fatalf("mergeParquetFiles: %v", err) @@ -1479,7 +1479,7 @@ func TestMergeParquetFilesDictionaryEncodedInput(t *testing.T) { merged, count, err := mergeParquetFiles( context.Background(), client, "test-bucket", "ns/tbl", - entries, nil, nil, nil, 0, + entries, nil, nil, nil, 0, false, ) if err != nil { t.Fatalf("mergeParquetFiles: %v", err) diff --git a/weed/worker/tasks/iceberg/operations.go b/weed/worker/tasks/iceberg/operations.go index 6a033f224..1ec2f9c18 100644 --- a/weed/worker/tasks/iceberg/operations.go +++ b/weed/worker/tasks/iceberg/operations.go @@ -566,8 +566,11 @@ func (h *Handler) rewriteManifests( } } + // v3 row lineage (row_lineage.go): the merged manifests are assigned first_row_id from the table's next-row-id; the + // data files keep the first_row_id they carry, so no row is renumbered + firstRowID := rowLineageFirstRowID(meta) var manifestListBuf bytes.Buffer - err = iceberg.WriteManifestList(version, &manifestListBuf, newSnapshotID, &snapshotID, &newSeqNum, 0, newManifests) + addedRows, err := writeManifestList(version, &manifestListBuf, newSnapshotID, &snapshotID, &newSeqNum, firstRowID, newManifests) if err != nil { return "", nil, fmt.Errorf("write manifest list: %w", err) } @@ -586,9 +589,9 @@ func (h *Handler) rewriteManifests( // Guard: verify table head hasn't advanced since we planned. // The merged manifest and manifest list were built against snapshotID; // if the head moved, they reference stale state. - cs := currentMeta.CurrentSnapshot() - if cs == nil || cs.SnapshotID != snapshotID { - return errStalePlan + cs, err := checkCommitPlan(currentMeta, snapshotID, version, firstRowID) + if err != nil { + return err } newSnapshot := &table.Snapshot{ @@ -604,6 +607,7 @@ func (h *Handler) rewriteManifests( return &id }(), } + setRowLineage(newSnapshot, version, firstRowID, addedRows) if err := builder.AddSnapshot(newSnapshot); err != nil { return err } diff --git a/weed/worker/tasks/iceberg/row_lineage.go b/weed/worker/tasks/iceberg/row_lineage.go new file mode 100644 index 000000000..48e725057 --- /dev/null +++ b/weed/worker/tasks/iceberg/row_lineage.go @@ -0,0 +1,393 @@ +package iceberg + +import ( + "bytes" + "context" + "fmt" + "io" + "reflect" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/parquet-go/parquet-go" +) + +// Row lineage (Iceberg v3, spec "Row Lineage" and "Snapshot Row IDs"). +// +// Every snapshot committed to a v3 table carries first-row-id, the table's next-row-id when the commit is attempted, +// and added-rows, the upper bound of the row IDs its manifest list assigns; committing advances next-row-id by +// added-rows. iceberg-go refuses a v3 snapshot without them ("invalid row lineage: first-row-id is required for v3 +// snapshots"), which is what every maintenance commit on a v3 table hit. The manifest list assigns first_row_id to the +// data manifests that have none, starting at the snapshot's first-row-id; passing 0 there (as the maintenance code did) +// would hand rows IDs other rows already hold. +// +// A rewrite that moves rows into a new data file (compaction) copies each row's _row_id and +// _last_updated_sequence_number into it: the value the file materializes, else the one inheritance gives it +// (the input file's first_row_id plus the row's position; the input entry's data sequence number). A compacted row keeps +// its identity, and a rewrite that modifies no row keeps its last-updated sequence number. + +const minFormatVersionRowLineage = 3 + +func hasRowLineage(version int) bool { return version >= minFormatVersionRowLineage } + +// rowLineageFirstRowID is the first-row-id a snapshot planned on meta starts at (0 below v3, where it is unused). +func rowLineageFirstRowID(meta table.Metadata) int64 { + if !hasRowLineage(meta.Version()) { + return 0 + } + return meta.NextRowID() +} + +// checkRowLineagePlan: a plan that assigned row IDs from next-row-id `planned` is stale once the table's next-row-id +// moved, even if the head did not (a commit to another branch advances it too); the spec has first-row-id reassigned on +// every attempt, which for a manifest list already written means planning again. +// +// A format-version change since planning (an upgrade to v3) is stale too: the manifests and the snapshot were built for +// the old version. +func checkRowLineagePlan(plannedVersion, currentVersion int, planned, current int64) error { + if currentVersion != plannedVersion { + return fmt.Errorf("%w: format version changed from %d to %d", errStalePlan, plannedVersion, currentVersion) + } + if hasRowLineage(plannedVersion) && planned != current { + return fmt.Errorf("%w: next-row-id moved from %d to %d", errStalePlan, planned, current) + } + return nil +} + +// checkCommitPlan is what every maintenance commit re-checks against the metadata it commits onto: the head it planned +// against, and the row-lineage state (format version, next-row-id) its manifest list was written for. +func checkCommitPlan(currentMeta table.Metadata, snapshotID int64, version int, firstRowID int64) (*table.Snapshot, error) { + cs := currentMeta.CurrentSnapshot() + if cs == nil || cs.SnapshotID != snapshotID { + return nil, errStalePlan + } + if err := checkRowLineagePlan(version, currentMeta.Version(), firstRowID, rowLineageFirstRowID(currentMeta)); err != nil { + return nil, err + } + return cs, nil +} + +// writeManifestList writes the snapshot's manifest list and returns its added-rows for a v3 table (nil below v3): the row +// IDs the writer assigned to manifests without a first_row_id, starting at firstRowID. +func writeManifestList(version int, out io.Writer, snapshotID int64, parentSnapshotID, sequenceNumber *int64, firstRowID int64, manifests []iceberg.ManifestFile) (*int64, error) { + if !hasRowLineage(version) { + return nil, iceberg.WriteManifestList(version, out, snapshotID, parentSnapshotID, sequenceNumber, firstRowID, manifests) + } + if sequenceNumber == nil { + return nil, fmt.Errorf("sequence number is required for v%d tables", version) + } + writer, err := iceberg.NewManifestListWriterV3(out, snapshotID, *sequenceNumber, firstRowID, parentSnapshotID) + if err != nil { + return nil, err + } + if err := writer.AddManifests(manifests); err != nil { + writer.Close() + return nil, err + } + if err := writer.Close(); err != nil { + return nil, err + } + next := writer.NextRowID() + if next == nil || *next < firstRowID { + return nil, fmt.Errorf("manifest list writer reported next-row-id %v below first-row-id %d", next, firstRowID) + } + addedRows := *next - firstRowID + return &addedRows, nil +} + +// setRowLineage puts first-row-id and added-rows on a snapshot of a v3 table. +func setRowLineage(snapshot *table.Snapshot, version int, firstRowID int64, addedRows *int64) { + if !hasRowLineage(version) || addedRows == nil { + return + } + first, added := firstRowID, *addedRows + snapshot.FirstRowID, snapshot.AddedRows = &first, &added +} + +// ---------------------------------------------------------------------------------------------- compaction rows + +// lineageSource is what inheritance gives the rows of one input data file. +type lineageSource struct { + firstRowID *int64 // the file's first_row_id; nil: its rows have no row ID to carry (a null is written) + dataSeq *int64 // the data sequence number of the file's manifest entry +} + +func lineageSourceOf(entry iceberg.ManifestEntry) lineageSource { + src := lineageSource{dataSeq: manifestEntrySeqNum(entry)} + if id := entry.DataFile().FirstRowID(); id != nil { + v := *id + src.firstRowID = &v + } + return src +} + +// lineageKind: 1 for _row_id, 2 for _last_updated_sequence_number, 0 for a data column. A column is a lineage column by +// its reserved field id, or by its reserved name when it has no field id; a reserved name under another field id is an +// error (kept as data, the output would have two columns of that name). +func lineageKind(f parquet.Field) (int, error) { + switch f.ID() { + case iceberg.RowIDFieldID: + return 1, nil + case iceberg.LastUpdatedSequenceNumberFieldID: + return 2, nil + } + named := 0 + switch f.Name() { + case iceberg.RowIDColumnName: + named = 1 + case iceberg.LastUpdatedSequenceNumberColumnName: + named = 2 + } + if named != 0 && f.ID() != 0 { + return 0, fmt.Errorf("column %s has field id %d, not the reserved id of that metadata column", f.Name(), f.ID()) + } + return named, nil +} + +// lineageLayout is the output of a v3 compaction: the reference file's own columns in their order (any lineage columns +// it has set aside), then _row_id and _last_updated_sequence_number, optional longs with the spec's field ids. +type lineageLayout struct { + schema *parquet.Schema + reference *parquet.Schema // the schema the layout was built from (the bin's first file) + baseFields []parquet.Field + baseOffsets []int // output leaf index of each base field's first leaf + baseLeaves int +} + +func newLineageLayout(reference *parquet.Schema) (*lineageLayout, error) { + var base []parquet.Field + var offsets []int + leaves := 0 + for _, f := range reference.Fields() { + kind, err := lineageKind(f) + if err != nil { + return nil, err + } + if kind != 0 { + continue + } + base = append(base, f) + offsets = append(offsets, leaves) + leaves += leafCount(f) + } + fields := append(append([]parquet.Field{}, base...), + lineageField{Node: parquet.FieldID(parquet.Optional(parquet.Leaf(parquet.Int64Type)), iceberg.RowIDFieldID), name: iceberg.RowIDColumnName}, + lineageField{Node: parquet.FieldID(parquet.Optional(parquet.Leaf(parquet.Int64Type)), iceberg.LastUpdatedSequenceNumberFieldID), name: iceberg.LastUpdatedSequenceNumberColumnName}, + ) + return &lineageLayout{ + schema: parquet.NewSchema(reference.Name(), &orderedGroup{Node: reference, fields: fields}), + reference: reference, + baseFields: base, + baseOffsets: offsets, + baseLeaves: leaves, + }, nil +} + +// mapperFor matches an input file to the layout: its non-lineage columns must be the reference's, in the same order +// (what the merge required of whole schemas before); its lineage columns, if any, are read wherever they are. +func (l *lineageLayout) mapperFor(input *parquet.Schema, src lineageSource) (*lineageMapper, error) { + m := &lineageMapper{rowIDLeaf: -1, seqLeaf: -1, outRowID: l.baseLeaves, outSeq: l.baseLeaves + 1, src: src} + next, base := 0, 0 + for _, f := range input.Fields() { + n := leafCount(f) + kind, err := lineageKind(f) + if err != nil { + return nil, err + } + if kind != 0 { + if n != 1 || !f.Leaf() { + return nil, fmt.Errorf("lineage column %s is not a primitive", f.Name()) + } + target := &m.rowIDLeaf + if kind == 2 { + target = &m.seqLeaf + } + if *target >= 0 { + return nil, fmt.Errorf("more than one %s column", f.Name()) + } + *target = next + m.leafMap = append(m.leafMap, -1) + next++ + continue + } + if base >= len(l.baseFields) || f.Name() != l.baseFields[base].Name() || !parquet.EqualNodes(f, l.baseFields[base]) { + return nil, fmt.Errorf("cannot merge files with different schemas") + } + for i := 0; i < n; i++ { + m.leafMap = append(m.leafMap, l.baseOffsets[base]+i) + } + next += n + base++ + } + if base != len(l.baseFields) { + return nil, fmt.Errorf("cannot merge files with different schemas") + } + return m, nil +} + +// lineageMapper turns one input file's rows into the layout's rows. +type lineageMapper struct { + leafMap []int // input leaf index -> output leaf index; -1 for an input lineage column + rowIDLeaf int // input leaf index of _row_id, -1 when the file has none + seqLeaf int // input leaf index of _last_updated_sequence_number, -1 when the file has none + outRowID, outSeq int + src lineageSource + rows []parquet.Row // reused per batch: both writers copy the values they are handed + values []parquet.Value // backing store of rows +} + +// mapRows maps rows read at the given positions of the input file (their position before any delete applied). +func (m *lineageMapper) mapRows(rows []parquet.Row, positions []int64) []parquet.Row { + need := 0 + for _, row := range rows { + need += len(row) + 2 + } + if cap(m.values) < need { + m.values = make([]parquet.Value, 0, need) + } + m.values = m.values[:0] + m.rows = m.rows[:0] + for i, row := range rows { + start := len(m.values) + mapped := m.values[start:start:cap(m.values)] + var rowID, seq *parquet.Value + for j := range row { + v := row[j] + switch c := v.Column(); { + case c == m.rowIDLeaf: + rowID = &row[j] + case c == m.seqLeaf: + seq = &row[j] + default: + mapped = append(mapped, v.Level(v.RepetitionLevel(), v.DefinitionLevel(), m.leafMap[c])) + } + } + var inheritedRowID *int64 + if m.src.firstRowID != nil { + id := *m.src.firstRowID + positions[i] + inheritedRowID = &id + } + mapped = append(mapped, lineageValue(rowID, inheritedRowID, m.outRowID), lineageValue(seq, m.src.dataSeq, m.outSeq)) + m.values = m.values[:start+len(mapped)] + m.rows = append(m.rows, parquet.Row(mapped)) + } + return m.rows +} + +// lineageValue: the value the file materializes when it is not null, else the inherited one, else a null. +func lineageValue(materialized *parquet.Value, inherited *int64, column int) parquet.Value { + if materialized != nil && !materialized.IsNull() { + return parquet.Int64Value(materialized.Int64()).Level(0, 1, column) + } + if inherited != nil { + return parquet.Int64Value(*inherited).Level(0, 1, column) + } + return parquet.NullValue().Level(0, 0, column) +} + +func leafCount(n parquet.Node) int { + if n.Leaf() { + return 1 + } + total := 0 + for _, f := range n.Fields() { + total += leafCount(f) + } + return total +} + +// orderedGroup is a group node with the fields given, in that order (parquet.Group sorts its fields by name). +type orderedGroup struct { + parquet.Node + fields []parquet.Field +} + +func (g *orderedGroup) Fields() []parquet.Field { return g.fields } +func (g *orderedGroup) Leaf() bool { return false } + +// GoType: rows are written as parquet.Row, never from Go values, so the base's type is never read for these columns. +func (g *orderedGroup) GoType() reflect.Type { return g.Node.GoType() } + +// lineageField names a lineage column in the layout. +type lineageField struct { + parquet.Node + name string +} + +func (f lineageField) Name() string { return f.name } +func (f lineageField) Value(base reflect.Value) reflect.Value { return reflect.Value{} } + +// prepareLineageInput: below v3 (layout nil) the file is merged as it is, with the equality-delete columns resolved on +// the first file. On v3 the file is mapped onto the layout, and its equality-delete columns are resolved on its own +// schema, since a file's lineage columns may sit anywhere among its columns. +func prepareLineageInput(layout *lineageLayout, reader *parquet.Reader, entry iceberg.ManifestEntry, firstFileEqGroups []resolvedEqDeleteGroup, + eqDeleteGroups []equalityDeleteGroup, icebergSchema *iceberg.Schema) (*lineageMapper, []resolvedEqDeleteGroup, error) { + if layout == nil { + return nil, firstFileEqGroups, nil + } + source := entry.DataFile().FilePath() + mapper, err := layout.mapperFor(reader.Schema(), lineageSourceOf(entry)) + if err != nil { + return nil, nil, fmt.Errorf("schema mismatch in %s: %w", source, err) + } + if reader.Schema() == layout.reference { + return mapper, firstFileEqGroups, nil // resolved on this very schema already + } + eqGroups, err := resolveEqualityDeleteGroupsForSchema(reader.Schema(), eqDeleteGroups, icebergSchema) + if err != nil { + return nil, nil, fmt.Errorf("resolve equality columns in %s: %w", source, err) + } + return mapper, eqGroups, nil +} + +// copyFileRows drains one input file into write: deletes filtered and, on a v3 table (layout set), mapped onto the +// layout with each row's lineage. Both merges, bin-packed and sorted, use it. +func copyFileRows(ctx context.Context, reader *parquet.Reader, entry iceberg.ManifestEntry, layout *lineageLayout, bucketName, dataPath string, + positionDeletes map[string][]int64, firstFileEqGroups []resolvedEqDeleteGroup, eqDeleteGroups []equalityDeleteGroup, + icebergSchema *iceberg.Schema, write func([]parquet.Row) error) (int64, error) { + mapper, eqGroups, err := prepareLineageInput(layout, reader, entry, firstFileEqGroups, eqDeleteGroups, icebergSchema) + if err != nil { + reader.Close() + return 0, err + } + return visitFilteredParquetRows(ctx, reader, entry.DataFile().FilePath(), bucketName, dataPath, positionDeletes, eqGroups, func(filtered []parquet.Row, positions []int64) error { + if mapper != nil { + filtered = mapper.mapRows(filtered, positions) + } + return write(filtered) + }) +} + +// mergedFileFirstRowID is the _row_id of the merged output's first row, or nil when it carries none: the value the +// manifest entry's first_row_id must name, since rewritten rows keep their IDs rather than taking fresh ones. +func mergedFileFirstRowID(data []byte) (*int64, error) { + reader := parquet.NewReader(bytes.NewReader(data)) + defer reader.Close() + schema := reader.Schema() + if schema == nil { + return nil, nil + } + leaf, ok := schema.Lookup(iceberg.RowIDColumnName) + if !ok { + return nil, nil + } + rows := make([]parquet.Row, 1) + n, err := reader.ReadRows(rows) + if err != nil && err != io.EOF { + return nil, err + } + if n == 0 { + return nil, nil + } + for _, v := range rows[0] { + if v.Column() != leaf.ColumnIndex { + continue + } + if v.IsNull() { + return nil, nil + } + id := v.Int64() + return &id, nil + } + return nil, nil +} diff --git a/weed/worker/tasks/iceberg/row_lineage_test.go b/weed/worker/tasks/iceberg/row_lineage_test.go new file mode 100644 index 000000000..83bc1bed1 --- /dev/null +++ b/weed/worker/tasks/iceberg/row_lineage_test.go @@ -0,0 +1,384 @@ +package iceberg + +import ( + "bytes" + "context" + "io" + "strings" + "testing" + + "github.com/apache/iceberg-go" + "github.com/parquet-go/parquet-go" + + filer_pb "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +type lineageDataRow struct { + ID int64 `parquet:"id"` + Name string `parquet:"name"` +} + +func putDataFileRows[T any](t *testing.T, fs *fakeFilerServer, name string, rows []T) { + t.Helper() + var buf bytes.Buffer + w := parquet.NewWriter(&buf, parquet.SchemaOf(new(T))) + for _, r := range rows { + if err := w.Write(&r); err != nil { + t.Fatalf("write row: %v", err) + } + } + if err := w.Close(); err != nil { + t.Fatalf("close: %v", err) + } + fs.putEntry("/buckets/test-bucket/ns/tbl/data", name, &filer_pb.Entry{ + Name: name, + Content: buf.Bytes(), + }) +} + +func lineageEntry(t *testing.T, path string, firstRowID, dataSeq, recordCount int64) iceberg.ManifestEntry { + t.Helper() + dfb, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, iceberg.EntryContentData, + path, iceberg.ParquetFile, map[int]any{}, nil, nil, recordCount, 1024) + if err != nil { + t.Fatalf("build data file: %v", err) + } + snapID := int64(1) + return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &dataSeq, nil, + dfb.FirstRowID(firstRowID).Build()) +} + +// mergedColumns returns each merged row's (id, name, _row_id, +// _last_updated_sequence_number): the two lineage columns the v3 merge +// appends after the data columns. +func mergedColumns(t *testing.T, data []byte) (ids, names, rowIDs, seqs []any) { + t.Helper() + f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data))) + if err != nil { + t.Fatalf("open merged: %v", err) + } + fields := f.Schema().Fields() + if len(fields) != 4 { + t.Fatalf("merged schema has %d fields, want 4 (id, name, _row_id, _last_updated_sequence_number)", len(fields)) + } + if fields[2].Name() != iceberg.RowIDColumnName || fields[2].ID() != iceberg.RowIDFieldID { + t.Fatalf("field 2 = %s id %d, want %s id %d", fields[2].Name(), fields[2].ID(), iceberg.RowIDColumnName, iceberg.RowIDFieldID) + } + if fields[3].Name() != iceberg.LastUpdatedSequenceNumberColumnName || fields[3].ID() != iceberg.LastUpdatedSequenceNumberFieldID { + t.Fatalf("field 3 = %s id %d, want %s id %d", fields[3].Name(), fields[3].ID(), iceberg.LastUpdatedSequenceNumberColumnName, iceberg.LastUpdatedSequenceNumberFieldID) + } + for _, rg := range f.RowGroups() { + rows := rg.Rows() + buf := make([]parquet.Row, 8) + for { + n, err := rows.ReadRows(buf) + for _, row := range buf[:n] { + ids = append(ids, row[0].Int64()) + names = append(names, row[1].String()) + if row[2].IsNull() { + rowIDs = append(rowIDs, nil) + } else { + rowIDs = append(rowIDs, row[2].Int64()) + } + if row[3].IsNull() { + seqs = append(seqs, nil) + } else { + seqs = append(seqs, row[3].Int64()) + } + } + if err == io.EOF { + break + } + if err != nil { + t.Fatalf("read merged rows: %v", err) + } + } + rows.Close() + } + return +} + +// On a v3 table the merged file carries every row's identity: the +// materialized value when the input has it, else the one inheritance +// gives it (the input file's first_row_id plus the row's position, the +// entry's data sequence number). +func TestMergeParquetFilesRowLineage(t *testing.T) { + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "file1.parquet", []lineageDataRow{{1, "alice"}, {2, "bob"}, {3, "carl"}}) + putDataFileRows(t, fs, "file2.parquet", []lineageDataRow{{4, "dave"}, {5, "eve"}}) + + entries := []iceberg.ManifestEntry{ + lineageEntry(t, "data/file1.parquet", 100, 7, 3), + lineageEntry(t, "data/file2.parquet", 200, 9, 2), + } + + merged, count, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl", + entries, nil, nil, nil, 0, true) + if err != nil { + t.Fatalf("mergeParquetFiles: %v", err) + } + if count != 5 { + t.Fatalf("expected 5 merged rows, got %d", count) + } + + _, _, rowIDs, seqs := mergedColumns(t, merged) + wantIDs := []any{int64(100), int64(101), int64(102), int64(200), int64(201)} + wantSeqs := []any{int64(7), int64(7), int64(7), int64(9), int64(9)} + for i := range wantIDs { + if rowIDs[i] != wantIDs[i] { + t.Errorf("row %d _row_id = %v, want %v", i, rowIDs[i], wantIDs[i]) + } + if seqs[i] != wantSeqs[i] { + t.Errorf("row %d _last_updated_sequence_number = %v, want %v", i, seqs[i], wantSeqs[i]) + } + } +} + +// A position delete removes a row without renumbering the survivors: +// each kept row's _row_id is the file's first_row_id plus its position +// before the delete. +func TestMergeParquetFilesRowLineagePositionDelete(t *testing.T) { + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "file1.parquet", []lineageDataRow{{1, "alice"}, {2, "bob"}, {3, "carl"}}) + + entries := []iceberg.ManifestEntry{lineageEntry(t, "data/file1.parquet", 100, 7, 3)} + posDeletes := map[string][]int64{"ns/tbl/data/file1.parquet": {1}} + + merged, count, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl", + entries, posDeletes, nil, nil, 0, true) + if err != nil { + t.Fatalf("mergeParquetFiles: %v", err) + } + if count != 2 { + t.Fatalf("expected 2 merged rows, got %d", count) + } + ids, _, rowIDs, _ := mergedColumns(t, merged) + if ids[0] != int64(1) || ids[1] != int64(3) { + t.Fatalf("merged ids = %v, want [1 3]", ids) + } + if rowIDs[0] != int64(100) || rowIDs[1] != int64(102) { + t.Fatalf("merged _row_id = %v, want [100 102]", rowIDs) + } +} + +// A file an engine already wrote with lineage columns materializes its +// rows' values; merged with a plain file, each input keeps its own +// provenance. +func TestMergeParquetFilesRowLineageMaterialized(t *testing.T) { + fs, client := startFakeFiler(t) + + type materializedRow struct { + ID int64 `parquet:"id"` + Name string `parquet:"name"` + RowID int64 `parquet:"_row_id,id(2147483540)"` + LastUp int64 `parquet:"_last_updated_sequence_number,id(2147483539)"` + } + putDataFileRows(t, fs, "plain.parquet", []lineageDataRow{{1, "alice"}}) + putDataFileRows(t, fs, "mats.parquet", []materializedRow{{2, "bob", 5000, 42}}) + + entries := []iceberg.ManifestEntry{ + lineageEntry(t, "data/plain.parquet", 100, 7, 1), + lineageEntry(t, "data/mats.parquet", 200, 9, 1), + } + merged, count, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl", + entries, nil, nil, nil, 0, true) + if err != nil { + t.Fatalf("mergeParquetFiles: %v", err) + } + if count != 2 { + t.Fatalf("expected 2 merged rows, got %d", count) + } + _, _, rowIDs, seqs := mergedColumns(t, merged) + if rowIDs[0] != int64(100) || rowIDs[1] != int64(5000) { + t.Errorf("_row_id = %v, want [100 5000]", rowIDs) + } + if seqs[0] != int64(7) || seqs[1] != int64(42) { + t.Errorf("_last_updated_sequence_number = %v, want [7 42]", seqs) + } +} + +// The v3 manifest list assigns first_row_id to manifests that have +// none, starting at the snapshot's first-row-id, and reports the range +// it assigned as the snapshot's added-rows. +func TestWriteManifestListV3RowIDs(t *testing.T) { + schema := newTestSchema() + spec := *iceberg.UnpartitionedSpec + snapID := int64(12) + dataSeq := int64(3) + + entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &dataSeq, nil, + mustDataFile(t, spec, "data/a.parquet", 4)) + var manifestBuf bytes.Buffer + mf, err := iceberg.WriteManifest("metadata/m0.avro", &manifestBuf, 3, spec, schema, snapID, + []iceberg.ManifestEntry{entry}) + if err != nil { + t.Fatalf("write manifest: %v", err) + } + + var listBuf bytes.Buffer + parent := int64(10) + seqNum := int64(3) + addedRows, err := writeManifestList(3, &listBuf, snapID, &parent, &seqNum, 500, []iceberg.ManifestFile{mf}) + if err != nil { + t.Fatalf("writeManifestList: %v", err) + } + if addedRows == nil || *addedRows != 4 { + t.Fatalf("addedRows = %v, want 4", addedRows) + } + + readBack, err := iceberg.ReadManifestList(&listBuf) + if err != nil { + t.Fatalf("read manifest list: %v", err) + } + if len(readBack) != 1 || readBack[0].FirstRowID() == nil || *readBack[0].FirstRowID() != 500 { + t.Fatalf("manifest first_row_id = %v, want 500", readBack) + } +} + +func mustDataFile(t *testing.T, spec iceberg.PartitionSpec, path string, recordCount int64) iceberg.DataFile { + t.Helper() + dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, path, iceberg.ParquetFile, + map[int]any{}, nil, nil, recordCount, 1024) + if err != nil { + t.Fatalf("build data file: %v", err) + } + return dfb.Build() +} + +// A plan that assigned row IDs from a next-row-id the table no longer +// has is stale, like a moved head; so is a format version that changed +// since planning. +func TestCheckRowLineagePlan(t *testing.T) { + if err := checkRowLineagePlan(3, 2, 0, 0); err == nil { + t.Error("version change must be stale") + } + if err := checkRowLineagePlan(3, 3, 100, 200); err == nil { + t.Error("moved next-row-id must be stale") + } + if err := checkRowLineagePlan(2, 2, 0, 0); err != nil { + t.Errorf("v2 plan is never row-lineage stale: %v", err) + } + if err := checkRowLineagePlan(3, 3, 100, 100); err != nil { + t.Errorf("unchanged next-row-id must not be stale: %v", err) + } +} + +// A data column named like a lineage column under a different field id +// refuses the bin rather than writing a doubled column. +func TestLineageLayoutRejectsReservedNameUnderOtherID(t *testing.T) { + type badRow struct { + ID int64 `parquet:"id"` + RowID int64 `parquet:"_row_id,id(7)"` + } + schema := parquet.SchemaOf(new(badRow)) + if _, err := newLineageLayout(schema); err == nil { + t.Fatal("reserved name under a non-reserved field id must fail the layout") + } +} + +// Compaction on a v3 table reads row lineage the way a reader does: the +// entries in the stored manifest carry no first_row_id, so the ones +// compactDataFiles merges are the inherited values (the manifest-list +// first_row_id plus the preceding files' record counts). The merged file +// materializes those IDs and its new entry names its own first row's. +func TestCompactDataFilesV3RowLineageManifestRead(t *testing.T) { + fs, client := startFakeFiler(t) + ctx := context.Background() + + setup := tableSetup{BucketName: "test-bucket", Namespace: "ns", TableName: "tbl"} + spec := *iceberg.UnpartitionedSpec + snapID := int64(1) + + putDataFileRows(t, fs, "f1.parquet", []lineageDataRow{{1, "a"}, {2, "b"}, {3, "c"}}) + putDataFileRows(t, fs, "f2.parquet", []lineageDataRow{{4, "d"}, {5, "e"}}) + + entries := []iceberg.ManifestEntry{ + iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, + mustDataFile(t, spec, "data/f1.parquet", 3)), + iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, + mustDataFile(t, spec, "data/f2.parquet", 2)), + } + writeV3Table(t, fs, setup, entries, nil, 0, 5) + + handler := NewHandler(nil) + config := Config{ + TargetFileSizeBytes: 4096, + MinInputFiles: 2, + ApplyDeletes: true, + } + result, _, err := handler.compactDataFiles(ctx, client, setup.BucketName, setup.tablePath(), config, nil) + if err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + if !strings.Contains(result, "compacted 2 files into 1") { + t.Fatalf("unexpected result: %q", result) + } + + state, err := loadCurrentMetadata(ctx, client, setup.BucketName, setup.tablePath()) + if err != nil { + t.Fatalf("loadCurrentMetadata: %v", err) + } + newSnap := state.Metadata.CurrentSnapshot() + if newSnap == nil || newSnap.SnapshotID == snapID { + t.Fatalf("no new snapshot committed: %+v", newSnap) + } + if newSnap.FirstRowID == nil || *newSnap.FirstRowID != 5 { + t.Fatalf("new snapshot first-row-id = %v, want 5", newSnap.FirstRowID) + } + + manifestListData, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, newSnap.ManifestList) + if err != nil { + t.Fatalf("read new manifest list: %v", err) + } + manifests, err := s3tables.ReadManifestList(manifestListData) + if err != nil { + t.Fatalf("parse new manifest list: %v", err) + } + if len(manifests) != 1 { + t.Fatalf("new snapshot has %d manifests, want 1", len(manifests)) + } + manifestData, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, manifests[0].FilePath()) + if err != nil { + t.Fatalf("read new manifest: %v", err) + } + newEntries, err := s3tables.ReadManifest(manifests[0], manifestData, false, specByID(state.Metadata), state.Metadata.CurrentSchema()) + if err != nil { + t.Fatalf("parse new manifest: %v", err) + } + + var merged iceberg.DataFile + var deletedPaths int + for _, e := range newEntries { + switch e.Status() { + case iceberg.EntryStatusADDED: + merged = e.DataFile() + case iceberg.EntryStatusDELETED: + deletedPaths++ + } + } + if deletedPaths != 2 { + t.Fatalf("new manifest has %d deleted entries, want 2", deletedPaths) + } + if merged == nil { + t.Fatal("new manifest has no added entry") + } + if merged.FirstRowID() == nil || *merged.FirstRowID() != 0 { + t.Fatalf("merged entry first_row_id = %v, want 0", merged.FirstRowID()) + } + + mergedData, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, merged.FilePath()) + if err != nil { + t.Fatalf("read merged file: %v", err) + } + _, _, rowIDs, seqs := mergedColumns(t, mergedData) + wantRowIDs := []any{int64(0), int64(1), int64(2), int64(3), int64(4)} + for i, want := range wantRowIDs { + if rowIDs[i] != want { + t.Fatalf("row %d _row_id = %v, want %v", i, rowIDs[i], want) + } + if seqs[i] != int64(1) { + t.Fatalf("row %d _last_updated_sequence_number = %v, want 1", i, seqs[i]) + } + } +}