diff --git a/weed/parquet_pushdown/footer/cache.go b/weed/parquet_pushdown/footer/cache.go new file mode 100644 index 000000000..0222da547 --- /dev/null +++ b/weed/parquet_pushdown/footer/cache.go @@ -0,0 +1,99 @@ +package footer + +import ( + "errors" + "fmt" + "sync/atomic" + + lru "github.com/hashicorp/golang-lru" +) + +// Identity uniquely names a Parquet file's content per the design's +// Index Consistency rules. The cache key is the same identity used +// to invalidate side indexes when a file's content changes. +// +// Iceberg manifest fields are preferred when available (Path, +// SizeBytes, RecordCount + ETag) because they pin the file by both +// its bytes and its declared row count — defending against the +// pathological case where a write produces a file with the same +// length but different content. ETag fills in for raw-S3 access +// without an Iceberg manifest. Mtime is intentionally not part of +// identity (see PARQUET_PUSHDOWN_DESIGN.md "Index Consistency"). +type Identity struct { + Path string + SizeBytes int64 + RecordCount int64 // 0 when unknown (non-Iceberg-managed files) + ETag string +} + +// Cache is a concurrent LRU of parsed footers keyed by Identity. +// Capacity is in entries, not bytes; Phase 1 footers are small +// enough that entry-count budgeting is easier to reason about. +type Cache struct { + c *lru.Cache + hits atomic.Int64 + misses atomic.Int64 +} + +// NewCache returns a Cache holding up to size entries. Returns an +// error when size is non-positive (a zero-size LRU is too easy to +// misconfigure to silently disable caching). +func NewCache(size int) (*Cache, error) { + if size <= 0 { + return nil, fmt.Errorf("cache size must be positive, got %d", size) + } + c, err := lru.New(size) + if err != nil { + return nil, fmt.Errorf("new lru: %w", err) + } + return &Cache{c: c}, nil +} + +// Get returns the cached ParsedFooter for id, or nil + false on miss. +func (c *Cache) Get(id Identity) (*ParsedFooter, bool) { + if v, ok := c.c.Get(id); ok { + c.hits.Add(1) + return v.(*ParsedFooter), true + } + c.misses.Add(1) + return nil, false +} + +// Add stores pf under id. Existing entries are replaced. +func (c *Cache) Add(id Identity, pf *ParsedFooter) { + c.c.Add(id, pf) +} + +// Stats returns cumulative hit and miss counters. +func (c *Cache) Stats() (hits, misses int64) { + return c.hits.Load(), c.misses.Load() +} + +// Len returns the current entry count. +func (c *Cache) Len() int { + return c.c.Len() +} + +// GetOrLoad atomically returns a cached entry or invokes load to +// produce one and inserts it. Two concurrent loads for the same id +// may both call load — the cache is not a singleflight; that +// optimization is deferred until profiling shows it matters. The +// returned pointer is the one stored in the cache so callers must +// treat it as read-only. +func (c *Cache) GetOrLoad(id Identity, load func() (*ParsedFooter, error)) (*ParsedFooter, error) { + if pf, ok := c.Get(id); ok { + return pf, nil + } + if load == nil { + return nil, errors.New("nil load func") + } + pf, err := load() + if err != nil { + return nil, err + } + if pf == nil { + return nil, errors.New("load returned nil ParsedFooter") + } + c.Add(id, pf) + return pf, nil +} diff --git a/weed/parquet_pushdown/footer/cache_test.go b/weed/parquet_pushdown/footer/cache_test.go new file mode 100644 index 000000000..9c7e267f8 --- /dev/null +++ b/weed/parquet_pushdown/footer/cache_test.go @@ -0,0 +1,142 @@ +package footer + +import ( + "errors" + "sync" + "testing" +) + +func TestCache_NewSizeRejectsZero(t *testing.T) { + if _, err := NewCache(0); err == nil { + t.Error("size=0 should be rejected") + } + if _, err := NewCache(-1); err == nil { + t.Error("negative size should be rejected") + } +} + +func TestCache_HitsAndMissesTracked(t *testing.T) { + c, err := NewCache(8) + if err != nil { + t.Fatalf("new: %v", err) + } + id := Identity{Path: "a", SizeBytes: 1} + pf := &ParsedFooter{NumRows: 10} + + if _, ok := c.Get(id); ok { + t.Fatal("first Get should miss") + } + c.Add(id, pf) + if got, ok := c.Get(id); !ok || got != pf { + t.Fatalf("expected cached pointer, got=%v ok=%v", got, ok) + } + hits, misses := c.Stats() + if hits != 1 || misses != 1 { + t.Errorf("hits=%d misses=%d, want 1,1", hits, misses) + } +} + +// Two distinct identities must not alias. Same path, different size +// is still a different file from the cache's POV. +func TestCache_IdentityIsCompositeKey(t *testing.T) { + c, err := NewCache(8) + if err != nil { + t.Fatalf("new: %v", err) + } + a := Identity{Path: "x", SizeBytes: 100, RecordCount: 1} + b := Identity{Path: "x", SizeBytes: 200, RecordCount: 1} + c.Add(a, &ParsedFooter{NumRows: 1}) + c.Add(b, &ParsedFooter{NumRows: 2}) + + gotA, _ := c.Get(a) + gotB, _ := c.Get(b) + if gotA == gotB { + t.Errorf("entries with different sizes must not alias") + } + if gotA.NumRows != 1 || gotB.NumRows != 2 { + t.Errorf("entries swapped: gotA=%d gotB=%d", gotA.NumRows, gotB.NumRows) + } +} + +// Eviction is the LRU's job; the test just verifies the cache obeys +// its size cap and doesn't grow without bound. +func TestCache_EvictsAboveCap(t *testing.T) { + c, err := NewCache(2) + if err != nil { + t.Fatalf("new: %v", err) + } + c.Add(Identity{Path: "a", SizeBytes: 1}, &ParsedFooter{NumRows: 1}) + c.Add(Identity{Path: "b", SizeBytes: 1}, &ParsedFooter{NumRows: 2}) + c.Add(Identity{Path: "c", SizeBytes: 1}, &ParsedFooter{NumRows: 3}) + + if got := c.Len(); got != 2 { + t.Fatalf("Len = %d, want 2 after eviction", got) + } +} + +func TestCache_GetOrLoadCachesAndDedupes(t *testing.T) { + c, err := NewCache(8) + if err != nil { + t.Fatalf("new: %v", err) + } + id := Identity{Path: "a", SizeBytes: 1} + + calls := 0 + load := func() (*ParsedFooter, error) { + calls++ + return &ParsedFooter{NumRows: 42}, nil + } + + first, err := c.GetOrLoad(id, load) + if err != nil { + t.Fatalf("first GetOrLoad: %v", err) + } + second, err := c.GetOrLoad(id, load) + if err != nil { + t.Fatalf("second GetOrLoad: %v", err) + } + if first != second { + t.Errorf("second GetOrLoad returned different pointer") + } + if calls != 1 { + t.Errorf("load called %d times, want 1", calls) + } +} + +func TestCache_GetOrLoadPropagatesError(t *testing.T) { + c, err := NewCache(8) + if err != nil { + t.Fatalf("new: %v", err) + } + wantErr := errors.New("io boom") + _, err = c.GetOrLoad(Identity{Path: "a"}, func() (*ParsedFooter, error) { return nil, wantErr }) + if !errors.Is(err, wantErr) { + t.Errorf("err = %v, want %v", err, wantErr) + } + if c.Len() != 0 { + t.Errorf("failed load should not insert; Len=%d", c.Len()) + } +} + +// Parallel reads on the same key should all observe a hit after the +// first writer wins, and the cache must not race-detect. +func TestCache_ConcurrentSafe(t *testing.T) { + c, err := NewCache(64) + if err != nil { + t.Fatalf("new: %v", err) + } + id := Identity{Path: "a", SizeBytes: 1} + c.Add(id, &ParsedFooter{NumRows: 1}) + + var wg sync.WaitGroup + for i := 0; i < 32; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for k := 0; k < 256; k++ { + _, _ = c.Get(id) + } + }() + } + wg.Wait() +}