Join Pipelines

Available since 15.1 Southside release

Joins two pipelines together.
Takes another source as the pipeline.

Joining pipelines only work for Analytics (formerly called PriceAnalyzer).

Code

def qapi = api.queryApi()

def customerDs = qapi.tables().dataSource("Customer")
def p1 = qapi.source(customerDs, [customerDs.CustomerID, customerDs.name.as("Customer Name")])
        
def tx = qapi.tables().dataSource("SalesTransactions")
def p2 = qapi.source(tx, [tx.InvoicePrice, tx.CustomerID])
        .aggregateBy({ prev -> [prev.CustomerID] }, { prev -> [prev.CustomerID, qapi.exprs().sum(prev.InvoicePrice).as("sumInvoice")] })

return p1
        .innerJoin(p2, { left, right -> [right.sumInvoice.as("SUM Invoice Price")] }, { left, right -> left.CustomerID.equal(right.CustomerID) })
        .stream { it.toList() }

Result

CustomerID	Customer Name	SUM Invoice Price

CustomerID	Customer Name	SUM Invoice Price
CD-00001	Appetito Mz	78929.793
CD-00002	Spagetti M	328578.885

Join Pipelines

Code

Result

See Also