apache / doris-flink-connector

Flink Connector for Apache Doris
https://doris.apache.org/
Apache License 2.0
315 stars 221 forks source link

[Fix](source) fix IndexOutOfBoundsException for union all #495

Closed JNSimba closed 1 week ago

JNSimba commented 1 week ago

Proposed changes

Issue Number: close #xxx

Problem Summary:

Sometimes when using flinksql for unionall, the following error is reported

select * from doris_source where name in ('lisi','wangwu') union all SELECT * from doris_source where name = 'zhangsan' 
Caused by: java.lang.IndexOutOfBoundsException: Index: 1, Size: 1
    at java.util.ArrayList.rangeCheck(ArrayList.java:659)
    at java.util.ArrayList.get(ArrayList.java:435)
    at org.apache.doris.flink.deserialization.converter.DorisRowConverter.convertInternal(DorisRowConverter.java:109)
    at org.apache.doris.flink.deserialization.RowDataDeserializationSchema.deserialize(RowDataDeserializationSchema.java:51)
    at org.apache.doris.flink.source.reader.DorisRecordEmitter.emitRecord(DorisRecordEmitter.java:44)
    at org.apache.doris.flink.source.reader.DorisRecordEmitter.emitRecord(DorisRecordEmitter.java:30)
    at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:203)
    at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:422)
    at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
    at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:638)
    at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:973)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:917)
    at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:970)
    at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:949)
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763)

Checklist(Required)

  1. Does it affect the original behavior: (Yes/No/I Don't know)
  2. Has unit tests been added: (Yes/No/No Need)
  3. Has document been added or modified: (Yes/No/No Need)
  4. Does it need to update dependencies: (Yes/No)
  5. Are there any changes that cannot be rolled back: (Yes/No)

Further comments

If this is a relatively large or complex change, kick off the discussion at dev@doris.apache.org by explaining why you chose the solution you did and what alternatives you considered, etc...