From bf62caf4f56fb33cfeabaeee4b0ade9a4c4a7788 Mon Sep 17 00:00:00 2001 From: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:05:23 -0700 Subject: [PATCH] iceberg maintenance: read only the parquet footer choosing a bin's writer schema The union-schema pass downloaded every input in full, then the merge downloaded each again. The footer suffices for the schema, and a chunked file's ReadAt pulls only the chunks the footer spans. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/worker/tasks/iceberg/compact.go | 18 ++++----- weed/worker/tasks/iceberg/filer_io.go | 58 +++++++++++++++++++++++---- 2 files changed, 58 insertions(+), 18 deletions(-) diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go index fc0fc24188..e63e9e4a4a 100644 --- a/weed/worker/tasks/iceberg/compact.go +++ b/weed/worker/tasks/iceberg/compact.go @@ -1145,10 +1145,11 @@ func mergeParquetFiles( } // binWriterSchema is the writer schema for a v1/v2 bin-pack merge: the union -// of the files' schemas by field id. Files whose schemas cannot union (no -// field ids, nested or repeated columns, a field id or name carrying -// different types) merge only under the old rule — every file's schema equal -// to the first — and the bin is refused as before otherwise. +// of the files' schemas by field id. Only each file's footer is read to learn +// its schema; the merge then downloads each file once. Files whose schemas +// cannot union (no field ids, nested or repeated columns, a field id or name +// carrying different types) merge only under the old rule — every file's +// schema equal to the first — and the bin is refused as before otherwise. func binWriterSchema( ctx context.Context, filerClient filer_pb.SeaweedFilerClient, @@ -1160,15 +1161,10 @@ func binWriterSchema( unionable, allEqual := true, true for _, entry := range entries { source := entry.DataFile().FilePath() - data, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, source) + schema, err := parquetSchemaByIcebergPath(ctx, filerClient, bucketName, dataPath, source) if err != nil { - return nil, fmt.Errorf("read parquet file %s: %w", source, err) + return nil, fmt.Errorf("schema of %s: %w", source, err) } - f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data))) - if err != nil { - return nil, fmt.Errorf("open parquet file %s: %w", source, err) - } - schema := f.Schema() if schema == nil { return nil, fmt.Errorf("no parquet schema found in %s", source) } diff --git a/weed/worker/tasks/iceberg/filer_io.go b/weed/worker/tasks/iceberg/filer_io.go index a1c8e1c5be..482c8dc80a 100644 --- a/weed/worker/tasks/iceberg/filer_io.go +++ b/weed/worker/tasks/iceberg/filer_io.go @@ -1,6 +1,7 @@ package iceberg import ( + "bytes" "context" "crypto/rand" "encoding/hex" @@ -14,6 +15,7 @@ import ( "sync" "time" + "github.com/parquet-go/parquet-go" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/operation" @@ -160,9 +162,9 @@ func loadCurrentMetadata(ctx context.Context, client filer_pb.SeaweedFilerClient return state, nil } -// loadFileByIcebergPath loads a file from the filer given an Iceberg-style path. -// dataPath is the bucket-relative directory the table's files live in. -func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) ([]byte, error) { +// lookupFileByIcebergPath returns the filer entry holding an Iceberg-style +// path. dataPath is the bucket-relative directory the table's files live in. +func lookupFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) (*filer_pb.Entry, error) { fullPath, err := icebergFilerPath(bucketName, dataPath, icebergPath) if err != nil { return nil, err @@ -179,23 +181,65 @@ func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClie if resp == nil || resp.Entry == nil { return nil, fmt.Errorf("file not found: %s/%s", dir, fileName) } + return resp.Entry, nil +} - if len(resp.Entry.Content) > 0 || len(resp.Entry.Chunks) == 0 { - return resp.Entry.Content, nil +// loadFileByIcebergPath loads a file from the filer given an Iceberg-style path. +func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) ([]byte, error) { + entry, err := lookupFileByIcebergPath(ctx, client, bucketName, dataPath, icebergPath) + if err != nil { + return nil, err + } + + if len(entry.Content) > 0 || len(entry.Chunks) == 0 { + return entry.Content, nil } initGlobalHTTPClientOnce.Do(util_http.InitGlobalHttpClient) - reader := filer.NewFileReader(singleFilerClient{client: client}, resp.Entry) + reader := filer.NewFileReader(singleFilerClient{client: client}, entry) if closer, ok := reader.(io.Closer); ok { defer closer.Close() } data, err := io.ReadAll(reader) if err != nil { - return nil, fmt.Errorf("read chunked file %s/%s: %w", dir, fileName, err) + return nil, fmt.Errorf("read chunked file %s: %w", icebergPath, err) } return data, nil } +// parquetSchemaByIcebergPath reads a file's parquet schema from its footer, +// so the file's body is never fetched: a chunked file's ReadAt pulls only the +// chunks the footer spans. +func parquetSchemaByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) (*parquet.Schema, error) { + entry, err := lookupFileByIcebergPath(ctx, client, bucketName, dataPath, icebergPath) + if err != nil { + return nil, err + } + + var readerAt io.ReaderAt + size := int64(entry.GetAttributes().GetFileSize()) + if len(entry.Content) > 0 || len(entry.Chunks) == 0 { + readerAt = bytes.NewReader(entry.Content) + size = int64(len(entry.Content)) + } else { + initGlobalHTTPClientOnce.Do(util_http.InitGlobalHttpClient) + reader := filer.NewChunkStreamReader(singleFilerClient{client: client}, entry.Chunks) + defer reader.Close() + readerAt = reader + if size <= 0 { + if size, err = reader.Seek(0, io.SeekEnd); err != nil { + return nil, fmt.Errorf("size of %s: %w", icebergPath, err) + } + } + } + + f, err := parquet.OpenFile(readerAt, size) + if err != nil { + return nil, fmt.Errorf("open parquet file %s: %w", icebergPath, err) + } + return f.Schema(), nil +} + // normalizeIcebergPath converts an Iceberg path (an S3 URL, an absolute filer // path, or a path relative to the table root) into a path relative to the // bucket root, e.g. "ns/table/metadata/snap-1.avro". Absolute references are