[
https://issues.apache.org/jira/browse/FLINK-15383?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu updated FLINK-15383:
----------------------------
Priority: Blocker (was: Major)
> StreamExecSink should use sink Schema field name rather than query Schema
> field name
> ------------------------------------------------------------------------------------
>
> Key: FLINK-15383
> URL: https://issues.apache.org/jira/browse/FLINK-15383
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 1.10.0
> Reporter: Leonard Xu
> Priority: Blocker
> Fix For: 1.10.0
>
>
> when user define a upsert table with primary key underlying(eg. mysql)
> {code:java}
> // primary key (log_per_min, item, currency_time)
> String sinkTableDDL = "CREATE TABLE gmv (\n" +
> " log_per_min STRING,\n" +
> " item STRING,\n" +
> " order_cnt BIGINT,\n" +
> " currency_time TIMESTAMP(3),\n" +
> " gmv DECIMAL(38, 18)," +
> " timestamp9 TIMESTAMP(6),\n" +
> " time9 TIME(6),\n" +
> " gdp DECIMAL(8, 4)\n" +
> ") WITH (\n" +
> " 'connector.type' = 'jdbc',\n" +
> " 'connector.url' = 'jdbc:mysql://localhost:3306/test',\n" +
> " 'connector.username' = 'root'," +
> " 'connector.table' = 'gmv',\n" +
> " 'connector.driver' = 'com.mysql.jdbc.Driver',\n" +
> " 'connector.write.flush.max-rows' = '5000', \n" +
> " 'connector.write.flush.interval' = '2s', \n" +
> " 'connector.write.max-retries' = '3'" +
> ")";
> {code}
> If user‘s query field name is different with sinktable's field name. For
> example, user defined a field `log_ts` which not equals `log_per_min` as
> following:
> {code:java}
> //
> insert into gmv \n" +
> "select log_ts,\n" +
> " item, COUNT(order_id) as order_cnt, currency_time,
> cast(sum(amount_kg) * max(rate) as DECIMAL(38, 4)) as gmv,\n" +
> " max(timestamp9), max(time9), max(gdp) \n" +
> " from ( \n" +
> " select cast(o.ts as VARCHAR) as log_ts, o.item as item, o.order_id
> as order_id, c.currency_time as currency_time,\n" +
> " o.amount_kg as amount_kg, c.rate as rate, c.timestamp9 as
> timestamp9, c.time9 as time9, c.gdp as gdp \n" +
> " from orders as o \n" +
> " join currency FOR SYSTEM_TIME AS OF o.proc_time c \n" +
> " on o.currency = c.currency_name \n" +
> " ) a group by log_ts, item, currency_time
> {code}
> The query will execute fail:
> {code:java}
> //
> Caused by: java.lang.ArrayIndexOutOfBoundsException: -1Caused by:
> java.lang.ArrayIndexOutOfBoundsException: -1 at
> org.apache.flink.api.java.io.jdbc.writer.UpsertWriter.lambda$create$0(UpsertWriter.java:59)
> at java.util.stream.IntPipeline$3$1.accept(IntPipeline.java:233) at
> java.util.Spliterators$IntArraySpliterator.forEachRemaining(Spliterators.java:1032)
> at java.util.Spliterator$OfInt.forEachRemaining(Spliterator.java:693) at
> java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
> at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:545) at
> java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
> at java.util.stream.IntPipeline.toArray(IntPipeline.java:502) at
> org.apache.flink.api.java.io.jdbc.writer.UpsertWriter.create(UpsertWriter.java:59)
> at
> org.apache.flink.api.java.io.jdbc.JDBCUpsertOutputFormat.open(JDBCUpsertOutputFormat.java:104)
> at
> org.apache.flink.api.java.io.jdbc.JDBCUpsertSinkFunction.open(JDBCUpsertSinkFunction.java:42)
> at
> org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
> at
> org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
> at
> org.apache.flink.streaming.api.operators.StreamSink.open(StreamSink.java:48)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:1023)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:454)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:449)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:461)
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:702) at
> org.apache.flink.runtime.taskmanager.Task.run(Task.java:527) at
> java.lang.Thread.run(Thread.java:748)
> {code}
> The root cause is that we should use sink Schema field name rather than query
> Schema field name in StreamExecSink
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
