@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 testTransformFunctionWithWrongInput() {
    Schema pinotSchema = new Schema();
    DimensionFieldSpec dimensionFieldSpec = new DimensionFieldSpec("x", FieldSpec.DataType.INT, true);
    pinotSchema.addField(dimensionFieldSpec);
    List<TransformConfig> transformConfigs = Collections.singletonList(
        new TransformConfig("y", "plus(x, 10)"));
    IngestionConfig ingestionConfig = new IngestionConfig();
    ingestionConfig.setTransformConfigs(transformConfigs);
    TableConfig tableConfig =
        new TableConfigBuilder(TableType.REALTIME).setTableName("testTransformFunctionWithWrongInput")
            .setIngestionConfig(ingestionConfig)
            .build();
    ExpressionTransformer expressionTransformer = new ExpressionTransformer(tableConfig, pinotSchema);
    // Valid case: x is int, y is int
    GenericRow genericRow = new GenericRow();
    genericRow.putValue("x", 10);
    expressionTransformer.transform(genericRow);
    Assert.assertEquals(genericRow.getValue("y"), 20.0);
    // Invalid case: x is string, y is int
    genericRow = new GenericRow();
    genericRow.putValue("x", "abcd");
    try {
      expressionTransformer.transform(genericRow);
      Assert.fail();
    } catch (Exception e) {
      Assert.assertEquals(e.getCause().getMessage(), "Caught exception while executing function: plus(x,'10')");
    }
  }