[SEDONA-142] Add ST_Collect to Flink Catalog #662
Conversation
| new Functions.ST_MakeValid(), | ||
| new Functions.ST_Intersection(), | ||
| new Functions.ST_Difference(), |
There was a problem hiding this comment.
Can you just make a separate PR for these and add them to the new common module created in #647? Even though they're pretty simple it will help not to maintain both definitions in flink and spark
There was a problem hiding this comment.
Yeah, just saw that #647 got merged earlier this morning, so i have to merge it in my branch anyways. I'll clean it up and make a seperate PR for them.
There was a problem hiding this comment.
I pushed a rebased version with #647 only containing the ST_Collect function.
Apparently #647 broke the build process for me, see: #647 (comment). I did NOT include the fix in my branch/this commit.
Elephantusparvus
force-pushed
the
feature/collect
branch
from
cfb0589
to
6de88bb
Compare
|
Thanks for your contribution. I have a much neat implementation of spatial aggregation in Flink using the UDAF api. It has been there for a while and I just forgot to create a PR. I will create a new PR in one or two days. Then you can follow my implementation to tweak your ST_Collect a bit. |
Sounds good. As stated earlier this really just is a copy/minimal port of flinks built in CollectAggFunction with a different output type (GeometryCollection instead of a MultiSet ) using the AggregateFunction Interface instead of BuiltInAggregateFunction. |
|
Please see PR #672 for the updates |
|
Hi, sorry for the late response. |
Elephantusparvus
force-pushed
the
feature/collect
branch
from
231f98b
to
3c2ee1f
Compare
| /** | ||
| * This is needed because for Resolving the RAW Geometry DataType in ST_Collect (MapView in Accumulator) | ||
| * the serializer needs to be known. To get the registered Serializer the ExecutionEnvironment is needed and | ||
| * to add the UDF the TableEnvironment ist needed. |
There was a problem hiding this comment.
| * to add the UDF the TableEnvironment ist needed. | |
| * to add the UDF the TableEnvironment is needed. |
Did you read the Contributor Guide?
Yes, I have read Contributor Rules and Contributor Development Guide
Yes.
Is this PR related to a JIRA ticket?
[SEDONA-142] Add ST_Collect to Flink Catalog.I actually created to JIRA tickets by accident (SEDONA-141). One could/should be deleted. Sorry for that.
What changes were proposed in this PR?
This is the first attempt to add ST_Collect aggregation to the flink api.
How was this patch tested?
In progress
For myself I only tested aggregations with group by.
I guess it would be necessary to test window/over aggregations as well. Does anyone have specific data/queries for these in mind?
Did this PR include necessary documentation updates?
Currently not.
In advance this pull request is mainly done to discuss if this is a feasible approach to implement ST_Collect.
The PR would have to be cleaned up (split up), tests would need to be added and it would need to be documented.
I basically took the Internal Collect Aggregator from flink-table and adjusted it's types, letting it return a GeometryCollection.
The getTypeInference needs to be overriden otherwise reflection fails for the mapview / selecting the right serializer fails.
One Problem is/was that this function is not instantiated from a BuiltInFunctionDefinition/,
so it is not possible to implement it as BuiltInAggregateFunction and let it pass a SpecializedFunction.SpecializedContext to its constructor. And flinks collect function is constructed through its AggFunctionFactory which also passes the argType.
So I needed to pass the Datatype for the geometry manually when constructing and adding it to the catalog, therefore needing a util function in the registrator.
If needed I add separate prs to ST_MakeValid which is an almost 1-t port/copy from the spark version, and the ST_Intersection, ST_Difference which just wrap the jts locationtech functions.