Documentation
¶
Index ¶
- func BuildRuntimeTasks(analysisID int64, params map[string]any, requestParam map[string]any, ...) (map[string]any, error)
- func FilterRequestParamByHandles(requestParam map[string]any, inputs map[string]any) map[string]any
- type ClassifyStage
- type CompileContext
- type DecorateStage
- type EdgeStage
- type EdgeStrategy
- type ExpandStage
- type NodeExpandStrategy
- type NodeRuntimeState
- type OutputStage
- type ResolveStage
- type ResolveStrategy
- type RuntimeCompiler
- type RuntimeEdgeBuilder
- type RuntimeNodeBuilder
- type SampleEdgeStrategy
- type SampleExpandStrategy
- type SampleResolveStrategy
- type SingletonEdgeStrategy
- type SingletonExpandStrategy
- type SingletonResolveStrategy
- type Stage
- type TopologyStage
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func BuildRuntimeTasks ¶
func BuildRuntimeTasks(analysisID int64, params map[string]any, requestParam map[string]any, dagDefinition map[string]any) (map[string]any, error)
BuildRuntimeTasks 展开 DAG 定义成 runtime 节点/边。
requestParam 是本次分析的原始 request_param(analysis.request_param 解析而来)。 编译器把它作为每个节点的 analysis_node.request_param 原样下发,语义上与 params 派生 resolved_inputs 一致:两者都是由编译器为节点生成的字段。
func FilterRequestParamByHandles ¶ added in v0.1.5
FilterRequestParamByHandles 从 analysis 的原始 request_param 中,只保留键与节点 输入句柄(inputs_patterns 的 name)相同的项。
这与 inputs_patterns 的填充口径一致:inputs_patterns 按脚本 form json(io_schema 的 inputs)里的 name 建键,节点的 request_param 也用同一组 name 过滤,所以节点 只携带脚本声明过的输入对应的请求值,而不是整份 analysis 请求。
导出版本供其他调度器(如 dataflow V3)复用,保证各调度器的节点 request_param 口径完全一致。
Types ¶
type ClassifyStage ¶
type ClassifyStage struct{}
func (*ClassifyStage) Name ¶
func (s *ClassifyStage) Name() string
func (*ClassifyStage) Run ¶
func (s *ClassifyStage) Run(ctx *CompileContext) error
type CompileContext ¶
type CompileContext struct {
AnalysisID int64
Params map[string]any
// RequestParam 是本次分析的原始 request_param(已解析成 map)。它由调用方
// 从 analysis.request_param 透传进来,用于为每个 runtime 节点生成
// analysis_node.request_param,语义与 Params 派生 resolved_inputs 一致。
RequestParam map[string]any
Dag map[string]any
Abort bool
Nodes []map[string]any
Edges []map[string]any
NodeMap map[string]map[string]any
Incoming map[string][]map[string]any
Outgoing map[string][]map[string]any
Order []string
NodeKind map[string]string
NodeLabel map[string][]string
NodeSamplesMap map[string]map[string]map[string]any
OutputCache map[cacheKey]any
NodeStates map[string][]*NodeRuntimeState
StateOrder []*NodeRuntimeState
AnalysisNodes []map[string]any
AnalysisEdges []map[string]any
}
func NewCompileContext ¶
type DecorateStage ¶
type DecorateStage struct{}
func (*DecorateStage) Name ¶
func (s *DecorateStage) Name() string
func (*DecorateStage) Run ¶
func (s *DecorateStage) Run(ctx *CompileContext) error
type EdgeStage ¶
type EdgeStage struct {
// contains filtered or unexported fields
}
func (*EdgeStage) Run ¶
func (s *EdgeStage) Run(ctx *CompileContext) error
type EdgeStrategy ¶
type EdgeStrategy interface {
Build(ctx *CompileContext, nid string, node map[string]any, states []*NodeRuntimeState, builder *RuntimeEdgeBuilder) []map[string]any
}
type ExpandStage ¶
type ExpandStage struct {
// contains filtered or unexported fields
}
func (*ExpandStage) Name ¶
func (s *ExpandStage) Name() string
func (*ExpandStage) Run ¶
func (s *ExpandStage) Run(ctx *CompileContext) error
type NodeExpandStrategy ¶
type NodeExpandStrategy interface {
Expand(ctx *CompileContext, nid string, node map[string]any) ([]*NodeRuntimeState, error)
}
type NodeRuntimeState ¶
type NodeRuntimeState struct {
OriginalNodeID string
Node map[string]any
Kind string
NodeIDBase string
NodeID string
NodeName string
ScriptID string
SampleLabel string
Sample map[string]any
Inputs map[string]any
Outputs map[string]any
ResolvedInputs map[string]any
// RequestParam 汇总节点自身的原始请求值(按输入句柄 name 建键)。目前只由散列
// (scatter each) 节点写入:item 里按句柄名命中的原始值放到这里,而不是直接写进
// resolved_inputs,落库后由 nodebuild.Materialize 的解析器据此生成结构化
// resolved_inputs(与 select-assay 输入同一口径)。
RequestParam map[string]any
ResolvedOutput map[string]any
InputErrors []string
MaxRetry int
}
type OutputStage ¶
type OutputStage struct {
// contains filtered or unexported fields
}
func (*OutputStage) Name ¶
func (s *OutputStage) Name() string
func (*OutputStage) Run ¶
func (s *OutputStage) Run(ctx *CompileContext) error
type ResolveStage ¶
type ResolveStage struct {
// contains filtered or unexported fields
}
func (*ResolveStage) Name ¶
func (s *ResolveStage) Name() string
func (*ResolveStage) Run ¶
func (s *ResolveStage) Run(ctx *CompileContext) error
type ResolveStrategy ¶
type ResolveStrategy interface {
Resolve(ctx *CompileContext, state *NodeRuntimeState) error
}
type RuntimeCompiler ¶
type RuntimeCompiler struct {
// contains filtered or unexported fields
}
func NewRuntimeCompiler ¶
func NewRuntimeCompiler() *RuntimeCompiler
type RuntimeEdgeBuilder ¶
type RuntimeEdgeBuilder struct{}
type RuntimeNodeBuilder ¶
type RuntimeNodeBuilder struct{}
func (*RuntimeNodeBuilder) Build ¶
func (b *RuntimeNodeBuilder) Build(ctx *CompileContext, state *NodeRuntimeState) map[string]any
type SampleEdgeStrategy ¶
type SampleEdgeStrategy struct{}
func (*SampleEdgeStrategy) Build ¶
func (s *SampleEdgeStrategy) Build(ctx *CompileContext, nid string, node map[string]any, states []*NodeRuntimeState, builder *RuntimeEdgeBuilder) []map[string]any
type SampleExpandStrategy ¶
type SampleExpandStrategy struct{}
func (*SampleExpandStrategy) Expand ¶
func (s *SampleExpandStrategy) Expand(ctx *CompileContext, nid string, node map[string]any) ([]*NodeRuntimeState, error)
type SampleResolveStrategy ¶
type SampleResolveStrategy struct{}
func (*SampleResolveStrategy) Resolve ¶
func (s *SampleResolveStrategy) Resolve(ctx *CompileContext, state *NodeRuntimeState) error
type SingletonEdgeStrategy ¶
type SingletonEdgeStrategy struct{}
func (*SingletonEdgeStrategy) Build ¶
func (s *SingletonEdgeStrategy) Build(ctx *CompileContext, nid string, node map[string]any, states []*NodeRuntimeState, builder *RuntimeEdgeBuilder) []map[string]any
type SingletonExpandStrategy ¶
type SingletonExpandStrategy struct{}
func (*SingletonExpandStrategy) Expand ¶
func (s *SingletonExpandStrategy) Expand(ctx *CompileContext, nid string, node map[string]any) ([]*NodeRuntimeState, error)
type SingletonResolveStrategy ¶
type SingletonResolveStrategy struct{}
func (*SingletonResolveStrategy) Resolve ¶
func (s *SingletonResolveStrategy) Resolve(ctx *CompileContext, state *NodeRuntimeState) error
type Stage ¶
type Stage interface {
Name() string
Run(ctx *CompileContext) error
}
type TopologyStage ¶
type TopologyStage struct{}
func (*TopologyStage) Name ¶
func (s *TopologyStage) Name() string
func (*TopologyStage) Run ¶
func (s *TopologyStage) Run(ctx *CompileContext) error
Click to show internal directories.
Click to hide internal directories.