Workaround to avoid NameId conflicts

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

@@ -23,6 +23,7 @@
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
 import org.elasticsearch.xpack.esql.session.EsqlConfiguration;
+import org.elasticsearch.xpack.ql.expression.NameId;
 
 import java.io.IOException;
 import java.util.List;
@@ -59,6 +60,8 @@ final class DataNodeRequest extends TransportRequest implements IndicesRequest {
         this.configuration = new EsqlConfiguration(in);
         this.shardIds = in.readCollectionAsList(ShardId::new);
         this.aliasFilters = in.readMap(Index::new, AliasFilter::readFrom);
+        // TODO: remove this shared local counter
+        NameId.advanceGlobalId(in.readVLong());
         this.plan = new PlanStreamInput(in, planNameRegistry, in.namedWriteableRegistry(), configuration).readPhysicalPlanNode();
     }
 
@@ -69,6 +72,8 @@ public void writeTo(StreamOutput out) throws IOException {
         configuration.writeTo(out);
         out.writeCollection(shardIds);
         out.writeMap(aliasFilters);
+        // TODO: remove this shared local counter
+        out.writeVLong(new NameId().getId());
         new PlanStreamOutput(out, planNameRegistry).writePhysicalPlanNode(plan);
     }
 

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/expression/NameId.java

@@ -30,6 +30,23 @@ public NameId(long id) {
         this.id = id;
     }
 
+    /**
+     * TODO:
+     * This is a workaround to prevent NameId conflicts in multi-node environments. Currently, when a NameId is serialized,
+     * its local ID is also serialized to the remote node. If the remote node generates a new NameId, it could potentially
+     * have the same ID as an existing one that was deserialized, leading to an incorrect local plan on the remote node.
+     * <p>
+     * It's important to note that while conflicts might still occur between multiple computations with this approach, there
+     * is no conflicts within a single computation. We only need to ensure no conflict in a single computation to ensure the correctness.
+     */
+    public static void advanceGlobalId(long other) {
+        COUNTER.accumulateAndGet(other, Math::max);
+    }
+
+    public long getId() {
+        return id;
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(id);