ExternalGroupOperatorDescriptor extends AbstractOperatorDescriptor { public static int calculateGroupByTableCardinality(long memoryBudgetByteSize, int numberOfGroupByColumns, int frameSize) { int tupleByteSize = 4 + 8 * numberOfGroupByColumns; long maxNumberOfTuplesInDataTable = memoryBudgetByteSize / tupleByteSize; int numberOfBits = Math.min(61, numberOfGroupByColumns * 4 * 8); long possibleNumberOfHashEntries = 2L << numberOfBits; long groupByTableCardinality = Math.min(possibleNumberOfHashEntries, maxNumberOfTuplesInDataTable); long groupByTableByteSize = SerializableHashTable.getExpectedTableByteSize(groupByTableCardinality, frameSize); double hashTableRatio = (double) groupByTableByteSize / (groupByTableByteSize + memoryBudgetByteSize); long finalGroupByTableByteSize = (long) (hashTableRatio * memoryBudgetByteSize); long finalGroupByTableCardinality = finalGroupByTableByteSize / SerializableHashTable.getExpectedByteSizePerHashValue(); return finalGroupByTableCardinality > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) finalGroupByTableCardinality; } ExternalGroupOperatorDescriptor(IOperatorDescriptorRegistry spec, int inputSizeInTuple, long inputFileSize,
            int[] keyFields, int framesLimit, IBinaryComparatorFactory[] comparatorFactories,
            INormalizedKeyComputerFactory firstNormalizerFactory, IAggregatorDescriptorFactory partialAggregatorFactory,
            IAggregatorDescriptorFactory intermediateAggregateFactory, RecordDescriptor partialAggRecordDesc,
            RecordDescriptor outRecordDesc, ISpillableTableFactory spillableTableFactory); @Override void contributeActivities(IActivityGraphBuilder builder); static int calculateGroupByTableCardinality(long memoryBudgetByteSize, int numberOfGroupByColumns,
            int frameSize);  }
@Test public void testCalculateGroupByTableCardinality() throws Exception { IOperatorDescriptorRegistry spec = new JobSpecification(32768); ExternalGroupOperatorDescriptor eGByOp = new ExternalGroupOperatorDescriptor(spec, 0, 0, null, 4, null, null, null, null, null, null, null); long memoryBudgetInBytes = 512; int numberOfGroupByColumns = 1; int frameSize = 256; int resultCardinality = (int) PA.invokeMethod(eGByOp, "calculateGroupByTableCardinality(long,int,int)", memoryBudgetInBytes, numberOfGroupByColumns, frameSize); Assert.assertTrue(resultCardinality == 10); frameSize = 128 * 1024; memoryBudgetInBytes = 1024 * 1024; resultCardinality = (int) PA.invokeMethod(eGByOp, "calculateGroupByTableCardinality(long,int,int)", memoryBudgetInBytes, numberOfGroupByColumns, frameSize); Assert.assertTrue(resultCardinality == 20388); memoryBudgetInBytes = 1024 * 1024 * 100; resultCardinality = (int) PA.invokeMethod(eGByOp, "calculateGroupByTableCardinality(long,int,int)", memoryBudgetInBytes, numberOfGroupByColumns, frameSize); Assert.assertTrue(resultCardinality == 2016724); memoryBudgetInBytes = 1024 * 1024 * 1024; resultCardinality = (int) PA.invokeMethod(eGByOp, "calculateGroupByTableCardinality(long,int,int)", memoryBudgetInBytes, numberOfGroupByColumns, frameSize); Assert.assertTrue(resultCardinality == 20649113); memoryBudgetInBytes = 1024 * 1024 * 1024 * 10L; resultCardinality = (int) PA.invokeMethod(eGByOp, "calculateGroupByTableCardinality(long,int,int)", memoryBudgetInBytes, numberOfGroupByColumns, frameSize); Assert.assertTrue(resultCardinality == 206489044); memoryBudgetInBytes = 1024 * 1024 * 1024 * 100L; resultCardinality = (int) PA.invokeMethod(eGByOp, "calculateGroupByTableCardinality(long,int,int)", memoryBudgetInBytes, numberOfGroupByColumns, frameSize); Assert.assertTrue(resultCardinality == 2045222521); memoryBudgetInBytes = 1024 * 1024 * 1024 * 1024L; frameSize = 128 * 1024; resultCardinality = (int) PA.invokeMethod(eGByOp, "calculateGroupByTableCardinality(long,int,int)", memoryBudgetInBytes, numberOfGroupByColumns, frameSize); Assert.assertTrue(resultCardinality == 2147483647); memoryBudgetInBytes = 1024 * 1024; numberOfGroupByColumns = 2; resultCardinality = (int) PA.invokeMethod(eGByOp, "calculateGroupByTableCardinality(long,int,int)", memoryBudgetInBytes, numberOfGroupByColumns, frameSize); Assert.assertTrue(resultCardinality == 17825); memoryBudgetInBytes = 1024 * 1024; numberOfGroupByColumns = 3; resultCardinality = (int) PA.invokeMethod(eGByOp, "calculateGroupByTableCardinality(long,int,int)", memoryBudgetInBytes, numberOfGroupByColumns, frameSize); Assert.assertTrue(resultCardinality == 16227); memoryBudgetInBytes = 1024 * 1024; numberOfGroupByColumns = 4; resultCardinality = (int) PA.invokeMethod(eGByOp, "calculateGroupByTableCardinality(long,int,int)", memoryBudgetInBytes, numberOfGroupByColumns, frameSize); Assert.assertTrue(resultCardinality == 14563); memoryBudgetInBytes = 1024 * 1024 * 32L; numberOfGroupByColumns = 4; resultCardinality = (int) PA.invokeMethod(eGByOp, "calculateGroupByTableCardinality(long,int,int)", memoryBudgetInBytes, numberOfGroupByColumns, frameSize); Assert.assertTrue(resultCardinality == 441913); }