fast unspent queries for RethinkDB

bigchaindb/backend/rethinkdb/query.py

@@ -120,14 +120,16 @@ def get_spent(connection, transaction_id, output):
 
 
 @register_query(RethinkDBConnection)
-def get_owned_ids(connection, owner):
-    return connection.run(
-            r.table('bigchain', read_mode=READ_MODE)
+def get_owned_ids(connection, owner, unwrap=True):
+    query = (r.table('bigchain', read_mode=READ_MODE)
              .get_all(owner, index='outputs')
              .distinct()
-             .concat_map(lambda doc: doc['block']['transactions'])
-             .filter(lambda tx: tx['outputs'].contains(
+             .concat_map(unroll_block_transactions)
+             .filter(lambda doc: doc['block']['transactions']['outputs'].contains(
                 lambda c: c['public_keys'].contains(owner))))
+    if unwrap:
+        query = query.map(lambda doc: doc['block']['transactions'])
+    return connection.run(query)
 
 
 @register_query(RethinkDBConnection)
@@ -253,3 +255,29 @@ def get_unvoted_blocks(connection, node_pubkey):
     #        database level. Solving issue #444 can help untangling the situation
     unvoted_blocks = filter(lambda block: not utils.is_genesis_block(block), unvoted)
     return unvoted_blocks
+
+
+@register_query(RethinkDBConnection)
+def get_votes_for_blocks_by_voter(connection, block_ids, node_pubkey):
+    return connection.run(
+        r.table('votes')
+        .filter(lambda row: r.expr(block_ids).contains(row['vote']['voting_for_block']))
+        .filter(lambda row: row['node_pubkey'] == node_pubkey))
+
+
+def unroll_block_transactions(block):
+    """ Simulate unrolling a transaction into block in MongoDB """
+    return block['block']['transactions'].map(
+             lambda tx: block.merge({'block': {'transactions': tx}}))
+
+
+@register_query(RethinkDBConnection)
+def get_spending_transactions(connection, links):
+    query = (
+        r.table('bigchain')
+        .get_all(*[(l['txid'], l['output']) for l in links], index='inputs')
+        .concat_map(unroll_block_transactions)
+        .filter(lambda doc: r.expr(links).set_intersection(
+            doc['block']['transactions']['inputs'].map(lambda i: i['fulfills'])))
+    )
+    return connection.run(query)

bigchaindb/fastquery.py

@@ -48,8 +48,8 @@ class FastQuery:
             outputs: list of TransactionLink
         """
         links = [o.to_dict() for o in outputs]
-        wrapped = self.filter_valid_blocks(
-                list(query.get_spending_transactions(self.connection, links)))
+        spending_txs = query.get_spending_transactions(self.connection, links)
+        wrapped = self.filter_valid_blocks(list(spending_txs))
         spends = {TransactionLink.from_dict(input_['fulfills'])
                   for block in wrapped
                   for input_ in block['block']['transactions']['inputs']}

tests/test_fastquery.py

@@ -51,27 +51,32 @@ def test_get_outputs_by_pubkey(b, user_pk, user2_pk, blockdata):
 def test_filter_spent_outputs(b, user_pk):
     out = [([user_pk], 1)]
     tx1 = Transaction.create([user_pk], out * 3)
+
+    # There are 3 inputs
     inputs = tx1.to_inputs()
+
+    # Each spent individually
     tx2 = Transaction.transfer([inputs[0]], out, tx1.id)
     tx3 = Transaction.transfer([inputs[1]], out, tx1.id)
     tx4 = Transaction.transfer([inputs[2]], out, tx1.id)
 
+    # The CREATE and first TRANSFER are valid. tx2 produces a new unspent.
     for tx in [tx1, tx2]:
         block = Block([tx])
         b.write_block(block)
         b.write_vote(b.vote(block.id, '', True))
 
-    # mark invalid
+    # The second TRANSFER is invalid. inputs[1] remains unspent.
     block = Block([tx3])
     b.write_block(block)
     b.write_vote(b.vote(block.id, '', False))
 
-    # undecided
+    # The third TRANSFER is undecided. It procuces a new unspent.
     block = Block([tx4])
     b.write_block(block)
 
-    unspents = b.fastquery.filter_spent_outputs(
-             b.fastquery.get_outputs_by_pubkey(user_pk))
+    outputs = b.fastquery.get_outputs_by_pubkey(user_pk)
+    unspents = b.fastquery.filter_spent_outputs(outputs)
 
     assert set(unspents) == {
         inputs[1].fulfills,