Add assertion for exchanges

Author: dnhatn

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java

@@ -312,14 +312,14 @@ private PhysicalOperation planOutput(OutputExec outputExec, LocalExecutionPlanne
         return source.withSink(
             new OutputOperatorFactory(
                 Expressions.names(output),
-                alignPageToAttributes(output, source.layout),
+                alignPageToAttributes(false, output, source.layout),
                 outputExec.getPageConsumer()
             ),
             source.layout
         );
     }
 
-    private static Function<Page, Page> alignPageToAttributes(List<Attribute> attrs, Layout layout) {
+    private static Function<Page, Page> alignPageToAttributes(boolean exchange, List<Attribute> attrs, Layout layout) {
         // align the page layout with the operator output
         // extraction order - the list ordinal is the same as the column one
         // while the value represents the position in the original page
@@ -330,17 +330,21 @@ private static Function<Page, Page> alignPageToAttributes(List<Attribute> attrs,
             mappedPosition[++index] = layout.get(attribute.id()).channel();
             transformRequired |= mappedPosition[index] != index;
         }
-        Function<Page, Page> transformer = transformRequired ? p -> {
+        if (transformRequired == false) {
+            return Function.identity();
+        }
+        if (exchange) {
+            throw new AssertionError("require alignment attributes [" + attrs + "] layout [" + layout.inverse() + "]");
+        }
+        return p -> {
             var blocks = new Block[mappedPosition.length];
             for (int i = 0; i < blocks.length; i++) {
                 blocks[i] = p.getBlock(mappedPosition[i]);
                 blocks[i].incRef();
             }
             p.releaseBlocks();
             return new Page(blocks);
-        } : Function.identity();
-
-        return transformer;
+        };
     }
 
     private PhysicalOperation planExchange(ExchangeExec exchangeExec, LocalExecutionPlannerContext context) {
@@ -355,7 +359,7 @@ private PhysicalOperation planExchangeSink(ExchangeSinkExec exchangeSink, LocalE
 
         Function<Page, Page> transformer = exchangeSink.isIntermediateAgg()
             ? Function.identity()
-            : alignPageToAttributes(exchangeSink.output(), source.layout);
+            : alignPageToAttributes(true, exchangeSink.output(), source.layout);
 
         return source.withSink(new ExchangeSinkOperatorFactory(exchangeSinkSupplier, transformer), source.layout);
     }