Merge branch 'rightlib' into mergelibs-241120-1113

127 files changed, 2410 insertions, 0 deletions

diff --git a/yql/essentials/tests/s-expressions/suites/Udf/AllFieldsAfterRemap.cfg b/yql/essentials/tests/s-expressions/suites/Udf/AllFieldsAfterRemap.cfg
new file mode 100644
index 0000000000..9bed8d5096
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/AllFieldsAfterRemap.cfg
@@ -0,0 +1,3 @@
+in Input input4.txt
+res result.txt
+udf dummylog
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/AllFieldsAfterRemap.yql b/yql/essentials/tests/s-expressions/suites/Udf/AllFieldsAfterRemap.yql
new file mode 100644
index 0000000000..77573dbc12
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/AllFieldsAfterRemap.yql
@@ -0,0 +1,11 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) (Void) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) table1 '('('type))))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/AutoMap.yql b/yql/essentials/tests/s-expressions/suites/Udf/AutoMap.yql
new file mode 100644
index 0000000000..cd809a5881
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/AutoMap.yql
@@ -0,0 +1,56 @@
+(
+(let world (block '(
+  (let x (Read! world (DataSource '"yt" '"plato") (Key '('table (String '"Input"))) '('"key") '()))
+  (let world (Left! x))
+  (let table0 (Right! x))
+  (let output (block '(
+    (let select (block '(
+      (let core table0)
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (Struct))
+        (let res (AddMember res '"column0" ("Apply" ("Udf" '"SimpleUdf.Increment") (Cast (Member row '"key") 'Uint32) (Int64 '"1"))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let x (Read! world (DataSource '"yt" '"plato") (Key '('table (String '"Input"))) '('"key") '()))
+  (let world (Left! x))
+  (let table0 (Right! x))
+  (let output (block '(
+    (let select (block '(
+      (let core table0)
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (Struct))
+        (let res (AddMember res '"column0" ("Apply" ("Udf" '"SimpleUdf.IncrementOpt") (Cast (Member row '"key") 'Uint32) (Int64 '"2"))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let plato_sink (DataSink '"yt" '"plato"))
+  (let world (Commit! world plato_sink))
+  (return world)
+)))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/AutoMapMany.cfg b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapMany.cfg
new file mode 100644
index 0000000000..0bbf3ad9f0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapMany.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf python2_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/AutoMapMany.sql b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapMany.sql
new file mode 100644
index 0000000000..74baf57176
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapMany.sql
@@ -0,0 +1,18 @@
+USE plato;
+
+$script = @@
+def f(x,y,z):
+  return x + y + z
+@@;
+
+$udf1=Python::f("(Int32{Flags:AutoMap},Int32{Flags:AutoMap},Int32{Flags:AutoMap})->Int32",$script);
+select $udf1(1,2,3);
+select $udf1(YQL::Just(1),2,3);
+select $udf1(YQL::Just(1),YQL::Just(2),3);
+select $udf1(YQL::Just(1),YQL::Just(2),YQL::Just(3));
+
+$udf2=Python::f("(Int32{Flags:AutoMap},Int32{Flags:AutoMap},Int32{Flags:AutoMap})->Int32?",$script);
+select $udf2(4,5,6);
+select $udf2(YQL::Just(4),5,6);
+select $udf2(YQL::Just(4),YQL::Just(5),6);
+select $udf2(YQL::Just(4),YQL::Just(5),YQL::Just(6));
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/AutoMapMany.yql b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapMany.yql
new file mode 100644
index 0000000000..e179419a5c
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapMany.yql
@@ -0,0 +1,206 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let scriptudf0 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf1 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf2 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf3 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf4 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf5 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf6 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf7 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(import aggregate_module '"/lib/yql/aggregate.yql")
+(import window_module '"/lib/yql/window.yql")
+(import core_module '"/lib/yql/core.yql")
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("Apply" (ScriptUdf 'Python '"f" (CallableType '() '((DataType 'Int32)) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1)) scriptudf0) (Int32 '"1") (Int32 '"2") (Int32 '"3")))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("Apply" (ScriptUdf 'Python '"f" (CallableType '() '((DataType 'Int32)) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1)) scriptudf1) ("Just" (Int32 '"1")) (Int32 '"2") (Int32 '"3")))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("Apply" (ScriptUdf 'Python '"f" (CallableType '() '((DataType 'Int32)) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1)) scriptudf2) ("Just" (Int32 '"1")) ("Just" (Int32 '"2")) (Int32 '"3")))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("Apply" (ScriptUdf 'Python '"f" (CallableType '() '((DataType 'Int32)) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1)) scriptudf3) ("Just" (Int32 '"1")) ("Just" (Int32 '"2")) ("Just" (Int32 '"3"))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("Apply" (ScriptUdf 'Python '"f" (CallableType '() '((OptionalType (DataType 'Int32))) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1)) scriptudf4) (Int32 '"4") (Int32 '"5") (Int32 '"6")))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("Apply" (ScriptUdf 'Python '"f" (CallableType '() '((OptionalType (DataType 'Int32))) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1)) scriptudf5) ("Just" (Int32 '"4")) (Int32 '"5") (Int32 '"6")))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("Apply" (ScriptUdf 'Python '"f" (CallableType '() '((OptionalType (DataType 'Int32))) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1)) scriptudf6) ("Just" (Int32 '"4")) ("Just" (Int32 '"5")) (Int32 '"6")))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("Apply" (ScriptUdf 'Python '"f" (CallableType '() '((OptionalType (DataType 'Int32))) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1) '((DataType 'Int32) '"" '1)) scriptudf7) ("Just" (Int32 '"4")) ("Just" (Int32 '"5")) ("Just" (Int32 '"6"))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let plato_sink (DataSink '"yt" '"plato"))
+  (let world (Commit! world plato_sink))
+  (return world)
+)))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/AutoMapManyNamed.cfg b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapManyNamed.cfg
new file mode 100644
index 0000000000..0bbf3ad9f0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapManyNamed.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf python2_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/AutoMapManyNamed.sql b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapManyNamed.sql
new file mode 100644
index 0000000000..fff1174f57
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapManyNamed.sql
@@ -0,0 +1,22 @@
+USE plato;
+
+$script = @@
+def f(x,y,z):
+  return x + y + z
+@@;
+
+$udf1=Python::f("(x:Int32{Flags:AutoMap},y:Int32{Flags:AutoMap},z:Int32{Flags:AutoMap})->Int32",$script);
+select $udf1(1 as x,2 as y,3 as z);
+select $udf1(YQL::Just(1) as x,2 as y,3 as z);
+select $udf1(YQL::Just(1) as x,YQL::Just(2) as y,3 as z);
+select $udf1(YQL::Just(1) as x,YQL::Just(2) as y,YQL::Just(3) as z);
+select $udf1(YQL::Just(1),YQL::Just(2) as y,YQL::Just(3) as z);
+select $udf1(YQL::Just(1),YQL::Just(2),YQL::Just(3) as z);
+
+$udf2=Python::f("(x:Int32{Flags:AutoMap},y:Int32{Flags:AutoMap},z:Int32{Flags:AutoMap})->Int32?",$script);
+select $udf2(4 as x,5 as y,6 as z);
+select $udf2(YQL::Just(4) as x,5 as y,6 as z);
+select $udf2(YQL::Just(4) as x,YQL::Just(5) as y,6 as z);
+select $udf2(YQL::Just(4) as x,YQL::Just(5) as y,YQL::Just(6) as z);
+select $udf2(YQL::Just(4),YQL::Just(5) as y,YQL::Just(6) as z);
+select $udf2(YQL::Just(4),YQL::Just(5),YQL::Just(6) as z);
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/AutoMapManyNamed.yql b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapManyNamed.yql
new file mode 100644
index 0000000000..2225ad9b7e
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/AutoMapManyNamed.yql
@@ -0,0 +1,302 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let scriptudf0 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf1 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf10 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf11 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf2 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf3 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf4 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf5 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf6 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf7 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf8 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(let scriptudf9 (String '@@
+def f(x,y,z):
+  return x + y + z
+@@))
+(import aggregate_module '"/lib/yql/aggregate.yql")
+(import window_module '"/lib/yql/window.yql")
+(import core_module '"/lib/yql/core.yql")
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((DataType 'Int32)) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf0) '() (AsStruct '('x (Int32 '"1")) '('y (Int32 '"2")) '('z (Int32 '"3")))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((DataType 'Int32)) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf1) '() (AsStruct '('x ("Just" (Int32 '"1"))) '('y (Int32 '"2")) '('z (Int32 '"3")))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((DataType 'Int32)) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf2) '() (AsStruct '('x ("Just" (Int32 '"1"))) '('y ("Just" (Int32 '"2"))) '('z (Int32 '"3")))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((DataType 'Int32)) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf3) '() (AsStruct '('x ("Just" (Int32 '"1"))) '('y ("Just" (Int32 '"2"))) '('z ("Just" (Int32 '"3"))))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((DataType 'Int32)) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf4) '(("Just" (Int32 '"1"))) (AsStruct '('y ("Just" (Int32 '"2"))) '('z ("Just" (Int32 '"3"))))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((DataType 'Int32)) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf5) '(("Just" (Int32 '"1")) ("Just" (Int32 '"2"))) (AsStruct '('z ("Just" (Int32 '"3"))))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((OptionalType (DataType 'Int32))) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf6) '() (AsStruct '('x (Int32 '"4")) '('y (Int32 '"5")) '('z (Int32 '"6")))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((OptionalType (DataType 'Int32))) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf7) '() (AsStruct '('x ("Just" (Int32 '"4"))) '('y (Int32 '"5")) '('z (Int32 '"6")))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((OptionalType (DataType 'Int32))) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf8) '() (AsStruct '('x ("Just" (Int32 '"4"))) '('y ("Just" (Int32 '"5"))) '('z (Int32 '"6")))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((OptionalType (DataType 'Int32))) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf9) '() (AsStruct '('x ("Just" (Int32 '"4"))) '('y ("Just" (Int32 '"5"))) '('z ("Just" (Int32 '"6"))))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((OptionalType (DataType 'Int32))) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf10) '(("Just" (Int32 '"4"))) (AsStruct '('y ("Just" (Int32 '"5"))) '('z ("Just" (Int32 '"6"))))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let output (block '(
+    (let select (block '(
+      (let core (AsList (Uint32 '0)))
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (AsStruct '('"column0" ("NamedApply" (ScriptUdf 'Python '"f" (CallableType '() '((OptionalType (DataType 'Int32))) '((DataType 'Int32) 'x '1) '((DataType 'Int32) 'y '1) '((DataType 'Int32) 'z '1)) scriptudf11) '(("Just" (Int32 '"4")) ("Just" (Int32 '"5"))) (AsStruct '('z ("Just" (Int32 '"6"))))))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let plato_sink (DataSink '"yt" '"plato"))
+  (let world (Commit! world plato_sink))
+  (return world)
+)))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/CachedUdfType.yql b/yql/essentials/tests/s-expressions/suites/Udf/CachedUdfType.yql
new file mode 100644
index 0000000000..178b280ad6
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/CachedUdfType.yql
@@ -0,0 +1,14 @@
+(
+#comment
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let res_sink (DataSink 'result))
+(let cachedType (CallableType '() '((DataType 'String)) '((OptionalType (DataType 'String)))))
+(let udf (Udf 'SimpleUdf.Echo (Void) (Void) '"" cachedType))
+(let x (String 'aaa))
+(let data (Apply udf x))
+(let world (Write! world res_sink (Key) data '()))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/CallOtherCallable.cfg b/yql/essentials/tests/s-expressions/suites/Udf/CallOtherCallable.cfg
new file mode 100644
index 0000000000..49968d37d7
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/CallOtherCallable.cfg
@@ -0,0 +1,2 @@
+udf callables_udf
+res result.txt
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/CallOtherCallable.yql b/yql/essentials/tests/s-expressions/suites/Udf/CallOtherCallable.yql
new file mode 100644
index 0000000000..5d5a0989e0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/CallOtherCallable.yql
@@ -0,0 +1,14 @@
+(
+#comment
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let res_sink (DataSink 'result))
+(let fromString (Udf 'Callables.FromString))
+(let mul (Udf 'Callables.Mul))
+(let list (AsList (String '1) (String '2) (String '3) (String '4) (String '5)))
+(let data (Apply mul list fromString))
+(let world (Write! world res_sink (Key) data '()))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/CallableAsLambda.cfg b/yql/essentials/tests/s-expressions/suites/Udf/CallableAsLambda.cfg
new file mode 100644
index 0000000000..57571ef968
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/CallableAsLambda.cfg
@@ -0,0 +1,3 @@
+in Input input.txt
+out Output output.txt
+udf simple_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/CallableAsLambda.yql b/yql/essentials/tests/s-expressions/suites/Udf/CallableAsLambda.yql
new file mode 100644
index 0000000000..aebf755663
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/CallableAsLambda.yql
@@ -0,0 +1,12 @@
+(
+#comment
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let res_sink (DataSink 'result))
+(let udf (Udf 'SimpleUdf.Echo))
+(let data (Map (AsList (String 'a) (String 'b)) udf))
+(let world (Write! world res_sink (Key) data '()))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/CallableType.yql b/yql/essentials/tests/s-expressions/suites/Udf/CallableType.yql
new file mode 100644
index 0000000000..817a744385
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/CallableType.yql
@@ -0,0 +1,18 @@
+(
+#comment
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let res_sink (DataSink 'result))
+(let t1 (DataType 'Uint32))
+(let o1 (OptionalType t1))
+(let str (DataType 'String))
+(let s1 (StructType '('key str) '('subkey str) '('value str)))
+(let c1 (CallableType '() '(t1) '(o1) '(s1)))
+(let world (Write! world res_sink (Key) (FormatType c1) '()))
+(let world (Write! world res_sink (Key) (FormatType (CallableResultType c1)) '()))
+(let world (Write! world res_sink (Key) (FormatType (CallableArgumentType c1 '0)) '()))
+(let world (Write! world res_sink (Key) (FormatType (CallableArgumentType c1 '1)) '()))
+(let world (Commit! world res_sink))
+(return world)
+)
\ No newline at end of file
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/CustomPython.cfg b/yql/essentials/tests/s-expressions/suites/Udf/CustomPython.cfg
new file mode 100644
index 0000000000..f493bfe9eb
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/CustomPython.cfg
@@ -0,0 +1,5 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+file custom_python.so yql/udfs/examples/custom_python/libcustom_python_udf.so
+os linux
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/CustomPython.yql b/yql/essentials/tests/s-expressions/suites/Udf/CustomPython.yql
new file mode 100644
index 0000000000..a683757f68
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/CustomPython.yql
@@ -0,0 +1,24 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+# prepare python udf
+(let world (Configure! world (DataSource 'config) 'ImportUdfs 'custom_python.so))
+
+(let json (DataType 'Json))
+(let udfType (CallableType '() '(json) '(json)))
+(let udfScript (String '"from custom_python.foo import bar"))
+(let udf (ScriptUdf 'CustomPython 'bar udfType udfScript))
+
+# call udf
+(let x (Json '@@{"abc":1}@@))
+(let result (Apply udf x))
+
+# output result with type
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) result '( '('type) )))
+
+# finish
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/ImportUdfs.cfg b/yql/essentials/tests/s-expressions/suites/Udf/ImportUdfs.cfg
new file mode 100644
index 0000000000..240cfb2bf5
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/ImportUdfs.cfg
@@ -0,0 +1,3 @@
+file myfile yql/essentials/udfs/test/test_import/libtest_import_udf.so
+res result.txt
+os linux
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/ImportUdfs.yql b/yql/essentials/tests/s-expressions/suites/Udf/ImportUdfs.yql
new file mode 100644
index 0000000000..e9da004a74
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/ImportUdfs.yql
@@ -0,0 +1,23 @@
+# not supported on windows
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+# (1) import udfs from 'myfile'
+(let config (DataSource 'config))
+(let world (Configure! world config 'ImportUdfs 'myfile))
+# (2) execute 'TestImportUdf.Concat'
+(let name (String 'Jamel))
+(let hello (Udf 'TestImportUdf.Concat))
+(let foo_data (Apply hello (String '"Hello, ") name))
+
+# (3) execute 'TestImportUdf.Repeat'
+(let bar_data (Apply (Udf 'TestImportUdf.Repeat) (String 'x) (Uint64 '3)))
+
+# (4) output results
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) foo_data '('('type))))
+(let world (Write! world res_sink (Key) bar_data '('('type))))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/LambdaAsCallable.cfg b/yql/essentials/tests/s-expressions/suites/Udf/LambdaAsCallable.cfg
new file mode 100644
index 0000000000..3d92a5a733
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/LambdaAsCallable.cfg
@@ -0,0 +1,2 @@
+udf type_inspection_udf
+res result.txt
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/LambdaAsCallable.yql b/yql/essentials/tests/s-expressions/suites/Udf/LambdaAsCallable.yql
new file mode 100644
index 0000000000..082da79b04
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/LambdaAsCallable.yql
@@ -0,0 +1,25 @@
+(
+#comment
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let res_sink (DataSink 'result))
+(let list (AsList (String 'A) (String 'B) (String 'C) (String 'D) (String 'E)))
+(let state (String '""))
+(let fold (Udf 'TypeInspection.Fold (Void)
+    (TupleType
+        (TupleType
+            (TypeOf list)
+            (TypeOf state)
+            (CallableType '() '((DataType 'String)) '((DataType 'String)) '((DataType 'String)))
+        )
+        (StructType)
+        (TupleType)
+    )
+))
+(let func (lambda '(item state) (Concat state item)))
+(let data (Apply fold list state func))
+(let world (Write! world res_sink (Key) data '()))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/NamedApply.yql b/yql/essentials/tests/s-expressions/suites/Udf/NamedApply.yql
new file mode 100644
index 0000000000..adfa44066a
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/NamedApply.yql
@@ -0,0 +1,66 @@
+(
+#comment
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let res_sink (DataSink 'result))
+(let f1 (lambda '(x y) (block '(
+    (return (+ x (Coalesce y (Uint32 '100))))
+))))
+
+(let f1dep (lambda '(x dep) (block '(
+    (return (+ x x))
+))))
+
+(let tuple (lambda '() '((Uint32 '1) (Uint32 '2))))
+(let tuple_ctype (CallableType '() '((TupleType (DataType 'Uint32) (DataType 'Uint32)))))
+(let tuple_call (Callable tuple_ctype tuple))
+
+(let ctype (CallableType '() '((DataType 'Uint32)) '((DataType 'Uint32) 'Arg1) '((DataType 'Uint32) 'Arg2)))
+(let call (Callable ctype f1))
+
+(let optTuple1 (lambda '() '((Uint32 '1) (Just (Uint32 '2)))))
+(let optTuple2 (lambda '() '((Uint32 '1) (Nothing (OptionalType (DataType 'Uint32))))))
+(let optCtype (CallableType '('1) 
+  '((DataType 'Uint32)) 
+  '((DataType 'Uint32) 'Arg1) 
+  '((OptionalType (DataType 'Uint32)) 'Arg2)))
+(let optCall (Callable optCtype f1))
+
+(let s1 (AsStruct '('Arg2 (Uint32 '2)) '('Arg1 (Uint32 '1))))
+(let strType (CallableType '() '((TypeOf s1))))
+(let strCall (Callable strType (lambda '() s1)))
+
+# lambdas
+(let world (Write! world res_sink (Key) (NamedApply f1 '((Uint32 '1) (Uint32 '2)) (AsStruct)) '()))
+(let world (Write! world res_sink (Key) (NamedApply f1dep '((Uint32 '1)) (AsStruct) (DependsOn (Uint32 '2))) '()))
+
+# callables
+(let world (Write! world res_sink (Key) (NamedApply call '((Uint32 '1) (Uint32 '2)) (AsStruct)) '()))
+(let world (Write! world res_sink (Key) (NamedApply call (Apply tuple_call) (AsStruct)) '()))
+
+# opt. positional
+(let world (Write! world res_sink (Key) (NamedApply optCall '((Uint32 '1) (Uint32 '2)) (AsStruct)) '()))
+(let world (Write! world res_sink (Key) (NamedApply optCall (Apply optTuple1) (AsStruct)) '()))
+(let world (Write! world res_sink (Key) (NamedApply optCall (Apply optTuple2) (AsStruct)) '()))
+(let world (Write! world res_sink (Key) (NamedApply optCall '((Uint32 '1)) (AsStruct)) '()))
+
+# named
+(let world (Write! world res_sink (Key) (NamedApply call '((Uint32 '1)) (AsStruct 
+'('Arg2 (Uint32 '2))
+)) '()))
+
+(let world (Write! world res_sink (Key) (NamedApply call '() (AsStruct 
+'('Arg2 (Uint32 '2)) '('Arg1 (Uint32 '1))
+)) '()))
+
+(let world (Write! world res_sink (Key) (NamedApply call '() (Apply strCall)) '()))
+
+#opt. named
+(let world (Write! world res_sink (Key) (NamedApply optCall '() (AsStruct 
+'('Arg1 (Uint32 '1))
+)) '()))
+
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/NamedArgUdf.yql b/yql/essentials/tests/s-expressions/suites/Udf/NamedArgUdf.yql
new file mode 100644
index 0000000000..b32401ac2c
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/NamedArgUdf.yql
@@ -0,0 +1,44 @@
+(
+#comment
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let udf (Udf 'SimpleUdf.NamedArgs ))
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) (FormatType (TypeOf udf)) '('('type))))
+(let world (Write! world res_sink (Key) (NamedApply udf 
+  '((Uint32 '1)) 
+  (AsStruct)) 
+'()))
+(let world (Write! world res_sink (Key) (NamedApply udf 
+  '((Uint32 '1) (Uint32 '2)) 
+  (AsStruct)) 
+'()))
+(let world (Write! world res_sink (Key) (NamedApply udf 
+  '((Uint32 '1)) 
+  (AsStruct '('C (Uint32 '3)))) 
+'()))
+(let world (Write! world res_sink (Key) (NamedApply udf 
+  '((Uint32 '1)) 
+  (AsStruct '('C (Uint32 '3)) '('D (Uint32 '4)))) 
+'()))
+(let world (Write! world res_sink (Key) (NamedApply udf 
+  '((Uint32 '1)) 
+  (AsStruct '('D (Uint32 '4)))) 
+'()))
+(let world (Write! world res_sink (Key) (NamedApply udf 
+  '((Uint32 '1) (Uint32 '2)) 
+  (AsStruct '('D (Uint32 '4)))) 
+'()))
+(let world (Write! world res_sink (Key) (NamedApply udf 
+  '((Uint32 '1) (Uint32 '2) (Uint32 '3)) 
+  (AsStruct)) 
+'()))
+(let world (Write! world res_sink (Key) (NamedApply udf 
+  '((Uint32 '1) (Uint32 '2) (Uint32 '3) (Uint32 '4)) 
+  (AsStruct))  
+'()))
+
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/NamedAutoMapName.yql b/yql/essentials/tests/s-expressions/suites/Udf/NamedAutoMapName.yql
new file mode 100644
index 0000000000..719a56a72d
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/NamedAutoMapName.yql
@@ -0,0 +1,60 @@
+(
+(let world (block '(
+  (let x (Read! world (DataSource '"yt" '"plato") (Key '('table (String '"Input"))) '('"key") '()))
+  (let world (Left! x))
+  (let table0 (Right! x))
+  (let output (block '(
+    (let select (block '(
+      (let core table0)
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (Struct))
+        (let res (AddMember res '"column0" ("NamedApply" ("Udf" '"SimpleUdf.Increment") '() (AsStruct
+          '('Arg1 (Cast (Member row '"key") 'Uint32)) '('Arg2 (Uint32 '"1"))
+        ))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let x (Read! world (DataSource '"yt" '"plato") (Key '('table (String '"Input"))) '('"key") '()))
+  (let world (Left! x))
+  (let table0 (Right! x))
+  (let output (block '(
+    (let select (block '(
+      (let core table0)
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (Struct))
+        (let res (AddMember res '"column0" ("NamedApply" ("Udf" '"SimpleUdf.IncrementOpt") '() (AsStruct
+          '('Arg1 (Cast (Member row '"key") 'Uint32)) '('Arg2 (Uint32 '"2"))
+        ))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let plato_sink (DataSink '"yt" '"plato"))
+  (let world (Commit! world plato_sink))
+  (return world)
+)))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/NamedAutoMapPos.yql b/yql/essentials/tests/s-expressions/suites/Udf/NamedAutoMapPos.yql
new file mode 100644
index 0000000000..37c8674711
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/NamedAutoMapPos.yql
@@ -0,0 +1,56 @@
+(
+(let world (block '(
+  (let x (Read! world (DataSource '"yt" '"plato") (Key '('table (String '"Input"))) '('"key") '()))
+  (let world (Left! x))
+  (let table0 (Right! x))
+  (let output (block '(
+    (let select (block '(
+      (let core table0)
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (Struct))
+        (let res (AddMember res '"column0" ("NamedApply" ("Udf" '"SimpleUdf.Increment") '((Cast (Member row '"key") 'Uint32) (Uint32 '"1")) (AsStruct))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let x (Read! world (DataSource '"yt" '"plato") (Key '('table (String '"Input"))) '('"key") '()))
+  (let world (Left! x))
+  (let table0 (Right! x))
+  (let output (block '(
+    (let select (block '(
+      (let core table0)
+      (let core (FlatMap core (lambda '(row) (block '(
+        (let res (Struct))
+        (let res (AddMember res '"column0" ("NamedApply" ("Udf" '"SimpleUdf.IncrementOpt") '((Cast (Member row '"key") 'Uint32) (Uint32 '"2")) (AsStruct))))
+        (let res (AsList res))
+        (return res)
+      )))))
+      (return core)
+    )))
+    (return select)
+  )))
+  (let world (block '(
+    (let result_sink (DataSink 'result))
+    (let world (Write! world result_sink (Key) output '('('type) '('autoref) '('columns '('"column0")))))
+    (return (Commit! world result_sink))
+  )))
+  (return world)
+)))
+(let world (block '(
+  (let plato_sink (DataSink '"yt" '"plato"))
+  (let world (Commit! world plato_sink))
+  (return world)
+)))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/OptionalReduceWithResource.cfg b/yql/essentials/tests/s-expressions/suites/Udf/OptionalReduceWithResource.cfg
new file mode 100644
index 0000000000..956cb08312
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/OptionalReduceWithResource.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+udf stat_udf
+res result.txt
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/OptionalReduceWithResource.yql b/yql/essentials/tests/s-expressions/suites/Udf/OptionalReduceWithResource.yql
new file mode 100644
index 0000000000..affd7f4d55
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/OptionalReduceWithResource.yql
@@ -0,0 +1,45 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let udfTDigest_Create (Udf 'Stat.TDigest_Create))
+(let udfTDigest_AddValue (Udf 'Stat.TDigest_AddValue))
+(let udfTDigest_GetPercentile (Udf 'Stat.TDigest_GetPercentile))
+(let udfTDigest_Serialize (Udf 'Stat.TDigest_Serialize))
+(let udfTDigest_Deserialize (Udf 'Stat.TDigest_Deserialize))
+(let udfTDigest_Merge (Udf 'Stat.TDigest_Merge))
+(let table1map (Map table1 (lambda '(item) (block '(
+   (let value (Member item 'value))
+   (let x (Apply udfTDigest_Create (Double '1.0)))
+   (let y (Apply udfTDigest_Create (Double '2.0)))
+   (let jx (Just x))
+   (let jy (Just y))
+   (let n (Nothing (OptionalType (TypeOf x))))
+   (let handler (lambda '(x y) (Apply udfTDigest_Merge x y)))
+   (let newValue value)
+   (let newValue (Concat newValue (String '" => ")))
+   (let newValue (Concat newValue (Concat (String '" ") (Coalesce (Map
+       (OptionalReduce jx jy handler) (lambda '(r) 
+       (ToString (Apply udfTDigest_GetPercentile r (Double '0.75))
+       ))) (String '?)))))
+   (let newValue (Concat newValue (Concat (String '" ") (Coalesce (Map
+       (OptionalReduce jx n handler) (lambda '(r) 
+       (ToString (Apply udfTDigest_GetPercentile r (Double '0.75))
+       ))) (String '?)))))
+   (let newValue (Concat newValue (Concat (String '" ") (Coalesce (Map
+       (OptionalReduce n n handler) (lambda '(r) 
+       (ToString (Apply udfTDigest_GetPercentile r (Double '0.75))
+       ))) (String '?)))))
+   (let s (Struct))
+   (let s (AddMember s 'key (Member item 'key)))
+   (let s (AddMember s 'subkey (Member item 'subkey)))
+   (let s (AddMember s 'value newValue))
+   (return s)
+)))))
+(let mr_sink (DataSink 'yt (quote plato)))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) table1map '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/OptionalRunConfig.cfg b/yql/essentials/tests/s-expressions/suites/Udf/OptionalRunConfig.cfg
new file mode 100644
index 0000000000..57571ef968
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/OptionalRunConfig.cfg
@@ -0,0 +1,3 @@
+in Input input.txt
+out Output output.txt
+udf simple_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/OptionalRunConfig.yql b/yql/essentials/tests/s-expressions/suites/Udf/OptionalRunConfig.yql
new file mode 100644
index 0000000000..8703e76e49
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/OptionalRunConfig.yql
@@ -0,0 +1,16 @@
+(
+#comment
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let res_sink (DataSink 'result))
+(let x (String 'bbb))
+(let udf (Udf 'SimpleUdf.EchoWithRunPrefix (String 'aaa) (Void) '""))
+(let data (Apply udf x))
+(let world (Write! world res_sink (Key) data '()))
+(let udf (Udf 'SimpleUdf.EchoWithRunPrefix))
+(let data (Apply udf x))
+(let world (Write! world res_sink (Key) data '()))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PartialCall.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PartialCall.cfg
new file mode 100644
index 0000000000..57571ef968
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PartialCall.cfg
@@ -0,0 +1,3 @@
+in Input input.txt
+out Output output.txt
+udf simple_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PartialCall.yql b/yql/essentials/tests/s-expressions/suites/Udf/PartialCall.yql
new file mode 100644
index 0000000000..ce81c3df0d
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PartialCall.yql
@@ -0,0 +1,16 @@
+(
+#comment
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let res_sink (DataSink 'result))
+(let udf (Udf 'SimpleUdf.EchoWithPrefix))
+(let x (String 'aaa))
+(let y (String 'bbb))
+(let data (Apply udf x))
+(let world (Write! world res_sink (Key) data '()))
+(let data (Apply udf x y))
+(let world (Write! world res_sink (Key) data '()))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PartialFieldsAfterRemap.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PartialFieldsAfterRemap.cfg
new file mode 100644
index 0000000000..9bed8d5096
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PartialFieldsAfterRemap.cfg
@@ -0,0 +1,3 @@
+in Input input4.txt
+res result.txt
+udf dummylog
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PartialFieldsAfterRemap.yql b/yql/essentials/tests/s-expressions/suites/Udf/PartialFieldsAfterRemap.yql
new file mode 100644
index 0000000000..4e5766f779
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PartialFieldsAfterRemap.yql
@@ -0,0 +1,11 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) table1 '('('type))))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.cfg b/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.cfg
new file mode 100644
index 0000000000..4de5e2339f
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.cfg
@@ -0,0 +1,4 @@
+in Input Percentiles.txt
+out Output output.txt
+udf stat_udf
+res result.txt
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.txt b/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.txt
new file mode 100644
index 0000000000..b500ea3b02
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.txt
@@ -0,0 +1,16 @@
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="b";"subkey"=".";"value"="10"};
+{"key"="b";"subkey"=".";"value"="2"};
+{"key"="a";"subkey"=".";"value"="3"};
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="b";"subkey"=".";"value"="1"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.txt.attr
new file mode 100644
index 0000000000..b6100e5fd0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.txt.attr
@@ -0,0 +1,30 @@
+{
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
\ No newline at end of file
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.yql b/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.yql
new file mode 100644
index 0000000000..0a15c6095a
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/Percentiles.yql
@@ -0,0 +1,45 @@
+(
+#comment
+(let udfTDigest_Create (Udf 'Stat.TDigest_Create))
+(let udfTDigest_AddValue (Udf 'Stat.TDigest_AddValue))
+(let udfTDigest_GetPercentile (Udf 'Stat.TDigest_GetPercentile))
+(let udfTDigest_Serialize (Udf 'Stat.TDigest_Serialize))
+(let udfTDigest_Deserialize (Udf 'Stat.TDigest_Deserialize))
+(let udfTDigest_Merge (Udf 'Stat.TDigest_Merge))
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let preMap (lambda '(item) (Just item)))
+(let keyExtractor (lambda '(item) (Member item 'key)))
+(let init (lambda '(key item) (Apply udfTDigest_Create (Unwrap (FromString (Member item 'value) 'Double)))))
+(let update (lambda '(key item state) (Apply udfTDigest_AddValue state (Unwrap (FromString (Member item 'value) 'Double)))))
+(let finish (lambda '(key state) (block '(
+  (let s (Struct))
+  (let s (AddMember s 'key key))
+  (let s (AddMember s 'subkey (String '.)))
+  (let s (AddMember s 'value (Apply udfTDigest_Serialize state)))
+  (return (Just s))
+))))
+(let table2 (CombineByKey table1 preMap keyExtractor init update finish))
+(let listHandler (lambda '(stream) (FlatMap stream (lambda '(pair) (block '(
+  (let key (Nth pair '0))
+  (let list (ForwardList (Nth pair '1)))
+  (let init (lambda '(item) (Apply udfTDigest_Deserialize (Member item 'value))))
+  (let update (lambda '(item state) (Apply udfTDigest_Merge (Apply udfTDigest_Deserialize (Member item 'value)) state)))
+  (let state (Unwrap (Fold1 list init update)))
+  (let s (Struct))
+  (let s (AddMember s 'key key))
+  (let s (AddMember s 'subkey (String '.)))
+  (let g1 (ToString (Apply udfTDigest_GetPercentile state (Double '0.75))))
+  (let g2 (ToString (Apply udfTDigest_GetPercentile state (Double '0.999))))
+  (let s (AddMember s 'value (Concat (Concat g1 (String '" ")) g2)))
+  (let ret (Just s))
+  (return ret)
+))))))
+(let table3 (PartitionByKey table2 keyExtractor (Void) (Void) listHandler))
+(let mr_sink (DataSink 'yt 'plato))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) table3 '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PireMatchInMap.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PireMatchInMap.cfg
new file mode 100644
index 0000000000..8bd0a53757
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PireMatchInMap.cfg
@@ -0,0 +1,3 @@
+in Input input.txt
+out Output output.txt
+udf pire_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PireMatchInMap.yql b/yql/essentials/tests/s-expressions/suites/Udf/PireMatchInMap.yql
new file mode 100644
index 0000000000..54c06596ad
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PireMatchInMap.yql
@@ -0,0 +1,16 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let udf (Udf 'Pire.Match (String '\d+)))
+(let table1map (Filter table1 (lambda '(item) (block '(
+   (let value (Member item 'value))
+   (return (Apply udf value))
+)))))
+(let mr_sink (DataSink 'yt (quote plato)))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) table1map '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PireMatchRes.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PireMatchRes.cfg
new file mode 100644
index 0000000000..d01e7622a0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PireMatchRes.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf pire_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PireMatchRes.yql b/yql/essentials/tests/s-expressions/suites/Udf/PireMatchRes.yql
new file mode 100644
index 0000000000..90391c2264
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PireMatchRes.yql
@@ -0,0 +1,13 @@
+(
+#comment
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let res_sink (DataSink 'result))
+(let udf (Udf 'Pire.Match (String '.?\d.?)))
+(let world (Write! world res_sink (Key) (Apply udf (String 'abc)) '()))
+(let world (Write! world res_sink (Key) (Apply udf (String 'x6y)) '()))
+(let world (Write! world res_sink (Key) (FormatType (TypeOf udf)) '()))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/ProtoField.cfg b/yql/essentials/tests/s-expressions/suites/Udf/ProtoField.cfg
new file mode 100644
index 0000000000..8fa549ddb5
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/ProtoField.cfg
@@ -0,0 +1,3 @@
+in Input input_proto.txt
+udf protobuf_udf
+res result.txt
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/ProtoField.yql b/yql/essentials/tests/s-expressions/suites/Udf/ProtoField.yql
new file mode 100644
index 0000000000..c3b00f1716
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/ProtoField.yql
@@ -0,0 +1,11 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) table1 '('('type))))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldOpt.cfg b/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldOpt.cfg
new file mode 100644
index 0000000000..8dd001f619
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldOpt.cfg
@@ -0,0 +1,3 @@
+in Input input_proto_opt.txt
+udf protobuf_udf
+res result.txt
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldOpt.yql b/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldOpt.yql
new file mode 100644
index 0000000000..c3b00f1716
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldOpt.yql
@@ -0,0 +1,11 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) table1 '('('type))))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldScheme.cfg b/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldScheme.cfg
new file mode 100644
index 0000000000..8fa549ddb5
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldScheme.cfg
@@ -0,0 +1,3 @@
+in Input input_proto.txt
+udf protobuf_udf
+res result.txt
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldScheme.yql b/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldScheme.yql
new file mode 100644
index 0000000000..89ab89671f
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/ProtoFieldScheme.yql
@@ -0,0 +1,11 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('tablescheme (String 'Input))) (Void) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) table1 '('('type))))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonAvg.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PythonAvg.cfg
new file mode 100644
index 0000000000..0bbf3ad9f0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonAvg.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf python2_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonAvg.yql b/yql/essentials/tests/s-expressions/suites/Udf/PythonAvg.yql
new file mode 100644
index 0000000000..62dc07db05
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonAvg.yql
@@ -0,0 +1,57 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+# prepare python udf
+(let ui32 (DataType 'Uint32))
+(let dbl (DataType 'Double))
+(let rt (ResourceType 'Python2))
+
+(let udfScript (String '@@
+class AvgCalc:
+    def __init__(self):
+        self.__count = 0
+        self.__sum = 0
+
+    def add(self, value):
+        self.__sum = self.__sum + value
+        self.__count = self.__count + 1
+
+    def result(self):
+        return self.__sum / float(self.__count)
+
+def avg_create():
+    return AvgCalc()
+
+def avg_add(avg, value):
+    avg.add(value)
+    return avg
+
+def avg_result(avg):
+    return avg.result()
+
+@@))
+
+(let avgCreate (ScriptUdf 'Python 'avg_create (CallableType '() '(rt)) udfScript))
+(let avgAdd (ScriptUdf 'Python 'avg_add (CallableType '() '(rt) '(rt) '(ui32)) udfScript))
+(let avgResult (ScriptUdf 'Python 'avg_result (CallableType '() '(dbl) '(rt)) udfScript))
+
+# data
+(let x (Uint32 '10))
+(let y (Uint32 '30))
+(let z (Uint32 '5))
+
+# call udfs
+(let avg (Apply avgCreate))
+(let avg (Apply avgAdd avg x))
+(let avg (Apply avgAdd avg y))
+(let avg (Apply avgAdd avg z))
+(let result (Apply avgResult avg))
+
+# output result with type
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) result '( '('type) )))
+
+# finish
+(return (Commit! world res_sink))
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsArg.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsArg.cfg
new file mode 100644
index 0000000000..0bffbd4f08
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsArg.cfg
@@ -0,0 +1,5 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf python2_udf
+udf simple_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsArg.yql b/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsArg.yql
new file mode 100644
index 0000000000..36a70677e4
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsArg.yql
@@ -0,0 +1,31 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+# common types
+(let ui64 (DataType 'Uint64))
+(let str (DataType 'String))
+
+# callable func
+(let funcType (CallableType '() '(str) '(str) '(ui64)))
+(let func (Udf 'SimpleUdf.Repeat))
+
+# python udf
+(let udfType (CallableType '() '(str) '(funcType) '(ui64)))
+(let udfScript (String '@@
+def new_string(func, x):
+    return func('x', x) + ':' + func('y', x)
+@@))
+(let udf (ScriptUdf 'Python 'new_string udfType udfScript))
+
+# call udf
+(let x (Uint64 '3))
+(let result (Apply udf func x))
+
+# output result with type
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) result '( '('type) )))
+
+# finish
+(return (Commit! world res_sink))
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsResult.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsResult.cfg
new file mode 100644
index 0000000000..0bbf3ad9f0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsResult.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf python2_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsResult.yql b/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsResult.yql
new file mode 100644
index 0000000000..7a7a6a995a
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonCallableAsResult.yql
@@ -0,0 +1,39 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+# prepare python udf
+(let ui64 (DataType 'Uint64))
+(let str (DataType 'String))
+(let funcType (CallableType '() '(ui64) '(ui64)))
+
+(let udfType (CallableType '() '(funcType) '(ui64)))
+(let udfScript (String '@@
+def create_counter(start):
+    def counter(step):
+        v = counter.x
+        counter.x += step
+        return v
+
+    counter.x = start
+    return counter
+@@))
+
+(let udf (ScriptUdf 'Python 'create_counter udfType udfScript))
+
+# call udf
+(let counter (Apply udf (Uint64 '1)))
+(let result (AsList
+    (Apply counter (Uint64 '1))
+    (Apply counter (Uint64 '2))
+    (Apply counter (Uint64 '3))
+    (Apply counter (Uint64 '4))
+))
+
+# output result with type
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) result '( '('type) )))
+
+# finish
+(return (Commit! world res_sink))
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonGenerator.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PythonGenerator.cfg
new file mode 100644
index 0000000000..0bbf3ad9f0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonGenerator.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf python2_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonGenerator.yql b/yql/essentials/tests/s-expressions/suites/Udf/PythonGenerator.yql
new file mode 100644
index 0000000000..d225a14c1b
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonGenerator.yql
@@ -0,0 +1,26 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+# prepare python udf
+(let i64 (DataType 'Int64))
+(let udfType (CallableType '() '((ListType i64))))
+(let udfScript (String '@@
+def g0():
+    def gen():
+        for x in xrange(1, 4):
+            yield long(x + 42)
+    return gen
+@@))
+(let udf (ScriptUdf 'Python 'g0 udfType udfScript))
+
+# call udf
+(let result (Apply udf))
+
+# output result with type
+(let sink (DataSink 'result))
+(let world (Write! world sink (Key) result '( '('type) )))
+
+# finish
+(return (Commit! world sink))
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorExprWithClosure.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorExprWithClosure.cfg
new file mode 100644
index 0000000000..0bbf3ad9f0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorExprWithClosure.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf python2_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorExprWithClosure.yql b/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorExprWithClosure.yql
new file mode 100644
index 0000000000..86a1c9a0bb
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorExprWithClosure.yql
@@ -0,0 +1,25 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+# prepare python udf
+(let i64 (DataType 'Int64))
+(let udfType (CallableType '() '((ListType i64)) '((ListType i64))))
+(let udfScript (String '@@
+def g(input):
+    def gen():
+        return (long(x + 42) for x in input)
+    return gen
+@@))
+(let udf (ScriptUdf 'Python 'g udfType udfScript))
+
+# call udf
+(let result (Apply udf (AsList (Int64 '1) (Int64 '2) (Int64 '3))))
+
+# output result with type
+(let sink (DataSink 'result))
+(let world (Write! world sink (Key) result '( '('type) )))
+
+# finish
+(return (Commit! world sink))
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorWithClosure.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorWithClosure.cfg
new file mode 100644
index 0000000000..0bbf3ad9f0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorWithClosure.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf python2_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorWithClosure.yql b/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorWithClosure.yql
new file mode 100644
index 0000000000..87799cbf69
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonGeneratorWithClosure.yql
@@ -0,0 +1,26 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+# prepare python udf
+(let i64 (DataType 'Int64))
+(let udfType (CallableType '() '((ListType i64)) '((ListType i64))))
+(let udfScript (String '@@
+def h(input):
+    def gen():
+        for x in input:
+            yield long(x + 42)
+    return gen
+@@))
+(let udf (ScriptUdf 'Python 'h udfType udfScript))
+
+# call udf
+(let result (Apply udf (AsList (Int64 '1) (Int64 '2) (Int64 '3))))
+
+# output result with type
+(let sink (DataSink 'result))
+(let world (Write! world sink (Key) result '( '('type) )))
+
+# finish
+(return (Commit! world sink))
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonPartialCall.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PythonPartialCall.cfg
new file mode 100644
index 0000000000..0bbf3ad9f0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonPartialCall.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf python2_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonPartialCall.yql b/yql/essentials/tests/s-expressions/suites/Udf/PythonPartialCall.yql
new file mode 100644
index 0000000000..f45e2d9157
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonPartialCall.yql
@@ -0,0 +1,22 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+# prepare python udf
+(let ui32 (DataType 'Uint32))
+(let udfType (CallableType '('1) '(ui32) '(ui32) '((OptionalType ui32))))
+(let udfScript (String '"def Add(x, y): return x + (y if y is not None else 1)"))
+(let udf (ScriptUdf 'Python 'Add udfType udfScript))
+
+# call udf
+(let x (Uint32 '20))
+(let result (Apply udf x))
+
+# output result with type
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) result '( '('type) )))
+
+# finish
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonStruct.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PythonStruct.cfg
new file mode 100644
index 0000000000..0bbf3ad9f0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonStruct.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf python2_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonStruct.yql b/yql/essentials/tests/s-expressions/suites/Udf/PythonStruct.yql
new file mode 100644
index 0000000000..38a484a4c3
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonStruct.yql
@@ -0,0 +1,33 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+# prepare python udf
+(let ui32 (DataType 'Uint32))
+(let str (DataType 'String))
+(let personType (StructType '('name str) '('age ui32)))
+(let udfType (CallableType '() '(personType) '(str) '(ui32)))
+(let udfScript (String '@@
+
+class Person:
+    def __init__(self, name, age):
+        self.name = name
+        self.age = age
+
+def NewPerson(name, age):
+    return Person(name, age)
+
+@@))
+(let udf (ScriptUdf 'Python 'NewPerson udfType udfScript))
+
+# call udf
+(let result (Apply udf (String 'Jamel) (Uint32 '99)))
+
+# output result with type
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) result '()))
+
+# finish
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonSum.cfg b/yql/essentials/tests/s-expressions/suites/Udf/PythonSum.cfg
new file mode 100644
index 0000000000..0bbf3ad9f0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonSum.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf python2_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/PythonSum.yql b/yql/essentials/tests/s-expressions/suites/Udf/PythonSum.yql
new file mode 100644
index 0000000000..b240dc0f58
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/PythonSum.yql
@@ -0,0 +1,23 @@
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+# prepare python udf
+(let ui32 (DataType 'Uint32))
+(let udfType (CallableType '() '(ui32) '(ui32) '(ui32)))
+(let udfScript (String '"def Sum(x, y): return x + y"))
+(let udf (ScriptUdf 'Python 'Sum udfType udfScript))
+
+# call udf
+(let x (Uint32 '10))
+(let y (Uint32 '32))
+(let result (Apply udf x y))
+
+# output result with type
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) result '( '('type) )))
+
+# finish
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapCopy.cfg b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapCopy.cfg
new file mode 100644
index 0000000000..ce0f9208d8
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapCopy.cfg
@@ -0,0 +1,3 @@
+in Input input2.txt
+out Output output.txt
+udf dummylog
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapCopy.yql b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapCopy.yql
new file mode 100644
index 0000000000..894e1a15a1
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapCopy.yql
@@ -0,0 +1,11 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let mr_sink (DataSink 'yt (quote plato)))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) table1 '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter.cfg b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter.cfg
new file mode 100644
index 0000000000..cb892c3e09
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter.cfg
@@ -0,0 +1,4 @@
+in Input input2.txt
+out Output output.txt
+udf dummylog
+udf pire_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter.yql b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter.yql
new file mode 100644
index 0000000000..c3752b674e
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter.yql
@@ -0,0 +1,16 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let udf (Udf 'Pire.Match (String '@@.*\d.*@@)))
+(let table1map (Filter table1 (lambda '(item) (block '(
+   (let value (Member item 'value))
+   (return (Apply udf value))
+)))))
+(let mr_sink (DataSink 'yt (quote plato)))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) table1map '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter2.cfg b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter2.cfg
new file mode 100644
index 0000000000..b6d398353a
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter2.cfg
@@ -0,0 +1,5 @@
+in Input input2.txt
+in Input2 input3.txt
+out Output output.txt
+udf dummylog
+udf pire_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter2.yql b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter2.yql
new file mode 100644
index 0000000000..58c57253c1
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapFilter2.yql
@@ -0,0 +1,22 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source 
+'(
+(Key '('table (String 'Input))) 
+(Key '('table (String 'Input2))) 
+)
+
+'('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let udf (Udf 'Pire.Match (String '@@.*\d.*@@)))
+(let table1map (Filter table1 (lambda '(item) (block '(
+   (let value (Member item 'value))
+   (return (Apply udf value))
+)))))
+(let mr_sink (DataSink 'yt (quote plato)))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) table1map '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdf.cfg b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdf.cfg
new file mode 100644
index 0000000000..cf41c46471
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdf.cfg
@@ -0,0 +1,6 @@
+in Input1 input_many1.txt
+in Input2 input_many2.txt
+in Input3 input_many3.txt
+in Input4 input_many4.txt
+out Output output.txt
+udf dummylog
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdf.yql b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdf.yql
new file mode 100644
index 0000000000..fad593cd98
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdf.yql
@@ -0,0 +1,14 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source 
+(Key '('table (MrTableRange '"" (lambda '($i) (And (>= $i (String '"Input1")) (<= $i (String '"Input~")))) '""))) 
+'('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table (Right! x))
+(let tableLimit (Take table (Uint64 '9)))
+(let mr_sink (DataSink 'yt (quote plato)))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) tableLimit '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdfPartial.cfg b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdfPartial.cfg
new file mode 100644
index 0000000000..227f4e34f7
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdfPartial.cfg
@@ -0,0 +1,7 @@
+in Input1 input_many1.txt
+in Input2 input_many1.txt
+in Input3 input_many2.txt
+in Input4 input_many2.txt
+out Output output.txt
+udf dummylog
+udf pire_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdfPartial.yql b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdfPartial.yql
new file mode 100644
index 0000000000..d86392d08a
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeDiffUdfPartial.yql
@@ -0,0 +1,19 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source 
+(Key '('table (MrTableRange '"" (lambda '($i) (And (>= $i (String '"Input1")) (<= $i (String '"Input~")))) '""))) 
+'('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table (Right! x))
+(let udf (Udf 'Pire.Match (String '@@.*\d.*@@)))
+(let table (Filter table (lambda '(item) (block '(
+   (let value (Member item 'value))
+   (return (Apply udf value))
+)))))
+(let tableLimit (Take table (Uint64 '9)))
+(let mr_sink (DataSink 'yt (quote plato)))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) tableLimit '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeSameUdf.cfg b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeSameUdf.cfg
new file mode 100644
index 0000000000..076884ed54
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeSameUdf.cfg
@@ -0,0 +1,6 @@
+in Input1 input_many1.txt
+in Input2 input_many1.txt
+in Input3 input_many1.txt
+in Input4 input_many1.txt
+out Output output.txt
+udf dummylog
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeSameUdf.yql b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeSameUdf.yql
new file mode 100644
index 0000000000..fad593cd98
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapRangeSameUdf.yql
@@ -0,0 +1,14 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source 
+(Key '('table (MrTableRange '"" (lambda '($i) (And (>= $i (String '"Input1")) (<= $i (String '"Input~")))) '""))) 
+'('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table (Right! x))
+(let tableLimit (Take table (Uint64 '9)))
+(let mr_sink (DataSink 'yt (quote plato)))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) tableLimit '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapReduce.cfg b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapReduce.cfg
new file mode 100644
index 0000000000..ce0f9208d8
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapReduce.cfg
@@ -0,0 +1,3 @@
+in Input input2.txt
+out Output output.txt
+udf dummylog
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapReduce.yql b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapReduce.yql
new file mode 100644
index 0000000000..df1bf89415
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapReduce.yql
@@ -0,0 +1,25 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let listHandler (lambda '(stream) (FlatMap stream (lambda '(pair) (block '(
+  (let key (Nth pair '0))
+  (let list (Collect (Nth pair '1)))
+  (let r (FlatMap list (lambda '(x) (block '(
+    (let s (Struct))
+    (let s (AddMember s 'key key))
+    (let s (AddMember s 'subkey (String '.)))
+    (let s (AddMember s 'value (ToString (Length list))))
+    (return (AsList s))
+  )))))
+  (return r)
+))))))
+(let keySelector (lambda '(x) (Member x 'key)))
+(let reducedTable (PartitionByKey table1 keySelector (Void) (Void) listHandler))
+(let mr_sink (DataSink 'yt 'plato))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) reducedTable '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapSort.cfg b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapSort.cfg
new file mode 100644
index 0000000000..ce0f9208d8
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapSort.cfg
@@ -0,0 +1,3 @@
+in Input input2.txt
+out Output output.txt
+udf dummylog
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapSort.yql b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapSort.yql
new file mode 100644
index 0000000000..01df591aa8
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapSort.yql
@@ -0,0 +1,14 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let output (Sort table1 (Bool 'true) (lambda '(item) (block '(
+   (return (Member item 'key))
+)))))
+(let mr_sink (DataSink 'yt (quote plato)))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) output '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWeakRange.cfg b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWeakRange.cfg
new file mode 100644
index 0000000000..43da70bb81
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWeakRange.cfg
@@ -0,0 +1,4 @@
+in Input input2.txt
+out Output output.txt
+udf dummylog
+res result.txt
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWeakRange.yql b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWeakRange.yql
new file mode 100644
index 0000000000..09a346ac6d
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWeakRange.yql
@@ -0,0 +1,11 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (MrTableRange '"" (lambda '($i) (And (>= $i (String '"Input")) (<= $i (String '"Input~")))) '""))) '('key) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) table1 '('('type))))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigCopy.cfg b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigCopy.cfg
new file mode 100644
index 0000000000..5f1288aa57
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigCopy.cfg
@@ -0,0 +1,3 @@
+in Input input4.txt
+out Output output.txt
+udf dummylog
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigCopy.yql b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigCopy.yql
new file mode 100644
index 0000000000..ffc63f7bb5
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigCopy.yql
@@ -0,0 +1,11 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let mr_sink (DataSink 'yt (quote plato)))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) table1 '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigResult.cfg b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigResult.cfg
new file mode 100644
index 0000000000..9bed8d5096
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigResult.cfg
@@ -0,0 +1,3 @@
+in Input input4.txt
+res result.txt
+udf dummylog
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigResult.yql b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigResult.yql
new file mode 100644
index 0000000000..b39a63ed36
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/RecordRemapWithConfigResult.yql
@@ -0,0 +1,11 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) table1 '('('type))))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/ResultWithRemapConfig.cfg b/yql/essentials/tests/s-expressions/suites/Udf/ResultWithRemapConfig.cfg
new file mode 100644
index 0000000000..c60ed039b2
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/ResultWithRemapConfig.cfg
@@ -0,0 +1,2 @@
+in Input input5.txt
+udf structs_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/ResultWithRemapConfig.yql b/yql/essentials/tests/s-expressions/suites/Udf/ResultWithRemapConfig.yql
new file mode 100644
index 0000000000..2f14a89709
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/ResultWithRemapConfig.yql
@@ -0,0 +1,11 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) (Void) '()))
+(let world (Left! x))
+(let table (Right! x))
+(let sink (DataSink 'result))
+(let world (Write! world sink (Key) table '('('type))))
+(let world (Commit! world sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/Sleep.cfg b/yql/essentials/tests/s-expressions/suites/Udf/Sleep.cfg
new file mode 100644
index 0000000000..7cb3e39eea
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/Sleep.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf simple_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/Sleep.yql b/yql/essentials/tests/s-expressions/suites/Udf/Sleep.yql
new file mode 100644
index 0000000000..86d036c80f
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/Sleep.yql
@@ -0,0 +1,35 @@
+(
+(let sleep (Udf 'SimpleUdf.Sleep))
+(let 01sec (Uint64 '100000))
+(let 05sec (Uint64 '500000))
+(let 03sec (Uint64 '300000))
+
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let table1map (Map table1 (lambda '(item) (block '(
+   (let data1 (Apply sleep (+ 05sec 01sec)))
+   (let s (Struct))
+   (let s (AddMember s 'key (Member item 'key)))
+   (let s (AddMember s 'subkey (String  '1)))
+   (let s (AddMember s 'value (ToString data1)))
+   (return s)
+)))))
+(let table2map (Map table1 (lambda '(item) (block '(
+   (let data1 (Apply sleep (+ 03sec 03sec)))
+   (let s (Struct))
+   (let s (AddMember s 'key (Member item 'key)))
+   (let s (AddMember s 'subkey (String  '2)))
+   (let s (AddMember s 'value (ToString data1)))
+   (return s)
+)))))
+(let mr_sink (DataSink 'yt 'plato))
+(let res_sink (DataSink 'result))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) table1map '()))
+(let world (Write! world res_sink (Key) table2map '()))
+(let world (Commit! world mr_sink))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/StringJoinFromList.cfg b/yql/essentials/tests/s-expressions/suites/Udf/StringJoinFromList.cfg
new file mode 100644
index 0000000000..386ab9ac7f
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/StringJoinFromList.cfg
@@ -0,0 +1,3 @@
+in Input input.txt
+out Output output.txt
+udf string_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/StringJoinFromList.yql b/yql/essentials/tests/s-expressions/suites/Udf/StringJoinFromList.yql
new file mode 100644
index 0000000000..6085f8d620
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/StringJoinFromList.yql
@@ -0,0 +1,21 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
+(let world (Left! x))
+(let table1 (Right! x))
+(let udf (Udf 'String.JoinFromList))
+(let table1map (Map table1 (lambda '(item) (block '(
+   (let key (Member item 'key))
+   (let value (Apply udf (AsList (Member item 'key) (Member item 'subkey) (Member item 'value)) (String '"_")))
+   (let s (Struct))
+   (let s (AddMember s 'key key))
+   (let s (AddMember s 'subkey (String '.)))
+   (let s (AddMember s 'value value))
+   (return s)
+)))))
+(let mr_sink (DataSink 'yt (quote plato)))
+(let world (Write! world mr_sink (Key '('table (String 'Output))) table1map '('('mode 'append))))
+(let world (Commit! world mr_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/TableSchemeWithRemap.cfg b/yql/essentials/tests/s-expressions/suites/Udf/TableSchemeWithRemap.cfg
new file mode 100644
index 0000000000..9bed8d5096
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/TableSchemeWithRemap.cfg
@@ -0,0 +1,3 @@
+in Input input4.txt
+res result.txt
+udf dummylog
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/TableSchemeWithRemap.yql b/yql/essentials/tests/s-expressions/suites/Udf/TableSchemeWithRemap.yql
new file mode 100644
index 0000000000..dd7da1dfb0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/TableSchemeWithRemap.yql
@@ -0,0 +1,11 @@
+(
+#comment
+(let mr_source (DataSource 'yt 'plato))
+(let x (Read! world mr_source (Key '('tablescheme (String 'Input))) (Void) '()))
+(let world (Left! x))
+(let scheme (Right! x))
+(let res_sink (DataSink 'result))
+(let world (Write! world res_sink (Key) scheme '('('type))))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.cfg b/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.cfg
new file mode 100644
index 0000000000..38a048bf8d
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.cfg
@@ -0,0 +1,4 @@
+in Input TopFreq.txt
+out Output output.txt
+udf topfreq_udf
+res result.txt
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.txt b/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.txt
new file mode 100644
index 0000000000..b500ea3b02
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.txt
@@ -0,0 +1,16 @@
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="b";"subkey"=".";"value"="10"};
+{"key"="b";"subkey"=".";"value"="2"};
+{"key"="a";"subkey"=".";"value"="3"};
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="a";"subkey"=".";"value"="2"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="b";"subkey"=".";"value"="1"};
+{"key"="b";"subkey"=".";"value"="1"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.txt.attr
new file mode 100644
index 0000000000..b6100e5fd0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.txt.attr
@@ -0,0 +1,30 @@
+{
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
\ No newline at end of file
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.yql b/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.yql
new file mode 100644
index 0000000000..ff322d1a1e
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/TopFreq.yql
@@ -0,0 +1,47 @@
+#program
+(
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(import agg_module '"/lib/yql/aggregate.yql")
+(let list (AsList
+(AsStruct '('key (Int32 '1)) '('value (Just (Int32 '1))))
+(AsStruct '('key (Int32 '1)) '('value (Just (Int32 '4))))
+(AsStruct '('key (Int32 '1)) '('value (Just (Int32 '8))))
+(AsStruct '('key (Int32 '1)) '('value (Just (Int32 '8))))
+(AsStruct '('key (Int32 '1)) '('value (Just (Int32 '3))))
+(AsStruct '('key (Int32 '1)) '('value (Just (Int32 '3))))
+(AsStruct '('key (Int32 '1)) '('value (Just (Int32 '3))))
+(AsStruct '('key (Int32 '1)) '('value (Just (Int32 '8))))
+(AsStruct '('key (Int32 '1)) '('value (Just (Int32 '1))))
+(AsStruct '('key (Int32 '2)) '('value (Just (Int32 '2))))
+(AsStruct '('key (Int32 '2)) '('value (Just (Int32 '2))))
+(AsStruct '('key (Int32 '2)) '('value (Just (Int32 '2))))
+(AsStruct '('key (Int32 '2)) '('value (Just (Int32 '8))))
+(AsStruct '('key (Int32 '2)) '('value (Just (Int32 '3))))
+(AsStruct '('key (Int32 '2)) '('value (Just (Int32 '3))))
+(AsStruct '('key (Int32 '2)) '('value (Just (Int32 '3))))
+(AsStruct '('key (Int32 '2)) '('value (Just (Int32 '8))))
+(AsStruct '('key (Int32 '2)) '('value (Just (Int32 '1))))
+(AsStruct '('key (Int32 '3)) '('value (Just (Int32 '1))))
+(AsStruct '('key (Int32 '3)) '('value (Just (Int32 '2))))
+(AsStruct '('key (Int32 '3)) '('value (Just (Int32 '3))))
+(AsStruct '('key (Int32 '3)) '('value (Just (Int32 '4))))
+(AsStruct '('key (Int32 '3)) '('value (Just (Int32 '5))))
+(AsStruct '('key (Int32 '3)) '('value (Just (Int32 '6))))
+(AsStruct '('key (Int32 '3)) '('value (Just (Int32 '7))))
+(AsStruct '('key (Int32 '3)) '('value (Just (Int32 '8))))
+(AsStruct '('key (Int32 '3)) '('value (Just (Int32 '9))))
+))
+(let res_sink (DataSink 'result))
+(let topfreq_traits (Apply (bind agg_module 'topfreq_traits_factory) (TypeOf list) (lambda '(row) (Member row 'value)) (Uint32 '2) (Uint32 '2)))
+(let resKey (Aggregate list '('key) '('('topfreq_value topfreq_traits))))
+(let world (Write! world res_sink (Key) resKey '('('type))))
+
+(let topfreq_traits2 (Apply (bind agg_module 'topfreq_traits_factory) (TypeOf list) (lambda '(row) (Unwrap (ToString (Member row 'value)))) (Uint32 '2) (Uint32 '2)))
+(let resKey (Aggregate list '('key) '('('topfreq_value topfreq_traits2))))
+(let world (Write! world res_sink (Key) resKey '('('type))))
+
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/TypeInspection.cfg b/yql/essentials/tests/s-expressions/suites/Udf/TypeInspection.cfg
new file mode 100644
index 0000000000..3d92a5a733
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/TypeInspection.cfg
@@ -0,0 +1,2 @@
+udf type_inspection_udf
+res result.txt
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/TypeInspection.yql b/yql/essentials/tests/s-expressions/suites/Udf/TypeInspection.yql
new file mode 100644
index 0000000000..eecfa7e41d
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/TypeInspection.yql
@@ -0,0 +1,32 @@
+(
+#comment
+(let config (DataSource 'config))
+(let world (Configure! world config 'PureDataSource 'yt))
+
+(let res_sink (DataSink 'result))
+(let list1 (AsList (String 'A) (String 'B) (String 'C) (String 'D) (String 'E)))
+(let list2 (AsList (Uint32 '1) (Uint32 '2) (Uint32 '3)))
+(let zip (Udf 'TypeInspection.Zip (Void)
+	(TupleType
+		(TypeOf '(list1 list2))
+		(StructType)
+		(TupleType)
+	)
+))
+(let data (Apply zip list1 list2))
+(let world (Write! world res_sink (Key) data '('('type))))
+
+(let stream1 (Iterator list1))
+(let stream2 (Iterator list2))
+(let interleave (Udf 'TypeInspection.Interleave (Void)
+	(TupleType
+		(TypeOf '(stream1 stream2))
+		(StructType)
+		(TupleType)
+	)
+))
+(let data (Apply interleave stream1 stream2))
+(let world (Write! world res_sink (Key) (Collect data) '('('type))))
+(let world (Commit! world res_sink))
+(return world)
+)
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/block_4_columns.csv b/yql/essentials/tests/s-expressions/suites/Udf/block_4_columns.csv
new file mode 100644
index 0000000000..15956fb4fe
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/block_4_columns.csv
@@ -0,0 +1,7 @@
+a,b,c,d
+1,2,3,4
+one,two,three,four
+A,B,C,D
+
+E,F,G,dgf
+G,H,I,J
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/default.cfg b/yql/essentials/tests/s-expressions/suites/Udf/default.cfg
new file mode 100644
index 0000000000..7cb3e39eea
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/default.cfg
@@ -0,0 +1,4 @@
+in Input input.txt
+out Output output.txt
+res result.txt
+udf simple_udf
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/file_3_columns.csv b/yql/essentials/tests/s-expressions/suites/Udf/file_3_columns.csv
new file mode 100644
index 0000000000..e4324930c7
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/file_3_columns.csv
@@ -0,0 +1,8 @@
+key,subkey,value
+String,String,String
+k1,sk1,v1
+k2,sk2,v2
+
+
+this line will be ignored
+
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input.txt b/yql/essentials/tests/s-expressions/suites/Udf/input.txt
new file mode 100644
index 0000000000..df76c64125
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input.txt
@@ -0,0 +1,4 @@
+{"key"="075";"subkey"="1";"value"="a7c"};
+{"key"="800";"subkey"="2";"value"="ddd"};
+{"key"="020";"subkey"="3";"value"="q"};
+{"key"="150";"subkey"="4";"value"="q8z"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/input.txt.attr
new file mode 100644
index 0000000000..b6100e5fd0
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input.txt.attr
@@ -0,0 +1,30 @@
+{
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
\ No newline at end of file
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input2.txt b/yql/essentials/tests/s-expressions/suites/Udf/input2.txt
new file mode 100644
index 0000000000..df76c64125
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input2.txt
@@ -0,0 +1,4 @@
+{"key"="075";"subkey"="1";"value"="a7c"};
+{"key"="800";"subkey"="2";"value"="ddd"};
+{"key"="020";"subkey"="3";"value"="q"};
+{"key"="150";"subkey"="4";"value"="q8z"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input2.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/input2.txt.attr
new file mode 100644
index 0000000000..c6f7f36696
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input2.txt.attr
@@ -0,0 +1,31 @@
+{
+    "_yql_read_udf"="DummyLog.ReadRecord";
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input3.txt b/yql/essentials/tests/s-expressions/suites/Udf/input3.txt
new file mode 100644
index 0000000000..090fe1cefd
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input3.txt
@@ -0,0 +1,4 @@
+{"key"="834";"subkey"="5";"value"="aa"};
+{"key"="987";"subkey"="6";"value"="798"};
+{"key"="124";"subkey"="7";"value"="001"};
+{"key"="623";"subkey"="8";"value"="002"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input3.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/input3.txt.attr
new file mode 100644
index 0000000000..c6f7f36696
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input3.txt.attr
@@ -0,0 +1,31 @@
+{
+    "_yql_read_udf"="DummyLog.ReadRecord";
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input4.txt b/yql/essentials/tests/s-expressions/suites/Udf/input4.txt
new file mode 100644
index 0000000000..090fe1cefd
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input4.txt
@@ -0,0 +1,4 @@
+{"key"="834";"subkey"="5";"value"="aa"};
+{"key"="987";"subkey"="6";"value"="798"};
+{"key"="124";"subkey"="7";"value"="001"};
+{"key"="623";"subkey"="8";"value"="002"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input4.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/input4.txt.attr
new file mode 100644
index 0000000000..5c5444bff8
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input4.txt.attr
@@ -0,0 +1,33 @@
+{
+    "_yql_read_udf"="DummyLog.ReadRecord2";
+    "_yql_read_udf_type_config"="AAA";
+    "_yql_read_udf_run_config"="BBB";
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input5.txt b/yql/essentials/tests/s-expressions/suites/Udf/input5.txt
new file mode 100644
index 0000000000..de514b7e8b
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input5.txt
@@ -0,0 +1,2 @@
+{"key"="Vasya";"subkey"="Pupkin";"value"="33"};
+{"key"="Naum";"subkey"="Prikhoder";"value"="66"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input5.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/input5.txt.attr
new file mode 100644
index 0000000000..6c072165b5
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input5.txt.attr
@@ -0,0 +1,32 @@
+{
+    "_yql_read_udf"="Person.New";
+    "_yql_read_udf_type_config"="RemapKSV";
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_many1.txt b/yql/essentials/tests/s-expressions/suites/Udf/input_many1.txt
new file mode 100644
index 0000000000..2fa12a4a52
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_many1.txt
@@ -0,0 +1,5 @@
+{"key"="075";"subkey"="1_1";"value"="a7c"};
+{"key"="800";"subkey"="2_1";"value"="ddd"};
+{"key"="020";"subkey"="3_1";"value"="q"};
+{"key"="150";"subkey"="4_1";"value"="q8z"};
+{"key"="200";"subkey"="5_1";"value"="a7a"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_many1.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/input_many1.txt.attr
new file mode 100644
index 0000000000..5f9afadd11
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_many1.txt.attr
@@ -0,0 +1,32 @@
+{
+    "_yql_read_udf"="DummyLog.ReadRecord";
+    "_yql_read_udf_type_config"="1";
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_many2.txt b/yql/essentials/tests/s-expressions/suites/Udf/input_many2.txt
new file mode 100644
index 0000000000..2b1bab42f8
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_many2.txt
@@ -0,0 +1,5 @@
+{"key"="075";"subkey"="1_2";"value"="a7c"};
+{"key"="800";"subkey"="2_2";"value"="ddd"};
+{"key"="020";"subkey"="3_2";"value"="q"};
+{"key"="150";"subkey"="4_2";"value"="q8z"};
+{"key"="200";"subkey"="5_2";"value"="a7a"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_many2.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/input_many2.txt.attr
new file mode 100644
index 0000000000..4e7e0e6b62
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_many2.txt.attr
@@ -0,0 +1,32 @@
+{
+    "_yql_read_udf"="DummyLog.ReadRecord";
+    "_yql_read_udf_type_config"="2";
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_many3.txt b/yql/essentials/tests/s-expressions/suites/Udf/input_many3.txt
new file mode 100644
index 0000000000..2ce03b8925
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_many3.txt
@@ -0,0 +1,4 @@
+{"key"="075";"subkey"="1_3";"value"="a7c"};
+{"key"="800";"subkey"="2_3";"value"="ddd"};
+{"key"="020";"subkey"="3_3";"value"="q"};
+{"key"="150";"subkey"="4_3";"value"="q8z"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_many3.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/input_many3.txt.attr
new file mode 100644
index 0000000000..f09165629a
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_many3.txt.attr
@@ -0,0 +1,32 @@
+{
+    "_yql_read_udf"="DummyLog.ReadRecord";
+    "_yql_read_udf_type_config"="3";
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_many4.txt b/yql/essentials/tests/s-expressions/suites/Udf/input_many4.txt
new file mode 100644
index 0000000000..c9f1e005d9
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_many4.txt
@@ -0,0 +1,4 @@
+{"key"="075";"subkey"="1_4";"value"="a7c"};
+{"key"="800";"subkey"="2_4";"value"="ddd"};
+{"key"="020";"subkey"="3_4";"value"="q"};
+{"key"="150";"subkey"="4_4";"value"="q8z"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_many4.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/input_many4.txt.attr
new file mode 100644
index 0000000000..f491d07f59
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_many4.txt.attr
@@ -0,0 +1,32 @@
+{
+    "_yql_read_udf"="DummyLog.ReadRecord";
+    "_yql_read_udf_type_config"="4";
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_proto.txt b/yql/essentials/tests/s-expressions/suites/Udf/input_proto.txt
new file mode 100644
index 0000000000..d2eaef7f29
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_proto.txt
@@ -0,0 +1 @@
+{"key"="1";"subkey"="\n\x10\x08\7\x12\x0C\n\ntext value";"value"="\n\x10\x08\7\x12\x0C\n\ntext value"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_proto.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/input_proto.txt.attr
new file mode 100644
index 0000000000..f11824f2b8
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_proto.txt.attr
@@ -0,0 +1,32 @@
+{
+    "_yql_proto_field_value"="{\"name\":\"TA\",\"meta\":\"H4sIAAAAAAAAA42Puw6CQBBFszw0DEp0Y4E2GmyolriJHyAPS6rVflFMTBCUR8Hn+KcuD0ui3WRy59wzcIJ1co9yntfOM8/KLKpuTv1KnKp0rqTdWBQk5uM5SEFhKhvZNqhKmE8CPAHF5yU30QbZmqUDCvAIpONuhmA/jL30WCywXsMI+SPuGJAOn0XdGVZDFhfl6qf1VuBdPAX1zJOq46vYAOSbkhh1Kosf4I2GC/mfhb3YcOD78EoYHfACkNva6FQjLZow11rCmJ3SqoivjaPXB4Sj9wEBmoL3ogEAAA==\",\"lists\":{\"optional\":true}}";
+    "_yql_proto_field_subkey"="{\"name\":\"TA\",\"meta\":\"H4sIAAAAAAAAA42Puw6CQBBFszw0DEp0Y4E2GmyolriJHyAPS6rVflFMTBCUR8Hn+KcuD0ui3WRy59wzcIJ1co9yntfOM8/KLKpuTv1KnKp0rqTdWBQk5uM5SEFhKhvZNqhKmE8CPAHF5yU30QbZmqUDCvAIpONuhmA/jL30WCywXsMI+SPuGJAOn0XdGVZDFhfl6qf1VuBdPAX1zJOq46vYAOSbkhh1Kosf4I2GC/mfhb3YcOD78EoYHfACkNva6FQjLZow11rCmJ3SqoivjaPXB4Sj9wEBmoL3ogEAAA==\",\"lists\":{\"optional\":true}}";
+    "_yql_row_spec" = {
+        "Type" = [
+            "StructType";
+            [
+                [
+                    "key";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "subkey";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ];
+                [
+                    "value";
+                    [
+                        "DataType";
+                        "String"
+                    ]
+                ]
+            ]
+        ]
+    }
+}
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_proto_opt.txt b/yql/essentials/tests/s-expressions/suites/Udf/input_proto_opt.txt
new file mode 100644
index 0000000000..d2eaef7f29
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_proto_opt.txt
@@ -0,0 +1 @@
+{"key"="1";"subkey"="\n\x10\x08\7\x12\x0C\n\ntext value";"value"="\n\x10\x08\7\x12\x0C\n\ntext value"};
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/input_proto_opt.txt.attr b/yql/essentials/tests/s-expressions/suites/Udf/input_proto_opt.txt.attr
new file mode 100644
index 0000000000..ad3ff15521
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/input_proto_opt.txt.attr
@@ -0,0 +1,5 @@
+{
+    "_yql_row_spec"={"Type"=["StructType";[["key";["OptionalType";["DataType";"String"]]];["subkey";["OptionalType";["DataType";"String"]]];["value";["OptionalType";["DataType";"String"]]]]]};
+    "_yql_proto_field_value"="{\"name\":\"TA\",\"meta\":\"H4sIAAAAAAAAA42Puw6CQBBFszw0DEp0Y4E2GmyolriJHyAPS6rVflFMTBCUR8Hn+KcuD0ui3WRy59wzcIJ1co9yntfOM8/KLKpuTv1KnKp0rqTdWBQk5uM5SEFhKhvZNqhKmE8CPAHF5yU30QbZmqUDCvAIpONuhmA/jL30WCywXsMI+SPuGJAOn0XdGVZDFhfl6qf1VuBdPAX1zJOq46vYAOSbkhh1Kosf4I2GC/mfhb3YcOD78EoYHfACkNva6FQjLZow11rCmJ3SqoivjaPXB4Sj9wEBmoL3ogEAAA==\",\"lists\":{\"optional\":true}}";
+    "_yql_proto_field_subkey"="{\"name\":\"TA\",\"meta\":\"H4sIAAAAAAAAA42Puw6CQBBFszw0DEp0Y4E2GmyolriJHyAPS6rVflFMTBCUR8Hn+KcuD0ui3WRy59wzcIJ1co9yntfOM8/KLKpuTv1KnKp0rqTdWBQk5uM5SEFhKhvZNqhKmE8CPAHF5yU30QbZmqUDCvAIpONuhmA/jL30WCywXsMI+SPuGJAOn0XdGVZDFhfl6qf1VuBdPAX1zJOq46vYAOSbkhh1Kosf4I2GC/mfhb3YcOD78EoYHfACkNva6FQjLZow11rCmJ3SqoivjaPXB4Sj9wEBmoL3ogEAAA==\",\"lists\":{\"optional\":true}}";
+}
diff --git a/yql/essentials/tests/s-expressions/suites/Udf/struct_4_fields.csv b/yql/essentials/tests/s-expressions/suites/Udf/struct_4_fields.csv
new file mode 100644
index 0000000000..fab478d14a
--- /dev/null
+++ b/yql/essentials/tests/s-expressions/suites/Udf/struct_4_fields.csv
@@ -0,0 +1,5 @@
+F1I,F2OI,F3N,F4ON
+Int32,Uint32?,Float,Double?
+1,,1.5,3.1415926
+5,1,2.5,
+-12,100,44,105
\ No newline at end of file