Implement SQLite incremental BLOB/TEXT I/O

16 files changed, 837 insertions, 76 deletions

diff --git a/odb/sqlite/blob-stream.hxx b/odb/sqlite/blob-stream.hxx
new file mode 100644
index 0000000..58be6ba
--- /dev/null
+++ b/odb/sqlite/blob-stream.hxx
@@ -0,0 +1,33 @@
+// file      : odb/sqlite/blob-stream.hxx
+// copyright : Copyright (c) 2005-2015 Code Synthesis Tools CC
+// license   : GNU GPL v2; see accompanying LICENSE file
+
+#ifndef ODB_SQLITE_BLOB_STREAM_HXX
+#define ODB_SQLITE_BLOB_STREAM_HXX
+
+#include <odb/pre.hxx>
+
+#include <odb/sqlite/blob.hxx>
+#include <odb/sqlite/stream.hxx>
+#include <odb/sqlite/details/export.hxx>
+
+namespace odb
+{
+  namespace sqlite
+  {
+    class LIBODB_SQLITE_EXPORT blob_stream: public stream
+    {
+    public:
+      blob_stream (const blob& b, bool rw)
+          : stream (b.db ().c_str (),
+                    b.table ().c_str (),
+                    b.column ().c_str (),
+                    b.rowid (),
+                    rw) {}
+    };
+  }
+}
+
+#include <odb/post.hxx>
+
+#endif // ODB_SQLITE_BLOB_STREAM_HXX
diff --git a/odb/sqlite/blob.hxx b/odb/sqlite/blob.hxx
new file mode 100644
index 0000000..eccb3ab
--- /dev/null
+++ b/odb/sqlite/blob.hxx
@@ -0,0 +1,70 @@
+// file      : odb/sqlite/blob.hxx
+// copyright : Copyright (c) 2005-2015 Code Synthesis Tools CC
+// license   : GNU GPL v2; see accompanying LICENSE file
+
+#ifndef ODB_SQLITE_BLOB_HXX
+#define ODB_SQLITE_BLOB_HXX
+
+#include <odb/pre.hxx>
+
+#include <string>
+#include <cstddef> // std::size_t
+
+// Carefully allow this header to be included into the ODB compilation.
+//
+#ifndef ODB_COMPILER
+#  include <odb/sqlite/forward.hxx>
+#  include <odb/sqlite/details/export.hxx>
+#endif
+
+namespace odb
+{
+  namespace sqlite
+  {
+#ifdef ODB_COMPILER
+    #pragma db sqlite:type("BLOB STREAM")
+    class blob
+#else
+    class LIBODB_SQLITE_EXPORT blob
+#endif
+    {
+    public:
+      // BLOB size to provision for. Set before calling persist() or update().
+      //
+      explicit
+      blob (std::size_t size = 0): size_ (size) {}
+
+      std::size_t size () const {return size_;}
+      void size (std::size_t s) {size_ = s;}
+
+      const std::string& db () const {return db_;}
+      const std::string& table () const {return table_;}
+      const std::string& column () const {return column_;}
+      long long rowid () const {return rowid_;}
+
+      void
+      clear ()
+      {
+        size_ = 0;
+        db_.clear ();
+        table_.clear ();
+        column_.clear ();
+        rowid_ = 0;
+      }
+
+    private:
+#ifndef ODB_COMPILER
+      friend struct default_value_traits<blob, id_blob_stream>;
+#endif
+      std::size_t size_;
+      mutable std::string db_;
+      mutable std::string table_;
+      mutable std::string column_;
+      mutable long long rowid_;
+    };
+  }
+}
+
+#include <odb/post.hxx>
+
+#endif // ODB_SQLITE_BLOB_HXX
diff --git a/odb/sqlite/connection.cxx b/odb/sqlite/connection.cxx
index dd30f1f..c44d648 100644
--- a/odb/sqlite/connection.cxx
+++ b/odb/sqlite/connection.cxx
@@ -33,7 +33,7 @@ namespace odb
         : odb::connection (db),
           db_ (db),
           unlock_cond_ (unlock_mutex_),
-          statements_ (0)
+          active_objects_ (0)
     {
       int f (db.flags () | extra_flags);
       const string& n (db.name ());
@@ -86,7 +86,7 @@ namespace odb
           db_ (db),
           handle_ (handle),
           unlock_cond_ (unlock_mutex_),
-          statements_ (0)
+          active_objects_ (0)
     {
       init ();
     }
@@ -184,11 +184,11 @@ namespace odb
     void connection::
     clear ()
     {
-      // The current first statement will remove itself from the list
-      // and make the second statement (if any) the new first.
+      // The current first active_object will remove itself from the list
+      // and make the second object (if any) the new first.
       //
-      while (statements_ != 0)
-        statements_->reset ();
+      while (active_objects_ != 0)
+        active_objects_->clear ();
     }
   }
 }
diff --git a/odb/sqlite/connection.hxx b/odb/sqlite/connection.hxx
index 8e54388..f8378fb 100644
--- a/odb/sqlite/connection.hxx
+++ b/odb/sqlite/connection.hxx
@@ -34,6 +34,39 @@ namespace odb
     class connection;
     typedef details::shared_ptr<connection> connection_ptr;
 
+    // SQLite "active object", i.e., an object that needs to be
+    // "cleared" before the transaction can be committed and the
+    // connection release. These form a doubly-linked list.
+    //
+    class active_object
+    {
+    public:
+      // This function should remove the object from the list, since
+      // it shall no longer be "active".
+      //
+      virtual void
+      clear () = 0;
+
+    protected:
+      active_object (connection& c): prev_ (0), next_ (this), conn_ (c) {}
+
+      void
+      list_add ();
+
+      void
+      list_remove ();
+
+    protected:
+      // prev_ == 0 means we are the first element.
+      // next_ == 0 means we are the last element.
+      // next_ == this means we are not on the list (prev_ should be 0).
+      //
+      active_object* prev_;
+      active_object* next_;
+
+      connection& conn_;
+    };
+
     class LIBODB_SQLITE_EXPORT connection: public odb::connection
     {
     public:
@@ -165,12 +198,12 @@ namespace odb
     private:
       friend class transaction_impl; // invalidate_results()
 
-      // Linked list of active statements currently associated
+      // Linked list of active objects currently associated
       // with this connection.
       //
     private:
-      friend class statement;
-      statement* statements_;
+      friend class active_object;
+      active_object* active_objects_;
     };
   }
 }
diff --git a/odb/sqlite/connection.ixx b/odb/sqlite/connection.ixx
index e0ed736..23df27c 100644
--- a/odb/sqlite/connection.ixx
+++ b/odb/sqlite/connection.ixx
@@ -6,6 +6,32 @@ namespace odb
 {
   namespace sqlite
   {
+    // active_objects
+    //
+    inline void active_object::
+    list_add ()
+    {
+      next_ = conn_.active_objects_;
+      conn_.active_objects_ = this;
+
+      if (next_ != 0)
+        next_->prev_ = this;
+    }
+
+    inline void active_object::
+    list_remove ()
+    {
+      (prev_ == 0 ? conn_.active_objects_ : prev_->next_) = next_;
+
+      if (next_ != 0)
+        next_->prev_ = prev_;
+
+      prev_ = 0;
+      next_ = this;
+    }
+
+    // connection
+    //
     template <typename T>
     inline prepared_query<T> connection::
     prepare_query (const char* n, const char* q)
diff --git a/odb/sqlite/forward.hxx b/odb/sqlite/forward.hxx
index 4d9a063..1bc5da5 100644
--- a/odb/sqlite/forward.hxx
+++ b/odb/sqlite/forward.hxx
@@ -39,6 +39,19 @@ namespace odb
 
     // Implementation details.
     //
+    enum database_type_id
+    {
+      id_integer,
+      id_real,
+      id_text,
+      id_blob,
+      id_text_stream,
+      id_blob_stream
+    };
+
+    template <typename T, database_type_id>
+    struct default_value_traits;
+
     enum statement_kind
     {
       statement_select,
diff --git a/odb/sqlite/makefile b/odb/sqlite/makefile
index cd80e53..e2ea0e9 100644
--- a/odb/sqlite/makefile
+++ b/odb/sqlite/makefile
@@ -18,6 +18,7 @@ simple-object-statements.cxx \
 statement.cxx                \
 statements-base.cxx          \
 statement-cache.cxx          \
+stream.cxx                   \
 tracer.cxx                   \
 traits.cxx                   \
 transaction.cxx              \
diff --git a/odb/sqlite/query.hxx b/odb/sqlite/query.hxx
index 6d7dd0d..bb1f50e 100644
--- a/odb/sqlite/query.hxx
+++ b/odb/sqlite/query.hxx
@@ -1565,6 +1565,24 @@ namespace odb
       details::buffer buffer_;
       std::size_t size_;
     };
+
+    // TEXT STREAM (reduce to id_text).
+    //
+    template <typename T>
+    struct query_param_impl<T, id_text_stream>: query_param_impl<T, id_text>
+    {
+      query_param_impl (ref_bind<T> r) : query_param_impl<T, id_text> (r) {}
+      query_param_impl (val_bind<T> v) : query_param_impl<T, id_text> (v) {}
+    };
+
+    // BLOB STREAM (reduce to id_blob).
+    //
+    template <typename T>
+    struct query_param_impl<T, id_blob_stream>: query_param_impl<T, id_blob>
+    {
+      query_param_impl (ref_bind<T> r) : query_param_impl<T, id_blob> (r) {}
+      query_param_impl (val_bind<T> v) : query_param_impl<T, id_blob> (v) {}
+    };
   }
 }
 
diff --git a/odb/sqlite/sqlite-types.hxx b/odb/sqlite/sqlite-types.hxx
index 38376d9..80ffbd2 100644
--- a/odb/sqlite/sqlite-types.hxx
+++ b/odb/sqlite/sqlite-types.hxx
@@ -7,6 +7,7 @@
 
 #include <odb/pre.hxx>
 
+#include <string>
 #include <cstddef>  // std::size_t
 
 namespace odb
@@ -24,7 +25,9 @@ namespace odb
         real,    // Buffer is double; size, capacity, truncated are unused.
         text,    // Buffer is a UTF-8 char array.
         text16,  // Buffer is a UTF-16 2-byte char array (sizes in bytes).
-        blob     // Buffer is a char array.
+        blob,    // Buffer is a char array.
+        stream   // Buffer is stream_buffers. Size specifies the BLOB size
+                 // (input only). Capacity and truncated unused.
       };
 
       buffer_type type;
@@ -34,6 +37,19 @@ namespace odb
       bool* is_null;
       bool* truncated;
     };
+
+    // The "out" values should be set in set_image() to point to
+    // variables that will be receiving the data. The "in" values
+    // are used in set_value() and contain the data that needs to
+    // be copied over.
+    //
+    struct stream_buffers
+    {
+      union {std::string* out; const char* in;} db;
+      union {std::string* out; const char* in;} table;
+      union {std::string* out; const char* in;} column;
+      union {long long* out; long long in;} rowid;
+    };
   }
 }
 
diff --git a/odb/sqlite/statement.cxx b/odb/sqlite/statement.cxx
index 3bd5e4c..f7ce632 100644
--- a/odb/sqlite/statement.cxx
+++ b/odb/sqlite/statement.cxx
@@ -42,6 +42,12 @@ namespace odb
     }
 
     void statement::
+    clear ()
+    {
+      reset ();
+    }
+
+    void statement::
     init (const char* text,
           std::size_t text_size,
           statement_kind sk,
@@ -49,8 +55,6 @@ namespace odb
           bool optimize)
     {
       active_ = false;
-      prev_ = 0;
-      next_ = this;
 
       string tmp;
       if (proc != 0)
@@ -58,23 +62,25 @@ namespace odb
         switch (sk)
         {
         case statement_select:
-          process_select (text,
+          process_select (tmp,
+                          text,
                           &proc->bind->buffer, proc->count, sizeof (bind),
                           '"', '"',
-                          optimize,
-                          tmp);
+                          optimize);
           break;
         case statement_insert:
-          process_insert (text,
+          process_insert (tmp,
+                          text,
                           &proc->bind->buffer, proc->count, sizeof (bind),
                           '?',
-                          tmp);
+                          '$');
           break;
         case statement_update:
-          process_update (text,
+          process_update (tmp,
+                          text,
                           &proc->bind->buffer, proc->count, sizeof (bind),
                           '?',
-                          tmp);
+                          '$');
           break;
         case statement_delete:
         case statement_generic:
@@ -102,9 +108,10 @@ namespace odb
         {
           // Temporarily store the statement text in prev_ so that
           // text() which may be called by the tracer can access it.
+          // Dirty but efficient.
           //
 #if SQLITE_VERSION_NUMBER >= 3005003
-          prev_ = reinterpret_cast<statement*> (const_cast<char*> (text));
+          prev_ = reinterpret_cast<active_object*> (const_cast<char*> (text));
 #endif
           t->prepare (conn_, *this);
 #if SQLITE_VERSION_NUMBER >= 3005003
@@ -159,10 +166,11 @@ namespace odb
 #endif
     }
 
-    void statement::
+    bool statement::
     bind_param (const bind* p, size_t n)
     {
       int e (SQLITE_OK);
+      bool r (false);
 
       // SQLite parameters are counted from 1.
       //
@@ -231,11 +239,25 @@ namespace odb
                                    SQLITE_STATIC);
             break;
           }
+        case bind::stream:
+          {
+#if SQLITE_VERSION_NUMBER >= 3004000
+            e = sqlite3_bind_zeroblob (stmt_,
+                                       c,
+                                       static_cast<int> (*b.size));
+            r = true;
+#else
+            assert (false);
+#endif
+            break;
+          }
         }
       }
 
       if (e != SQLITE_OK)
         translate_error (e, conn_);
+
+      return r;
     }
 
     bool statement::
@@ -254,6 +276,9 @@ namespace odb
 
         int c (col++);
 
+        if (b.type == bind::stream)
+          col++; // Skip ROWID value that follows.
+
         if (truncated && (b.truncated == 0 || !*b.truncated))
           continue;
 
@@ -320,6 +345,27 @@ namespace odb
             memcpy (b.buffer, d, *b.size);
             break;
           }
+        case bind::stream:
+          {
+            stream_buffers& sb (*static_cast<stream_buffers*> (b.buffer));
+
+            // SQLite documentation states that these are valid until the
+            // statement is finalized (or reprepared). For our case, we
+            // only need it to stay alive until we call set_value() which
+            // we do while executing the statement (i.e., we don't copy
+            // images for later processing).
+            //
+            sb.db.in = sqlite3_column_database_name (stmt_, c);
+            sb.table.in = sqlite3_column_table_name (stmt_, c);
+            sb.column.in = sqlite3_column_origin_name (stmt_, c);
+
+            // The ROWID comes in the following column.
+            //
+            sb.rowid.in = static_cast<long long> (
+              sqlite3_column_int64 (stmt_, c + 1));
+
+            break;
+          }
         }
       }
 
@@ -333,6 +379,61 @@ namespace odb
       return r;
     }
 
+    void statement::
+    stream_param (const bind* p, size_t n, const stream_data& d)
+    {
+      // Code similar to bind_param().
+      //
+      for (size_t i (0), j (1); i < n; ++i)
+      {
+        const bind& b (p[i]);
+
+        if (b.buffer == 0) // Skip NULL entries.
+          continue;
+
+        int c (static_cast<int> (j++));
+
+        if ((b.is_null != 0 && *b.is_null) || b.type != bind::stream)
+          continue;
+
+        // Get column name.
+        //
+        const char* col (sqlite3_bind_parameter_name (stmt_, c));
+        assert (col != 0); // Statement doesn't contain column name.
+
+        stream_buffers& sb (*static_cast<stream_buffers*> (b.buffer));
+
+        *sb.db.out = d.db;
+        *sb.table.out = d.table;
+        *sb.column.out = col + 1; // Skip '$'.
+        *sb.rowid.out = d.rowid;
+      }
+    }
+
+    inline void
+    update_hook (void* v, const char* db, const char* table, long long rowid)
+    {
+      statement::stream_data& d (*static_cast<statement::stream_data*> (v));
+      d.db = db;
+      d.table = table;
+      d.rowid = rowid;
+    }
+
+    extern "C" void
+    odb_sqlite_update_hook (void* v,
+                            int,
+                            const char* db,
+                            const char* table,
+#if SQLITE_VERSION_NUMBER >= 3005000
+                                    sqlite3_int64 rowid
+#else
+                                    sqlite_int64 rowid
+#endif
+    )
+    {
+      update_hook (v, db, table, static_cast<long long> (rowid));
+    }
+
     // generic_statement
     //
 
@@ -357,7 +458,7 @@ namespace odb
     generic_statement::
     generic_statement (connection_type& conn,
                        const char* text,
-                       std::size_t text_size)
+                       size_t text_size)
         : statement (conn,
                      text, text_size, statement_generic,
                      0, false),
@@ -640,12 +741,16 @@ namespace odb
           t->execute (conn_, *this);
       }
 
-      bind_param (param_.bind, param_.count);
+      sqlite3* h (conn_.handle ());
+      bool stream (bind_param (param_.bind, param_.count));
+
+      stream_data sd;
+      if (stream)
+        sqlite3_update_hook (h, &odb_sqlite_update_hook, &sd);
 
       int e;
 
 #ifdef LIBODB_SQLITE_HAVE_UNLOCK_NOTIFY
-      sqlite3* h (conn_.handle ());
       while ((e = sqlite3_step (stmt_)) == SQLITE_LOCKED)
       {
         if (sqlite3_extended_errcode (h) != SQLITE_LOCKED_SHAREDCACHE)
@@ -658,6 +763,9 @@ namespace odb
       e = sqlite3_step (stmt_);
 #endif
 
+      if (stream)
+        sqlite3_update_hook (h, 0, 0); // Clear the hook.
+
       // sqlite3_step() will return a detailed error code only if we used
       // sqlite3_prepare_v2(). Otherwise, sqlite3_reset() returns the
       // error.
@@ -684,6 +792,11 @@ namespace odb
           translate_error (e, conn_);
       }
 
+      // Stream parameters, if any.
+      //
+      if (stream)
+        stream_param (param_.bind, param_.count, sd);
+
       if (returning_ != 0)
       {
         bind& b (returning_->bind[0]);
@@ -691,7 +804,7 @@ namespace odb
         *b.is_null = false;
         *static_cast<long long*> (b.buffer) =
           static_cast<long long> (
-            sqlite3_last_insert_rowid (conn_.handle ()));
+            sqlite3_last_insert_rowid (h));
       }
 
       return true;
@@ -735,10 +848,14 @@ namespace odb
           t->execute (conn_, *this);
       }
 
-      bind_param (param_.bind, param_.count);
+      sqlite3* h (conn_.handle ());
+      bool stream (bind_param (param_.bind, param_.count));
+
+      stream_data sd;
+      if (stream)
+        sqlite3_update_hook (h, &odb_sqlite_update_hook, &sd);
 
       int e;
-      sqlite3* h (conn_.handle ());
 
 #ifdef LIBODB_SQLITE_HAVE_UNLOCK_NOTIFY
       while ((e = sqlite3_step (stmt_)) == SQLITE_LOCKED)
@@ -753,6 +870,9 @@ namespace odb
       e = sqlite3_step (stmt_);
 #endif
 
+      if (stream)
+        sqlite3_update_hook (h, 0, 0); // Clear the hook.
+
       // sqlite3_step() will return a detailed error code only if we used
       // sqlite3_prepare_v2(). Otherwise, sqlite3_reset() returns the
       // error.
@@ -768,7 +888,14 @@ namespace odb
 #endif
         translate_error (e, conn_);
 
-      return static_cast<unsigned long long> (sqlite3_changes (h));
+      int r (sqlite3_changes (h));
+
+      // Stream parameters, if any.
+      //
+      if (stream && r != 0)
+        stream_param (param_.bind, param_.count, sd);
+
+      return static_cast<unsigned long long> (r);
     }
 
     // delete_statement
diff --git a/odb/sqlite/statement.hxx b/odb/sqlite/statement.hxx
index 1be0cab..7344daf 100644
--- a/odb/sqlite/statement.hxx
+++ b/odb/sqlite/statement.hxx
@@ -30,7 +30,8 @@ namespace odb
   {
     class connection;
 
-    class LIBODB_SQLITE_EXPORT statement: public odb::statement
+    class LIBODB_SQLITE_EXPORT statement: public odb::statement,
+                                          public active_object
     {
     public:
       typedef sqlite::connection connection_type;
@@ -72,7 +73,7 @@ namespace odb
                  statement_kind sk,
                  const binding* process,
                  bool optimize)
-        : conn_ (conn)
+          : active_object (conn)
       {
         init (text.c_str (), text.size (), sk, process, optimize);
       }
@@ -82,7 +83,7 @@ namespace odb
                  statement_kind sk,
                  const binding* process,
                  bool optimize)
-        : conn_ (conn)
+          : active_object (conn)
       {
         init (text, std::strlen (text), sk, process, optimize);
       }
@@ -93,13 +94,15 @@ namespace odb
                  statement_kind sk,
                  const binding* process,
                  bool optimize)
-        : conn_ (conn)
+          : active_object (conn)
       {
         init (text, text_size, sk, process, optimize);
       }
 
     protected:
-      void
+      // Return true if we bound any stream parameters.
+      //
+      bool
       bind_param (const bind*, std::size_t count);
 
       // Extract row columns into the bound buffers. If the truncated
@@ -110,6 +113,21 @@ namespace odb
       bool
       bind_result (const bind*, std::size_t count, bool truncated = false);
 
+      // Stream (so to speak) parameters.
+      //
+      struct stream_data
+      {
+        std::string db;
+        std::string table;
+        long long rowid;
+      };
+
+      void
+      stream_param (const bind*, std::size_t count, const stream_data&);
+
+      friend void
+      update_hook (void*, const char*, const char*, long long);
+
       // Active state.
       //
     protected:
@@ -146,10 +164,12 @@ namespace odb
         return r;
       }
 
-    protected:
-      friend class sqlite::connection;
+      // The active_object interface.
+      //
+      virtual void
+      clear ();
 
-      connection_type& conn_;
+    protected:
       auto_handle<sqlite3_stmt> stmt_;
 
 #if SQLITE_VERSION_NUMBER < 3005003
@@ -165,38 +185,6 @@ namespace odb
             statement_kind,
             const binding* process,
             bool optimize);
-
-      // Doubly-linked list of active statements.
-      //
-    protected:
-      void
-      list_add ()
-      {
-        next_ = conn_.statements_;
-        conn_.statements_ = this;
-
-        if (next_ != 0)
-          next_->prev_ = this;
-      }
-
-      void
-      list_remove ()
-      {
-        (prev_ == 0 ? conn_.statements_ : prev_->next_) = next_;
-
-        if (next_ != 0)
-          next_->prev_ = prev_;
-
-        prev_ = 0;
-        next_ = this;
-      }
-
-      // prev_ == 0 means we are the first element.
-      // next_ == 0 means we are the last element.
-      // next_ == this means we are not on the list (prev_ should be 0).
-      //
-      statement* prev_;
-      statement* next_;
     };
 
     class LIBODB_SQLITE_EXPORT generic_statement: public statement
diff --git a/odb/sqlite/stream.cxx b/odb/sqlite/stream.cxx
new file mode 100644
index 0000000..8de199e
--- /dev/null
+++ b/odb/sqlite/stream.cxx
@@ -0,0 +1,121 @@
+// file      : odb/sqlite/stream.cxx
+// copyright : Copyright (c) 2005-2015 Code Synthesis Tools CC
+// license   : GNU GPL v2; see accompanying LICENSE file
+
+#include <sqlite3.h>
+
+#if SQLITE_VERSION_NUMBER >= 3004000
+
+#include <odb/sqlite/stream.hxx>
+
+#include <stdexcept> // invalid_argument
+
+#include <odb/sqlite/error.hxx>
+#include <odb/sqlite/database.hxx>
+#include <odb/sqlite/transaction.hxx>
+
+using namespace std;
+
+namespace odb
+{
+  namespace sqlite
+  {
+    stream::
+    stream (const char* db,
+            const char* table,
+            const char* column,
+            long long rowid,
+            bool rw)
+        : active_object (transaction::current ().connection ())
+    {
+      int e (sqlite3_blob_open (conn_.handle (),
+                                db,
+                                table,
+                                column,
+                                static_cast<sqlite_int64> (rowid),
+                                rw,
+                                &h_));
+
+      if (e != SQLITE_OK)
+        translate_error (e, conn_);
+
+      list_add (); // Add ourselves to the active objects list.
+    }
+
+    size_t stream::
+    size () const
+    {
+      return static_cast<size_t> (sqlite3_blob_bytes (h_));
+    }
+
+    void stream::
+    read (void* buf, size_t n, size_t o)
+    {
+      int e (sqlite3_blob_read (
+               h_, buf, static_cast<int> (n), static_cast<int> (o)));
+
+      if (e != SQLITE_OK)
+      {
+        if (e == SQLITE_ERROR)
+          throw invalid_argument ("read past end");
+        else
+          translate_error (e, conn_);
+      }
+    }
+
+    void stream::
+    write (const void* buf, size_t n, size_t o)
+    {
+      int e (sqlite3_blob_write (
+               h_, buf, static_cast<int> (n), static_cast<int> (o)));
+
+      if (e != SQLITE_OK)
+      {
+        if (e == SQLITE_ERROR)
+          throw invalid_argument ("write past end");
+        else
+          translate_error (e, conn_);
+      }
+    }
+
+    void stream::
+    close (bool check)
+    {
+      if (h_ != 0)
+      {
+        list_remove ();
+
+        int e (sqlite3_blob_close (h_));
+        h_ = 0; // No use trying again.
+
+        if (check && e != SQLITE_OK)
+          translate_error (e, conn_);
+      }
+    }
+
+#if SQLITE_VERSION_NUMBER >= 3007004
+    void stream::
+    reopen (long long rowid)
+    {
+      int e (sqlite3_blob_reopen (h_, rowid));
+
+      if (e != SQLITE_OK)
+        // According to the SQLite documentation, the handle is now
+        // considered aborted, which means we still need to close it.
+        //
+        translate_error (e, conn_);
+    }
+#endif
+
+    void stream::
+    clear ()
+    {
+      // This call can be part of the stack unwinding, so don't check
+      // for the errors.
+      //
+      close (false);
+    }
+  }
+}
+
+#endif // SQLITE_VERSION_NUMBER >= 3004000
diff --git a/odb/sqlite/stream.hxx b/odb/sqlite/stream.hxx
new file mode 100644
index 0000000..387dcf0
--- /dev/null
+++ b/odb/sqlite/stream.hxx
@@ -0,0 +1,82 @@
+// file      : odb/sqlite/stream.hxx
+// copyright : Copyright (c) 2005-2015 Code Synthesis Tools CC
+// license   : GNU GPL v2; see accompanying LICENSE file
+
+#ifndef ODB_SQLITE_STREAM_HXX
+#define ODB_SQLITE_STREAM_HXX
+
+#include <odb/pre.hxx>
+
+#include <sqlite3.h>
+
+#include <cstddef>   // std::size_t
+
+#include <odb/sqlite/connection.hxx>
+#include <odb/sqlite/details/export.hxx>
+
+namespace odb
+{
+  namespace sqlite
+  {
+    // SQLite incremental BLOB/TEXT I/O stream. Available since
+    // SQLite 3.4.0.
+    //
+    class LIBODB_SQLITE_EXPORT stream: public active_object
+    {
+    public:
+      stream (const char* db,
+              const char* table,
+              const char* column,
+              long long rowid,
+              bool rw);
+
+      std::size_t
+      size () const;
+
+      // The following two functions throw std::invalid_argument if
+      // offset + n is past size().
+      //
+      void
+      read (void* buf, std::size_t n, std::size_t offset = 0);
+
+      void
+      write (const void* buf, std::size_t n, std::size_t offset = 0);
+
+      sqlite3_blob*
+      handle () const {return h_;}
+
+      // Close without reporting errors, if any.
+      //
+      virtual
+      ~stream () {close (false);}
+
+      // Close, by default with reporting errors, if any.
+      //
+      void
+      close (bool check = true);
+
+      // Open the same BLOB but in a different row. Can be faster
+      // than creating a new stream instance. Note that the stream
+      // must be in the open state prior to calling this function.
+      // Only available since SQLite 3.7.4.
+      //
+#if SQLITE_VERSION_NUMBER >= 3007004
+      void
+      reopen (long long rowid);
+#endif
+
+    protected:
+      // The active_object interface.
+      //
+      virtual void
+      clear ();
+
+    private:
+      sqlite3_blob* h_;
+    };
+  }
+}
+
+#include <odb/post.hxx>
+
+#endif // ODB_SQLITE_STREAM_HXX
diff --git a/odb/sqlite/text-stream.hxx b/odb/sqlite/text-stream.hxx
new file mode 100644
index 0000000..71d3053
--- /dev/null
+++ b/odb/sqlite/text-stream.hxx
@@ -0,0 +1,33 @@
+// file      : odb/sqlite/text-stream.hxx
+// copyright : Copyright (c) 2005-2015 Code Synthesis Tools CC
+// license   : GNU GPL v2; see accompanying LICENSE file
+
+#ifndef ODB_SQLITE_TEXT_STREAM_HXX
+#define ODB_SQLITE_TEXT_STREAM_HXX
+
+#include <odb/pre.hxx>
+
+#include <odb/sqlite/text.hxx>
+#include <odb/sqlite/stream.hxx>
+#include <odb/sqlite/details/export.hxx>
+
+namespace odb
+{
+  namespace sqlite
+  {
+    class LIBODB_SQLITE_EXPORT text_stream: public stream
+    {
+    public:
+      text_stream (const text& b, bool rw)
+          : stream (b.db ().c_str (),
+                    b.table ().c_str (),
+                    b.column ().c_str (),
+                    b.rowid (),
+                    rw) {}
+    };
+  }
+}
+
+#include <odb/post.hxx>
+
+#endif // ODB_SQLITE_TEXT_STREAM_HXX
diff --git a/odb/sqlite/text.hxx b/odb/sqlite/text.hxx
new file mode 100644
index 0000000..1e7e25f
--- /dev/null
+++ b/odb/sqlite/text.hxx
@@ -0,0 +1,70 @@
+// file      : odb/sqlite/text.hxx
+// copyright : Copyright (c) 2005-2015 Code Synthesis Tools CC
+// license   : GNU GPL v2; see accompanying LICENSE file
+
+#ifndef ODB_SQLITE_TEXT_HXX
+#define ODB_SQLITE_TEXT_HXX
+
+#include <odb/pre.hxx>
+
+#include <string>
+#include <cstddef> // std::size_t
+
+// Carefully allow this header to be included into the ODB compilation.
+//
+#ifndef ODB_COMPILER
+#  include <odb/sqlite/forward.hxx>
+#  include <odb/sqlite/details/export.hxx>
+#endif
+
+namespace odb
+{
+  namespace sqlite
+  {
+#ifdef ODB_COMPILER
+    #pragma db sqlite:type("TEXT STREAM")
+    class text
+#else
+    class LIBODB_SQLITE_EXPORT text
+#endif
+    {
+    public:
+      // TEXT size to provision for. Set before calling persist() or update().
+      //
+      explicit
+      text (std::size_t size = 0): size_ (size) {}
+
+      std::size_t size () const {return size_;}
+      void size (std::size_t s) {size_ = s;}
+
+      const std::string& db () const {return db_;}
+      const std::string& table () const {return table_;}
+      const std::string& column () const {return column_;}
+      long long rowid () const {return rowid_;}
+
+      void
+      clear ()
+      {
+        size_ = 0;
+        db_.clear ();
+        table_.clear ();
+        column_.clear ();
+        rowid_ = 0;
+      }
+
+    private:
+#ifndef ODB_COMPILER
+      friend struct default_value_traits<text, id_text_stream>;
+#endif
+      std::size_t size_;
+      mutable std::string db_;
+      mutable std::string table_;
+      mutable std::string column_;
+      mutable long long rowid_;
+    };
+  }
+}
+
+#include <odb/post.hxx>
+
+#endif // ODB_SQLITE_TEXT_HXX
diff --git a/odb/sqlite/traits.hxx b/odb/sqlite/traits.hxx
index de8e3c3..dc4d113 100644
--- a/odb/sqlite/traits.hxx
+++ b/odb/sqlite/traits.hxx
@@ -25,22 +25,18 @@
 #include <odb/details/buffer.hxx>
 #include <odb/details/wrapper-p.hxx>
 
+#include <odb/sqlite/forward.hxx>
 #include <odb/sqlite/version.hxx>
 #include <odb/sqlite/sqlite-types.hxx>
 #include <odb/sqlite/details/export.hxx>
 
+#include <odb/sqlite/text.hxx>
+#include <odb/sqlite/blob.hxx>
+
 namespace odb
 {
   namespace sqlite
   {
-    enum database_type_id
-    {
-      id_integer,
-      id_real,
-      id_text,
-      id_blob
-    };
-
     //
     // image_traits
     //
@@ -67,6 +63,18 @@ namespace odb
     template <typename T>
     struct image_traits<T, id_blob> {typedef details::buffer image_type;};
 
+    template <typename T>
+    struct image_traits<T, id_text_stream>
+    {
+      typedef stream_buffers image_type;
+    };
+
+    template <typename T>
+    struct image_traits<T, id_blob_stream>
+    {
+      typedef stream_buffers image_type;
+    };
+
     //
     // value_traits
     //
@@ -140,6 +148,20 @@ namespace odb
       {
         vtraits::set_image (b, n, is_null, wtraits::get_ref (v));
       }
+
+      // TEXT and BLOB STREAM.
+      //
+      static void
+      set_value (W& v, const stream_buffers& b, std::size_t n, bool is_null)
+      {
+        vtraits::set_value (wtraits::set_ref (v), b, n, is_null);
+      }
+
+      static void
+      set_image (stream_buffers& b, std::size_t& n, bool& is_null, const W& v)
+      {
+        vtraits::set_image (b, n, is_null, wtraits::get_ref (v));
+      }
     };
 
     template <typename W, database_type_id ID>
@@ -191,6 +213,26 @@ namespace odb
         if (!is_null)
           vtraits::set_image (b, n, is_null, wtraits::get_ref (v));
       }
+
+      // TEXT and BLOB STREAM.
+      //
+      static void
+      set_value (W& v, const stream_buffers& b, std::size_t n, bool is_null)
+      {
+        if (is_null)
+          wtraits::set_null (v);
+        else
+          vtraits::set_value (wtraits::set_ref (v), b, n, is_null);
+      }
+
+      static void
+      set_image (stream_buffers& b, std::size_t& n, bool& is_null, const W& v)
+      {
+        is_null = wtraits::get_null (v);
+
+        if (!is_null)
+          vtraits::set_image (b, n, is_null, wtraits::get_ref (v));
+      }
     };
 
     template <typename T, database_type_id ID>
@@ -811,6 +853,82 @@ namespace odb
     };
 #endif
 
+    // text (stream) specialization.
+    //
+    template <>
+    struct LIBODB_SQLITE_EXPORT default_value_traits<text, id_text_stream>
+    {
+    public:
+      typedef text value_type;
+      typedef std::string query_type;
+      typedef stream_buffers image_type;
+
+      static void
+      set_value (text& v, const stream_buffers& b, std::size_t, bool is_null)
+      {
+        if (!is_null)
+        {
+          v.db_ = b.db.in;
+          v.table_ = b.table.in;
+          v.column_ = b.column.in;
+          v.rowid_ = b.rowid.in;
+        }
+      }
+
+      static void
+      set_image (stream_buffers& b,
+                 std::size_t& n,
+                 bool& is_null,
+                 const text& v)
+      {
+        is_null = false;
+        n = v.size_;
+
+        b.db.out = &v.db_;
+        b.table.out = &v.table_;
+        b.column.out = &v.column_;
+        b.rowid.out = &v.rowid_;
+      }
+    };
+
+    // blob (stream) specialization.
+    //
+    template <>
+    struct LIBODB_SQLITE_EXPORT default_value_traits<blob, id_blob_stream>
+    {
+    public:
+      typedef blob value_type;
+      typedef std::vector<char> query_type;
+      typedef stream_buffers image_type;
+
+      static void
+      set_value (blob& v, const stream_buffers& b, std::size_t, bool is_null)
+      {
+        if (!is_null)
+        {
+          v.db_ = b.db.in;
+          v.table_ = b.table.in;
+          v.column_ = b.column.in;
+          v.rowid_ = b.rowid.in;
+        }
+      }
+
+      static void
+      set_image (stream_buffers& b,
+                 std::size_t& n,
+                 bool& is_null,
+                 const blob& v)
+      {
+        is_null = false;
+        n = v.size_;
+
+        b.db.out = &v.db_;
+        b.table.out = &v.table_;
+        b.column.out = &v.column_;
+        b.rowid.out = &v.rowid_;
+      }
+    };
+
     //
     // type_traits
     //
@@ -970,6 +1088,18 @@ namespace odb
       static const database_type_id db_type_id = id_blob;
     };
 #endif
+
+    template <>
+    struct default_type_traits<text>
+    {
+      static const database_type_id db_type_id = id_text_stream;
+    };
+
+    template <>
+    struct default_type_traits<blob>
+    {
+      static const database_type_id db_type_id = id_blob_stream;
+    };
   }
 }