Repository navigation
Add collect/merge_map aggregations and binary merge optimizations. - #2388
platinumhamburg wants to merge 1 commit into
Conversation
Extend tests and docs to cover new functions and array/map utilities.
69ca1b2 to
4520c55
Compare
|
Hi @fresh-borzoni, would you have time to review this PR? It adds new aggregate functions to the aggregation merge engine. Given your work on last_n, your feedback on the function semantics, edge cases, and fit with the existing aggregation framework would be especially helpful. Thanks! |
| return null; | ||
| } | ||
|
|
||
| if (accumulator == null) { |
There was a problem hiding this comment.
With distinct=true, the first write for a key is stored as-is, so duplicates survive until the next update. Maybe override requiresAggOnFirstWrite() to return distinct, and skip this shortcut when distinct is on?
| collectDistinctBinary(collection, seen, seenNull, inputArray); | ||
| return new GenericArray(collection.toArray()); | ||
| } else { | ||
| Set<Object> seen = new HashSet<>(capacity); |
There was a problem hiding this comment.
HashSet doesn't keep insertion order, so the array can be reshuffled on every update. Consumers would see wrong changes. The other two paths keep insertion order. Maybe use a LinkedHashSet?
| for (int i = 0; i < keyArray.size(); i++) { | ||
| Object key = keyGetter.getElementOrNull(keyArray, i); | ||
| Object value = valueGetter.getElementOrNull(valueArray, i); | ||
| map.put(key, value); |
There was a problem hiding this comment.
For BYTES/BINARY keys, byte[] is compared by reference, so the same key can appear twice in the merged map. One option is to wrap the keys the way collect does. Another is to reject binary key types in the factory.
|
|
||
| protected final DataType fieldType; | ||
| protected final DataTypeRoot typeRoot; | ||
| protected BinaryRowFormat rowFormat = BinaryRowFormat.COMPACTED; |
There was a problem hiding this comment.
This adds mutable state to the base class, and only collect uses it. Do we have numbers showing the direct binary concat helps? If the gain is small, returning a GenericArray would let us remove this field and the setter loop in AggregationContext. If we keep it, passing the format through the factory or constructor might be cleaner.
| * @return the parsed boolean value or defaultValue if missing | ||
| * @throws IllegalArgumentException if the parameter value is not "true" or "false" | ||
| */ | ||
| public boolean getBooleanParameter(String key, boolean defaultValue) { |
There was a problem hiding this comment.
Parameters are already validated at table creation through validateParameter, so this strict check repeats that work. Maybe the factory could parse the value directly, as last_n does, and we avoid adding a new public method here?
| ) WITH ( | ||
| 'table.merge-engine' = 'aggregation', | ||
| 'fields.tags.agg' = 'collect', | ||
| 'fields.tags.distinct' = 'true' |
There was a problem hiding this comment.
I thing the parser expects fields.<col>.<func>.<param>, so this would be fields.tags.collect.distinct?
| Collects array elements into a single array. | ||
|
|
||
| - **Supported Data Types**: `ARRAY<T>` | ||
| - **Behavior**: Concatenates array elements; with `distinct=true`, duplicates are removed |
There was a problem hiding this comment.
It might be worth adding a note like the one on last_n about write and changelog amplification, since collect has no size limit. An optional max-size parameter could also be worth considering.
| // =================================================================================== | ||
|
|
||
| @Test | ||
| void testCollectAggregationNonDistinct() { |
There was a problem hiding this comment.
It would be great to have tests for the issues above. Coverage for non-primitive element types would also help. A Flink ITCase, like the last_n one, might also help
| } | ||
|
|
||
| @Override | ||
| public Object aggReversed(Object accumulator, Object inputField) { |
There was a problem hiding this comment.
Nit: I think a reversed collect should put the input first. The base default already does that, so could we drop this override?
|
|
||
| private InternalArrayUtils() {} | ||
|
|
||
| public static Object[] toObjectArray(InternalArray array, DataType elementType) { |
There was a problem hiding this comment.
This method and InternalMapUtils seem to be used only by tests. Should they live in test utils instead of fluss-common main?
|
Thanks @platinumhamburg. I've added a few comments and suggestions. 👍 |
Extend tests and docs to cover new functions and array/map utilities.
Purpose
Linked issue: close #2387
Brief change log
Tests
API and Format
Documentation