mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-18 22:14:33 +00:00
parquet_pushdown(M1-C1): add footer parser
Add weed/parquet_pushdown/footer/parser.go: a thin wrapper around parquet-go's OpenFile that returns the cache-friendly subset M1 needs — per-row-group, per-column-chunk byte ranges plus row counts and dotted column paths. Notes: - SkipPageIndex(true) and SkipBloomFilters(true) are passed because M1 only does file/range-level pushdown; ColumnIndex/OffsetIndex consumption is M6 work and skipping them avoids the extra thrift-decode pass per parse. - columnChunkBytes prefers DictionaryPageOffset when present so the byte range covers both the dictionary page and the data pages of the column chunk, per the design's Page-Level Index rule about dictionary pages. Tests cover row count, column count, column paths, byte-range sanity, and bad-input rejection.
This commit is contained in:
Chris Lu
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user