lvyanquan commented on code in PR #4391:
URL: https://github.com/apache/flink-cdc/pull/4391#discussion_r4068346124
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/SchemaChangeProvider.java:
##########
@@ -103,15 +141,102 @@ public static List<SchemaChange> add(
return result;
}
+ /**
+ * Convert CDC VARBINARY/BINARY/VARCHAR/STRING type to Paimon BLOB type if
configured.
+ *
+ * @param column The CDC column definition.
+ * @param tableOptions The table options containing blob-field
configuration.
+ * @return The Paimon DataType (BLOB if configured, otherwise original
converted type).
+ */
+ public static org.apache.paimon.types.DataType convertToBlobIfNeeded(
+ Column column, Map<String, String> tableOptions) {
+ // Check if this field should be converted to BLOB type using Paimon's
CoreOptions
+ List<String> blobFields = CoreOptions.blobField(tableOptions);
+ if (!blobFields.isEmpty() && isSupportedTypeForBlob(column.getType()))
{
+ if (blobFields.contains(column.getName())) {
+ // Convert VARBINARY/BINARY/VARCHAR/STRING to BLOB type
+ // BLOB type is always nullable in Paimon
+ return DataTypes.BLOB();
+ }
+ }
+
+ // Use TypeUtils.toPaimonDataType which handles VARIANT type properly
+ return TypeUtils.toPaimonDataType(column.getType());
+ }
+
+ /** Check if DataType can be converted to BLOB (BINARY, VARBINARY, or
VARCHAR). */
+ private static boolean isSupportedTypeForBlob(DataType dataType) {
+ DataTypeRoot typeRoot = dataType.getTypeRoot();
+ return typeRoot == DataTypeRoot.BINARY
+ || typeRoot == DataTypeRoot.VARBINARY
+ || typeRoot == DataTypeRoot.VARCHAR;
Review Comment:
Addressed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
