JoinerConfig extends PluginConfig { List<JoinField> getSelectedFields(FailureCollector failureCollector) { if (Strings.isNullOrEmpty(selectedFields)) { failureCollector.addFailure("Must select at least one field", null).withConfigProperty(SELECTED_FIELDS); throw failureCollector.getOrThrowException(); } List<JoinField> selectedJoinFields = new ArrayList<>(); for (String selectedField : Splitter.on(',').trimResults().omitEmptyStrings().split(selectedFields)) { Iterable<String> stageOldNameAliasPair = Splitter.on(" as ").trimResults().omitEmptyStrings() .split(selectedField); Iterable<String> stageOldNamePair = Splitter.on('.').trimResults().omitEmptyStrings(). split(Iterables.get(stageOldNameAliasPair, 0)); if (Iterables.size(stageOldNamePair) != 2) { failureCollector.addFailure(String.format("Invalid syntax. Selected Fields must be of syntax " + "<stageName>.<oldFieldName> as <alias>, but found %s", selectedField), null) .withConfigProperty(SELECTED_FIELDS); continue; } String stageName = Iterables.get(stageOldNamePair, 0); String oldFieldName = Iterables.get(stageOldNamePair, 1); String alias = Iterables.size(stageOldNameAliasPair) == 1 ? oldFieldName : Iterables.get(stageOldNameAliasPair, 1); selectedJoinFields.add(new JoinField(stageName, oldFieldName, alias)); } failureCollector.getOrThrowException(); return selectedJoinFields; } 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; }
@Test public void testJoinerConfigWithSelectedFields() { JoinerConfig config = new JoinerConfig("film.film_id=filmActor.film_id=filmCategory.film_id&" + "film.film_name=filmActor.film_name=filmCategory.film_name", SELECTED_FIELDS, "film,filmActor,filmCategory"); List<JoinField> expected = Arrays.asList( new JoinField("film", "film_id", "film_id"), new JoinField("film", "film_name", "film_name"), new JoinField("filmActor", "actor_name", "renamed_actor"), new JoinField("filmCategory", "category_name", "renamed_category")); Assert.assertEquals(expected, config.getSelectedFields(new MockFailureCollector())); }
@Test public void testJoinerConfigWithoutSelectedFields() { JoinerConfig config = new JoinerConfig("film.film_id=filmActor.film_id=filmCategory.film_id&" + "film.film_name=filmActor.film_name=filmCategory.film_name", "", "film,filmActor,filmCategory"); FailureCollector failureCollector = new MockFailureCollector(); try { config.getSelectedFields(failureCollector); } catch (ValidationException e) { Assert.assertEquals(1, e.getFailures().size()); ValidationFailure failure = e.getFailures().get(0); Assert.assertEquals(1, failure.getCauses().size()); Cause cause = failure.getCauses().get(0); Assert.assertEquals(JoinerConfig.SELECTED_FIELDS, cause.getAttribute(CauseAttributes.STAGE_CONFIG)); } }
@Test public void testJoinerConfigWithoutFieldsToRename() { String selectedFields = "film.film_id, film.film_name, " + "filmActor.actor_name as renamed_actor, filmCategory.category_name as renamed_category"; JoinerConfig config = new JoinerConfig("film.film_id=filmActor.film_id=filmCategory.film_id&" + "film.film_name=filmActor.film_name=filmCategory.film_name", selectedFields, "film,filmActor,filmCategory"); FailureCollector failureCollector = new MockFailureCollector(); List<JoinField> actual = config.getSelectedFields(failureCollector); List<JoinField> expected = Arrays.asList( new JoinField("film", "film_id", "film_id"), new JoinField("film", "film_name", "film_name"), new JoinField("filmActor", "actor_name", "renamed_actor"), new JoinField("filmCategory", "category_name", "renamed_category")); Assert.assertEquals(expected, actual); }