@Override
	public Tuple exec(Tuple aInput) throws IOException {
		if (aInput == null)
			return null;

		TupleFactory tTupleFactory = TupleFactory.getInstance();
		BagFactory tBagFactory = DefaultBagFactory.getInstance();

		// processing target | 処理対象
		DataBag tTarget = DataType.toBag(aInput.get(0));
		DataBag tGroupColumn = DataType.toBag(aInput.get(1));

		// InnerGroup container | InnerGroup の器
		GroupFilterFormat tGroupFilterFormat = mGroupFilterFormat;
		List<GroupFilterFormatGroup> tGroupList = tGroupFilterFormat.getGroupList();
		LinkedHashMap<GroupFilterFormatGroup, List<Tuple>> tInnerGroupMap = new LinkedHashMap<GroupFilterFormatGroup, List<Tuple>>();
		for (GroupFilterFormatGroup tCurrentGFFGroup : tGroupList) {
			tInnerGroupMap.put(tCurrentGFFGroup, new ArrayList<Tuple>());
		}

		// Grouping
		Iterator<Tuple> tTargetIterator = tTarget.iterator();
		Iterator<Tuple> tGroupColumnIterator = tGroupColumn.iterator();
		int tOriginalColumnNum = mOriginalColumnNum;
		while (tGroupColumnIterator.hasNext()) {
			String tCurrentGroupColumnValue = (String) tGroupColumnIterator.next().get(0);
			Tuple tCurrentTargetTupleOrigin = tTargetIterator.next();
			// The contents of Bag influence it mutually for some reason when Pig executes eval UDF which appointed Bag in parallel.
			// As coping, I record the number of the columns from a schema of original Bag at the time of outputSchema, and only the number succeeds a column.
			// The interference of the value disappears in this at least UDF implementing. It is all right to execute InnerGroup in parallel.
			//
			// NOTE なぜだか Pig は Bag を指定した eval UDF を並列に動かすと、Bag の中身が相互に影響しあう。
			// 対処として、outputSchema の時点で元の Bag のスキーマからカラム数を記録し、その数だけカラムを引き継ぐ。
			// 少なくともこの実装をする UDF 同士では値の干渉はなくなる。InnerGroup を並列に実行するのは大丈夫。
			ArrayList<Object> tOriginBagColumns = new ArrayList<Object>();
			Iterator<Object> tColumnIterator = tCurrentTargetTupleOrigin.getAll().iterator();
			for (int tColumnCount = 0; tColumnCount < tOriginalColumnNum; tColumnCount++) {
				if (tColumnIterator.hasNext())
					tOriginBagColumns.add(tColumnIterator.next());
				else
					tOriginBagColumns.add(null);
			}
			int[] tJoinedValueBagIndexes = mJoinedValueBagIndexes;
			int[] tJoinedValueBagSizes = mJoinedValueBagSizes;
			// When target JoinedValueBag does not exist, overwrite tModeOfTakeOutJoinedValue in ALL_GROUP.
			// 対象とする JoinedValueBag が存在しない場合は tModeOfTakeOutJoinedValue を ALL_GROUP で上書きする。
			ModeOfTakeOutJoinedValue tModeOfTakeOutJoinedValue =
					tJoinedValueBagIndexes == null ? ModeOfTakeOutJoinedValue.ALL_GROUP : mModeOfTakeOutJoinedValue;
			for (GroupFilterFormatGroup tCurrentGFFGroup : tGroupList) {
				if (tCurrentGFFGroup.isMember(tCurrentGroupColumnValue)) {
					tInnerGroupMap.get(tCurrentGFFGroup).add(tTupleFactory.newTupleNoCopy(
							composeTransformedTuple(tCurrentGFFGroup, tModeOfTakeOutJoinedValue, tOriginBagColumns,
									tJoinedValueBagIndexes, tJoinedValueBagSizes)));
				}
			}
		}

		// create return value | 戻り値作成
		ArrayList<Object> tResultTupleList = new ArrayList<Object>();
		for (GroupFilterFormatGroup tCurrentGFFGroup : tGroupList) {
			tResultTupleList.add(tCurrentGFFGroup.getGroupName());
			tResultTupleList.add(tBagFactory.newDefaultBag(tInnerGroupMap.get(tCurrentGFFGroup)));
		}
		return tTupleFactory.newTupleNoCopy(tResultTupleList);
	}