tscSecondaryMerge.h

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#ifndef TDENGINE_TSCSECONARYMERGE_H
#define TDENGINE_TSCSECONARYMERGE_H

#ifdef __cplusplus
extern "C" {
#endif

#include "taosmsg.h"
#include "textbuffer.h"
#include "tinterpolation.h"
#include "tlosertree.h"
#include "tsclient.h"

#define MAX_NUM_OF_SUBQUERY_RETRY 3

/*
 * @version 0.1
 * @date   2018/01/05
 * @author liaohj
 * management of client-side reducer for metric query
 */

struct SQLFunctionCtx;

typedef struct SLocalDataSource {
  tExtMemBuffer *pMemBuffer;
  int32_t        flushoutIdx;
  int32_t        pageId;
  int32_t        rowIdx;
  tFilePage      filePage;
} SLocalDataSource;

enum {
  TSC_LOCALREDUCE_READY = 0x0,
  TSC_LOCALREDUCE_IN_PROGRESS = 0x1,
  TSC_LOCALREDUCE_TOBE_FREED = 0x2,
};

typedef struct SLocalReducer {
  SLocalDataSource **    pLocalDataSrc;
  int32_t                numOfBuffer;
  int32_t                numOfCompleted;
  int32_t                numOfVnode;
  SLoserTreeInfo *       pLoserTree;
  char *                 prevRowOfInput;
  tFilePage *            pResultBuf;
  int32_t                nResultBufSize;
  char *                 pBufForInterpo;  // intermediate buffer for interpolation
  tFilePage *            pTempBuffer;
  struct SQLFunctionCtx *pCtx;
  int32_t                rowSize;     // size of each intermediate result.
  int32_t                status;      // denote it is in reduce process, in reduce process, it
  bool                   hasPrevRow;  // cannot be released
  bool                   hasUnprocessedRow;
  tOrderDescriptor *     pDesc;
  SColumnModel *            resColModel;
  tExtMemBuffer **       pExtMemBuffer;      // disk-based buffer
  SInterpolationInfo     interpolationInfo;  // interpolation support structure
  char *                 pFinalRes;          // result data after interpo
  tFilePage *            discardData;
  SResultInfo *          pResInfo;
  bool                   discard;
  int32_t                offset;             // limit offset value
} SLocalReducer;

typedef struct SSubqueryState {
  /*
   * the number of completed retrieval subquery, once this value equals to numOfVnodes,
   * all retrieval are completed.Local merge is launched.
   */
  int32_t  numOfCompleted;
  int32_t  numOfTotal;          // number of total sub-queries
  int32_t  code;                // code from subqueries
  uint64_t numOfRetrievedRows;  // total number of points in this query
} SSubqueryState;

typedef struct SRetrieveSupport {
  tExtMemBuffer **  pExtMemBuffer;     // for build loser tree
  tOrderDescriptor *pOrderDescriptor;
  SColumnModel *    pFinalColModel;    // colModel for final result
  SSubqueryState *  pState;
  int32_t           subqueryIndex;     // index of current vnode in vnode list
  SSqlObj *         pParentSqlObj;
  tFilePage *       localBuffer;       // temp buffer, there is a buffer for each vnode to
  uint32_t          numOfRetry;        // record the number of retry times
  pthread_mutex_t   queryMutex;
} SRetrieveSupport;

int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOrderDescriptor **pDesc,
                                 SColumnModel **pFinalModel, uint32_t nBufferSize);

void tscLocalReducerEnvDestroy(tExtMemBuffer **pMemBuffer, tOrderDescriptor *pDesc, SColumnModel *pFinalModel,
                               int32_t numOfVnodes);

int32_t saveToBuffer(tExtMemBuffer *pMemoryBuf, tOrderDescriptor *pDesc, tFilePage *pPage, void *data,
                     int32_t numOfRows, int32_t orderType);

int32_t tscFlushTmpBuffer(tExtMemBuffer *pMemoryBuf, tOrderDescriptor *pDesc, tFilePage *pPage, int32_t orderType);

/*
 * create local reducer to launch the second-stage reduce process at client site
 */
void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrderDescriptor *pDesc,
                           SColumnModel *finalModel, SSqlCmd *pSqlCmd, SSqlRes *pRes);

void tscDestroyLocalReducer(SSqlObj *pSql);

int32_t tscDoLocalreduce(SSqlObj *pSql);

#ifdef __cplusplus
}
#endif

#endif  // TDENGINE_TSCSECONARYMERGE_H