未验证 提交 590dacab 编写于 作者: H Haojun Liao 提交者: GitHub

Merge pull request #9548 from taosdata/fix/TS-802-D1

[TS-802]<fix>(query): first last API support order by desc in sub query
......@@ -1479,6 +1479,18 @@ void handleDownstreamOperator(SSqlObj** pSqlObjList, int32_t numOfUpstream, SQue
break;
}
}
// set input data order to param[1]
if(pex->base.functionId == TSDB_FUNC_FIRST || pex->base.functionId == TSDB_FUNC_FIRST_DST ||
pex->base.functionId == TSDB_FUNC_LAST || pex->base.functionId == TSDB_FUNC_LAST_DST) {
// set input order
SQueryInfo* pInputQI = pSqlObjList[0]->cmd.pQueryInfo;
if(pInputQI) {
pex->base.numOfParams = 3;
pex->base.param[2].nType = TSDB_DATA_TYPE_INT;
pex->base.param[2].i64 = pInputQI->order.order;
}
}
}
tscDebug("0x%"PRIx64" create QInfo 0x%"PRIx64" to execute the main query while all nest queries are ready", pSql->self, pSql->self);
......
......@@ -1620,33 +1620,65 @@ static bool first_last_function_setup(SQLFunctionCtx *pCtx, SResultRowCellInfo*
// todo opt for null block
static void first_function(SQLFunctionCtx *pCtx) {
if (pCtx->order == TSDB_ORDER_DESC) {
return;
}
SResultRowCellInfo* pResInfo = GET_RES_INFO(pCtx);
int32_t notNullElems = 0;
// handle the null value
for (int32_t i = 0; i < pCtx->size; ++i) {
char *data = GET_INPUT_DATA(pCtx, i);
if (pCtx->hasNull && isNull(data, pCtx->inputType)) {
continue;
}
memcpy(pCtx->pOutput, data, pCtx->inputBytes);
if (pCtx->ptsList != NULL) {
TSKEY k = GET_TS_DATA(pCtx, i);
DO_UPDATE_TAG_COLUMNS(pCtx, k);
int32_t step = 1;
int32_t i = 0;
bool inputAsc = true;
// input data come from sub query, input data order equal to sub query order
if(pCtx->numOfParams == 3) {
if(pCtx->param[2].nType == TSDB_DATA_TYPE_INT && pCtx->param[2].i64 == TSDB_ORDER_DESC) {
step = -1;
i = pCtx->size - 1;
inputAsc = false;
}
} else if (pCtx->order == TSDB_ORDER_DESC) {
return ;
}
if(pCtx->order == TSDB_ORDER_ASC && inputAsc) {
for (int32_t m = 0; m < pCtx->size; ++m, i+=step) {
char *data = GET_INPUT_DATA(pCtx, i);
if (pCtx->hasNull && isNull(data, pCtx->inputType)) {
continue;
}
memcpy(pCtx->pOutput, data, pCtx->inputBytes);
if (pCtx->ptsList != NULL) {
TSKEY k = GET_TS_DATA(pCtx, i);
DO_UPDATE_TAG_COLUMNS(pCtx, k);
}
SResultRowCellInfo *pInfo = GET_RES_INFO(pCtx);
pInfo->hasResult = DATA_SET_FLAG;
pInfo->complete = true;
notNullElems++;
break;
}
} else { // desc order
for (int32_t m = 0; m < pCtx->size; ++m, i+=step) {
char *data = GET_INPUT_DATA(pCtx, i);
if (pCtx->hasNull && isNull(data, pCtx->inputType) && (!pCtx->requireNull)) {
continue;
}
SResultRowCellInfo *pInfo = GET_RES_INFO(pCtx);
pInfo->hasResult = DATA_SET_FLAG;
pInfo->complete = true;
notNullElems++;
break;
TSKEY ts = pCtx->ptsList ? GET_TS_DATA(pCtx, i) : 0;
char* buf = GET_ROWCELL_INTERBUF(pResInfo);
if (pResInfo->hasResult != DATA_SET_FLAG || (*(TSKEY*)buf) > ts) {
pResInfo->hasResult = DATA_SET_FLAG;
memcpy(pCtx->pOutput, data, pCtx->inputBytes);
*(TSKEY*)buf = ts;
DO_UPDATE_TAG_COLUMNS(pCtx, ts);
}
notNullElems++;
break;
}
}
SET_VAL(pCtx, notNullElems, 1);
}
......@@ -1730,16 +1762,23 @@ static void first_dist_func_merge(SQLFunctionCtx *pCtx) {
* least one data in this block that is not null.(TODO opt for this case)
*/
static void last_function(SQLFunctionCtx *pCtx) {
if (pCtx->order != pCtx->param[0].i64) {
SResultRowCellInfo* pResInfo = GET_RES_INFO(pCtx);
int32_t notNullElems = 0;
int32_t step = -1;
int32_t i = pCtx->size - 1;
// input data come from sub query, input data order equal to sub query order
if(pCtx->numOfParams == 3) {
if(pCtx->param[2].nType == TSDB_DATA_TYPE_INT && pCtx->param[2].i64 == TSDB_ORDER_DESC) {
step = 1;
i = 0;
}
} else if (pCtx->order != pCtx->param[0].i64) {
return;
}
SResultRowCellInfo* pResInfo = GET_RES_INFO(pCtx);
int32_t notNullElems = 0;
if (pCtx->order == TSDB_ORDER_DESC) {
for (int32_t i = pCtx->size - 1; i >= 0; --i) {
for (int32_t m = pCtx->size - 1; m >= 0; --m, i += step) {
char *data = GET_INPUT_DATA(pCtx, i);
if (pCtx->hasNull && isNull(data, pCtx->inputType) && (!pCtx->requireNull)) {
continue;
......@@ -1756,7 +1795,7 @@ static void last_function(SQLFunctionCtx *pCtx) {
break;
}
} else { // ascending order
for (int32_t i = pCtx->size - 1; i >= 0; --i) {
for (int32_t m = pCtx->size - 1; m >= 0; --m, i += step) {
char *data = GET_INPUT_DATA(pCtx, i);
if (pCtx->hasNull && isNull(data, pCtx->inputType) && (!pCtx->requireNull)) {
continue;
......
......@@ -5612,6 +5612,18 @@ static int32_t getTableScanOrder(STableScanInfo* pTableScanInfo) {
return pTableScanInfo->order;
}
// check all SQLFunctionCtx is completed
static bool allCtxCompleted(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx) {
// only one false, return false
for(int32_t i = 0; i < pOperator->numOfOutput; i++) {
if(pCtx[i].resultInfo == NULL)
return false;
if(!pCtx[i].resultInfo->complete)
return false;
}
return true;
}
// this is a blocking operator
static SSDataBlock* doAggregate(void* param, bool* newgroup) {
SOperatorInfo* pOperator = (SOperatorInfo*) param;
......@@ -5650,6 +5662,9 @@ static SSDataBlock* doAggregate(void* param, bool* newgroup) {
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order);
doAggregateImpl(pOperator, pQueryAttr->window.skey, pInfo->pCtx, pBlock);
// if all pCtx is completed, then query should be over
if(allCtxCompleted(pOperator, pInfo->pCtx))
break;
}
doSetOperatorCompleted(pOperator);
......
......@@ -229,7 +229,8 @@ python3 test.py -f tools/taosdemoAllTest/taosdemoTestInsertAllType.py
python3 test.py -f tools/taosdemoAllTest/taosdemoTestInsertShell.py
#query
python3 test.py -f query/distinctOneColTb.py
python3 ./test.py -f query/queryBase.py
python3 ./test.py -f query/distinctOneColTb.py
python3 ./test.py -f query/filter.py
python3 ./test.py -f query/filterCombo.py
python3 ./test.py -f query/queryNormal.py
......
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
#
# query base function test case
#
import sys
from numpy.lib.function_base import insert
import taos
from util.log import *
from util.cases import *
from util.sql import *
import numpy as np
# constant define
WAITS = 5 # wait seconds
class TDTestCase:
#
# --------------- main frame -------------------
#
def caseDescription(self):
'''
Query moudle base api or keyword test case:
case1: api first() last()
case2: none
'''
return
# init
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor())
tdSql.prepare()
self.create_tables();
self.ts = 1500000000000
# run case
def run(self):
# insert data
self.insert_data("t1", self.ts, 1*10000, 30000, 0);
self.insert_data("t2", self.ts, 2*10000, 30000, 100000);
self.insert_data("t3", self.ts, 3*10000, 30000, 200000);
# test base case
self.case_first()
tdLog.debug(" QUERYBASE first() api ............ [OK]")
# test advance case
self.case_last()
tdLog.debug(" QUERYBASE last() api ............ [OK]")
# stop
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
#
# --------------- case -------------------
#
# create table
def create_tables(self):
# super table
tdSql.execute("create table st(ts timestamp, i1 int) tags(area int)");
# child table
tdSql.execute("create table t1 using st tags(1)");
tdSql.execute("create table t2 using st tags(2)");
tdSql.execute("create table t3 using st tags(3)");
return
# insert data1
def insert_data(self, tbname, ts_start, count, batch_num, base):
pre_insert = "insert into %s values"%tbname
sql = pre_insert
tdLog.debug("doing insert table %s rows=%d ..."%(tbname, count))
for i in range(count):
sql += " (%d,%d)"%(ts_start + i*1000, base + i)
if i >0 and i%batch_num == 0:
tdSql.execute(sql)
sql = pre_insert
# end sql
if sql != pre_insert:
tdSql.execute(sql)
tdLog.debug("INSERT TABLE DATA ............ [OK]")
return
# first case base
def case_first(self):
#
# last base function
#
# base t1 table
sql = "select first(*) from t1 where ts>='2017-07-14 12:40:00' order by ts asc;"
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 7200)
sql = "select first(*) from t1 where ts>='2017-07-14 12:40:00' order by ts desc;" # desc
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 7200)
# super table st
sql = "select first(*) from st where ts>='2017-07-14 11:40:00' and ts<='2017-07-14 12:40:00' and tbname in('t1') order by ts;"
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 3600)
sql = "select first(*) from st where ts>='2017-07-14 11:40:00' and ts<='2017-07-14 12:40:00' and tbname in('t1') order by ts desc;" # desc
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 3600)
# sub query
sql = "select first(*) from ( select sum(i1) from st where ts>='2017-07-14 11:40:00' and ts<'2017-07-14 12:40:00' interval(10m) order by ts asc );"
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 187019100)
sql = "select first(*) from ( select sum(i1) from st where ts>='2017-07-14 11:40:00' and ts<'2017-07-14 12:40:00' interval(10m) order by ts desc );" # desc
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 187019100)
return
# last case
def case_last(self):
#
# last base test
#
# base t1 table
sql = "select last(*) from t1 where ts<='2017-07-14 12:40:00' order by ts asc;"
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 7200)
sql = "select last(*) from t1 where ts<='2017-07-14 12:40:00' order by ts desc;" # desc
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 7200)
# super table st
sql = "select last(*) from st where ts>='2017-07-14 11:40:00' and ts<='2017-07-14 12:40:00' and tbname in('t1') order by ts;"
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 7200)
sql = "select last(*) from st where ts>='2017-07-14 11:40:00' and ts<='2017-07-14 12:40:00' and tbname in('t1') order by ts desc;" # desc
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 7200)
# sub query
sql = "select last(*) from ( select sum(i1) from st where ts>='2017-07-14 11:40:00' and ts<'2017-07-14 12:40:00' interval(10m) order by ts asc );"
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 192419100)
sql = "select last(*) from ( select sum(i1) from st where ts>='2017-07-14 11:40:00' and ts<'2017-07-14 12:40:00' interval(10m) order by ts desc );" # desc
tdSql.waitedQuery(sql, 1, WAITS)
tdSql.checkData(0, 1, 192419100)
#
# add case with filename
#
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册