diff --git a/lib/ch/row_binary.ex b/lib/ch/row_binary.ex
index 5ba2a7b4..6747449a 100644
--- a/lib/ch/row_binary.ex
+++ b/lib/ch/row_binary.ex
@@ -140,13 +140,9 @@ defmodule Ch.RowBinary do
   defp encoding_type({:nullable = n, t}), do: {n, encoding_type(t)}
   defp encoding_type({:low_cardinality, t}), do: encoding_type(t)
 
-  defp encoding_type({:decimal, p, s}) do
-    case decimal_size(p) do
-      32 -> {:decimal32, s}
-      64 -> {:decimal64, s}
-      128 -> {:decimal128, s}
-      256 -> {:decimal256, s}
-    end
+  defp encoding_type({:decimal, precision, scale} = type) do
+    validate_decimal_type!(precision, scale)
+    type
   end
 
   defp encoding_type({d, _scale} = t)
@@ -164,6 +160,7 @@ defmodule Ch.RowBinary do
   defp encoding_type({:time64 = t, p}), do: {t, time_unit(p)}
 
   defp encoding_type({e, mappings}) when e in [:enum8, :enum16] do
+    Ch.Types.encode({e, mappings})
     {e, Map.new(mappings)}
   end
 
@@ -258,47 +255,52 @@ defmodule Ch.RowBinary do
     end
   end
 
-  for size <- [32, 64] do
+  for {size, max} <- [{32, 3.4028234663852886e38}, {64, 1.7976931348623157e308}] do
     type = :"f#{size}"
 
-    def encode(unquote(type), f) when is_number(f) do
+    def encode(unquote(type), f) when is_number(f) and f >= -unquote(max) and f <= unquote(max) do
       <<f::unquote(size)-little-signed-float>>
     end
 
     def encode(unquote(type), nil), do: <<0::unquote(size)>>
+
+    def encode(unquote(type), term) do
+      raise ArgumentError, "invalid Float#{unquote(size)}: #{inspect(term)}"
+    end
   end
 
-  def encode({:decimal, precision, scale}, decimal) do
-    type =
-      case decimal_size(precision) do
-        32 -> :decimal32
-        64 -> :decimal64
-        128 -> :decimal128
-        256 -> :decimal256
-      end
+  def encode({:decimal, precision, scale}, %Decimal{} = decimal) do
+    validate_decimal_type!(precision, scale)
+    size = decimal_size(precision)
+    coefficient = decimal_coefficient!(decimal, scale, size)
 
-    encode({type, scale}, decimal)
+    if abs(coefficient) >= Integer.pow(10, precision) do
+      raise ArgumentError,
+            "Decimal value #{Decimal.to_string(decimal)} exceeds precision #{precision}"
+    end
+
+    encode_decimal!(decimal, coefficient, size, scale)
+  end
+
+  def encode({:decimal, precision, scale}, nil) do
+    validate_decimal_type!(precision, scale)
+    <<0::size(decimal_size(precision))>>
   end
 
   for size <- [32, 64, 128, 256] do
     type = :"decimal#{size}"
+    precision = %{32 => 9, 64 => 18, 128 => 38, 256 => 76}[size]
 
-    def encode({unquote(type), scale} = t, %Decimal{sign: sign, coef: coef, exp: exp} = d) do
-      cond do
-        scale == -exp ->
-          i = sign * coef
-          <<i::unquote(size)-little>>
-
-        exp >= 0 ->
-          i = sign * coef * Integer.pow(10, exp + scale)
-          <<i::unquote(size)-little>>
-
-        true ->
-          encode(t, Decimal.round(d, scale))
-      end
+    def encode({unquote(type), scale}, %Decimal{} = decimal) do
+      validate_decimal_scale!(unquote(type), scale, unquote(precision))
+      coefficient = decimal_coefficient!(decimal, scale, unquote(size))
+      encode_decimal!(decimal, coefficient, unquote(size), scale)
     end
 
-    def encode({unquote(type), _scale}, nil), do: <<0::unquote(size)>>
+    def encode({unquote(type), scale}, nil) do
+      validate_decimal_scale!(unquote(type), scale, unquote(precision))
+      <<0::unquote(size)>>
+    end
   end
 
   def encode(:boolean, true), do: 1
@@ -346,29 +348,36 @@ defmodule Ch.RowBinary do
 
   def encode(:datetime, %NaiveDateTime{} = datetime) do
     {seconds, _micros} = NaiveDateTime.to_gregorian_seconds(datetime)
-    <<seconds - @epoch_gregorian_seconds::32-little>>
+    encode_fixed_integer!(seconds - @epoch_gregorian_seconds, 32, :unsigned, "DateTime")
   end
 
   def encode(:datetime, %DateTime{} = datetime) do
-    <<DateTime.to_unix(datetime, :second)::32-little>>
+    datetime
+    |> DateTime.to_unix(:second)
+    |> encode_fixed_integer!(32, :unsigned, "DateTime")
   end
 
   def encode(:datetime, nil), do: <<0::32>>
 
   def encode({:datetime64, time_unit}, %NaiveDateTime{} = datetime) do
     {seconds, micros} = NaiveDateTime.to_gregorian_seconds(datetime)
-
-    <<(seconds - @epoch_gregorian_seconds) * time_unit + div(micros * time_unit, 1_000_000)::64-little-signed>>
+    ticks = (seconds - @epoch_gregorian_seconds) * time_unit + div(micros * time_unit, 1_000_000)
+    encode_fixed_integer!(ticks, 64, :signed, "DateTime64")
   end
 
   def encode({:datetime64, time_unit}, %DateTime{} = datetime) do
-    <<DateTime.to_unix(datetime, time_unit)::64-little-signed>>
+    datetime
+    |> DateTime.to_unix(time_unit)
+    |> encode_fixed_integer!(64, :signed, "DateTime64")
   end
 
   def encode({:datetime64, _time_unit}, nil), do: <<0::64>>
 
   def encode(:date, %Date{} = date) do
-    <<Date.to_gregorian_days(date) - @epoch_gregorian_days::16-little>>
+    date
+    |> Date.to_gregorian_days()
+    |> Kernel.-(@epoch_gregorian_days)
+    |> encode_fixed_integer!(16, :unsigned, "Date")
   end
 
   def encode(:date, nil), do: <<0::16>>
@@ -420,13 +429,29 @@ defmodule Ch.RowBinary do
 
   def encode(:uuid, nil), do: <<0::128>>
 
-  def encode(:ipv4, {a, b, c, d}), do: [d, c, b, a]
+  def encode(:ipv4, {a, b, c, d})
+      when is_integer(a) and a in 0..255 and is_integer(b) and b in 0..255 and is_integer(c) and
+             c in 0..255 and is_integer(d) and d in 0..255,
+      do: [d, c, b, a]
+
+  def encode(:ipv4, {_, _, _, _} = address) do
+    raise ArgumentError, "invalid IPv4 address: #{inspect(address)}"
+  end
+
   def encode(:ipv4, nil), do: <<0::32>>
 
-  def encode(:ipv6, {b1, b2, b3, b4, b5, b6, b7, b8}) do
+  def encode(:ipv6, {b1, b2, b3, b4, b5, b6, b7, b8})
+      when is_integer(b1) and b1 in 0..65_535 and is_integer(b2) and b2 in 0..65_535 and
+             is_integer(b3) and b3 in 0..65_535 and is_integer(b4) and b4 in 0..65_535 and
+             is_integer(b5) and b5 in 0..65_535 and is_integer(b6) and b6 in 0..65_535 and
+             is_integer(b7) and b7 in 0..65_535 and is_integer(b8) and b8 in 0..65_535 do
     <<b1::16, b2::16, b3::16, b4::16, b5::16, b6::16, b7::16, b8::16>>
   end
 
+  def encode(:ipv6, {_, _, _, _, _, _, _, _} = address) do
+    raise ArgumentError, "invalid IPv6 address: #{inspect(address)}"
+  end
+
   def encode(:ipv6, <<_::128>> = encoded), do: encoded
   def encode(:ipv6, nil), do: <<0::128>>
 
@@ -1477,6 +1502,86 @@ defmodule Ch.RowBinary do
     {:lists.reverse(rows), bin, {:cont, types_rest, row}}
   end
 
+  defp decimal_coefficient!(decimal, scale, size) do
+    unless is_integer(decimal.coef) do
+      raise ArgumentError, "ClickHouse Decimal values must be finite"
+    end
+
+    %Decimal{sign: sign, coef: coefficient, exp: exponent} = decimal
+    shift = exponent + scale
+    max_digits = size |> then(&((1 <<< (&1 - 1)) - 1)) |> Integer.digits() |> length()
+
+    cond do
+      coefficient == 0 ->
+        0
+
+      shift >= 0 and length(Integer.digits(coefficient)) + shift > max_digits ->
+        raise ArgumentError,
+              "Decimal#{size}(#{scale}) value #{Decimal.to_string(decimal)} is out of range"
+
+      shift >= 0 ->
+        sign * coefficient * Integer.pow(10, shift)
+
+      -shift > length(Integer.digits(coefficient)) ->
+        0
+
+      true ->
+        divisor = Integer.pow(10, -shift)
+        quotient = div(coefficient, divisor)
+        remainder = rem(coefficient, divisor)
+        rounded = if remainder * 2 >= divisor, do: quotient + 1, else: quotient
+        sign * rounded
+    end
+  end
+
+  defp encode_decimal!(decimal, coefficient, size, scale) do
+    encode_fixed_integer!(
+      coefficient,
+      size,
+      :signed,
+      "Decimal#{size}(#{scale}) value #{Decimal.to_string(decimal)}"
+    )
+  end
+
+  defp validate_decimal_type!(precision, scale)
+       when is_integer(precision) and precision in 1..76 and is_integer(scale) and scale >= 0 and
+              scale <= precision,
+       do: :ok
+
+  defp validate_decimal_type!(precision, scale) do
+    raise ArgumentError,
+          "invalid Decimal precision and scale: precision=#{inspect(precision)}, scale=#{inspect(scale)}"
+  end
+
+  defp validate_decimal_scale!(_type, scale, precision)
+       when is_integer(scale) and scale >= 0 and scale <= precision,
+       do: :ok
+
+  defp validate_decimal_scale!(type, scale, precision) do
+    raise ArgumentError,
+          "invalid #{decimal_type_name(type)} scale #{inspect(scale)}; expected 0..#{precision}"
+  end
+
+  defp decimal_type_name(type) do
+    type |> Atom.to_string() |> String.replace_prefix("decimal", "Decimal")
+  end
+
+  defp encode_fixed_integer!(integer, size, :unsigned, _type)
+       when is_integer(integer) and integer >= 0 and integer < 1 <<< size do
+    <<integer::little-unsigned-size(size)>>
+  end
+
+  defp encode_fixed_integer!(integer, size, :signed, _type)
+       when is_integer(integer) and integer >= -(1 <<< (size - 1)) and
+              integer < 1 <<< (size - 1) do
+    <<integer::little-signed-size(size)>>
+  end
+
+  defp encode_fixed_integer!(integer, size, signedness, type) do
+    raise ArgumentError,
+          "#{type} is out of range for a #{size}-bit #{signedness} integer: #{inspect(integer)}"
+  end
+
   @compile inline: [decimal_size: 1]
   # https://clickhouse.com/docs/en/sql-reference/data-types/decimal/
   defp decimal_size(precision) when is_integer(precision) do
diff --git a/lib/ch/types.ex b/lib/ch/types.ex
index 29a41d1e..3c03894b 100644
--- a/lib/ch/types.ex
+++ b/lib/ch/types.ex
@@ -419,7 +419,7 @@ defmodule Ch.Types do
   end
 
   defp decode([:string | stack], <<?', rest::bytes>>, acc) do
-    decode_string(rest, 0, rest, stack, acc)
+    decode_string(rest, 0, rest, [], stack, acc)
   end
 
   defp decode([:int | stack], <<rest::bytes>>, acc) do
@@ -482,20 +482,28 @@ defmodule Ch.Types do
   defp build_type(:map = m, [v, k]), do: {m, k, v}
   defp build_type(:nullable = n, [t]), do: {n, t}
   defp build_type(:low_cardinality = l, [t]), do: {l, t}
-  defp build_type(:enum8 = e, mapping), do: {e, build_enum_mapping(mapping)}
-  defp build_type(:enum16 = e, mapping), do: {e, build_enum_mapping(mapping)}
+  defp build_type(:enum8 = e, mapping), do: {e, build_enum_mapping!(e, mapping)}
+  defp build_type(:enum16 = e, mapping), do: {e, build_enum_mapping!(e, mapping)}
   defp build_type(:simple_aggregate_function = saf, [t, f]), do: {saf, f, t}
-  defp build_type(:decimal32 = d, [s]), do: {d, s}
-  defp build_type(:decimal64 = d, [s]), do: {d, s}
-  defp build_type(:decimal128 = d, [s]), do: {d, s}
-  defp build_type(:decimal256 = d, [s]), do: {d, s}
-  defp build_type(:decimal = d, [s, p]), do: {d, p, s}
+  defp build_type(:decimal32 = d, [s]), do: build_decimal_type!(d, s, 9)
+  defp build_type(:decimal64 = d, [s]), do: build_decimal_type!(d, s, 18)
+  defp build_type(:decimal128 = d, [s]), do: build_decimal_type!(d, s, 38)
+  defp build_type(:decimal256 = d, [s]), do: build_decimal_type!(d, s, 76)
+  defp build_type(:decimal = d, [s, p]), do: validate_decimal_type!({d, p, s})
   defp build_type(:time64 = t, [precision]), do: {t, precision}
   defp build_type(:variant = v, ts), do: {v, build_variant(ts)}
   defp build_type(:dynamic, _max_types), do: :dynamic
 
-  defp build_enum_mapping(mapping) do
-    mapping |> :lists.reverse() |> Enum.chunk_every(2) |> Enum.map(fn [k, v] -> {k, v} end)
+  defp build_enum_mapping!(type, mapping) do
+    mapping =
+      mapping |> :lists.reverse() |> Enum.chunk_every(2) |> Enum.map(fn [k, v] -> {k, v} end)
+
+    validate_enum_mapping!(type, mapping)
+  end
+
+  defp build_decimal_type!(type, scale, precision) do
+    validate_decimal_type!({:decimal, precision, scale})
+    {type, scale}
   end
 
   defp build_tuple([type, name | rest]) when is_binary(name) do
@@ -514,17 +522,34 @@ defmodule Ch.Types do
     Enum.sort_by(types, fn t -> IO.iodata_to_binary(encode(t)) end)
   end
 
-  # TODO '', \'
+  defp decode_string(<<?', rest::bytes>>, len, original, parts, stack, acc) do
+    part = :binary.part(original, 0, len)
+    string = parts |> then(&[part | &1]) |> Enum.reverse() |> IO.iodata_to_binary()
+    decode(stack, rest, [string | acc])
+  end
 
-  defp decode_string(<<?', rest::bytes>>, len, original, stack, acc) do
+  defp decode_string(<<?\\, escaped, rest::bytes>>, len, original, parts, stack, acc) do
     part = :binary.part(original, 0, len)
-    decode(stack, rest, [:binary.copy(part) | acc])
+    decoded = decode_escaped_char(escaped)
+    decode_string(rest, 0, rest, [decoded, part | parts], stack, acc)
+  end
+
+  defp decode_string(<<u::utf8, rest::bytes>>, len, original, parts, stack, acc) do
+    decode_string(rest, len + utf8_size(u), original, parts, stack, acc)
   end
 
-  defp decode_string(<<u::utf8, rest::bytes>>, len, original, stack, acc) do
-    decode_string(rest, len + utf8_size(u), original, stack, acc)
+  defp decode_string(<<>>, _len, _original, _parts, _stack, _acc) do
+    raise ArgumentError, "unexpected end of quoted string while decoding"
   end
 
+  defp decode_escaped_char(?0), do: <<0>>
+  defp decode_escaped_char(?b), do: <<8>>
+  defp decode_escaped_char(?f), do: <<12>>
+  defp decode_escaped_char(?n), do: <<10>>
+  defp decode_escaped_char(?r), do: <<13>>
+  defp decode_escaped_char(?t), do: <<9>>
+  defp decode_escaped_char(char), do: <<char>>
+
   @compile inline: [utf8_size: 1]
   defp utf8_size(codepoint) when codepoint <= 0x7F, do: 1
   defp utf8_size(codepoint) when codepoint <= 0x7FF, do: 2
@@ -609,6 +634,7 @@ defmodule Ch.Types do
   end
 
   def encode({:decimal, precision, scale}) do
+    {:decimal, precision, scale} = validate_decimal_type!({:decimal, precision, scale})
     ["Decimal(", Integer.to_string(precision), ", ", Integer.to_string(scale), ?)]
   end
 
@@ -629,6 +655,7 @@ defmodule Ch.Types do
   end
 
   def encode({:enum8, mapping}) do
+    validate_enum_mapping!(:enum8, mapping)
     ["Enum8('", encode_mapping(mapping), ?)]
   end
 
@@ -637,6 +664,7 @@ defmodule Ch.Types do
   end
 
   def encode({:enum16, mapping}) do
+    validate_enum_mapping!(:enum16, mapping)
     ["Enum16('", encode_mapping(mapping), ?)]
   end
 
@@ -655,10 +683,68 @@ defmodule Ch.Types do
   defp encode_intersperse([] = empty, _separator), do: empty
 
   defp encode_mapping([{k, v}]) when is_binary(k) do
-    [k, "' = ", Integer.to_string(v)]
+    [encode_quoted_string(k), "' = ", Integer.to_string(v)]
   end
 
   defp encode_mapping([{k, v} | mapping]) when is_binary(k) do
-    [k, "' = ", Integer.to_string(v), ", '" | encode_mapping(mapping)]
+    [encode_quoted_string(k), "' = ", Integer.to_string(v), ", '" | encode_mapping(mapping)]
+  end
+
+  defp encode_quoted_string(string) do
+    for <<char <- string>> do
+      case char do
+        0 -> "\\0"
+        8 -> "\\b"
+        9 -> "\\t"
+        10 -> "\\n"
+        12 -> "\\f"
+        13 -> "\\r"
+        ?' -> "\\'"
+        ?\\ -> "\\\\"
+        _ -> char
+      end
+    end
   end
+
+  defp validate_decimal_type!({:decimal, precision, scale} = type)
+       when is_integer(precision) and precision in 1..76 and is_integer(scale) and scale >= 0 and
+              scale <= precision,
+       do: type
+
+  defp validate_decimal_type!({:decimal, precision, scale}) do
+    raise ArgumentError,
+          "invalid Decimal precision and scale: precision=#{inspect(precision)}, scale=#{inspect(scale)}"
+  end
+
+  defp validate_enum_mapping!(type, mapping) do
+    {name, min, max} = enum_metadata(type)
+
+    if mapping == [] do
+      raise ArgumentError, "#{name} requires at least one mapping"
+    end
+
+    unless is_list(mapping) and
+             Enum.all?(mapping, fn
+               {label, value}
+               when is_binary(label) and is_integer(value) and value >= min and value <= max ->
+                 true
+
+               _ ->
+                 false
+             end) do
+      raise ArgumentError, "invalid #{name} mapping: #{inspect(mapping)}"
+    end
+
+    labels = Enum.map(mapping, &elem(&1, 0))
+    values = Enum.map(mapping, &elem(&1, 1))
+
+    if Enum.uniq(labels) != labels or Enum.uniq(values) != values do
+      raise ArgumentError, "#{name} mapping labels and values must be unique"
+    end
+
+    mapping
+  end
+
+  defp enum_metadata(:enum8), do: {"Enum8", -128, 127}
+  defp enum_metadata(:enum16), do: {"Enum16", -32_768, 32_767}
 end
diff --git a/test/ch/connection_integration_test.exs b/test/ch/connection_integration_test.exs
new file mode 100644
index 00000000..9482c2c2
--- /dev/null
+++ b/test/ch/connection_integration_test.exs
@@ -0,0 +1,70 @@
+defmodule Ch.ConnectionIntegrationTest do
+  use ExUnit.Case, async: true
+
+  setup do
+    {:ok, pool: start_supervised!(Ch)}
+  end
+
+  test "runs concurrent queries", %{pool: pool} do
+    parent = self()
+
+    for _ <- 1..10 do
+      spawn_link(fn -> send(parent, Ch.query!(pool, "SELECT sleep(0.05)").rows) end)
+    end
+
+    assert Ch.query!(pool, "SELECT 42").rows == [[42]]
+
+    for _ <- 1..10 do
+      assert_receive [[0]]
+    end
+  end
+
+  test "identifier params can address tables", %{pool: pool} do
+    Help.query!("CREATE TABLE connection_integration_identifier_params (a UInt8) ENGINE Memory")
+
+    on_exit(fn ->
+      Help.query!("DROP TABLE connection_integration_identifier_params")
+    end)
+
+    Ch.query!(pool, "INSERT INTO {table:Identifier} VALUES (1), (2)", %{
+      "table" => "connection_integration_identifier_params"
+    })
+
+    assert Ch.query!(pool, "SELECT sum(a) FROM {table:Identifier}", %{
+             "table" => "connection_integration_identifier_params"
+           }).rows == [[3]]
+  end
+
+  test "supports RowBinaryWithNamesAndTypes payloads", %{pool: pool} do
+    Help.query!("""
+    CREATE TABLE connection_integration_rowbinary_names_types (
+      country_code FixedString(2),
+      rare_string LowCardinality(String),
+      maybe_int32 Nullable(Int32)
+    ) ENGINE Memory
+    """)
+
+    on_exit(fn ->
+      Help.query!("DROP TABLE connection_integration_rowbinary_names_types")
+    end)
+
+    names = ["country_code", "rare_string", "maybe_int32"]
+    types = ["FixedString(2)", "LowCardinality(String)", "Nullable(Int32)"]
+    rows = [["AB", "rare", -42], ["CD", "another", nil]]
+
+    rowbinary = [
+      Ch.RowBinary.encode_names_and_types(names, types)
+      | Ch.RowBinary.encode_rows(rows, types)
+    ]
+
+    Ch.query!(pool, [
+      "INSERT INTO connection_integration_rowbinary_names_types FORMAT RowBinaryWithNamesAndTypes\n"
+      | rowbinary
+    ])
+
+    assert Ch.query!(
+             pool,
+             "SELECT * FROM connection_integration_rowbinary_names_types ORDER BY country_code"
+           ).rows == rows
+  end
+end
diff --git a/test/ch/connection_property_test.exs b/test/ch/connection_property_test.exs
deleted file mode 100644
index 5c945e59..00000000
--- a/test/ch/connection_property_test.exs
+++ /dev/null
@@ -1,209 +0,0 @@
-defmodule Ch.ConnectionPropertyTest do
-  use ExUnit.Case, async: true
-  use ExUnitProperties
-
-  setup do
-    {:ok, pool: start_supervised!(Ch)}
-  end
-
-  describe "query/4" do
-    test "selects rows and column names", %{pool: pool} do
-      assert %{names: ["one", "two"], rows: [[1, 2]]} =
-               Ch.query!(pool, "SELECT 1 AS one, 2 AS two")
-    end
-
-    test "accepts iodata statements", %{pool: pool} do
-      assert Ch.query!(pool, ["S", ?E, ["LEC" | "T"], " ", ~c"123"]).rows == [[123]]
-    end
-
-    test "returns ClickHouse errors", %{pool: pool} do
-      assert {:error, %Ch.Error{message: message}} = Ch.query(pool, "wat")
-      assert message =~ "Code: 62"
-      assert message =~ "SYNTAX_ERROR"
-    end
-
-    test "reuses the pool after a query error", %{pool: pool} do
-      assert {:error, %Ch.Error{}} = Ch.query(pool, "SELECT 123 + 'a'")
-      assert Ch.query!(pool, "SELECT 42").rows == [[42]]
-    end
-
-    test "runs concurrent queries", %{pool: pool} do
-      parent = self()
-
-      for _ <- 1..10 do
-        spawn_link(fn -> send(parent, Ch.query!(pool, "SELECT sleep(0.05)").rows) end)
-      end
-
-      assert Ch.query!(pool, "SELECT 42").rows == [[42]]
-
-      for _ <- 1..10 do
-        assert_receive [[0]]
-      end
-    end
-  end
-
-  describe "query params" do
-    property "scalar params round-trip through ClickHouse", %{pool: pool} do
-      check all {type, value, expected} <- scalar_param() do
-        assert Ch.query!(pool, "SELECT {value:#{type}}", %{"value" => value}).rows == [[expected]]
-      end
-    end
-
-    property "array params round-trip through ClickHouse", %{pool: pool} do
-      check all {type, values, expected} <- array_param() do
-        assert Ch.query!(pool, "SELECT {value:Array(#{type})}", %{"value" => values}).rows == [
-                 [expected]
-               ]
-      end
-    end
-
-    test "identifier params can address tables", %{pool: pool} do
-      Help.query!("CREATE TABLE connection_property_identifier_params (a UInt8) ENGINE Memory")
-      on_exit(fn -> Help.query!("DROP TABLE connection_property_identifier_params") end)
-
-      Ch.query!(pool, "INSERT INTO {table:Identifier} VALUES (1), (2)", %{
-        "table" => "connection_property_identifier_params"
-      })
-
-      assert Ch.query!(pool, "SELECT sum(a) FROM {table:Identifier}", %{
-               "table" => "connection_property_identifier_params"
-             }).rows == [[3]]
-    end
-  end
-
-  describe "RowBinary inserts" do
-    property "rows encoded as RowBinary can be inserted and selected", %{pool: pool} do
-      Help.query!("""
-      CREATE TABLE connection_property_rowbinary (
-        id UInt8,
-        name String,
-        active Bool
-      ) ENGINE Memory
-      """)
-
-      on_exit(fn -> Help.query!("DROP TABLE connection_property_rowbinary") end)
-
-      check all rows <- rowbinary_rows() do
-        Ch.query!(pool, "TRUNCATE TABLE connection_property_rowbinary")
-
-        rowbinary = Ch.RowBinary.encode_rows(rows, ["UInt8", "String", "Bool"])
-
-        Ch.query!(pool, [
-          "INSERT INTO connection_property_rowbinary FORMAT RowBinary\n" | rowbinary
-        ])
-
-        assert Ch.query!(pool, "SELECT * FROM connection_property_rowbinary ORDER BY id").rows ==
-                 Enum.sort_by(rows, &List.first/1)
-      end
-    end
-
-    test "supports RowBinaryWithNamesAndTypes payloads", %{pool: pool} do
-      Help.query!("""
-      CREATE TABLE connection_property_rowbinary_names_types (
-        country_code FixedString(2),
-        rare_string LowCardinality(String),
-        maybe_int32 Nullable(Int32)
-      ) ENGINE Memory
-      """)
-
-      on_exit(fn ->
-        Help.query!("DROP TABLE connection_property_rowbinary_names_types")
-      end)
-
-      names = ["country_code", "rare_string", "maybe_int32"]
-      types = ["FixedString(2)", "LowCardinality(String)", "Nullable(Int32)"]
-      rows = [["AB", "rare", -42], ["CD", "another", nil]]
-
-      rowbinary = [
-        Ch.RowBinary.encode_names_and_types(names, types)
-        | Ch.RowBinary.encode_rows(rows, types)
-      ]
-
-      Ch.query!(pool, [
-        "INSERT INTO connection_property_rowbinary_names_types FORMAT RowBinaryWithNamesAndTypes\n"
-        | rowbinary
-      ])
-
-      assert Ch.query!(
-               pool,
-               "SELECT * FROM connection_property_rowbinary_names_types ORDER BY country_code"
-             ).rows ==
-               rows
-    end
-  end
-
-  defp scalar_param do
-    one_of([
-      gen_constant("UInt8", integer(0..255)),
-      gen_constant("Int16", integer(-32_768..32_767)),
-      gen_constant("Bool", boolean()),
-      gen_constant("String", safe_string()),
-      gen_constant("Date", date_gen()),
-      gen_constant("Date32", date32_gen()),
-      gen_constant("Decimal(18, 4)", decimal_gen())
-    ])
-  end
-
-  defp gen_constant(type, generator) do
-    gen all value <- generator do
-      expected =
-        case type do
-          "Decimal(18, 4)" -> Decimal.round(value, 4)
-          _ -> value
-        end
-
-      {type, value, expected}
-    end
-  end
-
-  defp array_param do
-    one_of([
-      gen_array("UInt8", integer(0..255)),
-      gen_array("Int16", integer(-32_768..32_767)),
-      gen_array("Bool", boolean()),
-      gen_array("String", safe_string()),
-      gen_array("Date", date_gen())
-    ])
-  end
-
-  defp gen_array(type, generator) do
-    gen all values <- list_of(generator, max_length: 8) do
-      {type, values, values}
-    end
-  end
-
-  defp rowbinary_rows do
-    uniq_list_of(
-      fixed_list([
-        integer(0..255),
-        safe_string(),
-        boolean()
-      ]),
-      max_length: 12
-    )
-  end
-
-  defp safe_string do
-    string(:printable, max_length: 32)
-  end
-
-  defp date_gen do
-    gen all days <- integer(0..20_000) do
-      Date.add(~D[1970-01-01], days)
-    end
-  end
-
-  defp date32_gen do
-    gen all days <- integer(-25_567..120_529) do
-      Date.add(~D[1970-01-01], days)
-    end
-  end
-
-  defp decimal_gen do
-    gen all sign <- member_of([1, -1]),
-            coef <- integer(0..999_999_999),
-            exp <- integer(-4..4) do
-      Decimal.new(sign, coef, exp)
-    end
-  end
-end
diff --git a/test/ch/decimal_param_test.exs b/test/ch/decimal_param_test.exs
index 0e4d5e6e..69ae30f6 100644
--- a/test/ch/decimal_param_test.exs
+++ b/test/ch/decimal_param_test.exs
@@ -2,6 +2,8 @@ defmodule Ch.DecimalParamTest do
   use ExUnit.Case, async: true
   use ExUnitProperties
 
+  alias Ch.RowBinary
+
   setup do
     {:ok, pool: start_supervised!(Ch), query_options: []}
   end
@@ -73,6 +75,37 @@ defmodule Ch.DecimalParamTest do
     assert_decimal_param(ctx, Decimal.new(1, 1, -77), "Decimal(76, 76)", Decimal.new("0E-76"))
   end
 
+  test "decodes every RowBinary decimal width", %{pool: pool} do
+    assert Ch.query!(pool, """
+           SELECT
+             toDecimal32(2, 4),
+             toDecimal64(2, 4),
+             toDecimal128(2, 4),
+             toDecimal256(2, 4)
+           """).rows == [
+             List.duplicate(Decimal.new("2.0000"), 4)
+           ]
+  end
+
+  test "RowBinary decimal inserts apply scale and rounding", %{pool: pool} do
+    Help.query!("CREATE TABLE decimal_param_rowbinary(d Decimal32(4)) ENGINE Memory")
+    on_exit(fn -> Help.query!("DROP TABLE decimal_param_rowbinary") end)
+
+    rowbinary =
+      RowBinary.encode_rows(
+        [[Decimal.new("2.66")], [Decimal.new("2.6666")], [Decimal.new("2.66666")]],
+        ["Decimal32(4)"]
+      )
+
+    Ch.query!(pool, ["INSERT INTO decimal_param_rowbinary FORMAT RowBinary\n" | rowbinary])
+
+    assert Ch.query!(pool, "SELECT * FROM decimal_param_rowbinary").rows == [
+             [Decimal.new("2.6600")],
+             [Decimal.new("2.6666")],
+             [Decimal.new("2.6667")]
+           ]
+  end
+
   property "compact exponent Decimal integer params round-trip", ctx do
     check all decimal <- compact_decimal_integer() do
       assert_decimal_param(ctx, decimal, "Decimal(76, 0)", decimal)
diff --git a/test/ch/insert_test.exs b/test/ch/insert_test.exs
new file mode 100644
index 00000000..3edecd15
--- /dev/null
+++ b/test/ch/insert_test.exs
@@ -0,0 +1,118 @@
+defmodule Ch.InsertTest do
+  use ExUnit.Case, async: true
+
+  alias Ch.RowBinary
+
+  setup do
+    {:ok, pool: start_supervised!(Ch)}
+  end
+
+  test "inserts heterogeneous rows as one RowBinary stream", %{pool: pool} do
+    Help.query!("""
+    CREATE TABLE insert_matrix (
+      id UInt8,
+      signed Int64,
+      unsigned UInt64,
+      float64 Float64,
+      decimal Decimal(18, 4),
+      active Bool,
+      string String,
+      fixed FixedString(4),
+      nullable Nullable(String),
+      uuid UUID,
+      date Date,
+      datetime DateTime64(6, 'UTC'),
+      ints Array(Int16),
+      tuple Tuple(String, Int8),
+      map Map(String, UInt8)
+    ) ENGINE Memory
+    """)
+
+    on_exit(fn -> Help.query!("DROP TABLE insert_matrix") end)
+
+    uuid = Base.decode16!("417DDC5DE5564D2795DDA34D84E46A50")
+    zero_uuid = <<0::128>>
+
+    rows = [
+      [
+        1,
+        -42,
+        42,
+        1.5,
+        Decimal.new("12.3400"),
+        true,
+        "line\nwith\ttabs",
+        "AB",
+        nil,
+        uuid,
+        ~D[2024-02-29],
+        ~U[2024-02-29 12:34:56.123456Z],
+        [-2, -1, 0, 1, 2],
+        {"tuple", -8},
+        %{"a" => 1, "b" => 2}
+      ],
+      [
+        2,
+        0,
+        0,
+        0.0,
+        Decimal.new("0.0000"),
+        false,
+        "",
+        "",
+        "",
+        zero_uuid,
+        ~D[1970-01-01],
+        ~U[1970-01-01 00:00:00.000000Z],
+        [],
+        {"", 0},
+        %{}
+      ],
+      [
+        3,
+        -9_223_372_036_854_775_808,
+        18_446_744_073_709_551_615,
+        1.7976931348623157e308,
+        Decimal.new("99999999999999.9999"),
+        true,
+        <<0, 255>>,
+        "WXYZ",
+        nil,
+        uuid,
+        ~D[2100-01-01],
+        ~U[2100-01-01 23:59:59.999999Z],
+        [-32_768, 32_767],
+        {"edge", 127},
+        %{"max" => 255}
+      ]
+    ]
+
+    types = [
+      "UInt8",
+      "Int64",
+      "UInt64",
+      "Float64",
+      "Decimal(18, 4)",
+      "Bool",
+      "String",
+      "FixedString(4)",
+      "Nullable(String)",
+      "UUID",
+      "Date",
+      "DateTime64(6, 'UTC')",
+      "Array(Int16)",
+      "Tuple(String, Int8)",
+      "Map(String, UInt8)"
+    ]
+
+    rowbinary = RowBinary.encode_rows(rows, types)
+    Ch.query!(pool, ["INSERT INTO insert_matrix FORMAT RowBinary\n" | rowbinary])
+
+    expected =
+      rows
+      |> Enum.map(fn row -> List.update_at(row, 7, &String.pad_trailing(&1, 4, <<0>>)) end)
+      |> Enum.sort_by(&List.first/1)
+
+    assert Ch.query!(pool, "SELECT * FROM insert_matrix ORDER BY id").rows == expected
+  end
+end
diff --git a/test/ch/query_string_test.exs b/test/ch/query_string_test.exs
index 6033ec32..e1afcb3a 100644
--- a/test/ch/query_string_test.exs
+++ b/test/ch/query_string_test.exs
@@ -231,11 +231,7 @@ defmodule Ch.QueryStringTest do
              ).rows == [[strings, {"'", "\\", "x'), 1; select 1 --"}, map]]
     end
 
-    test "string parameters are escaped", %{pool: pool} do
-      for s <- ["\t", "\n", "\\", "'", "\b", "\f", "\r", "\0"] do
-        assert Ch.query!(pool, "select {s:String}", %{"s" => s}).rows == [[s]]
-      end
-
+    test "escaped string params work as splitByChar arguments", %{pool: pool} do
       assert Ch.query!(pool, "select splitByChar('\t', 'abc\t123')").rows ==
                [[["abc", "123"]]]
 
diff --git a/test/ch/row_binary_array_test.exs b/test/ch/row_binary_array_test.exs
index 3e9662f6..2d778a90 100644
--- a/test/ch/row_binary_array_test.exs
+++ b/test/ch/row_binary_array_test.exs
@@ -9,14 +9,6 @@ defmodule Ch.RowBinaryArrayTest do
     {:ok, pool: start_supervised!(Ch)}
   end
 
-  property "array params round-trip through ClickHouse across integer widths", %{pool: pool} do
-    check all {type, values, expected} <- integer_array_param() do
-      assert Ch.query!(pool, "SELECT {value:Array(#{type})}", %{"value" => values}).rows == [
-               [expected]
-             ]
-    end
-  end
-
   property "array params round-trip through ClickHouse across element kinds", %{pool: pool} do
     check all {type, values, expected} <- array_param() do
       assert Ch.query!(pool, "SELECT {value:Array(#{type})}", %{"value" => values}).rows == [
@@ -217,23 +209,6 @@ defmodule Ch.RowBinaryArrayTest do
     assert message =~ "UInt8"
   end
 
-  defp integer_array_param do
-    one_of([
-      typed_array("Int8", integer(-128..127)),
-      typed_array("Int16", integer(-32_768..32_767)),
-      typed_array("Int32", integer(-2_147_483_648..2_147_483_647)),
-      typed_array("Int64", integer(-9_007_199_254_740_992..9_007_199_254_740_991)),
-      typed_array("Int128", signed_integer(128)),
-      typed_array("Int256", signed_integer(256)),
-      typed_array("UInt8", integer(0..255)),
-      typed_array("UInt16", integer(0..65_535)),
-      typed_array("UInt32", integer(0..4_294_967_295)),
-      typed_array("UInt64", integer(0..9_007_199_254_740_991)),
-      typed_array("UInt128", unsigned_integer(128)),
-      typed_array("UInt256", unsigned_integer(256))
-    ])
-  end
-
   defp integer_width_examples do
     [
       {"Int8", [-128, 0, 127]},
@@ -322,19 +297,6 @@ defmodule Ch.RowBinaryArrayTest do
     end
   end
 
-  defp signed_integer(bits) do
-    gen all unsigned <- unsigned_integer(bits) do
-      signed_limit = 1 <<< (bits - 1)
-      if unsigned >= signed_limit, do: unsigned - (1 <<< bits), else: unsigned
-    end
-  end
-
-  defp unsigned_integer(bits) do
-    gen all bytes <- binary(length: div(bits, 8)) do
-      :binary.decode_unsigned(bytes, :little)
-    end
-  end
-
   defp decimal_gen do
     gen all sign <- member_of([1, -1]),
             coef <- integer(0..999_999_999),
diff --git a/test/ch/row_binary_decimal_test.exs b/test/ch/row_binary_decimal_test.exs
new file mode 100644
index 00000000..8a702a31
--- /dev/null
+++ b/test/ch/row_binary_decimal_test.exs
@@ -0,0 +1,129 @@
+defmodule Ch.RowBinaryDecimalTest do
+  use ExUnit.Case, async: true
+  use ExUnitProperties
+
+  alias Ch.RowBinary
+
+  import Bitwise
+
+  @decimal_types [decimal32: 32, decimal64: 64, decimal128: 128, decimal256: 256]
+
+  property "valid coefficients preserve every signed bit pattern" do
+    check all {type, size, bytes, coefficient} <- decimal_coefficient() do
+      decimal = Decimal.new(coefficient)
+      encoded = type |> RowBinary.encode(decimal) |> IO.iodata_to_binary()
+
+      assert encoded == bytes
+      assert [[decoded]] = RowBinary.decode_rows(encoded, [type])
+      assert Decimal.equal?(decoded, decimal)
+      assert bit_size(encoded) == size
+    end
+  end
+
+  test "rejects coefficients outside each storage width" do
+    for {name, size} <- @decimal_types do
+      type = {name, 0}
+      min = -(1 <<< (size - 1))
+      max = (1 <<< (size - 1)) - 1
+
+      assert encoded_binary(type, Decimal.new(min)) == <<min::little-signed-size(size)>>
+      assert encoded_binary(type, Decimal.new(max)) == <<max::little-signed-size(size)>>
+
+      assert_raise ArgumentError, ~r/out of range/, fn ->
+        RowBinary.encode(type, Decimal.new(min - 1))
+      end
+
+      assert_raise ArgumentError, ~r/out of range/, fn ->
+        RowBinary.encode(type, Decimal.new(max + 1))
+      end
+    end
+  end
+
+  test "regression: Decimal32 overflow never wraps or becomes zero" do
+    for value <- ["2147483648", "4294967296"] do
+      assert_raise ArgumentError, ~r/out of range/, fn ->
+        RowBinary.encode({:decimal32, 0}, Decimal.new(value))
+      end
+    end
+  end
+
+  test "rejects overflow introduced by scaling or rounding" do
+    type = {:decimal32, 2}
+
+    assert_raise ArgumentError, ~r/out of range/, fn ->
+      RowBinary.encode(type, Decimal.new("21474836.48"))
+    end
+
+    assert_raise ArgumentError, ~r/out of range/, fn ->
+      RowBinary.encode({:decimal32, 0}, Decimal.new("2147483647.5"))
+    end
+
+    assert_raise ArgumentError, ~r/out of range/, fn ->
+      RowBinary.encode({:decimal32, 0}, Decimal.new("-2147483648.5"))
+    end
+  end
+
+  test "declared precision rejects coefficients that still fit the storage width" do
+    for precision <- [1, 9, 10, 18, 19, 38, 39, 76] do
+      max = Integer.pow(10, precision) - 1
+      type = {:decimal, precision, 0}
+
+      assert Decimal.equal?(
+               type
+               |> RowBinary.encode(Decimal.new(max))
+               |> IO.iodata_to_binary()
+               |> then(&RowBinary.decode_rows(&1, [type]))
+               |> get_in([Access.at(0), Access.at(0)]),
+               Decimal.new(max)
+             )
+
+      assert_raise ArgumentError, ~r/exceeds precision/, fn ->
+        RowBinary.encode(type, Decimal.new(max + 1))
+      end
+
+      assert_raise ArgumentError, ~r/exceeds precision/, fn ->
+        RowBinary.encode(type, Decimal.new(-max - 1))
+      end
+    end
+  end
+
+  test "rejects non-finite decimals" do
+    for decimal <- [Decimal.new("NaN"), Decimal.new("Infinity"), Decimal.new("-Infinity")] do
+      assert_raise ArgumentError, "ClickHouse Decimal values must be finite", fn ->
+        RowBinary.encode({:decimal32, 0}, decimal)
+      end
+    end
+  end
+
+  test "rejects invalid precision and scale before encoding rows" do
+    for type <- [
+          {:decimal, 0, 0},
+          {:decimal, 77, 0},
+          {:decimal, 9, -1},
+          {:decimal, 9, 10},
+          {:decimal32, 10},
+          {:decimal64, 19},
+          {:decimal128, 39},
+          {:decimal256, 77}
+        ] do
+      assert_raise ArgumentError, ~r/invalid Decimal/, fn ->
+        RowBinary.encode_rows([[Decimal.new(0)]], [type])
+      end
+    end
+  end
+
+  defp decimal_coefficient do
+    gen all {name, size} <- member_of(@decimal_types),
+            bytes <- binary(length: div(size, 8)) do
+      unsigned = :binary.decode_unsigned(bytes, :little)
+      signed_limit = 1 <<< (size - 1)
+      coefficient = if unsigned >= signed_limit, do: unsigned - (1 <<< size), else: unsigned
+
+      {{name, 0}, size, bytes, coefficient}
+    end
+  end
+
+  defp encoded_binary(type, decimal) do
+    type |> RowBinary.encode(decimal) |> IO.iodata_to_binary()
+  end
+end
diff --git a/test/ch/row_binary_float_test.exs b/test/ch/row_binary_float_test.exs
index 0b97c175..8f80090c 100644
--- a/test/ch/row_binary_float_test.exs
+++ b/test/ch/row_binary_float_test.exs
@@ -85,6 +85,20 @@ defmodule Ch.RowBinaryFloatTest do
     assert message =~ "query parameter"
   end
 
+  test "RowBinary rejects values that overflow their float width" do
+    for value <- [3.5e38, -3.5e38, 1.0e300, -1.0e300] do
+      assert_raise ArgumentError, ~r/invalid Float32/, fn ->
+        RowBinary.encode(:f32, value)
+      end
+    end
+
+    huge_integer = Integer.pow(10, 1_000)
+
+    assert_raise ArgumentError, ~r/invalid Float64/, fn ->
+      RowBinary.encode(:f64, huge_integer)
+    end
+  end
+
   property "RowBinary float inserts round-trip through ClickHouse", %{pool: pool} do
     Help.query!("""
     CREATE TABLE row_binary_float_property (
diff --git a/test/ch/row_binary_integer_test.exs b/test/ch/row_binary_integer_test.exs
index e6f41d56..02c13bd0 100644
--- a/test/ch/row_binary_integer_test.exs
+++ b/test/ch/row_binary_integer_test.exs
@@ -24,6 +24,16 @@ defmodule Ch.RowBinaryIntegerTest do
     i256: 256
   ]
 
+  setup do
+    {:ok, pool: start_supervised!(Ch)}
+  end
+
+  property "params round-trip through ClickHouse across integer widths", %{pool: pool} do
+    check all {type, value} <- integer_param() do
+      assert Ch.query!(pool, "SELECT {value:#{type}}", %{"value" => value}).rows == [[value]]
+    end
+  end
+
   describe "unsigned integers" do
     property "encode and decode all bit patterns as little-endian values" do
       check all {type, bits, bytes, value} <- uint_value() do
@@ -109,6 +119,23 @@ defmodule Ch.RowBinaryIntegerTest do
     end
   end
 
+  defp integer_param do
+    one_of([
+      typed_integer("Int8", -128..127),
+      typed_integer("Int16", -32_768..32_767),
+      typed_integer("Int32", -2_147_483_648..2_147_483_647),
+      typed_integer("Int64", -9_007_199_254_740_992..9_007_199_254_740_991),
+      typed_integer("UInt8", 0..255),
+      typed_integer("UInt16", 0..65_535),
+      typed_integer("UInt32", 0..4_294_967_295),
+      typed_integer("UInt64", 0..9_007_199_254_740_991)
+    ])
+  end
+
+  defp typed_integer(type, range) do
+    gen(all value <- integer(range), do: {type, value})
+  end
+
   defp int_value do
     gen all {type, bits} <- member_of(@int_types),
             bytes <- binary(length: div(bits, 8)) do
diff --git a/test/ch/row_binary_network_test.exs b/test/ch/row_binary_network_test.exs
new file mode 100644
index 00000000..d00c38df
--- /dev/null
+++ b/test/ch/row_binary_network_test.exs
@@ -0,0 +1,47 @@
+defmodule Ch.RowBinaryNetworkTest do
+  use ExUnit.Case, async: true
+  use ExUnitProperties
+
+  alias Ch.RowBinary
+
+  property "IPv4 octets round-trip without reordering or truncation" do
+    check all octets <- fixed_list(List.duplicate(integer(0..255), 4)) do
+      address = List.to_tuple(octets)
+      assert RowBinary.decode_rows(encoded(:ipv4, address), [:ipv4]) == [[address]]
+    end
+  end
+
+  property "IPv6 segments preserve every 16-bit value" do
+    check all bytes <- binary(length: 16) do
+      address =
+        bytes |> :binary.bin_to_list() |> Enum.chunk_every(2) |> Enum.map(&decode_segment/1)
+
+      address = List.to_tuple(address)
+
+      assert encoded(:ipv6, address) == bytes
+      assert RowBinary.decode_rows(bytes, [:ipv6]) == [[address]]
+    end
+  end
+
+  test "rejects invalid IPv4 octets instead of returning invalid iodata" do
+    for invalid <- [-1, 256, "1", nil] do
+      assert_raise ArgumentError, ~r/invalid IPv4/, fn ->
+        RowBinary.encode(:ipv4, {127, 0, 0, invalid})
+      end
+    end
+  end
+
+  test "rejects invalid IPv6 segments instead of truncating them" do
+    for invalid <- [-1, 65_536, "1", nil] do
+      assert_raise ArgumentError, ~r/invalid IPv6/, fn ->
+        RowBinary.encode(:ipv6, {0, 0, 0, 0, 0, 0, 0, invalid})
+      end
+    end
+  end
+
+  defp decode_segment([high, low]), do: high * 256 + low
+
+  defp encoded(type, value) do
+    type |> RowBinary.encode(value) |> IO.iodata_to_binary()
+  end
+end
diff --git a/test/ch/row_binary_string_test.exs b/test/ch/row_binary_string_test.exs
index 203f6b09..7c94b846 100644
--- a/test/ch/row_binary_string_test.exs
+++ b/test/ch/row_binary_string_test.exs
@@ -12,6 +12,16 @@ defmodule Ch.RowBinaryStringTest do
     {:ok, pool: start_supervised!(Ch)}
   end
 
+  property "FixedString params are padded to their declared size", %{pool: pool} do
+    check all size <- integer(1..12),
+              value <- string(:alphanumeric, max_length: size) do
+      padding = :binary.copy(<<0>>, size - byte_size(value))
+
+      assert Ch.query!(pool, "SELECT {value:FixedString(#{size})}", %{"value" => value}).rows ==
+               [[value <> padding]]
+    end
+  end
+
   property "String values inserted as RowBinary round-trip through ClickHouse", %{pool: pool} do
     Help.query!("CREATE TABLE #{@string_table}(id UInt8, s String) ENGINE Memory")
     on_exit(fn -> Help.query!("DROP TABLE #{@string_table}") end)
diff --git a/test/ch/row_binary_temporal_boundary_test.exs b/test/ch/row_binary_temporal_boundary_test.exs
new file mode 100644
index 00000000..62afb750
--- /dev/null
+++ b/test/ch/row_binary_temporal_boundary_test.exs
@@ -0,0 +1,58 @@
+defmodule Ch.RowBinaryTemporalBoundaryTest do
+  use ExUnit.Case, async: true
+
+  alias Ch.RowBinary
+
+  import Bitwise
+
+  @epoch ~D[1970-01-01]
+
+  test "Date accepts its UInt16 boundaries without wrapping" do
+    max_date = Date.add(@epoch, (1 <<< 16) - 1)
+
+    assert RowBinary.decode_rows(encoded(:date, @epoch), [:date]) == [[@epoch]]
+    assert RowBinary.decode_rows(encoded(:date, max_date), [:date]) == [[max_date]]
+
+    for date <- [Date.add(@epoch, -1), Date.add(max_date, 1)] do
+      assert_raise ArgumentError, ~r/out of range/, fn -> RowBinary.encode(:date, date) end
+    end
+  end
+
+  test "DateTime accepts its UInt32 boundaries without wrapping" do
+    max_seconds = (1 <<< 32) - 1
+    epoch = DateTime.from_unix!(0)
+    max_datetime = DateTime.from_unix!(max_seconds)
+
+    assert RowBinary.decode_rows(encoded(:datetime, epoch), [:datetime]) ==
+             [[DateTime.to_naive(epoch)]]
+
+    assert RowBinary.decode_rows(encoded(:datetime, max_datetime), [:datetime]) ==
+             [[DateTime.to_naive(max_datetime)]]
+
+    for datetime <- [DateTime.from_unix!(-1), DateTime.from_unix!(max_seconds + 1)] do
+      assert_raise ArgumentError, ~r/out of range/, fn ->
+        RowBinary.encode(:datetime, datetime)
+      end
+
+      assert_raise ArgumentError, ~r/out of range/, fn ->
+        RowBinary.encode(:datetime, DateTime.to_naive(datetime))
+      end
+    end
+  end
+
+  test "DateTime64 rejects tick counts outside Int64" do
+    datetime = ~U[9999-12-31 23:59:59.999999Z]
+
+    assert_raise ArgumentError, ~r/out of range/, fn ->
+      RowBinary.encode({:datetime64, 1_000_000_000}, datetime)
+    end
+
+    assert_raise ArgumentError, ~r/out of range/, fn ->
+      RowBinary.encode({:datetime64, 1_000_000_000}, DateTime.to_naive(datetime))
+    end
+  end
+
+  defp encoded(type, value) do
+    type |> RowBinary.encode(value) |> IO.iodata_to_binary()
+  end
+end
diff --git a/test/ch/row_binary_test.exs b/test/ch/row_binary_test.exs
index 1e0cd5ca..f019be70 100644
--- a/test/ch/row_binary_test.exs
+++ b/test/ch/row_binary_test.exs
@@ -8,85 +8,17 @@ defmodule Ch.RowBinaryTest do
     type |> encode(value) |> IO.iodata_to_binary()
   end
 
-  test "encode -> decode" do
+  test "heterogeneous values compose into one RowBinary stream" do
     spec = [
-      {:string, ""},
-      {:string, "a"},
-      {:string, String.duplicate("a", 500)},
-      {:string, String.duplicate("a", 15000)},
-      {{:fixed_string, 2}, <<0, 0>>},
-      {{:fixed_string, 2}, "a" <> <<0>>},
-      {{:fixed_string, 2}, "aa"},
-      {:u8, 0},
-      {:u8, 0xFF},
-      {:u16, 0},
-      {:u16, 0xFFFF},
-      {:u32, 0},
-      {:u32, 0xFFFFFFFF},
-      {:u64, 0},
+      {:string, "mixed"},
+      {{:fixed_string, 2}, "xy"},
       {:u64, 0xFFFFFFFFFFFFFFFF},
-      {:u128, 0},
-      {:u128, 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF},
-      {:u256, 0},
-      {:u256, 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF},
-      {:i8, -0x80},
-      {:i8, 0},
-      {:i8, 0x7F},
-      {:i16, -0x8000},
-      {:i16, 0},
-      {:i16, 0x7FFF},
-      {:i32, -0x80000000},
-      {:i32, 0},
-      {:i32, 0x7FFFFFFF},
-      {:i64, -0x800000000000000},
-      {:i64, 0},
-      {:i64, 0x7FFFFFFFFFFFFFFF},
-      {:i128, -0x800000000000000000000000000000},
-      {:i128, 0},
-      {:i128, 0x7FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF},
-      {:i256, -0x800000000000000000000000000000000000000000000000000000000000},
-      {:i256, 0},
-      {:i256, 0x7FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF},
-      {:f32, 1.2345678806304932},
-      {:f64, 1.234567898762738492830000503040030202020433},
+      {:i64, -0x8000000000000000},
+      {:f64, 1.5},
       {:date, ~D[2022-01-01]},
-      {:date, ~D[2042-01-01]},
-      {:date, ~D[1970-01-01]},
-      {:date32, ~D[1960-01-01]},
-      {:date32, ~D[2100-01-01]},
-      {:datetime, ~N[1970-01-01 00:00:00]},
       {:datetime, ~N[2022-01-01 00:00:00]},
-      {:datetime, ~N[2042-01-01 00:00:00]},
-      {{:array, :string}, []},
-      {{:array, :string},
-       [
-         "",
-         "a",
-         String.duplicate("a", 500),
-         String.duplicate("a", 15000)
-       ]},
-      {{:array, :u8}, [0, 0xFF]},
-      {{:array, :u16}, [0, 0xFFFF]},
-      {{:array, :u32}, [0, 0xFFFFFFFF]},
-      {{:array, :u64}, [0, 0xFFFFFFFFFFFFFFFF]},
-      {{:array, :i8}, [-0x80, 0, 0x7F]},
-      {{:array, :i16}, [-0x8000, 0, 0x7FFF]},
-      {{:array, :i32}, [-0x80000000, 0, 0x7FFFFFFF]},
-      {{:array, :i64}, [-0x800000000000000, 0, 0x7FFFFFFFFFFFFFFF]},
-      {{:array, :f32}, [-1.2345678806304932, 0, 1.2345678806304932]},
-      {{:array, :f64},
-       [
-         -1.234567898762738492830000503040030202020433,
-         0,
-         1.234567898762738492830000503040030202020433
-       ]},
-      {{:array, :date}, [~D[2022-01-01], ~D[2042-01-01], ~D[1970-01-01]]},
-      {{:array, :datetime},
-       [~N[1970-01-01 12:23:34], ~N[2022-01-01 22:12:59], ~N[2042-01-01 04:23:01]]},
       {{:array, {:array, :string}}, [["a"], [], ["a", "b"]]},
       {{:nullable, :string}, nil},
-      {{:nullable, :string}, "string"},
-      {:point, {10, 10}},
       {:point, {10.5, 11}},
       {{:map, :string, :string}, %{"a" => "b", "c" => "d"}}
     ]
@@ -118,10 +50,10 @@ defmodule Ch.RowBinaryTest do
       assert encoding_types([{:datetime, "UTC"}]) == [:datetime]
       assert encoding_types([{:datetime64, 6}]) == [datetime64: 1_000_000]
       assert encoding_types([{:datetime64, 3, "UTC"}]) == [datetime64: 1000]
-      assert encoding_types([{:decimal, 9, 4}]) == [decimal32: 4]
-      assert encoding_types([{:decimal, 18, 4}]) == [decimal64: 4]
-      assert encoding_types([{:decimal, 38, 4}]) == [decimal128: 4]
-      assert encoding_types([{:decimal, 76, 4}]) == [decimal256: 4]
+      assert encoding_types([{:decimal, 9, 4}]) == [{:decimal, 9, 4}]
+      assert encoding_types([{:decimal, 18, 4}]) == [{:decimal, 18, 4}]
+      assert encoding_types([{:decimal, 38, 4}]) == [{:decimal, 38, 4}]
+      assert encoding_types([{:decimal, 76, 4}]) == [{:decimal, 76, 4}]
       assert encoding_types([{:simple_aggregate_function, "any", :u8}]) == [:u8]
 
       # See https://github.com/plausible/ch/issues/353
@@ -443,6 +375,12 @@ defmodule Ch.RowBinaryTest do
       assert_raise ArgumentError, "invalid Int8: \"a\"", fn -> encode(:i8, "a") end
     end
   end
+end
+
+defmodule Ch.RowBinaryStreamingCompositeTest do
+  use ExUnit.Case, async: true
+
+  import Ch.RowBinary
 
   describe "decode_header/1" do
     test "byte-by-byte" do
@@ -731,7 +669,32 @@ defmodule Ch.RowBinaryTest do
                ]
              ]
     end
+  end
+end
 
+defmodule Ch.RowBinaryStreamingScalarTest do
+  use ExUnit.Case, async: true
+
+  import Ch.RowBinary
+  import Bitwise
+
+  defp byte_by_byte(data, types) do
+    binary = IO.iodata_to_binary(data)
+    byte_by_byte(binary, decoding_types(types), _rows = [], _buffer = "", _state = nil)
+  end
+
+  defp byte_by_byte(<<byte, rest::bytes>>, types, rows, buffer, state) do
+    {new_rows, buffer, state} = decode_rows_continue(<<buffer::bytes, byte>>, types, state)
+    byte_by_byte(rest, types, rows ++ new_rows, buffer, state)
+  end
+
+  defp byte_by_byte(<<>>, _types, rows, buffer, state) do
+    assert buffer == ""
+    assert state == nil
+    rows
+  end
+
+  describe "decode_rows_continue/3 (byte-by-byte)" do
     test "integers" do
       types = [
         "UInt8",
diff --git a/test/ch/select_test.exs b/test/ch/select_test.exs
index 2646e7c4..074103bf 100644
--- a/test/ch/select_test.exs
+++ b/test/ch/select_test.exs
@@ -29,7 +29,10 @@ defmodule Ch.SelectTest do
     end
   end
 
-  test "decodes edge case selected values", %{pool: pool} do
+  test "selects heterogeneous parameter values as one RowBinary stream", %{pool: pool} do
+    uuid_text = "417ddc5d-e556-4d27-95dd-a34d84e46a50"
+    uuid = Base.decode16!("417DDC5DE5564D2795DDA34D84E46A50")
+
     assert %{names: names, rows: [row], data: data} =
              Ch.query!(
                pool,
@@ -37,7 +40,16 @@ defmodule Ch.SelectTest do
                SELECT
                  {empty:String} AS empty_string,
                  {special:String} AS special_string,
+                 {signed:Int64} AS signed,
+                 {unsigned:UInt64} AS unsigned,
+                 {float:Float64} AS float,
+                 {decimal:Decimal(18, 4)} AS decimal,
+                 {active:Bool} AS active,
+                 {fixed:FixedString(4)} AS fixed,
                  {nil:Nullable(String)} AS nil_string,
+                 {uuid:UUID} AS uuid,
+                 {date:Date} AS date,
+                 {datetime:DateTime64(6, 'UTC')} AS datetime,
                  {ints:Array(Int16)} AS ints,
                  {map:Map(String, UInt8)} AS map,
                  {tuple:Tuple(Int8, String)} AS tuple
@@ -45,7 +57,16 @@ defmodule Ch.SelectTest do
                %{
                  "empty" => "",
                  "special" => "line\n tab\t ampersand& equals= quote'",
+                 "signed" => -9_223_372_036_854_775_808,
+                 "unsigned" => 18_446_744_073_709_551_615,
+                 "float" => 1.5,
+                 "decimal" => Decimal.new("12.3400"),
+                 "active" => true,
+                 "fixed" => "AB",
                  "nil" => nil,
+                 "uuid" => uuid_text,
+                 "date" => ~D[2024-02-29],
+                 "datetime" => ~U[2024-02-29 12:34:56.123456Z],
                  "ints" => [-2, -1, 0, 1, 2],
                  "map" => %{"a" => 1, "b" => 2},
                  "tuple" => {-8, "tuple-value"}
@@ -55,7 +76,16 @@ defmodule Ch.SelectTest do
     assert names == [
              "empty_string",
              "special_string",
+             "signed",
+             "unsigned",
+             "float",
+             "decimal",
+             "active",
+             "fixed",
              "nil_string",
+             "uuid",
+             "date",
+             "datetime",
              "ints",
              "map",
              "tuple"
@@ -64,7 +94,16 @@ defmodule Ch.SelectTest do
     assert row == [
              "",
              "line\n tab\t ampersand& equals= quote'",
+             -9_223_372_036_854_775_808,
+             18_446_744_073_709_551_615,
+             1.5,
+             Decimal.new("12.3400"),
+             true,
+             "AB" <> <<0, 0>>,
              nil,
+             uuid,
+             ~D[2024-02-29],
+             ~U[2024-02-29 12:34:56.123456Z],
              [-2, -1, 0, 1, 2],
              %{"a" => 1, "b" => 2},
              {-8, "tuple-value"}
diff --git a/test/ch/type_integration_test.exs b/test/ch/type_integration_test.exs
index 2292ee8c..34c43af2 100644
--- a/test/ch/type_integration_test.exs
+++ b/test/ch/type_integration_test.exs
@@ -8,29 +8,7 @@ defmodule Ch.TypeIntegrationTest do
     {:ok, pool: start_supervised!(Ch)}
   end
 
-  property "integer params round-trip across ClickHouse integer widths", %{pool: pool} do
-    check all {type, value} <- integer_param() do
-      assert Ch.query!(pool, "SELECT {value:#{type}}", %{"value" => value}).rows == [[value]]
-    end
-  end
-
-  property "fixed string params are padded to their declared size", %{pool: pool} do
-    check all {size, value} <- fixed_string_param() do
-      padding = :binary.copy(<<0>>, size - byte_size(value))
-
-      assert Ch.query!(pool, "SELECT {value:FixedString(#{size})}", %{"value" => value}).rows ==
-               [[value <> padding]]
-    end
-  end
-
-  property "decimal params preserve Decimal(18, 4) scale", %{pool: pool} do
-    check all value <- decimal_param() do
-      assert Ch.query!(pool, "SELECT {value:Decimal(18, 4)}", %{"value" => value}).rows ==
-               [[Decimal.round(value, 4)]]
-    end
-  end
-
-  property "uuid params accept canonical text and decode to 16 bytes", %{pool: pool} do
+  property "UUID params accept canonical text and decode to 16 bytes", %{pool: pool} do
     check all {uuid_text, uuid_bin} <- uuid_param() do
       assert Ch.query!(pool, "SELECT {value:UUID}, toString({value:UUID})", %{
                "value" => uuid_text
@@ -38,83 +16,7 @@ defmodule Ch.TypeIntegrationTest do
     end
   end
 
-  property "DateTime64 UTC params preserve microseconds", %{pool: pool} do
-    check all dt <- utc_datetime64() do
-      assert Ch.query!(pool, "SELECT {value:DateTime64(6, 'UTC')}", %{"value" => dt}).rows ==
-               [[dt]]
-    end
-  end
-
-  property "map and tuple params round-trip", %{pool: pool} do
-    check all map <- map_of(safe_string(), integer(0..255), max_length: 8),
-              tuple <- tuple_param() do
-      assert Ch.query!(pool, "SELECT {map:Map(String, UInt8)}, {tuple:Tuple(Int8, String)}", %{
-               "map" => map,
-               "tuple" => tuple
-             }).rows == [[map, tuple]]
-    end
-  end
-
-  test "fixed strings", %{pool: pool} do
-    assert Ch.query!(
-             pool,
-             "SELECT {empty:FixedString(2)}, {one:FixedString(2)}, {two:FixedString(2)}",
-             %{
-               "empty" => "",
-               "one" => "a",
-               "two" => "aa"
-             }
-           ).rows == [[<<0, 0>>, "a" <> <<0>>, "aa"]]
-
-    Help.query!("CREATE TABLE type_integration_fixed_string(a FixedString(3)) ENGINE Memory")
-    on_exit(fn -> Help.query!("DROP TABLE type_integration_fixed_string") end)
-
-    rowbinary = RowBinary.encode_rows([[""], ["a"], ["aa"], ["aaa"]], ["FixedString(3)"])
-    Ch.query!(pool, ["INSERT INTO type_integration_fixed_string FORMAT RowBinary\n" | rowbinary])
-
-    assert Ch.query!(pool, "SELECT * FROM type_integration_fixed_string").rows == [
-             [<<0, 0, 0>>],
-             ["a" <> <<0, 0>>],
-             ["aa" <> <<0>>],
-             ["aaa"]
-           ]
-  end
-
-  test "decimals", %{pool: pool} do
-    assert Ch.query!(pool, """
-           SELECT
-             toDecimal32(2, 4),
-             toDecimal64(2, 4),
-             toDecimal128(2, 4),
-             toDecimal256(2, 4)
-           """).rows == [
-             [
-               Decimal.new("2.0000"),
-               Decimal.new("2.0000"),
-               Decimal.new("2.0000"),
-               Decimal.new("2.0000")
-             ]
-           ]
-
-    Help.query!("CREATE TABLE type_integration_decimal(d Decimal32(4)) ENGINE Memory")
-    on_exit(fn -> Help.query!("DROP TABLE type_integration_decimal") end)
-
-    rowbinary =
-      RowBinary.encode_rows(
-        [[Decimal.new("2.66")], [Decimal.new("2.6666")], [Decimal.new("2.66666")]],
-        ["Decimal32(4)"]
-      )
-
-    Ch.query!(pool, ["INSERT INTO type_integration_decimal FORMAT RowBinary\n" | rowbinary])
-
-    assert Ch.query!(pool, "SELECT * FROM type_integration_decimal").rows == [
-             [Decimal.new("2.6600")],
-             [Decimal.new("2.6666")],
-             [Decimal.new("2.6667")]
-           ]
-  end
-
-  test "booleans", %{pool: pool} do
+  test "booleans preserve ClickHouse coercion semantics", %{pool: pool} do
     Help.query!("CREATE TABLE type_integration_bool(a Int64, b Bool) ENGINE Memory")
     on_exit(fn -> Help.query!("DROP TABLE type_integration_bool") end)
 
@@ -150,13 +52,10 @@ defmodule Ch.TypeIntegrationTest do
     end
   end
 
-  test "uuid", %{pool: pool} do
+  test "UUID defaults and RowBinary inserts", %{pool: pool} do
     uuid = "417ddc5d-e556-4d27-95dd-a34d84e46a50"
     uuid_bin = uuid |> String.replace("-", "") |> Base.decode16!(case: :lower)
 
-    assert Ch.query!(pool, "SELECT {uuid:UUID}, toString({uuid:UUID})", %{"uuid" => uuid}).rows ==
-             [[uuid_bin, uuid]]
-
     Help.query!("CREATE TABLE type_integration_uuid(x UUID, y String) ENGINE Memory")
     on_exit(fn -> Help.query!("DROP TABLE type_integration_uuid") end)
 
@@ -175,7 +74,7 @@ defmodule Ch.TypeIntegrationTest do
     assert byte_size(generated_uuid) == 16
   end
 
-  test "enum8", %{pool: pool} do
+  test "Enum8 accepts labels and numeric representations", %{pool: pool} do
     Help.query!(
       "CREATE TABLE type_integration_enum(i UInt8, x Enum('hello' = 1, 'world' = 2)) ENGINE Memory"
     )
@@ -207,28 +106,46 @@ defmodule Ch.TypeIntegrationTest do
              ]
   end
 
-  test "map and tuple", %{pool: pool} do
-    assert Ch.query!(pool, "SELECT {map:Map(String, UInt8)}, {tuple:Tuple(Int8, String)}", %{
-             "map" => %{"pg" => 13, "hello" => 100},
-             "tuple" => {-1, "abs"}
-           }).rows == [[%{"hello" => 100, "pg" => 13}, {-1, "abs"}]]
+  test "Enum labels with quotes and backslashes round-trip through type headers", %{pool: pool} do
+    type =
+      {:enum8,
+       [
+         {"can't", 1},
+         {"back\\slash", 2},
+         {"comma, equals= parens()", 3},
+         {"line\nbreak", 4},
+         {"null\0byte", 5}
+       ]}
 
-    Help.query!("CREATE TABLE type_integration_tuple(a Tuple(String, Int64)) ENGINE Memory")
-    on_exit(fn -> Help.query!("DROP TABLE type_integration_tuple") end)
+    encoded_type = type |> Ch.Types.encode() |> IO.iodata_to_binary()
 
-    Ch.query!(pool, "INSERT INTO type_integration_tuple VALUES (('y', 10)), (('x', -10))")
-    rowbinary = RowBinary.encode_rows([[{"a", 20}], [{"b", 30}]], ["Tuple(String, Int64)"])
-    Ch.query!(pool, ["INSERT INTO type_integration_tuple FORMAT RowBinary\n" | rowbinary])
+    Help.query!([
+      "CREATE TABLE type_integration_enum_escaping(x ",
+      encoded_type,
+      ") ENGINE Memory"
+    ])
 
-    assert Ch.query!(pool, "SELECT a FROM type_integration_tuple ORDER BY a.1").rows == [
-             [{"a", 20}],
-             [{"b", 30}],
-             [{"x", -10}],
-             [{"y", 10}]
-           ]
+    on_exit(fn -> Help.query!("DROP TABLE type_integration_enum_escaping") end)
+
+    rows = [
+      ["can't"],
+      ["back\\slash"],
+      ["comma, equals= parens()"],
+      ["line\nbreak"],
+      ["null\0byte"]
+    ]
+
+    rowbinary = RowBinary.encode_rows(rows, [type])
+    Ch.query!(pool, ["INSERT INTO type_integration_enum_escaping FORMAT RowBinary\n" | rowbinary])
+
+    assert Ch.query!(
+             pool,
+             "SELECT x FROM type_integration_enum_escaping ORDER BY CAST(x, 'Int8')"
+           ).rows ==
+             rows
   end
 
-  test "datetime and datetime64 with timezone", %{pool: pool} do
+  test "DateTime and DateTime64 preserve declared timezones", %{pool: pool} do
     Help.query!("""
     CREATE TABLE type_integration_datetime(
       timestamp DateTime('Asia/Istanbul'),
@@ -267,40 +184,6 @@ defmodule Ch.TypeIntegrationTest do
              ]
   end
 
-  defp integer_param do
-    one_of([
-      typed_integer("Int8", -128..127),
-      typed_integer("Int16", -32_768..32_767),
-      typed_integer("Int32", -2_147_483_648..2_147_483_647),
-      typed_integer("Int64", -9_007_199_254_740_992..9_007_199_254_740_991),
-      typed_integer("UInt8", 0..255),
-      typed_integer("UInt16", 0..65_535),
-      typed_integer("UInt32", 0..4_294_967_295),
-      typed_integer("UInt64", 0..9_007_199_254_740_991)
-    ])
-  end
-
-  defp typed_integer(type, range) do
-    gen all value <- integer(range) do
-      {type, value}
-    end
-  end
-
-  defp fixed_string_param do
-    gen all size <- integer(1..12),
-            value <- string(:alphanumeric, max_length: size) do
-      {size, value}
-    end
-  end
-
-  defp decimal_param do
-    gen all sign <- member_of([1, -1]),
-            coef <- integer(0..999_999_999),
-            exp <- integer(-4..4) do
-      Decimal.new(sign, coef, exp)
-    end
-  end
-
   defp uuid_param do
     gen all bytes <- binary(length: 16) do
       <<a::binary-size(4), b::binary-size(2), c::binary-size(2), d::binary-size(2),
@@ -314,37 +197,10 @@ defmodule Ch.TypeIntegrationTest do
     end
   end
 
-  defp utc_datetime64 do
-    gen all date <- date_gen(),
-            hour <- integer(0..23),
-            minute <- integer(0..59),
-            second <- integer(0..59),
-            microsecond <- integer(0..999_999) do
-      DateTime.new!(date, Time.new!(hour, minute, second, {microsecond, 6}), "Etc/UTC")
-    end
-  end
-
-  defp tuple_param do
-    gen all n <- integer(-128..127),
-            string <- safe_string() do
-      {n, string}
-    end
-  end
-
   defp bool_rows do
     gen all ids <- uniq_list_of(integer(0..255), max_length: 32),
             values <- list_of(boolean(), length: length(ids)) do
       Enum.zip_with(ids, values, fn id, value -> [id, value] end)
     end
   end
-
-  defp date_gen do
-    gen all days <- integer(0..20_000) do
-      Date.add(~D[1970-01-01], days)
-    end
-  end
-
-  defp safe_string do
-    string(:printable, max_length: 32)
-  end
 end
diff --git a/test/ch/types_property_test.exs b/test/ch/types_property_test.exs
new file mode 100644
index 00000000..e01d3f3b
--- /dev/null
+++ b/test/ch/types_property_test.exs
@@ -0,0 +1,21 @@
+defmodule Ch.TypesPropertyTest do
+  use ExUnit.Case, async: true
+  use ExUnitProperties
+
+  alias Ch.Types
+
+  property "enum type expressions round-trip generated hostile labels" do
+    check all labels <- uniq_list_of(enum_label(), min_length: 1, max_length: 8) do
+      type = {:enum16, labels |> Enum.with_index() |> Enum.map(fn {label, i} -> {label, i} end)}
+      assert type |> Types.encode() |> IO.iodata_to_binary() |> Types.decode() == type
+    end
+  end
+
+  defp enum_label do
+    codepoints = Enum.to_list(?a..?z) ++ [0, 8, 9, 10, 12, 13, ?', ?\\, ?,, ?=, ?(, ?), ?\s]
+
+    gen all chars <- list_of(member_of(codepoints), max_length: 24) do
+      List.to_string(chars)
+    end
+  end
+end
diff --git a/test/ch/types_test.exs b/test/ch/types_test.exs
index d3fae2aa..a75834a5 100644
--- a/test/ch/types_test.exs
+++ b/test/ch/types_test.exs
@@ -200,6 +200,27 @@ defmodule Ch.TypesTest do
   end
 
   describe "encode/1" do
+    test "escapes enum labels and decodes them losslessly" do
+      for type <- [
+            {:enum8,
+             [
+               {"", 0},
+               {"can't", 1},
+               {"back\\slash", 2},
+               {"comma, equals= parens()", 3},
+               {"é€𐍈", 4},
+               {"controls\0\b\t\n\f\r", 5}
+             ]},
+            {:enum16, [{"quote' and slash\\", -1}, {"plain", 2}]}
+          ] do
+        encoded = type |> encode() |> IO.iodata_to_binary()
+
+        assert encoded =~ "\\'"
+        assert encoded =~ "\\\\"
+        assert decode(encoded) == type
+      end
+    end
+
     test "rejects empty enum mappings" do
       assert_raise ArgumentError, "Enum8 requires at least one mapping", fn ->
         encode({:enum8, []})
@@ -209,5 +230,57 @@ defmodule Ch.TypesTest do
         encode({:enum16, []})
       end
     end
+
+    test "rejects invalid decimal definitions" do
+      for type <- [
+            {:decimal, 0, 0},
+            {:decimal, 77, 0},
+            {:decimal, 9, -1},
+            {:decimal, 9, 10},
+            {:decimal32, 10},
+            {:decimal64, 19},
+            {:decimal128, 39},
+            {:decimal256, 77}
+          ] do
+        assert_raise ArgumentError, ~r/invalid Decimal precision and scale/, fn ->
+          encode(type)
+        end
+      end
+
+      for type <- ["Decimal(0, 0)", "Decimal(77, 0)", "Decimal(9, -1)", "Decimal32(10)"] do
+        assert_raise ArgumentError, ~r/invalid Decimal precision and scale/, fn ->
+          decode(type)
+        end
+      end
+    end
+
+    test "rejects invalid or ambiguous enum mappings" do
+      for type <- [
+            {:enum8, [{"too-large", 128}]},
+            {:enum8, [{"too-small", -129}]},
+            {:enum16, [{"too-large", 32_768}]},
+            {:enum16, [{"too-small", -32_769}]},
+            {:enum8, [{"not-an-integer", "1"}]},
+            {:enum8, [{:not_a_string, 1}]}
+          ] do
+        assert_raise ArgumentError, ~r/invalid Enum/, fn -> encode(type) end
+      end
+
+      for type <- [
+            "Enum8('too-large' = 128)",
+            "Enum16('too-small' = -32769)",
+            "Enum8('same' = 1, 'same' = 2)",
+            "Enum16('one' = 1, 'two' = 1)"
+          ] do
+        assert_raise ArgumentError, fn -> decode(type) end
+      end
+
+      for type <- [
+            {:enum8, [{"same", 1}, {"same", 2}]},
+            {:enum16, [{"one", 1}, {"two", 1}]}
+          ] do
+        assert_raise ArgumentError, ~r/must be unique/, fn -> encode(type) end
+      end
+    end
   end
 end