Repository navigation
feat(pipes): resolve table dependencies for cache invalidation #343
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
taitelee
wants to merge
32
commits into
main
Choose a base branch
from
pipe-cache-invalidation
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
Changes from 12 commits
Commits
Show all changes
32 commits
Select commit
Hold shift + click to select a range
1d8d350
feat(cache): namespace-versioned cache keys for queries and pipes
taitelee 218cad0
condensed new namespace query functions
taitelee 289fa11
refactor(cache): cut over to namespace-versioned Get/Set/Invalidate
taitelee 13427bc
make fix
taitelee ad1c0ff
Merge remote-tracking branch 'origin/main' into pipe-cache-invalidation
taitelee 1eeb929
perf(cache): skip redundant per-scope bumps on whole-table invalidation
taitelee 377636c
docs(changelog): add Unreleased entry for the cache cutover; encode q…
taitelee 97bbbc9
refactor(ingest): collapse whole-table-subsumed scopes in the worker,…
taitelee b614ea3
merge main
taitelee 3bae7ec
changelog
taitelee 27ae1f1
refactor(pipes): drop declared param-type enforcement
taitelee 038a205
fix(pipes): handle escaped backticks in query-tree identifier parsing…
taitelee 142dc00
merge main
taitelee 43ac5c3
fix(pipes): resolve through-view table deps via system.tables
taitelee b1e4bcc
feat(pipes): log dep-resolution outcome, document TTL-only fallbacks,…
taitelee 0d7097e
fix(pipes): resolve through registry-discovered views so ingest still…
taitelee 394420d
Merge remote-tracking branch 'origin/main' into pipe-cache-invalidation
taitelee 90e24b4
fix(pipes): drop view/MV names from resolved deps, keeping only their…
taitelee 13803e1
test(pipes): table-driven dep-resolution tests, document cache invali…
taitelee f762148
test(integration): allowlist-sanitize createTable names so subtest pu…
taitelee d6f447a
fix(pipes): dummy-bind array params for table-dep resolution; tighten…
taitelee a24be1a
feat(pipes): invalidate cached results via table dependencies resolve…
taitelee 5727a3b
Merge remote-tracking branch 'origin/main' into pipe-cache-invalidation
taitelee 7f13a70
fix(pipes): TTL-floor cached results with unresolvable deps (unfoldab…
taitelee 09a37ee
Merge remote-tracking branch 'origin/main' into pipe-cache-invalidation
taitelee 1083703
feat(pipes): meter degraded dependency resolution; add weird-identifi…
taitelee d6f047d
doc updates
taitelee 786e768
fix(pipes,cache): snapshot version-folded keys pre-query (#382); data…
taitelee 8267f7a
Merge remote-tracking branch 'origin/main' into pipe-cache-invalidation
taitelee e7ed7fa
feat(discovery): cascade source writes through MV TO targets via trig…
taitelee acf93a1
test(e2e): cover MV TO-target pipe invalidation end-to-end (fixes e2e…
taitelee 4d856a5
docs(architecture): cover discovery's dependency-cascade role and exp…
taitelee File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Some comments aren't visible on the classic Files Changed page.
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,215 @@ | ||
| package api | ||
|
|
||
| import ( | ||
| "context" | ||
| "sort" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/ClickHouse/clickhouse-go/v2/lib/driver" | ||
| "github.com/Wave-RF/WaveHouse/internal/cache" | ||
| "github.com/Wave-RF/WaveHouse/internal/discovery" | ||
| "github.com/Wave-RF/WaveHouse/internal/pipes" | ||
| "github.com/Wave-RF/WaveHouse/internal/query" | ||
| ) | ||
|
|
||
| // pipeResolveTimeout bounds the EXPLAIN round-trip a pipe Put() makes to resolve | ||
| // its table dependencies, so a slow or stuck ClickHouse can't hang pipe | ||
| // creation. Generous because EXPLAIN QUERY TREE only analyzes (it scans no data) | ||
| // and Put() is a rare admin write, not a per-request cost. | ||
| const pipeResolveTimeout = 10 * time.Second | ||
|
|
||
| // resolvePipeDeps resolves the ingested base tables a pipe reads, for cache | ||
| // invalidation. Best-effort: if no schema registry / ClickHouse connection is | ||
| // wired, or resolution fails, it returns nil and the pipe falls back to TTL-only | ||
| // caching. Because the table set depends only on the SQL (not on per-request | ||
| // parameter values), resolving here at Put() — a rare admin write — keeps it off | ||
| // the per-request read path and means it is computed once per definition. | ||
|
taitelee marked this conversation as resolved.
Resolved
|
||
| func (h *PipesHandler) resolvePipeDeps(ctx context.Context, q *pipes.NamedQuery) []string { | ||
| if h.Registry == nil || h.CHConn == nil { | ||
| return nil | ||
| } | ||
| ctx, cancel := context.WithTimeout(ctx, pipeResolveTimeout) | ||
| defer cancel() | ||
| return resolvePipeTables(ctx, h.CHConn, h.Registry, h.Database, q) | ||
| } | ||
|
|
||
| // resolvePipeTables binds the pipe with dummy parameters, asks ClickHouse to | ||
| // resolve the query — through views and aliases — to the tables it actually | ||
| // reads via EXPLAIN QUERY TREE, then keeps only the names the schema registry | ||
| // knows in the configured database (i.e. tables the ingest worker writes and can | ||
| // therefore version-invalidate). Any failure returns nil so the caller degrades | ||
| // to TTL-only with no error surfaced. | ||
| func resolvePipeTables(ctx context.Context, conn driver.Conn, registry *discovery.SchemaRegistry, database string, q *pipes.NamedQuery) []string { | ||
| if conn == nil || registry == nil || q == nil { | ||
| return nil | ||
| } | ||
| boundSQL, err := pipes.DummyBind(q) | ||
| if err != nil { | ||
| return nil | ||
| } | ||
| raw, err := explainQueryTreeTables(ctx, conn, boundSQL) | ||
| if err != nil { | ||
| return nil | ||
| } | ||
| return filterKnownTables(raw, database, registry) | ||
| } | ||
|
|
||
| // explainQueryTreeTables runs EXPLAIN QUERY TREE over sql and returns the raw | ||
| // `database.table` identifiers it references. Running the analyzer's passes | ||
| // (run_passes = 1 runs them all) resolves views and aliases down to their | ||
| // underlying tables, which static SQL parsing cannot do — the whole reason this | ||
| // goes through ClickHouse rather than parsing the FROM clause. Read-only, so it | ||
| // is safe to issue for any pipe (pipes are read queries). | ||
| func explainQueryTreeTables(ctx context.Context, conn driver.Conn, sql string) ([]string, error) { | ||
| rows, err := conn.Query(ctx, "EXPLAIN QUERY TREE run_passes = 1 "+sql) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| defer func() { _ = rows.Close() }() | ||
|
|
||
| var lines []string | ||
| for rows.Next() { | ||
| var line string | ||
| if err := rows.Scan(&line); err != nil { | ||
| return nil, err | ||
| } | ||
| lines = append(lines, line) | ||
| } | ||
| if err := rows.Err(); err != nil { | ||
| return nil, err | ||
| } | ||
| return parseQueryTreeTables(lines), nil | ||
| } | ||
|
|
||
| // parseQueryTreeTables pulls table identifiers out of an EXPLAIN QUERY TREE dump. | ||
| // After the analyzer passes run, every table the query reads — including those | ||
| // reached through views and aliases — appears as a TABLE node carrying a | ||
| // `table_name: <database>.<table>` field. We collect those raw identifiers; the | ||
| // database qualifier and backtick quoting are handled by filterKnownTables. | ||
| func parseQueryTreeTables(lines []string) []string { | ||
| const marker = "table_name:" | ||
| var out []string | ||
| for _, line := range lines { | ||
| _, after, found := strings.Cut(line, marker) | ||
| if !found { | ||
| continue | ||
| } | ||
| if id := identifierField(after); id != "" { | ||
| out = append(out, id) | ||
| } | ||
| } | ||
| return out | ||
| } | ||
|
|
||
| // identifierField returns the leading identifier of a query-tree field value: | ||
| // everything up to the first comma that is not inside backticks, trimmed. The | ||
| // dump separates a node's fields with ", " and quotes identifiers containing | ||
| // special characters with backticks, so a naive comma split would truncate a | ||
| // quoted name that itself contains a comma. A backslash-escaped backtick (\`) | ||
| // inside a quoted name does not end the quoting. | ||
| func identifierField(s string) string { | ||
| s = strings.TrimSpace(s) | ||
| inTick := false | ||
| for i := 0; i < len(s); i++ { | ||
| switch s[i] { | ||
| case '\\': | ||
| if inTick { | ||
| i++ // skip the escaped char (\` or \\) so it can't flip inTick | ||
| } | ||
| case '`': | ||
| inTick = !inTick | ||
| case ',': | ||
| if !inTick { | ||
| return strings.TrimSpace(s[:i]) | ||
| } | ||
| } | ||
| } | ||
| return strings.TrimSpace(s) | ||
| } | ||
|
|
||
| // filterKnownTables normalizes raw `database.table` identifiers to bare table | ||
| // names and keeps only those the schema registry knows in the configured | ||
| // database — i.e. tables the ingest worker writes and can version-invalidate. A | ||
| // qualifier naming a different database is dropped (ingest can't signal changes | ||
| // to it), as is anything the registry hasn't discovered (views, system tables, | ||
| // unknown names): a pipe may still read those, but WaveHouse has no change signal | ||
| // for them, so they stay covered by the TTL floor. Results are deduped and | ||
| // sorted for a stable, order-independent dependency set. | ||
| func filterKnownTables(raw []string, database string, registry *discovery.SchemaRegistry) []string { | ||
| seen := make(map[string]struct{}) | ||
| out := make([]string, 0, len(raw)) | ||
| for _, r := range raw { | ||
| db, table := splitQualified(r) | ||
| if db != "" && db != database { | ||
| continue // a table in another database — outside the ingest universe | ||
|
taitelee marked this conversation as resolved.
Resolved
|
||
| } | ||
| if table == "" || registry.Get(table) == nil { | ||
| continue | ||
| } | ||
| if _, ok := seen[table]; ok { | ||
| continue | ||
| } | ||
| seen[table] = struct{}{} | ||
| out = append(out, table) | ||
| } | ||
| sort.Strings(out) | ||
|
taitelee marked this conversation as resolved.
Resolved
|
||
| return out | ||
| } | ||
|
|
||
| // splitQualified splits a ClickHouse `database.table` (or bare `table`) | ||
| // identifier into its parts, tolerating the backtick quoting ClickHouse uses for | ||
| // names with special characters. Only the first dot outside backticks separates | ||
| // the database qualifier, so a quoted table name containing dots stays intact. A | ||
| // backslash-escaped backtick (\`) inside a quoted part does not end the quoting. | ||
| // A bare name returns db "". | ||
| func splitQualified(s string) (db, table string) { | ||
| s = strings.TrimSpace(s) | ||
| inTick := false | ||
| for i := 0; i < len(s); i++ { | ||
| switch s[i] { | ||
| case '\\': | ||
| if inTick { | ||
| i++ // skip the escaped char (\` or \\) so it can't flip inTick | ||
| } | ||
| case '`': | ||
| inTick = !inTick | ||
| case '.': | ||
| if !inTick { | ||
| return unquoteIdent(s[:i]), unquoteIdent(s[i+1:]) | ||
| } | ||
| } | ||
| } | ||
| return "", unquoteIdent(s) | ||
| } | ||
|
|
||
| // unquoteIdent strips ClickHouse backtick quoting from a single identifier and | ||
| // unescapes the `\“ and `\\` sequences inside it, so the result matches the | ||
| // bare name the schema registry is keyed by. | ||
| 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, "\\\\", "\\") | ||
| } | ||
| return s | ||
| } | ||
|
taitelee marked this conversation as resolved.
Resolved
|
||
|
|
||
| // pipeDeps maps a pipe's resolved base tables to cache dependency namespaces. | ||
| // The scope is empty (whole-table): a pipe can't know which scope of a table it | ||
| // reads, and a scoped write bumps the whole-table view too, so a whole-table | ||
| // dependency is still correctly invalidated by any write to the table. Each name | ||
| // is NATS-encoded exactly as the ingest worker encodes it (worker.go | ||
| // handleSuccess), so the read and invalidation sides build identical keys. nil | ||
| // in, nil out — keeping pipes keyed by sha alone (TTL-only). | ||
| func pipeDeps(tables []string) []cache.Namespace { | ||
| if len(tables) == 0 { | ||
| return nil | ||
| } | ||
| deps := make([]cache.Namespace, 0, len(tables)) | ||
| for _, t := range tables { | ||
| deps = append(deps, cache.Namespace{Table: query.SafeEncodeNATS(t)}) | ||
| } | ||
| return deps | ||
| } | ||
|
taitelee marked this conversation as resolved.
Resolved
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.