Merge pull request #15799 from taosdata/szhou/fix/nullgroup

fix: all null value group with all null out

include/libs/function/functionMgt.h

@@ -202,6 +202,7 @@ bool fmIsForbidStreamFunc(int32_t funcId);
 bool fmIsIntervalInterpoFunc(int32_t funcId);
 bool fmIsInterpFunc(int32_t funcId);
 bool fmIsLastRowFunc(int32_t funcId);
+bool fmIsReturnNotNullFunc(int32_t funcId);
 bool fmIsSelectValueFunc(int32_t funcId);
 bool fmIsSystemInfoFunc(int32_t funcId);
 bool fmIsImplicitTsFunc(int32_t funcId);

source/libs/executor/src/executorimpl.c

@@ -1437,7 +1437,8 @@ static void setExecutionContext(SOperatorInfo* pOperator, int32_t numOfOutput, u
   pAggInfo->groupId = groupId;
 }
 
-static void doUpdateNumOfRows(SResultRow* pRow, int32_t numOfExprs, const int32_t* rowCellOffset) {
+static void doUpdateNumOfRows(SqlFunctionCtx* pCtx, SResultRow* pRow, int32_t numOfExprs, const int32_t* rowCellOffset) {
+  bool returnNotNull = false;
   for (int32_t j = 0; j < numOfExprs; ++j) {
     struct SResultRowEntryInfo* pResInfo = getResultEntryInfo(pRow, j, rowCellOffset);
     if (!isRowEntryInitialized(pResInfo)) {
@@ -1447,6 +1448,15 @@ static void doUpdateNumOfRows(SResultRow* pRow, int32_t numOfExprs, const int32_
     if (pRow->numOfRows < pResInfo->numOfRes) {
       pRow->numOfRows = pResInfo->numOfRes;
     }
+
+    if (fmIsReturnNotNullFunc(pCtx[j].functionId)) {
+      returnNotNull = true;
+    }
+  }
+  // if all expr skips all blocks, e.g. all null inputs for max function, output one row in final result.
+  //  except for first/last, which require not null output, output no rows
+  if (pRow->numOfRows == 0 && !returnNotNull) {
+    pRow->numOfRows = 1;
   }
 }
 
@@ -1458,7 +1468,7 @@ int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosi
   SFilePage*  page = getBufPage(pBuf, resultRowPosition->pageId);
   SResultRow* pRow = (SResultRow*)((char*)page + resultRowPosition->offset);
 
-  doUpdateNumOfRows(pRow, numOfExprs, rowCellOffset);
+  doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowCellOffset);
   if (pRow->numOfRows == 0) {
     releaseBufPage(pBuf, page);
     return 0;
@@ -1514,7 +1524,7 @@ int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprI
 
     SResultRow* pRow = (SResultRow*)((char*)page + pPos->pos.offset);
 
-    doUpdateNumOfRows(pRow, numOfExprs, rowCellOffset);
+    doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowCellOffset);
     if (pRow->numOfRows == 0) {
       pGroupResInfo->index += 1;
       releaseBufPage(pBuf, page);

source/libs/function/src/builtinsimpl.c

@@ -64,6 +64,9 @@ typedef struct SMinmaxResInfo {
   bool      assign;  // assign the first value or not
   int64_t   v;
   STuplePos tuplePos;
+
+  STuplePos nullTuplePos;
+  bool      nullTupleSaved;
 } SMinmaxResInfo;
 
 typedef struct STopBotResItem {
@@ -75,6 +78,10 @@ typedef struct STopBotResItem {
 typedef struct STopBotRes {
   int32_t         maxSize;
   int16_t         type;
+
+  STuplePos       nullTuplePos;
+  bool            nullTupleSaved;
+
   STopBotResItem* pItems;
 } STopBotRes;
 
@@ -221,6 +228,10 @@ typedef struct SSampleInfo {
   int32_t    numSampled;
   uint8_t    colType;
   int16_t    colBytes;
+
+  STuplePos  nullTuplePos;
+  bool       nullTupleSaved;
+
   char*      data;
   STuplePos* tuplePos;
 } SSampleInfo;
@@ -1134,6 +1145,9 @@ bool minmaxFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo)
   SMinmaxResInfo* buf = GET_ROWCELL_INTERBUF(pResultInfo);
   buf->assign = false;
   buf->tuplePos.pageId = -1;
+
+  buf->nullTupleSaved = false;
+  buf->nullTuplePos.pageId = -1;
   return true;
 }
 
@@ -1575,6 +1589,10 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
   }
 
 _min_max_over:
+  if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pBuf->nullTupleSaved ) {
+    doSaveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pBuf->nullTuplePos);
+    pBuf->nullTupleSaved = true;
+  }
   return numOfElems;
 }
 
@@ -1615,7 +1633,7 @@ int32_t minmaxFunctionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   if (pEntryInfo->numOfRes > 0) {
     setSelectivityValue(pCtx, pBlock, &pRes->tuplePos, currentRow);
   } else {
-    setNullSelectivityValue(pCtx, pBlock, currentRow);
+    setSelectivityValue(pCtx, pBlock, &pRes->nullTuplePos, currentRow);
   }
 
   return pEntryInfo->numOfRes;
@@ -3366,6 +3384,8 @@ bool topBotFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
 
   pRes->maxSize = pCtx->param[1].param.i;
 
+  pRes->nullTupleSaved = false;
+  pRes->nullTuplePos.pageId = -1;
   return true;
 }
 
@@ -3403,6 +3423,10 @@ int32_t topFunction(SqlFunctionCtx* pCtx) {
     doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, pRes->type, pInput->uid, pResInfo, true);
   }
 
+  if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pRes->nullTupleSaved) {
+    doSaveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pRes->nullTuplePos);
+    pRes->nullTupleSaved = true;
+  }
   return TSDB_CODE_SUCCESS;
 }
 
@@ -3427,6 +3451,11 @@ int32_t bottomFunction(SqlFunctionCtx* pCtx) {
     doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, pRes->type, pInput->uid, pResInfo, false);
   }
 
+  if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pRes->nullTupleSaved) {
+    doSaveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pRes->nullTuplePos);
+    pRes->nullTupleSaved = true;
+  }
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -3625,6 +3654,11 @@ int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
 
   // todo assign the tag value and the corresponding row data
   int32_t currentRow = pBlock->info.rows;
+  if (pEntryInfo->numOfRes <= 0) {
+    colDataAppendNULL(pCol, currentRow);
+    setSelectivityValue(pCtx, pBlock, &pRes->nullTuplePos, currentRow);
+    return pEntryInfo->numOfRes;
+  }
   for (int32_t i = 0; i < pEntryInfo->numOfRes; ++i) {
     STopBotResItem* pItem = &pRes->pItems[i];
     if (type == TSDB_DATA_TYPE_FLOAT) {
@@ -4897,7 +4931,8 @@ bool sampleFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo)
   pInfo->numSampled = 0;
   pInfo->colType = pCtx->resDataInfo.type;
   pInfo->colBytes = pCtx->resDataInfo.bytes;
-
+  pInfo->nullTuplePos.pageId = -1;
+  pInfo->nullTupleSaved = false;
   pInfo->data = (char*)pInfo + sizeof(SSampleInfo);
   pInfo->tuplePos = (STuplePos*)((char*)pInfo + sizeof(SSampleInfo) + pInfo->samples * pInfo->colBytes);
 
@@ -4943,6 +4978,11 @@ int32_t sampleFunction(SqlFunctionCtx* pCtx) {
     doReservoirSample(pCtx, pInfo, data, i);
   }
 
+  if (pInfo->numSampled == 0 && pCtx->subsidiaries.num > 0 && !pInfo->nullTupleSaved) {
+    doSaveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pInfo->nullTuplePos);
+    pInfo->nullTupleSaved = true;
+  }
+
   SET_VAL(pResInfo, pInfo->numSampled, pInfo->numSampled);
   return TSDB_CODE_SUCCESS;
 }
@@ -4957,6 +4997,11 @@ int32_t sampleFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
 
   int32_t currentRow = pBlock->info.rows;
+  if (pInfo->numSampled == 0) {
+    colDataAppendNULL(pCol, currentRow);
+    setSelectivityValue(pCtx, pBlock, &pInfo->nullTuplePos, currentRow);
+    return pInfo->numSampled;
+  }
   for (int32_t i = 0; i < pInfo->numSampled; ++i) {
     colDataAppend(pCol, currentRow + i, pInfo->data + i * pInfo->colBytes, false);
     setSelectivityValue(pCtx, pBlock, &pInfo->tuplePos[i], currentRow + i);

source/libs/function/src/functionMgt.c

@@ -221,6 +221,18 @@ bool fmIsLastRowFunc(int32_t funcId) {
   return FUNCTION_TYPE_LAST_ROW == funcMgtBuiltins[funcId].type;
 }
 
+bool fmIsReturnNotNullFunc(int32_t funcId) {
+  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
+    return false;
+  }
+  return FUNCTION_TYPE_LAST == funcMgtBuiltins[funcId].type ||
+         FUNCTION_TYPE_LAST_PARTIAL == funcMgtBuiltins[funcId].type ||
+         FUNCTION_TYPE_LAST_MERGE == funcMgtBuiltins[funcId].type ||
+         FUNCTION_TYPE_FIRST == funcMgtBuiltins[funcId].type ||
+         FUNCTION_TYPE_FIRST_PARTIAL == funcMgtBuiltins[funcId].type ||
+         FUNCTION_TYPE_FIRST_MERGE == funcMgtBuiltins[funcId].type;
+}
+
 bool fmIsSelectValueFunc(int32_t funcId) {
   if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
     return false;

tests/script/tsim/parser/function.sim

@@ -327,8 +327,8 @@ endi
 
 print =================>td-2610
 sql select stddev(k) from tm2 where ts='2020-12-29 18:46:19.109'
-if $rows != 0 then
-  print expect 0, actual:$rows
+if $rows != 1 then
+  print expect 1, actual:$rows
   return -1
 endi
 sql select twa(k) from tm2 where ts='2020-12-29 18:46:19.109'
@@ -406,7 +406,7 @@ if $data00 != 1.378704626 then
 endi
 
 sql select stddev(c) from m1
-if $rows != 0 then
+if $rows != 1 then
   return -1
 endi
 

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

@@ -295,7 +295,7 @@ class TDTestCase:
         tdSql.checkData(0, 0, 4.500000000)
 
         tdSql.query(f" select avg(c1) from {dbname}.stb1 where c1 is null ")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
 
     def avg_func_filter(self, dbname="db"):

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

@@ -86,7 +86,7 @@ class TDTestCase:
     def distribute_agg_query(self, dbname="testdb"):
         # basic filter
         tdSql.query(f"select apercentile(c1 , 20) from {dbname}.stb1 where c1 is null")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
         tdSql.query(f"select apercentile(c1 , 20) from {dbname}.stb1 where t1=1")
         tdSql.checkData(0,0,2.800000000)

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

@@ -168,7 +168,7 @@ class TDTestCase:
     def distribute_agg_query(self, dbname="testdb"):
         # basic filter
         tdSql.query(f"select max(c1) from {dbname}.stb1 where c1 is null")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
         tdSql.query(f"select max(c1) from {dbname}.stb1 where t1=1")
         tdSql.checkData(0,0,10)

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

@@ -167,7 +167,7 @@ class TDTestCase:
     def distribute_agg_query(self, dbname="testdb"):
         # basic filter
         tdSql.query(f"select min(c1) from {dbname}.stb1 where c1 is null")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
         tdSql.query(f"select min(c1) from {dbname}.stb1 where t1=1")
         tdSql.checkData(0,0,2)

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

@@ -195,7 +195,7 @@ class TDTestCase:
     def distribute_agg_query(self):
         # basic filter
         tdSql.query("select spread(c1) from stb1 where c1 is null")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
         tdSql.query("select spread(c1) from stb1 where t1=1")
         tdSql.checkData(0,0,8.000000000)

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

@@ -743,7 +743,7 @@ class TDTestCase:
         # filter data
 
         tdSql.query(" select sample(c1, 20 ) from t1 where c1 is null ")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
         tdSql.query(" select sample(c1, 20 ) from t1 where c1 =6 ")
         tdSql.checkRows(1)
@@ -891,4 +891,4 @@ class TDTestCase:
         tdLog.success("%s successfully executed" % __file__)
 
 tdCases.addWindows(__file__, TDTestCase())
-tdCases.addLinux(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file