EventTimeHistoryPlanner implements RandomPlanner, ProvidesAlias, ProvidesValidations, InstantiatesComponents { private Row carryForwardWhenNull(Row into, Row from) { if (!config.hasPath(CARRY_FORWARD_CONFIG_NAME) || !config.getBoolean(CARRY_FORWARD_CONFIG_NAME)) { return into; } for (StructField field : into.schema().fields()) { String fieldName = field.name(); if (into.getAs(fieldName) == null && from.getAs(fieldName) != null) { into = RowUtils.set(into, fieldName, from.getAs(fieldName)); } } return into; }  @Override void configure(Config config); @Override Set<MutationType> getEmittedMutationTypes(); @Override List<Row> planMutationsForKey(Row key, List<Row> arrivingForKey, List<Row> existingForKey); @Override List<String> getKeyFieldNames(); @Override String getAlias(); @Override Validations getValidations(); @Override Set<InstantiatedComponent> getComponents(Config config, boolean configure); static final String KEY_FIELD_NAMES_CONFIG_NAME; static final String VALUE_FIELD_NAMES_CONFIG_NAME; static final String TIMESTAMP_FIELD_NAMES_CONFIG_NAME; static final String EFFECTIVE_FROM_FIELD_NAMES_CONFIG_NAME; static final String EFFECTIVE_TO_FIELD_NAMES_CONFIG_NAME; static final String CURRENT_FLAG_FIELD_NAME_CONFIG_NAME; static final String SURROGATE_KEY_FIELD_NAME_CONFIG_NAME; static final String CURRENT_FLAG_YES_CONFIG_NAME; static final String CURRENT_FLAG_NO_CONFIG_NAME; static final String LAST_UPDATED_FIELD_NAME_CONFIG_NAME; static final String CARRY_FORWARD_CONFIG_NAME; static final String EVENT_TIME_MODEL_CONFIG_NAME; static final String LAST_UPDATED_TIME_MODEL_CONFIG_NAME; static final String CURRENT_FLAG_DEFAULT_YES; static final String CURRENT_FLAG_DEFAULT_NO; }
@Test public void testCarryForwardWhenNull() { p = new EventTimeHistoryPlanner(); config = config.withValue(EventTimeHistoryPlanner.CARRY_FORWARD_CONFIG_NAME, ConfigValueFactory.fromAnyRef(true)); assertNoValidationFailures(p, config); p.configure(config); existing.add(new RowWithSchema(existingSchema, "a", "hello", 100L, 100L, 253402214400000L, EventTimeHistoryPlanner.CURRENT_FLAG_DEFAULT_YES, "")); arriving.add(new RowWithSchema(arrivingSchema, "a", null, 200L)); key = new RowWithSchema(keySchema, "a"); List<Row> planned = p.planMutationsForKey(key, arriving, existing); assertEquals(planned.size(), 2); assertEquals(PlannerUtils.getMutationType(planned.get(0)), MutationType.UPDATE); assertEquals(planned.get(0).getAs("value"), "hello"); assertEquals(planned.get(0).getAs("startdate"), 100L); assertEquals(planned.get(0).getAs("enddate"), 199L); assertEquals(planned.get(0).getAs("currentflag"), EventTimeHistoryPlanner.CURRENT_FLAG_DEFAULT_NO); assertEquals(PlannerUtils.getMutationType(planned.get(1)), MutationType.INSERT); assertEquals(planned.get(1).getAs("value"), "hello"); assertEquals(planned.get(1).getAs("startdate"), 200L); assertEquals(planned.get(1).getAs("enddate"), 253402214400000L); assertEquals(planned.get(1).getAs("currentflag"), EventTimeHistoryPlanner.CURRENT_FLAG_DEFAULT_YES); }