diff --git a/bugfix-swarm/review_scratch/function_plugin_filter.result b/bugfix-swarm/review_scratch/function_plugin_filter.result
new file mode 100644
index 0000000000000..ba7a0b4316b49
--- /dev/null
+++ b/bugfix-swarm/review_scratch/function_plugin_filter.result
@@ -0,0 +1,56 @@
+#
+# MDEV-41352 SIGSEGV in Item_field::used_tables on a plugin aggregate
+#            with FILTER
+#
+CREATE TABLE t (c INT);
+SELECT test_plugin_count(1) FILTER (WHERE c) FROM t;
+test_plugin_count(1) FILTER (WHERE c)
+0
+DROP TABLE t;
+CREATE TABLE t (id INT PRIMARY KEY, g INT, v INT);
+INSERT INTO t VALUES (1,1,10),(2,1,20),(3,2,30),(4,2,NULL),(5,3,50);
+SELECT test_plugin_count(v) FILTER (WHERE v > 15) FROM t;
+test_plugin_count(v) FILTER (WHERE v > 15)
+3
+SELECT count(v) FILTER (WHERE v > 15) AS c,
+test_plugin_count(v) FILTER (WHERE v > 15) AS pc,
+test_plugin_first(v) FILTER (WHERE v > 15) AS pf
+FROM t;
+c	pc	pf
+3	3	20
+SELECT g, count(v) FILTER (WHERE id > 1) AS c,
+test_plugin_count(v) FILTER (WHERE id > 1) AS pc,
+test_plugin_first(v) FILTER (WHERE id > 1) AS pf
+FROM t GROUP BY g ORDER BY g;
+g	c	pc	pf
+1	1	1	20
+2	1	1	30
+3	1	1	50
+SELECT count(DISTINCT g) FILTER (WHERE v > 15) AS c,
+test_plugin_count(DISTINCT g) FILTER (WHERE v > 15) AS pc
+FROM t;
+c	pc
+3	3
+SELECT id, count(v) FILTER (WHERE v > 15) OVER w AS c,
+test_plugin_count(v) FILTER (WHERE v > 15) OVER w AS pc
+FROM t
+WINDOW w AS (ORDER BY id ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING)
+ORDER BY id;
+id	c	pc
+1	1	1
+2	2	2
+3	2	2
+4	2	2
+5	1	1
+SELECT test_plugin_count(v) FILTER (WHERE count(v) > 1) FROM t;
+ERROR HY000: Incorrect usage of aggregate function and FILTER
+CREATE VIEW v1 AS
+SELECT test_plugin_count(v) FILTER (WHERE v > 15) AS pc FROM t;
+SHOW CREATE VIEW v1;
+View	Create View	character_set_client	collation_connection
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS select test_plugin_count(`t`.`v`) FILTER(WHERE `t`.`v` > 15) AS `pc` from `t`	latin1	latin1_swedish_ci
+SELECT * FROM v1;
+pc
+3
+DROP VIEW v1;
+DROP TABLE t;
diff --git a/bugfix-swarm/review_scratch/function_plugin_filter.test b/bugfix-swarm/review_scratch/function_plugin_filter.test
new file mode 100644
index 0000000000000..0259841c404de
--- /dev/null
+++ b/bugfix-swarm/review_scratch/function_plugin_filter.test
@@ -0,0 +1,38 @@
+--source include/have_udf.inc
+
+--echo #
+--echo # MDEV-41352 SIGSEGV in Item_field::used_tables on a plugin aggregate
+--echo #            with FILTER
+--echo #
+
+CREATE TABLE t (c INT);
+SELECT test_plugin_count(1) FILTER (WHERE c) FROM t;
+DROP TABLE t;
+
+CREATE TABLE t (id INT PRIMARY KEY, g INT, v INT);
+INSERT INTO t VALUES (1,1,10),(2,1,20),(3,2,30),(4,2,NULL),(5,3,50);
+SELECT test_plugin_count(v) FILTER (WHERE v > 15) FROM t;
+SELECT count(v) FILTER (WHERE v > 15) AS c,
+       test_plugin_count(v) FILTER (WHERE v > 15) AS pc,
+       test_plugin_first(v) FILTER (WHERE v > 15) AS pf
+FROM t;
+SELECT g, count(v) FILTER (WHERE id > 1) AS c,
+       test_plugin_count(v) FILTER (WHERE id > 1) AS pc,
+       test_plugin_first(v) FILTER (WHERE id > 1) AS pf
+FROM t GROUP BY g ORDER BY g;
+SELECT count(DISTINCT g) FILTER (WHERE v > 15) AS c,
+       test_plugin_count(DISTINCT g) FILTER (WHERE v > 15) AS pc
+FROM t;
+SELECT id, count(v) FILTER (WHERE v > 15) OVER w AS c,
+       test_plugin_count(v) FILTER (WHERE v > 15) OVER w AS pc
+FROM t
+WINDOW w AS (ORDER BY id ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING)
+ORDER BY id;
+--error ER_WRONG_USAGE
+SELECT test_plugin_count(v) FILTER (WHERE count(v) > 1) FROM t;
+CREATE VIEW v1 AS
+SELECT test_plugin_count(v) FILTER (WHERE v > 15) AS pc FROM t;
+SHOW CREATE VIEW v1;
+SELECT * FROM v1;
+DROP VIEW v1;
+DROP TABLE t;
diff --git a/bugfix-swarm/review_scratch/function_plugin_filter_rv.result b/bugfix-swarm/review_scratch/function_plugin_filter_rv.result
new file mode 100644
index 0000000000000..ae05fc5545df2
--- /dev/null
+++ b/bugfix-swarm/review_scratch/function_plugin_filter_rv.result
@@ -0,0 +1,91 @@
+CREATE TABLE t (id INT PRIMARY KEY, g INT, v INT);
+INSERT INTO t VALUES (1,1,10),(2,1,20),(3,2,30),(4,2,NULL),(5,3,50);
+CREATE TABLE t2 (x INT);
+INSERT INTO t2 VALUES (20),(50);
+# prepared statement, executed twice
+PREPARE s FROM 'SELECT test_plugin_count(v) FILTER (WHERE v > ?) FROM t';
+SET @a=15;
+EXECUTE s USING @a;
+test_plugin_count(v) FILTER (WHERE v > ?)
+3
+SET @a=25;
+EXECUTE s USING @a;
+test_plugin_count(v) FILTER (WHERE v > ?)
+2
+DEALLOCATE PREPARE s;
+# stored procedure called twice
+CREATE PROCEDURE p()
+BEGIN
+SELECT g, test_plugin_count(v) FILTER (WHERE v > 15),
+test_plugin_first(v) FILTER (WHERE v > 15) FROM t GROUP BY g ORDER BY g;
+END|
+CALL p();
+g	test_plugin_count(v) FILTER (WHERE v > 15)	test_plugin_first(v) FILTER (WHERE v > 15)
+1	1	20
+2	1	30
+3	1	50
+CALL p();
+g	test_plugin_count(v) FILTER (WHERE v > 15)	test_plugin_first(v) FILTER (WHERE v > 15)
+1	1	20
+2	1	30
+3	1	50
+DROP PROCEDURE p;
+# subquery in filter
+SELECT test_plugin_count(v) FILTER (WHERE v IN (SELECT x FROM t2)) FROM t;
+test_plugin_count(v) FILTER (WHERE v IN (SELECT x FROM t2))
+2
+SELECT count(v) FILTER (WHERE v IN (SELECT x FROM t2)) FROM t;
+count(v) FILTER (WHERE v IN (SELECT x FROM t2))
+2
+# outer reference in filter of subquery aggregate
+SELECT id, (SELECT test_plugin_count(x) FILTER (WHERE x > t.v) FROM t2) AS pc,
+(SELECT count(x) FILTER (WHERE x > t.v) FROM t2) AS c
+FROM t ORDER BY id;
+id	pc	c
+1	2	2
+2	1	1
+3	1	1
+4	0	0
+5	0	0
+# window function in filter -> error
+SELECT test_plugin_count(v) FILTER (WHERE row_number() OVER () > 1) FROM t;
+ERROR HY000: Incorrect usage of window function and FILTER
+# unknown column in filter -> error
+SELECT test_plugin_count(v) FILTER (WHERE nosuch > 1) FROM t;
+ERROR 42S22: Unknown column 'nosuch' in 'SELECT'
+# HAVING and ORDER BY referencing filtered plugin aggregate
+SELECT g, test_plugin_count(v) FILTER (WHERE v > 15) AS pc FROM t GROUP BY g
+HAVING test_plugin_count(v) FILTER (WHERE v > 15) > 0 ORDER BY pc DESC, g;
+g	pc
+1	1
+2	1
+3	1
+# window, filtered first over sliding frame
+SELECT id, test_plugin_first(v) FILTER (WHERE v > 15) OVER (ORDER BY id) AS pf,
+test_plugin_count(v) FILTER (WHERE id <> 3) OVER (ORDER BY id ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS pc,
+count(v) FILTER (WHERE id <> 3) OVER (ORDER BY id ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS c
+FROM t ORDER BY id;
+id	pf	pc	c
+1	NULL	1	1
+2	20	2	2
+3	20	2	2
+4	20	1	1
+5	20	1	1
+# ROLLUP
+SELECT g, test_plugin_count(v) FILTER (WHERE v > 15) AS pc, count(v) FILTER (WHERE v > 15) AS c FROM t GROUP BY g WITH ROLLUP;
+g	pc	c
+1	1	1
+2	1	1
+3	1	1
+NULL	3	3
+# CREATE TABLE ... SELECT
+CREATE TABLE t3 AS SELECT test_plugin_count(v) FILTER (WHERE v > 15) AS pc FROM t;
+SELECT * FROM t3;
+pc
+3
+DROP TABLE t3;
+# filter always false / empty
+SELECT test_plugin_count(v) FILTER (WHERE 0), test_plugin_first(v) FILTER (WHERE 0) FROM t;
+test_plugin_count(v) FILTER (WHERE 0)	test_plugin_first(v) FILTER (WHERE 0)
+0	NULL
+DROP TABLE t, t2;
diff --git a/bugfix-swarm/review_scratch/function_plugin_filter_rv.test b/bugfix-swarm/review_scratch/function_plugin_filter_rv.test
new file mode 100644
index 0000000000000..e4722370f465c
--- /dev/null
+++ b/bugfix-swarm/review_scratch/function_plugin_filter_rv.test
@@ -0,0 +1,66 @@
+--source include/have_udf.inc
+
+CREATE TABLE t (id INT PRIMARY KEY, g INT, v INT);
+INSERT INTO t VALUES (1,1,10),(2,1,20),(3,2,30),(4,2,NULL),(5,3,50);
+CREATE TABLE t2 (x INT);
+INSERT INTO t2 VALUES (20),(50);
+
+--echo # prepared statement, executed twice
+PREPARE s FROM 'SELECT test_plugin_count(v) FILTER (WHERE v > ?) FROM t';
+SET @a=15;
+EXECUTE s USING @a;
+SET @a=25;
+EXECUTE s USING @a;
+DEALLOCATE PREPARE s;
+
+--echo # stored procedure called twice
+DELIMITER |;
+CREATE PROCEDURE p()
+BEGIN
+  SELECT g, test_plugin_count(v) FILTER (WHERE v > 15),
+         test_plugin_first(v) FILTER (WHERE v > 15) FROM t GROUP BY g ORDER BY g;
+END|
+DELIMITER ;|
+CALL p();
+CALL p();
+DROP PROCEDURE p;
+
+--echo # subquery in filter
+SELECT test_plugin_count(v) FILTER (WHERE v IN (SELECT x FROM t2)) FROM t;
+SELECT count(v) FILTER (WHERE v IN (SELECT x FROM t2)) FROM t;
+
+--echo # outer reference in filter of subquery aggregate
+SELECT id, (SELECT test_plugin_count(x) FILTER (WHERE x > t.v) FROM t2) AS pc,
+           (SELECT count(x) FILTER (WHERE x > t.v) FROM t2) AS c
+FROM t ORDER BY id;
+
+--echo # window function in filter -> error
+--error ER_WRONG_USAGE
+SELECT test_plugin_count(v) FILTER (WHERE row_number() OVER () > 1) FROM t;
+
+--echo # unknown column in filter -> error
+--error ER_BAD_FIELD_ERROR
+SELECT test_plugin_count(v) FILTER (WHERE nosuch > 1) FROM t;
+
+--echo # HAVING and ORDER BY referencing filtered plugin aggregate
+SELECT g, test_plugin_count(v) FILTER (WHERE v > 15) AS pc FROM t GROUP BY g
+HAVING test_plugin_count(v) FILTER (WHERE v > 15) > 0 ORDER BY pc DESC, g;
+
+--echo # window, filtered first over sliding frame
+SELECT id, test_plugin_first(v) FILTER (WHERE v > 15) OVER (ORDER BY id) AS pf,
+           test_plugin_count(v) FILTER (WHERE id <> 3) OVER (ORDER BY id ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS pc,
+           count(v) FILTER (WHERE id <> 3) OVER (ORDER BY id ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS c
+FROM t ORDER BY id;
+
+--echo # ROLLUP
+SELECT g, test_plugin_count(v) FILTER (WHERE v > 15) AS pc, count(v) FILTER (WHERE v > 15) AS c FROM t GROUP BY g WITH ROLLUP;
+
+--echo # CREATE TABLE ... SELECT
+CREATE TABLE t3 AS SELECT test_plugin_count(v) FILTER (WHERE v > 15) AS pc FROM t;
+SELECT * FROM t3;
+DROP TABLE t3;
+
+--echo # filter always false / empty
+SELECT test_plugin_count(v) FILTER (WHERE 0), test_plugin_first(v) FILTER (WHERE 0) FROM t;
+
+DROP TABLE t, t2;
diff --git a/include/mysql/plugin_function.h b/include/mysql/plugin_function.h
index 117acd43222ef..89dfa1c47d354 100644
--- a/include/mysql/plugin_function.h
+++ b/include/mysql/plugin_function.h
@@ -29,6 +29,15 @@
 
 /*
   API for function plugins. (MariaDB_FUNCTION_PLUGIN)
+
+  An aggregate function's Create_func must create an Item_sum_plugin descendant.
+  Aggregate implementations consume rows in add_row() and use aggregation_arg()
+  to read values, including values replayed by DISTINCT. FILTER (WHERE ...) is
+  applied by the server: add_row() and remove_row() see only passing rows.
+  DISTINCT does not guarantee replay order; order-sensitive implementations
+  must override supports_distinct() to return false. Aggregates can implement
+  supports_removal() and remove_row() for moving window frames, and
+  val_native() to preserve a pluggable result type.
 */
 #define MariaDB_FUNCTION_INTERFACE_VERSION (MYSQL_VERSION_ID << 8)
 
diff --git a/plugin/func_test/mysql-test/func_test/function_plugin.result b/plugin/func_test/mysql-test/func_test/function_plugin.result
new file mode 100644
index 0000000000000..7f276b616cb0c
--- /dev/null
+++ b/plugin/func_test/mysql-test/func_test/function_plugin.result
@@ -0,0 +1,184 @@
+CREATE TABLE t1 (a INT, b INT);
+INSERT INTO t1 VALUES (1, 1), (2, 1), (NULL, 1), (3, 2);
+SELECT test_plugin_count(a) FROM t1;
+test_plugin_count(a)
+3
+SELECT b, test_plugin_count(a) FROM t1 GROUP BY b ORDER BY b;
+b	test_plugin_count(a)
+1	2
+2	1
+SELECT test_plugin_count(DISTINCT a) FROM
+(SELECT 1 AS a UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT NULL) dt;
+test_plugin_count(DISTINCT a)
+2
+SELECT test_plugin_count(a) FROM t1 WHERE FALSE;
+test_plugin_count(a)
+0
+SELECT test_plugin_count(a) FROM t1 WHERE a IS NULL;
+test_plugin_count(a)
+0
+CREATE TABLE t_empty (a INT);
+SELECT test_plugin_count(a) AS count_value,
+test_plugin_first(a) AS first_value
+FROM t_empty;
+Catalog	Database	Table	Table_alias	Column	Column_alias	Type	Length	Max length	Is_null	Flags	Decimals	Charsetnr
+def					count_value	8	21	1	N	32897	0	63
+def					first_value	3	11	0	Y	32896	0	63
+count_value	first_value
+0	NULL
+CREATE TABLE t_empty_result AS
+SELECT test_plugin_count(a) AS count_value,
+test_plugin_first(a) AS first_value
+FROM t_empty;
+SHOW CREATE TABLE t_empty_result;
+Table	Create Table
+t_empty_result	CREATE TABLE `t_empty_result` (
+  `count_value` bigint(21) NOT NULL,
+  `first_value` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci
+SELECT count_value, first_value,
+count_value IS NULL, first_value IS NULL
+FROM t_empty_result;
+count_value	first_value	count_value IS NULL	first_value IS NULL
+0	NULL	0	1
+DROP TABLE t_empty_result, t_empty;
+SET @save_optimizer_switch= @@optimizer_switch;
+SET optimizer_switch= 'derived_merge=off';
+SELECT b, test_plugin_count(a)
+FROM (SELECT a, b FROM t1) AS dt
+GROUP BY b ORDER BY b;
+b	test_plugin_count(a)
+1	2
+2	1
+SET optimizer_switch= @save_optimizer_switch;
+PREPARE stmt FROM
+'SELECT test_plugin_count(a) FROM t1 WHERE b = ?';
+SET @b= 1;
+EXECUTE stmt USING @b;
+test_plugin_count(a)
+2
+SET @b= 2;
+EXECUTE stmt USING @b;
+test_plugin_count(a)
+1
+SET @b= 3;
+EXECUTE stmt USING @b;
+test_plugin_count(a)
+0
+DEALLOCATE PREPARE stmt;
+SELECT a, test_plugin_count(a) OVER (ORDER BY a) FROM t1 ORDER BY a;
+a	test_plugin_count(a) OVER (ORDER BY a)
+NULL	0
+1	1
+2	2
+3	3
+CREATE TABLE t_window (id INT, part INT, value INT);
+INSERT INTO t_window VALUES
+(1, 1, 10), (2, 1, NULL), (3, 1, 30), (4, 1, 40),
+(5, 2, NULL), (6, 2, 60);
+SELECT id, part, value,
+test_plugin_count(value) OVER (
+PARTITION BY part ORDER BY id
+ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) AS cnt
+FROM t_window ORDER BY id;
+id	part	value	cnt
+1	1	10	1
+2	1	NULL	2
+3	1	30	2
+4	1	40	2
+5	2	NULL	1
+6	2	60	1
+SELECT test_plugin_count(DISTINCT a) OVER () FROM t1;
+ERROR 42000: This version of MariaDB doesn't yet support 'plugin aggregate with DISTINCT as window function'
+SELECT sysconst_test() OVER ();
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'OVER ()' at line 1
+SELECT strnxfrm(DISTINCT 'a', 1, 1, 0);
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near '' at line 1
+CREATE TABLE t_uuid (id INT, grp INT, value UUID);
+INSERT INTO t_uuid VALUES
+(1, 1, NULL),
+(2, 1, '00000000-0000-0000-0000-000000000001'),
+(3, 2, '00000000-0000-0000-0000-000000000002'),
+(4, 2, NULL),
+(5, 3, NULL);
+SELECT test_plugin_first(value) FROM t_uuid WHERE grp = 1;
+test_plugin_first(value)
+00000000-0000-0000-0000-000000000001
+SELECT grp, test_plugin_first(value) FROM t_uuid GROUP BY grp ORDER BY grp;
+grp	test_plugin_first(value)
+1	00000000-0000-0000-0000-000000000001
+2	00000000-0000-0000-0000-000000000002
+3	NULL
+SELECT id, grp, test_plugin_first(value) OVER (
+PARTITION BY grp ORDER BY id
+ROWS BETWEEN CURRENT ROW AND 1 FOLLOWING) AS value
+FROM t_uuid ORDER BY id;
+id	grp	value
+1	1	00000000-0000-0000-0000-000000000001
+2	1	00000000-0000-0000-0000-000000000001
+3	2	00000000-0000-0000-0000-000000000002
+4	2	NULL
+5	3	NULL
+SELECT test_plugin_count(DISTINCT value) FROM t_uuid;
+test_plugin_count(DISTINCT value)
+2
+SELECT test_plugin_first(DISTINCT value) FROM
+(SELECT value FROM t_uuid WHERE id = 2
+UNION ALL
+SELECT value FROM t_uuid WHERE id = 2) AS dt;
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+SELECT test_plugin_first(DISTINCT value) =
+(SELECT value FROM t_uuid WHERE id = 2)
+FROM t_uuid WHERE grp = 1;
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+SELECT test_plugin_first(value) IS NULL FROM t_uuid WHERE FALSE;
+test_plugin_first(value) IS NULL
+1
+SELECT test_plugin_first(value) IS NULL FROM t_uuid WHERE grp = 3;
+test_plugin_first(value) IS NULL
+1
+SELECT test_plugin_first(value) =
+(SELECT value FROM t_uuid WHERE id = 2)
+FROM t_uuid WHERE grp = 1;
+test_plugin_first(value) =
+(SELECT value FROM t_uuid WHERE id = 2)
+1
+SET @save_optimizer_switch= @@optimizer_switch;
+SET optimizer_switch= 'derived_merge=off';
+SELECT value, value = (SELECT value FROM t_uuid WHERE id = 3)
+FROM
+(SELECT test_plugin_first(value) AS value
+FROM t_uuid WHERE grp = 2) AS dt;
+value	value = (SELECT value FROM t_uuid WHERE id = 3)
+00000000-0000-0000-0000-000000000002	1
+SET optimizer_switch= @save_optimizer_switch;
+CREATE TABLE t_uuid_aggregate AS
+SELECT grp, test_plugin_first(value) AS value
+FROM t_uuid GROUP BY grp;
+SHOW CREATE TABLE t_uuid_aggregate;
+Table	Create Table
+t_uuid_aggregate	CREATE TABLE `t_uuid_aggregate` (
+  `grp` int(11) DEFAULT NULL,
+  `value` uuid DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci
+SELECT * FROM t_uuid_aggregate ORDER BY grp;
+grp	value
+1	00000000-0000-0000-0000-000000000001
+2	00000000-0000-0000-0000-000000000002
+3	NULL
+CREATE TABLE t_uuid_result (value UUID);
+INSERT INTO t_uuid_result
+SELECT test_plugin_first(value) FROM t_uuid WHERE grp = 1;
+SELECT value, value = (SELECT value FROM t_uuid WHERE id = 2)
+FROM t_uuid_result;
+value	value = (SELECT value FROM t_uuid WHERE id = 2)
+00000000-0000-0000-0000-000000000001	1
+CREATE TABLE t_blob (value TEXT);
+INSERT INTO t_blob VALUES ('a'), ('a'), ('b'), (NULL);
+SELECT test_plugin_count(DISTINCT value) FROM t_blob;
+test_plugin_count(DISTINCT value)
+2
+SELECT test_plugin_first(DISTINCT value) IS NOT NULL FROM t_blob;
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+DROP TABLE t_blob, t_uuid_result, t_uuid_aggregate, t_uuid, t_window;
+DROP TABLE t1;
diff --git a/plugin/func_test/mysql-test/func_test/function_plugin.test b/plugin/func_test/mysql-test/func_test/function_plugin.test
new file mode 100644
index 0000000000000..cc7c6e6d7212a
--- /dev/null
+++ b/plugin/func_test/mysql-test/func_test/function_plugin.test
@@ -0,0 +1,127 @@
+--source include/have_udf.inc
+
+CREATE TABLE t1 (a INT, b INT);
+INSERT INTO t1 VALUES (1, 1), (2, 1), (NULL, 1), (3, 2);
+
+SELECT test_plugin_count(a) FROM t1;
+SELECT b, test_plugin_count(a) FROM t1 GROUP BY b ORDER BY b;
+SELECT test_plugin_count(DISTINCT a) FROM
+  (SELECT 1 AS a UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT NULL) dt;
+SELECT test_plugin_count(a) FROM t1 WHERE FALSE;
+SELECT test_plugin_count(a) FROM t1 WHERE a IS NULL;
+
+CREATE TABLE t_empty (a INT);
+--disable_ps_protocol
+--disable_view_protocol
+--enable_metadata
+SELECT test_plugin_count(a) AS count_value,
+       test_plugin_first(a) AS first_value
+FROM t_empty;
+--disable_metadata
+--enable_view_protocol
+--enable_ps_protocol
+CREATE TABLE t_empty_result AS
+SELECT test_plugin_count(a) AS count_value,
+       test_plugin_first(a) AS first_value
+FROM t_empty;
+SHOW CREATE TABLE t_empty_result;
+SELECT count_value, first_value,
+       count_value IS NULL, first_value IS NULL
+FROM t_empty_result;
+DROP TABLE t_empty_result, t_empty;
+
+SET @save_optimizer_switch= @@optimizer_switch;
+SET optimizer_switch= 'derived_merge=off';
+SELECT b, test_plugin_count(a)
+FROM (SELECT a, b FROM t1) AS dt
+GROUP BY b ORDER BY b;
+SET optimizer_switch= @save_optimizer_switch;
+
+PREPARE stmt FROM
+  'SELECT test_plugin_count(a) FROM t1 WHERE b = ?';
+SET @b= 1;
+EXECUTE stmt USING @b;
+SET @b= 2;
+EXECUTE stmt USING @b;
+SET @b= 3;
+EXECUTE stmt USING @b;
+DEALLOCATE PREPARE stmt;
+
+SELECT a, test_plugin_count(a) OVER (ORDER BY a) FROM t1 ORDER BY a;
+
+CREATE TABLE t_window (id INT, part INT, value INT);
+INSERT INTO t_window VALUES
+  (1, 1, 10), (2, 1, NULL), (3, 1, 30), (4, 1, 40),
+  (5, 2, NULL), (6, 2, 60);
+SELECT id, part, value,
+       test_plugin_count(value) OVER (
+         PARTITION BY part ORDER BY id
+         ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) AS cnt
+FROM t_window ORDER BY id;
+
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_count(DISTINCT a) OVER () FROM t1;
+
+--error ER_PARSE_ERROR
+SELECT sysconst_test() OVER ();
+--error ER_PARSE_ERROR
+SELECT strnxfrm(DISTINCT 'a', 1, 1, 0);
+
+CREATE TABLE t_uuid (id INT, grp INT, value UUID);
+INSERT INTO t_uuid VALUES
+  (1, 1, NULL),
+  (2, 1, '00000000-0000-0000-0000-000000000001'),
+  (3, 2, '00000000-0000-0000-0000-000000000002'),
+  (4, 2, NULL),
+  (5, 3, NULL);
+
+SELECT test_plugin_first(value) FROM t_uuid WHERE grp = 1;
+SELECT grp, test_plugin_first(value) FROM t_uuid GROUP BY grp ORDER BY grp;
+SELECT id, grp, test_plugin_first(value) OVER (
+         PARTITION BY grp ORDER BY id
+         ROWS BETWEEN CURRENT ROW AND 1 FOLLOWING) AS value
+FROM t_uuid ORDER BY id;
+SELECT test_plugin_count(DISTINCT value) FROM t_uuid;
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT value) FROM
+  (SELECT value FROM t_uuid WHERE id = 2
+   UNION ALL
+   SELECT value FROM t_uuid WHERE id = 2) AS dt;
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT value) =
+       (SELECT value FROM t_uuid WHERE id = 2)
+FROM t_uuid WHERE grp = 1;
+SELECT test_plugin_first(value) IS NULL FROM t_uuid WHERE FALSE;
+SELECT test_plugin_first(value) IS NULL FROM t_uuid WHERE grp = 3;
+SELECT test_plugin_first(value) =
+       (SELECT value FROM t_uuid WHERE id = 2)
+FROM t_uuid WHERE grp = 1;
+
+SET @save_optimizer_switch= @@optimizer_switch;
+SET optimizer_switch= 'derived_merge=off';
+SELECT value, value = (SELECT value FROM t_uuid WHERE id = 3)
+FROM
+  (SELECT test_plugin_first(value) AS value
+   FROM t_uuid WHERE grp = 2) AS dt;
+SET optimizer_switch= @save_optimizer_switch;
+
+CREATE TABLE t_uuid_aggregate AS
+  SELECT grp, test_plugin_first(value) AS value
+  FROM t_uuid GROUP BY grp;
+SHOW CREATE TABLE t_uuid_aggregate;
+SELECT * FROM t_uuid_aggregate ORDER BY grp;
+
+CREATE TABLE t_uuid_result (value UUID);
+INSERT INTO t_uuid_result
+  SELECT test_plugin_first(value) FROM t_uuid WHERE grp = 1;
+SELECT value, value = (SELECT value FROM t_uuid WHERE id = 2)
+FROM t_uuid_result;
+
+CREATE TABLE t_blob (value TEXT);
+INSERT INTO t_blob VALUES ('a'), ('a'), ('b'), (NULL);
+SELECT test_plugin_count(DISTINCT value) FROM t_blob;
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT value) IS NOT NULL FROM t_blob;
+
+DROP TABLE t_blob, t_uuid_result, t_uuid_aggregate, t_uuid, t_window;
+DROP TABLE t1;
diff --git a/plugin/func_test/mysql-test/func_test/function_plugin_extra.result b/plugin/func_test/mysql-test/func_test/function_plugin_extra.result
new file mode 100644
index 0000000000000..5eea014de5352
--- /dev/null
+++ b/plugin/func_test/mysql-test/func_test/function_plugin_extra.result
@@ -0,0 +1,302 @@
+SELECT test_plugin_count();
+ERROR 42000: Incorrect parameter count in the call to native function 'test_plugin_count'
+SELECT test_plugin_count(1, 2);
+ERROR 42000: Incorrect parameter count in the call to native function 'test_plugin_count'
+SELECT test_plugin_first();
+ERROR 42000: Incorrect parameter count in the call to native function 'test_plugin_first'
+CREATE TABLE t_group (g INT, v INT);
+INSERT INTO t_group VALUES
+(1, NULL), (1, NULL),
+(2, 10), (2, NULL),
+(3, 20), (3, 30);
+SELECT * FROM t_group WHERE test_plugin_count(v) > 0;
+ERROR HY000: Invalid use of group function
+SELECT test_plugin_count(test_plugin_count(v)) FROM t_group;
+ERROR HY000: Invalid use of group function
+SELECT test_plugin_count(v) FROM t_group;
+test_plugin_count(v)
+3
+SELECT test_plugin_count(v) FROM t_group
+HAVING test_plugin_count(v) > 0;
+test_plugin_count(v)
+3
+SELECT g, test_plugin_count(v) AS c
+FROM t_group GROUP BY g HAVING c > 0 ORDER BY c, g;
+g	c
+2	1
+3	2
+SELECT test_plugin_count(v), test_plugin_count(v),
+test_plugin_count(v + 1), test_plugin_first(v)
+FROM t_group;
+test_plugin_count(v)	test_plugin_count(v)	test_plugin_count(v + 1)	test_plugin_first(v)
+3	3	3	10
+SELECT g, test_plugin_count(v), test_plugin_first(v)
+FROM t_group GROUP BY g ORDER BY g;
+g	test_plugin_count(v)	test_plugin_first(v)
+1	0	NULL
+2	1	10
+3	2	20
+SELECT g, test_plugin_count(v)
+FROM t_group GROUP BY g WITH ROLLUP;
+g	test_plugin_count(v)
+1	0
+2	1
+3	2
+NULL	3
+SET @save_optimizer_switch= @@optimizer_switch;
+SET optimizer_switch= 'derived_merge=off';
+SELECT * FROM
+(SELECT g, test_plugin_count(v) AS c
+FROM t_group GROUP BY g) AS dt
+ORDER BY g;
+g	c
+1	0
+2	1
+3	2
+WITH cte AS
+(SELECT g, test_plugin_count(v) AS c
+FROM t_group GROUP BY g)
+SELECT * FROM cte ORDER BY g;
+g	c
+1	0
+2	1
+3	2
+SET optimizer_switch= @save_optimizer_switch;
+SELECT g, test_plugin_count(DISTINCT v)
+FROM t_group GROUP BY g ORDER BY g;
+g	test_plugin_count(DISTINCT v)
+1	0
+2	1
+3	2
+SELECT test_plugin_count(DISTINCT 1);
+test_plugin_count(DISTINCT 1)
+1
+SELECT test_plugin_count(DISTINCT 1.5);
+test_plugin_count(DISTINCT 1.5)
+1
+SELECT test_plugin_count(DISTINCT 1.5e0);
+test_plugin_count(DISTINCT 1.5e0)
+1
+SELECT test_plugin_first(DISTINCT 2.5);
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+SELECT test_plugin_first(DISTINCT 2.5e0);
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+SELECT test_plugin_count(DISTINCT v) FROM t_group WHERE v IS NULL;
+test_plugin_count(DISTINCT v)
+0
+SELECT test_plugin_first(DISTINCT v) IS NULL FROM t_group WHERE v IS NULL;
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+SELECT test_plugin_count(DISTINCT v) FROM t_group WHERE FALSE;
+test_plugin_count(DISTINCT v)
+0
+SELECT test_plugin_first(DISTINCT v) IS NULL FROM t_group WHERE FALSE;
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+CREATE TABLE t_distinct (v INT);
+INSERT INTO t_distinct VALUES (1),(1),(2),(2),(2),(3),(3),(NULL),(4);
+SELECT COUNT(DISTINCT v), test_plugin_count(DISTINCT v) FROM t_distinct;
+COUNT(DISTINCT v)	test_plugin_count(DISTINCT v)
+4	4
+CREATE VIEW v_distinct AS
+SELECT COUNT(DISTINCT v) AS b,
+test_plugin_count(DISTINCT v) AS pl
+FROM t_distinct;
+SHOW CREATE VIEW v_distinct;
+View	Create View	character_set_client	collation_connection
+v_distinct	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v_distinct` AS select count(distinct `t_distinct`.`v`) AS `b`,test_plugin_count(distinct `t_distinct`.`v`) AS `pl` from `t_distinct`	latin1	latin1_swedish_ci
+SELECT b, pl FROM v_distinct;
+b	pl
+4	4
+EXPLAIN EXTENDED SELECT test_plugin_count(DISTINCT v) AS b FROM t_distinct;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
+1	SIMPLE	t_distinct	ALL	NULL	NULL	NULL	NULL	#	100.00	
+Warnings:
+Note	1003	select test_plugin_count(distinct `test`.`t_distinct`.`v`) AS `b` from `test`.`t_distinct`
+SHOW WARNINGS;
+Level	Code	Message
+Note	1003	select test_plugin_count(distinct `test`.`t_distinct`.`v`) AS `b` from `test`.`t_distinct`
+DROP VIEW v_distinct;
+DROP TABLE t_distinct;
+CREATE TABLE t_blob (value LONGTEXT);
+INSERT INTO t_blob VALUES
+(REPEAT('a', 10000)), (REPEAT('a', 10000)),
+(REPEAT('b', 10000)), (NULL);
+SELECT test_plugin_count(DISTINCT value) FROM t_blob;
+test_plugin_count(DISTINCT value)
+2
+SELECT LENGTH(test_plugin_first(DISTINCT value)) FROM t_blob;
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+CREATE TABLE t_distinct_order (i INT, vc VARCHAR(16), tx TEXT);
+INSERT INTO t_distinct_order VALUES
+(9,'zzz','zzz'), (5,'mmm','mmm'), (9,'zzz','zzz'),
+(1,'aaa','aaa'), (5,'mmm','mmm');
+SELECT test_plugin_first(DISTINCT i) FROM t_distinct_order;
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+SELECT test_plugin_first(DISTINCT vc) FROM t_distinct_order;
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+SELECT test_plugin_first(DISTINCT tx) FROM t_distinct_order;
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+DROP TABLE t_distinct_order;
+SET @save_tmp_table_size= @@tmp_table_size;
+SET @save_max_heap_table_size= @@max_heap_table_size;
+SET tmp_table_size= 16384;
+SET max_heap_table_size= 16384;
+CREATE TABLE t_spill AS
+WITH RECURSIVE seq AS
+(SELECT 1 AS value UNION ALL
+SELECT value + 1 FROM seq WHERE value < 1000)
+SELECT value FROM seq;
+INSERT INTO t_spill SELECT value FROM t_spill;
+SELECT test_plugin_count(DISTINCT value) FROM t_spill;
+test_plugin_count(DISTINCT value)
+1000
+DROP TABLE t_spill;
+SET tmp_table_size= @save_tmp_table_size;
+SET max_heap_table_size= @save_max_heap_table_size;
+INSTALL SONAME 'type_test';
+CREATE TABLE t_type (g INT, i test_int8, d test_double);
+INSERT INTO t_type VALUES
+(1, 10, 1.5), (1, 20, 2.5), (2, NULL, NULL);
+SELECT g, test_plugin_first(i), test_plugin_first(d)
+FROM t_type GROUP BY g ORDER BY g;
+g	test_plugin_first(i)	test_plugin_first(d)
+1	10	1.5
+2	NULL	NULL
+CREATE TABLE type_int_result AS
+SELECT test_plugin_first(i) AS value FROM t_type;
+SHOW CREATE TABLE type_int_result;
+Table	Create Table
+type_int_result	CREATE TABLE `type_int_result` (
+  `value` test_int8(20) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci
+CREATE TABLE type_double_result AS
+SELECT test_plugin_first(d) AS value FROM t_type;
+SHOW CREATE TABLE type_double_result;
+Table	Create Table
+type_double_result	CREATE TABLE `type_double_result` (
+  `value` test_double DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci
+CREATE VIEW type_view AS
+SELECT test_plugin_first(i) AS value FROM t_type;
+SHOW CREATE VIEW type_view;
+View	Create View	character_set_client	collation_connection
+type_view	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `type_view` AS select test_plugin_first(`t_type`.`i`) AS `value` from `t_type`	latin1	latin1_swedish_ci
+SELECT * FROM type_view;
+value
+10
+CREATE TEMPORARY TABLE type_tmp AS
+SELECT g, test_plugin_first(i) AS value
+FROM t_type GROUP BY g;
+SHOW CREATE TABLE type_tmp;
+Table	Create Table
+type_tmp	CREATE TEMPORARY TABLE `type_tmp` (
+  `g` int(11) DEFAULT NULL,
+  `value` test_int8(20) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci
+SELECT * FROM type_tmp ORDER BY g;
+g	value
+1	10
+2	NULL
+CREATE TABLE type_dst (value test_int8);
+INSERT INTO type_dst SELECT test_plugin_first(i) FROM t_type;
+SELECT * FROM type_dst;
+value
+10
+DROP VIEW type_view;
+DROP TEMPORARY TABLE type_tmp;
+DROP TABLE type_dst, type_int_result, type_double_result, t_type;
+UNINSTALL SONAME 'type_test';
+CREATE TABLE t_uuid (id INT, g INT, value UUID);
+INSERT INTO t_uuid VALUES
+(1, 1, '00000000-0000-0000-0000-000000000001'),
+(2, 1, '00000000-0000-0000-0000-000000000001'),
+(3, 1, '00000000-0000-0000-0000-000000000002'),
+(4, 2, NULL),
+(5, 2, '00000000-0000-0000-0000-000000000003');
+SELECT test_plugin_count(DISTINCT value) FROM t_uuid;
+test_plugin_count(DISTINCT value)
+3
+SELECT test_plugin_first(value) =
+(SELECT value FROM t_uuid WHERE id = 1)
+FROM t_uuid WHERE g = 1;
+test_plugin_first(value) =
+(SELECT value FROM t_uuid WHERE id = 1)
+1
+CREATE TABLE uuid_union AS
+SELECT test_plugin_first(value) AS value FROM t_uuid
+UNION ALL
+SELECT value FROM t_uuid WHERE id = 1;
+SHOW CREATE TABLE uuid_union;
+Table	Create Table
+uuid_union	CREATE TABLE `uuid_union` (
+  `value` uuid DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci
+SELECT CASE WHEN 1 THEN test_plugin_first(value)
+ELSE CAST(NULL AS UUID) END AS value
+FROM t_uuid;
+value
+00000000-0000-0000-0000-000000000001
+CREATE TABLE t_window (id INT, g INT, v INT, u UUID);
+INSERT INTO t_window VALUES
+(1, 1, 10, '00000000-0000-0000-0000-000000000001'),
+(2, 1, NULL, NULL),
+(3, 1, 30, '00000000-0000-0000-0000-000000000003'),
+(4, 1, 40, '00000000-0000-0000-0000-000000000004'),
+(5, 2, NULL, NULL),
+(6, 2, 60, '00000000-0000-0000-0000-000000000006');
+SELECT id,
+test_plugin_count(v) OVER (
+PARTITION BY g ORDER BY id
+ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS preceding_count,
+test_plugin_count(v) OVER (
+PARTITION BY g ORDER BY id
+ROWS BETWEEN CURRENT ROW AND 2 FOLLOWING) AS following_count,
+test_plugin_count(v) OVER (
+PARTITION BY g ORDER BY id
+ROWS BETWEEN 2 FOLLOWING AND 3 FOLLOWING) AS empty_count
+FROM t_window ORDER BY id;
+id	preceding_count	following_count	empty_count
+1	1	2	2
+2	1	2	1
+3	2	2	0
+4	2	1	0
+5	0	1	0
+6	1	1	0
+SELECT id,
+test_plugin_count(v) OVER w AS c1,
+test_plugin_count(v) OVER w AS c2,
+test_plugin_first(u) OVER w AS first_uuid
+FROM t_window
+WINDOW w AS (
+PARTITION BY g ORDER BY id
+ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING)
+ORDER BY id;
+id	c1	c2	first_uuid
+1	1	1	00000000-0000-0000-0000-000000000001
+2	2	2	00000000-0000-0000-0000-000000000001
+3	2	2	00000000-0000-0000-0000-000000000003
+4	2	2	00000000-0000-0000-0000-000000000003
+5	1	1	00000000-0000-0000-0000-000000000006
+6	1	1	00000000-0000-0000-0000-000000000006
+CREATE TABLE window_uuid_result AS
+SELECT test_plugin_first(u) OVER (
+PARTITION BY g ORDER BY id
+ROWS BETWEEN CURRENT ROW AND 1 FOLLOWING) AS value
+FROM t_window;
+SHOW CREATE TABLE window_uuid_result;
+Table	Create Table
+window_uuid_result	CREATE TABLE `window_uuid_result` (
+  `value` uuid DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci
+SELECT test_plugin_first(DISTINCT u) OVER () FROM t_window;
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+PREPARE reprepare_stmt FROM
+'SELECT test_plugin_count(v) FROM t_group';
+EXECUTE reprepare_stmt;
+test_plugin_count(v)
+3
+ALTER TABLE t_group ADD COLUMN extra INT;
+EXECUTE reprepare_stmt;
+test_plugin_count(v)
+3
+DEALLOCATE PREPARE reprepare_stmt;
+DROP TABLE window_uuid_result, t_window, uuid_union, t_uuid, t_blob, t_group;
diff --git a/plugin/func_test/mysql-test/func_test/function_plugin_extra.test b/plugin/func_test/mysql-test/func_test/function_plugin_extra.test
new file mode 100644
index 0000000000000..ee405fecc3a98
--- /dev/null
+++ b/plugin/func_test/mysql-test/func_test/function_plugin_extra.test
@@ -0,0 +1,205 @@
+--source include/have_udf.inc
+
+--error ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT
+SELECT test_plugin_count();
+--error ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT
+SELECT test_plugin_count(1, 2);
+--error ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT
+SELECT test_plugin_first();
+
+CREATE TABLE t_group (g INT, v INT);
+INSERT INTO t_group VALUES
+  (1, NULL), (1, NULL),
+  (2, 10), (2, NULL),
+  (3, 20), (3, 30);
+
+--error ER_INVALID_GROUP_FUNC_USE
+SELECT * FROM t_group WHERE test_plugin_count(v) > 0;
+--error ER_INVALID_GROUP_FUNC_USE
+SELECT test_plugin_count(test_plugin_count(v)) FROM t_group;
+
+SELECT test_plugin_count(v) FROM t_group;
+SELECT test_plugin_count(v) FROM t_group
+HAVING test_plugin_count(v) > 0;
+SELECT g, test_plugin_count(v) AS c
+FROM t_group GROUP BY g HAVING c > 0 ORDER BY c, g;
+SELECT test_plugin_count(v), test_plugin_count(v),
+       test_plugin_count(v + 1), test_plugin_first(v)
+FROM t_group;
+SELECT g, test_plugin_count(v), test_plugin_first(v)
+FROM t_group GROUP BY g ORDER BY g;
+SELECT g, test_plugin_count(v)
+FROM t_group GROUP BY g WITH ROLLUP;
+
+SET @save_optimizer_switch= @@optimizer_switch;
+SET optimizer_switch= 'derived_merge=off';
+SELECT * FROM
+  (SELECT g, test_plugin_count(v) AS c
+   FROM t_group GROUP BY g) AS dt
+ORDER BY g;
+WITH cte AS
+  (SELECT g, test_plugin_count(v) AS c
+   FROM t_group GROUP BY g)
+SELECT * FROM cte ORDER BY g;
+SET optimizer_switch= @save_optimizer_switch;
+
+SELECT g, test_plugin_count(DISTINCT v)
+FROM t_group GROUP BY g ORDER BY g;
+SELECT test_plugin_count(DISTINCT 1);
+SELECT test_plugin_count(DISTINCT 1.5);
+SELECT test_plugin_count(DISTINCT 1.5e0);
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT 2.5);
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT 2.5e0);
+SELECT test_plugin_count(DISTINCT v) FROM t_group WHERE v IS NULL;
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT v) IS NULL FROM t_group WHERE v IS NULL;
+SELECT test_plugin_count(DISTINCT v) FROM t_group WHERE FALSE;
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT v) IS NULL FROM t_group WHERE FALSE;
+
+CREATE TABLE t_distinct (v INT);
+INSERT INTO t_distinct VALUES (1),(1),(2),(2),(2),(3),(3),(NULL),(4);
+SELECT COUNT(DISTINCT v), test_plugin_count(DISTINCT v) FROM t_distinct;
+CREATE VIEW v_distinct AS
+  SELECT COUNT(DISTINCT v) AS b,
+         test_plugin_count(DISTINCT v) AS pl
+  FROM t_distinct;
+SHOW CREATE VIEW v_distinct;
+SELECT b, pl FROM v_distinct;
+--replace_column 9 #
+EXPLAIN EXTENDED SELECT test_plugin_count(DISTINCT v) AS b FROM t_distinct;
+SHOW WARNINGS;
+DROP VIEW v_distinct;
+DROP TABLE t_distinct;
+
+CREATE TABLE t_blob (value LONGTEXT);
+INSERT INTO t_blob VALUES
+  (REPEAT('a', 10000)), (REPEAT('a', 10000)),
+  (REPEAT('b', 10000)), (NULL);
+SELECT test_plugin_count(DISTINCT value) FROM t_blob;
+--error ER_NOT_SUPPORTED_YET
+SELECT LENGTH(test_plugin_first(DISTINCT value)) FROM t_blob;
+
+CREATE TABLE t_distinct_order (i INT, vc VARCHAR(16), tx TEXT);
+INSERT INTO t_distinct_order VALUES
+  (9,'zzz','zzz'), (5,'mmm','mmm'), (9,'zzz','zzz'),
+  (1,'aaa','aaa'), (5,'mmm','mmm');
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT i) FROM t_distinct_order;
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT vc) FROM t_distinct_order;
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT tx) FROM t_distinct_order;
+DROP TABLE t_distinct_order;
+
+SET @save_tmp_table_size= @@tmp_table_size;
+SET @save_max_heap_table_size= @@max_heap_table_size;
+SET tmp_table_size= 16384;
+SET max_heap_table_size= 16384;
+CREATE TABLE t_spill AS
+  WITH RECURSIVE seq AS
+    (SELECT 1 AS value UNION ALL
+     SELECT value + 1 FROM seq WHERE value < 1000)
+  SELECT value FROM seq;
+INSERT INTO t_spill SELECT value FROM t_spill;
+SELECT test_plugin_count(DISTINCT value) FROM t_spill;
+DROP TABLE t_spill;
+SET tmp_table_size= @save_tmp_table_size;
+SET max_heap_table_size= @save_max_heap_table_size;
+
+INSTALL SONAME 'type_test';
+CREATE TABLE t_type (g INT, i test_int8, d test_double);
+INSERT INTO t_type VALUES
+  (1, 10, 1.5), (1, 20, 2.5), (2, NULL, NULL);
+SELECT g, test_plugin_first(i), test_plugin_first(d)
+FROM t_type GROUP BY g ORDER BY g;
+CREATE TABLE type_int_result AS
+  SELECT test_plugin_first(i) AS value FROM t_type;
+SHOW CREATE TABLE type_int_result;
+CREATE TABLE type_double_result AS
+  SELECT test_plugin_first(d) AS value FROM t_type;
+SHOW CREATE TABLE type_double_result;
+CREATE VIEW type_view AS
+  SELECT test_plugin_first(i) AS value FROM t_type;
+SHOW CREATE VIEW type_view;
+SELECT * FROM type_view;
+CREATE TEMPORARY TABLE type_tmp AS
+  SELECT g, test_plugin_first(i) AS value
+  FROM t_type GROUP BY g;
+SHOW CREATE TABLE type_tmp;
+SELECT * FROM type_tmp ORDER BY g;
+CREATE TABLE type_dst (value test_int8);
+INSERT INTO type_dst SELECT test_plugin_first(i) FROM t_type;
+SELECT * FROM type_dst;
+DROP VIEW type_view;
+DROP TEMPORARY TABLE type_tmp;
+DROP TABLE type_dst, type_int_result, type_double_result, t_type;
+UNINSTALL SONAME 'type_test';
+
+CREATE TABLE t_uuid (id INT, g INT, value UUID);
+INSERT INTO t_uuid VALUES
+  (1, 1, '00000000-0000-0000-0000-000000000001'),
+  (2, 1, '00000000-0000-0000-0000-000000000001'),
+  (3, 1, '00000000-0000-0000-0000-000000000002'),
+  (4, 2, NULL),
+  (5, 2, '00000000-0000-0000-0000-000000000003');
+SELECT test_plugin_count(DISTINCT value) FROM t_uuid;
+SELECT test_plugin_first(value) =
+       (SELECT value FROM t_uuid WHERE id = 1)
+FROM t_uuid WHERE g = 1;
+CREATE TABLE uuid_union AS
+  SELECT test_plugin_first(value) AS value FROM t_uuid
+  UNION ALL
+  SELECT value FROM t_uuid WHERE id = 1;
+SHOW CREATE TABLE uuid_union;
+SELECT CASE WHEN 1 THEN test_plugin_first(value)
+            ELSE CAST(NULL AS UUID) END AS value
+FROM t_uuid;
+
+CREATE TABLE t_window (id INT, g INT, v INT, u UUID);
+INSERT INTO t_window VALUES
+  (1, 1, 10, '00000000-0000-0000-0000-000000000001'),
+  (2, 1, NULL, NULL),
+  (3, 1, 30, '00000000-0000-0000-0000-000000000003'),
+  (4, 1, 40, '00000000-0000-0000-0000-000000000004'),
+  (5, 2, NULL, NULL),
+  (6, 2, 60, '00000000-0000-0000-0000-000000000006');
+SELECT id,
+  test_plugin_count(v) OVER (
+    PARTITION BY g ORDER BY id
+    ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS preceding_count,
+  test_plugin_count(v) OVER (
+    PARTITION BY g ORDER BY id
+    ROWS BETWEEN CURRENT ROW AND 2 FOLLOWING) AS following_count,
+  test_plugin_count(v) OVER (
+    PARTITION BY g ORDER BY id
+    ROWS BETWEEN 2 FOLLOWING AND 3 FOLLOWING) AS empty_count
+FROM t_window ORDER BY id;
+SELECT id,
+  test_plugin_count(v) OVER w AS c1,
+  test_plugin_count(v) OVER w AS c2,
+  test_plugin_first(u) OVER w AS first_uuid
+FROM t_window
+WINDOW w AS (
+  PARTITION BY g ORDER BY id
+  ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING)
+ORDER BY id;
+CREATE TABLE window_uuid_result AS
+SELECT test_plugin_first(u) OVER (
+  PARTITION BY g ORDER BY id
+  ROWS BETWEEN CURRENT ROW AND 1 FOLLOWING) AS value
+FROM t_window;
+SHOW CREATE TABLE window_uuid_result;
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT u) OVER () FROM t_window;
+
+PREPARE reprepare_stmt FROM
+  'SELECT test_plugin_count(v) FROM t_group';
+EXECUTE reprepare_stmt;
+ALTER TABLE t_group ADD COLUMN extra INT;
+EXECUTE reprepare_stmt;
+DEALLOCATE PREPARE reprepare_stmt;
+
+DROP TABLE window_uuid_result, t_window, uuid_union, t_uuid, t_blob, t_group;
diff --git a/plugin/func_test/mysql-test/func_test/function_plugin_filter.result b/plugin/func_test/mysql-test/func_test/function_plugin_filter.result
new file mode 100644
index 0000000000000..ba7a0b4316b49
--- /dev/null
+++ b/plugin/func_test/mysql-test/func_test/function_plugin_filter.result
@@ -0,0 +1,56 @@
+#
+# MDEV-41352 SIGSEGV in Item_field::used_tables on a plugin aggregate
+#            with FILTER
+#
+CREATE TABLE t (c INT);
+SELECT test_plugin_count(1) FILTER (WHERE c) FROM t;
+test_plugin_count(1) FILTER (WHERE c)
+0
+DROP TABLE t;
+CREATE TABLE t (id INT PRIMARY KEY, g INT, v INT);
+INSERT INTO t VALUES (1,1,10),(2,1,20),(3,2,30),(4,2,NULL),(5,3,50);
+SELECT test_plugin_count(v) FILTER (WHERE v > 15) FROM t;
+test_plugin_count(v) FILTER (WHERE v > 15)
+3
+SELECT count(v) FILTER (WHERE v > 15) AS c,
+test_plugin_count(v) FILTER (WHERE v > 15) AS pc,
+test_plugin_first(v) FILTER (WHERE v > 15) AS pf
+FROM t;
+c	pc	pf
+3	3	20
+SELECT g, count(v) FILTER (WHERE id > 1) AS c,
+test_plugin_count(v) FILTER (WHERE id > 1) AS pc,
+test_plugin_first(v) FILTER (WHERE id > 1) AS pf
+FROM t GROUP BY g ORDER BY g;
+g	c	pc	pf
+1	1	1	20
+2	1	1	30
+3	1	1	50
+SELECT count(DISTINCT g) FILTER (WHERE v > 15) AS c,
+test_plugin_count(DISTINCT g) FILTER (WHERE v > 15) AS pc
+FROM t;
+c	pc
+3	3
+SELECT id, count(v) FILTER (WHERE v > 15) OVER w AS c,
+test_plugin_count(v) FILTER (WHERE v > 15) OVER w AS pc
+FROM t
+WINDOW w AS (ORDER BY id ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING)
+ORDER BY id;
+id	c	pc
+1	1	1
+2	2	2
+3	2	2
+4	2	2
+5	1	1
+SELECT test_plugin_count(v) FILTER (WHERE count(v) > 1) FROM t;
+ERROR HY000: Incorrect usage of aggregate function and FILTER
+CREATE VIEW v1 AS
+SELECT test_plugin_count(v) FILTER (WHERE v > 15) AS pc FROM t;
+SHOW CREATE VIEW v1;
+View	Create View	character_set_client	collation_connection
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS select test_plugin_count(`t`.`v`) FILTER(WHERE `t`.`v` > 15) AS `pc` from `t`	latin1	latin1_swedish_ci
+SELECT * FROM v1;
+pc
+3
+DROP VIEW v1;
+DROP TABLE t;
diff --git a/plugin/func_test/mysql-test/func_test/function_plugin_filter.test b/plugin/func_test/mysql-test/func_test/function_plugin_filter.test
new file mode 100644
index 0000000000000..0259841c404de
--- /dev/null
+++ b/plugin/func_test/mysql-test/func_test/function_plugin_filter.test
@@ -0,0 +1,38 @@
+--source include/have_udf.inc
+
+--echo #
+--echo # MDEV-41352 SIGSEGV in Item_field::used_tables on a plugin aggregate
+--echo #            with FILTER
+--echo #
+
+CREATE TABLE t (c INT);
+SELECT test_plugin_count(1) FILTER (WHERE c) FROM t;
+DROP TABLE t;
+
+CREATE TABLE t (id INT PRIMARY KEY, g INT, v INT);
+INSERT INTO t VALUES (1,1,10),(2,1,20),(3,2,30),(4,2,NULL),(5,3,50);
+SELECT test_plugin_count(v) FILTER (WHERE v > 15) FROM t;
+SELECT count(v) FILTER (WHERE v > 15) AS c,
+       test_plugin_count(v) FILTER (WHERE v > 15) AS pc,
+       test_plugin_first(v) FILTER (WHERE v > 15) AS pf
+FROM t;
+SELECT g, count(v) FILTER (WHERE id > 1) AS c,
+       test_plugin_count(v) FILTER (WHERE id > 1) AS pc,
+       test_plugin_first(v) FILTER (WHERE id > 1) AS pf
+FROM t GROUP BY g ORDER BY g;
+SELECT count(DISTINCT g) FILTER (WHERE v > 15) AS c,
+       test_plugin_count(DISTINCT g) FILTER (WHERE v > 15) AS pc
+FROM t;
+SELECT id, count(v) FILTER (WHERE v > 15) OVER w AS c,
+       test_plugin_count(v) FILTER (WHERE v > 15) OVER w AS pc
+FROM t
+WINDOW w AS (ORDER BY id ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING)
+ORDER BY id;
+--error ER_WRONG_USAGE
+SELECT test_plugin_count(v) FILTER (WHERE count(v) > 1) FROM t;
+CREATE VIEW v1 AS
+SELECT test_plugin_count(v) FILTER (WHERE v > 15) AS pc FROM t;
+SHOW CREATE VIEW v1;
+SELECT * FROM v1;
+DROP VIEW v1;
+DROP TABLE t;
diff --git a/plugin/func_test/mysql-test/func_test/function_plugin_negative.result b/plugin/func_test/mysql-test/func_test/function_plugin_negative.result
new file mode 100644
index 0000000000000..45de99297a58d
--- /dev/null
+++ b/plugin/func_test/mysql-test/func_test/function_plugin_negative.result
@@ -0,0 +1,117 @@
+#
+# Setup
+#
+CREATE TABLE t1 (id INT PRIMARY KEY, g INT, v INT, u UUID);
+INSERT INTO t1 VALUES
+(1, 1, 10, '00000000-0000-0000-0000-000000000001'),
+(2, 1, NULL, NULL),
+(3, 2, 30, '00000000-0000-0000-0000-000000000003');
+#
+# Aggregate functions in forbidden SQL contexts
+#
+CREATE TABLE bad_check (v INT, CHECK (test_plugin_count(v) > 0));
+ERROR HY000: Function or expression 'test_plugin_count()' cannot be used in the CHECK clause of `CONSTRAINT_1`
+CREATE TABLE bad_default (v INT DEFAULT (test_plugin_count(1)));
+ERROR HY000: Function or expression 'test_plugin_count()' cannot be used in the DEFAULT clause of `v`
+CREATE TABLE bad_generated
+(v INT, c INT AS (test_plugin_count(v)));
+ERROR HY000: Function or expression 'test_plugin_count()' cannot be used in the GENERATED ALWAYS AS clause of `c`
+SELECT test_plugin_count(SUM(v)) FROM t1;
+ERROR HY000: Invalid use of group function
+SELECT SUM(test_plugin_count(v)) FROM t1;
+ERROR HY000: Invalid use of group function
+SELECT test_plugin_first(test_plugin_count(v)) FROM t1;
+ERROR HY000: Invalid use of group function
+SELECT test_plugin_count(v) FROM t1;
+test_plugin_count(v)
+2
+#
+# Invalid window usage
+#
+SELECT * FROM t1 WHERE test_plugin_count(v) OVER () > 0;
+ERROR HY000: Window function is allowed only in SELECT list and ORDER BY clause
+SELECT test_plugin_count(test_plugin_count(v) OVER ()) FROM t1;
+ERROR HY000: Window functions can not be used as arguments to group functions.
+SELECT test_plugin_count(DISTINCT v) OVER (
+ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)
+FROM t1;
+ERROR 42000: This version of MariaDB doesn't yet support 'plugin aggregate with DISTINCT as window function'
+SELECT test_plugin_first(DISTINCT u) OVER (
+PARTITION BY g ORDER BY id)
+FROM t1;
+ERROR 42000: This version of MariaDB doesn't yet support 'DISTINCT for plugin aggregate test_plugin_first()'
+SELECT test_plugin_count(v) OVER (ORDER BY id) FROM t1;
+test_plugin_count(v) OVER (ORDER BY id)
+1
+1
+2
+#
+# Invalid argument lists
+#
+SELECT test_plugin_count(DISTINCT) FROM t1;
+ERROR 42000: Incorrect parameter count in the call to native function 'test_plugin_count'
+SELECT test_plugin_count(DISTINCT v, g) FROM t1;
+ERROR 42000: Incorrect parameter count in the call to native function 'test_plugin_count'
+SELECT sysconst_test(DISTINCT 1);
+ERROR 42000: Incorrect parameter count in the call to native function 'sysconst_test'
+SELECT test_plugin_count(DISTINCT v) FROM t1;
+test_plugin_count(DISTINCT v)
+2
+#
+# Invalid pluggable data type values
+#
+SET @save_sql_mode= @@sql_mode;
+SET sql_mode= 'STRICT_ALL_TABLES';
+INSERT INTO t1 VALUES (4, 2, 40, 'not-a-uuid');
+ERROR 22007: Incorrect uuid value: 'not-a-uuid' for column `test`.`t1`.`u` at row 1
+SELECT test_plugin_first(CAST('not-a-uuid' AS UUID)) IS NULL;
+test_plugin_first(CAST('not-a-uuid' AS UUID)) IS NULL
+1
+Warnings:
+Warning	1292	Incorrect uuid value: 'not-a-uuid'
+SHOW WARNINGS;
+Level	Code	Message
+Warning	1292	Incorrect uuid value: 'not-a-uuid'
+SET sql_mode= @save_sql_mode;
+#
+# Errors while storing aggregate results
+#
+CREATE TABLE uuid_dst (value UUID PRIMARY KEY);
+INSERT INTO uuid_dst
+SELECT test_plugin_first(u) FROM t1 WHERE g = 1;
+INSERT INTO uuid_dst
+SELECT test_plugin_first(u) FROM t1 WHERE g = 1;
+ERROR 23000: Duplicate entry '00000000-0000-0000-0000-000000000001' for key 'PRIMARY'
+CREATE TABLE not_null_dst (value INT NOT NULL);
+INSERT INTO not_null_dst
+SELECT test_plugin_first(v) FROM t1 WHERE FALSE;
+ERROR 23000: Column 'value' cannot be null
+SELECT test_plugin_count(v) FROM t1;
+test_plugin_count(v)
+2
+#
+# Failed automatic reprepare
+#
+PREPARE reprepare_drop FROM
+'SELECT test_plugin_count(v) FROM t1';
+ALTER TABLE t1 DROP COLUMN v;
+EXECUTE reprepare_drop;
+ERROR 42S22: Unknown column 'v' in 'SELECT'
+DEALLOCATE PREPARE reprepare_drop;
+ALTER TABLE t1 ADD COLUMN v INT;
+#
+# Repeated result access and cleanup
+#
+SELECT test_plugin_first(u),
+test_plugin_first(u) IS NULL,
+CAST(test_plugin_first(u) AS CHAR)
+FROM t1;
+test_plugin_first(u)	test_plugin_first(u) IS NULL	CAST(test_plugin_first(u) AS CHAR)
+00000000-0000-0000-0000-000000000001	0	00000000-0000-0000-0000-000000000001
+SELECT test_plugin_count(v) AS c
+FROM t1 HAVING c > 0 ORDER BY c;
+c
+#
+# Cleanup
+#
+DROP TABLE not_null_dst, uuid_dst, t1;
diff --git a/plugin/func_test/mysql-test/func_test/function_plugin_negative.test b/plugin/func_test/mysql-test/func_test/function_plugin_negative.test
new file mode 100644
index 0000000000000..d5826999e38fd
--- /dev/null
+++ b/plugin/func_test/mysql-test/func_test/function_plugin_negative.test
@@ -0,0 +1,108 @@
+--source include/have_udf.inc
+
+--echo #
+--echo # Setup
+--echo #
+CREATE TABLE t1 (id INT PRIMARY KEY, g INT, v INT, u UUID);
+INSERT INTO t1 VALUES
+  (1, 1, 10, '00000000-0000-0000-0000-000000000001'),
+  (2, 1, NULL, NULL),
+  (3, 2, 30, '00000000-0000-0000-0000-000000000003');
+
+--echo #
+--echo # Aggregate functions in forbidden SQL contexts
+--echo #
+--error ER_GENERATED_COLUMN_FUNCTION_IS_NOT_ALLOWED
+CREATE TABLE bad_check (v INT, CHECK (test_plugin_count(v) > 0));
+--error ER_GENERATED_COLUMN_FUNCTION_IS_NOT_ALLOWED
+CREATE TABLE bad_default (v INT DEFAULT (test_plugin_count(1)));
+--error ER_GENERATED_COLUMN_FUNCTION_IS_NOT_ALLOWED
+CREATE TABLE bad_generated
+  (v INT, c INT AS (test_plugin_count(v)));
+--error ER_INVALID_GROUP_FUNC_USE
+SELECT test_plugin_count(SUM(v)) FROM t1;
+--error ER_INVALID_GROUP_FUNC_USE
+SELECT SUM(test_plugin_count(v)) FROM t1;
+--error ER_INVALID_GROUP_FUNC_USE
+SELECT test_plugin_first(test_plugin_count(v)) FROM t1;
+SELECT test_plugin_count(v) FROM t1;
+
+--echo #
+--echo # Invalid window usage
+--echo #
+--error ER_WRONG_PLACEMENT_OF_WINDOW_FUNCTION
+SELECT * FROM t1 WHERE test_plugin_count(v) OVER () > 0;
+--error ER_SUM_FUNC_WITH_WINDOW_FUNC_AS_ARG
+SELECT test_plugin_count(test_plugin_count(v) OVER ()) FROM t1;
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_count(DISTINCT v) OVER (
+  ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)
+FROM t1;
+--error ER_NOT_SUPPORTED_YET
+SELECT test_plugin_first(DISTINCT u) OVER (
+  PARTITION BY g ORDER BY id)
+FROM t1;
+SELECT test_plugin_count(v) OVER (ORDER BY id) FROM t1;
+
+--echo #
+--echo # Invalid argument lists
+--echo #
+--error ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT
+SELECT test_plugin_count(DISTINCT) FROM t1;
+--error ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT
+SELECT test_plugin_count(DISTINCT v, g) FROM t1;
+--error ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT
+SELECT sysconst_test(DISTINCT 1);
+SELECT test_plugin_count(DISTINCT v) FROM t1;
+
+--echo #
+--echo # Invalid pluggable data type values
+--echo #
+SET @save_sql_mode= @@sql_mode;
+SET sql_mode= 'STRICT_ALL_TABLES';
+--error ER_TRUNCATED_WRONG_VALUE
+INSERT INTO t1 VALUES (4, 2, 40, 'not-a-uuid');
+SELECT test_plugin_first(CAST('not-a-uuid' AS UUID)) IS NULL;
+SHOW WARNINGS;
+SET sql_mode= @save_sql_mode;
+
+--echo #
+--echo # Errors while storing aggregate results
+--echo #
+CREATE TABLE uuid_dst (value UUID PRIMARY KEY);
+INSERT INTO uuid_dst
+  SELECT test_plugin_first(u) FROM t1 WHERE g = 1;
+--error ER_DUP_ENTRY
+INSERT INTO uuid_dst
+  SELECT test_plugin_first(u) FROM t1 WHERE g = 1;
+CREATE TABLE not_null_dst (value INT NOT NULL);
+--error ER_BAD_NULL_ERROR
+INSERT INTO not_null_dst
+  SELECT test_plugin_first(v) FROM t1 WHERE FALSE;
+SELECT test_plugin_count(v) FROM t1;
+
+--echo #
+--echo # Failed automatic reprepare
+--echo #
+PREPARE reprepare_drop FROM
+  'SELECT test_plugin_count(v) FROM t1';
+ALTER TABLE t1 DROP COLUMN v;
+--error ER_BAD_FIELD_ERROR
+EXECUTE reprepare_drop;
+DEALLOCATE PREPARE reprepare_drop;
+ALTER TABLE t1 ADD COLUMN v INT;
+
+--echo #
+--echo # Repeated result access and cleanup
+--echo #
+SELECT test_plugin_first(u),
+       test_plugin_first(u) IS NULL,
+       CAST(test_plugin_first(u) AS CHAR)
+FROM t1;
+SELECT test_plugin_count(v) AS c
+FROM t1 HAVING c > 0 ORDER BY c;
+
+--echo #
+--echo # Cleanup
+--echo #
+DROP TABLE not_null_dst, uuid_dst, t1;
diff --git a/plugin/func_test/plugin.cc b/plugin/func_test/plugin.cc
index 7c93b47e8e59a..4d64b9d53cebc 100644
--- a/plugin/func_test/plugin.cc
+++ b/plugin/func_test/plugin.cc
@@ -18,6 +18,7 @@
 
 #include <my_global.h>
 #include <sql_class.h>
+#include <item_sum.h>
 #include <mysql/plugin_function.h>
 
 class Item_func_sysconst_test :public Item_func_sysconst
@@ -275,6 +276,297 @@ class Item_func_strnxfrm: public Item_str_func
   }
 };
 
+
+class Item_sum_test_plugin_first: public Item_sum_plugin,
+                                  public Type_handler_hybrid_field_type
+{
+  using Self= Item_sum_test_plugin_first;
+  Item_cache *value;
+
+  void endup()
+  {
+    if (aggr)
+      aggr->endup();
+  }
+
+  bool setup_cache(THD *thd, Item *source)
+  {
+    if (!(value= args[0]->get_cache(thd)) || value->setup(thd, args[0]))
+      return true;
+    if (source)
+    {
+      value->store(source);
+      value->cache_value();
+      null_value= value->null_value;
+    }
+    return false;
+  }
+
+public:
+  Item_sum_test_plugin_first(THD *thd, Item *item):
+    Item_sum_plugin(thd, item),
+    Type_handler_hybrid_field_type(&type_handler_string), value(nullptr)
+  {}
+  Item_sum_test_plugin_first(THD *thd, Self *item):
+    Item_sum_plugin(thd, item), Type_handler_hybrid_field_type(item),
+    value(nullptr)
+  {}
+  Item_sum_test_plugin_first(const Self &item):
+    Item_sum_plugin(item), Type_handler_hybrid_field_type(item), value(nullptr)
+  {}
+  void clear() override
+  {
+    value->clear();
+    value->set_null();
+    null_value= true;
+  }
+  bool add_row() override
+  {
+    if (!null_value)
+      return false;
+    value->store(aggregation_arg(0));
+    value->cache_value();
+    null_value= value->null_value;
+    return false;
+  }
+  bool supports_distinct() const override { return false; }
+  double val_real() override
+  {
+    endup();
+    if (null_value)
+      return 0.0;
+    double result= value->val_real();
+    null_value= value->null_value;
+    return result;
+  }
+  longlong val_int() override
+  {
+    endup();
+    if (null_value)
+      return 0;
+    longlong result= value->val_int();
+    null_value= value->null_value;
+    return result;
+  }
+  String *val_str(String *str) override
+  {
+    endup();
+    if (null_value)
+      return nullptr;
+    String *result= value->val_str(str);
+    null_value= value->null_value;
+    return result;
+  }
+  my_decimal *val_decimal(my_decimal *to) override
+  {
+    endup();
+    if (null_value)
+      return nullptr;
+    my_decimal *result= value->val_decimal(to);
+    null_value= value->null_value;
+    return result;
+  }
+  bool get_date(THD *thd, MYSQL_TIME *ltime,
+                date_mode_t fuzzydate) override
+  {
+    endup();
+    if (null_value)
+      return true;
+    bool result= value->get_date(thd, ltime, fuzzydate);
+    null_value= value->null_value;
+    return result;
+  }
+  bool val_native(THD *thd, Native *to) override
+  {
+    endup();
+    if (null_value)
+      return true;
+    return val_native_from_item(thd, value, to);
+  }
+  bool is_null() override
+  {
+    endup();
+    return null_value;
+  }
+  const Type_handler *type_handler() const override
+  {
+    return Type_handler_hybrid_field_type::type_handler();
+  }
+  const Type_handler *real_type_handler() const override
+  {
+    return args[0]->real_type_handler();
+  }
+  const Type_extra_attributes type_extra_attributes() const override
+  {
+    return args[0]->type_extra_attributes();
+  }
+  bool fix_length_and_dec(THD *thd) override
+  {
+    Type_std_attributes::set(args[0]);
+    set_handler(args[0]->type_handler());
+    set_maybe_null();
+    null_value= true;
+    return setup_cache(thd, nullptr);
+  }
+  void cleanup() override
+  {
+    value= nullptr;
+    null_value= true;
+    Item_sum_plugin::cleanup();
+  }
+  LEX_CSTRING func_name_cstring() const override
+  {
+    return "test_plugin_first("_LEX_CSTRING;
+  }
+  Item *copy_or_same(THD *thd) override
+  {
+    Self *item= new (thd->mem_root) Self(thd, this);
+    if (!item || item->setup_cache(thd, value))
+      return nullptr;
+    return item;
+  }
+
+  class Create_func: public Create_native_func
+  {
+  public:
+    bool is_aggregate() const override { return true; }
+    Item *create_native(THD *thd, const LEX_CSTRING *name,
+                        List<Item> *item_list) override
+    {
+      uint arg_count= item_list ? item_list->elements : 0;
+      if (arg_count != 1)
+      {
+        my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name->str);
+        return nullptr;
+      }
+      return new (thd->mem_root) Self(thd, item_list->head());
+    }
+  };
+
+  static Plugin_function *plugin_descriptor()
+  {
+    static Create_func creator;
+    static Plugin_function descriptor(&creator);
+    return &descriptor;
+  }
+protected:
+  Item *shallow_copy(THD *thd) const override
+  {
+    Self *item= new (thd->mem_root) Self(thd, const_cast<Self *>(this));
+    if (!item || item->setup_cache(thd, value))
+      return nullptr;
+    return item;
+  }
+};
+
+
+class Item_sum_test_plugin_count: public Item_sum_plugin
+{
+  using Self= Item_sum_test_plugin_count;
+  longlong count;
+public:
+  Item_sum_test_plugin_count(THD *thd, Item *item):
+    Item_sum_plugin(thd, item), count(0)
+  {}
+  Item_sum_test_plugin_count(THD *thd, Self *item):
+    Item_sum_plugin(thd, item), count(0)
+  {}
+  Item_sum_test_plugin_count(const Self &item):
+    Item_sum_plugin(item), count(0)
+  {}
+  void clear() override
+  {
+    count= 0;
+    null_value= false;
+  }
+  bool add_row() override
+  {
+    if (!aggr->arg_is_null(false))
+      count++;
+    return false;
+  }
+  bool supports_removal() const override { return true; }
+  void remove_row() override
+  {
+    if (aggr->arg_is_null(false))
+      return;
+    if (count > 0)
+      count--;
+  }
+  longlong val_int() override
+  {
+    DBUG_ASSERT(fixed());
+    if (aggr)
+      aggr->endup();
+    null_value= false;
+    return count;
+  }
+  double val_real() override { return static_cast<double>(val_int()); }
+  String *val_str(String *str) override { return val_string_from_int(str); }
+  my_decimal *val_decimal(my_decimal *to) override
+  {
+    return val_decimal_from_int(to);
+  }
+  bool get_date(THD *thd, MYSQL_TIME *ltime,
+                date_mode_t fuzzydate) override
+  {
+    return get_date_from_int(thd, ltime, fuzzydate);
+  }
+  const Type_handler *type_handler() const override
+  {
+    return &type_handler_slonglong;
+  }
+  bool fix_length_and_dec(THD *thd) override
+  {
+    decimals= 0;
+    max_length= 21;
+    null_value= false;
+    return false;
+  }
+  void cleanup() override
+  {
+    count= 0;
+    Item_sum_plugin::cleanup();
+  }
+  LEX_CSTRING func_name_cstring() const override
+  {
+    return "test_plugin_count("_LEX_CSTRING;
+  }
+  Item *copy_or_same(THD *thd) override
+  {
+    return new (thd->mem_root) Self(thd, this);
+  }
+
+  class Create_func: public Create_native_func
+  {
+  public:
+    bool is_aggregate() const override { return true; }
+    Item *create_native(THD *thd, const LEX_CSTRING *name,
+                        List<Item> *item_list) override
+    {
+      uint arg_count= item_list ? item_list->elements : 0;
+      if (arg_count != 1)
+      {
+        my_error(ER_WRONG_PARAMCOUNT_TO_NATIVE_FCT, MYF(0), name->str);
+        return nullptr;
+      }
+      return new (thd->mem_root) Self(thd, item_list->head());
+    }
+  };
+
+  static Plugin_function *plugin_descriptor()
+  {
+    static Create_func creator;
+    static Plugin_function descriptor(&creator);
+    return &descriptor;
+  }
+protected:
+  Item *shallow_copy(THD *thd) const override
+  {
+    return get_item_copy<Self>(thd, this);
+  }
+};
+
 /*************************************************************************/
 
 maria_declare_plugin(type_test)
@@ -337,5 +629,35 @@ maria_declare_plugin(type_test)
   NULL,                         // System variables
   "1.0",                        // String version representation
   MariaDB_PLUGIN_MATURITY_EXPERIMENTAL // Maturity
+},
+{
+  MariaDB_FUNCTION_PLUGIN,       // the plugin type
+  Item_sum_test_plugin_first::plugin_descriptor(),
+  "test_plugin_first",          // plugin name
+  "MariaDB Corporation",        // plugin author
+  "Function TEST_PLUGIN_FIRST()", // the plugin description
+  PLUGIN_LICENSE_GPL,            // the plugin license
+  0,                             // Pointer to plugin initialization function
+  0,                             // Pointer to plugin deinitialization function
+  0x0100,                        // Numeric version 0xAABB means AA.BB version
+  NULL,                          // Status variables
+  NULL,                          // System variables
+  "1.0",                        // String version representation
+  MariaDB_PLUGIN_MATURITY_EXPERIMENTAL // Maturity
+},
+{
+  MariaDB_FUNCTION_PLUGIN,       // the plugin type
+  Item_sum_test_plugin_count::plugin_descriptor(),
+  "test_plugin_count",          // plugin name
+  "MariaDB Corporation",        // plugin author
+  "Function TEST_PLUGIN_COUNT()", // the plugin description
+  PLUGIN_LICENSE_GPL,            // the plugin license
+  0,                             // Pointer to plugin initialization function
+  0,                             // Pointer to plugin deinitialization function
+  0x0100,                        // Numeric version 0xAABB means AA.BB version
+  NULL,                          // Status variables
+  NULL,                          // System variables
+  "1.0",                        // String version representation
+  MariaDB_PLUGIN_MATURITY_EXPERIMENTAL // Maturity
 }
 maria_declare_plugin_end;
diff --git a/sql/item_create.h b/sql/item_create.h
index 019acde5b41ef..be6378e4f5811 100644
--- a/sql/item_create.h
+++ b/sql/item_create.h
@@ -61,6 +61,12 @@ class Create_func
   virtual Item *create_func(THD *thd, const LEX_CSTRING *name,
                             List<Item> *item_list) = 0;
 
+  /**
+    Whether the builder creates an aggregate (set) function.
+    Aggregate builders must create an Item_sum descendant.
+  */
+  virtual bool is_aggregate() const { return false; }
+
 protected:
   /** Constructor */
   Create_func() = default;
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index 52bffa1a348ab..f3ac356d2a852 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -505,6 +505,47 @@ Item_sum::Item_sum(THD *thd, Item_sum *item):
 }
 
 
+bool Item_sum_plugin::fix_fields(THD *thd, Item **ref)
+{
+  DBUG_ASSERT(fixed() == 0);
+
+  if (has_with_distinct() && !supports_distinct())
+  {
+    const LEX_CSTRING name= func_name_cstring();
+    const size_t name_length= name.length && name.str[name.length - 1] == '(' ?
+                              name.length - 1 : name.length;
+    char feature[NAME_LEN + 64];
+    my_snprintf(feature, sizeof(feature),
+                "DISTINCT for plugin aggregate %.*s()",
+                (int) name_length, name.str);
+    my_error(ER_NOT_SUPPORTED_YET, MYF(0), feature);
+    return true;
+  }
+
+  if (init_sum_func_check(thd))
+    return true;
+
+  for (uint i= 0; i < arg_count; i++)
+  {
+    if (args[i]->fix_fields_if_needed_for_scalar(thd, &args[i]))
+      return true;
+    with_flags|= args[i]->with_flags & ~item_with_t::FIELD;
+  }
+
+  if (fix_filter(thd))
+    return true;
+
+  result_field= NULL;
+  if (fix_length_and_dec(thd) || check_sum_func(thd, ref))
+    return true;
+
+  if (arg_count)
+    memcpy(orig_args, args, sizeof(Item *) * arg_count);
+  base_flags|= item_base_t::FIXED;
+  return false;
+}
+
+
 void Item_sum::mark_as_sum_func()
 {
   SELECT_LEX *cur_select= current_thd->lex->current_select;
@@ -521,6 +562,8 @@ void Item_sum::print(String *str, enum_query_type query_type)
   /* orig_args is not filled with valid values until fix_fields() */
   Item **pargs= fixed() ? orig_args : args;
   str->append(func_name_cstring());
+  if (sum_func() == PLUGIN_SUM_FUNC && has_with_distinct())
+    str->append(STRING_WITH_LEN("distinct "));
   /*
     TODO:
     The fact that func_name() may return a name with an extra '('
@@ -818,8 +861,9 @@ bool Aggregator_distinct::setup(THD *thd)
 
   if (item_sum->setup(thd))
     return TRUE;
-  if (item_sum->sum_func() == Item_sum::COUNT_FUNC || 
-      item_sum->sum_func() == Item_sum::COUNT_DISTINCT_FUNC)
+  if (item_sum->sum_func() == Item_sum::COUNT_FUNC ||
+      item_sum->sum_func() == Item_sum::COUNT_DISTINCT_FUNC ||
+      item_sum->sum_func() == Item_sum::PLUGIN_SUM_FUNC)
   {
     List<Item> list;
     SELECT_LEX *select_lex= thd->lex->current_select;
@@ -855,6 +899,24 @@ bool Aggregator_distinct::setup(THD *thd)
       return TRUE;
     table->file->extra(HA_EXTRA_NO_ROWS);		// Don't update rows
     table->no_rows=1;
+    if (item_sum->sum_func() == Item_sum::PLUGIN_SUM_FUNC)
+    {
+      if (!(distinct_args= thd->alloc<Item *>(item_sum->get_arg_count())))
+        return TRUE;
+      Field **field= table->field;
+      for (uint i= 0; i < item_sum->get_arg_count(); i++)
+      {
+        Item *arg= item_sum->get_arg(i);
+        if (arg->const_item())
+          distinct_args[i]= arg;
+        else
+        {
+          if (!*field || !(distinct_args[i]= new (thd->mem_root)
+                                              Item_field(thd, *field++)))
+            return TRUE;
+        }
+      }
+    }
 
     if (table->s->db_type() == heap_hton && !table->s->blob_fields)
     {
@@ -1000,8 +1062,9 @@ void Aggregator_distinct::clear()
   if (tree)
     tree->reset();
   /* tree and table can be both null only if always_null */
-  if (item_sum->sum_func() == Item_sum::COUNT_FUNC || 
-      item_sum->sum_func() == Item_sum::COUNT_DISTINCT_FUNC)
+  if (item_sum->sum_func() == Item_sum::COUNT_FUNC ||
+      item_sum->sum_func() == Item_sum::COUNT_DISTINCT_FUNC ||
+      item_sum->sum_func() == Item_sum::PLUGIN_SUM_FUNC)
   {
     if (!tree && table)
     {
@@ -1038,8 +1101,9 @@ bool Aggregator_distinct::add()
   if (always_null)
     return 0;
 
-  if (item_sum->sum_func() == Item_sum::COUNT_FUNC || 
-      item_sum->sum_func() == Item_sum::COUNT_DISTINCT_FUNC)
+  if (item_sum->sum_func() == Item_sum::COUNT_FUNC ||
+      item_sum->sum_func() == Item_sum::COUNT_DISTINCT_FUNC ||
+      item_sum->sum_func() == Item_sum::PLUGIN_SUM_FUNC)
   {
     int error;
     copy_fields(tmp_table_param);
@@ -1150,6 +1214,27 @@ void Aggregator_distinct::endup()
     }
   }
 
+  if (!tree && table &&
+      item_sum->sum_func() == Item_sum::PLUGIN_SUM_FUNC && !endup_done)
+  {
+    int error;
+    bool add_error= false;
+    use_distinct_values= true;
+    if (!(error= table->file->ha_rnd_init_with_error(true)))
+    {
+      while (!(error= table->file->ha_rnd_next(table->record[0])))
+      {
+        if ((add_error= item_sum->add()))
+          break;
+      }
+      if (!add_error && error != HA_ERR_END_OF_FILE)
+        table->file->print_error(error, MYF(0));
+      table->file->ha_rnd_end();
+    }
+    use_distinct_values= false;
+    endup_done= true;
+  }
+
  /*
    We don't have a tree only if 'setup()' hasn't been called;
    this is the case of sql_executor.cc:return_zero_rows.
@@ -1161,7 +1246,8 @@ void Aggregator_distinct::endup()
      Note that value of field is changed as we walk the tree, in
      Aggregator_distinct::unique_walk_function, but it's always not NULL.
    */
-   table->field[0]->set_notnull();
+    if (table->field[0])
+      table->field[0]->set_notnull();
     /* go over the tree of distinct keys and calculate the aggregate value */
     use_distinct_values= TRUE;
     tree_walk_action func;
@@ -1902,7 +1988,8 @@ void Item_sum_sum::remove()
   
 bool Aggregator_distinct::unique_walk_function(void *element)
 {
-  memcpy(table->field[0]->ptr, element, tree_key_length);
+  if (tree_key_length)
+    memcpy(table->field[0]->ptr, element, tree_key_length);
   item_sum->add();
   return 0;
 }
@@ -1982,15 +2069,19 @@ bool Aggregator_simple::arg_is_null(bool use_null_value)
 
 my_decimal *Aggregator_distinct::arg_val_decimal(my_decimal * value)
 {
-  return use_distinct_values ? table->field[0]->val_decimal(value) :
-    item_sum->args[0]->val_decimal(value);
+  if (!use_distinct_values)
+    return item_sum->args[0]->val_decimal(value);
+  return item_sum->sum_func() == Item_sum::PLUGIN_SUM_FUNC ?
+    distinct_args[0]->val_decimal(value) : table->field[0]->val_decimal(value);
 }
 
 
 double Aggregator_distinct::arg_val_real()
 {
-  return use_distinct_values ? table->field[0]->val_real() :
-    item_sum->args[0]->val_real();
+  if (!use_distinct_values)
+    return item_sum->args[0]->val_real();
+  return item_sum->sum_func() == Item_sum::PLUGIN_SUM_FUNC ?
+    distinct_args[0]->val_real() : table->field[0]->val_real();
 }
 
 
@@ -1998,7 +2089,8 @@ bool Aggregator_distinct::arg_is_null(bool use_null_value)
 {
   if (use_distinct_values)
   {
-    const bool rc= table->field[0]->is_null();
+    const bool rc= item_sum->sum_func() == Item_sum::PLUGIN_SUM_FUNC ?
+                   distinct_args[0]->is_null() : table->field[0]->is_null();
     DBUG_ASSERT(!rc); // NULLs are never stored in 'tree'
     return rc;
   }
@@ -2008,6 +2100,13 @@ bool Aggregator_distinct::arg_is_null(bool use_null_value)
 }
 
 
+Item *Aggregator_distinct::arg_item(uint i)
+{
+  DBUG_ASSERT(i < item_sum->get_arg_count());
+  return use_distinct_values ? distinct_args[i] : item_sum->get_arg(i);
+}
+
+
 Item *Item_sum_count::copy_or_same(THD* thd)
 {
   DBUG_ENTER("Item_sum_count::copy_or_same");
diff --git a/sql/item_sum.h b/sql/item_sum.h
index b9c6e4d89de79..41be63a6bb857 100644
--- a/sql/item_sum.h
+++ b/sql/item_sum.h
@@ -110,6 +110,7 @@ class Aggregator : public Sql_alloc
     was already applied during the collection phase.
   */
   virtual bool is_in_endup_phase() const { return false; }
+  virtual Item *arg_item(uint i) = 0;
 };
 
 
@@ -383,7 +384,8 @@ class Item_sum :public Item_func_or_sum
   enum Sumfunctype
   { COUNT_FUNC, COUNT_DISTINCT_FUNC, SUM_FUNC, SUM_DISTINCT_FUNC, AVG_FUNC,
     AVG_DISTINCT_FUNC, MIN_FUNC, MAX_FUNC, STD_FUNC,
-    VARIANCE_FUNC, SUM_BIT_FUNC, UDF_SUM_FUNC, GROUP_CONCAT_FUNC,
+    VARIANCE_FUNC, SUM_BIT_FUNC, UDF_SUM_FUNC, PLUGIN_SUM_FUNC,
+    GROUP_CONCAT_FUNC,
     ROW_NUMBER_FUNC, RANK_FUNC, DENSE_RANK_FUNC, PERCENT_RANK_FUNC,
     CUME_DIST_FUNC, NTILE_FUNC, FIRST_VALUE_FUNC, LAST_VALUE_FUNC,
     NTH_VALUE_FUNC, LEAD_FUNC, LAG_FUNC, PERCENTILE_CONT_FUNC,
@@ -466,6 +468,7 @@ class Item_sum :public Item_func_or_sum
     case VARIANCE_FUNC:
     case SUM_BIT_FUNC:
     case UDF_SUM_FUNC:
+    case PLUGIN_SUM_FUNC:
     case GROUP_CONCAT_FUNC:
     case JSON_ARRAYAGG_FUNC:
     case GEOMETRY_COLLECT_FUNC:
@@ -647,6 +650,45 @@ class Item_sum :public Item_func_or_sum
 };
 
 
+/**
+  Base class for native aggregate function plugins.
+
+  Derived classes implement clear(), add_row(), result accessors and copy
+  methods. aggregation_arg() must be used instead of args[] while consuming
+  rows, so DISTINCT replay can substitute values from its internal temporary
+  storage. DISTINCT replay order is unspecified; order-sensitive
+  implementations must override supports_distinct(). Implement
+  supports_removal() and remove_row() when the state is invertible.
+  FILTER (WHERE ...) is applied by add() and remove(): add_row() and
+  remove_row() are called only for rows that passed the filter.
+*/
+class Item_sum_plugin : public Item_sum
+{
+public:
+  Item_sum_plugin(THD *thd, Item *item): Item_sum(thd, item)
+  { quick_group= false; }
+  Item_sum_plugin(THD *thd, Item_sum_plugin *item): Item_sum(thd, item)
+  { quick_group= false; }
+  enum Sumfunctype sum_func() const override { return PLUGIN_SUM_FUNC; }
+  Field *create_field_for_create_select(MEM_ROOT *root, TABLE *table) override
+  { return create_table_field_from_handler(root, table); }
+  bool fix_fields(THD *thd, Item **ref) override;
+  Item *aggregation_arg(uint i) { return aggr->arg_item(i); }
+  virtual bool supports_distinct() const { return true; }
+  bool add() override final { return filter_passed() ? add_row() : false; }
+  void remove() override final
+  {
+    if (filter_passed())
+      remove_row();
+  }
+  void reset_field() override { DBUG_ASSERT(0); }
+  void update_field() override { DBUG_ASSERT(0); }
+protected:
+  virtual bool add_row()= 0;
+  virtual void remove_row() { DBUG_ASSERT(0); }
+};
+
+
 class Unique;
 
 
@@ -732,11 +774,12 @@ class Aggregator_distinct : public Aggregator
     instead of calling the relevant val_..() method.
   */
   bool use_distinct_values;
+  Item **distinct_args;
 
 public:
   Aggregator_distinct (Item_sum *sum) :
     Aggregator(sum), table(NULL), tmp_table_param(NULL), tree(NULL),
-    always_null(false), use_distinct_values(false) {}
+    always_null(false), use_distinct_values(false), distinct_args(NULL) {}
   virtual ~Aggregator_distinct ();
   Aggregator_type Aggrtype() override { return DISTINCT_AGGREGATOR; }
 
@@ -748,6 +791,7 @@ class Aggregator_distinct : public Aggregator
   double arg_val_real() override;
   bool arg_is_null(bool use_null_value) override;
   bool is_in_endup_phase() const override { return use_distinct_values; }
+  Item *arg_item(uint i) override;
 
   bool unique_walk_function(void *element);
   bool unique_walk_function_for_count(void *element);
@@ -775,6 +819,7 @@ class Aggregator_simple : public Aggregator
   my_decimal *arg_val_decimal(my_decimal * value) override;
   double arg_val_real() override;
   bool arg_is_null(bool use_null_value) override;
+  Item *arg_item(uint i) override { return item_sum->get_arg(i); }
 };
 
 
diff --git a/sql/sql_type.h b/sql/sql_type.h
index 152ee4622b60b..99f7426c351f0 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -8075,7 +8075,7 @@ extern Named_type_handler<Type_handler_bit>         type_handler_bit;
 extern Named_type_handler<Type_handler_enum>        type_handler_enum;
 extern Named_type_handler<Type_handler_set>         type_handler_set;
 
-extern Named_type_handler<Type_handler_string>      type_handler_string;
+extern MYSQL_PLUGIN_IMPORT Named_type_handler<Type_handler_string>      type_handler_string;
 extern Named_type_handler<Type_handler_var_string>  type_handler_var_string;
 extern MYSQL_PLUGIN_IMPORT Named_type_handler<Type_handler_varchar>     type_handler_varchar;
 extern Named_type_handler<Type_handler_varchar_compressed> type_handler_varchar_compressed;
diff --git a/sql/sql_window.cc b/sql/sql_window.cc
index 09c1e2faaea74..a1728fedc73fa 100644
--- a/sql/sql_window.cc
+++ b/sql/sql_window.cc
@@ -3239,6 +3239,12 @@ bool Window_func_runner::add_function_to_run(Item_window_func *win_func)
   sum_func->setup_window_func(current_thd, win_func->window_spec);
 
   Item_sum::Sumfunctype type= win_func->window_func()->sum_func();
+  if (type == Item_sum::PLUGIN_SUM_FUNC && sum_func->has_with_distinct())
+  {
+    my_error(ER_NOT_SUPPORTED_YET, MYF(0),
+             "plugin aggregate with DISTINCT as window function");
+    return true;
+  }
 
   switch (type)
   {
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index e11a018cba3d0..12c555ed946c5 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -355,6 +355,12 @@ void _CONCAT_UNDERSCORED(turn_parser_debug_on,yyparse)()
   class Window_frame *window_frame;
   class Window_frame_bound *window_frame_bound;
   udf_func *udf;
+  struct
+  {
+    Create_func *builder;
+    udf_func *udf;
+    bool aggregate;
+  } generic_function;
   st_trg_execution_order trg_execution_order;
 
   /* enums */
@@ -11375,40 +11381,49 @@ function_call_qualified_filterable:
 function_call_generic:
           ident_cli_func '('
           {
+            const Lex_ident_sys sysname(thd, &$1);
+            udf_func *udf= NULL;
+            bool aggregate= false;
+            if (sysname.is_null())
+              MYSQL_YYABORT; // EOM
+            Create_func *native_builder= Schema::find_implied(thd)->
+              find_native_function_builder(thd, sysname);
+            aggregate= native_builder && native_builder->is_aggregate();
 #ifdef HAVE_DLOPEN
-            udf_func *udf= 0;
-            LEX *lex= Lex;
-            if (using_udf_functions)
+            if (!native_builder && using_udf_functions)
             {
               // find_udf expectes a 0-terminated string
-              const Lex_ident_sys sysname(thd, &$1);
-              if (sysname.is_null())
-                MYSQL_YYABORT; // EOM
-              if ((udf= find_udf(sysname.str, sysname.length)) &&
-                  udf->type == UDFTYPE_AGGREGATE)
-              {
-                if (unlikely(lex->current_select->inc_in_sum_expr()))
-                {
-                  thd->parse_error();
-                  MYSQL_YYABORT;
-                }
-              }
+              udf= find_udf(sysname.str, sysname.length);
+              if (udf && udf->type == UDFTYPE_AGGREGATE)
+                aggregate= true;
             }
-            /* Temporary placing the result of find_udf in $3 */
-            $<udf>$= udf;
 #endif
+            if (aggregate &&
+                unlikely(Lex->current_select->inc_in_sum_expr()))
+            {
+              thd->parse_error();
+              MYSQL_YYABORT;
+            }
+            $<generic_function>$.builder= native_builder;
+            $<generic_function>$.udf= udf;
+            $<generic_function>$.aggregate= aggregate;
           }
 
-          opt_udf_expr_list_or_join_operator ')' opt_object_member_access opt_filter_expr
+          opt_distinct opt_udf_expr_list_or_join_operator ')'
+          opt_object_member_access opt_filter_expr
           {
             const Type_handler *h;
-            Create_func *builder;
+            Create_func *builder= $<generic_function>3.builder;
             Item *item= NULL;
             const sp_type_def *tdef= NULL;
             const Lex_ident_sys ident(thd, &$1);
             const Sp_rcontext_handler *rh;
             sp_variable *spv= NULL;
             bool allow_field_accessor= false;
+            bool native_item= false;
+
+            if ($<generic_function>3.aggregate)
+              Select->in_sum_expr--;
 
             if (unlikely(ident.is_null() ||
                          Lex_ident_routine::check_name_with_error(ident)))
@@ -11424,10 +11439,10 @@ function_call_generic:
 
               This will be revised with WL#2128 (SQL PATH)
             */
-            if ($4 && $4->elements == 1 &&
-                dynamic_cast<Item_join_operator_plus*>($4->head()))
+            if ($5 && $5->elements == 1 &&
+                dynamic_cast<Item_join_operator_plus*>($5->head()))
             {
-              if ($6.str)
+              if ($7.str)
               {
                 thd->parse_error();
                 MYSQL_YYABORT;
@@ -11435,29 +11450,29 @@ function_call_generic:
               item= Lex->create_item_ident(thd, &$1);
               if (!item || Lex->mark_item_ident_for_ora_join(thd, item))
                 MYSQL_YYABORT;
-            } else if ((builder= Schema::find_implied(thd)->
-                        find_native_function_builder(thd, ident)))
+            } else if (builder)
             {
-              item= builder->create_func(thd, &ident, $4);
+              item= builder->create_func(thd, &ident, $5);
+              native_item= true;
             }
             else if ((h= Type_handler::handler_by_name(thd, ident)) &&
-                     (item= h->make_constructor_item(thd, $4)))
+                     (item= h->make_constructor_item(thd, $5)))
             {
               // Found a constructor with a proper argument count
             }
             else if (Lex->spcont && (tdef= Lex->find_type_def(ident)))
             {
-              item= tdef->make_constructor_item(thd, $4);
+              item= tdef->make_constructor_item(thd, $5);
             }
             else if (Lex->spcont &&
                     (spv= Lex->find_variable(&ident, &rh)) &&
                     spv->type_handler()->has_functors())
             {
-              const char *end= $6.str ? $6.end() : $5.end();
+              const char *end= $7.str ? $7.end() : $6.end();
               const Lex_ident_cli name_cli($1.pos(), end - $1.pos());
-              auto ident2= $6.str ? Lex_ident_sys(thd, &$6) : Lex_ident_sys();
-              if (($6.str && ident2.is_null()) ||
-                  !(item= Lex->create_item_functor(thd, ident, $4,
+              auto ident2= $7.str ? Lex_ident_sys(thd, &$7) : Lex_ident_sys();
+              if (($7.str && ident2.is_null()) ||
+                  !(item= Lex->create_item_functor(thd, ident, $5,
                                                    ident2, name_cli)))
                 MYSQL_YYABORT;
               item->set_name(thd, $1.pos(), end - $1.pos(), thd->charset());
@@ -11469,32 +11484,38 @@ function_call_generic:
             {
 #ifdef HAVE_DLOPEN
               /* Retrieving the result of find_udf */
-              udf_func *udf= $<udf>3;
+              udf_func *udf= $<generic_function>3.udf;
 
               if (udf)
-              {
-                if (udf->type == UDFTYPE_AGGREGATE)
-                {
-                  Select->in_sum_expr--;
-                }
-
-                item= Create_udf_func::s_singleton.create(thd, udf, $4);
-              }
+                item= Create_udf_func::s_singleton.create(thd, udf, $5);
               else
 #endif
               {
                 builder= find_qualified_function_builder(thd);
                 DBUG_ASSERT(builder);
-                item= builder->create_func(thd, &ident, $4);
+                item= builder->create_func(thd, &ident, $5);
+              }
+            }
+
+            Item_sum *sum_item= (item &&
+                                 item->type() == Item::SUM_FUNC_ITEM) ?
+                                static_cast<Item_sum *>(item) : NULL;
+            if ($4)
+            {
+              if (!native_item || !sum_item)
+              {
+                thd->parse_error();
+                MYSQL_YYABORT;
               }
+              sum_item->set_distinct(true);
             }
 
-            if (item && attach_func_filter(item, $7))
+            if (item && attach_func_filter(item, $8))
               MYSQL_YYABORT;
 
-            if ($6.str && !allow_field_accessor)
+            if ($7.str && !allow_field_accessor)
             {
-              Lex_ident_sys field_sys(thd, &$6);
+              Lex_ident_sys field_sys(thd, &$7);
               my_error(ER_BAD_FIELD_ERROR, MYF(0), field_sys.str, ident.str);
               MYSQL_YYABORT;
             }
@@ -11910,11 +11931,10 @@ window_func:
         |
           function_call_generic
           {
-            Item* item = (Item*)$1;
-            /* Only UDF aggregate here possible */
-            if ((item == NULL) ||
-                (item->type() != Item::SUM_FUNC_ITEM)
-                || (((Item_sum *)item)->sum_func() != Item_sum::UDF_SUM_FUNC))
+            Item *item= (Item *) $1;
+            if (!item || item->type() != Item::SUM_FUNC_ITEM ||
+                (((Item_sum *) item)->sum_func() != Item_sum::UDF_SUM_FUNC &&
+                 ((Item_sum *) item)->sum_func() != Item_sum::PLUGIN_SUM_FUNC))
             {
               thd->parse_error();
               MYSQL_YYABORT;