From 595f7dc701535ad7561f68a517f870f04c418f82 Mon Sep 17 00:00:00 2001 From: Rodolphe Marques Date: Fri, 6 Jan 2017 11:23:21 +0100 Subject: [PATCH] several fixes to mongodb queries --- bigchaindb/backend/mongodb/query.py | 108 +++++++++++++++++++++++----- tests/pipelines/test_vote.py | 30 ++++---- tests/test_config_utils.py | 5 +- 3 files changed, 108 insertions(+), 35 deletions(-) diff --git a/bigchaindb/backend/mongodb/query.py b/bigchaindb/backend/mongodb/query.py index 8765d110..658eb3ca 100644 --- a/bigchaindb/backend/mongodb/query.py +++ b/bigchaindb/backend/mongodb/query.py @@ -1,6 +1,7 @@ """Query implementation for MongoDB""" from time import time +from itertools import chain from pymongo import ReturnDocument @@ -36,7 +37,8 @@ def delete_transaction(conn, *transaction_id): @register_query(MongoDBConnection) def get_stale_transactions(conn, reassign_delay): return conn.db['backlog']\ - .find({'assignment_timestamp': {'$lt': time() - reassign_delay}}) + .find({'assignment_timestamp': {'$lt': time() - reassign_delay}}, + projection={'_id': False}) @register_query(MongoDBConnection) @@ -58,7 +60,10 @@ def get_transaction_from_block(conn, transaction_id, block_id): @register_query(MongoDBConnection) def get_transaction_from_backlog(conn, transaction_id): - return conn.db['backlog'].find_one({'id': transaction_id}) + return conn.db['backlog']\ + .find_one({'id': transaction_id}, + projection={'_id': False, 'assignee': False, + 'assignment_timestamp': False}) @register_query(MongoDBConnection) @@ -70,33 +75,83 @@ def get_blocks_status_from_transaction(conn, transaction_id): @register_query(MongoDBConnection) def get_txids_by_asset_id(conn, asset_id): - return conn.db['bigchain']\ - .find({'block.transactions.asset.id': asset_id}, - projection=['id']) + # get the txid of the create transaction for asset_id + cursor = conn.db['bigchain'].aggregate([ + {'$match': { + 'block.transactions.id': asset_id, + 'block.transactions.operation': 'CREATE' + }}, + {'$unwind': '$block.transactions'}, + {'$match': { + 'block.transactions.id': asset_id, + 'block.transactions.operation': 'CREATE' + }}, + {'$project': {'block.transactions.id': True}} + ]) + create_tx_txids = (elem['block']['transactions']['id'] for elem in cursor) + + # get txids of transfer transaction with asset_id + cursor = conn.db['bigchain'].aggregate([ + {'$match': { + 'block.transactions.asset.id': asset_id + }}, + {'$unwind': '$block.transactions'}, + {'$match': { + 'block.transactions.asset.id': asset_id + }}, + {'$project': {'block.transactions.id': True}} + ]) + transfer_tx_ids = (elem['block']['transactions']['id'] for elem in cursor) + + return chain(create_tx_txids, transfer_tx_ids) @register_query(MongoDBConnection) def get_asset_by_id(conn, asset_id): - return conn.db['bigchain']\ - .find_one({'block.transactions.asset.id': asset_id, - 'block.transactions.asset.operation': 'CREATE'}, - projection=['block.transactions.asset']) + cursor = conn.db['bigchain'].aggregate([ + {'$match': { + 'block.transactions.id': asset_id, + 'block.transactions.operation': 'CREATE' + }}, + {'$unwind': '$block.transactions'}, + {'$match': { + 'block.transactions.id': asset_id, + 'block.transactions.operation': 'CREATE' + }}, + {'$project': {'block.transactions.asset': True}} + ]) + # we need to access some nested fields before returning so lets use a + # generator to avoid having to read all records on the cursor at this point + return (elem['block']['transactions'] for elem in cursor) @register_query(MongoDBConnection) def get_spent(conn, transaction_id, condition_id): - return conn.db['bigchain']\ - .find_one({'block.transactions.fulfillments.input.txid': - transaction_id, - 'block.transactions.fulfillments.input.cid': - condition_id}) + cursor = conn.db['bigchain'].aggregate([ + {'$unwind': '$block.transactions'}, + {'$match': { + 'block.transactions.fulfillments.input.txid': transaction_id, + 'block.transactions.fulfillments.input.cid': condition_id + }} + ]) + # we need to access some nested fields before returning so lets use a + # generator to avoid having to read all records on the cursor at this point + return (elem['block']['transactions'] for elem in cursor) @register_query(MongoDBConnection) def get_owned_ids(conn, owner): - return conn.db['bigchain']\ - .find({'block.transactions.transaction.conditions.owners_after': - owner}) + cursor = conn.db['bigchain'].aggregate([ + {'$unwind': '$block.transactions'}, + {'$match': { + 'block.transactions.conditions.owners_after': { + '$elemMatch': {'$eq': owner} + } + }} + ]) + # we need to access some nested fields before returning so lets use a + # generator to avoid having to read all records on the cursor at this point + return (elem['block']['transactions'] for elem in cursor) @register_query(MongoDBConnection) @@ -121,7 +176,8 @@ def write_block(conn, block): @register_query(MongoDBConnection) def get_block(conn, block_id): - return conn.db['bigchain'].find_one({'id': block_id}) + return conn.db['bigchain'].find_one({'id': block_id}, + projection={'_id': False}) @register_query(MongoDBConnection) @@ -184,4 +240,18 @@ def get_last_voted_block(conn, node_pubkey): @register_query(MongoDBConnection) def get_unvoted_blocks(conn, node_pubkey): - pass + return conn.db['bigchain'].aggregate([ + {'$lookup': { + 'from': 'votes', + 'localField': 'id', + 'foreignField': 'vote.voting_for_block', + 'as': 'votes' + }}, + {'$match': { + 'votes.node_pubkey': {'$ne': node_pubkey}, + 'block.transactions.operation': {'$ne': 'GENESIS'} + }}, + {'$project': { + 'votes': False, '_id': False + }} + ]) diff --git a/tests/pipelines/test_vote.py b/tests/pipelines/test_vote.py index 1c46c065..e0b27f50 100644 --- a/tests/pipelines/test_vote.py +++ b/tests/pipelines/test_vote.py @@ -279,7 +279,7 @@ def test_valid_block_voting_with_transfer_transactions(monkeypatch, tx = Transaction.create([b.me], [([test_user_pub], 1)]) tx = tx.sign([b.me_private]) - monkeypatch.setattr('time.time', lambda: 1111111111) + monkeypatch.setattr('time.time', lambda: 1000000000) block = b.create_block([tx]) b.write_block(block) @@ -289,7 +289,7 @@ def test_valid_block_voting_with_transfer_transactions(monkeypatch, asset_id=tx.id) tx2 = tx2.sign([test_user_priv]) - monkeypatch.setattr('time.time', lambda: 2222222222) + monkeypatch.setattr('time.time', lambda: 2000000000) block2 = b.create_block([tx2]) b.write_block(block2) @@ -314,7 +314,7 @@ def test_valid_block_voting_with_transfer_transactions(monkeypatch, 'previous_block': genesis_block.id, 'is_block_valid': True, 'invalid_reason': None, - 'timestamp': '2222222222'} + 'timestamp': '2000000000'} serialized_vote = utils.serialize(vote_doc['vote']).encode() assert vote_doc['node_pubkey'] == b.me @@ -328,7 +328,7 @@ def test_valid_block_voting_with_transfer_transactions(monkeypatch, 'previous_block': block.id, 'is_block_valid': True, 'invalid_reason': None, - 'timestamp': '2222222222'} + 'timestamp': '2000000000'} serialized_vote2 = utils.serialize(vote2_doc['vote']).encode() assert vote2_doc['node_pubkey'] == b.me @@ -498,15 +498,15 @@ def test_voter_considers_unvoted_blocks_when_single_node(monkeypatch, b): outpipe = Pipe() - monkeypatch.setattr('time.time', lambda: 1111111111) + monkeypatch.setattr('time.time', lambda: 1000000000) block_ids = [] # insert blocks in the database while the voter process is not listening # (these blocks won't appear in the changefeed) - monkeypatch.setattr('time.time', lambda: 2222222222) + monkeypatch.setattr('time.time', lambda: 1000000020) block_1 = dummy_block(b) block_ids.append(block_1.id) - monkeypatch.setattr('time.time', lambda: 3333333333) + monkeypatch.setattr('time.time', lambda: 1000000030) b.write_block(block_1) block_2 = dummy_block(b) block_ids.append(block_2.id) @@ -522,7 +522,7 @@ def test_voter_considers_unvoted_blocks_when_single_node(monkeypatch, b): outpipe.get() # create a new block that will appear in the changefeed - monkeypatch.setattr('time.time', lambda: 4444444444) + monkeypatch.setattr('time.time', lambda: 1000000040) block_3 = dummy_block(b) block_ids.append(block_3.id) b.write_block(block_3) @@ -546,15 +546,15 @@ def test_voter_chains_blocks_with_the_previous_ones(monkeypatch, b): outpipe = Pipe() - monkeypatch.setattr('time.time', lambda: 1111111111) + monkeypatch.setattr('time.time', lambda: 1000000000) block_ids = [] - monkeypatch.setattr('time.time', lambda: 2222222222) + monkeypatch.setattr('time.time', lambda: 1000000020) block_1 = dummy_block(b) block_ids.append(block_1.id) b.write_block(block_1) - monkeypatch.setattr('time.time', lambda: 3333333333) + monkeypatch.setattr('time.time', lambda: 1000000030) block_2 = dummy_block(b) block_ids.append(block_2.id) b.write_block(block_2) @@ -588,9 +588,9 @@ def test_voter_checks_for_previous_vote(monkeypatch, b): inpipe = Pipe() outpipe = Pipe() - monkeypatch.setattr('time.time', lambda: 1111111111) + monkeypatch.setattr('time.time', lambda: 1000000000) - monkeypatch.setattr('time.time', lambda: 2222222222) + monkeypatch.setattr('time.time', lambda: 1000000020) block_1 = dummy_block(b) inpipe.put(block_1.to_dict()) assert len(list(query.get_votes_by_block_id(b.connection, block_1.id))) == 0 @@ -603,11 +603,11 @@ def test_voter_checks_for_previous_vote(monkeypatch, b): outpipe.get() # queue block for voting AGAIN - monkeypatch.setattr('time.time', lambda: 3333333333) + monkeypatch.setattr('time.time', lambda: 1000000030) inpipe.put(block_1.to_dict()) # queue another block - monkeypatch.setattr('time.time', lambda: 4444444444) + monkeypatch.setattr('time.time', lambda: 1000000040) block_2 = dummy_block(b) inpipe.put(block_2.to_dict()) diff --git a/tests/test_config_utils.py b/tests/test_config_utils.py index 2a326147..e190f622 100644 --- a/tests/test_config_utils.py +++ b/tests/test_config_utils.py @@ -102,7 +102,10 @@ def test_env_config(monkeypatch): def test_autoconfigure_read_both_from_file_and_env(monkeypatch, request): file_config = { - 'database': {'host': 'test-host'}, + 'database': { + 'host': 'test-host', + 'backend': request.config.getoption('--database-backend') + }, 'backlog_reassign_delay': 5 } monkeypatch.setattr('bigchaindb.config_utils.file_config', lambda *args, **kwargs: file_config)