Ensure schema-less tables have all rows respected in load! (#260)

Fixes #259. This one feels bad. For the schema-less input table case, we
iterated the first row to get the names so we could do some schema
validation, but then threw the row away, assuming that as we iterate
later, we would somehow get all the rows anyway. That probably was fine
in cases like a DataFrame, where iteration would correctly restart, but
for most schema-less tables, it's not uncommon to be a stateful iterator
that is forward-pass only (like `SQLite.Query`!). We fix this here by
ensuring that row stays intact and is inserted before iterating further
rows.

Author: quinnj

src/tables.jl

@@ -148,7 +148,8 @@ function createtable!(db::DB, name::AbstractString, ::Tables.Schema{names, types
     columns = [string(esc_id(String(names[i])), ' ',
                       sqlitetype(types !== nothing ? fieldtype(types, i) : Any))
                for i in eachindex(names)]
-    return execute(db, "CREATE $temp TABLE $ifnotexists $(esc_id(string(name))) ($(join(columns, ',')))")
+    sql = "CREATE $temp TABLE $ifnotexists $(esc_id(string(name))) ($(join(columns, ',')))"
+    return execute(db, sql)
 end
 
 # table info for load!():
@@ -213,7 +214,7 @@ function checknames(::Tables.Schema{names}, db_names::AbstractVector{String}) wh
     return true
 end
 
-function load!(sch::Tables.Schema, rows, db::DB, name::AbstractString, db_tableinfo::Union{NamedTuple, Nothing};
+function load!(sch::Tables.Schema, rows, db::DB, name::AbstractString, db_tableinfo::Union{NamedTuple, Nothing}, row=nothing, st=nothing;
                temp::Bool=false, ifnotexists::Bool=false, analyze::Bool=false)
     # check for case-insensitive duplicate column names (sqlite doesn't allow)
     checkdupnames(sch.names)
@@ -229,12 +230,20 @@ function load!(sch::Tables.Schema, rows, db::DB, name::AbstractString, db_tablei
     stmt = _Stmt(db, "INSERT INTO $(esc_id(string(name))) ($columns) VALUES ($params)")
     # start a transaction for inserting rows
     transaction(db) do
-        for row in rows
+        if row === nothing
+            state = iterate(rows)
+            state === nothing && return
+            row, st = state
+        end
+        while true
             Tables.eachcolumn(sch, row) do val, col, _
                 bind!(stmt, col, val)
             end
             sqlite3_step(stmt.handle)
             sqlite3_reset(stmt.handle)
+            state = iterate(rows, st)
+            state === nothing && break
+            row, st = state
         end
     end
     _close!(stmt)
@@ -250,5 +259,5 @@ function load!(::Nothing, rows, db::DB, name::AbstractString,
     row, st = state
     names = propertynames(row)
     sch = Tables.Schema(names, nothing)
-    return load!(sch, rows, db, name, db_tableinfo; kwargs...)
+    return load!(sch, rows, db, name, db_tableinfo, row, st; kwargs...)
 end

test/runtests.jl

@@ -519,3 +519,25 @@ end
 end
 
 end # @testset
+
+struct UnknownSchemaTable
+end
+
+Tables.isrowtable(::Type{UnknownSchemaTable}) = true
+Tables.rows(x::UnknownSchemaTable) = x
+Base.length(x::UnknownSchemaTable) = 3
+Base.iterate(::UnknownSchemaTable, st=1) = st == 4 ? nothing : ((a=1, b=2, c=3), st + 1)
+
+@testset "misc" begin
+
+# https://github.com/JuliaDatabases/SQLite.jl/issues/259
+db = SQLite.DB()
+SQLite.load!(UnknownSchemaTable(), db, "tbl")
+tbl = DBInterface.execute(db, "select * from tbl") |> columntable
+@test tbl == (
+    a = [1, 1, 1],
+    b = [2, 2, 2],
+    c = [3, 3, 3]
+)
+
+end