From d67709f5a4b5d9fb2a3e34ce3bb3060009d1d216 Mon Sep 17 00:00:00 2001
From: Lukas Larsson <lukas@erlang.org>
Date: Mon, 2 Oct 2023 16:59:06 +0200
Subject: [PATCH] erts: Fix db_match map copy bug

---
 erts/emulator/beam/erl_db_util.c | 39 +++++++++++++++++---------------
 lib/stdlib/test/ets_SUITE.erl    | 23 +++++++++++++++++++
 2 files changed, 44 insertions(+), 18 deletions(-)

diff --git a/erts/emulator/beam/erl_db_util.c b/erts/emulator/beam/erl_db_util.c
index 8145f7083b60..60e7b46d7f81 100644
--- a/erts/emulator/beam/erl_db_util.c
+++ b/erts/emulator/beam/erl_db_util.c
@@ -3991,35 +3991,38 @@ dmc_map(DMCContext *context, DMCHeap *heap, DMC_STACK_TYPE(UWord) *text,
             return ret;
         }
 
+        if (constant_values) {
+            /* We may have to convert all values to individual matchPushC
+               instructions, if we do that then more stack will be needed
+               than estimated, so we artificially bumb the needed stack here
+               so that dmc_tuple thinks that dmc_array has used the needed stack. */
+            context->stack_used += nelems;
+        }
+
         if ((ret = dmc_tuple(context, heap, text, m->keys, &constant_keys)) != retOk) {
             return ret;
         }
 
+        if (constant_values) {
+            context->stack_used -= nelems;
+        }
+
         if (constant_values && constant_keys) {
             *constant = 1;
             return retOk;
         }
 
-        /* If all values were constants, then nothing was emitted by the
-           first dmc_array, so we reset the pc and emit all values as
-           constants and then re-emit the keys. */
         if (constant_values) {
-            DMC_STACK_NUM(*text) = textpos;
-            context->stack_used = stackpos;
-            ASSERT(!constant_keys);
-            for (int i = nelems; i--;) {
-                do_emit_constant(context, text, values[i]);
-            }
-            dmc_tuple(context, heap, text, m->keys, &constant_keys);
+            /* If all values were constants, then nothing was emitted by the
+               first dmc_array, so we insert the constants at the start of the
+               stack. */
+            dmc_rearrange_constants(context, text, textpos, values, nelems);
         } else if (constant_keys) {
-            Eterm *p = tuple_val(m->keys);
-            Uint nelems = arityval(*p);
-            ASSERT(!constant_values);
-            p++;
-            for (int i = nelems; i--;)
-                do_emit_constant(context, text, p[i]);
-            DMC_PUSH2(*text, matchMkTuple, nelems);
-            context->stack_used -= nelems - 1;
+            /* If all keys were constant we just want to emit the key tuple.
+               Since do_emit_constant expects tuples to be wrapped in 1 arity
+               tuples we need give do_emit_constant {keys} */
+            Eterm wrapTuple[2] = {make_arityval(1), m->keys};
+            do_emit_constant(context, text, make_tuple(wrapTuple));
         }
 
         DMC_PUSH2(*text, matchMkFlatMap, nelems);
diff --git a/lib/stdlib/test/ets_SUITE.erl b/lib/stdlib/test/ets_SUITE.erl
index 4a7c4653dbd9..f875c77d49a1 100644
--- a/lib/stdlib/test/ets_SUITE.erl
+++ b/lib/stdlib/test/ets_SUITE.erl
@@ -1914,6 +1914,29 @@ t_select_pam_stack_overflow_bug(Config) ->
     ets:delete(T),
     ok.
 
+t_select_flatmap_term_copy_bug(_Config) ->
+        T = ets:new(a,[]),
+    ets:insert(T, {list_to_binary(lists:duplicate(36,$a)),
+                   list_to_binary(lists:duplicate(36,$b)),
+                   list_to_binary(lists:duplicate(36,$c))}),
+    V1 = ets:select(T, [{{'$1','$2','$3'},[],[#{ '$1' => a }]}]),
+    erlang:garbage_collect(),
+    V1 = ets:select(T, [{{'$1','$2','$3'},[],[#{ '$1' => a }]}]),
+    erlang:garbage_collect(),
+    V2 = ets:select(T, [{{'$1','$2','$3'},[],[#{ a => '$1' }]}]),
+    erlang:garbage_collect(),
+    V2 = ets:select(T, [{{'$1','$2','$3'},[],[#{ a => '$1' }]}]),
+    erlang:garbage_collect(),
+    V3 = ets:select(T, [{{'$1','$2','$3'},[],[#{ '$1' => '$1' }]}]),
+    erlang:garbage_collect(),
+    V3 = ets:select(T, [{{'$1','$2','$3'},[],[#{ '$1' => '$1' }]}]),
+    erlang:garbage_collect(),
+    V3 = ets:select(T, [{{'$1','$2','$3'},[],[#{ a => a }]}]),
+    erlang:garbage_collect(),
+    V3 = ets:select(T, [{{'$1','$2','$3'},[],[#{ a => a }]}]),
+    erlang:garbage_collect(),
+    ets:delete(T),
+    ok.
 
 %% Test that partly bound keys gives faster matches.
 partly_bound(Config) when is_list(Config) ->