@Test public void optimize() { IterativeComQueue queue = new IterativeComQueue() .add(new MyComputeFunction()) .add(new MyComputeFunction()) .add(new MyCommunicateFunction()) .add(new MyCommunicateFunction()) .add(new MyComputeFunction()) .add(new MyComputeFunction()) .add(new MyCommunicateFunction()); queue.exec(); Assert.assertTrue( queue.toString().matches( "\\{\"completeResult\":null,\"maxIter\":2147483647,\"sessionId\":[0-9]*" + ",\"queue\":\"ChainedComputation,MyCommunicateFunction,MyCommunicateFunction," + "ChainedComputation,MyCommunicateFunction\",\"compareCriterion\":null\\}")); }
private void optimize() { if (queue.isEmpty()) { return; } int current = 0; for (int ahead = 1; ahead < queue.size(); ++ahead) { ComQueueItem curItem = queue.get(current); ComQueueItem aheadItem = queue.get(ahead); if (aheadItem instanceof ComputeFunction && curItem instanceof ComputeFunction) { if (curItem instanceof ChainedComputation) { queue.set(current, ((ChainedComputation) curItem).add((ComputeFunction) aheadItem)); } else { queue.set(current, new ChainedComputation() .add((ComputeFunction) curItem) .add((ComputeFunction) aheadItem) ); } } else { queue.set(++current, aheadItem); } } queue.subList(current + 1, queue.size()).clear(); }
BaseComQueue implements Serializable { private void optimize() { if (queue.isEmpty()) { return; } int current = 0; for (int ahead = 1; ahead < queue.size(); ++ahead) { ComQueueItem curItem = queue.get(current); ComQueueItem aheadItem = queue.get(ahead); if (aheadItem instanceof ComputeFunction && curItem instanceof ComputeFunction) { if (curItem instanceof ChainedComputation) { queue.set(current, ((ChainedComputation) curItem).add((ComputeFunction) aheadItem)); } else { queue.set(current, new ChainedComputation() .add((ComputeFunction) curItem) .add((ComputeFunction) aheadItem) ); } } else { queue.set(++current, aheadItem); } } queue.subList(current + 1, queue.size()).clear(); } }
BaseComQueue implements Serializable { private void optimize() { if (queue.isEmpty()) { return; } int current = 0; for (int ahead = 1; ahead < queue.size(); ++ahead) { ComQueueItem curItem = queue.get(current); ComQueueItem aheadItem = queue.get(ahead); if (aheadItem instanceof ComputeFunction && curItem instanceof ComputeFunction) { if (curItem instanceof ChainedComputation) { queue.set(current, ((ChainedComputation) curItem).add((ComputeFunction) aheadItem)); } else { queue.set(current, new ChainedComputation() .add((ComputeFunction) curItem) .add((ComputeFunction) aheadItem) ); } } else { queue.set(++current, aheadItem); } } queue.subList(current + 1, queue.size()).clear(); }  }
BaseComQueue implements Serializable { private void optimize() { if (queue.isEmpty()) { return; } int current = 0; for (int ahead = 1; ahead < queue.size(); ++ahead) { ComQueueItem curItem = queue.get(current); ComQueueItem aheadItem = queue.get(ahead); if (aheadItem instanceof ComputeFunction && curItem instanceof ComputeFunction) { if (curItem instanceof ChainedComputation) { queue.set(current, ((ChainedComputation) curItem).add((ComputeFunction) aheadItem)); } else { queue.set(current, new ChainedComputation() .add((ComputeFunction) curItem) .add((ComputeFunction) aheadItem) ); } } else { queue.set(++current, aheadItem); } } queue.subList(current + 1, queue.size()).clear(); }  Q add(ComQueueItem com); Q closeWith(CompleteResultFunction completeResult); Q initWithPartitionedData(String objName, DataSet<T> data); Q initWithBroadcastData(String objName, DataSet<T> data); Q initWithMLSessionId(Long sessionId); DataSet<Row> exec(); @Override String toString(); }
BaseComQueue implements Serializable { private void optimize() { if (queue.isEmpty()) { return; } int current = 0; for (int ahead = 1; ahead < queue.size(); ++ahead) { ComQueueItem curItem = queue.get(current); ComQueueItem aheadItem = queue.get(ahead); if (aheadItem instanceof ComputeFunction && curItem instanceof ComputeFunction) { if (curItem instanceof ChainedComputation) { queue.set(current, ((ChainedComputation) curItem).add((ComputeFunction) aheadItem)); } else { queue.set(current, new ChainedComputation() .add((ComputeFunction) curItem) .add((ComputeFunction) aheadItem) ); } } else { queue.set(++current, aheadItem); } } queue.subList(current + 1, queue.size()).clear(); }  Q add(ComQueueItem com); Q closeWith(CompleteResultFunction completeResult); Q initWithPartitionedData(String objName, DataSet<T> data); Q initWithBroadcastData(String objName, DataSet<T> data); Q initWithMLSessionId(Long sessionId); DataSet<Row> exec(); @Override String toString();  }