Class BuiltinOp

java.lang.Object
io.krabka.streams.columnar.BuiltinOp
All Implemented Interfaces:
ColumnarProcessor, StatefulColumnarProcessor
public final class BuiltinOp extends Object implements StatefulColumnarProcessor
Creates the built-in Arrow batch operators.

Five operators cover the common transformations: filter, select, withColumns, groupBy, and windowedGroupBy. Every factory takes the BufferAllocator that will own its output batches, and every operator forwards exactly one batch per input batch.

Add an operator to a topology with ColumnarTopology.addOperator(String, BuiltinOp, ColumnarNode); the operator is copied per logical partition there, so groupBy state never leaks across partitions. The grouping operators are stateful and cumulative: they retain groups across batches and participate in snapshot and restore.

Example

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);
var totals = topology.addOperator(
    "totals",
    BuiltinOp.groupBy(
        allocator,
        List.of("user"),
        new Aggregation("amount", "total", AggregateFunction.SUM)),
    large);
topology.addSink("sink", "user-totals", codec, totals);

  • Field Details

  • Method Details

    • filter

      public static BuiltinOp filter(org.apache.arrow.memory.BufferAllocator allocator, RowPredicate predicate)
      Creates an operator that keeps the rows a predicate accepts.

      The predicate runs once per row; passing rows are copied into a new batch with the same schema, in their original order. Metadata columns travel with the rows, so __offset still identifies the source record after filtering.

      Parameters:
      allocator - the allocator that owns the output batches
      predicate - the row predicate
      Returns:
      the filter operator

    • select

      public static BuiltinOp select(org.apache.arrow.memory.BufferAllocator allocator, String... columns)
      Creates an operator that projects the named payload columns.

      The output keeps the named columns in the order given, then appends whichever reserved metadata columns exist in the input — you do not list metadata columns to keep them. Duplicate names are ignored after the first; a name the input lacks makes the evaluation throw ColumnarException.

      Parameters:
      allocator - the allocator that owns the output batches
      columns - the payload columns to keep, in output order
      Returns:
      the projection operator

    • withColumns

      public static BuiltinOp withColumns(org.apache.arrow.memory.BufferAllocator allocator, DerivedColumn... columns)
      Creates an operator that adds or replaces derived columns.

      A derived column whose name matches an existing column replaces it in place, keeping its position; a new name is appended after the existing columns. Values are coerced to each column's declared Arrow type with overflow checking; Utf8 accepts any value via toString().

      Parameters:
      allocator - the allocator that owns the output batches
      columns - the columns to derive
      Returns:
      the with-columns operator
      Throws:
      ColumnarException - if a derived column uses a reserved metadata name

    • groupBy

      public static BuiltinOp groupBy(org.apache.arrow.memory.BufferAllocator allocator, Collection<String> keys, Aggregation... aggregations)
      Creates a cumulative group-by operator.

      Rows are grouped by the key columns' values and the groups are retained across every batch processed by the same built topology, so each output batch is the current cumulative result: key columns first, aggregate columns after, groups ordered by first appearance. Metadata columns are dropped from the output unless they are grouped by.

      An unknown key or input column, or an empty key list, makes the evaluation throw ColumnarException.

      Parameters:
      allocator - the allocator that owns the output batches
      keys - the payload columns to group by; at least one
      aggregations - the aggregate columns to compute
      Returns:
      the group-by operator

    • windowedGroupBy

      public static BuiltinOp windowedGroupBy(org.apache.arrow.memory.BufferAllocator allocator, Collection<String> keys, Duration windowSize, Aggregation... aggregations)
      Groups cumulatively into fixed event-time windows using the Kafka timestamp column.

      Each row is assigned to the fixed window containing its __timestamp; the output adds WINDOW_START_COLUMN and WINDOW_END_COLUMN as epoch milliseconds after the key columns. Closed windows are retained for one window size past stream time; use windowedGroupBy(BufferAllocator, Collection, java.time.Duration, java.time.Duration, Aggregation...) for a longer retention.

      Parameters:
      allocator - the allocator that owns the output batches
      keys - the payload columns to group by; at least one
      windowSize - the fixed window size; at least one millisecond
      aggregations - the aggregate columns to compute
      Returns:
      the windowed group-by operator
      Throws:
      IllegalArgumentException - if the window is shorter than one millisecond

    • windowedGroupBy

      public static BuiltinOp windowedGroupBy(org.apache.arrow.memory.BufferAllocator allocator, Collection<String> keys, Duration windowSize, Duration retention, Aggregation... aggregations)
      Groups into fixed event-time windows and retains closed windows for the supplied duration.

      Stream time is the largest __timestamp seen so far; windows whose end falls more than the retention behind stream time are pruned from state, and rows arriving for pruned windows are dropped. Use partition snapshots when window state must survive a rebalance or restart.

      Parameters:
      allocator - the allocator that owns the output batches
      keys - the payload columns to group by; at least one
      windowSize - the fixed window size; at least one millisecond
      retention - how long closed windows accept late rows; at least the window size
      aggregations - the aggregate columns to compute
      Returns:
      the windowed group-by operator
      Throws:
      IllegalArgumentException - if the window is shorter than one millisecond or the retention is shorter than the window

    • process

      public void process(ColumnarContext context, org.apache.arrow.vector.VectorSchemaRoot batch)
      Applies the operator and forwards exactly one output batch.
      Specified by:
      process in interface ColumnarProcessor
      Parameters:
      context - the collector the output batch is forwarded to
      batch - the input batch, owned by the framework

    • snapshot

      public byte[] snapshot()
      Serializes the operator's state.
      Specified by:
      snapshot in interface StatefulColumnarProcessor
      Returns:
      the grouping state for groupBy operators, or an empty array for the stateless operators

    • restore

      public void restore(byte[] snapshot)
      Replaces the operator's state with a previously taken snapshot.
      Specified by:
      restore in interface StatefulColumnarProcessor
      Parameters:
      snapshot - bytes previously returned by snapshot()
      Throws:
      ColumnarException - if the bytes cannot be restored or a non-empty snapshot is restored into a stateless operator