diff --git a/src/core.jl b/src/core.jl index 314e3ef7..de8b7303 100644 --- a/src/core.jl +++ b/src/core.jl @@ -2407,7 +2407,13 @@ function _normalizetypemap(typemap) end _maptype(tm, T) = tm === nothing || T === Missing ? T : get(tm, T, T) -@inline function _promotemapped(tm, current::Type, detected::Type) +# `parsed` is the type the chunk read the conflicting cell under. Another chunk +# can promote the column while this one is still parsing it: that conflict says +# nothing against the newer type, so the chunk re-parses under it. Joining the +# stale detection with a mapped current type (a `typemap` target outside the +# promotion order, such as Int32) would otherwise give String. +@inline function _promotemapped(tm, current::Type, parsed::Type, detected::Type) + parsed === current || return current joined = promote_kernel(current, detected) # A mapped result is already the selected parse type. Do not map it again # when a later result keeps the same type. @@ -4087,8 +4093,9 @@ function fusedchunk!(ci::ChunkIndex, buf::Vector{UInt8}, d::Dialect, ncols::Int, # single assignment: the lock closure captures it, and a captured # local that is reassigned is boxed (and shared across tasks) promoT = detected === T ? String : detected # a conflicting value must move the type + parsedT = T T = lock(promolock) do - promo[j] = _promotemapped(tm, promo[j], promoT) + promo[j] = _promotemapped(tm, promo[j], parsedT, promoT) end end end @@ -4400,8 +4407,9 @@ function directchunk!(ci::ChunkIndex, buf::Vector{UInt8}, d::Dialect, opts::Valu # single assignment: promoT is captured by the lock closure below, # and a captured-and-reassigned local boxes promoT = detected === T ? String : detected + parsedT = T T, dest = lock(promolock) do - joined = _promotemapped(tm, promo[j], promoT) + joined = _promotemapped(tm, promo[j], parsedT, promoT) if joined !== promo[j] promo[j] = joined finals[j] = _allocdirect(joined, ndata, buf, opts, d, j, diff --git a/test/api.jl b/test/api.jl index 8e442686..51ae63b6 100644 --- a/test/api.jl +++ b/test/api.jl @@ -1685,6 +1685,32 @@ end mappedpool = A.File(IOBuffer("a\n" * join((string(i % 3) for i in 1:300), '\n') * "\n"); typemap=IdDict(Int64 => String), pool=true) @test mappedpool.a isa PooledArrays.PooledArray{String} + # Sparse columns sampled as all-missing promote during the value wave, to + # a mapped type outside the promotion order. A chunk still parsing such a + # column as missing re-parses under the mapped type, so a parallel read + # matches the serial read and never turns the column into text. + sparse = let rng = Random.Xoshiro(42), io = IOBuffer() + println(io, join(("c$i" for i in 1:30), ','), ",t,u") + for r in 1:40_000 + for _ in 1:30 + print(io, rand(rng, 1:999), ',') + end + hit = r > 1 && r % 97 == 0 + println(io, hit ? string(rand(rng, 1:999)) : "", ',', + hit ? "2024-01-02T03:04:05" : "") + end + take!(io) + end + sparsetm = Dict(Int64 => Int32, Timestamp{Dates.Nanosecond} => DateTime) + serial = A.File(sparse; nsample=1, ntasks=1, typemap=sparsetm, chunkbytes=1 << 14) + @test eltype(serial.t) == Union{Int32, Missing} + @test eltype(serial.u) == Union{DateTime, Missing} + for _ in 1:8 + par = A.File(sparse; nsample=1, ntasks=Threads.nthreads(), typemap=sparsetm, + chunkbytes=1 << 14) + @test eltype(par.t) == eltype(serial.t) && isequal(par.t, serial.t) + @test eltype(par.u) == eltype(serial.u) && isequal(par.u, serial.u) + end # per-column dateformat input = "d1,d2\n03/04/2020,2020-01-02\n05/06/2021,2021-07-08\n"