enh:support join type

source/common/src/tdatablock.c

@@ -229,7 +229,11 @@ int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t currentRow,
                          bool trimValue) {
   int32_t len = pColumnInfoData->info.bytes;
   if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
-    len = varDataTLen(pData);
+    if (pColumnInfoData->info.type == TSDB_DATA_TYPE_JSON) {
+      len = getJsonValueLen(pData);
+    } else {
+      len = varDataTLen(pData);
+    }  
     if (pColumnInfoData->varmeta.allocLen < (numOfRows * len + pColumnInfoData->varmeta.length)) {
       int32_t code = colDataReserve(pColumnInfoData, (numOfRows * len + pColumnInfoData->varmeta.length));
       if (code != TSDB_CODE_SUCCESS) {

source/libs/executor/inc/mergejoin.h

@@ -57,6 +57,7 @@ typedef struct SMJoinColMap {
 typedef struct SMJoinColInfo {
   int32_t  srcSlot;
   int32_t  dstSlot;
+  bool     jsonData;
   bool     vardata;
   int32_t* offset;
   int32_t  bytes;

source/libs/executor/src/mergejoinoperator.c

@@ -60,6 +60,43 @@ int32_t mJoinBuildEqGrp(SMJoinTableCtx* pTable, int64_t timestamp, bool* wholeBl
 }
 
 
+void mJoinTrimKeepFirstRow(SSDataBlock* pBlock) {
+  int32_t bmLen = BitmapLen(pBlock->info.rows);
+  size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
+
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, i);
+    // it is a reserved column for scalar function, and no data in this column yet.
+    if (pDst->pData == NULL || (IS_VAR_DATA_TYPE(pDst->info.type) && pDst->varmeta.length == 0)) {
+      continue;
+    }
+
+    if (IS_VAR_DATA_TYPE(pDst->info.type)) {
+      pDst->varmeta.length = 0;
+
+      if (!colDataIsNull_var(pDst, 0)) {
+        char*   p1 = colDataGetVarData(pDst, 0);
+        int32_t len = 0;
+        if (pDst->info.type == TSDB_DATA_TYPE_JSON) {
+          len = getJsonValueLen(p1);
+        } else {
+          len = varDataTLen(p1);
+        }
+        pDst->varmeta.length = len;
+      }
+    } else {
+      bool isNull = colDataIsNull_f(pDst->nullbitmap, 0);
+
+      memset(pDst->nullbitmap, 0, bmLen);
+      if (isNull) {
+        colDataSetNull_f(pDst->nullbitmap, 0);
+      }
+    }
+  }
+
+  pBlock->info.rows = 1;
+}
+
 
 void mJoinTrimKeepOneRow(SSDataBlock* pBlock, int32_t totalRows, const bool* pBoolList) {
   //  int32_t totalRows = pBlock->info.rows;
@@ -341,6 +378,7 @@ int32_t mJoinFilterAndKeepSingleRow(SSDataBlock* pBlock, SFilterInfo* pFilterInf
   
   if (status == FILTER_RESULT_ALL_QUALIFIED) {
     pBlock->info.rows = 1;
+    mJoinTrimKeepFirstRow(pBlock);
   } else if (status == FILTER_RESULT_NONE_QUALIFIED) {
     pBlock->info.rows = 0;
   } else if (status == FILTER_RESULT_PARTIAL_QUALIFIED) {
@@ -713,6 +751,7 @@ static int32_t mJoinInitColsInfo(int32_t* colNum, int64_t* rowSize, SMJoinColInf
   FOREACH(pNode, pList) {
     SColumnNode* pColNode = (SColumnNode*)pNode;
     (*pCols)[i].srcSlot = pColNode->slotId;
+    (*pCols)[i].jsonData = TSDB_DATA_TYPE_JSON == pColNode->node.resType.type;
     (*pCols)[i].vardata = IS_VAR_DATA_TYPE(pColNode->node.resType.type);
     (*pCols)[i].bytes = pColNode->node.resType.bytes;
     *rowSize += pColNode->node.resType.bytes;
@@ -1257,7 +1296,10 @@ bool mJoinCopyKeyColsDataToBuf(SMJoinTableCtx* pTable, int32_t rowIdx, size_t *p
     if (colDataIsNull_s(pTable->keyCols[0].colData, rowIdx)) {
       return true;
     }
-    if (pTable->keyCols[0].vardata) {
+    if (pTable->keyCols[0].jsonData) {
+      pData = pTable->keyCols[0].data + pTable->keyCols[0].offset[rowIdx];
+      bufLen = getJsonValueLen(pData);
+    } else if (pTable->keyCols[0].vardata) {
       pData = pTable->keyCols[0].data + pTable->keyCols[0].offset[rowIdx];
       bufLen = varDataTLen(pData);
     } else {
@@ -1270,7 +1312,11 @@ bool mJoinCopyKeyColsDataToBuf(SMJoinTableCtx* pTable, int32_t rowIdx, size_t *p
       if (colDataIsNull_s(pTable->keyCols[i].colData, rowIdx)) {
         return true;
       }
-      if (pTable->keyCols[i].vardata) {
+      if (pTable->keyCols[0].jsonData) {
+        pData = pTable->keyCols[i].data + pTable->keyCols[i].offset[rowIdx];
+        memcpy(pTable->keyBuf + bufLen, pData, getJsonValueLen(pData));
+        bufLen += getJsonValueLen(pData);
+      } else if (pTable->keyCols[i].vardata) {
         pData = pTable->keyCols[i].data + pTable->keyCols[i].offset[rowIdx];
         memcpy(pTable->keyBuf + bufLen, pData, varDataTLen(pData));
         bufLen += varDataTLen(pData);

source/libs/executor/src/scanoperator.c

@@ -3767,7 +3767,7 @@ static int32_t stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {
     taosMemoryFree(pSubTblsInfo);
     pInfo->pSubTablesMergeInfo = NULL;
 
-    taosMemoryTrim(0);
+    //taosMemoryTrim(0);
   }
   return TSDB_CODE_SUCCESS;
 }

source/libs/function/src/builtins.c

@@ -545,6 +545,9 @@ static int32_t translatePercentile(SFunctionNode* pFunc, char* pErrBuf, int32_t
 
   for (int32_t i = 1; i < numOfParams; ++i) {
     SValueNode* pValue = (SValueNode*)nodesListGetNode(pFunc->pParameterList, i);
+    if (QUERY_NODE_VALUE != nodeType(pValue)) {
+      return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+    }
     pValue->notReserved = true;
 
     uint8_t paraType = getSDataTypeFromNode(nodesListGetNode(pFunc->pParameterList, i))->type;
@@ -3142,7 +3145,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "unique",
     .type = FUNCTION_TYPE_UNIQUE,
-    .classification = FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
+    .classification = FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
     .translateFunc = translateUnique,
     .getEnvFunc   = getUniqueFuncEnv,
     .initFunc     = uniqueFunctionSetup,

source/libs/parser/src/parTranslater.c

@@ -956,7 +956,7 @@ static bool isTimeLineAlignedQuery(SNode* pStmt) {
     return false;
   }
   SSelectStmt* pSub = (SSelectStmt*)((STempTableNode*)pSelect->pFromTable)->pSubquery;
-  if (nodesListMatch(pSelect->pPartitionByList, pSub->pPartitionByList)) {
+  if (pSelect->pPartitionByList && nodesListMatch(pSelect->pPartitionByList, pSub->pPartitionByList)) {
     return true;
   }
   return false;
@@ -5397,6 +5397,7 @@ static EDealRes appendTsForImplicitTsFuncImpl(SNode* pNode, void* pContext) {
       return DEAL_RES_ERROR;                               
     }
 
+/*
     SSelectStmt* pSelect = (SSelectStmt*)pCxt->pCurrStmt;
     if ((NULL != pSelect->pFromTable && QUERY_NODE_TEMP_TABLE == nodeType(pSelect->pFromTable) &&
         !isGlobalTimeLineQuery(((STempTableNode*)pSelect->pFromTable)->pSubquery) &&
@@ -5407,6 +5408,7 @@ static EDealRes appendTsForImplicitTsFuncImpl(SNode* pNode, void* pContext) {
                                      "%s function requires valid time series input", pFunc->functionName);
       return DEAL_RES_ERROR;                                     
     }
+*/
 
     SNode*         pPrimaryKey = NULL;
     SSHashObj*     pTableAlias = NULL;
@@ -5571,9 +5573,6 @@ static int32_t translateSelectFrom(STranslateContext* pCxt, SSelectStmt* pSelect
   if (TSDB_CODE_SUCCESS == code) {
     code = checkIsEmptyResult(pCxt, pSelect);
   }
-  if (TSDB_CODE_SUCCESS == code) {
-    code = appendTsForImplicitTsFunc(pCxt, pSelect);
-  }
   if (TSDB_CODE_SUCCESS == code) {
     resetSelectFuncNumWithoutDup(pSelect);
     code = checkAggColCoexist(pCxt, pSelect);
@@ -5587,6 +5586,9 @@ static int32_t translateSelectFrom(STranslateContext* pCxt, SSelectStmt* pSelect
   if (TSDB_CODE_SUCCESS == code) {
     code = translateInterp(pCxt, pSelect);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = appendTsForImplicitTsFunc(pCxt, pSelect);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = replaceOrderByAliasForSelect(pCxt, pSelect);
   }

source/libs/parser/src/parUtil.c

@@ -1169,7 +1169,7 @@ int32_t getTableCfgFromCache(SParseMetaCache* pMetaCache, const SName* pName, ST
   tNameExtractFullName(pName, fullName);
   STableCfg* pCfg = NULL;
   int32_t    code = getMetaDataFromHash(fullName, strlen(fullName), pMetaCache->pTableCfg, (void**)&pCfg);
-  if (TSDB_CODE_SUCCESS == code) {
+  if (TSDB_CODE_SUCCESS == code && NULL != pCfg) {
     *pOutput = tableCfgDup(pCfg);
     if (NULL == *pOutput) {
       code = TSDB_CODE_OUT_OF_MEMORY;

source/libs/parser/test/parSelectTest.cpp

@@ -464,7 +464,7 @@ TEST_F(ParserSelectTest, setOperator) {
 TEST_F(ParserSelectTest, setOperatorSemanticCheck) {
   useDb("root", "test");
 
-  run("SELECT c1, c2 FROM t1 UNION ALL SELECT c1, c2 FROM t1 ORDER BY ts", TSDB_CODE_PAR_INVALID_COLUMN);
+  run("SELECT c1, c2 FROM t1 UNION ALL SELECT c1, c2 FROM t1 ORDER BY ts", TSDB_CODE_PAR_ORDERBY_UNKNOWN_EXPR);
 }
 
 TEST_F(ParserSelectTest, informationSchema) {

source/util/src/terror.c

@@ -631,7 +631,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_WJOIN_HAVING_EXPR,  "Invalid window join
 TAOS_DEFINE_ERROR(TSDB_CODE_PAR_GRP_WINDOW_NOT_ALLOWED,     "GROUP BY/PARTITION BY/WINDOW-clause can't be used in WINDOW join")
 TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_WIN_OFFSET_UNIT,    "Invalid window offset unit")
 TAOS_DEFINE_ERROR(TSDB_CODE_PAR_VALID_PRIM_TS_REQUIRED,     "Valid primary timestamp required")
-TAOS_DEFINE_ERROR(TSDB_CODE_PAR_ORDERBY_UNKNOWN_EXPR,       "Unknown expr in order by clause")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_ORDERBY_UNKNOWN_EXPR,       "Invalid expr in order by clause")
 
 //planner
 TAOS_DEFINE_ERROR(TSDB_CODE_PLAN_INTERNAL_ERROR,            "Planner internal error")

tests/system-test/2-query/nestedQuery.py

@@ -1377,7 +1377,7 @@ class TDTestCase:
                 tdSql.query(sql)
                 self.cur1.execute(sql)
                 self.explain_sql(sql)
-            elif (mathlist == ['MAVG']) or (mathlist == ['TAIL']) or (mathlist == ['CSUM']) or (mathlist == ['UNIQUE']) or (mathlist == ['statecount','stateduration']) :
+            elif (mathlist == ['MAVG']) or (mathlist == ['TAIL']) or (mathlist == ['CSUM']) or (mathlist == ['statecount','stateduration']) :
                 sql = "select  count(asct1) from  ( select  "
                 sql += "%s as asct1 " % math_fun_join_2
                 sql += "from stable_1 t1 , stable_2 t2 where t1.ts = t2.ts and "
@@ -1389,7 +1389,7 @@ class TDTestCase:
                 tdLog.info(sql)
                 tdLog.info(len(sql))
                 tdSql.error(sql)
-            elif (mathlist == ['SAMPLE']) or (mathlist == ['HISTOGRAM']) or (mathlist == ['HYPERLOGLOG']) (mathlist == ['MODE']) :
+            elif (mathlist == ['SAMPLE']) or (mathlist == ['UNIQUE'])  or (mathlist == ['HISTOGRAM']) or (mathlist == ['HYPERLOGLOG']) or (mathlist == ['MODE']) :
                 sql = "select  count(asct1) from  ( select  "
                 sql += "%s as asct1 " % math_fun_join_2
                 sql += "from stable_1 t1 , stable_2 t2 where t1.ts = t2.ts and "
@@ -1631,9 +1631,7 @@ class TDTestCase:
                 or (mathlist == ['HYPERLOGLOG']) or (mathlist == ['UNIQUE']) or (mathlist == ['MODE']) or (mathlist == ['statecount','stateduration']) :
                 sql = "select  count(asct1) from  ( select  "
                 sql += "%s as asct1 " % math_fun_join_2
-                sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts and  "
+                sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts "
-                sql += "%s " % random.choice(self.t_join_where)
-                sql += " and %s " % random.choice(self.qt_u_or_where)
                 sql += "%s " % random.choice(self.partiton_where_j)
                 sql += "%s " % random.choice(self.slimit1_where)
                 sql += ") "
@@ -3701,9 +3699,7 @@ class TDTestCase:
                 sql = "select   asct1,(now()),(now()),asct2 ,now(),today(),timezone() from  ( select "
                 sql += "%s as asct2, " % time_fun_join_1
                 sql += "%s as asct1 " % time_fun_join_2
-                sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts and  "
+                sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts "
-                sql += "%s " % random.choice(self.t_join_where)
-                sql += " and %s " % random.choice(self.qt_u_or_where)
                 sql += "%s " % random.choice(self.partiton_where_j)
                 sql += "%s " % random.choice(self.slimit1_where)
                 sql += ") "
@@ -4853,8 +4849,7 @@ class TDTestCase:
         for i in range(self.fornum):
             sql = "select  * from ( select "
             sql += "%s " % random.choice(self.calc_calculate_regular_j)
-            sql += "  from stable_1 t1, stable_2 t2 where t1.ts = t2.ts and  "
+            sql += "  from stable_1 t1, stable_2 t2 where t1.ts = t2.ts "
-            sql += "%s " % random.choice(self.t_join_where)
             sql += "%s " % random.choice(self.partiton_where_j)
             sql += ") "
             sql += "%s " % random.choice([self.limit_where[2] , self.limit_where[3]] )

tests/system-test/2-query/para_tms.py

@@ -1449,8 +1449,19 @@ class TDTestCase:
                 tdLog.info(sql)
                 tdSql.query(sql)
                 self.explain_sql(sql)
-            elif (mathlist == ['MAVG']) or (mathlist == ['SAMPLE']) or (mathlist == ['TAIL']) or (mathlist == ['CSUM']) or (mathlist == ['HISTOGRAM']) \
+            elif (mathlist == ['MAVG']) or (mathlist == ['TAIL']) or (mathlist == ['CSUM']) or (mathlist == ['UNIQUE']) or (mathlist == ['statecount','stateduration']) :
-                or (mathlist == ['HYPERLOGLOG']) or (mathlist == ['UNIQUE']) or (mathlist == ['MODE']) or (mathlist == ['statecount','stateduration']) :
+                sql = "select /*+ para_tables_sort() */ count(asct1) from  ( select /*+ para_tables_sort() */ "
+                sql += "%s as asct1 " % math_fun_join_2
+                sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts and  "
+                sql += "%s " % random.choice(self.t_join_where)
+                sql += " and %s " % random.choice(self.qt_u_or_where)
+                sql += "%s " % random.choice(self.partiton_where_j)
+                sql += "%s " % random.choice(self.slimit1_where)
+                sql += ") "
+                sql += "%s ;" % random.choice(self.limit_u_where)
+                tdLog.info(sql)
+                tdSql.error(sql)
+            elif (mathlist == ['SAMPLE']) or (mathlist == ['HISTOGRAM']) or (mathlist == ['HYPERLOGLOG']) or (mathlist == ['MODE']) :
                 sql = "select /*+ para_tables_sort() */ count(asct1) from  ( select /*+ para_tables_sort() */ "
                 sql += "%s as asct1 " % math_fun_join_2
                 sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts and  "
@@ -3350,9 +3361,7 @@ class TDTestCase:
                 sql = "select /*+ para_tables_sort() */  asct1,(now()),(now()),asct2 ,now(),today(),timezone() from  ( select /*+ para_tables_sort() */"
                 sql += "%s as asct2, " % time_fun_join_1
                 sql += "%s as asct1 " % time_fun_join_2
-                sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts and  "
+                sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts "
-                sql += "%s " % random.choice(self.t_join_where)
-                sql += " and %s " % random.choice(self.qt_u_or_where)
                 sql += "%s " % random.choice(self.partiton_where_j)
                 sql += "%s " % random.choice(self.slimit1_where)
                 sql += ") "

tests/system-test/2-query/para_tms2.py

@@ -1640,9 +1640,7 @@ class TDTestCase:
                 or (mathlist == ['HYPERLOGLOG']) or (mathlist == ['UNIQUE']) or (mathlist == ['MODE']) or (mathlist == ['statecount','stateduration']) :
                 sql = "select /*+ para_tables_sort() */ count(asct1) from  ( select /*+ para_tables_sort() */ "
                 sql += "%s as asct1 " % math_fun_join_2
-                sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts and  "
+                sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts "
-                sql += "%s " % random.choice(self.t_join_where)
-                sql += " and %s " % random.choice(self.qt_u_or_where)
                 sql += "%s " % random.choice(self.partiton_where_j)
                 sql += "%s " % random.choice(self.slimit1_where)
                 sql += ") "

tests/system-test/2-query/stablity.py

@@ -1239,7 +1239,7 @@ class TDTestCase:
                 tdLog.info(sql)
                 tdSql.query(sql)
                 self.explain_sql(sql)
-            elif (mathlist == ['SAMPLE']) or (mathlist == ['HISTOGRAM']) or (mathlist == ['HYPERLOGLOG']) or (mathlist == ['MODE']) :
+            elif (mathlist == ['SAMPLE']) or (mathlist == ['TAIL']) or (mathlist == ['HISTOGRAM']) or (mathlist == ['HYPERLOGLOG']) or (mathlist == ['MODE']) or (mathlist == ['UNIQUE'])  :
                 sql = "select  %s as asct1 " % math_fun_join_2
                 sql += "from stable_1 t1 , stable_2 t2 where t1.ts = t2.ts and "
                 sql += "%s " % random.choice(self.t_join_where)
@@ -1249,7 +1249,7 @@ class TDTestCase:
                 tdLog.info(sql)
                 tdSql.query(sql)
                 self.explain_sql(sql)
-            elif (mathlist == ['MAVG']) or (mathlist == ['TAIL']) or (mathlist == ['CSUM']) or (mathlist == ['UNIQUE']) or (mathlist == ['statecount','stateduration']) :
+            elif (mathlist == ['MAVG']) or (mathlist == ['CSUM']) or (mathlist == ['statecount','stateduration']) :
                 sql = "select  %s as asct1 " % math_fun_join_2
                 sql += "from stable_1 t1 , stable_2 t2 where t1.ts = t2.ts and "
                 sql += "%s " % random.choice(self.t_join_where)
@@ -1449,8 +1449,8 @@ class TDTestCase:
                 tdLog.info(sql)
                 tdSql.query(sql)
                 self.explain_sql(sql)
-            elif (mathlist == ['MAVG']) or (mathlist == ['SAMPLE']) or (mathlist == ['TAIL']) or (mathlist == ['CSUM']) or (mathlist == ['HISTOGRAM']) \
+            elif (mathlist == ['SAMPLE']) or (mathlist == ['TAIL']) or (mathlist == ['HISTOGRAM']) \
-                or (mathlist == ['HYPERLOGLOG']) or (mathlist == ['UNIQUE']) or (mathlist == ['MODE']) or (mathlist == ['statecount','stateduration']) :
+                or (mathlist == ['HYPERLOGLOG']) or (mathlist == ['UNIQUE']) or (mathlist == ['MODE']) :
                 sql = "select  count(asct1) from  ( select  "
                 sql += "%s as asct1 " % math_fun_join_2
                 sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts and  "
@@ -1463,6 +1463,20 @@ class TDTestCase:
                 tdLog.info(sql)
                 tdSql.query(sql)
                 self.explain_sql(sql)
+            elif (mathlist == ['MAVG']) or (mathlist == ['CSUM']) \
+                or (mathlist == ['statecount','stateduration']) :
+                sql = "select  count(asct1) from  ( select  "
+                sql += "%s as asct1 " % math_fun_join_2
+                sql += "from stable_1  t1, stable_2 t2 where t1.ts = t2.ts and  "
+                sql += "%s " % random.choice(self.t_join_where)
+                sql += " and %s " % random.choice(self.qt_u_or_where)
+                sql += "%s " % random.choice(self.partiton_where_j)
+                sql += "%s " % random.choice(self.slimit1_where)
+                sql += ") "
+                sql += "%s ;" % random.choice(self.limit_u_where)
+                tdLog.info(sql)
+                tdSql.error(sql)
+                                
 
         #taos -f sql
         # startTime_taosf = time.time()