Reduce allocations when decoding scalar and numeric results

benchmark/decoding.jl

@@ -0,0 +1,73 @@
+# Run in an environment containing this checkout and BenchmarkTools. An
+# optional port enables prepared-query measurements against a disposable
+# local PostgreSQL server with a trust-authenticated postgres account:
+# julia --project=/path/to/env benchmark/decoding.jl [port]
+using Postgres, BenchmarkTools
+using Postgres: DBInterface, StructUtils, Parsers, Tables
+using Postgres.API: DataDecimals
+
+const API = Postgres.API
+const STYLE = Postgres.PostgresStyle()
+const BENCH_ROW = NamedTuple{(:id, :small, :big, :ratio, :flag), Tuple{Int32, Int16, Int64, Float64, Bool}}
+
+function wire_row(values, oids; names=[Symbol("c", i) for i in eachindex(values)])
+    io = IOBuffer()
+    write(io, hton(Int16(length(values))))
+    for value in values
+        if value === nothing
+            write(io, hton(Int32(-1)))
+        else
+            write(io, hton(Int32(sizeof(value))), value)
+        end
+    end
+    return API.DataRow(take!(io), names, oids, Dict(API.DEFAULT_TYPE_REGISTRY))
+end
+
+function untyped(row)
+    data = Vector{Any}(undef, length(row.names))
+    types = Type[Any for _ in row.names]
+    StructUtils.applyeach(STYLE, Postgres.RowClosure(data, types, 1), row)
+    return data
+end
+typed(row) = StructUtils.make(BENCH_ROW, row, STYLE)
+
+function report(label, benchmark)
+    trial = run(benchmark; samples=200, seconds=2)
+    estimate = median(trial)
+    times = sort(trial.times)
+    println(label, " ns=", estimate.time, " bytes=", estimate.memory, " allocs=", estimate.allocs,
+            " min_ns=", first(times), " p90_ns=", times[cld(9length(times), 10)], " max_ns=", last(times))
+end
+
+function main()
+    println("Julia ", VERSION, " Postgres=", pathof(Postgres), " Parsers=", pkgversion(Parsers), " StructUtils=", pkgversion(StructUtils))
+    row = wire_row(["123456", "-12345", "9223372036854775806", "12345.6789012345", "t"], [23,21,20,701,16]; names=collect(fieldnames(BENCH_ROW)))
+    @assert typed(row) == (id=123456, small=-12345, big=9223372036854775806, ratio=12345.6789012345, flag=true)
+    report("untyped_scalar_row", @benchmarkable untyped($row) evals=100)
+    report("typed_scalar_row", @benchmarkable typed($row) evals=100)
+    for (label, oid, value) in (("int64",20,"9223372036854775806"), ("float64",701,"12345.6789012345"), ("bool",16,"t"), ("numeric",1700,"123456789.0123400"), ("int_array",1007,"{1,2,3,4,5,6,7,8,9}"), ("json",3802,"{\"a\":123,\"b\":[1,2,3]}"))
+        row = wire_row(fill(value, 16), fill(oid,16))
+        report("untyped_16_"*label, @benchmarkable untyped($row) evals=100)
+    end
+    length(ARGS) == 0 && return
+    conn = DBInterface.connect(Postgres.Connection,"127.0.0.1","postgres",nothing;dbname="postgres",port=parse(Int,ARGS[1]),sslmode="disable")
+    try
+        sql = "SELECT i::int4 AS id, (i % 30000)::int2 AS small, (i::int8 * 1000000000000)::int8 AS big, (i::float8 / 7) AS ratio, i % 2 = 0 AS flag FROM generate_series(1,10000) i"
+        stmt = DBInterface.prepare(conn,sql)
+        @assert length(DBInterface.execute(stmt)) == 10000
+        @assert length(DBInterface.execute(stmt,nothing,Vector{BENCH_ROW})) == 10000
+        report("live_untyped_10k", @benchmarkable DBInterface.execute($stmt) evals=1)
+        report("live_typed_10k", @benchmarkable DBInterface.execute($stmt,nothing,Vector{BENCH_ROW}) evals=1)
+        DBInterface.close!(stmt)
+        dtype = NamedTuple{(:amount,), Tuple{API.PGDecimal}}
+        stmt = DBInterface.prepare(conn, "SELECT (i::numeric / 100)::numeric(20,4) AS amount FROM generate_series(1,10000) i")
+        @assert length(DBInterface.execute(stmt)) == 10000
+        @assert length(DBInterface.execute(stmt,nothing,Vector{dtype})) == 10000
+        report("live_untyped_decimal_10k", @benchmarkable DBInterface.execute($stmt) evals=1)
+        report("live_typed_decimal_10k", @benchmarkable DBInterface.execute($stmt,nothing,Vector{$dtype}) evals=1)
+        DBInterface.close!(stmt)
+    finally
+        DBInterface.close!(conn)
+    end
+end
+main()

src/api/API.jl

@@ -1004,9 +1004,9 @@ function StructUtils.applyeach(::AbstractPostgresStyle, f, dr::DataRow)
                 f(dr.names[i], nothing)
             else
                 (len >= 0 && pos + len - 1 <= nbuf) || throw(Error("truncated DataRow message from server"))
-                str = unsafe_string(pointer(buf, pos), len)
+                bytes = @view buf[pos:pos+len-1]
                 pos += len
-                @inbounds applycast(f, dr.names[i], dr.typeIds[i], str, dr.type_registry)
+                @inbounds applycast(f, dr.names[i], dr.typeIds[i], bytes, dr.type_registry)
             end
         end
         pos == nbuf + 1 || throw(Error("DataRow message has trailing bytes"))

src/api/types.jl

@@ -448,6 +448,16 @@ end
 end
 
 function parse_numeric(val::String, on_overflow::Symbol=:warn)::NumericValue
+    @static if pkgversion(Parsers) >= v"3"
+        # PostgreSQL emits fixed-point numeric text. The native decimal parser
+        # preserves its coefficient and scale without an intermediate BigInt.
+        # Longer coefficients and exponents retain the scale, bounds and error
+        # behavior of the fallback below.
+        if sizeof(val) <= 77 && !any(c -> c == 'e' || c == 'E', val)
+            value = Parsers.tryparse(PGDecimal, val)
+            value !== nothing && DataDecimals.scale(value) <= 16383 && return value
+        end
+    end
     num = parse_numeric_parts(val)
     if num !== nothing && num.scale <= 16383 && MIN_DECIMAL_COEFFICIENT <= num.coeff <= MAX_DECIMAL_COEFFICIENT
         return PGDecimal(DataDecimals.Int256(num.coeff), num.scale)
@@ -460,6 +470,13 @@ function parse_numeric(val::String, on_overflow::Symbol=:warn)::NumericValue
 end
 
 function parse_decimal(::Type{T}, val::String) where {T<:DataDecimals.AbstractDecimal}
+    @static if pkgversion(Parsers) >= v"3"
+        if T <: Union{DataDecimals.Decimal, DataDecimals.DecimalValue} &&
+           sizeof(val) <= 77 && !any(c -> c == 'e' || c == 'E', val)
+            value = Parsers.tryparse(T, val; rounding=DataDecimals.RoundExact)
+            value !== nothing && return value
+        end
+    end
     num = parse_numeric_parts(val)
     num === nothing && throw(InexactError(:parse_decimal, T, val))
     # Preserve the requested scale for variable-scale values. Fixed-scale
@@ -853,31 +870,29 @@ end
     return decode_bytea_escape(val)
 end
 
-function parse_boolean(val::String)
-    (val == "t" || val == "1") && return true
-    (val == "f" || val == "0") && return false
+function parse_boolean(val::Union{String, AbstractVector{UInt8}})
+    bytes = val isa String ? codeunits(val) : val
+    if length(bytes) == 1
+        (bytes[1] == UInt8('t') || bytes[1] == UInt8('1')) && return true
+        (bytes[1] == UInt8('f') || bytes[1] == UInt8('0')) && return false
+    end
     throw(PostgresInterfaceError("postgres value cannot be represented as Bool; select multi-bit bit(n) values as text"))
 end
 
-function parse_value(typeId::Int, val::String, registry::Dict{Int, TypeInfo})
+# DataRow supplies contiguous, bounds-checked views. Copy only at the boundary
+# where a parser or result needs an owned String.
+const FieldBytes = SubArray{UInt8, 1, Vector{UInt8}, Tuple{UnitRange{Int}}, true}
+_field_text(val::String) = val
+_field_text(val::FieldBytes) = GC.@preserve val unsafe_string(pointer(val), length(val))
+
+function parse_value(typeId::Int, val::Union{String, FieldBytes}, registry::Dict{Int, TypeInfo})
     info = type_info(registry, typeId)
     if info.parser !== nothing
-        return info.parser(val, registry)
+        return info.parser(_field_text(val), registry)
     end
     T = info.julia_type
     if T == Bool
         return parse_boolean(val)
-    elseif T == Char
-        return pg_parse_char(val)
-    elseif T == PGTimestamp
-        return pg_parse_timestamp(val)
-    elseif T == DateTime
-        if typeId == 1184
-            return parse_timestamptz(val)
-        end
-        return pg_parse_datetime(val)
-    elseif T == UUID
-        return UUID(val)
     elseif T == Int16
         return Parsers.parse(Int16, val)
     elseif T == Int32
@@ -890,6 +905,21 @@ function parse_value(typeId::Int, val::String, registry::Dict{Int, TypeInfo})
         return Parsers.parse(Float32, val)
     elseif T == Float64
         return Parsers.parse(Float64, val)
+    end
+    # Text, lazy JSON and custom parsers own their bytes. Fixed-width scalar
+    # decoders above can read directly from the row's bounded byte view.
+    val = _field_text(val)
+    if T == Char
+        return pg_parse_char(val)
+    elseif T == PGTimestamp
+        return pg_parse_timestamp(val)
+    elseif T == DateTime
+        if typeId == 1184
+            return parse_timestamptz(val)
+        end
+        return pg_parse_datetime(val)
+    elseif T == UUID
+        return UUID(val)
     elseif T == Date
         return pg_parse_date(val)
     elseif T == Time
@@ -923,49 +953,49 @@ function parse_value(typeId::Int, val::String, registry::Dict{Int, TypeInfo})
     return val
 end
 
-@inline function applycast(f, name, typeId, val::String, registry::Dict{Int, TypeInfo})
+@inline function applycast(f, name, typeId, val, registry::Dict{Int, TypeInfo})
     f(name, parse_value(typeId, val, registry))
     return
 end
 
 # Typed fields can require a wider range than the default OID mapping. Defer
 # decoding until StructUtils supplies the declared field type. Other targets,
 # including Any and registered composites, retain the OID parser behavior.
-struct FieldValue
-    text::String
+struct FieldValue{V}
+    text::V
     oid::Int
     registry::Dict{Int, TypeInfo}
 end
 
 const WireFieldScalar = Union{Durations.Timestamp, DataDecimals.AbstractDecimal, DateTime}
 _wire_field(::Type{T}) where {T} = T <: Union{Missing, WireFieldScalar}
 _wire_field(::Type{<:AbstractVector{T}}) where {T} = T <: WireFieldScalar
-_field_source(::Type{T}, v::FieldValue) where {T} = _wire_field(T) ? v.text : parse_value(v.oid, v.text, v.registry)
+_field_source(::Type{T}, v::FieldValue) where {T} = _wire_field(T) ? _field_text(v.text) : parse_value(v.oid, v.text, v.registry)
 StructUtils.make(st::AbstractPostgresStyle, ::Type{T}, v::FieldValue) where {T} = StructUtils.make(st, T, _field_source(T, v))
 StructUtils.make(st::AbstractPostgresStyle, ::Type{T}, v::FieldValue, tags) where {T} = StructUtils.make(st, T, _field_source(T, v), tags)
 
 @static if isdefined(StructUtils, :InterpClosure) && isdefined(StructUtils, :HotStructClosure)
-    @inline function applycast(f::Union{StructUtils.InterpClosure, StructUtils.HotStructClosure}, name, typeId, val::String, registry::Dict{Int, TypeInfo})
+    @inline function applycast(f::Union{StructUtils.InterpClosure, StructUtils.HotStructClosure}, name, typeId, val, registry::Dict{Int, TypeInfo})
         f(name, FieldValue(val, typeId, registry))
         return
     end
 end
 
 @static if isdefined(StructUtils, :StructClosure)
-    @inline function applycast(f::StructUtils.StructClosure, name, typeId, val::String, registry::Dict{Int, TypeInfo})
+    @inline function applycast(f::StructUtils.StructClosure, name, typeId, val, registry::Dict{Int, TypeInfo})
         f(name, FieldValue(val, typeId, registry))
         return
     end
 end
 
 @static if isdefined(StructUtils, :FieldSink)
-    @inline function applycast(f::StructUtils.FieldSink, name, typeId, val::String, registry::Dict{Int, TypeInfo})
+    @inline function applycast(f::StructUtils.FieldSink, name, typeId, val, registry::Dict{Int, TypeInfo})
         f(name, FieldValue(val, typeId, registry))
         return
     end
 end
 
-@inline function applycast(f::StructUtils.TupleClosure, name, typeId, val::String, registry::Dict{Int, TypeInfo})
+@inline function applycast(f::StructUtils.TupleClosure, name, typeId, val, registry::Dict{Int, TypeInfo})
     f(name, FieldValue(val, typeId, registry))
     return
 end

test/decimals.jl

@@ -50,6 +50,61 @@ function test_decimals()
         @test Postgres.API.parse_array("{1.23,4.56}", DataDecimals.Decimal64{2}) == DataDecimals.Decimal64{2}.(["1.23", "4.56"])
         @test Postgres._param(DataDecimals.Decimal64{4}("1.2300")) == "1.2300"
     end
+    @testset "Native decimal parsing preserves coefficient and scale" begin
+        texts = ["0", "-0", "+0", "-0.0000", "+0.0000", "  +12.3400\t", "\u200312.3400\u2003",
+                 ".5", "1.", "-.5", "+.5", "00012.3400", "0." * repeat("0", 100),
+                 repeat("1", 76) * ".0000", "0." * repeat("0", 16382) * "1"]
+        for n in 76:79
+            push!(texts, repeat("1", n), "-" * repeat("1", n), "0." * repeat("1", n), "0." * repeat("0", n))
+        end
+        rng = MersenneTwister(0xdec0de)
+        for _ in 1:200
+            digits = String(rand(rng, '0':'9', rand(rng, 1:80)))
+            split_at = rand(rng, 0:length(digits))
+            push!(texts, string(rand(rng, ("", "+", "-")), digits[1:split_at], ".", digits[split_at+1:end]))
+        end
+        for text in texts
+            # BigInt arithmetic is an independent oracle for the bounded
+            # decimal scanner, including scale that trailing zeros preserve.
+            stripped = strip(text)
+            negative = startswith(stripped, "-")
+            unsigned = (negative || startswith(stripped, "+")) ? stripped[2:end] : stripped
+            parts = split(unsigned, '.'; limit=2)
+            scale = length(parts) == 2 ? length(parts[2]) : 0
+            coeff = parse(BigInt, join(parts)) * (negative ? -1 : 1)
+            if typemin(DataDecimals.Int256) <= coeff <= typemax(DataDecimals.Int256) && scale <= 16383
+                value = Postgres.API.parse_numeric(text, :error)
+                @test value isa WireDecimal
+                @test BigInt(DataDecimals.unscaled(value)) == coeff
+                @test DataDecimals.scale(value) == scale
+                typed = Postgres.API.parse_decimal(WireDecimal, text)
+                @test BigInt(DataDecimals.unscaled(typed)) == coeff
+                @test DataDecimals.scale(typed) == scale
+            else
+                @test_throws Postgres.PostgresInterfaceError Postgres.API.parse_numeric(text, :error)
+                @test_throws InexactError Postgres.API.parse_decimal(WireDecimal, text)
+            end
+            T = DataDecimals.Decimal64{2}
+            expected = try
+                T(coeff // big(10)^scale)
+            catch err
+                err
+            end
+            if expected isa Exception
+                @test_throws typeof(expected) Postgres.API.parse_decimal(T, text)
+            else
+                @test Postgres.API.parse_decimal(T, text) === expected
+            end
+        end
+        for text in ("", " ", "+", "-", ".", "1..2", "1,2", "1/2", "1x", "1\0", "1.2.3", "1e", "1e+")
+            @test_throws Exception Postgres.API.parse_numeric(text, :error)
+            @test_throws Exception Postgres.API.parse_decimal(WireDecimal, text)
+        end
+        for text in ("0e100001", "0e-100001", "1e100001", "1e-100001")
+            @test_throws Postgres.PostgresInterfaceError Postgres.API.parse_numeric(text, :error)
+            @test_throws Postgres.PostgresInterfaceError Postgres.API.parse_decimal(WireDecimal, text)
+        end
+    end
 end
 
 function test_decimal_roundtrips(conn)