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
There was a problem hiding this comment.
Actionable comments posted: 10
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
internal/api/ingest.go (1)
58-61: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winInject the deduplication counter instead of creating global state.
dedupeMissingIDCounteris package-global and binds to the global OpenTelemetry meter during package initialization. Add the counter to the metrics/observer wiring and pass it intoNewIngestHandleras an*IngestHandlerdependency. This avoids uncontrolled global OpenTelemetry instrumentation and keeps the handler dependencies explicit.Source: Coding guidelines
AGENTS.md (1)
29-31: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueExtend the
chsql/gloss to cover the NATS encoder.This PR moves
SafeEncodeNATS/SafeDecodeNATSintointernal/chsql/nats.go, anddocs/src/content/docs/architecture.mdnow documents them. Thechsql/bullet in this package list and theinternal/chsql/line in the File Structure block still describe onlyQuoteIdentandBindUnsafe. Add the NATS-safe name encoder (andQuoteString) so the two documents agree.As per coding guidelines: "Every code change must update its corresponding documentation and
CHANGELOG.mdin the same PR."Source: Coding guidelines
CHANGELOG.md (1)
77-77: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winUpdate the trailing sentence: it contradicts the new entry in the same release.
The last sentence of this bullet states that pipes pass no dependencies and rely on TTL-only invalidation until
#178lands. TheAddedentry at line 14 in the sameUnreleasedsection closes#178and documents per-table versioned namespaces for pipes. A reader of one release section gets two opposite statements.♻️ Proposed change
-Pipes currently pass no dependencies (TTL-only invalidation) until they can report the tables they read ([`#178`](https://github.com/Wave-RF/WaveHouse/issues/178)). +Pipes gained their table dependencies in the same release — see the dependency-aware pipe invalidation entry under `Added` ([`#178`](https://github.com/Wave-RF/WaveHouse/issues/178)).
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 7c338981-aac0-4953-b5c4-d56094681546
📒 Files selected for processing (36)
AGENTS.mdCHANGELOG.mdcmd/wavehouse/main.godocs/src/content/docs/api.mddocs/src/content/docs/architecture.mddocs/src/content/docs/pipes.mdxinternal/api/boot_chain_test.gointernal/api/dlq.gointernal/api/errors_test.gointernal/api/ingest.gointernal/api/pipe_deps_test.gointernal/api/pipes.gointernal/api/pipes_test.gointernal/api/stream.gointernal/api/structured_query.gointernal/cache/cache.gointernal/cache/local.gointernal/cache/local_test.gointernal/cache/version_manager.gointernal/chsql/chsql.gointernal/chsql/nats.gointernal/chsql/nats_test.gointernal/chsql/quote_test.gointernal/discovery/deps_test.gointernal/discovery/discovery.gointernal/discovery/discovery_test.gointernal/discovery/explain.gointernal/discovery/explain_test.gointernal/discovery/fuzz_deps_test.gointernal/ingest/worker.gointernal/ingest/worker_test.gointernal/pipes/pipes.gotests/e2e/sdk/cache.test.tstests/integration/dlq_test.gotests/integration/pipe_cache_test.gotests/integration/setup_test.go
📜 Review details
🧰 Additional context used
📓 Path-based instructions (10)
**/*
📄 CodeRabbit inference engine (AGENTS.md)
**/*: Runmake cilocally before every push, using the documented background execution method and a Docker daemon.
Every code change must update its corresponding documentation andCHANGELOG.mdin the same PR.
Address every review finding with a substantive reply, a fix or tracking issue, and resolved review threads before merge.
Create agent PRs as drafts with Conventional-Commits titles of at most 72 characters; validate titles withscripts/lint-pr-title.sh.
Never force-push or rebase PR branches; mergeorigin/maininstead.
Do not hand-write review markers or bypass hooks with--no-verify; use the prescribed tooling.
Files:
internal/chsql/quote_test.godocs/src/content/docs/api.mdinternal/api/boot_chain_test.goAGENTS.mdinternal/api/dlq.gointernal/discovery/explain_test.gointernal/api/ingest.gointernal/api/errors_test.godocs/src/content/docs/architecture.mdtests/integration/setup_test.gotests/e2e/sdk/cache.test.tsinternal/pipes/pipes.goCHANGELOG.mddocs/src/content/docs/pipes.mdxinternal/api/stream.gointernal/api/pipes_test.gotests/integration/dlq_test.gointernal/chsql/nats_test.gointernal/discovery/discovery_test.gointernal/api/pipe_deps_test.gointernal/chsql/nats.gointernal/ingest/worker.gotests/integration/pipe_cache_test.gointernal/discovery/deps_test.gointernal/discovery/explain.gointernal/cache/local_test.gocmd/wavehouse/main.gointernal/cache/version_manager.gointernal/discovery/fuzz_deps_test.gointernal/api/structured_query.gointernal/cache/local.gointernal/ingest/worker_test.gointernal/cache/cache.gointernal/api/pipes.gointernal/chsql/chsql.gointernal/discovery/discovery.go
**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*.go: Use Go 1.26 with strictgofumptformatting.
Return errors instead of panicking, and wrap errors withfmt.Errorf("context: %w", err).
Use explicit dependency passing and constructor injection; do not introduce global state.
Use structured logging withlog/slogand JSON handlers.
Files:
internal/chsql/quote_test.gointernal/api/boot_chain_test.gointernal/api/dlq.gointernal/discovery/explain_test.gointernal/api/ingest.gointernal/api/errors_test.gotests/integration/setup_test.gointernal/pipes/pipes.gointernal/api/stream.gointernal/api/pipes_test.gotests/integration/dlq_test.gointernal/chsql/nats_test.gointernal/discovery/discovery_test.gointernal/api/pipe_deps_test.gointernal/chsql/nats.gointernal/ingest/worker.gotests/integration/pipe_cache_test.gointernal/discovery/deps_test.gointernal/discovery/explain.gointernal/cache/local_test.gocmd/wavehouse/main.gointernal/cache/version_manager.gointernal/discovery/fuzz_deps_test.gointernal/api/structured_query.gointernal/cache/local.gointernal/ingest/worker_test.gointernal/cache/cache.gointernal/api/pipes.gointernal/chsql/chsql.gointernal/discovery/discovery.go
internal/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
internal/**/*.go: Keep internal packages lowercase and single-word or abbreviated; useinternal/to enforce module privacy.
Core interchangeable behaviors should be represented by interfaces, includingCache,Deduplicator,Publisher, andSubscriber.
Files:
internal/chsql/quote_test.gointernal/api/boot_chain_test.gointernal/api/dlq.gointernal/discovery/explain_test.gointernal/api/ingest.gointernal/api/errors_test.gointernal/pipes/pipes.gointernal/api/stream.gointernal/api/pipes_test.gointernal/chsql/nats_test.gointernal/discovery/discovery_test.gointernal/api/pipe_deps_test.gointernal/chsql/nats.gointernal/ingest/worker.gointernal/discovery/deps_test.gointernal/discovery/explain.gointernal/cache/local_test.gointernal/cache/version_manager.gointernal/discovery/fuzz_deps_test.gointernal/api/structured_query.gointernal/cache/local.gointernal/ingest/worker_test.gointernal/cache/cache.gointernal/api/pipes.gointernal/chsql/chsql.gointernal/discovery/discovery.go
**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*_test.go: Use table-driven tests witht.Run(tt.name, ...)for multiple scenarios, and add tests for every new function.
Reuse shared test helpers and mocks frominternal/testutil/, including JWT, schema, policy, pipe, response, and mock helpers.
Files:
internal/chsql/quote_test.gointernal/api/boot_chain_test.gointernal/discovery/explain_test.gointernal/api/errors_test.gotests/integration/setup_test.gointernal/api/pipes_test.gotests/integration/dlq_test.gointernal/chsql/nats_test.gointernal/discovery/discovery_test.gointernal/api/pipe_deps_test.gotests/integration/pipe_cache_test.gointernal/discovery/deps_test.gointernal/cache/local_test.gointernal/discovery/fuzz_deps_test.gointernal/ingest/worker_test.go
**/*.md
📄 CodeRabbit inference engine (AGENTS.md)
Keep
CLAUDE.mdas a short pointer toAGENTS.mdand do not duplicate the full agent instructions.
Files:
docs/src/content/docs/api.mdAGENTS.mddocs/src/content/docs/architecture.mdCHANGELOG.md
internal/api/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
internal/api/**/*.go: Use Chi v5 for HTTP routing, and ensure all/v1/*routes run the always-on JWT middleware.
Maintain bearer-token-only CORS: never emitAccess-Control-Allow-Credentials, and do not reintroduce cookie or session authentication.
Enforce column-level access control on every read path, including structured queries and live streams, through the shared policy decision function.
Files:
internal/api/boot_chain_test.gointernal/api/dlq.gointernal/api/ingest.gointernal/api/errors_test.gointernal/api/stream.gointernal/api/pipes_test.gointernal/api/pipe_deps_test.gointernal/api/structured_query.gointernal/api/pipes.go
tests/e2e/sdk/*.test.ts
📄 CodeRabbit inference engine (AGENTS.md)
tests/e2e/sdk/*.test.ts: E2E tests must use the TypeScript SDK and suite-specific tables obtained fromsuiteTables; never use bare shared table names.
Keep E2E tests sequential withmaxWorkers: 1because they share mutable global policy state; policy-mutating tests must snapshot and restore the full policy.
Files:
tests/e2e/sdk/cache.test.ts
internal/pipes/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Named pipes must enforce exact
allowed_rolesmembership, with no wildcard authorization and admin-only behavior when roles are absent; pipe SQL runs as-is and unresolved dependencies must fall back safely to database-wide invalidation or a TTL cap.
Files:
internal/pipes/pipes.go
docs/src/content/docs/**/*.mdx
📄 CodeRabbit inference engine (AGENTS.md)
Author Mermaid diagrams vertically by default, avoid large side-by-side diagrams, and keep labels short and readable.
Files:
docs/src/content/docs/pipes.mdx
internal/ingest/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Ingest payloads must be validated against discovered ClickHouse schemas, and failed batch inserts must publish to the enabled DLQ without silent data loss.
Files:
internal/ingest/worker.gointernal/ingest/worker_test.go
🧠 Learnings (8)
📚 Learning: 2026-06-26T12:23:22.696Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 346
File: internal/stream/subscriber_test.go:9-28
Timestamp: 2026-06-26T12:23:22.696Z
Learning: In this Go repository, prefer table-driven tests (e.g., `[]struct{...}` with `t.Run(...)`) only for tests that cover multiple scenarios/inputs and can be cleanly enumerated. Do not artificially rewrite a clear single-scenario sequential behavioral-flow test into a table-driven form just to fit the pattern; if there’s only one meaningful scenario, keep the test as a straightforward linear flow (as in `TestSubscriber_SendDeliversThenDropsWhenFull`).
Applied to files:
internal/chsql/quote_test.gointernal/api/boot_chain_test.gointernal/discovery/explain_test.gointernal/api/errors_test.gotests/integration/setup_test.gointernal/api/pipes_test.gotests/integration/dlq_test.gointernal/chsql/nats_test.gointernal/discovery/discovery_test.gointernal/api/pipe_deps_test.gotests/integration/pipe_cache_test.gointernal/discovery/deps_test.gointernal/cache/local_test.gointernal/discovery/fuzz_deps_test.gointernal/ingest/worker_test.go
📚 Learning: 2026-07-07T12:38:12.052Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 378
File: internal/auth/auth.go:119-132
Timestamp: 2026-07-07T12:38:12.052Z
Learning: In this repo, do not add or recommend logging/tracing client IP addresses using naive or untrusted sources (e.g., `r.RemoteAddr` or directly trusting/deriving `X-Forwarded-For`) anywhere in the Go codebase. `middleware.RealIP` was removed due to IP-spoofing risks, and proper trusted-proxy-aware client-IP handling is intentionally deferred to issue `#333`. During code review, if proposed changes would record client IPs (including in audit paths such as `internal/auth/auth.go`), reject/redirect until `#333` lands with correct trusted-proxy configuration and safeguards.
Applied to files:
internal/chsql/quote_test.gointernal/api/boot_chain_test.gointernal/api/dlq.gointernal/discovery/explain_test.gointernal/api/ingest.gointernal/api/errors_test.gotests/integration/setup_test.gointernal/pipes/pipes.gointernal/api/stream.gointernal/api/pipes_test.gotests/integration/dlq_test.gointernal/chsql/nats_test.gointernal/discovery/discovery_test.gointernal/api/pipe_deps_test.gointernal/chsql/nats.gointernal/ingest/worker.gotests/integration/pipe_cache_test.gointernal/discovery/deps_test.gointernal/discovery/explain.gointernal/cache/local_test.gocmd/wavehouse/main.gointernal/cache/version_manager.gointernal/discovery/fuzz_deps_test.gointernal/api/structured_query.gointernal/cache/local.gointernal/ingest/worker_test.gointernal/cache/cache.gointernal/api/pipes.gointernal/chsql/chsql.gointernal/discovery/discovery.go
📚 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:
docs/src/content/docs/api.mdAGENTS.mddocs/src/content/docs/architecture.mdCHANGELOG.md
📚 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/boot_chain_test.gointernal/api/errors_test.gointernal/api/pipes_test.gointernal/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/boot_chain_test.gointernal/api/errors_test.gointernal/api/pipes_test.gointernal/api/pipe_deps_test.go
📚 Learning: 2026-07-09T16:20:50.620Z
Learnt from: taitelee
Repo: Wave-RF/WaveHouse PR: 402
File: internal/api/ingest_test.go:1434-1448
Timestamp: 2026-07-09T16:20:50.620Z
Learning: In WaveHouse’s Go schema/timestamp handling (internal/discovery), resolve any timestamp column zone/precision requirements exactly once during SchemaRegistry.Refresh() and cache the resolved specs on the column. Do not resolve zones per-record on the hot path. Ensure timezone resolution uses embedded tzdata (e.g., via cmd/wavehouse’s time.LoadLocation + embedded tzdata) so minimal containers don’t silently fall back to UTC for non-UTC ClickHouse servers, which would corrupt stored instants. Treat timezone-resolution failures during refresh as non-fatal: keep boot alive, RetryRefresh should continue retrying, and readiness/health (/livez) should surface a degraded state. If a column’s timezone cannot be resolved after refresh, emit a warning and reject only that column’s timestamp values per-record rather than failing the entire refresh.
Applied to files:
internal/discovery/explain_test.gointernal/discovery/discovery_test.gointernal/discovery/deps_test.gointernal/discovery/explain.gointernal/discovery/fuzz_deps_test.gointernal/discovery/discovery.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_test.gointernal/api/pipes.go
📚 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_test.gointernal/cache/version_manager.gointernal/cache/local.gointernal/cache/cache.go
🪛 ast-grep (0.45.0)
tests/integration/pipe_cache_test.go
[warning] 73-74: Detected a SQL statement built with 'fmt.Sprintf' and passed directly to 'db.Exec'/'db.ExecContext'. Interpolating values into a query string lets an attacker inject arbitrary SQL. Use parameterized queries instead: pass the SQL with placeholders ('?' or '') as the query argument and supply the values as separate arguments, e.g. 'db.Exec("UPDATE t SET x = ? WHERE id = ?", x, id)'.
Context: e.chConn.Exec(ctx,
fmt.Sprintf("INSERT INTO %s SELECT number, number%%3 FROM numbers(60)", base))
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-exec-sprintf-go)
[warning] 77-78: Detected a SQL statement built with 'fmt.Sprintf' and passed directly to 'db.Exec'/'db.ExecContext'. Interpolating values into a query string lets an attacker inject arbitrary SQL. Use parameterized queries instead: pass the SQL with placeholders ('?' or '') as the query argument and supply the values as separate arguments, e.g. 'db.Exec("UPDATE t SET x = ? WHERE id = ?", x, id)'.
Context: e.chConn.Exec(ctx,
fmt.Sprintf("CREATE VIEW %s AS SELECT user_id, org_id FROM %s", view, base))
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-exec-sprintf-go)
[warning] 120-121: Detected a SQL statement built with 'fmt.Sprintf' and passed directly to 'db.Exec'/'db.ExecContext'. Interpolating values into a query string lets an attacker inject arbitrary SQL. Use parameterized queries instead: pass the SQL with placeholders ('?' or '') as the query argument and supply the values as separate arguments, e.g. 'db.Exec("UPDATE t SET x = ? WHERE id = ?", x, id)'.
Context: e.chConn.Exec(ctx,
fmt.Sprintf("INSERT INTO %s SELECT number FROM numbers(60)", tbl))
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-exec-sprintf-go)
[warning] 183-184: Detected a SQL statement built with 'fmt.Sprintf' and passed directly to 'db.Exec'/'db.ExecContext'. Interpolating values into a query string lets an attacker inject arbitrary SQL. Use parameterized queries instead: pass the SQL with placeholders ('?' or '') as the query argument and supply the values as separate arguments, e.g. 'db.Exec("UPDATE t SET x = ? WHERE id = ?", x, id)'.
Context: e.chConn.Exec(ctx,
fmt.Sprintf("INSERT INTO %s SELECT number, number FROM numbers(100)", src))
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-exec-sprintf-go)
[warning] 187-189: Detected a SQL statement built with 'fmt.Sprintf' and passed directly to 'db.Exec'/'db.ExecContext'. Interpolating values into a query string lets an attacker inject arbitrary SQL. Use parameterized queries instead: pass the SQL with placeholders ('?' or '') as the query argument and supply the values as separate arguments, e.g. 'db.Exec("UPDATE t SET x = ? WHERE id = ?", x, id)'.
Context: e.chConn.Exec(ctx, fmt.Sprintf(
"CREATE MATERIALIZED VIEW %s ENGINE=AggregatingMergeTree ORDER BY id AS SELECT id, sumState(v) AS s FROM %s GROUP BY id",
mv, src))
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-exec-sprintf-go)
[warning] 229-231: Detected a SQL statement built with 'fmt.Sprintf' and passed directly to 'db.Exec'/'db.ExecContext'. Interpolating values into a query string lets an attacker inject arbitrary SQL. Use parameterized queries instead: pass the SQL with placeholders ('?' or '') as the query argument and supply the values as separate arguments, e.g. 'db.Exec("UPDATE t SET x = ? WHERE id = ?", x, id)'.
Context: e.chConn.Exec(ctx, fmt.Sprintf(
"CREATE MATERIALIZED VIEW %s TO %s AS SELECT id FROM %s",
mv, chQuoteIdent(target), src))
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-exec-sprintf-go)
[error] 79-79: SQL query is built by concatenating a string literal with a variable and passed to a database/sql call (Query, Exec, QueryRow, Prepare, or their Context variants). String concatenation lets attacker-controlled input alter the query structure, enabling SQL injection. Use parameterized queries with placeholders ('?' or '') and pass the values as separate arguments instead of concatenating them into the query string.
Context: e.chConn.Exec(context.Background(), "DROP VIEW IF EXISTS "+view+"")
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-query-string-concat-go)
[error] 190-190: SQL query is built by concatenating a string literal with a variable and passed to a database/sql call (Query, Exec, QueryRow, Prepare, or their Context variants). String concatenation lets attacker-controlled input alter the query structure, enabling SQL injection. Use parameterized queries with placeholders ('?' or '') and pass the values as separate arguments instead of concatenating them into the query string.
Context: e.chConn.Exec(context.Background(), "DROP VIEW IF EXISTS "+mv+"")
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-query-string-concat-go)
[error] 232-232: SQL query is built by concatenating a string literal with a variable and passed to a database/sql call (Query, Exec, QueryRow, Prepare, or their Context variants). String concatenation lets attacker-controlled input alter the query structure, enabling SQL injection. Use parameterized queries with placeholders ('?' or '') and pass the values as separate arguments instead of concatenating them into the query string.
Context: e.chConn.Exec(context.Background(), "DROP VIEW IF EXISTS "+mv+"")
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-query-string-concat-go)
internal/discovery/explain.go
[error] 70-70: SQL query is built by concatenating a string literal with a variable and passed to a database/sql call (Query, Exec, QueryRow, Prepare, or their Context variants). String concatenation lets attacker-controlled input alter the query structure, enabling SQL injection. Use parameterized queries with placeholders ('?' or '') and pass the values as separate arguments instead of concatenating them into the query string.
Context: conn.Query(ctx, "EXPLAIN QUERY TREE "+sql)
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-query-string-concat-go)
internal/discovery/fuzz_deps_test.go
[error] 60-60: SQL query is built by concatenating a string literal with a variable and passed to a database/sql call (Query, Exec, QueryRow, Prepare, or their Context variants). String concatenation lets attacker-controlled input alter the query structure, enabling SQL injection. Use parameterized queries with placeholders ('?' or '') and pass the values as separate arguments instead of concatenating them into the query string.
Context: boot.Exec(ctx, "CREATE DATABASE IF NOT EXISTS "+fdb)
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-query-string-concat-go)
[error] 61-61: SQL query is built by concatenating a string literal with a variable and passed to a database/sql call (Query, Exec, QueryRow, Prepare, or their Context variants). String concatenation lets attacker-controlled input alter the query structure, enabling SQL injection. Use parameterized queries with placeholders ('?' or '') and pass the values as separate arguments instead of concatenating them into the query string.
Context: boot.Exec(ctx, "CREATE DATABASE IF NOT EXISTS "+fother)
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-query-string-concat-go)
[error] 105-105: SQL query is built by concatenating a string literal with a variable and passed to a database/sql call (Query, Exec, QueryRow, Prepare, or their Context variants). String concatenation lets attacker-controlled input alter the query structure, enabling SQL injection. Use parameterized queries with placeholders ('?' or '') and pass the values as separate arguments instead of concatenating them into the query string.
Context: conn.Exec(ctx, "DROP DATABASE IF EXISTS "+fdb)
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-query-string-concat-go)
[error] 106-106: SQL query is built by concatenating a string literal with a variable and passed to a database/sql call (Query, Exec, QueryRow, Prepare, or their Context variants). String concatenation lets attacker-controlled input alter the query structure, enabling SQL injection. Use parameterized queries with placeholders ('?' or '') and pass the values as separate arguments instead of concatenating them into the query string.
Context: conn.Exec(ctx, "DROP DATABASE IF EXISTS "+fother)
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-query-string-concat-go)
[error] 249-249: SQL query is built by concatenating a string literal with a variable and passed to a database/sql call (Query, Exec, QueryRow, Prepare, or their Context variants). String concatenation lets attacker-controlled input alter the query structure, enabling SQL injection. Use parameterized queries with placeholders ('?' or '') and pass the values as separate arguments instead of concatenating them into the query string.
Context: conn.Query(ctx, "EXPLAIN QUERY TREE "+sql)
Note: [CWE-89] Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection').
(sql-injection-query-string-concat-go)
🪛 LanguageTool
docs/src/content/docs/api.md
[style] ~480-~480: Redundant conjunctions can lead to confusion; consider removing a conjunction here.
Context: ...meters can be supplied via query string and/or JSON body. Results are cached in the sh...
(AND_OR)
docs/src/content/docs/architecture.md
[style] ~73-~73: Since ownership is already implied, this phrasing may be redundant.
Context: ...e JWT/JWKS authentication middleware is its own package, [auth/](#auth--authenticatio...
(PRP_OWN)
[style] ~157-~157: Since ownership is already implied, this phrasing may be redundant.
Context: ...ared by query/ and policy/, kept in their own package to break an import cycle. `Quot...
(PRP_OWN)
docs/src/content/docs/pipes.mdx
[typographical] ~178-~178: The word ‘WHERE’ starts a question. Add a question mark (“?”) at the end of the sentence.
Context: ...its data, so this is precise, not stale). The source=mobile result is resolved ...
(WRB_QUESTION_MARK)
[style] ~178-~178: Try elevating your writing by using a synonym here.
Context: ...solved and cached separately, depending only on mobile_events. The pruning is deli...
(ONLY_SOLELY)
[style] ~184-~184: Since ownership is already implied, this phrasing may be redundant.
Context: ...e detects the read (a table function is its own node in the query tree) and **caps the ...
(PRP_OWN)
🔇 Additional comments (33)
internal/api/stream.go (1)
24-25: 🩺 Stability & AvailabilityNo change needed. Production wiring assigns
streamHandler.Metrics = sseMetricsafter construction, and production metrics instrumentation is nil-safe.internal/api/pipes.go (6)
26-63: LGTM!
74-88: LGTM!Also applies to: 98-114
251-266: LGTM!
307-343: LGTM!Also applies to: 355-403
405-448: LGTM!Also applies to: 450-461
247-249: 🩺 Stability & AvailabilityNo change needed.
pipes.BindParamsinlines supplied{{name}}and{{name:default}}placeholders as literals before dependency resolution runs, soresolveDepspasses a fully bound query toEXPLAIN QUERY TREE.internal/api/structured_query.go (1)
15-15: LGTM!Also applies to: 154-168, 177-177, 221-227
internal/api/pipe_deps_test.go (1)
5-119: LGTM!Also applies to: 121-135, 137-186, 188-228, 230-257, 259-311, 313-346
internal/api/pipes_test.go (1)
45-45: LGTM!Also applies to: 62-62, 77-77, 91-91, 107-107, 131-131, 158-158, 179-179, 200-200, 224-224, 249-249, 278-278, 297-297, 317-317, 334-334, 353-353, 370-370, 392-392, 415-415, 441-441, 462-462, 483-483
internal/api/errors_test.go (1)
17-17: LGTM!Also applies to: 140-140
internal/cache/cache.go (1)
10-72: LGTM!internal/cache/version_manager.go (1)
32-84: LGTM!internal/cache/local_test.go (1)
137-239: LGTM!internal/discovery/discovery.go (2)
246-367: LGTM!Also applies to: 571-664
461-464: 🎯 Functional Correctness
viewDefis declared once. No change is needed.internal/discovery/explain.go (1)
158-309: LGTM!Also applies to: 340-435
internal/discovery/deps_test.go (1)
61-134: LGTM!Also applies to: 199-290, 297-374, 504-602
internal/discovery/discovery_test.go (1)
158-160: LGTM!Also applies to: 178-180
internal/discovery/explain_test.go (1)
217-301: LGTM!Also applies to: 303-343
internal/discovery/fuzz_deps_test.go (1)
1-357: LGTM!internal/cache/local.go (1)
81-101: 🩺 Stability & AvailabilityNo change needed. Production invalidation passes encoded table inputs where cached dependency keys are NATS-encoded.
cmd/wavehouse/main.go (1)
186-204: LGTM!Also applies to: 269-273, 282-282, 297-301, 344-368, 421-422
internal/api/boot_chain_test.go (1)
115-118: LGTM!AGENTS.md (1)
56-62: LGTM!CHANGELOG.md (1)
14-14: LGTM!docs/src/content/docs/api.md (1)
480-480: LGTM!docs/src/content/docs/architecture.md (1)
55-55: LGTM!Also applies to: 117-117, 148-148, 157-158
docs/src/content/docs/pipes.mdx (1)
122-122: LGTM!Also applies to: 174-192
tests/e2e/sdk/cache.test.ts (2)
2-10: LGTM!
74-95: 🩺 Stability & AvailabilityClarify whether
suiteTablesshould create materialized-view dependencies.
tests/e2e/sdk/cache.test.tscreatesmv_src_${stamp},mv_tgt_${stamp}, andmv_${stamp}with raw DDL even thoughT = suiteTables("cache")exists. The suite-table path only creates predefinedclicks/events/userstables, so either document these raw object names as acceptable for this edge case or extendsuiteTablesto include cache-specific dependencies if all ClickHouse objects should be owned centrally.tests/integration/pipe_cache_test.go (1)
26-57: LGTM!Also applies to: 66-105, 176-209, 222-256, 271-298, 306-322
tests/integration/setup_test.go (1)
81-89: LGTM!Also applies to: 320-333
| registry.SetOnRefresh(func(snap discovery.DependencySnapshot) { | ||
| resultCache.SetDependents(snap.Cascade) | ||
| if len(snap.ChangedViews) > 0 { | ||
| nss := make([]cache.Namespace, len(snap.ChangedViews)) | ||
| for i, v := range snap.ChangedViews { | ||
| nss[i] = cache.Namespace{Table: v} | ||
| } | ||
| _, _ = resultCache.Invalidate(ctx, nss) | ||
| } | ||
| pipesHandler.ClearResolvedDeps() | ||
| }) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Log the failure of the refresh-time invalidation.
resultCache.Invalidate returns an error that this callback discards. If invalidation fails after a view definition changes, pipes keep serving results keyed on the pre-change version and no signal reaches the operator. Log the error at WARN with the affected namespaces.
♻️ Proposed change
if len(snap.ChangedViews) > 0 {
nss := make([]cache.Namespace, len(snap.ChangedViews))
for i, v := range snap.ChangedViews {
nss[i] = cache.Namespace{Table: v}
}
- _, _ = resultCache.Invalidate(ctx, nss)
+ if _, err := resultCache.Invalidate(ctx, nss); err != nil {
+ logger.Warn("invalidate changed view namespaces", "error", err, "views", len(nss))
+ }
}As per coding guidelines: "Use structured logging with log/slog and JSON handlers."
Source: Coding guidelines
| ### `discovery/` — Schema Discovery & Validation | ||
|
|
||
| - **discovery.go** — `SchemaRegistry` queries `system.columns` to discover ClickHouse table schemas. Supports periodic auto-refresh, on-demand refresh, and `RetryRefresh` (boot-time exponential backoff loop used by `cmd/wavehouse` so a transiently unreachable ClickHouse doesn't crash-loop the binary). Thread-safe via `sync.RWMutex`. | ||
| - **discovery.go** — `SchemaRegistry` queries `system.columns` to discover ClickHouse table schemas. Supports periodic auto-refresh, on-demand refresh, and `RetryRefresh` (boot-time exponential backoff loop used by `cmd/wavehouse` so a transiently unreachable ClickHouse doesn't crash-loop the binary). The registry also owns the cache-invalidation dependency graph: on each content-changed refresh it resolves every view and materialized-view definition (via `ResolveTables`, below) into a base-table→dependents cascade — carrying an MV's trigger edge through to its `TO` target, parsed off the rendered `create_table_query` (`parseMVTarget`) and propagated along `system.tables.dependencies_table` — exposed as `Dependents()` and pushed into the query cache via `SetOnRefresh`; `IsKnown` reports whether a name resolved cleanly into that cascade (the pipe handler's test for a dependency version invalidation can be trusted to watch — see the `pipes/` section for the full invalidation model). Thread-safe via `sync.RWMutex`. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Fix the garbled clause describing IsKnown.
The parenthetical reads "the pipe handler's test for a dependency version invalidation can be trusted to watch". The sentence has no coherent subject-verb structure, so the meaning of IsKnown is lost at the point the reader needs it.
✏️ Proposed wording
-`IsKnown` reports whether a name resolved cleanly into that cascade (the pipe handler's test for a dependency version invalidation can be trusted to watch — see the `pipes/` section for the full invalidation model)
+`IsKnown` reports whether a name resolved cleanly into that cascade — the pipe handler's test for whether a dependency's version can be trusted to invalidate on write (see the `pipes/` section for the full invalidation model)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| - **discovery.go** — `SchemaRegistry` queries `system.columns` to discover ClickHouse table schemas. Supports periodic auto-refresh, on-demand refresh, and `RetryRefresh` (boot-time exponential backoff loop used by `cmd/wavehouse` so a transiently unreachable ClickHouse doesn't crash-loop the binary). The registry also owns the cache-invalidation dependency graph: on each content-changed refresh it resolves every view and materialized-view definition (via `ResolveTables`, below) into a base-table→dependents cascade — carrying an MV's trigger edge through to its `TO` target, parsed off the rendered `create_table_query` (`parseMVTarget`) and propagated along `system.tables.dependencies_table` — exposed as `Dependents()` and pushed into the query cache via `SetOnRefresh`; `IsKnown` reports whether a name resolved cleanly into that cascade (the pipe handler's test for a dependency version invalidation can be trusted to watch — see the `pipes/` section for the full invalidation model). Thread-safe via `sync.RWMutex`. | |
| - **discovery.go** — `SchemaRegistry` queries `system.columns` to discover ClickHouse table schemas. Supports periodic auto-refresh, on-demand refresh, and `RetryRefresh` (boot-time exponential backoff loop used by `cmd/wavehouse` so a transiently unreachable ClickHouse doesn't crash-loop the binary). The registry also owns the cache-invalidation dependency graph: on each content-changed refresh it resolves every view and materialized-view definition (via `ResolveTables`, below) into a base-table→dependents cascade — carrying an MV's trigger edge through to its `TO` target, parsed off the rendered `create_table_query` (`parseMVTarget`) and propagated along `system.tables.dependencies_table` — exposed as `Dependents()` and pushed into the query cache via `SetOnRefresh`; `IsKnown` reports whether a name resolved cleanly into that cascade — the pipe handler's test for whether a dependency's version can be trusted to invalidate on write (see the `pipes/` section for the full invalidation model). Thread-safe via `sync.RWMutex`. |
| <-started | ||
| // The first resolution is now blocked in flight; give the remaining waiters | ||
| // time to reach the singleflight and join it. A straggler that misses even | ||
| // this window would still be served by the memo after the flight completes, | ||
| // so the assertion below stays stable. | ||
| time.Sleep(100 * time.Millisecond) | ||
| close(release) | ||
| wg.Wait() | ||
|
|
||
| assert.Equal(t, int32(1), calls.Load(), "concurrent cold misses must coalesce onto one EXPLAIN") |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
The sleep-based barrier can produce a false failure, and the comment overstates the guarantee.
The comment says a straggler is still served by the memo. That is not true for this assertion. resolveDeps reads the memo BEFORE it joins resolveSF. A goroutine that reads the empty memo and is then descheduled past close(release) reaches resolveSF.Do after the first flight finished, so it starts a second EXPLAIN. calls becomes 2 and assert.Equal(t, int32(1), ...) fails. A loaded CI machine makes this reachable.
Gate the release on all waiters actually reaching resolveDeps, and keep the sleep only as a short settle window.
🩹 Proposed fix
const waiters = 8
var wg sync.WaitGroup
+ var entered sync.WaitGroup
+ entered.Add(waiters)
results := make([][]cache.Namespace, waiters)
for i := range waiters {
wg.Add(1)
go func() {
defer wg.Done()
+ entered.Done()
results[i], _ = h.resolveDeps(ctx, "SELECT * FROM events")
}()
}
<-started
- // The first resolution is now blocked in flight; give the remaining waiters
- // time to reach the singleflight and join it. A straggler that misses even
- // this window would still be served by the memo after the flight completes,
- // so the assertion below stays stable.
+ // Every waiter goroutine is now scheduled and the first resolution is blocked
+ // in flight; give the remaining waiters time to reach the singleflight and
+ // join it before the flight is allowed to complete.
+ entered.Wait()
time.Sleep(100 * time.Millisecond)
close(release)| if h.Cache != nil { | ||
| _ = h.Cache.Set(r.Context(), cacheKey, nil, data, ttl) | ||
| ttl := cache.QueryTimeToTTL(queryDuration) | ||
| if depsUnresolved { | ||
| // The result can't be reliably version-invalidated — an unfoldable | ||
| // view, an external read (table function/cross-database), or a | ||
| // dead-branch-pruned set (belt-and-suspenders): cap the TTL so it | ||
| // self-expires rather than serving stale on a write the folded | ||
| // versions never see. | ||
| ttl = min(ttl, cache.UnresolvedDepsTTLCap) | ||
| } | ||
| _ = h.Cache.Set(r.Context(), entryKey, data, ttl) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Cache writes are bound to the leader request's cancellable context in both handlers. Both Set calls run inside a singleflight closure owned by whichever request won the race. If that leader disconnects after the query completes, its context is already canceled, so the write is dropped while the waiters still receive the data. The next reader then re-executes the query, which weakens the caching this PR is built around.
internal/api/pipes.go#L284-L294: wrap theSetcontext withcontext.WithoutCancel(r.Context()); keep thedepsUnresolvedTTL cap as written.internal/api/structured_query.go#L229-L230: wrap theSetcontext withcontext.WithoutCancel(r.Context()); keep theIsKnownTTL cap as written.
📍 Affects 2 files
internal/api/pipes.go#L284-L294(this comment)internal/api/structured_query.go#L229-L230
| // Coalesce concurrent cold misses for the same bound query: one EXPLAIN, | ||
| // every waiter shares the outcome. The generation is captured INSIDE the | ||
| // flight, before the EXPLAIN runs, so it dates the schema state the | ||
| // resolution was computed against for every waiter alike. | ||
| v, _, _ := h.resolveSF.Do(boundSQL, func() (any, error) { | ||
| h.pipeDepsMu.RLock() | ||
| gen := h.pipeDepsGen | ||
| h.pipeDepsMu.RUnlock() | ||
| rd, memoize := h.resolvePipe(ctx, boundSQL) | ||
| return resolveOutcome{rd: rd, memoize: memoize, gen: gen}, nil | ||
| }) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
The shared resolution inherits only the leader's request context.
resolveSF.Do runs resolvePipe with ctx from whichever request won the race. resolvePipe derives its 5s deadline from that context. If the leader client disconnects mid-EXPLAIN, the resolution is canceled for every coalesced waiter. All of them then serve on the database-version fallback, and the transient classification means nothing is memoized, so the next request pays another EXPLAIN.
The resolution is already bounded to 5s inside resolvePipe, so detaching cancellation does not create an unbounded operation. Pass a cancellation-detached context into the flight.
🩹 Proposed fix
v, _, _ := h.resolveSF.Do(boundSQL, func() (any, error) {
h.pipeDepsMu.RLock()
gen := h.pipeDepsGen
h.pipeDepsMu.RUnlock()
- rd, memoize := h.resolvePipe(ctx, boundSQL)
+ // The flight is shared, so it must not die with the leader's request.
+ // resolvePipe still bounds itself to 5s.
+ rd, memoize := h.resolvePipe(context.WithoutCancel(ctx), boundSQL)
return resolveOutcome{rd: rd, memoize: memoize, gen: gen}, nil
})📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // Coalesce concurrent cold misses for the same bound query: one EXPLAIN, | |
| // every waiter shares the outcome. The generation is captured INSIDE the | |
| // flight, before the EXPLAIN runs, so it dates the schema state the | |
| // resolution was computed against for every waiter alike. | |
| v, _, _ := h.resolveSF.Do(boundSQL, func() (any, error) { | |
| h.pipeDepsMu.RLock() | |
| gen := h.pipeDepsGen | |
| h.pipeDepsMu.RUnlock() | |
| rd, memoize := h.resolvePipe(ctx, boundSQL) | |
| return resolveOutcome{rd: rd, memoize: memoize, gen: gen}, nil | |
| }) | |
| // Coalesce concurrent cold misses for the same bound query: one EXPLAIN, | |
| // every waiter shares the outcome. The generation is captured INSIDE the | |
| // flight, before the EXPLAIN runs, so it dates the schema state the | |
| // resolution was computed against for every waiter alike. | |
| v, _, _ := h.resolveSF.Do(boundSQL, func() (any, error) { | |
| h.pipeDepsMu.RLock() | |
| gen := h.pipeDepsGen | |
| h.pipeDepsMu.RUnlock() | |
| // The flight is shared, so it must not die with the leader's request. | |
| // resolvePipe still bounds itself to 5s. | |
| rd, memoize := h.resolvePipe(context.WithoutCancel(ctx), boundSQL) | |
| return resolveOutcome{rd: rd, memoize: memoize, gen: gen}, nil | |
| }) |
| tests := []struct{ in, want string }{ | ||
| {"acme", "'acme'"}, | ||
| {"a'b", "'a''b'"}, | ||
| {"a\\b", "'a\\\\b'"}, | ||
| {"org_id = '1'", "'org_id = ''1'''"}, | ||
| } | ||
| for _, tt := range tests { | ||
| if got := QuoteString(tt.in); got != tt.want { | ||
| t.Errorf("QuoteString(%q) = %q, want %q", tt.in, got, tt.want) | ||
| } | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Run each quoting case as a named subtest.
This table has multiple scenarios but does not call t.Run(tt.name, ...). Add a name field and run each assertion in its own subtest.
Proposed change
- tests := []struct{ in, want string }{
- {"acme", "'acme'"},
- {"a'b", "'a''b'"},
- {"a\\b", "'a\\\\b'"},
- {"org_id = '1'", "'org_id = ''1'''"},
+ tests := []struct{ name, in, want string }{
+ {"plain", "acme", "'acme'"},
+ {"quote", "a'b", "'a''b'"},
+ {"backslash", "a\\b", "'a\\\\b'"},
+ {"sql_like", "org_id = '1'", "'org_id = ''1'''"},
}
for _, tt := range tests {
- if got := QuoteString(tt.in); got != tt.want {
- t.Errorf("QuoteString(%q) = %q, want %q", tt.in, got, tt.want)
- }
+ t.Run(tt.name, func(t *testing.T) {
+ if got := QuoteString(tt.in); got != tt.want {
+ t.Errorf("QuoteString(%q) = %q, want %q", tt.in, got, tt.want)
+ }
+ })
}As per coding guidelines, “Use table-driven tests with t.Run(tt.name, ...) for multiple scenarios.”
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| tests := []struct{ in, want string }{ | |
| {"acme", "'acme'"}, | |
| {"a'b", "'a''b'"}, | |
| {"a\\b", "'a\\\\b'"}, | |
| {"org_id = '1'", "'org_id = ''1'''"}, | |
| } | |
| for _, tt := range tests { | |
| if got := QuoteString(tt.in); got != tt.want { | |
| t.Errorf("QuoteString(%q) = %q, want %q", tt.in, got, tt.want) | |
| } | |
| } | |
| tests := []struct{ name, in, want string }{ | |
| {"plain", "acme", "'acme'"}, | |
| {"quote", "a'b", "'a''b'"}, | |
| {"backslash", "a\\b", "'a\\\\b'"}, | |
| {"sql_like", "org_id = '1'", "'org_id = ''1'''"}, | |
| } | |
| for _, tt := range tests { | |
| t.Run(tt.name, func(t *testing.T) { | |
| if got := QuoteString(tt.in); got != tt.want { | |
| t.Errorf("QuoteString(%q) = %q, want %q", tt.in, got, tt.want) | |
| } | |
| }) | |
| } |
Source: Coding guidelines
| } | ||
| return infos, rows.Err() | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Unwrapped rows.Err() returns in the new discovery paths. Three new row-iteration paths return the driver error without context, while every sibling error path in the same functions wraps with fmt.Errorf. All three propagate up through Refresh or ResolveTables, so an operator sees a bare driver message with no indication of which read failed.
internal/discovery/discovery.go#L457-L459: wrap thediscoverColumnsreturn asfmt.Errorf("iterate system.columns rows: %w", err).internal/discovery/discovery.go#L520-L524: wraprerrindiscoverViewMetaasfmt.Errorf("iterate system.tables rows: %w", rerr).internal/discovery/explain.go#L85-L87: wrap theResolveTablesreturn asfmt.Errorf("iterate explain rows: %w", err).
As per coding guidelines: "Return errors instead of panicking, and wrap errors with fmt.Errorf("context: %w", err)."
📍 Affects 2 files
internal/discovery/discovery.go#L457-L459(this comment)internal/discovery/discovery.go#L520-L524internal/discovery/explain.go#L85-L87
Source: Coding guidelines
| { | ||
| name: "dictGet: dictionary captured; the projection's stray '' before it is ignored", | ||
| lines: []string{ | ||
| " CONSTANT id: 2, constant_value: '', constant_value_type: String", | ||
| " FUNCTION id: 3, function_name: dictGet, function_type: ordinary, result_type: String", | ||
| " CONSTANT id: 5, constant_value: 'default.mydict', constant_value_type: String", | ||
| " CONSTANT id: 6, constant_value: 'val', constant_value_type: String", | ||
| }, | ||
| db: "default", | ||
| wantDicts: []string{"mydict"}, | ||
| }, | ||
| { | ||
| name: "dictGet cross-db dropped and marks the read external", | ||
| lines: []string{ | ||
| " FUNCTION id: 3, function_name: dictGet, function_type: ordinary, result_type: String", | ||
| " CONSTANT id: 5, constant_value: 'otherdb.mydict', constant_value_type: String", | ||
| }, | ||
| db: "default", | ||
| wantExternal: true, | ||
| }, | ||
| { | ||
| name: "joinGet and dictGet together", | ||
| lines: []string{ | ||
| " TABLE id: 3, table_name: default.users", | ||
| " FUNCTION id: 4, function_name: joinGet, function_type: ordinary, result_type: String", | ||
| " CONSTANT id: 6, constant_value: 'default.jt', constant_value_type: String", | ||
| " FUNCTION id: 8, function_name: dictGetOrDefault, function_type: ordinary, result_type: String", | ||
| " CONSTANT id: 9, constant_value: 'default.mydict', constant_value_type: String", | ||
| }, | ||
| db: "default", | ||
| wantTables: []string{"jt", "users"}, | ||
| wantDicts: []string{"mydict"}, | ||
| }, |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Add unit cases for dictHas and dictIsIn.
parseExplainTables matches three dict prefixes: dictGet, dictHas, and dictIsIn. The table only covers dictGet and dictGetOrDefault. dictHas and dictIsIn are covered only by the fuzzdeps suite, which is build-tagged and needs a live ClickHouse. A regression that drops either match would pass make ci. The earlier review thread settled on keeping these two names outside the dictGet prefix, so pin that decision here.
💚 Proposed additional cases
{
name: "dictGet cross-db dropped and marks the read external",Add before that entry:
{
name: "dictHas captures the dictionary",
lines: []string{
" FUNCTION id: 3, function_name: dictHas, function_type: ordinary, result_type: UInt8",
" CONSTANT id: 5, constant_value: 'default.mydict', constant_value_type: String",
},
db: "default",
wantDicts: []string{"mydict"},
},
{
name: "dictIsIn captures the dictionary",
lines: []string{
" FUNCTION id: 3, function_name: dictIsIn, function_type: ordinary, result_type: UInt8",
" CONSTANT id: 5, constant_value: 'default.mydict', constant_value_type: String",
},
db: "default",
wantDicts: []string{"mydict"},
},📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| { | |
| name: "dictGet: dictionary captured; the projection's stray '' before it is ignored", | |
| lines: []string{ | |
| " CONSTANT id: 2, constant_value: '', constant_value_type: String", | |
| " FUNCTION id: 3, function_name: dictGet, function_type: ordinary, result_type: String", | |
| " CONSTANT id: 5, constant_value: 'default.mydict', constant_value_type: String", | |
| " CONSTANT id: 6, constant_value: 'val', constant_value_type: String", | |
| }, | |
| db: "default", | |
| wantDicts: []string{"mydict"}, | |
| }, | |
| { | |
| name: "dictGet cross-db dropped and marks the read external", | |
| lines: []string{ | |
| " FUNCTION id: 3, function_name: dictGet, function_type: ordinary, result_type: String", | |
| " CONSTANT id: 5, constant_value: 'otherdb.mydict', constant_value_type: String", | |
| }, | |
| db: "default", | |
| wantExternal: true, | |
| }, | |
| { | |
| name: "joinGet and dictGet together", | |
| lines: []string{ | |
| " TABLE id: 3, table_name: default.users", | |
| " FUNCTION id: 4, function_name: joinGet, function_type: ordinary, result_type: String", | |
| " CONSTANT id: 6, constant_value: 'default.jt', constant_value_type: String", | |
| " FUNCTION id: 8, function_name: dictGetOrDefault, function_type: ordinary, result_type: String", | |
| " CONSTANT id: 9, constant_value: 'default.mydict', constant_value_type: String", | |
| }, | |
| db: "default", | |
| wantTables: []string{"jt", "users"}, | |
| wantDicts: []string{"mydict"}, | |
| }, | |
| { | |
| name: "dictGet: dictionary captured; the projection's stray '' before it is ignored", | |
| lines: []string{ | |
| " CONSTANT id: 2, constant_value: '', constant_value_type: String", | |
| " FUNCTION id: 3, function_name: dictGet, function_type: ordinary, result_type: String", | |
| " CONSTANT id: 5, constant_value: 'default.mydict', constant_value_type: String", | |
| " CONSTANT id: 6, constant_value: 'val', constant_value_type: String", | |
| }, | |
| db: "default", | |
| wantDicts: []string{"mydict"}, | |
| }, | |
| { | |
| name: "dictHas captures the dictionary", | |
| lines: []string{ | |
| " FUNCTION id: 3, function_name: dictHas, function_type: ordinary, result_type: UInt8", | |
| " CONSTANT id: 5, constant_value: 'default.mydict', constant_value_type: String", | |
| }, | |
| db: "default", | |
| wantDicts: []string{"mydict"}, | |
| }, | |
| { | |
| name: "dictIsIn captures the dictionary", | |
| lines: []string{ | |
| " FUNCTION id: 3, function_name: dictIsIn, function_type: ordinary, result_type: UInt8", | |
| " CONSTANT id: 5, constant_value: 'default.mydict', constant_value_type: String", | |
| }, | |
| db: "default", | |
| wantDicts: []string{"mydict"}, | |
| }, | |
| { | |
| name: "dictGet cross-db dropped and marks the read external", | |
| lines: []string{ | |
| " FUNCTION id: 3, function_name: dictGet, function_type: ordinary, result_type: String", | |
| " CONSTANT id: 5, constant_value: 'otherdb.mydict', constant_value_type: String", | |
| }, | |
| db: "default", | |
| wantExternal: true, | |
| }, | |
| { | |
| name: "joinGet and dictGet together", | |
| lines: []string{ | |
| " TABLE id: 3, table_name: default.users", | |
| " FUNCTION id: 4, function_name: joinGet, function_type: ordinary, result_type: String", | |
| " CONSTANT id: 6, constant_value: 'default.jt', constant_value_type: String", | |
| " FUNCTION id: 8, function_name: dictGetOrDefault, function_type: ordinary, result_type: String", | |
| " CONSTANT id: 9, constant_value: 'default.mydict', constant_value_type: String", | |
| }, | |
| db: "default", | |
| wantTables: []string{"jt", "users"}, | |
| wantDicts: []string{"mydict"}, | |
| }, |
| await admin.pipes.delete(pipeName); | ||
| await chQuery(`DROP VIEW IF EXISTS default.\`${mv}\``); | ||
| await chQuery(`DROP TABLE IF EXISTS default.\`${tgt}\``); | ||
| await chQuery(`DROP TABLE IF EXISTS default.\`${src}\``); | ||
| }, 30_000); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Move the teardown so it also runs when an assertion fails.
The four cleanup calls sit at the end of the test body. If any expect fails or either waitForCondition times out, the test throws before line 147. The pipe stays registered in the pipes KV store, and the two tables plus the materialized view stay in ClickHouse. This suite runs with maxWorkers: 1 against shared global state, so leaked objects accumulate across runs and a leaked pipe remains visible to later pipes.list assertions.
Wrap the body in try/finally, or register the teardown right after each object is created.
♻️ Proposed change
- await admin.pipes.delete(pipeName);
- await chQuery(`DROP VIEW IF EXISTS default.\`${mv}\``);
- await chQuery(`DROP TABLE IF EXISTS default.\`${tgt}\``);
- await chQuery(`DROP TABLE IF EXISTS default.\`${src}\``);
}, 30_000);and wrap the steps that follow object creation:
try {
// refresh, pipe set, exec/assert, ingest, waits, final assertions
} finally {
await admin.pipes.delete(pipeName);
await chQuery(`DROP VIEW IF EXISTS default.\`${mv}\``);
await chQuery(`DROP TABLE IF EXISTS default.\`${tgt}\``);
await chQuery(`DROP TABLE IF EXISTS default.\`${src}\``);
}As per path instructions: "Keep E2E tests sequential with maxWorkers: 1 because they share mutable global policy state."
Source: Path instructions
| boundSQL := strings.ReplaceAll(sql, "{{source:web}}", "'web'") | ||
| res, err := discovery.ResolveTables(ctx, e.chConn, testCHDatabase, boundSQL) | ||
| require.NoError(t, err) | ||
| require.Contains(t, res.Tables, web, "the live arm's table stays tracked") | ||
| require.NotContains(t, res.Tables, mobile, "the constant-false arm's table must be pruned") | ||
| require.True(t, res.Pruned, "ResolveTables must report the prune (it drives the TTL cap)") |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Locate the pipe parameter binder and its call site in the handler.
set -euo pipefail
fd -t f 'pipes.go' internal | while IFS= read -r f; do
echo "=== outline $f ==="
ast-grep outline "$f" --items all
done
echo "=== BindParams definition ==="
ast-grep run --pattern 'func $$$BindParams($$$) { $$$ }' --lang go internal || true
rg -nP --type=go -C 5 '\bfunc\s+(\([^)]*\)\s+)?BindParams\s*\('
echo "=== BindParams call sites ==="
rg -nP --type=go -C 4 '\bBindParams\s*\('Repository: Wave-RF/WaveHouse
Length of output: 3856
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== tests/integration/pipe_cache_test.go focus ==="
sed -n '120,170p' tests/integration/pipe_cache_test.go
echo "=== internal/pipes/pipes.go BindParams and formatParamValue ==="
sed -n '140,245p' internal/pipes/pipes.go
echo "=== internal/api/pipes.go Execute resolution path ==="
sed -n '180,320p' internal/api/pipes.go
echo "=== tests references for pin/coupling SQL handling ==="
rg -n --type=go -C 3 'pin|prune|BindParams|ResolveTables|ResolveTablesFn|source:web' tests internalRepository: Wave-RF/WaveHouse
Length of output: 50374
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== internal/pipes/pipes.go imports and QuoteString ==="
sed -n '1,20p' internal/pipes/pipes.go
rg -n --type=go -C 4 'func QuoteString|QuoteString\(' internal
echo "=== pipes helper used by tests maybe ==="
sed -n '1,170p' tests/integration/pipe_cache_test.goRepository: Wave-RF/WaveHouse
Length of output: 11041
Bind the pin SQL with pipes.BindParams.
streams.ReplaceAll(sql, "{{source:web}}", "'web'") duplicates the binder’s current output. Use pipes.BindParams(&pipes.NamedQuery{SQL: sql}, nil) so the pruning pin stays tied to the same BindParams rendering that /v1/pipes/{name}/execute uses.
github-project-automation
Bot
moved this from Ready
to In review
in WaveHouse Task Board
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