class StreamArrowPythonGroupWindowAggregateFunctionOperator extends AbstractArrowPythonAggregateFunctionOperator implements Triggerable<K, W> {

	@Override
	public void open() throws Exception {
		userDefinedFunctionOutputType = new RowType(
			outputType.getFields().subList(groupingSet.length, outputType.getFieldCount() - namedProperties.length));
		windowSerializer = windowAssigner.getWindowSerializer(new ExecutionConfig());

		internalTimerService = getInternalTimerService("window-timers", windowSerializer, this);

		triggerContext = new TriggerContext();
		triggerContext.open();

		StateDescriptor<ListState<RowData>, List<RowData>> windowStateDescriptor = new ListStateDescriptor<>(
			"window-input",
			new RowDataSerializer(inputType));
		StateDescriptor<ListState<RowData>, List<RowData>> dataRetractStateDescriptor = new ListStateDescriptor<>(
			"data-retract",
			new RowDataSerializer(inputType));
		this.windowAccumulateData = (InternalListState<K, W, RowData>) getOrCreateKeyedState(windowSerializer, windowStateDescriptor);
		this.windowRetractData = (InternalListState<K, W, RowData>) getOrCreateKeyedState(windowSerializer, dataRetractStateDescriptor);
		inputKeyAndWindow = new LinkedList<>();
		windowProperty = new GenericRowData(namedProperties.length);
		windowAggResult = new JoinedRowData();

		WindowContext windowContext = new WindowContext();
		windowAssigner.open(windowContext);
		super.open();
	}

	public  StreamArrowPythonGroupWindowAggregateFunctionOperator(
		Configuration config,
		PythonFunctionInfo[] pandasAggFunctions,
		RowType inputType,
		RowType outputType,
		int inputTimeFieldIndex,
		WindowAssigner<W> windowAssigner,
		Trigger<W> trigger,
		long allowedLateness,
		int[] namedProperties,
		int[] groupingSet,
		int[] udafInputOffsets);

}