Fix source-boundary preprocessing and chunk endings (#1194)

* fix(source): honor bounded input preprocessing

Use logical source bounds for BOM, gzip, and reverse footer scans so SubArray and positioned IO inputs do not inspect their parent buffers. Exclude comment rows from footerskip counts as documented, including trailing comment runs.

* fix(chunks): preserve final source byte

Keep the inclusive end position for the final CSV.Chunks partition. Earlier partitions still exclude the shared boundary byte, matching the established multithreaded CSV.File behavior.

Author: quinnj

src/chunks.jl

@@ -93,7 +93,7 @@ function Base.iterate(x::Chunks, i=1)
     names = copy(x.ctx.names)
     columns = [Column(col) for col in x.ctx.columns]
     datapos = x.ctx.chunkpositions[i]
-    len = x.ctx.chunkpositions[i + 1] - 1
+    len = x.ctx.chunkpositions[i + 1] - (i != x.ctx.ntasks)
     rowsguess = cld(x.ctx.rowsguess, x.ctx.ntasks)
     threaded = false
     ntasks = 1

src/context.jl

@@ -386,7 +386,7 @@ end
         throw(ArgumentError("delimited source to parse too large; must be < $MAX_INPUT_SIZE bytes"))
     end
     # skip over initial BOM character, if present
-    pos = consumeBOM(buf, pos)
+    pos = consumeBOM(buf, pos, len)
 
     oq = something(openquotechar, quotechar) % UInt8
     eq = escapechar % UInt8
@@ -408,11 +408,23 @@ end
     end
     cmt = comment === nothing ? nothing : (pointer(comment), sizeof(comment))
 
-    if footerskip > 0 && len > 0
-        lastbyte = buf[end]
+    if footerskip > 0 && pos <= len
+        lastbyte = buf[len]
         endpos = (lastbyte == UInt8('\r') || lastbyte == UInt8('\n')) +
-            (lastbyte == UInt8('\n') && buf[end - 1] == UInt8('\r'))
-        revlen = skiptorow(ReversedBuf(buf), 1 + endpos, len, oq, eq, cq, cmt, ignoreemptyrows, 0, footerskip) - 2
+            (lastbyte == UInt8('\n') && len > pos && buf[len - 1] == UInt8('\r'))
+        revbuf = ReversedBuf(buf, pos, len)
+        revbuflen = length(revbuf)
+        rows_to_skip = footerskip
+        revlen = 0
+        while true
+            revlen = skiptorow(revbuf, 1 + endpos, revbuflen, oq, eq, cq, nothing, ignoreemptyrows, 0, rows_to_skip) - 2
+            comments = countcomments(buf, len - revlen + 1, len, oq, eq, cq, cmt)
+            adjusted_rows_to_skip = footerskip + comments
+            rows_to_skip == adjusted_rows_to_skip && break
+            rows_to_skip = adjusted_rows_to_skip > rows_to_skip ?
+                max(adjusted_rows_to_skip, 2 * rows_to_skip) :
+                adjusted_rows_to_skip
+        end
         len -= revlen
         debug && println("adjusted for footerskip, len = $(len + revlen - 1) => $len")
     end

src/detection.jl

@@ -320,6 +320,32 @@ function checkcommentandemptyline(buf, pos, len, @nospecialize(cmt), ignoreempty
     return pos
 end
 
+function countcomments(buf, pos, len, oq, eq, cq, @nospecialize(cmt))
+    cmt === nothing && return 0
+    cmtptr, cmtlen = cmt
+    comments = 0
+    while pos <= len
+        if cmtlen > 0 && (pos + cmtlen - 1) <= len && Parsers.memcmp(pointer(buf, pos), cmtptr, cmtlen)
+            comments += 1
+            pos += cmtlen
+            while pos <= len
+                @inbounds b = buf[pos]
+                pos += 1
+                if b == UInt8('\n')
+                    break
+                elseif b == UInt8('\r')
+                    pos <= len && buf[pos] == UInt8('\n') && (pos += 1)
+                    break
+                end
+            end
+        else
+            newpos = skiptorow(buf, pos, len, oq, eq, cq, nothing, false, 0, 1)
+            pos = newpos > pos ? newpos : len + 1
+        end
+    end
+    return comments
+end
+
 struct ColumnProperties
     typecode::UInt8
     maxstringsize::UInt8

src/utils.jl

@@ -243,7 +243,7 @@ function chaincolumns!(@nospecialize(a), @nospecialize(b))
 end
 
 # one-liner suggested from ScottPJones
-consumeBOM(buf, pos) = (length(buf) >= 3 && buf[pos] == 0xef && buf[pos + 1] == 0xbb && buf[pos + 2] == 0xbf) ? pos + 3 : pos
+consumeBOM(buf, pos, len) = (len - pos + 1 >= 3 && buf[pos] == 0xef && buf[pos + 1] == 0xbb && buf[pos + 2] == 0xbf) ? pos + 3 : pos
 
 if isdefined(Base, :Memory)
     if isdefined(Base,:wrap)
@@ -301,15 +301,16 @@ end
 
 function getsource(@nospecialize(x), buffer_in_memory)
     buf, pos, len, tfile = getbytebuffer(x, buffer_in_memory)::Tuple{Vector{UInt8},Int,Int,Union{Nothing,String}}
-    if length(buf) >= 2 && buf[1] == 0x1f && buf[2] == 0x8b
+    if len - pos + 1 >= 2 && buf[pos] == 0x1f && buf[pos + 1] == 0x8b
         # gzipped source, gunzip it
+        compressed = pos == 1 && len == length(buf) ? buf : @view(buf[pos:len])
         if buffer_in_memory
-            buf = transcode(GzipDecompressor, buf)
+            buf = transcode(GzipDecompressor, compressed isa Vector{UInt8} ? compressed : collect(compressed))
         else
             # 917; if we already buffered input to tempfile, make sure the compressed tempfile is
             # cleaned up since we're only passing the *uncompressed* tempfile up for removal post-parsing
             tfile1 = tfile === nothing ? nothing : tfile
-            buf, tfile = buffer_to_tempfile(GzipDecompressor(), IOBuffer(buf))
+            buf, tfile = buffer_to_tempfile(GzipDecompressor(), IOBuffer(compressed))
             if tfile1 !== nothing
                 rm(tfile1; force=true)
             end
@@ -564,12 +565,15 @@ end
 # and skips lines backwards
 struct ReversedBuf <: AbstractVector{UInt8}
     buf::Vector{UInt8}
+    first::Int
+    last::Int
 end
+ReversedBuf(buf::Vector{UInt8}) = ReversedBuf(buf, firstindex(buf), lastindex(buf))
 
-Base.size(a::ReversedBuf) = size(a.buf)
+Base.size(a::ReversedBuf) = (a.last - a.first + 1,)
 Base.IndexStyle(::Type{ReversedBuf}) = Base.IndexLinear()
-Base.getindex(a::ReversedBuf, i::Int) = a.buf[end + 1 - i]
-Base.pointer(a::ReversedBuf, pos::Integer=1) = pointer(a.buf, length(a.buf) + 1 - pos)
+Base.getindex(a::ReversedBuf, i::Int) = a.buf[a.last + 1 - i]
+Base.pointer(a::ReversedBuf, pos::Integer=1) = pointer(a.buf, a.last + 1 - pos)
 
 memset!(ptr, value, num) = ccall(:memset, Ptr{Cvoid}, (Ptr{Cvoid}, Cint, Csize_t), ptr, value, num)
 

test/basics.jl

@@ -449,6 +449,15 @@ f = CSV.File(IOBuffer("x\r\n1\r\n2\r\n3\r\n4\r\n5\r\n"), footerskip=3)
 @test length(f) == 2
 @test f[1][1] == 1
 
+# Comment rows do not count towards footerskip.
+for newline in ("\n", "\r\n", "\r")
+    csv = join(("a,b", "1,2", "3,4", "# trailing comment 1", "# trailing comment 2"), newline) * newline
+    f = CSV.File(IOBuffer(csv); comment="#", footerskip=1)
+    @test Tables.rowtable(f) == [(a=1, b=2)]
+    rows = CSV.Rows(IOBuffer(csv); comment="#", footerskip=1, types=Int)
+    @test Tables.rowtable(rows) == [(a=1, b=2)]
+end
+
 # 578
 f = CSV.File(IOBuffer("h1234567890123456\t"^2262 * "lasthdr\r\n" * "dummy dummy dummy\r\n" * ("1.23\t"^2262 * "2.46\r\n")^10), skipto=3, ntasks=1);
 @test (length(f), length(f.names)) == (10, 2263)
@@ -737,6 +746,37 @@ f = CSV.File(@view(data[:]))
 @test length(f) == 2
 @test f.column_name == ["foobar", "bazbat"]
 
+# Preprocessing must honor SubArray and IOBuffer source bounds.
+data = Vector{UInt8}("a,b\n1,2\n3,4\n")
+compressed = transcode(GzipCompressor, data)
+prefix = Vector{UInt8}("ignored prefix")
+suffix = Vector{UInt8}("ignored suffix")
+
+parent = vcat(prefix, compressed, suffix)
+firstbyte = length(prefix) + 1
+lastbyte = firstbyte + length(compressed) - 1
+for buffer_in_memory in (false, true)
+    f = CSV.File(@view(parent[firstbyte:lastbyte]); buffer_in_memory)
+    @test Tables.rowtable(f) == [(a=1, b=2), (a=3, b=4)]
+end
+
+io = IOBuffer(vcat(prefix, compressed))
+seek(io, length(prefix))
+f = CSV.File(io)
+@test Tables.rowtable(f) == [(a=1, b=2), (a=3, b=4)]
+
+parent = vcat(compressed, data)
+firstbyte = length(compressed) + 1
+f = CSV.File(@view(parent[firstbyte:end]))
+@test Tables.rowtable(f) == [(a=1, b=2), (a=3, b=4)]
+
+parent = vcat(prefix, data, suffix)
+firstbyte = length(prefix) + 1
+lastbyte = firstbyte + length(data) - 1
+f = CSV.File(@view(parent[firstbyte:lastbyte]); footerskip=1)
+@test Tables.rowtable(f) == [(a=1, b=2)]
+@test isempty(CSV.File(UInt8[0xef, 0xbb, 0xbf]; footerskip=1))
+
 # 901; nonstandard types passed via types=T
 f = CSV.File(IOBuffer("a,b,c\n1.2,3.4,5.6\n"); types=Float32)
 @test length(f) == 1

test/runtests.jl

@@ -240,6 +240,11 @@ state = iterate(chunks, st)
 f, st = state
 @test length(f) == 34914
 
+data = Vector{UInt8}("a,b\n" * join(("$(i),value$(i)" for i in 1:40), "\n") * "X")
+chunks = collect(CSV.Chunks(data; ntasks=2, rows_to_check=5, pool=false))
+@test sum(length, chunks) == 40
+@test last(last(chunks)).b == "value40X"
+
 end
 
 function strs(x::Vector, e=nothing)