# flink-table-runtime Contains 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. ## Key Directory Structure - `functions/scalar/` — Scalar function implementations (47+) - `functions/aggregate/` — Aggregate function implementations - `functions/table/` — Table function implementations - `functions/ptf/` — Process table function implementations - `functions/ml/` — Machine learning function implementations - `operators/` — Runtime operators organized by type: - `join/` (hash, sort-merge, lookup, temporal, interval, delta, adaptive, stream/multi-join) - `aggregate/` (group, window) - `window/` (TVF windows, group windows) - `deduplicate/`, `rank/`, `sort/` - `sink/`, `source/` - `correlate/` (including `async/` for async table functions) - `calc/`, `match/`, `over/`, `process/`, `ml/`, `search/` ## Common Change Patterns ### Adding a built-in function Base classes by function type: - **Scalar:** Extend `BuiltInScalarFunction` in `functions/scalar/` - **Table:** Extend `BuiltInTableFunction` in `functions/table/` - **Aggregate:** Extend `BuiltInAggregateFunction` in `functions/aggregate/` - **Process Table Function:** Extend `BuiltInProcessTableFunction` in `functions/ptf/` All are constructed from `BuiltInFunctionDefinition#specialize(SpecializedContext)` and work on internal data structures by default. Some 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. ### Adding a runtime operator - **1 or 2 inputs:** Extend `TableStreamOperator` (which extends `AbstractStreamOperator`) and implement `OneInputStreamOperator` or `TwoInputStreamOperator` - **3+ inputs:** Extend `AbstractStreamOperatorV2` and implement `MultipleInputStreamOperator` (see `StreamingMultiJoinOperator`) - `TableStreamOperator` provides watermark tracking (`currentWatermark`), memory size computation, and a `ContextImpl` for timer services ### Async operators and runners - **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 - **Async correlate:** `operators/correlate/async/` for async table function support - **Runner abstraction:** `AbstractFunctionRunner` and `AbstractAsyncFunctionRunner` provide base classes for code-generated function invocations (used by lookup join, ML predict, vector search runners) ### State serializer migrations - When modifying state serializers, create a `TypeSerializerSnapshot` with version bumping - Migration test resources follow naming: `migration-flink----snapshot` - Rescaling tests verify state redistribution across parallelism changes (see `SinkUpsertMaterializerMigrationTest`, `SinkUpsertMaterializerRescalingTest`) ## Testing Patterns - **Harness tests:** Use `OneInputStreamOperatorTestHarness` with `RowDataHarnessAssertor` for output validation. See `operators/join/LookupJoinHarnessTest.java` as a reference. - **Test utilities:** `StreamRecordUtils.insertRecord()` for test records, `RowDataHarnessAssertor` for assertions - **Operator test base classes:** Module has dedicated base classes per operator type (e.g., `TemporalTimeJoinOperatorTestBase`, `Int2HashJoinOperatorTestBase`, `WindowAggOperatorTestBase`) - **State migration tests:** Use snapshot files per Flink version and state backend type to verify forward/backward compatibility