Disable ets:select_replace/2 for bags

The existing implementation presented both
semantic inconsistencies and performance issues.

4 files changed, 117 insertions, 169 deletions

diff --git a/erts/emulator/beam/erl_db.c b/erts/emulator/beam/erl_db.c
index 81fe79a92e..d77c585628 100644
--- a/erts/emulator/beam/erl_db.c
+++ b/erts/emulator/beam/erl_db.c
@@ -2982,6 +2982,14 @@ BIF_RETTYPE ets_select_replace_2(BIF_ALIST_2)
     if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE_REC)) == NULL) {
         BIF_ERROR(BIF_P, BADARG);
     }
+
+    if (tb->common.status & DB_BAG) {
+        /* Bag implementation presented both semantic consistency
+           and performance issues */
+        db_unlock(tb, LCK_WRITE_REC);
+        BIF_ERROR(BIF_P, BADARG);
+    }
+
     safety = ITERATION_SAFETY(BIF_P,tb);
     if (safety == ITER_UNSAFE) {
         local_fix_table(tb);
diff --git a/erts/emulator/beam/erl_db_hash.c b/erts/emulator/beam/erl_db_hash.c
index a4ca6dce77..96d74bb050 100644
--- a/erts/emulator/beam/erl_db_hash.c
+++ b/erts/emulator/beam/erl_db_hash.c
@@ -2009,7 +2009,7 @@ static int db_select_replace_hash(Process *p,
     DbTableHash *tb = &tbl->hash;
     struct mp_info mpi;
     Uint slot_ix = 0;
-    HashDbTerm **current = NULL, **initial = NULL, **iter = NULL;
+    HashDbTerm **current = NULL;
     HashDbTerm *new = NULL, *next = NULL;
     HashValue hval = INVALID_HASH;
     unsigned current_list_pos = 0;
@@ -2036,6 +2036,8 @@ static int db_select_replace_hash(Process *p,
 	return RetVal;				\
     } while(0)
 
+    /* Bag implementation presented both semantic consistency and performance issues */
+    ASSERT(!(tb->common.status & DB_BAG));
 
     if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
         RET_TO_BIF(NIL,errcode);
@@ -2060,7 +2062,6 @@ static int db_select_replace_hash(Process *p,
     }
 
 
-    initial = current;
     for(;;) {
         if ((*current) == NULL) {
             if (mpi.key_given) {  /* Key is bound */
@@ -2084,43 +2085,10 @@ static int db_select_replace_hash(Process *p,
                 }
                 current = &BUCKET(tb,slot_ix);
             }
-            initial = current;
         }
         else if ((*current)->hvalue == INVALID_HASH) {
             current = &((*current)->next);
         }
-        else if (tb->common.status & DB_BAG) {
-            match_res = db_match_dbterm(&tb->common, p, mpi.mp, 0,
-                                        &(*current)->dbterm, NULL, 0);
-            if (is_value(match_res) &&
-                is_value(key = db_getkey(tb->common.keypos, match_res)) &&
-                eq(key, GETKEY(tb, (*current)->dbterm.tpl)))
-            {
-                // we need to make sure we don't end up with duplicate objects;
-                // it's quite inefficient
-                int object_exists = 0;
-                for (iter = initial; *iter != NULL; iter = &((*iter)->next))
-                    if (((*iter)->hvalue != INVALID_HASH)
-                            && (*iter != *current)
-                            && db_eq(&tb->common, match_res, &(*iter)->dbterm))
-                    {
-                        object_exists = 1;
-                        break;
-                    }
-
-                if (!object_exists) {
-                    next = (*current)->next;
-                    hval = (*current)->hvalue;
-                    new = replace_dbterm(tb, *current, match_res);
-                    new->next = next;
-                    new->hvalue = hval;
-                    *current = new;
-                    ++got;
-                }
-            }
-            --num_left;
-            current = &((*current)->next);
-        }
         else {
             match_res = db_match_dbterm(&tb->common, p, mpi.mp, 0,
                                         &(*current)->dbterm, NULL, 0);
@@ -2178,7 +2146,7 @@ static int db_select_replace_continue_hash(Process *p,
 {
     DbTableHash *tb = &tbl->hash;
     Uint slot_ix;
-    HashDbTerm **current = NULL, **initial = NULL, **iter = NULL;
+    HashDbTerm **current = NULL;
     HashDbTerm *new = NULL, *next = NULL;
     HashValue hval = INVALID_HASH;
     Eterm key;
@@ -2213,7 +2181,6 @@ static int db_select_replace_continue_hash(Process *p,
         goto done;
     }
     current = &BUCKET(tb,slot_ix);
-    initial = current;
 
     for(;;) {
         if ((*current) == NULL) {
@@ -2225,43 +2192,10 @@ static int db_select_replace_continue_hash(Process *p,
                 goto trap;
             }
             current = &BUCKET(tb,slot_ix);
-            initial = current;
         }
         else if ((*current)->hvalue == INVALID_HASH) {
             current = &((*current)->next);
         }
-        else if (tb->common.status & DB_BAG) {
-            match_res = db_match_dbterm(&tb->common, p, mp, 0,
-                                        &(*current)->dbterm, NULL, 0);
-            if (is_value(match_res) &&
-                is_value(key = db_getkey(tb->common.keypos, match_res)) &&
-                eq(key, GETKEY(tb, (*current)->dbterm.tpl)))
-            {
-                // we need to make sure we don't end up with duplicate objects;
-                // it's quite inefficient
-                int object_exists = 0;
-                for (iter = initial; *iter != NULL; iter = &((*iter)->next))
-                    if (((*iter)->hvalue != INVALID_HASH)
-                            && (*iter != *current)
-                            && db_eq(&tb->common, match_res, &(*iter)->dbterm))
-                    {
-                        object_exists = 1;
-                        break;
-                    }
-
-                if (!object_exists) {
-                    next = (*current)->next;
-                    hval = (*current)->hvalue;
-                    new = replace_dbterm(tb, *current, match_res);
-                    new->next = next;
-                    new->hvalue = hval;
-                    *current = new;
-                    ++got;
-                }
-            }
-            --num_left;
-            current = &((*current)->next);
-        }
         else {
             match_res = db_match_dbterm(&tb->common, p, mp, 0,
                                         &(*current)->dbterm, NULL, 0);
diff --git a/lib/stdlib/doc/src/ets.xml b/lib/stdlib/doc/src/ets.xml
index 36c803b40c..3e9ad89b26 100644
--- a/lib/stdlib/doc/src/ets.xml
+++ b/lib/stdlib/doc/src/ets.xml
@@ -1495,6 +1495,10 @@ is_integer(X), is_integer(Y), X + Y < 4711]]></code>
       <fsummary>Match the objects in an ETS table against a match_spec and
         replaces matching objects with the match_spec result</fsummary>
       <desc>
+        <warning>
+          <p>For the moment, due to performance and semantic constraints,
+            tables of type <c>bag</c> are not yet supported.</p>
+        </warning>
         <p>Matches the objects in the table <c><anno>Tab</anno></c> using a
           <seealso marker="#match_spec">match_spec</seealso>. If the
           an object is matched, and the match_spec result is an object with the
diff --git a/lib/stdlib/test/ets_SUITE.erl b/lib/stdlib/test/ets_SUITE.erl
index e0e8436a8c..fed3c3ac61 100644
--- a/lib/stdlib/test/ets_SUITE.erl
+++ b/lib/stdlib/test/ets_SUITE.erl
@@ -1140,113 +1140,115 @@ t_select_delete(Config) when is_list(Config) ->
 t_select_replace(Config) when is_list(Config) ->
     EtsMem = etsmem(),
     Tables = fill_sets_int(10000) ++ fill_sets_int(10000, [{write_concurrency,true}]),
-    lists:foreach(
-      fun(Table) ->
-              TableType = ets:info(Table, type),
-
-              % Replacements are differently-sized objects
-              MatchSpec1_A = [{{'$1','$2'},
-                             [{'<', {'rem', '$1', 5}, 2}],
-                             [{{'$1', [$x | '$2'], stuff}}]}],
-              MatchSpec1_B = [{{'$1','$2','_'},
-                             [],
-                             [{{'$1','$2'}}]}],
-              4000 = ets:select_replace(Table, MatchSpec1_A),
-              4000 = ets:select_replace(Table, MatchSpec1_B),
-
-              % Replacement changes key to float equivalent
-              MatchSpec2 = [{{'$1', '$2'},
-                             [{'=:=', {'band', '$1', 2#11}, 2#11},
-                              {'=/=', {'hd', '$2'}, $x}],
-                             [{{{'*', '$1', 1.0}, '$2'}}]}],
-              case TableType of
-                  ordered_set ->   1500 = ets:select_replace(Table, MatchSpec2);
-                  set ->           0 = ets:select_replace(Table, MatchSpec2);
-                  bag ->           0 = ets:select_replace(Table, MatchSpec2);
-                  duplicate_bag -> 0 = ets:select_replace(Table, MatchSpec2)
-              end,
 
-              % Replacement is an equal object
-              MatchSpec3 = [{{'$1', '$2'},
-                             [{'>', {'rem', '$1', 5}, 3}],
-                             [{{'$1', '$2'}}]}],
-              case TableType of
-                  ordered_set   -> 1500 = ets:select_replace(Table, MatchSpec3);
-                  set           -> 2000 = ets:select_replace(Table, MatchSpec3);
-                  bag           -> 2000 = ets:select_replace(Table, MatchSpec3);
-                  duplicate_bag -> 2000 = ets:select_replace(Table, MatchSpec3)
-              end,
+    TestFun = fun (Table, TableType) when TableType =:= bag ->
+                      % Operation not supported; bag implementation
+                      % presented both semantic consistency and performance issues.
+                      10000 = ets:select_delete(Table, [{'_',[],[true]}]);
+
+                  (Table, TableType) ->
+                      % Replacements are differently-sized objects
+                      MatchSpec1_A = [{{'$1','$2'},
+                                       [{'<', {'rem', '$1', 5}, 2}],
+                                       [{{'$1', [$x | '$2'], stuff}}]}],
+                      MatchSpec1_B = [{{'$1','$2','_'},
+                                       [],
+                                       [{{'$1','$2'}}]}],
+                      4000 = ets:select_replace(Table, MatchSpec1_A),
+                      4000 = ets:select_replace(Table, MatchSpec1_B),
+
+                      % Replacement changes key to float equivalent
+                      MatchSpec2 = [{{'$1', '$2'},
+                                     [{'=:=', {'band', '$1', 2#11}, 2#11},
+                                      {'=/=', {'hd', '$2'}, $x}],
+                                     [{{{'*', '$1', 1.0}, '$2'}}]}],
+                      case TableType of
+                          ordered_set ->   1500 = ets:select_replace(Table, MatchSpec2);
+                          set ->           0 = ets:select_replace(Table, MatchSpec2);
+                          duplicate_bag -> 0 = ets:select_replace(Table, MatchSpec2)
+                      end,
 
-              check(Table,
-                    fun ({N, [$x, C | _]}) when ((N rem 5) < 2) -> (C >= $0) andalso (C =< $9);
-                        ({N, [C | _]}) when is_float(N) -> (C >= $0) andalso (C =< $9);
-                        ({N, [C | _]}) when ((N rem 5) > 3) -> (C >= $0) andalso (C =< $9);
-                        ({_, [C | _]}) -> (C >= $0) andalso (C =< $9)
-                    end,
-                    10000),
-
-              % Replace unbound range (>)
-              MatchSpec4 = [{{'$1', '$2'},
-                             [{'>', '$1', 7000}],
-                             [{{'$1', {{gt_range, '$2'}}}}]}],
-              case TableType of
-                  ordered_set   -> 3000 = ets:select_replace(Table, MatchSpec4);
-                  set           -> 3000 = ets:select_replace(Table, MatchSpec4);
-                  bag           -> 3000 = ets:select_replace(Table, MatchSpec4);
-                  duplicate_bag -> 3000 = ets:select_replace(Table, MatchSpec4)
-              end,
+                      % Replacement is an equal object
+                      MatchSpec3 = [{{'$1', '$2'},
+                                     [{'>', {'rem', '$1', 5}, 3}],
+                                     [{{'$1', '$2'}}]}],
+                      case TableType of
+                          ordered_set   -> 1500 = ets:select_replace(Table, MatchSpec3);
+                          set           -> 2000 = ets:select_replace(Table, MatchSpec3);
+                          duplicate_bag -> 2000 = ets:select_replace(Table, MatchSpec3)
+                      end,
 
-              % Replace unbound range (<)
-              MatchSpec5 = [{{'$1', '$2'},
-                             [{'<', '$1', 3000}],
-                             [{{'$1', {{le_range, '$2'}}}}]}],
-              case TableType of
-                  ordered_set ->   2999 = ets:select_replace(Table, MatchSpec5);
-                  set ->           2999 = ets:select_replace(Table, MatchSpec5);
-                  bag ->           2998 = ets:select_replace(Table, MatchSpec5);
-                  duplicate_bag -> 2998 = ets:select_replace(Table, MatchSpec5)
-              end,
+                      check(Table,
+                            fun ({N, [$x, C | _]}) when ((N rem 5) < 2) -> (C >= $0) andalso (C =< $9);
+                                ({N, [C | _]}) when is_float(N) -> (C >= $0) andalso (C =< $9);
+                                ({N, [C | _]}) when ((N rem 5) > 3) -> (C >= $0) andalso (C =< $9);
+                                ({_, [C | _]}) -> (C >= $0) andalso (C =< $9)
+                            end,
+                            10000),
+
+                      % Replace unbound range (>)
+                      MatchSpec4 = [{{'$1', '$2'},
+                                     [{'>', '$1', 7000}],
+                                     [{{'$1', {{gt_range, '$2'}}}}]}],
+                      case TableType of
+                          ordered_set   -> 3000 = ets:select_replace(Table, MatchSpec4);
+                          set           -> 3000 = ets:select_replace(Table, MatchSpec4);
+                          duplicate_bag -> 3000 = ets:select_replace(Table, MatchSpec4)
+                      end,
 
-              % Replace bound range
-              MatchSpec6 = [{{'$1', '$2'},
-                             [{'>=', '$1', 3001},
-                              {'<', '$1', 7000}],
-                             [{{'$1', {{range, '$2'}}}}]}],
-              case TableType of
-                  ordered_set ->   3999 = ets:select_replace(Table, MatchSpec6);
-                  set ->           3999 = ets:select_replace(Table, MatchSpec6);
-                  bag ->           3998 = ets:select_replace(Table, MatchSpec6);
-                  duplicate_bag -> 3998 = ets:select_replace(Table, MatchSpec6)
-              end,
+                      % Replace unbound range (<)
+                      MatchSpec5 = [{{'$1', '$2'},
+                                     [{'<', '$1', 3000}],
+                                     [{{'$1', {{le_range, '$2'}}}}]}],
+                      case TableType of
+                          ordered_set ->   2999 = ets:select_replace(Table, MatchSpec5);
+                          set ->           2999 = ets:select_replace(Table, MatchSpec5);
+                          duplicate_bag -> 2998 = ets:select_replace(Table, MatchSpec5)
+                      end,
 
-              % Replace particular keys
-              MatchSpec7 = [{{'$1', '$2'},
-                             [{'==', '$1', 3000}],
-                             [{{'$1', {{specific1, '$2'}}}}]},
-                            {{'$1', '$2'},
-                             [{'==', '$1', 7000}],
-                             [{{'$1', {{specific2, '$2'}}}}]}],
-              case TableType of
-                  ordered_set ->   2 = ets:select_replace(Table, MatchSpec7);
-                  set ->           2 = ets:select_replace(Table, MatchSpec7);
-                  bag ->           4 = ets:select_replace(Table, MatchSpec7);
-                  duplicate_bag -> 4 = ets:select_replace(Table, MatchSpec7)
+                      % Replace bound range
+                      MatchSpec6 = [{{'$1', '$2'},
+                                     [{'>=', '$1', 3001},
+                                      {'<', '$1', 7000}],
+                                     [{{'$1', {{range, '$2'}}}}]}],
+                      case TableType of
+                          ordered_set ->   3999 = ets:select_replace(Table, MatchSpec6);
+                          set ->           3999 = ets:select_replace(Table, MatchSpec6);
+                          duplicate_bag -> 3998 = ets:select_replace(Table, MatchSpec6)
+                      end,
+
+                      % Replace particular keys
+                      MatchSpec7 = [{{'$1', '$2'},
+                                     [{'==', '$1', 3000}],
+                                     [{{'$1', {{specific1, '$2'}}}}]},
+                                    {{'$1', '$2'},
+                                     [{'==', '$1', 7000}],
+                                     [{{'$1', {{specific2, '$2'}}}}]}],
+                      case TableType of
+                          ordered_set ->   2 = ets:select_replace(Table, MatchSpec7);
+                          set ->           2 = ets:select_replace(Table, MatchSpec7);
+                          duplicate_bag -> 4 = ets:select_replace(Table, MatchSpec7)
+                      end,
+
+                      check(Table,
+                            fun ({N, {gt_range, _}}) -> N > 7000;
+                                ({N, {le_range, _}}) -> N < 3000;
+                                ({N, {range, _}}) -> (N >= 3001) andalso (N < 7000);
+                                ({N, {specific1, _}}) -> N == 3000;
+                                ({N, {specific2, _}}) -> N == 7000
+                            end,
+                            10000),
+
+                      10000 = ets:select_delete(Table, [{'_',[],[true]}]),
+                      check(Table, fun (_) -> false end, 0)
               end,
 
-              check(Table,
-                    fun ({N, {gt_range, _}}) -> N > 7000;
-                        ({N, {le_range, _}}) -> N < 3000;
-                        ({N, {range, _}}) -> (N >= 3001) andalso (N < 7000);
-                        ({N, {specific1, _}}) -> N == 3000;
-                        ({N, {specific2, _}}) -> N == 7000
-                    end,
-                    10000),
-
-              10000 = ets:select_delete(Table, [{'_',[],[true]}]),
-              check(Table, fun (_) -> false end, 0)
+    lists:foreach(
+      fun(Table) ->
+              TestFun(Table, ets:info(Table, type)),
+              ets:delete(Table)
       end,
       Tables),
-    lists:foreach(fun(Tab) -> ets:delete(Tab) end,Tables),
     verify_etsmem(EtsMem).
 
 %% Test that partly bound keys gives faster matches.
@@ -5566,7 +5568,7 @@ smp_select_replace(Config) when is_list(Config) ->
               0 = ets:info(T, size),
               ets:delete(T)
       end,
-      [ordered_set, set, bag, duplicate_bag]).
+      [ordered_set, set, duplicate_bag]).
 
 %% Test different types.
 types(Config) when is_list(Config) ->