taosdata / TDengine / #5005

#include "streamMsg.h"

#include "tjson.h"

static const char* jkFieldName     = "name";

static const char* jkFieldType     = "type";

static const char* jkFieldFlags    = "flags";

static const char* jkFieldBytes    = "bytes";

static const char* jkFieldCompress = "compress";

static const char* jkFieldTypeMod  = "typeMod";

  }

    pJson, jkFieldFlags, pField->flags));

    pJson, jkFieldBytes, pField->bytes));

    pJson, jkFieldCompress, pField->compress));

    pJson, jkFieldTypeMod, pField->typeMod));

}

    pJson, jkFieldCompress, &pField->compress));

}

    pJson, jkFieldFlags, pField->flags));

    pJson, jkFieldBytes, pField->bytes));

  // TAOS_CHECK_RETURN(tjsonAddIntegerToObject(

  //   pJson, jkFieldCompress, pField->compress));

  // TAOS_CHECK_RETURN(tjsonAddIntegerToObject(

  //   pJson, jkFieldTypeMod, pField->typeMod));

}

  // TAOS_CHECK_RETURN(tjsonGetUIntValue(

  //   pJson, jkFieldCompress, &pField->compress));

  // TAOS_CHECK_RETURN(tjsonGetIntValue(pJson, jkFieldTypeMod, &pField->typeMod));

}

static const char* jkSessionTriggerSlotId     = "slotId";

static const char* jkSessionTriggerSessionVal = "sessionVal";

    pJson, jkSessionTriggerSlotId, pTrigger->slotId));

    pJson, jkSessionTriggerSessionVal, pTrigger->sessionVal));

}

    pJson, jkSessionTriggerSlotId, &pTrigger->slotId));

    pJson, jkSessionTriggerSessionVal, &pTrigger->sessionVal));

}

static const char* jkStateTriggerSlotId           = "slotId";

static const char* jkStateTriggerExtend           = "extend";

static const char* jkStateTriggerZeroth           = "zeroth";

static const char* jkStateTriggerTrueForType      = "trueForType";

static const char* jkStateTriggerTrueForCount     = "trueForCount";

static const char* jkStateTriggerTrueForDuration  = "trueForDuration";

static const char* jkStateTriggerExpr             = "expr";

    pJson, jkStateTriggerSlotId, pTrigger->slotId));

    pJson, jkStateTriggerExtend, pTrigger->extend));

      pJson, jkStateTriggerZeroth, (const char*)pTrigger->zeroth));

  }

      pJson, jkStateTriggerExpr, (const char*)pTrigger->expr));

  }

}

    tjsonGetSmallIntValue(pJson, jkStateTriggerSlotId, &pTrigger->slotId));

    tjsonGetSmallIntValue(pJson, jkStateTriggerExtend, &pTrigger->extend));

    pJson, jkStateTriggerZeroth, (char**)&pTrigger->zeroth));

    pJson, jkStateTriggerExpr, (char**)&pTrigger->expr));

}

static const char* jkSlidingTriggerIntervalUnit = "intervalUnit";

static const char* jkSlidingTriggerSlidingUnit  = "slidingUnit";

static const char* jkSlidingTriggerOffsetUnit   = "offsetUnit";

static const char* jkSlidingTriggerSoffsetUnit  = "soffsetUnit";

static const char* jkSlidingTriggerPrecision    = "precision";

static const char* jkSlidingTriggerInterval     = "interval";

static const char* jkSlidingTriggerOffset       = "offset";

static const char* jkSlidingTriggerSliding      = "sliding";

static const char* jkSlidingTriggerSoffset      = "soffset";

static const char* jkSlidingTriggerOverlap      = "overlap";

    pJson, jkSlidingTriggerIntervalUnit, pTrigger->intervalUnit));

    pJson, jkSlidingTriggerSlidingUnit, pTrigger->slidingUnit));

    pJson, jkSlidingTriggerOffsetUnit, pTrigger->offsetUnit));

    pJson, jkSlidingTriggerSoffsetUnit, pTrigger->soffsetUnit));

    pJson, jkSlidingTriggerPrecision, pTrigger->precision));

    pJson, jkSlidingTriggerInterval, pTrigger->interval));

    pJson, jkSlidingTriggerSliding, pTrigger->sliding));

    pJson, jkSlidingTriggerOffset, pTrigger->offset));

    pJson, jkSlidingTriggerSoffset, pTrigger->soffset));

    pJson, jkSlidingTriggerOverlap, pTrigger->overlap));

}

    pJson, jkSlidingTriggerIntervalUnit, &pTrigger->intervalUnit));

    pJson, jkSlidingTriggerSlidingUnit, &pTrigger->slidingUnit));

    pJson, jkSlidingTriggerOffsetUnit, &pTrigger->offsetUnit));

    pJson, jkSlidingTriggerSoffsetUnit, &pTrigger->soffsetUnit));

    pJson, jkSlidingTriggerPrecision, &pTrigger->precision));

    pJson, jkSlidingTriggerInterval, &pTrigger->interval));

    pJson, jkSlidingTriggerSliding, &pTrigger->sliding));

    pJson, jkSlidingTriggerOffset, &pTrigger->offset));

    pJson, jkSlidingTriggerSoffset, &pTrigger->soffset));

    pJson, jkSlidingTriggerOverlap, &pTrigger->overlap));

}

static const char* jkEventTriggerStartCond       = "startCond";

static const char* jkEventTriggerEndCond         = "endCond";

static const char* jkEventTriggerTrueForType     = "trueForType";

static const char* jkEventTriggerTrueForCount    = "trueForCount";

static const char* jkEventTriggerTrueForDuration = "trueForDuration";

      pJson, jkEventTriggerStartCond, (const char*)pTrigger->startCond));

  }

      pJson, jkEventTriggerEndCond, (const char*)pTrigger->endCond));

  }

}

    pJson, jkEventTriggerStartCond, (char**)&pTrigger->startCond));

    pJson, jkEventTriggerEndCond, (char**)&pTrigger->endCond));

}

static const char* jkCountTriggerCountVal = "countVal";

static const char* jkCountTriggerSliding  = "sliding";

static const char* jkCountTriggerCondCols = "condCols";

      pJson, jkCountTriggerCondCols, (const char*)pTrigger->condCols));

  }

}

    pJson, jkCountTriggerCountVal, &pTrigger->countVal));

    pJson, jkCountTriggerSliding, &pTrigger->sliding));

    pJson, jkCountTriggerCondCols, (char**)&pTrigger->condCols));

}

static const char* jkPeriodTriggerPeriodUnit = "periodUnit";

static const char* jkPeriodTriggerOffsetUnit = "offsetUnit";

static const char* jkPeriodTriggerPrecision  = "precision";

static const char* jkPeriodTriggerPeriod     = "period";

static const char* jkPeriodTriggerOffset     = "offset";

    pJson, jkPeriodTriggerPeriodUnit, pTrigger->periodUnit));

    pJson, jkPeriodTriggerOffsetUnit, pTrigger->offsetUnit));

    pJson, jkPeriodTriggerPrecision, pTrigger->precision));

    pJson, jkPeriodTriggerPeriod, pTrigger->period));

    pJson, jkPeriodTriggerOffset, pTrigger->offset));

}

    pJson, jkPeriodTriggerPeriodUnit, &pTrigger->periodUnit));

    pJson, jkPeriodTriggerOffsetUnit, &pTrigger->offsetUnit));

    pJson, jkPeriodTriggerPrecision, &pTrigger->precision));

    pJson, jkPeriodTriggerPeriod, &pTrigger->period));

    pJson, jkPeriodTriggerOffset, &pTrigger->offset));

}

}

}

}

}

static const char* jkSstreamCalcScanVgList        = "vgList";

static const char* jkSstreamCalcScanReadFromCache = "readFromCache";

static const char* jkSstreamCalcScanScanPlan      = "scanPlan";

    pJson, jkSstreamCalcScanVgList, int32ToJson,

    pPlan->vgList ? TARRAY_GET_ELEM(pPlan->vgList, 0) : NULL, sizeof(int32_t),

    pPlan->vgList ? pPlan->vgList->size : 0));

    pJson, jkSstreamCalcScanReadFromCache, pPlan->readFromCache));

      pJson, jkSstreamCalcScanScanPlan, (const char*)pPlan->scanPlan));

  }

}

    pJson, jkSstreamCalcScanVgList, jsonToInt32,

    &pPlan->vgList, sizeof(int32_t)));

    pJson, jkSstreamCalcScanReadFromCache, &pPlan->readFromCache));

    pJson, jkSstreamCalcScanScanPlan, (char**)&pPlan->scanPlan));

}

static const char* jkSDataTypeType      = "type";

static const char* jkSDataTypePrecision = "precision";

static const char* jkSDataTypeScale     = "scale";

static const char* jkSDataTypeBytes     = "bytes";

    pJson, jkSDataTypeType, pType->type));

    pJson, jkSDataTypePrecision, pType->precision));

    pJson, jkSDataTypeScale, pType->scale));

    pJson, jkSDataTypeBytes, pType->bytes));

}

    pJson, jkSDataTypeType, &pType->type));

    pJson, jkSDataTypePrecision, &pType->precision));

    pJson, jkSDataTypeScale, &pType->scale));

    pJson, jkSDataTypeBytes, &pType->bytes));

}

static const char* jkSStreamOutColExpr = "expr";

static const char* jkSStreamOutColType = "type";

      pJson, jkSStreamOutColExpr, (const char*)pCol->expr));

  }

    pJson, jkSStreamOutColType, sDataTypeToJson, &pCol->type));

}

    pJson, jkSStreamOutColExpr, (char**)&pCol->expr));

    pJson, jkSStreamOutColType, jsonToSDataType, &pCol->type));

}

}

}

static const char* jkCreateStreamReqName                 = "name";

static const char* jkCreateStreamReqStreamId             = "streamId";

static const char* jkCreateStreamReqSql                  = "sql";

static const char* jkCreateStreamReqStreamDB             = "streamDB";

static const char* jkCreateStreamReqTriggerDB            = "triggerDB";

static const char* jkCreateStreamReqOutDB                = "outDB";

static const char* jkCreateStreamReqCalcDB               = "calcDB";

static const char* jkCreateStreamReqTriggerTblName       = "triggerTblName";

static const char* jkCreateStreamReqOutTblName           = "outTblName";

static const char* jkCreateStreamReqIgExists             = "igExists";

static const char* jkCreateStreamReqTriggerType          = "triggerType";

static const char* jkCreateStreamReqIgDisorder           = "igDisorder";

static const char* jkCreateStreamReqDeleteReCalc         = "deleteReCalc";

static const char* jkCreateStreamReqDeleteOutTbl         = "deleteOutTbl";

static const char* jkCreateStreamReqFillHistory          = "fillHistory";

static const char* jkCreateStreamReqFillHistoryFirst     = "fillHistoryFirst";

static const char* jkCreateStreamReqCalcNotifyOnly       = "calcNotifyOnly";

static const char* jkCreateStreamReqLowLatencyCalc       = "lowLatencyCalc";

static const char* jkCreateStreamReqIgNoDataTrigger      = "igNoDataTrigger";

static const char* jkCreateStreamReqPNotifyAddrUrls      = "pNotifyAddrUrls";

static const char* jkCreateStreamReqNotifyEventTypes     = "notifyEventTypes";

static const char* jkCreateStreamReqAddOptions           = "addOptions";

static const char* jkCreateStreamReqNotifyHistory        = "notifyHistory";

static const char* jkCreateStreamReqTriggerFilterCols    = "triggerFilterCols";

static const char* jkCreateStreamReqTriggerCols          = "triggerCols";

static const char* jkCreateStreamReqPartitionCols        = "partitionCols";

static const char* jkCreateStreamReqOutCols              = "outCols";

static const char* jkCreateStreamReqOutTags              = "outTags";

static const char* jkCreateStreamReqMaxDelay             = "maxDelay";

static const char* jkCreateStreamReqFillHistoryStartTime = 

  "fillHistoryStartTime";

static const char* jkCreateStreamReqWatermark            = "watermark";

static const char* jkCreateStreamReqExpiredTime          = "expiredTime";

static const char* jkCreateStreamReqIdleTimeoutMs        = "idleTimeoutMs";

static const char* jkCreateStreamReqTrigger              = "trigger";

static const char* jkCreateStreamReqTriggerTblType       = "triggerTblType";

static const char* jkCreateStreamReqTriggerTblUid        = "triggerTblUid";

static const char* jkCreateStreamReqTriggerTblSuid       = "triggerTblSuid";

static const char* jkCreateStreamReqTriggerPrec          = "triggerPrec";

static const char* jkCreateStreamReqVtableCalc           = "vtableCalc";

static const char* jkCreateStreamReqOutTblType           = "outTblType";

static const char* jkCreateStreamReqOutStbExists         = "outStbExists";

static const char* jkCreateStreamReqOutStbUid            = "outStbUid";

static const char* jkCreateStreamReqOutStbSversion       = "outStbSversion";

static const char* jkCreateStreamReqEventTypes           = "eventTypes";

static const char* jkCreateStreamReqFlags                = "flags";

static const char* jkCreateStreamReqTsmaId               = "tsmaId";

static const char* jkCreateStreamReqPlaceHolderBitmap    = "placeHolderBitmap";

static const char* jkCreateStreamReqCalcTsSlotId         = "calcTsSlotId";

static const char* jkCreateStreamReqTriTsSlotId          = "triTsSlotId";

static const char* jkCreateStreamReqCalcPkSlotId         = "calcPkSlotId";

static const char* jkCreateStreamReqTriPkSlotId          = "triPkSlotId";

static const char* jkCreateStreamReqTriggerTblVgId       = "triggerTblVgId";

static const char* jkCreateStreamReqOutTblVgId           = "outTblVgId";

static const char* jkCreateStreamReqTriggerScanPlan      = "triggerScanPlan";

static const char* jkCreateStreamReqCalcScanPlanList     = "calcScanPlanList";

static const char* jkCreateStreamReqTriggerHasPF         = "triggerHasPF";

static const char* jkCreateStreamReqTriggerPrevFilter    = "triggerPrevFilter";

static const char* jkCreateStreamReqNumOfCalcSubplan     = "numOfCalcSubplan";

static const char* jkCreateStreamReqCalcPlan             = "calcPlan";

static const char* jkCreateStreamReqSubTblNameExpr       = "subTblNameExpr";

static const char* jkCreateStreamReqTagValueExpr         = "tagValueExpr";

static const char* jkCreateStreamReqForceOutCols         = "forceOutCols";

static const char* jkCreateStreamReqColCids = "colCids";

static const char* jkCreateStreamReqTagCids = "tagCids";

      pJson, jkCreateStreamReqName, pReq->name));

  }

    pJson, jkCreateStreamReqStreamId, pReq->streamId));

      pJson, jkCreateStreamReqSql, pReq->sql));

  }

      pJson, jkCreateStreamReqStreamDB, pReq->streamDB));

  }

      pJson, jkCreateStreamReqTriggerDB, pReq->triggerDB));

  }

      pJson, jkCreateStreamReqOutDB, pReq->outDB));

  }

    pJson, jkCreateStreamReqCalcDB, stringToJson,

    pReq->calcDB ? TARRAY_GET_ELEM(pReq->calcDB, 0) : NULL,

    pReq->calcDB ? pReq->calcDB->elemSize : 0,

    pReq->calcDB ? pReq->calcDB->size : 0));

      pJson, jkCreateStreamReqTriggerTblName, pReq->triggerTblName));

  }

      pJson, jkCreateStreamReqOutTblName, pReq->outTblName));

  }

  // trigger contol part

    pJson, jkCreateStreamReqIgExists, pReq->igExists));

    pJson, jkCreateStreamReqTriggerType, pReq->triggerType));

    pJson, jkCreateStreamReqIgDisorder, pReq->igDisorder));

    pJson, jkCreateStreamReqDeleteReCalc, pReq->deleteReCalc));

    pJson, jkCreateStreamReqDeleteOutTbl, pReq->deleteOutTbl));

    pJson, jkCreateStreamReqFillHistory, pReq->fillHistory));

    pJson, jkCreateStreamReqFillHistoryFirst, pReq->fillHistoryFirst));

    pJson, jkCreateStreamReqCalcNotifyOnly, pReq->calcNotifyOnly));

    pJson, jkCreateStreamReqLowLatencyCalc, pReq->lowLatencyCalc));

    pJson, jkCreateStreamReqIgNoDataTrigger, pReq->igNoDataTrigger));

  // notify part

    pJson, jkCreateStreamReqPNotifyAddrUrls, stringToJson,

    pReq->pNotifyAddrUrls ? TARRAY_GET_ELEM(pReq->pNotifyAddrUrls, 0) : NULL,

    pReq->pNotifyAddrUrls ? pReq->pNotifyAddrUrls->elemSize : 0,

    pReq->pNotifyAddrUrls ? pReq->pNotifyAddrUrls->size : 0));

    pJson, jkCreateStreamReqNotifyEventTypes, pReq->notifyEventTypes));

    pJson, jkCreateStreamReqAddOptions, pReq->addOptions));

    pJson, jkCreateStreamReqNotifyHistory, pReq->notifyHistory));

  // out table part

  // trigger cols and partition cols

      pJson, jkCreateStreamReqTriggerFilterCols,

      (const char*)pReq->triggerFilterCols));

  }

      pJson, jkCreateStreamReqTriggerCols, (const char*)pReq->triggerCols));

  }

      pJson, jkCreateStreamReqPartitionCols, (const char*)pReq->partitionCols));

  }

  // out cols

    pJson, jkCreateStreamReqOutCols, sfieldWithOptionsToJson,

    pReq->outCols ? TARRAY_GET_ELEM(pReq->outCols, 0) : NULL,

    pReq->outCols ? pReq->outCols->elemSize : 0,

    pReq->outCols ? pReq->outCols->size : 0));

  // out tags

    pJson, jkCreateStreamReqOutTags, stagFieldWithOptionsToJson,

    pReq->outTags ? TARRAY_GET_ELEM(pReq->outTags, 0) : NULL,

    pReq->outTags ? pReq->outTags->elemSize : 0,

    pReq->outTags ? pReq->outTags->size : 0));

    pJson, jkCreateStreamReqMaxDelay, pReq->maxDelay));

    pJson, jkCreateStreamReqFillHistoryStartTime, pReq->fillHistoryStartTime));

    pJson, jkCreateStreamReqWatermark, pReq->watermark));

    pJson, jkCreateStreamReqExpiredTime, pReq->expiredTime));

    pJson, jkCreateStreamReqIdleTimeoutMs, pReq->idleTimeoutMs));

  // trigger

        sessionTriggerToJson, &pReq->trigger));

        stateTriggerToJson, &pReq->trigger));

        slidingTriggerToJson, &pReq->trigger));

        eventTriggerToJson, &pReq->trigger));

        countTriggerToJson, &pReq->trigger));

        periodTriggerToJson, &pReq->trigger));

  }

    pJson, jkCreateStreamReqTriggerTblType, pReq->triggerTblType));

    pJson, jkCreateStreamReqTriggerTblUid, pReq->triggerTblUid));

    pJson, jkCreateStreamReqTriggerTblSuid, pReq->triggerTblSuid));

    pJson, jkCreateStreamReqTriggerPrec, pReq->triggerPrec));

    pJson, jkCreateStreamReqVtableCalc, pReq->vtableCalc));

    pJson, jkCreateStreamReqOutTblType, pReq->outTblType));

    pJson, jkCreateStreamReqOutStbExists, pReq->outStbExists));

    pJson, jkCreateStreamReqOutStbUid, pReq->outStbUid));

    pJson, jkCreateStreamReqOutStbSversion, pReq->outStbSversion));

    pJson, jkCreateStreamReqEventTypes, pReq->eventTypes));

    pJson, jkCreateStreamReqFlags, pReq->flags));

    pJson, jkCreateStreamReqTsmaId, pReq->tsmaId));

    pJson, jkCreateStreamReqPlaceHolderBitmap, pReq->placeHolderBitmap));

    pJson, jkCreateStreamReqCalcTsSlotId, pReq->calcTsSlotId));

    pJson, jkCreateStreamReqTriTsSlotId, pReq->triTsSlotId));

    pJson, jkCreateStreamReqCalcPkSlotId, pReq->calcPkSlotId));

    pJson, jkCreateStreamReqTriPkSlotId, pReq->triPkSlotId));

    pJson, jkCreateStreamReqTriggerTblVgId, pReq->triggerTblVgId));

    pJson, jkCreateStreamReqOutTblVgId, pReq->outTblVgId));

      pJson, jkCreateStreamReqTriggerScanPlan, (const char*)pReq->triggerScanPlan));

  }

    pJson, jkCreateStreamReqCalcScanPlanList, calcScanPlanToJson,

    pReq->calcScanPlanList ? TARRAY_GET_ELEM(pReq->calcScanPlanList, 0) : NULL,

    pReq->calcScanPlanList ? pReq->calcScanPlanList->elemSize : 0,

    pReq->calcScanPlanList ? pReq->calcScanPlanList->size : 0));

    pJson, jkCreateStreamReqTriggerHasPF, pReq->triggerHasPF));

      pJson, jkCreateStreamReqTriggerPrevFilter,

      (const char*)pReq->triggerPrevFilter));

  }

    pJson, jkCreateStreamReqNumOfCalcSubplan, pReq->numOfCalcSubplan));

      pJson, jkCreateStreamReqCalcPlan, (const char*)pReq->calcPlan));

  }

      jkCreateStreamReqSubTblNameExpr, (const char*)pReq->subTblNameExpr));

  }

      pJson, jkCreateStreamReqTagValueExpr, (const char*)pReq->tagValueExpr));

  }

    pJson, jkCreateStreamReqForceOutCols, sStreamOutColToJson,

    pReq->forceOutCols ? TARRAY_GET_ELEM(pReq->forceOutCols, 0) : NULL,

    pReq->forceOutCols ? pReq->forceOutCols->elemSize : 0,

    pReq->forceOutCols ? pReq->forceOutCols->size : 0));

      pJson, jkCreateStreamReqColCids, int16ToJson,

      pReq->colCids ? TARRAY_GET_ELEM(pReq->colCids, 0) : NULL,

      pReq->colCids ? pReq->colCids->elemSize : 0,

      pReq->colCids ? pReq->colCids->size : 0));

      pJson, jkCreateStreamReqTagCids, int16ToJson,

      pReq->tagCids ? TARRAY_GET_ELEM(pReq->tagCids, 0) : NULL,

      pReq->tagCids ? pReq->tagCids->elemSize : 0,

      pReq->tagCids ? pReq->tagCids->size : 0));

}

  const SCMCreateStreamReq* pReq, bool format, char** ppStr, int32_t* pStrLen) {

    pStrLen, code, lino, _end, TSDB_CODE_MND_STREAM_INTERNAL_ERROR);

    } else {

    }

  }

      "failed to convert SCMCreateStreamReq to json, lino: %d, since %s",

      lino, tstrerror(code));

  }

}

    pJson, jkCreateStreamReqName, (char**)&pReq->name));

    pJson, jkCreateStreamReqStreamId, &pReq->streamId));

    pJson, jkCreateStreamReqSql, (char**)&pReq->sql));

    pJson, jkCreateStreamReqStreamDB, (char**)&pReq->streamDB));

    pJson, jkCreateStreamReqTriggerDB, (char**)&pReq->triggerDB));

    pJson, jkCreateStreamReqOutDB, (char**)&pReq->outDB));

    pJson, jkCreateStreamReqCalcDB, jsonToString,

    &pReq->calcDB, POINTER_BYTES));

    pJson, jkCreateStreamReqTriggerTblName, (char**)&pReq->triggerTblName));

    pJson, jkCreateStreamReqOutTblName, (char**)&pReq->outTblName));

  // trigger control part

    pJson, jkCreateStreamReqIgExists, &pReq->igExists));

    pJson, jkCreateStreamReqTriggerType, &pReq->triggerType));

    pJson, jkCreateStreamReqIgDisorder, &pReq->igDisorder));

    pJson, jkCreateStreamReqDeleteReCalc, &pReq->deleteReCalc));

    pJson, jkCreateStreamReqDeleteOutTbl, &pReq->deleteOutTbl));

    pJson, jkCreateStreamReqFillHistory, &pReq->fillHistory));

    pJson, jkCreateStreamReqFillHistoryFirst, &pReq->fillHistoryFirst));

    pJson, jkCreateStreamReqCalcNotifyOnly, &pReq->calcNotifyOnly));

    pJson, jkCreateStreamReqLowLatencyCalc, &pReq->lowLatencyCalc));

    pJson, jkCreateStreamReqIgNoDataTrigger, &pReq->igNoDataTrigger));

  // notify part

    pJson, jkCreateStreamReqPNotifyAddrUrls, jsonToString,

    &pReq->pNotifyAddrUrls, POINTER_BYTES));

    pJson, jkCreateStreamReqNotifyEventTypes, &pReq->notifyEventTypes));

    pJson, jkCreateStreamReqAddOptions, &pReq->addOptions));

    pJson, jkCreateStreamReqNotifyHistory, &pReq->notifyHistory));

    pJson, jkCreateStreamReqTriggerFilterCols,

    (char**)&pReq->triggerFilterCols));

    pJson, jkCreateStreamReqTriggerCols, (char**)&pReq->triggerCols));

    pJson, jkCreateStreamReqPartitionCols, (char**)&pReq->partitionCols));

  // out cols

    pJson, jkCreateStreamReqOutCols, jsonToSFieldWithOptions,

    &pReq->outCols, sizeof(SFieldWithOptions)));

  // out tags

    pJson, jkCreateStreamReqOutTags, jsonToSTagFieldWithOptions,

    &pReq->outTags, sizeof(SFieldWithOptions)));

    pJson, jkCreateStreamReqMaxDelay, &pReq->maxDelay));

    pJson, jkCreateStreamReqFillHistoryStartTime, &pReq->fillHistoryStartTime));

    pJson, jkCreateStreamReqWatermark, &pReq->watermark));

    pJson, jkCreateStreamReqExpiredTime, &pReq->expiredTime));

    pJson, jkCreateStreamReqIdleTimeoutMs, &pReq->idleTimeoutMs));

  // trigger

        pJson, jkCreateStreamReqTrigger, jsonToSessionTrigger, &pReq->trigger));

        pJson, jkCreateStreamReqTrigger, jsonToStateTrigger, &pReq->trigger));

        pJson, jkCreateStreamReqTrigger, jsonToSlidingTrigger, &pReq->trigger));

        pJson, jkCreateStreamReqTrigger, jsonToEventTrigger, &pReq->trigger));

        pJson, jkCreateStreamReqTrigger, jsonToCountTrigger, &pReq->trigger));

        pJson, jkCreateStreamReqTrigger, jsonToPeriodTrigger, &pReq->trigger));

  }

    pJson, jkCreateStreamReqTriggerTblType, &pReq->triggerTblType));

    pJson, jkCreateStreamReqTriggerTblUid, &pReq->triggerTblUid));

    pJson, jkCreateStreamReqTriggerTblSuid, &pReq->triggerTblSuid));

    pJson, jkCreateStreamReqTriggerPrec, &pReq->triggerPrec));

    pJson, jkCreateStreamReqVtableCalc, &pReq->vtableCalc));

    pJson, jkCreateStreamReqOutTblType, &pReq->outTblType));

    pJson, jkCreateStreamReqOutStbExists, &pReq->outStbExists));

    pJson, jkCreateStreamReqOutStbUid, &pReq->outStbUid));

    pJson, jkCreateStreamReqOutStbSversion, &pReq->outStbSversion));

    pJson, jkCreateStreamReqEventTypes, &pReq->eventTypes));

    pJson, jkCreateStreamReqFlags, &pReq->flags));

    pJson, jkCreateStreamReqTsmaId, &pReq->tsmaId));

    pJson, jkCreateStreamReqPlaceHolderBitmap, &pReq->placeHolderBitmap));

    pJson, jkCreateStreamReqCalcTsSlotId, &pReq->calcTsSlotId));

    pJson, jkCreateStreamReqTriTsSlotId, &pReq->triTsSlotId));

    pJson, jkCreateStreamReqCalcPkSlotId, &pReq->calcPkSlotId));

    pJson, jkCreateStreamReqTriPkSlotId, &pReq->triPkSlotId));

    pJson, jkCreateStreamReqTriggerTblVgId, &pReq->triggerTblVgId));

    pJson, jkCreateStreamReqOutTblVgId, &pReq->outTblVgId));

    pJson, jkCreateStreamReqTriggerScanPlan, (char**)&pReq->triggerScanPlan));