Modify AbstractSessionContextImpl.generateTempIdTable(...) so that it chunks the insertions into the temp id table to work around limitations in some versions of SQL server of a maximum of 1000 rows in an INSERT statement

Author: realityforge

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ### Unreleased
 
+* Modify `AbstractSessionContextImpl.generateTempIdTable(...)` so that it chunks the insertions into the temp id table to work around limitations in some versions of SQL server of a maximum of 1000 rows in an INSERT statement.
+
 ### [v6.121](https://github.com/replicant4j/replicant/tree/v6.121) (2022-04-07) · [Full Changelog](https://github.com/spritz/spritz/compare/v6.120...v6.121)
 
 Changes in this release:

server/src/main/java/org/realityforge/replicant/server/ee/AbstractSessionContextImpl.java

@@ -6,7 +6,10 @@
 import java.sql.Statement;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import javax.persistence.EntityManager;
@@ -73,13 +76,26 @@ protected String generateTempIdTable( @Nonnull final List<ChannelAddress> addres
   {
     return
       "DECLARE @Ids TABLE ( Id INTEGER NOT NULL );\n" +
-      "INSERT INTO @Ids VALUES " +
-      addresses.stream()
-        .map( a -> "(" + a.getSubChannelId() + ")" )
-        .collect( Collectors.joining( "," ) ) +
+      chunked( addresses.stream().map( ChannelAddress::getSubChannelId ), 900 )
+        .map( ids ->
+                "INSERT INTO @Ids VALUES " +
+                ids.stream().map( id -> "(" + id + ")" ).collect( Collectors.joining( "," ) ) ).
+        collect( Collectors.joining( "\n" ) ) +
       "\n";
   }
 
+  @SuppressWarnings( "SameParameterValue" )
+  private static <T> Stream<List<T>> chunked( @Nonnull final Stream<T> stream, final int chunkSize )
+  {
+    final AtomicInteger index = new AtomicInteger( 0 );
+
+    return stream
+      .collect( Collectors.groupingBy( x -> index.getAndIncrement() / chunkSize ) )
+      .entrySet().stream()
+      .sorted( Map.Entry.comparingByKey() )
+      .map( Map.Entry::getValue );
+  }
+
   protected void bulkLinkFromSourceGraphToTargetGraph( @Nonnull final ReplicantSession session,
                                                        @Nullable final Object filter,
                                                        @Nonnull final ChangeSet changeSet,