function Table()

in src/table.jl [415:552]

• 131 lines of code
• 19 McCabe index (conditional complexity)

function Table(blobs::Vector{ArrowBlob}; convert::Bool=true)
    t = Table()
    sch = nothing
    dictencodings = Dict{Int64,DictEncoding}() 
    dictencoded = Dict{Int64,Meta.Field}() 
    sync = OrderedSynchronizer()
    tsks = Channel{Any}(Inf)
    tsk = @wkspawn begin
        i = 1
        for cols in tsks
            if i == 1
                foreach(x -> push!(columns(t), x), cols)
            elseif i == 2
                foreach(1:length(cols)) do i
                    columns(t)[i] = ChainedVector([columns(t)[i], cols[i]])
                end
            else
                foreach(1:length(cols)) do i
                    append!(columns(t)[i], cols[i])
                end
            end
            i += 1
        end
    end
    anyrecordbatches = false
    rbi = 1
    @sync for blob in blobs
        for batch in BatchIterator(blob)
            
            header = batch.msg.header
            if header isa Meta.Schema
                @debugv 1 "parsing schema message"
                
                
                if sch === nothing
                    for (i, field) in enumerate(header.fields)
                        push!(names(t), Symbol(field.name))
                        
                        getdictionaries!(dictencoded, field)
                        @debugv 1 "parsed column from schema: field = $field"
                    end
                    sch = header
                    schema(t)[] = sch
                elseif sch != header
                    throw(
                        ArgumentError(
                            "mismatched schemas between different arrow batches: $sch != $header",
                        ),
                    )
                end
            elseif header isa Meta.DictionaryBatch
                id = header.id
                recordbatch = header.data
                @debugv 1 "parsing dictionary batch message: id = $id, compression = $(recordbatch.compression)"
                if haskey(dictencodings, id) && header.isDelta
                    
                    field = dictencoded[id]
                    values, _, _ = build(
                        field,
                        field.type,
                        batch,
                        recordbatch,
                        dictencodings,
                        Int64(1),
                        Int64(1),
                        convert,
                    )
                    dictencoding = dictencodings[id]
                    if typeof(dictencoding.data) <: ChainedVector
                        append!(dictencoding.data, values)
                    else
                        A = ChainedVector([dictencoding.data, values])
                        S =
                            field.dictionary.indexType === nothing ? Int32 :
                            juliaeltype(field, field.dictionary.indexType, false)
                        dictencodings[id] = DictEncoding{eltype(A),S,typeof(A)}(
                            id,
                            A,
                            field.dictionary.isOrdered,
                            values.metadata,
                        )
                    end
                    continue
                end
                
                field = dictencoded[id]
                values, _, _ = build(
                    field,
                    field.type,
                    batch,
                    recordbatch,
                    dictencodings,
                    Int64(1),
                    Int64(1),
                    convert,
                )
                A = values
                S =
                    field.dictionary.indexType === nothing ? Int32 :
                    juliaeltype(field, field.dictionary.indexType, false)
                dictencodings[id] = DictEncoding{eltype(A),S,typeof(A)}(
                    id,
                    A,
                    field.dictionary.isOrdered,
                    values.metadata,
                )
                @debugv 1 "parsed dictionary batch message: id=$id, data=$values\n"
            elseif header isa Meta.RecordBatch
                anyrecordbatches = true
                @debugv 1 "parsing record batch message: compression = $(header.compression)"
                @wkspawn begin
                    cols = collect(VectorIterator(sch, $batch, dictencodings, convert))
                    put!(() -> put!(tsks, cols), sync, $(rbi))
                end
                rbi += 1
            else
                throw(ArgumentError("unsupported arrow message type: $(typeof(header))"))
            end
        end
    end
    close(tsks)
    wait(tsk)
    lu = lookup(t)
    ty = types(t)
    
    if !anyrecordbatches && !isnothing(sch)
        for field in sch.fields
            T = juliaeltype(field, buildmetadata(field), convert)
            push!(columns(t), T[])
        end
    end
    for (nm, col) in zip(names(t), columns(t))
        lu[nm] = col
        push!(ty, eltype(col))
    end
    getfield(t, :metadata)[] = buildmetadata(sch)
    return t
end