G0 D5: queryd DuckDB SELECT over Parquet via httpfs · 4 scrum fixes

Phase G0 Day 5 ships queryd: in-memory DuckDB with custom Connector
that runs INSTALL httpfs / LOAD httpfs / CREATE OR REPLACE SECRET
(TYPE S3) on every new connection, sourced from SecretsProvider +
shared.S3Config. SetMaxOpenConns(1) so registrar's CREATE VIEWs and
handler's SELECTs serialize through one connection (avoids cross-
connection MVCC visibility edge cases).

Registrar.Refresh reads catalogd /catalog/list, runs CREATE OR
REPLACE VIEW "name" AS SELECT * FROM read_parquet('s3://bucket/key')
per manifest, drops views for removed manifests, skips on unchanged
updated_at (the implicit etag). Drop pass runs BEFORE create pass so
a poison manifest can't block other manifest refreshes (post-scrum
C1 fix).

POST /sql with JSON body {"sql":"…"} returns
{"columns":[{"name":"id","type":"BIGINT"},…], "rows":[[…]],
"row_count":N}. []byte → string conversion so VARCHAR rows
JSON-encode as text. 30s default refresh ticker, configurable via
[queryd].refresh_every.

Cross-lineage scrum on shipped code:
  - Opus 4.7 (opencode):                      1 BLOCK + 4 WARN + 4 INFO
  - Kimi K2-0905 (openrouter):                2 BLOCK + 2 WARN + 1 INFO
  - Qwen3-coder (openrouter):                 2 BLOCK + 1 WARN + 1 INFO

Fixed (4):
  C1 (Opus + Kimi convergent): Refresh aborts on first per-view error
    → drop pass first, collect errors, errors.Join. Poison manifest
    no longer blocks the rest of the catalog from re-syncing.
  B-CTX (Opus BLOCK): bootstrap closure captured OpenDB's ctx →
    cancelled-ctx silently fails every reconnect. context.Background()
    inside closure; passed ctx only for initial Ping.
  B-LEAK (Kimi BLOCK): firstLine(stmt) truncated CREATE SECRET to 80
    chars but those 80 chars contained KEY_ID + SECRET prefix → log
    aggregator captures credentials. Stable per-statement labels +
    redactCreds() filter on wrapped DuckDB errors.
  JSON-ERR (Opus WARN): swallowed json.Encode error → silent
    truncated 200 on unsupported column types. slog.Warn the failure.

Dismissed (4 false positives):
  Qwen BLOCK "bootstrap not transactional" — DuckDB DDL is auto-commit
  Qwen BLOCK "MaxBytesReader after Decode" — false, applied before
  Kimi BLOCK "concurrent Refresh + user SELECT deadlock" — not a
    deadlock, just serialization, by design with 10s timeout retry
  Kimi WARN "dropView leaves r.known inconsistent" — current code
    returns before the delete; the entry persists for retry

Critical reviewer behavior: 1 convergent BLOCK between Opus + Kimi
on the per-view error blocking, plus two independent single-reviewer
BLOCKs (B-CTX, B-LEAK) that smoke could never have caught. The
B-LEAK fix uses defense-in-depth: never pass SQL into the error
path AND redact known cred values from DuckDB's own error message.

DuckDB cgo path: github.com/duckdb/duckdb-go/v2 v2.10502.0 (per
ADR-001 §1) on Go 1.25 + arrow-go. Smoke 6/6 PASS after every
fix round.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

internal/queryd/db.go

@@ -0,0 +1,140 @@
+// Package queryd is the SQL execution surface — a DuckDB engine that
+// reads Parquet directly from S3-compatible storage (MinIO in dev)
+// via DuckDB's httpfs extension. Views are registered from catalogd
+// manifests so user SQL just references dataset names.
+//
+// db.go owns the *sql.DB lifecycle: a custom Connector with a
+// bootstrapper that runs INSTALL httpfs / LOAD httpfs / CREATE OR
+// REPLACE SECRET on every new connection, plus a SetMaxOpenConns(1)
+// pin so the registrar's CREATE VIEWs and the handler's user SQL
+// serialize through one connection (avoids cross-connection visibility
+// edge cases for G0; lift to a pool when concurrency wins matter).
+package queryd
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"fmt"
+	"strings"
+
+	"github.com/duckdb/duckdb-go/v2"
+
+	"git.agentview.dev/profit/golangLAKEHOUSE/internal/secrets"
+	"git.agentview.dev/profit/golangLAKEHOUSE/internal/shared"
+)
+
+// OpenDB returns a *sql.DB backed by an in-memory DuckDB whose
+// connections are pre-loaded with httpfs + an S3 secret derived from
+// the shared S3 config + secrets provider. The bucket parameter names
+// the logical bucket whose credentials are pulled.
+//
+// Caller is responsible for Close()ing the returned db.
+func OpenDB(ctx context.Context, s3 shared.S3Config, prov secrets.Provider, bucketLogicalName string) (*sql.DB, error) {
+	creds, err := prov.S3Credentials(bucketLogicalName)
+	if err != nil {
+		return nil, fmt.Errorf("queryd: secrets: %w", err)
+	}
+
+	bootstrap := buildBootstrap(s3, creds)
+	// Stable labels per bootstrap statement. Per scrum B-LEAK (Kimi):
+	// the prior firstLine(stmt) truncated CREATE OR REPLACE SECRET
+	// to 80 chars, which contained both KEY_ID and the start of
+	// SECRET — a log aggregator would capture credentials. Using
+	// stable labels avoids putting the SQL into the error path at all.
+	labels := []string{"install httpfs", "load httpfs", "create secret"}
+
+	connector, err := duckdb.NewConnector("", func(execer driver.ExecerContext) error {
+		// Per scrum B-CTX (Opus): use Background() inside the bootstrap
+		// closure rather than capturing the OpenDB-call ctx. The
+		// connector callback runs on EVERY new physical connection,
+		// including reconnects long after OpenDB returned. A captured
+		// short-lived ctx would silently fail every reconnect's
+		// bootstrap. The passed ctx is only for the initial Ping below.
+		for i, stmt := range bootstrap {
+			if _, err := execer.ExecContext(context.Background(), stmt, nil); err != nil {
+				return fmt.Errorf("queryd bootstrap %s: %s", labels[i], redactCreds(err.Error(), creds))
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("queryd: new connector: %w", err)
+	}
+
+	db := sql.OpenDB(connector)
+	// One connection: the registrar's view CREATEs are visible to the
+	// query handler's SELECTs without worrying about MVCC/timing.
+	// Lift this for G2+ when concurrent reads matter.
+	db.SetMaxOpenConns(1)
+	if err := db.PingContext(ctx); err != nil {
+		_ = db.Close()
+		return nil, fmt.Errorf("queryd: ping: %w", err)
+	}
+	return db, nil
+}
+
+// buildBootstrap returns the SQL statements that initialize a fresh
+// DuckDB connection: install + load httpfs, then create-or-replace
+// the unnamed S3 secret. Endpoint is normalized — DuckDB wants
+// `host:port` without the scheme; the http/https split is controlled
+// by USE_SSL.
+func buildBootstrap(s3 shared.S3Config, creds secrets.S3Credentials) []string {
+	endpoint := s3.Endpoint
+	useSSL := true
+	switch {
+	case strings.HasPrefix(endpoint, "http://"):
+		endpoint = strings.TrimPrefix(endpoint, "http://")
+		useSSL = false
+	case strings.HasPrefix(endpoint, "https://"):
+		endpoint = strings.TrimPrefix(endpoint, "https://")
+		useSSL = true
+	}
+	urlStyle := "vhost"
+	if s3.UsePathStyle {
+		urlStyle = "path"
+	}
+
+	// CREATE OR REPLACE so a reconnect doesn't error on the existing
+	// secret. Single-quoted SQL string literals; we escape ' → '' for
+	// belt-and-braces (creds shouldn't contain ' but a future SSO
+	// token might).
+	createSecret := fmt.Sprintf(
+		"CREATE OR REPLACE SECRET (TYPE S3, KEY_ID '%s', SECRET '%s', REGION '%s', ENDPOINT '%s', URL_STYLE '%s', USE_SSL %t)",
+		sqlEscape(creds.AccessKeyID),
+		sqlEscape(creds.SecretAccessKey),
+		sqlEscape(s3.Region),
+		sqlEscape(endpoint),
+		sqlEscape(urlStyle),
+		useSSL,
+	)
+
+	return []string{
+		"INSTALL httpfs",
+		"LOAD httpfs",
+		createSecret,
+	}
+}
+
+// sqlEscape doubles single quotes in a SQL string literal value.
+// SQL identifier escaping (doubling " for quoted identifiers) lives
+// next to its use site in registrar.go.
+func sqlEscape(v string) string {
+	return strings.ReplaceAll(v, "'", "''")
+}
+
+// redactCreds replaces known credential values inside an error
+// message with placeholder tokens. DuckDB's parser/exec error may
+// echo the offending statement; this scrubs the secret values
+// regardless. Per scrum B-LEAK (Kimi): a 503 from a typo in CREATE
+// SECRET would otherwise drop both KEY_ID and SECRET into the log
+// pipeline.
+func redactCreds(msg string, creds secrets.S3Credentials) string {
+	if creds.AccessKeyID != "" {
+		msg = strings.ReplaceAll(msg, creds.AccessKeyID, "[REDACTED-KEY]")
+	}
+	if creds.SecretAccessKey != "" {
+		msg = strings.ReplaceAll(msg, creds.SecretAccessKey, "[REDACTED-SECRET]")
+	}
+	return msg
+}

internal/queryd/registrar.go

@@ -0,0 +1,181 @@
+// registrar.go — turn catalogd's manifest list into DuckDB views.
+//
+// Refresh() reads /catalog/list, runs CREATE OR REPLACE VIEW for
+// every manifest, drops views for manifests that disappeared, and
+// skips any whose updated_at hasn't changed since the prior refresh
+// (per kickoff D5.3 spec — "don't re-CREATE on every request,
+// Opus review flagged that as the perf cliff").
+//
+// Identifier safety: all view names are double-quoted in SQL with
+// internal " escaped to "" — catalogd accepts user-supplied dataset
+// names that wouldn't pass SQL bare-identifier rules (hyphens,
+// digit-leading, reserved words). Quoting makes them unambiguous.
+package queryd
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"git.agentview.dev/profit/golangLAKEHOUSE/internal/catalogd"
+)
+
+// CatalogLister is what registrar needs from the catalog client —
+// only List, no Register. Defined as an interface so unit tests can
+// inject a fake without spinning up a real catalogd.
+type CatalogLister interface {
+	List(ctx context.Context) ([]*catalogd.Manifest, error)
+}
+
+// Execer is the subset of *sql.DB that registrar actually uses.
+// Same testability win — unit tests can record the SQL strings
+// without booting DuckDB. *sql.DB satisfies this directly.
+type Execer interface {
+	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
+}
+
+// Registrar holds the running view state and the dependencies needed
+// to refresh it.
+type Registrar struct {
+	exec    Execer
+	catalog CatalogLister
+	bucket  string
+
+	// known tracks the current view names + the updated_at we last
+	// rebuilt them from. Used by Refresh to skip unchanged views and
+	// drop ones that disappeared from the catalog.
+	known map[string]time.Time
+}
+
+// NewRegistrar builds a fresh registrar. bucket is the S3 bucket
+// name used to construct s3:// URLs for read_parquet().
+func NewRegistrar(exec Execer, catalog CatalogLister, bucket string) *Registrar {
+	return &Registrar{
+		exec:    exec,
+		catalog: catalog,
+		bucket:  bucket,
+		known:   make(map[string]time.Time),
+	}
+}
+
+// RefreshStats reports what Refresh did — for logs + tests.
+type RefreshStats struct {
+	Created int
+	Updated int
+	Dropped int
+	Skipped int
+}
+
+// Refresh syncs DuckDB's view catalog with catalogd's manifest list.
+// Returns counts of {Created, Updated, Dropped, Skipped} so the
+// caller can decide whether to log loudly.
+//
+// Per scrum C1 (Opus + Kimi convergent): drop pass runs BEFORE create
+// pass so a failure in one view's CREATE doesn't block another view's
+// DROP. Per-iteration errors are collected and the refresh continues
+// — a single poison manifest must not block the whole catalog from
+// re-syncing. The collected errors are joined into the return value;
+// callers see "everything that happened to fail this round."
+func (r *Registrar) Refresh(ctx context.Context) (RefreshStats, error) {
+	var stats RefreshStats
+
+	manifests, err := r.catalog.List(ctx)
+	if err != nil {
+		return stats, fmt.Errorf("registrar list: %w", err)
+	}
+
+	wantNames := make(map[string]struct{}, len(manifests))
+	for _, m := range manifests {
+		wantNames[m.Name] = struct{}{}
+	}
+
+	var errs []error
+
+	// 1. Drop views whose manifests disappeared. Run first so the
+	//    create pass can't block removals on a poison-create error.
+	for name := range r.known {
+		if _, still := wantNames[name]; still {
+			continue
+		}
+		if err := r.dropView(ctx, name); err != nil {
+			errs = append(errs, err)
+			continue
+		}
+		delete(r.known, name)
+		stats.Dropped++
+	}
+
+	// 2. Create / update views for current manifests.
+	for _, m := range manifests {
+		prior, exists := r.known[m.Name]
+		if exists && prior.Equal(m.UpdatedAt) {
+			stats.Skipped++
+			continue
+		}
+		if err := r.createOrReplaceView(ctx, m); err != nil {
+			errs = append(errs, err)
+			continue
+		}
+		if exists {
+			stats.Updated++
+		} else {
+			stats.Created++
+		}
+		r.known[m.Name] = m.UpdatedAt
+	}
+
+	if len(errs) > 0 {
+		return stats, errors.Join(errs...)
+	}
+	return stats, nil
+}
+
+// createOrReplaceView builds the DuckDB SQL for one manifest and
+// runs it. G0 manifests have exactly one Object per dataset; if more
+// land later (G2 multi-part), `read_parquet([...])` accepts a list.
+func (r *Registrar) createOrReplaceView(ctx context.Context, m *catalogd.Manifest) error {
+	if len(m.Objects) == 0 {
+		return fmt.Errorf("registrar: manifest %q has no objects", m.Name)
+	}
+	urls := make([]string, len(m.Objects))
+	for i, obj := range m.Objects {
+		urls[i] = fmt.Sprintf("'%s'", sqlEscape(buildS3URL(r.bucket, obj.Key)))
+	}
+	var fromExpr string
+	if len(urls) == 1 {
+		fromExpr = "read_parquet(" + urls[0] + ")"
+	} else {
+		fromExpr = "read_parquet([" + strings.Join(urls, ", ") + "])"
+	}
+	sql := fmt.Sprintf("CREATE OR REPLACE VIEW %s AS SELECT * FROM %s",
+		quoteIdent(m.Name), fromExpr)
+	if _, err := r.exec.ExecContext(ctx, sql); err != nil {
+		return fmt.Errorf("create view %q: %w", m.Name, err)
+	}
+	return nil
+}
+
+func (r *Registrar) dropView(ctx context.Context, name string) error {
+	sql := "DROP VIEW IF EXISTS " + quoteIdent(name)
+	if _, err := r.exec.ExecContext(ctx, sql); err != nil {
+		return fmt.Errorf("drop view %q: %w", name, err)
+	}
+	return nil
+}
+
+// quoteIdent wraps a SQL identifier in double quotes and escapes
+// internal " by doubling. This is the SQL-standard rule — works in
+// every engine that accepts quoted identifiers.
+func quoteIdent(name string) string {
+	return `"` + strings.ReplaceAll(name, `"`, `""`) + `"`
+}
+
+// buildS3URL composes the s3://bucket/key URL DuckDB's httpfs
+// extension consumes. Keys aren't URL-escaped because read_parquet
+// accepts S3-style literal paths.
+func buildS3URL(bucket, key string) string {
+	return "s3://" + bucket + "/" + key
+}

internal/queryd/registrar_test.go

@@ -0,0 +1,192 @@
+package queryd
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"git.agentview.dev/profit/golangLAKEHOUSE/internal/catalogd"
+)
+
+// fakeExecer records every SQL string Exec'd against it.
+type fakeExecer struct {
+	calls []string
+}
+
+func (f *fakeExecer) ExecContext(_ context.Context, q string, _ ...any) (sql.Result, error) {
+	f.calls = append(f.calls, q)
+	return nil, nil
+}
+
+// fakeCatalog returns a fixed manifest list and an optional error.
+type fakeCatalog struct {
+	manifests []*catalogd.Manifest
+	err       error
+}
+
+func (f *fakeCatalog) List(_ context.Context) ([]*catalogd.Manifest, error) {
+	return f.manifests, f.err
+}
+
+func mkManifest(name, fp string, updatedAt time.Time, key string) *catalogd.Manifest {
+	return &catalogd.Manifest{
+		Name:              name,
+		DatasetID:         "id-" + name,
+		SchemaFingerprint: fp,
+		Objects:           []catalogd.Object{{Key: key, Size: 1024}},
+		CreatedAt:         updatedAt,
+		UpdatedAt:         updatedAt,
+	}
+}
+
+func TestRefresh_CreatesViewsForNewManifests(t *testing.T) {
+	exec := &fakeExecer{}
+	cat := &fakeCatalog{manifests: []*catalogd.Manifest{
+		mkManifest("workers", "fp1", time.Unix(100, 0), "datasets/workers/fp1.parquet"),
+	}}
+	r := NewRegistrar(exec, cat, "lakehouse-go-primary")
+
+	stats, err := r.Refresh(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stats.Created != 1 || stats.Updated != 0 || stats.Dropped != 0 {
+		t.Errorf("stats: got %+v, want {Created:1}", stats)
+	}
+	if len(exec.calls) != 1 {
+		t.Fatalf("calls: got %d, want 1", len(exec.calls))
+	}
+	got := exec.calls[0]
+	wantHas := []string{
+		`CREATE OR REPLACE VIEW "workers"`,
+		`read_parquet('s3://lakehouse-go-primary/datasets/workers/fp1.parquet')`,
+	}
+	for _, w := range wantHas {
+		if !strings.Contains(got, w) {
+			t.Errorf("missing %q in %q", w, got)
+		}
+	}
+}
+
+func TestRefresh_SkipsUnchangedUpdatedAt(t *testing.T) {
+	exec := &fakeExecer{}
+	t1 := time.Unix(100, 0)
+	cat := &fakeCatalog{manifests: []*catalogd.Manifest{
+		mkManifest("workers", "fp1", t1, "k1"),
+	}}
+	r := NewRegistrar(exec, cat, "b")
+
+	if _, err := r.Refresh(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	// Second refresh with same updated_at — should skip.
+	stats, err := r.Refresh(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stats.Skipped != 1 || stats.Created != 0 || stats.Updated != 0 {
+		t.Errorf("second refresh stats: got %+v, want {Skipped:1}", stats)
+	}
+	// Only the first refresh should have produced an Exec call.
+	if len(exec.calls) != 1 {
+		t.Errorf("calls: got %d, want 1 (skip should not Exec)", len(exec.calls))
+	}
+}
+
+func TestRefresh_RebuildsOnUpdatedAtBump(t *testing.T) {
+	exec := &fakeExecer{}
+	t1 := time.Unix(100, 0)
+	cat := &fakeCatalog{manifests: []*catalogd.Manifest{
+		mkManifest("workers", "fp1", t1, "k1"),
+	}}
+	r := NewRegistrar(exec, cat, "b")
+	_, _ = r.Refresh(context.Background())
+
+	// Bump updated_at — same name + fp, but updated_at changed (idempotent re-register on ingestd).
+	t2 := time.Unix(200, 0)
+	cat.manifests[0].UpdatedAt = t2
+	stats, err := r.Refresh(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stats.Updated != 1 {
+		t.Errorf("expected Updated:1 on updated_at bump, got %+v", stats)
+	}
+	if len(exec.calls) != 2 {
+		t.Errorf("calls: got %d, want 2 (initial + rebuild)", len(exec.calls))
+	}
+}
+
+func TestRefresh_DropsViewForRemovedManifest(t *testing.T) {
+	exec := &fakeExecer{}
+	t1 := time.Unix(100, 0)
+	cat := &fakeCatalog{manifests: []*catalogd.Manifest{
+		mkManifest("workers", "fp1", t1, "k1"),
+		mkManifest("retired", "fp2", t1, "k2"),
+	}}
+	r := NewRegistrar(exec, cat, "b")
+	_, _ = r.Refresh(context.Background())
+
+	// "retired" disappears from the catalog → registrar should DROP its view.
+	cat.manifests = cat.manifests[:1]
+	stats, err := r.Refresh(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stats.Dropped != 1 {
+		t.Errorf("expected Dropped:1, got %+v", stats)
+	}
+	// Find the DROP VIEW call.
+	var found bool
+	for _, c := range exec.calls {
+		if strings.HasPrefix(c, `DROP VIEW IF EXISTS "retired"`) {
+			found = true
+		}
+	}
+	if !found {
+		t.Errorf("DROP VIEW not emitted; calls=%v", exec.calls)
+	}
+}
+
+func TestRefresh_QuotesAndEscapesNames(t *testing.T) {
+	exec := &fakeExecer{}
+	weird := `my "weird" dataset` // contains internal double quotes
+	cat := &fakeCatalog{manifests: []*catalogd.Manifest{
+		mkManifest(weird, "fp", time.Unix(1, 0), "k"),
+	}}
+	r := NewRegistrar(exec, cat, "b")
+
+	if _, err := r.Refresh(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	got := exec.calls[0]
+	wantQuoted := `"my ""weird"" dataset"`
+	if !strings.Contains(got, wantQuoted) {
+		t.Errorf("identifier not properly quoted: got %q, want substring %q", got, wantQuoted)
+	}
+}
+
+func TestRefresh_PropagatesCatalogError(t *testing.T) {
+	exec := &fakeExecer{}
+	cat := &fakeCatalog{err: errors.New("catalog down")}
+	r := NewRegistrar(exec, cat, "b")
+	_, err := r.Refresh(context.Background())
+	if err == nil || !strings.Contains(err.Error(), "catalog down") {
+		t.Errorf("expected wrapped catalog error, got %v", err)
+	}
+}
+
+func TestRefresh_ManifestWithNoObjectsErrors(t *testing.T) {
+	exec := &fakeExecer{}
+	m := mkManifest("empty", "fp", time.Unix(1, 0), "k")
+	m.Objects = nil // pathological — registrar shouldn't synthesize an empty SQL.
+	cat := &fakeCatalog{manifests: []*catalogd.Manifest{m}}
+	r := NewRegistrar(exec, cat, "b")
+	_, err := r.Refresh(context.Background())
+	if err == nil || !strings.Contains(err.Error(), "no objects") {
+		t.Errorf("expected 'no objects' error, got %v", err)
+	}
+}