text:
//---------------------------------------------------------------------------
//	Greenplum Database
//	Copyright (C) 2012 EMC Corp.
//
//	@filename:
//		CQueryMutators.cpp
//
//	@doc:
//		Implementation of methods used during translating a GPDB Query object into a
//		DXL Tree
//
//	@test:
//
//---------------------------------------------------------------------------

#include "postgres.h"

#include "nodes/plannodes.h"
#include "nodes/parsenodes.h"
#include "nodes/makefuncs.h"
#include "optimizer/walkers.h"

#include "gpopt/base/CUtils.h"
#include "gpopt/mdcache/CMDAccessor.h"
#include "gpopt/mdcache/CMDAccessorUtils.h"
#include "gpopt/translate/CQueryMutators.h"
#include "gpopt/translate/CTranslatorDXLToPlStmt.h"

#include "naucrates/md/IMDScalarOp.h"
#include "naucrates/md/IMDAggregate.h"
#include "naucrates/md/IMDTypeBool.h"

#include "gpopt/gpdbwrappers.h"

using namespace gpdxl;
using namespace gpmd;

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::FNeedsPrLNormalization
//
//	@doc:
//		Is the group by project list flat (contains only aggregates
//		and grouping columns)
//---------------------------------------------------------------------------
BOOL
CQueryMutators::FNeedsPrLNormalization
	(
	const Query *pquery
	)
{
	if (!pquery->hasAggs && NULL == pquery->groupClause)
	{
		return false;
	}

	SContextTLWalker ctxTLWalker(pquery->targetList, pquery->groupClause);

	ListCell *plc = NULL;
	ForEach (plc, pquery->targetList)
	{
		TargetEntry *pte  = (TargetEntry*) lfirst(plc);

		if (FNeedsToFallback((Node *) pte->expr, &ctxTLWalker))
		{
			// TODO: remove temporary fix (revert exception to assert) to avoid crash during algebrization
			GPOS_RAISE(gpdxl::ExmaDXL, gpdxl::ExmiQuery2DXLError, GPOS_WSZ_LIT("No attribute"));
		}

		// Normalize when there is an expression that is neither used for grouping
		// nor is an aggregate function
		if (!IsA(pte->expr, Aggref) && !IsA(pte->expr, GroupingFunc) && !CTranslatorUtils::FGroupingColumn( (Node*) pte->expr, pquery->groupClause, pquery->targetList))
		{
			return true;
		}
	}

	return false;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::FNeedsToFallback
//
//	@doc:
//		Fall back when the target list refers to a attribute which algebrizer
//		at this point cannot resolve
//---------------------------------------------------------------------------
BOOL
CQueryMutators::FNeedsToFallback
	(
	Node *pnode,
	void *pvCtx
	)
{
	if (NULL == pnode)
	{
		return false;
	}

	if (IsA(pnode, Const) || IsA(pnode, Aggref) || IsA(pnode, GroupingFunc) || IsA(pnode, SubLink))
	{
		return false;
	}

	SContextTLWalker *pctx = (SContextTLWalker *) pvCtx;

	TargetEntry *pteFound = gpdb::PteMember(pnode, pctx->m_plTE);
	if (NULL != pteFound && CTranslatorUtils::FGroupingColumn( (Node *) pteFound->expr, pctx->m_groupClause, pctx->m_plTE))
	{
		return false;
	}

	if (IsA(pnode, SubLink))
	{
		return false;
	}

	if (IsA(pnode, Var))
	{
		Var *pvar = (Var *) pnode;
		if (0 == pvar->varlevelsup)
		{
			// if we reach a Var that was not a grouping column then there is an equivalent column
			// which the algebrizer at this point cannot resolve
			// example: consider two table r(a,b) and s(c,d) and the following query
			// SELECT a from r LEFT JOIN s on (r.a = s.c) group by r.a
			// In the query object, generated by the parse, the output columns refer to the output of
			// the left outer join while the grouping column refers to the base table column.
			// While r.a and a are equivalent, the algebrizer at this point cannot detect this.
			// Therefore, we fall back.
			return true;
		}

		return false;
	}

	return gpdb::FWalkExpressionTree(pnode, (PfFallback) CQueryMutators::FNeedsToFallback, pvCtx);
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PqueryNormalizeGrpByPrL
//
//	@doc:
// 		Flatten expressions in project list to contain only aggregates and
//		grouping columns
//		ORGINAL QUERY:
//			SELECT * from r where r.a > (SELECT max(c) + min(d) FROM t where r.b = t.e)
// 		NEW QUERY:
//			SELECT * from r where r.a > (SELECT x1+x2 as x3
//										 FROM (SELECT max(c) as x2, min(d) as x2
//											   FROM t where r.b = t.e) t2)
//---------------------------------------------------------------------------
Query *
CQueryMutators::PqueryNormalizeGrpByPrL
	(
	IMemoryPool *pmp,
	CMDAccessor *pmda,
	const Query *pquery
	)
{
	Query *pqueryCopy = (Query *) gpdb::PvCopyObject(const_cast<Query*>(pquery));

	if (!FNeedsPrLNormalization(pqueryCopy))
	{
		return pqueryCopy;
	}

	Query *pqueryNew = PqueryConvertToDerivedTable(pqueryCopy, false /*fFixTargetList*/, true /*fFixHavingQual*/);
	gpdb::GPDBFree(pqueryCopy);

	GPOS_ASSERT(1 == gpdb::UlListLength(pqueryNew->rtable));
	Query *pqueryDrdTbl = (Query *) ((RangeTblEntry *) gpdb::PvListNth(pqueryNew->rtable, 0))->subquery;
	SContextGrpbyPlMutator ctxGbPrLMutator(pmp, pmda, pqueryDrdTbl, NULL);
	List *plTEcopy = (List*) gpdb::PvCopyObject(pqueryDrdTbl->targetList);
	ListCell *plc = NULL;

	// first normalize grouping columns
	ForEach (plc, plTEcopy)
	{
		TargetEntry *pte  = (TargetEntry*) lfirst(plc);
		GPOS_ASSERT(NULL != pte);

		if (CTranslatorUtils::FGroupingColumn(pte, pqueryDrdTbl->groupClause))
		{
			pte->expr = (Expr*) PnodeFixGrpCol( (Node*) pte->expr, pte, &ctxGbPrLMutator);
		}
	}

	plc = NULL;
	// normalize remaining project elements
	ForEach (plc, plTEcopy)
	{
		TargetEntry *pte  = (TargetEntry*) lfirst(plc);
		GPOS_ASSERT(NULL != pte);

		BOOL fGroupingCol = CTranslatorUtils::FGroupingColumn(pte, pqueryDrdTbl->groupClause);
		if (!fGroupingCol)
		{
			pte->expr = (Expr*) PnodeGrpbyPrLMutator( (Node*) pte->expr, &ctxGbPrLMutator);
			GPOS_ASSERT
				(
				!IsA(pte->expr, Aggref) && !IsA(pte->expr, GroupingFunc) &&
				"New target list entry should not contain any Aggrefs"
				);
		}
	}

	pqueryDrdTbl->targetList = ctxGbPrLMutator.m_plTENewGroupByQuery;
	pqueryNew->targetList = plTEcopy;

	ReassignSortClause(pqueryNew, pqueryDrdTbl);

	return pqueryNew;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PnodeIncrementLevelsupMutator
//
//	@doc:
//		Increment any the query levels up of any outer reference by one
//---------------------------------------------------------------------------
Node *
CQueryMutators::PnodeIncrementLevelsupMutator
	(
	Node *pnode,
	void *pvCtx
	)
{
	if (NULL == pnode)
	{
		return NULL;
	}

	SContextIncLevelsupMutator *pctxIncLvlMutator = (SContextIncLevelsupMutator *) pvCtx;

	if (IsA(pnode, Var))
	{
		Var *pvar = (Var *) gpdb::PvCopyObject(pnode);

		// Consider the following use case:
		//	ORGINAL QUERY:
		//		SELECT * from r where r.a > (SELECT max(c) + min(d)
		//									 FROM t where r.b = t.e)
		// NEW QUERY:
		//		SELECT * from r where r.a > (SELECT x1+x2 as x3
		//									 FROM (SELECT max(c) as x2, min(d) as x2
		//										   FROM t where r.b = t.e) t2)
		//
		// In such a scenario, we need increment the levels up for the
		// correlation variable r.b in the subquery by 1.

		if (pvar->varlevelsup > pctxIncLvlMutator->m_ulCurrLevelsUp)
		{
			pvar->varlevelsup++;
			return (Node *) pvar;
		}
		return (Node *) pvar;
	}

	if (IsA(pnode, CommonTableExpr))
	{
		CommonTableExpr *pcte = (CommonTableExpr *) gpdb::PvCopyObject(pnode);
		GPOS_ASSERT(IsA(pcte->ctequery, Query));

		Query *pqueryCte = (Query *) pcte->ctequery;

		pctxIncLvlMutator->m_ulCurrLevelsUp++;
		pcte->ctequery = PnodeIncrementLevelsupMutator((Node *) pqueryCte, pctxIncLvlMutator);
		pctxIncLvlMutator->m_ulCurrLevelsUp--;

		gpdb::GPDBFree(pqueryCte);

		return (Node *) pcte;
	}

	if (IsA(pnode, SubLink))
	{
		SubLink *psublink = (SubLink *) gpdb::PvCopyObject(pnode);
		GPOS_ASSERT(IsA(psublink->subselect, Query));

		Query *pquerySublink = (Query *) psublink->subselect;

		pctxIncLvlMutator->m_ulCurrLevelsUp++;
		psublink->subselect = PnodeIncrementLevelsupMutator( (Node *) pquerySublink, pctxIncLvlMutator);
		pctxIncLvlMutator->m_ulCurrLevelsUp--;
		gpdb::GPDBFree(pquerySublink);

		return (Node *) psublink;
	}

	if (IsA(pnode, TargetEntry) && 0 == pctxIncLvlMutator->m_ulCurrLevelsUp && !pctxIncLvlMutator->m_fFixTargetListTopLevel)
	{
		return (Node *) gpdb::PvCopyObject(pnode);
	}

	// recurse into query structure
	if (IsA(pnode, Query))
	{
		Query *pquery = gpdb::PqueryMutateQueryTree
								(
								(Query *) pnode,
								(Pfnode) CQueryMutators::PnodeIncrementLevelsupMutator,
								pctxIncLvlMutator,
								1 // flag -- do not mutate range table entries
								);

		// fix the outer reference in derived table entries
		List *plRtable = pquery->rtable;
		ListCell *plc = NULL;
		ForEach (plc, plRtable)
		{
			RangeTblEntry *prte = (RangeTblEntry *) lfirst(plc);

			if (RTE_SUBQUERY == prte->rtekind)
			{
				Query *pquerySubquery = prte->subquery;
				// since we did not walk inside derived tables
				pctxIncLvlMutator->m_ulCurrLevelsUp++;
				prte->subquery = (Query *) PnodeIncrementLevelsupMutator( (Node *) pquerySubquery, pctxIncLvlMutator);
				pctxIncLvlMutator->m_ulCurrLevelsUp--;
				gpdb::GPDBFree(pquerySubquery);
			}
		}

		return (Node *) pquery;
	}

	return gpdb::PnodeMutateExpressionTree(pnode, (Pfnode) CQueryMutators::PnodeIncrementLevelsupMutator, pvCtx);
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PnodeFixCTELevelsupMutator
//
//	@doc:
//		Increment any the query levels up of any CTE range table reference by one
//---------------------------------------------------------------------------
Node *
CQueryMutators::PnodeFixCTELevelsupMutator
	(
	Node *pnode,
	void *pvCtx
	)
{
	if (NULL == pnode)
	{
		return NULL;
	}

	SContextIncLevelsupMutator *pctxinclvlmutator = (SContextIncLevelsupMutator *) pvCtx;

	// recurse into query structure
	if (IsA(pnode, Query))
	{
		Query *pquery = gpdb::PqueryMutateQueryTree
								(
								(Query *) pnode,
								(Pfnode) CQueryMutators::PnodeFixCTELevelsupMutator,
								pvCtx,
								1 // flag -- do not mutate range table entries
								);

		List *plRtable = pquery->rtable;
		ListCell *plc = NULL;
		ForEach (plc, plRtable)
		{
			RangeTblEntry *prte = (RangeTblEntry *) lfirst(plc);
			if (RTE_CTE == prte->rtekind  && FNeedsLevelsUpCorrection(pctxinclvlmutator, prte->ctelevelsup))
			{
				// fix the levels up for CTE range table entry when needed
				// the walker in GPDB does not walk range table entries of type CTE
				prte->ctelevelsup++;
			}

			if (RTE_SUBQUERY == prte->rtekind)
			{
				Query *pquerySubquery = prte->subquery;
				// since we did not walk inside derived tables
				pctxinclvlmutator->m_ulCurrLevelsUp++;
				prte->subquery = (Query *) PnodeFixCTELevelsupMutator( (Node *) pquerySubquery, pctxinclvlmutator);
				pctxinclvlmutator->m_ulCurrLevelsUp--;
				gpdb::GPDBFree(pquerySubquery);
			}
		}

		return (Node *) pquery;
	}

	if (IsA(pnode, CommonTableExpr))
	{
		CommonTableExpr *pcte = (CommonTableExpr *) gpdb::PvCopyObject(pnode);
		GPOS_ASSERT(IsA(pcte->ctequery, Query));

		Query *pqueryCte = (Query *) pcte->ctequery;
		pcte->ctequery = NULL;

		pctxinclvlmutator->m_ulCurrLevelsUp++;
		pcte->ctequery = PnodeFixCTELevelsupMutator((Node *) pqueryCte, pctxinclvlmutator);
		pctxinclvlmutator->m_ulCurrLevelsUp--;

		gpdb::GPDBFree(pqueryCte);

		return (Node *) pcte;
	}

	// recurse into a query attached to sublink
	if (IsA(pnode, SubLink))
	{
		SubLink *psublink = (SubLink *) gpdb::PvCopyObject(pnode);
		GPOS_ASSERT(IsA(psublink->subselect, Query));

		Query *pquerySublink = (Query *) psublink->subselect;
		psublink->subselect = NULL;

		pctxinclvlmutator->m_ulCurrLevelsUp++;
		psublink->subselect = PnodeFixCTELevelsupMutator((Node *) pquerySublink, pctxinclvlmutator);
		pctxinclvlmutator->m_ulCurrLevelsUp--;

		gpdb::GPDBFree(pquerySublink);

		return (Node *) psublink;
	}

	return gpdb::PnodeMutateExpressionTree(pnode, (Pfnode) CQueryMutators::PnodeFixCTELevelsupMutator, pctxinclvlmutator);
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::FCorrectCTELevelsup
//
//	@doc:
//		Check if the cte levels up is the expected query level
//---------------------------------------------------------------------------
BOOL
CQueryMutators::FNeedsLevelsUpCorrection
	(
	SContextIncLevelsupMutator *pctxinclvlmutator,
	Index idxCtelevelsup
	)
{
	// when converting the query to derived table, all references to cte defined at the current level
	// or above needs to be incremented
	return idxCtelevelsup >= pctxinclvlmutator->m_ulCurrLevelsUp;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PnodeGrpbyPrLMutator
//
//	@doc:
// 		Traverse the project list of a groupby operator and extract all aggregate
//		functions in an arbitrarily complex project element
//---------------------------------------------------------------------------
Node *
CQueryMutators::PnodeGrpbyPrLMutator
	(
	Node *pnode,
	void *pctx
	)
{
	if (NULL == pnode)
	{
		return NULL;
	}

	if (IsA(pnode, Const))
	{
		return (Node *) gpdb::PvCopyObject(pnode);
	}

	SContextGrpbyPlMutator *pctxGrpByMutator = (SContextGrpbyPlMutator *) pctx;

	if (IsA(pnode, Var) && pctxGrpByMutator->m_fAggregateArg)
	{
		// if we are mutating an aggregate argument do nothing since the aggregate will be place in the derived table's target list
		// fix any outer references in the grouping column expression or arguments of an aggregate
		return (Node *) PvarOuterReferenceIncrLevelsUp((Var*) pnode);
	}

	// if we find an aggregate or precentile expression then insert into the new derived table
	// and refer to it in the top-level query

	if (IsA(pnode, Aggref))
	{
		Aggref *paggrefOld = (Aggref*) pnode;
		Aggref *paggref = PaggrefFlatCopy(paggrefOld);
		paggref->agglevelsup = paggrefOld->agglevelsup;

		List *plArgsNew = NIL;
		ListCell *plc = NULL;

		BOOL fAggregate = pctxGrpByMutator->m_fAggregateArg;
		pctxGrpByMutator->m_fAggregateArg = true;

		ForEach (plc, paggrefOld->args)
		{
			Node *pnodeArg = (Node *) gpdb::PvCopyObject((Node*) lfirst(plc));
			GPOS_ASSERT(NULL != pnodeArg);
			// traverse each argument and fix levels up when needed
			plArgsNew = gpdb::PlAppendElement
						(
						plArgsNew,
						gpdb::PnodeMutateQueryOrExpressionTree
							(
							pnodeArg,
							(Pfnode) CQueryMutators::PnodeGrpbyPrLMutator,
							(void *) pctx,
							0 // flags -- mutate into cte-lists
							)
						);
		}
		pctxGrpByMutator->m_fAggregateArg = fAggregate;
		paggref->args = plArgsNew;

		const ULONG ulAttno = gpdb::UlListLength(pctxGrpByMutator->m_plTENewGroupByQuery) + 1;
		TargetEntry *pte = PteAggregateExpr(pctxGrpByMutator->m_pmp, pctxGrpByMutator->m_pmda, (Node *) paggref, ulAttno);

		// Add a new target entry to the query
		pctxGrpByMutator->m_plTENewGroupByQuery = gpdb::PlAppendElement(pctxGrpByMutator->m_plTENewGroupByQuery, pte);

		Var *pvarNew = gpdb::PvarMakeVar
						(
						1, // varno
						(AttrNumber) ulAttno,
						gpdb::OidExprType(pnode),
						gpdb::IExprTypeMod(pnode),
						0 // query levelsup
						);

		return (Node*) pvarNew;
	}

	if (IsA(pnode, GroupingFunc))
	{
		Node *pnodeCopy = (Node *) gpdb::PvCopyObject(pnode);

		const ULONG ulAttno = gpdb::UlListLength(pctxGrpByMutator->m_plTENewGroupByQuery) + 1;
		TargetEntry *pte = PteAggregateExpr(pctxGrpByMutator->m_pmp, pctxGrpByMutator->m_pmda, pnodeCopy, ulAttno);

		// Add a new target entry to the query
		pctxGrpByMutator->m_plTENewGroupByQuery = gpdb::PlAppendElement(pctxGrpByMutator->m_plTENewGroupByQuery, pte);

		Var *pvarNew = gpdb::PvarMakeVar
								(
								1, // varno
								(AttrNumber) ulAttno,
								gpdb::OidExprType(pnode),
								gpdb::IExprTypeMod(pnode),
								0 // query levelsup
								);

		return (Node*) pvarNew;
	}

	if (!pctxGrpByMutator->m_fAggregateArg)
	{
		// if we find a target entry in the new derived table then return the appropriate var
		// else investigate it to see if it needs to be added to the new derived table

		TargetEntry *pteFound = gpdb::PteMember(pnode, pctxGrpByMutator->m_plTENewGroupByQuery);

		if (NULL != pteFound)
		{
			pteFound->resjunk = false;
			Var *pvarNew = gpdb::PvarMakeVar
					(
							1, // varno
							pteFound->resno,
							gpdb::OidExprType((Node*) pteFound->expr),
							gpdb::IExprTypeMod( (Node*) pteFound->expr),
							0 // query levelsup
					);

			return (Node*) pvarNew;
		}

		// if it is grouping column then we have already added it to the derived table
		// so merely refer to it in the top-level query
		TargetEntry *pteFoundNewDrvdTable = gpdb::PteMember(pnode, pctxGrpByMutator->m_plTENewGroupByQuery);
		if (NULL != pteFoundNewDrvdTable)
		{
			return (Node *) gpdb::PvarMakeVar
							(
							1, // varno
							(AttrNumber) pteFoundNewDrvdTable->resno,
							gpdb::OidExprType( (Node*) pteFoundNewDrvdTable->expr),
							gpdb::IExprTypeMod( (Node*) pteFoundNewDrvdTable->expr),
							0 // query levelsup
							);
		}
	}

	// do not traverse into sub queries as they will be inserted into top-level query as is
	if (IsA(pnode, SubLink))
	{
		return (Node *) gpdb::PvCopyObject(pnode);
	}

	return gpdb::PnodeMutateExpressionTree(pnode, (Pfnode) CQueryMutators::PnodeGrpbyPrLMutator, pctx);
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PnodeGrpColMutator
//
//	@doc:
// 		Mutate the grouping columns, fix levels up when necessary
//
//---------------------------------------------------------------------------
Node *
CQueryMutators::PnodeGrpColMutator
	(
	Node *pnode,
	void *pctx
	)
{
	if (NULL == pnode)
	{
		return NULL;
	}

	if (IsA(pnode, Const))
	{
		return (Node *) gpdb::PvCopyObject(pnode);
	}

	SContextGrpbyPlMutator *pctxGrpByMutator = (SContextGrpbyPlMutator *) pctx;

	if (IsA(pnode, Var))
	{
		Var *pvarCopy = (Var *) gpdb::PvCopyObject(pnode);

		if (pvarCopy->varlevelsup > pctxGrpByMutator->m_ulCurrLevelsUp)
		{
			pvarCopy->varlevelsup++;
		}

		return (Node *) pvarCopy;
	}

	if (IsA(pnode, Aggref))
	{
		// merely fix the arguments of an aggregate
		Aggref *paggrefOld = (Aggref*) pnode;
		Aggref *paggref = PaggrefFlatCopy(paggrefOld);
		paggref->agglevelsup = paggrefOld->agglevelsup;

		List *plArgsNew = NIL;
		ListCell *plc = NULL;

		BOOL fAggregate = pctxGrpByMutator->m_fAggregateArg;
		pctxGrpByMutator->m_fAggregateArg = true;

		ForEach (plc, paggrefOld->args)
		{
			Node *pnodeArg = (Node *) gpdb::PvCopyObject((Node*) lfirst(plc));
			GPOS_ASSERT(NULL != pnodeArg);
			// traverse each argument and fix levels up when needed
			plArgsNew = gpdb::PlAppendElement
						(
						plArgsNew,
						gpdb::PnodeMutateQueryOrExpressionTree
							(
							pnodeArg,
							(Pfnode) CQueryMutators::PnodeGrpColMutator,
							(void *) pctxGrpByMutator,
							0 // flags -- mutate into cte-lists
							)
						);
		}
		pctxGrpByMutator->m_fAggregateArg = fAggregate;
		paggref->args = plArgsNew;

		return (Node*) paggref;
	}

	if (IsA(pnode, GroupingFunc))
	{
		return (Node *) gpdb::PvCopyObject(pnode);
	}

	if (IsA(pnode, SubLink))
	{
		SubLink *psublinkOld = (SubLink *) pnode;

		SubLink *psublinkNew = MakeNode(SubLink);
		psublinkNew->subLinkType = psublinkOld->subLinkType;
		psublinkNew->location = psublinkOld->location;
		psublinkNew->operName = (List *) gpdb::PvCopyObject(psublinkOld->operName);

		psublinkNew->testexpr =	gpdb::PnodeMutateQueryOrExpressionTree
										(
										psublinkOld->testexpr,
										(Pfnode) CQueryMutators::PnodeGrpColMutator,
										(void *) pctxGrpByMutator,
										0 // flags -- mutate into cte-lists
										);
		pctxGrpByMutator->m_ulCurrLevelsUp++;

		GPOS_ASSERT(IsA(psublinkOld->subselect, Query));

		psublinkNew->subselect = gpdb::PnodeMutateQueryOrExpressionTree
										(
										psublinkOld->subselect,
										(Pfnode) CQueryMutators::PnodeGrpColMutator,
										(void *) pctxGrpByMutator,
										0 // flags -- mutate into cte-lists
										);

		pctxGrpByMutator->m_ulCurrLevelsUp--;

		return (Node *) psublinkNew;
	}

	if (IsA(pnode, CommonTableExpr))
	{
		CommonTableExpr *pcte = (CommonTableExpr *) gpdb::PvCopyObject(pnode);
		pctxGrpByMutator->m_ulCurrLevelsUp++;

		GPOS_ASSERT(IsA(pcte->ctequery, Query));

		pcte->ctequery = gpdb::PnodeMutateQueryOrExpressionTree
									(
									pcte->ctequery,
									(Pfnode) CQueryMutators::PnodeGrpColMutator,
									(void *) pctxGrpByMutator,
									0 // flags --- mutate into cte-lists
									);

		pctxGrpByMutator->m_ulCurrLevelsUp--;
		return (Node *) pcte;
	}

	// recurse into query structure
	if (IsA(pnode, Query))
	{
		Query *pquery = gpdb::PqueryMutateQueryTree
								(
								(Query *) pnode,
								(Pfnode) CQueryMutators::PnodeGrpColMutator,
								pctxGrpByMutator,
								1 // flag -- do not mutate range table entries
								);

		// fix the outer reference in derived table entries
		List *plRtable = pquery->rtable;
		ListCell *plc = NULL;
		ForEach (plc, plRtable)
		{
			RangeTblEntry *prte = (RangeTblEntry *) lfirst(plc);

			if (RTE_SUBQUERY == prte->rtekind)
			{
				Query *pquerySubquery = prte->subquery;
				// since we did not walk inside derived tables
				pctxGrpByMutator->m_ulCurrLevelsUp++;
				prte->subquery = (Query *) PnodeGrpColMutator( (Node *) pquerySubquery, pctxGrpByMutator);
				pctxGrpByMutator->m_ulCurrLevelsUp--;
				gpdb::GPDBFree(pquerySubquery);
			}
		}

		return (Node *) pquery;
	}

	return gpdb::PnodeMutateExpressionTree(pnode, (Pfnode) CQueryMutators::PnodeGrpColMutator, pctx);
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PnodeFixGrpCol
//
//	@doc:
// 		Mutate the grouping columns, fix levels up when necessary
//---------------------------------------------------------------------------
Node *
CQueryMutators::PnodeFixGrpCol
	(
	Node *pnode,
	TargetEntry *pteOriginal,
	SContextGrpbyPlMutator *pctx
	)
{
	GPOS_ASSERT(NULL != pnode);

	ULONG ulArity = gpdb::UlListLength(pctx->m_plTENewGroupByQuery) + 1;

	// fix any outer references in the grouping column expression
	Node *pnodeExpr = (Node *) PnodeGrpColMutator( pnode, pctx);

	CHAR* szName = CQueryMutators::SzTEName(pteOriginal,pctx->m_pquery);
	TargetEntry *pteNew = gpdb::PteMakeTargetEntry((Expr*) pnodeExpr, (AttrNumber) ulArity, szName, false /*resjunk */);

	pteNew->ressortgroupref = pteOriginal->ressortgroupref;
	pteNew->resjunk = false;

	pctx->m_plTENewGroupByQuery = gpdb::PlAppendElement(pctx->m_plTENewGroupByQuery, pteNew);

	Var *pvarNew = gpdb::PvarMakeVar
			(
					1, // varno
					(AttrNumber) ulArity,
					gpdb::OidExprType( (Node*) pteOriginal->expr),
					gpdb::IExprTypeMod( (Node*) pteOriginal->expr),
					0 // query levelsup
			);

	return (Node*) pvarNew;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PteAggregateExpr
//
//	@doc:
// 		Return a target entry for an aggregate expression
//---------------------------------------------------------------------------
TargetEntry *
CQueryMutators::PteAggregateExpr
	(
	IMemoryPool *pmp,
	CMDAccessor *pmda,
	Node *pnode,
	ULONG ulAttno
	)
{
	GPOS_ASSERT(IsA(pnode, Aggref) || IsA(pnode, GroupingFunc));

	// get the function/aggregate name
	CHAR *szName = NULL;
	if (IsA(pnode, GroupingFunc))
	{
		szName = CTranslatorUtils::SzFromWsz(GPOS_WSZ_LIT("grouping"));
	}
	else
	{
		Aggref *paggref = (Aggref*) pnode;

		CMDIdGPDB *pmdidAgg = GPOS_NEW(pmp) CMDIdGPDB(paggref->aggfnoid);
		const IMDAggregate *pmdagg = pmda->Pmdagg(pmdidAgg);
		pmdidAgg->Release();

		const CWStringConst *pstr = pmdagg->Mdname().Pstr();
		szName = CTranslatorUtils::SzFromWsz(pstr->Wsz());
	}
	GPOS_ASSERT(NULL != szName);

	return gpdb::PteMakeTargetEntry((Expr*) pnode, (AttrNumber) ulAttno, szName, false);
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PnodeHavingQualMutator
//
//	@doc:
// 		This mutator function checks to see if the current node is an AggRef
//		or a correlated variable from the derived table.
//		If it is an Aggref:
//			Then replaces it with the appropriate attribute from the top-level query.
//		If it is a correlated Var:
//			Then replaces it with a attribute from the top-level query.
//---------------------------------------------------------------------------
Node *
CQueryMutators::PnodeHavingQualMutator
	(
	Node *pnode,
	void *pctx
	)
{
	if (NULL == pnode)
	{
		return NULL;
	}

	if (IsA(pnode, Const))
	{
		return (Node *) gpdb::PvCopyObject(pnode);
	}

	SContextHavingQualMutator *pctxHavingQualMutator = (SContextHavingQualMutator *) pctx;

	// check to see if the node is in the target list of the derived table.
	// check if we have the corresponding pte entry in derived tables target list
	if (0 == pctxHavingQualMutator->m_ulCurrLevelsUp)
	{
		if (IsA(pnode, Var) && pctxHavingQualMutator->m_fAggregateArg)
		{
			// fix outer references used in the aggregates
			return (Node *) PvarOuterReferenceIncrLevelsUp((Var*) pnode);
		}

		// check if an entry already exists, if so no need for duplicate
		Node *pnodeFound = PnodeFind(pnode, pctxHavingQualMutator);
		if (NULL != pnodeFound)
		{
			return pnodeFound;
		}

		if (IsA(pnode, GroupingFunc))
		{
			// create a new entry in the derived table and return its corresponding var
			Node *pnodeCopy = (Node*) gpdb::PvCopyObject(pnode);
			return (Node *) PvarInsertIntoDerivedTable(pnodeCopy, pctxHavingQualMutator);
		}
	}

	if (IsA(pnode, Aggref))
	{
		Aggref *paggrefOld = (Aggref *) pnode;
		if (paggrefOld->agglevelsup == pctxHavingQualMutator->m_ulCurrLevelsUp)
		{
			Aggref *paggrefNew = PaggrefFlatCopy(paggrefOld);

			BOOL fAggregateOld = pctxHavingQualMutator->m_fAggregateArg;
			ULONG ulAggregateLevelUp = pctxHavingQualMutator->m_ulAggregateLevelUp;

			pctxHavingQualMutator->m_fAggregateArg = true;
			pctxHavingQualMutator->m_ulAggregateLevelUp = paggrefOld->agglevelsup;

			List *plargsNew = NIL;
			ListCell *plc = NULL;

			ForEach (plc, paggrefOld->args)
			{
				Node *pnodeArg = (Node*) lfirst(plc);
				GPOS_ASSERT(NULL != pnodeArg);
				// traverse each argument and fix levels up when needed
				plargsNew = gpdb::PlAppendElement
									(
									plargsNew,
									gpdb::PnodeMutateQueryOrExpressionTree
											(
											pnodeArg,
											(Pfnode) CQueryMutators::PnodeHavingQualMutator,
											(void *) pctxHavingQualMutator,
											0 // mutate into cte-lists
											)
									);
			}
			paggrefNew->args = plargsNew;
			pctxHavingQualMutator->m_fAggregateArg = fAggregateOld;
			pctxHavingQualMutator->m_ulAggregateLevelUp = ulAggregateLevelUp;

			// check if an entry already exists, if so no need for duplicate
			Node *pnodeFound = PnodeFind((Node*) paggrefNew, pctxHavingQualMutator);
			if (NULL != pnodeFound)
			{
				return pnodeFound;
			}

			// create a new entry in the derived table and return its corresponding var
			return (Node *) PvarInsertIntoDerivedTable((Node *) paggrefNew, pctxHavingQualMutator);
		}
	}

	if (IsA(pnode, Var))
	{
		Var *pvar = (Var *) gpdb::PvCopyObject(pnode);
		if (pvar->varlevelsup == pctxHavingQualMutator->m_ulCurrLevelsUp)
		{
			// process outer references
			if (pvar->varlevelsup == pctxHavingQualMutator->m_ulAggregateLevelUp)
			{
				// an argument of an outer aggregate
				pvar->varlevelsup = 0;

				return (Node *) pvar;
			}

			pvar->varlevelsup = 0;
			TargetEntry *pteFound = gpdb::PteMember( (Node*) pvar, pctxHavingQualMutator->m_plTENewGroupByQuery);

			if (NULL == pteFound)
			{
				// Consider two table r(a,b) and s(c,d) and the following query
				// SELECT 1 from r LEFT JOIN s on (r.a = s.c) group by r.a having count(*) > a
				// The having clause refers to the output of the left outer join while the
				// grouping column refers to the base table column.
				// While r.a and a are equivalent, the algebrizer at this point cannot detect this.
				// Therefore, pteFound will be NULL and we fall back.

				pctxHavingQualMutator->m_fFallbackToPlanner = true;
				return NULL;
			}

			pvar->varlevelsup = pctxHavingQualMutator->m_ulCurrLevelsUp;
			pvar->varno = 1;
			pvar->varattno = pteFound->resno;
			pteFound->resjunk = false;

			return (Node*) pvar;
		}
		return (Node *) pvar;
	}

	if (IsA(pnode, CommonTableExpr))
	{
		CommonTableExpr *pcte = (CommonTableExpr *) gpdb::PvCopyObject(pnode);
		pctxHavingQualMutator->m_ulCurrLevelsUp++;

		GPOS_ASSERT(IsA(pcte->ctequery, Query));

		pcte->ctequery = gpdb::PnodeMutateQueryOrExpressionTree
									(
									pcte->ctequery,
									(Pfnode) CQueryMutators::PnodeHavingQualMutator,
									(void *) pctxHavingQualMutator,
									0 // flags --- mutate into cte-lists
									);

		pctxHavingQualMutator->m_ulCurrLevelsUp--;
		return (Node *) pcte;
	}

	if (IsA(pnode, SubLink))
	{
		SubLink *psublinkOld = (SubLink *) pnode;

		SubLink *psublinkNew = MakeNode(SubLink);
		psublinkNew->subLinkType = psublinkOld->subLinkType;
		psublinkNew->location = psublinkOld->location;
		psublinkNew->operName = (List *) gpdb::PvCopyObject(psublinkOld->operName);

		psublinkNew->testexpr =	gpdb::PnodeMutateQueryOrExpressionTree
										(
										psublinkOld->testexpr,
										(Pfnode) CQueryMutators::PnodeHavingQualMutator,
										(void *) pctxHavingQualMutator,
										0 // flags -- mutate into cte-lists
										);
		pctxHavingQualMutator->m_ulCurrLevelsUp++;

		GPOS_ASSERT(IsA(psublinkOld->subselect, Query));

		psublinkNew->subselect = gpdb::PnodeMutateQueryOrExpressionTree
										(
										psublinkOld->subselect,
										(Pfnode) CQueryMutators::PnodeHavingQualMutator,
										(void *) pctxHavingQualMutator,
										0 // flags -- mutate into cte-lists
										);

		pctxHavingQualMutator->m_ulCurrLevelsUp--;

		return (Node *) psublinkNew;
	}

	return gpdb::PnodeMutateExpressionTree(pnode, (Pfnode) CQueryMutators::PnodeHavingQualMutator, pctxHavingQualMutator);
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PvarInsertIntoDerivedTable
//
//	@doc:
//		Create a new entry in the derived table and return its corresponding var
//---------------------------------------------------------------------------
Var *
CQueryMutators::PvarInsertIntoDerivedTable
	(
	Node *pnode,
	SContextHavingQualMutator *context
	)
{
	GPOS_ASSERT(NULL != pnode);
	GPOS_ASSERT(NULL != context);
	GPOS_ASSERT(IsA(pnode, Aggref) || IsA(pnode, GroupingFunc));

	const ULONG ulAttno = gpdb::UlListLength(context->m_plTENewGroupByQuery) + 1;
	TargetEntry *pte = PteAggregateExpr(context->m_pmp, context->m_pmda, (Node *) pnode, ulAttno);
	context->m_plTENewGroupByQuery = gpdb::PlAppendElement(context->m_plTENewGroupByQuery, pte);

	Var *pvarNew = gpdb::PvarMakeVar
					(
					1, // varno
					ulAttno,
					gpdb::OidExprType((Node*) pnode),
					gpdb::IExprTypeMod((Node*) pnode),
					context->m_ulCurrLevelsUp
					);

	return pvarNew;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PnodeFind
//
//	@doc:
//		Check if a matching entry already exists in the list of target
//		entries, if yes return its corresponding var, otherwise return NULL
//---------------------------------------------------------------------------
Node *
CQueryMutators::PnodeFind
	(
	Node *pnode,
	SContextHavingQualMutator *context
	)
{
	GPOS_ASSERT(NULL != pnode);
	GPOS_ASSERT(NULL != context);

	TargetEntry *pteFound = gpdb::PteMember(pnode, context->m_plTENewGroupByQuery);
	if (NULL != pteFound)
	{
		gpdb::GPDBFree(pnode);
		Var *pvarNew = gpdb::PvarMakeVar
						(
						1, // varno
						pteFound->resno,
						gpdb::OidExprType( (Node*) pteFound->expr),
						gpdb::IExprTypeMod( (Node*) pteFound->expr),
						context->m_ulCurrLevelsUp
						);

		pteFound->resjunk = false;
		return (Node*) pvarNew;
	}

	return NULL;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PaggrefFlatCopy
//
//	@doc:
//		Make a copy of the aggref (minus the arguments)
//---------------------------------------------------------------------------
Aggref *
CQueryMutators::PaggrefFlatCopy
	(
	Aggref *paggrefOld
	)
{
	Aggref *paggrefNew = MakeNode(Aggref);

	*paggrefNew = *paggrefOld;

	paggrefNew->agglevelsup = 0;
	// This is not strictly necessary: we seem to ALWAYS assgin to args from
	// the callers
	// Explicitly setting this both to be safe and to be clear that we are
	// intentionally NOT copying the args
	paggrefNew->args = NIL;

	return paggrefNew;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PvarOuterReferenceIncrLevelsUp
//
//	@doc:
//		Increment the levels up of outer references
//---------------------------------------------------------------------------
Var *
CQueryMutators::PvarOuterReferenceIncrLevelsUp
	(
	Var *pvar
	)
{
	GPOS_ASSERT(NULL != pvar);

	Var *pvarCopy = (Var *) gpdb::PvCopyObject(pvar);
	if (0 != pvarCopy->varlevelsup)
	{
		pvarCopy->varlevelsup++;
	}

	return pvarCopy;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PqueryNormalizeHaving
//
//	@doc:
//		Pull up having qual into a select and fix correlated references
//		to the top-level query
//---------------------------------------------------------------------------
Query *
CQueryMutators::PqueryNormalizeHaving
	(
	IMemoryPool *pmp,
	CMDAccessor *pmda,
	const Query *pquery
	)
{
	Query *pqueryCopy = (Query *) gpdb::PvCopyObject(const_cast<Query*>(pquery));

	if (NULL == pquery->havingQual)
	{
		return pqueryCopy;
	}

	Query *pqueryNew = PqueryConvertToDerivedTable(pqueryCopy, true /*fFixTargetList*/, false /*fFixHavingQual*/);
	gpdb::GPDBFree(pqueryCopy);

	RangeTblEntry *prte = ((RangeTblEntry *) gpdb::PvListNth(pqueryNew->rtable, 0));
	Query *pqueryDrdTbl = (Query *) prte->subquery;

	// Add all necessary target list entries of subquery
	// into the target list of the RTE as well as the new top most query
	ListCell *plc = NULL;
	ULONG ulTECount = 1;
	ForEach (plc, pqueryDrdTbl->targetList)
	{
		TargetEntry *pte  = (TargetEntry*) lfirst(plc);
		GPOS_ASSERT(NULL != pte);

		// Add to the target lists:
		// 	(1) All grouping / sorting columns even if they do not appear in the subquery output (resjunked)
		//	(2) All non-resjunked target list entries
		if (CTranslatorUtils::FGroupingColumn(pte, pqueryDrdTbl->groupClause) ||
			CTranslatorUtils::FSortingColumn(pte, pqueryDrdTbl->sortClause) || !pte->resjunk)
		{
			TargetEntry *pteNew = Pte(pte, ulTECount);
			pqueryNew->targetList = gpdb::PlAppendElement(pqueryNew->targetList, pteNew);
			// Ensure that such target entries is not suppressed in the target list of the RTE
			// and has a name
			pte->resname = SzTEName(pte, pqueryDrdTbl);
			pte->resjunk = false;
			pteNew->ressortgroupref = pte->ressortgroupref;

			ulTECount++;
		}
	}

	SContextHavingQualMutator ctxHavingQualMutator(pmp, pmda, ulTECount, pqueryDrdTbl->targetList);

	// fix outer references in the qual
	pqueryNew->jointree->quals = PnodeHavingQualMutator(pqueryDrdTbl->havingQual, &ctxHavingQualMutator);

	if (ctxHavingQualMutator.m_fFallbackToPlanner)
	{
		// TODO: Oct 14 2013, remove temporary fix (revert exception to assert) to avoid crash during algebrization
		GPOS_RAISE(gpdxl::ExmaDXL, gpdxl::ExmiQuery2DXLError, GPOS_WSZ_LIT("No attribute"));
	}

	pqueryDrdTbl->havingQual = NULL;

	ReassignSortClause(pqueryNew, prte->subquery);

	if (!prte->subquery->hasAggs && NIL == prte->subquery->groupClause)
	{
		// if the derived table has no grouping columns or aggregates then the
		// subquery is equivalent to select XXXX FROM CONST-TABLE
		// (where XXXX is the original subquery's target list)

		Query *pqueryNewSubquery = MakeNode(Query);

		pqueryNewSubquery->commandType = CMD_SELECT;
		pqueryNewSubquery->targetList = NIL;

		pqueryNewSubquery->hasAggs = false;
		pqueryNewSubquery->hasWindowFuncs = false;
		pqueryNewSubquery->hasSubLinks = false;

		ListCell *plc = NULL;
		ForEach (plc, prte->subquery->targetList)
		{
			TargetEntry *pte  = (TargetEntry*) lfirst(plc);
			GPOS_ASSERT(NULL != pte);

			GPOS_ASSERT(!pte->resjunk);

			pqueryNewSubquery->targetList =  gpdb::PlAppendElement
													(
													pqueryNewSubquery->targetList,
													(TargetEntry *) gpdb::PvCopyObject(const_cast<TargetEntry*>(pte))
													);
		}

		gpdb::GPDBFree(prte->subquery);

		prte->subquery = pqueryNewSubquery;
		prte->subquery->jointree = MakeNode(FromExpr);
		prte->subquery->groupClause = NIL;
		prte->subquery->sortClause = NIL;
		prte->subquery->windowClause = NIL;
	}

	return pqueryNew;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PqueryNormalize
//
//	@doc:
//		Normalize queries with having and group by clauses
//---------------------------------------------------------------------------
Query *
CQueryMutators::PqueryNormalize
	(
	IMemoryPool *pmp,
	CMDAccessor *pmda,
	const Query *pquery,
	ULONG ulQueryLevel
	)
{
	// flatten join alias vars defined at the current level of the query
	Query *pqueryResolveJoinVarReferences = gpdb::PqueryFlattenJoinAliasVar(const_cast<Query*>(pquery), ulQueryLevel);

	// eliminate distinct clause
	Query *pqueryEliminateDistinct = CQueryMutators::PqueryEliminateDistinctClause(pqueryResolveJoinVarReferences);
	GPOS_ASSERT(NULL == pqueryEliminateDistinct->distinctClause);
	gpdb::GPDBFree(pqueryResolveJoinVarReferences);

	// normalize window operator's project list
	Query *pqueryWindowPlNormalized = CQueryMutators::PqueryNormalizeWindowPrL(pmp, pmda, pqueryEliminateDistinct);
	gpdb::GPDBFree(pqueryEliminateDistinct);

	// pull-up having quals into a select
	Query *pqueryHavingNormalized = CQueryMutators::PqueryNormalizeHaving(pmp, pmda, pqueryWindowPlNormalized);
	GPOS_ASSERT(NULL == pqueryHavingNormalized->havingQual);
	gpdb::GPDBFree(pqueryWindowPlNormalized);

	// normalize the group by project list
	Query *pqueryNew = CQueryMutators::PqueryNormalizeGrpByPrL(pmp, pmda, pqueryHavingNormalized);
	gpdb::GPDBFree(pqueryHavingNormalized);

	return pqueryNew;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::Pte
//
//	@doc:
//		Given an Target list entry in the derived table, create a new
//		TargetEntry to be added to the top level query. This function allocates
//		memory
//---------------------------------------------------------------------------
TargetEntry *
CQueryMutators::Pte
	(
	TargetEntry *pteOld,
	ULONG ulVarAttno
	)
{
	Var *pvarNew = gpdb::PvarMakeVar
							(
							1,
							(AttrNumber) ulVarAttno,
							gpdb::OidExprType( (Node*) pteOld->expr),
							gpdb::IExprTypeMod( (Node*) pteOld->expr),
							0 // query levelsup
							);

	TargetEntry *pteNew = gpdb::PteMakeTargetEntry((Expr*) pvarNew, (AttrNumber) ulVarAttno, pteOld->resname, pteOld->resjunk);

	return pteNew;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::SzTEName
//
//	@doc:
//		Return the column name of the target list entry
//---------------------------------------------------------------------------
CHAR *
CQueryMutators::SzTEName
	(
	TargetEntry *pte,
	Query *pquery
	)
{
	if (NULL != pte->resname)
	{
		return pte->resname;
	}

	// Since a resjunked target list entry will not have a column name create a dummy column name
	CWStringConst strUnnamedCol(GPOS_WSZ_LIT("?column?"));

	return CTranslatorUtils::SzFromWsz(strUnnamedCol.Wsz());
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PqueryConvertToDerivedTable
//
//	@doc:
//		Converts query into a derived table and return the new top-level query
//---------------------------------------------------------------------------
Query *
CQueryMutators::PqueryConvertToDerivedTable
	(
	const Query *pquery,
	BOOL fFixTargetList,
	BOOL fFixHavingQual
	)
{
	Query *pqueryCopy = (Query *) gpdb::PvCopyObject(const_cast<Query*>(pquery));

	// fix any outer references
	SContextIncLevelsupMutator ctxIncLvlMutator(0, fFixTargetList);

	Node *pnodeHavingQual = NULL;
	if (!fFixHavingQual)
	{
		pnodeHavingQual = pqueryCopy->havingQual;
		pqueryCopy->havingQual = NULL;
	}

	// fix outer references
	Query *pqueryDrvd = (Query *) PnodeIncrementLevelsupMutator((Node*) pqueryCopy, &ctxIncLvlMutator);
	gpdb::GPDBFree(pqueryCopy);

	// fix the CTE levels up -- while the old query is converted into a derived table, its cte list
	// is re-assigned to the new top-level query. The references to the ctes listed in the old query
	// as well as those listed before the current query level are accordingly adjusted in the new
	// derived table.
	List *plCteOriginal = pqueryDrvd->cteList;
	pqueryDrvd->cteList = NIL;

	SContextIncLevelsupMutator ctxinclvlmutator(0 /*starting level */, fFixTargetList);

	Query *pqueryDrvdNew  = (Query *) PnodeFixCTELevelsupMutator( (Node *) pqueryDrvd, &ctxinclvlmutator);
	gpdb::GPDBFree(pqueryDrvd);
	pqueryDrvd = pqueryDrvdNew;

	// create a range table entry for the query node
	RangeTblEntry *prte = MakeNode(RangeTblEntry);
	prte->rtekind = RTE_SUBQUERY;

	prte->subquery = pqueryDrvd;
	prte->inFromCl = true;
	prte->subquery->cteList = NIL;

	if (NULL != pnodeHavingQual)
	{
		pqueryDrvd->havingQual = pnodeHavingQual;
	}

	// create a new range table reference for the new RTE
	RangeTblRef *prtref = MakeNode(RangeTblRef);
	prtref->rtindex = 1;

	// intoClause, if not null, must be set on the top query, not on the derived table
	IntoClause *origIntoClause = pqueryDrvd->intoClause;
	pqueryDrvd->intoClause = NULL;
	struct GpPolicy* origIntoPolicy = pqueryDrvd->intoPolicy;
	pqueryDrvd->intoPolicy = NULL;

	// create a new top-level query with the new RTE in its from clause
	Query *pqueryNew = MakeNode(Query);
	pqueryNew->cteList = plCteOriginal;
	pqueryNew->hasAggs = false;
	pqueryNew->rtable = gpdb::PlAppendElement(pqueryNew->rtable, prte);
	pqueryNew->intoClause = origIntoClause;
	pqueryNew->intoPolicy = origIntoPolicy;

	FromExpr *pfromexpr = MakeNode(FromExpr);
	pfromexpr->quals = NULL;
	pfromexpr->fromlist = gpdb::PlAppendElement(pfromexpr->fromlist, prtref);

	pqueryNew->jointree = pfromexpr;
	pqueryNew->commandType = CMD_SELECT;

	GPOS_ASSERT(1 == gpdb::UlListLength(pqueryNew->rtable));
	return pqueryNew;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PqueryEliminateDistinctClause
//
//	@doc:
//		Eliminate distinct columns by translating it into a grouping columns
//---------------------------------------------------------------------------
Query *
CQueryMutators::PqueryEliminateDistinctClause
	(
	const Query *pquery
	)
{
	if (0 == gpdb::UlListLength(pquery->distinctClause))
	{
		return (Query*) gpdb::PvCopyObject(const_cast<Query*>(pquery));
	}

	// create a derived table out of the previous query
	Query *pqueryNew = PqueryConvertToDerivedTable(pquery, true /*fFixTargetList*/, true /*fFixHavingQual*/);

	GPOS_ASSERT(1 == gpdb::UlListLength(pqueryNew->rtable));
	Query *pqueryDrdTbl = (Query *) ((RangeTblEntry *) gpdb::PvListNth(pqueryNew->rtable, 0))->subquery;

	ReassignSortClause(pqueryNew, pqueryDrdTbl);

	pqueryNew->targetList = NIL;
	List *plTE = pqueryDrdTbl->targetList;
	ListCell *plc = NULL;

	// build the project list of the new top-level query
	ForEach (plc, plTE)
	{
		ULONG ulResNo = gpdb::UlListLength(pqueryNew->targetList) + 1;
		TargetEntry *pte  = (TargetEntry*) lfirst(plc);
		GPOS_ASSERT(NULL != pte);

		if (!pte->resjunk)
		{
			// create a new target entry that points to the corresponding entry in the derived table
			Var *pvarNew = gpdb::PvarMakeVar
									(
									1,
									pte->resno,
									gpdb::OidExprType((Node*) pte->expr),
									gpdb::IExprTypeMod((Node*) pte->expr),
									0 // query levels up
									);
			TargetEntry *pteNew= gpdb::PteMakeTargetEntry((Expr*) pvarNew, (AttrNumber) ulResNo, pte->resname, false);

			pteNew->ressortgroupref =  pte->ressortgroupref;
			pqueryNew->targetList = gpdb::PlAppendElement(pqueryNew->targetList, pteNew);
		}

		if (0 < pte->ressortgroupref &&
			!CTranslatorUtils::FGroupingColumn(pte, pqueryDrdTbl->groupClause) &&
			!CTranslatorUtils::FWindowSpec(pte, pqueryDrdTbl->windowClause))
		{
			// initialize the ressortgroupref of target entries not used in the grouping clause
			 pte->ressortgroupref = 0;
		}
	}

	if (gpdb::UlListLength(pqueryNew->targetList) != gpdb::UlListLength(pquery->distinctClause))
	{
		GPOS_RAISE(gpdxl::ExmaDXL, gpdxl::ExmiQuery2DXLUnsupportedFeature, GPOS_WSZ_LIT("DISTINCT operation on a subset of target list columns"));
	}

	ListCell *plcDistinctCl = NULL;
	ForEach (plcDistinctCl, pquery->distinctClause)
	{
		SortGroupClause *psortcl  = (SortGroupClause*) lfirst(plcDistinctCl);
		GPOS_ASSERT(NULL != psortcl);

		SortGroupClause *pgrpcl = MakeNode(SortGroupClause);
		pgrpcl->tleSortGroupRef = psortcl->tleSortGroupRef;
		pgrpcl->eqop = psortcl->eqop;
		pgrpcl->sortop = psortcl->sortop;
		pgrpcl->nulls_first = psortcl->nulls_first;
		pqueryNew->groupClause = gpdb::PlAppendElement(pqueryNew->groupClause, pgrpcl);
	}
	pqueryNew->distinctClause = NIL;
	pqueryDrdTbl->distinctClause = NIL;

	return pqueryNew;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::FNeedsWindowPrLNormalization
//
//	@doc:
//		Check whether the window operator's project list only contains
//		window functions and columns used in the window specification
//---------------------------------------------------------------------------
BOOL
CQueryMutators::FNeedsWindowPrLNormalization
	(
	const Query *pquery
	)
{
	if (!pquery->hasWindowFuncs)
	{
		return false;
	}

	ListCell *plc = NULL;
	ForEach (plc, pquery->targetList)
	{
		TargetEntry *pte  = (TargetEntry*) lfirst(plc);

		if (!CTranslatorUtils::FWindowSpec( (Node *) pte->expr, pquery->windowClause, pquery->targetList) && !IsA(pte->expr, WindowFunc) && !IsA(pte->expr, Var))
		{
			// computed columns in the target list that is not
			// used in the order by or partition by of the window specification(s)
			return true;
		}
	}

	return false;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PqueryNormalizeWindowPrL
//
//	@doc:
// 		Flatten expressions in project list to contain only window functions and
//		columns used in the window specification
//
//		ORGINAL QUERY:
//			SELECT row_number() over() + rank() over(partition by a+b order by a-b) from foo
//
// 		NEW QUERY:
//			SELECT rn+rk from (SELECT row_number() over() as rn rank() over(partition by a+b order by a-b) as rk FROM foo) foo_new
//---------------------------------------------------------------------------
Query *
CQueryMutators::PqueryNormalizeWindowPrL
	(
	IMemoryPool *pmp,
	CMDAccessor *pmda,
	const Query *pquery
	)
{
	Query *pqueryCopy = (Query *) gpdb::PvCopyObject(const_cast<Query*>(pquery));

	if (!FNeedsWindowPrLNormalization(pquery))
	{
		return pqueryCopy;
	}

	// we do not fix target list of the derived table since we will be mutating it below
	// to ensure that it does not have operations with window function
	Query *pqueryNew = PqueryConvertToDerivedTable(pqueryCopy, false /*fFixTargetList*/, true /*fFixHavingQual*/);
	gpdb::GPDBFree(pqueryCopy);

	GPOS_ASSERT(1 == gpdb::UlListLength(pqueryNew->rtable));
	Query *pqueryDrdTbl = (Query *) ((RangeTblEntry *) gpdb::PvListNth(pqueryNew->rtable, 0))->subquery;

	SContextGrpbyPlMutator ctxWindowPrLMutator(pmp, pmda, pqueryDrdTbl, NULL);
	ListCell *plc = NULL;
	List *plTE = pqueryDrdTbl->targetList;
	ForEach (plc, plTE)
	{
		TargetEntry *pte  = (TargetEntry*) lfirst(plc);
		const ULONG ulResNoNew = gpdb::UlListLength(pqueryNew->targetList) + 1;

		if (CTranslatorUtils::FWindowSpec(pte, pquery->windowClause))
		{
			// insert the target list entry used in the window specification as is
			TargetEntry *pteNew = (TargetEntry *) gpdb::PvCopyObject(pte);
			pteNew->resno = gpdb::UlListLength(ctxWindowPrLMutator.m_plTENewGroupByQuery) + 1;
			ctxWindowPrLMutator.m_plTENewGroupByQuery = gpdb::PlAppendElement(ctxWindowPrLMutator.m_plTENewGroupByQuery, pteNew);

			if (!pte->resjunk || CTranslatorUtils::FSortingColumn(pte, pquery->sortClause))
			{
				// if the target list entry used in the window specification is present
				// in the query output then add it to the target list of the new top level query
				Var *pvarNew = gpdb::PvarMakeVar
										(
										1,
										pteNew->resno,
										gpdb::OidExprType((Node*) pte->expr),
										gpdb::IExprTypeMod((Node*) pte->expr),
										0 // query levels up
										);
				TargetEntry *pteNewCopy = gpdb::PteMakeTargetEntry((Expr*) pvarNew, ulResNoNew, pte->resname, pte->resjunk);

				// Copy the resortgroupref and resjunk information for the top-level target list entry
				// Set target list entry of the derived table to be non-resjunked
				pteNewCopy->resjunk = pteNew->resjunk;
				pteNewCopy->ressortgroupref = pteNew->ressortgroupref;
				pteNew->resjunk = false;

				pqueryNew->targetList = gpdb::PlAppendElement(pqueryNew->targetList, pteNewCopy);
			}
		}
		else
		{
			// normalize target list entry
			ctxWindowPrLMutator.m_ulRessortgroupref = pte->ressortgroupref;
			Expr *pexprNew = (Expr*) PnodeWindowPrLMutator( (Node*) pte->expr, &ctxWindowPrLMutator);
			TargetEntry *pteNew = gpdb::PteMakeTargetEntry(pexprNew, ulResNoNew, pte->resname, pte->resjunk);
			pteNew->ressortgroupref = pte->ressortgroupref;
			pqueryNew->targetList = gpdb::PlAppendElement(pqueryNew->targetList, pteNew);
		}
	}
	pqueryDrdTbl->targetList = ctxWindowPrLMutator.m_plTENewGroupByQuery;

	GPOS_ASSERT(gpdb::UlListLength(pqueryNew->targetList) <= gpdb::UlListLength(pquery->targetList));

	pqueryNew->hasWindowFuncs = false;
	ReassignSortClause(pqueryNew, pqueryDrdTbl);

	return pqueryNew;
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::PnodeWindowPrLMutator
//
//	@doc:
// 		Traverse the project list of extract all window functions in an
//		arbitrarily complex project element
//---------------------------------------------------------------------------
Node *
CQueryMutators::PnodeWindowPrLMutator
	(
	Node *pnode,
	void *pctx
	)
{
	if (NULL == pnode)
	{
		return NULL;
	}

	// do not traverse into sub queries as they will be inserted are inserted into
	// top-level query as is
	if (IsA(pnode, SubLink))
	{
		return (Node *) gpdb::PvCopyObject(pnode);
	}

	SContextGrpbyPlMutator *pctxWindowPrLMutator = (SContextGrpbyPlMutator *) pctx;
	const ULONG ulResNo = gpdb::UlListLength(pctxWindowPrLMutator->m_plTENewGroupByQuery) + 1;

	if (IsA(pnode, WindowFunc))
	{
		// insert window operator into the derived table
        // and refer to it in the top-level query's target list
		WindowFunc *pwindowfunc = (WindowFunc *) gpdb::PvCopyObject(pnode);

		// get the function name and add it to the target list
		CMDIdGPDB *pmdidFunc = GPOS_NEW(pctxWindowPrLMutator->m_pmp) CMDIdGPDB(pwindowfunc->winfnoid);
		const CWStringConst *pstr = CMDAccessorUtils::PstrWindowFuncName(pctxWindowPrLMutator->m_pmda, pmdidFunc);
		pmdidFunc->Release();

		TargetEntry *pte = gpdb::PteMakeTargetEntry
								(
								(Expr*) gpdb::PvCopyObject(pnode),
								(AttrNumber) ulResNo,
								CTranslatorUtils::SzFromWsz(pstr->Wsz()),
								false /* resjunk */
								);
		pctxWindowPrLMutator->m_plTENewGroupByQuery = gpdb::PlAppendElement(pctxWindowPrLMutator->m_plTENewGroupByQuery, pte);

		// return a variable referring to the new derived table's corresponding target list entry
		Var *pvarNew = gpdb::PvarMakeVar
								(
								1,
								(AttrNumber) ulResNo,
								gpdb::OidExprType(pnode),
								gpdb::IExprTypeMod(pnode),
								0 // query levelsup
								);

		return (Node*) pvarNew;
	}

	if (IsA(pnode, Var))
	{
		Var *pvarNew = NULL;

		TargetEntry *pteFound = gpdb::PteMember(pnode, pctxWindowPrLMutator->m_plTENewGroupByQuery);
		if (NULL == pteFound)
		{
			// insert target entry into the target list of the derived table
			CWStringConst strUnnamedCol(GPOS_WSZ_LIT("?column?"));
			TargetEntry *pte = gpdb::PteMakeTargetEntry
									(
									(Expr*) gpdb::PvCopyObject(pnode),
									(AttrNumber) ulResNo,
									CTranslatorUtils::SzFromWsz(strUnnamedCol.Wsz()),
									false /* resjunk */
									);
			pctxWindowPrLMutator->m_plTENewGroupByQuery = gpdb::PlAppendElement(pctxWindowPrLMutator->m_plTENewGroupByQuery, pte);

			pvarNew = gpdb::PvarMakeVar
								(
								1,
								(AttrNumber) ulResNo,
								gpdb::OidExprType(pnode),
								gpdb::IExprTypeMod(pnode),
								0 // query levelsup
								);
		}
		else
		{
			pteFound->resjunk = false; // ensure that the derived target list is not resjunked
			pvarNew = gpdb::PvarMakeVar
								(
								1,
								pteFound->resno,
								gpdb::OidExprType(pnode),
								gpdb::IExprTypeMod(pnode),
								0 // query levelsup
								);
		}

		return (Node*) pvarNew;
	}

	return gpdb::PnodeMutateExpressionTree(pnode, (Pfnode) CQueryMutators::PnodeWindowPrLMutator, pctx);
}

//---------------------------------------------------------------------------
//	@function:
//		CQueryMutators::ReassignSortClause
//
//	@doc:
//		Reassign the sorting clause from the derived table to the new top-level query
//---------------------------------------------------------------------------
void
CQueryMutators::ReassignSortClause
	(
	Query *pqueryNew,
	Query *pqueryDrdTbl
	)
{
	pqueryNew->sortClause = pqueryDrdTbl->sortClause;
	pqueryNew->limitOffset = pqueryDrdTbl->limitOffset;
	pqueryNew->limitCount = pqueryDrdTbl->limitCount;
	pqueryDrdTbl->sortClause = NULL;
	pqueryDrdTbl->limitOffset = NULL;
	pqueryDrdTbl->limitCount = NULL;
}

// EOF


text:
fullName
<input name="name" ng-model="fullName" type="text" class="form-control" ng-change="loadPage(1, fullName, post, date)" pattern="[A-zА-я^\s]*">
post
<select class="form-control" ng-model="post" ng-change="loadPage(1, fullName, post, date)">
	<option value=""></option>
	<option value="DIRECTOR" translate>director</option>
	<option value="ADMINISTRATOR" translate>administrator</option>
	<option value="CLEENER" translate>cleener</option>
</select>
date

		<button type="button" class="btn btn-default" ng-click="popup1 = true"></button>

	<input name="date" ng-model="date" type="text" class="form-control" uib-datepicker-popup="yyyy-MM-dd" is-open="popup1" ng-required="true" close-text="Close" alt-input-formats="'yyyy-MM-dd'" ng-click="popup1 = true" ng-value="dateValue" ng-change="loadPage(1, fullName, post, date)" readonly>

		<button type="button" class="btn btn-default" ng-click="date = null; loadPage(1, fullName, post, date)">Clear</button>

<button class="btn btn-success full-width" ng-click="openAdd()"> add</button>

redpajama_set_name: RedPajamaGithub


text:
class Attribution::Report
  attr_accessor :portfolio, :start_date, :end_date, :securities,
                :cumulative_security_performances, :cumulative_portfolio_performance, :cumulative_security_contributions,
                :companies

  def initialize( opts={} )
    @portfolio = opts[:portfolio]
    @start_date = opts[:start_date]
    @end_date = opts[:end_date]
    @companies = Hash.new { |h, k| h[k] = {} }
  end

  def calculate
    ensure_inputs_are_set!
    ensure_days_are_completely_downloaded
    ensure_security_days_are_completely_calculated
    calculate_cumulative_security_performances
    calculate_cumulative_portfolio_performance
    calculate_cumulative_security_contributions
    @results = { :security => security_stats }
  end

  def ensure_inputs_are_set!
    [:start_date, :end_date, :portfolio].each do |attrib|
      if self.send(attrib).blank?
        raise "#{attrib} was not set! Cannot calculate."
      end
    end
  end

  def ensure_security_days_are_completely_calculated
    days.sort_by(&:date).each do |d|
      if d.security_days.empty?
        # ActiveRecord::Base.transaction do
          calculator = Attribution::SecurityPerformanceCalculator.new :day => d
          calculator.calculate
        # end
      end
    end
  end

  def calculate_cumulative_security_performances
    self.securities = Hash.new { |h, k| h[k] = [] }
    days.sort_by(&:date).each do |d|
      d.security_days.each do |sd|
        securities[sd.company_id] << sd
      end
    end

    @cumulative_security_performances = {}
    securities.each do |company_id, security_days|
      company = Attribution::Company.find( company_id )
      puts "#{company.tag.ljust(10)} | #{security_days.inspect}"
      cumulative_perf = geo_link security_days.map(&:performance)

      @companies[company][:performance] = cumulative_perf
      @cumulative_security_performances[company.tag] = cumulative_perf
    end
    @cumulative_security_performances
  end

  def calculate_cumulative_security_contributions
    @cumulative_security_contributions = {}
    security_days_by_company = Hash.new { |h, k| h[k] = [] }
    contribution_days_by_company = Hash.new { |h, k| h[k] = [] }

    days.sort_by(&:date).each do |d|
      d.security_days.each do |sd|
        security_days_by_company[sd.company_id] << sd
      end
    end

    security_days_by_company.each do |company_id, security_days|
      company = Attribution::Company.find( company_id )
      contrib = security_contribution( security_days )
      @companies[company][:contribution] = contrib
      @cumulative_security_contributions[company.tag] = contrib
    end
  end

  def calculate_cumulative_portfolio_performance
    range = (@start_date..@end_date).select(&:trading_day?)
    daily_performances = range.to_a.map do |d|
      puts "PORT PERF: #{d} | #{@portfolio.day( d ).portfolio_day}"
      @portfolio.day( d ).portfolio_day.performance
    end
    @cumulative_portfolio_performance = geo_link daily_performances
  end

  def days
    @portfolio.days.where( :date => @start_date..@end_date )
  end

  def ensure_days_are_completely_downloaded
    trading_days = (@start_date.prev_trading_day..@end_date).select(&:trading_day?)
    trading_days.each do |date|
      ensure_day_is_downloaded( date )
    end

    trading_days[1..-1].each do |date|
      ensure_day_is_computed( date )
    end
  end

  def ensure_day_is_downloaded( date )
    day = @portfolio.days.where(:date => date).first_or_create
    raise "no portfolio day for #{date}" if day.nil?
    day.download unless day.downloaded?
  end

  def ensure_day_is_computed( date )
    day = @portfolio.days.where(:date => date).first
    raise "no portfolio day for #{date}" if day.nil?
    day.compute_portfolio_day unless day.completed?
  end

  def security_stats
    hsh = Hash.new { |h, k| h[k] = {} }
    securities.inject(hsh) do |stats, cusip_and_days|
      cusip = cusip_and_days.first
      sec_days = cusip_and_days.last

      stats[cusip][:performance]  = security_performance( sec_days )
      stats[cusip][:contribution] = security_contribution( sec_days )
      stats
    end
  end

  def security_return_stats
    hsh = Hash.new { |h, k| h[k] = {} }
    securities.inject(hsh) do |stats, cusip_and_days|
      cusip = cusip_and_days.first
      ticker = Attribution::Holding.where( :cusip => cusip ).first.ticker
      sec_days = cusip_and_days.last

      stats[ticker][:performance]  = security_performance( sec_days )
      stats
    end
  end

  def portfolio_days
    days.inject({}) do |hsh, day|
      hsh[day.date] = day.portfolio_day
      hsh
    end
  end

  def days
    @portfolio.days.where( :date => [@start_date..@end_date] )
  end

  def pv_multiplier( security_day )
    future_days = (@start_date..@end_date).select(&:trading_day?).to_a.select { |d| d > security_day.date }
    future_performance_days = future_days.map{ |d| portfolio_days[d].performance }
    geo_link future_performance_days
  end

  def security_performance( security_days )
    geo_link( security_days.map { |sd| sd.performance } )
  end

  def security_contribution( security_days )
    sum( security_days.map { |sd| contribution_addend( sd ) } )
  end

  def contribution_addend( security_day )
    (security_day.contribution || 0) * pv_multiplier( security_day )
  end

  def security_results
    @results[:security]
  end

  def audit(ticker=nil)
    if ticker.nil?
      audit_portfolio
    else
      audit_security( ticker )
    end
  end

  def audit_security( tag )
    tag_holdings = days.map do |d|
      d.holdings.find { |h| h.company.tag.upcase == tag.upcase }
    end.compact

    tag_holdings.each do |h|
      h.audit
    end
  end

  def audit_portfolio
    cumulative_security_performances.each do |company_tag, perf|
      puts "#{company_tag.ljust(10)} | #{percentagize( perf ).round(4)}"
    end
  end

  def print_results
    tickers = @results[:security].keys.inject( {} ) do |hsh, cusip|
      ticker = Attribution::Holding.where( :cusip => cusip ).first.ticker
      hsh[cusip] = ticker
      hsh
    end

    puts "ATTRIBUTION REPORT FOR #{portfolio.name}"
    puts "Start date: #{start_date}"
    puts "End date: #{end_date}"
    puts "TOTAL RETURN: #{sprintf('%.5f', percentagize( cumulative_portfolio_return ))}"
    puts "======================================"
    puts "Ticker |   Return | Contribution"
    @results[:security].each do |cusip, stats|
      ticker = tickers[cusip]
      perf = stats[:performance]
      contrib = stats[:contribution]
      puts "#{(ticker || "(na)").ljust(6)} | #{sprintf( '%.5f', percentagize(perf)).rjust(8)} | #{sprintf( '%.5f', contrib*100).rjust(8)}"
    end

    puts "value of cumulative_portfolio_return is: #{'%.5f' % percentagize(cumulative_portfolio_return)}"
    summed_contribs = @results[:security].values.inject( BigDecimal("0.0") ) { |s, x| s += x[:contribution] }
    puts "value of summed_contribs is: #{'%.5f' % summed_contribs}"
  end

  # def cumulative_portfolio_return
  #   range = (@start_date..@end_date)
  #   geo_link range.to_a.map { |d| @portfolio.days.where( :date => d ).first.portfolio_day.performance }
  # end
  #
  # NOTE: this assumes it takes depercentagized numbers!
  def geo_link( nums )
    nums.compact.inject( BigDecimal("1.0") ) { |product, factor| product *= factor }
  end

  # "normalize" with respect to what is right mathematically
  # eg.,  depercentagize("5") => 1.05
  def depercentagize( num )
    BigDecimal(num.to_s) / 100 + 1
  end

  # 1.05 => 5
  def percentagize( num )
    (BigDecimal(num.to_s) - 1.0)*100
  end

  def sum( nums )
    nums.inject( BigDecimal("0.0") ) { |sum, addend| sum += addend }
  end

  def lookup_security_days( tag )
    c = Attribution::Company.find_by_tag( tag )
    return nil unless c
    c.security_days.where( :day_id => days.map(&:id) )
  end
end


text:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using DesignConsistencyChecker;
using DesignConsistencyChecker.DesignRule;
using CyPhyCOMInterfaces;
using GME.MGA;
using CyPhy = ISIS.GME.Dsml.CyPhyML.Interfaces;
using CyPhyClasses = ISIS.GME.Dsml.CyPhyML.Classes;
using GME.CSharp;

namespace CyPhy2Modelica_v2.Rules
{
    public class Checker
    {
        private CyPhyGUIs.IInterpreterMainParameters mainParameters { get; set; }
        public bool Success { get; set; }
        public CyPhyGUIs.GMELogger Logger { get; set; }
        public IMgaTraceability Traceability { get; set; }
        public int NbrOfErrors { get; set; }
        public int NbrOfWarnings { get; set; }
        public int NbrOfSuccesses { get; set; }

        public Checker(CyPhyGUIs.IInterpreterMainParameters parameters, CyPhyGUIs.GMELogger logger, IMgaTraceability traceability, CyPhyTypeFactory factory)
        {
            this.mainParameters = parameters;
            this.Logger = logger;
            this.Traceability = traceability;
            this.Factory = factory;
        }

        public CyPhyTypeFactory Factory { get; set; }

        public void Check()
        {
            CyPhy2Modelica_v2.Rules.Global.externalPackageNames = null;
            CyPhy2Modelica_v2.Rules.Global.ProjectDir = this.mainParameters.ProjectDirectory;

            // check current context
            if (this.mainParameters.CurrentFCO == null ||
                this.mainParameters.CurrentFCO.Meta.Name != typeof(CyPhy.TestBench).Name)
            {
                this.Logger.WriteError("A test bench must be opened.");
                this.Success = false;
                this.NbrOfErrors = 1;
                return;
            }
            else
            {
                var testBench = CyPhyClasses.TestBench.Cast(this.mainParameters.CurrentFCO);
                foreach (var tlsut in testBench.Children.TopLevelSystemUnderTestCollection)
                {
                    if (tlsut.AllReferred is CyPhy.DesignContainer)
                    {
                        this.Logger.WriteError("CyPhy2Modelica cannot be directly invoked on a Test Bench referencing a Design Container, "
                            + " please use the Master Interpreter.");
                        this.Success = false;
                        this.NbrOfErrors = 1;
                        return;
                    }
                }
            }

            var ruleMap = this.GetRulesByKind();

            Global.Factory = this.Factory;

            var filter = this.mainParameters.Project.CreateFilter();
            var childObjects = (this.mainParameters.CurrentFCO as MgaModel).GetDescendantFCOs(filter);
            this.Success = true;
            foreach (MgaFCO child in childObjects)
            {
                if (ruleMap.ContainsKey(child.MetaBase.MetaRef))
                {
                    foreach (var rule in ruleMap[child.MetaBase.MetaRef])
                    {
                        this.ReportFeedBack(rule.Invoke(child));
                    }
                }
                else
                {
                    // we are not checking this object
                }
            }
        }

        private void ReportFeedBack(IEnumerable<RuleFeedbackBase> feedBacks)
        {
            bool noFeedBack = true;
            foreach (var feedBack in feedBacks)
            {
                noFeedBack = false;
                if (feedBack.FeedbackType == FeedbackTypes.Error)
                {
                    this.Logger.WriteFailed(feedBack.Message);
                    if (feedBack.InvolvedObjectsByRole != null)
                    {
                        this.Logger.WriteFailed("Involved Objects : ");
                        foreach (var fco in feedBack.InvolvedObjectsByRole)
                        {
                            if (fco != null)
                            {
                                // TODO: turn object references to valid objects using Traceability map
                                this.Logger.WriteFailed("[ {0} ]", fco.ToMgaHyperLink(this.Traceability));
                            }
                            else
                            {
                                this.Logger.WriteError("[ Could not get fco from InvolvedObjectsByRole for this error. (fco was null) ]");
                            }
                        }
                    }
                    this.Success = false;
                    this.NbrOfErrors += 1;
                }
                else if (feedBack.FeedbackType == FeedbackTypes.Warning)
                {
                    this.Logger.WriteWarning(feedBack.Message);
                    if (feedBack.InvolvedObjectsByRole != null)
                    {
                        this.Logger.WriteWarning("Involved Objects : ");
                        foreach (var fco in feedBack.InvolvedObjectsByRole)
                        {
                            if (fco != null)
                            {
                                // TODO: turn object references to valid objects using Traceability map
                                this.Logger.WriteWarning("[ {0} ]", fco.ToMgaHyperLink(this.Traceability));
                            }
                            else
                            {
                                this.Logger.WriteError("[ Could not get fco from InvolvedObjectsByRole for this error. (fco was null) ]");
                            }
                        }
                    }

                    this.NbrOfWarnings += 1;
                }
            }

            if (noFeedBack)
            {
                this.NbrOfSuccesses += 1;
            }
        }

        public void PrintDetails()
        {
            if (this.NbrOfErrors > 0)
            {
                this.Logger.WriteError("Errors: {0}, Warnings: {1}, Successes : {2}", this.NbrOfErrors, this.NbrOfWarnings, this.NbrOfSuccesses);
            }
            else if (this.NbrOfWarnings > 0)
            {
                this.Logger.WriteWarning("Errors: {0}, Warnings: {1}, Successes : {2}", this.NbrOfErrors, this.NbrOfWarnings, this.NbrOfSuccesses);
            }
            else
            {
                this.Logger.WriteDebug("Errors: {0}, Warnings: {1}, Successes : {2}", this.NbrOfErrors, this.NbrOfWarnings, this.NbrOfSuccesses);
            }
        }

        public Dictionary<int, List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>> GetRulesByKind()
        {
            var results = new Dictionary<int, List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>>();

            // ComponentAssembly
            results.Add(this.Factory.ComponentAssemblyMeta, new List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>());

            results[this.Factory.ComponentAssemblyMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ComponentAssemblyCannotShareNameWithItsParent));
            results[this.Factory.ComponentAssemblyMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ComponentParametersHaveValues));
            results[this.Factory.ComponentAssemblyMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ComponentPropertiesHaveValues));
            results[this.Factory.ComponentAssemblyMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ConnectorWithinComponentAssembly));
            results[this.Factory.ComponentAssemblyMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ElaboratorMustNotLeaveRefsInComponentAssembly));
            results[this.Factory.ComponentAssemblyMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.NamesMustBeValid));

            // Component
            results.Add(this.Factory.ComponentMeta, new List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>());

            results[this.Factory.ComponentMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ComponentIsDefinedInComponentFolder));
            results[this.Factory.ComponentMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ComponentParametersHaveValues));
            results[this.Factory.ComponentMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ComponentPropertiesHaveValues));
            results[this.Factory.ComponentMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ConnectorWithinComponent));
            results[this.Factory.ComponentMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ModelicaModelHasValidURI));
            results[this.Factory.ComponentMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.NamesMustBeValid));

            // TestComponent
            results.Add(this.Factory.TestComponentMeta, new List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>());

            results[this.Factory.TestComponentMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ComponentParametersHaveValues));
            results[this.Factory.TestComponentMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ComponentPropertiesHaveValues));
            results[this.Factory.TestComponentMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ConnectorWithinComponent));
            results[this.Factory.TestComponentMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ModelicaModelHasValidURI));
            results[this.Factory.TestComponentMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.NamesMustBeValid));

            // TestBench
            results.Add(this.Factory.TestBenchMeta, new List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>());

            results[this.Factory.TestBenchMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ComponentParametersHaveValues));
            results[this.Factory.TestBenchMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ComponentPropertiesHaveValues));
            results[this.Factory.TestBenchMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ComponentTreeHasValidStructure));
            results[this.Factory.TestBenchMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ElaboratorMustNotLeaveRefsInTestBench));
            results[this.Factory.TestBenchMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.NamesMustBeValid));
            results[this.Factory.TestBenchMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.OneCAInTestBench));
            results[this.Factory.TestBenchMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.SolverSettings));
            results[this.Factory.TestBenchMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.Workflow));
            results[this.Factory.TestBenchMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.TestBenchNameMustBeValid));

            // Connector
            results.Add(this.Factory.ConnectorMeta, new List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>());

            results[this.Factory.ConnectorMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ConnectorsMustHaveOneValidConnection));
            results[this.Factory.ConnectorMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ConnectorsShouldHaveChildren));
            results[this.Factory.ConnectorMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.NamesMustBeValid));
            results[this.Factory.ConnectorMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.NoMoreThanOneModelicaConnectorWithinConnectors));

            // ModelicaConnector
            results.Add(this.Factory.ModelicaConnectorMeta, new List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>());

            results[this.Factory.ModelicaConnectorMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ModelicaConnectorHasValidURI));
            results[this.Factory.ModelicaConnectorMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ModelicaConnectorsMustHaveOneValidConnection));
            results[this.Factory.ModelicaConnectorMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ModelicaParametersHaveValues));
            results[this.Factory.ModelicaConnectorMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.NamesMustBeValid));

            // ModelicaModel
            results.Add(this.Factory.ModelicaModelMeta, new List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>());

            results[this.Factory.ModelicaModelMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ModelicaParametersHaveValues));
            results[this.Factory.ModelicaModelMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.NamesMustBeValid));
            results[this.Factory.ModelicaModelMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.PortParameterUpdates));

            // Parameter
            results.Add(this.Factory.ParameterMeta, new List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>());

            results[this.Factory.ParameterMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.ParametersHaveValidIncomingFlowsAndRanges));

            // Property
            results.Add(this.Factory.PropertyMeta, new List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>());

            results[this.Factory.PropertyMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.PropertiesMustNotGetValuesFromParameters));

            // PostProcessing
            results.Add(this.Factory.PostProcessingMeta, new List<Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>>());

            results[this.Factory.PostProcessingMeta].Add(new Func<GME.MGA.MgaFCO, IEnumerable<RuleFeedbackBase>>(Global.PostProcessingPathIsValid));

            return results;
        }

    }
}


text:
#include <algorithm>
#include <limits>
#include <vector>
using namespace std;

/*
Solution: inspired by this post --> https://leetcode.com/problems/removing-minimum-number-of-magic-beans/discuss/1767106/C%2B%2B-oror-Easy-Solution-oror-Explained-oror

Time complexity: O(n) [where n is the length of beans]
Space complexity: O(n)
*/

class Solution
{
    public:
        long long minimumRemoval(vector<int> & beans)
        {
            long long result=numeric_limits<long long>::max();

            int n=beans.size();

            sort(begin(beans), end(beans));

            vector<long long> prefixSums(n, 0);

            prefixSums[0]=beans[0];

            for(int i=1;i<n;i++)
            {
                prefixSums[i]=prefixSums[i - 1] + beans[i];
            }

            for(int i=0;i<n;i++)
            {
                long long sumOfSmaller=(i==0) ? 0 : prefixSums[i - 1];

                long long sumOfLarger=prefixSums[n - 1] - prefixSums[i];

                long long numberOfLargerValues=n - i - 1;

                long long removalsFromLargerValues=numberOfLargerValues * static_cast<long long>(beans[i]);

                long long beansToRemove=sumOfLarger - removalsFromLargerValues + sumOfSmaller;

                result=min(result, beansToRemove);
            }

            return result;
        }
};


text:
#ifndef VTKVVIEWFRAME_H
#define VTKVVIEWFRAME_H

#include "vtkViewFrame.h"

class vtkViewPane;
class vtkViewObjectCollection;

//! Provides a convenient automatic vertical packing of child vtkViewObject objects.
class vtkVViewFrame : public vtkViewFrame
{
public:
  //! Provides a way to instantiate a vtkVViewFrame.
  static vtkVViewFrame *New();

  //! Standard VTK macro to allow for traversals of VTK class hierarchies.
  vtkTypeMacro(vtkVViewFrame, vtkViewFrame);

  //! Updates the viewport of child at index to conform to a vertical packing.
  /*!
   *  \param index Index of the child in the collection of vtkViewObjects.
  */
  void UpdateChildView(int index);

protected:
  vtkVViewFrame();
  ~vtkVViewFrame();

private:
  vtkVViewFrame(const vtkVViewFrame&); // Not implemented.
  void operator=(const vtkVViewFrame&); // Not implemented.
};

#endif /* end of include guard: VTKVVIEWFRAME_H */


text:
package kafka.admin

import joptsimple._
import java.util.Properties
import kafka.common.{Topic, AdminCommandFailedException}
import kafka.utils.CommandLineUtils
import kafka.utils._
import org.I0Itec.zkclient.ZkClient
import org.I0Itec.zkclient.exception.ZkNodeExistsException
import scala.collection._
import scala.collection.JavaConversions._
import kafka.log.LogConfig
import kafka.consumer.Whitelist
import kafka.server.{ConfigType, OffsetManager}
import org.apache.kafka.common.utils.Utils
import kafka.coordinator.ConsumerCoordinator

object TopicCommand extends Logging {

  def main(args: Array[String]): Unit = {

    val opts = new TopicCommandOptions(args)

    if(args.length == 0)
      CommandLineUtils.printUsageAndDie(opts.parser, "Create, delete, describe, or change a topic.")

    // should have exactly one action
    val actions = Seq(opts.createOpt, opts.listOpt, opts.alterOpt, opts.describeOpt, opts.deleteOpt).count(opts.options.has _)
    if(actions != 1)
      CommandLineUtils.printUsageAndDie(opts.parser, "Command must include exactly one action: --list, --describe, --create, --alter or --delete")

    opts.checkArgs()

    val zkClient = ZkUtils.createZkClient(opts.options.valueOf(opts.zkConnectOpt), 30000, 30000)
    var exitCode = 0
    try {
      if(opts.options.has(opts.createOpt))
        createTopic(zkClient, opts)
      else if(opts.options.has(opts.alterOpt))
        alterTopic(zkClient, opts)
      else if(opts.options.has(opts.listOpt))
        listTopics(zkClient, opts)
      else if(opts.options.has(opts.describeOpt))
        describeTopic(zkClient, opts)
      else if(opts.options.has(opts.deleteOpt))
        deleteTopic(zkClient, opts)
    } catch {
      case e: Throwable =>
        println("Error while executing topic command : " + e.getMessage)
        error(Utils.stackTrace(e))
        exitCode = 1
    } finally {
      zkClient.close()
      System.exit(exitCode)
    }

  }

  private def getTopics(zkClient: ZkClient, opts: TopicCommandOptions): Seq[String] = {
    val allTopics = ZkUtils.getAllTopics(zkClient).sorted
    if (opts.options.has(opts.topicOpt)) {
      val topicsSpec = opts.options.valueOf(opts.topicOpt)
      val topicsFilter = new Whitelist(topicsSpec)
      allTopics.filter(topicsFilter.isTopicAllowed(_, excludeInternalTopics = false))
    } else
      allTopics
  }

  def createTopic(zkClient: ZkClient, opts: TopicCommandOptions) {
    val topic = opts.options.valueOf(opts.topicOpt)
    val configs = parseTopicConfigsToBeAdded(opts)
    if (Topic.hasCollisionChars(topic))
      println("WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.")
    if (opts.options.has(opts.replicaAssignmentOpt)) {
      val assignment = parseReplicaAssignment(opts.options.valueOf(opts.replicaAssignmentOpt))
      AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkClient, topic, assignment, configs, update = false)
    } else {
      CommandLineUtils.checkRequiredArgs(opts.parser, opts.options, opts.partitionsOpt, opts.replicationFactorOpt)
      val partitions = opts.options.valueOf(opts.partitionsOpt).intValue
      val replicas = opts.options.valueOf(opts.replicationFactorOpt).intValue
      AdminUtils.createTopic(zkClient, topic, partitions, replicas, configs)
    }
    println("Created topic \"%s\".".format(topic))
  }

  def alterTopic(zkClient: ZkClient, opts: TopicCommandOptions) {
    val topics = getTopics(zkClient, opts)
    if (topics.length == 0) {
      throw new IllegalArgumentException("Topic %s does not exist on ZK path %s".format(opts.options.valueOf(opts.topicOpt),
          opts.options.valueOf(opts.zkConnectOpt)))
    }
    topics.foreach { topic =>
      if(opts.options.has(opts.partitionsOpt)) {
        if (topic == ConsumerCoordinator.OffsetsTopicName) {
          throw new IllegalArgumentException("The number of partitions for the offsets topic cannot be changed.")
        }
        println("WARNING: If partitions are increased for a topic that has a key, the partition " +
          "logic or ordering of the messages will be affected")
        val nPartitions = opts.options.valueOf(opts.partitionsOpt).intValue
        val replicaAssignmentStr = opts.options.valueOf(opts.replicaAssignmentOpt)
        AdminUtils.addPartitions(zkClient, topic, nPartitions, replicaAssignmentStr)
        println("Adding partitions succeeded!")
      }
    }
  }

  def listTopics(zkClient: ZkClient, opts: TopicCommandOptions) {
    val topics = getTopics(zkClient, opts)
    for(topic <- topics) {
      if (ZkUtils.pathExists(zkClient,ZkUtils.getDeleteTopicPath(topic))) {
        println("%s - marked for deletion".format(topic))
      } else {
        println(topic)
      }
    }
  }

  def deleteTopic(zkClient: ZkClient, opts: TopicCommandOptions) {
    val topics = getTopics(zkClient, opts)
    if (topics.length == 0) {
      throw new IllegalArgumentException("Topic %s does not exist on ZK path %s".format(opts.options.valueOf(opts.topicOpt),
          opts.options.valueOf(opts.zkConnectOpt)))
    }
    topics.foreach { topic =>
      try {
        if (Topic.InternalTopics.contains(topic)) {
          throw new AdminOperationException("Topic %s is a kafka internal topic and is not allowed to be marked for deletion.".format(topic))
        } else {
          ZkUtils.createPersistentPath(zkClient, ZkUtils.getDeleteTopicPath(topic))
          println("Topic %s is marked for deletion.".format(topic))
          println("Note: This will have no impact if delete.topic.enable is not set to true.")
        }
      } catch {
        case e: ZkNodeExistsException =>
          println("Topic %s is already marked for deletion.".format(topic))
        case e: AdminOperationException =>
          throw e
        case e: Throwable =>
          throw new AdminOperationException("Error while deleting topic %s".format(topic))
      }
    }
  }

  def describeTopic(zkClient: ZkClient, opts: TopicCommandOptions) {
    val topics = getTopics(zkClient, opts)
    val reportUnderReplicatedPartitions = if (opts.options.has(opts.reportUnderReplicatedPartitionsOpt)) true else false
    val reportUnavailablePartitions = if (opts.options.has(opts.reportUnavailablePartitionsOpt)) true else false
    val reportOverriddenConfigs = if (opts.options.has(opts.topicsWithOverridesOpt)) true else false
    val liveBrokers = ZkUtils.getAllBrokersInCluster(zkClient).map(_.id).toSet
    for (topic <- topics) {
      ZkUtils.getPartitionAssignmentForTopics(zkClient, List(topic)).get(topic) match {
        case Some(topicPartitionAssignment) =>
          val describeConfigs: Boolean = !reportUnavailablePartitions && !reportUnderReplicatedPartitions
          val describePartitions: Boolean = !reportOverriddenConfigs
          val sortedPartitions = topicPartitionAssignment.toList.sortWith((m1, m2) => m1._1 < m2._1)
          if (describeConfigs) {
            val configs = AdminUtils.fetchEntityConfig(zkClient, ConfigType.Topic, topic)
            if (!reportOverriddenConfigs || configs.size() != 0) {
              val numPartitions = topicPartitionAssignment.size
              val replicationFactor = topicPartitionAssignment.head._2.size
              println("Topic:%s\tPartitionCount:%d\tReplicationFactor:%d\tConfigs:%s"
                .format(topic, numPartitions, replicationFactor, configs.map(kv => kv._1 + "=" + kv._2).mkString(",")))
            }
          }
          if (describePartitions) {
            for ((partitionId, assignedReplicas) <- sortedPartitions) {
              val inSyncReplicas = ZkUtils.getInSyncReplicasForPartition(zkClient, topic, partitionId)
              val leader = ZkUtils.getLeaderForPartition(zkClient, topic, partitionId)
              if ((!reportUnderReplicatedPartitions && !reportUnavailablePartitions) ||
                  (reportUnderReplicatedPartitions && inSyncReplicas.size < assignedReplicas.size) ||
                  (reportUnavailablePartitions && (!leader.isDefined || !liveBrokers.contains(leader.get)))) {
                print("\tTopic: " + topic)
                print("\tPartition: " + partitionId)
                print("\tLeader: " + (if(leader.isDefined) leader.get else "none"))
                print("\tReplicas: " + assignedReplicas.mkString(","))
                println("\tIsr: " + inSyncReplicas.mkString(","))
              }
            }
          }
        case None =>
          println("Topic " + topic + " doesn't exist!")
      }
    }
  }

  def parseTopicConfigsToBeAdded(opts: TopicCommandOptions): Properties = {
    val configsToBeAdded = opts.options.valuesOf(opts.configOpt).map(_.split("""\s*=\s*"""))
    require(configsToBeAdded.forall(config => config.length == 2),
      "Invalid topic config: all configs to be added must be in the format \"key=val\".")
    val props = new Properties
    configsToBeAdded.foreach(pair => props.setProperty(pair(0).trim, pair(1).trim))
    LogConfig.validate(props)
    props
  }

  def parseReplicaAssignment(replicaAssignmentList: String): Map[Int, List[Int]] = {
    val partitionList = replicaAssignmentList.split(",")
    val ret = new mutable.HashMap[Int, List[Int]]()
    for (i <- 0 until partitionList.size) {
      val brokerList = partitionList(i).split(":").map(s => s.trim().toInt)
      val duplicateBrokers = CoreUtils.duplicates(brokerList)
      if (duplicateBrokers.nonEmpty)
        throw new AdminCommandFailedException("Partition replica lists may not contain duplicate entries: %s".format(duplicateBrokers.mkString(",")))
      ret.put(i, brokerList.toList)
      if (ret(i).size != ret(0).size)
        throw new AdminOperationException("Partition " + i + " has different replication factor: " + brokerList)
    }
    ret.toMap
  }

  class TopicCommandOptions(args: Array[String]) {
    val parser = new OptionParser
    val zkConnectOpt = parser.accepts("zookeeper", "REQUIRED: The connection string for the zookeeper connection in the form host:port. " +
                                      "Multiple URLS can be given to allow fail-over.")
                           .withRequiredArg
                           .describedAs("urls")
                           .ofType(classOf[String])
    val listOpt = parser.accepts("list", "List all available topics.")
    val createOpt = parser.accepts("create", "Create a new topic.")
    val deleteOpt = parser.accepts("delete", "Delete a topic")
    val alterOpt = parser.accepts("alter", "Alter the number of partitions and/or replica assignment for a topic")
    val describeOpt = parser.accepts("describe", "List details for the given topics.")
    val helpOpt = parser.accepts("help", "Print usage information.")
    val topicOpt = parser.accepts("topic", "The topic to be create, alter or describe. Can also accept a regular " +
                                           "expression except for --create option")
                         .withRequiredArg
                         .describedAs("topic")
                         .ofType(classOf[String])
    val nl = System.getProperty("line.separator")
    val configOpt = parser.accepts("config", "A configuration override for the topic being created."  +
                                                         "The following is a list of valid configurations: " + nl + LogConfig.configNames.map("\t" + _).mkString(nl) + nl +
                                                         "See the Kafka documentation for full details on the topic configs.")
                          .withRequiredArg
                          .describedAs("name=value")
                          .ofType(classOf[String])
    val partitionsOpt = parser.accepts("partitions", "The number of partitions for the topic being created or " +
      "altered (WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected")
                           .withRequiredArg
                           .describedAs("# of partitions")
                           .ofType(classOf[java.lang.Integer])
    val replicationFactorOpt = parser.accepts("replication-factor", "The replication factor for each partition in the topic being created.")
                           .withRequiredArg
                           .describedAs("replication factor")
                           .ofType(classOf[java.lang.Integer])
    val replicaAssignmentOpt = parser.accepts("replica-assignment", "A list of manual partition-to-broker assignments for the topic being created or altered.")
                           .withRequiredArg
                           .describedAs("broker_id_for_part1_replica1 : broker_id_for_part1_replica2 , " +
                                        "broker_id_for_part2_replica1 : broker_id_for_part2_replica2 , ...")
                           .ofType(classOf[String])
    val reportUnderReplicatedPartitionsOpt = parser.accepts("under-replicated-partitions",
                                                            "if set when describing topics, only show under replicated partitions")
    val reportUnavailablePartitionsOpt = parser.accepts("unavailable-partitions",
                                                            "if set when describing topics, only show partitions whose leader is not available")
    val topicsWithOverridesOpt = parser.accepts("topics-with-overrides",
                                                "if set when describing topics, only show topics that have overridden configs")

    val options = parser.parse(args : _*)

    val allTopicLevelOpts: Set[OptionSpec[_]] = Set(alterOpt, createOpt, describeOpt, listOpt)

    def checkArgs() {
      // check required args
      CommandLineUtils.checkRequiredArgs(parser, options, zkConnectOpt)
      if (!options.has(listOpt) && !options.has(describeOpt))
        CommandLineUtils.checkRequiredArgs(parser, options, topicOpt)

      // check invalid args
      CommandLineUtils.checkInvalidArgs(parser, options, configOpt, allTopicLevelOpts -- Set(alterOpt, createOpt))
      CommandLineUtils.checkInvalidArgs(parser, options, partitionsOpt, allTopicLevelOpts -- Set(alterOpt, createOpt))
      CommandLineUtils.checkInvalidArgs(parser, options, replicationFactorOpt, allTopicLevelOpts -- Set(createOpt))
      CommandLineUtils.checkInvalidArgs(parser, options, replicaAssignmentOpt, allTopicLevelOpts -- Set(createOpt,alterOpt))
      // Topic configs cannot be changed with alterTopic
      CommandLineUtils.checkInvalidArgs(parser, options, alterOpt, Set(configOpt))
      if(options.has(createOpt))
          CommandLineUtils.checkInvalidArgs(parser, options, replicaAssignmentOpt, Set(partitionsOpt, replicationFactorOpt))
      CommandLineUtils.checkInvalidArgs(parser, options, reportUnderReplicatedPartitionsOpt,
        allTopicLevelOpts -- Set(describeOpt) + reportUnavailablePartitionsOpt + topicsWithOverridesOpt)
      CommandLineUtils.checkInvalidArgs(parser, options, reportUnavailablePartitionsOpt,
        allTopicLevelOpts -- Set(describeOpt) + reportUnderReplicatedPartitionsOpt + topicsWithOverridesOpt)
      CommandLineUtils.checkInvalidArgs(parser, options, topicsWithOverridesOpt,
        allTopicLevelOpts -- Set(describeOpt) + reportUnderReplicatedPartitionsOpt + reportUnavailablePartitionsOpt)
    }
  }

}