Seeking Advice on Handling Data Type Mismatches Between PostgreSQL Source and StarRocks Sink in Flink CDC Pipelines

[liuyq2step]

I am encountering issues while using Flink CDC pipelines to synchronize data from PostgreSQL to StarRocks. The root cause lies in the data type compatibility between the source (PostgreSQL) and the sink (StarRocks). Specifically, the StarRocks sink connector performs data type checks during SchemaChangeEvent processing and throws errors when unsupported types are encountered. For example, PostgreSQL types like timestamptz and time are not supported by StarRocks, leading to synchronization failures.

To address this, I have considered two potential solutions, both aimed at ensuring that the schema semantics and actual data in Flink CDC events align with the sink's expectations:

1.Handling Common PostgreSQL Types (e.g., timestamptz):

StarRocks only supports the timezone-agnostic datetime type. To resolve this, I modified the Flink CDC source code to treat timestamptz as a regular timestamp during JDBC-to-Flink-CDC type conversion. Specifically, I updated org.apache.flink.cdc.connectors.postgres.utils.PostgresTypeUtils#convertFromColumn:

case PgOid.TIMESTAMPTZ:  
    // Original: return new ZonedTimestampType(scale);  
    handleTimestampWithTemporalMode(temporalPrecisionMode, scale);

Additionally, I implemented a custom Debezium CustomConverter to complete the value converting.

public class PostgresTimestamptzConverter
        implements CustomConverter<SchemaBuilder, RelationalColumn> {

    private static final Logger LOGGER =
            LoggerFactory.getLogger(PostgresTimestamptzConverter.class);

    private Configuration config;
    private int targetOffsetHours = 8; 
    private ZoneOffset targetOffset; 

    @Override
    public void configure(Properties properties) {}

    @Override
    public void converterFor(
            RelationalColumn column, ConverterRegistration<SchemaBuilder> registration) {
        if (column.nativeType() == PgOid.TIMESTAMPTZ) {
            LOGGER.debug(
                    "Registering custom converter for TIMESTAMPTZ column (OID): {}", column.name());
            **SchemaBuilder builder = MicroTimestamp.builder();**
            if (column.isOptional()) {
                builder.optional();
            }

            registration.register(
                    builder,
                    new Converter() {
                        @Override
                        public Object convert(Object originalValue) {
                            return convertJdbcTimestampTzToAdjustedMicroTimestamp(originalValue);
                        }
                    });
        }
    }

    public static Long convertJdbcTimestampTzToAdjustedMicroTimestamp(Object originalValue) {
        if (originalValue == null) {
            return null;
        }

        Instant instant;

        if (originalValue instanceof OffsetDateTime) {
            instant = ((OffsetDateTime) originalValue).toInstant();
        } else if (originalValue instanceof LocalDateTime) {
            instant = ((LocalDateTime) originalValue).atZone(ZoneOffset.UTC).toInstant();
        } else if (originalValue instanceof Date) {
            instant = ((Date) originalValue).toInstant();
        } else if (originalValue instanceof String) {
            try {
                instant = Instant.parse((String) originalValue);
            } catch (Exception e) {
                LOGGER.warn("Failed to parse timestamptz string: {}", originalValue, e);
                return null;
            }
        } else {
            LOGGER.warn("Unsupported timestamptz value type: {}", originalValue.getClass());
            return null;
        }

        return instant.toEpochMilli() * 1_000 + 8 * 3_600_000_000L;
    }

}

This avoids to define transform rules for every table in the YAML job configuration.

2.Handling Less Common PostgreSQL Types (e.g., time):

Since StarRocks lacks a corresponding time type, I used a UserDefinedFunction (UDF) in the transform node of the YAML job to convert time data to varchar before saving it to StarRocks.

transform:
  - source-table: pg.public.t_flinktest
    projection: id,txt,dt,content,time_format(timecol) as timecol
pipeline:
  user-defined-function:
    - name: time_format
      classpath: com.sync.flink.cdc.udf.CastTimeAsString

public class CastTimeAsString implements UserDefinedFunction {

    public Object eval(Object obj) {
        if (obj == null) {
            return null;
        }

        if (obj instanceof TimeData) {
            return ((TimeData) obj).toLocalTime().toString();
        } else {
            throw new RuntimeException(
                    "CastTimeAsString is not supported："
                            + obj.getClass().getName()
                            + ", Value: "
                            + obj.toString());
        }
    }

    public DataType getReturnType() {
        return DataTypes.STRING();
    }
}

Does Flink CDC already provide built-in APIs or configurations to handle such data type mismatches that I might have overlooked? If not, I believe it would be valuable for Flink CDC to incorporate native support for resolving these compatibility issues, as they are common in real-world use cases.

Thank you for your insights and assistance!