@Test public void testTransformConfigsFromSchema() { 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("map1__KEYS", FieldSpec.DataType.INT) .addMultiValueDimension("map1__VALUES", FieldSpec.DataType.STRING).addMetric("cost", FieldSpec.DataType.DOUBLE) .addDateTime("hoursSinceEpoch", FieldSpec.DataType.LONG, "1:HOURS:EPOCH", "1:HOURS").build(); pinotSchema.getFieldSpecFor("maxBid").setTransformFunction("Groovy({bids.max{ it.toBigDecimal() }}, bids)"); pinotSchema.getFieldSpecFor("hoursSinceEpoch").setTransformFunction("Groovy({timestamp/(1000)}, timestamp)"); List<TransformConfig> transformConfigs = new ArrayList<>(); transformConfigs.add(new TransformConfig("userId", "Groovy({user_id}, user_id)")); transformConfigs.add(new TransformConfig("fullName", "Groovy({firstName+' '+lastName}, firstName, lastName)")); transformConfigs.add(new TransformConfig("hoursSinceEpoch", "Groovy({timestamp/(1000*60*60)}, timestamp)")); TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTransformFunctions") .setIngestionConfig(new IngestionConfig(null, transformConfigs)).build(); ExpressionTransformer expressionTransformer = new ExpressionTransformer(tableConfig, pinotSchema); 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<>(); 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); expressionTransformer.transform(genericRow); Assert.assertEquals(genericRow.getValue("userId"), 1L); Assert.assertEquals(genericRow.getValue("fullName"), "John Denver"); Assert.assertTrue(((List) genericRow.getValue("bids")).containsAll(Arrays.asList(10, 20))); Assert.assertEquals(genericRow.getValue("maxBid"), 20); ArrayList map1__keys = (ArrayList) genericRow.getValue("map1__KEYS"); Assert.assertEquals(map1__keys.get(0), "200"); Assert.assertEquals(map1__keys.get(1), "30"); ArrayList map1__values = (ArrayList) genericRow.getValue("map1__VALUES"); Assert.assertEquals(map1__values.get(0), "bar"); Assert.assertEquals(map1__values.get(1), "foo"); Assert.assertEquals(genericRow.getValue("cost"), 1000.0); Assert.assertEquals(genericRow.getValue("hoursSinceEpoch").toString(), "437222.2222222222"); }
@Override public GenericRow transform(GenericRow record) { for (Map.Entry<String, FunctionEvaluator> entry : _expressionEvaluators.entrySet()) { String column = entry.getKey(); FunctionEvaluator transformFunctionEvaluator = entry.getValue(); if (record.getValue(column) == null) { Object result = transformFunctionEvaluator.evaluate(record); record.putValue(column, result); } } return record; }
ExpressionTransformer implements RecordTransformer { @Override public GenericRow transform(GenericRow record) { for (Map.Entry<String, FunctionEvaluator> entry : _expressionEvaluators.entrySet()) { String column = entry.getKey(); FunctionEvaluator transformFunctionEvaluator = entry.getValue(); if (record.getValue(column) == null) { Object result = transformFunctionEvaluator.evaluate(record); record.putValue(column, result); } } return record; } }
ExpressionTransformer implements RecordTransformer { @Override public GenericRow transform(GenericRow record) { for (Map.Entry<String, FunctionEvaluator> entry : _expressionEvaluators.entrySet()) { String column = entry.getKey(); FunctionEvaluator transformFunctionEvaluator = entry.getValue(); if (record.getValue(column) == null) { Object result = transformFunctionEvaluator.evaluate(record); record.putValue(column, result); } } return record; } ExpressionTransformer(TableConfig tableConfig, Schema schema); }
ExpressionTransformer implements RecordTransformer { @Override public GenericRow transform(GenericRow record) { for (Map.Entry<String, FunctionEvaluator> entry : _expressionEvaluators.entrySet()) { String column = entry.getKey(); FunctionEvaluator transformFunctionEvaluator = entry.getValue(); if (record.getValue(column) == null) { Object result = transformFunctionEvaluator.evaluate(record); record.putValue(column, result); } } return record; } ExpressionTransformer(TableConfig tableConfig, Schema schema); @Override GenericRow transform(GenericRow record); }
ExpressionTransformer implements RecordTransformer { @Override public GenericRow transform(GenericRow record) { for (Map.Entry<String, FunctionEvaluator> entry : _expressionEvaluators.entrySet()) { String column = entry.getKey(); FunctionEvaluator transformFunctionEvaluator = entry.getValue(); if (record.getValue(column) == null) { Object result = transformFunctionEvaluator.evaluate(record); record.putValue(column, result); } } return record; } ExpressionTransformer(TableConfig tableConfig, Schema schema); @Override GenericRow transform(GenericRow record);  }