Ensure CannedSourceOperator::deepCopyOf properly release blocks on failure.

Author: afoucret

x-pack/plugin/esql/compute/test/src/main/java/org/elasticsearch/compute/test/CannedSourceOperator.java

@@ -84,14 +84,20 @@ public static List<Page> deepCopyOf(BlockFactory blockFactory, List<Page> pages)
         try {
             for (Page p : pages) {
                 Block[] blocks = new Block[p.getBlockCount()];
-                for (int b = 0; b < blocks.length; b++) {
-                    Block orig = p.getBlock(b);
-                    try (Block.Builder builder = orig.elementType().newBlockBuilder(p.getPositionCount(), blockFactory)) {
-                        builder.copyFrom(orig, 0, p.getPositionCount());
-                        blocks[b] = builder.build();
+                try {
+                    for (int b = 0; b < blocks.length; b++) {
+                        Block orig = p.getBlock(b);
+                        try (Block.Builder builder = orig.elementType().newBlockBuilder(p.getPositionCount(), blockFactory)) {
+                            builder.copyFrom(orig, 0, p.getPositionCount());
+                            blocks[b] = builder.build();
+                        }
                     }
+                    out.add(new Page(blocks));
+                } catch (Exception e) {
+                    // Something went wrong, release the blocks.
+                    Releasables.closeExpectNoException(blocks);
+                    throw e;
                 }
-                out.add(new Page(blocks));
             }
         } finally {
             if (pages.size() != out.size()) {