001 /*
002 * Copyright (c) 2007-2014 Concurrent, Inc. All Rights Reserved.
003 *
004 * Project and contact information: http://www.cascading.org/
005 *
006 * This file is part of the Cascading project.
007 *
008 * Licensed under the Apache License, Version 2.0 (the "License");
009 * you may not use this file except in compliance with the License.
010 * You may obtain a copy of the License at
011 *
012 * http://www.apache.org/licenses/LICENSE-2.0
013 *
014 * Unless required by applicable law or agreed to in writing, software
015 * distributed under the License is distributed on an "AS IS" BASIS,
016 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
017 * See the License for the specific language governing permissions and
018 * limitations under the License.
019 */
020
021 package cascading.pipe;
022
023 import java.lang.reflect.Type;
024 import java.util.ArrayList;
025 import java.util.Comparator;
026 import java.util.HashMap;
027 import java.util.HashSet;
028 import java.util.Iterator;
029 import java.util.LinkedHashMap;
030 import java.util.List;
031 import java.util.Map;
032 import java.util.Set;
033
034 import cascading.flow.FlowElement;
035 import cascading.flow.planner.DeclaresResults;
036 import cascading.flow.planner.Scope;
037 import cascading.pipe.joiner.BufferJoin;
038 import cascading.pipe.joiner.InnerJoin;
039 import cascading.pipe.joiner.Joiner;
040 import cascading.tuple.Fields;
041 import cascading.tuple.FieldsResolverException;
042 import cascading.tuple.TupleException;
043 import cascading.tuple.coerce.Coercions;
044 import cascading.tuple.type.CoercibleType;
045 import cascading.util.Util;
046
047 import static java.util.Arrays.asList;
048
049 /**
050 * The base class for {@link GroupBy}, {@link CoGroup}, {@link Merge}, and {@link HashJoin}. This class should not be used directly.
051 *
052 * @see GroupBy
053 * @see CoGroup
054 * @see Merge
055 * @see HashJoin
056 */
057 public class Splice extends Pipe
058 {
059 static enum Kind
060 {
061 GroupBy, CoGroup, Merge, Join
062 }
063
064 private Kind kind;
065 /** Field spliceName */
066 private String spliceName;
067 /** Field pipes */
068 private final List<Pipe> pipes = new ArrayList<Pipe>();
069 /** Field groupFieldsMap */
070 protected final Map<String, Fields> keyFieldsMap = new LinkedHashMap<String, Fields>(); // keep order
071 /** Field sortFieldsMap */
072 protected Map<String, Fields> sortFieldsMap = new LinkedHashMap<String, Fields>(); // keep order
073 /** Field reverseOrder */
074 private boolean reverseOrder = false;
075 /** Field declaredFields */
076 protected Fields declaredFields;
077 /** Field resultGroupFields */
078 protected Fields resultGroupFields;
079 /** Field repeat */
080 private int numSelfJoins = 0;
081 /** Field coGrouper */
082 private Joiner joiner;
083
084 /** Field pipePos */
085 private transient Map<String, Integer> pipePos;
086
087 /**
088 * Constructor Splice creates a new Splice instance.
089 *
090 * @param lhs of type Pipe
091 * @param lhsGroupFields of type Fields
092 * @param rhs of type Pipe
093 * @param rhsGroupFields of type Fields
094 * @param declaredFields of type Fields
095 */
096 protected Splice( Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields, Fields declaredFields )
097 {
098 this( lhs, lhsGroupFields, rhs, rhsGroupFields, declaredFields, null, null );
099 }
100
101 /**
102 * Constructor Splice creates a new Splice instance.
103 *
104 * @param lhs of type Pipe
105 * @param lhsGroupFields of type Fields
106 * @param rhs of type Pipe
107 * @param rhsGroupFields of type Fields
108 * @param declaredFields of type Fields
109 * @param resultGroupFields of type Fields
110 */
111 protected Splice( Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields, Fields declaredFields, Fields resultGroupFields )
112 {
113 this( lhs, lhsGroupFields, rhs, rhsGroupFields, declaredFields, resultGroupFields, null );
114 }
115
116 /**
117 * Constructor Splice creates a new Splice instance.
118 *
119 * @param lhs of type Pipe
120 * @param lhsGroupFields of type Fields
121 * @param rhs of type Pipe
122 * @param rhsGroupFields of type Fields
123 * @param declaredFields of type Fields
124 * @param joiner of type CoGrouper
125 */
126 protected Splice( Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields, Fields declaredFields, Joiner joiner )
127 {
128 this( Pipe.pipes( lhs, rhs ), Fields.fields( lhsGroupFields, rhsGroupFields ), declaredFields, joiner );
129 }
130
131 /**
132 * Constructor Splice creates a new Splice instance.
133 *
134 * @param lhs of type Pipe
135 * @param lhsGroupFields of type Fields
136 * @param rhs of type Pipe
137 * @param rhsGroupFields of type Fields
138 * @param declaredFields of type Fields
139 * @param resultGroupFields of type Fields
140 * @param joiner of type Joiner
141 */
142 protected Splice( Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields, Fields declaredFields, Fields resultGroupFields, Joiner joiner )
143 {
144 this( Pipe.pipes( lhs, rhs ), Fields.fields( lhsGroupFields, rhsGroupFields ), declaredFields, resultGroupFields, joiner );
145 }
146
147 /**
148 * Constructor Splice creates a new Splice instance.
149 *
150 * @param lhs of type Pipe
151 * @param lhsGroupFields of type Fields
152 * @param rhs of type Pipe
153 * @param rhsGroupFields of type Fields
154 * @param joiner of type CoGrouper
155 */
156 protected Splice( Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields, Joiner joiner )
157 {
158 this( lhs, lhsGroupFields, rhs, rhsGroupFields, null, joiner );
159 }
160
161 /**
162 * Constructor Splice creates a new Splice instance.
163 *
164 * @param lhs of type Pipe
165 * @param lhsGroupFields of type Fields
166 * @param rhs of type Pipe
167 * @param rhsGroupFields of type Fields
168 */
169 protected Splice( Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields )
170 {
171 this( Pipe.pipes( lhs, rhs ), Fields.fields( lhsGroupFields, rhsGroupFields ) );
172 }
173
174 /**
175 * Constructor Splice creates a new Splice instance.
176 *
177 * @param pipes of type Pipe...
178 */
179 protected Splice( Pipe... pipes )
180 {
181 this( pipes, (Fields[]) null );
182 }
183
184 /**
185 * Constructor Splice creates a new Splice instance.
186 *
187 * @param pipes of type Pipe[]
188 * @param groupFields of type Fields[]
189 */
190 protected Splice( Pipe[] pipes, Fields[] groupFields )
191 {
192 this( null, pipes, groupFields, null, null );
193 }
194
195
196 /**
197 * Constructor Splice creates a new Splice instance.
198 *
199 * @param spliceName of type String
200 * @param pipes of type Pipe[]
201 * @param groupFields of type Fields[]
202 */
203 protected Splice( String spliceName, Pipe[] pipes, Fields[] groupFields )
204 {
205 this( spliceName, pipes, groupFields, null, null );
206 }
207
208 /**
209 * Constructor Splice creates a new Splice instance.
210 *
211 * @param spliceName of type String
212 * @param pipes of type Pipe[]
213 * @param groupFields of type Fields[]
214 * @param declaredFields of type Fields
215 */
216 protected Splice( String spliceName, Pipe[] pipes, Fields[] groupFields, Fields declaredFields )
217 {
218 this( spliceName, pipes, groupFields, declaredFields, null );
219 }
220
221 /**
222 * Constructor Splice creates a new Splice instance.
223 *
224 * @param spliceName of type String
225 * @param pipes of type Pipe[]
226 * @param groupFields of type Fields[]
227 * @param declaredFields of type Fields
228 * @param resultGroupFields of type Fields
229 */
230 protected Splice( String spliceName, Pipe[] pipes, Fields[] groupFields, Fields declaredFields, Fields resultGroupFields )
231 {
232 this( spliceName, pipes, groupFields, declaredFields, resultGroupFields, null );
233 }
234
235 /**
236 * Constructor Splice creates a new Splice instance.
237 *
238 * @param pipes of type Pipe[]
239 * @param groupFields of type Fields[]
240 * @param declaredFields of type Fields
241 * @param joiner of type CoGrouper
242 */
243 protected Splice( Pipe[] pipes, Fields[] groupFields, Fields declaredFields, Joiner joiner )
244 {
245 this( null, pipes, groupFields, declaredFields, null, joiner );
246 }
247
248 /**
249 * Constructor Splice creates a new Splice instance.
250 *
251 * @param pipes of type Pipe[]
252 * @param groupFields of type Fields[]
253 * @param declaredFields of type Fields
254 * @param resultGroupFields of type Fields
255 * @param joiner of type Joiner
256 */
257 protected Splice( Pipe[] pipes, Fields[] groupFields, Fields declaredFields, Fields resultGroupFields, Joiner joiner )
258 {
259 this( null, pipes, groupFields, declaredFields, resultGroupFields, joiner );
260 }
261
262 /**
263 * Constructor Splice creates a new Splice instance.
264 *
265 * @param spliceName of type String
266 * @param pipes of type Pipe[]
267 * @param groupFields of type Fields[]
268 * @param declaredFields of type Fields
269 * @param joiner of type CoGrouper
270 */
271 protected Splice( String spliceName, Pipe[] pipes, Fields[] groupFields, Fields declaredFields, Fields resultGroupFields, Joiner joiner )
272 {
273 setKind();
274 this.spliceName = spliceName;
275
276 int uniques = new HashSet<Pipe>( asList( Pipe.resolvePreviousAll( pipes ) ) ).size();
277
278 if( pipes.length > 1 && uniques == 1 )
279 {
280 if( new HashSet<Fields>( asList( groupFields ) ).size() != 1 )
281 throw new IllegalArgumentException( "all groupFields must be identical" );
282
283 addPipe( pipes[ 0 ] );
284 this.numSelfJoins = pipes.length - 1;
285 this.keyFieldsMap.put( pipes[ 0 ].getName(), groupFields[ 0 ] );
286
287 if( resultGroupFields != null && groupFields[ 0 ].size() * pipes.length != resultGroupFields.size() )
288 throw new IllegalArgumentException( "resultGroupFields and cogroup joined fields must be same size" );
289 }
290 else
291 {
292 int last = -1;
293 for( int i = 0; i < pipes.length; i++ )
294 {
295 addPipe( pipes[ i ] );
296
297 if( groupFields == null || groupFields.length == 0 )
298 {
299 addGroupFields( pipes[ i ], Fields.FIRST );
300 continue;
301 }
302
303 if( last != -1 && last != groupFields[ i ].size() )
304 throw new IllegalArgumentException( "all groupFields must be same size" );
305
306 last = groupFields[ i ].size();
307 addGroupFields( pipes[ i ], groupFields[ i ] );
308 }
309
310 if( resultGroupFields != null && last * pipes.length != resultGroupFields.size() )
311 throw new IllegalArgumentException( "resultGroupFields and cogroup resulting joined fields must be same size" );
312 }
313
314 this.declaredFields = declaredFields;
315 this.resultGroupFields = resultGroupFields;
316 this.joiner = joiner;
317
318 verifyCoGrouper();
319 }
320
321 /**
322 * Constructor Splice creates a new Splice instance.
323 *
324 * @param spliceName of type String
325 * @param lhs of type Pipe
326 * @param lhsGroupFields of type Fields
327 * @param rhs of type Pipe
328 * @param rhsGroupFields of type Fields
329 * @param declaredFields of type Fields
330 */
331 protected Splice( String spliceName, Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields, Fields declaredFields )
332 {
333 this( lhs, lhsGroupFields, rhs, rhsGroupFields, declaredFields );
334 this.spliceName = spliceName;
335 }
336
337 /**
338 * Constructor Splice creates a new Splice instance.
339 *
340 * @param spliceName of type String
341 * @param lhs of type Pipe
342 * @param lhsGroupFields of type Fields
343 * @param rhs of type Pipe
344 * @param rhsGroupFields of type Fields
345 * @param declaredFields of type Fields
346 * @param resultGroupFields of type Fields
347 */
348 protected Splice( String spliceName, Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields, Fields declaredFields, Fields resultGroupFields )
349 {
350 this( lhs, lhsGroupFields, rhs, rhsGroupFields, declaredFields, resultGroupFields );
351 this.spliceName = spliceName;
352 }
353
354 /**
355 * Constructor Splice creates a new Splice instance.
356 *
357 * @param spliceName of type String
358 * @param lhs of type Pipe
359 * @param lhsGroupFields of type Fields
360 * @param rhs of type Pipe
361 * @param rhsGroupFields of type Fields
362 * @param declaredFields of type Fields
363 * @param joiner of type CoGrouper
364 */
365 protected Splice( String spliceName, Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields, Fields declaredFields, Joiner joiner )
366 {
367 this( lhs, lhsGroupFields, rhs, rhsGroupFields, declaredFields, joiner );
368 this.spliceName = spliceName;
369 }
370
371 /**
372 * Constructor Splice creates a new Splice instance.
373 *
374 * @param spliceName of type String
375 * @param lhs of type Pipe
376 * @param lhsGroupFields of type Fields
377 * @param rhs of type Pipe
378 * @param rhsGroupFields of type Fields
379 * @param declaredFields of type Fields
380 * @param resultGroupFields of type Fields
381 * @param joiner of type Joiner
382 */
383 protected Splice( String spliceName, Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields, Fields declaredFields, Fields resultGroupFields, Joiner joiner )
384 {
385 this( lhs, lhsGroupFields, rhs, rhsGroupFields, declaredFields, resultGroupFields, joiner );
386 this.spliceName = spliceName;
387 }
388
389 /**
390 * Constructor Splice creates a new Splice instance.
391 *
392 * @param spliceName of type String
393 * @param lhs of type Pipe
394 * @param lhsGroupFields of type Fields
395 * @param rhs of type Pipe
396 * @param rhsGroupFields of type Fields
397 * @param joiner of type CoGrouper
398 */
399 protected Splice( String spliceName, Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields, Joiner joiner )
400 {
401 this( lhs, lhsGroupFields, rhs, rhsGroupFields, joiner );
402 this.spliceName = spliceName;
403 }
404
405 /**
406 * Constructor Splice creates a new Splice instance.
407 *
408 * @param spliceName of type String
409 * @param lhs of type Pipe
410 * @param lhsGroupFields of type Fields
411 * @param rhs of type Pipe
412 * @param rhsGroupFields of type Fields
413 */
414 protected Splice( String spliceName, Pipe lhs, Fields lhsGroupFields, Pipe rhs, Fields rhsGroupFields )
415 {
416 this( lhs, lhsGroupFields, rhs, rhsGroupFields );
417 this.spliceName = spliceName;
418 }
419
420 /**
421 * Constructor Splice creates a new Splice instance.
422 *
423 * @param spliceName of type String
424 * @param pipes of type Pipe...
425 */
426 protected Splice( String spliceName, Pipe... pipes )
427 {
428 this( pipes );
429 this.spliceName = spliceName;
430 }
431
432 /**
433 * Constructor Splice creates a new Splice instance.
434 *
435 * @param pipe of type Pipe
436 * @param groupFields of type Fields
437 * @param numSelfJoins of type int
438 * @param declaredFields of type Fields
439 */
440 protected Splice( Pipe pipe, Fields groupFields, int numSelfJoins, Fields declaredFields )
441 {
442 this( pipe, groupFields, numSelfJoins );
443 this.declaredFields = declaredFields;
444 }
445
446 /**
447 * Constructor Splice creates a new Splice instance.
448 *
449 * @param pipe of type Pipe
450 * @param groupFields of type Fields
451 * @param numSelfJoins of type int
452 * @param declaredFields of type Fields
453 * @param resultGroupFields of type Fields
454 */
455 protected Splice( Pipe pipe, Fields groupFields, int numSelfJoins, Fields declaredFields, Fields resultGroupFields )
456 {
457 this( pipe, groupFields, numSelfJoins );
458 this.declaredFields = declaredFields;
459 this.resultGroupFields = resultGroupFields;
460
461 if( resultGroupFields != null && groupFields.size() * numSelfJoins != resultGroupFields.size() )
462 throw new IllegalArgumentException( "resultGroupFields and cogroup resulting join fields must be same size" );
463 }
464
465 /**
466 * Constructor Splice creates a new Splice instance.
467 *
468 * @param pipe of type Pipe
469 * @param groupFields of type Fields
470 * @param numSelfJoins of type int
471 * @param declaredFields of type Fields
472 * @param joiner of type CoGrouper
473 */
474 protected Splice( Pipe pipe, Fields groupFields, int numSelfJoins, Fields declaredFields, Joiner joiner )
475 {
476 this( pipe, groupFields, numSelfJoins, declaredFields );
477 this.joiner = joiner;
478
479 verifyCoGrouper();
480 }
481
482 /**
483 * Constructor Splice creates a new Splice instance.
484 *
485 * @param pipe of type Pipe
486 * @param groupFields of type Fields
487 * @param numSelfJoins of type int
488 * @param declaredFields of type Fields
489 * @param resultGroupFields of type Fields
490 * @param joiner of type Joiner
491 */
492 protected Splice( Pipe pipe, Fields groupFields, int numSelfJoins, Fields declaredFields, Fields resultGroupFields, Joiner joiner )
493 {
494 this( pipe, groupFields, numSelfJoins, declaredFields, resultGroupFields );
495 this.joiner = joiner;
496
497 verifyCoGrouper();
498 }
499
500 /**
501 * Constructor Splice creates a new Splice instance.
502 *
503 * @param pipe of type Pipe
504 * @param groupFields of type Fields
505 * @param numSelfJoins of type int
506 * @param joiner of type CoGrouper
507 */
508 protected Splice( Pipe pipe, Fields groupFields, int numSelfJoins, Joiner joiner )
509 {
510 setKind();
511 addPipe( pipe );
512 this.keyFieldsMap.put( pipe.getName(), groupFields );
513 this.numSelfJoins = numSelfJoins;
514 this.joiner = joiner;
515
516 verifyCoGrouper();
517 }
518
519 /**
520 * Constructor Splice creates a new Splice instance.
521 *
522 * @param pipe of type Pipe
523 * @param groupFields of type Fields
524 * @param numSelfJoins of type int
525 */
526 protected Splice( Pipe pipe, Fields groupFields, int numSelfJoins )
527 {
528 this( pipe, groupFields, numSelfJoins, (Joiner) null );
529 }
530
531 /**
532 * Constructor Splice creates a new Splice instance.
533 *
534 * @param spliceName of type String
535 * @param pipe of type Pipe
536 * @param groupFields of type Fields
537 * @param numSelfJoins of type int
538 * @param declaredFields of type Fields
539 */
540 protected Splice( String spliceName, Pipe pipe, Fields groupFields, int numSelfJoins, Fields declaredFields )
541 {
542 this( pipe, groupFields, numSelfJoins, declaredFields );
543 this.spliceName = spliceName;
544 }
545
546 /**
547 * Constructor Splice creates a new Splice instance.
548 *
549 * @param spliceName of type String
550 * @param pipe of type Pipe
551 * @param groupFields of type Fields
552 * @param numSelfJoins of type int
553 * @param declaredFields of type Fields
554 * @param resultGroupFields of type Fields
555 */
556 protected Splice( String spliceName, Pipe pipe, Fields groupFields, int numSelfJoins, Fields declaredFields, Fields resultGroupFields )
557 {
558 this( pipe, groupFields, numSelfJoins, declaredFields, resultGroupFields );
559 this.spliceName = spliceName;
560 }
561
562 /**
563 * Constructor Splice creates a new Splice instance.
564 *
565 * @param spliceName of type String
566 * @param pipe of type Pipe
567 * @param groupFields of type Fields
568 * @param numSelfJoins of type int
569 * @param declaredFields of type Fields
570 * @param joiner of type CoGrouper
571 */
572 protected Splice( String spliceName, Pipe pipe, Fields groupFields, int numSelfJoins, Fields declaredFields, Joiner joiner )
573 {
574 this( pipe, groupFields, numSelfJoins, declaredFields, joiner );
575 this.spliceName = spliceName;
576 }
577
578 /**
579 * Constructor Splice creates a new Splice instance.
580 *
581 * @param spliceName of type String
582 * @param pipe of type Pipe
583 * @param groupFields of type Fields
584 * @param numSelfJoins of type int
585 * @param declaredFields of type Fields
586 * @param resultGroupFields of type Fields
587 * @param joiner of type Joiner
588 */
589 protected Splice( String spliceName, Pipe pipe, Fields groupFields, int numSelfJoins, Fields declaredFields, Fields resultGroupFields, Joiner joiner )
590 {
591 this( pipe, groupFields, numSelfJoins, declaredFields, resultGroupFields, joiner );
592 this.spliceName = spliceName;
593 }
594
595 /**
596 * Constructor Splice creates a new Splice instance.
597 *
598 * @param spliceName of type String
599 * @param pipe of type Pipe
600 * @param groupFields of type Fields
601 * @param numSelfJoins of type int
602 * @param joiner of type CoGrouper
603 */
604 protected Splice( String spliceName, Pipe pipe, Fields groupFields, int numSelfJoins, Joiner joiner )
605 {
606 this( pipe, groupFields, numSelfJoins, joiner );
607 this.spliceName = spliceName;
608 }
609
610 /**
611 * Constructor Splice creates a new Splice instance.
612 *
613 * @param spliceName of type String
614 * @param pipe of type Pipe
615 * @param groupFields of type Fields
616 * @param numSelfJoins of type int
617 */
618 protected Splice( String spliceName, Pipe pipe, Fields groupFields, int numSelfJoins )
619 {
620 this( pipe, groupFields, numSelfJoins );
621 this.spliceName = spliceName;
622 }
623
624 ////////////
625 // GROUPBY
626 ////////////
627
628 /**
629 * Constructor Splice creates a new Splice instance where grouping occurs on {@link Fields#ALL} fields.
630 *
631 * @param pipe of type Pipe
632 */
633 protected Splice( Pipe pipe )
634 {
635 this( null, pipe, Fields.ALL, null, false );
636 }
637
638 /**
639 * Constructor Splice creates a new Splice instance.
640 *
641 * @param pipe of type Pipe
642 * @param groupFields of type Fields
643 */
644 protected Splice( Pipe pipe, Fields groupFields )
645 {
646 this( null, pipe, groupFields, null, false );
647 }
648
649 /**
650 * Constructor Splice creates a new Splice instance.
651 *
652 * @param spliceName of type String
653 * @param pipe of type Pipe
654 * @param groupFields of type Fields
655 */
656 protected Splice( String spliceName, Pipe pipe, Fields groupFields )
657 {
658 this( spliceName, pipe, groupFields, null, false );
659 }
660
661 /**
662 * Constructor Splice creates a new Splice instance.
663 *
664 * @param pipe of type Pipe
665 * @param groupFields of type Fields
666 * @param sortFields of type Fields
667 */
668 protected Splice( Pipe pipe, Fields groupFields, Fields sortFields )
669 {
670 this( null, pipe, groupFields, sortFields, false );
671 }
672
673 /**
674 * Constructor Splice creates a new Splice instance.
675 *
676 * @param spliceName of type String
677 * @param pipe of type Pipe
678 * @param groupFields of type Fields
679 * @param sortFields of type Fields
680 */
681 protected Splice( String spliceName, Pipe pipe, Fields groupFields, Fields sortFields )
682 {
683 this( spliceName, pipe, groupFields, sortFields, false );
684 }
685
686 /**
687 * Constructor Splice creates a new Splice instance.
688 *
689 * @param pipe of type Pipe
690 * @param groupFields of type Fields
691 * @param sortFields of type Fields
692 * @param reverseOrder of type boolean
693 */
694 protected Splice( Pipe pipe, Fields groupFields, Fields sortFields, boolean reverseOrder )
695 {
696 this( null, pipe, groupFields, sortFields, reverseOrder );
697 }
698
699 /**
700 * Constructor Splice creates a new Splice instance.
701 *
702 * @param spliceName of type String
703 * @param pipe of type Pipe
704 * @param groupFields of type Fields
705 * @param sortFields of type Fields
706 * @param reverseOrder of type boolean
707 */
708 protected Splice( String spliceName, Pipe pipe, Fields groupFields, Fields sortFields, boolean reverseOrder )
709 {
710 this( spliceName, Pipe.pipes( pipe ), groupFields, sortFields, reverseOrder );
711 }
712
713 /**
714 * Constructor Splice creates a new Splice instance.
715 *
716 * @param pipes of type Pipe
717 * @param groupFields of type Fields
718 */
719 protected Splice( Pipe[] pipes, Fields groupFields )
720 {
721 this( null, pipes, groupFields, null, false );
722 }
723
724 /**
725 * Constructor Splice creates a new Splice instance.
726 *
727 * @param spliceName of type String
728 * @param pipes of type Pipe
729 * @param groupFields of type Fields
730 */
731 protected Splice( String spliceName, Pipe[] pipes, Fields groupFields )
732 {
733 this( spliceName, pipes, groupFields, null, false );
734 }
735
736 /**
737 * Constructor Splice creates a new Splice instance.
738 *
739 * @param pipes of type Pipe
740 * @param groupFields of type Fields
741 * @param sortFields of type Fields
742 */
743 protected Splice( Pipe[] pipes, Fields groupFields, Fields sortFields )
744 {
745 this( null, pipes, groupFields, sortFields, false );
746 }
747
748 /**
749 * Constructor Splice creates a new Splice instance.
750 *
751 * @param spliceName of type String
752 * @param pipe of type Pipe
753 * @param groupFields of type Fields
754 * @param sortFields of type Fields
755 */
756 protected Splice( String spliceName, Pipe[] pipe, Fields groupFields, Fields sortFields )
757 {
758 this( spliceName, pipe, groupFields, sortFields, false );
759 }
760
761 /**
762 * Constructor Splice creates a new Splice instance.
763 *
764 * @param pipes of type Pipe
765 * @param groupFields of type Fields
766 * @param sortFields of type Fields
767 * @param reverseOrder of type boolean
768 */
769 protected Splice( Pipe[] pipes, Fields groupFields, Fields sortFields, boolean reverseOrder )
770 {
771 this( null, pipes, groupFields, sortFields, reverseOrder );
772 }
773
774 /**
775 * Constructor Splice creates a new Splice instance.
776 *
777 * @param spliceName of type String
778 * @param pipes of type Pipe[]
779 * @param groupFields of type Fields
780 * @param sortFields of type Fields
781 * @param reverseOrder of type boolean
782 */
783 protected Splice( String spliceName, Pipe[] pipes, Fields groupFields, Fields sortFields, boolean reverseOrder )
784 {
785 setKind();
786 this.spliceName = spliceName;
787
788 for( Pipe pipe : pipes )
789 {
790 addPipe( pipe );
791 this.keyFieldsMap.put( pipe.getName(), groupFields );
792
793 if( sortFields != null )
794 this.sortFieldsMap.put( pipe.getName(), sortFields );
795 }
796
797 this.reverseOrder = reverseOrder;
798 this.joiner = new InnerJoin();
799 }
800
801 private void verifyCoGrouper()
802 {
803 if( isJoin() && joiner instanceof BufferJoin )
804 throw new IllegalArgumentException( "invalid joiner, may not use BufferJoiner in a HashJoin" );
805
806 if( joiner == null )
807 {
808 joiner = new InnerJoin();
809 return;
810 }
811
812 if( joiner.numJoins() == -1 )
813 return;
814
815 int joins = Math.max( numSelfJoins, keyFieldsMap.size() - 1 ); // joining two streams is one join
816
817 if( joins != joiner.numJoins() )
818 throw new IllegalArgumentException( "invalid joiner, only accepts " + joiner.numJoins() + " joins, there are: " + joins );
819 }
820
821 private void setKind()
822 {
823 if( this instanceof GroupBy )
824 kind = Kind.GroupBy;
825 else if( this instanceof CoGroup )
826 kind = Kind.CoGroup;
827 else if( this instanceof Merge )
828 kind = Kind.Merge;
829 else
830 kind = Kind.Join;
831 }
832
833 /**
834 * Method getDeclaredFields returns the declaredFields of this Splice object.
835 *
836 * @return the declaredFields (type Fields) of this Splice object.
837 */
838 public Fields getDeclaredFields()
839 {
840 return declaredFields;
841 }
842
843 private void addPipe( Pipe pipe )
844 {
845 if( pipe.getName() == null )
846 throw new IllegalArgumentException( "each input pipe must have a name" );
847
848 pipes.add( pipe ); // allow same pipe
849 }
850
851 private void addGroupFields( Pipe pipe, Fields fields )
852 {
853 if( keyFieldsMap.containsKey( pipe.getName() ) )
854 throw new IllegalArgumentException( "each input pipe branch must be uniquely named" );
855
856 keyFieldsMap.put( pipe.getName(), fields );
857 }
858
859 @Override
860 public String getName()
861 {
862 if( spliceName != null )
863 return spliceName;
864
865 StringBuffer buffer = new StringBuffer();
866
867 for( Pipe pipe : pipes )
868 {
869 if( buffer.length() != 0 )
870 {
871 if( isGroupBy() || isMerge() )
872 buffer.append( "+" );
873 else if( isCoGroup() || isJoin() )
874 buffer.append( "*" ); // more semantically correct
875 }
876
877 buffer.append( pipe.getName() );
878 }
879
880 spliceName = buffer.toString();
881
882 return spliceName;
883 }
884
885 @Override
886 public Pipe[] getPrevious()
887 {
888 return pipes.toArray( new Pipe[ pipes.size() ] );
889 }
890
891 /**
892 * Method getGroupingSelectors returns the groupingSelectors of this Splice object.
893 *
894 * @return the groupingSelectors (type Map<String, Fields>) of this Splice object.
895 */
896 public Map<String, Fields> getKeySelectors()
897 {
898 return keyFieldsMap;
899 }
900
901 /**
902 * Method getSortingSelectors returns the sortingSelectors of this Splice object.
903 *
904 * @return the sortingSelectors (type Map<String, Fields>) of this Splice object.
905 */
906 public Map<String, Fields> getSortingSelectors()
907 {
908 return sortFieldsMap;
909 }
910
911 /**
912 * Method isSorted returns true if this Splice instance is sorting values other than the group fields.
913 *
914 * @return the sorted (type boolean) of this Splice object.
915 */
916 public boolean isSorted()
917 {
918 return !sortFieldsMap.isEmpty();
919 }
920
921 /**
922 * Method isSortReversed returns true if sorting is reversed.
923 *
924 * @return the sortReversed (type boolean) of this Splice object.
925 */
926 public boolean isSortReversed()
927 {
928 return reverseOrder;
929 }
930
931 public synchronized Map<String, Integer> getPipePos()
932 {
933 if( pipePos != null )
934 return pipePos;
935
936 pipePos = new HashMap<String, Integer>();
937
938 int pos = 0;
939 for( Object pipe : pipes )
940 pipePos.put( ( (Pipe) pipe ).getName(), pos++ );
941
942 return pipePos;
943 }
944
945 public Joiner getJoiner()
946 {
947 return joiner;
948 }
949
950 /**
951 * Method isGroupBy returns true if this Splice instance will perform a GroupBy operation.
952 *
953 * @return the groupBy (type boolean) of this Splice object.
954 */
955 public final boolean isGroupBy()
956 {
957 return kind == Kind.GroupBy;
958 }
959
960 public final boolean isCoGroup()
961 {
962 return kind == Kind.CoGroup;
963 }
964
965 public final boolean isMerge()
966 {
967 return kind == Kind.Merge;
968 }
969
970 public final boolean isJoin()
971 {
972 return kind == Kind.Join;
973 }
974
975 public int getNumSelfJoins()
976 {
977 return numSelfJoins;
978 }
979
980 boolean isSelfJoin()
981 {
982 return numSelfJoins != 0;
983 }
984
985 // FIELDS
986
987 @Override
988 public Scope outgoingScopeFor( Set<Scope> incomingScopes )
989 {
990 Map<String, Fields> groupingSelectors = resolveGroupingSelectors( incomingScopes );
991 Map<String, Fields> sortingSelectors = resolveSortingSelectors( incomingScopes );
992 Fields declared = resolveDeclared( incomingScopes );
993
994 Fields outGroupingFields = resultGroupFields;
995
996 if( outGroupingFields == null && isCoGroup() )
997 outGroupingFields = createJoinFields( incomingScopes, groupingSelectors, declared );
998
999 // for Group, the outgoing fields are the same as those declared
1000 return new Scope( getName(), declared, outGroupingFields, groupingSelectors, sortingSelectors, declared, isGroupBy() );
1001 }
1002
1003 private Fields createJoinFields( Set<Scope> incomingScopes, Map<String, Fields> groupingSelectors, Fields declared )
1004 {
1005 if( declared.isNone() )
1006 declared = Fields.UNKNOWN;
1007
1008 Map<String, Fields> incomingFields = new HashMap<String, Fields>();
1009
1010 for( Scope scope : incomingScopes )
1011 incomingFields.put( scope.getName(), scope.getIncomingSpliceFields() );
1012
1013 Fields outGroupingFields = Fields.NONE;
1014
1015 int offset = 0;
1016 for( Pipe pipe : pipes ) // need to retain order of pipes
1017 {
1018 String pipeName = pipe.getName();
1019 Fields pipeGroupingSelector = groupingSelectors.get( pipeName );
1020 Fields incomingField = incomingFields.get( pipeName );
1021
1022 if( !pipeGroupingSelector.isNone() )
1023 {
1024 Fields offsetFields = incomingField.selectPos( pipeGroupingSelector, offset );
1025 Fields resolvedSelect = declared.select( offsetFields );
1026
1027 outGroupingFields = outGroupingFields.append( resolvedSelect );
1028 }
1029
1030 offset += incomingField.size();
1031 }
1032
1033 return outGroupingFields;
1034 }
1035
1036 Map<String, Fields> resolveGroupingSelectors( Set<Scope> incomingScopes )
1037 {
1038 try
1039 {
1040 Map<String, Fields> groupingSelectors = getKeySelectors();
1041 Map<String, Fields> groupingFields = resolveSelectorsAgainstIncoming( incomingScopes, groupingSelectors, "grouping" );
1042
1043 if( !verifySameSize( groupingFields ) )
1044 throw new OperatorException( this, "all grouping fields must be same size: " + toString() );
1045
1046 verifySameTypes( groupingSelectors, groupingFields );
1047
1048 return groupingFields;
1049 }
1050 catch( FieldsResolverException exception )
1051 {
1052 throw new OperatorException( this, OperatorException.Kind.grouping, exception.getSourceFields(), exception.getSelectorFields(), exception );
1053 }
1054 catch( RuntimeException exception )
1055 {
1056 throw new OperatorException( this, "could not resolve grouping selector in: " + this, exception );
1057 }
1058 }
1059
1060 private boolean verifySameTypes( Map<String, Fields> groupingSelectors, Map<String, Fields> groupingFields )
1061 {
1062 // create array of field positions with comparators from the grouping selectors
1063 // unsure which side has the comparators declared so make a union
1064 boolean[] hasComparator = new boolean[ groupingFields.values().iterator().next().size() ];
1065
1066 for( Map.Entry<String, Fields> entry : groupingSelectors.entrySet() )
1067 {
1068 Comparator[] comparatorsArray = entry.getValue().getComparators();
1069
1070 for( int i = 0; i < comparatorsArray.length; i++ )
1071 hasComparator[ i ] = hasComparator[ i ] || comparatorsArray[ i ] != null;
1072 }
1073
1074 // compare all the rhs fields with the lhs (lhs and rhs are arbitrary here)
1075 Iterator<Fields> iterator = groupingFields.values().iterator();
1076 Fields lhsFields = iterator.next();
1077 Type[] lhsTypes = lhsFields.getTypes();
1078
1079 // if types are null, no basis for comparison
1080 if( lhsTypes == null )
1081 return true;
1082
1083 while( iterator.hasNext() )
1084 {
1085 Fields rhsFields = iterator.next();
1086 Type[] rhsTypes = rhsFields.getTypes();
1087
1088 // if types are null, no basis for comparison
1089 if( rhsTypes == null )
1090 return true;
1091
1092 for( int i = 0; i < lhsTypes.length; i++ )
1093 {
1094 if( hasComparator[ i ] )
1095 continue;
1096
1097 Type lhs = lhsTypes[ i ];
1098 Type rhs = rhsTypes[ i ];
1099
1100 lhs = getCanonicalType( lhs );
1101 rhs = getCanonicalType( rhs );
1102
1103 if( lhs.equals( rhs ) )
1104 continue;
1105
1106 Fields lhsError = new Fields( lhsFields.get( i ), lhsFields.getType( i ) );
1107 Fields rhsError = new Fields( rhsFields.get( i ), rhsFields.getType( i ) );
1108
1109 throw new OperatorException( this, "grouping fields must declare same types:" + lhsError.printVerbose() + " not same as " + rhsError.printVerbose() );
1110 }
1111 }
1112
1113 return true;
1114 }
1115
1116 private Type getCanonicalType( Type type )
1117 {
1118 if( type instanceof CoercibleType )
1119 type = ( (CoercibleType) type ).getCanonicalType();
1120
1121 // if one side is primitive, normalize to its primitive wrapper type
1122 if( type instanceof Class )
1123 type = Coercions.asNonPrimitive( (Class) type );
1124
1125 return type;
1126 }
1127
1128 private boolean verifySameSize( Map<String, Fields> groupingFields )
1129 {
1130 Iterator<Fields> iterator = groupingFields.values().iterator();
1131 int size = iterator.next().size();
1132
1133 while( iterator.hasNext() )
1134 {
1135 Fields groupingField = iterator.next();
1136
1137 if( groupingField.size() != size )
1138 return false;
1139
1140 size = groupingField.size();
1141 }
1142
1143 return true;
1144 }
1145
1146 private Map<String, Fields> resolveSelectorsAgainstIncoming( Set<Scope> incomingScopes, Map<String, Fields> selectors, String type )
1147 {
1148 Map<String, Fields> resolvedFields = new HashMap<String, Fields>();
1149
1150 for( Scope incomingScope : incomingScopes )
1151 {
1152 Fields selector = selectors.get( incomingScope.getName() );
1153
1154 if( selector == null )
1155 throw new OperatorException( this, "no " + type + " selector found for: " + incomingScope.getName() );
1156
1157 Fields incomingFields;
1158
1159 if( selector.isNone() )
1160 incomingFields = Fields.NONE;
1161 else if( selector.isAll() )
1162 incomingFields = incomingScope.getIncomingSpliceFields();
1163 else if( selector.isGroup() )
1164 incomingFields = incomingScope.getOutGroupingFields();
1165 else if( selector.isValues() )
1166 incomingFields = incomingScope.getOutValuesFields().subtract( incomingScope.getOutGroupingFields() );
1167 else
1168 incomingFields = incomingScope.getIncomingSpliceFields().select( selector );
1169
1170 resolvedFields.put( incomingScope.getName(), incomingFields );
1171 }
1172
1173 return resolvedFields;
1174 }
1175
1176 Map<String, Fields> resolveSortingSelectors( Set<Scope> incomingScopes )
1177 {
1178 try
1179 {
1180 if( getSortingSelectors().isEmpty() )
1181 return null;
1182
1183 return resolveSelectorsAgainstIncoming( incomingScopes, getSortingSelectors(), "sorting" );
1184 }
1185 catch( FieldsResolverException exception )
1186 {
1187 throw new OperatorException( this, OperatorException.Kind.sorting, exception.getSourceFields(), exception.getSelectorFields(), exception );
1188 }
1189 catch( RuntimeException exception )
1190 {
1191 throw new OperatorException( this, "could not resolve sorting selector in: " + this, exception );
1192 }
1193 }
1194
1195 @Override
1196 public Fields resolveIncomingOperationPassThroughFields( Scope incomingScope )
1197 {
1198 return incomingScope.getIncomingSpliceFields();
1199 }
1200
1201 Fields resolveDeclared( Set<Scope> incomingScopes )
1202 {
1203 try
1204 {
1205 Fields declaredFields = getJoinDeclaredFields();
1206
1207 // Fields.NONE is a flag to the CoGroup the following Buffer will use the JoinerClosure directly
1208 if( declaredFields != null && declaredFields.isNone() )
1209 {
1210 if( !isCoGroup() )
1211 throw new IllegalArgumentException( "Fields.NONE may only be declared as the join fields when using a CoGroup" );
1212
1213 return Fields.NONE;
1214 }
1215
1216 if( declaredFields != null ) // null for GroupBy
1217 {
1218 if( incomingScopes.size() != pipes.size() && isSelfJoin() )
1219 throw new OperatorException( this, "self joins without intermediate operators are not permitted, see 'numSelfJoins' constructor or identity function" );
1220
1221 int size = 0;
1222 boolean foundUnknown = false;
1223
1224 List<Fields> appendableFields = getOrderedResolvedFields( incomingScopes );
1225
1226 for( Fields fields : appendableFields )
1227 {
1228 foundUnknown = foundUnknown || fields.isUnknown();
1229 size += fields.size();
1230 }
1231
1232 // we must relax field checking in the face of unknown fields
1233 if( !foundUnknown && declaredFields.size() != size * ( numSelfJoins + 1 ) )
1234 {
1235 if( isSelfJoin() )
1236 throw new OperatorException( this, "declared grouped fields not same size as grouped values, declared: " + declaredFields.printVerbose() + " != size: " + size * ( numSelfJoins + 1 ) );
1237 else
1238 throw new OperatorException( this, "declared grouped fields not same size as grouped values, declared: " + declaredFields.printVerbose() + " resolved: " + Util.print( appendableFields, "" ) );
1239 }
1240
1241 int i = 0;
1242 for( Fields appendableField : appendableFields )
1243 {
1244 Type[] types = appendableField.getTypes();
1245
1246 if( types == null )
1247 {
1248 i += appendableField.size();
1249 continue;
1250 }
1251
1252 for( Type type : types )
1253 {
1254 if( type != null )
1255 declaredFields = declaredFields.applyType( i, type );
1256
1257 i++;
1258 }
1259 }
1260
1261 return declaredFields;
1262 }
1263
1264 // support merge or cogrouping here
1265 if( isGroupBy() || isMerge() )
1266 {
1267 Iterator<Scope> iterator = incomingScopes.iterator();
1268 Fields commonFields = iterator.next().getIncomingSpliceFields();
1269
1270 while( iterator.hasNext() )
1271 {
1272 Scope incomingScope = iterator.next();
1273 Fields fields = incomingScope.getIncomingSpliceFields();
1274
1275 if( !commonFields.equalsFields( fields ) )
1276 throw new OperatorException( this, "merged streams must declare the same field names, in the same order, expected: " + commonFields.printVerbose() + " found: " + fields.printVerbose() );
1277 }
1278
1279 return commonFields;
1280 }
1281 else
1282 {
1283 List<Fields> appendableFields = getOrderedResolvedFields( incomingScopes );
1284 Fields appendedFields = new Fields();
1285
1286 try
1287 {
1288 // will fail on name collisions
1289 for( Fields appendableField : appendableFields )
1290 appendedFields = appendedFields.append( appendableField );
1291 }
1292 catch( TupleException exception )
1293 {
1294 String fields = "";
1295
1296 for( Fields appendableField : appendableFields )
1297 fields += appendableField.print();
1298
1299 throw new OperatorException( this, "found duplicate field names in joined tuple stream: " + fields, exception );
1300 }
1301
1302 return appendedFields;
1303 }
1304 }
1305 catch( OperatorException exception )
1306 {
1307 throw exception;
1308 }
1309 catch( RuntimeException exception )
1310 {
1311 throw new OperatorException( this, "could not resolve declared fields in: " + this, exception );
1312 }
1313 }
1314
1315 public Fields getJoinDeclaredFields()
1316 {
1317 Fields declaredFields = getDeclaredFields();
1318
1319 if( !( joiner instanceof DeclaresResults ) )
1320 return declaredFields;
1321
1322 if( declaredFields == null && ( (DeclaresResults) joiner ).getFieldDeclaration() != null )
1323 declaredFields = ( (DeclaresResults) joiner ).getFieldDeclaration();
1324
1325 return declaredFields;
1326 }
1327
1328 private List<Fields> getOrderedResolvedFields( Set<Scope> incomingScopes )
1329 {
1330 Map<String, Scope> scopesMap = new HashMap<String, Scope>();
1331
1332 for( Scope incomingScope : incomingScopes )
1333 scopesMap.put( incomingScope.getName(), incomingScope );
1334
1335 List<Fields> appendableFields = new ArrayList<Fields>();
1336
1337 for( Pipe pipe : pipes )
1338 appendableFields.add( scopesMap.get( pipe.getName() ).getIncomingSpliceFields() );
1339 return appendableFields;
1340 }
1341
1342 @Override
1343 public boolean isEquivalentTo( FlowElement element )
1344 {
1345 boolean equivalentTo = super.isEquivalentTo( element );
1346
1347 if( !equivalentTo )
1348 return equivalentTo;
1349
1350 Splice splice = (Splice) element;
1351
1352 if( !keyFieldsMap.equals( splice.keyFieldsMap ) )
1353 return false;
1354
1355 if( !pipes.equals( splice.pipes ) )
1356 return false;
1357
1358 return true;
1359 }
1360
1361 // OBJECT OVERRIDES
1362
1363 @Override
1364 @SuppressWarnings({"RedundantIfStatement"})
1365 public boolean equals( Object object )
1366 {
1367 if( this == object )
1368 return true;
1369 if( object == null || getClass() != object.getClass() )
1370 return false;
1371 if( !super.equals( object ) )
1372 return false;
1373
1374 Splice splice = (Splice) object;
1375
1376 if( spliceName != null ? !spliceName.equals( splice.spliceName ) : splice.spliceName != null )
1377 return false;
1378 if( keyFieldsMap != null ? !keyFieldsMap.equals( splice.keyFieldsMap ) : splice.keyFieldsMap != null )
1379 return false;
1380 if( pipes != null ? !pipes.equals( splice.pipes ) : splice.pipes != null )
1381 return false;
1382
1383 return true;
1384 }
1385
1386 @Override
1387 public int hashCode()
1388 {
1389 int result = super.hashCode();
1390 result = 31 * result + ( pipes != null ? pipes.hashCode() : 0 );
1391 result = 31 * result + ( keyFieldsMap != null ? keyFieldsMap.hashCode() : 0 );
1392 result = 31 * result + ( spliceName != null ? spliceName.hashCode() : 0 );
1393 return result;
1394 }
1395
1396 @Override
1397 public String toString()
1398 {
1399 StringBuilder buffer = new StringBuilder( super.toString() );
1400
1401 buffer.append( "[by:" );
1402
1403 for( String name : keyFieldsMap.keySet() )
1404 {
1405 if( keyFieldsMap.size() > 1 )
1406 buffer.append( " " ).append( name ).append( ":" );
1407
1408 buffer.append( keyFieldsMap.get( name ).printVerbose() );
1409 }
1410
1411 if( isSelfJoin() )
1412 buffer.append( "[numSelfJoins:" ).append( numSelfJoins ).append( "]" );
1413
1414 buffer.append( "]" );
1415
1416 return buffer.toString();
1417 }
1418
1419 @Override
1420 protected void printInternal( StringBuffer buffer, Scope scope )
1421 {
1422 super.printInternal( buffer, scope );
1423 Map<String, Fields> map = scope.getKeySelectors();
1424
1425 if( map != null )
1426 {
1427 buffer.append( "[by:" );
1428
1429 // important to retain incoming pipe order
1430 for( Map.Entry<String, Fields> entry : keyFieldsMap.entrySet() )
1431 {
1432 String name = entry.getKey();
1433
1434 if( map.size() > 1 )
1435 buffer.append( name ).append( ":" );
1436
1437 buffer.append( map.get( name ).print() ); // get resolved keys
1438 }
1439
1440 if( isSelfJoin() )
1441 buffer.append( "[numSelfJoins:" ).append( numSelfJoins ).append( "]" );
1442
1443 buffer.append( "]" );
1444 }
1445 }
1446 }