Fix use of limit in multithreaded parsing (#964)

* Fix use of limit in multithreaded parsing

Fixes #963. The issue here is that although we were adjusting the # of
rows to a provided limit when multithreaded parsing, we failed to adjust
the actual column arrays to the correct size. This was an issue when we
converted from the old `CSV.Column` custom array type to returning
"normal" arrays in the 0.7 -> 0.8 transition. With `CSV.Column`, we just
passed the final row total and it adjusted the size dynamically, without
physically resizing the underlying array. With regular arrays, however,
we need to ensure the array gets resized appropriately. This became more
apparent in the recent pooling change that was released since it
actually became a silenced BoundsError because of the use of `@inbounds`
in the new `checkpooled!` routine. I've taken out those `@inbounds` uses
for now to be more conservative. The fix is fairly straightforward in
that if we adjust our final row down to a user-provided limit, then we
loop over the parsing tasks and "accumulate" rows until we hit the limit
and then resize or `empty!` columns as appropriate.

* Fix

Author: quinnj

src/file.jl

@@ -243,6 +243,27 @@ function File(ctx::Context, @nospecialize(chunking::Bool=false))
         finalrows = sum(rows)
         if ctx.limit < finalrows
             finalrows = ctx.limit
+            # adjust columns according to limit
+            acc = 0
+            for i = 1:ntasks
+                if acc + rows[i] > finalrows
+                    # need to resize this tasks columns down
+                    if finalrows - acc > 0
+                        for col in pertaskcolumns[i]
+                            if isdefined(col, :column)
+                                resize!(col.column, finalrows - acc)
+                            end
+                        end
+                    else
+                        for col in pertaskcolumns[i]
+                            if isdefined(col, :column)
+                                empty!(col.column)
+                            end
+                        end
+                    end
+                end
+                acc += rows[i]
+            end
         end
         # ok, all the parsing tasks have finished and we've promoted their types w/ the top-level columns
         # so now we just need to finish processing each column by making ChainedVectors of the individual columns
@@ -465,33 +486,33 @@ function checkpooled!(::Type{T}, pertaskcolumns, col, j, ntasks, nrows, ctx) whe
         for x in column
             if x isa PosLen
                 if x.missingvalue
-                    @inbounds refs[k] = get!(pool, missing) do
+                    refs[k] = get!(pool, missing) do
                         lastref[] += UInt32(1)
                     end
                 elseif x.escapedvalue
                     val = S === PosLenString ? S(ctx.buf, x, ctx.options.e) : Parsers.getstring(ctx.buf, x, ctx.options.e)
-                    @inbounds refs[k] = get!(pool, val) do
+                    refs[k] = get!(pool, val) do
                         lastref[] += UInt32(1)
                     end
                 else
                     val = PointerString(pointer(ctx.buf, x.pos), x.len)
                     index = Base.ht_keyindex2!(pool, val)
                     if index > 0
-                        @inbounds found_key = pool.vals[index]
+                        found_key = pool.vals[index]
                         ref = found_key::UInt32
                     else
                         new = lastref[] += UInt32(1)
                         if S === PosLenString
-                            @inbounds Base._setindex!(pool, new, S(ctx.buf, x, ctx.options.e), -index)
+                            Base._setindex!(pool, new, S(ctx.buf, x, ctx.options.e), -index)
                         else
-                            @inbounds Base._setindex!(pool, new, S(val), -index)
+                            Base._setindex!(pool, new, S(val), -index)
                         end
                         ref = new
                     end
-                    @inbounds refs[k] = ref
+                    refs[k] = ref
                 end
             else
-                @inbounds refs[k] = get!(pool, x) do
+                refs[k] = get!(pool, x) do
                     lastref[] += UInt32(1)
                 end
             end

test/basics.jl

@@ -769,4 +769,8 @@ f = CSV.File(IOBuffer(data); delim='|', normalizenames=true, stripwhitespace=fal
 f = CSV.File(IOBuffer(data); delim='|', stripwhitespace=true)
 @test f.Name[2] == "Mary Anne"
 
+# 963
+f = CSV.File(IOBuffer(join((rand(("a,$(rand())", "b,$(rand())")) for _ = 1:10^6), "\n")), header=false, limit=10000)
+@test length(f) == 10000
+
 end