mavg.py

###################################################################
#           Copyright (c) 2016 by TAOS Technologies, Inc.
#                     All rights reserved.
#
#  This file is proprietary and confidential to TAOS Technologies.
#  No part of this file may be reproduced, stored, transmitted,
#  disclosed or used in any form or by any means other than as
#  expressly provided by the written permission from Jianhui Tao
#
###################################################################

# -*- coding: utf-8 -*-

import sys
import subprocess
import random
import math
import numpy as np
import inspect
import re
import taos

from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *


class TDTestCase:
    def init(self, conn, logSql):
        tdLog.debug("start to execute %s" % __file__)
        tdSql.init(conn.cursor())

    def mavg_query_form(self, sel="select", func="mavg(", col="c1", m_comm =",", k=1,r_comm=")", alias="", fr="from",table_expr="t1", condition=""):
        '''
        mavg function:

        :param sel:         string, must be "select", required parameters;
        :param func:        string, in this case must be "mavg(", otherwise return other function, required parameters;
        :param col:         string, column name, required parameters;
        :param m_comm:      string, comma between col and k , required parameters;
        :param k:           int/float，the width of the  sliding window, [1,100], required parameters;
        :param r_comm:      string, must be ")", use with "(" in func, required parameters;
        :param alias:       string, result column another name，or add other funtion;
        :param fr:          string, must be "from", required parameters;
        :param table_expr:  string or expression, data source（eg,table/stable name, result set）, required parameters;
        :param condition:   expression；
        :return:            mavg query statement,default: select mavg(c1, 1) from t1
        '''

        return f"{sel} {func} {col} {m_comm} {k} {r_comm} {alias} {fr} {table_expr} {condition}"

    def checkmavg(self,sel="select", func="mavg(", col="c1", m_comm =",", k=1,r_comm=")", alias="", fr="from",table_expr="t1", condition=""):
        # print(self.mavg_query_form(sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
        #                            table_expr=table_expr, condition=condition))
        line = sys._getframe().f_back.f_lineno

        if not all([sel , func , col , m_comm , k , r_comm , fr , table_expr]):
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))

        sql = "select * from t1"
        collist =  tdSql.getColNameList(sql)

        if not isinstance(col, str):
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))

        if len([x for x in col.split(",") if x.strip()]) != 1:
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))

        col = col.replace(",", "").replace(" ", "")

        if any([re.compile('^[a-zA-Z]{1}.*$').match(col) is None , not col.replace(".","").isalnum()]):
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))

        # if all(["," in col , len(col.split(",")) != 2]):
        #     print(f"case in {line}: ", end='')
        #     return tdSql.error(self.mavg_query_form(
        #         sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
        #         table_expr=table_expr, condition=condition
        #     ))
        #
        # if ("," in col):
        #     if (not col.split(",")[0].strip()) ^ (not col.split(",")[1].strip()):
        #         col = col.strip().split(",")[0] if not col.split(",")[1].strip() else col.strip().split(",")[1]
        #     else:
        #         print(f"case in {line}: ", end='')
        #         return tdSql.error(self.mavg_query_form(
        #             sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
        #             table_expr=table_expr, condition=condition
        #         ))
        #     pass

        if '.' in col:
            if any([col.split(".")[0] not in table_expr, col.split(".")[1] not in collist]):
                print(f"case in {line}: ", end='')
                return tdSql.error(self.mavg_query_form(
                    sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                    table_expr=table_expr, condition=condition
                ))
            pass

        if "." not in col:
            if col not in  collist:
                print(f"case in {line}: ", end='')
                return tdSql.error(self.mavg_query_form(
                    sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                    table_expr=table_expr, condition=condition
                ))
            pass

        colname = col if "." not in col else col.split(".")[1]
        col_index = collist.index(colname)
        if any([tdSql.cursor.istype(col_index, "TIMESTAMP"), tdSql.cursor.istype(col_index, "BOOL")]):
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))

        if  any([tdSql.cursor.istype(col_index, "BINARY") , tdSql.cursor.istype(col_index,"NCHAR")]):
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))

        if any( [func != "mavg(" , r_comm != ")" , fr != "from",  sel != "select"]):
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))

        if all(["(" not in table_expr, "stb" in table_expr, "group" not in condition.lower()]):
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))

        if "order by tbname" in condition.lower():
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))                

        if all(["group" in condition.lower(), "tbname" not in condition.lower()]):
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))

        alias_list = ["tbname", "_c0", "st", "ts"]
        if all([alias, "," not in alias, not alias.isalnum()]):
            # actually， column alias also support "_"， but in this case，forbidden that。
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))

        if all([alias, "," in alias]):
            if  all(parm != alias.lower().split(",")[1].strip() for parm in alias_list):
                print(f"case in {line}: ", end='')
                return tdSql.error(self.mavg_query_form(
                    sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                    table_expr=table_expr, condition=condition
                ))
            pass

        condition_exception = [ "~", "^", "insert", "distinct",
                           "count", "avg", "twa", "irate", "sum", "stddev", "leastquares",
                           "min", "max", "first", "last", "top", "bottom", "percentile",
                           "apercentile", "last_row", "interp", "diff", "derivative",
                           "spread", "ceil", "floor", "round", "interval", "fill", "slimit", "soffset"]
        if "union" not in condition.lower():
            if any(parm in condition.lower().strip() for parm in condition_exception):

                print(f"case in {line}: ", end='')
                return tdSql.error(self.mavg_query_form(
                    sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                    table_expr=table_expr, condition=condition
                ))
            pass

        if not any([isinstance(k, int) ,  isinstance(k, float)])  :
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                col=col, k=k, alias=alias, table_expr=table_expr, condition=condition
            ))

        if not(1 <= k < 1001):
            print(f"case in {line}: ", end='')
            return tdSql.error(self.mavg_query_form(
                col=col, k=k, alias=alias, table_expr=table_expr, condition=condition
            ))

        k = int(k // 1)
        pre_sql = re.sub("mavg\([a-z0-9 .,]*\)", f"count({col})", self.mavg_query_form(
             col=col, table_expr=table_expr, condition=condition
        ))
        tdSql.query(pre_sql)

        if tdSql.queryRows == 0:
            tdSql.query(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))
            print(f"case in {line}: ", end='')
            tdSql.checkRows(0)
            return

        if "group" in condition:
            tb_condition = condition.split("group by")[1].split(" ")[1]
            tdSql.query(f"select distinct {tb_condition} from {table_expr}")
            query_result = tdSql.queryResult
            query_rows = tdSql.queryRows
            clear_condition = re.sub('order by [0-9a-z]*|slimit [0-9]*|soffset [0-9]*', "", condition)

            pre_row = 0
            for i in range(query_rows):
                group_name = query_result[i][0]
                if "where" in clear_condition:
                    pre_condition = re.sub('group by [0-9a-z]*', f"{tb_condition}='{group_name}'", clear_condition)
                else:
                    pre_condition = "where " + re.sub('group by [0-9a-z]*',f"{tb_condition}='{group_name}'", clear_condition)

                tdSql.query(f"select {col} {alias} from {table_expr} {pre_condition}")
                pre_data = np.array(tdSql.queryResult)[np.array(tdSql.queryResult) != None]
                if (platform.system().lower() == 'windows' and pre_data.dtype == 'int32'):
                    pre_data = np.array(pre_data, dtype = 'int64')
                pre_mavg = np.convolve(pre_data, np.ones(k), "valid")/k
                tdSql.query(self.mavg_query_form(
                    sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                    table_expr=table_expr, condition=condition
                ))
                for j in range(len(pre_mavg)):
                    print(f"case in {line}:", end='')
                    tdSql.checkData(pre_row+j, 0, pre_mavg[j])
                pre_row += len(pre_mavg)
            return
        elif "union" in condition:
            union_sql_0 = self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ).split("union all")[0]

            union_sql_1 = self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ).split("union all")[1]

            tdSql.query(union_sql_0)
            union_mavg_0 = tdSql.queryResult
            row_union_0 = tdSql.queryRows

            tdSql.query(union_sql_1)
            union_mavg_1 = tdSql.queryResult

            tdSql.query(self.mavg_query_form(
                sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                table_expr=table_expr, condition=condition
            ))
            for i in range(tdSql.queryRows):
                print(f"case in {line}: ", end='')
                if i < row_union_0:
                    tdSql.checkData(i, 0, union_mavg_0[i][0])
                else:
                    tdSql.checkData(i, 0, union_mavg_1[i-row_union_0][0])
            return

        else:
            tdSql.query(f"select {col} from {table_expr} {re.sub('limit [0-9]*|offset [0-9]*','',condition)}")
            offset_val = condition.split("offset")[1].split(" ")[1] if "offset" in condition else 0
            # print(f"select {col} from {table_expr} {re.sub('limit [0-9]*|offset [0-9]*','',condition)}")
            if not tdSql.queryResult:
                pre_result = np.array(tdSql.queryResult)[np.array(tdSql.queryResult) != None]
                if (platform.system().lower() == 'windows' and pre_result.dtype == 'int32'):
                    pre_result = np.array(pre_result, dtype = 'int64')
            
                pre_mavg = pre_mavg = np.convolve(pre_result, np.ones(k), "valid")[offset_val:]/k
                tdSql.query(self.mavg_query_form(
                    sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr,
                    table_expr=table_expr, condition=condition
                ))
                for i in range(tdSql.queryRows):
                    print(f"case in {line}: ", end='')
                    tdSql.checkData(i, 0, pre_mavg[i])

        pass

    def mavg_current_query(self) :

        # table schema :ts timestamp, c1 int, c2 float, c3 timestamp, c4 binary(16), c5 double, c6 bool
        #                 c7 bigint, c8 smallint, c9 tinyint, c10 nchar(16)

        # case1～6： numeric col:int/bigint/tinyint/smallint/float/double
        self.checkmavg()
        case2 =  {"col": "c2"}
        self.checkmavg(**case2)
        case3 =  {"col": "c5"}
        self.checkmavg(**case3)
        case4 =  {"col": "c7"}
        self.checkmavg(**case4)
        case5 =  {"col": "c8"}
        self.checkmavg(**case5)
        case6 =  {"col": "c9"}
        self.checkmavg(**case6)

        # # case7~8: nested query
        # case7 = {"table_expr": "(select c1 from stb1)"}
        # self.checkmavg(**case7)
        # case8 = {"table_expr": "(select mavg(c1, 1) c1 from stb1 group by tbname)"}
        # self.checkmavg(**case8)

        # case9~10: mix with tbname/ts/tag/col
        # case9 = {"alias": ", tbname"}
        # self.checkmavg(**case9)
        # case10 = {"alias": ", _c0"}
        # self.checkmavg(**case10)
        # case11 = {"alias": ", st1"}
        # self.checkmavg(**case11)
        # case12 = {"alias": ", c1"}
        # self.checkmavg(**case12)

        # case13~15: with  single condition
        case13 = {"condition": "where c1 <= 10"}
        self.checkmavg(**case13)
        case14 = {"condition": "where c6 in (0, 1)"}
        self.checkmavg(**case14)
        case15 = {"condition": "where c1 between 1 and 10"}
        self.checkmavg(**case15)

        # case16:  with multi-condition
        case16 = {"condition": "where c6=1 or c6 =0"}
        self.checkmavg(**case16)

        # case17: only support normal table join
        case17 = {
            "col": "t1.c1",
            "table_expr": "t1, t2",
            "condition": "where t1.ts=t2.ts"
        }
        self.checkmavg(**case17)
        # # case18~19: with group by
        # case19 = {
        #     "table_expr": "stb1",
        #     "condition": "partition by tbname"
        # }
        # self.checkmavg(**case19)

        # case20~21: with order by
        # case20 = {"condition": "order by ts"}
        # self.checkmavg(**case20)
        #case21 = {
        #    "table_expr": "stb1",
        #    "condition": "group by tbname order by tbname"
        #}
        #self.checkmavg(**case21)

        # # case22: with union
        # case22 = {
        #     "condition": "union all select mavg( c1 , 1 ) from t2"
        # }
        # self.checkmavg(**case22)

        # case23: with limit/slimit
        case23 = {
            "condition": "limit 1"
        }
        self.checkmavg(**case23)

        # case24: value k range[1, 100], can be int or float, k = floor(k)
        case24 = {"k": 3}
        self.checkmavg(**case24)
        case25 = {"k": 2.999}
        self.checkmavg(**case25)
        case26 = {"k": 1000}
        self.checkmavg(**case26)

        pass

    def mavg_error_query(self) -> None :
        # unusual test

        # form test
        err1 =  {"col": ""}
        self.checkmavg(**err1)          # no col
        err2 = {"sel": ""}
        self.checkmavg(**err2)          # no select
        err3 = {"func": "mavg", "col": "", "m_comm": "", "k": "", "r_comm": ""}
        self.checkmavg(**err3)          # no mavg condition: select mavg from
        err4 = {"col": "", "m_comm": "", "k": ""}
        self.checkmavg(**err4)          # no mavg condition: select mavg() from
        err5 = {"func": "mavg", "r_comm": ""}
        self.checkmavg(**err5)          # no brackets: select mavg col, k from
        err6 = {"fr": ""}
        self.checkmavg(**err6)          # no from
        err7 = {"k": ""}
        self.checkmavg(**err7)          # no k
        err8 = {"table_expr": ""}
        self.checkmavg(**err8)          # no table_expr

        # err9 = {"col": "st1"}
        # self.checkmavg(**err9)          # col: tag
        # err10 = {"col": 1}
        # self.checkmavg(**err10)         # col: value
        err11 = {"col": "NULL"}
        self.checkmavg(**err11)         # col: NULL
        err12 = {"col": "%_"}
        self.checkmavg(**err12)         # col: %_
        err13 = {"col": "c3"}
        self.checkmavg(**err13)         # col: timestamp col
        err14 = {"col": "_c0"}
        self.checkmavg(**err14)         # col: Primary key
        err15 = {"col": "avg(c1)"}
        self.checkmavg(**err15)         # expr col
        err16 = {"col": "c4"}
        self.checkmavg(**err16)         # binary col
        err17 = {"col": "c10"}
        self.checkmavg(**err17)         # nchar col
        err18 = {"col": "c6"}
        self.checkmavg(**err18)         # bool col
        err19 = {"col": "'c1'"}
        self.checkmavg(**err19)         # col: string
        err20 = {"col": None}
        self.checkmavg(**err20)         # col: None
        err21 = {"col": "''"}
        self.checkmavg(**err21)         # col: ''
        err22 = {"col": "tt1.c1"}
        self.checkmavg(**err22)         # not table_expr col
        err23 = {"col": "t1"}
        self.checkmavg(**err23)         # tbname
        err24 = {"col": "stb1"}
        self.checkmavg(**err24)         # stbname
        err25 = {"col": "db"}
        self.checkmavg(**err25)         # datbasename
        err26 = {"col": "True"}
        self.checkmavg(**err26)         # col: BOOL 1
        err27 = {"col": True}
        self.checkmavg(**err27)         # col: BOOL 2
        err28 = {"col": "*"}
        self.checkmavg(**err28)         # col: all col
        err29 = {"func": "mavg[", "r_comm": "]"}
        self.checkmavg(**err29)         # form: mavg[col, k]
        err30 = {"func": "mavg{", "r_comm": "}"}
        self.checkmavg(**err30)         # form: mavg{col, k}
        err31 = {"col": "[c1]"}
        self.checkmavg(**err31)         # form: mavg([col], k)
        err32 = {"col": "c1, c2"}
        self.checkmavg(**err32)         # form: mavg(col, col2, k)
        err33 = {"col": "c1, 2"}
        self.checkmavg(**err33)         # form: mavg(col, k1, k2)
        err34 = {"alias": ", count(c1)"}
        self.checkmavg(**err34)         # mix with aggregate function 1
        err35 = {"alias": ", avg(c1)"}
        self.checkmavg(**err35)         # mix with aggregate function 2
        err36 = {"alias": ", min(c1)"}
        self.checkmavg(**err36)         # mix with select function 1
        err37 = {"alias": ", top(c1, 5)"}
        self.checkmavg(**err37)         # mix with select function 2
        err38 = {"alias": ", spread(c1)"}
        self.checkmavg(**err38)         # mix with calculation function  1
        err39 = {"alias": ", diff(c1)"}
        self.checkmavg(**err39)         # mix with calculation function  2
        # err40 = {"alias": "+ 2"}
        # self.checkmavg(**err40)         # mix with arithmetic 1
        #tdSql.query(" select mavg( c1 , 1 ) + 2 from t1 ")
        err41 = {"alias": "+ avg(c1)"}
        self.checkmavg(**err41)         # mix with arithmetic 2
        err42 = {"alias": ", c1"}
        self.checkmavg(**err42)         # mix with other col
        # err43 = {"table_expr": "stb1"}
        # self.checkmavg(**err43)         # select stb directly
        err44 = {
            "col": "stb1.c1",
            "table_expr": "stb1, stb2",
            "condition": "where stb1.ts=stb2.ts and stb1.st1=stb2.st2 order by stb1.ts"
        }
        self.checkmavg(**err44)         # stb join
        err45 = {
            "condition": "where ts>0 and ts < now interval(1h) fill(next)"
        }
        self.checkmavg(**err45)         # interval
        err46 = {
            "table_expr": "t1",
            "condition": "group by c6"
        }
        self.checkmavg(**err46)         # group by normal col
        err47 = {
            "table_expr": "stb1",
            "condition": "group by tbname slimit 1 "
        }
        # self.checkmavg(**err47)         # with slimit
        err48 = {
            "table_expr": "stb1",
            "condition": "group by tbname slimit 1 soffset 1"
        }
        # self.checkmavg(**err48)         # with soffset
        err49 = {"k": "2021-01-01 00:00:00.000"}
        self.checkmavg(**err49)         # k: timestamp
        err50 = {"k": False}
        self.checkmavg(**err50)         # k: False
        err51 = {"k": "%"}
        self.checkmavg(**err51)         # k: special char
        err52 = {"k": ""}
        self.checkmavg(**err52)         # k: ""
        err53 = {"k": None}
        self.checkmavg(**err53)         # k: None
        err54 = {"k": "NULL"}
        self.checkmavg(**err54)         # k: null
        err55 = {"k": "binary(4)"}
        self.checkmavg(**err55)         # k: string
        err56 = {"k": "c1"}
        self.checkmavg(**err56)         # k: sring，col name
        err57 = {"col": "c1, 1, c2"}
        self.checkmavg(**err57)         # form: mavg(col1, k1, col2, k2)
        err58 = {"col": "c1 cc1"}
        self.checkmavg(**err58)         # form: mavg(col newname, k)
        err59 = {"k": "'1'"}
        # self.checkmavg(**err59)         # formL mavg(colm, "1")
        err60 = {"k": "-1-(-2)"}
        # self.checkmavg(**err60)         # formL mavg(colm, -1-2)
        err61 = {"k": 1001}
        self.checkmavg(**err61)         # k: right out of [1, 1000]
        err62 = {"k": -1}
        self.checkmavg(**err62)         # k: negative number
        err63 = {"k": 0}
        self.checkmavg(**err63)         # k: 0
        err64 = {"k": 2**63-1}
        self.checkmavg(**err64)         # k: max(bigint)
        err65 = {"k": 1-2**63}
        # self.checkmavg(**err65)         # k: min(bigint)
        err66 = {"k": -2**63}
        self.checkmavg(**err66)         # k: NULL
        err67 = {"k": 0.999999}
        self.checkmavg(**err67)         # k: left out of [1, 1000]
        err68 = {
            "table_expr": "stb1",
            "condition": "group by tbname order by tbname" # order by tbname not supported
        }
        self.checkmavg(**err68)

        pass

    def mavg_test_data(self, tbnum:int, data_row:int, basetime:int) -> None :
        for i in range(tbnum):
            for j in range(data_row):
                tdSql.execute(
                    f"insert into t{i} values ("
                    f"{basetime + (j+1)*10}, {random.randint(-200, -1)}, {random.uniform(200, -1)}, {basetime + random.randint(-200, -1)}, "
                    f"'binary_{j}', {random.uniform(-200, -1)}, {random.choice([0,1])}, {random.randint(-200,-1)}, "
                    f"{random.randint(-200, -1)}, {random.randint(-127, -1)}, 'nchar_{j}' )"
                )

                tdSql.execute(
                    f"insert into t{i} values ("
                    f"{basetime - (j+1) * 10}, {random.randint(1, 200)}, {random.uniform(1, 200)}, {basetime - random.randint(1, 200)}, "
                    f"'binary_{j}_1', {random.uniform(1, 200)}, {random.choice([0, 1])}, {random.randint(1,200)}, "
                    f"{random.randint(1,200)}, {random.randint(1,127)}, 'nchar_{j}_1' )"
                )
                tdSql.execute(
                    f"insert into tt{i} values ( {basetime-(j+1) * 10}, {random.randint(1, 200)} )"
                )

        pass

    def mavg_test_table(self,tbnum: int) -> None :
        tdSql.execute("drop database if exists db")
        tdSql.execute("create database  if not exists db keep 3650")
        tdSql.execute("use db")

        tdSql.execute(
            "create stable db.stb1 (\
                ts timestamp, c1 int, c2 float, c3 timestamp, c4 binary(16), c5 double, c6 bool, \
                c7 bigint, c8 smallint, c9 tinyint, c10 nchar(16)\
                ) \
            tags(st1 int)"
        )
        tdSql.execute(
            "create stable db.stb2 (ts timestamp, c1 int) tags(st2 int)"
        )
        for i in range(tbnum):
            tdSql.execute(f"create table t{i} using stb1 tags({i})")
            tdSql.execute(f"create table tt{i} using stb2 tags({i})")

        pass

    def mavg_test_run(self) :
        tdLog.printNoPrefix("==========TD-10594==========")
        tbnum = 10
        nowtime = int(round(time.time() * 1000))
        per_table_rows = 2
        self.mavg_test_table(tbnum)

        tdLog.printNoPrefix("######## no data test:")
        self.mavg_current_query()
        self.mavg_error_query()

        tdLog.printNoPrefix("######## insert only NULL test:")
        for i in range(tbnum):
            tdSql.execute(f"insert into t{i}(ts) values ({nowtime - 5})")
            tdSql.execute(f"insert into t{i}(ts) values ({nowtime + 5})")
        self.mavg_current_query()
        self.mavg_error_query()

        tdLog.printNoPrefix("######## insert data in the range near the max(bigint/double):")
        # self.mavg_test_table(tbnum)
        # tdSql.execute(f"insert into t1(ts, c1,c2,c5,c7) values "
        #               f"({nowtime - (per_table_rows + 1) * 10}, {2**31-1}, {3.4*10**38}, {1.7*10**308}, {2**63-1})")
        # tdSql.execute(f"insert into t1(ts, c1,c2,c5,c7) values "
        #               f"({nowtime - (per_table_rows + 2) * 10}, {2**31-1}, {3.4*10**38}, {1.7*10**308}, {2**63-1})")
        # self.mavg_current_query()
        # self.mavg_error_query()

        tdLog.printNoPrefix("######## insert data in the range near the min(bigint/double):")
        # self.mavg_test_table(tbnum)
        # tdSql.execute(f"insert into t1(ts, c1,c2,c5,c7) values "
        #               f"({nowtime - (per_table_rows + 1) * 10}, {1-2**31}, {-3.4*10**38}, {-1.7*10**308}, {1-2**63})")
        # tdSql.execute(f"insert into t1(ts, c1,c2,c5,c7) values "
        #               f"({nowtime - (per_table_rows + 2) * 10}, {1-2**31}, {-3.4*10**38}, {-1.7*10**308}, {512-2**63})")
        # self.mavg_current_query()
        # self.mavg_error_query()

        tdLog.printNoPrefix("######## insert data without NULL data test:")
        self.mavg_test_table(tbnum)
        self.mavg_test_data(tbnum, per_table_rows, nowtime)
        self.mavg_current_query()
        self.mavg_error_query()


        tdLog.printNoPrefix("######## insert data mix with NULL test:")
        for i in range(tbnum):
            tdSql.execute(f"insert into t{i}(ts) values ({nowtime})")
            tdSql.execute(f"insert into t{i}(ts) values ({nowtime-(per_table_rows+3)*10})")
            tdSql.execute(f"insert into t{i}(ts) values ({nowtime+(per_table_rows+3)*10})")
        self.mavg_current_query()
        self.mavg_error_query()



        tdLog.printNoPrefix("######## check after WAL test:")
        tdSql.query("show dnodes")
        index = tdSql.getData(0, 0)
        tdDnodes.stop(index)
        tdDnodes.start(index)
        self.mavg_current_query()
        self.mavg_error_query()
        tdSql.query("select mavg(1,1) from t1")
        tdSql.checkRows(7)
        tdSql.checkData(0,0,1.000000000)
        tdSql.checkData(1,0,1.000000000)
        tdSql.checkData(5,0,1.000000000)
        
        tdSql.query("select mavg(abs(c1),1) from t1")
        tdSql.checkRows(4)

    def mavg_support_stable(self):
        tdSql.query(" select mavg(1,3) from stb1 ")
        tdSql.checkRows(68)
        tdSql.checkData(0,0,1.000000000)
        tdSql.query("select mavg(c1,3) from stb1 partition by tbname ")
        tdSql.checkRows(38)
        # tdSql.query("select mavg(st1,3) from stb1 partition by tbname")
        # tdSql.checkRows(38)
        tdSql.query("select mavg(st1+c1,3) from stb1 partition by tbname")
        tdSql.checkRows(38)
        tdSql.query("select mavg(st1+c1,3) from stb1 partition by tbname")
        tdSql.checkRows(38)
        tdSql.query("select mavg(st1+c1,3) from stb1 partition by tbname")
        tdSql.checkRows(38)

        # # bug need fix 
        # tdSql.query("select mavg(st1+c1,3) from stb1 partition by tbname slimit 1 ")
        # tdSql.checkRows(2)
        # tdSql.error("select mavg(st1+c1,3) from stb1 partition by tbname limit 1 ")


        # bug need fix 
        tdSql.query("select mavg(st1+c1,3) from stb1 partition by tbname")
        tdSql.checkRows(38)        

        # bug need fix 
        # tdSql.query("select tbname , mavg(c1,3) from stb1 partition by tbname")
        # tdSql.checkRows(38)
        # tdSql.query("select tbname , mavg(st1,3) from stb1 partition by tbname")
        # tdSql.checkRows(38)
        # tdSql.query("select tbname , mavg(st1,3) from stb1 partition by tbname slimit 1")
        # tdSql.checkRows(2)

        # partition by tags 
        # tdSql.query("select st1 , mavg(c1,3) from stb1 partition by st1")
        # tdSql.checkRows(38)
        # tdSql.query("select mavg(c1,3) from stb1 partition by st1")
        # tdSql.checkRows(38)
        # tdSql.query("select st1 , mavg(c1,3) from stb1 partition by st1 slimit 1")
        # tdSql.checkRows(2)
        # tdSql.query("select mavg(c1,3) from stb1 partition by st1 slimit 1")
        # tdSql.checkRows(2)

        # partition by col
        # tdSql.query("select c1 , mavg(c1,3) from stb1 partition by c1")
        # tdSql.checkRows(38)
        # tdSql.query("select mavg(c1 ,3) from stb1 partition by c1")
        # tdSql.checkRows(38)
        # tdSql.query("select c1 , mavg(c1,3) from stb1 partition by st1 slimit 1")
        # tdSql.checkRows(2)
        # tdSql.query("select diff(c1) from stb1 partition by st1 slimit 1")
        # tdSql.checkRows(2)

    def run(self):
        import traceback
        try:
            # run in  develop branch
            self.mavg_test_run()
            self.mavg_support_stable()
            pass
        except Exception as e:
            traceback.print_exc()
            raise e


    def stop(self):
        tdSql.close()
        tdLog.success("%s successfully executed" % __file__)

tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())