[FLINK-39062][table] Support WATERMARK clause in CREATE VIEW statement

[featzhang]

What is the purpose of the change?

This PR FLINK-39062 adds support for specifying watermark strategies directly in CREATE VIEW statements, enabling users to define time attributes and watermark generation for views at creation time.

Brief change log

• Parser Layer: Updated SqlCreateView to include optional watermark field
• Grammar: Modified parserImpls.ftl to parse WATERMARK clause in CREATE VIEW syntax
• Converter Layer: Enhanced SqlCreateViewConverter to handle watermark conversion
• Schema Building: Updated SqlNodeConvertUtils.toCatalogView to process watermark and add it to view schema
• Compatibility: Updated SqlAlterViewAsConverter to handle new method signature

New Syntax

CREATE VIEW view_name
[WATERMARK FOR column_name AS watermark_expression]
AS SELECT ...

Examples

-- Basic watermark in view
CREATE VIEW my_view
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
AS SELECT event_id, event_time, data FROM source_table;

-- With field list
CREATE VIEW order_view (order_id, order_time, customer_id)
WATERMARK FOR order_time AS order_time - INTERVAL '10' SECOND
AS SELECT id, timestamp, customer FROM orders;

-- Temporary view with watermark
CREATE TEMPORARY VIEW temp_view
WATERMARK FOR ts AS ts - INTERVAL '1' MINUTE
AS SELECT * FROM events;

Verifying this change

This change added tests and can be verified as follows:

• Added test SQL file test_create_view_watermark.sql with 8 test cases
• Covers basic syntax, field lists, temporary views, comments, and complex expressions
• Backward compatibility maintained (WATERMARK clause is optional)

Manual testing:

1. Parser correctly handles the new WATERMARK clause
2. AST node properly represents the watermark specification
3. Converter translates to appropriate CatalogView with watermark in schema
4. No conflicts with existing CREATE VIEW statements

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? JavaDocs and inline code comments
• Future work will update official Flink SQL documentation

Related Issues

• Complements the existing ALTER VIEW SET WATERMARK functionality (PR [FLINK-39062][table] Support ALTER VIEW SET WATERMARK syntax #27570)
• Provides complete watermark support for views in Flink SQL

[flinkbot]

CI report:

• 5163df7 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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