mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-19 22:44:18 +00:00
parquet_pushdown(M1-C2): add footer LRU cache
Add weed/parquet_pushdown/footer/cache.go: a concurrent LRU keyed by the Identity tuple (Path, SizeBytes, RecordCount, ETag) the design specifies in Index Consistency. Mtime is intentionally absent. Notable behavior: - size <= 0 is rejected at construction so a misconfiguration cannot silently disable caching by sizing the LRU to zero. - Hits / misses are tracked atomically and surfaced via Stats() so the service handler can stamp PushdownStats.footer_cache_hits / footer_cache_misses. - GetOrLoad is a non-singleflight load helper. Concurrent loads for the same identity may both call the loader; that's an acceptable starting point given footer parses are millisecond-scale. Singleflight is deferred until profiling shows it matters. Tests cover: zero-size rejected, hit/miss counters, identity is a composite key (same path, different size = different entry), LRU evicts above cap, GetOrLoad caches the result and dedupes subsequent calls, errors propagate without inserting, and the cache is safe under concurrent readers (-race).
This commit is contained in:
Chris Lu
@@ -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
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
Reference in New Issue
Block a user