[FLINK-40170][table-planner] Infer update-producing changelog mode for early-fire interval join

.github/workflows/nightly-trigger.yml

@@ -53,7 +53,7 @@ jobs:
           branch: ${{ matrix.branch }}
 
       - name: Trigger Workflow
-        uses: actions/github-script@v7
+        uses: actions/github-script@v9
         with:
           script: |
             const branch = '${{ matrix.branch }}';

AGENTS.md

@@ -305,6 +305,15 @@ This section maps common types of Flink changes to the modules they touch and th
 - You must be able to explain the design, code, and tests, debug them, and respond to review feedback substantively
 - 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
 
+## Code Review Guidelines
+
+When reviewing a PR or diff against this repo:
+
+- Look for opportunities to simplify the code, scoped to the diff itself (not pre-existing code outside the change).
+- Flag comments that are obvious (restate what the code already says) or overly verbose.
+- 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.
+- 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`).
+
 ## Boundaries
 
 ### Ask first

docs/data/sql_functions.yml

@@ -1223,6 +1223,50 @@ json:
       -- [{"nested_json":{"value":42}}]
       JSON_ARRAY(JSON('{"nested_json": {"value": 42}}'))
       ```
+  - sql: JSON_LENGTH(json_doc[, path])
+    table: jsonLength(jsonObject[, path])
+    description: |
+      Returns the number of elements in a JSON document, or the length of the value at the specified path if one is provided.
+
+      The input can be a JSON STRING or a VARIANT. Returns NULL if the argument is NULL, the json is invalid, or the path is empty, malformed or does not locate a value.
+      The path must be a plain path literal such as '$.a.b'. A path carrying a 'lax'/'strict' path mode prefix raises an error.
+      eg.
+        -- 2
+        JSON_LENGTH('{"1": "hello", "2": "bye bye"}')
+
+        -- 5
+        JSON_LENGTH('[1,2,3,4,5]')
+
+        -- 1
+        JSON_LENGTH('"hello"')
+
+        -- 1
+        JSON_LENGTH('{"1": "hello", "2": "bye bye"}', '$.2')
+
+        -- NULL
+        JSON_LENGTH('{"1": "hello", "2": "BAD SYNTAX ->"', '$.2')
+
+        -- NULL
+        JSON_LENGTH('{"1": "hello", "2": "bye bye"}', '$.[')
+
+        -- error: JSON_LENGTH does not support the 'lax'/'strict' path mode prefix
+        JSON_LENGTH('{"1": "hello", "2": "bye bye"}', 'strict $.1')
+
+      The length is determined as follows:
+
+        - Scalar values (number, string, boolean): has length 1.
+        - Array: has a length equal to the number of its elements.
+        - Object: has a length equal to the number of its key-value pairs.
+
+      A wildcard path that matches 2 or more nodes returns NULL.
+      A NULL result is ambiguous - it means invalid JSON, no match, or a multi-match wildcard.
+
+      Pair JSON_LENGTH with a helper function to handle these cases explicitly:
+        -- IS JSON separates invalid input from a real result
+        SELECT CASE WHEN json_doc IS JSON THEN JSON_LENGTH(json_doc) END;
+
+        -- JSON_EXISTS separates an absent path from a present one
+        SELECT JSON_EXISTS(json_doc, '$.items[*]'), JSON_LENGTH(json_doc, '$.items[*]');
 
 variant:
   - sql: PARSE_JSON(json_string[, allow_duplicate_keys])

docs/data/sql_functions_zh.yml

@@ -1309,6 +1309,50 @@ json:
       -- '[[1]]'
       JSON_ARRAY(JSON_ARRAY(1))
       ```
+  - sql: JSON_LENGTH(json_doc[, path])
+    table: jsonLength(jsonObject[, path])
+    description: |
+      Returns the number of elements in a JSON document, or the length of the value at the specified path if one is provided.
+
+      The input can be a JSON STRING or a VARIANT. Returns NULL if the argument is NULL, the json is invalid, or the path is empty, malformed or does not locate a value.
+      The path must be a plain path literal such as '$.a.b'. A path carrying a 'lax'/'strict' path mode prefix raises an error.
+      eg.
+        -- 2
+        JSON_LENGTH('{"1": "hello", "2": "bye bye"}')
+
+        -- 5
+        JSON_LENGTH('[1,2,3,4,5]')
+
+        -- 1
+        JSON_LENGTH('"hello"')
+
+        -- 1
+        JSON_LENGTH('{"1": "hello", "2": "bye bye"}', '$.2')
+
+        -- NULL
+        JSON_LENGTH('{"1": "hello", "2": "BAD SYNTAX ->"', '$.2')
+
+        -- NULL
+        JSON_LENGTH('{"1": "hello", "2": "bye bye"}', '$.[')
+
+        -- error: JSON_LENGTH does not support the 'lax'/'strict' path mode prefix
+        JSON_LENGTH('{"1": "hello", "2": "bye bye"}', 'strict $.1')
+
+      The length is determined as follows:
+
+        - Scalar values (number, string, boolean): has length 1.
+        - Array: has a length equal to the number of its elements.
+        - Object: has a length equal to the number of its key-value pairs.
+
+      A wildcard path that matches 2 or more nodes returns NULL.
+      A NULL result is ambiguous - it means invalid JSON, no match, or a multi-match wildcard.
+
+      Pair JSON_LENGTH with a helper function to handle these cases explicitly:
+        -- IS JSON separates invalid input from a real result
+        SELECT CASE WHEN json_doc IS JSON THEN JSON_LENGTH(json_doc) END;
+
+        -- JSON_EXISTS separates an absent path from a present one
+        SELECT JSON_EXISTS(json_doc, '$.items[*]'), JSON_LENGTH(json_doc, '$.items[*]');
 
 variant:
   - sql: PARSE_JSON(json_string[, allow_duplicate_keys])

docs/layouts/shortcodes/generated/cluster_configuration.html

@@ -62,6 +62,12 @@
             <td>Duration</td>
             <td>The shutdown timeout for cluster services like executors.</td>
         </tr>
+        <tr>
+            <td><h5>cluster.thread-dump.default-mode</h5></td>
+            <td style="word-wrap: break-word;">FULL</td>
+            <td><p>Enum</p></td>
+            <td>Default granularity of the JobManager/TaskManager thread-dump REST endpoint when no explicit <code class="highlighter-rouge">mode</code> query parameter is supplied. The default is <code class="highlighter-rouge">FULL</code> to preserve historical behavior; operators of large clusters are strongly encouraged to switch to <code class="highlighter-rouge">LITE</code> to avoid heartbeat timeouts caused by long safepoint pauses.<br /><br />Possible values:<ul><li>"LITE": Stack traces only, without lock information. Negligible JVM pause.</li><li>"FULL": Additionally collects locked monitors and j.u.c. synchronizers, equivalent to jstack -l. Pauses the JVM in a safepoint for a duration that scales with heap size and thread count, which can take seconds on large TaskManagers.</li></ul></td>
+        </tr>
         <tr>
             <td><h5>cluster.thread-dump.stacktrace-max-depth</h5></td>
             <td style="word-wrap: break-word;">50</td>

docs/layouts/shortcodes/generated/expert_cluster_section.html

@@ -26,6 +26,12 @@
             <td>Boolean</td>
             <td>Whether processes should halt on fatal errors instead of performing a graceful shutdown. In some environments (e.g. Java 8 with the G1 garbage collector), a regular graceful shutdown can lead to a JVM deadlock. See <a href="https://issues.apache.org/jira/browse/FLINK-16510">FLINK-16510</a> for details.</td>
         </tr>
+        <tr>
+            <td><h5>cluster.thread-dump.default-mode</h5></td>
+            <td style="word-wrap: break-word;">FULL</td>
+            <td><p>Enum</p></td>
+            <td>Default granularity of the JobManager/TaskManager thread-dump REST endpoint when no explicit <code class="highlighter-rouge">mode</code> query parameter is supplied. The default is <code class="highlighter-rouge">FULL</code> to preserve historical behavior; operators of large clusters are strongly encouraged to switch to <code class="highlighter-rouge">LITE</code> to avoid heartbeat timeouts caused by long safepoint pauses.<br /><br />Possible values:<ul><li>"LITE": Stack traces only, without lock information. Negligible JVM pause.</li><li>"FULL": Additionally collects locked monitors and j.u.c. synchronizers, equivalent to jstack -l. Pauses the JVM in a safepoint for a duration that scales with heap size and thread count, which can take seconds on large TaskManagers.</li></ul></td>
+        </tr>
         <tr>
             <td><h5>cluster.thread-dump.stacktrace-max-depth</h5></td>
             <td style="word-wrap: break-word;">50</td>

docs/layouts/shortcodes/generated/expert_forst_section.html

@@ -20,6 +20,12 @@
             <td>Integer</td>
             <td>When the number of eviction that a block in hot link is moved to cold link reaches this value, the block will be blocked from being promoted to the head of the LRU list. The default value is '3'.</td>
         </tr>
+        <tr>
+            <td><h5>state.backend.forst.checkpoint.transfer-thread-num</h5></td>
+            <td style="word-wrap: break-word;">4</td>
+            <td>Integer</td>
+            <td>The number of threads used to transfer files during checkpoint (writing or copying files to the checkpoint storage) and restore (transferring state files back to the ForSt working directory). Consider increasing this value when snapshotting or restoring large state. Note that the pool is created per state backend instance. Setting it too high can saturate network bandwidth or trigger rate limiting depending on the remote storage; setting it too low can lead to long checkpoint durations or timeouts for large state. The default value is '4'.</td>
+        </tr>
         <tr>
             <td><h5>state.backend.forst.executor.inline-coordinator</h5></td>
             <td style="word-wrap: break-word;">false</td>

docs/layouts/shortcodes/generated/expert_rocksdb_section.html

@@ -20,6 +20,48 @@
             <td>String</td>
             <td>The local directory (on the TaskManager) where RocksDB puts its files. Per default, it will be &lt;WORKING_DIR&gt;/tmp. See <code class="highlighter-rouge">process.taskmanager.working-dir</code> for more details.</td>
         </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-auto-compactions</h5></td>
+            <td style="word-wrap: break-word;">30</td>
+            <td>Integer</td>
+            <td>The maximum number of automatic compactions running for manual compaction to start. If the actual number is higher, manual compaction won't be started to avoid delaying automatic ones.</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-file-size-to-compact</h5></td>
+            <td style="word-wrap: break-word;">50 kb</td>
+            <td>MemorySize</td>
+            <td>The maximum size of individual input files</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-files-to-compact</h5></td>
+            <td style="word-wrap: break-word;">30</td>
+            <td>Integer</td>
+            <td>The maximum number of input files to compact together in a single compaction run</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-output-file-size</h5></td>
+            <td style="word-wrap: break-word;">64 mb</td>
+            <td>MemorySize</td>
+            <td>The maximum output file size</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-parallel-compactions</h5></td>
+            <td style="word-wrap: break-word;">5</td>
+            <td>Integer</td>
+            <td>The maximum number of manual compactions to start. Note that only one of them can run at a time as of v8.10.0; all the others will be waiting</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.min-files-to-compact</h5></td>
+            <td style="word-wrap: break-word;">5</td>
+            <td>Integer</td>
+            <td>The minimum number of input files to compact together in a single compaction run</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.min-interval</h5></td>
+            <td style="word-wrap: break-word;">0 ms</td>
+            <td>Duration</td>
+            <td>The minimum interval between manual compactions. Zero disables manual compactions</td>
+        </tr>
         <tr>
             <td><h5>state.backend.rocksdb.options-factory</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

docs/layouts/shortcodes/generated/forst_configuration.html

@@ -38,6 +38,12 @@
             <td>MemorySize</td>
             <td>An upper-bound of the size that can be used for cache. User should specify at least one cache size limit to enable the cache, either this option or the 'state.backend.forst.cache.reserve-size' option. They can be set simultaneously, and in this case, cache will grow if meet the requirements of both two options. The default value is '0 bytes', meaning that this option is disabled. </td>
         </tr>
+        <tr>
+            <td><h5>state.backend.forst.checkpoint.transfer-thread-num</h5></td>
+            <td style="word-wrap: break-word;">4</td>
+            <td>Integer</td>
+            <td>The number of threads used to transfer files during checkpoint (writing or copying files to the checkpoint storage) and restore (transferring state files back to the ForSt working directory). Consider increasing this value when snapshotting or restoring large state. Note that the pool is created per state backend instance. Setting it too high can saturate network bandwidth or trigger rate limiting depending on the remote storage; setting it too low can lead to long checkpoint durations or timeouts for large state. The default value is '4'.</td>
+        </tr>
         <tr>
             <td><h5>state.backend.forst.executor.inline-coordinator</h5></td>
             <td style="word-wrap: break-word;">false</td>

docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

@@ -1383,6 +1383,16 @@
     <tr>
       <td colspan="2">Returns the thread dump of the JobManager.</td>
     </tr>
+    <tr>
+      <td colspan="2">Query parameters</td>
+    </tr>
+    <tr>
+      <td colspan="2">
+        <ul>
+<li><code>mode</code> (optional): Controls how much lock information is collected. Supported values: [LITE, FULL]. When omitted, cluster.thread-dump.default-mode is used.</li>
+        </ul>
+      </td>
+    </tr>
     <tr>
       <td colspan="2">
         <label>
@@ -7121,6 +7131,16 @@
         </ul>
       </td>
     </tr>
+    <tr>
+      <td colspan="2">Query parameters</td>
+    </tr>
+    <tr>
+      <td colspan="2">
+        <ul>
+<li><code>mode</code> (optional): Controls how much lock information is collected. Supported values: [LITE, FULL]. When omitted, cluster.thread-dump.default-mode is used.</li>
+        </ul>
+      </td>
+    </tr>
     <tr>
       <td colspan="2">
         <label>

docs/layouts/shortcodes/generated/rocksdb_manual_compaction_configuration.html

@@ -0,0 +1,54 @@
+<table class="configuration table table-bordered">
+    <thead>
+        <tr>
+            <th class="text-left" style="width: 20%">Key</th>
+            <th class="text-left" style="width: 15%">Default</th>
+            <th class="text-left" style="width: 10%">Type</th>
+            <th class="text-left" style="width: 55%">Description</th>
+        </tr>
+    </thead>
+    <tbody>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-auto-compactions</h5></td>
+            <td style="word-wrap: break-word;">30</td>
+            <td>Integer</td>
+            <td>The maximum number of automatic compactions running for manual compaction to start. If the actual number is higher, manual compaction won't be started to avoid delaying automatic ones.</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-file-size-to-compact</h5></td>
+            <td style="word-wrap: break-word;">50 kb</td>
+            <td>MemorySize</td>
+            <td>The maximum size of individual input files</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-files-to-compact</h5></td>
+            <td style="word-wrap: break-word;">30</td>
+            <td>Integer</td>
+            <td>The maximum number of input files to compact together in a single compaction run</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-output-file-size</h5></td>
+            <td style="word-wrap: break-word;">64 mb</td>
+            <td>MemorySize</td>
+            <td>The maximum output file size</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-parallel-compactions</h5></td>
+            <td style="word-wrap: break-word;">5</td>
+            <td>Integer</td>
+            <td>The maximum number of manual compactions to start. Note that only one of them can run at a time as of v8.10.0; all the others will be waiting</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.min-files-to-compact</h5></td>
+            <td style="word-wrap: break-word;">5</td>
+            <td>Integer</td>
+            <td>The minimum number of input files to compact together in a single compaction run</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.min-interval</h5></td>
+            <td style="word-wrap: break-word;">0 ms</td>
+            <td>Duration</td>
+            <td>The minimum interval between manual compactions. Zero disables manual compactions</td>
+        </tr>
+    </tbody>
+</table>

docs/static/generated/rest_v1_dispatcher.yml

@@ -457,6 +457,15 @@ paths:
     get:
       description: Returns the thread dump of the JobManager.
       operationId: getJobManagerThreadDump
+      parameters:
+      - name: mode
+        in: query
+        description: "Controls how much lock information is collected. Supported values:\
+          \ [LITE, FULL]. When omitted, cluster.thread-dump.default-mode is used."
+        required: false
+        style: form
+        schema:
+          $ref: "#/components/schemas/ThreadDumpMode"
       responses:
         "200":
           description: The request was successful.
@@ -1863,6 +1872,14 @@ paths:
         required: true
         schema:
           $ref: "#/components/schemas/ResourceID"
+      - name: mode
+        in: query
+        description: "Controls how much lock information is collected. Supported values:\
+          \ [LITE, FULL]. When omitted, cluster.thread-dump.default-mode is used."
+        required: false
+        style: form
+        schema:
+          $ref: "#/components/schemas/ThreadDumpMode"
       responses:
         "200":
           description: The request was successful.
@@ -3911,6 +3928,11 @@ components:
           type: array
           items:
             $ref: "#/components/schemas/ThreadInfo"
+    ThreadDumpMode:
+      type: string
+      enum:
+      - LITE
+      - FULL
     ThreadInfo:
       type: object
       properties:

flink-architecture-tests/flink-architecture-tests-test/src/main/java/org/apache/flink/architecture/TestCodeArchitectureTestBase.java

@@ -19,6 +19,7 @@
 package org.apache.flink.architecture;
 
 import org.apache.flink.architecture.rules.ITCaseRules;
+import org.apache.flink.architecture.rules.TestNamingRules;
 
 import com.tngtech.archunit.junit.ArchTest;
 import com.tngtech.archunit.junit.ArchTests;
@@ -33,4 +34,6 @@
 public class TestCodeArchitectureTestBase {
 
     @ArchTest public static final ArchTests ITCASE = ArchTests.in(ITCaseRules.class);
+
+    @ArchTest public static final ArchTests TEST_NAMING = ArchTests.in(TestNamingRules.class);
 }