Support for derived columns in realtime and offline

[kishoreg]

Objective: Pinot does not have the ability to generate columns based on other columns. The workaround has been for the users to pre-compute this as part of the input data. This adds additional processing - hadoop job in case of offline and a stream processing job in case of realtime.

Solution: Pinot currently supports this for time spec. The TimeFieldSpec allows one to specify the incoming field spec and outgoing field spec. We can generalize this other fields types as well (metric and dimensions). e.g. generate an HLL value from userId (derived metric), generate country from IP (derived dimension).

Option 1
Introduce a new FieldSpec in the schema - DerivedFieldSpec. This can look very similar to the existing fieldSpec the fieldType enum that can be DIMENSION, METRIC, TIME.

Option 2
Retain the dimensionFieldSpec, MetricFieldSpec, DateTimeFieldSpec and simply add a boolean isDerived to the base fieldSpec class.

I prefer option 2, since we will just add a simple boolean and rest of the query execution does not need any change.

In addition to above, we need the following to compute the derived column value.

• sourceFieldName: source field from which we can compute the derived field value.
• transformFunction: the function that needs to be applied to the source value to compute the derived value.
• transformFunctionParameters: Parameters to the transform function.

Here again, we have two options to store this configuration - Schema or TableConfig.

Thoughts/comments?

[mayankshriv]

What is the purpose of isDerived flag in option 2?

Could we not just get by with having the transform function info within the individual field specs?

[kishoreg]

Good point! I don't have any strong reason for having isDerived flag in option two. Might just help in validation and making the schema more readable.

[antumbde]

If derived fields are computed dynamically then this is same as supporting UDFs. There shouldn't be a need to update schema or table config.

[mayankshriv]

@antumbde I think the proposal here is to compute them statically (going by the example of time column quoted). But yes, would be good to call it out explicitly.

[mcvsubbu]

I believe we should go with option 1 proposed by Kishore. The (extra) json field elements in the derivedFieldSpec are not there in any other field spec, so it is best not to pollute every fieldspec with the new json elements.

[kishoreg]

@atumbde As Mayank, this is to compute the actual values during ingestion(realtime) and segment generation (offline).

How about the combination of the two. We have derivedFieldSpec only to describe the extra json fields (src, transform function, params) but we will also have the derived fields specified in dimensionFieldSpec and MetricFieldSpec and TimeFieldSpec.

@mcvsubbu Note that we haven't decided where to put the extra attributes - schema vs table config.

[antumbde]

@kishoreg @mayankshriv : I was confused by the objective then -

The workaround has been for the users to pre-compute this as part of the input data. This adds additional processing - hadoop job in case of offline and a stream processing job in case of realtime.

If derived columns are materialized columns there is still extra processing on hadoop and streams but it moves the burden from users to Pinot. Is that right ? (I thought outgoing timefield is computed dynamically based on the spec)

[kishoreg]

That's right. The computation moves from the user to Pinot but overall the amount of resources used will reduce especially in cases where a hadoop job or stream processing job is required just to apply the transformation function.

The outgoing time field is computed and materialized during the ingest time. It's not dynamically calculated during the query execution time.

[kishoreg]

We had an internal discussion and the conclusion was to just add a new field that specified a transformUDF. This function needs to be registered beforehand. In the initial version, we will only support UDF's defined in Pinot and must be loaded and registered at startup. Its easy to extend it to load dynamically.

We will support two types of UDF's - stateless and stateful. Eventually, we want to extend this to use it in query execution as well. But the scope of this PR will be to use it during ingestion (batch/real-time)

• A stateless UDF can be represented as a simple static method in the class.
• A stateful UDF - a separate init method that will be invoked at the beginning with the static arguments. This will avoid parsing the arguments and the setup cost for every invocation.

We will not enforce any interface for the UDF. We will derive most of it from the annotations. I will add sample code shortly.

[Jackie-Jiang]

Solved with #5509