add the option to disable the part of threading in Arrow.Table that leads to catastrophic negative scaling

Author: KristofferC

src/table.jl

@@ -465,7 +465,7 @@ Table(inputs::Vector; kw...) =
     Table([ArrowBlob(tobytes(x), 1, nothing) for x in inputs]; kw...)
 
 # will detect whether we're reading a Table from a file or stream
-function Table(blobs::Vector{ArrowBlob}; convert::Bool=true)
+function Table(blobs::Vector{ArrowBlob}; convert::Bool=true, threaded::Bool=true)
     t = Table()
     sch = nothing
     dictencodingslockable = Lockable(Dict{Int64,DictEncoding}()) # dictionary id => DictEncoding
@@ -580,10 +580,17 @@ function Table(blobs::Vector{ArrowBlob}; convert::Bool=true)
             elseif header isa Meta.RecordBatch
                 anyrecordbatches = true
                 @debug "parsing record batch message: compression = $(header.compression)"
-                @wkspawn begin
+                if threaded
+                    @wkspawn begin
+                        cols = collect(
+                            VectorIterator(sch, $batch, dictencodingslockable, convert),
+                        )
+                        put!(() -> put!(tsks, cols), sync, $(rbi))
+                    end
+                else
                     cols =
-                        collect(VectorIterator(sch, $batch, dictencodingslockable, convert))
-                    put!(() -> put!(tsks, cols), sync, $(rbi))
+                        collect(VectorIterator(sch, batch, dictencodingslockable, convert))
+                    put!(tsks, cols)
                 end
                 rbi += 1
             else

test/testtables.jl

@@ -310,6 +310,15 @@ function testtable(nm, t, writekw, readkw, extratests)
         @test all(isequal.(values(t), values(tt)))
         extratests !== nothing && extratests(tt)
         seekstart(io)
+
+        # Explicitly unthreaded
+        io = Arrow.tobuffer(t; writekw...)
+        tt = Arrow.Table(io; threaded=false, readkw...)
+        @test length(tt) == length(t)
+        @test all(isequal.(values(t), values(tt)))
+        extratests !== nothing && extratests(tt)
+        seekstart(io)
+
         str = Arrow.Stream(io; readkw...)
         tt = first(str)
         @test length(tt) == length(t)