feat: make UDAFs configurable and remove limit on COLLECT_LIST/SET #6851
Conversation
agavra
force-pushed
the
remove_collect_limit
branch
from
6201d8e
to
b35ea2b
Compare
ksqldb-common/src/main/java/io/confluent/ksql/function/AggregateFunctionInitArguments.java
Outdated
Show resolved
Hide resolved
| @Override | ||
| public void configure(final Map<String, ?> map) { | ||
| final Object limit = map.get(LIMIT_CONFIG); | ||
| this.limit = (limit == null) ? this.limit : ((Number) limit).intValue(); |
There was a problem hiding this comment.
good catch, I will treat negative as "no limit". I'll add that to the documentation
ksqldb-engine/src/main/java/io/confluent/ksql/function/udaf/array/CollectSetUdaf.java
Show resolved
Hide resolved
| @@ -67,7 +67,8 @@ | |||
| final List<FunctionCall> aggregations, | |||
| final Optional<WindowExpression> windowExpression, | |||
| final FormatInfo valueFormat, | |||
| final QueryContext.Stacker contextStacker | |||
| final Stacker contextStacker, | |||
| final KsqlConfig config | |||
There was a problem hiding this comment.
This config is not used here, right? I see SchemaKTable is created with ksqlConfig.
There was a problem hiding this comment.
good catch, I used it in SchemaKGroupedTable because I didn't realize it had access to the field ksqlConfig. I removed it
ksqldb-engine/src/test/java/io/confluent/ksql/function/udaf/array/CollectListUdafTest.java
Show resolved
Hide resolved
agavra
force-pushed
the
remove_collect_limit
branch
2 times, most recently
from
82937ff
to
40cc03a
Compare
| @@ -509,6 +509,10 @@ String getName() { | |||
| public static final Set<String> SSL_CONFIG_NAMES = sslConfigNames(); | |||
| public static final Set<String> STREAM_TOPIC_CONFIG_NAMES = streamTopicConfigNames(); | |||
|
|
|||
| public static KsqlConfig empty() { | |||
| return new KsqlConfig(ImmutableMap.of()); | |||
There was a problem hiding this comment.
What happened with the static EMPTY? That could be reused it, isn't it? Perhaps have the variable back and just return it from this method.
There was a problem hiding this comment.
It was causing issues because of circular static dependencies. Some things depended on it in a static block and then things wouldn't load. It was very weird...
ksqldb-engine/src/test/java/io/confluent/ksql/function/udaf/array/CollectListUdafTest.java
Show resolved
Hide resolved
agavra
force-pushed
the
remove_collect_limit
branch
from
40cc03a
to
93b73a3
Compare
Description
fixes #5738
fixes #6711
This commit adds the ability to configure UDAFs the same way that UDFs can be configured, by implementing the
Configurableinterface. By doing this, we can lift the restriction on the number of elements thatCOLLECT_LISTandCOLLECT_SEThave without opening the flood gates in the confluent cloud offering (we can tune that number later).The "meat" of the PR is the on-liner in
UdafFactoryInvoker.javaand the rest is plumbing to make sure that we can get theKsqlConfigdown to that level.Testing done
Reviewer checklist