diff --git a/mysql-test/main/func_gconcat.result b/mysql-test/main/func_gconcat.result
index 0e6c5295f3b67..900710dfb4468 100644
--- a/mysql-test/main/func_gconcat.result
+++ b/mysql-test/main/func_gconcat.result
@@ -863,7 +863,7 @@ group_concat(distinct a, c)
 00,01,10,11,31
 select group_concat(distinct a, c order by a) from t1;
 group_concat(distinct a, c order by a)
-00,01,11,10,31
+01,00,11,10,31
 select group_concat(distinct a, c) from t1;
 group_concat(distinct a, c)
 00,01,10,11,31
@@ -1169,21 +1169,20 @@ LENGTH(GROUP_CONCAT(f1 ORDER BY f2))
 1024
 Warnings:
 Warning	1260	Row 2 was cut by group_concat()
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
 SET group_concat_max_len= 499999;
 SELECT LENGTH(GROUP_CONCAT(f1 ORDER BY f2)) FROM t1 WHERE f2 = 0;
 LENGTH(GROUP_CONCAT(f1 ORDER BY f2))
 499999
 Warnings:
-Warning	1260	Row 1 was cut by group_concat()
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
 SELECT LENGTH(GROUP_CONCAT(f1 ORDER BY f2)) FROM t1 GROUP BY f2;
 LENGTH(GROUP_CONCAT(f1 ORDER BY f2))
 499999
 499999
 499999
 Warnings:
-Warning	1260	Row 1 was cut by group_concat()
-Warning	1260	Row 2 was cut by group_concat()
-Warning	1260	Row 3 was cut by group_concat()
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
 INSERT INTO t1 VALUES (REPEAT('a', 499999), 3), (REPEAT('b', 500000), 4);
 SELECT LENGTH(GROUP_CONCAT(f1 ORDER BY f2)) FROM t1 GROUP BY f2;
 LENGTH(GROUP_CONCAT(f1 ORDER BY f2))
@@ -1193,10 +1192,7 @@ LENGTH(GROUP_CONCAT(f1 ORDER BY f2))
 499999
 499999
 Warnings:
-Warning	1260	Row 1 was cut by group_concat()
-Warning	1260	Row 2 was cut by group_concat()
-Warning	1260	Row 3 was cut by group_concat()
-Warning	1260	Row 5 was cut by group_concat()
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
 DROP TABLE t1;
 SET group_concat_max_len= DEFAULT;
 set session group_concat_max_len=1024;
diff --git a/mysql-test/main/gconcat_cut_note.result b/mysql-test/main/gconcat_cut_note.result
new file mode 100644
index 0000000000000..0d90df1dd5cbd
--- /dev/null
+++ b/mysql-test/main/gconcat_cut_note.result
@@ -0,0 +1,63 @@
+CREATE TABLE t1 (grp INT, t TEXT);
+INSERT INTO t1 SELECT seq, REPEAT(CHAR(64 + seq), 200) FROM seq_1_to_3;
+SET SESSION group_concat_max_len= 100;
+#
+# One row per group, so the result of each group is exactly the cut
+# value and is never longer than the limit. Nothing is missing from
+# any answer, and the three cut values give one note.
+#
+SELECT grp, LENGTH(GROUP_CONCAT(t ORDER BY t)) AS len FROM t1 GROUP BY grp;
+grp	len
+1	100
+2	100
+3	100
+Warnings:
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
+#
+# Two aggregates in one statement, so one note each. The brackets
+# JSON_ARRAYAGG() puts around the value push its result past the
+# limit, so that one loses data as well and warns for it.
+#
+SELECT grp, LENGTH(GROUP_CONCAT(t ORDER BY t)) AS gc_len,
+LENGTH(JSON_ARRAYAGG(t ORDER BY t)) AS ja_len FROM t1 GROUP BY grp;
+grp	gc_len	ja_len
+1	100	102
+2	100	102
+3	100	102
+Warnings:
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
+Warning	1260	Row 1 was cut by json_arrayagg()
+Note	4265	Some values were cut while processing json_arrayagg(). Increase group_concat_max_len if you want to avoid the cut
+Warning	1260	Row 2 was cut by json_arrayagg()
+Warning	1260	Row 3 was cut by json_arrayagg()
+#
+# The same statement run again gets its own note.
+#
+SELECT grp, LENGTH(GROUP_CONCAT(t ORDER BY t)) AS len FROM t1 GROUP BY grp;
+grp	len
+1	100
+2	100
+3	100
+Warnings:
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
+#
+# All three rows in one group. Now the result is cut as well, and
+# values the user asked for really are missing, so the warning is
+# given for that on top of the note.
+#
+SELECT LENGTH(GROUP_CONCAT(t ORDER BY t)) AS len FROM t1;
+len
+100
+Warnings:
+Warning	1260	Row 2 was cut by group_concat()
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
+#
+# Nothing is cut, so nothing is said.
+#
+SET SESSION group_concat_max_len= DEFAULT;
+SELECT grp, LENGTH(GROUP_CONCAT(t ORDER BY t)) AS len FROM t1 GROUP BY grp;
+grp	len
+1	200
+2	200
+3	200
+DROP TABLE t1;
diff --git a/mysql-test/main/gconcat_cut_note.test b/mysql-test/main/gconcat_cut_note.test
new file mode 100644
index 0000000000000..c8c9e8c867e42
--- /dev/null
+++ b/mysql-test/main/gconcat_cut_note.test
@@ -0,0 +1,47 @@
+#
+# A TEXT value longer than group_concat_max_len is cut on its way into
+# blob_storage, while the group is being built. Whether that changed the
+# answer is not known: the result may have been cut in the same place
+# anyway. So it is a note, and one note is enough for the statement.
+#
+--source include/have_sequence.inc
+
+CREATE TABLE t1 (grp INT, t TEXT);
+INSERT INTO t1 SELECT seq, REPEAT(CHAR(64 + seq), 200) FROM seq_1_to_3;
+
+SET SESSION group_concat_max_len= 100;
+
+--echo #
+--echo # One row per group, so the result of each group is exactly the cut
+--echo # value and is never longer than the limit. Nothing is missing from
+--echo # any answer, and the three cut values give one note.
+--echo #
+SELECT grp, LENGTH(GROUP_CONCAT(t ORDER BY t)) AS len FROM t1 GROUP BY grp;
+
+--echo #
+--echo # Two aggregates in one statement, so one note each. The brackets
+--echo # JSON_ARRAYAGG() puts around the value push its result past the
+--echo # limit, so that one loses data as well and warns for it.
+--echo #
+SELECT grp, LENGTH(GROUP_CONCAT(t ORDER BY t)) AS gc_len,
+       LENGTH(JSON_ARRAYAGG(t ORDER BY t)) AS ja_len FROM t1 GROUP BY grp;
+
+--echo #
+--echo # The same statement run again gets its own note.
+--echo #
+SELECT grp, LENGTH(GROUP_CONCAT(t ORDER BY t)) AS len FROM t1 GROUP BY grp;
+
+--echo #
+--echo # All three rows in one group. Now the result is cut as well, and
+--echo # values the user asked for really are missing, so the warning is
+--echo # given for that on top of the note.
+--echo #
+SELECT LENGTH(GROUP_CONCAT(t ORDER BY t)) AS len FROM t1;
+
+--echo #
+--echo # Nothing is cut, so nothing is said.
+--echo #
+SET SESSION group_concat_max_len= DEFAULT;
+SELECT grp, LENGTH(GROUP_CONCAT(t ORDER BY t)) AS len FROM t1 GROUP BY grp;
+
+DROP TABLE t1;
diff --git a/mysql-test/main/gconcat_distinct_rewalk.result b/mysql-test/main/gconcat_distinct_rewalk.result
new file mode 100644
index 0000000000000..f960831152ffc
--- /dev/null
+++ b/mysql-test/main/gconcat_distinct_rewalk.result
@@ -0,0 +1,93 @@
+#
+# A plain aggregate, asked once and then asked twice. A HAVING
+# clause on the alias is the shortest statement that asks twice.
+#
+CREATE TABLE t1 (g INT, a VARCHAR(10));
+INSERT INTO t1 VALUES (1,'a'),(1,'b'),(1,'c'),(1,'d'),(2,'e'),(2,'f');
+SELECT GROUP_CONCAT(a ORDER BY a LIMIT 2 OFFSET 4) AS v FROM t1 WHERE g = 1;
+v
+
+SELECT GROUP_CONCAT(a ORDER BY a LIMIT 2 OFFSET 4) AS v FROM t1 WHERE g = 1
+HAVING v LIKE '%';
+v
+
+#
+# Two conditions ask for it a third time.
+#
+SELECT GROUP_CONCAT(a ORDER BY a LIMIT 2 OFFSET 4) AS v FROM t1 WHERE g = 1
+HAVING v LIKE '%' AND v NOT LIKE 'zz%';
+v
+
+#
+# DISTINCT reaches the walk by the other route.
+#
+SELECT GROUP_CONCAT(DISTINCT a LIMIT 2 OFFSET 4) AS v FROM t1 WHERE g = 1
+HAVING v LIKE '%';
+v
+
+#
+# One group spends the offset exactly and the other does not, so
+# both answers show up in the same statement.
+#
+SELECT g, GROUP_CONCAT(a ORDER BY a LIMIT 2 OFFSET 4) AS v FROM t1
+GROUP BY g HAVING v LIKE '%' ORDER BY g;
+g	v
+1	
+2	
+#
+# An offset that stops inside the group is not affected, the walk
+# having written a row, and neither is a group with no LIMIT.
+#
+SELECT GROUP_CONCAT(a ORDER BY a LIMIT 2 OFFSET 1) AS v FROM t1 WHERE g = 1
+HAVING v LIKE '%';
+v
+b,c
+SELECT GROUP_CONCAT(a ORDER BY a) AS v FROM t1 WHERE g = 1
+HAVING v LIKE '%';
+v
+a,b,c,d
+DROP TABLE t1;
+#
+# The same replay against a duplicate filter that has spilled to
+# disk, where the second walk is not merely wrong but unsupported.
+#
+CREATE TABLE t2 (g INT, a VARCHAR(100));
+INSERT INTO t2 SELECT seq MOD 4, LPAD(seq, 100, '0') FROM seq_1_to_2000;
+SELECT COUNT(*) AS rows_in_table, COUNT(DISTINCT a) AS distinct_values FROM t2;
+rows_in_table	distinct_values
+2000	2000
+SET @@tmp_memory_table_size=0;
+SELECT g, GROUP_CONCAT(DISTINCT a LIMIT 5 OFFSET 1000) AS gc
+FROM t2 GROUP BY g HAVING gc <> 'x';
+g	gc
+0	
+1	
+2	
+3	
+SELECT g, GROUP_CONCAT(DISTINCT a ORDER BY a LIMIT 5 OFFSET 1000) AS gc
+FROM t2 GROUP BY g HAVING gc <> 'x';
+g	gc
+0	
+1	
+2	
+3	
+#
+# An offset that skips only part of the group is not affected, and
+# the answer must not depend on whether the filter spilled.
+#
+SELECT g, LENGTH(GROUP_CONCAT(DISTINCT a ORDER BY a LIMIT 2 OFFSET 3)) AS gc_len
+FROM t2 GROUP BY g HAVING gc_len > 0;
+g	gc_len
+0	201
+1	201
+2	201
+3	201
+SET @@tmp_memory_table_size=DEFAULT;
+SELECT g, LENGTH(GROUP_CONCAT(DISTINCT a ORDER BY a LIMIT 2 OFFSET 3)) AS gc_len
+FROM t2 GROUP BY g HAVING gc_len > 0;
+g	gc_len
+0	201
+1	201
+2	201
+3	201
+DROP TABLE t2;
diff --git a/mysql-test/main/gconcat_distinct_rewalk.test b/mysql-test/main/gconcat_distinct_rewalk.test
new file mode 100644
index 0000000000000..f0d5b63da8ee4
--- /dev/null
+++ b/mysql-test/main/gconcat_distinct_rewalk.test
@@ -0,0 +1,75 @@
+#
+# Nothing says how many times a statement asks for the result of a group,
+# and the answer must not depend on it. When an OFFSET skips every row of
+# the group, the walk writes nothing, and the aggregate must still record
+# that it has run. Otherwise the next caller walks again with the offset
+# already spent, which replays the group into a buffer that was handed
+# over once already, and, once the duplicate filter has spilled to disk,
+# walks a Unique that can only be walked once.
+#
+--source include/have_sequence.inc
+
+--echo #
+--echo # A plain aggregate, asked once and then asked twice. A HAVING
+--echo # clause on the alias is the shortest statement that asks twice.
+--echo #
+CREATE TABLE t1 (g INT, a VARCHAR(10));
+INSERT INTO t1 VALUES (1,'a'),(1,'b'),(1,'c'),(1,'d'),(2,'e'),(2,'f');
+
+SELECT GROUP_CONCAT(a ORDER BY a LIMIT 2 OFFSET 4) AS v FROM t1 WHERE g = 1;
+SELECT GROUP_CONCAT(a ORDER BY a LIMIT 2 OFFSET 4) AS v FROM t1 WHERE g = 1
+  HAVING v LIKE '%';
+
+--echo #
+--echo # Two conditions ask for it a third time.
+--echo #
+SELECT GROUP_CONCAT(a ORDER BY a LIMIT 2 OFFSET 4) AS v FROM t1 WHERE g = 1
+  HAVING v LIKE '%' AND v NOT LIKE 'zz%';
+
+--echo #
+--echo # DISTINCT reaches the walk by the other route.
+--echo #
+SELECT GROUP_CONCAT(DISTINCT a LIMIT 2 OFFSET 4) AS v FROM t1 WHERE g = 1
+  HAVING v LIKE '%';
+
+--echo #
+--echo # One group spends the offset exactly and the other does not, so
+--echo # both answers show up in the same statement.
+--echo #
+SELECT g, GROUP_CONCAT(a ORDER BY a LIMIT 2 OFFSET 4) AS v FROM t1
+  GROUP BY g HAVING v LIKE '%' ORDER BY g;
+
+--echo #
+--echo # An offset that stops inside the group is not affected, the walk
+--echo # having written a row, and neither is a group with no LIMIT.
+--echo #
+SELECT GROUP_CONCAT(a ORDER BY a LIMIT 2 OFFSET 1) AS v FROM t1 WHERE g = 1
+  HAVING v LIKE '%';
+SELECT GROUP_CONCAT(a ORDER BY a) AS v FROM t1 WHERE g = 1
+  HAVING v LIKE '%';
+DROP TABLE t1;
+
+--echo #
+--echo # The same replay against a duplicate filter that has spilled to
+--echo # disk, where the second walk is not merely wrong but unsupported.
+--echo #
+CREATE TABLE t2 (g INT, a VARCHAR(100));
+INSERT INTO t2 SELECT seq MOD 4, LPAD(seq, 100, '0') FROM seq_1_to_2000;
+SELECT COUNT(*) AS rows_in_table, COUNT(DISTINCT a) AS distinct_values FROM t2;
+
+SET @@tmp_memory_table_size=0;
+SELECT g, GROUP_CONCAT(DISTINCT a LIMIT 5 OFFSET 1000) AS gc
+  FROM t2 GROUP BY g HAVING gc <> 'x';
+SELECT g, GROUP_CONCAT(DISTINCT a ORDER BY a LIMIT 5 OFFSET 1000) AS gc
+  FROM t2 GROUP BY g HAVING gc <> 'x';
+
+--echo #
+--echo # An offset that skips only part of the group is not affected, and
+--echo # the answer must not depend on whether the filter spilled.
+--echo #
+SELECT g, LENGTH(GROUP_CONCAT(DISTINCT a ORDER BY a LIMIT 2 OFFSET 3)) AS gc_len
+  FROM t2 GROUP BY g HAVING gc_len > 0;
+SET @@tmp_memory_table_size=DEFAULT;
+SELECT g, LENGTH(GROUP_CONCAT(DISTINCT a ORDER BY a LIMIT 2 OFFSET 3)) AS gc_len
+  FROM t2 GROUP BY g HAVING gc_len > 0;
+DROP TABLE t2;
diff --git a/mysql-test/main/gconcat_distinct_spill.result b/mysql-test/main/gconcat_distinct_spill.result
new file mode 100644
index 0000000000000..f93239aba1dbe
--- /dev/null
+++ b/mysql-test/main/gconcat_distinct_spill.result
@@ -0,0 +1,132 @@
+#
+# Each block records the answer computed with memory to spare, then
+# recomputes it with the duplicate filter starved, and compares.
+#
+CREATE TABLE t1 (pk INT AUTO_INCREMENT PRIMARY KEY, a VARCHAR(100) NOT NULL);
+INSERT INTO t1 (a) SELECT LPAD(seq, 4, '0') FROM seq_1_to_50;
+INSERT INTO t1 (a) SELECT a FROM t1 ORDER BY pk;
+SELECT COUNT(*) AS rows_in_table, COUNT(DISTINCT a) AS distinct_values FROM t1;
+rows_in_table	distinct_values
+100	50
+SELECT GROUP_CONCAT(DISTINCT a) INTO @gc FROM t1;
+SELECT GROUP_CONCAT(DISTINCT a ORDER BY a) INTO @gc_order FROM t1;
+SELECT JSON_ARRAYAGG(DISTINCT a) INTO @ja FROM t1;
+SELECT JSON_ARRAYAGG(DISTINCT a ORDER BY a) INTO @ja_order FROM t1;
+SET @@tmp_memory_table_size=0;
+SELECT GROUP_CONCAT(DISTINCT a) = @gc AS gc_unchanged FROM t1;
+gc_unchanged
+1
+SELECT GROUP_CONCAT(DISTINCT a ORDER BY a) = @gc_order AS gc_order_unchanged FROM t1;
+gc_order_unchanged
+1
+SELECT JSON_ARRAYAGG(DISTINCT a) = @ja AS ja_unchanged FROM t1;
+ja_unchanged
+1
+SELECT JSON_ARRAYAGG(DISTINCT a ORDER BY a) = @ja_order AS ja_order_unchanged FROM t1;
+ja_order_unchanged
+1
+SET @@tmp_memory_table_size=DEFAULT;
+DROP TABLE t1;
+#
+# Values wide enough that the filter flushes on nearly every row.
+# Here the ORDER BY case used to return a single value out of 30.
+#
+CREATE TABLE t2 (a VARCHAR(2000)) AS
+SELECT CONCAT(seq, REPEAT('.', 1990)) AS a FROM seq_1_to_30;
+SELECT COUNT(*) AS rows_in_table, COUNT(DISTINCT a) AS distinct_values FROM t2;
+rows_in_table	distinct_values
+30	30
+SELECT GROUP_CONCAT(DISTINCT a) INTO @gc FROM t2;
+SELECT GROUP_CONCAT(DISTINCT a ORDER BY a) INTO @gc_order FROM t2;
+SET @@tmp_memory_table_size=1000, @@max_heap_table_size=1000;
+Warnings:
+Warning	1292	Truncated incorrect tmp_memory_table_size value: '1000'
+Warning	1292	Truncated incorrect max_heap_table_size value: '1000'
+SELECT GROUP_CONCAT(DISTINCT a) = @gc AS gc_unchanged FROM t2;
+gc_unchanged
+1
+SELECT GROUP_CONCAT(DISTINCT a ORDER BY a) = @gc_order AS gc_order_unchanged FROM t2;
+gc_order_unchanged
+1
+SET @@tmp_memory_table_size=DEFAULT, @@max_heap_table_size=DEFAULT;
+DROP TABLE t2;
+#
+# ORDER BY does not order the rows that tie on the ordering
+# expression. Which of them comes first is not specified, but it
+# must not depend on the memory available or on the order the rows
+# are read in.
+#
+CREATE TABLE t3 (a BIT(2), b VARCHAR(10), c BIT);
+INSERT INTO t3 VALUES (1, 'a', 0), (0, 'b', 1), (0, 'c', 0), (3, 'd', 1),
+(1, 'e', 1), (3, 'f', 1), (0, 'g', 1);
+SELECT GROUP_CONCAT(DISTINCT a, c ORDER BY a) AS at_default FROM t3;
+at_default
+01,00,11,10,31
+SET @@tmp_memory_table_size=0;
+SELECT GROUP_CONCAT(DISTINCT a, c ORDER BY a) AS at_zero FROM t3;
+at_zero
+01,00,11,10,31
+SET @@tmp_memory_table_size=DEFAULT;
+DELETE FROM t3;
+INSERT INTO t3 VALUES (0, 'c', 0), (0, 'b', 1), (1, 'a', 0), (3, 'd', 1),
+(1, 'e', 1), (3, 'f', 1), (0, 'g', 1);
+SELECT GROUP_CONCAT(DISTINCT a, c ORDER BY a) AS reversed_scan_order FROM t3;
+reversed_scan_order
+01,00,11,10,31
+DROP TABLE t3;
+#
+# The sort tree can overflow too. Starving it makes repack_tree()
+# cut rows out of the group, which is expected, but the rows that
+# do come back must still be deduplicated and still be in order.
+#
+CREATE TABLE t4 (pk INT AUTO_INCREMENT PRIMARY KEY, a VARCHAR(20));
+INSERT INTO t4 (a) SELECT LPAD(seq, 6, '0') FROM seq_1_to_200;
+INSERT INTO t4 (a) SELECT a FROM t4 ORDER BY pk;
+SET @@tmp_memory_table_size=0, @@group_concat_max_len=4000;
+SELECT GROUP_CONCAT(DISTINCT a ORDER BY a) INTO @gc FROM t4;
+Warnings:
+Warning	1260	Row # was cut by group_concat()
+SELECT JSON_ARRAYAGG(DISTINCT a ORDER BY a) INTO @ja FROM t4;
+Warnings:
+Warning	1260	Row # was cut by json_arrayagg()
+SET @@tmp_memory_table_size=DEFAULT, @@group_concat_max_len=DEFAULT;
+SELECT COUNT(*) = COUNT(DISTINCT val) AS gc_no_duplicates,
+GROUP_CONCAT(val ORDER BY val) = @gc AS gc_ascending
+FROM (SELECT SUBSTRING_INDEX(SUBSTRING_INDEX(@gc, ',', seq), ',', -1) AS val
+FROM seq_1_to_500
+WHERE seq <= 1 + LENGTH(@gc) - LENGTH(REPLACE(@gc, ',', ''))) split;
+gc_no_duplicates	gc_ascending
+1	1
+SELECT JSON_VALID(@ja) AS ja_valid,
+COUNT(*) = COUNT(DISTINCT val) AS ja_no_duplicates,
+JSON_ARRAYAGG(val ORDER BY val) = @ja AS ja_ascending
+FROM (SELECT JSON_UNQUOTE(JSON_EXTRACT(@ja, CONCAT('$[', seq - 1, ']'))) AS val
+FROM seq_1_to_500 WHERE seq <= JSON_LENGTH(@ja)) split;
+ja_valid	ja_no_duplicates	ja_ascending
+1	1	1
+DROP TABLE t4;
+#
+# Running out of LIMIT stops the walk of the duplicate filter early,
+# but nothing is lost by it. It must not be reported as a cut value.
+#
+CREATE TABLE t5 (a VARCHAR(100));
+INSERT INTO t5 SELECT LPAD(seq MOD 200, 100, '0') FROM seq_1_to_600;
+SELECT COUNT(DISTINCT a) AS distinct_values FROM t5;
+distinct_values
+200
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a LIMIT 5)) AS gc_len FROM t5;
+gc_len
+504
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a ORDER BY a LIMIT 5)) AS gc_order_len FROM t5;
+gc_order_len
+504
+SET @@tmp_memory_table_size=0;
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a LIMIT 5)) AS gc_len_spilled FROM t5;
+gc_len_spilled
+504
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a ORDER BY a LIMIT 5)) AS gc_order_len_spilled
+FROM t5;
+gc_order_len_spilled
+504
+SET @@tmp_memory_table_size=DEFAULT;
+DROP TABLE t5;
diff --git a/mysql-test/main/gconcat_distinct_spill.test b/mysql-test/main/gconcat_distinct_spill.test
new file mode 100644
index 0000000000000..c9d0743dbfbf7
--- /dev/null
+++ b/mysql-test/main/gconcat_distinct_spill.test
@@ -0,0 +1,115 @@
+#
+# GROUP_CONCAT(DISTINCT ...) and JSON_ARRAYAGG(DISTINCT ...) filter
+# duplicates with a Unique object, which flushes to disk when it runs out
+# of memory. The answer must not depend on whether that flush happened.
+#
+--source include/have_sequence.inc
+
+--echo #
+--echo # Each block records the answer computed with memory to spare, then
+--echo # recomputes it with the duplicate filter starved, and compares.
+--echo #
+
+CREATE TABLE t1 (pk INT AUTO_INCREMENT PRIMARY KEY, a VARCHAR(100) NOT NULL);
+INSERT INTO t1 (a) SELECT LPAD(seq, 4, '0') FROM seq_1_to_50;
+INSERT INTO t1 (a) SELECT a FROM t1 ORDER BY pk;
+SELECT COUNT(*) AS rows_in_table, COUNT(DISTINCT a) AS distinct_values FROM t1;
+
+SELECT GROUP_CONCAT(DISTINCT a) INTO @gc FROM t1;
+SELECT GROUP_CONCAT(DISTINCT a ORDER BY a) INTO @gc_order FROM t1;
+SELECT JSON_ARRAYAGG(DISTINCT a) INTO @ja FROM t1;
+SELECT JSON_ARRAYAGG(DISTINCT a ORDER BY a) INTO @ja_order FROM t1;
+
+SET @@tmp_memory_table_size=0;
+SELECT GROUP_CONCAT(DISTINCT a) = @gc AS gc_unchanged FROM t1;
+SELECT GROUP_CONCAT(DISTINCT a ORDER BY a) = @gc_order AS gc_order_unchanged FROM t1;
+SELECT JSON_ARRAYAGG(DISTINCT a) = @ja AS ja_unchanged FROM t1;
+SELECT JSON_ARRAYAGG(DISTINCT a ORDER BY a) = @ja_order AS ja_order_unchanged FROM t1;
+SET @@tmp_memory_table_size=DEFAULT;
+DROP TABLE t1;
+
+--echo #
+--echo # Values wide enough that the filter flushes on nearly every row.
+--echo # Here the ORDER BY case used to return a single value out of 30.
+--echo #
+CREATE TABLE t2 (a VARCHAR(2000)) AS
+  SELECT CONCAT(seq, REPEAT('.', 1990)) AS a FROM seq_1_to_30;
+SELECT COUNT(*) AS rows_in_table, COUNT(DISTINCT a) AS distinct_values FROM t2;
+
+SELECT GROUP_CONCAT(DISTINCT a) INTO @gc FROM t2;
+SELECT GROUP_CONCAT(DISTINCT a ORDER BY a) INTO @gc_order FROM t2;
+
+SET @@tmp_memory_table_size=1000, @@max_heap_table_size=1000;
+SELECT GROUP_CONCAT(DISTINCT a) = @gc AS gc_unchanged FROM t2;
+SELECT GROUP_CONCAT(DISTINCT a ORDER BY a) = @gc_order AS gc_order_unchanged FROM t2;
+SET @@tmp_memory_table_size=DEFAULT, @@max_heap_table_size=DEFAULT;
+DROP TABLE t2;
+
+--echo #
+--echo # ORDER BY does not order the rows that tie on the ordering
+--echo # expression. Which of them comes first is not specified, but it
+--echo # must not depend on the memory available or on the order the rows
+--echo # are read in.
+--echo #
+CREATE TABLE t3 (a BIT(2), b VARCHAR(10), c BIT);
+INSERT INTO t3 VALUES (1, 'a', 0), (0, 'b', 1), (0, 'c', 0), (3, 'd', 1),
+(1, 'e', 1), (3, 'f', 1), (0, 'g', 1);
+SELECT GROUP_CONCAT(DISTINCT a, c ORDER BY a) AS at_default FROM t3;
+SET @@tmp_memory_table_size=0;
+SELECT GROUP_CONCAT(DISTINCT a, c ORDER BY a) AS at_zero FROM t3;
+SET @@tmp_memory_table_size=DEFAULT;
+
+DELETE FROM t3;
+INSERT INTO t3 VALUES (0, 'c', 0), (0, 'b', 1), (1, 'a', 0), (3, 'd', 1),
+(1, 'e', 1), (3, 'f', 1), (0, 'g', 1);
+SELECT GROUP_CONCAT(DISTINCT a, c ORDER BY a) AS reversed_scan_order FROM t3;
+DROP TABLE t3;
+
+--echo #
+--echo # The sort tree can overflow too. Starving it makes repack_tree()
+--echo # cut rows out of the group, which is expected, but the rows that
+--echo # do come back must still be deduplicated and still be in order.
+--echo #
+CREATE TABLE t4 (pk INT AUTO_INCREMENT PRIMARY KEY, a VARCHAR(20));
+INSERT INTO t4 (a) SELECT LPAD(seq, 6, '0') FROM seq_1_to_200;
+INSERT INTO t4 (a) SELECT a FROM t4 ORDER BY pk;
+
+SET @@tmp_memory_table_size=0, @@group_concat_max_len=4000;
+# Which row the tree overflows on depends on the size of a TREE_ELEMENT,
+# so the number in the warning differs between 32-bit and 64-bit builds.
+--replace_regex /Row [0-9]+ was cut/Row # was cut/
+SELECT GROUP_CONCAT(DISTINCT a ORDER BY a) INTO @gc FROM t4;
+--replace_regex /Row [0-9]+ was cut/Row # was cut/
+SELECT JSON_ARRAYAGG(DISTINCT a ORDER BY a) INTO @ja FROM t4;
+SET @@tmp_memory_table_size=DEFAULT, @@group_concat_max_len=DEFAULT;
+
+SELECT COUNT(*) = COUNT(DISTINCT val) AS gc_no_duplicates,
+       GROUP_CONCAT(val ORDER BY val) = @gc AS gc_ascending
+FROM (SELECT SUBSTRING_INDEX(SUBSTRING_INDEX(@gc, ',', seq), ',', -1) AS val
+      FROM seq_1_to_500
+      WHERE seq <= 1 + LENGTH(@gc) - LENGTH(REPLACE(@gc, ',', ''))) split;
+
+SELECT JSON_VALID(@ja) AS ja_valid,
+       COUNT(*) = COUNT(DISTINCT val) AS ja_no_duplicates,
+       JSON_ARRAYAGG(val ORDER BY val) = @ja AS ja_ascending
+FROM (SELECT JSON_UNQUOTE(JSON_EXTRACT(@ja, CONCAT('$[', seq - 1, ']'))) AS val
+      FROM seq_1_to_500 WHERE seq <= JSON_LENGTH(@ja)) split;
+DROP TABLE t4;
+
+--echo #
+--echo # Running out of LIMIT stops the walk of the duplicate filter early,
+--echo # but nothing is lost by it. It must not be reported as a cut value.
+--echo #
+CREATE TABLE t5 (a VARCHAR(100));
+INSERT INTO t5 SELECT LPAD(seq MOD 200, 100, '0') FROM seq_1_to_600;
+SELECT COUNT(DISTINCT a) AS distinct_values FROM t5;
+
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a LIMIT 5)) AS gc_len FROM t5;
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a ORDER BY a LIMIT 5)) AS gc_order_len FROM t5;
+
+SET @@tmp_memory_table_size=0;
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a LIMIT 5)) AS gc_len_spilled FROM t5;
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a ORDER BY a LIMIT 5)) AS gc_order_len_spilled
+FROM t5;
+SET @@tmp_memory_table_size=DEFAULT;
+DROP TABLE t5;
diff --git a/mysql-test/main/gconcat_distinct_walk_fail.result b/mysql-test/main/gconcat_distinct_walk_fail.result
new file mode 100644
index 0000000000000..36a7aaa1c19ab
--- /dev/null
+++ b/mysql-test/main/gconcat_distinct_walk_fail.result
@@ -0,0 +1,49 @@
+CREATE TABLE t1 (a VARCHAR(100));
+INSERT INTO t1 SELECT LPAD(seq MOD 200, 100, '0') FROM seq_1_to_600;
+#
+# Starve the duplicate filter so that it spills and the walk has to
+# merge, then make the merging walk fail.
+#
+SET @@tmp_memory_table_size=0;
+SET SESSION debug_dbug='+d,unique_walk_merge_fail';
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a)) AS gc_len FROM t1;
+gc_len
+0
+Warnings:
+Warning	1260	Row 0 was cut by group_concat()
+SELECT JSON_LENGTH(JSON_ARRAYAGG(DISTINCT a)) AS ja_len FROM t1;
+ja_len
+0
+Warnings:
+Warning	1260	Row 0 was cut by json_arrayagg()
+SET SESSION debug_dbug=DEFAULT;
+SET @@tmp_memory_table_size=DEFAULT;
+#
+# A walk that failed for a reason it reported itself, an out of
+# memory one for instance, needs no cut value warning: the user has
+# already been told, and the statement is failing anyway.
+#
+SET @@tmp_memory_table_size=0;
+SET SESSION debug_dbug='+d,unique_walk_merge_error';
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a)) AS gc_len FROM t1;
+ERROR HY000: Out of memory.
+SHOW COUNT(*) WARNINGS;
+@@session.warning_count
+1
+SELECT JSON_LENGTH(JSON_ARRAYAGG(DISTINCT a)) AS ja_len FROM t1;
+ERROR HY000: Out of memory.
+SHOW COUNT(*) WARNINGS;
+@@session.warning_count
+1
+SET SESSION debug_dbug=DEFAULT;
+SET @@tmp_memory_table_size=DEFAULT;
+#
+# Without the injected failure the same queries are complete and quiet.
+#
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a)) AS gc_len FROM t1;
+gc_len
+20199
+SELECT JSON_LENGTH(JSON_ARRAYAGG(DISTINCT a)) AS ja_len FROM t1;
+ja_len
+200
+DROP TABLE t1;
diff --git a/mysql-test/main/gconcat_distinct_walk_fail.test b/mysql-test/main/gconcat_distinct_walk_fail.test
new file mode 100644
index 0000000000000..cab5394069f9c
--- /dev/null
+++ b/mysql-test/main/gconcat_distinct_walk_fail.test
@@ -0,0 +1,46 @@
+#
+# GROUP_CONCAT(DISTINCT ...) builds its result by walking the Unique that
+# filtered the duplicates. That walk merges back what the Unique spilled to
+# disk, so it can fail on its own, without the callback that appends the
+# rows getting a chance to say anything. The result is then short and the
+# user has to be told about it.
+#
+--source include/have_debug.inc
+--source include/have_sequence.inc
+
+CREATE TABLE t1 (a VARCHAR(100));
+INSERT INTO t1 SELECT LPAD(seq MOD 200, 100, '0') FROM seq_1_to_600;
+
+--echo #
+--echo # Starve the duplicate filter so that it spills and the walk has to
+--echo # merge, then make the merging walk fail.
+--echo #
+SET @@tmp_memory_table_size=0;
+SET SESSION debug_dbug='+d,unique_walk_merge_fail';
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a)) AS gc_len FROM t1;
+SELECT JSON_LENGTH(JSON_ARRAYAGG(DISTINCT a)) AS ja_len FROM t1;
+SET SESSION debug_dbug=DEFAULT;
+SET @@tmp_memory_table_size=DEFAULT;
+
+--echo #
+--echo # A walk that failed for a reason it reported itself, an out of
+--echo # memory one for instance, needs no cut value warning: the user has
+--echo # already been told, and the statement is failing anyway.
+--echo #
+SET @@tmp_memory_table_size=0;
+SET SESSION debug_dbug='+d,unique_walk_merge_error';
+--error ER_OUT_OF_RESOURCES
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a)) AS gc_len FROM t1;
+SHOW COUNT(*) WARNINGS;
+--error ER_OUT_OF_RESOURCES
+SELECT JSON_LENGTH(JSON_ARRAYAGG(DISTINCT a)) AS ja_len FROM t1;
+SHOW COUNT(*) WARNINGS;
+SET SESSION debug_dbug=DEFAULT;
+SET @@tmp_memory_table_size=DEFAULT;
+
+--echo #
+--echo # Without the injected failure the same queries are complete and quiet.
+--echo #
+SELECT LENGTH(GROUP_CONCAT(DISTINCT a)) AS gc_len FROM t1;
+SELECT JSON_LENGTH(JSON_ARRAYAGG(DISTINCT a)) AS ja_len FROM t1;
+DROP TABLE t1;
diff --git a/mysql-test/main/gconcat_warn.result b/mysql-test/main/gconcat_warn.result
index b45a3739bcddd..a3932b2ce61ec 100644
--- a/mysql-test/main/gconcat_warn.result
+++ b/mysql-test/main/gconcat_warn.result
@@ -33,6 +33,7 @@ SELECT grp, GROUP_CONCAT(t ORDER BY t) FROM t1 GROUP BY grp;
 SHOW WARNINGS;
 Level	Code	Message
 Warning	1260	Row 2 was cut by group_concat()
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
 Warning	1260	Row 4 was cut by group_concat()
 Warning	1260	Row 6 was cut by group_concat()
 #
@@ -47,7 +48,9 @@ GROUP BY grp HAVING GROUP_CONCAT(t ORDER BY t) <> '';
 SHOW WARNINGS;
 Level	Code	Message
 Warning	1260	Row 2 was cut by group_concat()
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
 Warning	1260	Row 2 was cut by group_concat()
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
 Warning	1260	Row 4 was cut by group_concat()
 Warning	1260	Row 4 was cut by group_concat()
 Warning	1260	Row 6 was cut by group_concat()
@@ -63,6 +66,7 @@ GROUP BY grp HAVING g <> '';
 SHOW WARNINGS;
 Level	Code	Message
 Warning	1260	Row 2 was cut by group_concat()
+Note	4265	Some values were cut while processing group_concat(). Increase group_concat_max_len if you want to avoid the cut
 Warning	1260	Row 4 was cut by group_concat()
 Warning	1260	Row 6 was cut by group_concat()
 CREATE TABLE t2 (a INT);
diff --git a/mysql-test/suite/heap/blob_delete_reinsert_ceiling.result b/mysql-test/suite/heap/blob_delete_reinsert_ceiling.result
new file mode 100644
index 0000000000000..50a47399092f1
--- /dev/null
+++ b/mysql-test/suite/heap/blob_delete_reinsert_ceiling.result
@@ -0,0 +1,34 @@
+SET SESSION max_heap_table_size= 32768;
+CREATE TABLE t1 (
+pk INT PRIMARY KEY,
+b TEXT,
+c VARCHAR(8),
+UNIQUE(c)
+) ENGINE=MEMORY;
+INSERT INTO t1 VALUES (1,'one','foo'),(2,'two','bar');
+#
+# One 16K leaf for the records and one per hash index puts the
+# table over its own 32768-byte ceiling while holding two rows.
+#
+SELECT DATA_LENGTH, INDEX_LENGTH FROM INFORMATION_SCHEMA.TABLES
+WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH	INDEX_LENGTH
+16352	32704
+#
+# A targeted DELETE goes row by row, so the chains are parked and
+# then freed.  DELETE with no WHERE would be routed to
+# delete_all_rows(), which resets data_length and would hide this.
+#
+DELETE FROM t1 WHERE pk IN (1,2);
+INSERT INTO t1 VALUES (1,'one','foo'),(2,'two','bar');
+SELECT * FROM t1 ORDER BY pk;
+pk	b	c
+1	one	foo
+2	two	bar
+#
+# The ceiling still stops a table that genuinely needs more memory.
+#
+INSERT INTO t1 SELECT seq, REPEAT('x',255), CONCAT('c',seq)
+FROM seq_3_to_2000;
+ERROR HY000: The table 't1' is full
+DROP TABLE t1;
diff --git a/mysql-test/suite/heap/blob_delete_reinsert_ceiling.test b/mysql-test/suite/heap/blob_delete_reinsert_ceiling.test
new file mode 100644
index 0000000000000..1ce638af971ea
--- /dev/null
+++ b/mysql-test/suite/heap/blob_delete_reinsert_ceiling.test
@@ -0,0 +1,56 @@
+# A MEMORY table with a blob column must take back a row it has just
+# deleted, even when it is already over max_heap_table_size.
+#
+# Index leaves are not budgeted: hp_find_free_hash() allocates one
+# without consulting the ceiling, so a table with two hash indexes is
+# over a 32K ceiling from its first row onwards.  That is legal, because
+# the ceiling is only tested when the record cursor lands on a leaf
+# boundary, and the first row tests it before anything is allocated.
+#
+# Deleting a blob row parks its continuation chain; the next mutating
+# call frees the chain and hp_shrink_tail() walks the record cursor back
+# to a leaf boundary.  The leaf stays allocated and data_length keeps
+# counting it, so handing it back again adds no memory and must not be
+# refused.
+#
+# Block sizes are exact byte counts and depend on pointer width.
+
+--source include/have_64bit.inc
+--source include/have_sequence.inc
+
+SET SESSION max_heap_table_size= 32768;
+
+CREATE TABLE t1 (
+  pk INT PRIMARY KEY,
+  b TEXT,
+  c VARCHAR(8),
+  UNIQUE(c)
+) ENGINE=MEMORY;
+
+INSERT INTO t1 VALUES (1,'one','foo'),(2,'two','bar');
+
+--echo #
+--echo # One 16K leaf for the records and one per hash index puts the
+--echo # table over its own 32768-byte ceiling while holding two rows.
+--echo #
+SELECT DATA_LENGTH, INDEX_LENGTH FROM INFORMATION_SCHEMA.TABLES
+ WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+
+--echo #
+--echo # A targeted DELETE goes row by row, so the chains are parked and
+--echo # then freed.  DELETE with no WHERE would be routed to
+--echo # delete_all_rows(), which resets data_length and would hide this.
+--echo #
+DELETE FROM t1 WHERE pk IN (1,2);
+
+INSERT INTO t1 VALUES (1,'one','foo'),(2,'two','bar');
+SELECT * FROM t1 ORDER BY pk;
+
+--echo #
+--echo # The ceiling still stops a table that genuinely needs more memory.
+--echo #
+--error ER_RECORD_FILE_FULL
+INSERT INTO t1 SELECT seq, REPEAT('x',255), CONCAT('c',seq)
+  FROM seq_3_to_2000;
+
+DROP TABLE t1;
diff --git a/mysql-test/suite/heap/blob_replace_repl_ceiling.opt b/mysql-test/suite/heap/blob_replace_repl_ceiling.opt
new file mode 100644
index 0000000000000..8b9036f3e12dc
--- /dev/null
+++ b/mysql-test/suite/heap/blob_replace_repl_ceiling.opt
@@ -0,0 +1 @@
+--max-heap-table-size=32K
diff --git a/mysql-test/suite/heap/blob_replace_repl_ceiling.result b/mysql-test/suite/heap/blob_replace_repl_ceiling.result
new file mode 100644
index 0000000000000..d3f5a6b68da5f
--- /dev/null
+++ b/mysql-test/suite/heap/blob_replace_repl_ceiling.result
@@ -0,0 +1,21 @@
+include/master-slave.inc
+[connection master]
+CREATE TABLE t1 (
+pk INT PRIMARY KEY,
+b TEXT,
+c VARCHAR(8),
+UNIQUE(c)
+) ENGINE=MEMORY;
+INSERT INTO t1 VALUES (1,'one','foo');
+REPLACE INTO t1 SELECT * FROM t1;
+connection slave;
+# Slave applied the REPLACE
+SELECT * FROM t1;
+pk	b	c
+1	one	foo
+connection master;
+SELECT * FROM t1;
+pk	b	c
+1	one	foo
+DROP TABLE t1;
+include/rpl_end.inc
diff --git a/mysql-test/suite/heap/blob_replace_repl_ceiling.test b/mysql-test/suite/heap/blob_replace_repl_ceiling.test
new file mode 100644
index 0000000000000..550d0395bfdb3
--- /dev/null
+++ b/mysql-test/suite/heap/blob_replace_repl_ceiling.test
@@ -0,0 +1,46 @@
+# A REPLACE that the master accepts must also apply on the slave when
+# both run with a max_heap_table_size the table is already over.
+#
+# The two sides reach different engine states from the same statement.
+# REPLACE INTO t SELECT * FROM t feeds the row back out of the table, so
+# the record buffer's blob pointer still points into the chain that the
+# delete parked, the chain is adopted rather than freed, and the record
+# cursor never moves.  The slave builds the row from the replication
+# event buffer instead, so nothing points into the parked chain, it is
+# freed, and hp_shrink_tail() walks the cursor back onto a leaf
+# boundary.  Reclaiming that leaf adds no memory and must not be
+# refused, or the statement stops the SQL thread with
+# HA_ERR_RECORD_FILE_FULL.
+#
+# The table holds a single row on purpose.  REPLACE frees and rewrites
+# one row at a time, so with a second row present the freed records are
+# never the ones at the tail, the cursor stays put and the leaf boundary
+# is never reached.
+#
+# The ceiling is set in blob_replace_repl_ceiling.opt rather than by the
+# test, so that both servers hold the same value throughout.
+
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+
+CREATE TABLE t1 (
+  pk INT PRIMARY KEY,
+  b TEXT,
+  c VARCHAR(8),
+  UNIQUE(c)
+) ENGINE=MEMORY;
+
+INSERT INTO t1 VALUES (1,'one','foo');
+REPLACE INTO t1 SELECT * FROM t1;
+
+--sync_slave_with_master
+
+--echo # Slave applied the REPLACE
+SELECT * FROM t1;
+
+--connection master
+SELECT * FROM t1;
+
+DROP TABLE t1;
+
+--source include/rpl_end.inc
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index 08e83ee875c1b..3c25d9f2e4983 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -3828,6 +3828,25 @@ static void report_cut_value_error(THD *thd, uint row_count, const char *fname)
 }
 
 
+/*
+  Tell the user that a value was cut to group_concat_max_len on its way
+  into blob_storage.
+
+  This is only a note. Whether the answer would have been different with
+  a bigger limit is not known here: the result may well have been cut at
+  the same place anyway, in which case nothing was lost that the user
+  could have seen.
+*/
+
+static void report_cut_value_note(THD *thd, const char *fname)
+{
+  push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
+                      ER_CUT_VALUES_WHILE_PROCESSING,
+                      ER_THD(thd, ER_CUT_VALUES_WHILE_PROCESSING),
+                      fname, "group_concat_max_len");
+}
+
+
 void Item_func_group_concat::cut_max_length(String *result,
         uint old_length, uint max_length) const
 {
@@ -3858,29 +3877,27 @@ int dump_leaf_key(void* key_arg, element_count count __attribute__((unused)),
   uint max_length= table->in_use->gconcat_max_len();
   String tmp((char *)table->record[1], table->s->reclength,
              default_charset_info);
-  String tmp2;
   uchar *key= (uchar *) key_arg;
   String *result= &item->result;
   Item **arg= item->args, **arg_end= item->args + item->arg_count_field;
   uint old_length= result->length();
 
-  ulonglong *offset_limit= &item->copy_offset_limit;
   ulonglong *row_limit = &item->copy_row_limit;
   if (item->limit_clause && !(*row_limit))
   {
     item->result_finalized= true;
+    item->walk_stopped= true;
     return 1;
   }
-
-  tmp.length(0);
-
-  if (item->limit_clause && (*offset_limit))
+  if (item->copy_offset_limit)
   {
+    item->copy_offset_limit--;
     item->row_count++;
-    (*offset_limit)--;
     return 0;
   }
 
+  tmp.length(0);
+
   if (!item->result_finalized)
     item->result_finalized= true;
   else
@@ -3932,6 +3949,7 @@ int dump_leaf_key(void* key_arg, element_count count __attribute__((unused)),
       that the user gets one warning even if several things were cut.
     */
     item->result_cut= true;
+    item->walk_stopped= true;
     return 1;
   }
   return 0;
@@ -3961,7 +3979,9 @@ Item_func_group_concat(THD *thd, Name_resolution_context *context_arg,
    arg_count_field(select_list->elements),
    row_count(0),
    distinct(distinct_arg),
-   warning_for_row(FALSE), result_cut(FALSE), always_null(FALSE),
+   warning_for_row(FALSE), result_cut(FALSE), cut_note_given(FALSE),
+   walk_stopped(FALSE),
+   always_null(FALSE),
    force_copy_fields(0), row_limit(NULL),
    offset_limit(NULL), limit_clause(limit_clause),
    copy_offset_limit(0), copy_row_limit(0), original(0)
@@ -4029,6 +4049,8 @@ Item_func_group_concat::Item_func_group_concat(THD *thd,
   distinct(item->distinct),
   warning_for_row(item->warning_for_row),
   result_cut(item->result_cut),
+  cut_note_given(item->cut_note_given),
+  walk_stopped(item->walk_stopped),
   always_null(item->always_null),
   force_copy_fields(item->force_copy_fields),
   row_limit(item->row_limit), offset_limit(item->offset_limit),
@@ -4099,6 +4121,7 @@ void Item_func_group_concat::cleanup()
     row_count= 0;
     DBUG_ASSERT(tree == 0);
   }
+  cut_note_given= false;
   /*
     As the ORDER structures pointed to by the elements of the
     'order' array may be modified in find_order_in_list() called
@@ -4122,6 +4145,10 @@ Item *Item_func_group_concat::copy_or_same(THD* thd)
 }
 
 
+/*
+  Clear is called for the first element in a new group
+*/
+
 void Item_func_group_concat::clear()
 {
   result.length(0);
@@ -4130,8 +4157,10 @@ void Item_func_group_concat::clear()
   warning_for_row= FALSE;
   result_cut= FALSE;
   result_finalized= false;
+  copy_offset_limit= 0;
   if (offset_limit)
     copy_offset_limit= offset_limit->val_int();
+  /* copy_row_limit does not have to be zeroed if row_limit is not set */
   if (row_limit)
     copy_row_limit= row_limit->val_int();
   if (tree)
@@ -4319,6 +4348,60 @@ bool Item_func_group_concat::repack_tree(THD *thd)
 }
 
 
+/*
+  Insert one row into the ORDER BY tree, repacking it first if it has
+  grown past the memory we are allowed to use.
+
+  'key' is a temporary table record in the format produced by
+  get_record_pointer(). table->field[0] of that record holds the length
+  the row adds to the result, which is what repack_tree() adds up to
+  decide where to cut.
+
+  @return FALSE  row inserted
+  @return TRUE   out of memory
+*/
+
+bool Item_func_group_concat::insert_to_order_tree(uchar *key)
+{
+  DBUG_ASSERT(tree);
+  THD *thd= table->in_use;
+  /*
+    Repack when GCONCAT_TREE_REPACK_PARTS of the memory we may use is
+    gone. The rest is needed for the new tree that repack_tree()
+    allocates while the old one is still around.
+  */
+  if (tree->allocated >
+      max_tree_size / GCONCAT_TREE_PARTS * GCONCAT_TREE_REPACK_PARTS &&
+      tree->elements_in_tree > 1)
+    if (repack_tree(thd))
+      return TRUE;
+  /* check if there was enough memory to insert the row */
+  return !tree_insert(tree, key, 0, tree->custom_arg);
+}
+
+
+/*
+  Insert one deduplicated row into the ORDER BY tree.
+
+  Callback for Unique::walk(). The walk merges what unique_filter spilled
+  to disk back with what it still holds in memory, so it is the first
+  point at which the whole set of distinct rows is known. Every row it
+  visits belongs in the result.
+
+  @return 0  row inserted
+  @return 1  out of memory, which stops the walk
+*/
+
+int Item_func_group_concat::dump_leaf_key_to_tree(void *key_arg,
+                                                  element_count count
+                                                  __attribute__((unused)),
+                                                  void *item_arg)
+{
+  Item_func_group_concat *item= (Item_func_group_concat *) item_arg;
+  return item->insert_to_order_tree((uchar *) key_arg);
+}
+
+
 bool Item_func_group_concat::add(bool exclude_nulls)
 {
   if (always_null && exclude_nulls)
@@ -4360,7 +4443,15 @@ bool Item_func_group_concat::add(bool exclude_nulls)
   null_value= FALSE;
   bool row_eligible= TRUE;
 
-  if (distinct) 
+  /*
+    Store how much this row adds to the result in the record itself. With
+    DISTINCT the row does not reach the ORDER BY tree until val_str(), and
+    the record kept by unique_filter is all that is left of it by then.
+  */
+  if (tree)
+    table->field[0]->store(row_str_len, FALSE);
+
+  if (distinct)
   {
     /* Filter out duplicate rows. */
     uint count= unique_filter->elements_in_tree();
@@ -4368,25 +4459,17 @@ bool Item_func_group_concat::add(bool exclude_nulls)
     if (count == unique_filter->elements_in_tree())
       row_eligible= FALSE;
   }
-
-  TREE_ELEMENT *el= 0;                          // Only for safety
-  if (row_eligible && tree)
+  else
   {
-    THD *thd= table->in_use;
-    table->field[0]->store(row_str_len, FALSE);
     /*
-      Repack when GCONCAT_TREE_REPACK_PARTS of the memory we may use is
-      gone. The rest is needed for the new tree that repack_tree()
-      allocates while the old one is still around.
+      row_eligible only reflects the part of unique_filter that is
+      currently in memory, so as soon as unique_filter starts flushing
+      to disk it stops telling us whether a row is a
+      duplicate. val_str() fills the tree instead, from the
+      unique_filter walk that merges everything back together.
     */
-    if (tree->allocated >
-        max_tree_size / GCONCAT_TREE_PARTS * GCONCAT_TREE_REPACK_PARTS &&
-        tree->elements_in_tree > 1)
-      if (repack_tree(thd))
-        return 1;
-    el= tree_insert(tree, get_record_pointer(), 0, tree->custom_arg);
-    /* check if there was enough memory to insert the row */
-    if (!el)
+    if (row_eligible && tree &&
+        insert_to_order_tree(get_record_pointer()))
       return 1;
   }
 
@@ -4618,29 +4701,98 @@ String* Item_func_group_concat::val_str(String* str)
 
   if (!result_finalized) // Result yet to be written.
   {
-    if (tree != NULL) // order by
-      tree_walk(tree, &dump_leaf_key, this, left_root_right);
-    else if (distinct) // distinct (and no order by).
-      unique_filter->walk(table, &dump_leaf_key, this);
+    if (tree)                                   // Order by
+    {
+      if (distinct)                             // distinct and order by
+      {
+        /*
+          Sort the distinct rows now. add() could not do it, as a row is
+          only known not to be a duplicate once unique_filter has merged
+          everything it flushed to disk, which the walk below does.
+        */
+        if (unique_filter->walk(table, &dump_leaf_key_to_tree, this))
+        {
+          /*
+            Out of memory; mysys has reported it. The tree holds only part
+            of the group, so tell the user that the result was cut instead
+            of returning a short one silently.
+          */
+          result_cut= TRUE;
+        }
+        tree_walk(tree, &dump_leaf_key, this, left_root_right);
+      }
+      else                                      // Order by, no distinct
+      {
+        tree_walk(tree, &dump_leaf_key, this, left_root_right);
+      }
+    }
+    else if (distinct)                          // distinct (and no order by)
+    {
+      /*
+        walk() returns non-zero both when dump_leaf_key() stopped it and
+        when the walk itself failed. dump_leaf_key() reports what it did:
+        it sets result_cut when it cut the result, and it stops without
+        losing anything when the LIMIT is used up.
+
+        A walk that failed mostly reports itself: the merge buffer is
+        allocated with MY_WME and the spill file is opened with MY_WME,
+        so running out of memory or failing to read raises an error. Only
+        the guard at the top of merge_walk(), which refuses a merge
+        buffer too small to hold one key per chunk, returns quietly. Ask
+        for the cut value warning in that case alone. Where an error was
+        raised the user has already been told and the statement is
+        failing, so a warning about the length of a result nobody will
+        see would only be noise.
+      */
+      walk_stopped= FALSE;
+      if (unique_filter->walk(table, &dump_leaf_key, this) && !walk_stopped &&
+          !current_thd->is_error())
+        result_cut= TRUE;
+    }
     else if (row_limit && copy_row_limit == (ulonglong)row_limit->val_int())
+    {
+      /* No rows copied; Empty result */
       return &result;
+    }
     else
-      DBUG_ASSERT(false); // Can't happen
+      DBUG_ASSERT(false); // Can't happen. If it happens, no wrong result
+    /*
+      dump_leaf_key() sets this when it writes a row, but a group can end
+      without one: an OFFSET can eat every row, and a walk that failed
+      hands over nothing. val_str() can be called again for the same
+      group, and unique_filter must not be walked twice - Unique::walk()
+      leaves the object undefined and the second walk asserts in
+      merge_walk().
+    */
+    result_finalized= true;
   }
 
-  if (result_cut ||
-      (table && table->blob_storage &&
-       table->blob_storage->is_truncated_value()))
+  /*
+    The result itself came out short: it was cut at gconcat_max_len(), or
+    the sort tree could not keep every row it was given. Values the user
+    asked for are missing from the answer, so this is a warning.
+  */
+  if (result_cut)
   {
     warning_for_row= true;
     report_cut_value_error(current_thd, row_count, func_name());
     /*
-      Clear the marks so that we give only one warning per group, even
+      Clear the mark so that we give only one warning per group, even
       if val_str() is called more than once for this group.
     */
     result_cut= false;
-    if (table && table->blob_storage)
-      table->blob_storage->set_truncated_value(false);
+  }
+
+  /* A value was cut on its way into blob_storage: one note per statement. */
+  if (table && table->blob_storage &&
+      table->blob_storage->is_truncated_value())
+  {
+    if (!cut_note_given)
+    {
+      cut_note_given= true;
+      report_cut_value_note(current_thd, func_name());
+    }
+    table->blob_storage->set_truncated_value(false);
   }
 
   return &result;
diff --git a/sql/item_sum.h b/sql/item_sum.h
index 9a60a368dd1f7..cb307b0da09f5 100644
--- a/sql/item_sum.h
+++ b/sql/item_sum.h
@@ -2093,6 +2093,18 @@ class Item_func_group_concat : public Item_sum_str
     cut value warning for it.
   */
   bool result_cut;
+  /*
+    Set once the note about values cut on their way into blob_storage has
+    been given for this aggregate. One note per statement is enough,
+    however many groups had a value cut. cleanup() clears it, so a
+    statement that is run again gets its own note.
+  */
+  bool cut_note_given;
+  /*
+    Set by dump_leaf_key() when it is the one that stops a walk, so that
+    val_str() can tell that apart from the walk itself having failed.
+  */
+  bool walk_stopped;
   bool always_null;
   bool force_copy_fields;
   /** True if entire result of GROUP_CONCAT has been written to output buffer. */
@@ -2199,6 +2211,9 @@ class Item_func_group_concat : public Item_sum_str
   qsort_cmp2 get_comparator_function_for_order_by();
   uchar* get_record_pointer();
   uint get_null_bytes();
+  bool insert_to_order_tree(uchar *key);
+  static int dump_leaf_key_to_tree(void *key_arg, element_count count,
+                                   void *item_arg);
 
 protected:
   Item *shallow_copy(THD *thd) const override
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index ef4fd75fc205b..0246c9ba34b39 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -12406,3 +12406,5 @@ ER_WARN_QB_NAME_PATH_VIEW_NOT_FOUND
         eng "Hint %s is ignored. `%s` required at element #%u of the path is not found in the target query block."
 ER_WARN_QB_NAME_PATH_NOT_SUPPORTED_INSIDE_VIEW
         eng "Hint %s is ignored. QB_NAME hints with path are not supported inside view definitions."
+ER_CUT_VALUES_WHILE_PROCESSING
+        eng "Some values were cut while processing %s). Increase %s if you want to avoid the cut"
diff --git a/sql/uniques.cc b/sql/uniques.cc
index 1bfaf3327e7e8..ed4fc92f2d065 100644
--- a/sql/uniques.cc
+++ b/sql/uniques.cc
@@ -671,6 +671,11 @@ bool Unique::walk(TABLE *table, tree_walk_action action, void *walk_action_arg)
   if (elements == 0)                       /* the whole tree is in memory */
     return tree_walk(&tree, action, walk_action_arg, left_root_right);
 
+  DBUG_EXECUTE_IF("unique_walk_merge_fail", return 1;);
+  /* A failure that reported itself, as an out of memory one would */
+  DBUG_EXECUTE_IF("unique_walk_merge_error",
+                  { my_error(ER_OUT_OF_RESOURCES, MYF(0)); return 1; });
+
   sort.return_rows= elements+tree.elements_in_tree;
   /* flush current tree to the file to have some memory for merge buffer */
   if (flush())
diff --git a/storage/heap/hp_test_freelist-t.c b/storage/heap/hp_test_freelist-t.c
index 9edecdad551e8..7a845fb5546cd 100644
--- a/storage/heap/hp_test_freelist-t.c
+++ b/storage/heap/hp_test_freelist-t.c
@@ -1799,11 +1799,80 @@ static void test_block_to_block_coalesce(void)
 }
 
 
+/*
+  Test: a reclaimed leaf is handed back even when the table is over its
+  memory ceiling.
+
+  index_length is not budgeted anywhere - hp_find_free_hash() allocates a
+  leaf without consulting max_table_size - so a table is routinely over
+  its ceiling from its first row onwards.  That is legal, because the
+  ceiling is only tested when the record cursor lands on a leaf boundary,
+  and it is tested there before any allocation has happened.
+
+  hp_shrink_tail() then walks last_allocated back to 0, which puts the
+  cursor on a leaf boundary again.  The leaf itself stays allocated and
+  data_length keeps counting it, so handing it back adds no memory.  A
+  ceiling test on that branch would make the table refuse a row it held a
+  moment earlier.
+
+  A ceiling of 20000 is under the sum of the two leaves this table
+  allocates (one for records, one for the hash index) and over either one
+  alone, so the first write allocates freely and the second write is the
+  one that has to reclaim.
+*/
+
+static void test_reclaim_over_ceiling(void)
+{
+  HP_SHARE *share;
+  HP_INFO *info;
+  uchar rec[REC_LENGTH];
+  uchar blob_data[100];
+  ulonglong allocated;
+
+  memset(blob_data, 'R', sizeof(blob_data));
+
+  if (create_and_open_ceiling("test_reclaim_ceiling", 20000, &share, &info))
+  {
+    ok(0, "setup failed: %d", my_errno);
+    skip(5, "setup failed");
+    return;
+  }
+
+  build_record(rec, 1, blob_data, sizeof(blob_data));
+  ok(heap_write(info, rec) == 0, "insert blob row");
+
+  allocated= share->data_length + share->index_length;
+  ok(allocated >= share->max_table_size,
+     "table is over its ceiling after one row (allocated=%llu, ceiling=%llu)",
+     allocated, share->max_table_size);
+
+  {
+    uchar key[4];
+    int4store(key, 1);
+    ok(heap_rkey(info, rec, 0, key, 4, HA_READ_KEY_EXACT) == 0,
+       "found blob row");
+    ok(heap_delete(info, rec) == 0, "deleted blob row");
+  }
+  hp_flush_pending_blob_free(info);
+
+  ok(share->block.last_allocated == 0,
+     "hp_shrink_tail put the cursor back on a leaf boundary "
+     "(last_allocated=%lu)", (ulong) share->block.last_allocated);
+
+  build_record(rec, 2, blob_data, sizeof(blob_data));
+  ok(heap_write(info, rec) == 0,
+     "reinsert reclaims the leaf instead of reporting the table full");
+
+  heap_drop_table(info);
+  heap_close(info);
+}
+
+
 int main(int argc __attribute__((unused)),
          char **argv __attribute__((unused)))
 {
   MY_INIT("hp_test_freelist");
-  plan(252);
+  plan(258);
 
   diag("Test 1: free-list contiguity detects groups > 2 records");
   test_freelist_contiguity_multirecord();
@@ -1868,6 +1937,9 @@ int main(int argc __attribute__((unused)),
   diag("Test 21: block-to-block coalescing via adjacent blob chains");
   test_block_to_block_coalesce();
 
+  diag("Test 22: reclaimed leaf handed back over the memory ceiling");
+  test_reclaim_over_ceiling();
+
   my_end(0);
   return exit_status();
 }
diff --git a/storage/heap/hp_test_helpers.h b/storage/heap/hp_test_helpers.h
index 6a9c7ab46bdcf..e52571b5a8637 100644
--- a/storage/heap/hp_test_helpers.h
+++ b/storage/heap/hp_test_helpers.h
@@ -41,7 +41,9 @@ static void build_record(uchar *rec, int32 int_val,
 }
 
 
-static int create_and_open(const char *name, HP_SHARE **share, HP_INFO **info)
+static int create_and_open_ceiling(const char *name,
+                                   ulonglong max_table_size,
+                                   HP_SHARE **share, HP_INFO **info)
 {
   HP_KEYDEF keydef;
   HA_KEYSEG keyseg;
@@ -71,7 +73,7 @@ static int create_and_open(const char *name, HP_SHARE **share, HP_INFO **info)
   ci.reclength=      REC_LENGTH;
   ci.max_records=    1000;
   ci.min_records=    10;
-  ci.max_table_size= 1024 * 1024;
+  ci.max_table_size= max_table_size;
   ci.blob_descs=     &blob_desc;
   ci.blob_count=     1;
 
@@ -84,4 +86,10 @@ static int create_and_open(const char *name, HP_SHARE **share, HP_INFO **info)
   return 0;
 }
 
+
+static int create_and_open(const char *name, HP_SHARE **share, HP_INFO **info)
+{
+  return create_and_open_ceiling(name, 1024 * 1024, share, info);
+}
+
 #endif /* HP_TEST_HELPERS_H */
diff --git a/storage/heap/hp_write.c b/storage/heap/hp_write.c
index 96fba9b83132a..45a4f43118355 100644
--- a/storage/heap/hp_write.c
+++ b/storage/heap/hp_write.c
@@ -196,17 +196,11 @@ uchar *hp_alloc_from_tail(HP_SHARE *info, uint *blocks)
   if (!(block_pos= (uint)(info->block.last_allocated %
                            info->block.records_in_block)))
   {
-    if ((info->block.last_allocated > info->max_records &&
-         info->max_records) ||
-        (info->data_length + info->index_length >= info->max_table_size))
+    if (info->block.last_allocated > info->max_records && info->max_records)
     {
       DBUG_PRINT("error",
-                 ("record file full. last_allocated: %lu  max_records: %lu  "
-                  "data_length: %llu  index_length: %llu  "
-                  "max_table_size: %llu",
-                  info->block.last_allocated, info->max_records,
-                  info->data_length, info->index_length,
-                  info->max_table_size));
+                 ("record file full. last_allocated: %lu  max_records: %lu",
+                  info->block.last_allocated, info->max_records));
       my_errno= HA_ERR_RECORD_FILE_FULL;
       DBUG_RETURN(NULL);
     }
@@ -219,6 +213,26 @@ uchar *hp_alloc_from_tail(HP_SHARE *info, uint *blocks)
     }
     else
     {
+      /*
+        The table memory ceiling gates memory the table does not hold
+        yet, so it belongs here and not beside the max_records test.
+        The reclaim branch hands back a leaf that data_length already
+        counts, and hp_shrink_tail() reaches it whenever it empties the
+        tail of a table that has been over its ceiling since its first
+        row: index leaves are allocated without consulting the ceiling.
+        Testing the ceiling there as well would make such a table refuse
+        a row it held a moment earlier.
+      */
+      if (info->data_length + info->index_length >= info->max_table_size)
+      {
+        DBUG_PRINT("error",
+                   ("record file full. data_length: %llu  index_length: %llu  "
+                    "max_table_size: %llu",
+                    info->data_length, info->index_length,
+                    info->max_table_size));
+        my_errno= HA_ERR_RECORD_FILE_FULL;
+        DBUG_RETURN(NULL);
+      }
       /* No available blocks, allocate new ones */
       if (hp_get_new_block(info, &info->block, &length))
         DBUG_RETURN(NULL);