cdc with pipeline-transform failed because schema issue, is this a problem?

[liuyq2step]

test postgres db table with a timestamptz column , pipeline yaml is
`
source:
scan.startup.mode: committed-offset
....
transform:

• source-table: t_flinktest
  projection: id,txt,CAST(dt as VARCHAR) as dt
  --projection: id,txt,DATE_FORMAT(dt , ‘yyyy-MM-dd hh:mi:ss’) as dt
  --projection: id,txt,CAST(id as VARCHAR) as dt
  ....
  pipeline:
  name: test PGSQL to StarRocks Pieline
  schema.change.behavior: ignore
  parallelism: 1
  `

because starrocks dont have type that fit timetamptz, starrocks connector throws exception.

`
2025-10-24 13:47:37,065 ERROR org.apache.flink.cdc.runtime.operators.schema.common.SchemaRegistry [] - Current schema manager state: Schema Manager 427242635:
original schema map:
- table public.t_flinktest: {public.t_flinktest={0=columns={id BIGINT NOT NULL,txt VARCHAR(255),dt TIMESTAMP(6) WITH TIME ZONE now()}, primaryKeys=id, options=()}}
evolved schema map:

2025-10-24 13:47:37,071 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Trying to recover from a global failure.
org.apache.flink.util.FlinkException: Global failure triggered by OperatorCoordinator for 'Transform:Data -> SchemaOperator -> PrePartition' (operator 9899a42c64d67ef3172b7e3be3c1bbb9).
at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:651) ~[flink-dist-1.20.2.jar:1.20.2]
at org.apache.flink.cdc.runtime.operators.schema.common.SchemaRegistry.failJob(SchemaRegistry.java:392) ~[?:?]
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:249) ~[?:?]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[?:?]
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
at java.base/java.lang.Thread.run(Thread.java:842) ~[?:?]
Caused by: org.apache.flink.util.FlinkRuntimeException: Failed to apply schema change event.
... 6 more
Caused by: java.lang.UnsupportedOperationException: Unsupported CDC data type TIMESTAMP(6) WITH TIME ZONE
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils$CdcDataTypeTransformer.defaultMethod(StarRocksUtils.java:388) ~[?:?]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils$CdcDataTypeTransformer.defaultMethod(StarRocksUtils.java:241) ~[?:?]
at org.apache.flink.cdc.common.types.DataTypeDefaultVisitor.visit(DataTypeDefaultVisitor.java:105) ~[?:?]
at org.apache.flink.cdc.common.types.ZonedTimestampType.accept(ZonedTimestampType.java:110) ~[?:?]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils.toStarRocksDataType(StarRocksUtils.java:119) ~[?:?]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils.toStarRocksTable(StarRocksUtils.java:93) ~[?:?]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.applyCreateTable(StarRocksMetadataApplier.java:138) ~[?:?]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.lambda$applySchemaChange$2(StarRocksMetadataApplier.java:115) ~[?:?]
at org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:57) ~[?:?]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.applySchemaChange(StarRocksMetadataApplier.java:104) ~[?:?]
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:437) ~[?:?]
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:406) ~[?:?]
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:247) ~[?:?]
... 5 more
`

1. "schema.change.behavior: ignore" cannot make flink-cdc producing schemaChangeEvent.
2. schemaChangeEvent, the schema inside is source table schema, not after transform schema.
   is this a problem?

[yuxiqian]

Seems your upstream table doesn't hit that transform rule. Could you please try this?

 transform:
-  - source-table: t_flinktest
+  - source-table: public.t_flinktest
     projection: id,txt,CAST(dt as VARCHAR) as dt

source-table here should be fully qualified (including database, schema components).

[liuyq2step]

for test, i use CAST in yaml transform node.
transform:

• source-table: t_flinktest
  projection: id,txt,CAST(dt as VARCHAR) as dt

i set breakpoint at returning line of
org.apache.flink.cdc.debezium.event.DebeziumEventDeserializationSchema#convertToRecord
i find that ,the 'dt' column is converted to org.apache.flink.cdc.common.data.LocalZonedTimestampData, and then be generated to BinaryRecordData.

but in PreTransform part , BinaryRecordData.getZonedTimestamp get "", looks like serialize/deserialize issue?

[Mrart]

yes，I tried to reproduce the problem on the master branch.

[liuyq2step]

StarRocks does not have timestamp-with-time-zone data types. Therefore, I'm using the transform node in the YAML configuration to modify the schema in DatachangeEvent. Otherwise, the StarRocks sink connector throws the error
"
2025-10-27 15:45:37,447 INFO org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator [] - All sink subtask have flushed for table public.t_flinktest. Start to apply schema change request:
SchemaChangeRequest{tableId=public.t_flinktest, schemaChangeEvent=CreateTableEvent{tableId=public.t_flinktest, schema=columns={id BIGINT NOT NULL,txt VARCHAR(255),dt TIMESTAMP(6) WITH TIME ZONE 'now()'}, primaryKeys=id, options=()}, subTaskId=0}
that extracts to:
CreateTableEvent{tableId=bgwltest.t_flinktest, schema=columns={id BIGINT NOT NULL,txt VARCHAR(255),dt TIMESTAMP(6) WITH TIME ZONE 'now()'}, primaryKeys=id, options=()}
2025-10-27 15:45:37,447 ERROR org.apache.flink.cdc.runtime.operators.schema.common.SchemaRegistry [] - An exception was triggered from Schema change applying task. Job will fail now.
org.apache.flink.util.FlinkRuntimeException: Failed to apply schema change event.
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:249) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.base/java.lang.Thread.run(Thread.java:842) [?:?]
Caused by: java.lang.UnsupportedOperationException: Unsupported CDC data type TIMESTAMP(6) WITH TIME ZONE
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils$CdcDataTypeTransformer.defaultMethod(StarRocksUtils.java:388) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils$CdcDataTypeTransformer.defaultMethod(StarRocksUtils.java:241) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.common.types.DataTypeDefaultVisitor.visit(DataTypeDefaultVisitor.java:105) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at org.apache.flink.cdc.common.types.ZonedTimestampType.accept(ZonedTimestampType.java:110) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils.toStarRocksDataType(StarRocksUtils.java:119) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils.toStarRocksTable(StarRocksUtils.java:93) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.applyCreateTable(StarRocksMetadataApplier.java:138) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.lambda$applySchemaChange$2(StarRocksMetadataApplier.java:115) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:57) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.applySchemaChange(StarRocksMetadataApplier.java:104) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:437) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:406) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:247) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
... 5 more
".

However, I consider this approach (using transform nodes for schema modification) suboptimal, as it requires configuring transform nodes for every table containing timestamptz columns.

Fundamentally, data type compatibility between source and sink should be handled by the Flink CDC framework itself.

[Mrart]

I was looking in my development log. In the past, we needed to convert the date type to varchar. Using cast directly (dt as varchar) is syntactically not supported because flink sql plan does not support cast in this case. You can write a udf in java to convert date-related fields to Strings.

[Mrart]