Merge pull request #11788 from ydb-platform/mergelibs-241120-1113

Library import 241120-1113

56 files changed, 1030 insertions, 0 deletions

diff --git a/yql/essentials/tests/sql/suites/aggr_factory/avg.sql b/yql/essentials/tests/sql/suites/aggr_factory/avg.sql
new file mode 100644
index 0000000000..1352062811
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/avg.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("avg");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/avg_distinct_expr.sql b/yql/essentials/tests/sql/suites/aggr_factory/avg_distinct_expr.sql
new file mode 100644
index 0000000000..984c21a13f
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/avg_distinct_expr.sql
@@ -0,0 +1,10 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a), AsStruct(1 as a));
+$f = AGGREGATION_FACTORY("avg");
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+
+select AGGREGATE_BY(distinct cast(Unicode::ToLower(cast(a as Utf8) || "00"u) as Int), $f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/avg_if.sql b/yql/essentials/tests/sql/suites/aggr_factory/avg_if.sql
new file mode 100644
index 0000000000..7fc60b5b1c
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/avg_if.sql
@@ -0,0 +1,14 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("avg_if");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return AsTuple($z.a,$z.a<2)}))));
+
+use plato;
+insert into @a select AsTuple(a,a<2) as aa from as_table($t);
+commit;
+select AGGREGATE_BY(aa,$f) from @a;
+
+
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/bitand.sql b/yql/essentials/tests/sql/suites/aggr_factory/bitand.sql
new file mode 100644
index 0000000000..419b6bf336
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/bitand.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1u as a),AsStruct(2u as a));
+$f = AGGREGATION_FACTORY("bitand");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/bitor.sql b/yql/essentials/tests/sql/suites/aggr_factory/bitor.sql
new file mode 100644
index 0000000000..f9808d27c6
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/bitor.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1u as a),AsStruct(2u as a));
+$f = AGGREGATION_FACTORY("bitor");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/bitxor.sql b/yql/essentials/tests/sql/suites/aggr_factory/bitxor.sql
new file mode 100644
index 0000000000..fcb872944f
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/bitxor.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1u as a),AsStruct(2u as a));
+$f = AGGREGATION_FACTORY("bitxor");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/booland.sql b/yql/essentials/tests/sql/suites/aggr_factory/booland.sql
new file mode 100644
index 0000000000..03b483ee5d
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/booland.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(true as a),AsStruct(false as a));
+$f = AGGREGATION_FACTORY("booland");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/boolor.sql b/yql/essentials/tests/sql/suites/aggr_factory/boolor.sql
new file mode 100644
index 0000000000..c54cac9205
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/boolor.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(true as a),AsStruct(false as a));
+$f = AGGREGATION_FACTORY("boolor");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/bottom.sql b/yql/essentials/tests/sql/suites/aggr_factory/bottom.sql
new file mode 100644
index 0000000000..f663c39d63
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/bottom.sql
@@ -0,0 +1,24 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(
+    AsStruct(7 as a),
+    AsStruct(4 as a),
+    AsStruct(5 as a),
+    AsStruct(1 as a),
+    AsStruct(2 as a),
+    AsStruct(9 as a),
+    AsStruct(1 as a),
+    AsStruct(9 as a)
+);
+
+$f = AGGREGATION_FACTORY("bottom", 3);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"),
+    $f(ListItemType(TypeOf($t)), ($z)->{ return $z.a }))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+
+select AGGREGATE_BY(a, $f) from @a;
+select AGGREGATE_BY(distinct a, $f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/bottom_by.sql b/yql/essentials/tests/sql/suites/aggr_factory/bottom_by.sql
new file mode 100644
index 0000000000..f5ee53bd8c
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/bottom_by.sql
@@ -0,0 +1,21 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(
+    AsStruct(1 as key, 101 as value),
+    AsStruct(6 as key, 34 as value),
+    AsStruct(4 as key, 22 as value),
+    AsStruct(2 as key, 256 as value),
+    AsStruct(7 as key, 111 as value)
+);
+
+$f = AGGREGATION_FACTORY("bottomby", 3);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"),
+    $f(ListItemType(TypeOf($t)), ($z)->{ return AsTuple($z.value, $z.key) }))));
+
+use plato;
+insert into @a select AsTuple(value, key) as vk from as_table($t);
+commit;
+
+select AGGREGATE_BY(vk, $f) from @a;
+
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/container.sql b/yql/essentials/tests/sql/suites/aggr_factory/container.sql
new file mode 100644
index 0000000000..303e3035f1
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/container.sql
@@ -0,0 +1,61 @@
+/* syntax version 1 */
+/* postgres can not */
+$factory = AGGREGATION_FACTORY("sum");
+
+select ListAggregate(ListCreate(Int32), $factory);
+select ListAggregate(AsList(1, 2, 3), $factory);
+select ListAggregate(Just(AsList(1, 2, 3)), $factory);
+select ListAggregate(Nothing(ParseType("List<Int32>?")), $factory);
+
+$factory = AGGREGATION_FACTORY("count");
+
+select ListAggregate(ListCreate(Int32), $factory);
+select ListAggregate(AsList(1, 2, 3), $factory);
+select ListAggregate(Just(AsList(1, 2, 3)), $factory);
+select ListAggregate(Nothing(ParseType("List<Int32>?")), $factory);
+
+$factory = AGGREGATION_FACTORY("sum");
+
+select ListSort(DictItems(DictAggregate(
+    DictCreate(ParseType("String"), ParseType("List<Int32>"))
+    , $factory)));
+
+select ListSort(DictItems(DictAggregate(
+    AsDict(
+        AsTuple("foo", AsList(1, 3)), 
+        AsTuple("bar", AsList(2))
+    ), $factory)));
+
+
+select ListSort(DictItems(DictAggregate(
+    Just(AsDict(
+        AsTuple("foo", AsList(1, 3)), 
+        AsTuple("bar", AsList(2))
+    )), $factory)));
+
+select ListSort(DictItems(DictAggregate(
+    Nothing(ParseType("Dict<String, List<Int32>>?"))
+    , $factory)));
+
+$factory = AGGREGATION_FACTORY("count");
+
+select ListSort(DictItems(DictAggregate(
+    DictCreate(ParseType("String"), ParseType("List<Int32>"))
+    , $factory)));
+
+select ListSort(DictItems(DictAggregate(
+    AsDict(
+        AsTuple("foo", AsList(1, 3)), 
+        AsTuple("bar", AsList(2))
+    ), $factory)));
+
+
+select ListSort(DictItems(DictAggregate(
+    Just(AsDict(
+        AsTuple("foo", AsList(1, 3)), 
+        AsTuple("bar", AsList(2))
+    )), $factory)));
+
+select ListSort(DictItems(DictAggregate(
+    Nothing(ParseType("Dict<String, List<Int32>>?"))
+    , $factory)));
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/container_empty.sql b/yql/essentials/tests/sql/suites/aggr_factory/container_empty.sql
new file mode 100644
index 0000000000..d76fe8317f
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/container_empty.sql
@@ -0,0 +1,10 @@
+/* syntax version 1 */
+/* postgres can not */
+select ListAggregate([],AGGREGATION_FACTORY("sum"));
+select ListAggregate([1,2],AGGREGATION_FACTORY("sum"));
+select ListAggregate(Just([1,2]),AGGREGATION_FACTORY("sum"));
+select ListAggregate(null,AGGREGATION_FACTORY("sum"));
+select DictAggregate({},AGGREGATION_FACTORY("sum"));
+select DictAggregate({'a':[2,3]},AGGREGATION_FACTORY("sum"));
+select DictAggregate(Just({'a':[2,3]}),AGGREGATION_FACTORY("sum"));
+select DictAggregate(null,AGGREGATION_FACTORY("sum"));
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/corellation.sql b/yql/essentials/tests/sql/suites/aggr_factory/corellation.sql
new file mode 100644
index 0000000000..d42c5259ce
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/corellation.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("correlation");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return AsTuple($z.a,-$z.a)}))));
+
+use plato;
+insert into @a select AsTuple(a,-a) as aa from as_table($t);
+commit;
+select AGGREGATE_BY(aa,$f) from @a;
+
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/count.sql b/yql/essentials/tests/sql/suites/aggr_factory/count.sql
new file mode 100644
index 0000000000..3e209eb11f
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/count.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("count");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/count_if.sql b/yql/essentials/tests/sql/suites/aggr_factory/count_if.sql
new file mode 100644
index 0000000000..a3dbd155b0
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/count_if.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(true as a),AsStruct(false as a));
+$f = AGGREGATION_FACTORY("countif");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/def_value_full_table.sql b/yql/essentials/tests/sql/suites/aggr_factory/def_value_full_table.sql
new file mode 100644
index 0000000000..2d2af73022
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/def_value_full_table.sql
@@ -0,0 +1,63 @@
+$my_table =
+SELECT
+    1 AS id
+    , 1 AS ts
+    , 4 AS value1
+    , 5 AS value2
+UNION ALL
+SELECT
+    3 AS id
+    , 10 AS ts
+    , 40 AS value1
+    , NULL AS value2
+UNION ALL
+SELECT
+    2 AS id
+    , 1 AS ts
+    , NULL AS value1
+    , NULL AS value2
+UNION ALL
+SELECT
+    1 AS id
+    , 2 AS ts
+    , 4 AS value1
+    , 5 AS value2
+UNION ALL
+SELECT
+    3 AS id
+    , 2 AS ts
+    , 40 AS value1
+    , NULL AS value2
+UNION ALL
+SELECT
+    3 AS id
+    , 5 AS ts
+    , 2 AS value1
+    , 7 AS value2
+;
+
+-- Эмуляция агрегационной функции COUNT
+$cnt_create = ($_item, $_parent) -> { return 1ul };
+$cnt_add = ($state, $_item, $_parent) -> { return 1ul + $state };
+$cnt_merge = ($state1, $state2) -> { return $state1 + $state2 };
+$cnt_get_result = ($state) -> { return $state };
+$cnt_serialize = ($state) -> { return $state };
+$cnt_deserialize = ($state) -> { return $state };
+$cnt_default = 0l;
+
+$cnt_udaf_factory = AggregationFactory(
+	    "UDAF",
+	    $cnt_create,
+	    $cnt_add,
+	    $cnt_merge,
+	    $cnt_get_result,
+	    $cnt_serialize,
+	    $cnt_deserialize,
+	    $cnt_default
+);
+
+
+SELECT
+    AGGREGATE_BY(value1, $cnt_udaf_factory) AS cnt1
+FROM $my_table
+;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/def_value_with_keys.sql b/yql/essentials/tests/sql/suites/aggr_factory/def_value_with_keys.sql
new file mode 100644
index 0000000000..848ce00cf0
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/def_value_with_keys.sql
@@ -0,0 +1,66 @@
+$my_table =
+SELECT
+    1 AS id
+    , 1 AS ts
+    , 4 AS value1
+    , 5 AS value2
+UNION ALL
+SELECT
+    3 AS id
+    , 10 AS ts
+    , 40 AS value1
+    , NULL AS value2
+UNION ALL
+SELECT
+    2 AS id
+    , 1 AS ts
+    , NULL AS value1
+    , NULL AS value2
+UNION ALL
+SELECT
+    1 AS id
+    , 2 AS ts
+    , 4 AS value1
+    , 5 AS value2
+UNION ALL
+SELECT
+    3 AS id
+    , 2 AS ts
+    , 40 AS value1
+    , NULL AS value2
+UNION ALL
+SELECT
+    3 AS id
+    , 5 AS ts
+    , 2 AS value1
+    , 7 AS value2
+;
+
+-- Эмуляция агрегационной функции COUNT
+$cnt_create = ($_item, $_parent) -> { return 1ul };
+$cnt_add = ($state, $_item, $_parent) -> { return 1ul + $state };
+$cnt_merge = ($state1, $state2) -> { return $state1 + $state2 };
+$cnt_get_result = ($state) -> { return $state };
+$cnt_serialize = ($state) -> { return $state };
+$cnt_deserialize = ($state) -> { return $state };
+$cnt_default = 0ul;
+
+$cnt_udaf_factory = AggregationFactory(
+	    "UDAF",
+	    $cnt_create,
+	    $cnt_add,
+	    $cnt_merge,
+	    $cnt_get_result,
+	    $cnt_serialize,
+	    $cnt_deserialize,
+	    $cnt_default
+);
+
+
+SELECT
+    id,
+    AGGREGATE_BY(value1, $cnt_udaf_factory) AS cnt1
+FROM $my_table
+GROUP BY
+    id
+;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/default.cfg b/yql/essentials/tests/sql/suites/aggr_factory/default.cfg
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/default.cfg
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/every.sql b/yql/essentials/tests/sql/suites/aggr_factory/every.sql
new file mode 100644
index 0000000000..d44ae81d3b
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/every.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(true as a),AsStruct(false as a));
+$f = AGGREGATION_FACTORY("every");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/flatten.cfg b/yql/essentials/tests/sql/suites/aggr_factory/flatten.cfg
new file mode 100644
index 0000000000..8c22472a41
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/flatten.cfg
@@ -0,0 +1 @@
+udf set_udf
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/flatten.sql b/yql/essentials/tests/sql/suites/aggr_factory/flatten.sql
new file mode 100644
index 0000000000..e4b8be9bce
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/flatten.sql
@@ -0,0 +1,21 @@
+/* syntax version 1 */
+/* postgres can not */
+$f = AGGREGATION_FACTORY("sum");
+$g = AggregateFlatten($f);
+select
+  ListAggregate([1,2,3], $f),
+  ListAggregate(ListCreate(List<Int32>), $g),
+  ListAggregate([ListCreate(Int32)], $g),
+  ListAggregate([ListCreate(Int32),ListCreate(Int32)], $g),
+  ListAggregate([[1,2]], $g),
+  ListAggregate([[1,2],[3]], $g),
+  ListAggregate([ListCreate(Int32),[3]], $g),
+  ListAggregate([[1,2],ListCreate(Int32)], $g);
+
+$i = AGGREGATION_FACTORY("AGGREGATE_LIST_DISTINCT");
+$j = AggregateFlatten($i);
+select AggregateBy(x, $j) from (
+   select [1,2] as x 
+   union all
+   select [2,3] as x
+);
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/histogram.sql b/yql/essentials/tests/sql/suites/aggr_factory/histogram.sql
new file mode 100644
index 0000000000..1a058a8f2a
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/histogram.sql
@@ -0,0 +1,17 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("histogram");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return AsTuple($z.a,1.0)}))));
+
+$f = AGGREGATION_FACTORY("histogram", 5);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return AsTuple($z.a,1.0)}))));
+
+use plato;
+insert into @a select AsTuple(a, 1.0) as aa from as_table($t);
+commit;
+select AGGREGATE_BY(aa,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/hll.sql b/yql/essentials/tests/sql/suites/aggr_factory/hll.sql
new file mode 100644
index 0000000000..9a2c1f0716
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/hll.sql
@@ -0,0 +1,18 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("hll");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+$f = AGGREGATION_FACTORY("hll", 4);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/input.txt b/yql/essentials/tests/sql/suites/aggr_factory/input.txt
new file mode 100644
index 0000000000..70dd8b1eff
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/input.txt
@@ -0,0 +1,4 @@
+{"key"=1;"subkey"=10;"value"=5};
+{"key"=1;"subkey"=20;"value"=6};
+{"key"=1;"subkey"=30;"value"=7};
+{"key"=0;"subkey"=40;"value"=8};
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/input.txt.attr b/yql/essentials/tests/sql/suites/aggr_factory/input.txt.attr
new file mode 100644
index 0000000000..84791918db
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/input.txt.attr
@@ -0,0 +1,30 @@
+{
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "Int32"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "Int32"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "Int32"
+                    ]
+                ]
+            ]
+        ]
+    };
+}
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/linear_histogram.sql b/yql/essentials/tests/sql/suites/aggr_factory/linear_histogram.sql
new file mode 100644
index 0000000000..68db6ad402
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/linear_histogram.sql
@@ -0,0 +1,18 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("linearhistogram");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+$f = AGGREGATION_FACTORY("linearhistogram", 10, 0.0, 1000.0);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select a as aa from as_table($t);
+commit;
+select AGGREGATE_BY(aa,$f) from @a;
+
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/list.cfg b/yql/essentials/tests/sql/suites/aggr_factory/list.cfg
new file mode 100644
index 0000000000..5c248bff3d
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/list.cfg
@@ -0,0 +1 @@
+udf string_udf
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/list.sql b/yql/essentials/tests/sql/suites/aggr_factory/list.sql
new file mode 100644
index 0000000000..bc43f37032
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/list.sql
@@ -0,0 +1,18 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("aggregate_list");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+$f = AGGREGATION_FACTORY("aggregate_list", length(CAST(Unicode::ToUpper("xx"u) AS String)));
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select ListSort(AGGREGATE_BY(a,$f)) from @a;
+select ListSort(AGGREGATE_BY(distinct a,$f)) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/log_histogram.sql b/yql/essentials/tests/sql/suites/aggr_factory/log_histogram.sql
new file mode 100644
index 0000000000..1b5bd50199
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/log_histogram.sql
@@ -0,0 +1,18 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("loghistogram");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+$f = AGGREGATION_FACTORY("loghistogram", 10, 0.01, 1000.0);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select a as aa from as_table($t);
+commit;
+select AGGREGATE_BY(aa,$f) from @a;
+
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/logariphmic_histogram.sql b/yql/essentials/tests/sql/suites/aggr_factory/logariphmic_histogram.sql
new file mode 100644
index 0000000000..a11941be69
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/logariphmic_histogram.sql
@@ -0,0 +1,18 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("logarithmichistogram");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+$f = AGGREGATION_FACTORY("logarithmichistogram", 10, 0.01, 1000.0);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select a as aa from as_table($t);
+commit;
+select AGGREGATE_BY(aa,$f) from @a;
+
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/max.sql b/yql/essentials/tests/sql/suites/aggr_factory/max.sql
new file mode 100644
index 0000000000..a0e854663c
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/max.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("max");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/max_by.sql b/yql/essentials/tests/sql/suites/aggr_factory/max_by.sql
new file mode 100644
index 0000000000..0e7cfaf37f
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/max_by.sql
@@ -0,0 +1,21 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(
+    AsStruct(1 as key, 200 as value),
+    AsStruct(2 as key, 100 as value)
+);
+
+$f = AGGREGATION_FACTORY("maxby");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"),
+    $f(ListItemType(TypeOf($t)), ($z)->{ return AsTuple($z.value, $z.key) }))));
+
+$f = AGGREGATION_FACTORY("maxby", 10);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"),
+    $f(ListItemType(TypeOf($t)), ($z)->{ return AsTuple($z.value, $z.key) }))));
+
+use plato;
+insert into @a select AsTuple(value, key) as vk from as_table($t);
+commit;
+select AGGREGATE_BY(vk, $f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/median.sql b/yql/essentials/tests/sql/suites/aggr_factory/median.sql
new file mode 100644
index 0000000000..19921bbcbd
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/median.sql
@@ -0,0 +1,18 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("median");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+$f = AGGREGATION_FACTORY("percentile", 0.9);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/min.sql b/yql/essentials/tests/sql/suites/aggr_factory/min.sql
new file mode 100644
index 0000000000..d101099550
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/min.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("min");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/min_by.sql b/yql/essentials/tests/sql/suites/aggr_factory/min_by.sql
new file mode 100644
index 0000000000..4b1d240aa0
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/min_by.sql
@@ -0,0 +1,21 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(
+    AsStruct(1 as key, 200 as value),
+    AsStruct(2 as key, 100 as value)
+);
+
+$f = AGGREGATION_FACTORY("minby");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"),
+    $f(ListItemType(TypeOf($t)), ($z)->{ return AsTuple($z.value, $z.key) }))));
+
+$f = AGGREGATION_FACTORY("minby", 10);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"),
+    $f(ListItemType(TypeOf($t)), ($z)->{ return AsTuple($z.value, $z.key) }))));
+
+use plato;
+insert into @a select AsTuple(value, key) as vk from as_table($t);
+commit;
+select AGGREGATE_BY(vk, $f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/mode.sql b/yql/essentials/tests/sql/suites/aggr_factory/mode.sql
new file mode 100644
index 0000000000..496cd4e4d8
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/mode.sql
@@ -0,0 +1,18 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("mode");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+$f = AGGREGATION_FACTORY("topfreq", 10, 20);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select ListSort(AGGREGATE_BY(distinct a,$f), ($x)->{ return $x.Value }) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/multi.cfg b/yql/essentials/tests/sql/suites/aggr_factory/multi.cfg
new file mode 100644
index 0000000000..fad550d3a9
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/multi.cfg
@@ -0,0 +1,3 @@
+in Input input.txt
+udf top_udf
+udf set_udf
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/multi.sql b/yql/essentials/tests/sql/suites/aggr_factory/multi.sql
new file mode 100644
index 0000000000..e6b0638f14
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/multi.sql
@@ -0,0 +1,22 @@
+/* syntax version 1 */
+/* postgres can not */
+USE plato;
+
+$input = (select AsStruct(
+    key as key,
+    Just(subkey) as subkey,
+    Just(value) as value) as nums from Input);
+
+SELECT
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("count")) as count,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("min")) as min,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("max")) as max,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("sum")) as sum,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("avg")) as avg,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("stddev")) as stddev,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("percentile", 0.5)) as p50,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("aggregate_list")) as agg_list,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("aggregate_list_distinct")) as agg_list_distinct,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("mode")) as mode,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("top", 3)) as top,
+FROM $input;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/multi_list.sql b/yql/essentials/tests/sql/suites/aggr_factory/multi_list.sql
new file mode 100644
index 0000000000..7a142eb664
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/multi_list.sql
@@ -0,0 +1,24 @@
+/* syntax version 1 */
+/* postgres can not */
+
+$input = 
+select AsList(
+    1,
+    2,
+    3
+) as nums
+union all
+select AsList(
+    4,
+    5) as nums;
+
+SELECT
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("count")) as count,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("min")) as min,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("max")) as max,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("sum")) as sum,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("avg")) as avg,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("stddev")) as stddev,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("percentile", 0.5)) as p50,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("aggregate_list")) as agg_list
+FROM $input;
\ No newline at end of file
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/multi_list_distinct_expr.sql b/yql/essentials/tests/sql/suites/aggr_factory/multi_list_distinct_expr.sql
new file mode 100644
index 0000000000..4795014fa6
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/multi_list_distinct_expr.sql
@@ -0,0 +1,28 @@
+/* syntax version 1 */
+/* postgres can not */
+
+$input = 
+select AsList(
+    1,
+    2,
+    3
+) as nums
+union all
+select AsList(
+    4,
+    5) as nums
+union all
+select AsList(
+    1,
+    2,
+    3
+) as nums;
+
+SELECT
+    MULTI_AGGREGATE_BY(distinct ListExtend(nums, AsList(1,5)), AGGREGATION_FACTORY("count")) as count,
+    MULTI_AGGREGATE_BY(distinct ListExtend(nums, AsList(1,5)), AGGREGATION_FACTORY("min")) as min,
+    MULTI_AGGREGATE_BY(distinct ListExtend(nums, AsList(1,5)), AGGREGATION_FACTORY("max")) as max,
+    MULTI_AGGREGATE_BY(distinct ListExtend(nums, AsList(1,5)), AGGREGATION_FACTORY("sum")) as sum,
+    MULTI_AGGREGATE_BY(distinct ListExtend(nums, AsList(1,5)), AGGREGATION_FACTORY("percentile", 0.5)) as p50,
+    MULTI_AGGREGATE_BY(distinct ListExtend(nums, AsList(1,5)), AGGREGATION_FACTORY("aggregate_list")) as agg_list
+FROM $input;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/multi_list_nulls.sql b/yql/essentials/tests/sql/suites/aggr_factory/multi_list_nulls.sql
new file mode 100644
index 0000000000..103e92c047
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/multi_list_nulls.sql
@@ -0,0 +1,24 @@
+/* syntax version 1 */
+/* postgres can not */
+
+$data = AsList(
+    AsStruct(AsList(1.0, 2.0) as x),
+    AsStruct(AsList(3.0, 4.0) as x),
+    AsStruct(AsList(NULL, NULL) as x),
+    AsStruct(AsList(2.0, 3.0, 4.0) as x),
+);
+
+SELECT  
+    MULTI_AGGREGATE_BY(x, AggregationFactory("agg_list")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("avg")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("count")),
+FROM
+    AS_TABLE($data);
+
+SELECT
+    MULTI_AGGREGATE_BY(x, AggregationFactory("agg_list")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("avg")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("count")),
+FROM
+    (SELECT * FROM AS_TABLE($data) LIMIT 0);
+
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/multi_minmaxby.sql b/yql/essentials/tests/sql/suites/aggr_factory/multi_minmaxby.sql
new file mode 100644
index 0000000000..4d8628971b
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/multi_minmaxby.sql
@@ -0,0 +1,10 @@
+/* syntax version 1 */
+/* postgres can not */
+SELECT
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("minby"))
+FROM (select TableRow() as nums from AS_TABLE([<|x:(1,6)|>,<|x:(3,4)|>,<|x:(5,2)|>]));
+
+
+SELECT
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("maxby", 2))
+FROM (select TableRow() as nums from AS_TABLE([<|x:(1,6)|>,<|x:(3,4)|>,<|x:(5,2)|>]));
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/multi_struct_nulls.sql b/yql/essentials/tests/sql/suites/aggr_factory/multi_struct_nulls.sql
new file mode 100644
index 0000000000..11f7139d11
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/multi_struct_nulls.sql
@@ -0,0 +1,23 @@
+/* syntax version 1 */
+/* postgres can not */
+
+$data = AsList(
+    AsStruct(AsStruct(1.0 as a, 2 as b) as x),
+    AsStruct(AsStruct(3.0 as a, 4 as b) as x),
+    AsStruct(AsStruct(NULL as a, NULL as b) as x),
+    AsStruct(AsStruct(2.0 as a, 3 as b, 4.0 as c) as x),
+);
+
+SELECT  
+    MULTI_AGGREGATE_BY(x, AggregationFactory("agg_list")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("avg")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("count")),
+FROM
+    AS_TABLE($data);
+
+SELECT
+    MULTI_AGGREGATE_BY(x, AggregationFactory("agg_list")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("avg")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("count")),
+FROM
+    (SELECT * FROM AS_TABLE($data) LIMIT 0);
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/multi_tuple.sql b/yql/essentials/tests/sql/suites/aggr_factory/multi_tuple.sql
new file mode 100644
index 0000000000..f12d7460a2
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/multi_tuple.sql
@@ -0,0 +1,23 @@
+/* syntax version 1 */
+/* postgres can not */
+
+$input = 
+select AsTuple(
+    1,
+    Just(2),
+    Just(3)) as nums
+union all
+select AsTuple(4,
+    Just(5),
+    Just(6)) as nums;
+
+SELECT
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("count")) as count,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("min")) as min,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("max")) as max,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("sum")) as sum,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("avg")) as avg,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("stddev")) as stddev,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("percentile", 0.5)) as p50,
+    MULTI_AGGREGATE_BY(nums, AGGREGATION_FACTORY("aggregate_list")) as agg_list,
+FROM $input;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/multi_tuple_nulls.sql b/yql/essentials/tests/sql/suites/aggr_factory/multi_tuple_nulls.sql
new file mode 100644
index 0000000000..6760707800
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/multi_tuple_nulls.sql
@@ -0,0 +1,24 @@
+/* syntax version 1 */
+/* postgres can not */
+
+$data = AsList(
+    AsStruct(AsTuple(1.0, 2) as x),
+    AsStruct(AsTuple(3.0, 4) as x),
+    AsStruct(AsTuple(NULL, NULL) as x),
+    AsStruct(AsTuple(2.0, 3) as x),
+);
+
+SELECT  
+    MULTI_AGGREGATE_BY(x, AggregationFactory("agg_list")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("avg")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("count")),
+FROM
+    AS_TABLE($data);
+
+SELECT
+    MULTI_AGGREGATE_BY(x, AggregationFactory("agg_list")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("avg")),
+    MULTI_AGGREGATE_BY(x, AggregationFactory("count")),
+FROM
+    (SELECT * FROM AS_TABLE($data) LIMIT 0);
+
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/some.sql b/yql/essentials/tests/sql/suites/aggr_factory/some.sql
new file mode 100644
index 0000000000..83f5a52a29
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/some.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(1 as a));
+$f = AGGREGATION_FACTORY("some");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/stddev.sql b/yql/essentials/tests/sql/suites/aggr_factory/stddev.sql
new file mode 100644
index 0000000000..7e46f46e5c
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/stddev.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("stddev");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/sum_if.sql b/yql/essentials/tests/sql/suites/aggr_factory/sum_if.sql
new file mode 100644
index 0000000000..b48b38e153
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/sum_if.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("sum_if");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return AsTuple($z.a,$z.a>1)}))));
+
+use plato;
+insert into @a select AsTuple(a,a>1) as aa from as_table($t);
+commit;
+select AGGREGATE_BY(aa,$f) from @a;
+
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/top.sql b/yql/essentials/tests/sql/suites/aggr_factory/top.sql
new file mode 100644
index 0000000000..dd50c1351f
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/top.sql
@@ -0,0 +1,23 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(
+    AsStruct(7 as a),
+    AsStruct(4 as a),
+    AsStruct(5 as a),
+    AsStruct(1 as a),
+    AsStruct(2 as a),
+    AsStruct(9 as a),
+    AsStruct(1 as a),
+    AsStruct(9 as a));
+
+$f = AGGREGATION_FACTORY("top", 3);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"),
+    $f(ListItemType(TypeOf($t)), ($z)->{ return $z.a }))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+
+select AGGREGATE_BY(a, $f) from @a;
+select AGGREGATE_BY(distinct a, $f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/top_by.sql b/yql/essentials/tests/sql/suites/aggr_factory/top_by.sql
new file mode 100644
index 0000000000..f63186dcda
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/top_by.sql
@@ -0,0 +1,21 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(
+    AsStruct(1 as key, 101 as value),
+    AsStruct(6 as key, 34 as value),
+    AsStruct(4 as key, 22 as value),
+    AsStruct(2 as key, 256 as value),
+    AsStruct(7 as key, 111 as value)
+);
+
+$f = AGGREGATION_FACTORY("topby", 3);
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"),
+    $f(ListItemType(TypeOf($t)), ($z)->{ return AsTuple($z.value, $z.key) }))));
+
+use plato;
+insert into @a select AsTuple(value, key) as vk from as_table($t);
+commit;
+
+select AGGREGATE_BY(vk, $f) from @a;
+
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/transform_input.sql b/yql/essentials/tests/sql/suites/aggr_factory/transform_input.sql
new file mode 100644
index 0000000000..0a92f6d097
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/transform_input.sql
@@ -0,0 +1,8 @@
+/* syntax version 1 */
+/* postgres can not */
+$f = AGGREGATION_FACTORY("sum");
+$g = AggregateTransformInput($f, ($x)->(cast($x as Int32)));
+$h = AggregateTransformInput($f, ($x)->($x * 2));
+select ListAggregate([1,2,3], $f);
+select ListAggregate(["1","2","3"], $g);
+select ListAggregate([1,2,3], $h);
\ No newline at end of file
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/transform_output.sql b/yql/essentials/tests/sql/suites/aggr_factory/transform_output.sql
new file mode 100644
index 0000000000..4598192464
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/transform_output.sql
@@ -0,0 +1,8 @@
+/* syntax version 1 */
+/* postgres can not */
+$f = AGGREGATION_FACTORY("sum");
+$g = AggregateTransformOutput($f, ($x)->(cast($x as String)));
+$h = AggregateTransformOutput($f, ($x)->($x * 2));
+select ListAggregate([1,2,3], $f);
+select ListAggregate([1,2,3], $g);
+select ListAggregate([1,2,3], $h);
\ No newline at end of file
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/udaf.sql b/yql/essentials/tests/sql/suites/aggr_factory/udaf.sql
new file mode 100644
index 0000000000..d419f8767f
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/udaf.sql
@@ -0,0 +1,26 @@
+/* syntax version 1 */
+/* postgres can not */
+-- count example
+$create = ($_item, $_parent) -> { return 1 };
+$add = ($state, $_item, $_parent) -> { return 1 + $state };
+$merge = ($state1, $state2) -> { return $state1 + $state2 };
+$get_result = ($state) -> { return $state };
+$serialize = ($state) -> { return $state };
+$deserialize = ($state) -> { return $state };
+$default = 0;
+
+$f = AGGREGATION_FACTORY("udaf", $create, $add, $merge, $get_result, $serialize, $deserialize, $default);
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+$t = AsList(AsStruct(1/0 as a),AsStruct(2/0 as a));
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/udaf_distinct_expr.sql b/yql/essentials/tests/sql/suites/aggr_factory/udaf_distinct_expr.sql
new file mode 100644
index 0000000000..15a4fe7c64
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/udaf_distinct_expr.sql
@@ -0,0 +1,19 @@
+/* syntax version 1 */
+/* postgres can not */
+-- count example
+$create = ($_item, $_parent) -> { return 1 };
+$add = ($state, $_item, $_parent) -> { return 1 + $state };
+$merge = ($state1, $state2) -> { return $state1 + $state2 };
+$get_result = ($state) -> { return $state };
+$serialize = ($state) -> { return $state };
+$deserialize = ($state) -> { return $state };
+$default = 0;
+
+$f = AGGREGATION_FACTORY("udaf", $create, $add, $merge, $get_result, $serialize, $deserialize, $default);
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a), AsStruct(1 as a));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+
+select AGGREGATE_BY(distinct cast(Unicode::ToLower(cast(a as Utf8) || "00"u) as Int), $f) from @a;
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/udaf_in_udaf.sql b/yql/essentials/tests/sql/suites/aggr_factory/udaf_in_udaf.sql
new file mode 100644
index 0000000000..49194de507
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/udaf_in_udaf.sql
@@ -0,0 +1,25 @@
+/* syntax version 1 */
+
+$merge_dicts = ($dict1, $dict2) -> { return SetUnion($dict1, $dict2, ($_key, $a, $b) -> { RETURN Coalesce($a, 0) + Coalesce($b, 0) }) };
+
+$create_single_item_dict = ($item, $_parent) -> { return AsDict(AsTuple($item, 1)) };
+$count_values = AGGREGATION_FACTORY(
+    "UDAF",
+    $create_single_item_dict,
+    ($dict, $item, $parent) -> { return $merge_dicts($create_single_item_dict($item, $parent), $dict) },
+    $merge_dicts
+);
+
+$create_dict_from_list = ($list, $_parent) -> { return ListAggregate($list, $count_values) };
+$add_list_to_dict = ($dict, $list, $parent) -> { return $merge_dicts($create_dict_from_list($list, $parent), $dict) };
+$count_list_values = AGGREGATION_FACTORY(
+    "UDAF",
+    $create_dict_from_list,
+    $add_list_to_dict,
+    $merge_dicts
+);
+
+$test_data = AsList(AsList(1,2),AsList(3,2),AsList(3,3),AsList(1,3),AsList(3,1),AsList(2,2));
+SELECT
+    ListSort(DictItems(ListAggregate(AsList(1,2,3,2,3,3), $count_values))) AS count_values,
+    ListSort(DictItems(ListAggregate($test_data, $count_list_values))) AS count_list_values,
diff --git a/yql/essentials/tests/sql/suites/aggr_factory/variance.sql b/yql/essentials/tests/sql/suites/aggr_factory/variance.sql
new file mode 100644
index 0000000000..2573bd6e66
--- /dev/null
+++ b/yql/essentials/tests/sql/suites/aggr_factory/variance.sql
@@ -0,0 +1,13 @@
+/* syntax version 1 */
+/* postgres can not */
+$t = AsList(AsStruct(1 as a),AsStruct(2 as a));
+$f = AGGREGATION_FACTORY("variance");
+
+select Yql::Aggregate($t, AsTuple(), AsTuple(AsTuple(AsAtom("res"), $f(
+    ListItemType(TypeOf($t)), ($z)->{return $z.a}))));
+
+use plato;
+insert into @a select * from as_table($t);
+commit;
+select AGGREGATE_BY(a,$f) from @a;
+select AGGREGATE_BY(distinct a,$f) from @a;