// Copyright The kweaver.ai Authors.
//
// Licensed under the Apache License, Version 2.0.
// See the LICENSE file in the project root for details.

package resource

import (
	"context"
	"fmt"
	"net/http"
	"regexp"
	"slices"
	"strings"

	"github.com/dlclark/regexp2"
	"github.com/kweaver-ai/kweaver-go-lib/otel/oteltrace"
	"github.com/kweaver-ai/kweaver-go-lib/rest"
	"github.com/mitchellh/mapstructure"
	"go.opentelemetry.io/otel/codes"

	verrors "vega-backend/errors"
	"vega-backend/interfaces"
	fcond "vega-backend/logics/filter_condition"
)

// 创建和更新视图的一些通用操作
func (rs *resourceService) validateLogicDefinition(ctx context.Context, view *interfaces.ResourceRequest) (string, error) {
	ctx, span := oteltrace.StartNamedInternalSpan(ctx, "logic layer: Common operation for creating and updating views")
	defer span.End()

	// 自定义视图
	if view.LogicDefinition == nil {
		return "", rest.NewHTTPError(ctx, http.StatusBadRequest, rest.PublicError_BadRequest).
			WithErrorDetails("Logic definition is empty")
	}

	// 校验节点ID的唯一性
	nodeMap := make(map[string]struct{})
	for _, node := range view.LogicDefinition {
		if node.ID == "" {
			return "", rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("Node ID cannot be empty")
		}
		if _, exists := nodeMap[node.ID]; exists {
			return "", rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_Duplicated_NodeID).
				WithErrorDetails(fmt.Sprintf("Duplicate node ID found: %s", node.ID))
		}
		nodeMap[node.ID] = struct{}{}
	}

	resourceNodeCount := 0
	outputNodeCount := 0
	refResourceMap := make(map[string]*interfaces.Resource)

	for _, node := range view.LogicDefinition {
		// 节点不能自引用
		if slices.Contains(node.Inputs, node.ID) {
			return "", rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("Node '%s' cannot reference itself: %s", node.ID, node.ID))
		}

		switch node.Type {
		case interfaces.LogicDefinitionNodeType_Resource:
			// 校验资源节点
			err := validateResourceNode(ctx, rs, node, refResourceMap)
			if err != nil {
				return "", err
			}

			resourceNodeCount++
		case interfaces.LogicDefinitionNodeType_Join:
			err := validateJoinNode(ctx, rs, node, view.LogicDefinition, nodeMap)
			if err != nil {
				return "", err
			}
		case interfaces.LogicDefinitionNodeType_Union:
			err := validateUnionNode(ctx, rs, view.Category, node, view.LogicDefinition, nodeMap)
			if err != nil {
				return "", err
			}
		case interfaces.LogicDefinitionNodeType_Sql:
			err := validateSqlNode(ctx, rs, node, view.LogicDefinition, nodeMap)
			if err != nil {
				return "", err
			}
		case interfaces.LogicDefinitionNodeType_Output:
			err := validateOutputNode(ctx, rs, node, view.LogicDefinition, nodeMap)
			if err != nil {
				return "", err
			}

			outputNodeCount++

		default:
			return "", rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("The logic definition node type is invalid")
		}
	}

	// 判断视图类型：衍生视图还是组合视图
	logicType := determineLogicType(view.LogicDefinition)

	var refResourceCategory string
	refResourceCategoryMap := make(map[string]struct{})
	refResourceCatalogMap := make(map[string]struct{})
	for _, dsView := range refResourceMap {
		refResourceCatalogMap[dsView.CatalogID] = struct{}{}
		refResourceCategoryMap[dsView.Category] = struct{}{}
		refResourceCategory = dsView.Category
	}

	if len(refResourceCategoryMap) != 1 {
		return "", rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The source view of the custom view must have the same category")
	}

	// 如果数据源类型是opensearch，则不能跨opensearch数据源选择
	if refResourceCategory == interfaces.ResourceCategoryIndex && len(refResourceCatalogMap) > 1 {
		return "", rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The source view of query type DSL must have the same data source when create custom view")
	}

	span.SetStatus(codes.Ok, "")
	return logicType, nil
}

// determineLogicType 判断视图类型：衍生视图还是组合视图
// 衍生视图：输出节点只引用一个资源节点（没有经过 Join/Union/SQL 等多源处理节点）
// 组合视图：输出节点引用了多个资源节点，或经过了 Join/Union/SQL 等处理节点
func determineLogicType(nodes []*interfaces.LogicDefinitionNode) string {
	// 默认是组合视图
	logicType := interfaces.LogicType_Composite

	// 找到输出节点
	var outputNode *interfaces.LogicDefinitionNode
	for _, node := range nodes {
		if node.Type == interfaces.LogicDefinitionNodeType_Output {
			outputNode = node
			break
		}
	}

	if outputNode != nil && len(outputNode.Inputs) == 1 {
		// 输出节点只有一个输入，检查是否只引用了一个资源节点
		// 递归追踪输入节点，看是否最终只引用了一个资源节点，且没有经过 Join/Union/SQL 节点
		hasProcessingNode := false
		resourceNodeIDs := make(map[string]struct{})

		// 使用 BFS 遍历所有上游节点
		visited := make(map[string]struct{})
		queue := []string{outputNode.Inputs[0]}
		visited[outputNode.Inputs[0]] = struct{}{}

		for len(queue) > 0 {
			currentID := queue[0]
			queue = queue[1:]

			// 找到当前节点
			var currentNode *interfaces.LogicDefinitionNode
			for _, n := range nodes {
				if n.ID == currentID {
					currentNode = n
					break
				}
			}

			if currentNode == nil {
				continue
			}

			// 检查节点类型
			switch currentNode.Type {
			case interfaces.LogicDefinitionNodeType_Resource:
				// 记录资源节点
				resourceNodeIDs[currentNode.ID] = struct{}{}
			case interfaces.LogicDefinitionNodeType_Join,
				interfaces.LogicDefinitionNodeType_Union,
				interfaces.LogicDefinitionNodeType_Sql:
				// 遇到处理节点，标记为组合视图
				hasProcessingNode = true
			case interfaces.LogicDefinitionNodeType_Output:
				// 不应该出现，但忽略
				// break
			}

			// 将输入节点加入队列
			for _, inputID := range currentNode.Inputs {
				if _, ok := visited[inputID]; !ok {
					visited[inputID] = struct{}{}
					queue = append(queue, inputID)
				}
			}
		}

		// 如果只有一个资源节点且没有经过处理节点，则为衍生视图
		if !hasProcessingNode && len(resourceNodeIDs) == 1 {
			logicType = interfaces.LogicType_Derived
		}
	}

	return logicType
}

// 获取节点的输出字段映射（用于校验字段是否存在）
func getNodeOutputFieldsMap(ctx context.Context, rs *resourceService, nodeID string,
	allNodes []*interfaces.LogicDefinitionNode, nodeCache map[string]map[string]*interfaces.Property) (map[string]*interfaces.Property, error) {

	// 如果已经计算过，直接返回缓存结果
	if cached, ok := nodeCache[nodeID]; ok {
		return cached, nil
	}

	// 找到节点
	var node *interfaces.LogicDefinitionNode
	for _, n := range allNodes {
		if n.ID == nodeID {
			node = n
			break
		}
	}
	if node == nil {
		return nil, fmt.Errorf("node %s not found", nodeID)
	}

	fieldsMap := make(map[string]*interfaces.Property)

	switch node.Type {
	case interfaces.LogicDefinitionNodeType_Resource:
		// Resource 节点：从资源获取字段列表
		var cfg interfaces.ResourceNodeCfg
		if err := mapstructure.Decode(node.Config, &cfg); err != nil {
			return nil, err
		}
		resource, err := rs.GetByID(ctx, cfg.ResourceID)
		if err != nil {
			return nil, err
		}
		for _, field := range resource.SchemaDefinition {
			fieldsMap[field.Name] = field
		}
	default:
		// 其他节点：从 output_fields 中获取字段列表
		for _, field := range node.OutputFields {
			if field.Name == "*" {
				// 通配符模式：需要从上游节点获取所有字段
				for _, inputID := range node.Inputs {
					// 递归获取输入节点的字段
					inputFieldsMap, err := getNodeOutputFieldsMap(ctx, rs, inputID, allNodes, nodeCache)
					if err != nil {
						return nil, err
					}
					for name, f := range inputFieldsMap {
						fieldsMap[name] = f
					}
				}
			} else {
				// 非通配符：直接使用字段定义
				prop := &interfaces.Property{
					Name:        field.Name,
					Type:        field.Type,
					DisplayName: field.DisplayName,
				}
				fieldsMap[field.Name] = prop
			}
		}
	}

	// 缓存结果
	nodeCache[nodeID] = fieldsMap
	return fieldsMap, nil
}

func validateResourceNode(ctx context.Context, dvs *resourceService, node *interfaces.LogicDefinitionNode,
	refResourceMap map[string]*interfaces.Resource) error {
	// 资源节点输入节点必须为空
	if len(node.Inputs) != 0 {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The resource node must have no input node")
	}

	var cfg interfaces.ResourceNodeCfg
	err := mapstructure.Decode(node.Config, &cfg)
	if err != nil {
		return rest.NewHTTPError(ctx, http.StatusInternalServerError, rest.PublicError_InternalServerError).
			WithErrorDetails(fmt.Sprintf("decode resource node config failed, %v", err))
	}

	// 判断自定义视图的来源表是否存在，从这个函数能够拿到字段列表
	atomicView, err := dvs.GetByID(ctx, cfg.ResourceID)
	if err != nil {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails(fmt.Sprintf("get resource %s failed, %v", cfg.ResourceID, err))
	}

	// 校验来源视图的类型
	switch atomicView.Category {
	case interfaces.ResourceCategoryTable:
	case interfaces.ResourceCategoryFile:
	case interfaces.ResourceCategoryFileset:
	case interfaces.ResourceCategoryAPI:
	case interfaces.ResourceCategoryTopic:
	case interfaces.ResourceCategoryIndex:
	default:
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails(fmt.Sprintf("The source resource of the custom view '%s' is not supported", cfg.ResourceID))
	}

	refResourceMap[atomicView.ID] = atomicView

	// fieldsMap 是字段name和字段的映射
	fieldsMap := make(map[string]*interfaces.Property)
	for _, viewField := range atomicView.SchemaDefinition {
		fieldsMap[viewField.Name] = viewField
	}

	// 校验过滤条件
	httpErr := validateCond(ctx, cfg.Filters, fieldsMap)
	if httpErr != nil {
		return httpErr
	}

	// 校验去重配置, 只有 table 去重配置
	if cfg.Distinct {
		if atomicView.Category != interfaces.ResourceCategoryTable {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("The logic definition view category is not table, distinct config is not supported")
		}
	}

	// 校验输出字段格式：resource节点支持通配符模式和投影模式
	for _, field := range node.OutputFields {
		// 通配符模式：只允许 "*"
		if field.Name == "*" {
			// 通配符模式下，不应有其他字段配置
			if field.Type != "" || field.From != "" || field.FromNode != "" || len(field.FromList) > 0 {
				return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
					WithErrorDetails("Wildcard field '*' should not have additional configuration")
			}
			continue
		}

		// 投影模式：只允许字段名，不应有映射或对齐配置
		if field.From != "" || field.FromNode != "" || len(field.FromList) > 0 {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("Resource node output field '%s' should not have from, from_node or from_list configuration", field.Name))
		}

		// 校验字段是否存在于资源字段列表中
		if _, ok := fieldsMap[field.Name]; !ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, rest.PublicError_BadRequest).
				WithErrorDetails(fmt.Sprintf("The field '%s' is not in the view '%s' field list", field.Name, atomicView.Name))
		}
	}

	return nil
}

func validateJoinNode(ctx context.Context, rs *resourceService, node *interfaces.LogicDefinitionNode,
	allNodes []*interfaces.LogicDefinitionNode, nodeMap map[string]struct{}) error {
	// 仅支持两个视图join
	if len(node.Inputs) != 2 {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The logic definition join config is invalid, only support two views join")
	}

	// 校验输入节点是否重复
	inputNodesMap := make(map[string]struct{})
	for _, inputNode := range node.Inputs {
		if _, ok := inputNodesMap[inputNode]; ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("The logic definition join config is invalid, inputs must be unique")
		}
		inputNodesMap[inputNode] = struct{}{}
	}

	// 校验输入节点是否存在
	for _, inputNode := range node.Inputs {
		if _, ok := nodeMap[inputNode]; !ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("The logic definition join config is invalid, input '%s' is not exist", inputNode))
		}
	}

	// mapstructure 解析 join_on
	var cfg interfaces.JoinNodeCfg
	err := mapstructure.Decode(node.Config, &cfg)
	if err != nil {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The logic definition join config is invalid")
	}

	// join_type 只能为 inner, left, right
	if _, ok := interfaces.JoinTypeMap[cfg.JoinType]; !ok {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_JoinType).
			WithErrorDetails("The logic definition join config is invalid, join_type must be inner, left, right")
	}

	// join_on 校验
	if len(cfg.JoinOn) == 0 {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The logic definition join config is invalid, join_on must be set")
	}

	// join_on 校验
	for _, joinOn := range cfg.JoinOn {
		if joinOn.LeftField == "" || joinOn.RightField == "" {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("The logic definition join config is invalid, join_on left_field and right_field must be set")
		}

		// 操作符必须只为=
		if joinOn.Operator != "=" {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("The logic definition join config is invalid, join_on operator must be =")
		}
	}

	// 校验输出字段不能为空
	if len(node.OutputFields) == 0 {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("Join node must have output fields")
	}

	// 校验输出字段格式：join节点只支持映射模式
	// 先获取所有输入节点的输出字段
	nodeCache := make(map[string]map[string]*interfaces.Property)
	inputFieldsMap := make(map[string]map[string]*interfaces.Property)
	for _, inputID := range node.Inputs {
		fieldsMap, err := getNodeOutputFieldsMap(ctx, rs, inputID, allNodes, nodeCache)
		if err != nil {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("Failed to get output fields from input node '%s': %v", inputID, err))
		}
		inputFieldsMap[inputID] = fieldsMap
	}

	// 校验每个输出字段
	for _, field := range node.OutputFields {
		// Join节点不支持通配符模式
		if field.Name == "*" {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("Join node does not support wildcard field '*'")
		}

		// 映射模式：必须指定 from 和 from_node
		if field.From == "" || field.FromNode == "" {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("Join node output field '%s' must have 'from' and 'from_node' configuration", field.Name))
		}

		// 映射模式：不应有 FromList 配置
		if len(field.FromList) > 0 {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("Join node output field '%s' should not have 'from_list' configuration", field.Name))
		}

		// 校验 from_node 是否在输入节点中
		found := false
		for _, inputNode := range node.Inputs {
			if inputNode == field.FromNode {
				found = true
				break
			}
		}
		if !found {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("Join node output field '%s' references non-existent input node '%s'", field.Name, field.FromNode))
		}

		// 校验 from 字段是否存在于源节点中
		if sourceFields, ok := inputFieldsMap[field.FromNode]; ok {
			if _, exists := sourceFields[field.From]; !exists {
				return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
					WithErrorDetails(fmt.Sprintf("Join node output field '%s' references non-existent field '%s' in node '%s'",
						field.Name, field.From, field.FromNode))
			}
		}
	}

	return nil
}

func validateUnionNode(ctx context.Context, rs *resourceService, category string, node *interfaces.LogicDefinitionNode,
	allNodes []*interfaces.LogicDefinitionNode, nodeMap map[string]struct{}) error {
	// 当前仅支持两个视图union
	if len(node.Inputs) < 2 {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The logic definition union config is invalid, need at least two views union")
	}

	// 校验输入节点是否重复
	inputNodesMap := make(map[string]struct{})
	for _, inputNode := range node.Inputs {
		if _, ok := inputNodesMap[inputNode]; ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("The logic definition union config is invalid, inputs must be unique")
		}
		inputNodesMap[inputNode] = struct{}{}
	}

	// 校验输入节点是否存在
	for _, inputNode := range node.Inputs {
		if _, ok := nodeMap[inputNode]; !ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("The logic definition union config is invalid, input '%s' is not exist", inputNode))
		}
	}

	// mapstructure 解析 union config
	var cfg interfaces.UnionNodeCfg
	err := mapstructure.Decode(node.Config, &cfg)
	if err != nil {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The logic definition union config is invalid")
	}

	if _, ok := interfaces.UnionTypeMap[cfg.UnionType]; !ok {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The logic definition union config is invalid, union_type must be all, distinct")
	}

	// 如果是索引resource，只允许union all
	if category == interfaces.ResourceCategoryIndex {
		if cfg.UnionType != interfaces.UnionType_All {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("The logic definition union config is invalid, DSL or IndexBase view only support union all")
		}
	}

	// 校验输出字段不能为空
	if len(node.OutputFields) == 0 {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("Union node must have output fields")
	}

	// 校验输出字段格式：union节点只支持对齐模式
	// 先获取所有输入节点的输出字段
	nodeCache := make(map[string]map[string]*interfaces.Property)
	inputFieldsMap := make(map[string]map[string]*interfaces.Property)
	for _, inputID := range node.Inputs {
		fieldsMap, err := getNodeOutputFieldsMap(ctx, rs, inputID, allNodes, nodeCache)
		if err != nil {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("Failed to get output fields from input node '%s': %v", inputID, err))
		}
		inputFieldsMap[inputID] = fieldsMap
	}

	for _, field := range node.OutputFields {
		// Union节点不支持通配符模式
		if field.Name == "*" {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("Union node does not support wildcard field '*'")
		}

		// 对齐模式：必须有 FromList 配置
		if len(field.FromList) == 0 {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("Union node output field '%s' must have 'from_list' configuration", field.Name))
		}

		// 对齐模式：不应有单独的 from 和 from_node 配置（除非在FromList中使用）
		if field.From != "" || field.FromNode != "" {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("Union node output field '%s' should not have 'from' or 'from_node' at field level, use 'from_list' instead", field.Name))
		}

		// 校验 FromList 长度是否与 inputs 长度一致
		if len(field.FromList) != len(node.Inputs) {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("The union output field '%s' from list count (%d) not equal inputs count (%d)",
					field.Name, len(field.FromList), len(node.Inputs)))
		}

		// 校验 FromList 中的每个引用是否都指向有效的输入节点和字段
		for _, ref := range field.FromList {
			found := false
			for _, inputNode := range node.Inputs {
				if inputNode == ref.FromNode {
					found = true
					break
				}
			}
			if !found {
				return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
					WithErrorDetails(fmt.Sprintf("Union node output field '%s' references non-existent input node '%s' in from_list", field.Name, ref.FromNode))
			}

			// 校验 from 字段是否存在于源节点中
			if ref.From != "" {
				if sourceFields, ok := inputFieldsMap[ref.FromNode]; ok {
					if _, exists := sourceFields[ref.From]; !exists {
						return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
							WithErrorDetails(fmt.Sprintf("Union node output field '%s' references non-existent field '%s' in node '%s'",
								field.Name, ref.From, ref.FromNode))
					}
				}
			}
		}
	}

	return nil
}

func validateSqlNode(ctx context.Context, rs *resourceService, node *interfaces.LogicDefinitionNode,
	allNodes []*interfaces.LogicDefinitionNode, nodeMap map[string]struct{}) error {
	// 输入节点不能为空
	if len(node.Inputs) == 0 {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The logic definition sql config is invalid, inputs must be set")
	}

	// 校验输入节点是否重复
	inputNodesMap := make(map[string]struct{})
	for _, inputNode := range node.Inputs {
		if _, ok := inputNodesMap[inputNode]; ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("The logic definition sql config is invalid, inputs must be unique")
		}
		inputNodesMap[inputNode] = struct{}{}
	}

	// 校验输入节点是否存在
	for _, inputNode := range node.Inputs {
		if _, ok := nodeMap[inputNode]; !ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("The logic definition sql config is invalid, input '%s' is not exist", inputNode))
		}
	}

	// mapstructure 解析 sql config
	var cfg interfaces.SQLNodeCfg
	err := mapstructure.Decode(node.Config, &cfg)
	if err != nil {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The logic definition sql config is invalid")
	}

	// 校验 sql 是否为空
	if cfg.SQL == "" {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The logic definition sql config is invalid, sql must be set")
	}

	// 校验 SQL 语法是否正确
	if err := validateSQLSyntax(ctx, cfg.SQL); err != nil {
		return err
	}

	// 校验输出字段不能为空
	if len(node.OutputFields) == 0 {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("SQL node must have output fields")
	}

	// 校验输出字段格式：sql节点支持定义模式和通配符模式
	for _, field := range node.OutputFields {
		// 通配符模式：只允许 "*"
		if field.Name == "*" {
			// 通配符模式下，不应有其他字段配置（但允许 type 用于类型推断）
			if field.From != "" || field.FromNode != "" || len(field.FromList) > 0 {
				return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
					WithErrorDetails("Wildcard field '*' in SQL node should not have from, from_node or from_list configuration")
			}
			continue
		}

		// 定义模式：不应有映射或对齐配置（SQL节点自行定义字段）
		if field.From != "" || field.FromNode != "" || len(field.FromList) > 0 {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("SQL node output field '%s' should not have from, from_node or from_list configuration", field.Name))
		}
	}

	return nil
}

func validateOutputNode(ctx context.Context, rs *resourceService, node *interfaces.LogicDefinitionNode,
	allNodes []*interfaces.LogicDefinitionNode, nodeMap map[string]struct{}) error {
	// 输入节点只能有一个
	if len(node.Inputs) != 1 {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The output node must have one input node")
	}

	// 校验输入节点是否存在
	inputNode := node.Inputs[0]
	if _, ok := nodeMap[inputNode]; !ok {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails(fmt.Sprintf("The output node input '%s' is not exist", inputNode))
	}

	if len(node.OutputFields) == 0 {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("The output node must have output fields")
	}

	// 校验输出字段格式：output节点支持通配符模式和投影模式
	// 获取输入节点的输出字段
	nodeCache := make(map[string]map[string]*interfaces.Property)
	inputNodeID := node.Inputs[0]
	inputFieldsMap, err := getNodeOutputFieldsMap(ctx, rs, inputNodeID, allNodes, nodeCache)
	if err != nil {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails(fmt.Sprintf("Failed to get output fields from input node '%s': %v", inputNodeID, err))
	}

	for _, field := range node.OutputFields {
		// 通配符模式：只允许 "*"
		if field.Name == "*" {
			// 通配符模式下，不应有其他字段配置
			if field.Type != "" || field.From != "" || field.FromNode != "" || len(field.FromList) > 0 {
				return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
					WithErrorDetails("Wildcard field '*' should not have additional configuration")
			}
			continue
		}

		// 投影模式：只允许字段名，不应有映射或对齐配置
		if field.From != "" || field.FromNode != "" || len(field.FromList) > 0 {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("Output node field '%s' should not have from, from_node or from_list configuration", field.Name))
		}

		// 校验字段是否存在于输入节点中
		if _, ok := inputFieldsMap[field.Name]; !ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("Output node field '%s' is not in the input node '%s' output fields", field.Name, inputNodeID))
		}
	}

	// 校验name不能重复，display_name 不能重复
	nameMap := make(map[string]struct{})
	displayNameMap := make(map[string]struct{})
	for _, field := range node.OutputFields {
		if _, ok := nameMap[field.Name]; ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("The output node field name is repeated")
		}
		nameMap[field.Name] = struct{}{}

		if _, ok := displayNameMap[field.DisplayName]; ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails("The output node field display_name is repeated")
		}
		displayNameMap[field.DisplayName] = struct{}{}
	}

	return nil
}

// 相比handler层的校验，补充对过滤条件字段类型的校验
func validateCond(ctx context.Context, cfg *interfaces.FilterCondCfg, fieldsMap map[string]*interfaces.Property) error {
	if cfg == nil {
		return nil
	}

	// 判断过滤器是否为空对象 {}
	if cfg.Name == "" && cfg.Operation == "" && len(cfg.SubConds) == 0 && cfg.ValueFrom == "" && cfg.Value == nil {
		return nil
	}

	// 过滤条件字段不允许 __id 和 __routing
	if cfg.Name == "__id" || cfg.Name == "__routing" {
		return rest.NewHTTPError(ctx, http.StatusForbidden, verrors.VegaBackend_InvalidParameter_FilterCondition).
			WithErrorDetails("The filter field '__id' and '__routing' is not allowed")
	}

	// 过滤操作符
	if cfg.Operation == "" {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_NullParameter_FilterConditionOperation)
	}

	_, exists := fcond.OperationMap[cfg.Operation]
	if !exists {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_UnsupportFilterConditionOperation).
			WithErrorDetails(fmt.Sprintf("unsupport condition operation %s", cfg.Operation))
	}

	switch cfg.Operation {
	case fcond.OperationAnd, fcond.OperationOr:
		// 子过滤条件不能超过10个
		if len(cfg.SubConds) > interfaces.MaxSubCondition {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_CountExceeded_FilterConditionSubConds).
				WithErrorDetails(fmt.Sprintf("The number of subConditions exceeds %d", interfaces.MaxSubCondition))
		}

		for _, subCond := range cfg.SubConds {
			err := validateCond(ctx, subCond, fieldsMap)
			if err != nil {
				return err
			}
		}
	default:
		// 过滤字段名称不能为空
		if cfg.Name == "" {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_NullParameter_FilterConditionName)
		}
	}

	switch cfg.Operation {
	case fcond.OperationEqual, fcond.OperationNotEqual, fcond.OperationGt, fcond.OperationGte,
		fcond.OperationLt, fcond.OperationLte, fcond.OperationLike, fcond.OperationNotLike,
		fcond.OperationRegex, fcond.OperationMatch, fcond.OperationMatchPhrase, fcond.OperationCurrent:
		// 右侧值为单个值
		_, ok := cfg.Value.([]interface{})
		if ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterConditionValue).
				WithErrorDetails(fmt.Sprintf("[%s] operation's value should be a single value", cfg.Operation))
		}

		if cfg.Operation == fcond.OperationLike || cfg.Operation == fcond.OperationNotLike ||
			cfg.Operation == fcond.OperationPrefix || cfg.Operation == fcond.OperationNotPrefix {
			_, ok := cfg.Value.(string)
			if !ok {
				return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterConditionValue).
					WithErrorDetails("[like not_like prefix not_prefix] operation's value should be a string")
			}
		}

		if cfg.Operation == fcond.OperationRegex {
			val, ok := cfg.Value.(string)
			if !ok {
				return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterConditionValue).
					WithErrorDetails("[regex] operation's value should be a string")
			}

			_, err := regexp2.Compile(val, regexp2.RE2)
			if err != nil {
				return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterConditionValue).
					WithErrorDetails(fmt.Sprintf("[regex] operation regular expression error: %s", err.Error()))
			}

		}

	case fcond.OperationIn, fcond.OperationNotIn:
		// 当 operation 是 in, not_in 时，value 为任意基本类型的数组，且长度大于等于1；
		_, ok := cfg.Value.([]interface{})
		if !ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterConditionValue).
				WithErrorDetails("[in not_in] operation's value must be an array")
		}

		if len(cfg.Value.([]interface{})) <= 0 {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterConditionValue).
				WithErrorDetails("[in not_in] operation's value should contains at least 1 value")
		}
	case fcond.OperationRange, fcond.OperationOutRange, fcond.OperationBetween:
		// 当 operation 是 range 时，value 是个由范围的下边界和上边界组成的长度为 2 的数值型数组
		// 当 operation 是 out_range 时，value 是个长度为 2 的数值类型的数组，查询的数据范围为 (-inf, value[0]) || [value[1], +inf)
		v, ok := cfg.Value.([]interface{})
		if !ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterConditionValue).
				WithErrorDetails("[range, out_range, between] operation's value must be an array")
		}

		if len(v) != 2 {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterConditionValue).
				WithErrorDetails("[range, out_range, between] operation's value must contain 2 values")
		}
	case fcond.OperationBefore:
		// before时, 长度为2的数组，第一个值为时间长度，数值型；第二个值为时间单位，字符串
		_, ok := cfg.Value.(float64)
		if !ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterConditionValue).
				WithErrorDetails("[before] operation's value must be an array")
		}

		_, ok = cfg.RemainCfg["unit"]
		if !ok {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterConditionValue).
				WithErrorDetails("[before] operation's remain cfg must contain unit")
		}
	}

	switch cfg.Operation {
	case fcond.OperationAnd, fcond.OperationOr:
		for _, subCond := range cfg.SubConds {
			err := validateCond(ctx, subCond, fieldsMap)
			if err != nil {
				return err
			}
		}
	default:
		// 除 * 之外的过滤字段在视图字段列表里
		if cfg.Name != interfaces.AllField {
			cField, ok := fieldsMap[cfg.Name]
			if !ok {
				return rest.NewHTTPError(ctx, http.StatusForbidden, verrors.VegaBackend_InvalidParameter_FilterCondition).
					WithErrorDetails(fmt.Sprintf("Filter field '%s' is not in view fields list", cfg.Name))
			}

			fieldType := cField.Type
			// binary 类型的字段不支持过滤
			if fieldType == interfaces.DataType_Binary {
				return rest.NewHTTPError(ctx, http.StatusForbidden, verrors.VegaBackend_InvalidParameter_FilterCondition).
					WithErrorDetails("Binary fields do not support filtering")
			}

			// empty, not_empty 的字段类型必须为 string
			if cfg.Operation == fcond.OperationEmpty || cfg.Operation == fcond.OperationNotEmpty {
				if !interfaces.DataType_IsString(fieldType) {
					return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterCondition).
						WithErrorDetails("Filter field must be of string type when using 'empty' or 'not_empty' operation")
				}
			}
		} else {
			// 如果字段为 *，则只允许使用 match 和 match_phrase 操作符
			if cfg.Operation != fcond.OperationMatch && cfg.Operation != fcond.OperationMatchPhrase &&
				cfg.Operation != fcond.OperationMultiMatch {
				return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_InvalidParameter_FilterCondition).
					WithErrorDetails("Filter field '*' only supports 'match', 'match_phrase' and 'multi_match' operations")
			}
		}
	}

	return nil
}

// 解析 logicDefinition，生成 schemaDefinition
func (rs *resourceService) parseLogicDefinition(ctx context.Context,
	logicDefinition []*interfaces.LogicDefinitionNode) ([]*interfaces.Property, error) {

	// 1. 构建节点映射表
	nodes := make(map[string]*interfaces.LogicDefinitionNode)
	for _, node := range logicDefinition {
		nodes[node.ID] = node
	}

	// 2. 找到终端输出节点 (output 节点)
	var outputNode *interfaces.LogicDefinitionNode
	for _, node := range logicDefinition {
		if node.Type == interfaces.LogicDefinitionNodeType_Output {
			outputNode = node
			break
		}
	}

	if outputNode == nil {
		// 如果没显式定义 output 节点，兜底取最后一个节点
		if len(logicDefinition) > 0 {
			outputNode = logicDefinition[len(logicDefinition)-1]
		} else {
			return nil, fmt.Errorf("logic definition is empty")
		}
	}

	// 3. 递归解析字段元数据 (带缓存避免重复计算)
	memo := make(map[string][]*interfaces.Property)
	var resolve func(nodeID string) ([]*interfaces.Property, error)
	resolve = func(nodeID string) ([]*interfaces.Property, error) {
		if cached, ok := memo[nodeID]; ok {
			return cached, nil
		}

		node, ok := nodes[nodeID]
		if !ok {
			return nil, fmt.Errorf("node %s not found in logic definition", nodeID)
		}

		var result []*interfaces.Property
		var inputFieldsMap = make(map[string][]*interfaces.Property)
		var sourceResourceFields []*interfaces.Property

		// 处理叶子节点：Resource 节点
		if node.Type == interfaces.LogicDefinitionNodeType_Resource {
			var cfg interfaces.ResourceNodeCfg
			if err := mapstructure.Decode(node.Config, &cfg); err != nil {
				return nil, fmt.Errorf("decode resource node config failed: %w", err)
			}
			res, err := rs.GetByID(ctx, cfg.ResourceID)
			if err != nil {
				return nil, fmt.Errorf("get resource %s failed: %w", cfg.ResourceID, err)
			}
			sourceResourceFields = res.SchemaDefinition
		} else {
			// 解析所有输入节点的输出字段
			for _, inputID := range node.Inputs {
				fields, err := resolve(inputID)
				if err != nil {
					return nil, err
				}
				inputFieldsMap[inputID] = fields
			}
		}

		// 处理当前节点的 output_fields
		for _, vProp := range node.OutputFields {
			if vProp.Name == "*" {
				// 通配符模式：全量透传上游字段
				if node.Type == interfaces.LogicDefinitionNodeType_Resource {
					for _, f := range sourceResourceFields {
						result = append(result, copyProperty(f))
					}
				} else {
					for _, inputID := range node.Inputs {
						for _, f := range inputFieldsMap[inputID] {
							result = append(result, copyProperty(f))
						}
					}
				}
				continue
			}

			// 投影/映射/对齐/定义模式：构造 Property
			prop := &interfaces.Property{
				Name:         vProp.Name,
				Type:         vProp.Type,
				DisplayName:  vProp.DisplayName,
				OriginalName: vProp.OriginalName,
				Description:  vProp.Description,
				Features:     vProp.Features,
			}

			// 递归溯源补全元数据 (Type, DisplayName, Description, OriginalName, Features)
			var sourceProp *interfaces.Property
			if node.Type == interfaces.LogicDefinitionNodeType_Resource {
				// Resource 节点从物理 Schema 中找
				for _, f := range sourceResourceFields {
					if f.Name == vProp.Name {
						sourceProp = f
						break
					}
				}
			} else if vProp.From != "" && vProp.FromNode != "" {
				// 映射模式 (Join)：明确指定了来源节点和字段
				if sFields, ok := inputFieldsMap[vProp.FromNode]; ok {
					for _, f := range sFields {
						if f.Name == vProp.From {
							sourceProp = f
							break
						}
					}
				}
			} else if len(vProp.FromList) > 0 {
				// 对齐模式 (Union)：从匹配的第一个来源节点取元数据
				for _, ref := range vProp.FromList {
					if sFields, ok := inputFieldsMap[ref.FromNode]; ok {
						for _, f := range sFields {
							if f.Name == ref.From {
								sourceProp = f
								break
							}
						}
					}
					if sourceProp != nil {
						break
					}
				}
			} else {
				// 投影模式/SQL定义：按名称在上游输入中查找
				for _, inputID := range node.Inputs {
					if sFields, ok := inputFieldsMap[inputID]; ok {
						for _, f := range sFields {
							if f.Name == vProp.Name {
								sourceProp = f
								break
							}
						}
					}
					if sourceProp != nil {
						break
					}
				}
			}

			// 如果找到了源字段，则补全缺失的信息
			if sourceProp != nil {
				fillMissingMetadata(prop, sourceProp)
			}
			result = append(result, prop)
		}

		memo[nodeID] = result
		return result, nil
	}

	return resolve(outputNode.ID)
}

func copyProperty(p *interfaces.Property) *interfaces.Property {
	if p == nil {
		return nil
	}
	cp := *p
	if len(p.Features) > 0 {
		cp.Features = make([]interfaces.PropertyFeature, len(p.Features))
		copy(cp.Features, p.Features)
	}
	return &cp
}

func fillMissingMetadata(target, source *interfaces.Property) {
	if target.Type == "" {
		target.Type = source.Type
	}
	if target.DisplayName == "" {
		target.DisplayName = source.DisplayName
	}
	if target.Description == "" {
		target.Description = source.Description
	}
	if target.OriginalName == "" {
		target.OriginalName = source.OriginalName
	}
	if len(target.Features) == 0 {
		target.Features = source.Features
	}
}

// validateSQLSyntax 校验 SQL 语法是否正确
// 1. 先将 SQL 中的变量（如 .node1）替换为占位符
// 2. 再使用标准 SQL 语法规则校验
func validateSQLSyntax(ctx context.Context, sql string) error {
	if sql == "" {
		return nil // 空 SQL 已在前面的校验中处理
	}

	// 步骤 1: 替换 SQL 中的变量（如 .node1, .node2 等）为占位符
	// 匹配模式：点后跟标识符，例如 .node1, .my_table
	nodeVarRegex := regexp.MustCompile(`\.[a-zA-Z_][a-zA-Z0-9_]*`)
	cleanedSQL := nodeVarRegex.ReplaceAllString(sql, " placeholder_table ")

	// 步骤 2: 标准 SQL 语法校验
	// 2.1 检查是否以 SELECT 或 WITH 开头
	trimmedSQL := strings.TrimSpace(strings.ToUpper(cleanedSQL))
	if !strings.HasPrefix(trimmedSQL, "SELECT") && !strings.HasPrefix(trimmedSQL, "WITH") {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("SQL must start with SELECT or WITH clause")
	}

	// 2.2 检查括号是否匹配
	openParen := strings.Count(cleanedSQL, "(")
	closeParen := strings.Count(cleanedSQL, ")")
	if openParen != closeParen {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails(fmt.Sprintf("Unbalanced parentheses: %d opening vs %d closing", openParen, closeParen))
	}

	// 2.3 检查常见的语法错误
	if err := checkCommonSQLErrors(ctx, cleanedSQL); err != nil {
		return err
	}

	return nil
}

// checkCommonSQLErrors 检查常见的 SQL 语法错误
func checkCommonSQLErrors(ctx context.Context, sql string) error {
	upperSQL := strings.ToUpper(sql)
	trimmedSQL := strings.TrimSpace(sql)

	// 检查重复的关键字
	duplicatePatterns := []struct {
		pattern *regexp.Regexp
		message string
	}{
		{regexp.MustCompile(`\bFROM\s+FROM\b`), "Duplicate FROM keyword"},
		{regexp.MustCompile(`\bSELECT\s+SELECT\b`), "Duplicate SELECT keyword"},
		{regexp.MustCompile(`\bWHERE\s+WHERE\b`), "Duplicate WHERE keyword"},
		{regexp.MustCompile(`\bJOIN\s+JOIN\b`), "Duplicate JOIN keyword"},
		{regexp.MustCompile(`\bGROUP\s+BY\s+BY\b`), "Duplicate BY in GROUP BY"},
		{regexp.MustCompile(`\bORDER\s+BY\s+BY\b`), "Duplicate BY in ORDER BY"},
	}

	for _, dp := range duplicatePatterns {
		if dp.pattern.MatchString(upperSQL) {
			return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
				WithErrorDetails(fmt.Sprintf("SQL syntax error: %s", dp.message))
		}
	}

	// 检查 FROM 后是否有表名（优先检查）
	fromWithoutTable := regexp.MustCompile(`(?i)\bFROM\s*$`)
	if fromWithoutTable.MatchString(trimmedSQL) {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("SQL syntax error: FROM clause must specify a table")
	}

	// 检查 SELECT 后是否有 FROM（简单检查）
	if strings.HasPrefix(upperSQL, "SELECT") {
		// 检查是否包含 FROM 关键字
		if !strings.Contains(upperSQL, " FROM ") && !strings.HasSuffix(upperSQL, " FROM") {
			// 检查是否是 SELECT * 或 SELECT 1 这种简单形式（不含 FROM）
			simpleSelectRegex := regexp.MustCompile(`(?i)^SELECT\s+[*\d]+\s*$`)
			if !simpleSelectRegex.MatchString(trimmedSQL) {
				return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
					WithErrorDetails("SQL syntax error: SELECT statement must contain a FROM clause")
			}
		}
	}

	// 检查 WHERE 后是否有条件
	whereWithoutCondition := regexp.MustCompile(`(?i)\bWHERE\s*$`)
	if whereWithoutCondition.MatchString(trimmedSQL) {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("SQL syntax error: WHERE clause must have a condition")
	}

	// 检查 GROUP BY 后是否有列名
	groupByWithoutColumn := regexp.MustCompile(`(?i)\bGROUP\s+BY\s*$`)
	if groupByWithoutColumn.MatchString(trimmedSQL) {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("SQL syntax error: GROUP BY must have at least one column")
	}

	// 检查 ORDER BY 后是否有列名
	orderByWithoutColumn := regexp.MustCompile(`(?i)\bORDER\s+BY\s*$`)
	if orderByWithoutColumn.MatchString(trimmedSQL) {
		return rest.NewHTTPError(ctx, http.StatusBadRequest, verrors.VegaBackend_LogicView_InvalidParameter_LogicDefinition).
			WithErrorDetails("SQL syntax error: ORDER BY must have at least one column")
	}

	return nil
}
