ESQL: Ensure non-zero row size in EstimatesRowSize (elastic#122762) (elastic#123958)

Closes elastic#121535

Co-authored-by: kanoshiou <[email protected]>

Author: dnhatn

docs/changelog/122762.yaml

@@ -0,0 +1,6 @@
+pr: 122762
+summary: "ESQL: Remove estimated row size assertion"
+area: ES|QL
+type: bug
+issues:
+  - 121535

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlActionIT.java

@@ -587,6 +587,28 @@ public void testFilterWithNullAndEval() {
         }
     }
 
+    public void testSortWithNull() {
+        try (EsqlQueryResponse results = run("row a = null | sort a")) {
+            logger.info(results);
+            assertEquals(1, getValuesList(results).size());
+            int countIndex = results.columns().indexOf(new ColumnInfoImpl("a", "null"));
+            assertThat(results.columns().stream().map(ColumnInfo::name).toList(), contains("a"));
+            assertThat(results.columns().stream().map(ColumnInfoImpl::type).toList(), contains(DataType.NULL));
+            assertNull(getValuesList(results).getFirst().get(countIndex));
+        }
+    }
+
+    public void testStatsByNull() {
+        try (EsqlQueryResponse results = run("row a = null | stats by a")) {
+            logger.info(results);
+            assertEquals(1, getValuesList(results).size());
+            int countIndex = results.columns().indexOf(new ColumnInfoImpl("a", "null"));
+            assertThat(results.columns().stream().map(ColumnInfo::name).toList(), contains("a"));
+            assertThat(results.columns().stream().map(ColumnInfoImpl::type).toList(), contains(DataType.NULL));
+            assertNull(getValuesList(results).getFirst().get(countIndex));
+        }
+    }
+
     public void testStringLength() {
         try (EsqlQueryResponse results = run("from test | eval l = length(color)")) {
             logger.info(results);

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/AggregateExec.java

@@ -136,6 +136,7 @@ public Integer estimatedRowSize() {
     public PhysicalPlan estimateRowSize(State state) {
         state.add(false, aggregates);  // The groupings are contained within the aggregates
         int size = state.consumeAllFields(true);
+        size = Math.max(size, 1);
         return Objects.equals(this.estimatedRowSize, size)
             ? this
             : new AggregateExec(source(), child(), groupings, aggregates, mode, intermediateAttributes, size);

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/TopNExec.java

@@ -101,6 +101,7 @@ public PhysicalPlan estimateRowSize(State state) {
         final boolean needsSortedDocIds = output.stream().anyMatch(a -> a.dataType() == DataType.DOC_DATA_TYPE);
         state.add(needsSortedDocIds, output);
         int size = state.consumeAllFields(true);
+        size = Math.max(size, 1);
         return Objects.equals(this.estimatedRowSize, size) ? this : new TopNExec(source(), child(), order, limit, size);
     }