forked from elastic/elasticsearch
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fix NameId Mapper in PlanStreamInput (elastic#99295)
This commit fixes the NameId mapper in PlanStreamInput. The current mapper function just materializes a NameId instance with the stream's primitive long value. This can lead to a potential issue during re-planning if the local data node planner just happens to use an id, retrieved automatically from the global counter, during the course of its planning. This can lead to confusion, where the same id is used for different logical attributes. The updated Mapper keeps a map of seen stream Ids to NameId instances. The no-args NameId constructor is used for absent entries, as it will automatically select and increment an id from the global counter, thus avoiding potential conflicts between the id in the stream and id's during local re-planning on the data node.
- Loading branch information
1 parent
e8f8a0a
commit d3f5584
Showing
4 changed files
with
123 additions
and
30 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
89 changes: 89 additions & 0 deletions
89
...lugin/esql/src/test/java/org/elasticsearch/xpack/esql/io/stream/PlanStreamInputTests.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,89 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License | ||
| * 2.0; you may not use this file except in compliance with the Elastic License | ||
| * 2.0. | ||
| */ | ||
|
|
||
| package org.elasticsearch.xpack.esql.io.stream; | ||
|
|
||
| import org.elasticsearch.test.ESTestCase; | ||
| import org.elasticsearch.xpack.ql.expression.NameId; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.HashSet; | ||
| import java.util.LinkedHashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
|
|
||
| import static org.hamcrest.Matchers.equalTo; | ||
| import static org.hamcrest.Matchers.is; | ||
| import static org.hamcrest.Matchers.not; | ||
|
|
||
| public class PlanStreamInputTests extends ESTestCase { | ||
|
|
||
| public void testMapperSimple() { | ||
| var mapper = new PlanStreamInput.NameIdMapper(); | ||
|
|
||
| NameId first = mapper.apply(1L); | ||
| NameId second = mapper.apply(1L); | ||
| assertThat(second, equalTo(first)); | ||
|
|
||
| NameId third = mapper.apply(2L); | ||
| NameId fourth = mapper.apply(2L); | ||
| assertThat(third, not(equalTo(second))); | ||
| assertThat(fourth, equalTo(third)); | ||
|
|
||
| assertThat(mapper.seen.size(), is(2)); | ||
| } | ||
|
|
||
| public void testMapper() { | ||
| List<Long> longs = randomLongsListOfSize(100); | ||
| List<Long> nameIds = new ArrayList<>(); | ||
| for (long l : longs) { | ||
| nameIds.add(l); | ||
| if (randomBoolean()) { // randomly insert additional values from the known list | ||
| int idx = randomIntBetween(0, longs.size() - 1); | ||
| nameIds.add(longs.get(idx)); | ||
| } | ||
| } | ||
|
|
||
| var mapper = new PlanStreamInput.NameIdMapper(); | ||
| List<NameId> mappedIds = nameIds.stream().map(mapper::apply).toList(); | ||
| assertThat(mappedIds.size(), is(nameIds.size())); | ||
| // there must be exactly 100 distinct elements | ||
| assertThat(mapper.seen.size(), is(100)); | ||
| assertThat(mappedIds.stream().distinct().count(), is(100L)); | ||
|
|
||
| // The pre-mapped name id pattern must match that of the mapped one | ||
| Map<Long, List<Long>> nameIdsSeen = new LinkedHashMap<>(); // insertion order | ||
| for (int i = 0; i < nameIds.size(); i++) { | ||
| long value = nameIds.get(i); | ||
| nameIdsSeen.computeIfAbsent(value, k -> new ArrayList<>()); | ||
| nameIdsSeen.get(value).add((long) i); | ||
| } | ||
| assert nameIdsSeen.size() == 100; | ||
|
|
||
| Map<NameId, List<Long>> mappedSeen = new LinkedHashMap<>(); // insertion order | ||
| for (int i = 0; i < mappedIds.size(); i++) { | ||
| NameId nameId = mappedIds.get(i); | ||
| mappedSeen.computeIfAbsent(nameId, k -> new ArrayList<>()); | ||
| mappedSeen.get(nameId).add((long) i); | ||
| } | ||
| assert mappedSeen.size() == 100; | ||
|
|
||
| var mappedSeenItr = mappedSeen.values().iterator(); | ||
| for (List<Long> indexes : nameIdsSeen.values()) { | ||
| assertThat(indexes, equalTo(mappedSeenItr.next())); | ||
| } | ||
| } | ||
|
|
||
| List<Long> randomLongsListOfSize(int size) { | ||
| Set<Long> longs = new HashSet<>(); | ||
| while (longs.size() < size) { | ||
| longs.add(randomLong()); | ||
| } | ||
| return longs.stream().toList(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters