Skip to content

Commit e39e7dc

Browse files
quinnjquinnj
committed
Project Cleanup
1 parent 25531d3 commit e39e7dc

File tree

3 files changed

+37
-61
lines changed

3 files changed

+37
-61
lines changed

‎src/Sink.jl

Lines changed: 17 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -23,23 +23,20 @@ function Sink(db::DB, schema::Data.Schema; name::AbstractString="julia_"*randstr
2323
return Sink(schema, db, name, stmt, "")
2424
end
2525

26-
"constructs a new SQLite.Sink from the given `SQLite.Source`; uses `source` schema to create the SQLite table"
27-
function Sink(source::SQLite.Source; name::AbstractString="julia_"*randstring(), temp::Bool=false, ifnotexists::Bool=true)
28-
return Sink(source.db, source.schema; name=name, temp=temp, ifnotexists=ifnotexists)
29-
end
30-
"constructs a new SQLite.Sink from the given `Data.Source`; uses `source` schema to create the SQLite table"
31-
function Sink(db::DB, source; name::AbstractString="julia_"*randstring(), temp::Bool=false, ifnotexists::Bool=true)
32-
return Sink(db, Data.schema(source); name=name, temp=temp, ifnotexists=ifnotexists)
33-
end
34-
3526
# DataStreams interface
3627
Data.streamtypes{T<:SQLite.Sink}(::Type{T}) = [Data.Field]
3728

38-
function Sink{T}(source, ::Type{T}, append::Bool, db::DB, name::AbstractString="julia_" * randstring())
39-
sink = Sink(db, Data.schema(source); name=name, append=append)
29+
function Sink{T}(sch::Data.Schema, ::Type{T}, append::Bool, ref::Vector{UInt8}, db::DB, name::AbstractString="julia_" * randstring(); kwargs...)
30+
sink = Sink(db, sch; name=name, append=append, kwargs...)
31+
execute!(sink.db,"PRAGMA synchronous = OFF;")
32+
sink.transaction = string("SQLITE",randstring(10))
33+
transaction(sink.db, sink.transaction)
4034
return sink
4135
end
42-
function Sink{T}(sink, source, ::Type{T}, append::Bool)
36+
function Sink{T}(sink, sch::Data.Schema, ::Type{T}, append::Bool, ref::Vector{UInt8})
37+
execute!(sink.db,"PRAGMA synchronous = OFF;")
38+
sink.transaction = string("SQLITE",randstring(10))
39+
transaction(sink.db, sink.transaction)
4340
!append && execute!(sink.db, "delete from $(esc_id(sink.tablename))")
4441
return sink
4542
end
@@ -65,15 +62,7 @@ else
6562
getbind!{T}(val::T, col, stmt) = SQLite.bind!(stmt, col, val)
6663
end
6764

68-
function Data.open!(sink::SQLite.Sink, source)
69-
execute!(sink.db,"PRAGMA synchronous = OFF;")
70-
sink.transaction = string("SQLITE",randstring(10))
71-
transaction(sink.db, sink.transaction)
72-
return nothing
73-
end
74-
75-
function Data.streamfield!{T}(sink::SQLite.Sink, source, ::Type{T}, row, col, cols)
76-
val = Data.getfield(source, T, row ,col)
65+
function Data.stream!{T}(sink::SQLite.Sink, ::Type{Data.Field}, val::T, row, col, cols)
7766
getbind!(val, col, sink.stmt)
7867
if col == cols
7968
SQLite.sqlite3_step(sink.stmt.handle)
@@ -84,6 +73,7 @@ end
8473

8574
function Data.cleanup!(sink::SQLite.Sink)
8675
rollback(sink.db, sink.transaction)
76+
commit(sink.db, sink.transaction)
8777
execute!(sink.db, "PRAGMA synchronous = ON;")
8878
return nothing
8979
end
@@ -101,26 +91,16 @@ Load a Data.Source `source` into an SQLite table that will be named `tablename`
10191
`temp=true` will create a temporary SQLite table that will be destroyed automatically when the database is closed
10292
`ifnotexists=false` will throw an error if `tablename` already exists in `db`
10393
"""
104-
function load{T}(db, name, ::Type{T}, args...;
105-
temp::Bool=false,
106-
ifnotexists::Bool=true,
107-
append::Bool=false)
108-
source = T(args...)
109-
schema = Data.schema(source)
110-
sink = Sink(db, schema; name=name, temp=temp, ifnotexists=ifnotexists, append=append)
111-
Data.stream!(source, sink, append)
94+
function load{T}(db, name, ::Type{T}, args...; append::Bool=false, transforms::Dict=Dict{Int,Function}(), kwargs...)
95+
sink = Data.stream!(T(args...), SQLite.Sink, append, transforms, db, name; kwargs...)
11296
Data.close!(sink)
11397
return sink
11498
end
115-
function load{T}(db, name, source::T;
116-
temp::Bool=false,
117-
ifnotexists::Bool=true,
118-
append::Bool=false)
119-
sink = Sink(db, Data.schema(source); name=name, temp=temp, ifnotexists=ifnotexists, append=append)
120-
Data.stream!(source, sink, append)
99+
function load{T}(db, name, source::T; append::Bool=false, transforms::Dict=Dict{Int,Function}(), kwargs...)
100+
sink = Data.stream!(source, SQLite.Sink, append, transforms, db, name; kwargs...)
121101
Data.close!(sink)
122102
return sink
123103
end
124104

125-
load{T}(sink::Sink, ::Type{T}, args...; append::Bool=false) = (sink = Data.stream!(T(args...), sink, append); Data.close!(sink); return sink)
126-
load(sink::Sink, source; append::Bool=false) = (sink = Data.stream!(source, sink, append); Data.close!(sink); return sink)
105+
load{T}(sink::Sink, ::Type{T}, args...; append::Bool=false, transforms::Dict=Dict{Int,Function}()) = (sink = Data.stream!(T(args...), sink, append, transforms); Data.close!(sink); return sink)
106+
load(sink::Sink, source; append::Bool=false, transforms::Dict=Dict{Int,Function}()) = (sink = Data.stream!(source, sink, append, transforms); Data.close!(sink); return sink)

‎src/Source.jl

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ function Source(db::DB, sql::AbstractString, values=[]; rows::Int=-1, stricttype
2525
types[i] = stricttypes ? SQLite.juliatype(stmt.handle, i) : Any
2626
end
2727
end
28-
return SQLite.Source(Data.Schema(header, types, rows), stmt, status)
28+
return SQLite.Source(Data.Schema(Data.Field, header, types, rows), stmt, status)
2929
end
3030

3131
"""
@@ -106,21 +106,21 @@ Will bind `values` to any parameters in `sql`.
106106
`rows` is used to indicate how many rows to return in the query result if known beforehand. `rows=0` (the default) will return all possible rows.
107107
`stricttypes=false` will remove strict column typing in the result set, making each column effectively `Vector{Any}`
108108
"""
109-
function query(db::DB, sql::AbstractString, sink=DataFrame, args...; append::Bool=false, values=[], rows::Int=-1, stricttypes::Bool=true, nullable::Bool=true)
109+
function query(db::DB, sql::AbstractString, sink=DataFrame, args...; append::Bool=false, transforms::Dict=Dict{Int,Function}(), values=[], rows::Int=-1, stricttypes::Bool=true, nullable::Bool=true)
110110
source = Source(db, sql, values; rows=rows, stricttypes=stricttypes, nullable=nullable)
111-
sink = Data.stream!(source, sink, append, args...)
111+
sink = Data.stream!(source, sink, append, transforms, args...)
112112
Data.close!(sink)
113113
return sink
114114
end
115115

116-
function query{T}(db::DB, sql::AbstractString, sink::T; append::Bool=false, values=[], rows::Int=-1, stricttypes::Bool=true, nullable::Bool=true)
116+
function query{T}(db::DB, sql::AbstractString, sink::T; append::Bool=false, transforms::Dict=Dict{Int,Function}(), values=[], rows::Int=-1, stricttypes::Bool=true, nullable::Bool=true)
117117
source = Source(db, sql, values; rows=rows, stricttypes=stricttypes, nullable=nullable)
118-
sink = Data.stream!(source, sink, append)
118+
sink = Data.stream!(source, sink, append, transforms)
119119
Data.close!(sink)
120120
return sink
121121
end
122-
query(source::SQLite.Source, sink=DataFrame, args...; append::Bool=false) = (sink = Data.stream!(source, sink, append, args...); Data.close!(sink); return sink)
123-
query{T}(source::SQLite.Source, sink::T; append::Bool=false) = (sink = Data.stream!(source, sink, append); Data.close!(sink); return sink)
122+
query(source::SQLite.Source, sink=DataFrame, args...; append::Bool=false, transforms::Dict=Dict{Int,Function}()) = (sink = Data.stream!(source, sink, append, transforms, args...); Data.close!(sink); return sink)
123+
query{T}(source::SQLite.Source, sink::T; append::Bool=false, transforms::Dict=Dict{Int,Function}()) = (sink = Data.stream!(source, sink, append, transforms); Data.close!(sink); return sink)
124124

125125
"""
126126
`SQLite.tables(db, sink=DataFrame)`

‎test/runtests.jl

Lines changed: 13 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,8 @@ SQLite.DB(temp)
1414

1515
dbfile = joinpath(dirname(@__FILE__),"Chinook_Sqlite.sqlite")
1616
dbfile2 = joinpath(dirname(@__FILE__),"test.sqlite")
17-
# dbfile = joinpath("/Users/jacobquinn/.julia/v0.5/SQLite/test","Chinook_Sqlite.sqlite")
18-
# dbfile2 = joinpath("/Users/jacobquinn/.julia/v0.5/SQLite/test","test.sqlite")
17+
# dbfile = joinpath(Pkg.dir("SQLite"),"test/Chinook_Sqlite.sqlite")
18+
# dbfile2 = joinpath(Pkg.dir("SQLite"),"test/test.sqlite")
1919
cp(dbfile, dbfile2; remove_destination=true)
2020
db = SQLite.DB(dbfile2)
2121

@@ -163,8 +163,8 @@ r = SQLite.query(db,"select * from temp limit 10")
163163
@test all(Bool[get(x) == Date(2014,1,1) for x in r[:,5]])
164164
@test length(SQLite.query(db,"drop table temp").columns) == 0
165165

166-
dt = DataFrame(Data.Schema([Float64,Float64,Float64,Float64,Float64],5))
167-
sink = SQLite.Sink(db,dt)
166+
dt = DataFrame(Data.Schema(Data.Field, [Float64,Float64,Float64,Float64,Float64],5))
167+
sink = SQLite.Sink(db,Data.schema(dt, Data.Field))
168168
Data.stream!(dt,sink)
169169
Data.close!(sink)
170170
r = SQLite.query(db,"select * from $(sink.tablename)")
@@ -173,7 +173,7 @@ r = SQLite.query(db,"select * from $(sink.tablename)")
173173
SQLite.drop!(db,"$(sink.tablename)")
174174

175175
dt = DataFrame(zeros(5,5))
176-
sink = SQLite.Sink(db,dt)
176+
sink = SQLite.Sink(db,Data.schema(dt, Data.Field))
177177
Data.stream!(dt,sink)
178178
Data.close!(sink)
179179
r = SQLite.query(db,"select * from $(sink.tablename)")
@@ -183,7 +183,7 @@ r = SQLite.query(db,"select * from $(sink.tablename)")
183183
SQLite.drop!(db,"$(sink.tablename)")
184184

185185
dt = DataFrame(zeros(Int,5,5))
186-
sink = SQLite.Sink(db,dt)
186+
sink = SQLite.Sink(db,Data.schema(dt, Data.Field))
187187
Data.stream!(dt,sink)
188188
Data.close!(sink)
189189
r = SQLite.query(db,"select * from $(sink.tablename)")
@@ -202,7 +202,7 @@ SQLite.drop!(db,"$(sink.tablename)")
202202

203203
rng = Date(2013):Date(2013,1,5)
204204
dt = DataFrame([i for i = rng, j = rng])
205-
sink = SQLite.Sink(db,dt)
205+
sink = SQLite.Sink(db,Data.schema(dt, Data.Field))
206206
Data.stream!(dt,sink)
207207
Data.close!(sink)
208208
r = SQLite.query(db,"select * from $(sink.tablename)")
@@ -370,7 +370,7 @@ db = nothing; gc(); gc();
370370

371371
db = SQLite.DB()
372372
source = CSV.Source(temp)
373-
sink = SQLite.Sink(db,source; name="temp")
373+
sink = SQLite.Sink(db,Data.schema(source); name="temp")
374374
Data.stream!(source,sink)
375375
Data.close!(sink)
376376
source2 = SQLite.Source(sink)
@@ -379,7 +379,7 @@ dt = Data.stream!(source2,DataFrame)
379379
@test string(get(dt[1,2])) == "For Those About To Rock We Salute You"
380380
@test get(dt[1,3]) == 1
381381

382-
sink = SQLite.Sink(db, Data.schema(dt); name="temp2")
382+
sink = SQLite.Sink(db, Data.schema(dt, Data.Field); name="temp2")
383383
Data.stream!(dt,sink)
384384
Data.close!(sink)
385385
source3 = SQLite.Source(sink)
@@ -392,12 +392,9 @@ dt = Data.stream!(source3,DataFrame)
392392
db = SQLite.DB() #In case the order of tests is changed
393393
arr = Array(String,2)
394394
arr[1] = "1" #Now an array with the second value undefined
395-
nv = NullableArrays.NullableArray(arr, [false, true])
396-
schema = DataStreams.Data.Schema(["nv"], [String],2)
397-
d = Any[nv]
398-
dt = DataFrame(d)
395+
dt = DataFrame(Any[NullableArrays.NullableArray(arr, [false, true])])
399396
SQLite.drop!(db, "temp", ifexists=true)
400-
sink = SQLite.Sink(db, dt; name="temp")
397+
sink = SQLite.Sink(db, Data.schema(dt, Data.Field); name="temp")
401398
Data.stream!(dt, sink)
402399
Data.close!(sink)
403400
dt2 = SQLite.query(db, "Select * from temp")
@@ -411,11 +408,10 @@ ints = Int64[1,1,2,2,3]
411408
strs = String["A", "A", "B", "C", "C"]
412409
nvInts = NullableArrays.NullableArray(ints)
413410
nvStrs = NullableArrays.NullableArray(strs)
414-
schema = Data.Schema(["ints", "strs"], [Int64, String],5)
415411
d = Any[nvInts, nvStrs]
416-
dt = DataFrame(d,[:ints,:strs])
412+
dt = DataFrame(d, [:ints,:strs])
417413
SQLite.drop!(db, "temp", ifexists=true)
418-
sink = SQLite.Sink(db, dt; name="temp")
414+
sink = SQLite.Sink(db, Data.schema(dt, Data.Field); name="temp")
419415
Data.stream!(dt, sink)
420416
Data.close!(sink)
421417
SQLite.removeduplicates!(db, "temp", ["ints","strs"]) #New format

0 commit comments

Comments
 (0)