forked from GiteaTest2015/streamsql
rulego-team
600 lines
18 KiB
Go
600 lines
18 KiB
Go
package stream
|
|
|
|
import (
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/rulego/streamsql/expr"
|
|
"github.com/rulego/streamsql/functions"
|
|
"github.com/rulego/streamsql/logger"
|
|
"github.com/rulego/streamsql/utils/fieldpath"
|
|
)
|
|
|
|
// fieldProcessInfo field processing information for caching pre-compiled field processing logic
|
|
type fieldProcessInfo struct {
|
|
fieldName string // Original field name
|
|
outputName string // Output field name
|
|
isFunctionCall bool // Whether it's a function call
|
|
hasNestedField bool // Whether it contains nested fields
|
|
isSelectAll bool // Whether it's SELECT *
|
|
isStringLiteral bool // Whether it's a string literal
|
|
stringValue string // Pre-processed string literal value (quotes removed)
|
|
alias string // Field alias for quick access
|
|
}
|
|
|
|
// expressionProcessInfo expression processing information for caching pre-compiled expression processing logic
|
|
type expressionProcessInfo struct {
|
|
originalExpr string // Original expression
|
|
processedExpr string // Pre-processed expression
|
|
isFunctionCall bool // Whether it's a function call
|
|
hasNestedFields bool // Whether it contains nested fields
|
|
compiledExpr *expr.Expression // Pre-compiled expression object
|
|
needsBacktickPreprocess bool // Whether backtick preprocessing is needed
|
|
}
|
|
|
|
// compileFieldProcessInfo pre-compiles field processing information to avoid runtime re-parsing
|
|
func (s *Stream) compileFieldProcessInfo() {
|
|
s.compiledFieldInfo = make(map[string]*fieldProcessInfo)
|
|
s.compiledExprInfo = make(map[string]*expressionProcessInfo)
|
|
|
|
// Compile SimpleFields information
|
|
for _, fieldSpec := range s.config.SimpleFields {
|
|
info := s.compileSimpleFieldInfo(fieldSpec)
|
|
s.compiledFieldInfo[fieldSpec] = info
|
|
}
|
|
|
|
// Pre-compile expression field information
|
|
s.compileExpressionInfo()
|
|
}
|
|
|
|
// compileSimpleFieldInfo compiles simple field information
|
|
func (s *Stream) compileSimpleFieldInfo(fieldSpec string) *fieldProcessInfo {
|
|
info := &fieldProcessInfo{}
|
|
|
|
if fieldSpec == "*" {
|
|
info.isSelectAll = true
|
|
info.fieldName = ""
|
|
info.outputName = "*"
|
|
info.alias = "*"
|
|
return info
|
|
}
|
|
|
|
// Parse alias
|
|
parts := strings.Split(fieldSpec, ":")
|
|
info.fieldName = parts[0]
|
|
// Remove backticks from field name
|
|
if len(info.fieldName) >= 2 && info.fieldName[0] == '`' && info.fieldName[len(info.fieldName)-1] == '`' {
|
|
info.fieldName = info.fieldName[1 : len(info.fieldName)-1]
|
|
}
|
|
info.outputName = info.fieldName
|
|
if len(parts) > 1 {
|
|
info.outputName = parts[1]
|
|
// Remove backticks from output name
|
|
if len(info.outputName) >= 2 && info.outputName[0] == '`' && info.outputName[len(info.outputName)-1] == '`' {
|
|
info.outputName = info.outputName[1 : len(info.outputName)-1]
|
|
}
|
|
}
|
|
|
|
// Pre-determine field characteristics
|
|
info.isFunctionCall = strings.Contains(info.fieldName, "(") && strings.Contains(info.fieldName, ")")
|
|
info.hasNestedField = !info.isFunctionCall && fieldpath.IsNestedField(info.fieldName)
|
|
|
|
// Check if it's a string literal and preprocess value
|
|
info.isStringLiteral = (len(info.fieldName) >= 2 &&
|
|
((info.fieldName[0] == '\'' && info.fieldName[len(info.fieldName)-1] == '\'') ||
|
|
(info.fieldName[0] == '"' && info.fieldName[len(info.fieldName)-1] == '"')))
|
|
|
|
// Preprocess string literal value, remove quotes
|
|
if info.isStringLiteral && len(info.fieldName) >= 2 {
|
|
info.stringValue = info.fieldName[1 : len(info.fieldName)-1]
|
|
}
|
|
|
|
// Set alias for quick access
|
|
info.alias = info.outputName
|
|
|
|
return info
|
|
}
|
|
|
|
// compileExpressionInfo pre-compiles expression processing information
|
|
func (s *Stream) compileExpressionInfo() {
|
|
// Initialize unnest function detection flag
|
|
s.hasUnnestFunction = false
|
|
bridge := functions.GetExprBridge()
|
|
|
|
for fieldName, fieldExpr := range s.config.FieldExpressions {
|
|
exprInfo := &expressionProcessInfo{
|
|
originalExpr: fieldExpr.Expression,
|
|
}
|
|
|
|
// Preprocess expression
|
|
processedExpr := fieldExpr.Expression
|
|
if bridge.ContainsIsNullOperator(processedExpr) {
|
|
if processed, err := bridge.PreprocessIsNullExpression(processedExpr); err == nil {
|
|
processedExpr = processed
|
|
}
|
|
}
|
|
if bridge.ContainsLikeOperator(processedExpr) {
|
|
if processed, err := bridge.PreprocessLikeExpression(processedExpr); err == nil {
|
|
processedExpr = processed
|
|
}
|
|
}
|
|
exprInfo.processedExpr = processedExpr
|
|
|
|
// Pre-judge expression characteristics
|
|
exprInfo.isFunctionCall = strings.Contains(fieldExpr.Expression, "(") && strings.Contains(fieldExpr.Expression, ")")
|
|
exprInfo.hasNestedFields = !exprInfo.isFunctionCall && strings.Contains(fieldExpr.Expression, ".")
|
|
exprInfo.needsBacktickPreprocess = bridge.ContainsBacktickIdentifiers(fieldExpr.Expression)
|
|
|
|
// Check if expression contains unnest function
|
|
if exprInfo.isFunctionCall && strings.Contains(strings.ToLower(fieldExpr.Expression), "unnest(") {
|
|
s.hasUnnestFunction = true
|
|
}
|
|
|
|
// Pre-compile expression object (only for non-function call expressions)
|
|
if !exprInfo.isFunctionCall {
|
|
exprToCompile := fieldExpr.Expression
|
|
if exprInfo.needsBacktickPreprocess {
|
|
if processed, err := bridge.PreprocessBacktickIdentifiers(exprToCompile); err == nil {
|
|
exprToCompile = processed
|
|
}
|
|
}
|
|
if compiledExpr, err := expr.NewExpression(exprToCompile); err == nil {
|
|
exprInfo.compiledExpr = compiledExpr
|
|
}
|
|
}
|
|
|
|
s.compiledExprInfo[fieldName] = exprInfo
|
|
}
|
|
}
|
|
|
|
// processExpressionField processes expression field
|
|
func (s *Stream) processExpressionField(fieldName string, dataMap map[string]interface{}, result map[string]interface{}) {
|
|
exprInfo := s.compiledExprInfo[fieldName]
|
|
if exprInfo == nil {
|
|
// Fallback to original logic
|
|
s.processExpressionFieldFallback(fieldName, dataMap, result)
|
|
return
|
|
}
|
|
|
|
var evalResult interface{}
|
|
bridge := functions.GetExprBridge()
|
|
|
|
if exprInfo.isFunctionCall {
|
|
// For function calls, use bridge processor
|
|
exprResult, err := bridge.EvaluateExpression(exprInfo.processedExpr, dataMap)
|
|
if err != nil {
|
|
logger.Error("Function call evaluation failed for field %s: %v", fieldName, err)
|
|
result[fieldName] = nil
|
|
return
|
|
}
|
|
evalResult = exprResult
|
|
} else if exprInfo.hasNestedFields {
|
|
// Use pre-compiled expression object
|
|
if exprInfo.compiledExpr != nil {
|
|
// Use EvaluateValueWithNull to get actual value (including strings)
|
|
exprResult, isNull, err := exprInfo.compiledExpr.EvaluateValueWithNull(dataMap)
|
|
if err != nil {
|
|
logger.Error("Expression evaluation failed for field %s: %v", fieldName, err)
|
|
result[fieldName] = nil
|
|
return
|
|
}
|
|
if isNull {
|
|
evalResult = nil
|
|
} else {
|
|
evalResult = exprResult
|
|
}
|
|
} else {
|
|
// Fallback to dynamic compilation
|
|
s.processExpressionFieldFallback(fieldName, dataMap, result)
|
|
return
|
|
}
|
|
} else {
|
|
// Try using bridge processor for other expressions
|
|
exprResult, err := bridge.EvaluateExpression(exprInfo.processedExpr, dataMap)
|
|
if err != nil {
|
|
// If bridge fails, use pre-compiled expression object
|
|
if exprInfo.compiledExpr != nil {
|
|
// Use EvaluateValueWithNull to get actual value (including strings)
|
|
exprResult, isNull, evalErr := exprInfo.compiledExpr.EvaluateValueWithNull(dataMap)
|
|
if evalErr != nil {
|
|
logger.Error("Expression evaluation failed for field %s: %v", fieldName, evalErr)
|
|
result[fieldName] = nil
|
|
return
|
|
}
|
|
if isNull {
|
|
evalResult = nil
|
|
} else {
|
|
evalResult = exprResult
|
|
}
|
|
} else {
|
|
// Fallback to dynamic compilation
|
|
s.processExpressionFieldFallback(fieldName, dataMap, result)
|
|
return
|
|
}
|
|
} else {
|
|
evalResult = exprResult
|
|
}
|
|
}
|
|
|
|
result[fieldName] = evalResult
|
|
}
|
|
|
|
// processExpressionFieldFallback fallback logic for expression field processing
|
|
func (s *Stream) processExpressionFieldFallback(fieldName string, dataMap map[string]interface{}, result map[string]interface{}) {
|
|
fieldExpr, exists := s.config.FieldExpressions[fieldName]
|
|
if !exists {
|
|
result[fieldName] = nil
|
|
return
|
|
}
|
|
|
|
// Use bridge to calculate expression, supports IS NULL and other syntax
|
|
bridge := functions.GetExprBridge()
|
|
|
|
// Preprocess IS NULL and LIKE syntax in expression
|
|
processedExpr := fieldExpr.Expression
|
|
if bridge.ContainsIsNullOperator(processedExpr) {
|
|
if processed, err := bridge.PreprocessIsNullExpression(processedExpr); err == nil {
|
|
processedExpr = processed
|
|
}
|
|
}
|
|
if bridge.ContainsLikeOperator(processedExpr) {
|
|
if processed, err := bridge.PreprocessLikeExpression(processedExpr); err == nil {
|
|
processedExpr = processed
|
|
}
|
|
}
|
|
|
|
// Check if expression is a function call (contains parentheses)
|
|
isFunctionCall := strings.Contains(fieldExpr.Expression, "(") && strings.Contains(fieldExpr.Expression, ")")
|
|
|
|
// Check if expression contains nested fields (but exclude dots in function calls)
|
|
hasNestedFields := false
|
|
if !isFunctionCall && strings.Contains(fieldExpr.Expression, ".") {
|
|
hasNestedFields = true
|
|
}
|
|
|
|
var evalResult interface{}
|
|
|
|
if isFunctionCall {
|
|
// For function calls, prioritize bridge processor
|
|
exprResult, err := bridge.EvaluateExpression(processedExpr, dataMap)
|
|
if err != nil {
|
|
logger.Error("Function call evaluation failed for field %s: %v", fieldName, err)
|
|
result[fieldName] = nil
|
|
return
|
|
}
|
|
evalResult = exprResult
|
|
} else if hasNestedFields {
|
|
// Detected nested fields (non-function call), use custom expression engine
|
|
exprToUse := fieldExpr.Expression
|
|
if bridge.ContainsBacktickIdentifiers(exprToUse) {
|
|
if processed, err := bridge.PreprocessBacktickIdentifiers(exprToUse); err == nil {
|
|
exprToUse = processed
|
|
}
|
|
}
|
|
expression, parseErr := expr.NewExpression(exprToUse)
|
|
if parseErr != nil {
|
|
logger.Error("Expression parse failed for field %s: %v", fieldName, parseErr)
|
|
result[fieldName] = nil
|
|
return
|
|
}
|
|
|
|
// Use EvaluateValueWithNull to get actual value (including strings)
|
|
exprResult, isNull, err := expression.EvaluateValueWithNull(dataMap)
|
|
if err != nil {
|
|
logger.Error("Expression evaluation failed for field %s: %v", fieldName, err)
|
|
result[fieldName] = nil
|
|
return
|
|
}
|
|
if isNull {
|
|
evalResult = nil
|
|
} else {
|
|
evalResult = exprResult
|
|
}
|
|
} else {
|
|
// Try using bridge processor for other expressions
|
|
exprResult, err := bridge.EvaluateExpression(processedExpr, dataMap)
|
|
if err != nil {
|
|
// If bridge fails, fallback to original expression engine
|
|
exprToUse := fieldExpr.Expression
|
|
if bridge.ContainsBacktickIdentifiers(exprToUse) {
|
|
if processed, err := bridge.PreprocessBacktickIdentifiers(exprToUse); err == nil {
|
|
exprToUse = processed
|
|
}
|
|
}
|
|
expression, parseErr := expr.NewExpression(exprToUse)
|
|
if parseErr != nil {
|
|
logger.Error("Expression parse failed for field %s: %v", fieldName, parseErr)
|
|
result[fieldName] = nil
|
|
return
|
|
}
|
|
|
|
// Use EvaluateValueWithNull to get actual value (including strings)
|
|
exprResult, isNull, evalErr := expression.EvaluateValueWithNull(dataMap)
|
|
if evalErr != nil {
|
|
logger.Error("Expression evaluation failed for field %s: %v", fieldName, evalErr)
|
|
result[fieldName] = nil
|
|
return
|
|
}
|
|
if isNull {
|
|
evalResult = nil
|
|
} else {
|
|
evalResult = exprResult
|
|
}
|
|
} else {
|
|
evalResult = exprResult
|
|
}
|
|
}
|
|
|
|
result[fieldName] = evalResult
|
|
}
|
|
|
|
// processSimpleField processes simple field
|
|
func (s *Stream) processSimpleField(fieldSpec string, dataMap map[string]interface{}, data interface{}, result map[string]interface{}) {
|
|
info := s.compiledFieldInfo[fieldSpec]
|
|
if info == nil {
|
|
// If no pre-compiled info, fallback to original logic (safety guarantee)
|
|
s.processSingleFieldFallback(fieldSpec, dataMap, data, result)
|
|
return
|
|
}
|
|
|
|
if info.isSelectAll {
|
|
// SELECT *: batch copy all fields, skip expression fields
|
|
for k, v := range dataMap {
|
|
if _, isExpression := s.config.FieldExpressions[k]; !isExpression {
|
|
result[k] = v
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
// Skip fields already processed by expression fields
|
|
if _, isExpression := s.config.FieldExpressions[info.outputName]; isExpression {
|
|
return
|
|
}
|
|
|
|
if info.isStringLiteral {
|
|
// String literal processing: use pre-compiled string value
|
|
result[info.alias] = info.stringValue
|
|
} else if info.isFunctionCall {
|
|
// Execute function call
|
|
if funcResult, err := s.executeFunction(info.fieldName, dataMap); err == nil {
|
|
result[info.outputName] = funcResult
|
|
} else {
|
|
logger.Error("Function execution error %s: %v", info.fieldName, err)
|
|
result[info.outputName] = nil
|
|
}
|
|
} else {
|
|
// Ordinary field processing
|
|
var value interface{}
|
|
var exists bool
|
|
|
|
if info.hasNestedField {
|
|
value, exists = fieldpath.GetNestedField(data, info.fieldName)
|
|
} else {
|
|
value, exists = dataMap[info.fieldName]
|
|
}
|
|
|
|
if exists {
|
|
result[info.outputName] = value
|
|
} else {
|
|
result[info.outputName] = nil
|
|
}
|
|
}
|
|
}
|
|
|
|
// processSingleFieldFallback fallback processing for single field (when pre-compiled info is missing)
|
|
func (s *Stream) processSingleFieldFallback(fieldSpec string, dataMap map[string]interface{}, data interface{}, result map[string]interface{}) {
|
|
// Handle special case of SELECT *
|
|
if fieldSpec == "*" {
|
|
// SELECT *: return all fields, but skip fields already processed by expression fields
|
|
for k, v := range dataMap {
|
|
// If field already processed by expression field, skip, maintain expression calculation result
|
|
if _, isExpression := s.config.FieldExpressions[k]; !isExpression {
|
|
result[k] = v
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
// Handle alias
|
|
parts := strings.Split(fieldSpec, ":")
|
|
fieldName := parts[0]
|
|
outputName := fieldName
|
|
if len(parts) > 1 {
|
|
outputName = parts[1]
|
|
}
|
|
|
|
// Skip fields already processed by expression fields
|
|
if _, isExpression := s.config.FieldExpressions[outputName]; isExpression {
|
|
return
|
|
}
|
|
|
|
// Check if it's a function call
|
|
if strings.Contains(fieldName, "(") && strings.Contains(fieldName, ")") {
|
|
// Execute function call
|
|
if funcResult, err := s.executeFunction(fieldName, dataMap); err == nil {
|
|
result[outputName] = funcResult
|
|
} else {
|
|
logger.Error("Function execution error %s: %v", fieldName, err)
|
|
result[outputName] = nil
|
|
}
|
|
} else {
|
|
// Ordinary field - supports nested fields
|
|
var value interface{}
|
|
var exists bool
|
|
|
|
if fieldpath.IsNestedField(fieldName) {
|
|
value, exists = fieldpath.GetNestedField(data, fieldName)
|
|
} else {
|
|
value, exists = dataMap[fieldName]
|
|
}
|
|
|
|
if exists {
|
|
result[outputName] = value
|
|
} else {
|
|
result[outputName] = nil
|
|
}
|
|
}
|
|
}
|
|
|
|
// executeFunction executes function call
|
|
func (s *Stream) executeFunction(funcExpr string, data map[string]interface{}) (interface{}, error) {
|
|
// Check if it's a custom function
|
|
funcName := extractFunctionName(funcExpr)
|
|
if funcName != "" {
|
|
// Use function system directly
|
|
fn, exists := functions.Get(funcName)
|
|
if exists {
|
|
// Parse parameters
|
|
args, err := s.parseFunctionArgs(funcExpr, data)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Create function context
|
|
ctx := &functions.FunctionContext{Data: data}
|
|
|
|
// Execute function
|
|
return fn.Execute(ctx, args)
|
|
}
|
|
}
|
|
|
|
// For complex nested function calls, use ExprBridge directly
|
|
// This avoids the float64 type limitation of Expression.Evaluate
|
|
bridge := functions.GetExprBridge()
|
|
result, err := bridge.EvaluateExpression(funcExpr, data)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("evaluate function expression failed: %w", err)
|
|
}
|
|
|
|
return result, nil
|
|
}
|
|
|
|
// extractFunctionName extracts function name from expression
|
|
func extractFunctionName(expr string) string {
|
|
parenIndex := strings.Index(expr, "(")
|
|
if parenIndex == -1 {
|
|
return ""
|
|
}
|
|
funcName := strings.TrimSpace(expr[:parenIndex])
|
|
if strings.ContainsAny(funcName, " +-*/=<>!&|") {
|
|
return ""
|
|
}
|
|
return funcName
|
|
}
|
|
|
|
// parseFunctionArgs parses function arguments, supports nested function calls
|
|
func (s *Stream) parseFunctionArgs(funcExpr string, data map[string]interface{}) ([]interface{}, error) {
|
|
// Extract parameters within parentheses
|
|
start := strings.Index(funcExpr, "(")
|
|
end := strings.LastIndex(funcExpr, ")")
|
|
if start == -1 || end == -1 || end <= start {
|
|
return nil, fmt.Errorf("invalid function expression: %s", funcExpr)
|
|
}
|
|
|
|
argsStr := strings.TrimSpace(funcExpr[start+1 : end])
|
|
if argsStr == "" {
|
|
return []interface{}{}, nil
|
|
}
|
|
|
|
// Smart split arguments, handle nested functions and quotes
|
|
argParts, err := s.smartSplitArgs(argsStr)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
args := make([]interface{}, len(argParts))
|
|
|
|
for i, arg := range argParts {
|
|
arg = strings.TrimSpace(arg)
|
|
|
|
// If parameter is string constant (enclosed in quotes)
|
|
if strings.HasPrefix(arg, "'") && strings.HasSuffix(arg, "'") {
|
|
args[i] = strings.Trim(arg, "'")
|
|
} else if strings.HasPrefix(arg, "\"") && strings.HasSuffix(arg, "\"") {
|
|
args[i] = strings.Trim(arg, "\"")
|
|
} else if strings.Contains(arg, "(") {
|
|
// If parameter contains function call, execute recursively
|
|
result, err := s.executeFunction(arg, data)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to execute nested function '%s': %v", arg, err)
|
|
}
|
|
args[i] = result
|
|
} else if value, exists := data[arg]; exists {
|
|
// If it's a data field
|
|
args[i] = value
|
|
} else {
|
|
// Try to parse as number
|
|
if val, err := strconv.ParseFloat(arg, 64); err == nil {
|
|
args[i] = val
|
|
} else {
|
|
args[i] = arg
|
|
}
|
|
}
|
|
}
|
|
|
|
return args, nil
|
|
}
|
|
|
|
// smartSplitArgs intelligently splits arguments, considering bracket nesting and quotes
|
|
func (s *Stream) smartSplitArgs(argsStr string) ([]string, error) {
|
|
var args []string
|
|
var current strings.Builder
|
|
parenDepth := 0
|
|
inQuotes := false
|
|
quoteChar := byte(0)
|
|
|
|
for i := 0; i < len(argsStr); i++ {
|
|
ch := argsStr[i]
|
|
|
|
switch ch {
|
|
case '\'':
|
|
if !inQuotes {
|
|
inQuotes = true
|
|
quoteChar = ch
|
|
} else if quoteChar == ch {
|
|
inQuotes = false
|
|
quoteChar = 0
|
|
}
|
|
current.WriteByte(ch)
|
|
case '"':
|
|
if !inQuotes {
|
|
inQuotes = true
|
|
quoteChar = ch
|
|
} else if quoteChar == ch {
|
|
inQuotes = false
|
|
quoteChar = 0
|
|
}
|
|
current.WriteByte(ch)
|
|
case '(':
|
|
if !inQuotes {
|
|
parenDepth++
|
|
}
|
|
current.WriteByte(ch)
|
|
case ')':
|
|
if !inQuotes {
|
|
parenDepth--
|
|
}
|
|
current.WriteByte(ch)
|
|
case ',':
|
|
if !inQuotes && parenDepth == 0 {
|
|
// Found parameter separator
|
|
args = append(args, strings.TrimSpace(current.String()))
|
|
current.Reset()
|
|
} else {
|
|
current.WriteByte(ch)
|
|
}
|
|
default:
|
|
current.WriteByte(ch)
|
|
}
|
|
}
|
|
|
|
// Add the last parameter
|
|
if current.Len() > 0 {
|
|
args = append(args, strings.TrimSpace(current.String()))
|
|
}
|
|
|
|
return args, nil
|
|
}
|