[jira] [Created] (FLINK-15928) Batch mode in blink planner caused IndexOutOfBoundsException error

["Fanbin Bu (Jira)" <ji...@apache.org>]

Fanbin Bu created FLINK-15928:
---------------------------------

             Summary: Batch mode in blink planner caused IndexOutOfBoundsException error
                 Key: FLINK-15928
                 URL: https://issues.apache.org/jira/browse/FLINK-15928
             Project: Flink
          Issue Type: Bug
    Affects Versions: 1.9.2
            Reporter: Fanbin Bu


Flink version: 1.9.2

mode: Batch mode, running on EMR with YARN

The following is the details:

 

table source sample:

class SnowflakeTableSource(val schema: TableSchema,
 val parallelism: Int,
 val fetchSize: Int,
 val query: String,
 val options: SnowflakeOptions
 )
 extends StreamTableSource[Row] {

override def getDataStream(execEnv: StreamExecutionEnvironment): SingleOutputStreamOperator[Row] = {
 execEnv.createInput(getInputFormat, getReturnType).name("app_event_stream")
}

override def getReturnType: TypeInformation[Row] = schema.toRowType

override def getTableSchema: TableSchema = schema

override def isBounded: Boolean = true

private def getInputFormat: JDBCInputFormat = {
 JDBCInputFormat.buildJDBCInputFormat
 .setDrivername(options.driverName)
 .setDBUrl(options.dbUrl)
 .setUsername(options.username)
 .setPassword(options.password)
 .setQuery(query)
 .setRowTypeInfo(getInputRowTypeInfo)
 .setFetchSize(fetchSize)
 .setParametersProvider(new GenericParameterValuesProvider(buildQueryParams(parallelism)))
 .finish
}

}

 

Here is the sample setup code:

val settings = EnvironmentSettings.newInstance()
 .useBlinkPlanner()
 .inBatchMode()
 .build()

val tableEnv = TableEnvironment.create(settings)
val configurations = tableEnv.getConfig.getConfiguration

configurations.setString(
 TABLE_EXEC_RESOURCE_HASH_AGG_MEMORY.key, s"${Globals.TABLE_EXEC_RESOURCE_HASH_AGG_MEMORY} mb")

tableEnv.registerTableSource(tableName, tableSource)

queryResult = tableEnv.sqlQuery(sql)

tableEnv.execute()

 

Here is the sample SQL:

select 
ip_address
 , hop_end(created_at, interval '30' second, interval '1' minute) as bucket_ts
 , sum(case when name = 'signin' then 1 else 0 end) as signin_count_1m
, sum(case when name = 'signin_failure' then 1 else 0 end) as signin_failure_count_1m

...
from events
group by
ip_address
 , hop(created_at, interval '30' second, interval '1' minute)

 

Here is the stacktrace:

java.lang.IndexOutOfBoundsException at org.apache.flink.core.memory.MemorySegment.getInt(MemorySegment.java:701) at org.apache.flink.table.dataformat.BinaryRow.getInt(BinaryRow.java:264) at HashWinAggWithKeys$538.endInput(Unknown Source) at org.apache.flink.streaming.runtime.tasks.OperatorChain.endInput(OperatorChain.java:276) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.checkFinished(StreamOneInputProcessor.java:151) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:138) at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:279) at org.apache.flink.streaming.runtime.tasks.StreamTask.run(StreamTask.java:301) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:406) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530) at java.lang.Thread.run(Thread.java:748)

 

The fact that this same code works well with other sql and the stacktrace message suggests that this might be related to memory issue. And this only happens for blink planner in batch mode. I tried to use BatchTableEnvironment in old planner and it works.

 

 

 



--
This message was sent by Atlassian Jira
(v8.3.4#803005)