Add support for TVP [WIP]

lib/tds.ex

@@ -17,6 +17,11 @@ defmodule Tds do
     end
   end
 
+  def proc(pid, statement, params, opts \\ []) do
+    opts = Keyword.put_new(opts, :proc, statement)
+    query(pid, statement, params, opts)
+  end
+
   def query!(pid, statement, params, opts \\ []) do
     query = %Query{statement: statement}
     opts = Keyword.put_new(opts, :parameters, params)

lib/tds/column.ex

@@ -0,0 +1,9 @@
+defmodule Tds.Column do
+  @type t :: %__MODULE__{
+    name: String.t | nil,
+    type: Atom | nil,
+    opts: Keyword.t
+  }
+
+  defstruct [name: "", type: nil, opts: []]
+end

lib/tds/messages.ex

@@ -362,6 +362,11 @@ defmodule Tds.Messages do
   defp encode_rpc(:sp_unprepare, params) do
     <<0xFF, 0xFF, @tds_sp_unprepare::little-size(2)-unit(8), 0x00, 0x00>> <> encode_rpc_params(params, "")
   end
+  defp encode_rpc(proc, params) when is_binary(proc) do
+    rpc_size = byte_size(proc)
+    rpc_name = to_little_ucs2(proc)
+    <<rpc_size::little-size(16)>> <> rpc_name <> <<0x00, 0x00>> <> encode_rpc_params(params, "")
+  end
 
   # Finished processing params
   defp encode_rpc_params([], ret), do: ret

lib/tds/protocol.ex

@@ -90,11 +90,12 @@ defmodule Tds.Protocol do
 
   def handle_execute(%Query{statement: statement} = query, params, opts, %{sock: _sock} = s) do
     params = opts[:parameters] || params
+    proc = opts[:proc] || nil
 
-    if params != [] do
-      send_param_query(query, params, s)
-    else
-      send_query(statement, s)
+    cond do
+      params != [] and is_nil(proc) -> send_param_query(query, params, s)
+      not is_nil(proc) -> send_proc(proc, params, s)
+      true -> send_query(statement, s)
     end
   end
 
@@ -391,6 +392,20 @@ defmodule Tds.Protocol do
   #  {:ok, %{s | statement: nil, state: :ready}}
   #end
 
+  def send_proc(proc, params, s) do
+    params = Tds.Parameter.prepare_params(params)
+    msg = msg_rpc(proc: proc, params: params)
+
+    case msg_send(msg, s) do
+      {:ok, %{result: result} = s} ->
+        {:ok, result, %{s | state: :ready}}
+      {:error, err, %{transaction: :started} = s} ->
+        {:error, err, %{s | transaction: :failed}}
+      err ->
+        err
+    end
+  end
+
   def send_param_query(%Query{handle: handle} = _query, params, %{transaction: :started} = s) do
     params = [
       %Tds.Parameter{name: "@handle", type: :integer, direction: :input, value: handle}

lib/tds/types.ex

@@ -4,6 +4,7 @@ defmodule Tds.Types do
   use Bitwise
 
   alias Tds.Parameter
+  alias Tds.Column
   alias Tds.DateTime
   alias Tds.DateTime2
 
@@ -67,6 +68,7 @@ defmodule Tds.Types do
   @tds_data_type_nchar            0xEF
   @tds_data_type_xml              0xF1
   @tds_data_type_udt              0xF0
+  @tds_data_type_tvp              0xF3
   @tds_data_type_text             0x23
   @tds_data_type_image            0x22
   @tds_data_type_ntext            0x63
@@ -501,6 +503,7 @@ defmodule Tds.Types do
       :date -> encode_date_type(param)
       :time -> encode_time_type(param)
       :uuid -> encode_uuid_type(param)
+      :tvp -> encode_tvp_type(param)
       _ -> encode_string_type(param)
     end
   end
@@ -574,6 +577,11 @@ defmodule Tds.Types do
     encode_data_type(%{param | type: :datetimeoffset})
   end
 
+  def encode_tvp_type(%Parameter{}) do
+    type = @tds_data_type_tvp
+    data = <<type>> <> <<0, 0, 0>>
+    {type, data, []}
+  end
 
   def encode_binary_type(%Parameter{value: value} = param)
   when value == "" do
@@ -764,6 +772,7 @@ defmodule Tds.Types do
       :date -> "date"
       :time -> "time"
       :smalldatetime -> "smalldatetime"
+      :tvp -> "#{value.name} readonly"
       :binary -> encode_binary_descriptor(value)
       :string ->
         cond do
@@ -967,9 +976,95 @@ defmodule Tds.Types do
   def encode_data(@tds_data_type_bigvarbinary, value, _),
     do: <<byte_size(value)::little-unsigned-16>> <> value
 
+  @doc """
+  Data Encoding TVP type
+  """
+  def encode_data(@tds_data_type_tvp, %{columns: nil}, _attrs),
+    do: <<0xFF :: little-unsigned-16, 0x00, 0x00 >>
+
+  def encode_data(@tds_data_type_tvp, %{columns: columns, rows: rows}, _attrs) do
+    column_length = <<length(columns) :: little-unsigned-16>>
+    {column_attrs, column_meta} = Enum.reduce(columns, {[], <<>>}, fn (%Column{} = param, {attrs, acc_bin}) ->
+      {bin_type, data, attr} = encode_column_type(param)
+      bin = acc_bin <> <<0x00 :: little-unsigned-32, 0x00 :: little-unsigned-16 >> <> data <> <<0x00>>
+
+      {[{bin_type, attr} | attrs], bin}
+    end)
+
+    row_data = Enum.reduce(rows, <<>>, fn (params, row_acc) ->
+      row_bin = column_attrs
+      |> Enum.reverse
+      |> Enum.zip(params)
+      |> Enum.reduce(<<>>, fn ({{type, attr}, param}, acc) ->
+        acc <> encode_data(type, param, attr)
+      end)
+
+      row_acc <> << 0x01 >> <> row_bin
+    end)
+
+    column_length <> column_meta <> <<0x00>> <> row_data <> <<0x00>>
+  end
+
+  def encode_column_type(%Column{type: type} = col) when type != nil do
+    case type do
+      :varchar -> encode_bigvarchar_col_type(col)
+      :boolean -> encode_binary_type(%Parameter{type: :boolean, value: nil})
+      :varbinary -> encode_varbinary_col_type(col)
+      :int -> encode_integer_type(%Parameter{type: :integer, value: nil})
+      :decimal -> encode_decimal_type(%Parameter{type: :decimal, value: nil})
+      :float -> encode_float_type(%Parameter{type: :float, value: nil})
+      :datetime -> encode_datetime_type(%Parameter{type: :datetime, value: nil})
+      :smalldatetime -> encode_smalldatetime_type(%Parameter{type: :smalldatetime, value: nil})
+      :datetime2 -> encode_datetime2_type(%Parameter{type: :datetime2, value: nil})
+      :datetimeoffset -> encode_datetimeoffset_type(%Parameter{type: :datetimeoffset, value: nil})
+      :date -> encode_date_type(%Parameter{type: :datetimeoffset, value: nil})
+      :time -> encode_time_type(%Parameter{type: :time, value: nil})
+      :uuid -> encode_uuid_type(%Parameter{type: :uuid, value: nil})
+    end
+  end
+
+  def encode_bigvarchar_col_type(%Column{opts: opts} = col) do
+    type = @tds_data_type_bigvarchar
+    length = Keyword.get(opts, :length, 0)
+    collation = <<0x00, 0x00, 0x00, 0x00, 0x00>>
+    bin_length = if length <= 8000, do: <<8000::little-unsigned-16>>, else: <<0xFF, 0xFF>>
+    data = <<type>> <> bin_length <> collation
+
+    {type, data, length: length}
+  end
+
+  def encode_varbinary_col_type(%Column{opts: opts} = col) do
+    type = @tds_data_type_bigvarbinary
+    length = Keyword.get(opts, :length, 0)
+    bin_length = if length <= 8000, do: <<8000::little-unsigned-16>>, else: <<0xFF, 0xFF>>
+    data = <<type>> <> bin_length
+
+    {type, data, length: length}
+  end
+
   @doc """
   Data Encoding String Types
   """
+
+  def encode_data(@tds_data_type_bigvarchar, nil, opts) do
+    length = Keyword.get(opts, :length, 0)
+    if length <= 8000,
+      do: <<65535::little-unsigned-16>>,
+      else: <<@tds_plp_null::little-unsigned-64>>
+  end
+
+  def encode_data(@tds_data_type_bigvarchar, value, opts) do
+    value_size = byte_size(value)
+    cond do
+      value_size <= 0 ->
+        <<0x00::unsigned-64, 0x00::unsigned-32>>
+      value_size > 8000 ->
+        encode_plp(value)
+      true ->
+        <<value_size::little-size(2)-unit(8)>> <> value
+    end
+  end
+
   def encode_data(@tds_data_type_nvarchar, nil, _),
     do: <<@tds_plp_null::little-unsigned-64>>
   def encode_data(@tds_data_type_nvarchar, value, _) do

test/test_helper.exs

@@ -14,7 +14,7 @@ defmodule Tds.TestHelper do
 
   defmacro proc(proc, params, opts \\ []) do
     quote do
-      case Tds.Connection.proc(var!(context)[:pid], unquote(proc),
+      case Tds.proc(var!(context)[:pid], unquote(proc),
                                      unquote(params), unquote(opts)) do
         {:ok, %Tds.Result{rows: nil}} -> :ok
         {:ok, %Tds.Result{rows: []}} -> :ok

test/tvp_test.exs

@@ -0,0 +1,55 @@
+defmodule TvpTest do
+  import Tds.TestHelper
+  require Logger
+  use ExUnit.Case, async: true
+  alias Tds.Parameter
+  alias Tds.Column
+
+  @tag timeout: 50000
+
+  setup do
+    opts = Application.fetch_env!(:tds, :opts)
+    {:ok, pid} = Tds.start_link(opts)
+
+    {:ok, [pid: pid]}
+  end
+
+  test "TVP in stored proc", context do
+    assert :ok = query("BEGIN TRY DROP PROCEDURE __tvpTest DROP TYPE TvpTestType END TRY BEGIN CATCH END CATCH", [])
+    assert :ok = query("""
+      CREATE TYPE TvpTestType AS TABLE (
+        a int,
+        b uniqueidentifier,
+        c varchar(100),
+        d varbinary(max)
+      );
+    """, [])
+
+    assert :ok = query("""
+    CREATE PROCEDURE __tvpTest (@tvp TvpTestType readonly)
+      AS BEGIN
+        select * from @tvp
+      END
+    """, [])
+
+    rows = [1, <<158, 3, 157, 56, 133, 56, 73, 67, 128, 121, 126, 204, 115, 227, 162, 157>>, "foo", "{\"foo\":\"bar\",\"baz\":\"biz\"}"]
+    params = [
+      %Parameter{
+        name: "@tvp",
+        value: %{
+          name: "TvpTestType",
+          columns: [
+            %Column{name: "a", type: :int},
+            %Column{name: "b", type: :uuid},
+            %Column{name: "c", type: :varchar},
+            %Column{name: "d", type: :varbinary},
+          ],
+          rows: [rows]
+        },
+        type: :tvp
+      }
+    ]
+
+    assert [^rows] = proc("__tvpTest", params)
+  end
+end