Fixed issues with Aggregated Contribution data source

CHANGELOG.md

 # Version 1.118 (not yet released)
 
+* Fixed issues with Aggregated Contribution data source
+
 # Version 1.117
 
 * Fixed issue with clicking export button

Civi/DataProcessor/Source/AbstractCivicrmEntitySource.php

@@ -23,7 +23,6 @@ use Civi\DataProcessor\DataSpecification\FieldExistsException;
 use Civi\DataProcessor\DataSpecification\FieldSpecification;
 use Civi\DataProcessor\DataSpecification\Utils as DataSpecificationUtils;
 use Civi\DataProcessor\ProcessorType\AbstractProcessorType;
-
 use CRM_Dataprocessor_ExtensionUtil as E;
 use CRM_Dataprocessor_Utils_Date;
 
@@ -178,10 +177,19 @@ abstract class AbstractCivicrmEntitySource extends AbstractSource {
     if ($this->isAggregationEnabled()) {
       $this->getAggregationDataFlow();
 
-      $dataFlow = new CombinedSqlDataFlow('', $this->aggregationDateFlow->getPrimaryTable(), $this->aggregationDateFlow->getPrimaryTableAlias());
+      if ($this->aggregationDateFlow instanceof SubqueryDataFlow) {
+        $primaryTable = $this->aggregationDateFlow->getPrimaryTable();
+        $primaryTableAlias = $this->aggregationDateFlow->getPrimaryTableAlias();
+      } elseif ($this->aggregationDateFlow instanceof SqlTableDataFlow) {
+        $primaryTable = $this->aggregationDateFlow->getTable();
+        $primaryTableAlias = $this->aggregationDateFlow->getTableAlias();
+      }
+      $dataFlow = new CombinedSqlDataFlow('', $primaryTable, $primaryTableAlias);
       $dataFlow->addSourceDataFlow(new DataFlowDescription($this->aggregationDateFlow));
-      $dataFlowDescription = new DataFlowDescription($this->entityDataFlow, $this->getAggregationJoin($this->getEntityTableAlias()));
-      $dataFlow->addSourceDataFlow($dataFlowDescription);
+      if ($this->isAggregationJoinable()) {
+        $dataFlowDescription = new DataFlowDescription($this->entityDataFlow, $this->getAggregationJoin($this->getEntityTableAlias()));
+        $dataFlow->addSourceDataFlow($dataFlowDescription);
+      }
       return $dataFlow;
     } else {
       return $this->entityDataFlow;
@@ -200,8 +208,8 @@ abstract class AbstractCivicrmEntitySource extends AbstractSource {
       $aggrgeate_field_spec = clone $this->getAvailableFields()->getFieldSpecificationByName($this->getAggregateField());
       $aggrgeate_field_spec->setMySqlFunction($this->getAggregateFunction());
       $aggrgeate_field_spec->alias = $this->getAggregateField();
-
-      $this->aggretated_table_dataflow = new SqlTableDataFlow($this->getTable(), '_aggregated_'.$this->getSourceName());
+      $tableAlias = '_aggregated_' . $this->getSourceName();
+      $this->aggretated_table_dataflow = new SqlTableDataFlow($this->getTable(), $tableAlias);
       $this->aggretated_table_dataflow->getDataSpecification()->addFieldSpecification($aggrgeate_field_spec->name, $aggrgeate_field_spec);
       foreach ($groupByFields as $groupByField) {
         $this->aggretated_table_dataflow->getDataSpecification()
@@ -209,8 +217,7 @@ abstract class AbstractCivicrmEntitySource extends AbstractSource {
         $this->aggretated_table_dataflow->getGroupByDataSpecification()
           ->addFieldSpecification($groupByField->name, $groupByField);
       }
-
-      $this->aggregationDateFlow = new SubqueryDataFlow('_aggregated_'.$this->getSourceName(), $this->getTable(), '_aggregated_' . $this->getSourceName());
+      $this->aggregationDateFlow = new SubqueryDataFlow($tableAlias, $this->getTable(), $tableAlias);
       $this->aggregationDateFlow->addSourceDataFlow(new DataFlowDescription($this->aggretated_table_dataflow));
     }
 
@@ -241,6 +248,15 @@ abstract class AbstractCivicrmEntitySource extends AbstractSource {
     return $join;
   }
 
+  protected function isAggregationJoinable(): bool {
+    switch ($this->getAggregateFunction()) {
+      case 'MAX':
+      case 'MIN':
+        return TRUE;
+    }
+    return FALSE;
+  }
+
   /**
    * Load the fields from this entity.
    *
@@ -477,13 +493,19 @@ abstract class AbstractCivicrmEntitySource extends AbstractSource {
     }
 
     if ($this->isAggregationEnabled()) {
+      if ($this->aggregationDateFlow instanceof SubqueryDataFlow) {
+        $primaryTableAlias = $this->aggregationDateFlow->getPrimaryTableAlias();
+      } elseif ($this->aggregationDateFlow instanceof SqlTableDataFlow) {
+        $primaryTableAlias = $this->aggregationDateFlow->getTableAlias();
+      }
+
       if ($join instanceof SimpleJoin && $join->getLeftTable() == $this->getSourceName()) {
-        $join->setLeftTable($this->aggregationDateFlow->getPrimaryTableAlias());
-        $join->setLeftPrefix($this->aggregationDateFlow->getPrimaryTableAlias());
+        $join->setLeftTable($primaryTableAlias);
+        $join->setLeftPrefix($primaryTableAlias);
       }
       elseif ($join instanceof SimpleJoin && $join->getRightTable() == $this->getSourceName()) {
-        $join->setRightTable($this->aggregationDateFlow->getPrimaryTableAlias());
-        $join->setRightPrefix($this->aggregationDateFlow->getPrimaryTableAlias());
+        $join->setRightTable($primaryTableAlias);
+        $join->setRightPrefix($primaryTableAlias);
       }
     }
     return $this;
@@ -496,8 +518,10 @@ abstract class AbstractCivicrmEntitySource extends AbstractSource {
   public function getAvailableFields() {
     if (!$this->availableFields) {
       $this->availableFields = new DataSpecification();
-      $this->loadFields($this->availableFields, array());
-      $this->loadCustomGroupsAndFields($this->availableFields, false);
+      if (!$this->isAggregationEnabled() || $this->isAggregationJoinable()) {
+        $this->loadFields($this->availableFields, []);
+        $this->loadCustomGroupsAndFields($this->availableFields, FALSE);
+      }
     }
     return $this->availableFields;
   }
@@ -535,7 +559,7 @@ abstract class AbstractCivicrmEntitySource extends AbstractSource {
         if (!$dataFlow->getDataSpecification()->doesAliasExists($fieldSpecification->alias)) {
           $dataFlow->getDataSpecification()->addFieldSpecification($fieldSpecification->alias, $fieldSpecification);
         }
-      } elseif ($originalFieldSpecification) {
+      } elseif ($originalFieldSpecification && (!$this->isAggregationEnabled() || $this->getAggregateField() != $originalFieldSpecification->name)) {
         $this->ensureEntity();
         $this->entityDataFlow->getDataSpecification()->addFieldSpecification($fieldSpecification->alias, $fieldSpecification);
       }

Civi/DataProcessor/Source/Contribution/AggregatedContributionSource.php

@@ -78,12 +78,20 @@ class AggregatedContributionSource extends AbstractCivicrmEntitySource {
   public function getAvailableFields() {
     if (!$this->availableFields) {
       parent::getAvailableFields();
-      $id = new FieldSpecification('id', 'Integer', E::ts('Contribution ID'), null, $this->getSourceName().'_id');
-      $this->availableFields->addFieldSpecification('id', $id);
-      $totalAmount = new FieldSpecification('total_amount', 'Float', E::ts('Total amount'), null, $this->getSourceName().'_total_amount');
-      $this->availableFields->addFieldSpecification('total_amount', $totalAmount);
-      $receiveDate = new FieldSpecification('receive_date', 'Date', E::ts('Receive Date'), null, $this->getSourceName().'_receive_date');
-      $this->availableFields->addFieldSpecification('receive_date', $receiveDate);
+      switch ($this->getAggregateField()) {
+        case 'id':
+          $id = new FieldSpecification('id', 'Integer', E::ts('Contribution ID'), NULL, $this->getSourceName() . '_id');
+          $this->availableFields->addFieldSpecification('id', $id);
+          break;
+        case 'total_amount':
+          $totalAmount = new FieldSpecification('total_amount', 'Float', E::ts('Total amount'), NULL, $this->getSourceName() . '_total_amount');
+          $this->availableFields->addFieldSpecification('total_amount', $totalAmount);
+          break;
+        case 'receive_date':
+          $receiveDate = new FieldSpecification('receive_date', 'Date', E::ts('Receive Date'), NULL, $this->getSourceName() . '_receive_date');
+          $this->availableFields->addFieldSpecification('receive_date', $receiveDate);
+          break;
+      }
       $contactId = new FieldSpecification('contact_id', 'Float', E::ts('Contact ID'), null, $this->getSourceName().'_contact_id');
       $this->availableFields->addFieldSpecification('contact_id', $contactId);
       $count = new FieldSpecification('count', 'Integer', E::ts('Count'), null, $this->getSourceName().'_count');
@@ -132,10 +140,10 @@ class AggregatedContributionSource extends AbstractCivicrmEntitySource {
         } elseif ($this->getAvailableFields()->doesFieldExist($fieldSpecification->name)) {
           $originalFieldSpecification = $this->getAvailableFields()->getFieldSpecificationByName($fieldSpecification->name);
         }
-        if ($originalFieldSpecification) {
+        if ($originalFieldSpecification && $this->getAggregateField() == $originalFieldSpecification->name) {
           $originalFieldSpecification = clone $originalFieldSpecification;
-          $originalFieldSpecification->alias = $originalFieldSpecification->name;
-          $this->aggretated_table_dataflow->getDataSpecification()->addFieldSpecification($originalFieldSpecification->alias, $originalFieldSpecification);
+          $originalFieldSpecification->alias = $fieldSpecification->alias;
+          $this->aggregationDateFlow->getDataSpecification()->addFieldSpecification($fieldSpecification->alias, $originalFieldSpecification);
         }
       } catch (FieldExistsException $e) {
         // Do nothing.