Plugin system

​

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).
• 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

​

Core types and interfaces

// 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)
}

​

Data flow / execution flow

​

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 "${<table>}" 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 for details.

​

Operator (Plan)

​

Built-in Writer Plugins

​

devstub

​

State and lifecycle

​

Configuration

​

Extension points

​

Adding a Call Builder

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

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
}

​

Testing

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 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 (orchestrator and phase admission), Call execution subsystem (Calls phase and streaming dispatch), IO access subsystem (implementation and rate limiting of TaskIOReader / TaskIO), Bundle clusters (bundlescan / bundlewrite expanded), Task lifecycle.