> ## Documentation Index > Fetch the complete documentation index at: https://docs.blockx.chaintable.com/llms.txt > Use this file to discover all available pages before exploring further. # Plugin system > Call Builder and Writer Plugin (ResultHandler): the pluggable phases at both ends of a Task, and how to register, assemble, run, and extend them The plugin system consists of two groups of pluggable implementations inside the Worker process: the **Call Builder** turns the task configuration into a `CallList` before the Executor phase; the **Writer Plugin** (called ResultHandler on the task side) receives the return values of all calls after the Executor phase and persists or passes them through. They correspond to the Builder phase and the Writer phase of the main Task path in the [Architecture overview](/en/architecture/overview), and are driven by the [Worker](/en/components/worker) orchestrator. ## Responsibilities and boundaries * A Call Builder is only responsible for "reading the input and producing a `CallList`". It does not dispatch calls, fetch function code, or cache results. * A Writer Plugin is only responsible for "consuming `outputs` and writing out or returning the result". It takes no part in call retry or deduplication. * A task has exactly one `FunctionCallConfig` (`type/config`) and at most one `ResultHandler` (`type/config`). `type` is the registered name and `config` is a `json.RawMessage` that the corresponding implementation parses itself. * Neither Builders nor Plugins hold IO at construction time; the Worker passes `TaskIOReader` / `TaskIO` as arguments when calling `Build` / `Execute`, and IO rate limiting and retry are done in the IO subsystem (see [IO access subsystem](/en/components/io-subsystem)). * Which Builders / Plugins a Worker process registers is declared by the entry point's `app.Profile`. A `type` not in the Profile misses on lookup at runtime, and the task fails cleanly with `BUILDER_NOT_FOUND` / `PLUGIN_NOT_FOUND`. * Writer Plugins are fail-fast: if `Execute` returns an error, the task fails; whether upstream rescheduling is allowed is decided by `PluginError.Retryable`. * Every non-nil element of `outputs` must be a `json.RawMessage` containing either a single object (one row) or an array of objects (multiple rows). Table-writing plugins use `types.VisitOutputRows` for uniform validation and iteration. * A streaming Builder (`StreamingCallBuilder`) splits "static parsing" and "scanning and producing" into two phases: a `PrepareStream` failure is still an atomic failure with zero calls dispatched; if `Run` fails midway, earlier batches may already have executed. ## Code location | Path | Purpose | | ------------------------------------------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `internal/plugin/callbuilder/types/` | `CallBuilder` / `StreamingCallBuilder` interfaces, `Call` / `CallList`, `CallBuilderName` enum, inline code normalization, Substrait-style `Plan` (`operator.go`) | | `internal/plugin/callbuilder/registry.go` | `BuilderRegistry`: thread-safe named registry | | `internal/plugin/callbuilder/dbscan/` | `DBScanBuilder` (`BlockTableCallConfig`): reads rows from BlockDB by block | | `internal/plugin/callbuilder/call_list/` | `CallListCallConfigBuilder` (`CallListCallConfig`): explicit call list, no IO | | `internal/plugin/callbuilder/bundlescan/` | `BundleScanBuilder` (`BlockBundleCallConfig`): reads S3 parquet through DuckDB; the only builder that implements the streaming contract. Expanded in [Bundle clusters](/en/components/bundle) | | `internal/plugin/event/types/` | `WriterPlugin` interface, `PluginName` enum, `PluginResult`, `PluginError`, `VisitOutputRows` / `PreviewOutputs` | | `internal/plugin/event/registry.go` | `PluginRegistry` | | `internal/plugin/event/blockdbwrite/` | `BlockWriteResultHandler`: writes BlockDB tables by block | | `internal/plugin/event/returnvalue/` | `ReturnValueResultHandler`: concatenates outputs into a single JSON array and passes it through to the caller | | `internal/plugin/event/tableupserts/` | `TableUpsertsResultHandler`: writes regular L1 tables, with block and bundle write paths | | `internal/plugin/event/batchwrite/` | BlockDB BatchWrite job lifecycle (Init / presigned PUT / Commit), shared by tableupserts and bundlewrite | | `internal/plugin/event/bundlewrite/` | `BlockBundleWriteResultHandler`: overwrites by bundle. Expanded in [Bundle clusters](/en/components/bundle) | | `internal/plugin/schema/` | `ReadColumns` / `ReadTable`: reads logical-types table schemas through `TaskIOReader` | | `internal/worker/devstub/` | Stub builders (`static` / `payload`) and plugins (`log` / `test`) for dev / e2e | | `internal/worker/app/app.go` | `Profile.Builders` / `Profile.Plugins`; `newBuilderRegistry` / `newPluginRegistry` construct and register by name | | `internal/worker/adapters/orchestrator_phases.go` | Worker call sites: `runBuilderPhase`, `runWriterPhase` | | `cmd/worker/main.go`, `cmd/bundle_worker/main.go` | The Profile manifest of each of the two entry points | | `internal/plugin/**/*_test.go`, `internal/plugin/pipeline_integration_test.go` | Unit and integration tests | | `docs/specs/plugin-system.md` and others | Design specs, see the end of this page | ## Core types and interfaces The following signatures are copied from the code. ```go theme={null} // internal/plugin/callbuilder/types/types.go type CallBuilder interface { Name() CallBuilderName Build(ctx context.Context, taskCtx *commontypes.TaskCtx, io TaskIOReader) (CallList, error) } type StreamingCallBuilder interface { CallBuilder PrepareStream(ctx context.Context, taskCtx *commontypes.TaskCtx, io TaskIOReader) (CallStreamProducer, error) } type CallStreamProducer interface { Run(ctx context.Context, emit BatchEmit) error } type BatchEmit func(ctx context.Context, batch CallList) error // internal/plugin/event/types/types.go type WriterPlugin interface { Name() PluginName Execute(ctx context.Context, taskCtx *commontypes.TaskCtx, outputs []any, io TaskIO) (any, error) } ``` | Type / function | File | Description | | ----------------------------------------------------------- | ------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `CallBuilderName` | `callbuilder/types/types.go` | Registered name. Built-in values: `BuilderDBScan = "BlockTableCallConfig"`, `BuilderCallList = "CallListCallConfig"`, `BuilderBundleScan = "BlockBundleCallConfig"` | | `Call` | same as above | One unit of work: `FunctionID`, `Code` (inline), `Args` or `ArgsJSON` (compact pre-serialized form, takes precedence), `CallID`, `NoResultCache`, `ArgsPooled`. The last three fields are `json:"-"` and only used in-process | | `CompactCall` / `CompactArgs` | same as above | Used by scan-type builders to marshal each row's args into `ArgsJSON` once, avoiding keeping millions of row map object graphs resident | | `NormalizeInlineCode` / `InlineCodeFunctionID` | same as above | Trusted ingress for inline code: computes the digest once; `functionId` and a non-empty `code.sourceCode` are mutually exclusive; the FunctionID of inline code is `inline-` | | `ErrProductionStopped` / `CallStreamMarker` | same as above | Sentinels of the streaming contract: the former means the Worker stopped production in a controlled way; the latter is the `PhaseOutcome.Result` by which the Builder phase tells core to "enter the streaming Calls phase" | | `Plan` and `ApplyInMemory` / `Validate` / `ToFilterSQLArgs` | `callbuilder/types/operator.go` | Substrait-style `relations` array (read / filter / deduplicate); DBScan runs it with the in-memory interpreter, BundleScan translates it into SQL and pushes it down to DuckDB | | `TaskCtx` / `FunctionCallConfig` / `ResultHandler` | `internal/common/types/types.go` | `TaskCtx` embeds `TaskInput` (`ID`, `FunctionCallConfig{Type, Config}`, `ResultHandler *{Type, Config}`, `TaskTimeoutMs`) plus `Stage`, `DebugMode`, `Deadline`, `Ctx`, `Cancel` | | `BuilderRegistry` / `PluginRegistry` | `callbuilder/registry.go`, `event/registry.go` | `Register` errors on duplicate names, `Lookup` returns nil on a miss | | `PluginName` | `event/types/types.go` | `PluginBlockDBWrite = "BlockWriteResultHandler"`, `PluginReturnValue = "ReturnValueResultHandler"`, `PluginBundleWrite = "BlockBundleWriteResultHandler"`, `PluginTableUpserts = "TableUpsertsResultHandler"` | | `PluginError` / `IsRetryableError` | `event/types/errors.go` | Unified error model; the `Retryable` field decides whether the orchestrator marks the task as reschedulable. A bare error that does not use `PluginError` is treated as retryable by default | | `VisitOutputRows` / `ConsumeOutputRows` / `PreviewOutputs` | `event/types/output_rows.go`, `output_preview.go` | Validate and visit outputs row by row; `Preview` serves `debug:true` mode, which only counts rows and materializes at most 3 | | `PluginResult` | `internal/common/types/types.go` | Normalized result the Writer phase returns to the caller (`Success`, `FailureCode`, `Retryable`, `Result`) | | `schema.ReadColumns` | `internal/plugin/schema/columns.go` | Used by table-writing plugins and bundlescan to get the target table's column definitions; projects only the columns that appear in the output and belong to the target table | `event/types` also declares a `SchemaProvider` interface, but it currently has no production callers; all column definitions go through `schema.ReadColumns`. ## Data flow / execution flow ```mermaid theme={null} sequenceDiagram participant O as Orchestrator participant BR as BuilderRegistry participant B as CallBuilder participant D as Dispatcher / Executor participant PR as PluginRegistry participant P as WriterPlugin O->>BR: Lookup(FunctionCallConfig.Type) alt non-streaming or StreamBuild.Enabled=false O->>B: Build(ctx, taskCtx, ioScope) B-->>O: CallList else StreamingCallBuilder and the switch is on O->>B: PrepareStream(ctx, taskCtx, ioScope) B-->>O: CallStreamProducer O->>B: producer.Run(ctx, emit) under scanSem in the Calls phase end O->>D: dispatch calls, collect outputs []json.RawMessage O->>PR: Lookup(ResultHandler.Type) O->>P: Execute(ctx, taskCtx, outputs, ioScope) P-->>O: result any, error wrapped as PluginResult ``` `Orchestrator.activateAndRun` (`orchestrator_phases.go`) retrieves the parsed `TaskCtx` from core, injects `Ctx` / `Cancel`, and promotes `debug:true` in `ResultHandler.Config` to `TaskCtx.DebugMode`. `runBuilderPhase` first acquires `builderSem` (`Admission.BuilderSlots`), then does a `Lookup` in `builderRegistry` by `FunctionCallConfig.Type`. If it hits a `StreamingCallBuilder` and `StreamBuild.Enabled` is on, it goes through `PrepareStream`; otherwise `Build`. On success, `releaseBuilderConfigPayload` sets `FunctionCallConfig.Config` to nil to release the request JSON. Failure codes: `BUILDER_NOT_FOUND` / `BUILDER_FAILED`; `Retryable` depends on whether the error is a retryable `IOError`. Non-streaming: `runCallPhase` hands the whole `CallList` to the dispatcher. Streaming: `runCallStreamPhase` runs `producer.Run` under `scanSem`, and `makeBatchEmit` provides an `emit` callback with credit-based (`MaxOutstandingBatches`) backpressure. See [Call execution subsystem](/en/components/call-execution) for details. `runWriterPhase` acquires `pluginSem` (`Admission.PluginSlots`); if `ResultHandler` is non-nil, it does a `Lookup` and calls `plugin.Execute(resultCtx, ioTaskCtx, cmd.Outputs, ioScope)`; the return value is wrapped as `commontypes.PluginResult` and goes into `TaskResult.ExecuteResult.PluginResults`. When `ResultHandler` is nil, the phase succeeds immediately. ### Built-in Call Builders * **DBScanBuilder** (`dbscan/dbscan.go`): parses `DBScanBuilderDecl{Block, TriggerSources}`; one goroutine per `TriggerSource{Table, Operator, FunctionID, Code, Params}` (`errgroup.SetLimit(3)`). For each source: validate the table name identifier → `NormalizeInlineCode` → `Operator.Validate` → `io.Read` sends `blockdb.ReadOp[*GetBlockRowsRequest]` (`Op: "GetBlockRows"`, the logical table name is passed to BlockDB as-is) → `Operator.ApplyInMemory(rows)` (filter first, then deduplicate, in relation chain order) → build args from the `Params` template. Positions in `Params` equal to `"${}"` are replaced with the row map; when `Params` is empty, the whole row is the only argument; when `Params` is non-empty but has no placeholder, all calls share the same static `ArgsJSON`. The `CallID` format is `{taskID}-{table}-{height}-{i}`. * **CallListCallConfigBuilder** (`call_list/call_list.go`): parses `{function, code, callList: [][]any}`, does no IO, and generates `Call{FunctionID, Args, CallID: "{taskID}-{i}"}` for the i-th item. * **BundleScanBuilder** (`bundlescan/`): resolves parquet locations from `blockBundle.number` + `triggerSources[].tableId` (or the compatibility field `bundleBucket`), streams rows through DuckDB, and translates the `Operator` into SQL pushed down to it. It also implements `StreamingCallBuilder` (`stream.go`, batch boundaries `defaultBatchMaxRows = 8192` / `defaultBatchMaxArgsBytes = 32 MiB`) and sets `Call.NoResultCache` when row-key uniqueness can be proven. See [Bundle clusters](/en/components/bundle) for details. ### Operator (Plan) `TriggerSource.Operator` is a `*types.Plan` whose JSON looks like `{"relations": [{"read": ...}, {"filter": ...}, {"deduplicate": ...}]}`; relations are chained into a linear list through `input.relation_id` (array index). `Plan.Validate` requires exactly one `read`, a chain with no branches, and every `selection.field` within `read.base_schema.names`. filter uses SQL three-valued logic (comparisons with NULL yield `UNKNOWN`, only `TRUE` is kept), and `equal(col, null)` is interpreted as `IS NULL`; deduplicate keys use a deterministic encoding with type tags and length prefixes (`appendDedupValue`). Supported `function_reference` values: `and`, `or`, `not`, `equal`, `not_equal`, `gt`, `gte`, `lt`, `lte`, `is_null`, `is_not_null`. `Plan.UnmarshalJSON` also accepts a legacy form: an expression tree with a top-level `type` field (`Attribute` / `Literal` / `EqualTo` / `In` / `And` / `Or` / `Not`, etc.), which `legacyConditionToPlan` converts into a single filter relation. See `docs/specs/substrait_ast.md` in the blockx repo for the full format. ### Built-in Writer Plugins | Name | Package | Configuration (`ResultHandler.Config`) | Behavior | | ------------------------------- | -------------- | ------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `BlockWriteResultHandler` | `blockdbwrite` | `{targetTable, block{id,height,timestamp}}` | `schema.ReadColumns` gets the columns → `VisitOutputRows` projects into `WriteRow` → a single `blockdb.BlockWriteReq` (`Sync:false`). Returns `{written_rows, finish_time}` | | `ReturnValueResultHandler` | `returnvalue` | None | Concatenates outputs as-is into one JSON array (`json.RawMessage`) placed in `PluginResult.Result`; nil elements are written as `null` | | `TableUpsertsResultHandler` | `tableupserts` | `{targetTable, condition?{column, policy}}` | The block profile uses `TableWriteService.UpsertRows`; the bundle profile uses `batchwrite.Writer.Submit` (`NormalTableUpsertStrategy`). `policy` supports `UpdateIfSmaller` / `UpdateIfLarger` | | `BlockBundleWriteResultHandler` | `bundlewrite` | `{targetTable, blockBundle{number}}` | Overwrite through `batchwrite` doing Init / presigned PUT / Commit. See [Bundle clusters](/en/components/bundle) | When `TaskCtx.DebugMode` is true, all table-writing plugins only `PreviewOutputs` to count rows and emit a `table_write_debug` log; they neither read the schema nor send write requests. ### devstub `internal/worker/devstub/` provides `StaticCallBuilder` (name `static`, returns fixed calls), `PayloadCallBuilder` (`payload`, takes calls from `config.calls`, `config.builderFail` simulates failure), `LogWriterPlugin` (`log`), and `TestWriterPlugin` (`test`, `config.pluginFail` simulates failure). They are only listed in the `cmd/worker` Profile; `cmd/bundle_worker` does not register them. They are for e2e / perf (e.g. `e2e/perf/s2_call_count_payload_test.go`) and local integration testing; do not use them in production tasks. ## State and lifecycle The set of registered Builders and Writer Plugins is determined by `app.Profile` at process startup and does not change dynamically per task after startup. A regular Builder runs once in the Builder phase. A streaming Builder's producer continues into the Calls phase and ends when production completes, fails, or dispatch stops. A Writer Plugin runs once after the Calls phase converges successfully. The Worker owns phase state, timeouts, and terminal state; the plugin registry does not store them. Builders and Writer Plugins receive the current task's `TaskCtx` and IO interfaces on every invocation. They do not own Worker slots, task terminal state, or cross-task IO scopes. ## Configuration Plugins have no configuration file of their own; all related items live in `WorkerFullConfig` in `internal/worker/app/config.go`: | Field | Default | Purpose | | ------------------------------------------------ | ------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `PhaseIOTimeoutMs` | `10000` | The `ioTimeout` passed to `NewDBScanBuilder` / `NewBlockDBWritePlugin` etc., bounding a single IO in the Builder / Writer phases. Environment variable `PHASE_IO_TIMEOUT_MS` | | `Admission.BuilderSlots` | `16` | `builderSem`, number of concurrent Builder phases (`internal/worker/core/config.go`) | | `Admission.PluginSlots` | `16` | `pluginSem`, number of concurrent Writer phases | | `Admission.ScanSlots` | `8` | `scanSem`, number of concurrent streaming producers (within each producer, bundlescan runs at most 3 sources concurrently) | | `StreamBuild.Enabled` | `false` (turned on by `TuneDefaults` in `cmd/bundle_worker`) | Whether `StreamingCallBuilder` takes the two-phase streaming path | | `StreamBuild.BatchMaxRows` / `BatchMaxArgsBytes` | `8192` / `32 << 20` | Per-batch limits; whichever is reached first cuts the batch; a single call's `ArgsJSON` exceeding the byte limit is a production error | | `StreamBuild.MaxOutstandingBatches` | `2` | Producer credit window, the source of backpressure | | `BlockDB.BatchWriteAddr` | empty | Required when the Profile includes `PluginBundleWrite`, or when the bundle profile includes `PluginTableUpserts`; otherwise `validateProfileRuntimeConfig` fails at startup. Environment variable `BLOCKDB_BATCH_WRITE_ADDR` | ## Extension points ### Adding a Call Builder Add a `CallBuilderName` to the `const` block in `internal/plugin/callbuilder/types/types.go`; its value is the `functionCallConfig.type` string in the task. Create `internal/plugin/callbuilder//` and implement `types.CallBuilder`. In `Build`, parse `taskCtx.FunctionCallConfig.Config` with `sonic.Config{UseNumber: true}`, handle inline code with `types.NormalizeInlineCode`, and do all read IO through the passed-in `io.Read`. Produce per-row args with `types.CompactCall`. To support streaming, also implement `types.StreamingCallBuilder` (do all static validation in `PrepareStream`, `emit` batch by batch in `Run`, use a fresh backing array for each batch, and return `types.ErrProductionStopped` as-is when you encounter it). Add a case to the `switch name` in `newBuilderRegistry` in `internal/worker/app/app.go` to construct it; then add the name to the Profile that needs it (the `Builders` list in `cmd/worker/main.go` or `cmd/bundle_worker/main.go`). Without the case, startup fails with `unknown call builder`. In-package unit tests (see `dbscan/dbscan_test.go` and `call_list/call_list_test.go`: use a fake `TaskIOReader` to assert request types, CallID, args, and error classification); add the name to the `TestNewBuilderRegistry_ProfileMembership` series in `internal/worker/app/profile_test.go`; for end-to-end, see `e2e/system/logical_type/dbscan_operator_test.go`. ```go theme={null} package mybuilder import ( "context" "fmt" "github.com/bytedance/sonic" commontypes "github.com/Chaintable/blockx/internal/common/types" "github.com/Chaintable/blockx/internal/plugin/callbuilder/types" ) var jsonAPI = sonic.Config{UseNumber: true}.Froze() type decl struct { Function string `json:"function"` Rows []any `json:"rows"` } // BuilderMyScan must first be added to the CallBuilderName const block in internal/plugin/callbuilder/types/types.go. type Builder struct{} func (b *Builder) Name() types.CallBuilderName { return types.BuilderMyScan } func (b *Builder) Build(ctx context.Context, taskCtx *commontypes.TaskCtx, io types.TaskIOReader) (types.CallList, error) { var d decl if err := jsonAPI.Unmarshal(taskCtx.FunctionCallConfig.Config, &d); err != nil { return nil, fmt.Errorf("myScan: unmarshal config: %w", err) } calls := make(types.CallList, 0, len(d.Rows)) for i, row := range d.Rows { c, err := types.CompactCall(d.Function, nil, []any{row}, fmt.Sprintf("%s-%d", taskCtx.ID, i)) if err != nil { return nil, err } calls = append(calls, c) } return calls, nil } ``` ### Adding a Writer Plugin Add a `PluginName` to the `const` block in `internal/plugin/event/types/types.go`; its value is `resultHandler.type`. Create `internal/plugin/event//` and implement `types.WriterPlugin`. In `Execute`: return `*types.PluginError{Kind: types.PluginErrExecution, ...}` when `taskCtx.ResultHandler` is nil or config parsing fails; iterate outputs with `types.VisitOutputRows` (or `ConsumeOutputRows`); do write IO through the passed-in `io.Write`; use `types.IsRetryableError(err)` on downstream errors to decide `Retryable`. Respect `taskCtx.DebugMode`: only `PreviewOutputs`, no external IO. On success, return the data to pass through (`nil` if none). Add a case to `newPluginRegistry` in `internal/worker/app/app.go`; add the name to the `Plugins` list in `cmd/*/main.go`. If it depends on a new endpoint, add a startup check in `validateProfileRuntimeConfig`. In-package unit tests (see the `blockdbwrite` tests and `tableupserts/table_upserts_test.go`: single row / multiple rows / missing columns / explicit nil / write failure retryability / debug mode); cover the new error classification in `TestIsRetryableError` in `event/types/errors_test.go`; `TestNewPluginRegistry_ProfileMembership` in `internal/worker/app/profile_test.go`; the orchestrator-side `TestOrchestrator_EventPhase_*` tests already cover the generic path and usually need no changes. ```go theme={null} package myplugin import ( "context" commontypes "github.com/Chaintable/blockx/internal/common/types" "github.com/Chaintable/blockx/internal/plugin/event/types" ) // PluginMyWrite must first be added to the PluginName const block in internal/plugin/event/types/types.go. type Plugin struct{} func (p *Plugin) Name() types.PluginName { return types.PluginMyWrite } func (p *Plugin) Execute(ctx context.Context, taskCtx *commontypes.TaskCtx, outputs []any, io types.TaskIO) (any, error) { if taskCtx.ResultHandler == nil { return nil, &types.PluginError{Kind: types.PluginErrExecution, PluginName: p.Name(), Message: "result handler config is nil"} } rowCount, err := types.VisitOutputRows(outputs, func(row map[string]any) error { // project row into the write request; do not keep the row map itself return nil }) if err != nil { return nil, &types.PluginError{Kind: types.PluginErrExecution, PluginName: p.Name(), Message: err.Error()} } // Write IO goes through the passed-in io.Write; preserve the retryable classification of downstream errors: // if _, err := io.Write(ctx, req); err != nil { // return nil, &types.PluginError{Kind: types.PluginErrExecution, PluginName: p.Name(), // Message: "write failed", Cause: err, Retryable: types.IsRetryableError(err)} // } return map[string]any{"written_rows": rowCount}, nil } ``` `Call.ArgsJSON`, `NoResultCache`, and `ArgsPooled` are all in-process fields: `Call` must not be JSON round-tripped as a carrier for args, and `ArgsPooled` may only be set when exactly one call references the buffer. Builders that share static args never set `NoResultCache`. ## Testing ```bash theme={null} go test ./internal/plugin/... -short # builder / plugin / operator unit tests and in-package integration go test ./internal/worker/app/ -run 'Registry|Profile' # Profile assembly go test ./internal/worker/adapters/ -run 'Orchestrator_EventPhase|OrchestratorStream|BuilderFails' go test ./internal/worker/devstub/ ``` * `internal/plugin/callbuilder/callbuilder_test.go`, `internal/plugin/event/event_test.go`: registry and "runner"-style integration cases; `internal/plugin/pipeline_integration_test.go`: starts a gRPC mock BlockDB and runs the full DBScan → BlockDBWrite chain. * `internal/plugin/callbuilder/types/operator_test.go`: three-valued logic, exact big-integer comparison, legacy operator conversion, dedup keys. * `internal/plugin/callbuilder/bundlescan/stream_test.go`: streaming batch boundaries, single-row overflow, zero emits on `PrepareStream` static errors. * e2e: `e2e/system/logical_type/dbscan_operator_test.go` (real Worker process + gRPC mock BlockDB), `e2e/perf/s1_builder_dbscan_test.go`, `s5_plugin_return_value_test.go`. See [Testing](/en/development/testing) for an overview. ## Related docs * `docs/specs/plugin-system.md` in the blockx repo: interfaces, data structures, assembly, and the detailed design of DBScan / BlockDBWrite / BlockBundle / TableUpserts (the primary reference). * `docs/specs/architecture.md` §4.2.3 in the blockx repo: where the Plugin module sits in the overall architecture. * `docs/specs/substrait_ast.md` in the blockx repo: the Operator (Plan) JSON format and its differences from Substrait. * `docs/specs/2026-07-14-dbscan-in-memory-operator.md` in the blockx repo: decision record for switching DBScan to `GetBlockRows` + the in-memory interpreter. * On this site: [Worker](/en/components/worker) (orchestrator and phase admission), [Call execution subsystem](/en/components/call-execution) (Calls phase and streaming dispatch), [IO access subsystem](/en/components/io-subsystem) (implementation and rate limiting of `TaskIOReader` / `TaskIO`), [Bundle clusters](/en/components/bundle) (bundlescan / bundlewrite expanded), [Task lifecycle](/en/architecture/task-lifecycle).