concurrent reading/writing in "chunks" with RSQLite (need some help troubleshooting) - R-SIG-DB

Mon, Jun 28, 2010 2:15 PM #

A bit more troubleshooting, see the modded code below. ?I'm noticing
that dbExistsTable() appears to be failing. ?If you run the code
below, you'll see that the output is:

[1] "row extraction begins: 0"
[1] "making new table"
[1] "row extraction begins: 3"
[1] "making new table"
[1] "row extraction begins: 6"
[1] "making new table"
[1] "row extraction begins: 9"
[1] "making new table"

E.g. it keeps seeing dbExistsTable() as FALSE even though, after the
first pass, it should have created that table. ?I through in some
other things I thought might work (placing the dbConnect statement for
the writing inside the while() loop, and adding a dbCommit()
statement), but none of these worked either. ?Is it possible that the
dbDisconnect() statement is not behaving properly?

Thanks a ton for any help you can give me!

--j

***

require(RSQLite)
mysqldb='test.sqlite'
fetch_n=3

# Create a small table.
somedata=data.frame(data1=seq(1:10),data2=seq(1:10)*3)

# Make a table with it.
m <- dbDriver("SQLite",flags = SQLITE_RO)
con=dbConnect(m, dbname=mysqldb)
dbWriteTable(con,"TABLEA",somedata,overwrite=TRUE)
dbDisconnect(con)

# Now we want to read from TABLEA in "chunks" and write to TABLEB.
read_con=dbConnect(m, dbname=mysqldb)
read_query=dbSendQuery(read_con,"SELECT * FROM TABLEA")

while(!dbHasCompleted(read_query))
{
? ? ? ?print(dbGetRowCount(read_query))
? ? ? ?write_con=dbConnect(m, dbname=mysqldb)
? ? ? ?read_chunk=fetch(read_query,fetch_n)
? ? ? ?new_data=data.frame(data3=read_chunk$data1*4,data4=read_chunk$data2*4)
? ? ? ?# Let's check dbexists with a different connection.
? ? ? ?tables_exists_con=dbConnect(m, dbname=mysqldb)
? ? ? ?table_exists=dbExistsTable(tables_exists_con, "TABLEB")
? ? ? ?dbDisconnect(tables_exists_con)

? ? ? ?if(!table_exists)
? ? ? ?{
? ? ? ? ? ? ? ?print("making new table")
? ? ? ? ? ? ? ?dbWriteTable(write_con,"TABLEB",new_data)
? ? ? ?} else
? ? ? ?{
? ? ? ? ? ? ? ?print("append to existing table")
? ? ? ? ? ? ? ?dbWriteTable(write_con,"TABLEB",new_data, append=TRUE)
? ? ? ?}
? ? ? ?dbCommit(write_con)
? ? ? ?dbDisconnect(write_con)
}
dbClearResult(read_query)
dbDisconnect(read_con)

On Mon, Jun 28, 2010 at 1:52 PM, Seth Falcon <seth at userprimary.net> wrote:

Hi,

A couple of ideas below... ?I will try to reproduce this later.

On Mon, Jun 28, 2010 at 1:26 PM, Jonathan Greenberg
<greenberg at ucdavis.edu> wrote:

I'm trying to cycle through a database, reading a chunk of data from
one table in the DB, processing this data chunk, and then writing the
output to another table in the database. ?I'm having to set the
fetch(n=) option because the database is too large to read the entire
thing into memory at once. ?Here's my sample code:

***

require(RSQLite)
mysqldb='test.sqlite'
fetch_n=3

# Create a data table.
somedata=data.frame(data1=seq(1:10),data2=seq(1:10)*3)

# Make a table in our database with it.
m <- dbDriver("SQLite")
con=dbConnect(m, dbname=mysqldb)
dbWriteTable(con,"TABLEA",somedata,overwrite=TRUE)
dbDisconnect(con)

# Now we want to read from TABLEA in "chunks" and write to TABLEB.
read_con=dbConnect(m, dbname=mysqldb)
write_con=dbConnect(m, dbname=mysqldb)

You might try setting the flags on the read connection to indicate
that it is read only:

?read_con = dbConnect(m, dbname = mysqldb, flags = SQLITE_RO)

There are some details in ?sqliteSupport

I wonder if it would make a difference if you created the destination
table (TABLEB) earlier (say before opening read/write cons).

read_query=dbSendQuery(read_con,"SELECT * FROM TABLEA")

while(!dbHasCompleted(read_query))
{
? ? ? ?read_chunk=fetch(read_query,fetch_n)
? ? ? ?new_data=data.frame(data3=read_chunk$data1*4,data4=read_chunk$data2*4)
? ? ? ?if(!dbExistsTable(write_con, "TABLEB"))
? ? ? ?{
? ? ? ? ? ? ? ?dbWriteTable(write_con,"TABLEB",new_data)
? ? ? ?} else
? ? ? ?{
? ? ? ? ? ? ? ?dbWriteTable(write_con,"TABLEB",new_data, append=TRUE)
? ? ? ?}
}
dbDisconnect(read_con)
dbDisconnect(write_con)

**

I am getting this error:

Warning messages:
1: In sqliteWriteTable(conn, name, value, ...) :
?RS_SQLite_exec: could not execute1: database is locked
2: In sqliteWriteTable(conn, name, value, ...) :
?RS_SQLite_exec: could not execute1: database is locked
3: In sqliteWriteTable(conn, name, value, ...) :
?RS_SQLite_exec: could not execute1: database is locked

dbDisconnect(read_con)

[1] TRUE
Warning message:
In sqliteCloseConnection(conn, ...) :
?RS-DBI driver warning: (closing pending result sets before closing
this connection)

You can get rid of this warning by calling dbClearResult on your query
result object.

dbDisconnect(write_con)

[1] TRUE

Any suggestions on how to fix this? ?Thanks!

I will try and have a closer look over the next couple of days.

+ seth

--
Seth Falcon | @sfalcon | http://userprimary.net/

Seth Falcon

Mon, Jun 28, 2010 8:50 PM #

Hi again,

I had a chance to play with this a bit.  You may find it useful to
read up on the how SQLite does locking
(http://www.sqlite.org/lockingv3.html).

I think what's happening is that read_con with the open read_query is
holding a shared lock on the db.  While that shared lock is held,
writes are not allowed.  The discussion of transaction in the SQLite
doc referenced above gave me the idea to try putting the writes into a
transaction and waiting to commit until after the read_query is
complete.  This seems to work.  Here's an example:

library("RSQLite")
mysqldb='test.sqlite'
unlink(mysqldb)
fetch_n=3

# Create a data table.
somedata=data.frame(data1=seq(1:10),data2=seq(1:10)*3)

# Make a table in our database with it.
m <- dbDriver("SQLite")
con=dbConnect(m, dbname=mysqldb)
dbWriteTable(con,"TABLEA",somedata,overwrite=TRUE)
dbGetQuery(con, "CREATE table t2 (data1 INTEGER, data2 FLOAT)")
dbDisconnect(con)

# Now we want to read from TABLEA in "chunks" and write to TABLEB.
read_con=dbConnect(m, dbname=mysqldb)
write_con=dbConnect(m, dbname=mysqldb)

read_query=dbSendQuery(read_con,"SELECT * FROM TABLEA")
dbBeginTransaction(write_con)
while (!dbHasCompleted(read_query))
{
       read_chunk=fetch(read_query,fetch_n)
       new_data=data.frame(data3=read_chunk$data1*4,data4=read_chunk$data2*4)
       dbGetPreparedQuery(write_con, "INSERT into t2 values (?, ?)", new_data)
}
dbClearResult(read_query)
dbCommit(write_con)

dbDisconnect(read_con)
dbDisconnect(write_con)



Unfortunately, dbWriteTable does not currently allow you to interact
with transactions so I'm not sure there is a way to accomplish what
you want and still use the dbWriteTable convenience function.

Hope that helps,

+ seth

Seth Falcon | @sfalcon | http://userprimary.net/