functions.go 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190
  1. package xsql
  2. import (
  3. "github.com/emqx/kuiper/plugins"
  4. "strings"
  5. )
  6. // ONLY use NewFunctionValuer function to initialize
  7. type FunctionValuer struct {
  8. funcPlugins *funcPlugins
  9. }
  10. //Should only be called by stream to make sure a single instance for an operation
  11. func NewFunctionValuer(p *funcPlugins) *FunctionValuer {
  12. fv := &FunctionValuer{
  13. funcPlugins: p,
  14. }
  15. return fv
  16. }
  17. func (*FunctionValuer) Value(_ string) (interface{}, bool) {
  18. return nil, false
  19. }
  20. func (*FunctionValuer) Meta(_ string) (interface{}, bool) {
  21. return nil, false
  22. }
  23. var aggFuncMap = map[string]string{"avg": "",
  24. "count": "",
  25. "max": "", "min": "",
  26. "sum": "",
  27. "collect": "",
  28. "deduplicate": "",
  29. }
  30. var funcWithAsteriskSupportMap = map[string]string{
  31. "collect": "",
  32. "count": "",
  33. }
  34. var mathFuncMap = map[string]string{"abs": "", "acos": "", "asin": "", "atan": "", "atan2": "",
  35. "bitand": "", "bitor": "", "bitxor": "", "bitnot": "",
  36. "ceil": "", "cos": "", "cosh": "",
  37. "exp": "",
  38. "ln": "", "log": "",
  39. "mod": "",
  40. "power": "",
  41. "rand": "", "round": "",
  42. "sign": "", "sin": "", "sinh": "", "sqrt": "",
  43. "tan": "", "tanh": "",
  44. }
  45. var strFuncMap = map[string]string{"concat": "",
  46. "endswith": "",
  47. "format_time": "",
  48. "indexof": "",
  49. "length": "", "lower": "", "lpad": "", "ltrim": "",
  50. "numbytes": "",
  51. "regexp_matches": "", "regexp_replace": "", "regexp_substr": "", "rpad": "", "rtrim": "",
  52. "substring": "", "startswith": "", "split_value": "",
  53. "trim": "",
  54. "upper": "",
  55. }
  56. var convFuncMap = map[string]string{"concat": "", "cast": "", "chr": "",
  57. "encode": "",
  58. "trunc": "",
  59. }
  60. var hashFuncMap = map[string]string{"md5": "",
  61. "sha1": "", "sha256": "", "sha384": "", "sha512": "",
  62. }
  63. var jsonFuncMap = map[string]string{
  64. "json_path_query": "", "json_path_query_first": "", "json_path_exists": "",
  65. }
  66. var otherFuncMap = map[string]string{"isnull": "",
  67. "newuuid": "", "tstamp": "", "mqtt": "", "meta": "",
  68. }
  69. func (fv *FunctionValuer) Call(name string, args []interface{}) (interface{}, bool) {
  70. lowerName := strings.ToLower(name)
  71. if _, ok := mathFuncMap[lowerName]; ok {
  72. return mathCall(name, args)
  73. } else if _, ok := strFuncMap[lowerName]; ok {
  74. return strCall(lowerName, args)
  75. } else if _, ok := convFuncMap[lowerName]; ok {
  76. return convCall(lowerName, args)
  77. } else if _, ok := hashFuncMap[lowerName]; ok {
  78. return hashCall(lowerName, args)
  79. } else if _, ok := jsonFuncMap[lowerName]; ok {
  80. return jsonCall(lowerName, args)
  81. } else if _, ok := otherFuncMap[lowerName]; ok {
  82. return otherCall(lowerName, args)
  83. } else if _, ok := aggFuncMap[lowerName]; ok {
  84. return nil, false
  85. } else {
  86. nf, fctx, err := fv.funcPlugins.GetFuncFromPlugin(name)
  87. if err != nil {
  88. return err, false
  89. }
  90. if nf.IsAggregate() {
  91. return nil, false
  92. }
  93. logger := fctx.GetLogger()
  94. logger.Debugf("run func %s", name)
  95. result, ok := nf.Exec(args, fctx)
  96. logger.Debugf("run custom function %s, get result %v", name, result)
  97. return result, ok
  98. }
  99. }
  100. func IsAggStatement(node Node) bool {
  101. var r = false
  102. WalkFunc(node, func(n Node) {
  103. if f, ok := n.(*Call); ok {
  104. if ok := isAggFunc(f); ok {
  105. r = true
  106. return
  107. }
  108. } else if d, ok := n.(Dimensions); ok {
  109. ds := d.GetGroups()
  110. if ds != nil && len(ds) > 0 {
  111. r = true
  112. return
  113. }
  114. }
  115. })
  116. return r
  117. }
  118. func isAggFunc(f *Call) bool {
  119. fn := strings.ToLower(f.Name)
  120. if _, ok := aggFuncMap[fn]; ok {
  121. return true
  122. } else if _, ok := strFuncMap[fn]; ok {
  123. return false
  124. } else if _, ok := convFuncMap[fn]; ok {
  125. return false
  126. } else if _, ok := hashFuncMap[fn]; ok {
  127. return false
  128. } else if _, ok := otherFuncMap[fn]; ok {
  129. return false
  130. } else if _, ok := mathFuncMap[fn]; ok {
  131. return false
  132. } else {
  133. if nf, err := plugins.GetFunction(f.Name); err == nil {
  134. if nf.IsAggregate() {
  135. //Add cache
  136. aggFuncMap[fn] = ""
  137. return true
  138. }
  139. }
  140. }
  141. return false
  142. }
  143. func HasAggFuncs(node Node) bool {
  144. if node == nil {
  145. return false
  146. }
  147. var r = false
  148. WalkFunc(node, func(n Node) {
  149. if f, ok := n.(*Call); ok {
  150. if ok := isAggFunc(f); ok {
  151. r = true
  152. return
  153. }
  154. }
  155. })
  156. return r
  157. }
  158. func HasNoAggFuncs(node Node) bool {
  159. if node == nil {
  160. return false
  161. }
  162. var r = false
  163. WalkFunc(node, func(n Node) {
  164. if f, ok := n.(*Call); ok {
  165. if ok := isAggFunc(f); !ok {
  166. r = true
  167. return
  168. }
  169. }
  170. })
  171. return r
  172. }