Um array JSON pode ser enviado como um parâmetro de procedimento armazenado em modo de streaming?

Criei uma versão mais eficiente e simples do seu teste JSON e também o comparei com o teste TVP, usando o BenchmarkDotNet (o padrão ouro em benchmarking .NET).

Melhorias:

• Usamos a API de sincronização para o SqlClient, pois a API assíncrona atualmente tem uma séria desvantagem de desempenho.
• Para garantir que estamos realmente lendo todas as três colunas, adicionei uma subconsulta derivada SELECT DISTINCTsobre todas as colunas.

[Params(1, 100, 1000, 50000)]
public uint Length { get; set;}

[Benchmark]
public int JsonTest()
{
    const string query = """
                            SELECT COUNT(*)
                            FROM  (
                                SELECT DISTINCT Prop1, Prop2, Prop3
                                FROM OPENJSON(@JsonData)
                                WITH (
                                    Prop1 int,
                                    Prop2 datetime2(7),
                                    Prop3 nvarchar(1000)
                                ) j
                            ) t;
                         """;
    using var conn = new SqlConnection(connectionString);
    conn.Open();
    using var cmd = new SqlCommand(query, conn);
    cmd.CommandTimeout = 0;

    var data = GetEnumerableOfRandomDto(Length);
    using var reader = new StreamReader(GetJsonStream(data));
    cmd.Parameters.Add("@JsonData", SqlDbType.NVarChar, -1).Value = reader;

    return (int)cmd.ExecuteScalar();
}

• O DTO é criado usando valores idênticos para prop2e prop3, para evitar sobrecargas de Randome interpolação de strings.

static IEnumerable<Dto> GetEnumerableOfRandomDto(uint length)
{
    var rnd = new Random();
    var s = Guid.NewGuid().ToString();
    var dt = DateTime.Now.AddMinutes(rnd.Next(1, 10000));

    for (var i = 0; i < length; i++)
    {
        yield return new Dto(i,
            dt,
            s
        );
    }
}

• Obtemos o TextReaderuso da nova System.IO.PipelinesAPI da seguinte forma:
  • Crie um Pipe.
  • Em um thread separado, envie os DTOs para o pipe writer o mais rápido que ele puder.
  • Obtenha o leitor de pipe como a Streame passe-o para StreamReader.
  • StreamIsso evita ter que usar toda essa complicação de passagem falsa , já que o Pipefaz tudo isso para nós.

public static Stream GetJsonStream<T>(T obj, JsonSerializerOptions? options = null)
{
    var pipe = new Pipe();
    _ = Task.Run(() =>
    {
        try
        {
            var utf8writer = new Utf8JsonWriter(pipe.Writer);
            JsonSerializer.Serialize(utf8writer, obj);
            pipe.Writer.Complete();
        }
        catch(Exception ex)
        {
            pipe.Writer.Complete(ex);
        }
    });
    var stream = pipe.Reader.AsStream();
    return stream;
}

• Para TvpTest, você não deve recriar SqlDataRecord. Ao contrário de qualquer outro uso de IEnumerableno planeta, é recomendado reutilizar o SqlDataRecord. Observe que a DateTime2escala é explicitamente especificada.

static IEnumerable<SqlDataRecord> GetRecordsOfRandomDto(uint length)
{
    var list = GetEnumerableOfRandomDto(length);
    var record = new SqlDataRecord(
        new("prop1", SqlDbType.Int),
        new("prop2", SqlDbType.DateTime2, 0, 7),
        new("prop3", SqlDbType.NVarChar, 1000));

    foreach (var dto in list)
    {
        record.SetInt32(0, dto.Prop1);
        record.SetDateTime(1, dto.Prop2);
        record.SetString(2, dto.Prop3);
        yield return record;
    }
}

Os resultados do BenchmarkDotNet estão abaixo. Você pode ver que usar JSON é bem mais rápido para conjuntos de dados muito pequenos, mas os TVPs se distanciam massivamente na extremidade superior. Então parece melhor usar TVPs somente quando o conjunto de dados provavelmente atingirá 1000 linhas, devido a um custo inicial claramente maior.

Também testei com uma única Prop1 intcoluna em ambos, com a consulta SELECT MAX(Prop1) FROMpara cada um. Os resultados são interessantes: a versão TVP agora usa muito menos memória, mesmo para 50000 linhas, mas JSON usou o mesmo, e a diferença de velocidade em contagens de linhas altas também foi maior.

Outras notas:

• Usar a StreamReadertem a desvantagem de converter de UTF-8 para UTF-16. Eu vi uma pequena melhora em alguns casos usando a varbinarye escrevendo o UTF-8 Streamdiretamente no servidor, então CASTing server-side as varchare então como nvarchar, mas não foi consistente. Infelizmente, o SqlClient não aceita a Streamde bytes UTF-8 para um varcharvalor.
• Infelizmente, não há como usar System.Text.JsonUTF-16 para escrever, e OPENJSONele só aceita nvarchar, então é quase certo que haja alguma perda de eficiência na conversão.
• Também tentei usar XML e .nodesvários .valuesmétodos. Todos foram menos eficientes do que os outros dois métodos em todos os casos.
  • XmlReaderprecisa que stringvalores sejam passados, o que significa que muitas strings são geradas para os valores intand convertidos DateTime. Isso ainda pode ser uma opção para um caso de uso que envolva apenas strings.
  • XmlSerializerusará apenas uma XmlReadersaída.
  • XDocumentpode gerar um XmlReader, mas você ainda precisa gerar todas essas strings.
  • Tentei gerar o XML manualmente em UTF-8 e UTF-16 usando o Pipee lendo usando um StreamReader, bem como passando-o como um Streamdiretamente e CASTenviando-o para o servidor.

This was tested using SqlClient v6.0.1, and LocalDB 2019 RTM-CU29-GDR. I did not test using TDS v8.0, but some improvements have been made there to streaming large unknown-length parameters.

O tópico Exemplo: Streaming para SQL Server mostra como parâmetros podem ser definidos a partir de um fluxo.

Embora System.Text.Jsonseja possível serializar para um fluxo, não ficou claro para mim se/como é possível fazer com que ele serialize um enumerável e tenha a avaliação do enumerável adiada até ser usada como um valor de parâmetro para o procedimento armazenado equivalente.

CREATE OR ALTER PROC dbo.TestJsonPerf
@JsonData NVARCHAR(MAX)
AS

SELECT COUNT(*)
FROM OPENJSON(@JsonData)
WITH (Prop1 int, Prop2 datetime2, Prop3 varchar(1000))

Eu encontrei uma solução possível para isso nesta resposta do StackOverflow . O abaixo é (agora muito vagamente) baseado nisso e adia a avaliação do enumerável até depois que o ExecuteScalarAsyncé chamado. No processo do cliente, a memória é muito baixa durante todo o processo.

⚠️ O desempenho desse método ainda é pior do que usar TVPs. Com o código TVP na questão na minha máquina local, uma execução típica com 5 milhões de linhas levou de 8 a 9 segundos, a versão de streaming JSON levou cerca de 45 segundos. Dos quais cerca de 25 segundos pareceram ser gastos enviando o valor do parâmetro, então tanto a passagem do parâmetro quanto os tempos de execução da consulta foram muito piores!

O código de passagem de parâmetros provavelmente ainda tem algumas melhorias de desempenho que podem ser feitas, mas isso não ajudará no desempenho da consulta.

A consulta TVP foi executada em DOP 4 em modo batch na minha máquina e teve uma média de 80 ms por duração mostrada no armazenamento de consultas, enquanto a JSON obteve um plano serial e duração média de > 20 segundos. Na prática, porém, é improvável que alguém envie 5 milhões de elementos JSON para o SQL Server apenas para contá-los e a diferença pode ser bem menos gritante em casos de uso mais realistas.

using System.Data;
using System.Text;
using System.Text.Json;
using Microsoft.Data.SqlClient;

const string connectionString =
    @"...";

await JsonTest();

return;

static async Task JsonTest()
{
    await using var conn = new SqlConnection(connectionString);
    await conn.OpenAsync();
    await using var cmd = new SqlCommand("dbo.TestJsonPerf", conn);
    cmd.CommandType = CommandType.StoredProcedure;
    cmd.CommandTimeout = 300;


    var data = GetEnumerableOfRandomDto(5_000_000);
    var stream = new JsonArrayStream<Dto>(data);

    cmd.Parameters.Add("@JsonData", SqlDbType.NVarChar, -1).Value =
        new StreamReader(stream, Encoding.UTF8, bufferSize: 1_048_576);

    Console.WriteLine($"calling ExecuteScalarAsync at {DateTime.Now:O}");
    var result = await cmd.ExecuteScalarAsync();

    Console.WriteLine($"writing result at {DateTime.Now:O}. Result was {result}");
}

static IEnumerable<Dto> GetEnumerableOfRandomDto(uint length)
{
    var rnd = new Random();

    for (var i = 0; i < length; i++)
    {
        yield return new Dto(i,
            DateTime.UtcNow.AddMinutes(rnd.Next(1, 10000)),
            Guid.NewGuid().ToString()
        );

        if ((i + 1) % 100_000 == 0)
            Console.WriteLine($"Generated enumerable {i + 1} at {DateTime.Now:O}");
    }
}


internal record Dto(int Prop1, DateTime Prop2, string Prop3);

public class JsonArrayStream<T> : Stream, IDisposable
{
    private readonly IEnumerator<T> _input;
    private bool _arrayClosed;

    private byte[] _bytesToAppendNextRead = [(byte) '['];
    private bool _firstElement = true;
    private bool _disposed;

    public JsonArrayStream(IEnumerable<T> input)
    {
        _input = input.GetEnumerator();
    }

    public override bool CanRead => true;
    public override bool CanSeek => false;
    public override bool CanWrite => false;
    public override long Length => 0;
    public override long Position { get; set; }

    void IDisposable.Dispose()
    {
        if (_disposed)
            return;
        _disposed = true;
        _input.Dispose();
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        var bytesWrittenThisRead = 0;

        if (!TryAppendToBuffer(_bytesToAppendNextRead, buffer, ref offset, count, ref bytesWrittenThisRead))
            return bytesWrittenThisRead;

  
        while (_input.MoveNext())
        {
            if (!_firstElement)
            {
                //regardless of result of this call will proceed to the next step to ensure bytes for _input.Current are saved before MoveNext called
                TryAppendToBuffer([(byte) ','], buffer, ref offset, count, ref bytesWrittenThisRead);
            }

            _firstElement = false;

            if (!TryAppendToBuffer(JsonSerializer.SerializeToUtf8Bytes(_input.Current), buffer, ref offset, count, ref bytesWrittenThisRead))
                return bytesWrittenThisRead;
        }

        if (!_arrayClosed)
        {
            TryAppendToBuffer([(byte)']'], buffer, ref offset, count, ref bytesWrittenThisRead);
            _arrayClosed = true;
        }

        return bytesWrittenThisRead;
    }

    //Appends to stream buffer if possible. If no capacity saves any excess in _bytesToAppendNextRead
    private bool TryAppendToBuffer(byte[] bytesToAppend, byte[] buffer, ref int offset, int count,
        ref int bytesWrittenThisRead)
    {
        var bytesToAppendLength = bytesToAppend.Length;

        if (bytesToAppendLength > 0)
        {
            var lengthToWrite = Math.Min(bytesToAppendLength, count - bytesWrittenThisRead);
            Buffer.BlockCopy(bytesToAppend, 0, buffer, offset, lengthToWrite);
            offset += lengthToWrite;
            bytesWrittenThisRead += lengthToWrite;

            if (bytesToAppendLength > lengthToWrite)
            {
                _bytesToAppendNextRead = _bytesToAppendNextRead.Length == 0 ? 
                    bytesToAppend[lengthToWrite..] : 
                    _bytesToAppendNextRead.Concat(bytesToAppend[lengthToWrite..]).ToArray();

                return false;
            }

            _bytesToAppendNextRead = [];
        }

        return true;
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        throw new NotSupportedException();
    }

    public override void SetLength(long value)
    {
        throw new NotSupportedException();
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        throw new NotSupportedException();
    }

    public override void Flush()
    {
        throw new NotSupportedException();
    }
}