Read each chunk's column types under one lock

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
@@ -4355,13 +4362,18 @@ function directchunk!(ci::ChunkIndex, buf::Vector{UInt8}, d::Dialect, opts::Valu
     end
     segs = Vector{Any}(undef, ncols)
     st = Vector{Type}(undef, ncols)
+    # One snapshot of every column's type and final per chunk. A column that
+    # another chunk promotes after the snapshot parses here under its old type
+    # and is re-parsed after the wave, exactly as when the promotion lands
+    # while this chunk is parsing it.
+    types, dests = lock(() -> (copy(promo), copy(finals)), promolock)
     for j in 1:ncols
         if selected !== nothing && !selected[j]
             segs[j] = nothing
             st[j] = Missing
             continue
         end
-        T, dest = lock(() -> (promo[j], finals[j]), promolock)
+        T, dest = types[j], dests[j]
         attempts = 0
         lo = rowbase + 1
         hi = rowbase + min(n, reportlimit)
@@ -4400,8 +4412,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,

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"

test/kernel.jl

@@ -1859,6 +1859,30 @@ end
     @test any(p -> p.kind == :unclosed_quote, K.problems(malformedseq))
 end
 
+@testset "promotions that land while another chunk parses" begin
+    # nsample=1 types every column from the first data row, so chunks promote
+    # columns while other chunks are parsing them. A chunk that read a column's
+    # type before such a promotion finishes under the old type and is re-parsed
+    # after the wave; parallel reads must equal the serial read.
+    rng = Random.Xoshiro(1221)
+    ncols = 40
+    cell(j) = rand(rng) < 0.003 * j ? string(rand(rng)) :
+              rand(rng) < 0.0005 ? "x" : rand(rng) < 0.02 ? "" : string(rand(rng, 1:9))
+    rows = [join((cell(j) for j in 1:ncols), ',') for _ in 1:4000]
+    names = ["c$j" for j in 1:ncols]
+    input = join(names, ',') * "\n" * join(rows, '\n') * "\n"
+    # a typemap target outside the promotion order must survive a stale chunk too
+    for tm in (nothing, Dict(Int64 => Int32))
+        ref = K.parse(input; nsample=1, chunkbytes=512, parallel=false, typemap=tm)
+        for _ in 1:4
+            t = K.parse(input; nsample=1, chunkbytes=512, parallel=true, typemap=tm)
+            @test all(nm -> eltype(t[Symbol(nm)]) == eltype(ref[Symbol(nm)]) &&
+                            isequal(collect(t[Symbol(nm)]), collect(ref[Symbol(nm)])), names)
+            @test K.problems(t) == K.problems(ref)
+        end
+    end
+end
+
 @testset "direct final slice coverage" begin
     quote_csv(s) = "\"" * replace(s, "\"" => "\"\"") * "\""
     nums = Union{Int64, Missing}[]