fix mem.leak

Author: jan-elastic

test/framework/src/main/java/org/elasticsearch/common/util/MockBigArrays.java

@@ -100,9 +100,6 @@ public static void assertFitsIn(ByteSizeValue max, Function<BigArrays, Releasabl
     private static final ConcurrentMap<Object, Object> ACQUIRED_ARRAYS = new ConcurrentHashMap<>();
 
     public static void ensureAllArraysAreReleased() throws Exception {
-        // TODO: reenable + fix mem.leaks
-        if (true) return;
-
         final Map<Object, Object> masterCopy = new HashMap<>(ACQUIRED_ARRAYS);
         if (masterCopy.isEmpty() == false) {
             // not empty, we might be executing on a shared cluster that keeps on obtaining

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/ChangePointOperator.java

@@ -23,6 +23,8 @@
 
 public class ChangePointOperator implements Operator {
 
+    // TODO: close upon failure / interrupt
+
     public record Factory(int inputChannel) implements OperatorFactory {
         @Override
         public Operator get(DriverContext driverContext) {
@@ -70,15 +72,16 @@ public void finish() {
 
     @Override
     public boolean isFinished() {
-        return finished && outputPageIndex == inputPages.size();
+        return finished && outputPages.isEmpty();
     }
 
     @Override
     public Page getOutput() {
         if (finished == false) {
             return null;
         }
-        if (outputPageIndex == inputPages.size()) {
+        if (outputPageIndex == outputPages.size()) {
+            outputPages.clear();
             return null;
         }
         return outputPages.get(outputPageIndex++);
@@ -90,6 +93,7 @@ private void createOutputPages() {
             valuesCount += page.getPositionCount();
         }
 
+        // TODO: account for this memory?
         double[] values = new double[valuesCount];
         int valuesIndex = 0;
         for (Page inputPage : inputPages) {
@@ -131,11 +135,13 @@ private void createOutputPages() {
                 changePvalueBlock = blockFactory.newConstantNullBlock(inputPage.getPositionCount());
             }
 
+            // TODO: what about duplicate names??
             Page outputPage = inputPage.appendBlocks(new Block[] { changeTypeBlock, changePvalueBlock });
             outputPages.add(outputPage);
-
             pageStartIndex += inputPage.getPositionCount();
         }
+
+        inputPages.clear();
     }
 
     @Override

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/CsvTests.java

@@ -327,16 +327,13 @@ private void doTest() throws Exception {
         } finally {
             Releasables.close(() -> Iterators.map(actualResults.pages().iterator(), p -> p::releaseBlocks));
             // Give the breaker service some time to clear in case we got results before the rest of the driver had cleaned up
-            // TODO: reenable + fix mem.leaks
-            /*
             assertBusy(
                 () -> assertThat(
                     "Not all circuits were cleaned up",
                     bigArrays.breakerService().getBreaker(CircuitBreaker.REQUEST).getUsed(),
                     equalTo(0L)
                 )
             );
-            */
         }
     }