Send source in Join serialization, and add tests

Author: ivancea

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

@@ -396,6 +396,53 @@ emp_no:integer | language_code:integer | language_name:keyword
 10005          | [-2, 13]              | null
 ;
 
+mvJoinKeyOnTheLookupIndexAfterStats
+required_capability: join_lookup_v12
+required_capability: join_lookup_skip_mv_warnings
+
+FROM employees
+| WHERE 10003 < emp_no AND emp_no < 10008
+| EVAL language_code = emp_no % 10
+| STATS BY emp_no, language_code
+| LOOKUP JOIN languages_lookup_non_unique_key ON language_code
+| SORT emp_no, language_name
+| KEEP emp_no, language_code, language_name
+;
+
+warning:Line 5:3: evaluation of [LOOKUP JOIN languages_lookup_non_unique_key ON language_code] failed, treating result as null. Only first 20 failures recorded.
+warning:Line 5:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value
+
+emp_no:integer | language_code:integer | language_name:keyword
+10004          | 4                     | Quenya
+10005          | 5                     | null
+10006          | 6                     | null
+10007          | 7                     | null
+;
+
+mvJoinKeyOnFromAfterStats
+required_capability: join_lookup_v12
+required_capability: join_lookup_skip_mv_warnings
+
+FROM employees
+| WHERE emp_no < 10006
+| EVAL language_code = salary_change.int
+| STATS language_code = VALUES(language_code) BY emp_no
+| LOOKUP JOIN languages_lookup ON language_code
+| SORT emp_no
+| KEEP emp_no, language_code, language_name
+;
+
+warning:Line 5:3: evaluation of [LOOKUP JOIN languages_lookup ON language_code] failed, treating result as null. Only first 20 failures recorded.
+warning:Line 5:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value
+
+emp_no:integer | language_code:integer | language_name:keyword
+10001          | 1                     | English              
+10002          | [-7, 11]              | null                 
+10003          | [12, 14]              | null                 
+10004          | [0, 1, 3, 13]         | null                 
+10005          | [-2, 13]              | null
+;
+
 mvJoinKeyFromRow
 required_capability: join_lookup_v12
 required_capability: join_lookup_skip_mv_warnings

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/join/Join.java

@@ -63,7 +63,7 @@ public Join(StreamInput in) throws IOException {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        Source.EMPTY.writeTo(out);
+        source().writeTo(out);
         out.writeNamedWriteable(left());
         out.writeNamedWriteable(right());
         config.writeTo(out);