[FLINK-39092][SQL] Enhance EXPLAIN plan to display watermark specification

[featzhang]

What is the purpose of the change

This change enhances the EXPLAIN plan output for TableScan nodes by explicitly displaying watermark specifications. Currently, watermark logic is difficult to identify in EXPLAIN plans, which creates challenges for streaming users trying to understand and debug watermark strategies.

Before:

TableScan(table=[[default_catalog, default_database, orders]], fields=[user_id, order_time])

After:

TableScan:
  table: [[default_catalog, default_database, orders]]
  fields: user_id, order_time
  watermark: order_time - order_time - INTERVAL '5' SECOND

This improvement makes watermark strategies immediately visible in query plans, helping streaming users:

• Quickly understand watermark configurations
• Debug late data handling issues
• Verify watermark logic without examining table DDL
• Improve overall development and troubleshooting experience

Brief change log

• Modified FlinkLogicalTableSourceScan.explainTerms() to extract and display watermark specifications
• Added watermark information retrieval from ResolvedSchema via TableSourceTable
• Enhanced explain output to show rowtime attribute and watermark expression in a readable format
• Watermark is displayed only when present, maintaining backward compatibility for non-streaming tables

Verifying this change

This change can be verified by:

1. Compilation: The module compiles successfully without errors

   ./mvnw clean spotless:apply install -DskipTests -Pfast -pl flink-table/flink-table-planner

2. Create table with watermark:

   CREATE TABLE orders (
     user_id INT,
     order_time TIMESTAMP(3),
     WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
   ) WITH (...);

3. Verify EXPLAIN output:

   EXPLAIN SELECT * FROM orders;

   The output should display watermark information in the TableScan node.

4. Test without watermark: Verify that tables without watermarks still work normally

   CREATE TABLE batch_table (id INT, name STRING) WITH (...);
   EXPLAIN SELECT * FROM batch_table;

Does this pull request potentially affect

• Dependencies: No
• The public API: No (only changes internal explain output format)
• The serializers: No
• The runtime per-record code paths: No (only affects plan display)
• Anything that affects deployment or recovery: No

Documentation

• This change enhances the readability of query plan output for streaming jobs
• No user-facing API documentation changes required
• The improvement is visible in EXPLAIN plan output
• Maintains backward compatibility by only enhancing display format, not changing RelNode structure
• Particularly valuable for streaming processing users working with event time and watermarks

[flinkbot]

CI report:

• 01fa5eb Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[featzhang]

@flinkbot run azure