SqlServerCommandBuilder.cs

// Copyright (c) Microsoft. All rights reserved.

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq.Expressions;
using System.Text;
using System.Text.Json;
using Microsoft.Data.SqlClient;
using Microsoft.Data.SqlTypes;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.VectorData;
using Microsoft.Extensions.VectorData.ProviderServices;

#pragma warning disable CA2100 // Review SQL queries for security vulnerabilities

namespace Microsoft.SemanticKernel.Connectors.SqlServer;

internal static class SqlServerCommandBuilder
{
    internal static List<SqlCommand> CreateTable(
        SqlConnection connection,
        string? schema,
        string tableName,
        bool ifNotExists,
        CollectionModel model)
    {
        List<SqlCommand> commands = [];

        StringBuilder sb = new(200);
        if (ifNotExists)
        {
            sb.Append("IF OBJECT_ID(N'");
            sb.AppendTableName(schema, tableName);
            sb.AppendLine("', N'U') IS NULL");
        }
        sb.AppendLine("BEGIN");
        sb.Append("CREATE TABLE ");
        sb.AppendTableName(schema, tableName);
        sb.AppendLine(" (");

        var keyStoreType = Map(model.KeyProperty);
        sb.AppendIdentifier(model.KeyProperty.StorageName).Append(' ').Append(keyStoreType);
        if (model.KeyProperty.IsAutoGenerated)
        {
            switch (keyStoreType.ToUpperInvariant())
            {
                case "SMALLINT":
                case "INT":
                case "BIGINT":
                    sb.Append(" IDENTITY");
                    break;
                case "UNIQUEIDENTIFIER":
                    sb.Append(" DEFAULT NEWSEQUENTIALID()");
                    break;
                default:
                    throw new UnreachableException();
            }
        }

        sb.AppendLine(",");

        foreach (var property in model.DataProperties)
        {
            sb.AppendIdentifier(property.StorageName).Append(' ').Append(Map(property));
            if (!property.IsNullable)
            {
                sb.Append(" NOT NULL");
            }
            sb.AppendLine(",");
        }

        foreach (var property in model.VectorProperties)
        {
            sb.AppendIdentifier(property.StorageName).Append(" VECTOR(").Append(property.Dimensions).Append(')');
            if (!property.IsNullable)
            {
                sb.Append(" NOT NULL");
            }
            sb.AppendLine(",");
        }

        sb.Append("PRIMARY KEY (").AppendIdentifier(model.KeyProperty.StorageName).AppendLine(")");
        sb.AppendLine(");"); // end the table definition

        foreach (var dataProperty in model.DataProperties)
        {
            if (dataProperty.IsIndexed)
            {
                var sqlType = Map(dataProperty);
                if (sqlType == "JSON")
                {
                    sb.Append("CREATE JSON INDEX ");
                }
                else
                {
                    sb.Append("CREATE INDEX ");
                }
                sb.AppendIndexName(tableName, dataProperty.StorageName);
                sb.Append(" ON ").AppendTableName(schema, tableName);
                sb.Append('(').AppendIdentifier(dataProperty.StorageName).AppendLine(");");
            }
        }

        // Create full-text catalog and index for properties marked as IsFullTextIndexed
        var fullTextProperties = new List<DataPropertyModel>();
        foreach (var dataProperty in model.DataProperties)
        {
            if (dataProperty.IsFullTextIndexed)
            {
                fullTextProperties.Add(dataProperty);
            }
        }

        if (fullTextProperties.Count > 0)
        {
            // Generate a unique catalog name based on the table name
            var catalogName = $"ftcat_{tableName}".Replace(" ", "_");

            // Create full-text catalog if it doesn't exist
            sb.Append("IF NOT EXISTS (SELECT 1 FROM sys.fulltext_catalogs WHERE name = '").Append(catalogName.Replace("'", "''")).AppendLine("')");
            sb.Append("    CREATE FULLTEXT CATALOG ").AppendIdentifier(catalogName).AppendLine(";");

            // Create full-text index on the table using dynamic SQL to look up the PK constraint name
            // Full-text indexes require a unique index (we use the primary key)
            sb.AppendLine("DECLARE @pkIndexName NVARCHAR(128);");
            sb.Append("SELECT @pkIndexName = name FROM sys.indexes WHERE object_id = OBJECT_ID(N'");
            sb.AppendTableName(schema, tableName);
            sb.AppendLine("') AND is_primary_key = 1;");

            sb.AppendLine("DECLARE @ftSql NVARCHAR(MAX);");
            sb.Append("SET @ftSql = N'CREATE FULLTEXT INDEX ON ");
            sb.AppendTableName(schema, tableName).Append(" (");
            for (int i = 0; i < fullTextProperties.Count; i++)
            {
                sb.AppendIdentifier(fullTextProperties[i].StorageName);
                if (i < fullTextProperties.Count - 1)
                {
                    sb.Append(',');
                }
            }
            sb.Append(") KEY INDEX ' + QUOTENAME(@pkIndexName) + N' ON ");
            sb.AppendIdentifier(catalogName).AppendLine("';");
            sb.AppendLine("EXEC sp_executesql @ftSql;");
        }

        sb.Append("END;");

        commands.Add(connection.CreateCommand(sb));

        // CREATE VECTOR INDEX must be in a separate batch from CREATE TABLE.
        // It is also a preview feature in SQL Server 2025, requiring PREVIEW_FEATURES to be enabled.
        bool hasVectorIndex = false;
        foreach (var vectorProperty in model.VectorProperties)
        {
            switch (vectorProperty.IndexKind)
            {
                case IndexKind.Flat or null or "":
                    continue;

                case IndexKind.DiskAnn:
                    if (!hasVectorIndex)
                    {
                        SqlCommand enablePreview = connection.CreateCommand();
                        enablePreview.CommandText = "ALTER DATABASE SCOPED CONFIGURATION SET PREVIEW_FEATURES = ON;";
                        commands.Add(enablePreview);
                        hasVectorIndex = true;
                    }

                    string distanceFunction = vectorProperty.DistanceFunction ?? DistanceFunction.CosineDistance;
                    (string distanceMetric, _) = MapDistanceFunction(distanceFunction);

                    StringBuilder vectorIndexSb = new(200);
                    vectorIndexSb.Append("CREATE VECTOR INDEX ");
                    vectorIndexSb.AppendIndexName(tableName, vectorProperty.StorageName);
                    vectorIndexSb.Append(" ON ").AppendTableName(schema, tableName);
                    vectorIndexSb.Append('(').AppendIdentifier(vectorProperty.StorageName).Append(')');
                    vectorIndexSb.Append(" WITH (METRIC = '").Append(distanceMetric).AppendLine("', TYPE = 'DISKANN');");
                    commands.Add(connection.CreateCommand(vectorIndexSb));
                    break;

                default:
                    throw new NotSupportedException($"Index kind '{vectorProperty.IndexKind}' is not supported by the SQL Server connector.");
            }
        }

        return commands;
    }

    internal static SqlCommand DropTableIfExists(SqlConnection connection, string? schema, string tableName)
    {
        StringBuilder sb = new(50);
        sb.Append("DROP TABLE IF EXISTS ");
        sb.AppendTableName(schema, tableName);

        return connection.CreateCommand(sb);
    }

    internal static SqlCommand SelectTableName(SqlConnection connection, string? schema, string tableName)
    {
        SqlCommand command = connection.CreateCommand();
        command.CommandText = """
                SELECT TABLE_NAME
                FROM INFORMATION_SCHEMA.TABLES
                WHERE TABLE_TYPE = 'BASE TABLE'
                    AND (@schema is NULL or TABLE_SCHEMA = @schema)
                    AND TABLE_NAME = @tableName
                """;
        command.Parameters.AddWithValue("@schema", string.IsNullOrEmpty(schema) ? DBNull.Value : schema);
        command.Parameters.AddWithValue("@tableName", tableName); // the name is not escaped by us, just provided as parameter
        return command;
    }

    internal static SqlCommand SelectTableNames(SqlConnection connection, string? schema)
    {
        SqlCommand command = connection.CreateCommand();
        command.CommandText = """
                SELECT TABLE_NAME
                FROM INFORMATION_SCHEMA.TABLES
                WHERE TABLE_TYPE = 'BASE TABLE'
                    AND (@schema is NULL or TABLE_SCHEMA = @schema)
                """;
        command.Parameters.AddWithValue("@schema", string.IsNullOrEmpty(schema) ? DBNull.Value : schema);
        return command;
    }

    /// <summary>
    /// Checks if the key property uses SQL Server IDENTITY (for int/bigint) as opposed to DEFAULT (for GUID).
    /// IDENTITY columns require SET IDENTITY_INSERT ON to insert explicit values.
    /// </summary>
    private static bool UsesIdentity(KeyPropertyModel keyProperty)
    {
        if (!keyProperty.IsAutoGenerated)
        {
            return false;
        }

        var keyStoreType = Map(keyProperty).ToUpperInvariant();
        return keyStoreType is "SMALLINT" or "INT" or "BIGINT";
    }

    // Note: since keys may be auto-generated, we can't use a single multi-value MERGE statement, since that would return
    // the generated keys in undefined order (OUTPUT order is not guaranteed in MERGE).
    // Use a batch of single-row MERGE statements instead - each returns a separate result set.
    internal static bool Upsert<TKey>(
        SqlCommand command,
        string? schema,
        string tableName,
        CollectionModel model,
        IEnumerable<object> records,
        int firstRecordIndex,
        Dictionary<VectorPropertyModel, IReadOnlyList<Embedding>>? generatedEmbeddings)
    {
        var keyProperty = model.KeyProperty;
        StringBuilder sb = new(500);

        int rowIndex = 0, paramIndex = 0;

        foreach (var record in records)
        {
            // A record needs auto-generation if the key property is auto-generated AND the record has a default key value.
            var needsKeyGeneration = keyProperty.IsAutoGenerated && Equals(keyProperty.GetValueAsObject(record), default(TKey));
            // Skip key in INSERT when auto-generating (IDENTITY will provide the value)
            var skipKeyInInsert = needsKeyGeneration;
            // For explicit keys with IDENTITY columns, we need to enable IDENTITY_INSERT
            // (only for int/bigint, not for GUID which uses DEFAULT NEWSEQUENTIALID())
            var needsIdentityInsert = UsesIdentity(keyProperty) && !needsKeyGeneration;

            // Enable IDENTITY_INSERT if we're inserting an explicit value into an IDENTITY column
            if (needsIdentityInsert)
            {
                sb.Append("SET IDENTITY_INSERT ");
                sb.AppendTableName(schema, tableName);
                sb.AppendLine(" ON;");
            }

            sb.Append("MERGE INTO ");
            sb.AppendTableName(schema, tableName);
            sb.AppendLine(" AS t");
            sb.Append("USING (VALUES (");

            foreach (var property in model.Properties)
            {
                // Skip key in VALUES when auto-generating
                if (property is KeyPropertyModel && skipKeyInInsert)
                {
                    continue;
                }

                sb.AppendParameterName(property, ref paramIndex, out var paramName).Append(',');

                var value = property is VectorPropertyModel vectorProperty && generatedEmbeddings?.TryGetValue(vectorProperty, out var ge) == true
                    ? ge[firstRecordIndex + rowIndex]
                    : property.GetValueAsObject(record);

                command.AddParameter(property, paramName, value);
            }

            sb[sb.Length - 1] = ')'; // replace the last comma with a closing parenthesis
            sb.Append(") AS s (");
            sb.AppendIdentifiers(model.Properties, skipKey: skipKeyInInsert);
            sb.AppendLine(")");

            if (needsKeyGeneration)
            {
                // When auto-generating a key, we always insert (ON condition never matches).
                sb.AppendLine("ON (1=0)");
            }
            else
            {
                // For upsert, match on the key from the source
                sb.Append("ON (t.").AppendIdentifier(model.KeyProperty.StorageName).Append(" = s.").AppendIdentifier(model.KeyProperty.StorageName).AppendLine(")");
                sb.AppendLine("WHEN MATCHED THEN");
                sb.Append("UPDATE SET ");
                foreach (var property in model.Properties)
                {
                    if (property is not KeyPropertyModel) // don't update the key
                    {
                        sb.Append("t.").AppendIdentifier(property.StorageName).Append(" = s.").AppendIdentifier(property.StorageName).Append(',');
                    }
                }
                --sb.Length; // remove the last comma
                sb.AppendLine();
            }

            sb.AppendLine("WHEN NOT MATCHED THEN");
            sb.Append("INSERT (");
            sb.AppendIdentifiers(model.Properties, skipKey: skipKeyInInsert);
            sb.AppendLine(")");
            sb.Append("VALUES (");
            sb.AppendIdentifiers(model.Properties, prefix: "s.", skipKey: skipKeyInInsert);
            sb.AppendLine(")");
            sb.Append("OUTPUT inserted.").AppendIdentifier(model.KeyProperty.StorageName).AppendLine(";");

            // Disable IDENTITY_INSERT after the MERGE
            if (needsIdentityInsert)
            {
                sb.Append("SET IDENTITY_INSERT ");
                sb.AppendTableName(schema, tableName);
                sb.AppendLine(" OFF;");
            }

            sb.AppendLine();

            rowIndex++;
        }

        if (rowIndex == 0)
        {
            return false; // there is nothing to do!
        }

        command.CommandText = sb.ToString();
        return true;
    }

    internal static SqlCommand DeleteSingle(
        SqlConnection connection, string? schema, string tableName,
        KeyPropertyModel keyProperty, object key)
    {
        SqlCommand command = connection.CreateCommand();

        int paramIndex = 0;
        StringBuilder sb = new(100);
        sb.Append("DELETE FROM ");
        sb.AppendTableName(schema, tableName);
        sb.Append(" WHERE ").AppendIdentifier(keyProperty.StorageName).Append(" = ");
        sb.AppendParameterName(keyProperty, ref paramIndex, out string keyParamName);
        command.AddParameter(keyProperty, keyParamName, key);

        command.CommandText = sb.ToString();
        return command;
    }

    internal static bool DeleteMany<TKey>(
        SqlCommand command, string? schema, string tableName,
        KeyPropertyModel keyProperty, IEnumerable<TKey> keys)
    {
        StringBuilder sb = new(100);
        sb.Append("DELETE FROM ");
        sb.AppendTableName(schema, tableName);
        sb.Append(" WHERE ").AppendIdentifier(keyProperty.StorageName).Append(" IN (");
        sb.AppendKeyParameterList(keys, command, keyProperty, out bool emptyKeys);
        sb.Append(')'); // close the IN clause

        if (emptyKeys)
        {
            return false;
        }

        command.CommandText = sb.ToString();
        return true;
    }

    internal static SqlCommand SelectSingle(
        SqlConnection sqlConnection, string? schema, string collectionName,
        CollectionModel model,
        object key,
        bool includeVectors)
    {
        SqlCommand command = sqlConnection.CreateCommand();

        int paramIndex = 0;
        StringBuilder sb = new(200);
        sb.Append("SELECT ");
        sb.AppendIdentifiers(model.Properties, includeVectors: includeVectors);
        sb.AppendLine();
        sb.Append("FROM ");
        sb.AppendTableName(schema, collectionName);
        sb.AppendLine();
        sb.Append("WHERE ").AppendIdentifier(model.KeyProperty.StorageName).Append(" = ");
        sb.AppendParameterName(model.KeyProperty, ref paramIndex, out string keyParamName);
        command.AddParameter(model.KeyProperty, keyParamName, key);

        command.CommandText = sb.ToString();
        return command;
    }

    internal static bool SelectMany<TKey>(
        SqlCommand command, string? schema, string tableName,
        CollectionModel model,
        IEnumerable<TKey> keys,
        bool includeVectors)
    {
        StringBuilder sb = new(200);
        sb.Append("SELECT ");
        sb.AppendIdentifiers(model.Properties, includeVectors: includeVectors);
        sb.AppendLine();
        sb.Append("FROM ");
        sb.AppendTableName(schema, tableName);
        sb.AppendLine();
        sb.Append("WHERE ").AppendIdentifier(model.KeyProperty.StorageName).Append(" IN (");
        sb.AppendKeyParameterList(keys, command, model.KeyProperty, out bool emptyKeys);
        sb.Append(')'); // close the IN clause

        if (emptyKeys)
        {
            return false; // there is nothing to do!
        }

        command.CommandText = sb.ToString();
        return true;
    }

    internal static SqlCommand SelectVector<TRecord>(
        SqlConnection connection, string? schema, string tableName,
        VectorPropertyModel vectorProperty,
        CollectionModel model,
        int top,
        VectorSearchOptions<TRecord> options,
        SqlVector<float> vector)
    {
        string distanceFunction = vectorProperty.DistanceFunction ?? DistanceFunction.CosineDistance;
        (string distanceMetric, string sorting) = MapDistanceFunction(distanceFunction);

        return UseVectorSearch(vectorProperty)
            ? SelectVectorWithVectorSearch(connection, schema, tableName, vectorProperty, model, top, options, vector, distanceMetric, sorting)
            : SelectVectorWithVectorDistance(connection, schema, tableName, vectorProperty, model, top, options, vector, distanceMetric, sorting);
    }

    private static SqlCommand SelectVectorWithVectorDistance<TRecord>(
        SqlConnection connection, string? schema, string tableName,
        VectorPropertyModel vectorProperty,
        CollectionModel model,
        int top,
        VectorSearchOptions<TRecord> options,
        SqlVector<float> vector,
        string distanceMetric,
        string sorting)
    {
        SqlCommand command = connection.CreateCommand();
        command.Parameters.AddWithValue("@vector", vector);

        StringBuilder sb = new(200);

        sb.Append("SELECT ");
        sb.AppendIdentifiers(model.Properties, includeVectors: options.IncludeVectors);
        sb.AppendLine(",");
        sb.Append("VECTOR_DISTANCE('").Append(distanceMetric).Append("', ").AppendIdentifier(vectorProperty.StorageName)
            .Append(", CAST(@vector AS VECTOR(").Append(vector.Length).AppendLine("))) AS [score]");
        sb.Append("FROM ");
        sb.AppendTableName(schema, tableName);
        sb.AppendLine();

        if (options.Filter is not null)
        {
            int startParamIndex = command.Parameters.Count;

            SqlServerFilterTranslator translator = new(model, options.Filter, sb, startParamIndex: startParamIndex);
            translator.Translate(appendWhere: true);
            List<object> parameters = translator.ParameterValues;

            foreach (object parameter in parameters)
            {
                command.AddParameter(vectorProperty, $"@_{startParamIndex++}", parameter);
            }

            sb.AppendLine();
        }

        // If score threshold is specified, wrap in a subquery to filter on the pre-calculated score
        // This avoids calculating VECTOR_DISTANCE() twice.
        if (options.ScoreThreshold is not null)
        {
            // For SQL Server, all distance metrics return a distance (lower = more similar), so we filter with <=.
            command.Parameters.AddWithValue("@scoreThreshold", options.ScoreThreshold!.Value);

            var innerQuery = sb.ToString();
            sb.Clear();
            sb.Append("SELECT * FROM (").Append(innerQuery).AppendLine(") AS [inner]");
            sb.AppendLine("WHERE [score] <= @scoreThreshold");
        }

        sb.AppendFormat("ORDER BY [score] {0}", sorting);
        sb.AppendLine();
        // Negative Skip and Top values are rejected by the VectorSearchOptions property setters.
        // 0 is a legal value for OFFSET.
        sb.AppendFormat("OFFSET {0} ROWS FETCH NEXT {1} ROWS ONLY;", options.Skip, top);

        command.CommandText = sb.ToString();
        return command;
    }

    /// <summary>
    /// Generates a SELECT query using the VECTOR_SEARCH() function for approximate nearest neighbor search
    /// when the vector property has a vector index (e.g. DiskANN).
    /// </summary>
    private static SqlCommand SelectVectorWithVectorSearch<TRecord>(
        SqlConnection connection, string? schema, string tableName,
        VectorPropertyModel vectorProperty,
        CollectionModel model,
        int top,
        VectorSearchOptions<TRecord> options,
        SqlVector<float> vector,
        string distanceMetric,
        string sorting)
    {
        SqlCommand command = connection.CreateCommand();
        command.Parameters.AddWithValue("@vector", vector);

        StringBuilder sb = new(300);

        // When skip > 0, we need a subquery since TOP and OFFSET/FETCH can't coexist in the same SELECT.
        bool needsSubquery = options.Skip > 0;

        if (needsSubquery)
        {
            sb.Append("SELECT * FROM (");
        }

        // VECTOR_SEARCH returns all columns from the table plus a 'distance' column.
        // We select the needed columns from the table alias and alias 'distance' as 'score'.
        // The latest version vector indexes require SELECT TOP(N) WITH APPROXIMATE instead of the deprecated TOP_N parameter.
        sb.Append("SELECT TOP(").Append(top + options.Skip).Append(") WITH APPROXIMATE ");
        sb.AppendIdentifiers(model.Properties, prefix: "t.", includeVectors: options.IncludeVectors);
        sb.AppendLine(",");
        sb.AppendLine("s.[distance] AS [score]");
        sb.Append("FROM VECTOR_SEARCH(TABLE = ");
        sb.AppendTableName(schema, tableName);
        sb.Append(" AS t, COLUMN = ").AppendIdentifier(vectorProperty.StorageName);
        sb.Append(", SIMILAR_TO = @vector, METRIC = '").Append(distanceMetric).AppendLine("') AS s");

        // With latest version vector indexes, WHERE predicates are applied during the vector search process
        // (iterative filtering), not after retrieval.
        if (options.Filter is not null)
        {
            int startParamIndex = command.Parameters.Count;

            SqlServerFilterTranslator translator = new(model, options.Filter, sb, startParamIndex: startParamIndex, tableAlias: "t");
            translator.Translate(appendWhere: true);
            List<object> parameters = translator.ParameterValues;

            foreach (object parameter in parameters)
            {
                command.AddParameter(property: null, $"@_{startParamIndex++}", parameter);
            }

            sb.AppendLine();
        }

        if (options.ScoreThreshold is not null)
        {
            command.Parameters.AddWithValue("@scoreThreshold", options.ScoreThreshold!.Value);
            sb.Append(options.Filter is not null ? "AND " : "WHERE ");
            sb.AppendLine("s.[distance] <= @scoreThreshold");
        }

        sb.AppendFormat("ORDER BY [score] {0}", sorting);

        if (needsSubquery)
        {
            sb.AppendLine();
            sb.Append(") AS [inner]");
            sb.AppendLine();
            sb.AppendFormat("ORDER BY [score] {0}", sorting);
            sb.AppendLine();
            sb.AppendFormat("OFFSET {0} ROWS FETCH NEXT {1} ROWS ONLY;", options.Skip, top);
        }

        command.CommandText = sb.ToString();
        return command;
    }

    internal static SqlCommand SelectHybrid<TRecord>(
        SqlConnection connection, string? schema, string tableName,
        VectorPropertyModel vectorProperty,
        DataPropertyModel textProperty,
        CollectionModel model,
        int top,
        HybridSearchOptions<TRecord> options,
        SqlVector<float> vector,
        string keywords)
    {
        bool useVectorSearch = UseVectorSearch(vectorProperty);

        string distanceFunction = vectorProperty.DistanceFunction ?? DistanceFunction.CosineDistance;
        (string distanceMetric, _) = MapDistanceFunction(distanceFunction);

        SqlCommand command = connection.CreateCommand();
        command.Parameters.AddWithValue("@vector", vector);
        command.Parameters.AddWithValue("@keywords", keywords);

        // For RRF, we need to fetch more candidates from each search than the final top count
        // to allow proper merging. The number of candidates should be at least top + skip.
        // The RRF constant (k) is typically 60 in literature, but we use a smaller value
        // that still allows proper ranking while keeping the query efficient.
        int candidateCount = Math.Max(top + options.Skip, 20); // Fetch at least 20 candidates
        const int RrfK = 60; // Standard RRF constant

        command.Parameters.AddWithValue("@candidateCount", candidateCount);
        command.Parameters.AddWithValue("@rrfK", RrfK);

        StringBuilder sb = new(1000);

        // Build the hybrid search query using CTEs with Reciprocal Rank Fusion (RRF)
        // Reference: https://github.com/Azure-Samples/azure-sql-db-openai/blob/main/vector-embeddings/07-hybrid-search.sql

        // CTE 1: Keyword search using FREETEXTTABLE
        sb.AppendLine("WITH keyword_search AS (");
        sb.AppendLine("    SELECT TOP(@candidateCount)");
        sb.Append("        ").AppendIdentifier(model.KeyProperty.StorageName).AppendLine(",");
        sb.AppendLine("        RANK() OVER (ORDER BY ft_rank DESC) AS [rank]");
        sb.AppendLine("    FROM (");
        sb.AppendLine("        SELECT TOP(@candidateCount)");
        sb.Append("            w.").AppendIdentifier(model.KeyProperty.StorageName).AppendLine(",");
        sb.AppendLine("            ftt.[RANK] AS ft_rank");
        sb.Append("        FROM ").AppendTableName(schema, tableName).AppendLine(" w");
        sb.Append("        INNER JOIN FREETEXTTABLE(").AppendTableName(schema, tableName).Append(", ")
            .AppendIdentifier(textProperty.StorageName).AppendLine(", @keywords) AS ftt");
        sb.Append("            ON w.").AppendIdentifier(model.KeyProperty.StorageName).AppendLine(" = ftt.[KEY]");

        // Apply filter to keyword search if specified
        if (options.Filter is not null)
        {
            int startParamIndex = command.Parameters.Count;
            SqlServerFilterTranslator translator = new(model, options.Filter, sb, startParamIndex: startParamIndex, tableAlias: "w");
            translator.Translate(appendWhere: true);
            foreach (object parameter in translator.ParameterValues)
            {
                command.AddParameter(property: null, $"@_{startParamIndex++}", parameter);
            }
            sb.AppendLine();
        }

        sb.AppendLine("        ORDER BY ft_rank DESC");
        sb.AppendLine("    ) AS freetext_documents");
        sb.AppendLine("),");

        // CTE 2: Semantic/vector search
        if (useVectorSearch)
        {
            // Use VECTOR_SEARCH() for approximate nearest neighbor search with a vector index.
            // The latest version vector indexes require SELECT TOP(N) WITH APPROXIMATE instead of the deprecated TOP_N parameter.
            sb.AppendLine("semantic_search AS (");
            sb.AppendLine("    SELECT TOP(@candidateCount) WITH APPROXIMATE");
            sb.Append("        t.").AppendIdentifier(model.KeyProperty.StorageName).AppendLine(",");
            sb.AppendLine("        RANK() OVER (ORDER BY s.[distance]) AS [rank]");
            sb.AppendLine("    FROM VECTOR_SEARCH(TABLE = ");
            sb.Append("        ").AppendTableName(schema, tableName);
            sb.Append(" AS t, COLUMN = ").AppendIdentifier(vectorProperty.StorageName);
            sb.Append(", SIMILAR_TO = @vector, METRIC = '").Append(distanceMetric).AppendLine("') AS s");

            // With latest version vector indexes, WHERE predicates are applied during the vector search process
            // (iterative filtering), not after retrieval.
            if (options.Filter is not null)
            {
                int filterParamStart = command.Parameters.Count;
                SqlServerFilterTranslator translator = new(model, options.Filter, sb, startParamIndex: filterParamStart, tableAlias: "t");
                translator.Translate(appendWhere: true);
                foreach (object parameter in translator.ParameterValues)
                {
                    command.AddParameter(property: null, $"@_{filterParamStart++}", parameter);
                }
                sb.AppendLine();
            }

            sb.AppendLine("    ORDER BY s.[distance]");
            sb.AppendLine("),");
        }
        else
        {
            // Use VECTOR_DISTANCE() for exact brute-force search (flat index / no index)
            sb.AppendLine("semantic_search AS (");
            sb.AppendLine("    SELECT TOP(@candidateCount)");
            sb.Append("        ").AppendIdentifier(model.KeyProperty.StorageName).AppendLine(",");
            sb.AppendLine("        RANK() OVER (ORDER BY cosine_distance) AS [rank]");
            sb.AppendLine("    FROM (");
            sb.AppendLine("        SELECT TOP(@candidateCount)");
            sb.Append("            w.").AppendIdentifier(model.KeyProperty.StorageName).AppendLine(",");
            sb.Append("            VECTOR_DISTANCE('").Append(distanceMetric).Append("', ")
                .AppendIdentifier(vectorProperty.StorageName)
                .Append(", CAST(@vector AS VECTOR(").Append(vector.Length).AppendLine("))) AS cosine_distance");
            sb.Append("        FROM ").AppendTableName(schema, tableName).AppendLine(" w");

            // Apply filter to semantic search if specified
            if (options.Filter is not null)
            {
                // We need to re-translate the filter for the semantic search CTE
                // The parameters are already added from keyword search, so we start fresh for this CTE
                int filterParamStart = command.Parameters.Count;
                SqlServerFilterTranslator translator = new(model, options.Filter, sb, startParamIndex: filterParamStart, tableAlias: "w");
                translator.Translate(appendWhere: true);
                foreach (object parameter in translator.ParameterValues)
                {
                    command.AddParameter(property: null, $"@_{filterParamStart++}", parameter);
                }
                sb.AppendLine();
            }

            sb.AppendLine("        ORDER BY cosine_distance");
            sb.AppendLine("    ) AS similar_documents");
            sb.AppendLine("),");
        }

        // CTE 3: Combined results with RRF scoring
        sb.AppendLine("hybrid_result AS (");
        sb.AppendLine("    SELECT");
        sb.Append("        COALESCE(ss.").AppendIdentifier(model.KeyProperty.StorageName)
            .Append(", ks.").AppendIdentifier(model.KeyProperty.StorageName).AppendLine(") AS combined_key,");
        sb.AppendLine("        ss.[rank] AS semantic_rank,");
        sb.AppendLine("        ks.[rank] AS keyword_rank,");
        // Cast to FLOAT to match the expected return type in C# (double)
        // Use @rrfK as the RRF constant (typically 60)
        sb.AppendLine("        CAST(COALESCE(1.0 / (@rrfK + ss.[rank]), 0.0) + COALESCE(1.0 / (@rrfK + ks.[rank]), 0.0) AS FLOAT) AS [score]");
        sb.AppendLine("    FROM semantic_search ss");
        sb.Append("    FULL OUTER JOIN keyword_search ks ON ss.").AppendIdentifier(model.KeyProperty.StorageName)
            .Append(" = ks.").AppendIdentifier(model.KeyProperty.StorageName).AppendLine();
        sb.AppendLine(")");

        // Final SELECT joining back to the main table
        sb.Append("SELECT ");
        foreach (var property in model.Properties)
        {
            if (!options.IncludeVectors && property is VectorPropertyModel)
            {
                continue;
            }
            sb.Append("w.").AppendIdentifier(property.StorageName).Append(',');
        }
        sb.Length--; // remove trailing comma
        sb.AppendLine(",");
        sb.AppendLine("    hr.[score]");
        sb.AppendLine("FROM hybrid_result hr");
        sb.Append("INNER JOIN ").AppendTableName(schema, tableName).AppendLine(" w");
        sb.Append("    ON hr.combined_key = w.").AppendIdentifier(model.KeyProperty.StorageName).AppendLine();
        if (options.ScoreThreshold.HasValue)
        {
            command.Parameters.AddWithValue("@scoreThreshold", options.ScoreThreshold.Value);
            sb.AppendLine("WHERE hr.[score] >= @scoreThreshold");
        }
        sb.AppendLine("ORDER BY hr.[score] DESC");
        sb.AppendFormat("OFFSET {0} ROWS FETCH NEXT {1} ROWS ONLY;", options.Skip, top);

        command.CommandText = sb.ToString();
        return command;
    }

    internal static SqlCommand SelectWhere<TRecord>(
        Expression<Func<TRecord, bool>> filter,
        int top,
        FilteredRecordRetrievalOptions<TRecord> options,
        SqlConnection connection, string? schema, string tableName,
        CollectionModel model)
    {
        SqlCommand command = connection.CreateCommand();

        StringBuilder sb = new(200);
        sb.Append("SELECT ");
        sb.AppendIdentifiers(model.Properties, includeVectors: options.IncludeVectors);
        sb.AppendLine();
        sb.Append("FROM ");
        sb.AppendTableName(schema, tableName);
        sb.AppendLine();
        if (filter is not null)
        {
            int startParamIndex = command.Parameters.Count;

            SqlServerFilterTranslator translator = new(model, filter, sb, startParamIndex: startParamIndex);
            translator.Translate(appendWhere: true);
            List<object> parameters = translator.ParameterValues;

            foreach (object parameter in parameters)
            {
                command.AddParameter(property: null, $"@_{startParamIndex++}", parameter);
            }
            sb.AppendLine();
        }

        var orderBy = options.OrderBy?.Invoke(new()).Values;
        if (orderBy is { Count: > 0 })
        {
            sb.Append("ORDER BY ");

            var first = true;
            foreach (var sortInfo in orderBy)
            {
                if (!first)
                {
                    sb.Append(',');
                }
                first = false;
                sb.AppendIdentifier(model.GetDataOrKeyProperty(sortInfo.PropertySelector).StorageName)
                    .Append(sortInfo.Ascending ? " ASC" : " DESC");
            }

            sb.AppendLine();
        }
        else
        {
            // no order by properties, but we need to add something for OFFSET and NEXT to work
            sb.AppendLine("ORDER BY (SELECT 1)");
        }

        // Negative Skip and Top values are rejected by the GetFilteredRecordOptions property setters.
        // 0 is a legal value for OFFSET.
        sb.AppendFormat("OFFSET {0} ROWS FETCH NEXT {1} ROWS ONLY;", options.Skip, top);

        command.CommandText = sb.ToString();
        return command;
    }

    internal static StringBuilder AppendParameterName(this StringBuilder sb, PropertyModel property, ref int paramIndex, out string parameterName)
    {
        // In SQL Server, parameter names cannot be just a number like "@1".
        // Parameter names must start with an alphabetic character or an underscore
        // and can be followed by alphanumeric characters or underscores.
        // Since we can't guarantee that the value returned by StoragePropertyName and DataModelPropertyName
        // is valid parameter name (it can contain whitespaces, or start with a number),
        // we just append the ASCII letters, stop on the first non-ASCII letter
        // and append the index.
        int index = sb.Length;
        sb.Append('@');
        foreach (char character in property.StorageName)
        {
            // We don't call APIs like char.IsWhitespace as they are expensive
            // as they need to handle all Unicode characters.
            if (character is not (>= 'a' and <= 'z' or >= 'A' and <= 'Z'))
            {
                break;
            }
            sb.Append(character);
        }
        // In case the column name is empty or does not start with ASCII letters,
        // we provide the underscore as a prefix (allowed).
        sb.Append('_');
        // To ensure the generated parameter id is unique, we append the index.
        sb.Append(paramIndex++);
        parameterName = sb.ToString(index, sb.Length - index);

        return sb;
    }

    internal static StringBuilder AppendTableName(this StringBuilder sb, string? schema, string tableName)
    {
        // If the identifier contains a ], then escape it by doubling it.
        // "Name with [brackets]" becomes [Name with [brackets]]].

        if (!string.IsNullOrEmpty(schema))
        {
            sb.AppendIdentifier(schema!).Append('.');
        }

        return sb.AppendIdentifier(tableName);
    }

    /// <summary>
    /// Appends a properly quoted and escaped SQL Server identifier to the StringBuilder.
    /// If the identifier contains a ], it is escaped by doubling it.
    /// </summary>
    internal static StringBuilder AppendIdentifier(this StringBuilder sb, string identifier)
    {
        sb.Append('[');
        int index = sb.Length;
        sb.Append(identifier);
        sb.Replace("]", "]]", index, identifier.Length);
        sb.Append(']');
        return sb;
    }

    private static StringBuilder AppendIdentifiers(this StringBuilder sb,
        IEnumerable<PropertyModel> properties,
        string? prefix = null,
        bool includeVectors = true,
        bool skipKey = false)
    {
        bool any = false;
        foreach (var property in properties)
        {
            if (!includeVectors && property is VectorPropertyModel)
            {
                continue;
            }

            if (skipKey && property is KeyPropertyModel)
            {
                continue;
            }

            if (prefix is not null)
            {
                sb.Append(prefix);
            }
            sb.AppendIdentifier(property.StorageName).Append(',');
            any = true;
        }

        if (any)
        {
            --sb.Length; // remove the last comma
        }

        return sb;
    }

    private static StringBuilder AppendKeyParameterList<TKey>(this StringBuilder sb,
        IEnumerable<TKey> keys, SqlCommand command, KeyPropertyModel keyProperty, out bool emptyKeys)
    {
        int keyIndex = 0;
        foreach (TKey key in keys)
        {
            // The caller ensures that keys collection is not null.
            // We need to ensure that none of the keys is null.
            Verify.NotNull(key);

            sb.AppendParameterName(keyProperty, ref keyIndex, out string keyParamName);
            sb.Append(',');
            command.AddParameter(keyProperty, keyParamName, key);
        }

        emptyKeys = keyIndex == 0;
        sb.Length--; // remove the last comma
        return sb;
    }

    private static StringBuilder AppendIndexName(this StringBuilder sb, string tableName, string columnName)
    {
        int length = sb.Length;

        // "Index names must start with a letter or an underscore (_)."
        sb.Append("index");
        sb.Append('_');
        AppendAllowedOnly(tableName);
        sb.Append('_');
        AppendAllowedOnly(columnName);

        if (sb.Length > length + SqlServerConstants.MaxIndexNameLength)
        {
            sb.Length = length + SqlServerConstants.MaxIndexNameLength;
        }

        return sb;

        void AppendAllowedOnly(string value)
        {
            foreach (char c in value)
            {
                // Index names can include letters, numbers, and underscores.
                if (char.IsLetterOrDigit(c) || c == '_')
                {
                    sb.Append(c);
                }
            }
        }
    }

    private static SqlCommand CreateCommand(this SqlConnection connection, StringBuilder sb)
    {
        SqlCommand command = connection.CreateCommand();
        command.CommandText = sb.ToString();
        return command;
    }

    private static void AddParameter(this SqlCommand command, PropertyModel? property, string name, object? value)
    {
        switch (value)
        {
            case null when property?.Type == typeof(byte[]):
                command.Parameters.Add(name, System.Data.SqlDbType.VarBinary).Value = DBNull.Value;
                break;
            case null:
                command.Parameters.AddWithValue(name, DBNull.Value);
                break;
            case byte[] buffer:
                command.Parameters.Add(name, System.Data.SqlDbType.VarBinary).Value = buffer;
                break;
            case DateTime dateTime:
                command.Parameters.Add(name, System.Data.SqlDbType.DateTime2).Value = dateTime;
                break;

            // Note that SqlVector doesn't any transformation and can be passed as-is (default case below)
            case ReadOnlyMemory<float> vector:
                command.Parameters.AddWithValue(name, new SqlVector<float>(vector));
                break;
            case Embedding<float> { Vector: var vector }:
                command.Parameters.AddWithValue(name, new SqlVector<float>(vector));
                break;
            case float[] vectorArray:
                command.Parameters.AddWithValue(name, new SqlVector<float>(vectorArray));
                break;

            case string[] strings:
                command.Parameters.AddWithValue(name, JsonSerializer.Serialize(strings, SqlServerJsonSerializerContext.Default.StringArray));
                break;
            case List<string> strings:
                command.Parameters.AddWithValue(name, JsonSerializer.Serialize(strings, SqlServerJsonSerializerContext.Default.ListString));
                break;

            default:
                command.Parameters.AddWithValue(name, value);
                break;
        }
    }

    private static string Map(PropertyModel property)
        => (Nullable.GetUnderlyingType(property.Type) ?? property.Type) switch
        {
            Type t when t == typeof(byte) => "TINYINT",
            Type t when t == typeof(short) => "SMALLINT",
            Type t when t == typeof(int) => "INT",
            Type t when t == typeof(long) => "BIGINT",
            Type t when t == typeof(Guid) => "UNIQUEIDENTIFIER",
            Type t when t == typeof(string) && property is KeyPropertyModel => "NVARCHAR(4000)",
            Type t when t == typeof(string) && property is DataPropertyModel { IsIndexed: true } => "NVARCHAR(4000)",
            Type t when t == typeof(string) => "NVARCHAR(MAX)",
            Type t when t == typeof(byte[]) => "VARBINARY(MAX)",
            Type t when t == typeof(bool) => "BIT",
            Type t when t == typeof(DateTime) => "DATETIME2",
            Type t when t == typeof(DateTimeOffset) => "DATETIMEOFFSET",
#if NET
            Type t when t == typeof(DateOnly) => "DATE",
            Type t when t == typeof(TimeOnly) => "TIME",
#endif
            Type t when t == typeof(decimal) => "DECIMAL(18,2)",
            Type t when t == typeof(double) => "FLOAT",
            Type t when t == typeof(float) => "REAL",

            Type t when t == typeof(string[]) || t == typeof(List<string>) => "JSON",

            _ => throw new NotSupportedException($"Type {property.Type} is not supported.")
        };

    // Source: https://learn.microsoft.com/sql/t-sql/functions/vector-distance-transact-sql
    private static (string distanceMetric, string sorting) MapDistanceFunction(string name) => name switch
    {
        // A value of 0 indicates that the vectors are identical in direction (cosine similarity of 1),
        // while a value of 1 indicates that the vectors are orthogonal (cosine similarity of 0).
        DistanceFunction.CosineDistance => ("COSINE", "ASC"),
        // A value of 0 indicates that the vectors are identical, while larger values indicate greater dissimilarity.
        DistanceFunction.EuclideanDistance => ("EUCLIDEAN", "ASC"),
        // Smaller numbers indicate more similar vectors
        DistanceFunction.NegativeDotProductSimilarity => ("DOT", "ASC"),
        _ => throw new NotSupportedException($"Distance function {name} is not supported.")
    };

    /// <summary>
    /// Returns whether VECTOR_SEARCH() (approximate/indexed search) should be used for the given vector property,
    /// as opposed to VECTOR_DISTANCE() (exact/brute-force search).
    /// </summary>
    private static bool UseVectorSearch(VectorPropertyModel vectorProperty)
        => vectorProperty.IndexKind is not (null or "" or IndexKind.Flat);
}