Mapping & Validation

Every hop in a ProtoMolt pipeline can carry its own declarative logic: field mappings that reshape a document before and after a module runs, and CEL (Common Expression Language) expressions that filter documents, gate individual mapping rules, select source values, and pick routing edges. All of it is compiled and evaluated inside the engine JVM — there is no extra network hop per expression — and the same machinery backs the live gRPC doors used to preview mappings and validate graphs before they deploy.

Per-hop declarative logic

A GraphNode carries ordered pre_mappings and post_mappings (applied before and after the module runs), plus filter_conditions and a symmetric post-filter pair (post_filter_conditions, post_filter_action). Each filter condition is a CEL expression evaluated in order against a {document, stream} context; the first false short-circuits. The filter_action decides the outcome: DROP is an intentional terminal stop recorded as a typed DROP(FILTERED) accounting row, while SKIP bypasses the module and routes the document onward unchanged. FILTER_ACTION_UNSPECIFIED behaves as DROP so older graphs keep their behavior. Every per-condition decision is appended to the stream's processing logs (Filter 'cond' → result), echoed at INFO with a [DRY RUN] prefix on dry runs, so both telemetry and forensics carry the per-condition decisions. Edges carry their own CEL: GraphEdge.condition (field 5) gates which outgoing edge a document takes, with priority tie-breaking.

engine routing tests — edge conditions and hop guards

stream.hop_count < 10
document.size < 1000000

All compilation happens once. CelEvaluatorService compiles and caches CEL ASTs in a ConcurrentHashMap keyed by expression string, and the cache is pre-warmed asynchronously on every graph load: an onGraphLoaded observer collects every edge condition, node filter, and mapping filter/selector/expression from the graph. Evaluation has two failure modes: evaluate() returns false on any error (safe for edge selection), while evaluateOrFail() throws CelEvaluationException so a caller can quarantine a document instead of mistaking a configuration error for "filter didn't match". A blank expression means "always apply" everywhere — filters, validators, and mappings all treat blank as unconditional rather than an error.

The five mapping types

Mapping rules are ProcessingMapping messages with a MappingType: DIRECT (field-to-field copy), TRANSFORM (a named function), AGGREGATE (combine sources into one target), SPLIT (one source into several targets), and CEL (a computed value).

pipestream-protos/common/proto/ai/pipestream/data/v1/pipeline_core_types.proto:748-794

enum MappingType {
  MAPPING_TYPE_UNSPECIFIED = 0;
  MAPPING_TYPE_DIRECT = 1;      // Direct field-to-field copy.
  MAPPING_TYPE_TRANSFORM = 2;   // Apply transformation function to field.
  MAPPING_TYPE_AGGREGATE = 3;   // Combine multiple source fields into one target.
  MAPPING_TYPE_SPLIT = 4;       // Split one source field into multiple targets.
  MAPPING_TYPE_CEL = 5;         // Calculate value using CEL expression.
}

message CelConfig {
  // Available context variables:
  //   - document: The full PipeDoc object
  //   - stream: The PipeStream object (metadata, history, etc.)
  // Example expressions:
  //   - "document.search_metadata.relevance_score * 100"
  //   - "document.search_metadata.title + ' - ' + document.search_metadata.author"
  //   - "stream.metadata.context_params['priority'] == 'high'"
  string expression = 1;
}

Rules apply in order through a single chokepoint, MappingEngine, which is used by both the per-hop runtime path and the gRPC preview — so preview semantics match runtime by construction. The context is progressive: each rule's CEL sees the effects of the rules before it (pinned by the test laterRulesSeeEarlierRulesEffects, where rule 2's filter keys on a custom field that rule 1 wrote). A bare mapping path is sugar: normalizePath maps "headline" to search_metadata.custom_fields.headline; explicit dot paths address anything.

Per-rule CEL filters and selectors

Every mapping rule carries two optional CEL fields: a filter that gates whether the rule runs at all, and a selector that replaces the source field paths with a computed value.

pipestream-protos/common/proto/ai/pipestream/data/v1/pipeline_core_types.proto:730-744

// Per-rule CEL gate; blank = always apply. Evaluated against
// {document, stream} immediately before this rule runs — pre-mappings see
// the (hydrated) input document, post-mappings see the module's result
// document, and each rule sees the effects of the rules before it.
string filter = 9;
// Per-rule CEL source selector; when set, it overrides source_field_paths:
// the expression result (evaluated against {document, stream}) becomes the
// source value for this rule. Unlike dot-paths, CEL can traverse
// google.protobuf.Any payloads (e.g.
// document.parsed_metadata['tika'].data...).
string selector = 10;

The Any-traversal matters in practice: the runtime CEL environment binds document and stream as DYN and registers the TikaResponse and DoclingResponse message types for Any unpacking, so a selector can reach straight into a parser result packed inside google.protobuf.Any — verified by the test selectorReachesThroughAnyPayloads.

engine/.../mapping/MappingFilterSelectorTest.java:65-141

// node filter / per-rule filter
document.search_metadata.language == 'de'
document.search_metadata.custom_fields['stage'] == 'hello world'

// selector composing a value
document.search_metadata.title + ' (' + document.search_metadata.language + ')'

// selector reaching through a google.protobuf.Any payload
document.parsed_metadata['tika'].data.doc_id

The proto_rules escape hatch

For cases where declarative field paths are not enough, a TRANSFORM rule named proto_rules executes imperative string rules via ProtoFieldMapperImpl.mapInPlace. The rule engine supports ASSIGN, APPEND, and CLEAR, dot-paths, google.protobuf.Struct/Any handling, and type conversion.

core-services/pipestream-engine engine/mapping/MappingEngine.java:350-355 (javadoc)

search_metadata.custom_fields.output = search_metadata.custom_fields.headline
search_metadata.custom_fields.flag = true
-search_metadata.custom_fields.headline

The last form — a leading minus — clears the field. The mapping utilities this runs on (ProtoFieldMapperImpl, TypeConverter, AnyHandler) live under engine/mapping/util/ and are the engine's own copy, not a shared library.

Live doors: candidate fallback and ValidateCel

The engine exposes the mapping machinery over gRPC. MappingService.ApplyMapping previews a mapping against a document without running a pipeline, with candidate-fallback semantics: a MappingRule is a list of candidate mappings tried in order, first success wins.

pipestream-protos/admin/proto/ai/pipestream/mapping/v1/mapping_service.proto:32-37

// A MappingRule represents a single logical transformation that can have
// multiple fallback strategies. The service will attempt each
// 'candidate_mapping' in order until one succeeds.
message MappingRule {
  repeated ai.pipestream.data.v1.ProcessingMapping candidate_mappings = 1;
}

The companion door, ValidateCel, compiles an expression against CelEnvironments.runtime() — the same environment definition the engine evaluates with at runtime — so accept/reject can never drift from runtime behavior. A second pass against a strictly-typed advisory environment produces warnings only, because dynamic evaluation legitimately accepts Any-traversals that static typing rejects. The test fieldTypoIsOkButWarned pins the contract: document.serch_metadata.language == 'en' (note the typo) returns ok=true with a WARNING naming the unknown field.

pipestream-protos/admin/proto/ai/pipestream/mapping/v1/mapping_service.proto:20-29

// Validates a CEL expression against the SAME environment the engine
// evaluates with at runtime — accept/reject is identical by construction.
//
// Two passes are reported through one issues list:
// ERROR issues come from the literal runtime environment and gate `ok`...
// WARNING issues come from an advisory typed pass (e.g. a field name the
// declared types do not have)...
rpc ValidateCel(ValidateCelRequest) returns (ValidateCelResponse);

Edge and filter contexts must produce a boolean; mapping selectors may return anything.

The 12-rule deploy gate

Graph deploys are gated. PipelineGraphGrpcService.createAndActivateGraph always revalidates the exact submitted payload in PRODUCTION mode and refuses with FAILED_PRECONDITION before anything is inserted; draft autosaves validate in DRAFT mode and persist a validation snapshot with the draft. Validation runs through GraphRuleRegistry, which CDI-discovers every GraphRule bean, sorts rules by id so issue lists are deterministic, and dispatches by ValidationMode (DRAFT is structural only; DESIGN adds catalog and tenancy; PRODUCTION adds plan producibility; UNSPECIFIED runs everything). The twelve rules, by id:

• cel-expression — every CEL expression in the graph must compile.
• edge-endpoints — edges must point at nodes that exist.
• no-cycles — the graph must be acyclic (Johnson's algorithm for cycle enumeration).
• orphan-nodes — no unreachable nodes.
• intake-presence — an intake node must be present.
• intake-datasource — intake nodes must name a datasource.
• intake-duplicate-datasource — no two intake nodes sharing a datasource.
• sink-requires-plans — sink nodes must have index plans attached.
• directive-form — directive payloads must be well-formed.
• module-config-schema — module configs validated against JSON-Schema 2020-12 (via networknt; also unwraps OpenAPI documents).
• datasource-resolvable — referenced datasources must resolve.
• plan-producibility — delegates to opensearch-manager's ValidatePlanProducibility.

Separate from the gate, GraphValidationService also runs hot-path node-existence checks (cache-only, never I/O) and computes a save-time memory-frontier advisory: the widest antichain of the MEMORY-edge subgraph via Dilworth's theorem (max antichain = n − max bipartite matching over the reachability graph). It only advises — the cap is a runtime pacing bound, deliberately never a save veto, because an RTBF document forces every edge to MEMORY anyway, so a veto would guarantee nothing.

Per-datasource intake mappings

The same mapping model is applied at the front door. Each datasource can carry its own mapping rules, stored as protobuf-JSON in a Postgres intake_mappings table keyed by datasource_id — deliberately not graph-versioned, so intake rules are editable without cutting a new graph version. They are managed over gRPC (GetIntakeMapping / SaveIntakeMapping on PipelineGraphService), and the save path is strict: CEL must compile against the runtime environment, a selector is only allowed on DIRECT/TRANSFORM rules, and saving an empty rule list deletes the config.

On the read side the rules are served on the first-hop hot path through a 30-second Caffeine cache (quarkus.cache.caffeine."intake-mappings".expire-after-write=30S, 10k entries). At route-through, IntakeRouting.stampIntakeMappings merges the datasource's rules into the pointer's stream-level pre_mappings; the raw :intake account copy is never mutated, so re-crawls pick up the then-current config. MappingApplier runs stream-level pre-mappings before node pre-mappings so node rules observe defaulted documents, and runs post-mappings node-first with the stream-level rules appended.

CEL beyond the engine core

CEL is the platform's general-purpose decision language, and it shows up well outside the per-hop path:

• Vector-set source selection. VectorSet.source_cel (field 14) and per-directive VectorDirective.cel_selector choose which documents feed a vector set and where the source text comes from; engine pre-mapping stamps VectorSetDirectives onto documents, pinned as "the production path" by VectorSetDirectivesMappingTest.
• Quality scoring. The quality module ships five scoring profiles (quality-default-profile.json, plus profiles for news articles, scientific papers, technical docs, and web content), each with five CEL-scored dimensions — 25 shipping expressions — alongside native QualityChecker beans (DuplicationChecker, AuthorityChecker, TopicRelevanceChecker) for dimensions that don't fit CEL. Custom exp(x) and clamp(x,lo,hi) functions are registered, and results are coerced and clamped to [0,1].
• The designer's recipe palette. The frontend ships nine curated CEL recipes categorized by use (STEP_FILTER, POST_STEP_FILTER, MAPPING_FILTER, MAPPING_SELECTOR, MAPPING_EXPRESSION, EDGE_CONDITION), served to the pipeline designer via the ListCelRecipes RPC.

modules/module-quality quality-default-profile.json — shipped scoring expressions

(hasTitle ? 0.25 : 0.0) + (hasAuthor ? 0.15 : 0.0) + (bodyLength > 200 ? 0.30 : (bodyLength > 0 ? 0.15 : 0.0)) + (keywordCount > 0 ? 0.15 : 0.0) + (sectionCount >= 2 ? 0.15 : 0.0)
hasDate ? exp(-0.693 * ageDays / 365.0) : 0.5
clamp(1.0 - replacementCharRatio * 20.0, 0.0, 1.0) * (bodyLength > 0 ? 1.0 : 0.0)

frontend/pipestream-frontend src/lib/celRecipes.ts — designer recipes

document.search_metadata.source_mime_type == 'application/pdf'
stream.metadata.context_params['skip_processing'] != 'true'
document.search_metadata.title != '' ? document.search_metadata.title : document.search_metadata.source_uri

Scale, and honest limits

This machinery is not a thin veneer. The engine's mapping + validation + routing main source is about 7,870 lines of Java across roughly 25 classes, with 43 engine test files covering mapping, validation, filters, and CEL. Across all platform repos (core-services, modules, connectors, grpc-services) there are 31 main-source mapper/mapping classes with 29 mapper test files, and 35 main-source validator/rule classes with 34 validator test files — including per-service validators such as RegisterRequestValidator, AccountValidator, ConnectorValidationService, CrawlDefinitionValidator, VectorSetSourceCelValidator, and the connector-side ConfluenceMapper/ConfluenceValidator and MicrosoftMapper/MicrosoftValidator pairs.