[FLINK-6138] [table] Create the ListStateDescriptor with the aggregat…

…ionStateType instead of a serializer.

this closes apache#3581

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/UnboundedNonPartitionedProcessingOverProcessFunction.scala

@@ -98,9 +98,7 @@ class UnboundedNonPartitionedProcessingOverProcessFunction(
  }
 
  override def initializeState(context: FunctionInitializationContext): Unit = {
- val stateSerializer =
- aggregationStateType.createSerializer(getRuntimeContext.getExecutionConfig)
- val accumulatorsDescriptor = new ListStateDescriptor[Row]("overState", stateSerializer)
+ val accumulatorsDescriptor = new ListStateDescriptor[Row]("overState", aggregationStateType)
  state = context.getOperatorStateStore.getOperatorState(accumulatorsDescriptor)
  }
 }