Join Pipelines

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() }