More doc updates for the new DataStreams implementation

Author: quinnj

README.md

@@ -6,7 +6,7 @@ SQLite.jl
 [![Build Status](https://travis-ci.org/JuliaDB/SQLite.jl.svg?branch=master)](https://travis-ci.org/JuliaDB/SQLite.jl)
 [![Coverage Status](https://coveralls.io/repos/JuliaDB/SQLite.jl/badge.svg?branch=master&service=github)](https://coveralls.io/github/JuliaDB/SQLite.jl?branch=master)
 
-A Julia interface to the SQLite library and support for operations on DataFrames
+A Julia interface to the SQLite library and support for the `DataStreams` data processing framework.
 
 **Installation**: `julia> Pkg.add("SQLite")`
 

REQUIRE

@@ -1,5 +1,9 @@
-julia 0.3
+julia 0.4
 BinDeps
 Compat
+DataStreams
+NullableArrays
+CSV
+Libz
 @osx Homebrew
 @windows WinRPM

src/SQLite.jl

@@ -3,20 +3,7 @@ module SQLite
 using Compat, NullableArrays, CSV, Libz, DataStreams
 import CSV.PointerString
 
-#TODO
- # create old_ui.jl file w/ deprecations
-   # make all function renames
-   # make new types SQLite.DB/Stmt, get code from jq/updates
-   # deprecate all old function names/types
- # create Source.jl, Sink.jl files
-   # pull Source code from jq/updates
-   # stream!(SQLite.Source,DataStream)
-   # stream!(SQLite.Source,CSV.Sink)
-   # stream!(DataStream,SQLite.Sink)
-   # stream!(CSV.Source,SQLite.Sink)
- # rewrite tests
- # add new tests file for Source/Sink
-
+# Deprecated exports
 export NULL, SQLiteDB, SQLiteStmt, ResultSet,
        execute, query, tables, indices, columns, droptable, dropindex,
        create, createindex, append, deleteduplicates
@@ -35,7 +22,7 @@ immutable NullType end
 const NULL = NullType()
 show(io::IO,::NullType) = print(io,"#NULL")
 
-# internal wrapper type to, in-effect, mark something which has been serialized
+"internal wrapper type to, in-effect, mark something which has been serialized"
 immutable Serialization
     object
 end
@@ -66,7 +53,9 @@ type DB
         end
     end
 end
+"`SQLite.DB(file::AbstractString)` opens or creates an SQLite database with `file`"
 DB(f::AbstractString) = DB(utf8(f))
+"`SQLite.DB()` creates an in-memory SQLite database"
 DB() = DB(":memory:")
 
 function _close(db::DB)
@@ -77,6 +66,9 @@ end
 
 Base.show(io::IO, db::SQLite.DB) = print(io, string("SQLite.DB(",db.file == ":memory:" ? "in-memory" : "\"$(db.file)\"",")"))
 
+"""
+`SQLite.Stmt(db::DB, sql::AbstractString)` creates and prepares an SQLite statement
+"""
 type Stmt
     db::DB
     handle::Ptr{Void}
@@ -110,15 +102,15 @@ include("UDF.jl")
 include("old_ui.jl")
 export @sr_str, @register, register
 
-# bind a row to nameless parameters
+"bind a row (`values`) to nameless parameters by index"
 function bind!(stmt::Stmt, values::Vector)
     nparams = sqlite3_bind_parameter_count(stmt.handle)
     @assert nparams == length(values) "you must provide values for all placeholders"
     for i in 1:nparams
         @inbounds bind!(stmt, i, values[i])
     end
 end
-# bind a row to named parameters
+"bind a row (`Dict(:key => value)`) to named parameters"
 function bind!{V}(stmt::Stmt, values::Dict{Symbol, V})
     nparams = sqlite3_bind_parameter_count(stmt.handle)
     @assert nparams == length(values) "you must provide values for all placeholders"
@@ -131,23 +123,24 @@ function bind!{V}(stmt::Stmt, values::Dict{Symbol, V})
     end
 end
 # Binding parameters to SQL statements
+"bind `val` to the named parameter `name`"
 function bind!(stmt::Stmt,name::AbstractString,val)
     i = sqlite3_bind_parameter_index(stmt.handle,name)
     if i == 0
         throw(SQLiteException("SQL parameter $name not found in $stmt"))
     end
     return bind!(stmt,i,val)
 end
-bind!(stmt::Stmt,i::Int,val::AbstractFloat)  = sqlite3_bind_double(stmt.handle,i,Float64(val))
-bind!(stmt::Stmt,i::Int,val::Int32)          = sqlite3_bind_int(stmt.handle,i,val)
-bind!(stmt::Stmt,i::Int,val::Int64)          = sqlite3_bind_int64(stmt.handle,i,val)
-bind!(stmt::Stmt,i::Int,val::NullType)       = sqlite3_bind_null(stmt.handle,i)
-bind!(stmt::Stmt,i::Int,val::AbstractString) = sqlite3_bind_text(stmt.handle,i,val)
-bind!(stmt::Stmt,i::Int,val::PointerString)  = sqlite3_bind_text(stmt.handle,i,val.ptr,val.len)
-bind!(stmt::Stmt,i::Int,val::UTF16String)    = sqlite3_bind_text16(stmt.handle,i,val)
+bind!(stmt::Stmt,i::Int,val::AbstractFloat)  = (sqlite3_bind_double(stmt.handle,i,Float64(val)); return nothing)
+bind!(stmt::Stmt,i::Int,val::Int32)          = (sqlite3_bind_int(stmt.handle,i,val); return nothing)
+bind!(stmt::Stmt,i::Int,val::Int64)          = (sqlite3_bind_int64(stmt.handle,i,val); return nothing)
+bind!(stmt::Stmt,i::Int,val::NullType)       = (sqlite3_bind_null(stmt.handle,i); return nothing)
+bind!(stmt::Stmt,i::Int,val::AbstractString) = (sqlite3_bind_text(stmt.handle,i,val); return nothing)
+bind!(stmt::Stmt,i::Int,val::PointerString)  = (sqlite3_bind_text(stmt.handle,i,val.ptr,val.len); return nothing)
+bind!(stmt::Stmt,i::Int,val::UTF16String)    = (sqlite3_bind_text16(stmt.handle,i,val); return nothing)
 # We may want to track the new ByteVec type proposed at https://github.com/JuliaLang/julia/pull/8964
 # as the "official" bytes type instead of Vector{UInt8}
-bind!(stmt::Stmt,i::Int,val::Vector{UInt8})  = sqlite3_bind_blob(stmt.handle,i,val)
+bind!(stmt::Stmt,i::Int,val::Vector{UInt8})  = (sqlite3_bind_blob(stmt.handle,i,val); return nothing)
 # Fallback is BLOB and defaults to serializing the julia value
 function sqlserialize(x)
     t = IOBuffer()
@@ -158,12 +151,13 @@ function sqlserialize(x)
     serialize(t,s)
     return takebuf_array(t)
 end
+"bind `val` to the parameter at index `i`"
 bind!(stmt::Stmt,i::Int,val) = bind!(stmt,i,sqlserialize(val))
 #TODO:
  #int sqlite3_bind_zeroblob(sqlite3_stmt*, int, int n);
  #int sqlite3_bind_value(sqlite3_stmt*, int, const sqlite3_value*);
 
-# Execute SQL statements
+"Execute a prepared SQLite statement"
 function execute!(stmt::Stmt)
     r = sqlite3_step(stmt.handle)
     if r == SQLITE_DONE
@@ -173,6 +167,7 @@ function execute!(stmt::Stmt)
     end
     return r
 end
+"Prepare and execute an SQLite statement"
 function execute!(db::DB,sql::AbstractString)
     stmt = Stmt(db,sql)
     return execute!(stmt)
@@ -192,28 +187,26 @@ function sqldeserialize(r)
 end
 
 # Transaction-based commands
-function transaction(db, mode="DEFERRED")
-    #=
-     Begin a transaction in the spedified mode, default "DEFERRED".
+"""
+Begin a transaction in the spedified `mode`, default = "DEFERRED".
 
-     If mode is one of "", "DEFERRED", "IMMEDIATE" or "EXCLUSIVE" then a
-     transaction of that (or the default) type is started. Otherwise a savepoint
-     is created whose name is mode converted to AbstractString.
-    =#
+If `mode` is one of "", "DEFERRED", "IMMEDIATE" or "EXCLUSIVE" then a
+transaction of that (or the default) type is started. Otherwise a savepoint
+is created whose name is `mode` converted to AbstractString.
+"""
+function transaction(db, mode="DEFERRED")
+    execute!(db,"PRAGMA temp_store=MEMORY;")
     if uppercase(mode) in ["", "DEFERRED", "IMMEDIATE", "EXCLUSIVE"]
         execute!(db, "BEGIN $(mode) TRANSACTION;")
     else
         execute!(db, "SAVEPOINT $(mode);")
     end
 end
-
+"Execute the function `f` within a transaction."
 function transaction(f::Function, db)
-    #=
-     Execute the function f within a transaction.
-    =#
     # generate a random name for the savepoint
     name = string("SQLITE",randstring(10))
-    execute!(db,"PRAGMA synchronous = OFF")
+    execute!(db,"PRAGMA synchronous = OFF;")
     transaction(db, name)
     try
         f()
@@ -223,18 +216,21 @@ function transaction(f::Function, db)
     finally
         # savepoints are not released on rollback
         commit(db, name)
-        execute!(db,"PRAGMA synchronous = ON")
+        execute!(db,"PRAGMA synchronous = ON;")
     end
 end
 
-# commit a transaction or savepoint (if name is given)
+"commit a transaction or named savepoint"
 commit(db) = execute!(db, "COMMIT TRANSACTION;")
+"commit a transaction or named savepoint"
 commit(db, name) = execute!(db, "RELEASE SAVEPOINT $(name);")
 
-# rollback transaction or savepoint (if name is given)
+"rollback transaction or named savepoint"
 rollback(db) = execute!(db, "ROLLBACK TRANSACTION;")
+"rollback transaction or named savepoint"
 rollback(db, name) = execute!(db, "ROLLBACK TRANSACTION TO SAVEPOINT $(name);")
 
+"drop the SQLite table `table` from the database `db`; `ifexists=true` will not return an error if `table` doesn't exist"
 function drop!(db::DB,table::AbstractString;ifexists::Bool=false)
     exists = ifexists ? "if exists" : ""
     transaction(db) do
@@ -243,15 +239,19 @@ function drop!(db::DB,table::AbstractString;ifexists::Bool=false)
     execute!(db,"vacuum")
     return
 end
-
+"drop the SQLite index `index` from the database `db`; `ifexists=true` will not return an error if `index` doesn't exist"
 function dropindex!(db::DB,index::AbstractString;ifexists::Bool=false)
     exists = ifexists ? "if exists" : ""
     transaction(db) do
         execute!(db,"drop index $exists $index")
     end
     return
 end
-
+"""
+create the SQLite index `index` on the table `table` using `cols`, which may be a single column or comma-delimited list of columns.
+`unique` specifies whether the index will be unique or not.
+`ifnotexists=true` will not throw an error if the index already exists
+"""
 function createindex!(db::DB,table::AbstractString,index::AbstractString,cols
                     ;unique::Bool=true,ifnotexists::Bool=false)
     u = unique ? "unique" : ""
@@ -262,8 +262,8 @@ function createindex!(db::DB,table::AbstractString,index::AbstractString,cols
     execute!(db,"analyze $index")
     return
 end
-
-function deleteduplicates!(db,table::AbstractString,cols::AbstractString)
+"removes duplicate rows from `table` based on the values in `cols` which may be a single column or comma-delimited list of columns"
+function removeduplicates!(db,table::AbstractString,cols::AbstractString)
     transaction(db) do
         execute!(db,"delete from $table where rowid not in (select max(rowid) from $table group by $cols);")
     end
@@ -274,4 +274,4 @@ end
 include("Source.jl")
 include("Sink.jl")
 
-end #SQLite module
+end # module

src/Sink.jl

@@ -2,21 +2,26 @@ sqlitetype{T<:Integer}(::Type{T}) = "INT"
 sqlitetype{T<:AbstractFloat}(::Type{T}) = "REAL"
 sqlitetype{T<:AbstractString}(::Type{T}) = "TEXT"
 sqlitetype(x) = "BLOB"
-
+"SQLite.Sink implements the `Sink` interface in the `DataStreams` framework"
 type Sink <: Data.Sink # <: IO
     schema::Data.Schema
     db::DB
     tablename::UTF8String
     stmt::Stmt
 end
-
-function Source(sink::SQLite.Sink)
-    stmt = SQLite.Stmt(sink.db,"select * from $(sink.tablename)")
+"constructs an SQLite.Source from an SQLite.Sink; selects all rows/columns from the underlying Sink table by default"
+function Source(sink::SQLite.Sink,sql::AbstractString="select * from $(sink.tablename)")
+    stmt = SQLite.Stmt(sink.db,sql)
     status = SQLite.execute!(stmt)
     return SQLite.Source(sink.schema, stmt, status)
 end
 
-# independent Sink constructor for new or existing SQLite tables
+"""
+independent SQLite.Sink constructor to create a new or wrap an existing SQLite table with name `tablename`.
+can optionally provide a `Data.Schema` through the `schema` argument.
+`temp=true` will create a temporary SQLite table that will be destroyed automatically when the database is closed
+`ifnotexists=false` will throw an error if `tablename` already exists in `db`
+"""
 function Sink(db::DB,tablename::AbstractString="julia_"*randstring(),schema::Data.Schema=Data.EMPTYSCHEMA;temp::Bool=false,ifnotexists::Bool=true)
     rows, cols = size(schema)
     temp = temp ? "TEMP" : ""
@@ -27,59 +32,71 @@ function Sink(db::DB,tablename::AbstractString="julia_"*randstring(),schema::Dat
     stmt = SQLite.Stmt(db,"insert into $tablename values ($params)")
     return Sink(schema,db,utf8(tablename),stmt)
 end
+"constructs a new SQLite.Sink from the given `Data.Source`; uses `source` schema to create the SQLite table"
+function Sink(source::Data.Source, db::DB, tablename::AbstractString="julia_"*randstring();temp::Bool=false,ifnotexists::Bool=true)
+    sink = Sink(db, tablename, source.schema; temp=temp, ifnotexists=ifnotexists)
+    return Data.stream!(source,sink)
+end
 
 # create a new SQLite table
 # Data.Table
 function getbind!{T}(dt::NullableVector{T},row,col,stmt)
-    @inbounds SQLite.bind!(stmt,col,ifelse(dt.isnull[row], NULL, dt.values[row]::T))
+    @inbounds val, isnull = dt.values[row]::T, dt.isnull[row]
+    if isnull
+        SQLite.bind!(stmt,col,NULL)
+    else
+        SQLite.bind!(stmt,col,val)
+    end
     return
 end
-
+"stream the data in `dt` into the SQLite table represented by `sink`"
 function Data.stream!(dt::Data.Table,sink::SQLite.Sink)
     rows, cols = size(dt)
     types = Data.types(dt)
+    handle = sink.stmt.handle
     transaction(sink.db) do
         if rows*cols != 0
             for row = 1:rows
                 for col = 1:cols
                     @inbounds SQLite.getbind!(Data.column(dt,col,types[col]),row,col,sink.stmt)
                 end
-                SQLite.execute!(sink.stmt)
+                SQLite.sqlite3_step(handle)
+                SQLite.sqlite3_reset(handle)
             end
         end
     end
     SQLite.execute!(sink.db,"analyze $(sink.tablename)")
     return sink
 end
-function Sink(dt::Data.Table,db::DB,tablename::AbstractString="julia_"*randstring();temp::Bool=false,ifnotexists::Bool=false)
-    sink = Sink(db,tablename,dt.schema;temp=temp,ifnotexists=ifnotexists)
-    return Data.stream!(dt,sink)
-end
 # CSV.Source
 function getbind!{T}(io,::Type{T},opts,row,col,stmt)
     val, isnull = CSV.getfield(io,T,opts,row,col)
-    SQLite.bind!(stmt,col,ifelse(isnull,NULL,val))
+    if isnull
+        SQLite.bind!(stmt,col,NULL)
+    else
+        SQLite.bind!(stmt,col,val)
+    end
     return
 end
+"stream the data in `source` CSV file to the SQLite table represented by `sink`"
 function Data.stream!(source::CSV.Source,sink::SQLite.Sink)
     rows, cols = size(source)
     types = Data.types(source)
     io = source.data
     opts = source.options
+    stmt = sink.stmt
+    handle = stmt.handle
     transaction(sink.db) do
         if rows*cols != 0
             for row = 1:rows
                 for col = 1:cols
-                    @inbounds SQLite.getbind!(io, types[col], opts, row, col, sink.stmt)
+                    @inbounds SQLite.getbind!(io, types[col], opts, row, col, stmt)
                 end
-                SQLite.execute!(sink.stmt)
+                SQLite.sqlite3_step(handle)
+                SQLite.sqlite3_reset(handle)
             end
         end
     end
     SQLite.execute!(sink.db,"analyze $(sink.tablename)")
     return sink
 end
-function Sink(csv::CSV.Source,db::DB,tablename::AbstractString="julia_"*randstring();temp::Bool=false,ifnotexists::Bool=false)
-    sink = Sink(db,tablename,csv.schema;temp=temp,ifnotexists=ifnotexists)
-    return Data.stream!(csv,sink)
-end