parquet_pushdown(M1-C5): round-trip integration test

Generates a 3-file Parquet "table" in a temp dir using parquet-go,
boots SeaweedParquetPushdown on a bufconn listener with LocalLoader,
and submits real RPCs covering the M1 surface:

- TestPushdown_M1_RoundTrip: predicate-less, 2-column projection
  across 3 files. Asserts FileRanges reference the right file and
  fit inside it; first call has zero hits and one miss per file.
- TestPushdown_M1_FooterCacheHits: a second identical request must
  produce one cache hit per file and zero misses, the dev plan's
  acceptance bar for cache hit-rate.
- TestPushdown_M1_NoColumnProjection: empty Columns returns ranges
  for every column in every row group.
- TestPushdown_M1_UnknownColumn: unknown column path is rejected.
- TestPushdown_M1_FieldIdOnlyRejected: field-id without a path hint
  is rejected (catalog-schema resolution lands in M3).

Filer-process-level integration is deferred: M1 does not require a
filer, and the bufconn round-trip exercises every M1 code path end-
to-end. test/parquet_pushdown/ stays available for the heavier
filer-driven test that lands in or after M2.

weed/parquet_pushdown/m1_round_trip_test.go

@@ -0,0 +1,260 @@
+package parquet_pushdown
+
+// m1_round_trip_test.go is the M1 round-trip integration test from
+// PARQUET_PUSHDOWN_DEV_PLAN.md:
+//
+//   "Round-trip integration test against a 3-file Iceberg table built
+//   in test/parquet_pushdown/fixtures/.
+//
+//   Footer cache hit-rate test (second call avoids re-parse; verified
+//   via PushdownStats.FooterCacheHits)."
+//
+// It generates three Parquet files in a temp dir using parquet-go,
+// boots the SeaweedParquetPushdown service on a bufconn listener
+// with LocalLoader pointing at the temp dir, and submits real RPCs.
+// It does NOT require a filer; the filer-backed loader is M2+.
+
+import (
+	"bytes"
+	"context"
+	"net"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/parquet-go/parquet-go"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/test/bufconn"
+
+	pb "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb"
+)
+
+type fixtureRow struct {
+	ID    int64
+	Name  string
+	Value float64
+}
+
+func writeParquetFile(t *testing.T, path string, rows []fixtureRow) {
+	t.Helper()
+	var buf bytes.Buffer
+	w := parquet.NewGenericWriter[fixtureRow](&buf, parquet.SchemaOf(fixtureRow{}))
+	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)
+	}
+	if err := os.WriteFile(path, buf.Bytes(), 0o600); err != nil {
+		t.Fatalf("write file: %v", err)
+	}
+}
+
+func makeFixtureTable(t *testing.T) (dir string, files []string, sizes []int64) {
+	t.Helper()
+	dir = t.TempDir()
+	for i, count := range []int{500, 750, 1000} {
+		name := filepath.Join(dir, "part-0000"+string(rune('1'+i))+".parquet")
+		rows := make([]fixtureRow, count)
+		for k := range rows {
+			rows[k] = fixtureRow{ID: int64(i*10000 + k), Name: "r", Value: float64(k)}
+		}
+		writeParquetFile(t, name, rows)
+		st, err := os.Stat(name)
+		if err != nil {
+			t.Fatalf("stat: %v", err)
+		}
+		files = append(files, name)
+		sizes = append(sizes, st.Size())
+	}
+	return dir, files, sizes
+}
+
+func startServiceForFixtures(t *testing.T) (pb.SeaweedParquetPushdownClient, func()) {
+	t.Helper()
+	lis := bufconn.Listen(1 << 20)
+	srv := grpc.NewServer()
+	svc := New(Options{
+		Version:   "m1-test",
+		TrustMode: TrustModeCatalogValidated,
+		Loader:    LocalLoader{},
+		CacheSize: 32,
+	})
+	pb.RegisterSeaweedParquetPushdownServer(srv, svc)
+	go func() { _ = srv.Serve(lis) }()
+
+	dialer := func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }
+	conn, err := grpc.NewClient(
+		"passthrough:///bufnet",
+		grpc.WithContextDialer(dialer),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	if err != nil {
+		srv.Stop()
+		t.Fatalf("dial: %v", err)
+	}
+	return pb.NewSeaweedParquetPushdownClient(conn), func() {
+		_ = conn.Close()
+		srv.Stop()
+	}
+}
+
+func descriptorsFor(t *testing.T, files []string, sizes []int64) []*pb.DataFileDescriptor {
+	t.Helper()
+	out := make([]*pb.DataFileDescriptor, len(files))
+	for i, p := range files {
+		out[i] = &pb.DataFileDescriptor{Path: p, SizeBytes: sizes[i], RecordCount: 1}
+	}
+	return out
+}
+
+// Round-trip: three files, project two columns, no predicate.
+// Expect 3 files × N row groups × 2 columns of FileRanges, each with
+// a positive offset / length within the underlying file.
+func TestPushdown_M1_RoundTrip(t *testing.T) {
+	_, files, sizes := makeFixtureTable(t)
+	client, cleanup := startServiceForFixtures(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	resp, err := client.Pushdown(ctx, &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 1,
+		DataFiles:  descriptorsFor(t, files, sizes),
+		Columns:    []*pb.ColumnRef{{Path: "ID"}, {Path: "Value"}},
+	})
+	if err != nil {
+		t.Fatalf("pushdown: %v", err)
+	}
+	if len(resp.FileRanges) == 0 {
+		t.Fatal("expected file ranges, got none")
+	}
+
+	// Per file, every range must reference that file and fit inside it.
+	byFile := map[string]int64{}
+	for i, p := range files {
+		byFile[p] = sizes[i]
+	}
+	for i, r := range resp.FileRanges {
+		size, ok := byFile[r.File]
+		if !ok {
+			t.Fatalf("range %d references unknown file %q", i, r.File)
+		}
+		if r.Offset <= 0 || r.Length <= 0 {
+			t.Errorf("range %d has invalid bounds (offset=%d length=%d)", i, r.Offset, r.Length)
+		}
+		if r.Offset+r.Length > size {
+			t.Errorf("range %d overruns file %q (offset=%d length=%d size=%d)", i, r.File, r.Offset, r.Length, size)
+		}
+	}
+
+	// First call: every footer is a miss; no hits.
+	if got := resp.Stats.FooterCacheHits; got != 0 {
+		t.Errorf("first call FooterCacheHits = %d, want 0", got)
+	}
+	if got := resp.Stats.FooterCacheMisses; got != int64(len(files)) {
+		t.Errorf("first call FooterCacheMisses = %d, want %d", got, len(files))
+	}
+}
+
+// Footer cache hit-rate: a second identical request must hit the
+// cache for every file (zero misses).
+func TestPushdown_M1_FooterCacheHits(t *testing.T) {
+	_, files, sizes := makeFixtureTable(t)
+	client, cleanup := startServiceForFixtures(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	req := &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 1,
+		DataFiles:  descriptorsFor(t, files, sizes),
+		Columns:    []*pb.ColumnRef{{Path: "ID"}},
+	}
+
+	if _, err := client.Pushdown(ctx, req); err != nil {
+		t.Fatalf("warm-up: %v", err)
+	}
+
+	resp, err := client.Pushdown(ctx, req)
+	if err != nil {
+		t.Fatalf("second call: %v", err)
+	}
+	if got := resp.Stats.FooterCacheHits; got != int64(len(files)) {
+		t.Errorf("FooterCacheHits = %d, want %d", got, len(files))
+	}
+	if got := resp.Stats.FooterCacheMisses; got != 0 {
+		t.Errorf("FooterCacheMisses = %d, want 0", got)
+	}
+}
+
+// Empty Columns list returns ranges for every column in every row group.
+func TestPushdown_M1_NoColumnProjection(t *testing.T) {
+	_, files, sizes := makeFixtureTable(t)
+	client, cleanup := startServiceForFixtures(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	resp, err := client.Pushdown(ctx, &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 1,
+		DataFiles:  descriptorsFor(t, files, sizes),
+	})
+	if err != nil {
+		t.Fatalf("pushdown: %v", err)
+	}
+	// Every fixture has 3 columns; the count must be a multiple of 3.
+	if len(resp.FileRanges)%3 != 0 {
+		t.Errorf("FileRanges count %d not a multiple of column count 3", len(resp.FileRanges))
+	}
+}
+
+// An unknown column path is rejected with InvalidArgument before any
+// range is emitted.
+func TestPushdown_M1_UnknownColumn(t *testing.T) {
+	_, files, sizes := makeFixtureTable(t)
+	client, cleanup := startServiceForFixtures(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	_, err := client.Pushdown(ctx, &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 1,
+		DataFiles:  descriptorsFor(t, files, sizes),
+		Columns:    []*pb.ColumnRef{{Path: "definitely_not_a_real_column"}},
+	})
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+}
+
+// A field-id-only ColumnRef without a Path hint is unsupported in M1
+// because schema-based resolution lands in M3.
+func TestPushdown_M1_FieldIdOnlyRejected(t *testing.T) {
+	_, files, sizes := makeFixtureTable(t)
+	client, cleanup := startServiceForFixtures(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	_, err := client.Pushdown(ctx, &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 1,
+		DataFiles:  descriptorsFor(t, files, sizes),
+		Columns:    []*pb.ColumnRef{{FieldId: 1}}, // no Path
+	})
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+}