Workaround for exceeding governor limits using COUNT_DISTINCT in Batch Apex Class?

Question

So I have the following query that I am running through Batch Apex. I get the error:

System.LimitException: Too many query rows: 50001

I am receiving this because COUNT_DISTINCT() is subject to the same governor limits as other SOQL Queries. Basically I cannot think of a solution here, which is why I am posting this. The goal of the query is to see how many unique accounts an SDR is in contact with over a given time period. In this case, it is over a 120 day period. Would love some advice on how to achieve this.

String Contacts_Reached_TR_120_Query = 'SELECT Ownerid SDR_ID, COUNT_DISTINCT( whoid ) result FROM Task Where ActivityDate = LAST_N_DAYS:120 AND OwnerId in :ListOfIDs GROUP BY Ownerid';

DynamicBatchApexAR batch3_TR120      =   new DynamicBatchApexAR('SDR_ID', 'result', 'Contacts_Reached_TR_120__c', Contacts_Reached_TR_120_Query, SDR_IDs);

Then here is the batch class

global class DynamicBatchApexAR implements Database.Batchable<AggregateResult> {

public String sObjectIdKey;
public String sObjectResultKey;
public String sObjectFieldToUpdate;
public String query;
public List<Id> ListOfIDs;


global DynamicBatchApexAR(String arIdKey, String arResultKey, String fieldToUpdate, String soqlQuery, List<Id> whereInClause ) 
{


    sObjectIdKey            = arIdKey;
    sObjectResultKey        = arResultKey;
    sObjectFieldToUpdate    = fieldToUpdate;
    query                   = soqlQuery;
    ListOfIDs               = whereInClause;
}

global DynamicBatchApexAR(String arIdKey, String arResultKey, String fieldToUpdate, String soqlQuery ) 
{


    sObjectIdKey            = arIdKey;
    sObjectResultKey        = arResultKey;
    sObjectFieldToUpdate    = fieldToUpdate;
    query                   = soqlQuery;
}

global Iterable<AggregateResult> start(Database.BatchableContext BC) 
{
    return ListOfIDs == null ? new BulkIterable(query) : new BulkIterable(query, ListOfIDs);
}

global void execute(Database.BatchableContext BC, List<sObject> scope) 
{
    DynamicSObjectUpdater sObjectUpdater = new DynamicSObjectUpdater();

    for(sObject sObj: scope)
    {

        AggregateResult ar      = (AggregateResult)sObj;
        ID sObjectId            = (ID)ar.get(sObjectIdKey);
        Decimal arResult        = (Decimal)ar.get(sObjectResultKey);
        Map<String, Object> sObjectFieldsToUpdate = new Map<String, Object>{
            sObjectFieldToUpdate => arResult
        };

        sObjectUpdater.getUpdateSObject(sObjectId, sObjectFieldsToUpdate);
    }

    sObjectUpdater.updateSObjects();


}
global void finish(Database.BatchableContext BC) 
{

}

}

Here is the iterable

  global with sharing class BulkIterable implements Iterable<AggregateResult> {
        String query;
        List<Id> 

ListOfIDs;

    global BulkIterable(String soql, List<Id> WhereInClause)
    {
        query = soql;
        ListOfIDs = WhereInClause;
    }

    global BulkIterable(String soql)
    {
        query = soql;
    }

    global Iterator<AggregateResult> Iterator(){
        return ListOfIDs == null ? new BulkIterator(query) :  new BulkIterator(query, ListOfIDs);
    } 
}

Here is the iterator

    global class BulkIterator implements Iterator<AggregateResult> {

      AggregateResult[] results { get;set; }
      Integer index { get;set; }


List<Id> ListOfIDs;

    global BulkIterator(String query, List<Id> WhereInClause) 
    {
    index = 0;
    ListOfIDs = WhereInClause;
    results = Database.query(query);   
    }

  global BulkIterator(String query) {
    index = 0;
    results = Database.query(query);   
  }

  global Boolean hasNext(){ 
   return results != null && !results.isEmpty() && index < results.size(); 
  }    

  global AggregateResult next() { 
    return results[index++];
  }    

}

Is the error because I am running too many batches?

Answer 1

While this may not answer the question fully, the following points should be noted (several based on details from the governor limits documentation):

1. You are restricted to 5 concurrently executing batches per org, with up to a further 100 batches held in the flex queue. Having the number of batches you spawn from this code is potentially problematic, especially if this gets executed too frequently (so batch queuing overlaps) or there are other batches being used on the org
2. The use of Iterable with a batch means that, when the batch is iterating SObjects (directly or via Aggregate Results), the entire batch execution is limited by the number of rows queried per transaction, being 50000 rows across all SOQL queries in the transaction, rather than the number of rows permitted across Database.QueryLocator handling in a batch, which is 50 million rows.
3. As per the SOQL documentation, aggregate functions other than COUNT, consume rows queried not by results but by the rows containing data contributing to the aggregate result. So if you use COUNT_DISTINCT against a query that scans 100 records and those records have 3 distinct, non-null values, there will be 3 returned aggregate results but the query will have consumed 100 rows from the row limits.

Because this code uses COUNT_DISTINCT and there are no limits applied (limiting aggregate functions only works with GROUP BY and you probably would not get the results you want), it is likely the workaround is to convert these two queries into Database.QueryLocator based batches that hold their own state (implementing Database.Stateful) and perform their own aggregate counting. The Batchable documentation gives a hint about this. Essentially you would want a:

private Map<Id, Map<Id, Integer>> distinctCounts = new Map<Id, Map<Id, Integer>>();

where the key is the "owner ID" and the value is a map of "who ID" to the count. The map would be initialized empty. The query would be much as you defined, except without the COUNT_DISTINCT and GROUP BY, something like:

SELECT OwnerId, WhoId FROM Task Where ActivityDate = LAST_N_DAYS:120 AND OwnerId in :listOfIDs

The execute method would take the data and do something like:

for (Task task : scope) {
    Map<Id, Integer> countsByWhoId = distinctCounts.get(task.OwnerId);

    if (countsByWhoId == null) {
        distinctCounts.put(task.OwnerId, new Map<Id, Integer> {
            task.WhoId => 1
        });
    } else {
        Integer countForWhoId = countsByWhoId.get(task.WhoId);

        if (countForWhoId == null) {
            countsByWhoId.put(task.WhoId, 1);
        } else {
            countsByWhoId.put(task.WhoId, countForWhoId + 1);
        }
    }
}

The last piece is to finally process these WhoId by OwnerId counts in the finish method (not shown here, but you can just iterate the map keySets and get the values to process).

Some ancillary points of note:

• Global should only be used as an access modifier if absolutely essential, as per the documentation.