delete logging later

Author: limotova

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/AggregateMetricDoubleBlockBuilder.java

@@ -45,17 +45,20 @@ public AggregateMetricDoubleBlockBuilder(int estimatedSize, BlockFactory blockFa
 
     @Override
     protected int valuesLength() {
-        throw new UnsupportedOperationException("Not available on aggregate_metric_double");
+        return minBuilder.valuesLength();
     }
 
     @Override
     protected void growValuesArray(int newSize) {
-        throw new UnsupportedOperationException("Not available on aggregate_metric_double");
+        minBuilder.growValuesArray(newSize);
+        maxBuilder.growValuesArray(newSize);
+        sumBuilder.growValuesArray(newSize);
+        countBuilder.growValuesArray(newSize);
     }
 
     @Override
     protected int elementSize() {
-        throw new UnsupportedOperationException("Not available on aggregate_metric_double");
+        return minBuilder.elementSize() + maxBuilder.elementSize() + sumBuilder.elementSize() + countBuilder.elementSize();
     }
 
     @Override

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

@@ -7,6 +7,8 @@
 
 package org.elasticsearch.compute.operator;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ContextPreservingActionListener;
 import org.elasticsearch.action.support.SubscribableListener;
@@ -15,6 +17,7 @@
 import org.elasticsearch.compute.Describable;
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.exchange.ExchangeSinkOperator;
+import org.elasticsearch.compute.operator.topn.TopNOperator;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
@@ -248,6 +251,8 @@ public void abort(Exception reason, ActionListener<Void> listener) {
         }
     }
 
+    private static final Logger logger = LogManager.getLogger(Driver.class);
+
     private IsBlockedResult runSingleLoopIteration() {
         driverContext.checkForEarlyTermination();
         boolean movedPage = false;
@@ -283,6 +288,7 @@ private IsBlockedResult runSingleLoopIteration() {
                         page.releaseBlocks();
                         throw e;
                     }
+                    logger.warn("INPUTTING PAGE TO OPERATOR " + nextOp);
                     nextOp.addInput(page);
                     movedPage = true;
                 }
@@ -327,13 +333,15 @@ private int closeEarlyFinishedOperators(ListIterator<Operator> operators) {
                 while (finishedOperators.hasNext()) {
                     Operator op = finishedOperators.next();
                     statusOfCompletedOperators.add(new OperatorStatus(op.toString(), op.status()));
+                    logger.warn("CLOSING OPERATOR " + op);
                     op.close();
                     finishedOperators.remove();
                 }
 
                 // Finish the next operator, which is now the first operator.
                 if (activeOperators.isEmpty() == false) {
                     Operator newRootOperator = activeOperators.get(0);
+                    logger.warn("FINISHING OPERATOR " + newRootOperator);
                     newRootOperator.finish();
                 }
                 return index;

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

@@ -30,6 +30,11 @@ public void decodeKey(BytesRef keys) {
 
     @Override
     public void decodeValue(BytesRef values) {
+        int count = TopNEncoder.DEFAULT_UNSORTABLE.decodeVInt(values);
+        if (count == 0) {
+            builder.appendNull();
+            return;
+        }
         for (BlockLoader.DoubleBuilder subBuilder : List.of(builder.min(), builder.max(), builder.sum())) {
             if (TopNEncoder.DEFAULT_UNSORTABLE.decodeBoolean(values)) {
                 subBuilder.appendDouble(TopNEncoder.DEFAULT_UNSORTABLE.decodeDouble(values));
@@ -51,7 +56,7 @@ public Block build() {
 
     @Override
     public String toString() {
-        return "ValueExtractorForAggregateMetricDouble";
+        return "ResultBuilderForAggregateMetricDouble";
     }
 
     @Override

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

@@ -7,6 +7,8 @@
 
 package org.elasticsearch.compute.operator.topn;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.apache.lucene.util.Accountable;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.PriorityQueue;
@@ -47,6 +49,8 @@ public class TopNOperator implements Operator, Accountable {
     private static final byte SMALL_NULL = 0x01; // "null" representation for "nulls first"
     private static final byte BIG_NULL = 0x02; // "null" representation for "nulls last"
 
+    private static final Logger logger = LogManager.getLogger("TopNOperator");
+
     /**
      * Internal row to be used in the PriorityQueue instead of the full blown Page.
      * It mirrors somehow the Block build in the sense that it keeps around an array of offsets and a count of values (to account for
@@ -122,15 +126,20 @@ public void clearRefCounters() {
         }
     }
 
+
     static final class BytesOrder implements Releasable, Accountable {
         private static final long BASE_RAM_USAGE = RamUsageEstimator.shallowSizeOfInstance(BytesOrder.class);
         private final CircuitBreaker breaker;
         final List<SortOrder> sortOrders;
         final int[] endOffsets;
+        private static final Logger logger = LogManager.getLogger(BytesOrder.class);
+        private boolean open = true;
 
         BytesOrder(List<SortOrder> sortOrders, CircuitBreaker breaker, String label) {
             this.breaker = breaker;
             this.sortOrders = sortOrders;
+            logger.warn("ADDED " + memoryUsed(sortOrders.size()));
+            logger.warn("", new Throwable());
             breaker.addEstimateBytesAndMaybeBreak(memoryUsed(sortOrders.size()), label);
             this.endOffsets = new int[sortOrders.size()];
         }
@@ -160,6 +169,10 @@ public long ramBytesUsed() {
 
         @Override
         public void close() {
+//            assert open;
+            open = false;
+            logger.warn("REMOVING " + (-ramBytesUsed()));
+            logger.warn("", new Throwable());
             breaker.addWithoutBreaking(-ramBytesUsed());
         }
     }
@@ -180,6 +193,10 @@ static final class RowFiller {
                     page.getBlock(b)
                 );
             }
+            logger.warn("VALUE EXTRACTORS ARRAY LOOKS LIKE");
+            for (var v : valueExtractors) {
+                logger.warn("" + v);
+            }
             keyFactories = new KeyFactory[sortOrders.size()];
             for (int k = 0; k < keyFactories.length; k++) {
                 SortOrder so = sortOrders.get(k);
@@ -219,6 +236,7 @@ private void writeValues(int position, Row destination) {
                 if (refCounted != null) {
                     destination.setShardRefCountersAndShard(refCounted);
                 }
+                logger.warn("ENCODING VALUE FOR " + e +  " AT OFFSET " + destination.values.length());
                 e.writeValue(destination.values, position);
             }
         }
@@ -467,6 +485,11 @@ private Iterator<Page> toPages() {
                         );
                     }
                     p = 0;
+
+                    logger.warn("BUILDERS ARRAY LOOKS LIKE ");
+                    for (var b : builders) {
+                        logger.warn("" + b);
+                    }
                 }
 
                 try (Row row = list.get(i)) {
@@ -479,6 +502,7 @@ private Iterator<Page> toPages() {
                         }
                         keys.offset++;
                         keys.length--;
+                        logger.warn("DECODING KEY AT OFFSET " + keys.offset);
                         builders[so.channel].decodeKey(keys);
                     }
                     if (keys.length != 0) {
@@ -488,6 +512,7 @@ private Iterator<Page> toPages() {
                     BytesRef values = row.values.bytesRefView();
                     for (ResultBuilder builder : builders) {
                         builder.setNextRefCounted(row.shardRefCounter);
+                        logger.warn("DECODING VALUE FOR BUILDER " + builder + " AT OFFSET " + values.offset);
                         builder.decodeValue(values);
                     }
                     if (values.length != 0) {
@@ -517,6 +542,9 @@ private Iterator<Page> toPages() {
             assert builders == null;
             success = true;
             return result.iterator();
+        } catch (Throwable e) {
+            logger.warn("THREW AN EXCEPTION: ", e);
+            throw e;
         } finally {
             if (success == false) {
                 List<Releasable> close = new ArrayList<>(list);

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

@@ -24,6 +24,7 @@ public class ValueExtractorForAggregateMetricDouble implements ValueExtractor {
 
     @Override
     public void writeValue(BreakingBytesRefBuilder values, int position) {
+        TopNEncoder.DEFAULT_UNSORTABLE.encodeVInt(1, values);
         for (DoubleBlock doubleBlock : List.of(block.minBlock(), block.maxBlock(), block.sumBlock())) {
             if (doubleBlock.isNull(position)) {
                 TopNEncoder.DEFAULT_UNSORTABLE.encodeBoolean(false, values);

x-pack/plugin/esql/qa/server/multi-node/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/multi_node/GenerativeIT.java

@@ -7,10 +7,13 @@
 
 package org.elasticsearch.xpack.esql.qa.multi_node;
 
+import com.carrotsearch.randomizedtesting.annotations.Seed;
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
 
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.TestClustersThreadFilter;
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.rest.ESRestTestCase;
 import org.elasticsearch.xpack.esql.qa.rest.generative.GenerativeRestTest;
 import org.junit.ClassRule;
 
@@ -35,6 +38,17 @@ protected String getTestRestCluster() {
         return cluster.getHttpAddresses();
     }
 
+    @Override
+    protected final Settings restClientSettings() {
+        return Settings.builder()
+            .put(super.restClientSettings())
+            // increase the timeout here to 90 seconds to handle long waits for a green
+            // cluster health. the waits for green need to be longer than a minute to
+            // account for delayed shards
+            .put(ESRestTestCase.CLIENT_SOCKET_TIMEOUT, "10h")
+            .build();
+    }
+
     @Override
     protected boolean supportsSourceFieldMapping() {
         return false;