From b2eefb8a73b8a44c7fcd640bd008ed247148f5f6 Mon Sep 17 00:00:00 2001 From: Martijn Hoekstra Date: Sat, 10 Oct 2026 17:39:29 +0200 Subject: [PATCH] filer: fail the request, not the process, when chunks cannot be resolved (#11685) * filer: fail the request, not the process, when chunks cannot be resolved CompactFileChunks and ViewFromChunks discarded the error from NonOverlappingVisibleIntervals and iterated the nil interval list it returns on failure. A CreateEntry or UpdateEntry whose context was cancelled therefore panicked in SeparateGarbageChunks and took the whole filer down, losing the metadata log it had not yet flushed. Both functions now return the error. cleanupChunks fails the request with it, the mount flush paths return it, the replication sinks and the S3 SSE range reader return it, and a ChunkStreamReader that cannot build its view fails every read with it and reports it as its SourceError instead of reading as an empty file. Fixes #11682 Co-Authored-By: Claude Opus 5.5 (1M context) * filer: propagate chunk resolution failures in scan paths NonOverlappingVisibleIntervals can fail without a cancelled context; the query engine and log-store callers discarded the error and dereferenced nil intervals. * query: propagate chunk resolution errors instead of skipping files When a manifest cannot be resolved, ReadParquetStatistics, countLiveLogRowsExcludingParquetSources, and computeLiveLogMinMax logged a warning and skipped the failed file, so queries returned understated counts and wrong min/max values. Surface the errors instead: the fast-path aggregation collector now returns a DataSourceError so the engine falls back to a full scan, and the row-count helpers return the error rather than a partial total. --------- Co-authored-by: Claude Opus 5.5 (1M context) Co-authored-by: Chris Lu Co-authored-by: Chris Lu --- weed/filer/filechunk_manifest_test.go | 10 +++- weed/filer/filechunks.go | 22 +++++-- weed/filer/filechunks2_test.go | 5 +- weed/filer/filechunks_cancel_test.go | 58 +++++++++++++++++++ weed/filer/filechunks_test.go | 24 ++++++-- weed/filer/stream.go | 15 ++++- weed/mount/weedfs_file_sync.go | 5 +- weed/mount/weedfs_file_sync_test.go | 30 ++++++++-- weed/mount/weedfs_metadata_flush.go | 6 +- weed/mq/logstore/read_parquet_to_log.go | 5 +- weed/query/engine/aggregations.go | 14 +++-- weed/query/engine/engine.go | 54 +++++++++-------- weed/query/engine/hybrid_message_scanner.go | 9 +-- weed/replication/sink/azuresink/azure_sink.go | 7 ++- weed/replication/sink/b2sink/b2_sink.go | 5 +- weed/replication/sink/gcssink/gcs_sink.go | 5 +- weed/replication/sink/localsink/local_sink.go | 5 +- weed/s3api/s3api_object_handlers.go | 5 +- weed/server/filer_grpc_server.go | 5 +- .../filer_grpc_server_cleanup_chunks_test.go | 47 +++++++++++++++ 20 files changed, 271 insertions(+), 65 deletions(-) create mode 100644 weed/filer/filechunks_cancel_test.go create mode 100644 weed/server/filer_grpc_server_cleanup_chunks_test.go diff --git a/weed/filer/filechunk_manifest_test.go b/weed/filer/filechunk_manifest_test.go index 0cec6c8d1..e23bf5366 100644 --- a/weed/filer/filechunk_manifest_test.go +++ b/weed/filer/filechunk_manifest_test.go @@ -338,7 +338,10 @@ func TestCompactResolvedOverlappingManifests(t *testing.T) { } // CompactFileChunks on resolved sub-chunks (nil lookupFn is fine for non-manifest chunks) - compacted, garbage := CompactFileChunks(context.Background(), nil, dataChunks) + compacted, garbage, err := CompactFileChunks(context.Background(), nil, dataChunks) + if err != nil { + t.Fatal(err) + } if len(compacted) != 3 { t.Fatalf("expected 3 compacted chunks, got %d", len(compacted)) @@ -487,7 +490,10 @@ func TestCompactMultipleOverlappingManifestGenerations(t *testing.T) { } // Compact: only generation 5 (the newest 3 chunks) should survive - compacted, garbage := CompactFileChunks(context.Background(), nil, dataChunks) + compacted, garbage, err := CompactFileChunks(context.Background(), nil, dataChunks) + if err != nil { + t.Fatal(err) + } if len(compacted) != 3 { t.Errorf("expected 3 compacted (gen 5), got %d", len(compacted)) } diff --git a/weed/filer/filechunks.go b/weed/filer/filechunks.go index 954547296..3a61ed3b5 100644 --- a/weed/filer/filechunks.go +++ b/weed/filer/filechunks.go @@ -63,9 +63,15 @@ func ETagChunks(chunks []*filer_pb.FileChunk) (etag string) { return finalETag } -func CompactFileChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk) (compacted, garbage []*filer_pb.FileChunk) { +// CompactFileChunks splits chunks into those still visible and those fully +// covered by newer writes. It fails if the chunk manifests cannot be resolved, +// including when ctx is cancelled. +func CompactFileChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk) (compacted, garbage []*filer_pb.FileChunk, err error) { - visibles, _ := NonOverlappingVisibleIntervals(ctx, lookupFileIdFn, chunks, 0, math.MaxInt64) + visibles, err := NonOverlappingVisibleIntervals(ctx, lookupFileIdFn, chunks, 0, math.MaxInt64) + if err != nil { + return nil, nil, err + } compacted, garbage = SeparateGarbageChunks(visibles, chunks) @@ -194,11 +200,17 @@ func (cv *ChunkView) CanRangeFetch() bool { return cv.CipherKey == nil && !cv.IsGzipped } -func ViewFromChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, offset int64, size int64) (chunkViews *IntervalList[*ChunkView]) { +// ViewFromChunks returns the views of chunks that cover [offset, offset+size). +// It fails if the chunk manifests cannot be resolved, including when ctx is +// cancelled. +func ViewFromChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, offset int64, size int64) (chunkViews *IntervalList[*ChunkView], err error) { - visibles, _ := NonOverlappingVisibleIntervals(ctx, lookupFileIdFn, chunks, offset, offset+size) + visibles, err := NonOverlappingVisibleIntervals(ctx, lookupFileIdFn, chunks, offset, offset+size) + if err != nil { + return nil, err + } - return ViewFromVisibleIntervals(visibles, offset, size) + return ViewFromVisibleIntervals(visibles, offset, size), nil } diff --git a/weed/filer/filechunks2_test.go b/weed/filer/filechunks2_test.go index 9c8a916e6..e8b1f39d2 100644 --- a/weed/filer/filechunks2_test.go +++ b/weed/filer/filechunks2_test.go @@ -66,7 +66,10 @@ func TestCompactFileChunksRealCase(t *testing.T) { printChunks("before", chunks) - compacted, garbage := CompactFileChunks(context.Background(), nil, chunks) + compacted, garbage, err := CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } printChunks("compacted", compacted) printChunks("garbage", garbage) diff --git a/weed/filer/filechunks_cancel_test.go b/weed/filer/filechunks_cancel_test.go new file mode 100644 index 000000000..689f9478d --- /dev/null +++ b/weed/filer/filechunks_cancel_test.go @@ -0,0 +1,58 @@ +package filer + +import ( + "context" + "errors" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// cancelledChunks returns a cancelled context and plain data chunks. Resolving +// chunks under a cancelled context fails even without manifests, so these +// exercise the error path with no lookup function. +func cancelledChunks() (context.Context, []*filer_pb.FileChunk) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + return ctx, []*filer_pb.FileChunk{ + {FileId: "1,01", Offset: 0, Size: 100, ModifiedTsNs: 1}, + {FileId: "1,02", Offset: 0, Size: 100, ModifiedTsNs: 2}, + } +} + +func TestCompactFileChunksCancelledContext(t *testing.T) { + ctx, chunks := cancelledChunks() + + compacted, garbage, err := CompactFileChunks(ctx, nil, chunks) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + if compacted != nil || garbage != nil { + t.Fatalf("compacted = %v, garbage = %v, want both nil on error", compacted, garbage) + } +} + +func TestViewFromChunksCancelledContext(t *testing.T) { + ctx, chunks := cancelledChunks() + + views, err := ViewFromChunks(ctx, nil, chunks, 0, 100) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + if views != nil { + t.Fatalf("views = %v, want nil on error", views) + } +} + +func TestChunkStreamReaderCancelledContext(t *testing.T) { + ctx, chunks := cancelledChunks() + + reader := NewChunkStreamReaderFromLookup(ctx, nil, chunks) + n, err := reader.Read(make([]byte, 10)) + if n != 0 || !errors.Is(err, context.Canceled) { + t.Fatalf("Read = %d, %v, want 0, context.Canceled", n, err) + } + if !errors.Is(reader.SourceError(), context.Canceled) { + t.Fatalf("SourceError = %v, want context.Canceled", reader.SourceError()) + } +} diff --git a/weed/filer/filechunks_test.go b/weed/filer/filechunks_test.go index 4ae7d6133..716e077a3 100644 --- a/weed/filer/filechunks_test.go +++ b/weed/filer/filechunks_test.go @@ -22,7 +22,10 @@ func TestCompactFileChunks(t *testing.T) { {Offset: 110, Size: 200, FileId: "jkl", ModifiedTsNs: 300}, } - compacted, garbage := CompactFileChunks(context.Background(), nil, chunks) + compacted, garbage, err := CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } if len(compacted) != 3 { t.Fatalf("unexpected compacted: %d", len(compacted)) @@ -55,7 +58,10 @@ func TestCompactFileChunks2(t *testing.T) { }) } - compacted, garbage := CompactFileChunks(context.Background(), nil, chunks) + compacted, garbage, err := CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } if len(compacted) != 4 { t.Fatalf("unexpected compacted: %d", len(compacted)) @@ -427,7 +433,10 @@ func TestChunksReading(t *testing.T) { // continue } log.Printf("++++++++++ read test case %d ++++++++++++++++++++", i) - chunks := ViewFromChunks(context.Background(), nil, testcase.Chunks, testcase.Offset, testcase.Size) + chunks, err := ViewFromChunks(context.Background(), nil, testcase.Chunks, testcase.Offset, testcase.Size) + if err != nil { + t.Fatal(err) + } x := -1 for c := chunks.Front(); c != nil; c = c.Next { x++ @@ -474,7 +483,9 @@ func BenchmarkCompactFileChunks(b *testing.B) { } for n := 0; n < b.N; n++ { - CompactFileChunks(context.Background(), nil, chunks) + if _, _, err := CompactFileChunks(context.Background(), nil, chunks); err != nil { + b.Fatal(err) + } } } @@ -563,7 +574,10 @@ func TestCompactFileChunks3(t *testing.T) { {Offset: 300, Size: 100, FileId: "def", ModifiedTsNs: 200}, } - compacted, _ := CompactFileChunks(context.Background(), nil, chunks) + compacted, _, err := CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } if len(compacted) != 4 { t.Fatalf("unexpected compacted: %d", len(compacted)) diff --git a/weed/filer/stream.go b/weed/filer/stream.go index 47fccbb87..a34a6c548 100644 --- a/weed/filer/stream.go +++ b/weed/filer/stream.go @@ -386,9 +386,18 @@ var _ = io.ReadSeeker(&ChunkStreamReader{}) var _ = io.ReaderAt(&ChunkStreamReader{}) var _ = io.Closer(&ChunkStreamReader{}) +// doNewChunkStreamReader builds a reader over chunks. If their manifests cannot +// be resolved, the reader fails every read with that error and reports it as its +// SourceError. func doNewChunkStreamReader(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk) *ChunkStreamReader { - chunkViews := ViewFromChunks(ctx, lookupFileIdFn, chunks, 0, math.MaxInt64) + chunkViews, err := ViewFromChunks(ctx, lookupFileIdFn, chunks, 0, math.MaxInt64) + if err != nil { + return &ChunkStreamReader{ + lookupFileId: lookupFileIdFn, + sourceErr: err, + } + } var totalSize int64 for x := chunkViews.Front(); x != nil; x = x.Next { @@ -487,6 +496,10 @@ func insideChunk(offset int64, chunk *ChunkView) bool { } func (c *ChunkStreamReader) prepareBufferFor(offset int64) (err error) { + // a reader without chunk views has a source error only if ViewFromChunks failed + if c.head == nil && c.sourceErr != nil { + return c.sourceErr + } // stay in the same chunk if c.bufferOffset <= offset && offset < c.bufferOffset+int64(len(c.buffer)) { return nil diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 834e3827c..8c7d85242 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -238,7 +238,10 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(entry.GetChunks()) - chunks, _ := filer.CompactFileChunks(context.Background(), wfs.LookupFn(), nonManifestChunks) + chunks, _, err := filer.CompactFileChunks(context.Background(), wfs.LookupFn(), nonManifestChunks) + if err != nil { + return fmt.Errorf("compact chunks of %s: %w", fileFullPath, err) + } if mergedChunks, mergeErr := wfs.maybeMergeChunks(fileFullPath, chunks, manifestChunks); mergeErr != nil { glog.V(0).Infof("maybeMergeChunks %s: %v", fileFullPath, mergeErr) diff --git a/weed/mount/weedfs_file_sync_test.go b/weed/mount/weedfs_file_sync_test.go index ac8c51872..ed1ce122a 100644 --- a/weed/mount/weedfs_file_sync_test.go +++ b/weed/mount/weedfs_file_sync_test.go @@ -245,7 +245,10 @@ func TestCompactThenMergeCondition(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - compacted, _ := filer.CompactFileChunks(context.Background(), nil, tt.chunks) + compacted, _, err := filer.CompactFileChunks(context.Background(), nil, tt.chunks) + if err != nil { + t.Fatal(err) + } _, _, merge := shouldMergeChunks(compacted, nil) if merge != tt.wantMerge { var total uint64 @@ -294,7 +297,10 @@ func TestRandomWritesBloatDetection(t *testing.T) { } } - compacted, _ := filer.CompactFileChunks(context.Background(), nil, chunks) + compacted, _, err := filer.CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } totalCompacted, reportedFileSize, merge := shouldMergeChunks(compacted, nil) // Sanity: condition must be consistent with its own inputs. @@ -337,7 +343,10 @@ func TestFlushCycleManifestAccumulation(t *testing.T) { // --- flush pipeline (mirrors flushMetadataToFiler) --- manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(entryChunks) - compacted, _ := filer.CompactFileChunks(context.Background(), nil, nonManifestChunks) + compacted, _, err := filer.CompactFileChunks(context.Background(), nil, nonManifestChunks) + if err != nil { + t.Fatal(err) + } _, _, merge := shouldMergeChunks(compacted, manifestChunks) if merge { @@ -406,9 +415,18 @@ func TestVisibleContentPreservedAfterCompact(t *testing.T) { }) } - origViews := filer.ViewFromChunks(context.Background(), nil, chunks, 0, math.MaxInt64) - compacted, _ := filer.CompactFileChunks(context.Background(), nil, chunks) - compViews := filer.ViewFromChunks(context.Background(), nil, compacted, 0, math.MaxInt64) + origViews, err := filer.ViewFromChunks(context.Background(), nil, chunks, 0, math.MaxInt64) + if err != nil { + t.Fatal(err) + } + compacted, _, err := filer.CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } + compViews, err := filer.ViewFromChunks(context.Background(), nil, compacted, 0, math.MaxInt64) + if err != nil { + t.Fatal(err) + } // Collect all (offset, size) pairs from views. type viewKey struct { diff --git a/weed/mount/weedfs_metadata_flush.go b/weed/mount/weedfs_metadata_flush.go index 2a4f3eee9..bce2e9e8d 100644 --- a/weed/mount/weedfs_metadata_flush.go +++ b/weed/mount/weedfs_metadata_flush.go @@ -2,6 +2,7 @@ package mount import ( "context" + "fmt" "sync" "time" @@ -136,7 +137,10 @@ func (wfs *WFS) flushFileMetadata(fh *FileHandle) error { manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(snapshotChunks) // Compact chunks to remove fully overlapped ones - compactedChunks, _ := filer.CompactFileChunks(context.Background(), wfs.LookupFn(), nonManifestChunks) + compactedChunks, _, err := filer.CompactFileChunks(context.Background(), wfs.LookupFn(), nonManifestChunks) + if err != nil { + return fmt.Errorf("compact chunks of %s: %w", fileFullPath, err) + } // Try to create manifest chunks for large files // no chunk deleter here: a failed fold reports the blobs it saved diff --git a/weed/mq/logstore/read_parquet_to_log.go b/weed/mq/logstore/read_parquet_to_log.go index 87366d464..68d654112 100644 --- a/weed/mq/logstore/read_parquet_to_log.go +++ b/weed/mq/logstore/read_parquet_to_log.go @@ -98,7 +98,10 @@ func GenParquetReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p topic eachFileFn := func(entry *filer_pb.Entry, eachLogEntryFn log_buffer.EachLogEntryFuncType, starTsNs, stopTsNs int64) (processedTsNs int64, err error) { // create readerAt for the parquet file fileSize := filer.FileSize(entry) - visibleIntervals, _ := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + visibleIntervals, err := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + if err != nil { + return 0, fmt.Errorf("resolve chunks of %s: %w", entry.Name, err) + } chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, 0, int64(fileSize)) readerCache := filer.NewReaderCache(32, chunkCache, lookupFileIdFn, nil) readerAt := filer.NewChunkReaderAtFromClient(context.Background(), readerCache, chunkViews, int64(fileSize), filer.DefaultPrefetchCount) diff --git a/weed/query/engine/aggregations.go b/weed/query/engine/aggregations.go index 54130212e..e672e6174 100644 --- a/weed/query/engine/aggregations.go +++ b/weed/query/engine/aggregations.go @@ -116,8 +116,9 @@ func (opt *FastPathOptimizer) CollectDataSourcesWithTimeFilter(ctx context.Conte // Read parquet file statistics parquetStats, err := hybridScanner.ReadParquetStatistics(partitionPath) if err != nil { - if isDebugMode(ctx) { - fmt.Printf(" ERROR: Failed to read parquet statistics: %v\n", err) + return dataSources, DataSourceError{ + Source: "parquet_statistics", + Cause: fmt.Errorf("partition %s: %w", partitionPath, err), } } else if len(parquetStats) == 0 { if isDebugMode(ctx) { @@ -141,8 +142,9 @@ func (opt *FastPathOptimizer) CollectDataSourcesWithTimeFilter(ctx context.Conte parquetSources := opt.engine.extractParquetSourceFiles(dataSources.ParquetFiles[partitionPath]) liveLogCount, liveLogErr := opt.engine.countLiveLogRowsExcludingParquetSources(ctx, partitionPath, parquetSources) if liveLogErr != nil { - if isDebugMode(ctx) { - fmt.Printf(" ERROR: Failed to count live log rows: %v\n", liveLogErr) + return dataSources, DataSourceError{ + Source: "live_log_count", + Cause: fmt.Errorf("partition %s: %w", partitionPath, liveLogErr), } } else { dataSources.LiveLogRowCount += liveLogCount @@ -308,7 +310,7 @@ func (comp *AggregationComputer) computeGlobalMin(spec AggregationSpec, dataSour liveLogMin, _, err := comp.engine.computeLiveLogMinMax(partition, spec.Column, partitionParquetSources) if err != nil { - continue // Skip partitions with errors + return nil, err } if liveLogMin != nil { @@ -384,7 +386,7 @@ func (comp *AggregationComputer) computeGlobalMax(spec AggregationSpec, dataSour _, liveLogMax, err := comp.engine.computeLiveLogMinMax(partition, spec.Column, partitionParquetSources) if err != nil { - continue // Skip partitions with errors + return nil, err } if liveLogMax != nil { diff --git a/weed/query/engine/engine.go b/weed/query/engine/engine.go index 7a1c783ba..ec8707c0c 100644 --- a/weed/query/engine/engine.go +++ b/weed/query/engine/engine.go @@ -3894,8 +3894,7 @@ func (e *SQLEngine) computeLiveLogMinMax(partitionPath string, columnName string // Scan this log file for MIN/MAX values fileMin, fileMax, err := e.computeFileMinMax(filerClient, filePath, columnName) if err != nil { - fmt.Printf("Warning: failed to compute min/max for file %s: %v\n", filePath, err) - return nil // Continue with other files + return err } // Update global min/max @@ -4030,7 +4029,10 @@ func (e *SQLEngine) eachLogEntryInFile(filerClient filer_pb.FilerClient, filePat // Read file chunks and process them (pattern from countRowsInLogFile) fileSize := filer.FileSize(fileEntry) - visibleIntervals, _ := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, fileEntry.Chunks, 0, int64(fileSize)) + visibleIntervals, err := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, fileEntry.Chunks, 0, int64(fileSize)) + if err != nil { + return err + } chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, 0, int64(fileSize)) for x := chunkViews.Front(); x != nil; x = x.Next { @@ -4185,8 +4187,7 @@ func (e *SQLEngine) countLiveLogRowsExcludingParquetSources(ctx context.Context, // Count rows in live log file rowCount, err := e.countRowsInLogFile(filerClient, partitionPath, entry) if err != nil { - fmt.Printf("Warning: failed to count rows in %s/%s: %v\n", partitionPath, entry.Name, err) - return nil // Continue with other files + return err } totalRows += rowCount return nil @@ -4354,7 +4355,10 @@ func (e *SQLEngine) countRowsInLogFile(filerClient filer_pb.FilerClient, partiti // Read file chunks and process them (pattern from read_log_from_disk.go) fileSize := filer.FileSize(entry) - visibleIntervals, _ := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + visibleIntervals, err := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + if err != nil { + return 0, err + } chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, 0, int64(fileSize)) for x := chunkViews.Front(); x != nil; x = x.Next { @@ -4456,23 +4460,22 @@ func (e *SQLEngine) getTopicTotalRowCount(ctx context.Context, namespace, topicN // For each partition, count both parquet and live log rows for _, partition := range partitions { // Count parquet rows - parquetStats, parquetErr := hybridScanner.ReadParquetStatistics(partition) - if parquetErr == nil { - for _, stats := range parquetStats { - totalRowCount += stats.RowCount - } + parquetStats, err := hybridScanner.ReadParquetStatistics(partition) + if err != nil { + return 0, fmt.Errorf("read parquet statistics for %s: %w", partition, err) + } + for _, stats := range parquetStats { + totalRowCount += stats.RowCount } // Count live log rows (with deduplication) - parquetSourceFiles := make(map[string]bool) - if parquetErr == nil { - parquetSourceFiles = e.extractParquetSourceFiles(parquetStats) - } + parquetSourceFiles := e.extractParquetSourceFiles(parquetStats) - liveLogCount, liveLogErr := e.countLiveLogRowsExcludingParquetSources(ctx, partition, parquetSourceFiles) - if liveLogErr == nil { - totalRowCount += liveLogCount + liveLogCount, err := e.countLiveLogRowsExcludingParquetSources(ctx, partition, parquetSourceFiles) + if err != nil { + return 0, fmt.Errorf("count live log rows for %s: %w", partition, err) } + totalRowCount += liveLogCount } return totalRowCount, nil @@ -4509,17 +4512,18 @@ func (e *SQLEngine) getActualRowsScannedForFastPath(ctx context.Context, namespa // (parquet files use metadata/statistics, so they contribute 0 to scan count) for _, partition := range partitions { // Get parquet files to determine what was converted - parquetStats, parquetErr := hybridScanner.ReadParquetStatistics(partition) - parquetSourceFiles := make(map[string]bool) - if parquetErr == nil { - parquetSourceFiles = e.extractParquetSourceFiles(parquetStats) + parquetStats, err := hybridScanner.ReadParquetStatistics(partition) + if err != nil { + return 0, fmt.Errorf("read parquet statistics for %s: %w", partition, err) } + parquetSourceFiles := e.extractParquetSourceFiles(parquetStats) // Count only live log rows that haven't been converted to parquet - liveLogCount, liveLogErr := e.countLiveLogRowsExcludingParquetSources(ctx, partition, parquetSourceFiles) - if liveLogErr == nil { - totalScannedRows += liveLogCount + liveLogCount, err := e.countLiveLogRowsExcludingParquetSources(ctx, partition, parquetSourceFiles) + if err != nil { + return 0, fmt.Errorf("count live log rows for %s: %w", partition, err) } + totalScannedRows += liveLogCount // Note: Parquet files contribute 0 to scan count since we use their metadata/statistics } diff --git a/weed/query/engine/hybrid_message_scanner.go b/weed/query/engine/hybrid_message_scanner.go index 2b45e9cdc..e7836ee8c 100644 --- a/weed/query/engine/hybrid_message_scanner.go +++ b/weed/query/engine/hybrid_message_scanner.go @@ -1140,9 +1140,7 @@ func (h *HybridMessageScanner) ReadParquetStatistics(partitionPath string) ([]*P // Extract statistics from this parquet file stats, err := h.extractParquetFileStats(entry, lookupFileIdFn, chunkCache) if err != nil { - // Log error but continue processing other files - fmt.Printf("Warning: failed to extract stats from %s: %v\n", entry.Name, err) - return nil + return err } if stats != nil { @@ -1164,7 +1162,10 @@ func (h *HybridMessageScanner) extractParquetFileStats(entry *filer_pb.Entry, lo } // Create reader for the parquet file fileSize := filer.FileSize(entry) - visibleIntervals, _ := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + visibleIntervals, err := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + if err != nil { + return nil, fmt.Errorf("resolve chunks of %s: %w", entry.Name, err) + } chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, 0, int64(fileSize)) readerCache := filer.NewReaderCache(32, chunkCache, lookupFileIdFn, nil) readerAt := filer.NewChunkReaderAtFromClient(context.Background(), readerCache, chunkViews, int64(fileSize), filer.DefaultPrefetchCount) diff --git a/weed/replication/sink/azuresink/azure_sink.go b/weed/replication/sink/azuresink/azure_sink.go index 43374de44..1ce8d62cd 100644 --- a/weed/replication/sink/azuresink/azure_sink.go +++ b/weed/replication/sink/azuresink/azure_sink.go @@ -130,7 +130,10 @@ func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [] } totalSize := filer.FileSize(entry) - chunkViews := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + chunkViews, err := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + if err != nil { + return fmt.Errorf("resolve chunks of %s: %w", key, err) + } // Create append blob client appendBlobClient := g.client.ServiceClient().NewContainerClient(g.container).NewAppendBlobClient(key) @@ -138,7 +141,7 @@ func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [] // Try to create the blob first (without access conditions for initial creation) ctxCreate, cancelCreate := context.WithTimeout(context.Background(), azure.DefaultAzureOpTimeout) defer cancelCreate() - _, err := appendBlobClient.Create(ctxCreate, nil) + _, err = appendBlobClient.Create(ctxCreate, nil) needsWrite := true freshlyCreated := false diff --git a/weed/replication/sink/b2sink/b2_sink.go b/weed/replication/sink/b2sink/b2_sink.go index a18af0e4f..a3b59a7cf 100644 --- a/weed/replication/sink/b2sink/b2_sink.go +++ b/weed/replication/sink/b2sink/b2_sink.go @@ -104,7 +104,10 @@ func (g *B2Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int } totalSize := filer.FileSize(entry) - chunkViews := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + chunkViews, err := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + if err != nil { + return fmt.Errorf("resolve chunks of %s: %w", key, err) + } bucket, err := g.client.Bucket(context.Background(), g.bucket) if err != nil { diff --git a/weed/replication/sink/gcssink/gcs_sink.go b/weed/replication/sink/gcssink/gcs_sink.go index f2aa271b5..ff2879add 100644 --- a/weed/replication/sink/gcssink/gcs_sink.go +++ b/weed/replication/sink/gcssink/gcs_sink.go @@ -117,7 +117,10 @@ func (g *GcsSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []in } totalSize := filer.FileSize(entry) - chunkViews := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + chunkViews, err := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + if err != nil { + return fmt.Errorf("resolve chunks of %s: %w", key, err) + } ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/weed/replication/sink/localsink/local_sink.go b/weed/replication/sink/localsink/local_sink.go index f316b80ed..8debf7993 100644 --- a/weed/replication/sink/localsink/local_sink.go +++ b/weed/replication/sink/localsink/local_sink.go @@ -84,7 +84,10 @@ func (localsink *LocalSink) CreateEntry(key string, entry *filer_pb.Entry, signa glog.V(4).Infof("Create Entry key: %s", key) totalSize := filer.FileSize(entry) - chunkViews := filer.ViewFromChunks(context.Background(), localsink.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + chunkViews, err := filer.ViewFromChunks(context.Background(), localsink.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + if err != nil { + return fmt.Errorf("resolve chunks of %s: %w", key, err) + } dir := filepath.Dir(key) diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index a99c93b1c..0065eb863 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -1807,7 +1807,10 @@ func (s3a *S3ApiServer) streamDecryptedRangeFromChunks(ctx context.Context, w io if err != nil { return 0, err } - chunkViews := filer.ViewFromChunks(ctx, nil, resolvedChunks, offset, size) + chunkViews, err := filer.ViewFromChunks(ctx, nil, resolvedChunks, offset, size) + if err != nil { + return 0, err + } totalWritten := int64(0) targetOffset := offset diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 8759afbaa..862b9b4fa 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -803,7 +803,10 @@ func (fs *FilerServer) cleanupChunks(ctx context.Context, fullpath string, exist // files with manifest chunks are usually large and append only, skip calculating covered chunks manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(newEntry.GetChunks()) - chunks, coveredChunks := filer.CompactFileChunks(ctx, fs.lookupFileId, nonManifestChunks) + chunks, coveredChunks, err := filer.CompactFileChunks(ctx, fs.lookupFileId, nonManifestChunks) + if err != nil { + return newEntry.GetChunks(), nil, fmt.Errorf("CompactFileChunks: %w", err) + } garbage = append(garbage, coveredChunks...) if newEntry.Attributes != nil { diff --git a/weed/server/filer_grpc_server_cleanup_chunks_test.go b/weed/server/filer_grpc_server_cleanup_chunks_test.go new file mode 100644 index 000000000..7aeef26fd --- /dev/null +++ b/weed/server/filer_grpc_server_cleanup_chunks_test.go @@ -0,0 +1,47 @@ +package weed_server + +import ( + "context" + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// A CreateEntry whose context is cancelled before its chunks are compacted must +// fail the request and store nothing, rather than panic in CompactFileChunks. +func TestCreateEntryCancelledContextFailsRequest(t *testing.T) { + store := newRenameTestStore() + f := newRenameTestFiler(t, store) + + fs := &FilerServer{ + filer: f, + option: &FilerOption{}, + entryLockTable: util.NewLockTable[util.FullPath](), + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := fs.CreateEntry(ctx, &filer_pb.CreateEntryRequest{ + Directory: "/test", + SkipCheckParentDirectory: true, + Entry: &filer_pb.Entry{ + Name: "obj", + Attributes: &filer_pb.FuseAttributes{Mtime: 1700000000, FileMode: 0644, Inode: 1}, + Chunks: []*filer_pb.FileChunk{ + {FileId: "1,01", Offset: 0, Size: 100, ModifiedTsNs: 1}, + }, + }, + }) + if err == nil || !strings.Contains(err.Error(), "CompactFileChunks: "+context.Canceled.Error()) { + t.Fatalf("err = %v, want CompactFileChunks to fail with context canceled", err) + } + + store.mu.Lock() + defer store.mu.Unlock() + if _, found := store.entries["/test/obj"]; found { + t.Fatal("entry was stored despite the failed request") + } +}