//---------------------------------------------------------------------------
//	Greenplum Database
//	Copyright (C) 2012 EMC Corp.
//
//	@filename:
//		CPartPruneStepsBuilder.cpp
//
//	@doc:
//		Utility class to construct PartPruneInfos with appropriate
// 		PartPruningSteps from partitioning filter expressions
//---------------------------------------------------------------------------

extern "C" {
#include "postgres.h"

#include "nodes/nodeFuncs.h"
#include "nodes/parsenodes.h"
#include "partitioning/partdesc.h"
#include "utils/partcache.h"
#include "utils/rel.h"
}

#include "gpopt/gpdbwrappers.h"
#include "gpopt/translate/CPartPruneStepsBuilder.h"
#include "naucrates/dxl/operators/CDXLScalarBoolExpr.h"
#include "naucrates/dxl/operators/CDXLScalarCast.h"
#include "naucrates/dxl/operators/CDXLScalarComp.h"
#include "naucrates/exception.h"

using namespace gpdxl;

// ctor
CPartPruneStepsBuilder::CPartPruneStepsBuilder(
	Relation relation, Index rtindex, ULongPtrArray *part_indexes,
	List *child_rtindexes, CMappingColIdVarPlStmt *colid_var_mapping,
	CTranslatorDXLToScalar *translator_dxl_to_scalar)
	: m_relation(relation),
	  m_rtindex(rtindex),
	  m_part_indexes(part_indexes),
	  m_child_rtindexes(child_rtindexes),
	  m_colid_var_mapping(colid_var_mapping),
	  m_translator_dxl_to_scalar(translator_dxl_to_scalar)
{
}

List *
CPartPruneStepsBuilder::CreatePartPruneInfos(
	CDXLNode *filterNode, Relation relation, Index rtindex,
	ULongPtrArray *part_indexes, List *child_rtindexes,
	CMappingColIdVarPlStmt *colid_var_mapping,
	CTranslatorDXLToScalar *translator_dxl_to_scalar)
{
	CPartPruneStepsBuilder builder(relation, rtindex, part_indexes,
								   child_rtindexes, colid_var_mapping,
								   translator_dxl_to_scalar);

	// See comments over PartitionPruneInfo::prune_infos for more details.

	// ORCA only supports single-level partitioned tables for which only one
	// list of pruning steps is needed.
	// So, size of 2nd dimension of (prune_infos) = 1
	PartitionedRelPruneInfo *pinfo =
		builder.CreatePartPruneInfoForOneLevel(filterNode);
	List *prune_info_per_hierarchy = ListMake1(pinfo);

	// Since ORCA translates each DynamicTableScan to a different Append node,
	// there is always only one partition hierarchy per Append.
	// So, size of 1st dimension of (prune_infos) = 1
	return ListMake1(prune_info_per_hierarchy);
}

// Walk an expression tree collecting PARAM_EXEC param IDs into *paramids.
static bool
CollectExecParamWalker(Node *node, Bitmapset **paramids)
{
	if (node == nullptr)
		return false;
	if (IsA(node, Param))
	{
		Param *p = (Param *) node;
		if (p->paramkind == PARAM_EXEC)
			*paramids = bms_add_member(*paramids, p->paramid);
		return false;
	}
	return expression_tree_walker(node, CollectExecParamWalker, paramids);
}

PartitionedRelPruneInfo *
CPartPruneStepsBuilder::CreatePartPruneInfoForOneLevel(CDXLNode *filterNode)
{
	PartitionedRelPruneInfo *pinfo = MakeNode(PartitionedRelPruneInfo);
	pinfo->rtindex = m_rtindex;

	pinfo->nparts = gpdb::RelationGetPartitionDesc(m_relation, true)->nparts;

	pinfo->subpart_map = (int *) palloc(sizeof(int) * pinfo->nparts);
	pinfo->subplan_map = (int *) palloc(sizeof(int) * pinfo->nparts);
	pinfo->leafpart_rti_map = (int *) palloc0(sizeof(int) * pinfo->nparts);
	pinfo->relid_map = (Oid *) palloc(sizeof(Oid) * pinfo->nparts);

	// m_part_indexes contains the partition-descriptor positions that survived
	// static pruning; m_child_rtindexes[i] is the SeqScan RT index for the
	// i-th surviving partition (parallel arrays).
	ULONG part_ptr = 0;
	for (ULONG i = 0; (int) i < pinfo->nparts; ++i)
	{
		pinfo->subpart_map[i] = -1;
		if (part_ptr < m_part_indexes->Size() &&
			i == (*(*m_part_indexes)[part_ptr]))
		{
			// partition survived static pruning
			pinfo->subplan_map[i] = part_ptr;
			pinfo->relid_map[i] =
				gpdb::RelationGetPartitionDesc(m_relation, true)->oids[i];
			pinfo->present_parts = bms_add_member(pinfo->present_parts, i);

			// leafpart_rti_map: RT index of the child SeqScan for this partition
			if (m_child_rtindexes != NIL &&
				(int) part_ptr < list_length(m_child_rtindexes))
				pinfo->leafpart_rti_map[i] =
					list_nth_int(m_child_rtindexes, (int) part_ptr);

			++part_ptr;
		}
		else
		{
			// partition was pruned statically
			pinfo->subplan_map[i] = -1;
			pinfo->relid_map[i] = 0;
			pinfo->leafpart_rti_map[i] = 0;
		}
	}

	// Build pruning steps from the filter expression
	INT step_id = 0;
	List *all_steps = PartPruneStepsFromFilter(filterNode, &step_id, NIL);

	// Collect PARAM_EXEC IDs from step expressions to populate execparamids.
	// Steps with PARAM_EXEC refs must go to exec_pruning_steps (re-evaluated
	// per rescan); steps without go to initial_pruning_steps (startup only).
	Bitmapset *execparamids = nullptr;
	ListCell *lc;
	foreach(lc, all_steps)
	{
		PartitionPruneStep *step = (PartitionPruneStep *) lfirst(lc);
		if (IsA(step, PartitionPruneStepOp))
		{
			PartitionPruneStepOp *ostep = (PartitionPruneStepOp *) step;
			ListCell *lc2;
			foreach(lc2, ostep->exprs)
			{
				Node *expr = (Node *) lfirst(lc2);
				CollectExecParamWalker(expr, &execparamids);
			}
		}
	}

	if (execparamids != nullptr)
	{
		// Filter references outer join params → per-rescan exec pruning
		pinfo->exec_pruning_steps = all_steps;
		pinfo->execparamids = execparamids;
	}
	else
	{
		// Filter is constant → startup-only initial pruning
		pinfo->initial_pruning_steps = all_steps;
	}

	return pinfo;
}

List *
CPartPruneStepsBuilder::PartPruneStepFromScalarCmp(CDXLNode *node, int *step_id,
													List *steps_list)
{
	GPOS_ASSERT(nullptr != node);
	CDXLScalarComp *dxlop = CDXLScalarComp::Cast(node->GetOperator());
	Oid opno = CMDIdGPDB::CastMdid(dxlop->MDId())->Oid();
	Oid opfamily = RelationGetPartitionKey(m_relation)->partopfamily[0 /* col */];

	StrategyNumber strategy_num;
	Oid righttype = InvalidOid;

	// extract the strategy (<, >, = etc) of the operator in the scalar cmp
	// and confirm that it's usable given the partition column's opfamily
	gpdb::IndexOpProperties(opno, opfamily, &strategy_num, &righttype);

	if (InvalidOid == righttype)
	{
		GPOS_RAISE(
			gpdxl::ExmaDXL, gpdxl::ExmiDXL2PlStmtConversion,
			GPOS_WSZ_LIT("Could not find op in partition table's opfamily"));
	}

	// CPredicateUtils::ValidatePartPruningExpr() ensures that the LHS contains
	// the partition column, and RHS contains the translatable expression
	Expr *expr = m_translator_dxl_to_scalar->TranslateDXLToScalar(
		(*node)[1], m_colid_var_mapping);

	PartitionPruneStepOp *step = MakeNode(PartitionPruneStepOp);
	step->step.step_id = (*step_id)++;
	step->opstrategy = strategy_num;

	// Use cmpfns from the partitioned table, since the op was confirmed
	// to be part of partitioning column opfamily above.
	// ORCA doesn't support multi-key (a.k.a composite) partition keys. So these
	// lists will be of size 1.
	step->cmpfns = ListMake1Oid(RelationGetPartitionKey(m_relation)->partsupfunc[0].fn_oid);
	step->exprs = ListMake1(expr);

	return gpdb::LAppend(steps_list, (PartitionPruneStep *) step);
}

List *
CPartPruneStepsBuilder::PartPruneStepFromScalarBoolExpr(CDXLNode *node,
														int *step_id,
														List *steps_list)
{
	GPOS_ASSERT(nullptr != node);
	CDXLScalarBoolExpr *dxlop = CDXLScalarBoolExpr::Cast(node->GetOperator());

	PartitionPruneCombineOp combineOp;
	switch (dxlop->GetDxlBoolTypeStr())
	{
		case Edxlnot:
		{
			GPOS_RAISE(
				gpdxl::ExmaDXL, gpdxl::ExmiDXL2PlStmtConversion,
				GPOS_WSZ_LIT("NOT expressions in DPE filter expr unsupported"));
			break;
		}
		case Edxland:
		{
			GPOS_ASSERT(2 <= node->Arity());
			combineOp = PARTPRUNE_COMBINE_INTERSECT;
			break;
		}
		case Edxlor:
		{
			GPOS_ASSERT(2 <= node->Arity());
			combineOp = PARTPRUNE_COMBINE_UNION;
			break;
		}
		default:
		{
			GPOS_RTL_ASSERT(!"Boolean Operation: Must be either or/ and / not");
		}
	}

	List *stepids = NIL;
	for (ULONG ul = 0; ul < node->Arity(); ul++)
	{
		CDXLNode *child_node = (*node)[ul];
		steps_list = PartPruneStepsFromFilter(child_node, step_id, steps_list);

		PartitionPruneStep *last_step =
			(PartitionPruneStep *) lfirst(gpdb::ListTail(steps_list));
		stepids = gpdb::LAppendInt(stepids, last_step->step_id);
	}

	PartitionPruneStepCombine *step = MakeNode(PartitionPruneStepCombine);
	step->step.step_id = (*step_id)++;
	step->source_stepids = stepids;
	step->combineOp = combineOp;

	return gpdb::LAppend(steps_list, (PartitionPruneStep *) step);
}

List *
CPartPruneStepsBuilder::PartPruneStepsFromFilter(CDXLNode *node, INT *step_id,
												 List *steps_list)
{
	GPOS_ASSERT(nullptr != node);
	Edxlopid eopid = node->GetOperator()->GetDXLOperator();

	switch (eopid)
	{
		case EdxlopScalarCmp:
		{
			steps_list = PartPruneStepFromScalarCmp(node, step_id, steps_list);
			break;
		}
		case EdxlopScalarBoolExpr:
		{
			steps_list =
				PartPruneStepFromScalarBoolExpr(node, step_id, steps_list);
			break;
		}
		default:
			GPOS_RTL_ASSERT(
				!"Unsupported operator in PartPruneStepsFromFilter");
			break;
	}
	return steps_list;
}
// EOF