Window Linear Mode use smaller buffers

[mustafasrepo]

Which issue does this PR close?

Closes #.

Rationale for this change

Background: In Linear mode of the BoundedWindowAggExec none of the partition by expressions are ordered. In these cases, we can generate result for a partition as long as a new row with same partition is received. Otherwise, result cannot be generated for the partition. As an example consider the table,

sn	hash
0	2
1	2
2	2
3	2
4	1
5	1
6	1
7	1
8	0
9	0

Assume following query is executed on this table

SELECT *, COUNT(*) OVER(PARTITION BY hash ORDER BY sn RANGE BETWEEN CURRENT ROW AND 1 FOLLOWING) as count
FROM test;

datafusion will generate following plan

"ProjectionExec: expr=[sn@0 as sn,  hash@1 as hash, COUNT([Column { name: \"sn\", index: 0 }]) PARTITION BY: [[Column { name: \"hash\", index: 1 }]], ORDER BY: [[PhysicalSortExpr { expr: Column { name: \"hash\", index: 1 }, options: SortOptions { descending: false, nulls_first: true } }]]@2 as count]",
"  BoundedWindowAggExec: wdw=[COUNT([Column { name: \"sn\", index: 0 }]) PARTITION BY: [[Column { name: \"hash\", index: 1 }]], ORDER BY: [[PhysicalSortExpr { expr: Column { name: \"duplicated_hash\", index: 3 }, options: SortOptions { descending: false, nulls_first: true } }]]: Ok(Field { name: \"COUNT([Column { name: \\\"sn\\\", index: 0 }]) PARTITION BY: [[Column { name: \\\"duplicated_hash\\\", index: 1 }]], ORDER BY: [[PhysicalSortExpr { expr: Column { name: \\\"hash\\\", index: 1 }, options: SortOptions { descending: false, nulls_first: true } }]]\", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: CurrentRow, end_bound: Following(UInt64(1)) }], mode=[Linear]",
"    SourceExec: partition_sizes=1, projection=[sn, hash], output_ordering=[sn@0 ASC NULLS LAST]",

where BoundedWindowAggExec is in Linear mode. Partition with hash=2 receives following section from the input table

sn	hash
0	2
1	2
2	2
3	2

query above can generate following result for the section above

sn	hash	count
0	2	2
1	2	2
2	2
3	2

Since in query we have range between unbounded preceding and 1 following for sn=2 we cannot generate result until sn=4 is received (where it is guaranteed end range for the sn=3). Same thing applies to other partitions with different hash values. However, from the input data, it can be seen that for the partition: hash=2 possible future rows cannot have sn=3,sn=4, etc. (where most recent data the input is sn=9). If we can use this information, we can generate early results for different partitions. Also this enables us to use less memory. With this information we can generate the following
result

sn	hash
0	2
1	2
2	2
3	2
4	1
5	1
6	1
7	1
8
9

instead of current behaviour with result

sn	hash
0	2
1	2
2
3
4	1
5	1
6
7
8
9

This enables to use less memory when cardinality is high for partition by expressions, and window frame query is either RANGE or GROUPS query (For ROWS queries we need the new row that belong to same partition anyway.).

What changes are included in this PR?

Are these changes tested?

Yes.

Are there any user-facing changes?

[mustafasrepo]

these changes introduced to be able to test BoundedWindowAggExec with current row bound.

[ozankabak]

I reviewed this carefully and LGTM

[ozankabak]

There seems to be an issue with how we print plans making CI fail. I will let @mustafasrepo fix it and then merge