MDEV-7773: Aggregate stored functions

This commit implements aggregate stored functions. The basic idea behind
the feature is:

* Implement a special instruction FETCH GROUP NEXT ROW that will pause
the execution of the stored function. When the instruction is reached,
execution of the initial query resumes "as if" the function returned.
This gives the server the opportunity to advance to the next row in the
result set.

* Stored aggregates behave like regular aggregate functions. The
implementation of thus resides in the class Item_sum_sp. Because it is
an aggregate function, for each new row in the group, the
Item_sum_sp::add() method will be called. This is when execution resumes
and the function does another iteration to "add" one extra element to
the final result.

* When the end of group is reached, val_xxx() method will be called for
the item. This case is handled by another execute step for the stored
function, only with a special flag to force a call to the return
handler. See Item_sum_sp::execute() for details.

To allow this pause and resume semantic, we must preserve the function
context across executions. This is stored in Item_sp::sp_query_arena only for
aggregate stored functions, but has no impact for regular functions.

We also enforce aggregate functions to include the "FETCH GROUP NEXT ROW"
instruction.
Signed-off-by: Vicențiu Ciorbaru <vicentiu@mariadb.org>

mysql-test/r/create_drop_function.result

@@ -3,8 +3,8 @@ CREATE FUNCTION f1(str char(20))
 RETURNS CHAR(100)
 RETURN CONCAT('Hello, ', str, '!');
 SELECT * FROM mysql.proc WHERE name like 'f1';
-db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8
-test	f1	FUNCTION	f1	SQL	CONTAINS_SQL	NO	DEFINER	str char(20)	char(100) CHARSET latin1	RETURN CONCAT('Hello, ', str, '!')	root@localhost	2014-09-30 08:00:00	2014-09-30 08:00:00	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION		latin1	latin1_swedish_ci	latin1_swedish_ci	RETURN CONCAT('Hello, ', str, '!')
+db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8	aggregate
+test	f1	FUNCTION	f1	SQL	CONTAINS_SQL	NO	DEFINER	str char(20)	char(100) CHARSET latin1	RETURN CONCAT('Hello, ', str, '!')	root@localhost	2014-09-30 08:00:00	2014-09-30 08:00:00	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION		latin1	latin1_swedish_ci	latin1_swedish_ci	RETURN CONCAT('Hello, ', str, '!')	NONE
 SELECT f1('world');
 f1('world')
 Hello, world!

mysql-test/r/information_schema.result

@@ -665,6 +665,7 @@ proc	character_set_client	char(32)
 proc	collation_connection	char(32)
 proc	db_collation	char(32)
 proc	body_utf8	longblob
+proc	aggregate	enum('NONE','GROUP')
 drop table t115;
 create procedure p108 () begin declare c cursor for select data_type
 from information_schema.columns;  open c; open c; end;//
@@ -1270,7 +1271,7 @@ drop table t1;
 use mysql;
 INSERT INTO `proc` VALUES ('test','','PROCEDURE','','SQL','CONTAINS_SQL',
 'NO','DEFINER','','','BEGIN\r\n  \r\nEND','root@%','2006-03-02 18:40:03',
-'2006-03-02 18:40:03','','','utf8','utf8_general_ci','utf8_general_ci','n/a');
+'2006-03-02 18:40:03','','','utf8','utf8_general_ci','utf8_general_ci','n/a', 'NONE');
 select routine_name from information_schema.routines where ROUTINE_SCHEMA='test';
 routine_name
 

mysql-test/r/sp-destruct.result

-call mtr.add_suppression("Column count of mysql.proc is wrong. Expected 20, found 19. The table is probably corrupted");
+call mtr.add_suppression("Column count of mysql.proc is wrong. Expected 21, found 20. The table is probably corrupted");
 call mtr.add_suppression("Stored routine .test...bug14233_[123].: invalid value in column mysql.proc");
 flush table mysql.proc;
 use test;
@@ -14,13 +14,13 @@ create table t1 (id int);
 create trigger t1_ai after insert on t1 for each row call bug14233();
 alter table mysql.proc drop security_type;
 call bug14233();
-ERROR HY000: Column count of mysql.proc is wrong. Expected 20, found 19. The table is probably corrupted
+ERROR HY000: Column count of mysql.proc is wrong. Expected 21, found 20. The table is probably corrupted
 create view v1 as select bug14233_f();
-ERROR HY000: Column count of mysql.proc is wrong. Expected 20, found 19. The table is probably corrupted
+ERROR HY000: Column count of mysql.proc is wrong. Expected 21, found 20. The table is probably corrupted
 insert into t1 values (0);
-ERROR HY000: Column count of mysql.proc is wrong. Expected 20, found 19. The table is probably corrupted
+ERROR HY000: Column count of mysql.proc is wrong. Expected 21, found 20. The table is probably corrupted
 show procedure status;
-ERROR HY000: Column count of mysql.proc is wrong. Expected 20, found 19. The table is probably corrupted
+ERROR HY000: Column count of mysql.proc is wrong. Expected 21, found 20. The table is probably corrupted
 flush table mysql.proc;
 call bug14233();
 ERROR HY000: Incorrect information in file: './mysql/proc.frm'
@@ -146,7 +146,7 @@ alter table mysql.proc drop column security_type;
 # The below statement should not cause assertion failure.
 drop database mysqltest;
 Warnings:
-Error	1805	Column count of mysql.proc is wrong. Expected 20, found 19. The table is probably corrupted
+Error	1805	Column count of mysql.proc is wrong. Expected 21, found 20. The table is probably corrupted
 # Restore mysql.proc.
 drop table mysql.proc;
 #

mysql-test/r/sp-error.result

@@ -1212,7 +1212,7 @@ ERROR 42S02: Unknown table 'c' in field list
 drop procedure bug15091;
 drop function if exists bug16896;
 create aggregate function bug16896() returns int return 1;
-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 '() returns int return 1' at line 1
+ERROR HY000: Aggregate specific instruction(FETCH GROUP NEXT ROW) missing from the aggregate function
 DROP PROCEDURE IF EXISTS bug14702;
 CREATE IF NOT EXISTS PROCEDURE bug14702()
 BEGIN

mysql-test/r/sp.result

@@ -5324,7 +5324,7 @@ DROP PROCEDURE bug21414|
 set names utf8|
 drop database if exists това_е_дълго_име_за_база_данни_нали|
 create database това_е_дълго_име_за_база_данни_нали|
-INSERT INTO mysql.proc VALUES ('това_е_дълго_име_за_база_данни_нали','това_е_процедура_с_доста_дълго_име_нали_и_още_по_дълго','PROCEDURE','това_е_процедура_с_доста_дълго_име_нали_и_още_по_дълго','SQL','CONTAINS_SQL','NO','DEFINER','','','bad_body','root@localhost',now(), now(),'','', 'utf8', 'utf8_general_ci', 'utf8_general_ci', 'n/a')|
+INSERT INTO mysql.proc VALUES ('това_е_дълго_име_за_база_данни_нали','това_е_процедура_с_доста_дълго_име_нали_и_още_по_дълго','PROCEDURE','това_е_процедура_с_доста_дълго_име_нали_и_още_по_дълго','SQL','CONTAINS_SQL','NO','DEFINER','','','bad_body','root@localhost',now(), now(),'','', 'utf8', 'utf8_general_ci', 'utf8_general_ci', 'n/a', 'NONE')|
 call това_е_дълго_име_за_база_данни_нали.това_е_процедура_с_доста_дълго_име_нали_и_още_по_дълго()|
 ERROR HY000: Failed to load routine това_е_дълго_име_за_база_данни_нали.това_е_процедура_с_доста_дълго_име_нали_и_още_по_дълго. The table mysql.proc is missing, corrupt, or contains bad data (internal code -6)
 drop database това_е_дълго_име_за_база_данни_нали|

mysql-test/r/system_mysql_db.result

@@ -220,6 +220,7 @@ proc	CREATE TABLE `proc` (
   `collation_connection` char(32) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL,
   `db_collation` char(32) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL,
   `body_utf8` longblob DEFAULT NULL,
+  `aggregate` enum('NONE','GROUP') NOT NULL DEFAULT 'NONE',
   PRIMARY KEY (`db`,`name`,`type`)
 ) ENGINE=MyISAM DEFAULT CHARSET=utf8 COMMENT='Stored Procedures'
 show create table event;

mysql-test/r/system_mysql_db_fix40123.result

@@ -220,6 +220,7 @@ proc	CREATE TABLE `proc` (
   `collation_connection` char(32) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL,
   `db_collation` char(32) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL,
   `body_utf8` longblob DEFAULT NULL,
+  `aggregate` enum('NONE','GROUP') NOT NULL DEFAULT 'NONE',
   PRIMARY KEY (`db`,`name`,`type`)
 ) ENGINE=MyISAM DEFAULT CHARSET=utf8 COMMENT='Stored Procedures'
 show create table event;

mysql-test/suite/funcs_1/r/is_columns_mysql.result

@@ -133,6 +133,7 @@ def	mysql	innodb_table_stats	sum_of_other_index_sizes	6	NULL	NO	bigint	NULL	NULL
 def	mysql	innodb_table_stats	table_name	2	NULL	NO	varchar	64	192	NULL	NULL	NULL	utf8	utf8_bin	varchar(64)	PRI		select,insert,update,references		NEVER	NULL
 def	mysql	plugin	dl	2	''	NO	varchar	128	384	NULL	NULL	NULL	utf8	utf8_general_ci	varchar(128)			select,insert,update,references		NEVER	NULL
 def	mysql	plugin	name	1	''	NO	varchar	64	192	NULL	NULL	NULL	utf8	utf8_general_ci	varchar(64)	PRI		select,insert,update,references		NEVER	NULL
+def	mysql	proc	aggregate	21	'NONE'	NO	enum	5	15	NULL	NULL	NULL	utf8	utf8_general_ci	enum('NONE','GROUP')			select,insert,update,references		NEVER	NULL
 def	mysql	proc	body	11	NULL	NO	longblob	4294967295	4294967295	NULL	NULL	NULL	NULL	NULL	longblob			select,insert,update,references		NEVER	NULL
 def	mysql	proc	body_utf8	20	NULL	YES	longblob	4294967295	4294967295	NULL	NULL	NULL	NULL	NULL	longblob			select,insert,update,references		NEVER	NULL
 def	mysql	proc	character_set_client	17	NULL	YES	char	32	96	NULL	NULL	NULL	utf8	utf8_bin	char(32)			select,insert,update,references		NEVER	NULL
@@ -488,6 +489,7 @@ NULL	mysql	proc	modified	timestamp	NULL	NULL	NULL	NULL	timestamp
 3.0000	mysql	proc	collation_connection	char	32	96	utf8	utf8_bin	char(32)
 3.0000	mysql	proc	db_collation	char	32	96	utf8	utf8_bin	char(32)
 1.0000	mysql	proc	body_utf8	longblob	4294967295	4294967295	NULL	NULL	longblob
+3.0000	mysql	proc	aggregate	enum	5	15	utf8	utf8_general_ci	enum('NONE','GROUP')
 3.0000	mysql	procs_priv	Host	char	60	180	utf8	utf8_bin	char(60)
 3.0000	mysql	procs_priv	Db	char	64	192	utf8	utf8_bin	char(64)
 3.0000	mysql	procs_priv	User	char	80	240	utf8	utf8_bin	char(80)

mysql-test/suite/funcs_1/r/is_columns_mysql_embedded.result

@@ -119,6 +119,7 @@ def	mysql	index_stats	prefix_arity	4	NULL	NO	int	NULL	NULL	10	0	NULL	NULL	NULL	i
 def	mysql	index_stats	table_name	2	NULL	NO	varchar	64	192	NULL	NULL	NULL	utf8	utf8_bin	varchar(64)	PRI				NEVER	NULL
 def	mysql	plugin	dl	2	''	NO	varchar	128	384	NULL	NULL	NULL	utf8	utf8_general_ci	varchar(128)					NEVER	NULL
 def	mysql	plugin	name	1	''	NO	varchar	64	192	NULL	NULL	NULL	utf8	utf8_general_ci	varchar(64)	PRI				NEVER	NULL
+def	mysql	proc	aggregate	21	'NONE'	NO	enum	5	15	NULL	NULL	NULL	utf8	utf8_general_ci	enum('NONE','GROUP')					NEVER	NULL
 def	mysql	proc	body	11	NULL	NO	longblob	4294967295	4294967295	NULL	NULL	NULL	NULL	NULL	longblob					NEVER	NULL
 def	mysql	proc	body_utf8	20	NULL	YES	longblob	4294967295	4294967295	NULL	NULL	NULL	NULL	NULL	longblob					NEVER	NULL
 def	mysql	proc	character_set_client	17	NULL	YES	char	32	96	NULL	NULL	NULL	utf8	utf8_bin	char(32)					NEVER	NULL
@@ -472,6 +473,7 @@ NULL	mysql	proc	modified	timestamp	NULL	NULL	NULL	NULL	timestamp
 3.0000	mysql	proc	collation_connection	char	32	96	utf8	utf8_bin	char(32)
 3.0000	mysql	proc	db_collation	char	32	96	utf8	utf8_bin	char(32)
 1.0000	mysql	proc	body_utf8	longblob	4294967295	4294967295	NULL	NULL	longblob
+3.0000	mysql	proc	aggregate	enum	5	15	utf8	utf8_general_ci	enum('NONE','GROUP')
 3.0000	mysql	procs_priv	Host	char	60	180	utf8	utf8_bin	char(60)
 3.0000	mysql	procs_priv	Db	char	64	192	utf8	utf8_bin	char(64)
 3.0000	mysql	procs_priv	User	char	80	240	utf8	utf8_bin	char(80)

mysql-test/suite/innodb/r/instant_alter_debug.result

@@ -35,7 +35,7 @@ ALTER TABLE t4 ADD COLUMN b INT;
 SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS
 LEFT JOIN t4 ON (NUMERIC_SCALE = pk);
 COUNT(*)
-1734
+1735
 SET DEBUG_SYNC='innodb_inplace_alter_table_enter SIGNAL enter WAIT_FOR delete';
 ALTER TABLE t4 ADD COLUMN c INT;
 connect  dml,localhost,root,,;

mysql-test/suite/rpl/r/rpl_sp.result

@@ -17,7 +17,7 @@ insert into t1 values (b);
 insert into t1 values (unix_timestamp());
 end|
 select * from mysql.proc where name='foo' and db='mysqltest1';
-db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8
+db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8	aggregate
 mysqltest1	foo	PROCEDURE	foo	SQL	CONTAINS_SQL	NO	DEFINER			begin
 declare b int;
 set b = 8;
@@ -28,10 +28,10 @@ declare b int;
 set b = 8;
 insert into t1 values (b);
 insert into t1 values (unix_timestamp());
-end
+end	NONE
 connection slave;
 select * from mysql.proc where name='foo' and db='mysqltest1';
-db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8
+db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8	aggregate
 mysqltest1	foo	PROCEDURE	foo	SQL	CONTAINS_SQL	NO	DEFINER			begin
 declare b int;
 set b = 8;
@@ -42,7 +42,7 @@ declare b int;
 set b = 8;
 insert into t1 values (b);
 insert into t1 values (unix_timestamp());
-end
+end	NONE
 connection master;
 set timestamp=1000000000;
 call foo();
@@ -137,19 +137,19 @@ select * from t2;
 a
 20
 select * from mysql.proc where name="foo4" and db='mysqltest1';
-db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8
+db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8	aggregate
 mysqltest1	foo4	PROCEDURE	foo4	SQL	CONTAINS_SQL	YES	DEFINER			begin
 insert into t2 values(20),(20);
 end	root@localhost	#	#			latin1	latin1_swedish_ci	latin1_swedish_ci	begin
 insert into t2 values(20),(20);
-end
+end	NONE
 connection master;
 drop procedure foo4;
 select * from mysql.proc where name="foo4" and db='mysqltest1';
-db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8
+db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8	aggregate
 connection slave;
 select * from mysql.proc where name="foo4" and db='mysqltest1';
-db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8
+db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8	aggregate
 connection master;
 drop procedure foo;
 drop procedure foo2;
@@ -235,22 +235,22 @@ select fn3();
 fn3()
 0
 select * from mysql.proc where db='mysqltest1';
-db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8
+db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8	aggregate
 mysqltest1	fn1	FUNCTION	fn1	SQL	NO_SQL	NO	DEFINER		int(11)	begin
 return unix_timestamp();
 end	root@localhost	#	#			latin1	latin1_swedish_ci	latin1_swedish_ci	begin
 return unix_timestamp();
-end
+end	NONE
 mysqltest1	fn2	FUNCTION	fn2	SQL	NO_SQL	NO	DEFINER		int(11)	begin
 return unix_timestamp();
 end	zedjzlcsjhd@localhost	#	#			latin1	latin1_swedish_ci	latin1_swedish_ci	begin
 return unix_timestamp();
-end
+end	NONE
 mysqltest1	fn3	FUNCTION	fn3	SQL	READS_SQL_DATA	NO	DEFINER		int(11)	begin
 return 0;
 end	root@localhost	#	#			latin1	latin1_swedish_ci	latin1_swedish_ci	begin
 return 0;
-end
+end	NONE
 select * from t1;
 a
 1000000000
@@ -260,22 +260,22 @@ select * from t1;
 a
 1000000000
 select * from mysql.proc where db='mysqltest1';
-db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8
+db	name	type	specific_name	language	sql_data_access	is_deterministic	security_type	param_list	returns	body	definer	created	modified	sql_mode	comment	character_set_client	collation_connection	db_collation	body_utf8	aggregate
 mysqltest1	fn1	FUNCTION	fn1	SQL	NO_SQL	NO	DEFINER		int(11)	begin
 return unix_timestamp();
 end	root@localhost	#	#			latin1	latin1_swedish_ci	latin1_swedish_ci	begin
 return unix_timestamp();
-end
+end	NONE
 mysqltest1	fn2	FUNCTION	fn2	SQL	NO_SQL	NO	DEFINER		int(11)	begin
 return unix_timestamp();
 end	zedjzlcsjhd@localhost	#	#			latin1	latin1_swedish_ci	latin1_swedish_ci	begin
 return unix_timestamp();
-end
+end	NONE
 mysqltest1	fn3	FUNCTION	fn3	SQL	READS_SQL_DATA	NO	DEFINER		int(11)	begin
 return 0;
 end	root@localhost	#	#			latin1	latin1_swedish_ci	latin1_swedish_ci	begin
 return 0;
-end
+end	NONE
 connection master;
 delete from t2;
 alter table t2 add unique (a);

mysql-test/t/information_schema.test

@@ -867,7 +867,7 @@ drop table t1;
 use mysql;
 INSERT INTO `proc` VALUES ('test','','PROCEDURE','','SQL','CONTAINS_SQL',
 'NO','DEFINER','','','BEGIN\r\n  \r\nEND','root@%','2006-03-02 18:40:03',
-'2006-03-02 18:40:03','','','utf8','utf8_general_ci','utf8_general_ci','n/a');
+'2006-03-02 18:40:03','','','utf8','utf8_general_ci','utf8_general_ci','n/a', 'NONE');
 select routine_name from information_schema.routines where ROUTINE_SCHEMA='test';
 delete from proc where name='';
 use test;

mysql-test/t/sp-destruct.test

@@ -10,7 +10,7 @@
 -- source include/not_embedded.inc
 
 # Supress warnings written to the log file
-call mtr.add_suppression("Column count of mysql.proc is wrong. Expected 20, found 19. The table is probably corrupted");
+call mtr.add_suppression("Column count of mysql.proc is wrong. Expected 21, found 20. The table is probably corrupted");
 call mtr.add_suppression("Stored routine .test...bug14233_[123].: invalid value in column mysql.proc");
 
 # Backup proc table

mysql-test/t/sp-error.test

@@ -1746,7 +1746,7 @@ drop procedure bug15091;
 drop function if exists bug16896;
 --enable_warnings
 
---error ER_PARSE_ERROR
+--error ER_INVALID_AGGREGATE_FUNCTION
 create aggregate function bug16896() returns int return 1;
 
 #

mysql-test/t/sp.test

@@ -6317,7 +6317,7 @@ set names utf8|
 drop database if exists това_е_дълго_име_за_база_данни_нали|
 --enable_warnings
 create database това_е_дълго_име_за_база_данни_нали|
-INSERT INTO mysql.proc VALUES ('това_е_дълго_име_за_база_данни_нали','това_е_процедура_с_доста_дълго_име_нали_и_още_по_дълго','PROCEDURE','това_е_процедура_с_доста_дълго_име_нали_и_още_по_дълго','SQL','CONTAINS_SQL','NO','DEFINER','','','bad_body','root@localhost',now(), now(),'','', 'utf8', 'utf8_general_ci', 'utf8_general_ci', 'n/a')|
+INSERT INTO mysql.proc VALUES ('това_е_дълго_име_за_база_данни_нали','това_е_процедура_с_доста_дълго_име_нали_и_още_по_дълго','PROCEDURE','това_е_процедура_с_доста_дълго_име_нали_и_още_по_дълго','SQL','CONTAINS_SQL','NO','DEFINER','','','bad_body','root@localhost',now(), now(),'','', 'utf8', 'utf8_general_ci', 'utf8_general_ci', 'n/a', 'NONE')|
 --error ER_SP_PROC_TABLE_CORRUPT
 call това_е_дълго_име_за_база_данни_нали.това_е_процедура_с_доста_дълго_име_нали_и_още_по_дълго()|
 drop database това_е_дълго_име_за_база_данни_нали|

scripts/mysql_system_tables.sql

@@ -80,7 +80,7 @@ CREATE TABLE IF NOT EXISTS time_zone_transition_type (   Time_zone_id int unsign
 
 CREATE TABLE IF NOT EXISTS time_zone_leap_second (   Transition_time bigint signed NOT NULL, Correction int signed NOT NULL, PRIMARY KEY TranTime (Transition_time) ) engine=MyISAM CHARACTER SET utf8   comment='Leap seconds information for time zones';
 
-CREATE TABLE IF NOT EXISTS proc (db char(64) collate utf8_bin DEFAULT '' NOT NULL, name char(64) DEFAULT '' NOT NULL, type enum('FUNCTION','PROCEDURE') NOT NULL, specific_name char(64) DEFAULT '' NOT NULL, language enum('SQL') DEFAULT 'SQL' NOT NULL, sql_data_access enum( 'CONTAINS_SQL', 'NO_SQL', 'READS_SQL_DATA', 'MODIFIES_SQL_DATA') DEFAULT 'CONTAINS_SQL' NOT NULL, is_deterministic enum('YES','NO') DEFAULT 'NO' NOT NULL, security_type enum('INVOKER','DEFINER') DEFAULT 'DEFINER' NOT NULL, param_list blob NOT NULL, returns longblob NOT NULL, body longblob NOT NULL, definer char(141) collate utf8_bin DEFAULT '' NOT NULL, created timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, modified timestamp NOT NULL DEFAULT '0000-00-00 00:00:00', sql_mode set( 'REAL_AS_FLOAT', 'PIPES_AS_CONCAT', 'ANSI_QUOTES', 'IGNORE_SPACE', 'IGNORE_BAD_TABLE_OPTIONS', 'ONLY_FULL_GROUP_BY', 'NO_UNSIGNED_SUBTRACTION', 'NO_DIR_IN_CREATE', 'POSTGRESQL', 'ORACLE', 'MSSQL', 'DB2', 'MAXDB', 'NO_KEY_OPTIONS', 'NO_TABLE_OPTIONS', 'NO_FIELD_OPTIONS', 'MYSQL323', 'MYSQL40', 'ANSI', 'NO_AUTO_VALUE_ON_ZERO', 'NO_BACKSLASH_ESCAPES', 'STRICT_TRANS_TABLES', 'STRICT_ALL_TABLES', 'NO_ZERO_IN_DATE', 'NO_ZERO_DATE', 'INVALID_DATES', 'ERROR_FOR_DIVISION_BY_ZERO', 'TRADITIONAL', 'NO_AUTO_CREATE_USER', 'HIGH_NOT_PRECEDENCE', 'NO_ENGINE_SUBSTITUTION', 'PAD_CHAR_TO_FULL_LENGTH', 'EMPTY_STRING_IS_NULL') DEFAULT '' NOT NULL, comment text collate utf8_bin NOT NULL, character_set_client char(32) collate utf8_bin, collation_connection char(32) collate utf8_bin, db_collation char(32) collate utf8_bin, body_utf8 longblob, PRIMARY KEY (db,name,type)) engine=MyISAM character set utf8 comment='Stored Procedures';
+CREATE TABLE IF NOT EXISTS proc (db char(64) collate utf8_bin DEFAULT '' NOT NULL, name char(64) DEFAULT '' NOT NULL, type enum('FUNCTION','PROCEDURE') NOT NULL, specific_name char(64) DEFAULT '' NOT NULL, language enum('SQL') DEFAULT 'SQL' NOT NULL, sql_data_access enum( 'CONTAINS_SQL', 'NO_SQL', 'READS_SQL_DATA', 'MODIFIES_SQL_DATA') DEFAULT 'CONTAINS_SQL' NOT NULL, is_deterministic enum('YES','NO') DEFAULT 'NO' NOT NULL, security_type enum('INVOKER','DEFINER') DEFAULT 'DEFINER' NOT NULL, param_list blob NOT NULL, returns longblob NOT NULL, body longblob NOT NULL, definer char(141) collate utf8_bin DEFAULT '' NOT NULL, created timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, modified timestamp NOT NULL DEFAULT '0000-00-00 00:00:00', sql_mode set( 'REAL_AS_FLOAT', 'PIPES_AS_CONCAT', 'ANSI_QUOTES', 'IGNORE_SPACE', 'IGNORE_BAD_TABLE_OPTIONS', 'ONLY_FULL_GROUP_BY', 'NO_UNSIGNED_SUBTRACTION', 'NO_DIR_IN_CREATE', 'POSTGRESQL', 'ORACLE', 'MSSQL', 'DB2', 'MAXDB', 'NO_KEY_OPTIONS', 'NO_TABLE_OPTIONS', 'NO_FIELD_OPTIONS', 'MYSQL323', 'MYSQL40', 'ANSI', 'NO_AUTO_VALUE_ON_ZERO', 'NO_BACKSLASH_ESCAPES', 'STRICT_TRANS_TABLES', 'STRICT_ALL_TABLES', 'NO_ZERO_IN_DATE', 'NO_ZERO_DATE', 'INVALID_DATES', 'ERROR_FOR_DIVISION_BY_ZERO', 'TRADITIONAL', 'NO_AUTO_CREATE_USER', 'HIGH_NOT_PRECEDENCE', 'NO_ENGINE_SUBSTITUTION', 'PAD_CHAR_TO_FULL_LENGTH', 'EMPTY_STRING_IS_NULL') DEFAULT '' NOT NULL, comment text collate utf8_bin NOT NULL, character_set_client char(32) collate utf8_bin, collation_connection char(32) collate utf8_bin, db_collation char(32) collate utf8_bin, body_utf8 longblob, aggregate enum('NONE', 'GROUP') DEFAULT 'NONE' NOT NULL, PRIMARY KEY (db,name,type)) engine=MyISAM character set utf8 comment='Stored Procedures';
 
 CREATE TABLE IF NOT EXISTS procs_priv ( Host char(60) binary DEFAULT '' NOT NULL, Db char(64) binary DEFAULT '' NOT NULL, User char(80) binary DEFAULT '' NOT NULL, Routine_name char(64) COLLATE utf8_general_ci DEFAULT '' NOT NULL, Routine_type enum('FUNCTION','PROCEDURE') NOT NULL, Grantor char(141) DEFAULT '' NOT NULL, Proc_priv set('Execute','Alter Routine','Grant') COLLATE utf8_general_ci DEFAULT '' NOT NULL, Timestamp timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (Host,Db,User,Routine_name,Routine_type), KEY Grantor (Grantor) ) engine=MyISAM CHARACTER SET utf8 COLLATE utf8_bin   comment='Procedure privileges';
 

sql/item.cc

@@ -2735,8 +2735,12 @@ Item_sp::Item_sp(THD *thd, Name_resolution_context *context_arg,
   context(context_arg), m_name(name_arg), m_sp(NULL), func_ctx(NULL),
   sp_result_field(NULL)
 {
-  dummy_table= (TABLE*) thd->calloc(sizeof(TABLE) + sizeof(TABLE_SHARE));
+  dummy_table= (TABLE*) thd->calloc(sizeof(TABLE) + sizeof(TABLE_SHARE) +
+                                    sizeof(Query_arena));
   dummy_table->s= (TABLE_SHARE*) (dummy_table + 1);
+  /* TODO(cvicentiu) Move this sp_query_arena in the class as a direct member.
+     Currently it can not be done due to header include dependencies. */
+  sp_query_arena= (Query_arena *) (dummy_table->s + 1);
   memset(&sp_mem_root, 0, sizeof(sp_mem_root));
 }
 
@@ -2876,19 +2880,37 @@ Item_sp::execute_impl(THD *thd, Item **args, uint arg_count)
   */
   thd->reset_sub_statement_state(&statement_state, SUB_STMT_FUNCTION);
 
-  init_sql_alloc(&sp_mem_root, MEM_ROOT_BLOCK_SIZE, 0, MYF(0));
-  Query_arena call_arena(&sp_mem_root, Query_arena::STMT_INITIALIZED_FOR_SP);
+  /*
+     If this function is an aggregate function, we want to initialise the
+     mem_root only once per group. For a regular stored function, we will
+     initialise once for each call to execute_function.
+  */
+  m_sp->agg_type();
+  DBUG_ASSERT(m_sp->agg_type() == GROUP_AGGREGATE ||
+              (m_sp->agg_type() == NOT_AGGREGATE && !func_ctx));
+  if (!func_ctx)
+  {
+    init_sql_alloc(&sp_mem_root, MEM_ROOT_BLOCK_SIZE, 0, MYF(0));
+    *sp_query_arena= Query_arena(&sp_mem_root,
+                                 Query_arena::STMT_INITIALIZED_FOR_SP);
+  }
 
   bool err_status= m_sp->execute_function(thd, args, arg_count,
                                           sp_result_field, &func_ctx,
-                                          &call_arena);
-  /* Free Items allocated during function execution. */
-  delete func_ctx;
-  func_ctx= NULL;
-  call_arena.free_items();
-  free_root(&sp_mem_root, MYF(0));
-  memset(&sp_mem_root, 0, sizeof(sp_mem_root));
-
+                                          sp_query_arena);
+  /*
+     We free the function context when the function finished executing normally
+     (quit_func == TRUE) or the function has exited with an error.
+  */
+  if (err_status || func_ctx->quit_func)
+  {
+    /* Free Items allocated during function execution. */
+    delete func_ctx;
+    func_ctx= NULL;
+    sp_query_arena->free_items();
+    free_root(&sp_mem_root, MYF(0));
+    memset(&sp_mem_root, 0, sizeof(sp_mem_root));
+  }
   thd->restore_sub_statement_state(&statement_state);
 
   thd->security_ctx= save_security_ctx;

sql/item.h

@@ -4485,6 +4485,7 @@ class Item_sp
   uchar result_buf[64];
   sp_rcontext *func_ctx;
   MEM_ROOT sp_mem_root;
+  Query_arena *sp_query_arena;
 
   /*
      The result field of the stored function.

sql/item_func.cc

@@ -6441,6 +6441,36 @@ Item_func_sp::fix_fields(THD *thd, Item **ref)
   if (res)
     DBUG_RETURN(TRUE);
 
+  if (m_sp->agg_type() == GROUP_AGGREGATE)
+  {
+    List<Item> list;
+    list.empty();
+    for (uint i=0; i < arg_count; i++)
+      list.push_back(*(args+i));
+
+    Item_sum_sp *item_sp;
+    Query_arena *arena, backup;
+    arena= thd->activate_stmt_arena_if_needed(&backup);
+
+    if (arg_count)
+      item_sp= new (thd->mem_root) Item_sum_sp(thd, context, m_name, sp, list);
+    else
+      item_sp= new (thd->mem_root) Item_sum_sp(thd, context, m_name, sp);
+
+    if (arena)
+      thd->restore_active_arena(arena, &backup);
+    if (!item_sp)
+      DBUG_RETURN(TRUE);
+    *ref= item_sp;
+    item_sp->name= name;
+    bool err= item_sp->fix_fields(thd, ref);
+    if (err)
+      DBUG_RETURN(TRUE);
+
+    list.empty();
+    DBUG_RETURN(FALSE);
+  }
+
   res= Item_func::fix_fields(thd, ref);
 
   if (res)

sql/item_sum.cc

@@ -30,6 +30,10 @@
 #include "sql_priv.h"
 #include "sql_select.h"
 #include "uniques.h"
+#include "sp_rcontext.h"
+#include "sp.h"
+#include "sql_parse.h"
+#include "sp_head.h"
 
 /**
   Calculate the affordable RAM limit for structures like TREE or Unique
@@ -1239,6 +1243,158 @@ Field *Item_sum_hybrid::create_tmp_field(bool group, TABLE *table)
   DBUG_RETURN(tmp_table_field_from_field_type(table));
 }
 
+/***********************************************************************
+** Item_sum_sp class
+***********************************************************************/
+
+Item_sum_sp::Item_sum_sp(THD *thd, Name_resolution_context *context_arg,
+                           sp_name *name_arg, sp_head *sp, List<Item> &list)
+  :Item_sum(thd, list), Item_sp(thd, context_arg, name_arg)
+{
+  maybe_null= 1;
+  quick_group= 0;
+  m_sp= sp;
+}
+
+Item_sum_sp::Item_sum_sp(THD *thd, Name_resolution_context *context_arg,
+                           sp_name *name_arg, sp_head *sp)
+  :Item_sum(thd), Item_sp(thd, context_arg, name_arg)
+{
+  maybe_null= 1;
+  quick_group= 0;
+  m_sp= sp;
+}
+
+
+bool
+Item_sum_sp::fix_fields(THD *thd, Item **ref)
+{
+  DBUG_ASSERT(fixed == 0);
+  if (init_sum_func_check(thd))
+    return TRUE;
+  decimals= 0;
+
+  m_sp= m_sp ? m_sp : sp_handler_function.sp_find_routine(thd, m_name, true);
+
+  if (!m_sp)
+  {
+    my_missing_function_error(m_name->m_name, ErrConvDQName(m_name).ptr());
+    context->process_error(thd);
+    return TRUE;
+  }
+
+  if (init_result_field(thd, max_length, maybe_null, &null_value, &name))
+    return TRUE;
+
+  for (uint i= 0 ; i < arg_count ; i++)
+  {
+    if (args[i]->fix_fields(thd, args + i) || args[i]->check_cols(1))
+      return TRUE;
+    set_if_bigger(decimals, args[i]->decimals);
+    m_with_subquery|= args[i]->with_subquery();
+    with_window_func|= args[i]->with_window_func;
+  }
+  result_field= NULL;
+  max_length= float_length(decimals);
+  null_value= 1;
+  fix_length_and_dec();
+
+  if (check_sum_func(thd, ref))
+    return TRUE;
+
+  memcpy(orig_args, args, sizeof(Item *) * arg_count);
+  fixed= 1;
+  return FALSE;
+}
+
+/**
+  Execute function to store value in result field.
+  This is called when we need the value to be returned for the function.
+  Here we send a signal in form of the server status that all rows have been
+  fetched and now we have to exit from the function with the return value.
+  @return Function returns error status.
+  @retval FALSE on success.
+  @retval TRUE if an error occurred.
+*/
+
+bool
+Item_sum_sp::execute()
+{
+  THD *thd= current_thd;
+  bool res;
+  uint old_server_status= thd->server_status;
+
+  /* We set server status so we can send a signal to exit from the
+     function with the return value. */
+
+  thd->server_status= SERVER_STATUS_LAST_ROW_SENT;
+  res= Item_sp::execute(thd, &null_value, args, arg_count);
+  thd->server_status= old_server_status;
+  return res;
+}
+
+/**
+  Handles the aggregation of the values.
+  @note: See class description for more details on how and why this is done.
+  @return The error state.
+  @retval FALSE on success.
+  @retval TRUE if an error occurred.
+*/
+
+bool
+Item_sum_sp::add()
+{
+  return execute_impl(current_thd, args, arg_count);
+}
+
+
+void
+Item_sum_sp::clear()
+{
+  delete func_ctx;
+  func_ctx= NULL;
+  sp_query_arena->free_items();
+  free_root(&sp_mem_root, MYF(0));
+}
+
+const Type_handler *Item_sum_sp::type_handler() const
+{
+  DBUG_ENTER("Item_sum_sp::type_handler");
+  DBUG_PRINT("info", ("m_sp = %p", (void *) m_sp));
+  DBUG_ASSERT(sp_result_field);
+  // This converts ENUM/SET to STRING
+  const Type_handler *handler= sp_result_field->type_handler();
+  DBUG_RETURN(handler->type_handler_for_item_field());
+}
+
+void
+Item_sum_sp::cleanup()
+{
+  Item_sp::cleanup();
+  Item_sum::cleanup();
+}
+
+/**
+  Initialize local members with values from the Field interface.
+  @note called from Item::fix_fields.
+*/
+
+void
+Item_sum_sp::fix_length_and_dec()
+{
+  DBUG_ENTER("Item_sum_sp::fix_length_and_dec");
+  DBUG_ASSERT(sp_result_field);
+  Type_std_attributes::set(sp_result_field);
+  Item_sum::fix_length_and_dec();
+  DBUG_VOID_RETURN;
+}
+
+const char *
+Item_sum_sp::func_name() const
+{
+  THD *thd= current_thd;
+  return Item_sp::func_name(thd);
+}
 
 /***********************************************************************
 ** reset and add of sum_func

sql/item_sum.h

@@ -355,7 +355,7 @@ class Item_sum :public Item_func_or_sum
     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,
-    PERCENTILE_DISC_FUNC
+    PERCENTILE_DISC_FUNC, SP_AGGREGATE_FUNC
   };
 
   Item **ref_by; /* pointer to a ref to the object used to register it */
@@ -1224,6 +1224,132 @@ class Item_sum_xor :public Item_sum_bit
   void set_bits_from_counters();
 };
 
+class sp_head;
+class sp_name;
+class Query_arena;
+struct st_sp_security_context;
+
+/*
+  Item_sum_sp handles STORED AGGREGATE FUNCTIONS
+
+  Each Item_sum_sp represents a custom aggregate function. Inside the
+  function's body, we require at least one occurence of FETCH GROUP NEXT ROW
+  instruction. This cursor is what makes custom stored aggregates possible.
+
+  During computation the function's add method is called. This in turn performs
+  an execution of the function. The function will execute from the current
+  function context (and instruction), if one exists, or from the start if not.
+  See Item_sp for more details.
+
+  Upon encounter of FETCH GROUP NEXT ROW instruction, the function will pause
+  execution. We assume that the user has performed the necessary additions for
+  a row, between two encounters of FETCH GROUP NEXT ROW.
+
+  Example:
+  create aggregate function f1(x INT) returns int
+  begin
+    declare continue handler for not found return s;
+    declare s int default 0
+    loop
+      fetch group next row;
+      set s = s + x;
+    end loop;
+  end
+
+  The function will always stop after an encounter of FETCH GROUP NEXT ROW,
+  except (!) on first encounter, as the value for the first row in the
+  group is already set in the argument x. This behaviour is done so when
+  a user writes a function, he should "logically" include FETCH GROUP NEXT ROW
+  before any "add" instructions in the stored function. This means however that
+  internally, the first occurence doesn't stop the function. See the
+  implementation of FETCH GROUP NEXT ROW for details as to how it happens.
+
+  Either way, one should assume that after calling "Item_sum_sp::add()" that
+  the values for that particular row have been added to the aggregation.
+
+  To produce values for val_xxx methods we need an extra syntactic construct.
+  We require a continue handler when "no more rows are available". val_xxx
+  methods force a function return by executing the function again, while
+  setting a server flag that no more rows have been found. This implies
+  that val_xxx methods should only be called once per group however.
+
+  Example:
+  DECLARE CONTINUE HANDLER FOR NOT FOUND RETURN ret_val;
+*/
+class Item_sum_sp :public Item_sum,
+                   public Item_sp
+{
+ private:
+  bool execute();
+
+public:
+  Item_sum_sp(THD *thd, Name_resolution_context *context_arg, sp_name *name,
+              sp_head *sp);
+
+  Item_sum_sp(THD *thd, Name_resolution_context *context_arg, sp_name *name,
+              sp_head *sp, List<Item> &list);
+
+  enum Sumfunctype sum_func () const
+  {
+    return SP_AGGREGATE_FUNC;
+  }
+  void fix_length_and_dec();
+  bool fix_fields(THD *thd, Item **ref);
+  const char *func_name() const;
+  const Type_handler *type_handler() const;
+  bool add();
+
+  /* val_xx functions */
+  longlong val_int()
+  {
+    if(execute())
+      return 0;
+    return sp_result_field->val_int();
+  }
+
+  double val_real()
+  {
+    if(execute())
+      return 0.0;
+    return sp_result_field->val_real();
+  }
+
+  my_decimal *val_decimal(my_decimal *dec_buf)
+  {
+    if(execute())
+      return NULL;
+    return sp_result_field->val_decimal(dec_buf);
+  }
+
+  String *val_str(String *str)
+  {
+    String buf;
+    char buff[20];
+    buf.set(buff, 20, str->charset());
+    buf.length(0);
+    if (execute())
+      return NULL;
+    /*
+      result_field will set buf pointing to internal buffer
+      of the resul_field. Due to this it will change any time
+      when SP is executed. In order to prevent occasional
+      corruption of returned value, we make here a copy.
+    */
+    sp_result_field->val_str(&buf);
+    str->copy(buf);
+    return str;
+  }
+  void reset_field(){DBUG_ASSERT(0);}
+  void update_field(){DBUG_ASSERT(0);}
+  void clear();
+  void cleanup();
+  inline Field *get_sp_result_field()
+  {
+    return sp_result_field;
+  }
+  Item *get_copy(THD *thd)
+  { return get_item_copy<Item_sum_sp>(thd, this); }
+};
 
 /* Items to get the value of a stored sum function */
 

sql/share/errmsg-utf8.txt

@@ -7803,3 +7803,7 @@ ER_ARGUMENT_OUT_OF_RANGE
         eng "Argument to the %s function does not belong to the range [0,1]"
 ER_WRONG_TYPE_OF_ARGUMENT
         eng "%s function only accepts arguments that can be converted to numerical types"
+ER_NOT_AGGREGATE_FUNCTION
+        eng "Non-aggregate function contains aggregate specific instructions: (FETCH GROUP NEXT ROW)"
+ER_INVALID_AGGREGATE_FUNCTION
+        eng "Aggregate specific instruction(FETCH GROUP NEXT ROW) missing from the aggregate function"

sql/sp.cc

@@ -201,6 +201,11 @@ TABLE_FIELD_TYPE proc_table_fields[MYSQL_PROC_FIELD_COUNT] =
     { C_STRING_WITH_LEN("body_utf8") },
     { C_STRING_WITH_LEN("longblob") },
     { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("aggregate") },
+    { C_STRING_WITH_LEN("enum('NONE','GROUP')") },
+    { NULL, 0 }
   }
 };
 
@@ -583,6 +588,22 @@ bool st_sp_chistics::read_from_mysql_proc_row(THD *thd, TABLE *table)
     return true;
   suid= str.str[0] == 'I' ? SP_IS_NOT_SUID : SP_IS_SUID;
 
+  if (table->field[MYSQL_PROC_FIELD_AGGREGATE]->val_str_nopad(thd->mem_root,
+                                                              &str))
+    return true;
+
+  switch (str.str[0]) {
+  case 'N':
+    agg_type= NOT_AGGREGATE;
+    break;
+  case 'G':
+    agg_type= GROUP_AGGREGATE;
+    break;
+  default:
+    agg_type= DEFAULT_AGGREGATE;
+  }
+
+
   if (table->field[MYSQL_PROC_FIELD_COMMENT]->val_str_nopad(thd->mem_root,
                                                             &comment))
     return true;
@@ -1183,6 +1204,13 @@ Sp_handler::sp_create_routine(THD *thd, const sp_head *sp) const
       table->field[MYSQL_PROC_FIELD_NAME]->
         store(sp->m_name, system_charset_info);
 
+    if (sp->agg_type() != DEFAULT_AGGREGATE)
+    {
+      store_failed= store_failed ||
+        table->field[MYSQL_PROC_FIELD_AGGREGATE]->
+          store((longlong)sp->agg_type(),TRUE);
+    }
+
     store_failed= store_failed ||
       table->field[MYSQL_PROC_MYSQL_TYPE]->
         store((longlong) type(), true);
@@ -1494,6 +1522,9 @@ Sp_handler::sp_update_routine(THD *thd, const Database_qualified_name *name,
     if (chistics->comment.str)
       table->field[MYSQL_PROC_FIELD_COMMENT]->store(chistics->comment,
 						    system_charset_info);
+    if (chistics->agg_type != DEFAULT_AGGREGATE)
+      table->field[MYSQL_PROC_FIELD_AGGREGATE]->
+         store((longlong)chistics->agg_type, TRUE);
     if ((ret= table->file->ha_update_row(table->record[1],table->record[0])) &&
         ret != HA_ERR_RECORD_IS_THE_SAME)
       ret= SP_WRITE_ROW_FAILED;
@@ -2238,11 +2269,12 @@ Sp_handler::show_create_sp(THD *thd, String *buf,
                            sql_mode_t sql_mode) const
 {
   sql_mode_t old_sql_mode= thd->variables.sql_mode;
+  size_t agglen= (chistics.agg_type == GROUP_AGGREGATE)? 10 : 0;
   /* Make some room to begin with */
   if (buf->alloc(100 + db.length + 1 + name.length +
                  params.length + returns.length +
-		 chistics.comment.length + 10 /* length of " DEFINER= "*/ +
-                 USER_HOST_BUFF_SIZE))
+                 chistics.comment.length + 10 /* length of " DEFINER= "*/ +
+                 agglen + USER_HOST_BUFF_SIZE))
     return true;
 
   thd->variables.sql_mode= sql_mode;
@@ -2250,6 +2282,8 @@ Sp_handler::show_create_sp(THD *thd, String *buf,
   if (ddl_options.or_replace())
     buf->append(STRING_WITH_LEN("OR REPLACE "));
   append_definer(thd, buf, &definer.user, &definer.host);
+  if (chistics.agg_type == GROUP_AGGREGATE)
+    buf->append(STRING_WITH_LEN("AGGREGATE "));
   buf->append(type_lex_cstring());
   buf->append(STRING_WITH_LEN(" "));
   if (ddl_options.if_not_exists())

sql/sp.h

@@ -370,6 +370,7 @@ enum
   MYSQL_PROC_FIELD_COLLATION_CONNECTION,
   MYSQL_PROC_FIELD_DB_COLLATION,
   MYSQL_PROC_FIELD_BODY_UTF8,
+  MYSQL_PROC_FIELD_AGGREGATE,
   MYSQL_PROC_FIELD_COUNT
 };
 

sql/sp_head.cc

@@ -999,6 +999,12 @@ sp_head::execute(THD *thd, bool merge_da_on_success)
   bool err_status= FALSE;
   uint ip= 0;
   sql_mode_t save_sql_mode;
+
+  // TODO(cvicentiu) See if you can drop this bit. This is used to resume
+  // execution from where we left off.
+  if (m_chistics.agg_type == GROUP_AGGREGATE)
+    ip= thd->spcont->instr_ptr;
+
   bool save_abort_on_warning;
   Query_arena *old_arena;
   /* per-instruction arena */
@@ -1176,6 +1182,7 @@ sp_head::execute(THD *thd, bool merge_da_on_success)
 #if defined(ENABLED_PROFILING)
       thd->profiling.discard_current_query();
 #endif
+      thd->spcont->quit_func= TRUE;
       break;
     }
 
@@ -1245,7 +1252,8 @@ sp_head::execute(THD *thd, bool merge_da_on_success)
     /* Reset sp_rcontext::end_partial_result_set flag. */
     ctx->end_partial_result_set= FALSE;
 
-  } while (!err_status && !thd->killed && !thd->is_fatal_error);
+  } while (!err_status && !thd->killed && !thd->is_fatal_error &&
+           !thd->spcont->pause_state);
 
 #if defined(ENABLED_PROFILING)
   thd->profiling.finish_current_query();
@@ -1261,9 +1269,14 @@ sp_head::execute(THD *thd, bool merge_da_on_success)
 
   thd->restore_active_arena(&execute_arena, &backup_arena);
 
-  thd->spcont->pop_all_cursors(); // To avoid memory leaks after an error
+  /* Only pop cursors when we're done with group aggregate running. */
+  if (m_chistics.agg_type != GROUP_AGGREGATE ||
+      (m_chistics.agg_type == GROUP_AGGREGATE && thd->spcont->quit_func))
+    thd->spcont->pop_all_cursors(); // To avoid memory leaks after an error
 
   /* Restore all saved */
+  if (m_chistics.agg_type == GROUP_AGGREGATE)
+    thd->spcont->instr_ptr= ip;
   thd->server_status= (thd->server_status & ~status_backup_mask) | old_server_status;
   old_packet.swap(thd->packet);
   DBUG_ASSERT(thd->change_list.is_empty());
@@ -1856,7 +1869,7 @@ sp_head::execute_function(THD *thd, Item **argp, uint argcount,
     }
   }
 
-  if (!err_status)
+  if (!err_status && thd->spcont->quit_func)
   {
     /* We need result only in function but not in trigger */
 
@@ -2482,7 +2495,6 @@ sp_head::set_chistics(const st_sp_chistics &chistics)
                                          m_chistics.comment.length);
 }
 
-
 void
 sp_head::set_info(longlong created, longlong modified,
                   const st_sp_chistics &chistics, sql_mode_t sql_mode)
@@ -4217,6 +4229,35 @@ sp_instr_cfetch::print(String *str)
   }
 }
 
+int
+sp_instr_agg_cfetch::execute(THD *thd, uint *nextp)
+{
+  DBUG_ENTER("sp_instr_cfetch::execute");
+  int res= 0;
+  if (!thd->spcont->instr_ptr)
+  {
+    *nextp= m_ip+1;
+    thd->spcont->instr_ptr= m_ip + 1;
+  }
+  else if (!thd->spcont->pause_state)
+    thd->spcont->pause_state= TRUE;
+  else
+  {
+    thd->spcont->pause_state= FALSE;
+    if (thd->server_status == SERVER_STATUS_LAST_ROW_SENT)
+    {
+      my_message(ER_SP_FETCH_NO_DATA,
+                 ER_THD(thd, ER_SP_FETCH_NO_DATA), MYF(0));
+      res= -1;
+      thd->spcont->quit_func= TRUE;
+    }
+    else
+      *nextp= m_ip + 1;
+  }
+  DBUG_RETURN(res);
+}
+
+
 
 /*
   sp_instr_cursor_copy_struct class functions

sql/sp_head.h

@@ -164,7 +164,10 @@ class sp_head :private Query_arena,
     /*
       Marks routines that have column type references: DECLARE a t1.a%TYPE;
     */
-    HAS_COLUMN_TYPE_REFS= 8192
+    HAS_COLUMN_TYPE_REFS= 8192,
+    /* Set if has FETCH GROUP NEXT ROW instr. Used to ensure that only
+       functions with AGGREGATE keyword use the instr. */
+    HAS_AGGREGATE_INSTR= 16384
   };
 
   const Sp_handler *m_handler;
@@ -197,6 +200,7 @@ class sp_head :private Query_arena,
   enum_sp_suid_behaviour suid() const { return m_chistics.suid; }
   bool detistic() const { return m_chistics.detistic; }
   enum_sp_data_access daccess() const { return m_chistics.daccess; }
+  enum_sp_aggregate_type agg_type() const { return m_chistics.agg_type; }
   /**
     Is this routine being executed?
   */
@@ -720,6 +724,10 @@ class sp_head :private Query_arena,
                                           const LEX_CSTRING &table);
 
   void set_chistics(const st_sp_chistics &chistics);
+  inline void set_chistics_agg_type(enum enum_sp_aggregate_type type)
+  {
+    m_chistics.agg_type= type;
+  }
   void set_info(longlong created, longlong modified,
 		const st_sp_chistics &chistics, sql_mode_t sql_mode);
 
@@ -1822,6 +1830,32 @@ class sp_instr_cfetch : public sp_instr
 
 }; // class sp_instr_cfetch : public sp_instr
 
+/*
+This class is created for the special fetch instruction
+FETCH GROUP NEXT ROW, used in the user-defined aggregate
+functions
+*/
+
+class sp_instr_agg_cfetch : public sp_instr
+{
+  sp_instr_agg_cfetch(const sp_instr_cfetch &); /**< Prevent use of these */
+  void operator=(sp_instr_cfetch &);
+
+public:
+
+  sp_instr_agg_cfetch(uint ip, sp_pcontext *ctx)
+    : sp_instr(ip, ctx){}
+
+  virtual ~sp_instr_agg_cfetch()
+  {}
+
+  virtual int execute(THD *thd, uint *nextp);
+
+  virtual void print(String *str){};
+}; // class sp_instr_agg_cfetch : public sp_instr
+
+
+
 
 class sp_instr_error : public sp_instr
 {

sql/sp_rcontext.cc

@@ -40,6 +40,7 @@ sp_rcontext::sp_rcontext(const sp_head *owner,
                          Field *return_value_fld,
                          bool in_sub_stmt)
   :end_partial_result_set(false),
+   pause_state(false), quit_func(false), instr_ptr(0),
    m_sp(owner),
    m_root_parsing_ctx(root_parsing_ctx),
    m_var_table(NULL),

sql/sp_rcontext.h

@@ -178,6 +178,9 @@ class sp_rcontext : public Sql_alloc
   /// (if one is found). Otherwise the client will hang due to a violation
   /// of the client/server protocol.
   bool end_partial_result_set;
+  bool pause_state;
+  bool quit_func;
+  uint instr_ptr;
 
   /// The stored program for which this runtime context is created. Used for
   /// checking if correct runtime context is used for variable handling.

sql/sql_lex.h

@@ -226,6 +226,13 @@ enum enum_sp_data_access
   SP_MODIFIES_SQL_DATA
 };
 
+enum enum_sp_aggregate_type
+{
+  DEFAULT_AGGREGATE= 0,
+  NOT_AGGREGATE,
+  GROUP_AGGREGATE
+};
+
 const LEX_STRING sp_data_access_name[]=
 {
   { C_STRING_WITH_LEN("") },
@@ -1290,6 +1297,7 @@ struct st_sp_chistics
   enum enum_sp_suid_behaviour suid;
   bool detistic;
   enum enum_sp_data_access daccess;
+  enum enum_sp_aggregate_type agg_type;
   void init() { bzero(this, sizeof(*this)); }
   void set(const st_sp_chistics &other) { *this= other; }
   bool read_from_mysql_proc_row(THD *thd, TABLE *table);

sql/sql_yacc.yy

@@ -1911,7 +1911,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         opt_field_or_var_spec fields_or_vars opt_load_data_set_spec
         view_list_opt view_list view_select
         trigger_tail sp_tail sf_tail event_tail
-        udf_tail create_function_tail
+        udf_tail create_function_tail create_aggregate_function_tail
         install uninstall partition_entry binlog_base64_event
         normal_key_options normal_key_opts all_key_opt 
         spatial_key_options fulltext_key_options normal_key_opt 
@@ -2638,8 +2638,16 @@ create:
           event_tail
           { }
         | create_or_replace definer FUNCTION_SYM
-          { Lex->create_info.set($1); }
-          sf_tail
+          {
+            Lex->create_info.set($1);
+          }
+          sf_tail_not_aggregate
+          { }
+        | create_or_replace definer AGGREGATE_SYM FUNCTION_SYM
+          {
+            Lex->create_info.set($1);
+          }
+          sf_tail_aggregate
           { }
         | create_or_replace no_definer FUNCTION_SYM
           { Lex->create_info.set($1); }
@@ -2648,9 +2656,8 @@ create:
         | create_or_replace no_definer AGGREGATE_SYM FUNCTION_SYM
           {
             Lex->create_info.set($1);
-            Lex->udf.type= UDFTYPE_AGGREGATE;
           }
-          udf_tail
+          create_aggregate_function_tail
           { }
         | create_or_replace USER_SYM opt_if_not_exists clear_privileges grant_list
           opt_require_clause opt_resource_options
@@ -2677,11 +2684,36 @@ create:
           { }
         ;
 
+sf_tail_not_aggregate:
+        sf_tail
+        {
+          if (Lex->sphead->m_flags & sp_head::HAS_AGGREGATE_INSTR)
+          {
+            my_yyabort_error((ER_NOT_AGGREGATE_FUNCTION, MYF(0), ""));
+          }
+          Lex->sphead->set_chistics_agg_type(NOT_AGGREGATE);
+        }
+
+sf_tail_aggregate:
+        sf_tail
+        {
+          if (!(Lex->sphead->m_flags & sp_head::HAS_AGGREGATE_INSTR))
+          {
+            my_yyabort_error((ER_INVALID_AGGREGATE_FUNCTION, MYF(0), ""));
+          }
+          Lex->sphead->set_chistics_agg_type(GROUP_AGGREGATE);
+        }
+
 create_function_tail:
-          sf_tail { }
+          sf_tail_not_aggregate { }
         | udf_tail { Lex->udf.type= UDFTYPE_FUNCTION; }
         ;
 
+create_aggregate_function_tail:
+          sf_tail_aggregate
+          { }
+        | udf_tail { Lex->udf.type= UDFTYPE_AGGREGATE; }
+        ;
 opt_sequence:
          /* empty */ { }
         | sequence_defs
@@ -4011,7 +4043,19 @@ sp_proc_stmt_fetch_head:
         ;
 
 sp_proc_stmt_fetch:
-          sp_proc_stmt_fetch_head sp_fetch_list { }
+         sp_proc_stmt_fetch_head sp_fetch_list { }
+       | FETCH_SYM GROUP_SYM NEXT_SYM ROW_SYM
+         {
+            LEX *lex= Lex;
+            sp_head *sp= lex->sphead;
+            lex->sphead->m_flags|= sp_head::HAS_AGGREGATE_INSTR;
+            sp_instr_agg_cfetch *i=
+              new (thd->mem_root) sp_instr_agg_cfetch(sp->instructions(),
+                                                      lex->spcont);
+            if (i == NULL ||
+                sp->add_instr(i))
+              MYSQL_YYABORT;
+         }
         ;
 
 sp_proc_stmt_close: