AGENTS.md

1<!--
2Licensed to the Apache Software Foundation (ASF) under one
3or more contributor license agreements.  See the NOTICE file
4distributed with this work for additional information
5regarding copyright ownership.  The ASF licenses this file
6to you under the Apache License, Version 2.0 (the
7"License"); you may not use this file except in compliance
8with the License.  You may obtain a copy of the License at
9 
10  http://www.apache.org/licenses/LICENSE-2.0
11 
12Unless required by applicable law or agreed to in writing,
13software distributed under the License is distributed on an
14"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15KIND, either express or implied.  See the License for the
16specific language governing permissions and limitations
17under the License.
18-->
19 
20# flink-table-runtime
21 
22Contains classes required by TaskManagers for execution of table programs. Implements runtime operators, built-in functions, and code generation support. Bundles janino (Java compiler for code generation) and flink-shaded-jsonpath.
23 
24## Key Directory Structure
25 
26- `functions/scalar/` — Scalar function implementations (47+)
27- `functions/aggregate/` — Aggregate function implementations
28- `functions/table/` — Table function implementations
29- `functions/ptf/` — Process table function implementations
30- `functions/ml/` — Machine learning function implementations
31- `operators/` — Runtime operators organized by type:
32  - `join/` (hash, sort-merge, lookup, temporal, interval, delta, adaptive, stream/multi-join)
33  - `aggregate/` (group, window)
34  - `window/` (TVF windows, group windows)
35  - `deduplicate/`, `rank/`, `sort/`
36  - `sink/`, `source/`
37  - `correlate/` (including `async/` for async table functions)
38  - `calc/`, `match/`, `over/`, `process/`, `ml/`, `search/`
39 
40## Common Change Patterns
41 
42### Adding a built-in function
43 
44Base classes by function type:
45 
46- **Scalar:** Extend `BuiltInScalarFunction` in `functions/scalar/`
47- **Table:** Extend `BuiltInTableFunction` in `functions/table/`
48- **Aggregate:** Extend `BuiltInAggregateFunction` in `functions/aggregate/`
49- **Process Table Function:** Extend `BuiltInProcessTableFunction` in `functions/ptf/`
50 
51All are constructed from `BuiltInFunctionDefinition#specialize(SpecializedContext)` and work on internal data structures by default.
52 
53Some functions also require custom code generators in the planner (e.g., `JsonCallGen.scala` for JSON functions). Simple scalar functions typically don't need planner changes; the planner handles them uniformly through the function definition.
54 
55### Adding a runtime operator
56 
57- **1 or 2 inputs:** Extend `TableStreamOperator<RowData>` (which extends `AbstractStreamOperator<OUT>`) and implement `OneInputStreamOperator<RowData, RowData>` or `TwoInputStreamOperator<RowData, RowData, RowData>`
58- **3+ inputs:** Extend `AbstractStreamOperatorV2` and implement `MultipleInputStreamOperator` (see `StreamingMultiJoinOperator`)
59- `TableStreamOperator` provides watermark tracking (`currentWatermark`), memory size computation, and a `ContextImpl` for timer services
60 
61### Async operators and runners
62 
63- **Key-ordered async execution:** `operators/join/lookup/keyordered/` contains async execution controller infrastructure (`AecRecord`, `Epoch`, `EpochManager`, `KeyAccountingUnit`, `RecordsBuffer`) for ordering guarantees in async lookup joins
64- **Async correlate:** `operators/correlate/async/` for async table function support
65- **Runner abstraction:** `AbstractFunctionRunner` and `AbstractAsyncFunctionRunner` provide base classes for code-generated function invocations (used by lookup join, ML predict, vector search runners)
66 
67### State serializer migrations
68 
69- When modifying state serializers, create a `TypeSerializerSnapshot` with version bumping
70- Migration test resources follow naming: `migration-flink-<version>-<backend>-<variant>-snapshot`
71- Rescaling tests verify state redistribution across parallelism changes (see `SinkUpsertMaterializerMigrationTest`, `SinkUpsertMaterializerRescalingTest`)
72 
73## Testing Patterns
74 
75- **Harness tests:** Use `OneInputStreamOperatorTestHarness<RowData, RowData>` with `RowDataHarnessAssertor` for output validation. See `operators/join/LookupJoinHarnessTest.java` as a reference.
76- **Test utilities:** `StreamRecordUtils.insertRecord()` for test records, `RowDataHarnessAssertor` for assertions
77- **Operator test base classes:** Module has dedicated base classes per operator type (e.g., `TemporalTimeJoinOperatorTestBase`, `Int2HashJoinOperatorTestBase`, `WindowAggOperatorTestBase`)
78- **State migration tests:** Use snapshot files per Flink version and state backend type to verify forward/backward compatibility
79