@Test public void testJoinerConfigWithoutJoinKeys() { JoinerConfig config = new JoinerConfig("", SELECTED_FIELDS, "film,filmActor,filmCategory"); MockFailureCollector failureCollector = new MockFailureCollector(); try { config.getJoinKeys(failureCollector); } catch (ValidationException e) { Assert.assertEquals(1, e.getFailures().size()); ValidationFailure failure = e.getFailures().get(0); Assert.assertEquals(1, failure.getCauses().size()); } }
Set<JoinKey> getJoinKeys(FailureCollector failureCollector) { Map<String, List<String>> stageToKey = new LinkedHashMap<>(); if (Strings.isNullOrEmpty(joinKeys)) { failureCollector.addFailure("Join keys cannot be empty", null).withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } Iterable<String> multipleJoinKeys = Splitter.on('&').trimResults().omitEmptyStrings().split(joinKeys); if (Iterables.isEmpty(multipleJoinKeys)) { failureCollector.addFailure("Join keys cannot be empty", null).withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } int numJoinKeys = 0; for (String singleJoinKey : multipleJoinKeys) { KeyValueListParser kvParser = new KeyValueListParser("\\s*=\\s*", "\\."); Iterable<KeyValue<String, String>> keyValues = kvParser.parse(singleJoinKey); if (numJoinKeys == 0) { numJoinKeys = Iterables.size(keyValues); } else if (numJoinKeys != Iterables.size(keyValues)) { failureCollector.addFailure("There should be one join key from each of the stages", "Please add join keys for each stage.") .withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } for (KeyValue<String, String> keyValue : keyValues) { String stageName = keyValue.getKey(); String joinKey = keyValue.getValue(); if (!stageToKey.containsKey(stageName)) { stageToKey.put(stageName, new ArrayList<>()); } stageToKey.get(stageName).add(joinKey); } } return stageToKey.entrySet().stream() .map(entry -> new JoinKey(entry.getKey(), entry.getValue())) .collect(Collectors.toSet()); }
JoinerConfig extends PluginConfig { Set<JoinKey> getJoinKeys(FailureCollector failureCollector) { Map<String, List<String>> stageToKey = new LinkedHashMap<>(); if (Strings.isNullOrEmpty(joinKeys)) { failureCollector.addFailure("Join keys cannot be empty", null).withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } Iterable<String> multipleJoinKeys = Splitter.on('&').trimResults().omitEmptyStrings().split(joinKeys); if (Iterables.isEmpty(multipleJoinKeys)) { failureCollector.addFailure("Join keys cannot be empty", null).withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } int numJoinKeys = 0; for (String singleJoinKey : multipleJoinKeys) { KeyValueListParser kvParser = new KeyValueListParser("\\s*=\\s*", "\\."); Iterable<KeyValue<String, String>> keyValues = kvParser.parse(singleJoinKey); if (numJoinKeys == 0) { numJoinKeys = Iterables.size(keyValues); } else if (numJoinKeys != Iterables.size(keyValues)) { failureCollector.addFailure("There should be one join key from each of the stages", "Please add join keys for each stage.") .withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } for (KeyValue<String, String> keyValue : keyValues) { String stageName = keyValue.getKey(); String joinKey = keyValue.getValue(); if (!stageToKey.containsKey(stageName)) { stageToKey.put(stageName, new ArrayList<>()); } stageToKey.get(stageName).add(joinKey); } } return stageToKey.entrySet().stream() .map(entry -> new JoinKey(entry.getKey(), entry.getValue())) .collect(Collectors.toSet()); } }
JoinerConfig extends PluginConfig { Set<JoinKey> getJoinKeys(FailureCollector failureCollector) { Map<String, List<String>> stageToKey = new LinkedHashMap<>(); if (Strings.isNullOrEmpty(joinKeys)) { failureCollector.addFailure("Join keys cannot be empty", null).withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } Iterable<String> multipleJoinKeys = Splitter.on('&').trimResults().omitEmptyStrings().split(joinKeys); if (Iterables.isEmpty(multipleJoinKeys)) { failureCollector.addFailure("Join keys cannot be empty", null).withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } int numJoinKeys = 0; for (String singleJoinKey : multipleJoinKeys) { KeyValueListParser kvParser = new KeyValueListParser("\\s*=\\s*", "\\."); Iterable<KeyValue<String, String>> keyValues = kvParser.parse(singleJoinKey); if (numJoinKeys == 0) { numJoinKeys = Iterables.size(keyValues); } else if (numJoinKeys != Iterables.size(keyValues)) { failureCollector.addFailure("There should be one join key from each of the stages", "Please add join keys for each stage.") .withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } for (KeyValue<String, String> keyValue : keyValues) { String stageName = keyValue.getKey(); String joinKey = keyValue.getValue(); if (!stageToKey.containsKey(stageName)) { stageToKey.put(stageName, new ArrayList<>()); } stageToKey.get(stageName).add(joinKey); } } return stageToKey.entrySet().stream() .map(entry -> new JoinKey(entry.getKey(), entry.getValue())) .collect(Collectors.toSet()); } JoinerConfig(); @VisibleForTesting  JoinerConfig(String joinKeys, String selectedFields, String requiredInputs); }
JoinerConfig extends PluginConfig { Set<JoinKey> getJoinKeys(FailureCollector failureCollector) { Map<String, List<String>> stageToKey = new LinkedHashMap<>(); if (Strings.isNullOrEmpty(joinKeys)) { failureCollector.addFailure("Join keys cannot be empty", null).withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } Iterable<String> multipleJoinKeys = Splitter.on('&').trimResults().omitEmptyStrings().split(joinKeys); if (Iterables.isEmpty(multipleJoinKeys)) { failureCollector.addFailure("Join keys cannot be empty", null).withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } int numJoinKeys = 0; for (String singleJoinKey : multipleJoinKeys) { KeyValueListParser kvParser = new KeyValueListParser("\\s*=\\s*", "\\."); Iterable<KeyValue<String, String>> keyValues = kvParser.parse(singleJoinKey); if (numJoinKeys == 0) { numJoinKeys = Iterables.size(keyValues); } else if (numJoinKeys != Iterables.size(keyValues)) { failureCollector.addFailure("There should be one join key from each of the stages", "Please add join keys for each stage.") .withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } for (KeyValue<String, String> keyValue : keyValues) { String stageName = keyValue.getKey(); String joinKey = keyValue.getValue(); if (!stageToKey.containsKey(stageName)) { stageToKey.put(stageName, new ArrayList<>()); } stageToKey.get(stageName).add(joinKey); } } return stageToKey.entrySet().stream() .map(entry -> new JoinKey(entry.getKey(), entry.getValue())) .collect(Collectors.toSet()); } JoinerConfig(); @VisibleForTesting  JoinerConfig(String joinKeys, String selectedFields, String requiredInputs); @Nullable Integer getNumPartitions(); @Nullable Schema getOutputSchema(FailureCollector collector); @Nullable Integer getDistributionFactor(); @Nullable String getDistributionStageName(); boolean isDistributionValid(FailureCollector collector); boolean distributionContainsMacro(); }
JoinerConfig extends PluginConfig { Set<JoinKey> getJoinKeys(FailureCollector failureCollector) { Map<String, List<String>> stageToKey = new LinkedHashMap<>(); if (Strings.isNullOrEmpty(joinKeys)) { failureCollector.addFailure("Join keys cannot be empty", null).withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } Iterable<String> multipleJoinKeys = Splitter.on('&').trimResults().omitEmptyStrings().split(joinKeys); if (Iterables.isEmpty(multipleJoinKeys)) { failureCollector.addFailure("Join keys cannot be empty", null).withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } int numJoinKeys = 0; for (String singleJoinKey : multipleJoinKeys) { KeyValueListParser kvParser = new KeyValueListParser("\\s*=\\s*", "\\."); Iterable<KeyValue<String, String>> keyValues = kvParser.parse(singleJoinKey); if (numJoinKeys == 0) { numJoinKeys = Iterables.size(keyValues); } else if (numJoinKeys != Iterables.size(keyValues)) { failureCollector.addFailure("There should be one join key from each of the stages", "Please add join keys for each stage.") .withConfigProperty(JOIN_KEYS); throw failureCollector.getOrThrowException(); } for (KeyValue<String, String> keyValue : keyValues) { String stageName = keyValue.getKey(); String joinKey = keyValue.getValue(); if (!stageToKey.containsKey(stageName)) { stageToKey.put(stageName, new ArrayList<>()); } stageToKey.get(stageName).add(joinKey); } } return stageToKey.entrySet().stream() .map(entry -> new JoinKey(entry.getKey(), entry.getValue())) .collect(Collectors.toSet()); } JoinerConfig(); @VisibleForTesting  JoinerConfig(String joinKeys, String selectedFields, String requiredInputs); @Nullable Integer getNumPartitions(); @Nullable Schema getOutputSchema(FailureCollector collector); @Nullable Integer getDistributionFactor(); @Nullable String getDistributionStageName(); boolean isDistributionValid(FailureCollector collector); boolean distributionContainsMacro(); static final String SELECTED_FIELDS; static final String REQUIRED_INPUTS; static final String JOIN_KEYS; static final String OUTPUT_SCHEMA; static final String DISTRIBUTION_FACTOR; static final String DISTRIBUTION_STAGE; static final String MEMORY_INPUTS; }