EK9 in Depth

Stream Pipelines in Depth

Streams and Pipelines in Volume 1 gave you cat, filter, map and collect, and ended with a promise: sort returns in Using Operators; the rest of the vocabulary is for the second volume. This is the rest. If you know Java's Stream, Rust's iterator adaptors or LINQ, the verbs will be familiar; the Unix pipe is the nearer ancestor, since EK9 has tee, uniq, head and tail under their own names. Two things differ from what you know. Every stage takes a function value, by name or written where it is used (Dynamic Functions and Classes), never a lambda over an implicit scope. And the compiler holds the pipeline to rules a library cannot: it refuses a group with no sort ahead of it, and a stream that appends to the collection it is reading.

One data set, every verb

Eight orders, each a customer, a product and an amount, are enough to use every verb once. The whole program is one file; each block below is a slice of it and of its single run. Where a verb needs a function, the program has a small pure one: customerOf, productOf, amountOf, isLarge, two comparators (byCustomerThenAmount and byAmountDescending), and addAmounts.

Sort, branch, group, restructure

stdout.println("Top orders per customer:")
sorted <- List() of Order
cat orders
  | sort by byCustomerThenAmount
  | tee in sorted
  | group by customerOf
  | map with topOfGroup
  | flatten
  > stdout

Top orders per customer:
ada monitor 180.0
ada keyboard 45.0
bob monitor 220.0
bob keyboard 60.0
cy laptop 999.0
cy mouse 12.0

sort by takes a comparator and orders the stream. tee in sorted is the Unix tee: a copy of everything passing that point lands in the variable sorted, and the main flow carries on. group by customerOf collects adjacent items with equal keys into a List each, so the stream now carries lists of orders, not orders. That is why the sort by ahead of it is load-bearing, and the compiler agrees:

cat words | group by initialOf > stdout

EK9     : Error   : E10040: 'group' on line 16 position 18: 'group' requires a preceding 'sort' with no type-changing operations between them
             See: ek9 -h E10040

Coming from Java, group by is not Collectors.groupingBy. It yields a stream of lists, not a Map, and it works by adjacency; the price is the mandatory sort. What the adjacency buys is that grouping stays a stream stage like any other and composes with the rest.

map with topOfGroup then runs a function over each list. That function is itself a small pipeline, which is the usual way to keep one readable:

topOfGroup() as pure
  -> group as List of Order
  <- rtn as List of Order: cat group | sort by byAmountDescending | head TOP_PER_CUSTOMER | collect as List of Order

flatten turns a stream of lists back into a stream of their elements, and > stdout ends it. The order is the order of the groups, and within each group the largest amount first: the two biggest orders for each customer.

Taking and skipping

stdout.println("Three smallest:")
cat sorted | sort by byAmountDescending | tail 3 > stdout
stdout.println("All but the first in customer order:")
cat sorted | skip 1 | map with productOf > stdout

Three smallest:
ada mouse 15.5
cy mouse 12.0
bob cable 9.0
All but the first in customer order:
keyboard
monitor
cable
keyboard
monitor
mouse
laptop

head N, tail N and skip N take from the front, take from the back, and drop from the front. head is the answer to the loop that used break to stop early, and skip to the counter that used continue for the first few. tail 3 after a descending sort gives the three smallest, still in descending order. These two pipelines read sorted, the copy tee left behind.

Filtering out, de-duplicating, and the two sinks

products <- List() of String
cat orders | reject by isLarge | map with productOf | uniq | sort > products
stdout.println(`small-order products: ${products}`)
cat orders | filter by isLarge | map with productOf | uniq | sort >> products
stdout.println(`after adding large-order products: ${products}`)

small-order products: ["cable", "keyboard", "mouse"]
after adding large-order products: ["cable", "keyboard", "mouse", "laptop", "monitor"]

reject by is filter by inverted, so there is no need to write a not function. uniq drops repeats and sort with no by uses the type's own ordering. The two sinks are not the same. > replaces what the target holds, and >> adds to it. The first pipeline fills products from nothing; the second adds the large-order products to what is there. This works the same on a List, a Set, a Dict and a PriorityQueue, and on a file, where > clobbers and >> appends.

A replace is all or nothing. The pipeline fills a fresh collection and swaps it in only when the stream has finished, so a failure part-way leaves the target as it was:

#!ek9
defines module book.vol2.streams.sinks

  defines constant
    BAD_NUMBER <- 3

  defines function

    checked()
      -> number as Integer
      <- rtn as String: String()
      if number == BAD_NUMBER
        throw Exception("bad item")
      rtn: $number

  defines program

    ReplaceOrAppend()
      stdout <- Stdout()

      names <- ["p"]
      cat ["q", "r"] > names
      stdout.println(`after >: ${names}`)

      cat ["s"] >> names
      stdout.println(`after >>: ${names}`)

      kept <- ["old"]
      try
        for i in 1 ... 5 | map with checked > kept
      catch
        -> ex as Exception
        stdout.println(`'>' failed part-way, so kept is still: ${kept}`)

      scores <- [30, 10, 20]
      cat scores | sort > scores
      stdout.println(`'>' may replace the List the pipeline reads: ${scores}`)

after >: ["q", "r"]
after >>: ["q", "r", "s"]
'>' failed part-way, so kept is still: ["old"]
'>' may replace the List the pipeline reads: [10, 20, 30]

The last line is the consequence of the swap. Because > builds a new collection and replaces the old one at the end, cat scores | sort > scores is legal: the pipeline reads the old list and the result takes its place. >> has no such luxury, since it would be adding to the very list it is reading. In Java, scores.stream().forEach(scores::add) compiles and fails at run time with ConcurrentModificationException; EK9 refuses to compile the equivalent:

cat scores >> scores

EK9     : Error   : E08277: 'scores' on line 10 position 20: 'scores' is read by 'cat' and appended to by '>>': a variable a stream statement appends to ('>>' or 'tee') also plays another part in that same pipeline - read by its 'cat', or captured by or passed to a function the pipeline runs - so the pipeline uses what it is still writing; use the variable before or after the statement, read a copy instead, use '>' to replace it, or collect into a new variable
             See: ek9 -h E08277

The message lists the three ways out: use the variable before or after the statement, read a copy, or replace with >. The check is not only about cat. A variable that >> or tee writes may not also be captured by, or passed to, a function the pipeline runs.

Folding, splitting and calling

grandTotal <- cat orders | map with amountOf | join with addAmounts | collect as Float

manifest <- ["# Monday", "ada keyboard", "bob monitor", "# Tuesday", "cy laptop", "# Wednesday", "bob cable", "ada mouse"]
cat manifest | split with isDayHeader | map with daySummary > stdout

countLine <- (orders) is summaryLine as pure function
  rtn: `${length orders} orders in total`
totalLine <- (grandTotal) is summaryLine as pure function
  rtn: `grand total ${grandTotal}`
cat [countLine, totalLine] | call > stdout

# Monday (2)
# Tuesday (1)
# Wednesday (2)
8 orders in total
grand total 1540.5

join with addAmounts folds the stream pairwise into one value, here the total of every amount; it is the fold other languages call reduce. split with isDayHeader starts a new list at each item the predicate accepts and emits the lists, which daySummary then summarises: three headers, three lists. group and split take a stream down to lists and flatten brings it back up.

call is the odd one. It takes no function. The elements of the stream are the functions, and call runs each one in order and passes its result on. Here the elements are the two dynamic functions from the lines above, each capturing a different value. Give call or async a function and the compiler says so:

cat numbers | call with doubled > stdout

cat numbers | async by doubled > stdout

EK9     : Error   : E07880: 'call' on line 16 position 20: a function or function delegate is not required here
             See: ek9 -h E07880
EK9     : Error   : E07880: 'async' on line 18 position 20: a function or function delegate is not required here
             See: ek9 -h E07880

To apply a function to plain values, map with first turns the data into functions to call: cat numbers | map with toWorker | call.

JSON as a sink

Collections in Volume 1 said the second volume covers JSON in full; the parsing half is in A Web Service, Step by Step. The sink half belongs here, because JSON follows the same two rules as the collections. > replaces an array target's items; >> adds to them; collect as JSON makes a fresh array. The records are turned to JSON by $$, the serialising operator, which the record asks for with default operator $$:

#!ek9
defines module book.vol2.streams.jsonsink

  defines constant
    LARGE_ORDER <- 100.0

  defines record
    Order
      customer <- String()
      product <- String()
      amount <- Float()

      Order() as pure
        ->
          customer as String
          product as String
          amount as Float

        this.customer :=? String(customer)
        this.product :=? String(product)
        this.amount :=? Float(amount)

      default operator ?

      default operator $$

  defines function

    isLarge() as pure
      -> order as Order
      <- rtn as Boolean: order.amount >= LARGE_ORDER

    asJson() as pure
      -> order as Order
      <- rtn as JSON: $$order

  defines program

    OrdersAsJson()
      stdout <- Stdout()

      orders <- [Order("ada", "keyboard", 45.0), Order("bob", "monitor", 220.0), Order("cy", "laptop", 999.0)]

      report <- JSON().array()
      cat orders | filter by isLarge | map with asJson >> report
      stdout.println(`after >>: ${report}`)

      cat orders | reject by isLarge | map with asJson > report
      stdout.println(`after >: ${report}`)

      everything <- cat orders | map with asJson | collect as JSON
      stdout.println(everything.prettyPrint())

after >>: [{"customer":"bob","product":"monitor","amount":220.0},{"customer":"cy","product":"laptop","amount":999.0}]
after >: [{"customer":"ada","product":"keyboard","amount":45.0}]
[ {
  "customer" : "ada",
  "product" : "keyboard",
  "amount" : 45.0
}, {
  "customer" : "bob",
  "product" : "monitor",
  "amount" : 220.0
}, {
  "customer" : "cy",
  "product" : "laptop",
  "amount" : 999.0
} ]

The target's shape decides what pouring means. Into an unset or array target the stream collects; into an object target each streamed object is merged in; a non-object poured into an object, or anything poured into a single JSON value, is invalid, leaves the target unset, and later items do not revive it (ek9 -h JSON). That last case is the cost: a JSON sink does not raise an error when the shapes disagree, it goes unset, so the habit is to check the result with ? before using it.

async is a stage, not a thread

Put async in a pipeline whose elements are functions and each one is called concurrently. The syntax is all this chapter needs; Concurrency, Safely is where the safety rules for what those functions may share live.

#!ek9
defines module book.vol2.streams.async

  defines function

    priceLookup() as abstract
      <- rtn as String?

    lookupKeyboard() is priceLookup
      <- rtn <- "keyboard 45.0"

    lookupMonitor() is priceLookup
      <- rtn <- "monitor 220.0"

    lookupMouse() is priceLookup
      <- rtn <- "mouse 15.5"

    lookupLaptop() is priceLookup
      <- rtn <- "laptop 999.0"

  defines program

    AsyncLookups()
      stdout <- Stdout()
      lookups <- [lookupKeyboard, lookupMonitor, lookupMouse, lookupLaptop]

      stdout.println("in completion order:")
      cat lookups | async > stdout

      stdout.println("sorted after async:")
      cat lookups | async | sort > stdout

      stdout.println("async then head 2, then sorted:")
      cat lookups | async | head 2 | sort > stdout

in completion order:
mouse 15.5
laptop 999.0
monitor 220.0
keyboard 45.0
sorted after async:
keyboard 45.0
laptop 999.0
monitor 220.0
mouse 15.5
async then head 2, then sorted:
keyboard 45.0
monitor 220.0

Results come out in the order the calls complete, not the order they arrived, so the first block is not stable. Three more runs of the same program printed this as their first block:

mouse 15.5
keyboard 45.0
laptop 999.0
monitor 220.0

mouse 15.5
laptop 999.0
keyboard 45.0
monitor 220.0

keyboard 45.0
monitor 220.0
laptop 999.0
mouse 15.5

If you need an order, put it back with sort, as the second block does; it was the same on every run. The third block looks deterministic and is not a race at all: with only map, call or tee between async and head N, just N calls are started (ek9 -h async), so you get the first two in list order, not the first two to finish. A filter between them keeps a wider window.

What the compiler measures

Volume 1 said pipelines replace the loops that needed break and continue. The compiler can say what that buys, because it scores every function. The same job written both ways, titles by authors with at least three books after a cut-off year:

titlesByLoop()
  -> books as List of Book
  <- rtn <- List() of String

  for candidate in books
    if candidate.year > ERA_CUTOFF
      count <- Integer(0)
      for other in books
        if other.author == candidate.author
          if other.year > ERA_CUTOFF
            count++
      if count >= MINIMUM_BOOKS
        if not (rtn contains candidate.title)
          rtn += candidate.title

titlesByPipeline() as pure
  -> books as List of Book
  <-
    rtn as List of String: cat books
      | filter by isModern
      | sort by byAuthor
      | group by authorOf
      | filter by hasEnoughBooks
      | flatten
      | map with titleOf
      | uniq
      | sort
      | collect as List of String

Both are in a test project that asserts they agree. Run with ek9 -t6, which writes an HTML report, each function carries the figures the compiler computed. These are copied from that report:

titlesByLoop()         cyclomatic 13, cognitive 22, nesting depth 5
titlesByPipeline()     cyclomatic 14, cognitive 0, nesting depth 0
isModern()             cyclomatic 2, cognitive 0, nesting depth 0
byAuthor()             cyclomatic 2, cognitive 0, nesting depth 0
authorOf()             cyclomatic 1, cognitive 0, nesting depth 0
hasEnoughBooks()       cyclomatic 2, cognitive 0, nesting depth 0
titleOf()              cyclomatic 1, cognitive 0, nesting depth 0

Read the table precisely, because the claim it supports is narrower than "pipelines are simpler". Cognitive complexity and nesting fall to zero. Cyclomatic complexity does not fall: it rose from 13 to 14, since every stage is a path. The loop is harder to hold in your head; the pipeline has more things to count. The compiler limits each: cyclomatic 45 for a function (E11010), cognitive 35 (E11021) and nesting 6 (E11011) (ek9 -h complexity). Written out in full, with a second scan of the books to collect the titles, the loop form stops compiling:

EK9     : Error   : E11011: 'titlesByLoop' on line 28 position 4: Nesting depth of 7 exceeds maximum of 6: excessive nesting - extract function
             See: ek9 -h E11011
           forAI: qualityMetrics.nestingDepth

The five small functions beside the pipeline each score almost nothing, which is the shape the limits steer you towards: many tiny named functions, one pipeline to say how they combine.

The cost

A pipeline can be too clever. Nine stages on a line each of which is a function you have to go and find is harder to read than the loop it replaced, and the compiler's scores will not tell you: the pipeline above scores zero cognitive complexity and is still at the edge of what a reader takes in at once. A workable limit is the length at which you would have named an intermediate variable in the loop; past that, split the pipeline and name the halves. You also pay in small functions. Every verb wants a function, so a pipeline of seven stages is seven declarations, where a lambda-heavy Java stream puts the logic inline; the declarations are where the tests and the names live, but they are more typing. And the safety is partial: > is all or nothing, >> into what you read is refused, and group without a sort is refused, but async ordering is yours to restore and a JSON sink goes quietly unset when the shapes disagree. What the rules buy is that the failures you would find only at run time in another language, a list modified while you read it, groups that are not groups, are compile errors here.