Documentation
¶
Overview ¶
Package pipe parses .pipe project files into model.Pipe values, stores them in a hot-swappable in-memory registry (ADR 0020), and executes published endpoints against ClickHouse (ADRs 0003, 0009, 0012). Pipes are sequences of NODE blocks; the terminal node (TYPE endpoint, or the last node) is the published endpoint, with earlier nodes composed in as CTEs at query time.
Index ¶
- func Parse(name, raw string) (*model.Pipe, error)
- func ParseFile(path string) (*model.Pipe, error)
- type Executor
- func (e *Executor) EnableCopy(w model.CHWriter) *Executor
- func (e *Executor) EnableJobs(store model.JobStore) *Executor
- func (e *Executor) Run(ctx context.Context, name string, params url.Values) (body []byte, status int, err error)
- func (e *Executor) RunCopy(ctx context.Context, name string, params url.Values) (body []byte, status int, err error)
- func (e *Executor) RunFormat(ctx context.Context, name string, params url.Values, format model.OutputFormat) (body []byte, status int, err error)
- type Registry
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Executor ¶
type Executor struct {
// contains filtered or unexported fields
}
Executor runs published pipe endpoints against ClickHouse (model.PipeRunner). Query parameters are bound via ClickHouse's native {name:Type} placeholders, never string-interpolated, so endpoints are injection-proof (ADR 0003). Each run emits a best-effort observability stat to a StatsRecorder (ADR 0014).
func NewExecutor ¶
func NewExecutor(ch model.CHQuerier, pipes model.PipeRegistry, ds model.DatasourceRegistry, rec model.StatsRecorder) *Executor
NewExecutor wires the runner to its ClickHouse querier, registries, and the stats recorder (rec may be nil; ADR 0014 — observability is best-effort). The copy write path is off by default; call EnableCopy to wire it.
func (*Executor) EnableCopy ¶
EnableCopy wires the read-write path RunCopy needs (INSERT INTO ... SELECT for copy pipes). Kept off the constructor so existing query-only callers are unaffected. Returns e for chaining.
func (*Executor) EnableJobs ¶ added in v0.3.16
EnableJobs wires a JobStore so completed copy jobs are recorded for GET /v0/jobs / GET /v0/jobs/{id} (gap #8). Kept off the constructor like EnableCopy so existing callers are unaffected; nil store leaves RunCopy's response unchanged, it just isn't recorded anywhere.
func (*Executor) Run ¶
func (e *Executor) Run(ctx context.Context, name string, params url.Values) (body []byte, status int, err error)
Run executes the endpoint of the named pipe and returns the ClickHouse JSON body. It is the JSON shorthand for RunFormat (ADR 0003).
func (*Executor) RunCopy ¶
func (e *Executor) RunCopy(ctx context.Context, name string, params url.Values) (body []byte, status int, err error)
RunCopy triggers a TYPE copy pipe on demand: it composes the copy SQL, binds request params (identical pipeline to a query), and runs INSERT INTO <target> SELECT ... over the read-write path. Tinybird models this as an async job; TinyRaven runs it synchronously and returns a job-shaped body with a terminal status so existing copy clients parse the response unchanged.
ponytail: execution is synchronous even when triggered by internal/scheduler (COPY_SCHEDULE) rather than the HTTP handler — there is no queue/worker, the scheduler's own goroutine blocks on this call. Fine at MVP scale; a slow copy would just delay that tick's other due pipes. The returned job is already "done" (or surfaced as a 400 CH error), so GET /v0/jobs/{id} has nothing to transition to; it exists so job_url is a live link instead of shape-only parity. If e.jobs is wired, the record is persisted best-effort (a store failure never fails the copy response — the job already ran).
func (*Executor) RunFormat ¶
func (e *Executor) RunFormat(ctx context.Context, name string, params url.Values, format model.OutputFormat) (body []byte, status int, err error)
RunFormat executes the endpoint of the named pipe and returns the body in the requested output format. status is the HTTP status the API should send; err carries the failure reason for client-mappable cases (ADR 0012). Only the trailing ClickHouse FORMAT varies by format — param binding, control flow and validation are identical across formats.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry is an in-memory model.PipeRegistry with lock-free reads and an atomic snapshot swap for hot reload (ADR 0020). Get/List read the current snapshot without blocking writers; Put/Replace serialize via mu and publish a new immutable snapshot copy-on-write.
Source Files