[jira] [Created] (FLINK-15563) When using ParquetTableSource, The program hangs until OOM

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

sujun created FLINK-15563:
-----------------------------

             Summary:  When using ParquetTableSource, The program hangs until OOM
                 Key: FLINK-15563
                 URL: https://issues.apache.org/jira/browse/FLINK-15563
             Project: Flink
          Issue Type: Bug
          Components: Connectors / FileSystem
    Affects Versions: 1.9.1, 1.8.1
            Reporter: sujun


def main(args: Array[String]): Unit = \{
val env = StreamExecutionEnvironment.getExecutionEnvironment
 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
 val tableEnv = StreamTableEnvironment.create(env)
val schema = "{\"type\":\"record\",\"name\":\"root\",\"fields\":[{\"name\":\"log_id\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"city\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"log_from\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"ip\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"type\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"data_source\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"is_scan\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"result\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"timelong\",\"type\":[\"null\",\"long\"],\"default\":null},\{\"name\":\"is_sec\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"event_name\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"id\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"time_string\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"device\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"timestamp_string\",\"type\":[\"null\",\"string\"],\"default\":null},\{\"name\":\"occur_time\",\"type\":[\"null\",{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"}],\"default\":null},\{\"name\":\"row_time\",\"type\":[\"null\",{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"}],\"default\":null}]}"**
 val parquetTableSource: ParquetTableSource = ParquetTableSource
 .builder
 .forParquetSchema(new org.apache.parquet.avro.AvroSchemaConverter().convert(
 org.apache.avro.Schema.parse(schema, true)))
 .path("/path/to/login_data")
 .build
tableEnv.registerTableSource("source",parquetTableSource)

 val t1 = tableEnv.sqlQuery("select log_id,city from source where city = '274' ")
 tableEnv.registerTable("t1",t1)
val t4 = tableEnv.sqlQuery("select * from t1 where log_id='5927070661978133'")
 t1.toAppendStream[Row].print()
env.execute()
}



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