killbill-uncached

diff --git a/account/src/main/java/com/ning/billing/account/dao/AccountSqlDao.java b/account/src/main/java/com/ning/billing/account/dao/AccountSqlDao.java
index dd1d652..8dcaa59 100644
--- a/account/src/main/java/com/ning/billing/account/dao/AccountSqlDao.java
+++ b/account/src/main/java/com/ning/billing/account/dao/AccountSqlDao.java
@@ -16,14 +16,12 @@
 
 package com.ning.billing.account.dao;
 
-import java.util.Iterator;
 import java.util.UUID;
 
 import org.skife.jdbi.v2.sqlobject.Bind;
 import org.skife.jdbi.v2.sqlobject.BindBean;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
 import org.skife.jdbi.v2.sqlobject.SqlUpdate;
-import org.skife.jdbi.v2.sqlobject.customizers.FetchSize;
 
 import com.ning.billing.account.api.Account;
 import com.ning.billing.callcontext.InternalCallContext;
@@ -41,16 +39,6 @@ public interface AccountSqlDao extends EntitySqlDao<AccountModelDao, Account> {
                                            @BindBean final InternalTenantContext context);
 
     @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
-    public Iterator<AccountModelDao> searchAccounts(@Bind("searchKey") final String searchKey,
-                                                    @Bind("likeSearchKey") final String likeSearchKey,
-                                                    @Bind("offset") final Long offset,
-                                                    @Bind("rowCount") final Long rowCount,
-                                                    @BindBean final InternalTenantContext context);
-
-    @SqlQuery
     public UUID getIdFromKey(@Bind("externalKey") final String key,
                              @BindBean final InternalTenantContext context);
 

diff --git a/account/src/main/java/com/ning/billing/account/dao/DefaultAccountDao.java b/account/src/main/java/com/ning/billing/account/dao/DefaultAccountDao.java
index ea710c3..15809c7 100644
--- a/account/src/main/java/com/ning/billing/account/dao/DefaultAccountDao.java
+++ b/account/src/main/java/com/ning/billing/account/dao/DefaultAccountDao.java
@@ -112,7 +112,7 @@ public class DefaultAccountDao extends EntityDaoBase<AccountModelDao, Account, A
                                               new PaginationIteratorBuilder<AccountModelDao, Account, AccountSqlDao>() {
                                                   @Override
                                                   public Iterator<AccountModelDao> build(final AccountSqlDao accountSqlDao, final Long limit) {
-                                                      return accountSqlDao.searchAccounts(searchKey, String.format("%%%s%%", searchKey), offset, limit, context);
+                                                      return accountSqlDao.search(searchKey, String.format("%%%s%%", searchKey), offset, limit, context);
                                                   }
                                               },
                                               offset,

diff --git a/account/src/main/resources/com/ning/billing/account/dao/AccountSqlDao.sql.stg b/account/src/main/resources/com/ning/billing/account/dao/AccountSqlDao.sql.stg
index c1e5255..e3f279a 100644
--- a/account/src/main/resources/com/ning/billing/account/dao/AccountSqlDao.sql.stg
+++ b/account/src/main/resources/com/ning/billing/account/dao/AccountSqlDao.sql.stg
@@ -86,22 +86,12 @@ getAccountByKey() ::= <<
     where external_key = :externalKey <AND_CHECK_TENANT()>;
 >>
 
-searchAccounts() ::= <<
-select SQL_CALC_FOUND_ROWS
-<allTableFields("t.")>
-from <tableName()> t
-where 1 = 1
-and (
-     <idField("t.")> = :searchKey
-  or t.name like :likeSearchKey
-  or t.email like :likeSearchKey
-  or t.external_key like :likeSearchKey
-  or t.company_name like :likeSearchKey
-)
-<AND_CHECK_TENANT("t.")>
-order by <recordIdField("t.")> ASC
-limit :offset, :rowCount
-;
+searchQuery(prefix) ::= <<<
+     <idField(prefix)> = :searchKey
+  or <prefix>.name like :likeSearchKey
+  or <prefix>.email like :likeSearchKey
+  or <prefix>.external_key like :likeSearchKey
+  or <prefix>.company_name like :likeSearchKey
 >>
 
 getIdFromKey() ::= <<

diff --git a/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java b/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java
index 441eba2..e10b94a 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java
@@ -256,7 +256,7 @@ public class DefaultInvoiceDao extends EntityDaoBase<InvoiceModelDao, Invoice, I
                                               new PaginationIteratorBuilder<InvoiceModelDao, Invoice, InvoiceSqlDao>() {
                                                   @Override
                                                   public Iterator<InvoiceModelDao> build(final InvoiceSqlDao invoiceSqlDao, final Long limit) {
-                                                      return invoiceSqlDao.searchInvoices(searchKey, offset, limit, context);
+                                                      return invoiceSqlDao.search(searchKey, String.format("%%%s%%", searchKey), offset, limit, context);
                                                   }
                                               },
                                               offset,

diff --git a/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceSqlDao.java b/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceSqlDao.java
index a0824db..aebd7ab 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceSqlDao.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceSqlDao.java
@@ -16,14 +16,12 @@
 
 package com.ning.billing.invoice.dao;
 
-import java.util.Iterator;
 import java.util.List;
 import java.util.UUID;
 
 import org.skife.jdbi.v2.sqlobject.Bind;
 import org.skife.jdbi.v2.sqlobject.BindBean;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
-import org.skife.jdbi.v2.sqlobject.customizers.FetchSize;
 
 import com.ning.billing.callcontext.InternalTenantContext;
 import com.ning.billing.invoice.api.Invoice;
@@ -38,15 +36,6 @@ public interface InvoiceSqlDao extends EntitySqlDao<InvoiceModelDao, Invoice> {
                                                     @BindBean final InternalTenantContext context);
 
     @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
-    public Iterator<InvoiceModelDao> searchInvoices(@Bind("searchKey") final String searchKey,
-                                                    @Bind("offset") final Long offset,
-                                                    @Bind("rowCount") final Long rowCount,
-                                                    @BindBean final InternalTenantContext context);
-
-    @SqlQuery
     UUID getInvoiceIdByPaymentId(@Bind("paymentId") final String paymentId,
                                  @BindBean final InternalTenantContext context);
 }

diff --git a/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoiceSqlDao.sql.stg b/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoiceSqlDao.sql.stg
index 304da69..aabe831 100644
--- a/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoiceSqlDao.sql.stg
+++ b/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoiceSqlDao.sql.stg
@@ -36,21 +36,11 @@ getInvoicesBySubscription() ::= <<
   ;
 >>
 
-searchInvoices() ::= <<
-select SQL_CALC_FOUND_ROWS
-<allTableFields("t.")>
-from <tableName()> t
-where 1 = 1
-and (
-     <idField("t.")> = :searchKey
-  or <recordIdField("t.")> = :searchKey
-  or t.account_id = :searchKey
-  or t.currency = :searchKey
-)
-<AND_CHECK_TENANT("t.")>
-order by <recordIdField("t.")> ASC
-limit :offset, :rowCount
-;
+searchQuery(prefix) ::= <<
+     <idField(prefix)> = :searchKey
+  or <recordIdField(prefix)> = :searchKey
+  or <prefix>.account_id = :searchKey
+  or <prefix>.currency = :searchKey
 >>
 
 getInvoiceIdByPaymentId() ::= <<

diff --git a/payment/src/main/java/com/ning/billing/payment/dao/PaymentMethodSqlDao.java b/payment/src/main/java/com/ning/billing/payment/dao/PaymentMethodSqlDao.java
index 3ad3b72..a05f1c6 100644
--- a/payment/src/main/java/com/ning/billing/payment/dao/PaymentMethodSqlDao.java
+++ b/payment/src/main/java/com/ning/billing/payment/dao/PaymentMethodSqlDao.java
@@ -23,10 +23,10 @@ import org.skife.jdbi.v2.sqlobject.Bind;
 import org.skife.jdbi.v2.sqlobject.BindBean;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
 import org.skife.jdbi.v2.sqlobject.SqlUpdate;
-import org.skife.jdbi.v2.sqlobject.customizers.FetchSize;
 
 import com.ning.billing.callcontext.InternalCallContext;
 import com.ning.billing.callcontext.InternalTenantContext;
+import com.ning.billing.commons.jdbi.statement.SmartFetchSize;
 import com.ning.billing.payment.api.PaymentMethod;
 import com.ning.billing.util.audit.ChangeType;
 import com.ning.billing.util.entity.dao.Audited;
@@ -57,9 +57,7 @@ public interface PaymentMethodSqlDao extends EntitySqlDao<PaymentMethodModelDao,
     List<PaymentMethodModelDao> getByAccountIdIncludedDelete(@Bind("accountId") final String accountId, @BindBean final InternalTenantContext context);
 
     @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
+    @SmartFetchSize(shouldStream = true)
     public Iterator<PaymentMethodModelDao> getByPluginName(@Bind("pluginName") final String pluginName,
                                                            @Bind("offset") final Long offset,
                                                            @Bind("rowCount") final Long rowCount,

diff --git a/payment/src/main/java/com/ning/billing/payment/dao/PaymentSqlDao.java b/payment/src/main/java/com/ning/billing/payment/dao/PaymentSqlDao.java
index e234194..606d49a 100644
--- a/payment/src/main/java/com/ning/billing/payment/dao/PaymentSqlDao.java
+++ b/payment/src/main/java/com/ning/billing/payment/dao/PaymentSqlDao.java
@@ -25,11 +25,11 @@ import org.skife.jdbi.v2.sqlobject.Bind;
 import org.skife.jdbi.v2.sqlobject.BindBean;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
 import org.skife.jdbi.v2.sqlobject.SqlUpdate;
-import org.skife.jdbi.v2.sqlobject.customizers.FetchSize;
 
 import com.ning.billing.callcontext.InternalCallContext;
 import com.ning.billing.callcontext.InternalTenantContext;
 import com.ning.billing.catalog.api.Currency;
+import com.ning.billing.commons.jdbi.statement.SmartFetchSize;
 import com.ning.billing.payment.api.Payment;
 import com.ning.billing.util.audit.ChangeType;
 import com.ning.billing.util.entity.dao.Audited;
@@ -69,9 +69,7 @@ public interface PaymentSqlDao extends EntitySqlDao<PaymentModelDao, Payment> {
                                                 @BindBean final InternalTenantContext context);
 
     @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
+    @SmartFetchSize(shouldStream = true)
     public Iterator<PaymentModelDao> getByPluginName(@Bind("pluginName") final String pluginName,
                                                      @Bind("offset") final Long offset,
                                                      @Bind("rowCount") final Long rowCount,

diff --git a/payment/src/main/java/com/ning/billing/payment/dao/RefundSqlDao.java b/payment/src/main/java/com/ning/billing/payment/dao/RefundSqlDao.java
index 4c3d5cb..18388b0 100644
--- a/payment/src/main/java/com/ning/billing/payment/dao/RefundSqlDao.java
+++ b/payment/src/main/java/com/ning/billing/payment/dao/RefundSqlDao.java
@@ -24,11 +24,11 @@ import org.skife.jdbi.v2.sqlobject.Bind;
 import org.skife.jdbi.v2.sqlobject.BindBean;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
 import org.skife.jdbi.v2.sqlobject.SqlUpdate;
-import org.skife.jdbi.v2.sqlobject.customizers.FetchSize;
 
 import com.ning.billing.callcontext.InternalCallContext;
 import com.ning.billing.callcontext.InternalTenantContext;
 import com.ning.billing.catalog.api.Currency;
+import com.ning.billing.commons.jdbi.statement.SmartFetchSize;
 import com.ning.billing.payment.api.Refund;
 import com.ning.billing.util.audit.ChangeType;
 import com.ning.billing.util.entity.dao.Audited;
@@ -55,9 +55,7 @@ public interface RefundSqlDao extends EntitySqlDao<RefundModelDao, Refund> {
                                               @BindBean final InternalTenantContext context);
 
     @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
+    @SmartFetchSize(shouldStream = true)
     public Iterator<RefundModelDao> getByPluginName(@Bind("pluginName") final String pluginName,
                                                     @Bind("offset") final Long offset,
                                                     @Bind("rowCount") final Long rowCount,

diff --git a/subscription/src/main/java/com/ning/billing/subscription/engine/dao/BundleSqlDao.java b/subscription/src/main/java/com/ning/billing/subscription/engine/dao/BundleSqlDao.java
index 7746e81..2e86a92 100644
--- a/subscription/src/main/java/com/ning/billing/subscription/engine/dao/BundleSqlDao.java
+++ b/subscription/src/main/java/com/ning/billing/subscription/engine/dao/BundleSqlDao.java
@@ -17,14 +17,12 @@
 package com.ning.billing.subscription.engine.dao;
 
 import java.util.Date;
-import java.util.Iterator;
 import java.util.List;
 
 import org.skife.jdbi.v2.sqlobject.Bind;
 import org.skife.jdbi.v2.sqlobject.BindBean;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
 import org.skife.jdbi.v2.sqlobject.SqlUpdate;
-import org.skife.jdbi.v2.sqlobject.customizers.FetchSize;
 
 import com.ning.billing.callcontext.InternalCallContext;
 import com.ning.billing.callcontext.InternalTenantContext;
@@ -62,13 +60,4 @@ public interface BundleSqlDao extends EntitySqlDao<SubscriptionBundleModelDao, S
     @SqlQuery
     public List<SubscriptionBundleModelDao> getBundlesForKey(@Bind("externalKey") String externalKey,
                                                              @BindBean final InternalTenantContext context);
-
-    @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
-    public Iterator<SubscriptionBundleModelDao> searchBundles(@Bind("searchKey") final String searchKey,
-                                                              @Bind("offset") final Long offset,
-                                                              @Bind("rowCount") final Long rowCount,
-                                                              @BindBean final InternalTenantContext context);
 }

diff --git a/subscription/src/main/java/com/ning/billing/subscription/engine/dao/DefaultSubscriptionDao.java b/subscription/src/main/java/com/ning/billing/subscription/engine/dao/DefaultSubscriptionDao.java
index 3c876f8..b20328d 100644
--- a/subscription/src/main/java/com/ning/billing/subscription/engine/dao/DefaultSubscriptionDao.java
+++ b/subscription/src/main/java/com/ning/billing/subscription/engine/dao/DefaultSubscriptionDao.java
@@ -195,7 +195,7 @@ public class DefaultSubscriptionDao extends EntityDaoBase<SubscriptionBundleMode
                                               new PaginationIteratorBuilder<SubscriptionBundleModelDao, SubscriptionBaseBundle, BundleSqlDao>() {
                                                   @Override
                                                   public Iterator<SubscriptionBundleModelDao> build(final BundleSqlDao bundleSqlDao, final Long limit) {
-                                                      return bundleSqlDao.searchBundles(searchKey, offset, limit, context);
+                                                      return bundleSqlDao.search(searchKey, String.format("%%%s%%", searchKey), offset, limit, context);
                                                   }
                                               },
                                               offset,

diff --git a/subscription/src/main/resources/com/ning/billing/subscription/engine/dao/BundleSqlDao.sql.stg b/subscription/src/main/resources/com/ning/billing/subscription/engine/dao/BundleSqlDao.sql.stg
index f5f9ade..bc883f6 100644
--- a/subscription/src/main/resources/com/ning/billing/subscription/engine/dao/BundleSqlDao.sql.stg
+++ b/subscription/src/main/resources/com/ning/billing/subscription/engine/dao/BundleSqlDao.sql.stg
@@ -78,18 +78,8 @@ account_id = :accountId
 ;
 >>
 
-searchBundles() ::= <<
-select SQL_CALC_FOUND_ROWS
-<allTableFields("t.")>
-from <tableName()> t
-where 1 = 1
-and (
-     <idField("t.")> = :searchKey
-  or t.external_key = :searchKey
-  or t.account_id = :searchKey
-)
-<AND_CHECK_TENANT("t.")>
-order by <recordIdField("t.")> ASC
-limit :offset, :rowCount
-;
+searchQuery(prefix) ::= <<
+     <idField(prefix)> = :searchKey
+  or <prefix>.external_key = :searchKey
+  or <prefix>.account_id = :searchKey
 >>

diff --git a/util/src/main/java/com/ning/billing/util/customfield/dao/CustomFieldSqlDao.java b/util/src/main/java/com/ning/billing/util/customfield/dao/CustomFieldSqlDao.java
index c257b74..d0755f1 100644
--- a/util/src/main/java/com/ning/billing/util/customfield/dao/CustomFieldSqlDao.java
+++ b/util/src/main/java/com/ning/billing/util/customfield/dao/CustomFieldSqlDao.java
@@ -16,7 +16,6 @@
 
 package com.ning.billing.util.customfield.dao;
 
-import java.util.Iterator;
 import java.util.List;
 import java.util.UUID;
 
@@ -24,7 +23,6 @@ import org.skife.jdbi.v2.sqlobject.Bind;
 import org.skife.jdbi.v2.sqlobject.BindBean;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
 import org.skife.jdbi.v2.sqlobject.SqlUpdate;
-import org.skife.jdbi.v2.sqlobject.customizers.FetchSize;
 
 import com.ning.billing.ObjectType;
 import com.ning.billing.callcontext.InternalCallContext;
@@ -47,14 +45,4 @@ public interface CustomFieldSqlDao extends EntitySqlDao<CustomFieldModelDao, Cus
     List<CustomFieldModelDao> getCustomFieldsForObject(@Bind("objectId") UUID objectId,
                                                        @Bind("objectType") ObjectType objectType,
                                                        @BindBean InternalTenantContext internalTenantContext);
-
-    @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
-    public Iterator<CustomFieldModelDao> searchCustomFields(@Bind("searchKey") final String searchKey,
-                                                            @Bind("likeSearchKey") final String likeSearchKey,
-                                                            @Bind("offset") final Long offset,
-                                                            @Bind("rowCount") final Long rowCount,
-                                                            @BindBean final InternalTenantContext context);
 }

diff --git a/util/src/main/java/com/ning/billing/util/customfield/dao/DefaultCustomFieldDao.java b/util/src/main/java/com/ning/billing/util/customfield/dao/DefaultCustomFieldDao.java
index 092d95a..207a703 100644
--- a/util/src/main/java/com/ning/billing/util/customfield/dao/DefaultCustomFieldDao.java
+++ b/util/src/main/java/com/ning/billing/util/customfield/dao/DefaultCustomFieldDao.java
@@ -148,7 +148,7 @@ public class DefaultCustomFieldDao extends EntityDaoBase<CustomFieldModelDao, Cu
                                               new PaginationIteratorBuilder<CustomFieldModelDao, CustomField, CustomFieldSqlDao>() {
                                                   @Override
                                                   public Iterator<CustomFieldModelDao> build(final CustomFieldSqlDao customFieldSqlDao, final Long limit) {
-                                                      return customFieldSqlDao.searchCustomFields(searchKey, String.format("%%%s%%", searchKey), offset, limit, context);
+                                                      return customFieldSqlDao.search(searchKey, String.format("%%%s%%", searchKey), offset, limit, context);
                                                   }
                                               },
                                               offset,

diff --git a/util/src/main/java/com/ning/billing/util/dao/AuditSqlDao.java b/util/src/main/java/com/ning/billing/util/dao/AuditSqlDao.java
index 7f91aac..ee475f1 100644
--- a/util/src/main/java/com/ning/billing/util/dao/AuditSqlDao.java
+++ b/util/src/main/java/com/ning/billing/util/dao/AuditSqlDao.java
@@ -24,10 +24,10 @@ import org.skife.jdbi.v2.sqlobject.BindBean;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
 import org.skife.jdbi.v2.sqlobject.SqlUpdate;
 import org.skife.jdbi.v2.sqlobject.customizers.Define;
-import org.skife.jdbi.v2.sqlobject.customizers.FetchSize;
 
 import com.ning.billing.callcontext.InternalCallContext;
 import com.ning.billing.callcontext.InternalTenantContext;
+import com.ning.billing.commons.jdbi.statement.SmartFetchSize;
 import com.ning.billing.util.audit.dao.AuditLogModelDao;
 import com.ning.billing.util.cache.Cachable;
 import com.ning.billing.util.cache.Cachable.CacheType;
@@ -52,15 +52,11 @@ public interface AuditSqlDao {
                                            @BindBean final InternalCallContext context);
 
     @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
+    @SmartFetchSize(shouldStream = true)
     public Iterator<AuditLogModelDao> getAuditLogsForAccountRecordId(@BindBean final InternalTenantContext context);
 
     @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
+    @SmartFetchSize(shouldStream = true)
     public Iterator<AuditLogModelDao> getAuditLogsForTableNameAndAccountRecordId(@Bind("tableName") final String tableName,
                                                                                  @BindBean final InternalTenantContext context);
 

diff --git a/util/src/main/java/com/ning/billing/util/entity/dao/DefaultPaginationSqlDaoHelper.java b/util/src/main/java/com/ning/billing/util/entity/dao/DefaultPaginationSqlDaoHelper.java
index e245ffd..ddc138b 100644
--- a/util/src/main/java/com/ning/billing/util/entity/dao/DefaultPaginationSqlDaoHelper.java
+++ b/util/src/main/java/com/ning/billing/util/entity/dao/DefaultPaginationSqlDaoHelper.java
@@ -44,13 +44,7 @@ public class DefaultPaginationSqlDaoHelper {
             @Override
             public Long inTransaction(final EntitySqlDaoWrapperFactory<EntitySqlDao> entitySqlDaoWrapperFactory) throws Exception {
                 final EntitySqlDao<M, E> sqlDao = entitySqlDaoWrapperFactory.become(sqlDaoClazz);
-                // TODO lame cast, but couldn't make sqlDaoClazz a Class<? extends S>
-                final Iterator<M> dumbIterator = paginationIteratorBuilder.build((S) sqlDao, 1L);
-                // Make sure to go through the results to close the connection
-                while (dumbIterator.hasNext()) {
-                    dumbIterator.next();
-                }
-                return sqlDao.getFoundRows(context);
+                return sqlDao.getSearchCount(context);
             }
         });
 

diff --git a/util/src/main/java/com/ning/billing/util/entity/dao/EntityDaoBase.java b/util/src/main/java/com/ning/billing/util/entity/dao/EntityDaoBase.java
index a0721bc..a477efd 100644
--- a/util/src/main/java/com/ning/billing/util/entity/dao/EntityDaoBase.java
+++ b/util/src/main/java/com/ning/billing/util/entity/dao/EntityDaoBase.java
@@ -77,7 +77,7 @@ public abstract class EntityDaoBase<M extends EntityModelDao<E>, E extends Entit
     protected abstract U generateAlreadyExistsException(final M entity, final InternalCallContext context);
 
     protected String getNaturalOrderingColumns() {
-        return "recordId";
+        return "record_id";
     }
 
     @Override
@@ -124,7 +124,7 @@ public abstract class EntityDaoBase<M extends EntityModelDao<E>, E extends Entit
 
         // Note: we need to perform the count before streaming the results, as the connection
         // will be busy as we stream the results out. This is also why we cannot use
-        // SQL_CALC_FOUND_ROWS / FOUND_ROWS (which may ne be faster anyways).
+        // SQL_CALC_FOUND_ROWS / FOUND_ROWS (which may not be faster anyways).
         final Long count = sqlDao.getCount(context);
 
         final Iterator<M> results = sqlDao.getAll(context);

diff --git a/util/src/main/java/com/ning/billing/util/entity/dao/EntitySqlDao.java b/util/src/main/java/com/ning/billing/util/entity/dao/EntitySqlDao.java
index 44506de..7430327 100644
--- a/util/src/main/java/com/ning/billing/util/entity/dao/EntitySqlDao.java
+++ b/util/src/main/java/com/ning/billing/util/entity/dao/EntitySqlDao.java
@@ -23,13 +23,14 @@ import org.skife.jdbi.v2.sqlobject.Bind;
 import org.skife.jdbi.v2.sqlobject.BindBean;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
 import org.skife.jdbi.v2.sqlobject.SqlUpdate;
-import org.skife.jdbi.v2.sqlobject.customizers.FetchSize;
+import org.skife.jdbi.v2.sqlobject.customizers.Define;
 import org.skife.jdbi.v2.sqlobject.mixins.CloseMe;
 import org.skife.jdbi.v2.sqlobject.mixins.Transactional;
 import org.skife.jdbi.v2.sqlobject.mixins.Transmogrifier;
 
 import com.ning.billing.callcontext.InternalCallContext;
 import com.ning.billing.callcontext.InternalTenantContext;
+import com.ning.billing.commons.jdbi.statement.SmartFetchSize;
 import com.ning.billing.entity.EntityPersistenceException;
 import com.ning.billing.util.audit.ChangeType;
 import com.ning.billing.util.cache.Cachable;
@@ -70,21 +71,25 @@ public interface EntitySqlDao<M extends EntityModelDao<E>, E extends Entity> ext
                             @BindBean final InternalTenantContext context);
 
     @SqlQuery
-    public Long getFoundRows(@BindBean final InternalTenantContext context);
+    @SmartFetchSize(shouldStream = true)
+    public Iterator<M> search(@Bind("searchKey") final String searchKey,
+                              @Bind("likeSearchKey") final String likeSearchKey,
+                              @Bind("offset") final Long offset,
+                              @Bind("rowCount") final Long rowCount,
+                              @BindBean final InternalTenantContext context);
 
     @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
+    public Long getSearchCount(@BindBean final InternalTenantContext context);
+
+    @SqlQuery
+    @SmartFetchSize(shouldStream = true)
     public Iterator<M> getAll(@BindBean final InternalTenantContext context);
 
     @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
+    @SmartFetchSize(shouldStream = true)
     public Iterator<M> get(@Bind("offset") final Long offset,
                            @Bind("rowCount") final Long rowCount,
-                           @Bind("orderBy") final String orderBy,
+                           @Define("orderBy") final String orderBy,
                            @BindBean final InternalTenantContext context);
 
     @SqlQuery

diff --git a/util/src/main/java/com/ning/billing/util/tag/dao/DefaultTagDao.java b/util/src/main/java/com/ning/billing/util/tag/dao/DefaultTagDao.java
index 4a4abaa..7f9a1ca 100644
--- a/util/src/main/java/com/ning/billing/util/tag/dao/DefaultTagDao.java
+++ b/util/src/main/java/com/ning/billing/util/tag/dao/DefaultTagDao.java
@@ -223,7 +223,7 @@ public class DefaultTagDao extends EntityDaoBase<TagModelDao, Tag, TagApiExcepti
                                               new PaginationIteratorBuilder<TagModelDao, Tag, TagSqlDao>() {
                                                   @Override
                                                   public Iterator<TagModelDao> build(final TagSqlDao tagSqlDao, final Long limit) {
-                                                      return tagSqlDao.searchTags(searchKey, String.format("%%%s%%", searchKey), offset, limit, context);
+                                                      return tagSqlDao.search(searchKey, String.format("%%%s%%", searchKey), offset, limit, context);
                                                   }
                                               },
                                               offset,

diff --git a/util/src/main/java/com/ning/billing/util/tag/dao/TagSqlDao.java b/util/src/main/java/com/ning/billing/util/tag/dao/TagSqlDao.java
index 668553b..26caab0 100644
--- a/util/src/main/java/com/ning/billing/util/tag/dao/TagSqlDao.java
+++ b/util/src/main/java/com/ning/billing/util/tag/dao/TagSqlDao.java
@@ -16,7 +16,6 @@
 
 package com.ning.billing.util.tag.dao;
 
-import java.util.Iterator;
 import java.util.List;
 import java.util.UUID;
 
@@ -24,7 +23,6 @@ import org.skife.jdbi.v2.sqlobject.Bind;
 import org.skife.jdbi.v2.sqlobject.BindBean;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
 import org.skife.jdbi.v2.sqlobject.SqlUpdate;
-import org.skife.jdbi.v2.sqlobject.customizers.FetchSize;
 
 import com.ning.billing.ObjectType;
 import com.ning.billing.callcontext.InternalCallContext;
@@ -52,14 +50,4 @@ public interface TagSqlDao extends EntitySqlDao<TagModelDao, Tag> {
     List<TagModelDao> getTagsForObjectIncludedDeleted(@Bind("objectId") UUID objectId,
                                                       @Bind("objectType") ObjectType objectType,
                                                       @BindBean InternalTenantContext internalTenantContext);
-
-    @SqlQuery
-    // Magic value to force MySQL to stream from the database
-    // See http://dev.mysql.com/doc/refman/5.0/en/connector-j-reference-implementation-notes.html (ResultSet)
-    @FetchSize(Integer.MIN_VALUE)
-    public Iterator<TagModelDao> searchTags(@Bind("searchKey") final String searchKey,
-                                            @Bind("likeSearchKey") final String likeSearchKey,
-                                            @Bind("offset") final Long offset,
-                                            @Bind("rowCount") final Long rowCount,
-                                            @BindBean final InternalTenantContext context);
 }

diff --git a/util/src/main/resources/com/ning/billing/util/customfield/dao/CustomFieldSqlDao.sql.stg b/util/src/main/resources/com/ning/billing/util/customfield/dao/CustomFieldSqlDao.sql.stg
index 1d01add..fcdf9f5 100644
--- a/util/src/main/resources/com/ning/billing/util/customfield/dao/CustomFieldSqlDao.sql.stg
+++ b/util/src/main/resources/com/ning/billing/util/customfield/dao/CustomFieldSqlDao.sql.stg
@@ -52,19 +52,9 @@ and is_active
 ;
 >>
 
-searchCustomFields() ::= <<
-select SQL_CALC_FOUND_ROWS
-<allTableFields("t.")>
-from <tableName()> t
-where 1 = 1
-and (
-     <idField("t.")> = :searchKey
-  or t.object_type like :likeSearchKey
-  or t.field_name like :likeSearchKey
-  or t.field_value like :likeSearchKey
-)
-<AND_CHECK_TENANT("t.")>
-order by <recordIdField("t.")> ASC
-limit :offset, :rowCount
-;
+searchQuery(prefix) ::= <<
+     <idField(prefix)> = :searchKey
+  or <prefix>.object_type like :likeSearchKey
+  or <prefix>.field_name like :likeSearchKey
+  or <prefix>.field_value like :likeSearchKey
 >>

diff --git a/util/src/main/resources/com/ning/billing/util/entity/dao/EntitySqlDao.sql.stg b/util/src/main/resources/com/ning/billing/util/entity/dao/EntitySqlDao.sql.stg
index 6214825..55a801f 100644
--- a/util/src/main/resources/com/ning/billing/util/entity/dao/EntitySqlDao.sql.stg
+++ b/util/src/main/resources/com/ning/billing/util/entity/dao/EntitySqlDao.sql.stg
@@ -137,10 +137,6 @@ allHistoryTableValues() ::= <<
 CHECK_TENANT(prefix) ::= "<prefix>tenant_record_id = :tenantRecordId"
 AND_CHECK_TENANT(prefix) ::= "and <CHECK_TENANT(prefix)>"
 
-getFoundRows() ::= <<
-select FOUND_ROWS();
->>
-
 getAll() ::= <<
 select
 <allTableFields("t.")>
@@ -152,12 +148,12 @@ where <CHECK_TENANT("t.")>
 >>
 
 get(offset, rowCount, orderBy) ::= <<
-select SQL_CALC_FOUND_ROWS
+select
 <allTableFields("t.")>
 from <tableName()> t
 where <CHECK_TENANT("t.")>
 <andCheckSoftDeletionWithComma("t.")>
-order by :orderBy
+order by t.<orderBy>
 limit :offset, :rowCount
 ;
 >>
@@ -223,7 +219,6 @@ where <recordIdField("t.")> = :recordId
 ;
 >>
 
-
 getRecordId(id) ::= <<
 select
   <recordIdField("t.")>
@@ -254,7 +249,6 @@ where <targetRecordIdField("t.")> = :targetRecordId
 ;
 >>
 
-
 getHistoryRecordIdsForTable(historyTableName) ::= <<
 select
   <recordIdField("t.")>
@@ -265,6 +259,29 @@ where <targetRecordIdField("t.")> = :targetRecordId
 ;
 >>
 
+searchQuery(prefix) ::= <<
+1 = 1
+>>
+
+search() ::= <<
+select
+<allTableFields("t.")>
+from <tableName()> t
+where <searchQuery("t.")>
+<AND_CHECK_TENANT("t.")>
+order by <recordIdField("t.")> ASC
+limit :offset, :rowCount
+;
+>>
+
+getSearchCount() ::= <<
+select
+  count(1) as count
+from <tableName()> t
+where <searchQuery("t.")>
+<AND_CHECK_TENANT("t.")>
+;
+>>
 
 create() ::= <<
 insert into <tableName()> (

diff --git a/util/src/test/java/com/ning/billing/util/dao/TestPagination.java b/util/src/test/java/com/ning/billing/util/dao/TestPagination.java
index 01b7b8a..69dcc4d 100644
--- a/util/src/test/java/com/ning/billing/util/dao/TestPagination.java
+++ b/util/src/test/java/com/ning/billing/util/dao/TestPagination.java
@@ -44,12 +44,12 @@ public class TestPagination extends UtilTestSuiteWithEmbeddedDB {
 
         // Tests via SQL dao directly
         Assert.assertEquals(ImmutableList.<TagDefinitionModelDao>copyOf(tagDefinitionSqlDao.getAll(internalCallContext)).size(), 10);
-        Assert.assertEquals(ImmutableList.<TagDefinitionModelDao>copyOf(tagDefinitionSqlDao.get(0L, 100L, "recordId", internalCallContext)).size(), 10);
-        Assert.assertEquals(ImmutableList.<TagDefinitionModelDao>copyOf(tagDefinitionSqlDao.get(5L, 100L, "recordId", internalCallContext)).size(), 5);
-        Assert.assertEquals(ImmutableList.<TagDefinitionModelDao>copyOf(tagDefinitionSqlDao.get(5L, 10L, "recordId", internalCallContext)).size(), 5);
-        Assert.assertEquals(ImmutableList.<TagDefinitionModelDao>copyOf(tagDefinitionSqlDao.get(0L, 5L, "recordId", internalCallContext)).size(), 5);
+        Assert.assertEquals(ImmutableList.<TagDefinitionModelDao>copyOf(tagDefinitionSqlDao.get(0L, 100L, "record_id", internalCallContext)).size(), 10);
+        Assert.assertEquals(ImmutableList.<TagDefinitionModelDao>copyOf(tagDefinitionSqlDao.get(5L, 100L, "record_id", internalCallContext)).size(), 5);
+        Assert.assertEquals(ImmutableList.<TagDefinitionModelDao>copyOf(tagDefinitionSqlDao.get(5L, 10L, "record_id", internalCallContext)).size(), 5);
+        Assert.assertEquals(ImmutableList.<TagDefinitionModelDao>copyOf(tagDefinitionSqlDao.get(0L, 5L, "record_id", internalCallContext)).size(), 5);
         for (int i = 0; i < 10; i++) {
-            final List<TagDefinitionModelDao> tagDefinitions = ImmutableList.<TagDefinitionModelDao>copyOf(tagDefinitionSqlDao.get(0L, (long) i, "recordId", internalCallContext));
+            final List<TagDefinitionModelDao> tagDefinitions = ImmutableList.<TagDefinitionModelDao>copyOf(tagDefinitionSqlDao.get(0L, (long) i, "record_id", internalCallContext));
             Assert.assertEquals(tagDefinitions.size(), i);
 
             for (int j = 0; j < tagDefinitions.size(); j++) {

diff --git a/util/src/test/java/com/ning/billing/util/dao/TestStringTemplateInheritance.java b/util/src/test/java/com/ning/billing/util/dao/TestStringTemplateInheritance.java
index df148e1..7acbe91 100644
--- a/util/src/test/java/com/ning/billing/util/dao/TestStringTemplateInheritance.java
+++ b/util/src/test/java/com/ning/billing/util/dao/TestStringTemplateInheritance.java
@@ -114,7 +114,7 @@ public class TestStringTemplateInheritance extends UtilTestSuiteNoDB {
                                                                          "where t.tenant_record_id = :tenantRecordId\n" +
                                                                          "order by t.record_id ASC\n" +
                                                                          ";");
-        Assert.assertEquals(kombucha.getInstanceOf("get", ImmutableMap.<String, String>of("orderBy", "recordId", "offset", "3", "rowCount", "12")).toString(), "select SQL_CALC_FOUND_ROWS\n" +
+        Assert.assertEquals(kombucha.getInstanceOf("get", ImmutableMap.<String, String>of("orderBy", "record_id", "offset", "3", "rowCount", "12")).toString(), "select\n" +
                                                                                                                                                                "  t.record_id\n" +
                                                                                                                                                                ", t.id\n" +
                                                                                                                                                                ", t.tea\n" +
@@ -124,7 +124,7 @@ public class TestStringTemplateInheritance extends UtilTestSuiteNoDB {
                                                                                                                                                                ", t.tenant_record_id\n" +
                                                                                                                                                                "from kombucha t\n" +
                                                                                                                                                                "where t.tenant_record_id = :tenantRecordId\n" +
-                                                                                                                                                               "order by :orderBy\n" +
+                                                                                                                                                               "order by t.record_id\n" +
                                                                                                                                                                "limit :offset, :rowCount\n" +
                                                                                                                                                                ";");
         Assert.assertEquals(kombucha.getInstanceOf("test").toString(), "select\n" +