[jira] [Created] (FLINK-29373) DataStream to table not support BigDecimalTypeInfo
hk__lrzy created FLINK-29373: Summary: DataStream to table not support BigDecimalTypeInfo Key: FLINK-29373 URL: https://issues.apache.org/jira/browse/FLINK-29373 Project: Flink Issue Type: Bug Components: Table SQL / Planner Affects Versions: 1.17.0 Reporter: hk__lrzy Attachments: image-2022-09-21-15-12-11-082.png When we want to transfrom datastream to table, `TypeInfoDataTypeConverter` will try to convert `TypeInformation` to `DataType`. But if datastream's produce types contains `BigDecimalTypeInfo`, `TypeInfoDataTypeConverter` will convert it `RawDataType`. Then when we want tranfrom table to datastream again, exception will hapend. !image-2022-09-21-15-12-11-082.png! -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-29944) Support accumaltor in source reader
hk__lrzy created FLINK-29944:
Summary: Support accumaltor in source reader
Key: FLINK-29944
URL: https://issues.apache.org/jira/browse/FLINK-29944
Project: Flink
Issue Type: Improvement
Components: API / Core
Affects Versions: 1.17.0
Reporter: hk__lrzy
Source Reader is mainly for union batch and streaming logic in single
interface, it's good point for the developer, but in the {{SourceFunction}} we
can access {{runtimeconext}} to use accumulator before, now the
{{SourceReaderContext}} have no method for it, this PR is mainly to support it.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (FLINK-35421) Schema Operator blocking forever when Akka Rpc timeout
hk__lrzy created FLINK-35421: Summary: Schema Operator blocking forever when Akka Rpc timeout Key: FLINK-35421 URL: https://issues.apache.org/jira/browse/FLINK-35421 Project: Flink Issue Type: Bug Components: Flink CDC Reporter: hk__lrzy -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-35432) Support catch modify-event for mysql-cdc
hk__lrzy created FLINK-35432:
Summary: Support catch modify-event for mysql-cdc
Key: FLINK-35432
URL: https://issues.apache.org/jira/browse/FLINK-35432
Project: Flink
Issue Type: Improvement
Components: Flink CDC
Reporter: hk__lrzy
Now users common use sql like to modify the column type in MySQL.
{code:java}
Alter table MODIFY COLUMN `new_name` new_type{code}
Flink-CDC use *CustomAlterTableParserListener* to parse the ddl and wrap it as
ChangeEevent now. But i noticed that *CustomAlterTableParserListener* not
implement the method *enterAlterByModifyColumn* and
{*}exitAlterByModifyColumn{*}, and it will cause we can't received the
*AlterColumnTypeEvent* now
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (FLINK-35436) Job can't launch when setting the option schema.change.behavior to IGNORE
hk__lrzy created FLINK-35436:
Summary: Job can't launch when setting the option
schema.change.behavior to IGNORE
Key: FLINK-35436
URL: https://issues.apache.org/jira/browse/FLINK-35436
Project: Flink
Issue Type: Improvement
Components: Flink CDC
Reporter: hk__lrzy
Now in the 3.0 pipeline, *SchemaOperator* already was necessary operator in
the Flink DAG, both *PrePartitionOperator* and *DataSinkWriterOperator* have
connection with the *SchemaRegister* according the *schemaEvolutionClient,* but
when we set the option schema.change.behavior to ignore or exception, the
pipeline will add a filter operator instead of the *Schema Operator,* final
cause the job fail.
I think we still need keep the option for the schema.change.behavior to meet
the difference cases, so i advice to move schema.change.behavior to the
*SchemaRegister* to let *SchemaOperator* will be always in the DAG, and let to
*SchemaRegister* decided to apply the schema change or not.{*}{*}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (FLINK-35463) CDC job failed when restored from checkpoint when route rule changed.
hk__lrzy created FLINK-35463:
Summary: CDC job failed when restored from checkpoint when route
rule changed.
Key: FLINK-35463
URL: https://issues.apache.org/jira/browse/FLINK-35463
Project: Flink
Issue Type: Improvement
Components: Flink CDC
Reporter: hk__lrzy
*Exception:*
{code:java}
java.lang.IllegalStateException: Unable to get latest schema for table
"demo.partition_all" {code}
*Reason:*
Now *ScheamRegister* will stored the all original tables which from source and
all derivered table which from route rules.
Assume we have a table in mysql which named *demo.partition_all* and have route
rule *demo1.partition_all* for it.
1. Before first checkpoint be triggered, *ScheamRegister* will store both
*demo.partition_all* and {*}demo1.partition_all{*}'s schema in the
{*}SchemaManager{*}.
2. Stop the job, change route rule as {*}demo1.partition_all_1{*}, and restart
it with checkpoint.
3. According the logical follow:
{code:java}
if (request.getSchemaChangeEvent() instanceof CreateTableEvent
&& schemaManager.schemaExists(request.getTableId())) {
return CompletableFuture.completedFuture(
wrap(new SchemaChangeResponse(Collections.emptyList(;
} {code}
We will not create a schema for the table *demo1.partition_all_1* and job will
be failed when we request schema from {*}SchemaOperator{*}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
