Merge pull request #4539 from taosdata/feature/query

Feature/query

packaging/cfg/taos.cfg

@@ -29,8 +29,12 @@
 # number of threads per CPU core
 # numOfThreadsPerCore       1.0
 
-# the proportion of total threads responsible for query
-# ratioOfQueryThreads       0.5
+# the proportion of total CPU cores available for query processing
+# 2.0: the query threads will be set to double of the CPU cores.
+# 1.0: all CPU cores are available for query processing [default].
+# 0.5: only half of the CPU cores are available for query.
+# 0.0: only one core available.
+# tsRatioOfQueryCores       1.0
 
 # number of management nodes in the system
 # numOfMnodes               3
@@ -265,5 +269,5 @@
 # enable/disable stream (continuous query)
 # stream                1
 
-# only 50% CPU resources will be used in query processing
-# halfCoresForQuery     0
+# in retrieve blocking model, only in 50% query threads will be used in query processing in dnode
+# retrieveBlockingModel    0

src/common/inc/tglobal.h

@@ -46,7 +46,7 @@ extern int32_t  tsShellActivityTimer;
 extern uint32_t tsMaxTmrCtrl;
 extern float    tsNumOfThreadsPerCore;
 extern int32_t  tsNumOfCommitThreads;
-extern float    tsRatioOfQueryThreads;  // todo remove it
+extern float    tsRatioOfQueryCores;
 extern int8_t   tsDaylight;
 extern char     tsTimezone[];
 extern char     tsLocale[];
@@ -57,7 +57,7 @@ extern char     tsTempDir[];
 
 //query buffer management
 extern int32_t  tsQueryBufferSize;      // maximum allowed usage buffer for each data node during query processing
-extern int32_t  tsHalfCoresForQuery;         // only 50% will be used in query processing
+extern int32_t  tsRetrieveBlockingModel;         // only 50% will be used in query processing
 
 // client
 extern int32_t tsTableMetaKeepTimer;

src/common/src/tglobal.c

@@ -52,7 +52,7 @@ int32_t tsMaxConnections = 5000;
 int32_t tsShellActivityTimer  = 3;  // second
 float   tsNumOfThreadsPerCore = 1.0f;
 int32_t tsNumOfCommitThreads = 1;
-float   tsRatioOfQueryThreads = 0.5f;
+float   tsRatioOfQueryCores = 1.0f;
 int8_t  tsDaylight       = 0;
 char    tsTimezone[TSDB_TIMEZONE_LEN] = {0};
 char    tsLocale[TSDB_LOCALE_LEN] = {0};
@@ -107,8 +107,8 @@ int64_t tsMaxRetentWindow = 24 * 3600L;  // maximum time window tolerance
 // positive value (in MB)
 int32_t tsQueryBufferSize = -1;
 
-// only 50% cpu will be used in query processing in dnode
-int32_t tsHalfCoresForQuery = 0;
+// in retrieve blocking model, the retrieve threads will wait for the completion of the query processing.
+int32_t tsRetrieveBlockingModel = 0;
 
 // db parameters
 int32_t tsCacheBlockSize = TSDB_DEFAULT_CACHE_BLOCK_SIZE;
@@ -444,12 +444,12 @@ static void doInitGlobalConfig(void) {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
-  cfg.option = "ratioOfQueryThreads";
-  cfg.ptr = &tsRatioOfQueryThreads;
+  cfg.option = "ratioOfQueryCores";
+  cfg.ptr = &tsRatioOfQueryCores;
   cfg.valType = TAOS_CFG_VTYPE_FLOAT;
   cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG;
-  cfg.minValue = 0.1f;
-  cfg.maxValue = 0.9f;
+  cfg.minValue = 0.0f;
+  cfg.maxValue = 2.0f;
   cfg.ptrLength = 0;
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
@@ -887,8 +887,8 @@ static void doInitGlobalConfig(void) {
   cfg.unitType = TAOS_CFG_UTYPE_BYTE;
   taosInitConfigOption(cfg);
 
-  cfg.option = "halfCoresForQuery";
-  cfg.ptr = &tsHalfCoresForQuery;
+  cfg.option = "retrieveBlockingModel";
+  cfg.ptr = &tsRetrieveBlockingModel;
   cfg.valType = TAOS_CFG_VTYPE_INT32;
   cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
   cfg.minValue = 0;

src/dnode/src/dnodeShell.c

@@ -70,8 +70,7 @@ int32_t dnodeInitShell() {
 
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_NETWORK_TEST]   = dnodeSendStartupStep;
 
-  int32_t numOfThreads = tsNumOfCores * tsNumOfThreadsPerCore;
-  numOfThreads = (int32_t) ((1.0 - tsRatioOfQueryThreads) * numOfThreads / 2.0);
+  int32_t numOfThreads = (tsNumOfCores * tsNumOfThreadsPerCore) / 2.0;
   if (numOfThreads < 1) {
     numOfThreads = 1;
   }

src/dnode/src/dnodeVRead.c

@@ -26,16 +26,20 @@ static SWorkerPool tsVQueryWP;
 static SWorkerPool tsVFetchWP;
 
 int32_t dnodeInitVRead() {
+  const int32_t maxFetchThreads = 4;
+
+  // calculate the available query thread
+  float threadsForQuery = MAX(tsNumOfCores * tsRatioOfQueryCores, 1);
+
   tsVQueryWP.name = "vquery";
   tsVQueryWP.workerFp = dnodeProcessReadQueue;
-  tsVQueryWP.min = tsNumOfCores;
-  tsVQueryWP.max = tsNumOfCores/* * tsNumOfThreadsPerCore*/;
-//  if (tsVQueryWP.max <= tsVQueryWP.min * 2) tsVQueryWP.max = 2 * tsVQueryWP.min;
+  tsVQueryWP.min = (int32_t) threadsForQuery;
+  tsVQueryWP.max = tsVQueryWP.min;
   if (tWorkerInit(&tsVQueryWP) != 0) return -1;
 
   tsVFetchWP.name = "vfetch";
   tsVFetchWP.workerFp = dnodeProcessReadQueue;
-  tsVFetchWP.min = MIN(4, tsNumOfCores);
+  tsVFetchWP.min = MIN(maxFetchThreads, tsNumOfCores);
   tsVFetchWP.max = tsVFetchWP.min;
   if (tWorkerInit(&tsVFetchWP) != 0) return -1;
 

src/query/src/qExecutor.c

@@ -7635,7 +7635,7 @@ int32_t qRetrieveQueryResultInfo(qinfo_t qinfo, bool* buildRes, void* pRspContex
 
   int32_t code = TSDB_CODE_SUCCESS;
 
-  if (tsHalfCoresForQuery) {
+  if (tsRetrieveBlockingModel) {
     pQInfo->rspContext = pRspContext;
     tsem_wait(&pQInfo->ready);
     *buildRes = true;

src/query/src/qFilterfunc.c

@@ -21,6 +21,12 @@
 #include "tcompare.h"
 #include "tsqlfunction.h"
 
+#define FLT_EQUAL(_x, _y)        (fabs((_x) - (_y)) <= (4 * FLT_EPSILON))
+#define FLT_GREATER(_x, _y)      (!FLT_EQUAL((_x), (_y)) && ((_x) > (_y)))
+#define FLT_LESS(_x, _y)         (!FLT_EQUAL((_x), (_y)) && ((_x) < (_y)))
+#define FLT_GREATEREQUAL(_x, _y) (FLT_EQUAL((_x), (_y)) || ((_x) > (_y)))
+#define FLT_LESSEQUAL(_x, _y)    (FLT_EQUAL((_x), (_y)) || ((_x) < (_y)))
+
 bool less_i8(SColumnFilterElem *pFilter, char *minval, char *maxval) {
   return (*(int8_t *)minval < pFilter->filterInfo.upperBndi);
 }
@@ -38,35 +44,35 @@ bool less_i64(SColumnFilterElem *pFilter, char *minval, char *maxval) {
 }
 
 bool less_ds(SColumnFilterElem *pFilter, char *minval, char *maxval) {
-  return (*(float *)minval < pFilter->filterInfo.upperBndd);
+  return FLT_LESS(*(float*)minval, pFilter->filterInfo.upperBndd);
 }
 
 bool less_dd(SColumnFilterElem *pFilter, char *minval, char *maxval) {
-  return (*(double *)minval < pFilter->filterInfo.upperBndd);
+  return *(double *)minval < pFilter->filterInfo.upperBndd;
 }
 
 //////////////////////////////////////////////////////////////////
-bool large_i8(SColumnFilterElem *pFilter, char *minval, char *maxval) {
+bool larger_i8(SColumnFilterElem *pFilter, char *minval, char *maxval) {
   return (*(int8_t *)maxval > pFilter->filterInfo.lowerBndi);
 }
 
-bool large_i16(SColumnFilterElem *pFilter, char *minval, char *maxval) {
+bool larger_i16(SColumnFilterElem *pFilter, char *minval, char *maxval) {
   return (*(int16_t *)maxval > pFilter->filterInfo.lowerBndi);
 }
 
-bool large_i32(SColumnFilterElem *pFilter, char *minval, char *maxval) {
+bool larger_i32(SColumnFilterElem *pFilter, char *minval, char *maxval) {
   return (*(int32_t *)maxval > pFilter->filterInfo.lowerBndi);
 }
 
-bool large_i64(SColumnFilterElem *pFilter, char *minval, char *maxval) {
+bool larger_i64(SColumnFilterElem *pFilter, char *minval, char *maxval) {
   return (*(int64_t *)maxval > pFilter->filterInfo.lowerBndi);
 }
 
-bool large_ds(SColumnFilterElem *pFilter, char *minval, char *maxval) {
-  return (*(float *)maxval > pFilter->filterInfo.lowerBndd);
+bool larger_ds(SColumnFilterElem *pFilter, char *minval, char *maxval) {
+  return FLT_GREATER(*(float*)maxval, pFilter->filterInfo.lowerBndd);
 }
 
-bool large_dd(SColumnFilterElem *pFilter, char *minval, char *maxval) {
+bool larger_dd(SColumnFilterElem *pFilter, char *minval, char *maxval) {
   return (*(double *)maxval > pFilter->filterInfo.lowerBndd);
 }
 /////////////////////////////////////////////////////////////////////
@@ -88,10 +94,14 @@ bool lessEqual_i64(SColumnFilterElem *pFilter, char *minval, char *maxval) {
 }
 
 bool lessEqual_ds(SColumnFilterElem *pFilter, char *minval, char *maxval) {
-  return (*(float *)minval <= pFilter->filterInfo.upperBndd);
+  return FLT_LESSEQUAL(*(float*)minval, pFilter->filterInfo.upperBndd);
 }
 
 bool lessEqual_dd(SColumnFilterElem *pFilter, char *minval, char *maxval) {
+  if ((fabs(*(double*)minval) - pFilter->filterInfo.upperBndd) <= 2 * DBL_EPSILON) {
+    return true;
+  }
+
   return (*(double *)minval <= pFilter->filterInfo.upperBndd);
 }
 
@@ -113,11 +123,15 @@ bool largeEqual_i64(SColumnFilterElem *pFilter, char *minval, char *maxval) {
 }
 
 bool largeEqual_ds(SColumnFilterElem *pFilter, char *minval, char *maxval) {
-  return (*(float *)maxval >= pFilter->filterInfo.lowerBndd);
+  return FLT_GREATEREQUAL(*(float*)maxval, pFilter->filterInfo.lowerBndd);
 }
 
 bool largeEqual_dd(SColumnFilterElem *pFilter, char *minval, char *maxval) {
-  return (*(double *)maxval >= pFilter->filterInfo.lowerBndd);
+  if (fabs(*(double *)maxval - pFilter->filterInfo.lowerBndd) <= 2 * DBL_EPSILON) {
+    return true;
+  }
+
+  return (*(double *)maxval - pFilter->filterInfo.lowerBndd > (2 * DBL_EPSILON));
 }
 
 ////////////////////////////////////////////////////////////////////////
@@ -162,10 +176,12 @@ bool equal_i64(SColumnFilterElem *pFilter, char *minval, char *maxval) {
   }
 }
 
+// user specified input filter value and the original saved float value may needs to
+// increase the tolerance to obtain the correct result.
 bool equal_ds(SColumnFilterElem *pFilter, char *minval, char *maxval) {
   if (*(float *)minval == *(float *)maxval) {
-    return (fabs(*(float *)minval - pFilter->filterInfo.lowerBndd) <= FLT_EPSILON);
-  } else { /* range filter */
+    return FLT_EQUAL(*(float*)minval, pFilter->filterInfo.lowerBndd);
+  } else { // range filter
     assert(*(float *)minval < *(float *)maxval);
     return *(float *)minval <= pFilter->filterInfo.lowerBndd && *(float *)maxval >= pFilter->filterInfo.lowerBndd;
   }
@@ -173,10 +189,9 @@ bool equal_ds(SColumnFilterElem *pFilter, char *minval, char *maxval) {
 
 bool equal_dd(SColumnFilterElem *pFilter, char *minval, char *maxval) {
   if (*(double *)minval == *(double *)maxval) {
-    return (*(double *)minval == pFilter->filterInfo.lowerBndd);
-  } else { /* range filter */
+    return (fabs(*(double *)minval - pFilter->filterInfo.lowerBndd) <= 2 * DBL_EPSILON);
+  } else { // range filter
     assert(*(double *)minval < *(double *)maxval);
-
     return *(double *)minval <= pFilter->filterInfo.lowerBndi && *(double *)maxval >= pFilter->filterInfo.lowerBndi;
   }
 }
@@ -255,7 +270,7 @@ bool nequal_i64(SColumnFilterElem *pFilter, char *minval, char *maxval) {
 
 bool nequal_ds(SColumnFilterElem *pFilter, char *minval, char *maxval) {
   if (*(float *)minval == *(float *)maxval) {
-    return (*(float *)minval != pFilter->filterInfo.lowerBndd);
+    return !FLT_EQUAL(*(float *)minval,  pFilter->filterInfo.lowerBndd);
   }
 
   return true;
@@ -364,7 +379,8 @@ bool rangeFilter_i64_ei(SColumnFilterElem *pFilter, char *minval, char *maxval)
 
 ////////////////////////////////////////////////////////////////////////
 bool rangeFilter_ds_ii(SColumnFilterElem *pFilter, char *minval, char *maxval) {
-  return (*(float *)minval <= pFilter->filterInfo.upperBndd && *(float *)maxval >= pFilter->filterInfo.lowerBndd);
+  return FLT_LESSEQUAL(*(float *)minval, pFilter->filterInfo.upperBndd) &&
+         FLT_GREATEREQUAL(*(float *)maxval, pFilter->filterInfo.lowerBndd);
 }
 
 bool rangeFilter_ds_ee(SColumnFilterElem *pFilter, char *minval, char *maxval) {
@@ -376,7 +392,8 @@ bool rangeFilter_ds_ie(SColumnFilterElem *pFilter, char *minval, char *maxval) {
 }
 
 bool rangeFilter_ds_ei(SColumnFilterElem *pFilter, char *minval, char *maxval) {
-  return (*(float *)minval <= pFilter->filterInfo.upperBndd && *(float *)maxval > pFilter->filterInfo.lowerBndd);
+  return FLT_GREATER(*(float *)maxval, pFilter->filterInfo.lowerBndd) &&
+         FLT_LESSEQUAL(*(float *)minval, pFilter->filterInfo.upperBndd);
 }
 
 //////////////////////////////////////////////////////////////////////////
@@ -400,7 +417,7 @@ bool rangeFilter_dd_ei(SColumnFilterElem *pFilter, char *minval, char *maxval) {
 bool (*filterFunc_i8[])(SColumnFilterElem *pFilter, char *minval, char *maxval) = {
   NULL,
   less_i8,
-  large_i8,
+  larger_i8,
   equal_i8,
   lessEqual_i8,
   largeEqual_i8,
@@ -413,7 +430,7 @@ bool (*filterFunc_i8[])(SColumnFilterElem *pFilter, char *minval, char *maxval)
 bool (*filterFunc_i16[])(SColumnFilterElem *pFilter, char *minval, char *maxval) = {
   NULL,
   less_i16,
-  large_i16,
+  larger_i16,
   equal_i16,
   lessEqual_i16,
   largeEqual_i16,
@@ -426,7 +443,7 @@ bool (*filterFunc_i16[])(SColumnFilterElem *pFilter, char *minval, char *maxval)
 bool (*filterFunc_i32[])(SColumnFilterElem *pFilter, char *minval, char *maxval) = {
   NULL,
   less_i32,
-  large_i32,
+  larger_i32,
   equal_i32,
   lessEqual_i32,
   largeEqual_i32,
@@ -439,7 +456,7 @@ bool (*filterFunc_i32[])(SColumnFilterElem *pFilter, char *minval, char *maxval)
 bool (*filterFunc_i64[])(SColumnFilterElem *pFilter, char *minval, char *maxval) = {
   NULL,
   less_i64,
-  large_i64,
+  larger_i64,
   equal_i64,
   lessEqual_i64,
   largeEqual_i64,
@@ -452,7 +469,7 @@ bool (*filterFunc_i64[])(SColumnFilterElem *pFilter, char *minval, char *maxval)
 bool (*filterFunc_ds[])(SColumnFilterElem *pFilter, char *minval, char *maxval) = {
   NULL,
   less_ds,
-  large_ds,
+  larger_ds,
   equal_ds,
   lessEqual_ds,
   largeEqual_ds,
@@ -465,7 +482,7 @@ bool (*filterFunc_ds[])(SColumnFilterElem *pFilter, char *minval, char *maxval)
 bool (*filterFunc_dd[])(SColumnFilterElem *pFilter, char *minval, char *maxval) = {
   NULL,
   less_dd,
-  large_dd,
+  larger_dd,
   equal_dd,
   lessEqual_dd,
   largeEqual_dd,
@@ -551,7 +568,7 @@ bool (*rangeFilterFunc_dd[])(SColumnFilterElem *pFilter, char *minval, char *max
 
 __filter_func_t* getRangeFilterFuncArray(int32_t type) {
   switch(type) {
-    case TSDB_DATA_TYPE_BOOL:       return rangeFilterFunc_i8;
+    case TSDB_DATA_TYPE_BOOL:
     case TSDB_DATA_TYPE_TINYINT:    return rangeFilterFunc_i8;
     case TSDB_DATA_TYPE_SMALLINT:   return rangeFilterFunc_i16;
     case TSDB_DATA_TYPE_INT:        return rangeFilterFunc_i32;
@@ -565,7 +582,7 @@ __filter_func_t* getRangeFilterFuncArray(int32_t type) {
 
 __filter_func_t* getValueFilterFuncArray(int32_t type) {
   switch(type) {
-    case TSDB_DATA_TYPE_BOOL:       return filterFunc_i8;
+    case TSDB_DATA_TYPE_BOOL:
     case TSDB_DATA_TYPE_TINYINT:    return filterFunc_i8;
     case TSDB_DATA_TYPE_SMALLINT:   return filterFunc_i16;
     case TSDB_DATA_TYPE_INT:        return filterFunc_i32;

src/vnode/src/vnodeRead.c

@@ -281,7 +281,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
     vTrace("vgId:%d, QInfo:%p, dnode continues to exec query", pVnode->vgId, *qhandle);
 
     // In the retrieve blocking model, only 50% CPU will be used in query processing
-    if (tsHalfCoresForQuery) {
+    if (tsRetrieveBlockingModel) {
       qTableQuery(*qhandle);  // do execute query
       qReleaseQInfo(pVnode->qMgmt, (void **)&qhandle, false);
     } else {
@@ -380,7 +380,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
     freeHandle = true;
   } else {  // result is not ready, return immediately
     // Only effects in the non-blocking model
-    if (!tsHalfCoresForQuery) {
+    if (!tsRetrieveBlockingModel) {
       if (!buildRes) {
         assert(pRead->rpcHandle != NULL);
 

tests/script/general/parser/tags_filter.sim

@@ -149,4 +149,57 @@ if $rows != 2 then
   return -1
 endi
 
+print ==================>td-2424
+sql create table t1(ts timestamp, k float)
+sql insert into t1 values(now, 8.001)
+sql select * from t1 where k=8.001
+if $rows != 1 then
+  return -1
+endi
+
+sql select * from t1 where k<8.001
+if $rows != 0 then
+  return -1
+endi
+
+sql select * from t1 where k<=8.001
+if $rows != 1 then
+  return -1
+endi
+
+sql select * from t1 where k>8.001
+if $rows != 0 then
+  return -1
+endi
+
+sql select * from t1 where k>=8.001
+if $rows != 1 then
+  return -1
+endi
+
+sql select * from t1 where k<>8.001
+if $rows != 0 then
+  return -1
+endi
+
+sql select * from t1 where k>=8.001 and k<=8.001
+if $rows != 1 then
+  return -1
+endi
+
+sql select * from t1 where k>=8.0009999 and k<=8.001
+if $rows != 1 then
+  return -1
+endi
+
+sql select * from t1 where k>8.001 and k<=8.001
+if $rows != 0 then
+  return -1
+endi
+
+sql select * from t1 where k>=8.001 and k<8.001
+if $rows != 0 then
+  return -1
+endi
+
 system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file