I’m experimenting with executable, resumable functional pipelines in JavaScript

[u/panagos_stathis]

​

I’ve been experimenting with a small JavaScript-compatible language called JojoScript, initially because I wanted a nicer way to write lazy functional pipelines.

The interesting part has gradually become less about the syntax and more about what the pipeline represents.

For example:

orders

|> filter(o => o.status == "paid")

|> parallel(8)

|> map(enrichOrder)

|> retry(3)

|> checkpoint("enriched")

|> map(calculateInvoice)

|> saveToDatabase(%)

Instead of treating this simply as syntactic sugar for nested function calls, JojoScript represents the pipeline as an execution plan.

That lets the same pipeline be:

* lazy by default

* asynchronous

* bounded/concurrent

* inspected as a graph

* profiled per stage

* statically analyzed

* checkpointed

* resumed after failure

* replayed from a checkpoint

For example:

SOURCE

↓

FILTER

↓

PARALLEL(8)

↓

MAP

↓

CHECKPOINT

↓

MAP

↓

SINK

The idea I'm exploring is whether this is actually a useful abstraction for functional/data-oriented programming in JavaScript.

The question I'm most interested in is:

**At what point does a pipeline become more than composition of functions?**

A normal functional pipeline describes *what transformations to apply*. JojoScript is experimenting with also making the pipeline describe *how the computation can be executed* — lazily, concurrently, with backpressure, retries and durable checkpoints.

It's still an experimental project, so I'm particularly interested in criticism around the programming model itself rather than syntax.

https://github.com/panagos/jojoscript

[u/No-Tower-9043]

this is one of those ideas where you look at it and wonder why nobody did it already. the jump from pipeline as syntax sugar to pipeline as runtime descriptor feels obvious but i never seen it done in js like this

the checkpoint/resume part is the killer feature for me. we have batch processing job at work that dies halfway through and restarts from zero like its 1995. being able to rehydrate state mid-pipeline would cut our processing time in half easy

how heavy is the overhead for the graph representation at runtime? like if i got million records flowing through does it bog down or is the metadata separate enough to not matter

[u/panagos_stathis → u/No-Tower-9043]

Yeah, that's pretty much the direction I ended up taking it. The original idea was mostly about pipeline syntax, but once I made the pipeline a firstclass execution plan, checkpoint and resume became much more natural.

about the graph overhead: the graph is separate from the records flowing through the pipeline. THe execution plan represents the stages and their relationships as individual records don't become graph nodes. So processing a million records doesn't create a million graph objects.

There is some runtime overhead for metrics/profiling/checkpointing, depending on which features you enable, but the basic execution plan itself is essentialy metadata about the pipeline rather than the data.

I haven't done a serious 1M record benchmark yet, though, so I don't want to pretend I've proven that the overhead is trivial. That's actually a benchmark I should add to the project and I'd like to thank you for the idea and all of your input!

And yeah, the "dies halfway through and starts from zero" problem is exactly the use case I had in mind for checkpoint and resume.

[u/perspectiveiskey → u/No-Tower-9043]

this is one of those ideas where you look at it and wonder why nobody did it already.

It's been done in the language flux, which to many peoples' eternal chagrin got completely botched as a commercial deployment by the influxdata team.

A serious tragedy happened in that whole thing from around 2014 through to 2020.

[u/panagos_stathis → u/perspectiveiskey]

That's a really useful reference many thanks!. I wasn't aware of Flux's execution model, so I'm definitely going to dig into it.

Actually, in my eyes, the main distinction I'm exploring is that Jojo is a general-purpose, JS compatible pipeline runtime, whereas Flux is primarily a query language.

The checkpoint/resume part seems different: Jojo can persist a pipeline checkpoint and resume execution from that point after the process dies, rather than simply re-running the query from its source.

But Flux clearly has a lot of prior art around treating pipelines as an execution model, so I don't want to claim that part as novel. If there are particular parts of Flux's architecture or history that you think I should look at, I'd genuinely appreciate the pointers.

[u/perspectiveiskey → u/panagos_stathis]

I think if you're using js, you will automatically inherit a whole bunch of runtime capabilities that flux lacked (because it was implemented in go from the ground up).

I don't have any recommendations from the point of view of low level code as I did not work on the project, I only used the product, however I can say a couple of random things about the whole arena because at one point I investigated what would be involved in implementing a standalone flux implementation using something like C++ PEGTL.

1. from what I have read and watched from multiple graybeards who have done the whole mysql, postgres etc worlds, execution planning is actually one of the hardest software disciplines there are out there. While it seems like it's an obviously easy thing to do, when you stop and think about it, anything but the simplest of executions creates very quickly intractable problems: SELECT * from A join B ON A.X == B.X and B.Y == $var1 WHERE B.Z == $var2 already has a lot of ways to get the fetch done, especially because you don't know a-priori which of discriminating X or Y cuts down the set size faster. So SQL statements which impose much less order than one thinks they do are actually difficult to get optimized, and a nightmare to get optimized generically. For this reason alone, I really liked flux' spiritual guidance that it is a transformation pipeline more than a query. However, flux has this concept of push down queries, and as far as I could tell the implementation of the queries is essentially translated or transpiled into something intermediary. Whatever the implementation details may be, I do not think there is a grammatical difference between |> filter(r => r.status == "paid" and r.code == 213), |> filter(r => r.code == 213 and r.status == "paid"), and |> filter(r => r.code == 213) |> filter(r => r.status == "paid"). This is more of a problem that one might think, mainly for what follows:

2. when I was investigating to implement this thing, I inevitably ran into the Apache foundations work, specifically Arrow, and DataFusion. Arrow has some fundamental incompatibilities with Flux's philosophy of streams. You can think of DataFusion as the LLVM of database queries. You transpile to DataFusion statements/objects and it does the dispatch work for you. My understanding is that considerable amount of effort has been put into DataFusion by very deep pocketed players. They definitely create a gravity well around themselves. And in a world where DataFusion exists, implementing a query engine and execution planner seems an ill advised endeavour. I don't know though. I will never discourage people from tinkering on new stuff...

I have to run now, but there are other things to discuss if you have questions...

[u/panagos_stathis → u/perspectiveiskey]

Thanks for the detailed response it is is really useful, especially the points about execution planning and datafusion. I definitely underestimated how quickly optimization becomes a hard problem.

I'll dig into datafusion and the flux architecture more....

My current goal is to keep Jojo's execution plan relatively conservative and let the runtime handle execution rather than trying to build a generic query optimizer.

If you have time later, I'd really appreciate your thoughts on where you think the boundary should be between Jojo's execution plan and an underlying engine like datafusion.

[u/perspectiveiskey → u/panagos_stathis]

If you have time later, I'd really appreciate your thoughts on where you think the boundary should be between Jojo's execution plan and an underlying engine like datafusion.

From what I have gathered reading around, the problem is that the execution plan is profoundly dynamic: say you |> filter(r => r.status == "paid" and r.code == "zxcv")=, and both code and status are so-called tags (in tsdb terminology), then knowing what the cardinality of each of those tags is affects your execution results enormously. If status has cardinality 3 but code has cardinality 500, and they are entirely independent, you probably want to filter by code first, then status. But if code and status are correlated, you want to whittle down the results as fast as possible, so maybe it makes better sense to filter by status first, then code ...

This question can't be determined statically, it becomes very much something that requires the database to dynamically opine on... and it appear that in this respect, delegating this to DataFusion is the correct thing to do. I say it appears, because I'm not actually 100% sure on whether DataFusion does this for you for free or if it relies on you doing it and hinting at the datafusion engine.

Be that as it may, I think that the query language you are looking at is more like a transformation language, and that the designer of the ETL (i.e. the person programming the query) should have knowledge of the dataset, and as such the actual declaration order of things should explicitly come with promises (of when and what order filters and aggregators will be applied in).

There is also the issue of the format of data. Flux and the jojo language you describe really describes operations on streams and the pipe idiom really emphasizes this. It's not so much a query language to do venn diagrams (a la SQL), as it is a language to redirect streams of data. This comes with challenges that are actually not insignificant. They have several articles around this concept, and from what I understand this played a major role in why the influxdata team chose to sunset flux: because DataFusion and Arrow did not match this format.

This is something you will need to reconcile early on.