Implement SQLite incremental BLOB/TEXT I/O

14 files changed, 376 insertions, 58 deletions

diff --git a/odb/relational/context.hxx b/odb/relational/context.hxx
index 02adfc7..b172e45 100644
--- a/odb/relational/context.hxx
+++ b/odb/relational/context.hxx
@@ -20,6 +20,7 @@ namespace relational
     statement_select,
     statement_insert,
     statement_update,
+    statement_delete,
     statement_where   // WHERE clause.
   };
 
diff --git a/odb/relational/model.hxx b/odb/relational/model.hxx
index 13c67d7..3b53c67 100644
--- a/odb/relational/model.hxx
+++ b/odb/relational/model.hxx
@@ -115,8 +115,7 @@ namespace relational
                        (key_prefix_.empty () ? m.name () : key_prefix_));
 
         sema_rel::column& c (
-          model_.new_node<sema_rel::column> (
-            col_id, column_type (), null (m)));
+          model_.new_node<sema_rel::column> (col_id, type (m), null (m)));
         c.set ("cxx-location", m.location ());
         c.set ("member-path", member_path_);
         model_.new_edge<sema_rel::unames> (table_, c, name);
@@ -142,6 +141,12 @@ namespace relational
         return true;
       }
 
+      virtual string
+      type (semantics::data_member&)
+      {
+        return object_columns_base::column_type ();
+      }
+
       virtual bool
       null (semantics::data_member&)
       {
diff --git a/odb/relational/mssql/source.cxx b/odb/relational/mssql/source.cxx
index 42c29a7..4384761 100644
--- a/odb/relational/mssql/source.cxx
+++ b/odb/relational/mssql/source.cxx
@@ -24,7 +24,7 @@ namespace relational
         query_parameters (base const& x): base (x) {}
 
         virtual string
-        auto_id ()
+        auto_id (semantics::data_member&, const string&, const string&)
         {
           return "";
         }
diff --git a/odb/relational/oracle/source.cxx b/odb/relational/oracle/source.cxx
index 802bc99..b69b291 100644
--- a/odb/relational/oracle/source.cxx
+++ b/odb/relational/oracle/source.cxx
@@ -22,7 +22,7 @@ namespace relational
         query_parameters (base const& x): base (x), i_ (0) {}
 
         virtual string
-        next ()
+        next (semantics::data_member&, const string&, const string&)
         {
           ostringstream ss;
           ss << ":" << ++i_;
@@ -31,7 +31,7 @@ namespace relational
         }
 
         virtual string
-        auto_id ()
+        auto_id (semantics::data_member&, const string&, const string&)
         {
           return quote_id (sequence_name (table_)) + ".nextval";
         }
@@ -612,9 +612,15 @@ namespace relational
             // Top-level auto id.
             //
             if (id != 0 && !poly_derived && auto_ (*id))
-              r = "RETURNING " +
-                convert_from (column_qname (*id), *id->back ()) +
-                " INTO " + qp.next ();
+            {
+              semantics::data_member& idb (*id->back ());
+
+              const string& name (column_qname (*id));
+              const string& type (column_type (idb));
+
+              r = "RETURNING " + convert_from (name, type, idb) +
+                " INTO " + qp.next (idb, name, type);
+            }
           }
 
           return r;
diff --git a/odb/relational/pgsql/source.cxx b/odb/relational/pgsql/source.cxx
index b8270ac..b0155b7 100644
--- a/odb/relational/pgsql/source.cxx
+++ b/odb/relational/pgsql/source.cxx
@@ -24,7 +24,7 @@ namespace relational
         query_parameters (base const& x): base (x), i_ (0) {}
 
         virtual string
-        next ()
+        next (semantics::data_member&, const string&, const string&)
         {
           ostringstream ss;
           ss << "$" << ++i_;
@@ -33,7 +33,7 @@ namespace relational
         }
 
         virtual string
-        auto_id ()
+        auto_id (semantics::data_member&, const string&, const string&)
         {
           return "DEFAULT";
         }
diff --git a/odb/relational/source.cxx b/odb/relational/source.cxx
index 3b96ab5..e9ddd77 100644
--- a/odb/relational/source.cxx
+++ b/odb/relational/source.cxx
@@ -746,7 +746,7 @@ traverse_object (type& c)
       os << strlit (s) << endl;
     }
 
-    instance<query_parameters> qp (table);
+    instance<query_parameters> qp (statement_insert, table);
 
     string extra (persist_statement_extra (c, *qp, persist_after_columns));
 
@@ -854,7 +854,7 @@ traverse_object (type& c)
         }
 
         string where ("WHERE ");
-        instance<query_parameters> qp (table);
+        instance<query_parameters> qp (statement_select, table);
         for (object_columns_list::iterator b (id_cols->begin ()), i (b);
              i != id_cols->end (); ++i)
         {
@@ -862,7 +862,7 @@ traverse_object (type& c)
             where += " AND ";
 
           where += qtable + "." + quote_id (i->name) + "=" +
-            convert_to (qp->next (), i->type, *i->member);
+            convert_to (qp->next (*i), i->type, *i->member);
         }
 
         os << strlit (where);
@@ -936,7 +936,7 @@ traverse_object (type& c)
       os << strlit ("FROM " + qtable + " ") << endl;
 
       string where ("WHERE ");
-      instance<query_parameters> qp (table);
+      instance<query_parameters> qp (statement_select, table);
       for (object_columns_list::iterator b (id_cols->begin ()), i (b);
            i != id_cols->end (); ++i)
       {
@@ -944,7 +944,7 @@ traverse_object (type& c)
           where += " AND ";
 
         where += qtable + "." + quote_id (i->name) + "=" +
-          convert_to (qp->next (), i->type, *i->member);
+          convert_to (qp->next (*i), i->type, *i->member);
       }
 
       os << strlit (where) << ";"
@@ -957,7 +957,7 @@ traverse_object (type& c)
     {
       string sep (versioned ? "\n" : " ");
 
-      instance<query_parameters> qp (table);
+      instance<query_parameters> qp (statement_update, table);
 
       statement_columns sc;
       {
@@ -1005,15 +1005,18 @@ traverse_object (type& c)
           where += " AND ";
 
         where += quote_id (i->name) + "=" +
-          convert_to (qp->next (), i->type, *i->member);
+          convert_to (qp->next (*i), i->type, *i->member);
       }
 
       // Top-level version column.
       //
       if (opt != 0 && !poly_derived)
       {
-        where += " AND " + column_qname (*opt, column_prefix ()) + "=" +
-          convert_to (qp->next (), *opt);
+        string name (column_qname (*opt, column_prefix ()));
+        string type (column_type (*opt));
+
+        where += " AND " + name + "=" +
+          convert_to (qp->next (*opt, name, type), type, *opt);
       }
 
       os << strlit (where) << ";"
@@ -1023,7 +1026,7 @@ traverse_object (type& c)
     // erase_statement
     //
     {
-      instance<query_parameters> qp (table);
+      instance<query_parameters> qp (statement_delete, table);
       os << "const char " << traits << "::erase_statement[] =" << endl
          << strlit ("DELETE FROM " + qtable + " ") << endl;
 
@@ -1035,7 +1038,7 @@ traverse_object (type& c)
           where += " AND ";
 
         where += quote_id (i->name) + "=" +
-          convert_to (qp->next (), i->type, *i->member);
+          convert_to (qp->next (*i), i->type, *i->member);
       }
 
       os << strlit (where) << ";"
@@ -1044,7 +1047,7 @@ traverse_object (type& c)
 
     if (opt != 0 && !poly_derived)
     {
-      instance<query_parameters> qp (table);
+      instance<query_parameters> qp (statement_delete, table);
 
       os << "const char " << traits << "::optimistic_erase_statement[] " <<
         "=" << endl
@@ -1058,13 +1061,16 @@ traverse_object (type& c)
           where += " AND ";
 
         where += quote_id (i->name) + "=" +
-          convert_to (qp->next (), i->type, *i->member);
+          convert_to (qp->next (*i), i->type, *i->member);
       }
 
       // Top-level version column.
       //
-      where += " AND " + column_qname (*opt, column_prefix ()) + "=" +
-        convert_to (qp->next (), *opt);
+      string name (column_qname (*opt, column_prefix ()));
+      string type (column_type (*opt));
+
+      where += " AND " + name + "=" +
+        convert_to (qp->next (*opt, name, type), type, *opt);
 
       os << strlit (where) << ";"
          << endl;
diff --git a/odb/relational/source.hxx b/odb/relational/source.hxx
index f495cef..9b23b31 100644
--- a/odb/relational/source.hxx
+++ b/odb/relational/source.hxx
@@ -55,21 +55,39 @@ namespace relational
     {
       typedef query_parameters base;
 
-      query_parameters (qname const& table): table_ (table) {}
+      query_parameters (statement_kind sk, qname const& table)
+          : sk_ (sk), table_ (table) {}
 
       virtual string
-      next ()
+      next (semantics::data_member&,
+            const std::string& /*column*/, // Table qualified and quoted.
+            const std::string& /*sqlt*/)
       {
         return "?";
       }
 
       virtual string
-      auto_id ()
+      auto_id (semantics::data_member& m,
+               const std::string& column,
+               const std::string& sqlt)
       {
-        return next ();
+        return next (m, column, sqlt);
+      }
+
+      string
+      next (const object_columns_list::column& c)
+      {
+        return next (*c.member, quote_id (c.name), c.type);
+      }
+
+      string
+      next (const statement_column& c)
+      {
+        return next (*c.member, c.column, c.type);
       }
 
     protected:
+      statement_kind sk_;
       qname table_;
     };
 
@@ -338,7 +356,7 @@ namespace relational
         else if (param_ != 0)
         {
           r += '=';
-          r += convert_to (param_->next (), sqlt, m);
+          r += convert_to (param_->next (m, column, sqlt), sqlt, m);
         }
         else if (sk_ == statement_select)
           r = convert_from (r, sqlt, m);
@@ -3634,7 +3652,7 @@ namespace relational
               os << strlit (c + (++i != e ? "," : "") + sep) << endl;
             }
 
-            instance<query_parameters> qp (inv_table);
+            instance<query_parameters> qp (statement_select, inv_table);
             os << strlit ("FROM " + inv_qtable + sep) << endl;
 
             string where ("WHERE ");
@@ -3645,7 +3663,7 @@ namespace relational
                 where += " AND ";
 
               where += inv_qtable + "." + quote_id (i->name) + "=" +
-                convert_to (qp->next (), i->type, *i->member);
+                convert_to (qp->next (*i), i->type, *i->member);
             }
             os << strlit (where);
           }
@@ -3691,7 +3709,7 @@ namespace relational
               os << strlit (c + (++i != e ? "," : "") + sep) << endl;
             }
 
-            instance<query_parameters> qp (table);
+            instance<query_parameters> qp (statement_select, table);
             os << strlit ("FROM " + qtable + sep) << endl;
 
             string where ("WHERE ");
@@ -3702,7 +3720,7 @@ namespace relational
                 where += " AND ";
 
               where += qtable + "." + quote_id (i->name) + "=" +
-                convert_to (qp->next (), i->type, *i->member);
+                convert_to (qp->next (*i), i->type, *i->member);
             }
 
             if (ordered)
@@ -3781,7 +3799,7 @@ namespace relational
             os << strlit ("VALUES" + sep) << endl;
 
             string values ("(");
-            instance<query_parameters> qp (table);
+            instance<query_parameters> qp (statement_insert, table);
             for (statement_columns::const_iterator b (sc.begin ()), i (b),
                    e (sc.end ()); i != e; ++i)
             {
@@ -3791,7 +3809,7 @@ namespace relational
                 values += sep;
               }
 
-              values += convert_to (qp->next (), i->type, *i->member);
+              values += convert_to (qp->next (*i), i->type, *i->member);
             }
             values += ')';
 
@@ -3808,7 +3826,7 @@ namespace relational
                << strlit ("UPDATE " + qtable + sep) << endl
                << strlit ("SET" + sep) << endl;
 
-            instance<query_parameters> qp (table);
+            instance<query_parameters> qp (statement_update, table);
             statement_columns sc;
             {
               bool f (false);                       // Imperfect forwarding.
@@ -3834,14 +3852,14 @@ namespace relational
                 where += " AND ";
 
               where += quote_id (i->name) + "=" +
-                convert_to (qp->next (), i->type, *i->member);
+                convert_to (qp->next (*i), i->type, *i->member);
             }
 
             for (object_columns_list::iterator b (ik_cols->begin ()), i (b);
                  i != ik_cols->end (); ++i)
             {
               where += " AND " + quote_id (i->name) + "=" +
-                convert_to (qp->next (), i->type, *i->member);
+                convert_to (qp->next (*i), i->type, *i->member);
             }
 
             os << strlit (where) << ";"
@@ -3858,7 +3876,7 @@ namespace relational
                << endl;
           else
           {
-            instance<query_parameters> qp (table);
+            instance<query_parameters> qp (statement_delete, table);
 
             os << strlit ("DELETE FROM " + qtable + " ") << endl;
 
@@ -3870,7 +3888,7 @@ namespace relational
                 where += " AND ";
 
               where += quote_id (i->name) + "=" +
-                convert_to (qp->next (), i->type, *i->member);
+                convert_to (qp->next (*i), i->type, *i->member);
             }
 
             if (smart)
@@ -3880,7 +3898,7 @@ namespace relational
               {
                 where += " AND " + quote_id (i->name) +
                   (ck == ck_ordered ? ">=" : "=") +
-                  convert_to (qp->next (), i->type, *i->member);
+                  convert_to (qp->next (*i), i->type, *i->member);
               }
             }
 
@@ -5961,7 +5979,7 @@ namespace relational
           }
 
           string where ("WHERE ");
-          instance<query_parameters> qp (table);
+          instance<query_parameters> qp (statement_select, table);
           for (object_columns_list::iterator b (id_cols->begin ()), i (b);
                i != id_cols->end (); ++i)
           {
@@ -5969,7 +5987,7 @@ namespace relational
               where += " AND ";
 
             where += qtable + "." + quote_id (i->name) + "=" +
-              convert_to (qp->next (), i->type, *i->member);
+              convert_to (qp->next (*i), i->type, *i->member);
           }
 
           os << strlit (where) << ";"
@@ -5980,7 +5998,7 @@ namespace relational
         //
         if (update || update_opt)
         {
-          instance<query_parameters> qp (table);
+          instance<query_parameters> qp (statement_update, table);
 
           statement_columns sc;
           {
@@ -6029,13 +6047,16 @@ namespace relational
               where += " AND ";
 
             where += quote_id (i->name) + "=" +
-              convert_to (qp->next (), i->type, *i->member);
+              convert_to (qp->next (*i), i->type, *i->member);
           }
 
           if (s.optimistic ()) // Note: not update_opt.
           {
-            where += " AND " + column_qname (*opt, column_prefix ()) + "=" +
-              convert_to (qp->next (), *opt);
+            string name (column_qname (*opt, column_prefix ()));
+            string type (column_type (*opt));
+
+            where += " AND " + name + "=" +
+              convert_to (qp->next (*opt, name, type), type, *opt);
           }
 
           os << strlit (where) << ";"
@@ -6450,16 +6471,23 @@ namespace relational
       }
 
       virtual bool
-      traverse_column (semantics::data_member& m, string const&, bool first)
+      traverse_column (semantics::data_member& m,
+                       string const& name,
+                       bool first)
       {
         string p;
 
         if (version (m))
           p = version_value (m);
-        else if (auto_ (m)) // Only simple, direct id can be auto.
-          p = qp_.auto_id ();
         else
-          p = qp_.next ();
+        {
+          const string& qname (quote_id (name));
+          const string& type (column_type ());
+
+          p = auto_ (m) // Only simple, direct id can be auto.
+            ? qp_.auto_id (m, qname, type)
+            : qp_.next (m, qname, type);
+        }
 
         if (!p.empty ())
         {
diff --git a/odb/relational/sqlite/common.cxx b/odb/relational/sqlite/common.cxx
index 36a0e86..4ab8968 100644
--- a/odb/relational/sqlite/common.cxx
+++ b/odb/relational/sqlite/common.cxx
@@ -39,12 +39,18 @@ namespace relational
         }
       case sql_type::TEXT:
         {
-          traverse_text (mi);
+          if (mi.st->stream)
+            traverse_text_stream (mi);
+          else
+            traverse_text (mi);
           break;
         }
       case sql_type::BLOB:
         {
-          traverse_blob (mi);
+          if (mi.st->stream)
+            traverse_blob_stream (mi);
+          else
+            traverse_blob (mi);
           break;
         }
       case sql_type::invalid:
@@ -108,6 +114,12 @@ namespace relational
       type_ = "details::buffer";
     }
 
+    void member_image_type::
+    traverse_stream (member_info&)
+    {
+      type_ = "sqlite::stream_buffers";
+    }
+
     entry<member_image_type> member_image_type_;
 
     //
@@ -170,6 +182,18 @@ namespace relational
       type_id_ = "sqlite::id_blob";
     }
 
+    void member_database_type_id::
+    traverse_text_stream (member_info&)
+    {
+      type_id_ = "sqlite::id_text_stream";
+    }
+
+    void member_database_type_id::
+    traverse_blob_stream (member_info&)
+    {
+      type_id_ = "sqlite::id_blob_stream";
+    }
+
     entry<member_database_type_id> member_database_type_id_;
 
     //
diff --git a/odb/relational/sqlite/common.hxx b/odb/relational/sqlite/common.hxx
index bea0889..b5a61a1 100644
--- a/odb/relational/sqlite/common.hxx
+++ b/odb/relational/sqlite/common.hxx
@@ -56,6 +56,23 @@ namespace relational
       traverse_string (member_info&)
       {
       }
+
+      virtual void
+      traverse_text_stream (member_info& m)
+      {
+        traverse_stream (m);
+      }
+
+      virtual void
+      traverse_blob_stream (member_info& m)
+      {
+        traverse_stream (m);
+      }
+
+      virtual void
+      traverse_stream (member_info&)
+      {
+      }
     };
 
     struct member_image_type: relational::member_image_type,
@@ -82,6 +99,9 @@ namespace relational
       virtual void
       traverse_string (member_info&);
 
+      virtual void
+      traverse_stream (member_info&);
+
     private:
       string type_;
     };
@@ -114,6 +134,12 @@ namespace relational
       virtual void
       traverse_blob (member_info&);
 
+      virtual void
+      traverse_text_stream (member_info&);
+
+      virtual void
+      traverse_blob_stream (member_info&);
+
     private:
       string type_id_;
     };
diff --git a/odb/relational/sqlite/context.cxx b/odb/relational/sqlite/context.cxx
index 007d67f..94b8067 100644
--- a/odb/relational/sqlite/context.cxx
+++ b/odb/relational/sqlite/context.cxx
@@ -339,10 +339,23 @@ namespace relational
           if (ids_.empty ())
             return error ("expected SQLite type name");
 
+          // First check our own types.
+          //
+          if (ids_.size () == 2 && ids_[0] == "TEXT" && ids_[1] == "STREAM")
+          {
+            r.type = sql_type::TEXT;
+            r.stream = true;
+          }
+          if (ids_.size () == 2 && ids_[0] == "BLOB" && ids_[1] == "STREAM")
+          {
+            r.type = sql_type::BLOB;
+            r.stream = true;
+          }
+          //
           // Apply the first four rules of the SQLite type to affinity
           // conversion algorithm.
           //
-          if (find ("INT"))
+          else if (find ("INT"))
             r.type = sql_type::INTEGER;
           else if (find ("TEXT") || find ("CHAR") || find ("CLOB"))
             r.type = sql_type::TEXT;
diff --git a/odb/relational/sqlite/context.hxx b/odb/relational/sqlite/context.hxx
index 7c5107e..6fcdc47 100644
--- a/odb/relational/sqlite/context.hxx
+++ b/odb/relational/sqlite/context.hxx
@@ -26,9 +26,10 @@ namespace relational
         invalid
       };
 
-      sql_type (): type (invalid) {}
+      sql_type (): type (invalid), stream (false) {}
 
       core_type type;
+      bool stream; // TEXT or BLOB via sqlite3_blob_open().
 
       // Conversion expressions for custom database types.
       //
diff --git a/odb/relational/sqlite/header.cxx b/odb/relational/sqlite/header.cxx
index 9432825..e548413 100644
--- a/odb/relational/sqlite/header.cxx
+++ b/odb/relational/sqlite/header.cxx
@@ -48,6 +48,15 @@ namespace relational
              << "bool " << mi.var << "null;"
              << endl;
         }
+
+        virtual void
+        traverse_stream (member_info& mi)
+        {
+          os << image_type << " " << mi.var << "value;"
+             << "std::size_t " << mi.var << "size;"
+             << "bool " << mi.var << "null;"
+             << endl;
+        }
       };
       entry<image_member> image_member_;
     }
diff --git a/odb/relational/sqlite/model.cxx b/odb/relational/sqlite/model.cxx
index 47eb88b..45216b7 100644
--- a/odb/relational/sqlite/model.cxx
+++ b/odb/relational/sqlite/model.cxx
@@ -23,6 +23,27 @@ namespace relational
       {
         object_columns (base const& x): base (x) {}
 
+        virtual string
+        type (semantics::data_member& m)
+        {
+          // Translate BLOB|TEXT STREAM to just BLOB|TEXT.
+          //
+          string r (relational::object_columns::type (m));
+
+          sql_type const& t (parse_sql_type (r, m, false));
+          if (t.stream)
+          {
+            switch (t.type)
+            {
+            case sql_type::BLOB: r = "BLOB"; break;
+            case sql_type::TEXT: r = "TEXT"; break;
+            default: break;
+            }
+          }
+
+          return r;
+        }
+
         virtual bool
         null (semantics::data_member& m)
         {
diff --git a/odb/relational/sqlite/source.cxx b/odb/relational/sqlite/source.cxx
index 0d821ac..2f6623e 100644
--- a/odb/relational/sqlite/source.cxx
+++ b/odb/relational/sqlite/source.cxx
@@ -17,6 +17,72 @@ namespace relational
     {
       namespace relational = relational::source;
 
+      struct query_parameters: relational::query_parameters, context
+      {
+        query_parameters (base const& x): base (x) {}
+
+        virtual string
+        next (semantics::data_member& m,
+              const string& column,
+              const string& sqlt)
+        {
+          // Handle stream columns. Specifically, we somehow need to
+          // pass the column name to the code that runs in the
+          // statement. So what we are going to do is encode it
+          // in the parameter name.
+          //
+          if (sk_ == statement_insert || sk_ == statement_update)
+          {
+            const sql_type& t (parse_sql_type (sqlt, m, false));
+            if (t.stream)
+            {
+              // The column name is quoted.
+              //
+              string r (column);
+              r[0] = '$';               // Replace leading '"'.
+              r.resize (r.size () - 1); // Remove trailing '"'.
+
+              // Verify it only contains allowed characters.
+              //
+              for (size_t i (1); i != r.size (); ++i)
+              {
+                char c (r[i]);
+                if (c != '_' &&
+                    (c < '0' || c > '9') &&
+                    (c < 'a' || c > 'z') &&
+                    (c < 'A' || c > 'Z'))
+                {
+                  cerr << m.file () << ":" << m.line () << ":" << m.column ()
+                       << ": error: unsupported character '" << c << "' in "
+                       << sqlt << " column name " << column << endl;
+
+                  cerr << m.file () << ":" << m.line () << ":" << m.column ()
+                       << ": info: STREAM column can contain alpha-numeric "
+                       << "characters plus '_'" << endl;
+
+                  throw operation_failed ();
+                }
+              }
+
+              // For TEXT columns, since we use the *_bind_zeroblob()
+              // function (there is no *_bind_zerotext()), the value
+              // that will be stored is BLOB, not TEXT, unless we
+              // explicitly CAST it. The user better make sure the
+              // encoding of raw TEXT data they are going to write
+              // matches the database encoding.
+              //
+              if (t.type == sql_type::TEXT)
+                r = "CAST(" + r + " AS TEXT)";
+
+              return r;
+            }
+          }
+
+          return "?";
+        }
+      };
+      entry<query_parameters> query_parameters_;
+
       //
       // bind
       //
@@ -71,6 +137,15 @@ namespace relational
             "value.capacity ();"
              << b << ".is_null = &" << arg << "." << mi.var << "null;";
         }
+
+        virtual void
+        traverse_stream (member_info& mi)
+        {
+          os << b << ".type = sqlite::bind::stream;"
+             << b << ".buffer = &" << arg << "." << mi.var << "value;"
+             << b << ".size = &" << arg << "." << mi.var << "size;"
+             << b << ".is_null = &" << arg << "." << mi.var << "null;";
+        }
       };
       entry<bind_member> bind_member_;
 
@@ -110,6 +185,13 @@ namespace relational
              << "grew = true;"
              << "}";
         }
+
+        virtual void
+        traverse_stream (member_info&)
+        {
+          os << e << " = false;"
+             << endl;
+        }
       };
       entry<grow_member> grow_member_;
 
@@ -166,6 +248,17 @@ namespace relational
              << "i." << mi.var << "null = is_null;"
              << "grew = grew || (cap != i." << mi.var << "value.capacity ());";
         }
+
+        virtual void
+        traverse_stream (member_info& mi)
+        {
+          os << traits << "::set_image (" << endl
+             << "i." << mi.var << "value," << endl
+             << "i." << mi.var << "size," << endl
+             << "is_null," << endl
+             << member << ");"
+             << "i." << mi.var << "null = is_null;";
+        }
       };
       entry<init_image_member> init_image_member_;
 
@@ -220,10 +313,65 @@ namespace relational
              << "i." << mi.var << "null);"
              << endl;
         }
+
+        virtual void
+        traverse_stream (member_info& mi)
+        {
+          os << traits << "::set_value (" << endl
+             << member << "," << endl
+             << "i." << mi.var << "value," << endl
+             << "i." << mi.var << "size," << endl
+             << "i." << mi.var << "null);"
+             << endl;
+        }
       };
       entry<init_value_member> init_value_member_;
 
-      struct container_traits: relational::container_traits, context
+      struct statement_columns_common: context
+      {
+        void
+        process (relational::statement_columns& cs, statement_kind sk)
+        {
+          using relational::statement_columns;
+
+          // For SELECT statements, add _ROWID_ "follow-up" column to
+          // each stream column. The reason we need both, and not just
+          // ROWID is the NULL value. Let's hope that SELECT'ing a BLOB
+          // but not actually reading it with sqlite3_result_blob() is
+          // as fast as not SELECT'ing it.
+          //
+          if (sk != statement_select)
+            return;
+
+          for (statement_columns::iterator i (cs.begin ());
+               i != cs.end (); ++i)
+          {
+            if (parse_sql_type (i->type, *i->member).stream)
+            {
+              // Column is already table-qualified and quoted. Do some
+              // surgery to replace it with _ROWID_. That is, we want to
+              // transform "table"."column" to "table"."_ROWID_".
+              //
+              string c (i->column);
+              string::size_type n (c.size ()), p (c.rfind ('"', n - 2));
+              assert (p != string::npos);
+              string as (c, p + 1, n - p - 2);
+              c.resize (p);
+              c += "\"_ROWID_\"";
+
+              // We are going to pack this "tightly", without any newlines,
+              // so that the statement processing code treats them as a
+              // single column.
+              //
+              i->column += ',';
+              i->column += c;
+            }
+          }
+        }
+      };
+
+      struct container_traits: relational::container_traits,
+                               statement_columns_common
       {
         container_traits (base const& x): base (x) {}
 
@@ -234,10 +382,32 @@ namespace relational
           // interleaving statements.
           //
         }
+
+        virtual void
+        process_statement_columns (relational::statement_columns& cols,
+                                   statement_kind sk,
+                                   bool)
+        {
+          statement_columns_common::process (cols, sk);
+        }
       };
       entry<container_traits> container_traits_;
 
-      struct class_: relational::class_, context
+      struct section_traits: relational::section_traits,
+                             statement_columns_common
+      {
+        section_traits (base const& x): base (x) {}
+
+        virtual void
+        process_statement_columns (relational::statement_columns& cols,
+                                   statement_kind sk)
+        {
+          statement_columns_common::process (cols, sk);
+        }
+      };
+      entry<section_traits> section_traits_;
+
+      struct class_: relational::class_, statement_columns_common
       {
         class_ (base const& x): base (x) {}
 
@@ -274,6 +444,14 @@ namespace relational
 
           return base::join_syntax (vo);
         }
+
+        virtual void
+        process_statement_columns (relational::statement_columns& cols,
+                                   statement_kind sk,
+                                   bool)
+        {
+          statement_columns_common::process (cols, sk);
+        }
       };
       entry<class_> class_entry_;
     }