[flink] changelog read support for pk table without pushdown optimizations

[MehulBatra]

Purpose

Linked issue: close #2333

Brief change log

File: ChangelogFlinkTableSource.java
Purpose: Flink table source for $changelog virtual tables
────────────────────────────────────────
File: ChangelogDeserializationSchema.java
Purpose: Deserializes LogRecord → RowData with metadata
────────────────────────────────────────
File: ChangelogRowConverter.java
Purpose: Converts records, adds _change_type, _log_offset,
commit_timestamp
────────────────────────────────────────
File: RecordToFlinkRowConverter.java
Purpose: Interface for record converters
────────────────────────────────────────
File: FlinkTableFactory.java
Change: Routes to ChangelogFlinkTableSource for changelog
tables
────────────────────────────────────────
File: TableDescriptor.java
Change: Added CHANGELOG* column name constants
────────────────────────────────────────
File: TableDescriptorValidation.java
Change: Validates reserved column names
Test Coverage
Test: testChangelogVirtualTableSchema
Type: Schema
What it verifies: DESCRIBE shows correct columns
────────────────────────────────────────
File: FlinkCatalog.java
Change: Detects $changelog suffix, builds virtual table
schema

• Access via SELECT * FROM table$changelog
• Metadata columns: _change_type (VARCHAR), _log_offset (BIGINT), _commit_timestamp (TIMESTAMP)
• All records output as INSERT (change type in column)
• Supports PK tables with change types: +I, -U, +U, -D

Tests

File: ChangelogVirtualTableITCase.java
Purpose: Integration tests (7 tests)
────────────────────────────────────────
File: ChangelogRowConverterTest.java
Purpose: Unit tests for converter
────────────────────────────────────────

API and Format

Documentation

[wuchong]

Great! Thank you @MehulBatra , I will review it.