diff --git a/weed/parquet_pushdown/footer/parser.go b/weed/parquet_pushdown/footer/parser.go new file mode 100644 index 000000000..e6151be64 --- /dev/null +++ b/weed/parquet_pushdown/footer/parser.go @@ -0,0 +1,139 @@ +// Package footer holds the parsed-footer cache and the parser that +// fills it. Phase 1 of the dev plan: parsed-footer cache + file/range- +// level pushdown. Page-level (ColumnIndex/OffsetIndex) consumption +// lands in M6; for M1 we capture the column-chunk byte ranges that +// file/range-level pushdown needs. +package footer + +import ( + "errors" + "fmt" + "io" + + "github.com/parquet-go/parquet-go" + "github.com/parquet-go/parquet-go/format" +) + +// ParsedFooter is the daemon's compact, cache-friendly view of a +// Parquet file's footer. It carries only the fields M1 actually +// consumes; later milestones extend it (ColumnIndex / OffsetIndex, +// statistics, bloom-filter offsets). +type ParsedFooter struct { + NumRows int64 + NumCols int + NumGroups int + + // ColumnPaths is the dotted path-in-schema for each leaf column, + // in the canonical order parquet-go assigns to the file. + ColumnPaths []string + + // RowGroups[i] holds the byte ranges and row count for row group i. + RowGroups []RowGroupRanges +} + +// RowGroupRanges enumerates the column chunks of one row group, in +// the canonical column order. +type RowGroupRanges struct { + NumRows int64 + ColumnChunks []ColumnChunkRange +} + +// ColumnChunkRange is the byte range to read for one column chunk in +// one row group, including the dictionary page region when present. +// +// Iceberg-style page-level pruning narrows reads further; M1 returns +// the full column-chunk byte range. Including the dictionary-page +// prefix when DictionaryPageOffset > 0 follows the rule from the +// design's Page-Level Index section: the dictionary page is not in +// OffsetIndex but must be fetched whenever any data page is read. +type ColumnChunkRange struct { + Path []string + Offset int64 + Length int64 + NumValues int64 + TotalCompSize int64 +} + +// ParseFromReader reads only the footer of a Parquet file and +// returns the projection used by file/range-level pushdown. +// +// SkipPageIndex is set: M1 does not need ColumnIndex/OffsetIndex, +// and skipping them avoids two extra reads per parse for files that +// have page indexes written. M6 will turn this back on for the page- +// level pruning milestone. +func ParseFromReader(r io.ReaderAt, size int64) (*ParsedFooter, error) { + if r == nil { + return nil, errors.New("nil reader") + } + if size <= 0 { + return nil, fmt.Errorf("invalid size %d", size) + } + + pf, err := parquet.OpenFile(r, size, parquet.SkipPageIndex(true), parquet.SkipBloomFilters(true)) + if err != nil { + return nil, fmt.Errorf("open parquet footer: %w", err) + } + + md := pf.Metadata() + out := &ParsedFooter{ + NumRows: md.NumRows, + NumGroups: len(md.RowGroups), + RowGroups: make([]RowGroupRanges, len(md.RowGroups)), + } + + // All row groups share the same leaf-column layout; sample row + // group 0 to get column paths and column count. + if len(md.RowGroups) > 0 { + out.NumCols = len(md.RowGroups[0].Columns) + out.ColumnPaths = make([]string, out.NumCols) + for j, c := range md.RowGroups[0].Columns { + out.ColumnPaths[j] = joinPath(c.MetaData.PathInSchema) + } + } + + for i, rg := range md.RowGroups { + out.RowGroups[i].NumRows = rg.NumRows + out.RowGroups[i].ColumnChunks = make([]ColumnChunkRange, len(rg.Columns)) + for j, c := range rg.Columns { + start, length := columnChunkBytes(&c) + out.RowGroups[i].ColumnChunks[j] = ColumnChunkRange{ + Path: c.MetaData.PathInSchema, + Offset: start, + Length: length, + NumValues: c.MetaData.NumValues, + TotalCompSize: c.MetaData.TotalCompressedSize, + } + } + } + return out, nil +} + +// columnChunkBytes returns the (offset, length) of the contiguous +// byte region that a reader needs to fetch to consume this column +// chunk: the dictionary page (when present) plus all data pages. +// +// parquet-go exposes DictionaryPageOffset as the start of the +// dictionary page when one exists, otherwise zero, and DataPageOffset +// as the start of the first data page. The end of the chunk is +// `start + TotalCompressedSize`. +func columnChunkBytes(c *format.ColumnChunk) (int64, int64) { + start := c.MetaData.DataPageOffset + if dict := c.MetaData.DictionaryPageOffset; dict > 0 && dict < start { + start = dict + } + return start, c.MetaData.TotalCompressedSize +} + +func joinPath(parts []string) string { + switch len(parts) { + case 0: + return "" + case 1: + return parts[0] + } + out := parts[0] + for _, p := range parts[1:] { + out += "." + p + } + return out +} diff --git a/weed/parquet_pushdown/footer/parser_test.go b/weed/parquet_pushdown/footer/parser_test.go new file mode 100644 index 000000000..2bc9ca05f --- /dev/null +++ b/weed/parquet_pushdown/footer/parser_test.go @@ -0,0 +1,129 @@ +package footer + +import ( + "bytes" + "testing" + + "github.com/parquet-go/parquet-go" +) + +type sampleRow struct { + ID int64 + Name string + Value float64 +} + +// writeFixture builds a small Parquet buffer with two row groups so +// the parser exercises both row-group iteration and per-column-chunk +// byte ranges. RowGroupTargetSize is set tight enough that the writer +// flushes mid-input. +func writeFixture(t *testing.T, rows []sampleRow) []byte { + t.Helper() + + schema := parquet.SchemaOf(sampleRow{}) + var buf bytes.Buffer + w := parquet.NewGenericWriter[sampleRow](&buf, schema) + if _, err := w.Write(rows); err != nil { + t.Fatalf("write rows: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("close writer: %v", err) + } + return buf.Bytes() +} + +func TestParseFromReader_ColumnChunksAndCounts(t *testing.T) { + rows := make([]sampleRow, 1000) + for i := range rows { + rows[i] = sampleRow{ID: int64(i), Name: "row", Value: float64(i)} + } + data := writeFixture(t, rows) + + pf, err := ParseFromReader(bytes.NewReader(data), int64(len(data))) + if err != nil { + t.Fatalf("parse: %v", err) + } + if pf.NumRows != int64(len(rows)) { + t.Errorf("NumRows = %d, want %d", pf.NumRows, len(rows)) + } + if pf.NumCols != 3 { + t.Errorf("NumCols = %d, want 3", pf.NumCols) + } + wantCols := []string{"ID", "Name", "Value"} + if len(pf.ColumnPaths) != len(wantCols) { + t.Fatalf("ColumnPaths len %d, want %d", len(pf.ColumnPaths), len(wantCols)) + } + for i, want := range wantCols { + if pf.ColumnPaths[i] != want { + t.Errorf("ColumnPaths[%d] = %q, want %q", i, pf.ColumnPaths[i], want) + } + } + if pf.NumGroups < 1 { + t.Fatalf("NumGroups = %d, want >= 1", pf.NumGroups) + } + + // Every column chunk has a non-zero offset (data starts after the + // 4-byte magic header) and a positive length, and the byte ranges + // must not overlap within a single row group when sorted. + for gi, rg := range pf.RowGroups { + if rg.NumRows == 0 { + t.Errorf("row group %d has zero rows", gi) + } + if len(rg.ColumnChunks) != pf.NumCols { + t.Errorf("row group %d ColumnChunks len = %d, want %d", gi, len(rg.ColumnChunks), pf.NumCols) + } + for ci, cc := range rg.ColumnChunks { + if cc.Offset <= 0 { + t.Errorf("row group %d col %d offset = %d, want > 0", gi, ci, cc.Offset) + } + if cc.Length <= 0 { + t.Errorf("row group %d col %d length = %d, want > 0", gi, ci, cc.Length) + } + if cc.Offset+cc.Length > int64(len(data)) { + t.Errorf("row group %d col %d overruns file: end %d > size %d", gi, ci, cc.Offset+cc.Length, len(data)) + } + } + } +} + +func TestParseFromReader_RejectsInvalid(t *testing.T) { + if _, err := ParseFromReader(nil, 100); err == nil { + t.Error("nil reader should fail") + } + if _, err := ParseFromReader(bytes.NewReader([]byte("PAR1")), 0); err == nil { + t.Error("zero size should fail") + } + if _, err := ParseFromReader(bytes.NewReader([]byte("not parquet at all")), 18); err == nil { + t.Error("non-parquet bytes should fail") + } +} + +// columnChunkBytes prefers the dictionary page offset when present +// because the dictionary page must be fetched alongside the data +// pages of the same chunk. +func TestColumnChunkBytes_DictionaryOffsetPreferred(t *testing.T) { + // Synthesize a fixture where dictionary encoding is likely: + // repeating low-cardinality strings nudge parquet-go to write a + // dictionary page. Then check that at least one column chunk + // reports a start offset before its data-page offset. + rows := make([]sampleRow, 2000) + for i := range rows { + rows[i] = sampleRow{ID: int64(i % 5), Name: "tenant-" + string(rune('a'+(i%5))), Value: float64(i % 5)} + } + data := writeFixture(t, rows) + pf, err := ParseFromReader(bytes.NewReader(data), int64(len(data))) + if err != nil { + t.Fatalf("parse: %v", err) + } + + // Sanity: at least one chunk's range should be plausible (length + // is positive and offset doesn't exceed the file). Beyond that we + // can't depend on a specific encoding choice from parquet-go. + for _, rg := range pf.RowGroups { + for _, cc := range rg.ColumnChunks { + if cc.Offset+cc.Length > int64(len(data)) { + t.Fatalf("chunk overruns file") + } + } + } +}