protected void addSubKeysAndAggregator(Set<String> values,Set<String> keysOfCompositeAggregator,Set<String> subKeys,String aggregatorName,Map<Set<String>,Integer> keysToCombinationId){
  if (keysOfCompositeAggregator == null || subKeys == null || keysOfCompositeAggregator.isEmpty() || subKeys.isEmpty()) {
    throw new IllegalArgumentException("Both keys and subKeys can't be null or empty");
  }
  Set<String> allKeys=Sets.newHashSet();
  allKeys.addAll(keysOfCompositeAggregator);
  allKeys.addAll(subKeys);
  if (allKeys.size() != keysOfCompositeAggregator.size() + subKeys.size()) {
    throw new IllegalArgumentException("Should NOT have overlap between keys " + keysOfCompositeAggregator.toString() + " and subKeys "+ subKeys);
  }
  Integer combinationId=keysToCombinationId.get(allKeys);
  if (combinationId == null) {
    if (dimensionsDescriptorIDToKeys.add(new Fields(allKeys))) {
      combinationId=dimensionsDescriptorIDToKeys.size() - 1;
    }
 else {
      throw new RuntimeException("The keys " + allKeys + " already existed.");
    }
    keysToCombinationId.put(allKeys,combinationId);
    addValueToAggregatorToCombination(values,allKeys,aggregatorName);
  }
 else {
    Set<String> incrementalAggregatorNames;
    boolean isOTFAggregator=false;
    if (!isIncrementalAggregator(aggregatorName)) {
      incrementalAggregatorNames=getOTFDependedIncrementalAggregatorNames(aggregatorName);
      isOTFAggregator=true;
    }
 else {
      incrementalAggregatorNames=Sets.newHashSet();
      incrementalAggregatorNames.add(aggregatorName);
    }
    Map<String,Set<String>> newValueToIncrementalAggregators=Maps.newHashMap();
    Map<String,Set<String>> newValueToOTFAggregators=Maps.newHashMap();
    for (    String value : values) {
      newValueToIncrementalAggregators.put(value,incrementalAggregatorNames);
      if (isOTFAggregator) {
        newValueToOTFAggregators.put(value,Sets.newHashSet(aggregatorName));
      }
    }
    int ddid=combinationId * customTimeBucketsCombination.size();
    for (int index=0; index < customTimeBucketsCombination.size(); ++index, ++ddid) {
      mergeMaps(dimensionsDescriptorIDToValueToAggregator.get(ddid),newValueToIncrementalAggregators);
      mergeMaps(dimensionsDescriptorIDToValueToOTFAggregator.get(ddid),newValueToOTFAggregators);
    }
  }
}

protected void addSubKeysAndAggregator ( Set < String > values , Set < String > keysOfCompositeAggregator , Set < String > subKeys , String aggregatorName , Map < Set < String > , Integer > keysToCombinationId ) { if ( keysOfCompositeAggregator == null || subKeys == null || keysOfCompositeAggregator . isEmpty ( ) || subKeys . isEmpty ( ) ) { throw new IllegalArgumentException ( "Both keys and subKeys can't be null or empty" ) ; } Set < String > allKeys = Sets . newHashSet ( ) ; allKeys . addAll ( keysOfCompositeAggregator ) ; allKeys . addAll ( subKeys ) ; if ( allKeys . size ( ) != keysOfCompositeAggregator . size ( ) + subKeys . size ( ) ) { throw new IllegalArgumentException ( "Should NOT have overlap between keys " + keysOfCompositeAggregator . toString ( ) + " and subKeys " + subKeys ) ; } Integer combinationId = keysToCombinationId . get ( allKeys ) ; if ( combinationId == null ) { if ( dimensionsDescriptorIDToKeys . add ( new Fields ( allKeys ) ) ) { combinationId = dimensionsDescriptorIDToKeys . size ( ) - 1 ; } else { throw new RuntimeException ( "The keys " + allKeys + " already existed." ) ; } keysToCombinationId . put ( allKeys , combinationId ) ; addValueToAggregatorToCombination ( values , allKeys , aggregatorName ) ; } else { Set < String > incrementalAggregatorNames ; boolean isOTFAggregator = false ; if ( ! isIncrementalAggregator ( aggregatorName ) ) { incrementalAggregatorNames = getOTFDependedIncrementalAggregatorNames ( aggregatorName ) ; isOTFAggregator = true ; } else { incrementalAggregatorNames = Sets . newHashSet ( ) ; incrementalAggregatorNames . add ( aggregatorName ) ; } Map < String , Set < String > > newValueToIncrementalAggregators = Maps . newHashMap ( ) ; Map < String , Set < String > > newValueToOTFAggregators = Maps . newHashMap ( ) ; for ( String value : values ) { newValueToIncrementalAggregators . put ( value , incrementalAggregatorNames ) ; if ( isOTFAggregator ) { newValueToOTFAggregators . put ( value , Sets . newHashSet ( aggregatorName ) ) ; } } int ddid = combinationId * customTimeBucketsCombination . size ( ) ; for ( int index = 0 ; index < customTimeBucketsCombination . size ( ) ; ++ index , ++ ddid ) { mergeMaps ( dimensionsDescriptorIDToValueToAggregator . get ( ddid ) , newValueToIncrementalAggregators ) ; mergeMaps ( dimensionsDescriptorIDToValueToOTFAggregator . get ( ddid ) , newValueToOTFAggregators ) ; } } }
add sub - keys and aggregtor to the dimension description . add the aggregator to the dimension if it already existed , else add new dimension precondition : neither keys nor subkeys are empty . keys empty means all keys , composite aggregator should not apply to this case notes : keep the data integration
