Class OpenSearchSinkScaffold

java.lang.Object
ai.pipestream.module.pipelineprobe.moduletest.scaffold.OpenSearchSinkScaffold
All Implemented Interfaces:
StepScaffold
public class OpenSearchSinkScaffold extends Object implements StepScaffold
opensearch-sink scaffold, e2e-faithful: provisions exactly what a pipeline crawl provisions before the sink sees a doc —
  1. the full 2×2 vector-set registration (sentence-10-3 + sentences_internal × minilm + paraphrase-minilm) via EmbedderFixturesRegistrar, matching what the captured corpus docs actually carry;
  2. an IndexPlan with a PINNED id and the caller-chosen IndexingStrategy (NESTED / SEPARATE_INDICES / CHUNK_COMBINED — the same three the e2e exercises; default CHUNK_COMBINED, the e2e default), gated on READY like the crawl gates dispatch;
  3. the semantic centroid configs assigned to the index, so the sink can write the document/section/paragraph centroid SPRs the corpus carries ("KNN field not provisioned at bind time" otherwise).
The plan id + instance ride StreamMetadata.context_params — what OpenSearchSinkOptionsParser reads.

The index name is unique per session so every run writes into a clean index; cleanup() destroys the plan, all derived indices, and the redis streams (with the background sweep for late bulk writes).

  • Constructor Details

    • OpenSearchSinkScaffold

      public OpenSearchSinkScaffold(OpenSearchTestSetupService setup, EmbedderFixturesRegistrar registrar, ai.pipestream.opensearch.v1.IndexingStrategy strategy)
      Creates the opensearch-sink scaffold, choosing a unique per-session index name so each run writes into a clean index.
      Parameters:
      setup - the OpenSearch test setup service used to provision the plan and indices
      registrar - the embedder fixtures registrar used to register vector sets and semantic configs
      strategy - the indexing strategy to provision the index plan with

  • Method Details

    • module

      public String module()
      Description copied from interface: StepScaffold
      The module/step this scaffold targets (e.g. "chunker").
      Specified by:
      module in interface StepScaffold
      Returns:
      the module/step name this scaffold targets

    • prepare

      public void prepare()
      Description copied from interface: StepScaffold
      Provision external state (vector sets, fresh index, …). Called once per session.
      Specified by:
      prepare in interface StepScaffold

    • wrap

      public ai.pipestream.data.v1.PipeStream wrap(ai.pipestream.data.v1.PipeDoc doc)
      Description copied from interface: StepScaffold
      Turn a corpus doc into the step's expected input PipeStream (directives, context_params, …).
      Specified by:
      wrap in interface StepScaffold
      Parameters:
      doc - the corpus document to wrap
      Returns:
      the PipeStream shaped as the step expects

    • cleanup

      public void cleanup()
      Destroy this session's plan, index (all strategy-derived indices incl. centroid side-indices via the prefix sweep), vector-set bindings, and redis streams. The shared recipes survive for the next run.
      Specified by:
      cleanup in interface StepScaffold

    • info

      public Map<String,Object> info()
      Description copied from interface: StepScaffold
      What the scaffold provisioned, for display/logging.
      Specified by:
      info in interface StepScaffold
      Returns:
      a map describing what the scaffold provisioned; empty by default