PipelineUtils implements Serializable { public static Set<String> extractRequiredInputColumns(Transformer[] transformers) { Set<String> inputColumns = new HashSet<>(); for(Transformer t : transformers) { inputColumns.addAll(t.getInputKeys()); } for(Transformer t : transformers) { Set<String> setDifference = new HashSet<>(t.getOutputKeys()); setDifference.removeAll(t.getInputKeys()); inputColumns.removeAll(setDifference); } return inputColumns; }  static Set<String> extractRequiredInputColumns(Transformer[] transformers); static Set<String> extractRequiredOutputColumns(Transformer[] transformers);  }
@Test public void testExtractedInputForSingleTransformer() { Transformer t = createTransformer(Arrays.asList("a"), Arrays.asList("b")); Set<String> inputs = PipelineUtils.extractRequiredInputColumns(new Transformer[] {t}); assertEquals(inputs.size(), 1); assertTrue(inputs.containsAll(Arrays.asList("a"))); assertFalse(inputs.contains("b")); }
@Test public void testExtractedInputForSingleTransformerWithColumnModification() { Transformer t = createTransformer(Arrays.asList("a", "b", "c"), Arrays.asList("a1", "b1", "c")); Set<String> inputs = PipelineUtils.extractRequiredInputColumns(new Transformer[] {t}); assertEquals(inputs.size(), 3); assertTrue(inputs.containsAll(Arrays.asList("a","b","c"))); assertFalse(inputs.contains("a1")); assertFalse(inputs.contains("b1")); }
@Test public void testExtractedInputForIndependentTransformers() { 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> inputs = PipelineUtils.extractRequiredInputColumns(new Transformer[] {t1, t2}); assertEquals(inputs.size(), 6); assertTrue(inputs.containsAll(Arrays.asList("a","b","c","d","e","f"))); assertFalse(inputs.contains("a1")); assertFalse(inputs.contains("b1")); assertFalse(inputs.contains("d1")); assertFalse(inputs.contains("e1")); assertFalse(inputs.contains("f1")); }
@Test public void testExtractedInputForDependentTransformers() { 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> inputs = PipelineUtils.extractRequiredInputColumns(new Transformer[] {t1, t2, t3}); assertEquals(inputs.size(), 4); assertTrue(inputs.containsAll(Arrays.asList("a","b","c","f"))); assertFalse(inputs.contains("a1")); assertFalse(inputs.contains("b1")); assertFalse(inputs.contains("a2")); assertFalse(inputs.contains("c1")); assertFalse(inputs.contains("a3")); assertFalse(inputs.contains("a4")); }
@Test public void testExtractedInputForTransformersWithModification() { 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> inputs = PipelineUtils.extractRequiredInputColumns(new Transformer[] {t1, t2, t3}); assertEquals(inputs.size(), 4); assertTrue(inputs.containsAll(Arrays.asList("a","b","c","f"))); assertFalse(inputs.contains("a1")); assertFalse(inputs.contains("b1")); assertFalse(inputs.contains("a2")); assertFalse(inputs.contains("c1")); }