ESQL: Align RENAME behavior with EVAL for sequential processing

docs/changelog/122250.yaml

@@ -0,0 +1,6 @@
+pr: 122250
+summary: "ES|QL: Improve `RENAME` logic to swap names"
+area: ES|QL
+type: enhancement
+issues:
+  - 121739

x-pack/plugin/esql/qa/testFixtures/src/main/resources/rename.csv-spec

@@ -213,3 +213,29 @@ x:keyword
 Facello
 Simmel
 ;
+
+swappingNames
+FROM employees
+| SORT emp_no ASC
+| KEEP first_name, last_name
+| RENAME first_name AS last_name, last_name AS first_name, first_name as name
+| LIMIT 2
+;
+
+last_name:keyword   | name:keyword
+Georgi              | Facello
+Bezalel             | Simmel
+;
+
+complexSwappingNames
+FROM employees
+| SORT emp_no ASC
+| KEEP first_name, last_name, emp_no
+| RENAME first_name AS last_name, last_name AS first_name, first_name as emp_no, emp_no AS first_name
+| LIMIT 2
+;
+
+last_name:keyword   | emp_no:keyword    | first_name:INTEGER
+Georgi              | Facello           | 10001
+Bezalel             | Simmel            | 10002
+;

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java

@@ -897,11 +897,18 @@ public static List<NamedExpression> projectionsForRename(Rename rename, List<Att
             List<NamedExpression> unresolved = new ArrayList<>(renamingsCount);
             Map<String, String> reverseAliasing = new HashMap<>(renamingsCount); // `| rename a as x` => map(a: x)
 
+            Set<String> toRenameAttributes = rename.renamings()
+                .stream()
+                .map(Alias::child)
+                .filter(v -> v instanceof UnresolvedAttribute)
+                .map(attr -> ((UnresolvedAttribute) attr).name())
+                .collect(Collectors.toSet());
+
             rename.renamings().forEach(alias -> {
                 // skip NOPs: `| rename a as a`
                 if (alias.child() instanceof UnresolvedAttribute ua && alias.name().equals(ua.name()) == false) {
                     // remove attributes overwritten by a renaming: `| keep a, b, c | rename a as b`
-                    projections.removeIf(x -> x.name().equals(alias.name()));
+                    projections.removeIf(x -> x.name().equals(alias.name()) && toRenameAttributes.contains(x.name()) == false);
 
                     var resolved = maybeResolveAttribute(ua, childrenOutput, logger);
                     if (resolved instanceof UnsupportedAttribute || resolved.resolved()) {