ragflow/internal/engine/oceanbase/rows.go

//
//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the License.
//  You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
//

package oceanbase

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"
)

func (e *Engine) queryRows(ctx context.Context, query string, args ...interface{}) ([]map[string]interface{}, error) {
	rows, err := e.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	columns, err := rows.Columns()
	if err != nil {
		return nil, err
	}
	result := make([]map[string]interface{}, 0)
	for rows.Next() {
		values := make([]interface{}, len(columns))
		destinations := make([]interface{}, len(columns))
		for i := range values {
			destinations[i] = &values[i]
		}
		if err := rows.Scan(destinations...); err != nil {
			return nil, err
		}
		row := make(map[string]interface{}, len(columns))
		for i, column := range columns {
			if values[i] == nil {
				continue
			}
			if raw, ok := values[i].([]byte); ok {
				row[column] = string(raw)
			} else {
				row[column] = values[i]
			}
		}
		result = append(result, row)
	}
	return result, rows.Err()
}

func decodeLogicalRow(row map[string]interface{}, kind string) map[string]interface{} {
	result := make(map[string]interface{}, len(row))
	for rawColumn, value := range row {
		column := rawColumn
		if kind == "memory" {
			if vectorColumnPattern.MatchString(column) {
				column = "content_embed"
			} else if mapped, ok := memoryColumnToField[column]; ok {
				column = mapped
			}
		}
		if kind == "chunk" && column == "_order_id" {
			column = "chunk_order_int"
		}
		if kind == "memory" && column == "status" {
			switch status := value.(type) {
			case int64:
				value = status != 0
			case int:
				value = status != 0
			case string:
				value = status != "" && status != "0"
			}
		}
		storedColumn := rawColumn
		if (kind == "chunk" && arrayColumns[storedColumn]) || jsonColumns[storedColumn] || vectorColumnPattern.MatchString(storedColumn) {
			if text, ok := value.(string); ok {
				var decoded interface{}
				if err := json.Unmarshal([]byte(text), &decoded); err == nil {
					value = decoded
				}
			}
		}
		result[column] = value
	}
	return result
}

func scanCount(row *sql.Row) (int64, error) {
	var count int64
	if err := row.Scan(&count); err != nil {
		return 0, err
	}
	return count, nil
}

type selectField struct {
	column string
	alias  string
}

func (field selectField) expression(tableAlias string) string {
	prefix := ""
	if tableAlias != "" {
		prefix = quoteIdentifier(tableAlias) + "."
	}
	expression := prefix + quoteIdentifier(field.column)
	if field.alias != field.column {
		expression += " AS " + quoteIdentifier(field.alias)
	}
	return expression
}

func parseSelectField(field, kind string) (*selectField, error) {
	if field == "_score" {
		return nil, nil
	}
	if field == "row_id()" || field == "row_id" {
		return &selectField{column: identifierField(kind), alias: "row_id"}, nil
	}
	column := field
	if kind == "memory" {
		column = mapMemoryField(field)
	}
	if kind == "chunk" && field == "chunk_order_int" {
		column = "_order_id"
	}
	if field == "content_embed" && kind == "memory" {
		return nil, nil
	}
	if !vectorColumnPattern.MatchString(column) && !validColumns(kind)[column] {
		return nil, fmt.Errorf("unknown %s field: %s", kind, field)
	}
	return &selectField{column: column, alias: field}, nil
}

func selectExpression(field, kind string) (string, string, error) {
	parsed, err := parseSelectField(field, kind)
	if err != nil || parsed == nil {
		return "", "", err
	}
	return parsed.expression(""), parsed.alias, nil
}

func buildSelectFields(fields []string, kind string) (string, []string, error) {
	return buildSelectFieldsWithAlias(fields, kind, "")
}

func buildQualifiedSelectFields(fields []string, kind, tableAlias string) (string, []string, error) {
	return buildSelectFieldsWithAlias(fields, kind, tableAlias)
}

func buildSelectFieldsWithAlias(fields []string, kind, tableAlias string) (string, []string, error) {
	if len(fields) == 0 || containsString(fields, "*") {
		fields = defaultFields(kind)
	}
	if !containsString(fields, identifierField(kind)) {
		fields = append([]string{identifierField(kind)}, fields...)
	}
	expressions := make([]string, 0, len(fields))
	aliases := make([]string, 0, len(fields))
	seen := make(map[string]bool)
	for _, field := range fields {
		parsed, err := parseSelectField(field, kind)
		if err != nil {
			return "", nil, err
		}
		if parsed == nil || seen[parsed.alias] {
			continue
		}
		seen[parsed.alias] = true
		expressions = append(expressions, parsed.expression(tableAlias))
		aliases = append(aliases, parsed.alias)
	}
	return strings.Join(expressions, ", "), aliases, nil
}

func defaultFields(kind string) []string {
	var columns []columnDefinition
	switch kind {
	case "memory":
		columns = memoryColumns
	case "metadata":
		columns = metadataColumns
	case "skill":
		columns = skillColumns
	default:
		columns = chunkColumns
	}
	fields := make([]string, 0, len(columns))
	for _, column := range columns {
		field := column.name
		if kind == "memory" {
			if mapped, ok := memoryColumnToField[field]; ok {
				field = mapped
			}
		}
		if kind == "chunk" && field == "_order_id" {
			field = "chunk_order_int"
		}
		fields = append(fields, field)
	}
	return fields
}

func identifierField(kind string) string {
	if kind == "skill" {
		return "skill_id"
	}
	return "id"
}

func containsString(values []string, target string) bool {
	for _, value := range values {
		if value == target {
			return true
		}
	}
	return false
}