[connector] flink read support for fluss primary key table changelog auxiliary table/ cdc events

[MehulBatra]

Contribution Checklist

• Linked issue: close 356

Purpose

Enhanced the Flink connector by introducing support for the $changelog auxiliary table. enabling seamless CDC event capture and processing in streaming applications

Brief change log

• Touched FlinkCatalog GetTable (To support enablement of changelog Table)
• Touched CoordinatorService CreateTable (To add validation of system reserved columns)
• Extended FlussRowToFlinkRowConverter (ChangelogRowConverter to add support for emitting changelog records)
• Touched FlinkRecordEmitter (To add choice of special FlinkRow converter based on Changelog Flag)
• Touched FlinkTableFactory (To add createDynamicSource support for changlog table)
• Touched FlinkSource, FlinkSourceReader, FlinkTableSource (To add flag isChangelog in the source constructor of each class)

Tests

Yet to write (In progress)

API and Format

Yes , Flink Source Reader

Documentation

Yes post feedback need to add changes in the doc to support CDC reads via Flink Connector

[wuchong]

Thanks @MehulBatra , the direction looks good to me. I left some comments. Looking forward to a complete pull request!

Some tips for the following:

1. Add comprehensive tests for the feature, including

• FlinkCatalogTest for verifying the catalog table information
• integration tests in FlinkTableSourceITCase to support reading all columns,partial columns, partial metadata columns.
• tests to verify the $changelog doesn't support as a sink table.
• etc.

2. We may also need to adapt the FlinkTableSource to make it works with existing pushdown optimizations.

[MehulBatra]

Thanks @MehulBatra , the direction looks good to me. I left some comments. Looking forward to a complete pull request!

Some tips for the following:

1. Add comprehensive tests for the feature, including

• FlinkCatalogTest for verifying the catalog table information
• integration tests in FlinkTableSourceITCase to support reading all columns,partial columns, partial metadata columns.
• tests to verify the $changelog doesn't support as a sink table.
• etc.

2. We may also need to adapt the FlinkTableSource to make it works with existing pushdown optimizations.

Resolved the comments, made on the particular code blocks.

Will start with this now!

[MehulBatra]

Thanks @MehulBatra , the direction looks good to me. I left some comments. Looking forward to a complete pull request!

Some tips for the following:

1. Add comprehensive tests for the feature, including

• FlinkCatalogTest for verifying the catalog table information
• integration tests in FlinkTableSourceITCase to support reading all columns,partial columns, partial metadata columns.
• tests to verify the $changelog doesn't support as a sink table.
• etc.

2. We may also need to adapt the FlinkTableSource to make it works with existing pushdown optimizations.

• Wrote Unit and Integration test to support Flink Catalog changes, support reading all columns,partial columns, partial metadata columns and test to verify changelog doesn't support sink.

• Adapted FlinkTableSource for Projection.

[MehulBatra]

@wuchong Could you please help me with reviewing this PR