OLAP分区表替换示例表后使用twindow函数遇异常求助

操作步骤与问题 #

1. 创建OLAP数据库并导入币安聚合交易数据
   执行以下脚本创建OLAP库并导入数据：

dataFilePath = "/data/ddb/server/binance_historical_data/spot/monthly/aggTrades/BTCUSDT/BTCUSDT-aggTrades-2017-08.csv"
schemaTB = extractTextSchema(dataFilePath)

update schemaTB set name = [`trade_id,`price,`qty,`first_id,`last_id,`timestamp,`buyer_maker,`best_price_match];

update schemaTB set type = [`LONG,`DOUBLE,`DOUBLE,`LONG,`LONG,`LONG,`BOOL,`BOOL]


create database "dfs://binanceAggTradesOLAP"
partitioned by VALUE(2017.08.01..2017.09.01)
engine="OLAP"

create table "dfs://binanceAggTradesOLAP"."BTCUSDT"(
  trade_id LONG
  price DOUBLE
  qty DOUBLE
  first_id LONG
  last_id LONG
  timestamp NANOTIMESTAMP
  buyer_maker BOOL
  best_price_match BOOL
)
partitioned by timestamp

def myMapFunc(mutable t){
  t.replaceColumn!(`timestamp, iif(long(exec timestamp from t) > 9999999999999, nanotimestamp(exec timestamp from t * 1000), nanotimestamp(exec timestamp from t * 1000000)))
  tableInsert(loadTable("dfs://binanceAggTradesOLAP","BTCUSDT"), t)
}

dataDirPath = "/data/ddb/server/binance_historical_data/spot/monthly/aggTrades/BTCUSDT" 
fileList = exec filename from files(dataDirPath, "%.csv") where isDir = false
fileList = fileList.sort()

for(filename in fileList){
        dataFilePath = dataDirPath + "/" + filename
        ds = textChunkDS(dataFilePath, 200, schema=schemaTB)
        mr(ds, myMapFunc,,,false)
}

2. 生成5分钟OHLC表
   执行脚本生成5分钟K线表：

db = database("dfs://binanceAggTradesOLAP")
btc = db.loadTable("BTCUSDT")

model = select top 1 timestamp as barStart, price as open, price as high, price as low, price as close, qty from btc
if(existsTable("dfs://binanceAggTradesOLAP", "OHLC5m"))
 db.dropTable("OHLC5m")

ohlc5m = db.createPartitionedTable(model, `OHLC5m, partitionColumns=`barStart)

def calcOHLC(inputTable){
        barMinutes = 5m
        tmp=select first(price) as open, max(price) as high, min(price) as low, last(price) as close, sum(qty) as qty from inputTable group by dailyAlignedBar(timestamp, 00:00:00.000000000, barMinutes) as barStart
        loadTable("dfs://binanceAggTradesOLAP", `OHLC5m).append!(tmp)
        return tmp.size()
}
ds = sqlDS(<select * from btc where timestamp between nanotimestamp("2017.08.01 00:00:00.000") and nanotimestamp("2025.06.01 00:00:00.000")>)
mr(ds, calcOHLC, +)

3. 执行update语句报错
   为OHLC表新增bar_type字段后执行更新：

ohlc = loadTable("dfs://binanceAggTradesOLAP", "OHLC5m")
addColumn(ohlc, "bar_type", INT)

update ohlc set bar_type = twindow(myIndicator, [high,low], barStart, -5m:0m) where barStart >= 2017.08.01T00:00:00.000000000 and barStart <= 2025.06.01T00:00:00.000000000

报错信息：Aggregate or order-sensitive functions are not allowed without context by clause when updating a partitioned table

4. 添加context by后结果异常
   修改脚本添加context by barStart后执行：

update ohlc set bar_type = twindow(myIndicator, [high,low], barStart, -5m:0m) where barStart >= 2017.08.01T00:00:00.000000000 and barStart <= 2025.06.01T00:00:00.000000000 context by barStart

脚本无报错，但所有bar_type值均为1；调整twindow时间范围为-10m:0m或-5d:0d，结果仍无变化。

自定义函数myIndicator代码 #

defg myIndicator(high, low){
    prevHigh = first(high)
    prevLow = first(low)
    curHigh = last(high)
    curLow = last(low)
    if(curHigh <= prevHigh && curLow >= prevLow){
        return 1
    }else if(curHigh > prevHigh && curLow < prevLow){
        return 3
    }else {
        return 2
    }
}

n = 60
ts = 2025.06.09T09:30:00.000 + (1..n) * 1000 * 60
high = rand(700..900, n)\100
low = rand(600..700, n)\100
f1 = take(double(NULL), n)
t = table(ts, high, low, f1)

update t set f1 = twindow(myIndicator, [high,low], ts, -5m:0m) where ts >= 2025.06.09T09:31:00.000 and ts <= 2025.06.09T09:40:00.000

需求 #

解决上述OLAP表更新twindow函数结果异常的问题。

内容的提问来源于stack exchange，提问作者Drac Pet