@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 testTransformConfigsFromTableConfig() {
    Schema pinotSchema = new Schema.SchemaBuilder().addSingleValueDimension("userId", FieldSpec.DataType.LONG)
        .addSingleValueDimension("fullName", FieldSpec.DataType.STRING)
        .addMultiValueDimension("bids", FieldSpec.DataType.INT)
        .addSingleValueDimension("maxBid", FieldSpec.DataType.INT)
        .addMultiValueDimension("map2_keys", FieldSpec.DataType.STRING)
        .addMultiValueDimension("map2_values", FieldSpec.DataType.INT).addMetric("cost", FieldSpec.DataType.DOUBLE)
        .addDateTime("hoursSinceEpoch", FieldSpec.DataType.LONG, "1:HOURS:EPOCH", "1:HOURS").build();

    List<TransformConfig> transformConfigs = Arrays.asList(
        new TransformConfig("userId", "Groovy({user_id}, user_id)"),
        new TransformConfig("fullName", "Groovy({firstName+' '+lastName}, firstName, lastName)"),
        new TransformConfig("maxBid", "Groovy({bids.max{ it.toBigDecimal() }}, bids)"),
        new TransformConfig("map2_keys", "Groovy({map2.sort()*.key}, map2)"),
        new TransformConfig("map2_values", "Groovy({map2.sort()*.value}, map2)"),
        new TransformConfig("hoursSinceEpoch", "Groovy({timestamp/(1000*60*60)}, timestamp)"));
    IngestionConfig ingestionConfig = new IngestionConfig();
    ingestionConfig.setTransformConfigs(transformConfigs);
    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTransformFunctions")
        .setIngestionConfig(ingestionConfig).build();

    ExpressionTransformer expressionTransformer = new ExpressionTransformer(tableConfig, pinotSchema);
    DataTypeTransformer dataTypeTransformer = new DataTypeTransformer(tableConfig, pinotSchema);

    // test functions from schema
    GenericRow genericRow = new GenericRow();
    genericRow.putValue("user_id", 1L);
    genericRow.putValue("firstName", "John");
    genericRow.putValue("lastName", "Denver");
    genericRow.putValue("bids", Arrays.asList(10, 20));
    HashMap<String, String> map1 = new HashMap<>(); // keys in Map from avro are always in STRING
    map1.put("30", "foo");
    map1.put("200", "bar");
    genericRow.putValue("map1", map1);
    HashMap<String, Integer> map2 = new HashMap<>();
    map2.put("k1", 10);
    map2.put("k2", 20);
    genericRow.putValue("map2", map2);
    genericRow.putValue("cost", 1000.0);
    genericRow.putValue("timestamp", 1574000000000L);
    genericRow.putValue("lon", 1.0);
    genericRow.putValue("lat", 2.0);

    // expression transformer
    expressionTransformer.transform(genericRow);

    // extract userId
    Assert.assertEquals(genericRow.getValue("userId"), 1L);
    // concat fullName
    Assert.assertEquals(genericRow.getValue("fullName"), "John Denver");
    Assert.assertTrue(((List) genericRow.getValue("bids")).containsAll(Arrays.asList(10, 20)));
    // find max bid from bids
    Assert.assertEquals(genericRow.getValue("maxBid"), 20);
    // handle Map through transform functions
    ArrayList map2Keys = (ArrayList) genericRow.getValue("map2_keys");
    Assert.assertEquals(map2Keys.get(0), "k1");
    Assert.assertEquals(map2Keys.get(1), "k2");
    ArrayList map2Values = (ArrayList) genericRow.getValue("map2_values");
    Assert.assertEquals(map2Values.get(0), 10);
    Assert.assertEquals(map2Values.get(1), 20);
    Assert.assertEquals(genericRow.getValue("cost"), 1000.0);
    // calculate hoursSinceEpoch
    Assert.assertEquals(genericRow.getValue("hoursSinceEpoch").toString(), "437222.2222222222");

    // data type transformer
    dataTypeTransformer.transform(genericRow);

    Assert.assertEquals(genericRow.getValue("userId"), 1L);
    Assert.assertEquals(genericRow.getValue("fullName"), "John Denver");
    Assert.assertEquals(((Object[]) genericRow.getValue("bids")), new Integer[]{10, 20});
    Assert.assertEquals(genericRow.getValue("maxBid"), 20);
    // handle Map through transform functions
    Object[] map2KeysObject = (Object[]) genericRow.getValue("map2_keys");
    Assert.assertEquals(map2KeysObject[0], "k1");
    Assert.assertEquals(map2KeysObject[1], "k2");
    Object[] map2ValuesObject = (Object[]) genericRow.getValue("map2_values");
    Assert.assertEquals(map2ValuesObject[0], 10);
    Assert.assertEquals(map2ValuesObject[1], 20);
    Assert.assertEquals(genericRow.getValue("cost"), 1000.0);
    // convert to LONG
    Assert.assertEquals(genericRow.getValue("hoursSinceEpoch"), 437222L);
  }