implement Pipelet.._transactional_fetch(); use it in Pipelet..add_source() and Encapsulate; use this_query on first fetch in Pipelet..add_source()