APIFramework { private static AlgebricksAbsolutePartitionConstraint chooseLocations(IClusterInfoCollector clusterInfoCollector, int parallelismHint, AlgebricksAbsolutePartitionConstraint storageLocations) throws AlgebricksException { try { Map<String, NodeControllerInfo> ncMap = clusterInfoCollector.getNodeControllerInfos(); int totalNumCores = getTotalNumCores(ncMap); if (parallelismHint == CompilerProperties.COMPILER_PARALLELISM_AS_STORAGE && storageLocations.getLocations().length <= totalNumCores) { return storageLocations; } return getComputationLocations(ncMap, parallelismHint); } catch (HyracksException e) { throw new AlgebricksException(e); } } APIFramework(ILangCompilationProvider compilationProvider); Pair<IReturningStatement, Integer> reWriteQuery(List<FunctionDecl> declaredFunctions,
            MetadataProvider metadataProvider, IReturningStatement q, SessionOutput output, boolean inlineUdfs,
            Collection<VarIdentifier> externalVars, IWarningCollector warningCollector); JobSpecification compileQuery(IClusterInfoCollector clusterInfoCollector, MetadataProvider metadataProvider,
            Query query, int varCounter, String outputDatasetName, SessionOutput output,
            ICompiledDmlStatement statement, Map<VarIdentifier, IAObject> externalVars, IResponsePrinter printer,
            IWarningCollector warningCollector); void executeJobArray(IHyracksClientConnection hcc, JobSpecification[] specs, PrintWriter out); void executeJobArray(IHyracksClientConnection hcc, Job[] jobs, PrintWriter out); ExecutionPlans getExecutionPlans(); static AlgebricksAbsolutePartitionConstraint getJobLocations(JobSpecification spec,
            INodeJobTracker jobTracker, AlgebricksAbsolutePartitionConstraint clusterLocations); static final String PREFIX_INTERNAL_PARAMETERS; }
@Test public void testChooseLocations() throws Exception { IClusterInfoCollector clusterInfoCollector = mock(IClusterInfoCollector.class); Map<String, NodeControllerInfo> map = new HashMap<>(); NodeControllerInfo nc1Info = mock(NodeControllerInfo.class); when(nc1Info.getNumAvailableCores()).thenReturn(1); NodeControllerInfo nc2Info = mock(NodeControllerInfo.class); when(nc2Info.getNumAvailableCores()).thenReturn(1); String nc1 = "nc1"; String nc2 = "nc2"; map.put(nc1, nc1Info); map.put(nc2, nc2Info); when(clusterInfoCollector.getNodeControllerInfos()).thenReturn(map); APIFramework apiFramework = new APIFramework(mock(ILangCompilationProvider.class)); AlgebricksAbsolutePartitionConstraint storageLocations = new AlgebricksAbsolutePartitionConstraint(new String[] { "node1", "node1", "node2" }); AlgebricksAbsolutePartitionConstraint computationLocations = (AlgebricksAbsolutePartitionConstraint) PA.invokeMethod(apiFramework, "chooseLocations(" + IClusterInfoCollector.class.getName() + ",int," + AlgebricksAbsolutePartitionConstraint.class.getName() + ")", clusterInfoCollector, CompilerProperties.COMPILER_PARALLELISM_AS_STORAGE, storageLocations); Assert.assertTrue(computationLocations.getLocations().length == 2); storageLocations = new AlgebricksAbsolutePartitionConstraint(new String[] { "node1", "node2" }); computationLocations = (AlgebricksAbsolutePartitionConstraint) PA.invokeMethod(apiFramework, "chooseLocations(" + IClusterInfoCollector.class.getName() + ",int," + AlgebricksAbsolutePartitionConstraint.class.getName() + ")", clusterInfoCollector, CompilerProperties.COMPILER_PARALLELISM_AS_STORAGE, storageLocations); Assert.assertTrue(computationLocations.getLocations().length == 2); storageLocations = new AlgebricksAbsolutePartitionConstraint(new String[] { "node1" }); computationLocations = (AlgebricksAbsolutePartitionConstraint) PA.invokeMethod(apiFramework, "chooseLocations(" + IClusterInfoCollector.class.getName() + ",int," + AlgebricksAbsolutePartitionConstraint.class.getName() + ")", clusterInfoCollector, CompilerProperties.COMPILER_PARALLELISM_AS_STORAGE, storageLocations); Assert.assertTrue(computationLocations.getLocations().length == 1); verify(clusterInfoCollector, times(3)).getNodeControllerInfos(); }