multiTableRepartitionDS

Syntax

multiTableRepartitionDS(queries, [column], [partitionType], [partitionScheme], [local=true])

Details

Generate a tuple of data sources from multiple tables with a new partitioning design.

multiTableRepartitionDS generates aligned data sources for multiple SQL queries so that they can be processed together by mr.

If queries is a tuple and column, partitionType, and partitionScheme are omitted, the function applies sqlDS to each query. All queries must produce the same number of data sources. Data sources at the same partition index are grouped into an inner tuple.

If repartitioning arguments are specified, the function applies the same partitioning scheme to every query and groups the resulting data sources by partition index. When the result is passed to mr, the data sources in each inner tuple must come from distributed tables in the same database.

Parameters

queries is SQL metacode or a tuple of SQL metacode. When a tuple is specified, each query must produce the same number of data sources. When a single piece of metacode is specified, column is required. To generate data sources using the table's original partitioning scheme, use sqlDS instead.

column is a STRING scalar specifying the column used to repartition the data sources. It is required when the input is a single piece of SQL metacode and optional when the input is a metacode tuple.

partitionType is an optional parameter specifying the partitioning type. It can be VALUE or RANGE. HASH is also supported in the 3.00 series.

partitionScheme is an optional vector indicating the partitioning scheme. For details please refer to DistributedComputing. A scalar is also supported in the 3.00 series.

local is an optional BOOL scalar indicating whether to move the data sources to the local node for computing. The default value is true.

Returns

Returns a tuple containing data sources.

If queries is a metacode tuple, each element of the returned outer tuple is an inner tuple containing the data sources from the same partition index. The length of each inner tuple equals the length of queries.

Examples

The following examples use order and trade tables in the same database.

createCatalog("factor_pipeline")
go
use catalog factor_pipeline

create database market_data partitioned by VALUE(2026.01.02 2026.01.03), engine='OLAP'
go

create table market_data.orders (
    TradeDate DATE,
    SecurityID SYMBOL,
    TradeTime TIME,
    OrderNO LONG,
    OrderQty INT
)
partitioned by TradeDate
go

create table market_data.trades (
    TradeDate DATE,
    SecurityID SYMBOL,
    TradeTime TIME,
    TradePrice DOUBLE,
    OfferApplSeqNum LONG
)
partitioned by TradeDate
go

orderData = table(
    2026.01.02 2026.01.02 2026.01.02 2026.01.03 2026.01.03 2026.01.03 as TradeDate,
    `AAPL`AAPL`MSFT`AAPL`AAPL`MSFT as SecurityID,
    09:30:00.000 09:31:00.000 09:32:00.000 09:30:00.000 09:31:00.000 09:32:00.000 as TradeTime,
    1001 1002 1001 2001 2002 2001 as OrderNO,
    100 300 200 150 150 400 as OrderQty
)
tradeData = table(
    2026.01.02 2026.01.02 2026.01.02 2026.01.03 2026.01.03 2026.01.03 as TradeDate,
    `AAPL`AAPL`MSFT`AAPL`AAPL`MSFT as SecurityID,
    09:30:00.300 09:31:00.700 09:32:00.200 09:30:00.100 09:31:00.400 09:32:00.100 as TradeTime,
    0.0 0.0 0.0 0.0 0.0 12.0 as TradePrice,
    1001 1002 1001 2001 2002 2001 as OfferApplSeqNum
)
factor_pipeline.market_data.orders.append!(orderData)
factor_pipeline.market_data.trades.append!(tradeData)

Example 1: Generate aligned data sources based on the tables' original partitioning scheme, with column, partitionType, and partitionScheme omitted.

ds = multiTableRepartitionDS(queries=[
    <select * from market_data.orders>,
    <select * from market_data.trades>
])
ds.size()
// output: 2
ds[0].size()
// output: 2

Example 2: Partition the data sources by stock symbol.

ds = multiTableRepartitionDS(
    queries=[
        <select * from market_data.orders>,
        <select * from market_data.trades>
    ],
    column=`SecurityID,
    partitionType=VALUE,
    partitionScheme=symbol(`AAPL`MSFT)
)
ds.size()
// output: 2
ds[0].size()
// output: 2

Example 3: Partition the data sources by date range.

ds = multiTableRepartitionDS(
    queries=[
        <select * from market_data.orders>,
        <select * from market_data.trades>
    ],
    column=`TradeDate,
    partitionType=RANGE,
    partitionScheme=2026.01.02 2026.01.03 2026.01.04
)
ds.size()
// output: 2
ds[0].size()
// output: 2

Example 4: Pass aligned data sources generated from the original partitioning scheme to mr. The map function receives the orders and trades for the same trading day and calculates, for each stock, the ratio of order quantity canceled within 500 milliseconds to total order quantity. A trade record with TradePrice equal to 0 represents a cancellation.

ds = multiTableRepartitionDS(queries=[
    <select * from market_data.orders>,
    <select * from market_data.trades>
])

def calcCancelRatio(orderTB, tradeTB){
    startTime = 09:30:00.000
    endTime = 14:57:00.000
    cancelTB = select TradeDate, SecurityID, TradeTime as CancelTime, OfferApplSeqNum as OrderNO
        from tradeTB
        where TradeTime between startTime and endTime, TradePrice=0
    joinedTB = lj(orderTB, cancelTB, `TradeDate`SecurityID`OrderNO)
    return select TradeDate, SecurityID,
            1.0 * sum(iif(isNull(CancelTime), 0,
                iif((CancelTime-TradeTime>=0) and (CancelTime-TradeTime<500), OrderQty, 0))) /
                sum(OrderQty) as cancelRatio
        from joinedTB
        where TradeTime between startTime and endTime
        group by TradeDate, SecurityID
}

result = mr(ds=ds, mapFunc=calcCancelRatio, reduceFunc=unionAll)
result.sortBy!(`TradeDate`SecurityID)
result

The output is as follows:

TradeDate SecurityID cancelRatio
2026.01.02 AAPL 0.25
2026.01.02 MSFT 1
2026.01.03 AAPL 1
2026.01.03 MSFT 0

Related functions: repartitionDS, mr