feat(pipes): resolve table dependencies for cache invalidation#343
feat(pipes): resolve table dependencies for cache invalidation#343taitelee wants to merge 32 commits into
Conversation
… not Cache.Invalidate
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughThe PR implements dependency-aware pipe cache invalidation. ChangesPipe dependency-based cache invalidation
Sequence Diagram(s)sequenceDiagram
participant Client
participant PipesHandler
participant resolvePipeDeps
participant ClickHouse
participant SchemaRegistry
participant Cache
rect rgba(100, 149, 237, 0.5)
Note over Client,Cache: PUT /v1/pipes/{name}
Client->>PipesHandler: PUT pipe SQL
PipesHandler->>resolvePipeDeps: pipe definition
resolvePipeDeps->>ClickHouse: DummyBind SQL → EXPLAIN QUERY TREE
ClickHouse-->>resolvePipeDeps: table_name identifiers
resolvePipeDeps->>ClickHouse: system.tables as_select (view expansion)
ClickHouse-->>resolvePipeDeps: view SQL (recursive)
resolvePipeDeps->>SchemaRegistry: filterKnownTables
SchemaRegistry-->>resolvePipeDeps: resolved base table names
resolvePipeDeps-->>PipesHandler: ResolvedTables []string
PipesHandler->>Cache: store pipe with ResolvedTables
end
rect rgba(60, 179, 113, 0.5)
Note over Client,Cache: GET /v1/pipes/{name}/execute
Client->>PipesHandler: Execute request
PipesHandler->>PipesHandler: pipeDeps(q.ResolvedTables) → []Namespace
PipesHandler->>Cache: Get(sha, deps)
Cache-->>PipesHandler: HIT or MISS
alt MISS
PipesHandler->>ClickHouse: run pipe SQL
ClickHouse-->>PipesHandler: results
PipesHandler->>Cache: Set(sha, deps, results)
end
PipesHandler-->>Client: results + X-Cache header
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested labels
Suggested reviewers
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
✨ Simplify code
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
github-actions
Bot
added
go
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: d7ff65f4-c449-4c3d-a7b3-4ef586eb7bbe
📒 Files selected for processing (18)
CHANGELOG.mdcmd/wavehouse/main.gointernal/api/pipe_deps.gointernal/api/pipe_deps_test.gointernal/api/pipes.gointernal/api/structured_query.gointernal/cache/cache.gointernal/cache/cache_test.gointernal/cache/local.gointernal/cache/local_test.gointernal/cache/version_manager.gointernal/cache/version_manager_test.gointernal/ingest/worker.gointernal/ingest/worker_test.gointernal/pipes/dummybind_test.gointernal/pipes/pipes.gointernal/testutil/mocks.gotests/integration/pipe_deps_test.go
💤 Files with no reviewable changes (1)
- internal/cache/cache_test.go
📜 Review details
🧰 Additional context used
📓 Path-based instructions (7)
**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*.go: Use Go 1.26 with strict formatting enforced by gofumpt
Use structured logging with log/slog (JSON handler)
Use Chi v5 for HTTP routing
Return errors, don't panic. Wrap with fmt.Errorf("context: %w", err)
Use package naming: lowercase, single word (or abbreviated). internal/ enforces module privacy
No global state: Dependencies are passed explicitly (constructor injection)
Comment the why, not the what. Add a comment only when the reason isn't obvious from the code; a line that matches the surrounding pattern needs none. Keep comments to 1–2 lines
DRY — one source of truth. Before adding logic, look for an existing helper, type, or constant to reuse; before duplicating a rule, factor it into one place every caller reads
Leave it neater than you found it — within reason. Fix small, safe things in passing: a stale comment, an obvious typo, a misnamed local, dead code on your path
Files:
internal/api/structured_query.gointernal/testutil/mocks.gointernal/ingest/worker.gointernal/pipes/pipes.gointernal/pipes/dummybind_test.gocmd/wavehouse/main.gotests/integration/pipe_deps_test.gointernal/ingest/worker_test.gointernal/api/pipe_deps.gointernal/cache/local.gointernal/api/pipe_deps_test.gointernal/cache/local_test.gointernal/cache/cache.gointernal/api/pipes.gointernal/cache/version_manager.gointernal/cache/version_manager_test.go
internal/api/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Chi HTTP router, JWT/JWKS middleware (from auth/), ingest/query/structured-query/SSE/schema/DLQ/policy/pipes handlers, Hub
Files:
internal/api/structured_query.gointernal/api/pipe_deps.gointernal/api/pipe_deps_test.gointernal/api/pipes.go
internal/ingest/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Ingest worker pipeline (worker.go): JetStream input → per-table batch INSERT with DLQ output. The pipeline is insert-only. Wire format EventMessage carries {table_name, scope, received_timestamp, data} and nothing else
Files:
internal/ingest/worker.gointernal/ingest/worker_test.go
internal/pipes/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Named query pipes: NamedQuery type + NATS KV store (WAVEHOUSE_PIPES) + .sql file bootstrap. Pre-defined SQL templates with param binding + caching; per-pipe allowed_roles is the only execute-path gate via policy.RoleAllowed
Files:
internal/pipes/pipes.gointernal/pipes/dummybind_test.go
**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*_test.go: Use table-driven tests with tests := []struct{ name string; ... } and t.Run(tt.name, ...)
Use shared mocks from internal/testutil/ (MockPublisher, MockCache, MockDeduplicator, MockSubscriber) instead of creating ad-hoc mocks
Use testutil.MakeJWT(t, claims) and testutil.MakeExpiredJWT(t, claims) for auth tests
Use testutil.NewTestSchemaRegistry(tables) or discovery.NewSchemaRegistryFromMap(tables) for schema-aware tests
Use policy.NewMemoryStore(p) for in-memory policy testing without NATS
Use pipes.NewMemoryStore(queries...) for in-memory pipes testing without NATS
Use testutil.AssertJSONResponse(t, rec, status, expected) and testutil.AssertJSONContains(t, rec, status, substring) for response assertions
Files:
internal/pipes/dummybind_test.gotests/integration/pipe_deps_test.gointernal/ingest/worker_test.gointernal/api/pipe_deps_test.gointernal/cache/local_test.gointernal/cache/version_manager_test.go
tests/integration/**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
Go integration tests (//go:build integration; ClickHouse testcontainer); run via make test-integration
Files:
tests/integration/pipe_deps_test.go
internal/cache/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Implement Cache interface → LocalCache (Ristretto) + SharedCache (TBD) + TieredCache (singleflight)
Files:
internal/cache/local.gointernal/cache/local_test.gointernal/cache/cache.gointernal/cache/version_manager.gointernal/cache/version_manager_test.go
🧠 Learnings (5)
📚 Learning: 2026-06-10T15:01:09.027Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 312
File: docs/src/content/docs/development.md:0-0
Timestamp: 2026-06-10T15:01:09.027Z
Learning: In this repo’s Markdown review (all .md files), do not flag capitalization/style issues for literal paths starting with ".github/" (or any substring that is a path beginning with ".github/"). Treat ".github" as the correct lowercase dotfile directory name, even when it appears inside prose or code spans; automated checks such as LanguageTool’s "(GITHUB)" rule commonly produce false positives for this literal filesystem path.
Applied to files:
CHANGELOG.md
📚 Learning: 2026-05-25T11:24:21.130Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 180
File: internal/cache/local.go:0-0
Timestamp: 2026-05-25T11:24:21.130Z
Learning: In WaveHouse’s cache packages (e.g., internal/cache/local.go), it’s acceptable to define package-level `var` constants that hold immutable OpenTelemetry metric attribute sets / `metric.MeasurementOption` values (for example: `cacheL1Attrs = metric.WithAttributes(attribute.String("tier","L1"))`). Treat these as stateless, pre-allocated option values (analogous to `regexp.MustCompile(...)`), not mutable global state. When applying the AGENTS.md “no global state / constructor injection” guideline, apply it to application dependencies (e.g., Cache, Publisher, Deduplicator) rather than to these immutable OTel attribute/measurement option variables—do not flag them as constructor-injection violations.
Applied to files:
internal/cache/local.gointernal/cache/local_test.gointernal/cache/cache.gointernal/cache/version_manager.gointernal/cache/version_manager_test.go
📚 Learning: 2026-05-20T01:02:00.784Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 164
File: internal/api/router_test.go:289-350
Timestamp: 2026-05-20T01:02:00.784Z
Learning: In WaveHouse’s internal API tests (files matching internal/api/**/*_test.go), follow the existing separation-of-concerns convention for testing the RequireRole middleware: inject `ContextKeyRole` directly into the request `context.Context` instead of using `testutil.MakeJWT`/JWT-driven flows. Do not refactor role-gate tests to use JWT tokens—JWT parsing and token handling are covered separately in `middleware_test.go` (the dedicated JWT parsing tests), and mixing those concerns would expand the failure surface and reduce isolation.
Applied to files:
internal/api/pipe_deps_test.go
📚 Learning: 2026-05-23T01:23:59.268Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 174
File: internal/api/ingest_test.go:111-111
Timestamp: 2026-05-23T01:23:59.268Z
Learning: In WaveHouse Go tests in internal/api/**/*_test.go, use internal/testutil.AssertJSONErrorResponse(t, w) for HTTP error-path JSON assertions. Do not use (or reintroduce) package-local assertJSONErrorResponse helpers. AssertJSONErrorResponse verifies the response Content-Type is application/json, includes the X-Content-Type-Options: nosniff header, and that the JSON body contains an "error" field.
Applied to files:
internal/api/pipe_deps_test.go
📚 Learning: 2026-05-20T20:30:15.808Z
Learnt from: taitelee
Repo: Wave-RF/WaveHouse PR: 172
File: internal/api/pipes_test.go:106-118
Timestamp: 2026-05-20T20:30:15.808Z
Learning: For WaveHouse pipes authorization allowlist checks, fix the empty-role fail-open behavior by (1) removing any outer guard that prevents allowlist evaluation when the incoming `role` is `""` (e.g., don’t short-circuit with `if role != "" { ... }`), and (2) during allowlist scanning, ensure only non-empty allowlist entries can match—e.g., require `ar != "" && ar == role` (so a malformed allowlist like `["" ]` cannot grant access to an empty incoming role via `"" == ""`).
Applied to files:
internal/api/pipes.go
🔇 Additional comments (30)
internal/cache/cache.go (1)
8-32: LGTM!internal/cache/version_manager.go (3)
16-32: LGTM!
34-51: LGTM!
76-94: LGTM!internal/cache/version_manager_test.go (1)
9-72: LGTM!internal/cache/local.go (1)
31-75: LGTM!internal/cache/local_test.go (1)
12-161: LGTM!internal/api/structured_query.go (2)
129-143: LGTM!
172-174: LGTM!internal/testutil/mocks.go (1)
117-136: LGTM!internal/ingest/worker_test.go (4)
224-232: LGTM!
449-543: LGTM!
704-731: LGTM!
733-764: LGTM!internal/ingest/worker.go (1)
461-509: LGTM!internal/pipes/pipes.go (2)
29-37: LGTM!
204-243: LGTM!internal/api/pipe_deps.go (7)
28-35: LGTM!
43-56: LGTM!
64-83: LGTM!
90-103: LGTM!
134-153: LGTM!
196-205: LGTM!
179-187:⚠️ Potential issue | 🟠 Major | ⚡ Quick winIncorrect escape sequence handling for identifiers with both backslashes and backticks.
Lines 183-184 unescape ClickHouse backtick-quoted identifiers in the wrong order, causing incorrect results when an identifier contains both a literal backslash and a literal backtick. ClickHouse uses
\\for a backslash and ``` for a backtick inside backtick-quoted identifiers.Example: The ClickHouse identifier
a\b(a, backslash, backtick, b) is written in SQL as ``a\`b`` (where\→` and\`` → ````). After stripping the outer backticks, the string isa\\\b`. The current code:
- Line 183 replaces
\`` with ````:a\\\b` → `a\b`- Line 184 tries to replace
\\\\with\\: no match (only two backslashes remain)- Result:
a\\b(wrong; expecteda\b`)Sequential
ReplaceAllis unsafe here because the replacements can interact. The correct approach is to process escape sequences in a single pass:
- Iterate byte-by-byte; when
\is seen, check the next byte:
- If
\, output one\and advance two bytes- If
`, output one`and advance two bytes- Otherwise, output
\(or handle as an error)🔧 Proposed fix for correct escape handling
func unquoteIdent(s string) string { s = strings.TrimSpace(s) if len(s) >= 2 && s[0] == '`' && s[len(s)-1] == '`' { - s = s[1 : len(s)-1] - s = strings.ReplaceAll(s, "\\`", "`") - s = strings.ReplaceAll(s, "\\\\", "\\") + s = s[1 : len(s)-1] // strip outer backticks + // Unescape in a single pass to avoid interaction between \\ and \` + var out strings.Builder + for i := 0; i < len(s); i++ { + if s[i] == '\\' && i+1 < len(s) { + next := s[i+1] + if next == '\\' || next == '`' { + out.WriteByte(next) + i++ // skip the next byte (already consumed) + continue + } + } + out.WriteByte(s[i]) + } + s = out.String() } return s }> Likely an incorrect or invalid review comment.internal/api/pipes.go (3)
27-33: LGTM!
76-80: LGTM!
155-193: LGTM!cmd/wavehouse/main.go (1)
367-372: LGTM!internal/pipes/dummybind_test.go (1)
1-63: LGTM!internal/api/pipe_deps_test.go (1)
1-192: LGTM!
github-project-automation
Bot
moved this from Backlog
to In review
in WaveHouse Task Board
…le views); correct pipe cache docs
# Conflicts: # internal/api/stream.go
|
Docs note (posting as a normal comment —
|
EricAndrechek
left a comment
EricAndrechek
left a comment
There was a problem hiding this comment.
Overall much better than the last pass, but I think it still has some room for improvement. A few nit-picky things are included, some docs tweaks, etc, but also some efficiency concerns and the main bit being on caching and pipe dependencies. Specifically, I think we need to consider a method for pipe/table dependencies like we have for the namespace dependencies and versioning using a global table version for easy table-wide invalidation for pipe dependencies/tables and their schemas, potentially on the whole database or something if that makes sense.
github-project-automation
Bot
moved this from Ready
to In review
in WaveHouse Task Board
…base-version fallback; TTL-cap external/pruned reads; serialize schema refreshes Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
# Conflicts: # AGENTS.md # CHANGELOG.md # internal/api/stream.go
…ger edges + create_table_query parse
…lain.go in the package inventory
2 participants
Summary
Pipes were cached TTL-only because they couldn't report which tables they read, so an ingest never invalidated a stale pipe result (#178). This teaches pipes dependency-aware invalidation on the same namespace-versioned path structured queries already use — by asking ClickHouse, never by parsing SQL.
On a pipe's first execution with a given parameter binding,
PipesHandler.resolveDepsrunsEXPLAIN QUERY TREEover the bound SQL and reads the table set off ClickHouse's own analysis (discovery.ResolveTables). Resolution is per bound query because the table set can depend on parameter values: a parameter-gatedUNIONarm ClickHouse folds to constant-false is pruned, so?source=webdepends only onweb_events. Resolutions are memoized per bound SQL (singleflight-coalesced; a generation counter keeps a schema refresh that lands mid-resolution from being silently overwritten by the in-flight result) and re-resolve on every schema refresh. Each resolved table folds into the cache key as its own versioned namespace, encoded exactly as the ingest worker encodes them, so a write to any of them evicts the result; the registry's refresh-time base-table→view cascade extends the same eviction to views and materialized views, and refreshes are serialized end-to-end so an older refresh can't install its cascade over a newer one.Degraded paths fail toward expiry or over-invalidation, never staleness. A query ClickHouse can't analyze (a write/DDL pipe, a missing table, an unreachable server) falls back to a single database-wide version that every write bumps — any write evicts, O(1) per request — and a transient failure is retried on the next execution rather than memoized. A result that can't be reliably version-invalidated — an unfoldable view, an external read (a table function, a cross-database table, a non-local dictionary source), or a dead-branch-pruned dependency set — is TTL-capped (~10 s) so it self-expires. The version-folded cache key is snapshotted before the query executes and reused for its
Set, so a write landing mid-query orphans the entry instead of re-homing pre-write data under the post-bump key (#382, fixed for both cached read paths).Related Issues
Closes #178
Closes #382