Merge remote-tracking branch 'origin/feat/TS-4243-3.0' into feat/TS-4243-3.0

include/libs/nodes/querynodes.h

@@ -85,6 +85,7 @@ typedef struct SColumnNode {
   char        colName[TSDB_COL_NAME_LEN];
   int16_t     dataBlockId;
   int16_t     slotId;
+  int16_t     numOfPKs;
   bool        tableHasPk;
   bool        isPk;
 } SColumnNode;

include/libs/qcom/query.h

@@ -76,6 +76,7 @@ typedef struct STableComInfo {
   uint8_t  numOfTags;     // the number of tags in schema
   uint8_t  precision;     // the number of precision
   col_id_t numOfColumns;  // the number of columns
+  int16_t  numOfPKs;
   int32_t  rowSize;       // row size of the schema
 } STableComInfo;
 

source/common/src/tdataformat.c

@@ -1061,7 +1061,6 @@ _exit:
 static int32_t tRowKVUpsertColData(SRow *pRow, STSchema *pTSchema, SColData *aColData, int32_t nColData, int32_t flag) {
   int32_t code = 0;
 
-  SKVIdx   *pKVIdx = (SKVIdx *)pRow->data;
   uint8_t  *pv = NULL;
   int32_t   iColData = 0;
   SColData *pColData = &aColData[iColData];
@@ -1069,6 +1068,14 @@ static int32_t tRowKVUpsertColData(SRow *pRow, STSchema *pTSchema, SColData *aCo
   STColumn *pTColumn = &pTSchema->columns[iTColumn];
   int32_t   iCol = 0;
 
+  // primary keys
+  uint8_t         *data = pRow->data;
+  SPrimaryKeyIndex index;
+  for (int32_t i = 0; i < pRow->numOfPKs; i++) {
+    data += tGetPrimaryKeyIndex(data, &index);
+  }
+
+  SKVIdx *pKVIdx = (SKVIdx *)data;
   if (pRow->flag & KV_FLG_LIT) {
     pv = pKVIdx->idx + pKVIdx->nCol;
   } else if (pRow->flag & KV_FLG_MID) {

source/libs/function/src/builtinsimpl.c

@@ -2819,11 +2819,13 @@ static int32_t firstLastTransferInfoImpl(SFirstLastRes* pInput, SFirstLastRes* p
   pOutput->isNull = pInput->isNull;
   pOutput->ts = pInput->ts;
   pOutput->bytes = pInput->bytes;
+  pOutput->pkType = pInput->pkType;
 
   memcpy(pOutput->buf, pInput->buf, pOutput->bytes);
   if (pInput->pkData) {
     pOutput->pkBytes = pInput->pkBytes;
     memcpy(pOutput->buf+pOutput->bytes, pInput->pkData, pOutput->pkBytes);
+    pOutput->pkData = pOutput->buf + pOutput->bytes;
   }
   return TSDB_CODE_SUCCESS;
 }

source/libs/parser/src/parInsertSql.c

@@ -183,6 +183,8 @@ static int32_t parseBoundColumns(SInsertParseContext* pCxt, const char** pSql, E
 
   pBoundInfo->numOfBound = 0;
 
+  bool    hasPK = pTableMeta->tableInfo.numOfPKs;
+  int16_t numOfBoundPKs = 0;
   int16_t lastColIdx = -1;  // last column found
   int32_t code = TSDB_CODE_SUCCESS;
   while (TSDB_CODE_SUCCESS == code) {
@@ -219,14 +221,20 @@ static int32_t parseBoundColumns(SInsertParseContext* pCxt, const char** pSql, E
       pUseCols[index] = true;
       pBoundInfo->pColIndex[pBoundInfo->numOfBound] = index;
       ++pBoundInfo->numOfBound;
+      if (hasPK && (pSchema[index].flags & COL_IS_KEY)) ++numOfBoundPKs;
     }
   }
 
-  if (TSDB_CODE_SUCCESS == code && (BOUND_TAGS != boundColsType) && !pUseCols[0]) {
+  if (TSDB_CODE_SUCCESS == code && (BOUND_TAGS != boundColsType)) {
-    code = buildInvalidOperationMsg(&pCxt->msg, "primary timestamp column can not be null");
+    if (!pUseCols[0]) {
+      code = buildInvalidOperationMsg(&pCxt->msg, "Primary timestamp column should not be null");
+    }
+    if (numOfBoundPKs != pTableMeta->tableInfo.numOfPKs) {
+      code = buildInvalidOperationMsg(&pCxt->msg, "Primary key column should not be none");
+    }
   }
   if (TSDB_CODE_SUCCESS == code && (BOUND_ALL_AND_TBNAME == boundColsType) && !pUseCols[tbnameSchemaIndex]) {
-    code = buildInvalidOperationMsg(&pCxt->msg, "tbname column can not be null");
+    code = buildInvalidOperationMsg(&pCxt->msg, "tbname column should not be null");
   }
   taosMemoryFree(pUseCols);
 
@@ -469,13 +477,15 @@ static int32_t parseTagToken(const char** end, SToken* pToken, SSchema* pSchema,
   char*    endptr = NULL;
   int32_t  code = TSDB_CODE_SUCCESS;
 
+#if 0
   if (isNullValue(pSchema->type, pToken)) {
     if (TSDB_DATA_TYPE_TIMESTAMP == pSchema->type && PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
-      return buildSyntaxErrMsg(pMsgBuf, "primary timestamp should not be null", pToken->z);
+      return buildSyntaxErrMsg(pMsgBuf, "Primary timestamp column can not be null", pToken->z);
     }
 
     return TSDB_CODE_SUCCESS;
   }
+#endif
 
   //  strcpy(val->colName, pSchema->name);
   val->cid = pSchema->colId;
@@ -1643,7 +1653,10 @@ static int32_t parseValueToken(SInsertParseContext* pCxt, const char** pSql, STo
   int32_t code = checkAndTrimValue(pToken, pCxt->tmpTokenBuf, &pCxt->msg, pSchema->type);
   if (TSDB_CODE_SUCCESS == code && isNullValue(pSchema->type, pToken)) {
     if (TSDB_DATA_TYPE_TIMESTAMP == pSchema->type && PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
-      return buildSyntaxErrMsg(&pCxt->msg, "primary timestamp should not be null", pToken->z);
+      return buildSyntaxErrMsg(&pCxt->msg, "Primary timestamp column should not be null", pToken->z);
+    }
+    if (pSchema->flags & COL_IS_KEY) {
+      return buildSyntaxErrMsg(&pCxt->msg, "Primary key column should not be null", pToken->z);
     }
 
     pVal->flag = CV_FLAG_NULL;

source/libs/parser/src/parInsertStmt.c

@@ -267,6 +267,11 @@ int32_t qBindStmtColsValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, in
       pBind = bind + c;
     }
 
+    if(pBind->is_null && (pColSchema->flags & COL_IS_KEY)){
+      code = buildInvalidOperationMsg(&pBuf, "Primary key column should not be null");
+      goto _return;
+    }
+
     code = tColDataAddValueByBind(pCol, pBind, IS_VAR_DATA_TYPE(pColSchema->type) ? pColSchema->bytes - VARSTR_HEADER_SIZE: -1);
     if (code) {
       goto _return;
@@ -313,7 +318,12 @@ int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBu
     pBind = bind;
   }
 
-  tColDataAddValueByBind(pCol, pBind, IS_VAR_DATA_TYPE(pColSchema->type) ? pColSchema->bytes - VARSTR_HEADER_SIZE: -1);
+  if (pBind->is_null && (pColSchema->flags & COL_IS_KEY)) {
+    code = buildInvalidOperationMsg(&pBuf, "Primary key column should not be null");
+    goto _return;
+  }
+
+  tColDataAddValueByBind(pCol, pBind, IS_VAR_DATA_TYPE(pColSchema->type) ? pColSchema->bytes - VARSTR_HEADER_SIZE : -1);
 
   qDebug("stmt col %d bind %d rows data", colIdx, rowNum);
 

source/libs/parser/src/parTranslater.c

@@ -955,6 +955,7 @@ static void setColumnInfoBySchema(const SRealTableNode* pTable, const SSchema* p
   }
   pCol->tableHasPk = hasPkInTable(pTable->pMeta);
   pCol->isPk = (pCol->tableHasPk) && (pColSchema->flags & COL_IS_KEY);
+  pCol->numOfPKs = pTable->pMeta->tableInfo.numOfPKs;
 }
 
 static void setColumnInfoByExpr(STempTableNode* pTable, SExprNode* pExpr, SColumnNode** pColRef) {
@@ -5087,6 +5088,8 @@ static int32_t translateInsertProject(STranslateContext* pCxt, SInsertStmt* pIns
   SNode*  pPrimaryKeyExpr = NULL;
   SNode*  pBoundCol = NULL;
   SNode*  pProj = NULL;
+  int16_t numOfPKs = 0;
+  int16_t numOfBoundPKs = 0;
   FORBOTH(pBoundCol, pInsert->pCols, pProj, pProjects) {
     SColumnNode* pCol = (SColumnNode*)pBoundCol;
     SExprNode*   pExpr = (SExprNode*)pProj;
@@ -5102,12 +5105,18 @@ static int32_t translateInsertProject(STranslateContext* pCxt, SInsertStmt* pIns
     snprintf(pExpr->aliasName, sizeof(pExpr->aliasName), "%s", pCol->colName);
     if (PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId) {
       pPrimaryKeyExpr = (SNode*)pExpr;
+      numOfPKs = pCol->numOfPKs;
     }
+    if (pCol->isPk) ++numOfBoundPKs;
   }
 
   if (NULL == pPrimaryKeyExpr) {
     return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_COLUMNS_NUM,
-                                   "Primary timestamp column can not be null");
+                                   "Primary timestamp column should not be null");
+  }
+
+  if (numOfBoundPKs != numOfPKs) {
+    return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_SYNTAX_ERROR, "Primary key column should not be none");
   }
 
   return addOrderByPrimaryKeyToQuery(pCxt, pPrimaryKeyExpr, pInsert->pQuery);

source/libs/planner/src/planLogicCreater.c

@@ -294,6 +294,7 @@ static SNode* createFirstCol(uint64_t tableId, const SSchema* pSchema, const STa
   pCol->colType = COLUMN_TYPE_COLUMN;
   pCol->isPk = pSchema->flags & COL_IS_KEY;
   pCol->tableHasPk = hasPkInTable(pMeta);
+  pCol->numOfPKs = pMeta->tableInfo.numOfPKs;
   strcpy(pCol->colName, pSchema->name);
   return (SNode*)pCol;
 }

source/libs/qcom/src/querymsg.c

@@ -421,8 +421,16 @@ int32_t queryCreateTableMetaFromMsg(STableMetaRsp *msg, bool isStb, STableMeta *
 
   memcpy(pTableMeta->schema, msg->pSchemas, sizeof(SSchema) * total);
 
+  bool hasPK = (msg->numOfColumns > 1) && (pTableMeta->schema[1].flags & COL_IS_KEY);
   for (int32_t i = 0; i < msg->numOfColumns; ++i) {
     pTableMeta->tableInfo.rowSize += pTableMeta->schema[i].bytes;
+    if (hasPK && (i > 0)) {
+      if ((pTableMeta->schema[i].flags & COL_IS_KEY)) {
+        ++pTableMeta->tableInfo.numOfPKs;
+      } else {
+        hasPK = false;
+      }
+    }
   }
 
   qDebug("table %s uid %" PRIx64 " meta returned, type %d vgId:%d db %s stb %s suid %" PRIx64 " sver %d tver %d"