From ccbd507c3559ac54336211a47ff776ebf21bcb01 Mon Sep 17 00:00:00 2001
From: Jason Volk <jason@zemos.net>
Date: Thu, 14 Sep 2017 12:59:20 -0700
Subject: [PATCH] ircd::db: Fix issues with cell.

---
 include/ircd/db/cell.h |  6 +++--
 include/ircd/db/json.h |  2 +-
 include/ircd/db/row.h  |  3 ---
 ircd/db.cc             | 59 ++++++++++++++++++++++++++++--------------
 4 files changed, 44 insertions(+), 26 deletions(-)

diff --git a/include/ircd/db/cell.h b/include/ircd/db/cell.h
index 0d3b41495..273a58d5b 100644
--- a/include/ircd/db/cell.h
+++ b/include/ircd/db/cell.h
@@ -53,7 +53,6 @@ struct ircd::db::cell
 	struct delta;
 
 	column c;
-	string_view index;
 	database::snapshot ss;
 	std::unique_ptr<rocksdb::Iterator> it;
 
@@ -66,6 +65,7 @@ struct ircd::db::cell
 	explicit operator column &()                 { return c;                                      }
 
 	bool valid() const;                          // cell exists
+	bool valid(const string_view &eq) const;     // valid_equal
 	operator bool() const                        { return valid();                                }
 	bool operator!() const                       { return !valid();                               }
 
@@ -91,11 +91,13 @@ struct ircd::db::cell
 	string_view exchange(const string_view &desired);
 
 	// [GET] load cell only (returns valid)
-	bool load(gopts = {});
+	bool load(const string_view &index = {}, gopts = {});
 
+	cell(column, std::unique_ptr<rocksdb::Iterator>, gopts = {});
 	cell(column, const string_view &index, std::unique_ptr<rocksdb::Iterator>, gopts = {});
 	cell(column, const string_view &index, gopts = {});
 	cell(database &, const string_view &column, const string_view &index, gopts = {});
+	cell(database &, const string_view &column, gopts = {});
 	cell();
 	cell(cell &&) noexcept;
 	cell(const cell &) = delete;
diff --git a/include/ircd/db/json.h b/include/ircd/db/json.h
index fb442ccdb..4780980ca 100644
--- a/include/ircd/db/json.h
+++ b/include/ircd/db/json.h
@@ -74,7 +74,7 @@ ircd::json::tuple<T...> &
 ircd::db::set(json::tuple<T...> &tuple,
               const row &row)
 {
-	for(auto &cell : row)
+	for(const auto &cell : row)
 		if(cell.valid())
 			json::set(tuple, cell.col(), cell.val());
 		else
diff --git a/include/ircd/db/row.h b/include/ircd/db/row.h
index 1f781d265..cebd13dc1 100644
--- a/include/ircd/db/row.h
+++ b/include/ircd/db/row.h
@@ -76,9 +76,6 @@ struct ircd::db::row
     // [SET] Perform operation
 	void operator()(const op &, const string_view &col, const string_view &val = {}, const sopts & = {});
 
-	// All cells
-	void load(const gopts & = {});               // !DANGER! not atomic
-
 	row(std::vector<cell> cells = {})
 	:its{std::move(cells)}
 	{}
diff --git a/ircd/db.cc b/ircd/db.cc
index 0e7794ea1..4367979e8 100644
--- a/ircd/db.cc
+++ b/ircd/db.cc
@@ -1380,6 +1380,16 @@ ircd::db::cell::cell()
 {
 }
 
+ircd::db::cell::cell(database &d,
+                     const string_view &colname,
+                     gopts opts)
+:cell
+{
+	column(d[colname]), std::unique_ptr<rocksdb::Iterator>{}, std::move(opts)
+}
+{
+}
+
 ircd::db::cell::cell(database &d,
                      const string_view &colname,
                      const string_view &index,
@@ -1395,10 +1405,12 @@ ircd::db::cell::cell(column column,
                      const string_view &index,
                      gopts opts)
 :c{std::move(column)}
-,index{index}
 ,ss{opts.snapshot}
-,it{ss? seek(this->c, this->index, opts) : std::unique_ptr<rocksdb::Iterator>{}}
+,it{ss? seek(this->c, index, opts) : std::unique_ptr<rocksdb::Iterator>{}}
 {
+	if(bool(this->it))
+		if(!valid_equal(*this->it, index))
+			this->it.reset();
 }
 
 ircd::db::cell::cell(column column,
@@ -1406,7 +1418,18 @@ ircd::db::cell::cell(column column,
                      std::unique_ptr<rocksdb::Iterator> it,
                      gopts opts)
 :c{std::move(column)}
-,index{index}
+,ss{opts.snapshot}
+,it{std::move(it)}
+{
+	seek(*this, index);
+	if(!valid_equal(*this->it, index))
+		this->it.reset();
+}
+
+ircd::db::cell::cell(column column,
+                     std::unique_ptr<rocksdb::Iterator> it,
+                     gopts opts)
+:c{std::move(column)}
 ,ss{std::move(opts.snapshot)}
 ,it{std::move(it)}
 {
@@ -1416,7 +1439,6 @@ ircd::db::cell::cell(column column,
 ircd::db::cell::cell(cell &&o)
 noexcept
 :c{std::move(o.c)}
-,index{std::move(o.index)}
 ,ss{std::move(o.ss)}
 ,it{std::move(o.it)}
 {
@@ -1428,7 +1450,6 @@ ircd::db::cell::operator=(cell &&o)
 noexcept
 {
 	c = std::move(o.c);
-	index = std::move(o.index);
 	ss = std::move(o.ss);
 	it = std::move(o.it);
 
@@ -1442,7 +1463,8 @@ noexcept
 }
 
 bool
-ircd::db::cell::load(gopts opts)
+ircd::db::cell::load(const string_view &index,
+                     gopts opts)
 {
 	database &d(c);
 	if(valid() && !opts.snapshot && sequence(ss) == sequence(d))
@@ -1541,7 +1563,7 @@ ircd::db::cell::key()
 	if(!valid())
 		load();
 
-	return likely(valid())? db::key(*it) : index;
+	return likely(valid())? db::key(*it) : string_view{};
 }
 
 ircd::string_view
@@ -1555,14 +1577,21 @@ ircd::string_view
 ircd::db::cell::key()
 const
 {
-	return likely(valid())? db::key(*it) : index;
+	return likely(valid())? db::key(*it) : string_view{};
+}
+
+bool
+ircd::db::cell::valid(const string_view &eq)
+const
+{
+	return bool(it) && db::valid_equal(*it, eq);
 }
 
 bool
 ircd::db::cell::valid()
 const
 {
-	return it && (index.empty() || valid_equal(*it, index));
+	return bool(it) && db::valid(*it);
 }
 
 ///////////////////////////////////////////////////////////////////////////////
@@ -1831,16 +1860,6 @@ const
 	});
 }
 
-void
-ircd::db::row::load(const gopts &opts)
-{
-	std::for_each(std::begin(its), std::end(its), [&opts]
-	(auto &cell)
-	{
-		cell.load(opts);
-	});
-}
-
 ///////////////////////////////////////////////////////////////////////////////
 //
 // db/column.h
@@ -2450,7 +2469,7 @@ ircd::db::append(rocksdb::WriteBatch &batch,
 	append(batch, column, column::delta
 	{
 		std::get<op>(delta),
-		std::get<cell *>(delta)->index,
+		std::get<cell *>(delta)->key(),
 		std::get<string_view>(delta)
 	});
 }