[FLINK-38879][pipeline-connector][paimon] Add support for creating and writing table with variant type. #4228
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…d writing table with variant type.
Contributor
Author
|
Hi @yunfengzhou-hub, maybe you can help to review this as you are more familiar with Paimon Sink. |
There was a problem hiding this comment.
Pull request overview
This pull request adds support for creating and writing tables with the VARIANT data type in the Paimon connector for Flink CDC.
Changes:
- Introduced a new
TypeUtilsutility class to handle type conversions between CDC and Paimon systems with special handling for VARIANT types - Added field getter implementation for VARIANT type in
PaimonWriterHelperto convert between CDC'sBinaryVariantand Paimon'sGenericVariant - Refactored existing type conversion code to use the new
TypeUtilsclass for consistency and to support VARIANT type
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| TypeUtils.java | New utility class providing bidirectional type conversion between CDC and Paimon types with special handling for VARIANT |
| PaimonWriterHelper.java | Added VARIANT field getter and updated schema deduction to use TypeUtils for type conversions |
| SchemaChangeProvider.java | Refactored to use TypeUtils for consistent type conversion, eliminating duplicated conversion logic |
| PaimonMetadataApplier.java | Refactored to use TypeUtils for type conversion when applying schema changes |
| PaimonWriterHelperTest.java | Added comprehensive tests for VARIANT type handling in data conversion and schema deduction |
| PaimonMetadataApplierTest.java | Added tests for creating tables and adding columns with VARIANT type |
| PaimonHashFunctionTest.java | Added VARIANT type to hash function tests to ensure proper handling in partitioning logic |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
yunfengzhou-hub
left a comment
yunfengzhou-hub
left a comment
There was a problem hiding this comment.
Thanks for the PR. Left some comments as below.
...nector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/utils/TypeUtils.java
Show resolved
Hide resolved
...-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriterHelper.java
Outdated
Show resolved
Hide resolved
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Add support for creating and writing table with variant type.
Why not add end-to-end (e2e) tests?
Because the Flink version we're currently using is 1.20, and Flink SQL in this version does not yet support querying the VARIANT type. Therefore, we cannot use the catalog in Flink SQL to query and verify the results.
Support for this will only be available after we upgrade to Flink 2.2.