@Override
  public GenericRow transform(GenericRow record) {
    for (Map.Entry<String, FunctionEvaluator> entry : _expressionEvaluators.entrySet()) {
      String column = entry.getKey();
      FunctionEvaluator transformFunctionEvaluator = entry.getValue();
      Object existingValue = record.getValue(column);
      if (existingValue == null) {
        try {
          // Skip transformation if column value already exists
          // NOTE: column value might already exist for OFFLINE data,
          // For backward compatibility, The only exception here is that we will override nested field like array,
          // collection or map since they were not included in the record transformation before.
          record.putValue(column, transformFunctionEvaluator.evaluate(record));
        } catch (Exception e) {
          if (!_continueOnError) {
            throw new RuntimeException("Caught exception while evaluation transform function for column: " + column, e);
          } else {
            LOGGER.debug("Caught exception while evaluation transform function for column: {}", column, e);
            record.putValue(GenericRow.INCOMPLETE_RECORD_KEY, true);
          }
        }
      } else if (existingValue.getClass().isArray() || existingValue instanceof Collections
          || existingValue instanceof Map) {
        try {
          Object transformedValue = transformFunctionEvaluator.evaluate(record);
          // For backward compatibility, The only exception here is that we will override nested field like array,
          // collection or map since they were not included in the record transformation before.
          if (!isTypeCompatible(existingValue, transformedValue)) {
            record.putValue(column, transformedValue);
          }
        } catch (Exception e) {
          LOGGER.debug("Caught exception while evaluation transform function for column: {}", column, e);
        }
      }
    }
    return record;
  }
@Test
  public void testValueAlreadyExists() {
    Schema pinotSchema = new Schema();
    DimensionFieldSpec dimensionFieldSpec = new DimensionFieldSpec("fullName", FieldSpec.DataType.STRING, true);
    pinotSchema.addField(dimensionFieldSpec);
    List<TransformConfig> transformConfigs = Collections.singletonList(
        new TransformConfig("fullName", "Groovy({firstName + ' ' + lastName}, firstName, lastName)"));
    IngestionConfig ingestionConfig = new IngestionConfig();
    ingestionConfig.setTransformConfigs(transformConfigs);
    TableConfig tableConfig =
        new TableConfigBuilder(TableType.REALTIME).setTableName("testValueExists").setIngestionConfig(ingestionConfig)
            .build();
    ExpressionTransformer expressionTransformer = new ExpressionTransformer(tableConfig, pinotSchema);

    GenericRow genericRow = new GenericRow();
    genericRow.putValue("firstName", "John");
    genericRow.putValue("lastName", "Denver");
    genericRow.putValue("fullName", "John N Denver");

    // no transformation
    expressionTransformer.transform(genericRow);
    Assert.assertEquals(genericRow.getValue("fullName"), "John N Denver");

    pinotSchema = new Schema.SchemaBuilder()
        .addTime(new TimeGranularitySpec(FieldSpec.DataType.LONG, TimeUnit.MILLISECONDS, "incoming"),
            new TimeGranularitySpec(FieldSpec.DataType.INT, TimeUnit.DAYS, "outgoing")).build();
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName("testValueExists")
        .setIngestionConfig(new IngestionConfig()).build();
    expressionTransformer = new ExpressionTransformer(tableConfig, pinotSchema);

    genericRow = new GenericRow();
    genericRow.putValue("incoming", "123456789");
    genericRow.putValue("outgoing", "123");

    // no transformation
    expressionTransformer.transform(genericRow);
    Assert.assertEquals(genericRow.getValue("outgoing"), "123");
  }