Split large deletes into chunks when using IN

python/lsst/daf/butler/registry/interfaces/_database.py

@@ -45,6 +45,7 @@
     Type,
     Union,
 )
+import math
 import uuid
 import warnings
 
@@ -1459,11 +1460,35 @@ def delete(self, table: sqlalchemy.schema.Table, columns: Iterable[str], *rows:
                 # The set only has one element
                 clauses.append(column == v.pop())
 
-            # Now the IN operator
-            clauses.append(table.columns[name].in_(content[name]))
-
-            sql = sql.where(sqlalchemy.sql.and_(*clauses))
-            return self._connection.execute(sql).rowcount
+            # The IN operator will not work for "infinite" numbers of
+            # rows so must batch it up into distinct calls.
+            in_content = list(content[name])
+            n_elements = len(in_content)
+
+            # This is a soft limit which might be increased by 50%.
+            # For example with max=1001 if 1500 elements are involved
+            # the code will choose one loop of 1500. If 2000 elements are
+            # involved there will be two loops of 1000.
+            # The idea is to allow some adjustment of the limits so that
+            # if 1002 elements are present this is done in an even split
+            # and not as 1001 elements in one call and 1 in another.
+            #
+            MAX_ELEMENTS_PER_IN = 1000
+            n_loops = round(n_elements / MAX_ELEMENTS_PER_IN)
+            if n_loops == 0:
+                n_loops = 1
+            n_per_loop = math.ceil(n_elements / n_loops)
+
+            rowcount = 0
+            iposn = 0
+            while iposn < n_elements:
+                endpos = iposn + n_per_loop
+                in_clause = table.columns[name].in_(in_content[iposn:endpos])
+                iposn = endpos
+
+                newsql = sql.where(sqlalchemy.sql.and_(*clauses, in_clause))
+                rowcount += self._connection.execute(newsql).rowcount
+            return rowcount
         else:
             whereTerms = [table.columns[name] == sqlalchemy.sql.bindparam(name) for name in columns]
             if whereTerms: