PipelineUtils implements Serializable { public static Set<String> extractRequiredOutputColumns(Transformer[] transformers) { Set<String> outputColumns = new HashSet<>(); for(int i = transformers.length-1; i>=0; i--) { outputColumns.addAll(transformers[i].getOutputKeys()); } for(int i = transformers.length-1; i>=0; i--) { Set<String> setDifference = new HashSet<>(transformers[i].getInputKeys()); setDifference.removeAll(transformers[i].getOutputKeys()); outputColumns.removeAll(setDifference); } return outputColumns; }  static Set<String> extractRequiredInputColumns(Transformer[] transformers); static Set<String> extractRequiredOutputColumns(Transformer[] transformers);  }
@Test public void testExtractedOutputForSingleTransformer() { Transformer t = createTransformer(Arrays.asList("a"), Arrays.asList("b")); Set<String> outputs = PipelineUtils.extractRequiredOutputColumns(new Transformer[] {t}); assertEquals(outputs.size(), 1); assertTrue(outputs.containsAll(Arrays.asList("b"))); assertFalse(outputs.contains("a")); }
@Test public void testExtractedOutputForSingleTransformerWithColumnModification() { Transformer t = createTransformer(Arrays.asList("a", "b", "c"), Arrays.asList("a1", "b1", "c")); Set<String> outputs = PipelineUtils.extractRequiredOutputColumns(new Transformer[] {t}); assertEquals(outputs.size(), 3); assertTrue(outputs.containsAll(Arrays.asList("a1","b1","c"))); assertFalse(outputs.contains("a")); assertFalse(outputs.contains("b")); }
@Test public void testExtractedOutputForIndependentTransformers() { Transformer t1 = createTransformer(Arrays.asList("a", "b", "c"), Arrays.asList("a1", "b1", "c")); Transformer t2 = createTransformer(Arrays.asList("d", "e", "f"), Arrays.asList("d1", "e1", "f1")); Set<String> outputs = PipelineUtils.extractRequiredOutputColumns(new Transformer[] {t1, t2}); assertEquals(outputs.size(), 6); assertTrue(outputs.containsAll(Arrays.asList("a1","b1","c","d1","e1","f1"))); assertFalse(outputs.contains("a")); assertFalse(outputs.contains("b")); assertFalse(outputs.contains("d")); assertFalse(outputs.contains("e")); assertFalse(outputs.contains("f")); }
@Test public void testExtractedOutputForDependentTransformers() { Transformer t1 = createTransformer(Arrays.asList("a", "b", "c"), Arrays.asList("a1", "b1", "c")); Transformer t2 = createTransformer(Arrays.asList("a1", "c", "f"), Arrays.asList("a2", "c1")); Transformer t3 = createTransformer(Arrays.asList("a", "a1", "a2"), Arrays.asList("a3", "a4")); Set<String> outputs = PipelineUtils.extractRequiredOutputColumns(new Transformer[] {t1, t2, t3}); assertEquals(outputs.size(), 4); assertTrue(outputs.containsAll(Arrays.asList("b1","c1","a3","a4"))); assertFalse(outputs.contains("a")); assertFalse(outputs.contains("b")); assertFalse(outputs.contains("a1")); assertFalse(outputs.contains("c")); assertFalse(outputs.contains("f")); assertFalse(outputs.contains("a1")); assertFalse(outputs.contains("a2")); }
@Test public void testExtractedOutputForTransformersWithModification() { Transformer t1 = createTransformer(Arrays.asList("a", "b", "c"), Arrays.asList("a1", "b1", "c")); Transformer t2 = createTransformer(Arrays.asList("a1", "c", "f"), Arrays.asList("a2", "c1")); Transformer t3 = createTransformer(Arrays.asList("a", "a1", "a2"), Arrays.asList("a2")); Set<String> outputs = PipelineUtils.extractRequiredOutputColumns(new Transformer[] {t1, t2, t3}); assertEquals(outputs.size(), 3); assertTrue(outputs.containsAll(Arrays.asList("b1","a2","c1"))); assertFalse(outputs.contains("a")); assertFalse(outputs.contains("b")); assertFalse(outputs.contains("c")); assertFalse(outputs.contains("a1")); }