@Test public void testJoinerWithNullableSchema() { Schema filmCategorySchema = Schema.recordOf( "filmCategory", Schema.Field.of("film_id", Schema.of(Schema.Type.STRING)), Schema.Field.of("film_name", Schema.of(Schema.Type.STRING)), Schema.Field.of("category_name", Schema.nullableOf(Schema.of(Schema.Type.STRING)))); Map<String, Schema> inputSchemas = ImmutableMap.of("film", FILM_SCHEMA, "filmActor", FILM_ACTOR_SCHEMA, "filmCategory", filmCategorySchema); List<String> inputStages = Arrays.asList("film", "filmActor", "filmCategory"); String joinKeys = "film.film_id=filmActor.film_id=filmCategory.film_id"; String selectedFields = "film.film_id, film.film_name, filmActor.actor_name as renamed_actor, " + "filmCategory.category_name as renamed_category"; String requiredInputs = "film,filmActor"; JoinerConfig conf = new JoinerConfig(joinKeys, selectedFields, requiredInputs); Schema outputSchema = Schema.recordOf( "join.output", Schema.Field.of("film_id", Schema.of(Schema.Type.STRING)), Schema.Field.of("film_name", Schema.of(Schema.Type.STRING)), Schema.Field.of("renamed_actor", Schema.of(Schema.Type.STRING)), Schema.Field.of("renamed_category", filmCategorySchema.getField("category_name").getSchema())); Joiner joiner = new Joiner(conf); FailureCollector collector = new MockFailureCollector(); AutoJoinerContext autoJoinerContext = new MockAutoJoinerContext(INPUT_STAGES, collector); JoinDefinition joinDefinition = joiner.define(autoJoinerContext); Assert.assertEquals(outputSchema, joinDefinition.getOutputSchema()); Assert.assertEquals(0, collector.getValidationFailures().size()); }
@Nullable public Schema getOutputSchema(FailureCollector collector) { try { return Strings.isNullOrEmpty(schema) ? null : Schema.parseJson(schema); } catch (IOException e) { collector.addFailure("Invalid schema: " + e.getMessage(), null).withConfigProperty(OUTPUT_SCHEMA); } throw collector.getOrThrowException(); }
JoinerConfig extends PluginConfig { @Nullable public Schema getOutputSchema(FailureCollector collector) { try { return Strings.isNullOrEmpty(schema) ? null : Schema.parseJson(schema); } catch (IOException e) { collector.addFailure("Invalid schema: " + e.getMessage(), null).withConfigProperty(OUTPUT_SCHEMA); } throw collector.getOrThrowException(); } }
JoinerConfig extends PluginConfig { @Nullable public Schema getOutputSchema(FailureCollector collector) { try { return Strings.isNullOrEmpty(schema) ? null : Schema.parseJson(schema); } catch (IOException e) { collector.addFailure("Invalid schema: " + e.getMessage(), null).withConfigProperty(OUTPUT_SCHEMA); } throw collector.getOrThrowException(); } JoinerConfig(); @VisibleForTesting  JoinerConfig(String joinKeys, String selectedFields, String requiredInputs); }
JoinerConfig extends PluginConfig { @Nullable public Schema getOutputSchema(FailureCollector collector) { try { return Strings.isNullOrEmpty(schema) ? null : Schema.parseJson(schema); } catch (IOException e) { collector.addFailure("Invalid schema: " + e.getMessage(), null).withConfigProperty(OUTPUT_SCHEMA); } throw collector.getOrThrowException(); } 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 { @Nullable public Schema getOutputSchema(FailureCollector collector) { try { return Strings.isNullOrEmpty(schema) ? null : Schema.parseJson(schema); } catch (IOException e) { collector.addFailure("Invalid schema: " + e.getMessage(), null).withConfigProperty(OUTPUT_SCHEMA); } throw collector.getOrThrowException(); } 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; }