Two files, one repository
apache/flink ships 1 format across 3 indexed files. The question worth asking is whether the second one says anything the first does not.
| Dimension | Shared | Only in A | Only in B | Overlap |
|---|---|---|---|---|
| Sections | 1 | 38 | 13 | 2% |
| Commands | 0 | 0 | 0 | — |
| Section tags | 5 | 8 | 0 | 38% |
What each file covers
Sections
1 shared · 38 only in A · 13 only in B- − Flink AI Agent Instructions
- − Prerequisites
- − Commands
- − Build
- − Testing
- − Code Quality
- − Repository Structure
- − Core Infrastructure
- − SQL / Table API (recommended API for most users)
- − DataStream API (original streaming API)
- − DataStream API v2 (newer event-driven API)
- − Connectors (in-tree)
- − Formats
- − State Backends
- − File Systems
- − Queryable State
- − Deployment
- − Metrics
- − Libraries
- − Other
- − Architecture Boundaries
- − Adding a new SQL built-in function
- − Adding a new configuration option
- − Adding a new table operator (e.g., join type, aggregate)
- − Adding a new connector (Source or Sink)
- − Modifying state serializers
- − Introducing or changing user-facing APIs (`@Public`, `@PublicEvolving`, `@Experimental`)
- − Coding Standards
- − Testing Standards
- − Commits and PRs
- − Commit message format
- − Pull request conventions
- − AI-assisted contributions
- − Code Review Guidelines
- − Boundaries
- − Ask first
- − Never
- − References
- + flink-table-planner
- + Build Commands
- + Key Directory Structure
- + Key Abstractions
- + Adding a new table operator
- + Adding a planner optimization rule
- + Extending SQL syntax
- + Code generation changes
- + Plan serialization changes
- + ExecNode versioning
- + Configuration options
- + PTF conditional traits
- + Testing Patterns
- Common Change Patterns
Commands
neither file has anySection tags
5 shared · 8 only in A · 0 only in B- − setup
- − lint-format
- − types
- − git-pr
- − dependencies
- − api
- − do-not
- − agent-behaviour
- build
- test
- code-style
- architecture
- database
Line diff
apache/flink · AGENTS.md
@@ −17 @@
17under the License.
18-->
19
20# Flink AI Agent Instructions
21
22This file provides guidance for AI coding agents working with the Apache Flink codebase.
23
24## Prerequisites
25
26- Java 11, 17 (default), or 21. Java 11 syntax must be used in all modules. Java 17 syntax (records, sealed classes, pattern matching) is only permitted in the `flink-tests-java17` module.
27- Maven 3.8.6 (Maven wrapper `./mvnw` included; prefer it)
28- Git
29- Unix-like environment (Linux, macOS, WSL, Cygwin)
30
31## Commands
32
33### Build
34
35- Fast dev build: `./mvnw clean install -DskipTests -Dfast -Pskip-webui-build -T1C`
36- Full build (Java 17 default): `./mvnw clean package -DskipTests -Djdk17 -Pjava17-target`
37- Java 11: `./mvnw clean package -DskipTests -Djdk11 -Pjava11-target`
38- Java 21: `./mvnw clean package -DskipTests -Djdk21 -Pjava21-target`
39- Full build with tests: `./mvnw clean verify`
40- Single module: `./mvnw clean package -DskipTests -pl flink-core-api`
41- Single module with tests: `./mvnw clean verify -pl flink-core-api`
42
43### Testing
44
45- Single test class: `./mvnw -pl flink-core-api -Dtest=MemorySizeTest test`
46- Single test method: `./mvnw -pl flink-core-api -Dtest=MemorySizeTest#testParseBytes test`
47
48### Code Quality
49
50- Format code (Java + Scala): `./mvnw spotless:apply`
51- Check formatting: `./mvnw spotless:check`
52- Checkstyle: `./mvnw checkstyle:check -T1C`
53- Checkstyle config: `tools/maven/checkstyle.xml`
54
55## Repository Structure
56
57Every module from the root pom.xml, organized by function. Flink provides three main user-facing APIs (recommended in this order: SQL, Table API, DataStream API) plus a newer DataStream v2 API.
58
59### Core Infrastructure
60
61- `flink-annotations` — Stability annotations (`@Public`, `@PublicEvolving`, `@Internal`, `@Experimental`) and `@VisibleForTesting`
62- `flink-core-api` — Core API interfaces (functions, state, types) shared by all APIs
63- `flink-core` — Core implementation (type system, serialization, memory management, configuration)
64- `flink-runtime` — Distributed runtime (JobManager, TaskManager, scheduling, network, state)
65- `flink-clients` — CLI and client-side job submission
66- `flink-rpc/` — RPC framework
67 - `flink-rpc-core` — RPC interfaces
68 - `flink-rpc-akka`, `flink-rpc-akka-loader` — Pekko-based RPC implementation
69
70### SQL / Table API (recommended API for most users)
71
72- `flink-table/`
73 - `flink-sql-parser` — SQL parser (extends Calcite SQL parser)
74 - `flink-table-common` — Shared types, descriptors, catalog interfaces
75 - `flink-table-api-java` — Table API for Java
76 - `flink-table-api-scala` — Table API for Scala
77 - `flink-table-api-bridge-base`, `flink-table-api-java-bridge`, `flink-table-api-scala-bridge` — Bridges between Table and DataStream APIs
78 - `flink-table-api-java-uber` — Uber JAR for Table API
79 - `flink-table-planner` — SQL/Table query planning and optimization (Calcite-based)
80 - `flink-table-planner-loader`, `flink-table-planner-loader-bundle` — Classloader isolation for planner
81 - `flink-table-runtime` — Runtime operators for Table/SQL queries
82 - `flink-table-calcite-bridge` — Bridge to Apache Calcite
83 - `flink-sql-gateway-api`, `flink-sql-gateway` — SQL Gateway for remote SQL execution
84 - `flink-sql-client` — Interactive SQL CLI
85 - `flink-sql-jdbc-driver`, `flink-sql-jdbc-driver-bundle` — JDBC driver for SQL Gateway
86 - `flink-table-code-splitter` — Code generation utilities
87 - `flink-table-test-utils` — Test utilities for Table/SQL
88
89### DataStream API (original streaming API)
90
91- `flink-streaming-java` — DataStream API and stream processing operator implementations
92
93### DataStream API v2 (newer event-driven API)
94
95- `flink-datastream-api` — DataStream v2 API definitions
96- `flink-datastream` — DataStream v2 API implementation
97
98### Connectors (in-tree)
99
100- `flink-connectors/`
101 - `flink-connector-base` — Base classes for source/sink connectors
102 - `flink-connector-files` — Unified file system source and sink
103 - `flink-connector-datagen` — DataGen source for testing
104 - `flink-connector-datagen-test` — Tests for DataGen connector
105 - `flink-hadoop-compatibility` — Hadoop InputFormat/OutputFormat compatibility
106 - `flink-file-sink-common` — Common file sink utilities
107- Most connectors (Kafka, JDBC, Elasticsearch, etc.) live in separate repos under [github.com/apache](https://github.com/apache); see README.md for the full list
108
109### Formats
110
111- `flink-formats/`
112 - `flink-json`, `flink-csv`, `flink-avro`, `flink-parquet`, `flink-orc`, `flink-protobuf` — Serialization formats
113 - `flink-avro-confluent-registry` — Avro with Confluent Schema Registry
114 - `flink-sequence-file`, `flink-compress`, `flink-hadoop-bulk`, `flink-orc-nohive` — Hadoop-related formats
115 - `flink-format-common` — Shared format utilities
116 - `flink-sql-json`, `flink-sql-csv`, `flink-sql-avro`, `flink-sql-parquet`, `flink-sql-orc`, `flink-sql-protobuf` — SQL-layer format integrations
117 - `flink-sql-avro-confluent-registry` — SQL-layer Avro with Confluent Schema Registry
118
119### State Backends
120
121- `flink-state-backends/`
122 - `flink-statebackend-rocksdb` — RocksDB state backend
123 - `flink-statebackend-forst` — ForSt state backend (experimental; a fork of RocksDB)
124 - `flink-statebackend-heap-spillable` — Heap-based spillable state backend
125 - `flink-statebackend-changelog` — Changelog state backend
126 - `flink-statebackend-common` — Shared state backend utilities
127- `flink-dstl/flink-dstl-dfs` — State changelog storage (DFS-based persistent changelog for incremental checkpointing)
128
129### File Systems
130
131- `flink-filesystems/`
132 - `flink-hadoop-fs` — Hadoop FileSystem abstraction
133 - `flink-s3-fs-native`, `flink-s3-fs-hadoop`, `flink-s3-fs-presto`, `flink-s3-fs-base` — S3 file systems
134 - `flink-oss-fs-hadoop` — Alibaba OSS
135 - `flink-azure-fs-hadoop` — Azure Blob Storage
136 - `flink-gs-fs-hadoop` — Google Cloud Storage
137 - `flink-fs-hadoop-shaded` — Shaded Hadoop dependencies
138
139### Queryable State
140
141- `flink-queryable-state/`
142 - `flink-queryable-state-runtime` — Server-side queryable state service
143 - `flink-queryable-state-client-java` — Client for querying operator state from running jobs
144
145### Deployment
146
147- `flink-kubernetes` — Kubernetes integration
148- `flink-yarn` — YARN integration
149- `flink-dist`, `flink-dist-scala` — Distribution packaging
150- `flink-container` — Container entry-point and utilities for containerized deployments
151
152### Metrics
153
154- `flink-metrics/`
155 - `flink-metrics-core` — Metrics API and core implementation
156 - Reporter implementations: `flink-metrics-jmx`, `flink-metrics-prometheus`, `flink-metrics-datadog`, `flink-metrics-statsd`, `flink-metrics-graphite`, `flink-metrics-influxdb`, `flink-metrics-slf4j`, `flink-metrics-dropwizard`, `flink-metrics-otel`
157
158### Libraries
159
160- `flink-libraries/`
161 - `flink-cep` — Complex Event Processing
162 - `flink-state-processing-api` — Offline state access (savepoint reading/writing)
163
164### Other
165
166- `flink-models` — AI model integration (sub-modules: `flink-model-openai`, `flink-model-triton`)
167- `flink-python` — PyFlink (Python API)
168- `flink-runtime-web` — Web UI for JobManager dashboard
169- `flink-external-resources` — External resource management (e.g., GPU)
170- `docs/` — Documentation content (Hugo site). This is where user-facing docs are written.
171- `flink-docs` — Documentation build module (auto-generated config reference docs)
172- `flink-examples` — Example programs
173- `flink-quickstart` — Maven archetype for new projects
174- `flink-walkthroughs` — Tutorial walkthrough projects
175
176### Testing
177
178- `flink-tests` — Integration tests
179- `flink-end-to-end-tests` — End-to-end tests
180- `flink-test-utils-parent` — Test utility classes
181- `flink-yarn-tests` — YARN-specific tests
182- `flink-fs-tests` — FileSystem tests
183- `flink-architecture-tests` — ArchUnit architectural boundary tests
184- `tools/ci/flink-ci-tools` — CI tooling
185
186## Architecture Boundaries
187
1881. **Client** submits jobs to the cluster. Submission paths include the CLI (`bin/flink run` via `flink-clients`), the SQL Client (`bin/sql-client.sh` via `flink-sql-client`), the SQL Gateway (`flink-sql-gateway`, also accessible via JDBC driver), the REST API (direct HTTP to JobManager), programmatic execution (`StreamExecutionEnvironment.execute()` or `TableEnvironment.executeSql()`), and PyFlink (`flink-python`, wraps the Java APIs).
1892. **JobManager** (`flink-runtime`) orchestrates execution: receives jobs, creates the execution graph, manages scheduling, coordinates checkpoints, and handles failover. Never runs user code directly.
1903. **TaskManager** (`flink-runtime`) executes the user's operators in task slots. Manages network buffers, state backends, and I/O.
1914. **Table Planner** (`flink-table-planner`) translates SQL/Table API programs into DataStream programs. The planner is loaded in a separate classloader (`flink-table-planner-loader`) to isolate Calcite dependencies.
1925. **Connectors** communicate with external systems. Source connectors implement the `Source` API (FLIP-27); sinks implement the `Sink` API (package `sink2`). Most connectors are externalized to separate repositories.
1936. **State Backends** persist keyed state and operator state. RocksDB is the primary backend for production use.
1947. **Checkpointing** provides exactly-once guarantees. The JobManager coordinates barriers through the data stream; TaskManagers snapshot local state to a distributed file system.
195
196Key separations:
197
198- **Planner vs Runtime:** The table planner generates code and execution plans; the runtime executes them. Changes to planning logic live in `flink-table-planner`; changes to runtime operators live in `flink-table-runtime` or `flink-streaming-java`.
199- **Codegen vs hand-written operators:** Per-record expression logic (casts, projections, filters, function calls) is generated at planning time by cast rules in `flink-table-planner/.../functions/casting/` and call generators in `flink-table-planner/.../codegen/calls/`, then compiled by Janino into the surrounding operator class. Operators with fixed structure (joins, aggregations, source/sink runtime) are hand-written Java in `flink-table-runtime` or `flink-streaming-java`. New scalar functions usually only need a `BuiltInFunctionDefinitions` entry plus a `BuiltInScalarFunction` subclass - the planner wires up codegen automatically. New cast behaviour or a custom call shape needs a cast rule or call generator.
200- **API vs Implementation:** Public API surfaces (`flink-core-api`, `flink-datastream-api`, `flink-table-api-java`) are separate from implementation modules. API stability annotations control what users can depend on.
201- **ArchUnit enforcement:** `flink-architecture-tests/` contains ArchUnit tests that enforce module boundaries. New violations should be avoided; if unavoidable, follow the freeze procedure in `flink-architecture-tests/README.md`.
202
203## Common Change Patterns
204
205This section maps common types of Flink changes to the modules they touch and the verification they require.
206
207### Adding a new SQL built-in function
208
2091. Register in `flink-table-common` in `BuiltInFunctionDefinitions.java` (definition, input/output type strategies, runtime class reference)
2102. Implement in `flink-table-runtime` under `functions/` (extend the appropriate base class: `BuiltInScalarFunction`, `BuiltInTableFunction`, `BuiltInAggregateFunction`, or `BuiltInProcessTableFunction`)
2113. Add tests in `flink-table-planner` and `flink-table-runtime`
2124. Extend Table API support
2135. Document in `docs/`
2146. See [flink-table/flink-table-planner/AGENTS.md](flink-table/flink-table-planner/AGENTS.md) and [flink-table/flink-table-runtime/AGENTS.md](flink-table/flink-table-runtime/AGENTS.md) for detailed patterns
215
216### Adding a new configuration option
217
2181. Define `ConfigOption<T>` in the relevant config class (e.g., `ExecutionConfigOptions.java` in `flink-table-api-java`)
2192. Use `ConfigOptions.key("table.exec....")` builder with type, default value, and description
2203. Add `@Documentation.TableOption` annotation for auto-generated docs
2214. Document in `docs/` if user-facing
2225. Verify: unit test for default value, ITCase for behavior change
223
224### Adding a new table operator (e.g., join type, aggregate)
225
2261. Involves `flink-table-runtime` (operator), `flink-table-planner` (ExecNode, physical/logical rules), and tests across both
2272. See [flink-table/flink-table-planner/AGENTS.md](flink-table/flink-table-planner/AGENTS.md) and [flink-table/flink-table-runtime/AGENTS.md](flink-table/flink-table-runtime/AGENTS.md) for detailed development order and testing patterns
228
229### Adding a new connector (Source or Sink)
230
2311. Implement the `Source` API (`flink-connector-base`): `SplitEnumerator`, `SourceReader`, `SourceSplit`, serializers (`SimpleVersionedSerializer`)
2322. Or implement the `Sink` API (package `sink2`) for sinks
2333. Most new connectors go in separate repos under `github.com/apache`, not in the main Flink repo
2344. Verify: unit tests + ITCase with real or embedded external system
235
236### Modifying state serializers
237
2381. Changes to `TypeSerializer` require a corresponding `TypeSerializerSnapshot` for migration
2392. Bump version in `getCurrentVersion()`, handle old versions in `readSnapshot()`
2403. Snapshot must have no-arg constructor for reflection-based deserialization
2414. Implement `resolveSchemaCompatibility()` for upgrade paths
2425. Verify: serializer snapshot migration tests, checkpoint restore tests across versions
243
244### Introducing or changing user-facing APIs (`@Public`, `@PublicEvolving`, `@Experimental`)
245
2461. New user-facing API requires a voted FLIP (Flink Improvement Proposal); this applies to `@Public`, `@PublicEvolving`, and `@Experimental` since users build against all three
2472. Every user-facing API class and method must carry a stability annotation
2483. Changes to existing `@Public` or `@PublicEvolving` API must maintain backward compatibility
2494. `@Internal` APIs can be changed freely; users should not depend on them
2505. Update JavaDoc on the changed class/method
2516. Add to release notes
2527. Verify: ArchUnit tests pass, no new architecture violations
253
254## Coding Standards
255
256- **Format Java files with Spotless immediately after editing:** `./mvnw spotless:apply`. Uses google-java-format with AOSP style.
257- **Scala formatting:** Spotless + scalafmt (config at `.scalafmt.conf`, maxColumn 100).
258- **Checkstyle:** `tools/maven/checkstyle.xml` (version defined in root `pom.xml` as `checkstyle.version`). Some modules (flink-core, flink-optimizer, flink-runtime) are not covered by checkstyle enforcement, but conventions should still be followed.
259- **No new Scala code.** All Flink Scala APIs are deprecated per FLIP-265. Write all new code in Java.
260- **Apache License 2.0 header** required on all new files (enforced by Apache Rat). Use an HTML comment for markdown files.
261- **API stability annotations:** Every user-facing API class and method must have a stability annotation. `@Public` (stable across minor releases), `@PublicEvolving` (may change in minor releases), `@Experimental` (may change at any time). These are all part of the public API surface that users build against. `@Internal` marks APIs with no stability guarantees that users should not depend on.
262- **Logging:** Use parameterized log statements (SLF4J `{}` placeholders), never string concatenation.
263- **No Java serialization** for new features (except internal RPC message transport).
264- **Use `final`** for variables and fields where applicable.
265- **Comments:** Do not add unnecessary comments that restate what the code does. Add comments that explain "the why" where relevant.
266- **Reuse existing code.** Before implementing new utilities or abstractions, search for existing ones in the codebase. Prioritize architecture consistency and code reusability.
267- Full code style guide: https://flink.apache.org/how-to-contribute/code-style-and-quality-preamble/
268
269## Testing Standards
270
271- Add tests for new behavior, covering success, failure, and edge cases.
272- Use **JUnit 5** + **AssertJ** assertions. Do not use JUnit 4 or Hamcrest in new test code.
273- Prefer real test implementations over Mockito mocks where possible.
274- **Integration tests:** Name classes with `ITCase` suffix (e.g., `MyFeatureITCase.java`).
275- **Red-green verification:** For bug fixes, verify that new tests actually fail without the fix before confirming they pass with it.
276- **Test location** mirrors source structure within each module.
277- Follow the testing conventions at https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#7-testing
278
279## Commits and PRs
280
281### Commit message format
282
283- `[FLINK-XXXX][component] Description` where FLINK-XXXX is the JIRA issue number
284- `[hotfix][component] Description` for typo fixes without JIRA
285- Each commit must have a meaningful message including the JIRA ID. If you don't know the ticket number, ask.
286- Separate cleanup/refactoring from functional changes into distinct commits
287- When AI tools were used: add `Generated-by: <Tool Name and Version>` trailer per [ASF generative tooling guidance](https://www.apache.org/legal/generative-tooling.html)
288
289### Pull request conventions
290
291- Title format: `[FLINK-XXXX][component] Title of the pull request`
292- A corresponding JIRA issue is required (except hotfixes for typos)
293- Fill out the PR template completely but concisely: describe purpose, change log, testing approach, impact assessment
294- Each PR should address exactly one issue
295- Ensure `./mvnw clean verify` passes before opening a PR
296- Always push to your fork, not directly to `apache/flink`
297- Rebase onto the latest target branch before submitting
298- For user-visible behaviour changes, breaking changes, new SQL features, or new config options: fill in the **Release Notes** field on the JIRA ticket. The release manager consolidates these when cutting a release. The next version's `docs/content/release-notes/flink-X.Y.md` will be generated based of the jira tickets, so make sure to fill them in properly.
299
300### AI-assisted contributions
301
302- Disclose AI usage by checking the AI disclosure checkbox and uncommenting the `Generated-by` line in the PR template
303- Add `Generated-by: <Tool Name and Version>` to commit messages
304- Never add `Co-Authored-By` with an AI agent as co-author; agents are assistants, not authors
305- You must be able to explain the design, code, and tests, debug them, and respond to review feedback substantively
306- Reviewer-ready quality bar: the author owns PR quality. PRs that look AI-generated without author refinement (walls of unreviewed prose, scaffolding without behaviour, tests that do not exercise the change, padded commit messages) will be closed without review
307
308## Code Review Guidelines
309
310When reviewing a PR or diff against this repo:
311
312- Look for opportunities to simplify the code, scoped to the diff itself (not pre-existing code outside the change).
313- Flag comments that are obvious (restate what the code already says) or overly verbose.
314- In test code, look for potential flakiness — e.g. `Thread.sleep` used outside a retry/poll loop, or similar timing-dependent, non-deterministic patterns. Where applicable, suggest clock injection (e.g. a manually-advanced `Clock`/`ManualClock`) instead of relying on wall-clock time, or waiting for the actual condition in a loop with a timeout, for deterministic tests.
315- Check that each commit message conforms to Flink conventions: it must start with `[FLINK-XXXX]` or `[hotfix]`, and must specify a subsystem/component (e.g. `[FLINK-XXXX][runtime] Description`).
316- If a change introduces a new feature controlled by a config option/flag, check that the resolved state (enabled/disabled, and the effective value) is logged at INFO level when the feature initializes/activates.
317- Consider whether a change should be hidden behind a feature flag, especially if it's non-trivial (touches core paths, changes default behavior, or is hard to reason about in isolation). This is mandatory if the change is risky (correctness, performance, backward-compatibility, or data-safety risk) and no flag/kill-switch already exists.
318- For changes to configuration options, check if the corresponding documentation has been regenerated (this should be covered by tests but flagging it earlier speeds up development)
319
320## Boundaries
321
322### Ask first
323
324- Adding or changing `@Public`, `@PublicEvolving`, or `@Experimental` annotations (these are user-facing API commitments requiring a FLIP)
325- Large cross-module refactors
326- New dependencies
327- Changes to serialization formats (affects state compatibility)
328- Changes to checkpoint/savepoint behavior
329- Changes that could impact performance on hot paths (per-record processing, serialization, state access)
330
331### Never
332
333- Commit secrets, credentials, or tokens
334- Push directly to `apache/flink`; always work from your fork
335- Mix unrelated changes into one PR
336- Use Java serialization for new features
337- Edit generated files by hand when a generation workflow exists
338- Use the legacy `SourceFunction` or `SinkFunction` interfaces for connectors; use the `Source` API (FLIP-27) and `Sink` API (package `sink2`) instead
339- Add `Co-Authored-By` with an AI agent as co-author in commit messages; AI agents are assistants, not authors. Use `Generated-by: <Tool Name and Version>` instead.
340- Suppress or bypass checkstyle rules (no `CHECKSTYLE:ON`/`CHECKSTYLE:OFF` comments, no adding entries to `tools/maven/suppressions.xml`, no `@SuppressWarnings`). Fix the code to satisfy checkstyle instead.
341- Add, change, or remove classes outside the `org.apache.flink.*` package (for example, classes copied from Calcite)
342- Modify `Parser.jj` (Calcite's generated parser grammar; expected to be removed in future Calcite upgrades)
343- Use destructive git operations unless explicitly requested
344
345## References
346
347- [README.md](README.md) — Build instructions and project overview
348- [DEVELOPMENT.md](DEVELOPMENT.md) — IDE setup and development environment
349- [.github/CONTRIBUTING.md](.github/CONTRIBUTING.md) — Contribution process
350- [.github/PULL_REQUEST_TEMPLATE.md](.github/PULL_REQUEST_TEMPLATE.md) — PR checklist
351- [Code Style Guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-preamble/) — Detailed coding guidelines
352- [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) — AI tooling policy
353
apache/flink · flink-table/flink-table-planner/AGENTS.md
@@ +17 @@
17under the License.
18-->
19
20# flink-table-planner
21
22Translates and optimizes SQL/Table API programs into executable plans using Apache Calcite. Bridges the Table/SQL API and the runtime by generating code and execution plans. The planner is loaded in a separate classloader (`flink-table-planner-loader`) to isolate Calcite dependencies.
23
24See also [README.md](README.md) for Immutables rule config conventions and JSON plan test regeneration.
25
26## Build Commands
27
28Full table modules rebuild:
29
30```
31./mvnw clean install -T1C -DskipTests -Pskip-webui-build -pl flink-table/flink-table-common,flink-table/flink-sql-parser,flink-table/flink-table-planner-loader,flink-table/flink-table-planner,flink-table/flink-table-api-java -am
32```
33
34After the first full build, drop `-am` for faster rebuilds when you're only changing code within these modules.
35
36## Key Directory Structure
37
38- `plan/rules/physical/stream/` and `plan/rules/physical/batch/` — Physical planner rules
39- `plan/rules/logical/` — Logical optimization rules
40- `plan/nodes/exec/stream/` and `plan/nodes/exec/batch/` — ExecNodes (bridge between planner and runtime)
41- `plan/nodes/exec/spec/` — Serializable operator specifications (JoinSpec, WindowSpec, etc.)
42- `plan/nodes/physical/stream/` and `plan/nodes/physical/batch/` — Intermediate physical nodes (Calcite-based)
43- `plan/nodes/logical/` — Logical nodes (Calcite-based)
44- `codegen/` — Code generation
45- `codegen/calls/` — Custom code generators for specific functions (e.g., `JsonCallGen.scala`)
46- `functions/casting/` — Cast rules for code generation (e.g., `BinaryToBinaryCastRule`, `StringToTimeCastRule`)
47- `functions/` — Function management and inference
48- `catalog/` — Catalog integration
49
50## Key Abstractions
51
52- **ExecNode**: Bridge between planner and runtime. Annotated with `@ExecNodeMetadata(name, version, minPlanVersion, minStateVersion)` for versioning and backwards compatibility. Extends `ExecNodeBase<T>` and implements either `StreamExecNode<T>` (streaming) or `BatchExecNode<T>` (batch); `T` is typically `RowData`.
53- **Physical rules**: Extend `RelRule`, use Immutables `@Value.Immutable` for config. Transform logical nodes to physical nodes. Registered in `FlinkStreamRuleSets` and/or `FlinkBatchRuleSets`.
54- **Logical optimization rules**: Also extend `RelRule`, often use `RexShuttle` for expression rewriting. Registered in rule sets.
55- **Specs**: Serializable specifications in `plan/nodes/exec/spec/` (JoinSpec, WindowSpec, etc.) that carry operator configuration.
56
57## Common Change Patterns
58
59### Adding a new table operator
60
61Components involved (can be developed top-down or bottom-up):
62
631. **Runtime operator** in `flink-table-runtime` under `operators/` (extend `TableStreamOperator`, implement `OneInputStreamOperator` or `TwoInputStreamOperator`). Test with harness tests. See [flink-table-runtime AGENTS.md](../flink-table-runtime/AGENTS.md).
642. **ExecNode** in `plan/nodes/exec/stream/` and/or `plan/nodes/exec/batch/` (extend `ExecNodeBase<T>`; implement `StreamExecNode<T>` for streaming or `BatchExecNode<T>` for batch; annotate with `@ExecNodeMetadata`; `T` is typically `RowData`)
653. **Physical Node + Physical Rules** in `plan/rules/physical/stream/` and/or `plan/rules/physical/batch/` (physical rules usually extend `ConverterRule` via `Config.INSTANCE.withConversion(...)`; same-convention rewrites extend `RelRule` with an `@Value.Immutable` config)
664. **Logical Node + Planner rule**
675. Tests: semantic tests, plan tests, restore tests (if stateful)
68
69Both `stream/` and `batch/` directories exist for rules and ExecNodes. Consider whether your change applies to one or both.
70
71### Adding a planner optimization rule
72
73Pick the base class by what the rule does:
74- Converts a node from one calling convention to another (for example, logical → stream physical): extend `ConverterRule`.
75Call `ConverterRule.Config.INSTANCE.withConversion(...)` in the constructor, do not define your own config.
76- Rewrites nodes within the same convention (logical → logical, physical → physical): extend `RelRule` with an `@Value.Immutable` config.
77Some existing rules still use Calcite's older `RelOptRule`; prefer `RelRule` for new code.
78
79Then:
801. Register in `FlinkStreamRuleSets.scala` and/or `FlinkBatchRuleSets.scala`
812. Plan tests with XML golden files — when the test fails, copy the framework's generated log file over the reference `.xml` (cases are ordered alphabetically by method name)
823. A same-convention rewrite needs no runtime changes. A `ConverterRule` that produces a new physical node also needs the physical node, ExecNode, and runtime operator — see "Adding a new table operator" above.
83
84### Extending SQL syntax
85
861. Modify parser grammar in `flink-sql-parser` (`parserImpls.ftl`)
872. Add operation conversion logic in `SqlNodeToOperationConversion.java`
883. Test with parser tests and SQL gateway integration tests (`.q` files)
89
90### Code generation changes
91
92- Cast rules live in `functions/casting/`. Each extends `AbstractExpressionCodeGeneratorCastRule` or similar.
93- Custom call generators for functions live in `codegen/calls/` (e.g., `JsonCallGen.scala`). Simple scalar functions typically don't need these; the planner handles them uniformly through the function definition.
94- Immutables library is used for rule configs (`@Value.Immutable`, `@Value.Enclosing`). See [README.md](README.md).
95
96### Plan serialization changes
97
98- ExecNode specs use Jackson for JSON serialization. Source/sink specs should use `@JsonIgnoreProperties(ignoreUnknown = true)` for forward compatibility.
99- When adding new ExecNode features, update `RexNodeJsonDeserializer` or related serde classes if new function kinds or types are introduced.
100
101### ExecNode versioning
102
103When bumping an ExecNode version, update the `@ExecNodeMetadata` annotation's `version` and `minPlanVersion`/`minStateVersion` fields. Add restore test snapshots for the new version.
104
105### Configuration options
106
107New features often introduce `ExecutionConfigOptions` entries (in `flink-table-api-java`) for runtime tunability (e.g., cache sizes, timeouts, batch sizes).
108
109### PTF conditional traits
110
111A *conditional trait* lets a PTF's table-argument traits depend on the call site instead of being fixed at declaration. Example for `TO_CHANGELOG`: the `input` argument is row-semantic by default (single stream, no PARTITION BY), but switches to set-semantic when the user writes `PARTITION BY` so the runtime can co-locate state per key. One declaration, two effective signatures depending on the call.
112
113**Declaration.** Built-in functions add conditional rules in `BuiltInFunctionDefinitions` via `StaticArgument.withConditionalTrait(trait, condition)`. The condition (a `TraitCondition`) is a small value-comparable predicate evaluated against a `TraitContext`. Built-in factories live on `TraitCondition` (`hasPartitionBy()`, `argIsEqualTo(name, value)`, `not(c)`); under the hood they wrap into the package-private `BuiltInCondition` so equality cascades correctly through `StaticArgument.equals`.
114
115**Evaluation.** A `TraitCondition` reads two things: whether `PARTITION BY` is present on this table arg, and the literal value of named scalar args. Both come through `TraitContext`. There are two factories: `TraitContext.of(TableSemantics, CallContext, declared)` for the validation side (called from `SystemTypeInference.resolveStaticArgs`) and a planner-side adapter inside `BridgingSqlFunction.buildTraitContext` that sources the same data from a `RexCall` + `RexTableArgCall`. Same logical context, different inputs because the two layers don't share types.
116
117**Resolution.** Three call sites bake conditional traits into the operator's effective signature:
118
1191. **Validation** — `SystemTypeInference.resolveStaticArgs` runs once each from `inferInputTypes` and `inferType`. Twice per validation pass; can't dedupe across Calcite hooks because each gets a different `CallContext` instance.
1202. **Planning** — `BridgingSqlFunction.resolveCallTraits` is called from `FlinkLogicalTableFunctionScan.Converter.convert`. It rewrites the operator on the `RexCall` so all downstream readers see the resolved view via plain `function.getTypeInference().getStaticArguments()`.
1213. **Compiled-plan restore** — `BridgingSqlFunction.resolveCallTraits` is called again from `StreamExecProcessTableFunction.@JsonCreator`, because the JSON path skips the logical converter. Without this hook, restore would silently produce wrong results for any conditional-trait PTF.
122
123The payoff: downstream rules, exec nodes, codegen, and changelog inference all use ordinary `staticArg.is(SET_SEMANTIC_TABLE)` checks. No consumer needs to know that conditional traits exist. Why three sites and not one. The three resolution points exist because they sit in different lifecycles that can't share state.
124
125## Testing Patterns
126
127Choose test types based on what you're changing:
128
129- **Semantic tests** (for ExecNode/operator changes): Use `SemanticTestBase` (streaming) or `BatchSemanticTestBase` (batch) in `plan/nodes/exec/testutils/`. Extends `CommonSemanticTestBase` which implements `TableTestProgramRunner`. Prefer these over ITCase for operators and ExecNodes.
130- **Restore tests** (for stateful operators): Use `RestoreTestBase` or `BatchRestoreTestBase` in `plan/nodes/exec/testutils/`. Implements `TableTestProgramRunner`, uses `@ExtendWith(MiniClusterExtension.class)`. Required when your operator uses state. Tests savepoint creation and job restart in two phases: (1) generate compiled plans + savepoints, (2) verify recovery.
131- **Plan tests** (for optimization rules): Verify the generated execution plan using XML golden files. Used for logical and physical optimization rules.
132- **ITCase** (for built-in functions): Function tests typically use ITCase with `TestSetSpec` for end-to-end verification (e.g., `JsonFunctionsITCase`, `TimeFunctionsITCase`).
133- **JSON plan test regeneration:** Set `PLAN_TEST_FORCE_OVERWRITE=true` environment variable (documented in [README.md](README.md)).
134
@@ −17 +17 @@
1717 under the License.
1818 -->
1919
20−# Flink AI Agent Instructions
20+# flink-table-planner
2121
22−This file provides guidance for AI coding agents working with the Apache Flink codebase.
22+Translates and optimizes SQL/Table API programs into executable plans using Apache Calcite. Bridges the Table/SQL API and the runtime by generating code and execution plans. The planner is loaded in a separate classloader (`flink-table-planner-loader`) to isolate Calcite dependencies.
2323
24−## Prerequisites
24+See also [README.md](README.md) for Immutables rule config conventions and JSON plan test regeneration.
2525
26−- Java 11, 17 (default), or 21. Java 11 syntax must be used in all modules. Java 17 syntax (records, sealed classes, pattern matching) is only permitted in the `flink-tests-java17` module.
27−- Maven 3.8.6 (Maven wrapper `./mvnw` included; prefer it)
28−- Git
29−- Unix-like environment (Linux, macOS, WSL, Cygwin)
26+## Build Commands
3027
31−## Commands
28+Full table modules rebuild:
3229
33−### Build
30+```
31+./mvnw clean install -T1C -DskipTests -Pskip-webui-build -pl flink-table/flink-table-common,flink-table/flink-sql-parser,flink-table/flink-table-planner-loader,flink-table/flink-table-planner,flink-table/flink-table-api-java -am
32+```
3433
35−- Fast dev build: `./mvnw clean install -DskipTests -Dfast -Pskip-webui-build -T1C`
36−- Full build (Java 17 default): `./mvnw clean package -DskipTests -Djdk17 -Pjava17-target`
37−- Java 11: `./mvnw clean package -DskipTests -Djdk11 -Pjava11-target`
38−- Java 21: `./mvnw clean package -DskipTests -Djdk21 -Pjava21-target`
39−- Full build with tests: `./mvnw clean verify`
40−- Single module: `./mvnw clean package -DskipTests -pl flink-core-api`
41−- Single module with tests: `./mvnw clean verify -pl flink-core-api`
34+After the first full build, drop `-am` for faster rebuilds when you're only changing code within these modules.
4235
43−### Testing
36+## Key Directory Structure
4437
45−- Single test class: `./mvnw -pl flink-core-api -Dtest=MemorySizeTest test`
46−- Single test method: `./mvnw -pl flink-core-api -Dtest=MemorySizeTest#testParseBytes test`
38+- `plan/rules/physical/stream/` and `plan/rules/physical/batch/` — Physical planner rules
39+- `plan/rules/logical/` — Logical optimization rules
40+- `plan/nodes/exec/stream/` and `plan/nodes/exec/batch/` — ExecNodes (bridge between planner and runtime)
41+- `plan/nodes/exec/spec/` — Serializable operator specifications (JoinSpec, WindowSpec, etc.)
42+- `plan/nodes/physical/stream/` and `plan/nodes/physical/batch/` — Intermediate physical nodes (Calcite-based)
43+- `plan/nodes/logical/` — Logical nodes (Calcite-based)
44+- `codegen/` — Code generation
45+- `codegen/calls/` — Custom code generators for specific functions (e.g., `JsonCallGen.scala`)
46+- `functions/casting/` — Cast rules for code generation (e.g., `BinaryToBinaryCastRule`, `StringToTimeCastRule`)
47+- `functions/` — Function management and inference
48+- `catalog/` — Catalog integration
4749
48−### Code Quality
50+## Key Abstractions
4951
50−- Format code (Java + Scala): `./mvnw spotless:apply`
51−- Check formatting: `./mvnw spotless:check`
52−- Checkstyle: `./mvnw checkstyle:check -T1C`
53−- Checkstyle config: `tools/maven/checkstyle.xml`
52+- **ExecNode**: Bridge between planner and runtime. Annotated with `@ExecNodeMetadata(name, version, minPlanVersion, minStateVersion)` for versioning and backwards compatibility. Extends `ExecNodeBase<T>` and implements either `StreamExecNode<T>` (streaming) or `BatchExecNode<T>` (batch); `T` is typically `RowData`.
53+- **Physical rules**: Extend `RelRule`, use Immutables `@Value.Immutable` for config. Transform logical nodes to physical nodes. Registered in `FlinkStreamRuleSets` and/or `FlinkBatchRuleSets`.
54+- **Logical optimization rules**: Also extend `RelRule`, often use `RexShuttle` for expression rewriting. Registered in rule sets.
55+- **Specs**: Serializable specifications in `plan/nodes/exec/spec/` (JoinSpec, WindowSpec, etc.) that carry operator configuration.
5456
55−## Repository Structure
56−
57−Every module from the root pom.xml, organized by function. Flink provides three main user-facing APIs (recommended in this order: SQL, Table API, DataStream API) plus a newer DataStream v2 API.
58−
59−### Core Infrastructure
60−
61−- `flink-annotations` — Stability annotations (`@Public`, `@PublicEvolving`, `@Internal`, `@Experimental`) and `@VisibleForTesting`
62−- `flink-core-api` — Core API interfaces (functions, state, types) shared by all APIs
63−- `flink-core` — Core implementation (type system, serialization, memory management, configuration)
64−- `flink-runtime` — Distributed runtime (JobManager, TaskManager, scheduling, network, state)
65−- `flink-clients` — CLI and client-side job submission
66−- `flink-rpc/` — RPC framework
67− - `flink-rpc-core` — RPC interfaces
68− - `flink-rpc-akka`, `flink-rpc-akka-loader` — Pekko-based RPC implementation
69−
70−### SQL / Table API (recommended API for most users)
71−
72−- `flink-table/`
73− - `flink-sql-parser` — SQL parser (extends Calcite SQL parser)
74− - `flink-table-common` — Shared types, descriptors, catalog interfaces
75− - `flink-table-api-java` — Table API for Java
76− - `flink-table-api-scala` — Table API for Scala
77− - `flink-table-api-bridge-base`, `flink-table-api-java-bridge`, `flink-table-api-scala-bridge` — Bridges between Table and DataStream APIs
78− - `flink-table-api-java-uber` — Uber JAR for Table API
79− - `flink-table-planner` — SQL/Table query planning and optimization (Calcite-based)
80− - `flink-table-planner-loader`, `flink-table-planner-loader-bundle` — Classloader isolation for planner
81− - `flink-table-runtime` — Runtime operators for Table/SQL queries
82− - `flink-table-calcite-bridge` — Bridge to Apache Calcite
83− - `flink-sql-gateway-api`, `flink-sql-gateway` — SQL Gateway for remote SQL execution
84− - `flink-sql-client` — Interactive SQL CLI
85− - `flink-sql-jdbc-driver`, `flink-sql-jdbc-driver-bundle` — JDBC driver for SQL Gateway
86− - `flink-table-code-splitter` — Code generation utilities
87− - `flink-table-test-utils` — Test utilities for Table/SQL
88−
89−### DataStream API (original streaming API)
90−
91−- `flink-streaming-java` — DataStream API and stream processing operator implementations
92−
93−### DataStream API v2 (newer event-driven API)
94−
95−- `flink-datastream-api` — DataStream v2 API definitions
96−- `flink-datastream` — DataStream v2 API implementation
97−
98−### Connectors (in-tree)
99−
100−- `flink-connectors/`
101− - `flink-connector-base` — Base classes for source/sink connectors
102− - `flink-connector-files` — Unified file system source and sink
103− - `flink-connector-datagen` — DataGen source for testing
104− - `flink-connector-datagen-test` — Tests for DataGen connector
105− - `flink-hadoop-compatibility` — Hadoop InputFormat/OutputFormat compatibility
106− - `flink-file-sink-common` — Common file sink utilities
107−- Most connectors (Kafka, JDBC, Elasticsearch, etc.) live in separate repos under [github.com/apache](https://github.com/apache); see README.md for the full list
108−
109−### Formats
110−
111−- `flink-formats/`
112− - `flink-json`, `flink-csv`, `flink-avro`, `flink-parquet`, `flink-orc`, `flink-protobuf` — Serialization formats
113− - `flink-avro-confluent-registry` — Avro with Confluent Schema Registry
114− - `flink-sequence-file`, `flink-compress`, `flink-hadoop-bulk`, `flink-orc-nohive` — Hadoop-related formats
115− - `flink-format-common` — Shared format utilities
116− - `flink-sql-json`, `flink-sql-csv`, `flink-sql-avro`, `flink-sql-parquet`, `flink-sql-orc`, `flink-sql-protobuf` — SQL-layer format integrations
117− - `flink-sql-avro-confluent-registry` — SQL-layer Avro with Confluent Schema Registry
118−
119−### State Backends
120−
121−- `flink-state-backends/`
122− - `flink-statebackend-rocksdb` — RocksDB state backend
123− - `flink-statebackend-forst` — ForSt state backend (experimental; a fork of RocksDB)
124− - `flink-statebackend-heap-spillable` — Heap-based spillable state backend
125− - `flink-statebackend-changelog` — Changelog state backend
126− - `flink-statebackend-common` — Shared state backend utilities
127−- `flink-dstl/flink-dstl-dfs` — State changelog storage (DFS-based persistent changelog for incremental checkpointing)
128−
129−### File Systems
130−
131−- `flink-filesystems/`
132− - `flink-hadoop-fs` — Hadoop FileSystem abstraction
133− - `flink-s3-fs-native`, `flink-s3-fs-hadoop`, `flink-s3-fs-presto`, `flink-s3-fs-base` — S3 file systems
134− - `flink-oss-fs-hadoop` — Alibaba OSS
135− - `flink-azure-fs-hadoop` — Azure Blob Storage
136− - `flink-gs-fs-hadoop` — Google Cloud Storage
137− - `flink-fs-hadoop-shaded` — Shaded Hadoop dependencies
138−
139−### Queryable State
140−
141−- `flink-queryable-state/`
142− - `flink-queryable-state-runtime` — Server-side queryable state service
143− - `flink-queryable-state-client-java` — Client for querying operator state from running jobs
144−
145−### Deployment
146−
147−- `flink-kubernetes` — Kubernetes integration
148−- `flink-yarn` — YARN integration
149−- `flink-dist`, `flink-dist-scala` — Distribution packaging
150−- `flink-container` — Container entry-point and utilities for containerized deployments
151−
152−### Metrics
153−
154−- `flink-metrics/`
155− - `flink-metrics-core` — Metrics API and core implementation
156− - Reporter implementations: `flink-metrics-jmx`, `flink-metrics-prometheus`, `flink-metrics-datadog`, `flink-metrics-statsd`, `flink-metrics-graphite`, `flink-metrics-influxdb`, `flink-metrics-slf4j`, `flink-metrics-dropwizard`, `flink-metrics-otel`
157−
158−### Libraries
159−
160−- `flink-libraries/`
161− - `flink-cep` — Complex Event Processing
162− - `flink-state-processing-api` — Offline state access (savepoint reading/writing)
163−
164−### Other
165−
166−- `flink-models` — AI model integration (sub-modules: `flink-model-openai`, `flink-model-triton`)
167−- `flink-python` — PyFlink (Python API)
168−- `flink-runtime-web` — Web UI for JobManager dashboard
169−- `flink-external-resources` — External resource management (e.g., GPU)
170−- `docs/` — Documentation content (Hugo site). This is where user-facing docs are written.
171−- `flink-docs` — Documentation build module (auto-generated config reference docs)
172−- `flink-examples` — Example programs
173−- `flink-quickstart` — Maven archetype for new projects
174−- `flink-walkthroughs` — Tutorial walkthrough projects
175−
176−### Testing
177−
178−- `flink-tests` — Integration tests
179−- `flink-end-to-end-tests` — End-to-end tests
180−- `flink-test-utils-parent` — Test utility classes
181−- `flink-yarn-tests` — YARN-specific tests
182−- `flink-fs-tests` — FileSystem tests
183−- `flink-architecture-tests` — ArchUnit architectural boundary tests
184−- `tools/ci/flink-ci-tools` — CI tooling
185−
186−## Architecture Boundaries
187−
188−1. **Client** submits jobs to the cluster. Submission paths include the CLI (`bin/flink run` via `flink-clients`), the SQL Client (`bin/sql-client.sh` via `flink-sql-client`), the SQL Gateway (`flink-sql-gateway`, also accessible via JDBC driver), the REST API (direct HTTP to JobManager), programmatic execution (`StreamExecutionEnvironment.execute()` or `TableEnvironment.executeSql()`), and PyFlink (`flink-python`, wraps the Java APIs).
189−2. **JobManager** (`flink-runtime`) orchestrates execution: receives jobs, creates the execution graph, manages scheduling, coordinates checkpoints, and handles failover. Never runs user code directly.
190−3. **TaskManager** (`flink-runtime`) executes the user's operators in task slots. Manages network buffers, state backends, and I/O.
191−4. **Table Planner** (`flink-table-planner`) translates SQL/Table API programs into DataStream programs. The planner is loaded in a separate classloader (`flink-table-planner-loader`) to isolate Calcite dependencies.
192−5. **Connectors** communicate with external systems. Source connectors implement the `Source` API (FLIP-27); sinks implement the `Sink` API (package `sink2`). Most connectors are externalized to separate repositories.
193−6. **State Backends** persist keyed state and operator state. RocksDB is the primary backend for production use.
194−7. **Checkpointing** provides exactly-once guarantees. The JobManager coordinates barriers through the data stream; TaskManagers snapshot local state to a distributed file system.
195−
196−Key separations:
197−
198−- **Planner vs Runtime:** The table planner generates code and execution plans; the runtime executes them. Changes to planning logic live in `flink-table-planner`; changes to runtime operators live in `flink-table-runtime` or `flink-streaming-java`.
199−- **Codegen vs hand-written operators:** Per-record expression logic (casts, projections, filters, function calls) is generated at planning time by cast rules in `flink-table-planner/.../functions/casting/` and call generators in `flink-table-planner/.../codegen/calls/`, then compiled by Janino into the surrounding operator class. Operators with fixed structure (joins, aggregations, source/sink runtime) are hand-written Java in `flink-table-runtime` or `flink-streaming-java`. New scalar functions usually only need a `BuiltInFunctionDefinitions` entry plus a `BuiltInScalarFunction` subclass - the planner wires up codegen automatically. New cast behaviour or a custom call shape needs a cast rule or call generator.
200−- **API vs Implementation:** Public API surfaces (`flink-core-api`, `flink-datastream-api`, `flink-table-api-java`) are separate from implementation modules. API stability annotations control what users can depend on.
201−- **ArchUnit enforcement:** `flink-architecture-tests/` contains ArchUnit tests that enforce module boundaries. New violations should be avoided; if unavoidable, follow the freeze procedure in `flink-architecture-tests/README.md`.
202−
20357 ## Common Change Patterns
20458
205−This section maps common types of Flink changes to the modules they touch and the verification they require.
59+### Adding a new table operator
20660
207−### Adding a new SQL built-in function
61+Components involved (can be developed top-down or bottom-up):
20862
209−1. Register in `flink-table-common` in `BuiltInFunctionDefinitions.java` (definition, input/output type strategies, runtime class reference)
210−2. Implement in `flink-table-runtime` under `functions/` (extend the appropriate base class: `BuiltInScalarFunction`, `BuiltInTableFunction`, `BuiltInAggregateFunction`, or `BuiltInProcessTableFunction`)
211−3. Add tests in `flink-table-planner` and `flink-table-runtime`
212−4. Extend Table API support
213−5. Document in `docs/`
214−6. See [flink-table/flink-table-planner/AGENTS.md](flink-table/flink-table-planner/AGENTS.md) and [flink-table/flink-table-runtime/AGENTS.md](flink-table/flink-table-runtime/AGENTS.md) for detailed patterns
63+1. **Runtime operator** in `flink-table-runtime` under `operators/` (extend `TableStreamOperator`, implement `OneInputStreamOperator` or `TwoInputStreamOperator`). Test with harness tests. See [flink-table-runtime AGENTS.md](../flink-table-runtime/AGENTS.md).
64+2. **ExecNode** in `plan/nodes/exec/stream/` and/or `plan/nodes/exec/batch/` (extend `ExecNodeBase<T>`; implement `StreamExecNode<T>` for streaming or `BatchExecNode<T>` for batch; annotate with `@ExecNodeMetadata`; `T` is typically `RowData`)
65+3. **Physical Node + Physical Rules** in `plan/rules/physical/stream/` and/or `plan/rules/physical/batch/` (physical rules usually extend `ConverterRule` via `Config.INSTANCE.withConversion(...)`; same-convention rewrites extend `RelRule` with an `@Value.Immutable` config)
66+4. **Logical Node + Planner rule**
67+5. Tests: semantic tests, plan tests, restore tests (if stateful)
21568
216−### Adding a new configuration option
69+Both `stream/` and `batch/` directories exist for rules and ExecNodes. Consider whether your change applies to one or both.
21770
218−1. Define `ConfigOption<T>` in the relevant config class (e.g., `ExecutionConfigOptions.java` in `flink-table-api-java`)
219−2. Use `ConfigOptions.key("table.exec....")` builder with type, default value, and description
220−3. Add `@Documentation.TableOption` annotation for auto-generated docs
221−4. Document in `docs/` if user-facing
222−5. Verify: unit test for default value, ITCase for behavior change
71+### Adding a planner optimization rule
22372
224−### Adding a new table operator (e.g., join type, aggregate)
73+Pick the base class by what the rule does:
74+- Converts a node from one calling convention to another (for example, logical → stream physical): extend `ConverterRule`.
75+Call `ConverterRule.Config.INSTANCE.withConversion(...)` in the constructor, do not define your own config.
76+- Rewrites nodes within the same convention (logical → logical, physical → physical): extend `RelRule` with an `@Value.Immutable` config.
77+Some existing rules still use Calcite's older `RelOptRule`; prefer `RelRule` for new code.
22578
226−1. Involves `flink-table-runtime` (operator), `flink-table-planner` (ExecNode, physical/logical rules), and tests across both
227−2. See [flink-table/flink-table-planner/AGENTS.md](flink-table/flink-table-planner/AGENTS.md) and [flink-table/flink-table-runtime/AGENTS.md](flink-table/flink-table-runtime/AGENTS.md) for detailed development order and testing patterns
79+Then:
80+1. Register in `FlinkStreamRuleSets.scala` and/or `FlinkBatchRuleSets.scala`
81+2. Plan tests with XML golden files — when the test fails, copy the framework's generated log file over the reference `.xml` (cases are ordered alphabetically by method name)
82+3. A same-convention rewrite needs no runtime changes. A `ConverterRule` that produces a new physical node also needs the physical node, ExecNode, and runtime operator — see "Adding a new table operator" above.
22883
229−### Adding a new connector (Source or Sink)
84+### Extending SQL syntax
23085
231−1. Implement the `Source` API (`flink-connector-base`): `SplitEnumerator`, `SourceReader`, `SourceSplit`, serializers (`SimpleVersionedSerializer`)
232−2. Or implement the `Sink` API (package `sink2`) for sinks
233−3. Most new connectors go in separate repos under `github.com/apache`, not in the main Flink repo
234−4. Verify: unit tests + ITCase with real or embedded external system
86+1. Modify parser grammar in `flink-sql-parser` (`parserImpls.ftl`)
87+2. Add operation conversion logic in `SqlNodeToOperationConversion.java`
88+3. Test with parser tests and SQL gateway integration tests (`.q` files)
23589
236−### Modifying state serializers
90+### Code generation changes
23791
238−1. Changes to `TypeSerializer` require a corresponding `TypeSerializerSnapshot` for migration
239−2. Bump version in `getCurrentVersion()`, handle old versions in `readSnapshot()`
240−3. Snapshot must have no-arg constructor for reflection-based deserialization
241−4. Implement `resolveSchemaCompatibility()` for upgrade paths
242−5. Verify: serializer snapshot migration tests, checkpoint restore tests across versions
92+- Cast rules live in `functions/casting/`. Each extends `AbstractExpressionCodeGeneratorCastRule` or similar.
93+- Custom call generators for functions live in `codegen/calls/` (e.g., `JsonCallGen.scala`). Simple scalar functions typically don't need these; the planner handles them uniformly through the function definition.
94+- Immutables library is used for rule configs (`@Value.Immutable`, `@Value.Enclosing`). See [README.md](README.md).
24395
244−### Introducing or changing user-facing APIs (`@Public`, `@PublicEvolving`, `@Experimental`)
96+### Plan serialization changes
24597
246−1. New user-facing API requires a voted FLIP (Flink Improvement Proposal); this applies to `@Public`, `@PublicEvolving`, and `@Experimental` since users build against all three
247−2. Every user-facing API class and method must carry a stability annotation
248−3. Changes to existing `@Public` or `@PublicEvolving` API must maintain backward compatibility
249−4. `@Internal` APIs can be changed freely; users should not depend on them
250−5. Update JavaDoc on the changed class/method
251−6. Add to release notes
252−7. Verify: ArchUnit tests pass, no new architecture violations
98+- ExecNode specs use Jackson for JSON serialization. Source/sink specs should use `@JsonIgnoreProperties(ignoreUnknown = true)` for forward compatibility.
99+- When adding new ExecNode features, update `RexNodeJsonDeserializer` or related serde classes if new function kinds or types are introduced.
253100
254−## Coding Standards
101+### ExecNode versioning
255102
256−- **Format Java files with Spotless immediately after editing:** `./mvnw spotless:apply`. Uses google-java-format with AOSP style.
257−- **Scala formatting:** Spotless + scalafmt (config at `.scalafmt.conf`, maxColumn 100).
258−- **Checkstyle:** `tools/maven/checkstyle.xml` (version defined in root `pom.xml` as `checkstyle.version`). Some modules (flink-core, flink-optimizer, flink-runtime) are not covered by checkstyle enforcement, but conventions should still be followed.
259−- **No new Scala code.** All Flink Scala APIs are deprecated per FLIP-265. Write all new code in Java.
260−- **Apache License 2.0 header** required on all new files (enforced by Apache Rat). Use an HTML comment for markdown files.
261−- **API stability annotations:** Every user-facing API class and method must have a stability annotation. `@Public` (stable across minor releases), `@PublicEvolving` (may change in minor releases), `@Experimental` (may change at any time). These are all part of the public API surface that users build against. `@Internal` marks APIs with no stability guarantees that users should not depend on.
262−- **Logging:** Use parameterized log statements (SLF4J `{}` placeholders), never string concatenation.
263−- **No Java serialization** for new features (except internal RPC message transport).
264−- **Use `final`** for variables and fields where applicable.
265−- **Comments:** Do not add unnecessary comments that restate what the code does. Add comments that explain "the why" where relevant.
266−- **Reuse existing code.** Before implementing new utilities or abstractions, search for existing ones in the codebase. Prioritize architecture consistency and code reusability.
267−- Full code style guide: https://flink.apache.org/how-to-contribute/code-style-and-quality-preamble/
103+When bumping an ExecNode version, update the `@ExecNodeMetadata` annotation's `version` and `minPlanVersion`/`minStateVersion` fields. Add restore test snapshots for the new version.
268104
269−## Testing Standards
105+### Configuration options
270106
271−- Add tests for new behavior, covering success, failure, and edge cases.
272−- Use **JUnit 5** + **AssertJ** assertions. Do not use JUnit 4 or Hamcrest in new test code.
273−- Prefer real test implementations over Mockito mocks where possible.
274−- **Integration tests:** Name classes with `ITCase` suffix (e.g., `MyFeatureITCase.java`).
275−- **Red-green verification:** For bug fixes, verify that new tests actually fail without the fix before confirming they pass with it.
276−- **Test location** mirrors source structure within each module.
277−- Follow the testing conventions at https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#7-testing
107+New features often introduce `ExecutionConfigOptions` entries (in `flink-table-api-java`) for runtime tunability (e.g., cache sizes, timeouts, batch sizes).
278108
279−## Commits and PRs
109+### PTF conditional traits
280110
281−### Commit message format
111+A *conditional trait* lets a PTF's table-argument traits depend on the call site instead of being fixed at declaration. Example for `TO_CHANGELOG`: the `input` argument is row-semantic by default (single stream, no PARTITION BY), but switches to set-semantic when the user writes `PARTITION BY` so the runtime can co-locate state per key. One declaration, two effective signatures depending on the call.
282112
283−- `[FLINK-XXXX][component] Description` where FLINK-XXXX is the JIRA issue number
284−- `[hotfix][component] Description` for typo fixes without JIRA
285−- Each commit must have a meaningful message including the JIRA ID. If you don't know the ticket number, ask.
286−- Separate cleanup/refactoring from functional changes into distinct commits
287−- When AI tools were used: add `Generated-by: <Tool Name and Version>` trailer per [ASF generative tooling guidance](https://www.apache.org/legal/generative-tooling.html)
113+**Declaration.** Built-in functions add conditional rules in `BuiltInFunctionDefinitions` via `StaticArgument.withConditionalTrait(trait, condition)`. The condition (a `TraitCondition`) is a small value-comparable predicate evaluated against a `TraitContext`. Built-in factories live on `TraitCondition` (`hasPartitionBy()`, `argIsEqualTo(name, value)`, `not(c)`); under the hood they wrap into the package-private `BuiltInCondition` so equality cascades correctly through `StaticArgument.equals`.
288114
289−### Pull request conventions
115+**Evaluation.** A `TraitCondition` reads two things: whether `PARTITION BY` is present on this table arg, and the literal value of named scalar args. Both come through `TraitContext`. There are two factories: `TraitContext.of(TableSemantics, CallContext, declared)` for the validation side (called from `SystemTypeInference.resolveStaticArgs`) and a planner-side adapter inside `BridgingSqlFunction.buildTraitContext` that sources the same data from a `RexCall` + `RexTableArgCall`. Same logical context, different inputs because the two layers don't share types.
290116
291−- Title format: `[FLINK-XXXX][component] Title of the pull request`
292−- A corresponding JIRA issue is required (except hotfixes for typos)
293−- Fill out the PR template completely but concisely: describe purpose, change log, testing approach, impact assessment
294−- Each PR should address exactly one issue
295−- Ensure `./mvnw clean verify` passes before opening a PR
296−- Always push to your fork, not directly to `apache/flink`
297−- Rebase onto the latest target branch before submitting
298−- For user-visible behaviour changes, breaking changes, new SQL features, or new config options: fill in the **Release Notes** field on the JIRA ticket. The release manager consolidates these when cutting a release. The next version's `docs/content/release-notes/flink-X.Y.md` will be generated based of the jira tickets, so make sure to fill them in properly.
117+**Resolution.** Three call sites bake conditional traits into the operator's effective signature:
299118
300−### AI-assisted contributions
119+1. **Validation** — `SystemTypeInference.resolveStaticArgs` runs once each from `inferInputTypes` and `inferType`. Twice per validation pass; can't dedupe across Calcite hooks because each gets a different `CallContext` instance.
120+2. **Planning** — `BridgingSqlFunction.resolveCallTraits` is called from `FlinkLogicalTableFunctionScan.Converter.convert`. It rewrites the operator on the `RexCall` so all downstream readers see the resolved view via plain `function.getTypeInference().getStaticArguments()`.
121+3. **Compiled-plan restore** — `BridgingSqlFunction.resolveCallTraits` is called again from `StreamExecProcessTableFunction.@JsonCreator`, because the JSON path skips the logical converter. Without this hook, restore would silently produce wrong results for any conditional-trait PTF.
301122
302−- Disclose AI usage by checking the AI disclosure checkbox and uncommenting the `Generated-by` line in the PR template
303−- Add `Generated-by: <Tool Name and Version>` to commit messages
304−- Never add `Co-Authored-By` with an AI agent as co-author; agents are assistants, not authors
305−- You must be able to explain the design, code, and tests, debug them, and respond to review feedback substantively
306−- Reviewer-ready quality bar: the author owns PR quality. PRs that look AI-generated without author refinement (walls of unreviewed prose, scaffolding without behaviour, tests that do not exercise the change, padded commit messages) will be closed without review
123+The payoff: downstream rules, exec nodes, codegen, and changelog inference all use ordinary `staticArg.is(SET_SEMANTIC_TABLE)` checks. No consumer needs to know that conditional traits exist. Why three sites and not one. The three resolution points exist because they sit in different lifecycles that can't share state.
307124
308−## Code Review Guidelines
125+## Testing Patterns
309126
310−When reviewing a PR or diff against this repo:
127+Choose test types based on what you're changing:
311128
312−- Look for opportunities to simplify the code, scoped to the diff itself (not pre-existing code outside the change).
313−- Flag comments that are obvious (restate what the code already says) or overly verbose.
314−- In test code, look for potential flakiness — e.g. `Thread.sleep` used outside a retry/poll loop, or similar timing-dependent, non-deterministic patterns. Where applicable, suggest clock injection (e.g. a manually-advanced `Clock`/`ManualClock`) instead of relying on wall-clock time, or waiting for the actual condition in a loop with a timeout, for deterministic tests.
315−- Check that each commit message conforms to Flink conventions: it must start with `[FLINK-XXXX]` or `[hotfix]`, and must specify a subsystem/component (e.g. `[FLINK-XXXX][runtime] Description`).
316−- If a change introduces a new feature controlled by a config option/flag, check that the resolved state (enabled/disabled, and the effective value) is logged at INFO level when the feature initializes/activates.
317−- Consider whether a change should be hidden behind a feature flag, especially if it's non-trivial (touches core paths, changes default behavior, or is hard to reason about in isolation). This is mandatory if the change is risky (correctness, performance, backward-compatibility, or data-safety risk) and no flag/kill-switch already exists.
318−- For changes to configuration options, check if the corresponding documentation has been regenerated (this should be covered by tests but flagging it earlier speeds up development)
319−
320−## Boundaries
321−
322−### Ask first
323−
324−- Adding or changing `@Public`, `@PublicEvolving`, or `@Experimental` annotations (these are user-facing API commitments requiring a FLIP)
325−- Large cross-module refactors
326−- New dependencies
327−- Changes to serialization formats (affects state compatibility)
328−- Changes to checkpoint/savepoint behavior
329−- Changes that could impact performance on hot paths (per-record processing, serialization, state access)
330−
331−### Never
332−
333−- Commit secrets, credentials, or tokens
334−- Push directly to `apache/flink`; always work from your fork
335−- Mix unrelated changes into one PR
336−- Use Java serialization for new features
337−- Edit generated files by hand when a generation workflow exists
338−- Use the legacy `SourceFunction` or `SinkFunction` interfaces for connectors; use the `Source` API (FLIP-27) and `Sink` API (package `sink2`) instead
339−- Add `Co-Authored-By` with an AI agent as co-author in commit messages; AI agents are assistants, not authors. Use `Generated-by: <Tool Name and Version>` instead.
340−- Suppress or bypass checkstyle rules (no `CHECKSTYLE:ON`/`CHECKSTYLE:OFF` comments, no adding entries to `tools/maven/suppressions.xml`, no `@SuppressWarnings`). Fix the code to satisfy checkstyle instead.
341−- Add, change, or remove classes outside the `org.apache.flink.*` package (for example, classes copied from Calcite)
342−- Modify `Parser.jj` (Calcite's generated parser grammar; expected to be removed in future Calcite upgrades)
343−- Use destructive git operations unless explicitly requested
344−
345−## References
346−
347−- [README.md](README.md) — Build instructions and project overview
348−- [DEVELOPMENT.md](DEVELOPMENT.md) — IDE setup and development environment
349−- [.github/CONTRIBUTING.md](.github/CONTRIBUTING.md) — Contribution process
350−- [.github/PULL_REQUEST_TEMPLATE.md](.github/PULL_REQUEST_TEMPLATE.md) — PR checklist
351−- [Code Style Guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-preamble/) — Detailed coding guidelines
352−- [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) — AI tooling policy
129+- **Semantic tests** (for ExecNode/operator changes): Use `SemanticTestBase` (streaming) or `BatchSemanticTestBase` (batch) in `plan/nodes/exec/testutils/`. Extends `CommonSemanticTestBase` which implements `TableTestProgramRunner`. Prefer these over ITCase for operators and ExecNodes.
130+- **Restore tests** (for stateful operators): Use `RestoreTestBase` or `BatchRestoreTestBase` in `plan/nodes/exec/testutils/`. Implements `TableTestProgramRunner`, uses `@ExtendWith(MiniClusterExtension.class)`. Required when your operator uses state. Tests savepoint creation and job restart in two phases: (1) generate compiled plans + savepoints, (2) verify recovery.
131+- **Plan tests** (for optimization rules): Verify the generated execution plan using XML golden files. Used for logical and physical optimization rules.
132+- **ITCase** (for built-in functions): Function tests typically use ITCase with `TestSetSpec` for end-to-end verification (e.g., `JsonFunctionsITCase`, `TimeFunctionsITCase`).
133+- **JSON plan test regeneration:** Set `PLAN_TEST_FORCE_OVERWRITE=true` environment variable (documented in [README.md](README.md)).
353134
