apache / apache/doris-flink-connector
doris2.0.3、flink-doris-connector-1.14_2.12:1.1.1 在flink中建表时,DECIMAL类型和DATETIME类型无法映射到flink。出现空指针异常。
- Dominant language
- Java
- Stars
- 385
- Forks
- 279
- Avg merge
- 2d 2h
- Merged PRs (30d)
- 6
Description
### Search before asking
- [X] I had searched in the [issues](https://github.com/apache/incubator-doris/issues?q=is%3Aissue) and found no similar issues.
### Version
flink-doris-connector-1.14_2.12:1.1.1、doris2.0.3
### What's Wrong?
在flink中建表时,DECIMAL类型和DATETIME类型无法映射到flink。出现空指针异常。
建表如下:
CREATE TABLE IF NOT EXISTS `ods_project_proj_task_info` (
`id` BIGINT NOT NULL
,`version` INT
,`del_flag` BIGINT
,`remarks` STRING
,`create_by` BIGINT
,`create_date` TIMESTAMP
,`update_by` BIGINT
,`update_date` TIMESTAMP
,`tenant_id` STRING
,`task_name` STRING
,`node_id` BIGINT
,`parent_id` BIGINT
,`plan_start_time` TIMESTAMP
,`plan_end_time` TIMESTAMP
,`reality_start_time` TIMESTAMP
,`reality_end_time` TIMESTAMP
,`state` STRING
,`custom_columns` STRING
,`executor` BIGINT
,`files` STRING
,`priority` STRING
,`description` STRING
,`seq` INT
,`family_parent_ids` STRING
,`extend1` STRING
,`extend2` STRING
,`extend3` STRING
,`extend4` STRING
,`extend5` STRING
,`extend6` STRING
,`extend7` STRING
,`extend8` STRING
,`extend9` STRING
,`extend10` STRING
,`extend11` DECIMAL(18,7)
,`extend12` DECIMAL(18,7)
,`extend13` DECIMAL(18,7)
,`extend14` DECIMAL(18,7)
,`extend15` DECIMAL(18,7)
,`extend16` TIMESTAMP
,`extend17` TIMESTAMP
,`extend18` TIMESTAMP
,`extend19` TIMESTAMP
,`extend20` TIMESTAMP
,`plan_working_hours` DECIMAL(18,3)
,`reality_working_hours` DECIMAL(18,3)
,`task_identification` STRING
,`project_id` BIGINT
,`leaf_state` STRING
,`number_of_changes` INT
,`first_plan_start_time` TIMESTAMP
,`first_plan_end_time` TIMESTAMP
,`total_plan_working_hours` DECIMAL(18,3)
,`yield` DECIMAL(18,3)
,`parent_flag` STRING
,`task_rate` DECIMAL(18,3)
,`org_ids` STRING
,`task_type` STRING
,`device_id` BIGINT
,PRIMARY KEY ( `id` ) NOT ENFORCED
) WITH (
'connector' = 'doris',
'fenodes' = '192.168.7.45:8030',
'table.identifier' = 'ods.ods_project_proj_task_info',
'username' = 'root',
'password' = 'XXX'
);
报错如下:
Caused by: java.lang.RuntimeException: SplitFetcher thread 0 received unexpected exception while polling the records
at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:150)
at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:105)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
Caused by: java.lang.NullPointerException
at org.apache.doris.flink.rest.SchemaUtils.lambda$convertToSchema$0(SchemaUtils.java:36)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382)
at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:580)
at org.apache.doris.flink.rest.SchemaUtils.convertToSchema(SchemaUtils.java:36)
at org.apache.doris.flink.datastream.ScalaValueReader.(ScalaValueReader.scala:127)
at org.apache.doris.flink.source.reader.DorisSourceSplitReader.checkSplitOrStartNext(DorisSourceSplitReader.java:74)
at org.apache.doris.flink.source.reader.DorisSourceSplitReader.fetch(DorisSourceSplitReader.java:57)
at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58)
at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:142)
... 6 more
我看了源码之后,看到其中DECIMAL类型和TIMESTAMP类型的字段的type属性为空,导致的空指针异常
### What You Expected?
期望查找一下原因,修复这个连接器的bug。
### How to Reproduce?
_No response_
### Anything Else?
_No response_
### Are you willing to submit PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
Contributor guide
Research direction
Start with org.apache.doris.flink.rest.SchemaUtils.convertToSchema at SchemaUtils.java:36, then follow its use from ScalaValueReader.scala:127 while reproducing the supplied CREATE TABLE statement. Done means DECIMAL and TIMESTAMP fields no longer cause a null-pointer exception and are mapped successfully into the Flink schema.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, scala, sql
- Domain
- backend, database
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100