Write daily IRs for incremental aggregation #998
Conversation
| s"${aggregationPart.inputColumn}_$opSuffix${aggregationPart.window.suffix}${bucketSuffix}" | ||
|
|
||
| def incOutputColumnName = | ||
| s"${aggregationPart.inputColumn}_$opSuffix${bucketSuffix}" |
There was a problem hiding this comment.
should we still keep the aggregationPart.window.suffix? Otherwise, how do we reconstruct the final output column?
There was a problem hiding this comment.
@pengyu-hou not sure I get it. I can not use the window.suffix right as the intermediate incremental is daily aggregation.
| .toArray | ||
| .zip(columnAggregators.map(_.irType)) | ||
|
|
||
| val incSchema = aggregationParts |
There was a problem hiding this comment.
I am thinking that we should use a full incremental in the code and we can keep inc as the suffix for the table so the table names are not getting too long. What do you think?
| val incSchema = aggregationParts | |
| val incrementalSchema = aggregationParts |
| def cleanName: String = metaData.name.sanitize | ||
|
|
||
| def outputTable = s"${metaData.outputNamespace}.${metaData.cleanName}" | ||
| def incrementalOutputTable = s"${metaData.outputNamespace}.${metaData.cleanName}_inc" |
There was a problem hiding this comment.
| def incrementalOutputTable = s"${metaData.outputNamespace}.${metaData.cleanName}_inc" | |
| def incrementalOutputTable = s"${metaData.outputNamespace}.${metaData.cleanName}_daily_inc" |
There was a problem hiding this comment.
@abbywh I think we already have a daily here.
| } | ||
| } | ||
|
|
||
| class AverageIR extends SimpleAggregator[Array[Any], Array[Any], Double] { |
There was a problem hiding this comment.
It looks to me like this mostly exists to serialize aggregators. Is there something more generic we can do? @nikhilsimha and I talked about https://fory.apache.org/ for example, but it could be even simpler to moving this logic into a different class abstraction
There was a problem hiding this comment.
I agree that if we could somehow make it more general, then we would not have to implement for each operation.
There was a problem hiding this comment.
I was thinking of sticking with Avro because it is what we used for the online serving path so that we can keep the ser//de logic same.
There was a problem hiding this comment.
I think the issue with Avro is that it needs to schematize the object, whereas hopIR is just an Array[Any]. If you just reflect the hopIR, there is no need to schematize it.
3 participants
Summary
Why / Goal
Test Plan
Checklist
Reviewers