Prajwal-banakar opened a new pull request, #3398:
URL: https://github.com/apache/fluss/pull/3398
<!--
*Thank you very much for contributing to Fluss - we are happy that you want
to help us improve Fluss. To help the community review your contribution in the
best possible way, please go through the checklist below, which will get the
contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub
issue](https://github.com/apache/fluss/issues). Exceptions are made for typos
in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull
request", where *[component]* should be replaced by the name of the component
being changed. Typically, this corresponds to the component label assigned to
the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are
unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the
pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean
verify` passes.
- Each pull request should address only one issue, not mix up code from
multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were
used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: [Tool Name and Version] following [the
guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: Part of #3289
This PR adds the three Phase 1 bitmap aggregate functions and registers them
as built-in catalog functions in FlussCatalog, building on the infrastructure
from PR #3319
After `USE CATALOG fluss_catalog`, users can call these functions directly
in Flink SQL without any `CREATE TEMPORARY FUNCTION` statement.
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how
they address the issue -->
**New files in `fluss-flink/fluss-flink-common`:**
- `RbBuildAggFunction.java` — `rb_build_agg(INT) -> BYTES`. Aggregates a
stream of 32-bit integers into a serialized RoaringBitmap. Extends
`AbstractRbAggFunction` from PR #3319.
- `RbOrAggFunction.java` — `rb_or_agg(BYTES) -> BYTES`. Unions multiple
serialized RoaringBitmaps via bitwise OR across rows. Extends
`AbstractRbAggFunction`.
- `RbAndAggFunction.java` — `rb_and_agg(BYTES) -> BYTES`. Intersects
multiple serialized RoaringBitmaps via bitwise AND across rows. Unlike the
OR/build variants, AND requires a custom `Accumulator` POJO with an
`initialized` flag because an empty bitmap cannot serve as a "not yet
initialized" sentinel — once the AND result becomes empty, it must remain
empty. The inner `AccumulatorSerializer` follows the same pattern as
`RoaringBitmapSerializer`.
**Modified file: `FlinkCatalog.java`**
Overrides `listFunctions()`, `functionExists()`, and `getFunction()` to
serve the three aggregate functions as built-in catalog functions via a static
`BUILTIN_BITMAP_FUNCTIONS` map using `CatalogFunctionImpl` from
`flink-table-api-java`.
**Test file: `RbAggFunctionsTest.java`**
Unit tests covering:
- `rb_build_agg`: basic accumulation, null handling, duplicate
deduplication, merge, retract throws
- `rb_or_agg`: union correctness, null/empty input handling, merge
- `rb_and_agg`: intersection correctness, null inputs ignored, empty
intersection returns null, merge, reset, Accumulator serializer round-trip
**Catalog registration verified in
`FlinkCatalogTest.testBitmapFunctionsRegistered`**
Note: `AccumulatorSerializer` in `RbAndAggFunction` uses
`TypeSerializerSingleton`, which is deprecated in Flink 1.20 but is the same
approach used in `RoaringBitmapSerializer` (PR #3319, reviewed and approved).
This will be revisited when upgrading to Flink 2.x.
The scalar functions (rb_cardinality, rb_build, rb_contains, rb_to_array,
rb_or, rb_and) will follow in the next PR.
### Tests
<!-- List UT and IT cases to verify this change -->
Unit tests: `RbAggFunctionsTest` — 14 tests, all passing.
Catalog integration: `FlinkCatalogTest.testBitmapFunctionsRegistered` and
`testViewsAndFunctions` — verified against a live Fluss cluster.
Verified locally:
- `./mvnw spotless:apply` — 0 violations
- `./mvnw test -pl fluss-flink/fluss-flink-common
-Dtest="BitmapUtilsTest,RbAggFunctionsTest,RoaringBitmapSerializerTest,AbstractRbAggFunctionITCase,FlinkCatalogTest"`
— BUILD SUCCESS
Full test run: Tests run: 68, Failures: 0, Errors: 0, Skipped: 0
### API and Format
<!-- Does this change affect API or storage format -->
This change does not affect any storage format or wire protocol. The
functions operate on existing `BYTES` columns using the standard RoaringBitmap
serialization format already used by the server-side `FieldRoaringBitmap32Agg`.
### Documentation
<!-- Does this change introduce a new feature -->
No user-facing documentation yet. Documentation will be added after the
scalar functions are implemented and the full Phase 1 function set is available.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
