Class ColumnarTopology

java.lang.Object
io.krabka.streams.columnar.ColumnarTopology
public final class ColumnarTopology extends Object
A linear or branching graph whose edges carry Arrow batches.

Nodes are added in evaluation order: every add* method returns a ColumnarNode handle that downstream nodes name as their parent, so a parent always precedes its children. A topology needs at least one source and one sink; build() validates the graph and returns a reusable BuiltColumnarTopology that holds the per-partition processor state.

The topology itself is cheap and stateless; all processing state lives in the built form. Build once and keep the built topology for the lifetime of a consumer group member so stateful operators accumulate across polls.

Example

try (var allocator = new RootAllocator()) {
    var codec = new BlobCodec(allocator);
    var topology = new ColumnarTopology(allocator);

    var source = topology.addSource("source", List.of("transactions"), codec);
    var large = topology.addOperator(
        "large",
        BuiltinOp.filter(allocator, (batch, row) ->
            ((BigIntVector) batch.getVector("amount")).get(row) > 4),
        source);
    topology.addSink("sink", "large-transactions", codec, large);

    try (var built = topology.build()) {
        var output = built.runBatch("transactions", records);
    }
}

  • Constructor Details

    • ColumnarTopology

      public ColumnarTopology(org.apache.arrow.memory.BufferAllocator allocator)
      Creates an empty topology.
      Parameters:
      allocator - the allocator that owns every batch the topology creates; use one allocator per application and close it at shutdown

  • Method Details

    • addSource

      public ColumnarNode addSource(String name, Collection<String> topics, BatchCodec codec)
      Adds a source that decodes records from one or more topics.

      All topics of one source must share the codec and, per graph evaluation, their decoded batches flow separately into the source's children.

      Parameters:
      name - the unique node name
      topics - the topics the source consumes; at least one
      codec - the codec that decodes fetched records
      Returns:
      the handle downstream nodes use as their parent
      Throws:
      IllegalArgumentException - if topics is empty

    • addOperator

      public ColumnarNode addOperator(String name, BuiltinOp operator, ColumnarNode parent)
      Adds a built-in operator node.

      The operator is copied per logical partition, so the same BuiltinOp instance can be reused in several topologies and its state stays partition-local.

      Parameters:
      name - the unique node name; stateful operators are snapshotted under it
      operator - the built-in operator to apply
      parent - the upstream node whose batches the operator processes
      Returns:
      the handle downstream nodes use as their parent
      Throws:
      IllegalArgumentException - if parent is not a node of this topology

    • addOperator

      public ColumnarNode addOperator(String name, Supplier<? extends ColumnarProcessor> processor, ColumnarNode parent)
      Adds a custom operator node.

      The supplier is invoked once per logical partition, which is what makes instance fields of the processor partition-local state. Implement StatefulColumnarProcessor to participate in snapshot and restore.

      Parameters:
      name - the unique node name; stateful operators are snapshotted under it
      processor - the factory invoked once per logical partition
      parent - the upstream node whose batches the processor processes
      Returns:
      the handle downstream nodes use as their parent
      Throws:
      IllegalArgumentException - if parent is not a node of this topology

    • addMerge

      public ColumnarNode addMerge(String name, Collection<ColumnarNode> parents)
      Merges batches from two or more upstream branches with the same Arrow schema.

      Per evaluation, the parents' batches are concatenated into one batch in parent order. Parents whose schemas differ make the evaluation throw ColumnarException.

      Parameters:
      name - the unique node name
      parents - the branches to merge; at least two
      Returns:
      the handle downstream nodes use as their parent
      Throws:
      IllegalArgumentException - if fewer than two parents are given or one of them is not a node of this topology

    • addJoin

      public ColumnarNode addJoin(String name, ColumnarJoin join, ColumnarNode left, ColumnarNode right)
      Joins two co-partitioned branches by key within an event-time window.

      The join is stateful: rows are retained per logical partition until they age past the window, so late arrivals on either side still match. Join state participates in snapshot and restore under this node's name.

      Parameters:
      name - the unique node name; join state is snapshotted under it
      join - the key columns, window, and output prefixes
      left - the left branch
      right - the right branch
      Returns:
      the handle downstream nodes use as their parent
      Throws:
      IllegalArgumentException - if the parents are the same node or either is not a node of this topology

    • addSink

      public ColumnarNode addSink(String name, String topic, BatchCodec codec, ColumnarNode parent)
      Adds a sink that encodes its parent's batches to a topic.
      Parameters:
      name - the unique node name
      topic - the topic produced records are destined for
      codec - the codec that encodes batches into records
      parent - the upstream node whose batches the sink encodes
      Returns:
      the sink's handle
      Throws:
      IllegalArgumentException - if parent is not a node of this topology

    • addPassThroughSink

      public ColumnarNode addPassThroughSink(String name, String topic, ColumnarNode source)
      Copies source records byte-for-byte to a topic without decoding and re-encoding the sink.

      Use this for mirroring or teeing a source topic alongside processing: the source's records keep their exact key, value, timestamp, and headers. When the source feeds only pass-through sinks, its records are never decoded at all.

      Parameters:
      name - the unique node name
      topic - the topic the copies are destined for
      source - the source node whose records are copied
      Returns:
      the sink's handle
      Throws:
      IllegalArgumentException - if source is not a source node of this topology

    • sourceTopics

      public List<String> sourceTopics()
      Returns every topic consumed by the topology's sources.
      Returns:
      the source topics in node insertion order

    • validate

      public void validate()
      Checks the graph without building it.

      build() calls this automatically; call it directly to fail fast in tests or while assembling a topology dynamically.

      Throws:
      ColumnarException - if a node name repeats, a non-source node lacks a valid earlier parent, or the topology has no source or no sink

    • build

      public BuiltColumnarTopology build()
      Validates the graph and creates its executable form.

      Each build creates fresh processor state; nodes added to this topology after building are not part of the built instance.

      Returns:
      the built topology; close it to release processor state
      Throws:
      ColumnarException - if validation fails