| 706 | } |
| 707 | |
| 708 | void |
| 709 | EndMotionLayerNode(MotionLayerState *mlStates, int16 motNodeID, bool flushCommLayer) |
| 710 | { |
| 711 | MotionNodeEntry *pMNEntry; |
| 712 | ChunkSorterEntry *pCSEntry; |
| 713 | int i; |
| 714 | |
| 715 | pMNEntry = getMotionNodeEntry(mlStates, motNodeID); |
| 716 | |
| 717 | #ifdef AMS_VERBOSE_LOGGING |
| 718 | elog(DEBUG5, "Cleaning up Motion Layer details for motion node %d.", |
| 719 | motNodeID); |
| 720 | #endif |
| 721 | |
| 722 | /* |
| 723 | * Iterate through all entries in the motion layer's chunk-sort map, to |
| 724 | * see if we have gotten end-of-stream from all senders. |
| 725 | */ |
| 726 | if (pMNEntry->preserve_order && pMNEntry->ready_tuple_lists != NULL) |
| 727 | { |
| 728 | for (i = 0; i < pMNEntry->num_senders; i++) |
| 729 | { |
| 730 | pCSEntry = &pMNEntry->ready_tuple_lists[i]; |
| 731 | |
| 732 | /* |
| 733 | * QD should not expect end-of-stream comes from QEs who is not |
| 734 | * members of direct dispatch |
| 735 | */ |
| 736 | if (!pCSEntry->init) |
| 737 | continue; |
| 738 | |
| 739 | if (pMNEntry->preserve_order && |
| 740 | gp_log_interconnect >= GPVARS_VERBOSITY_DEBUG) |
| 741 | { |
| 742 | /* Print chunk-sorter entry statistics. */ |
| 743 | elog(DEBUG4, "Chunk-sorter entry [route=%d,node=%d] statistics:\n" |
| 744 | "\tAvailable Tuples High-Watermark: " UINT64_FORMAT, |
| 745 | i, pMNEntry->motion_node_id, |
| 746 | pMNEntry->stat_tuples_available_hwm); |
| 747 | } |
| 748 | |
| 749 | if (!pMNEntry->stopped && !pCSEntry->end_of_stream) |
| 750 | { |
| 751 | if (flushCommLayer) |
| 752 | { |
| 753 | elog(FATAL, "Motion layer node %d cleanup - did not receive" |
| 754 | " end-of-stream from sender %d.", motNodeID, i); |
| 755 | |
| 756 | /*** TODO - get chunks until end-of-stream comes in. ***/ |
| 757 | } |
| 758 | else |
| 759 | { |
| 760 | elog(LOG, "Motion layer node %d cleanup - did not receive" |
| 761 | " end-of-stream from sender %d.", motNodeID, i); |
| 762 | } |
| 763 | } |
| 764 | else |
| 765 | { |
no test coverage detected