text: /*-------------------------------------------------------------------------
   *
   * nodeWindowNew.c
   *	  Routines to handle window nodes.
   *
   *
   * Copyright (c) 2007-2012, Greenplum inc *
   *
   *-------------------------------------------------------------------------
   */
  
  #include "postgres.h"
  #include "miscadmin.h"
  
  /* XXX include list is speculative -- bhagenbuch */
  #include "access/heapam.h"
  #include "catalog/catquery.h"
  #include "catalog/pg_aggregate.h"
  #include "catalog/pg_namespace.h"
  #include "catalog/pg_operator.h"
  #include "catalog/pg_proc.h"
  #include "catalog/pg_type.h"
  #include "catalog/pg_window.h"
  #include "cdb/cdbvars.h"
  #include "executor/executor.h"
  #include "executor/nodeAgg.h"
  #include "executor/nodeWindow.h"
  #include "optimizer/clauses.h"
  #include "nodes/makefuncs.h"
  #include "parser/parse_agg.h"
  #include "parser/parse_coerce.h"
  #include "parser/parse_expr.h"
  #include "parser/parse_oper.h"
  #include "utils/acl.h"
  #include "utils/array.h"
  #include "utils/builtins.h"
  #include "utils/datum.h"
  #include "utils/debugutils.h"
  #include "utils/int8.h"
  #include "utils/lsyscache.h"
  #include "utils/memutils.h"
  #include "utils/numeric.h"
  #include "utils/syscache.h"
  #include "utils/tuplestorenew.h"
  
  /* The initial size (in bytes) for an entry in the frame buffer. */
  #define FRAMEBUFFER_ENTRY_SIZE 1024
  
  /* Check for aggregate functions that have only transition functions,
   * but not inverse preliminary functions or preliminary functions.
   */
  #define HAS_ONLY_TRANS_FUNC(funcstate) \
  	(funcstate->isAgg && \
  	 !OidIsValid(funcstate->invprelimfn_oid) && \
  	 !OidIsValid(funcstate->prelimfn_oid) && \
  	 OidIsValid(funcstate->transfn_oid))
  
  /*
   * A struct to hold all the data relevant to a value, used for
   * buffering.
   */
  typedef struct WindowValue
  {
  	/*
  	 * Total number of not NULL argument for the function associated with this
  	 * value.
  	 *
  	 * This is only used for aggregate functions that have inverse preliminary
  	 * functions. For this kind of functions, when their argument values are
  	 * all NULLs in the current frame, their result value should be NULL as
  	 * well. This value is used to tell if all argument values in the current
  	 * frame are all NULLs.
  	 */
  	uint64		numNotNulls;
  	Datum		value;
  	bool		valueIsNull;
  }	WindowValue;
  
  /*
   * A higher-level structure to represent an entry in the frame buffer.
   */
  typedef struct FrameBufferEntry
  {
  	/*
  	 * Keys for RANGE frames.
  	 *
  	 * This is only used for RANGE frames. For ROWS frame, these values are NULL.
  	 */
  	Datum	   *keys;
  	bool	   *nulls;
  
  	/*
  	 * A list of intermediate function values to be buffered in the frame
  	 * buffer. Each of values is of WindowValue type.
  	 */
  	List	   *func_values;
  } FrameBufferEntry;
  
  /*
   * WindowStatePerLevelData - per-level working state
   */
  typedef struct WindowStatePerLevelData
  {
  	Index		level;			/* index of WindowKey for this level */
  	FmgrInfo   *eqfunctions;	/* equality fns for partial key */
  	FmgrInfo   *ltfunctions;	/* less-than functions for partial key */
  	bool		need_peercount;
  
  	int64		group_index;
  	int64		prior_non_peer_count;
  	int64		peer_index;
  	int64		peer_count;
  	int64		rank;
  	int64		dense_rank;
  	int64		prior_rank;
  	int64		prior_dense_rank;
  
  	/* List of functions in this key level */
  	List	   *level_funcs;
  
  	/* The framing clause */
  	WindowFrame *frame;
  
  	/* Indicate if all functions have trivial frames. */
  	bool		trivial_frames_only;
  
  	/* if the user didn't specify a frame, we use the default */
  	bool		default_frame;
  
  	/* Indicate if the frame is a ROW frame or a RANGE frame. */
  	bool		is_rows;
  
  	/*
  	 * Indicate if the frame clause for this level has an edge of DELAY_BOUND
  	 * type.
  	 */
  	bool		has_delay_bound;
  
  	/*
  	 * Indicate if all functions in this level are aggregate functions that
  	 * have no preliminary functions or inverse preliminary functions.
  	 */
  	bool		has_only_trans_funcs;
  
  	/* user can specify ascending or descending order, record it here */
  	bool	   *col_sort_asc;
  
  	/*
  	 * The partial order by keys. There is at least one sort key in this
  	 * array. It is used for convenience to find the order by keys since we
  	 * need to store them in the frame buffer for RANGE-based framing.
  	 */
  	int			numSortCols;
  	AttrNumber *sortColIdx;
  	Oid		   *sortOperators;
  
  	/* column type data */
  	Oid		   *col_types;
  	int2	   *col_typlens;
  	bool	   *col_typbyvals;
  
  	/*
  	 * Frame is contradictory and no tuples will fall into the frame, for
  	 * example: BETWEEN 10 FOLLOWING AND 10 PRECEDING.
  	 */
  	bool		empty_frame;
  
  	/* XXX: merge the following 4 fields into 2 */
  	/* state for rows frame is simple (for now) */
  	long int	trail_rows;
  	long int	lead_rows;
  
  	/* state for range frame is more complex */
  	Datum		trail_range;
  	Datum		lead_range;
  
  	/*
  	 * The number of rows between the trailing edge and the current_row, and
  	 * between the leading edge and the current_row.
  	 *
  	 * For preceding edges, these numbers are 0 or negative.
  	 */
  	long int	num_trail_rows;
  	long int	num_lead_rows;
  
  	/* state for frame edges (for both ROWS and RANGE frames) */
  	ExprState  *trail_expr;
  	ExprState  *lead_expr;
  
  	/* state for RANGE-frame edge computation */
  	ExprState  *trail_range_expr;
  	ExprState  *lead_range_expr;
  
  	/* RANGE frame only; see comments on init_bound_frame_edge_expr */
  	ExprState  *trail_range_eq_expr;
  	ExprState  *lead_range_eq_expr;
  
  	/*
  	 * The frame buffer to buffer the intermediate results for the functions
  	 * in this level key.
  	 */
  	struct WindowFrameBufferData *frame_buffer;
  
  	/*
  	 * These two readers are pointing to the trailing and leading edges of
  	 * this frame, respectively.
  	 */
  	NTupleStoreAccessor *trail_reader;
  	NTupleStoreAccessor *lead_reader;
  
  	/*
  	 * Indicate if lead_reader is pointing to the exact edge position. With
  	 * this flag, we don't need to do the tuple comparison again during
  	 * checking if there are sufficient data to generate an output.
  	 */
  	bool		lead_ready;
  
  	/*
  	 * Indicate if funcstate->aggTransValue actually contains the aggregate
  	 * value.
  	 */
  	bool		agg_filled;
  
  	/*
  	 * FrameBufferEntry buffers to avoid pallocs/pfrees.
  	 */
  	FrameBufferEntry *curr_entry_buf;
  	FrameBufferEntry *trail_entry_buf;
  	FrameBufferEntry *lead_entry_buf;
  }	WindowStatePerLevelData;
  
  /*
   * WindowStatePerFunctionData
   */
  typedef struct WindowStatePerFunctionData
  {
  	WindowRefExprState *wrxstate;
  	WindowRef  *wref;
  	WindowStatePerLevelData *wlevel;
  	bool		isAgg;
  	bool		allowframe;
  
  	/* Indicate if this function requires a range cumulative frame. */
  	bool		cumul_frame;
  
  	bool		winpeercount;
  
  	int			numargs;		/* explicit, caller-supplied args */
  	int16		resulttypeLen;
  	bool		resulttypeByVal;
  
  	/* Ordinary window function */
  	Datum		win_value;
  	bool		win_value_is_null;
  	Oid			windowfn_oid;
  	FmgrInfo	windowfn;
  
  	/* Aggregate-derived window function */
  	Oid			transfn_oid;
  	Oid			finalfn_oid;
  	Oid			prelimfn_oid;
  	Oid			invtransfn_oid;
  	Oid			invprelimfn_oid;
  
  	FmgrInfo	transfn;
  	FmgrInfo	finalfn;
  	FmgrInfo	prelimfn;
  	FmgrInfo	invtransfn;
  	FmgrInfo	invprelimfn;
  
  	Datum		aggInitValue;
  	bool		aggInitValueIsNull;
  
  	int16		aggTranstypeLen;
  	bool		aggTranstypeByVal;
  
  	/* the intermediate transition value */
  	Datum		aggTransValue;
  	bool		aggTransValueIsNull;
  	bool		aggNoTransValue;
  
  	/* the final transition value for output tuples */
  	Datum		final_aggTransValue;
  	bool		final_aggTransValueIsNull;
  	bool		final_aggNoTransValue;
  
  	/*
  	 * The final transition value may be a copy of a value stored in the frame
  	 * buffre, or the above intermediate in-memory transition value. We should
  	 * free the copied value. This boolean is used to determine if
  	 * final_aggTransValue should be freed.
  	 */
  	bool		final_aggShouldFree;
  
  	/* frame does not require buffering and complexity of invokeWindowFuncs() */
  	bool		trivial_frame;
  
  	/*
  	 * The index for the intermediate value of this function when serializing
  	 * its value along with others before storing them into frame buffers.
  	 *
  	 * The initial value is -1.
  	 */
  	int			serial_index;
  
  	/*
  	 * The total number of not NULL arguments for this function so far.
  	 */
  	uint64		numNotNulls;
  }	WindowStatePerFunctionData;
  
  #define FRAME_TRAIL_ROWS	0
  #define FRAME_TRAIL_RANGE	1
  #define FRAME_LEAD_ROWS		2
  #define FRAME_LEAD_RANGE	3
  
  #define EDGE_IS_BOUND(e) \
  	(e->kind == WINDOW_BOUND_PRECEDING || \
  	 e->kind == WINDOW_BOUND_FOLLOWING || \
  	 e->kind == WINDOW_DELAYED_BOUND_FOLLOWING || \
  	 e->kind == WINDOW_DELAYED_BOUND_PRECEDING)
  #define EDGE_IS_CURRENT_ROW(e) \
  	 (e->kind == WINDOW_CURRENT_ROW)
  #define EDGE_IS_DELAYED(e) \
  	 (e->kind == WINDOW_DELAYED_BOUND_FOLLOWING || \
  	  e->kind == WINDOW_DELAYED_BOUND_PRECEDING)
  #define EDGE_IS_BOUND_FOLLOWING(e) \
  	 (e->kind == WINDOW_BOUND_FOLLOWING || \
  	  e->kind == WINDOW_DELAYED_BOUND_FOLLOWING)
  #define EDGE_IS_BOUND_PRECEDING(e) \
  	 (e->kind == WINDOW_BOUND_PRECEDING || \
  	  e->kind == WINDOW_DELAYED_BOUND_PRECEDING)
  #define EDGE_IS_DELAYED_BOUND(e) \
  	 (e->kind == WINDOW_DELAYED_BOUND_PRECEDING || \
  	  e->kind == WINDOW_DELAYED_BOUND_FOLLOWING)
  #define EDGE_EQ_CURRENT_ROW(level_state, wstate, e, is_lead) \
  		 (EDGE_IS_CURRENT_ROW(e) || \
  		  ((EDGE_IS_BOUND_FOLLOWING(e) || \
  			EDGE_IS_BOUND_PRECEDING(e)) && \
  			((!is_lead && \
  			   (((level_state)->is_rows && (level_state)->trail_rows == 0) || \
  				(!(level_state)->is_rows && exec_eq_exprstate((wstate), \
  															  (level_state)->trail_range_eq_expr)))) || \
  			 (is_lead && \
  			  (((level_state)->is_rows && (level_state)->lead_rows == 0) || \
  			   (!(level_state)->is_rows && exec_eq_exprstate((wstate), \
  															 (level_state)->lead_range_eq_expr)))))))
  
  /****************************************************
   * Window Input Buffer and its APIs
   ***************************************************/
  /*
   * WindowInputBuffer: this buffer holds input tuples that fit in the
   * partition which is currently processed by the window node. This
   * buffer also stores the first tuple in the next partition, which
   * is always the last tuple in the buffer.
   */
  typedef struct WindowInputBufferData
  {
  	NTupleStore *tuplestore;
  	NTupleStoreAccessor *writer;
  
  	/* The reader that is always pointed to the previous current_row */
  	NTupleStoreAccessor *current_row_reader;
  
  	/* The reader that reads tuples in the buffer */
  	NTupleStoreAccessor *reader;
  
  	/*
  	 * Indicate if the last tuple in the buffer is the tuple that breaks the
  	 * partition key.
  	 */
  	bool		part_break;
  
  	/* The total number of tuples in the store. */
  	int64		num_tuples;
  }	WindowInputBufferData;
  
  static void resetInputBuffer(WindowState * wstate);
  static void freeInputBuffer(WindowState * wstate);
  static void trimInputBuffer(WindowInputBuffer buffer);
  
  /********************************************************
   * Window Frame Buffer and its APIs
   *******************************************************/
  /*
   * WindowFrameBuffer: this buffer holds intermediate aggregate values
   * or input arguments in order to generate the final aggregates for
   * the next output row.
   *
   * This buffer will trim old values when they are outside the lower bound
   * of the given frame provided to this buffer. This frame can be a RANGE
   * frame or a ROW frame. Note that if this frame is a RANGE frame, the
   * actual number of values in the buffer may vary over time.
   *
   * This buffer is designed to be one for each key level.  We could
   * also share one buffer among key levels if they have the same list
   * functions.
   */
  typedef struct WindowFrameBufferData
  {
  	NTupleStore *tuplestore;
  	NTupleStoreAccessor *writer;
  
  	/*
  	 * This reader is used to scan through the buffer to compute the final
  	 * result.
  	 */
  	NTupleStoreAccessor *reader;
  
  	/* Indicate if this buffer is for a RANGE frame or a ROWS frame. */
  	bool		is_rows;
  
  	/*
  	 * The number of rows before and after the current_row. Note that
  	 * 'num_rows_after' counts the current_row.
  	 *
  	 * These two counters are used to adjust frame edges for frames with
  	 * expressions.
  	 *
  	 * Currently, these two counters do not be modified after a trim operation.
  	 */
  	long int	num_rows_before;
  	long int	num_rows_after;
  
  	/*
  	 * The trailing and leading number of rows from current_row if this frame
  	 * is a ROW frame.
  	 */
  	long int	trail_rows;
  	long int	lead_rows;
  
  	/*
  	 * The trailing and leading range from current_row if this frame is a
  	 * RANGE frame.
  	 */
  	Datum		trail_range;
  	Datum		lead_range;
  
  	/* The reader that always points to the current_row. */
  	NTupleStoreAccessor *current_row_reader;
  
  	/*
  	 * The accessor that defines the point before which all values in the
  	 * buffer can be trimmed.
  	 */
  	NTupleStoreAccessor *trim_reader;
  
  	/*
  	 * Pointer to the level state. Information about window functions can be
  	 * found here.
  	 */
  	WindowStatePerLevel level_state;
  }	WindowFrameBufferData;
  typedef WindowFrameBufferData *WindowFrameBuffer;
  
  static WindowFrameBuffer createRangeFrameBuffer(Datum trail_range,
  					   Datum lead_range,
  					   int bytes);
  static WindowFrameBuffer createRowsFrameBuffer(long int trail_rows,
  					  long int lead_rows,
  					  int bytes);
  static WindowFrameBuffer resetFrameBuffer(WindowFrameBuffer buffer);
  static void appendToFrameBuffer(WindowStatePerLevel level_state,
  					WindowState * wstate,
  					bool last_peer);
  static void trimFrameBuffer(WindowFrameBuffer buffer);
  static void incrementCurrentRow(WindowFrameBuffer buffer, WindowState * wstate);
  static bool hasEnoughDataInRange(WindowFrameBuffer buffer,
  					 WindowStatePerLevel level_state,
  					 WindowState * wstate,
  					 Datum tail_range,
  					 Datum lead_range);
  static bool hasEnoughDataInRows(WindowFrameBuffer buffer,
  					WindowStatePerLevel level_state,
  					WindowState * wstate,
  					long int trail_rows,
  					long int lead_rows);
  static void computeFrameValue(WindowStatePerLevel level_state,
  				  WindowState * wstate,
  				  NTupleStoreAccessor * trail_reader,
  				  NTupleStoreAccessor * lead_reader);
  
  static void createFrameBuffers(WindowState * wstate);
  static void resetFrameBuffers(WindowState * wstate);
  static void resetTransValues(WindowStatePerLevel level_state,
  				 WindowState * wstate);
  static void freeFrameBuffer(WindowFrameBuffer buffer);
  static void freeFrameBuffers(WindowState * wstate);
  
  /*
   * WindowBufferCursor
   * This is an abstract cursor to scan the frame buffer.  This holds transient
   * state of a window level, so you have to make sure the scan on the same window
   * level should happen once at a time.
   */
  typedef struct WindowBufferCursorData
  {
  	WindowState *wstate;		/* state of this window */
  	WindowStatePerLevel level_state;	/* state of this window level */
  	NTupleStoreAccessor *reader;	/* reader to scan rows */
  	NTupleStorePos orig_pos;	/* original position of the reader */
  	bool		use_last_agg;	/* should scan "last_agg"? */
  	bool		reader_eof;		/* is reader at end of scan? */
  	bool		has_tuples;		/* does frame has tuples? */
  	bool		forward;		/* scan direction */
  	bool		first;			/* is first fetch of tupstore? */
  }	WindowBufferCursorData;
  
  typedef struct WindowBufferCursorData *WindowBufferCursor;
  
  static WindowBufferCursor windowBufferBeginScan(WindowState * wstate,
  					  WindowStatePerLevel level_state, bool forward);
  static List *windowBufferNextTupleStore(WindowBufferCursor cursor);
  static List *windowBufferNextLastAgg(WindowBufferCursor cursor);
  static List *windowBufferNext(WindowBufferCursor cursor);
  static void windowBufferEndScan(WindowBufferCursor cursor);
  
  
  /* the functions for the Window node */
  static WindowState *makeWindowState(Window * window, EState *estate);
  static void initializePartition(WindowState * wstate);
  static bool checkOutputReady(WindowState * wstate);
  static TupleTableSlot *fetchTupleSlotThroughBuf(WindowState * wstate);
  static int initFcinfo(WindowRefExprState * wrxstate, FunctionCallInfoData *fcinfo,
  		   WindowStatePerFunction funcstate, ExprContext *econtext,
  		   bool check_nulls);
  static void adjustEdges(WindowFrameBuffer buffer, WindowState * wstate);
  static TupleTableSlot *fetchCurrentRow(WindowState * wstate);
  static TupleTableSlot *fetchTupleSlotThroughBuf(WindowState * wstate);
  static void processTupleSlot(WindowState * wstate, TupleTableSlot * slot, bool last_peer);
  static bool getCurrentValue(NTupleStoreAccessor * reader,
  				WindowStatePerLevel level_state,
  				FrameBufferEntry *entry_buf);
  static bool cmp_deformed_tuple(Datum *a, bool *a_nulls, Datum *b, bool *b_nulls,
  				   bool *asc_cols, int ncols,
  				   FmgrInfo *ltfuncs, FmgrInfo *eqfuncs,
  				   MemoryContext evalContext, bool is_equal);
  static FmgrInfo *get_ltfuncs(TupleDesc tupdesc, int numCols, AttrNumber *matchColIdx);
  static void advanceKeyLevelState(WindowState * wstate, int min_level);
  static void init_frames(WindowState * wstate);
  static void deform_window_tuple(TupleTableSlot * slot, int nattrs, AttrNumber *attnums,
  					Datum *values, bool *nulls, WindowState * wstate);
  static void add_tuple_to_trans(WindowStatePerFunction funcstate, WindowState * wstate,
  				   ExprContext *econtext, bool check_nulls);
  static bool hasTuplesInFrame(WindowStatePerLevel level_state,
  				 WindowState * wstate);
  static Datum last_value_internal(WindowRefExprState * wrxstate, bool *isnull);
  static Datum first_value_internal(WindowRefExprState * wrxstate, bool *isnull);
  static FrameBufferEntry *createFrameBufferEntry(WindowStatePerLevel level_state);
  static void freeFrameBufferEntry(FrameBufferEntry *entry);
  static void advanceEdgeForRange(WindowStatePerLevel level_state,
  					WindowState * wstate,
  					WindowFrameEdge * edge,
  					ExprState *edge_expr,
  					ExprState *edge_range_expr,
  					NTupleStoreAccessor * edge_reader,
  					bool is_lead_edge);
  static void forwardEdgeForRange(WindowStatePerLevel level_state,
  					WindowState * wstate,
  					WindowFrameEdge * edge,
  					ExprState *edge_expr,
  					ExprState *edge_range_expr,
  					NTupleStoreAccessor * edge_reader,
  					bool is_lead_edge);
  static bool checkLastRowForEdge(WindowStatePerLevel level_state,
  					WindowState * wstate,
  					WindowFrameEdge * edge,
  					NTupleStoreAccessor * edge_reader,
  					Datum new_edge_value,
  					bool new_edge_value_isnull,
  					bool is_lead_edge);
  static ExprState *make_eq_exprstate(WindowState * wstate,
  				  Expr *expr1, Expr *expr2);
  static bool exec_eq_exprstate(WindowState * wstate, ExprState *eq_exprstate);
  static void setEmptyFrame(WindowStatePerLevel level_state,
  			  WindowState * wstate);
  
  /*
   * initFrameBuffer -- initialize the frame buffer.
   *
   * Create the tuplestore and all accessors.
   */
  static void
  initFrameBuffer(WindowFrameBuffer buffer, int bytes)
  {
  	buffer->tuplestore = ntuplestore_create(bytes);
  
  	/* Create accessors. */
  	buffer->writer = ntuplestore_create_accessor(buffer->tuplestore, true);
  	buffer->reader = ntuplestore_create_accessor(buffer->tuplestore, false);
  
  	buffer->current_row_reader =
  		ntuplestore_create_accessor(buffer->tuplestore, false);
  	buffer->trim_reader =
  		ntuplestore_create_accessor(buffer->tuplestore, false);
  
  	buffer->num_rows_before = buffer->num_rows_after = 0;
  }
  
  /*
   * createRangeFrameBuffer -- create a new WindowFrameBuffer of the RANGE type.
   */
  static WindowFrameBuffer
  createRangeFrameBuffer(Datum trail_range, Datum lead_range, int bytes)
  {
  	WindowFrameBuffer buffer =
  	(WindowFrameBuffer) palloc0(sizeof(WindowFrameBufferData));
  
  	buffer->is_rows = false;
  	buffer->trail_range = trail_range;
  	buffer->lead_range = lead_range;
  
  	initFrameBuffer(buffer, bytes);
  
  	return buffer;
  }
  
  /*
   * createRowsFrameBuffer -- create a new WindowFrameBuffer of the ROWS type.
   */
  static WindowFrameBuffer
  createRowsFrameBuffer(long int trail_rows, long int lead_rows, int bytes)
  {
  	WindowFrameBuffer buffer =
  	(WindowFrameBuffer) palloc0(sizeof(WindowFrameBufferData));
  
  	buffer->is_rows = true;
  	buffer->trail_rows = trail_rows;
  	buffer->lead_rows = lead_rows;
  
  	initFrameBuffer(buffer, bytes);
  
  	return buffer;
  }
  
  /*
   * createFrameBuffers -- create frame buffers for all key levels inside
   * a given WindowState.
   */
  static void
  createFrameBuffers(WindowState * wstate)
  {
  	int			level;
  	int			bytes;
  
  	if (wstate->numlevels <= 0)
  		return;
  
  	bytes = ((PlanStateOperatorMemKB((PlanState *) wstate) * 1024L) / 2) / wstate->numlevels;
  
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		Assert(level_state->frame_buffer == NULL);
  		if (level_state->is_rows)
  			level_state->frame_buffer =
  				createRowsFrameBuffer(level_state->trail_rows,
  									  level_state->lead_rows,
  									  bytes);
  		else
  			level_state->frame_buffer =
  				createRangeFrameBuffer(level_state->trail_range,
  									   level_state->lead_range,
  									   bytes);
  
  		level_state->trail_reader =
  			ntuplestore_create_accessor(level_state->frame_buffer->tuplestore, false);
  		level_state->lead_reader =
  			ntuplestore_create_accessor(level_state->frame_buffer->tuplestore, false);
  
  		level_state->frame_buffer->level_state = level_state;
  	}
  }
  
  /*
   * resetFrameBuffer -- reset or initialize a window frame buffer.
   *
   * Reset the tuplestore in the buffer, and re-create all accessors.
   *
   * The input argument 'buffer' can not be NULL.
   */
  static WindowFrameBuffer
  resetFrameBuffer(WindowFrameBuffer buffer)
  {
  	Assert(buffer != NULL);
  
  	/* destroy all accessors */
  	ntuplestore_destroy_accessor(buffer->writer);
  	ntuplestore_destroy_accessor(buffer->reader);
  	ntuplestore_destroy_accessor(buffer->current_row_reader);
  	ntuplestore_destroy_accessor(buffer->trim_reader);
  
  	ntuplestore_reset(buffer->tuplestore);
  
  	/* (Re)create accessors. */
  	buffer->writer = ntuplestore_create_accessor(buffer->tuplestore, true);
  	buffer->reader = ntuplestore_create_accessor(buffer->tuplestore, false);
  
  	buffer->current_row_reader =
  		ntuplestore_create_accessor(buffer->tuplestore, false);
  	buffer->trim_reader =
  		ntuplestore_create_accessor(buffer->tuplestore, false);
  
  	buffer->num_rows_before = buffer->num_rows_after = 0;
  
  	return buffer;
  }
  
  /*
   * resetFrameBuffers -- reset all frame buffers in a WindowState.
   */
  static void
  resetFrameBuffers(WindowState * wstate)
  {
  	int			level;
  
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		if (level_state->frame_buffer)
  		{
  			/* destroy its accessors */
  			if (level_state->trail_reader)
  				ntuplestore_destroy_accessor(level_state->trail_reader);
  			if (level_state->lead_reader)
  				ntuplestore_destroy_accessor(level_state->lead_reader);
  
  			level_state->frame_buffer = resetFrameBuffer(level_state->frame_buffer);
  
  			/* (Re)create accessors */
  			level_state->trail_reader =
  				ntuplestore_create_accessor(level_state->frame_buffer->tuplestore, false);
  			level_state->lead_reader =
  				ntuplestore_create_accessor(level_state->frame_buffer->tuplestore, false);
  		}
  
  		level_state->num_trail_rows = 0;
  		level_state->num_lead_rows = 0;
  		level_state->lead_ready = false;
  	}
  }
  
  /*
   * freeFrameBuffer -- release the space for a given WindowFrameBuffer.
   */
  static void
  freeFrameBuffer(WindowFrameBuffer buffer)
  {
  	if (buffer == NULL)
  		return;
  	ntuplestore_destroy_accessor(buffer->writer);
  	ntuplestore_destroy_accessor(buffer->reader);
  	ntuplestore_destroy_accessor(buffer->current_row_reader);
  
  	ntuplestore_destroy(buffer->tuplestore);
  
  	pfree(buffer);
  }
  
  /*
   * freeFrameBuffers -- release the space for all frame buffers.
   */
  static void
  freeFrameBuffers(WindowState * wstate)
  {
  	int			level;
  
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		if (level_state->frame_buffer)
  		{
  			/* destroy its accessors */
  			if (level_state->trail_reader)
  				ntuplestore_destroy_accessor(level_state->trail_reader);
  			if (level_state->lead_reader)
  				ntuplestore_destroy_accessor(level_state->lead_reader);
  
  			freeFrameBuffer(level_state->frame_buffer);
  			level_state->frame_buffer = NULL;
  		}
  
  	}
  }
  
  /*
   * ensureSpace -- ensure that there is enough space in the buffer.
   *
   * This function returns the new written position if the array is
   * entended.
   */
  static inline char *
  ensureSpace(char **p_serial_entry, Size * p_max_size,
  			char *written_pos, Size written_len)
  {
  	Size		current_len = written_pos - *p_serial_entry;
  
  	while (current_len + written_len > *p_max_size)
  	{
  		*p_max_size += FRAMEBUFFER_ENTRY_SIZE;
  		*p_serial_entry = repalloc(*p_serial_entry, *p_max_size);
  	}
  
  	return (*p_serial_entry + current_len);
  }
  
  /*
   * serializeValue -- serialize a given value (Datum) into a char array.
   *
   * The start writing position in the given char array is specified by
   * 'start_pos'. If this value is too big to fit into the available
   * space in the array, the size of the array is increased in this
   * function.
   *
   * The return pointer points to the next char position in the
   * given char array after writing the serialized string of the
   * given value.
   */
  static char *
  serializeValue(Datum value, bool isnull,
  			   bool byvalue, int16 typelen,
  			   char **p_serial_entry, Size * p_max_size, Size * p_len,
  			   Size start_pos)
  {
  	char	   *written_pos = (*p_serial_entry) + start_pos;
  
  	written_pos = ensureSpace(p_serial_entry, p_max_size,
  							  written_pos, sizeof(bool));
  
  	memcpy(written_pos, &isnull, sizeof(bool));
  	written_pos += sizeof(bool);	/* isnull col */
  
  	if (!isnull)
  	{
  		if (byvalue)
  		{
  			written_pos = ensureSpace(p_serial_entry, p_max_size,
  									  written_pos, sizeof(Datum));
  
  			memcpy(written_pos, &value, sizeof(Datum));
  			written_pos += sizeof(Datum);
  		}
  		else
  		{
  			Size		real_size = datumGetSize(value, byvalue, typelen);
  
  			/*
  			 * The pointer needs to be properly aligned since the deserializer
  			 * will not make a copy for this Datum, but simply convert this
  			 * pointer address to a Datum.
  			 */
  			Size		alignmentBytes = ((char *)MAXALIGN(written_pos)) - written_pos;
  
  			written_pos = ensureSpace(p_serial_entry, p_max_size,
  									written_pos, real_size + alignmentBytes);
  			written_pos += alignmentBytes;
  
  			Assert(DatumGetPointer(value) != NULL);
  			memcpy(written_pos,
  				   DatumGetPointer(value),
  				   real_size);
  
  			written_pos += real_size;
  		}
  	}
  
  	return written_pos;
  }
  
  /*
   * serializeFuncs -- serialize values for all functions in a given
   * level and put it into a given space, starting from 'start_pos'.
   */
  static void
  serializeFuncs(WindowStatePerLevel level_state,
  			   ExprContext *econtext,
  			   char **p_serial_entry, Size * p_max_size, Size * p_len,
  			   Size start_pos)
  {
  	ListCell   *lc;
  	char	   *written_pos = (*p_serial_entry) + start_pos;
  	int			serial_index = -1;
  
  	/* the value for each function. */
  	foreach(lc, level_state->level_funcs)
  	{
  		WindowStatePerFunction funcstate = (WindowStatePerFunction) lfirst(lc);
  
  		if (funcstate->trivial_frame)
  		{
  			/* ignore trivial window functions */
  			continue;
  		}
  		else if (funcstate->winpeercount)
  		{
  			/* functions that require peer counts. */
  			int16		typelen;
  			bool		byvalue;
  
  			get_typlenbyval(CUME_DIST_PRELIM_TYPE,
  							&typelen, &byvalue);
  
  			written_pos =
  				serializeValue(funcstate->win_value,
  							   funcstate->win_value_is_null,
  							   byvalue, typelen,
  							   p_serial_entry, p_max_size,
  							   p_len, (written_pos - *p_serial_entry));
  		}
  		else if (IS_LEAD_LAG(funcstate->wrxstate->winkind) ||
  				 IS_FIRST_LAST(funcstate->wrxstate->winkind))
  		{
  			WindowRefExprState *wrxstate;
  			int			nargs;
  			int			argno = 1;
  
  			wrxstate = funcstate->wrxstate;
  			nargs = list_length(wrxstate->args);
  			Assert(2 <= nargs && nargs <= 4);
  
  			written_pos =
  				serializeValue(funcstate->aggTransValue,
  							   funcstate->aggTransValueIsNull,
  							   wrxstate->argtypbyval[argno],
  							   wrxstate->argtyplen[argno],
  							   p_serial_entry, p_max_size,
  							   p_len, (written_pos - *p_serial_entry));
  		}
  
  		/*
  		 * the function which has an inverse preliminary function or
  		 * preliminary function.
  		 */
  		else if (OidIsValid(funcstate->invprelimfn_oid) ||
  				 OidIsValid(funcstate->prelimfn_oid))
  		{
  			/*
  			 * Store number of not NULLS when the function has a inverse
  			 * preliminary function.
  			 */
  			if (OidIsValid(funcstate->invprelimfn_oid))
  			{
  				written_pos = ensureSpace(p_serial_entry, p_max_size,
  										  written_pos, sizeof(uint64));
  				memcpy(written_pos, &funcstate->numNotNulls, sizeof(uint64));
  				written_pos += sizeof(uint64);
  			}
  
  			written_pos =
  				serializeValue(funcstate->aggTransValue,
  							   funcstate->aggTransValueIsNull,
  							   funcstate->aggTranstypeByVal,
  							   funcstate->aggTranstypeLen,
  							   p_serial_entry, p_max_size,
  							   p_len, (written_pos - *p_serial_entry));
  		}
  		else
  		{
  			/* the general case */
  			WindowRefExprState *winref_state = funcstate->wrxstate;
  			ListCell   *ref_lc;
  			Oid			typid;
  			bool		byval;
  			int16		arglen;
  
  			/* Store the input arguments */
  			foreach(ref_lc, winref_state->args)
  			{
  				ExprState  *argstate = (ExprState *)lfirst(ref_lc);
  				Datum		value;
  				bool		isnull;
  				MemoryContext oldctx;
  
  				typid = exprType((Node *) argstate->expr);
  				get_typlenbyval(typid, &arglen, &byval);
  
  				oldctx = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
  				value = ExecEvalExpr(argstate, econtext, &isnull, NULL);
  				MemoryContextSwitchTo(oldctx);
  
  				written_pos =
  					serializeValue(value, isnull,
  								   byval, arglen,
  								   p_serial_entry, p_max_size,
  								   p_len, (written_pos - *p_serial_entry));
  			}
  		}
  
  		serial_index++;
  		funcstate->serial_index = serial_index;
  	}
  
  	*p_len = written_pos - *p_serial_entry;
  	Assert(*p_len <= *p_max_size);
  }
  
  /*
   * deserializeValue -- deserialize a Datum value from a given char array.
   *
   * This function reads the char array as specified by 'read_pos', and
   * deserialize one serialized Datum values in this array into a given
   * WindowValue. If a Datum is passed by reference, this function does _not_
   * make a copy of this Datum, but simply set the pointer to the
   * corresponding address of the given char array.
   *
   * The return pointer points to the char position in the given char array
   * after reading one seralized string of WindowValue.
   *
   * Note that 'value' is allocated by the caller.
   */
  static char *
  deserializeValue(char *read_pos, WindowValue * value,
  				 bool byvalue, int16 typelen)
  {
  	memcpy(&(value->valueIsNull), read_pos, sizeof(bool));
  	read_pos += sizeof(bool);
  
  	if (!value->valueIsNull)
  	{
  		if (byvalue)
  		{
  			memcpy(&(value->value), read_pos, sizeof(Datum));
  			read_pos += sizeof(Datum);
  		}
  		else
  		{
  			/*
  			 * Make read_pos properly aligned, since it is the start address
  			 * of a Datum that is passed by reference.
  			 */
  			read_pos = (char *)MAXALIGN(read_pos);
  
  			value->value = PointerGetDatum(read_pos);
  			Size		valueLen = datumGetSize(value->value, byvalue, typelen);
  
  			read_pos += valueLen;
  		}
  	}
  	else
  		value->value = 0;
  
  	return read_pos;
  }
  
  /*
   * deserializeFuncs -- deserialize Datum values from a given char array
   * for all functions in a key level.
   */
  static List *
  deserializeFuncs(WindowStatePerLevel level_state,
  				 FrameBufferEntry *entry_buf,
  				 char *start_pos,
  				 Size len)
  {
  	List	   *func_values = entry_buf->func_values;
  
  	ListCell   *lc;
  	char	   *read_pos = start_pos;
  
  	/* the value for each function. */
  	foreach(lc, level_state->level_funcs)
  	{
  		WindowStatePerFunction funcstate = (WindowStatePerFunction) lfirst(lc);
  		WindowValue *value;
  
  		/* skip window functions with trivial frames */
  		if (funcstate->trivial_frame)
  			continue;
  
  		/* functions that require peer counts. */
  		else if (funcstate->winpeercount)
  		{
  			int16		typelen;
  			bool		byvalue;
  
  			get_typlenbyval(CUME_DIST_PRELIM_TYPE,
  							&typelen, &byvalue);
  			value = list_nth(func_values, funcstate->serial_index);
  			read_pos = deserializeValue(read_pos, value, byvalue, typelen);
  		}
  		else if (IS_LEAD_LAG(funcstate->wrxstate->winkind) ||
  				 IS_FIRST_LAST(funcstate->wrxstate->winkind))
  		{
  			WindowRefExprState *wrxstate;
  			int			nargs;
  			int			argno = 1;
  
  			wrxstate = funcstate->wrxstate;
  			nargs = list_length(wrxstate->args);
  			Assert(2 <= nargs && nargs <= 4);
  
  			value = list_nth(func_values, funcstate->serial_index);
  			read_pos = deserializeValue(read_pos, value,
  										wrxstate->argtypbyval[argno],
  										wrxstate->argtyplen[argno]);
  		}
  		else if (OidIsValid(funcstate->invprelimfn_oid) ||
  				 OidIsValid(funcstate->prelimfn_oid))
  		{
  			/*
  			 * The function which has an inverse preliminary function.
  			 */
  			value = list_nth(func_values, funcstate->serial_index);
  
  			/*
  			 * Read number of not nulls if this function has a preliminary
  			 * function.
  			 */
  			if (OidIsValid(funcstate->invprelimfn_oid))
  			{
  				memcpy(&(value->numNotNulls), read_pos, sizeof(uint64));
  				read_pos += sizeof(uint64);
  			}
  
  			read_pos = deserializeValue(read_pos, value,
  										funcstate->aggTranstypeByVal,
  										funcstate->aggTranstypeLen);
  		}
  		else
  		{
  			/* the general case */
  			WindowRefExprState *winref_state = funcstate->wrxstate;
  			ListCell   *ref_lc;
  			Oid			typid;
  			bool		byval;
  			int16		arglen;
  			int			argno = 0;
  
  			foreach(ref_lc, winref_state->args)
  			{
  				ExprState  *argstate = (ExprState *)lfirst(ref_lc);
  
  				typid = exprType((Node *) argstate->expr);
  				get_typlenbyval(typid, &arglen, &byval);
  
  				value = list_nth(func_values, funcstate->serial_index + argno);
  				read_pos = deserializeValue(read_pos, value, byval, arglen);
  
  				argno++;
  			}
  		}
  	}
  
  	Assert((read_pos - start_pos) == len);
  
  	return func_values;
  }
  
  /*
   * serializeEntry -- construct a serialized version of function values for
   * a given key level using the given char array.
   *
   * If the data generated by this serialization have more bytes than the space
   * provided by the given char array, the size of this array is increased
   * in this function.
   */
  static void
  serializeEntry(WindowStatePerLevel level_state,
  			   ExprContext *econtext,
  			   char **p_serial_entry,
  			   Size * p_max_size,
  			   Size * p_len)
  {
  	int			key_no;
  	char	   *written_pos;
  	TupleTableSlot *slot = econtext->ecxt_scantuple;
  
  	*p_len = 0;
  	Assert(*p_serial_entry != NULL);
  
  	/* We rely on the address of the char array is maxaligned. */
  	Assert(*p_serial_entry == (char *)MAXALIGN(*p_serial_entry));
  
  	written_pos = *p_serial_entry;
  
  	if (!level_state->is_rows)
  	{
  		/* Copy the keys */
  		for (key_no = 0; key_no < level_state->numSortCols; key_no++)
  		{
  			AttrNumber	attnum = level_state->sortColIdx[key_no];
  			Datum		key;
  			bool		isnull;
  
  			key = slot_getattr(slot, attnum, &isnull);
  
  			written_pos = ensureSpace(p_serial_entry, p_max_size, written_pos,
  									  sizeof(bool));
  			memcpy(written_pos, &(isnull), sizeof(bool));
  			written_pos += sizeof(bool);
  
  			if (!isnull)
  			{
  				Size		keylen = 0;
  				Size		alignmentBytes = 0;
  
  				if (level_state->col_typbyvals[key_no])
  				{
  					keylen = sizeof(Datum);
  				}
  
  				else
  				{
  					alignmentBytes = ((char *)MAXALIGN(written_pos)) - written_pos;
  					keylen = datumGetSize(key, level_state->col_typbyvals[key_no],
  										  level_state->col_typlens[key_no]);
  				}
  
  				written_pos = ensureSpace(p_serial_entry, p_max_size, written_pos,
  										  keylen + alignmentBytes);
  
  				written_pos += alignmentBytes;
  
  				if (level_state->col_typbyvals[key_no])
  					memcpy(written_pos, &(key), keylen);
  				else
  					memcpy(written_pos, DatumGetPointer(key), keylen);
  				written_pos += keylen;
  			}
  		}
  	}
  
  	/* Copy function values */
  	serializeFuncs(level_state,
  				   econtext,
  				   p_serial_entry,
  				   p_max_size,
  				   p_len,
  				   (written_pos - *p_serial_entry));
  }
  
  /*
   * deserializeEntry -- deserialize a buffer entry.
   */
  static FrameBufferEntry *
  deserializeEntry(WindowStatePerLevel level_state,
  				 FrameBufferEntry *entry_buf,
  				 char *serial_entry, Size len)
  {
  	FrameBufferEntry *entry = entry_buf;
  	char	   *read_pos = serial_entry;
  	Size		keylen = 0;
  	int			key_no;
  
  	/* We rely on the address of serial_entry is maxaligned. */
  	Assert(serial_entry == (char *)MAXALIGN(serial_entry));
  
  	if (!level_state->is_rows)
  	{
  		for (key_no = 0; key_no < level_state->numSortCols; key_no++)
  		{
  			Size		curr_keylen = 0;
  
  			memcpy(&(entry->nulls[key_no]), read_pos, sizeof(bool));
  			read_pos += sizeof(bool);
  			keylen += sizeof(bool);
  
  			if (!entry->nulls[key_no])
  			{
  				if (level_state->col_typbyvals[key_no])
  				{
  					memcpy(&(entry->keys[key_no]), read_pos, sizeof(Datum));
  					curr_keylen = sizeof(Datum);
  				}
  
  				else
  				{
  					Size		alignmentBytes = ((char *)MAXALIGN(read_pos)) - read_pos;
  
  					read_pos += alignmentBytes;
  					keylen += alignmentBytes;
  
  					curr_keylen = datumGetSize(PointerGetDatum(read_pos),
  										  level_state->col_typbyvals[key_no],
  										   level_state->col_typlens[key_no]);
  					entry->keys[key_no] = PointerGetDatum(read_pos);
  				}
  				read_pos += curr_keylen;
  				keylen += curr_keylen;
  			}
  		}
  	}
  
  	/* deserial function values */
  	entry->func_values = deserializeFuncs(level_state, entry_buf,
  										  read_pos, len - keylen);
  
  	return entry;
  }
  
  /*
   * adjustEdgesAfterAppend -- adjust the trailing and leading edges for
   * a given level_state after appending a value into its frame buffer.
   */
  static void
  adjustEdgesAfterAppend(WindowStatePerLevel level_state,
  					   WindowState * wstate,
  					   bool last_peer)
  {
  	WindowFrameBuffer buffer = level_state->frame_buffer;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  	TupleTableSlot *inserting_tuple = econtext->ecxt_scantuple;
  
  	/*
  	 * If the current_row_reader for the buffer is not set, set it to point to
  	 * the last row.
  	 */
  	if (!ntuplestore_acc_tell(buffer->current_row_reader, NULL))
  		ntuplestore_acc_seek_last(buffer->current_row_reader);
  
  	/*
  	 * Adjust the trailing edge if it is "ROWS x FOLLOWING" or it is a delayed
  	 * edge.
  	 */
  	if (!(level_state->empty_frame &&
  		  EDGE_IS_DELAYED(level_state->frame->trail)))
  	{
  		if (level_state->is_rows &&
  			EDGE_IS_BOUND(level_state->frame->trail) &&
  			level_state->trail_rows > 0)
  		{
  			if (!ntuplestore_acc_tell(level_state->trail_reader, NULL))
  			{
  				if (!last_peer)
  					ntuplestore_acc_seek_last(level_state->trail_reader);
  			}
  			else if (level_state->num_trail_rows < level_state->trail_rows - 1)
  			{
  				ntuplestore_acc_advance(level_state->trail_reader, 1);
  				level_state->num_trail_rows++;
  			}
  		}
  		else if (!level_state->is_rows &&
  				 (EDGE_IS_BOUND_FOLLOWING(level_state->frame->trail) ||
  				  EDGE_IS_DELAYED(level_state->frame->trail)))
  		{
  			econtext->ecxt_scantuple = wstate->curslot;
  			forwardEdgeForRange(level_state, wstate,
  								level_state->frame->trail,
  								level_state->trail_expr,
  								level_state->trail_range_expr,
  								level_state->trail_reader,
  								false);
  
  			econtext->ecxt_scantuple = inserting_tuple;
  		}
  	}
  
  
  	if (!(level_state->empty_frame &&
  		  EDGE_IS_DELAYED(level_state->frame->lead)))
  	{
  		/* If the leading edge is "x PRECEDING" and x > 0, simply return. */
  		if (EDGE_IS_BOUND_PRECEDING(level_state->frame->lead) &&
  			ntuplestore_acc_tell(level_state->lead_reader, NULL))
  		{
  			level_state->lead_ready = true;
  			return;
  		}
  
  		/*
  		 * Adjust the leading edge when the edge is "x FOLLOWING" or "CURRENT
  		 * ROW".
  		 */
  		if (level_state->is_rows)
  		{
  			if (level_state->num_lead_rows < level_state->lead_rows)
  			{
  				Assert(!ntuplestore_acc_tell(level_state->lead_reader, NULL));
  
  				if (!last_peer)
  					level_state->num_lead_rows++;
  			}
  			else if ((EDGE_IS_BOUND(level_state->frame->lead) ||
  					  EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->lead, true)) &&
  					 level_state->num_lead_rows == level_state->lead_rows)
  			{
  				/* If the leading edge is not set, set it here. */
  				if (!ntuplestore_acc_tell(level_state->lead_reader, NULL))
  					ntuplestore_acc_seek_last(level_state->lead_reader);
  			}
  			else if (!EDGE_IS_BOUND(level_state->frame->lead) &&
  					 !EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->lead, true))
  			{
  				if (!ntuplestore_acc_tell(level_state->lead_reader, NULL))
  					ntuplestore_acc_seek_last(level_state->lead_reader);
  				else
  				{
  					bool		found = ntuplestore_acc_advance(level_state->lead_reader, 1);
  
  					if (found)
  						level_state->num_lead_rows++;
  				}
  			}
  		}
  		else
  		{
  			if (EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->lead,
  									true))
  			{
  				bool		found;
  				NTupleStorePos pos;
  
  				found = ntuplestore_acc_tell(buffer->current_row_reader, &pos);
  				Assert(found);
  				ntuplestore_acc_seek(level_state->lead_reader, &pos);
  
  				level_state->lead_ready = true;
  			}
  			else if (EDGE_IS_BOUND(level_state->frame->lead))
  			{
  				econtext->ecxt_scantuple = wstate->curslot;
  				forwardEdgeForRange(level_state, wstate,
  									level_state->frame->lead,
  									level_state->lead_expr,
  									level_state->lead_range_expr,
  									level_state->lead_reader,
  									true);
  				econtext->ecxt_scantuple = inserting_tuple;
  			}
  			else
  			{
  				ntuplestore_acc_advance(level_state->lead_reader, 1);
  				level_state->lead_ready = last_peer;
  			}
  		}
  	}
  }
  
  /*
   * appendToFrameBuffer -- append the intermediate values stored in level
   */
  static void
  appendToFrameBuffer(WindowStatePerLevel level_state,
  					WindowState * wstate,
  					bool last_peer)
  {
  	Size		len;
  	WindowFrameBuffer buffer = level_state->frame_buffer;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  
  	Assert(buffer->is_rows == level_state->is_rows);
  	MemSet(wstate->serial_array, 0, wstate->max_size);
  	serializeEntry(level_state, econtext,
  				   &(wstate->serial_array), &(wstate->max_size), &len);
  
  	ntuplestore_acc_put_data(buffer->writer, (void *)(wstate->serial_array), len);
  
  	adjustEdgesAfterAppend(level_state, wstate, last_peer);
  
  	buffer->num_rows_after++;
  }
  
  /*
   * trimFrameBuffer -- trim the old values in the buffer.
   */
  static void
  trimFrameBuffer(WindowFrameBuffer buffer)
  {
  	NTupleStorePos pos;
  	bool		found;
  
  	Assert(buffer != NULL);
  
  	found = ntuplestore_acc_tell(buffer->trim_reader, &pos);
  	if (!found)
  		return;
  
  	ntuplestore_trim(buffer->tuplestore, &pos);
  }
  
  /*
   * getCurrentValue -- obtain the value at the current position for
   * an accessor.
   *
   * The value is stored in the given 'entry_buf'. Note that 'entry_buf'
   * is allocated by the caller.
   *
   * If the current position of the accessor is an invalid position, this
   * function returns false.
   */
  static bool
  getCurrentValue(NTupleStoreAccessor * reader,
  				WindowStatePerLevel level_state,
  				FrameBufferEntry *entry_buf)
  {
  	void	   *data;
  	int			len;
  
  	bool		found = ntuplestore_acc_current_data(reader, &data, &len);
  
  	if (!found)
  		return false;
  
  	entry_buf = deserializeEntry(level_state, entry_buf, data, len);
  
  	return true;
  }
  
  /*
   * incrementCurrentRow -- increment the current_row in the buffer.
   */
  static void
  incrementCurrentRow(WindowFrameBuffer buffer,
  					WindowState * wstate)
  {
  	WindowStatePerLevel level_state = buffer->level_state;
  
  	/* Increment the current_row_reader */
  	ntuplestore_acc_advance(buffer->current_row_reader, 1);
  
  	buffer->num_rows_before++;
  	if (buffer->num_rows_after > 0)
  		buffer->num_rows_after--;
  
  	if (level_state->is_rows)
  	{
  		level_state->num_trail_rows--;
  		level_state->num_lead_rows--;
  	}
  
  	if (!buffer->level_state->trivial_frames_only)
  		/* Adjust all leading and trailig edges */
  		adjustEdges(buffer, wstate);
  
  	/* Set the trim_reader, and trim the buffer */
  	if (!level_state->has_delay_bound)
  	{
  		NTupleStorePos pos;
  		bool		found = ntuplestore_acc_tell(buffer->current_row_reader, &pos);
  
  		if (found)
  			ntuplestore_acc_seek(buffer->trim_reader, &pos);
  		else
  			ntuplestore_acc_set_invalid(buffer->trim_reader);
  
  		if (ntuplestore_acc_tell(level_state->trail_reader, NULL) &&
  			((ntuplestore_acc_tell(buffer->trim_reader, NULL) &&
  			  ntuplestore_acc_is_before(level_state->trail_reader,
  										buffer->trim_reader)) ||
  			 !ntuplestore_acc_tell(buffer->trim_reader, NULL)))
  		{
  			found = ntuplestore_acc_tell(level_state->trail_reader, &pos);
  			Assert(found);
  			ntuplestore_acc_seek(buffer->trim_reader, &pos);
  		}
  
  		if (!ntuplestore_acc_tell(level_state->trail_reader, NULL))
  			ntuplestore_acc_set_invalid(buffer->trim_reader);
  		else
  			ntuplestore_acc_advance(buffer->trim_reader, -1);
  
  		trimFrameBuffer(buffer);
  	}
  }
  
  /*
   * hasEnoughDataInRange -- return true if there is enough data in
   * the buffer that satisfy a given value range.
   */
  static bool
  hasEnoughDataInRange(WindowFrameBuffer buffer,
  					 WindowStatePerLevel level_state,
  					 WindowState * wstate,
  					 Datum trail_range, Datum lead_range)
  {
  	if (level_state->empty_frame)
  		return true;
  
  	if (!EDGE_IS_BOUND(level_state->frame->lead) &&
  	!EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->lead, true))
  		return false;
  
  	return level_state->lead_ready;
  }
  
  
  /*
   * hasEnoughDataInRows -- return true if there is enough data in the
   * buffer that satisfy	a given row range.
   */
  static bool
  hasEnoughDataInRows(WindowFrameBuffer buffer,
  					WindowStatePerLevel level_state,
  					WindowState * wstate,
  					long int trail_rows, long int lead_rows)
  {
  	if (level_state->empty_frame)
  		return true;
  
  	if (!EDGE_IS_BOUND(level_state->frame->lead) &&
  	!EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->lead, true))
  		return false;
  
  	if (trail_rows == 0 &&
  		lead_rows == 0)
  		return true;
  
  	if (level_state->num_lead_rows >= lead_rows)
  		return true;
  	return false;
  }
  
  static FrameBufferEntry *
  createFrameBufferEntry(WindowStatePerLevel level_state)
  {
  	FrameBufferEntry *entry = NULL;
  	ListCell   *lc;
  
  	entry = (FrameBufferEntry *)palloc0(sizeof(FrameBufferEntry));
  	if (!level_state->is_rows)
  	{
  		entry->keys = (Datum *)palloc0(level_state->numSortCols * sizeof(Datum));
  		entry->nulls = (bool *)palloc0(level_state->numSortCols * sizeof(bool));
  	}
  
  	entry->func_values = NIL;
  
  	/* Allocate the space for WindowValue */
  	foreach(lc, level_state->level_funcs)
  	{
  		WindowStatePerFunction funcstate = (WindowStatePerFunction) lfirst(lc);
  		WindowValue *value;
  
  		/* skip window functions with trivial frames */
  		if (funcstate->trivial_frame)
  			continue;
  
  		else if (funcstate->winpeercount ||
  				 (IS_LEAD_LAG(funcstate->wrxstate->winkind) ||
  				  IS_FIRST_LAST(funcstate->wrxstate->winkind)) ||
  				 (OidIsValid(funcstate->invprelimfn_oid) ||
  				  OidIsValid(funcstate->prelimfn_oid)))
  		{
  			value = (WindowValue *) palloc0(sizeof(WindowValue));
  			entry->func_values = lappend(entry->func_values, value);
  		}
  
  		else
  		{
  			WindowRefExprState *winref_state = funcstate->wrxstate;
  			int			argno;
  
  			for (argno = 0; argno < list_length(winref_state->args); argno++)
  			{
  				value = (WindowValue *) palloc0(sizeof(WindowValue));
  				entry->func_values = lappend(entry->func_values, value);
  			}
  		}
  	}
  
  	return entry;
  }
  
  static void
  freeFrameBufferEntry(FrameBufferEntry *entry)
  {
  	if (entry->keys != NULL)
  		pfree(entry->keys);
  
  	if (entry->nulls != NULL)
  		pfree(entry->nulls);
  
  	if (entry->func_values != NULL)
  		list_free_deep(entry->func_values);
  }
  
  /*
   * hasTuplesInFrame - Check if there are tuples between the trailing
   * and leading frame edge for the current_row.
   *
   * This function returns true if there are such tuples. Otherwise,
   * return false.
   */
  static bool
  hasTuplesInFrame(WindowStatePerLevel level_state,
  				 WindowState * wstate)
  {
  	bool		has_tuples = true;
  	NTupleStorePos orig_pos;
  	bool		trail_valid;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  
  	if (level_state->empty_frame)
  		return false;
  
  	/*
  	 * If both trail_reader and lead_reader point to the same position, there
  	 * are no tuples within the frame edges.
  	 */
  	if (ntuplestore_acc_tell(level_state->trail_reader, NULL) &&
  		ntuplestore_acc_tell(level_state->lead_reader, NULL) &&
  		(!ntuplestore_acc_is_before(level_state->lead_reader,
  									level_state->trail_reader) &&
  		 !ntuplestore_acc_is_before(level_state->trail_reader,
  									level_state->lead_reader)))
  		return false;
  
  	/* Save the position for the trail_reader */
  	trail_valid = ntuplestore_acc_tell(level_state->trail_reader, &orig_pos);
  
  	if (trail_valid)
  		ntuplestore_acc_advance(level_state->trail_reader, 1);
  	else
  	{
  		/*
  		 * When the leading edge and trailing edge are both RANGE x PRECEDING,
  		 * and the trail_reader points to an invalid position but the
  		 * lead_reader points to a valid position, it is possible that there
  		 * are no tuples in the current frame.
  		 */
  		if (!level_state->is_rows &&
  			EDGE_IS_BOUND_PRECEDING(level_state->frame->lead) &&
  			EDGE_IS_BOUND_PRECEDING(level_state->frame->trail) &&
  			ntuplestore_acc_tell(level_state->lead_reader, NULL) &&
  			!ntuplestore_acc_tell(level_state->trail_reader, NULL))
  		{
  			bool		is_less = true;
  			MemoryContext oldctx;
  			Datum		trail_edge_value;
  			bool		trail_edge_value_isnull;
  			bool		has_entry;
  			FrameBufferEntry *entry = level_state->curr_entry_buf;
  
  			oldctx = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
  			trail_edge_value =
  				ExecEvalExpr(level_state->trail_range_expr, econtext,
  							 &(trail_edge_value_isnull), NULL);
  
  			has_entry = getCurrentValue(level_state->lead_reader, level_state, entry);
  			Assert(has_entry);
  			is_less = cmp_deformed_tuple(entry->keys, entry->nulls,
  										 &trail_edge_value,
  										 &trail_edge_value_isnull,
  										 level_state->col_sort_asc,
  										 level_state->numSortCols,
  										 level_state->ltfunctions,
  										 level_state->eqfunctions,
  										 wstate->cmpcontext,
  										 false);
  			MemoryContextSwitchTo(oldctx);
  
  			return (!is_less);
  		}
  
  		if (!EDGE_IS_BOUND_FOLLOWING(level_state->frame->trail) ||
  			EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->trail, false))
  			ntuplestore_acc_seek_first(level_state->trail_reader);
  	}
  
  	/*
  	 * When both trail_reader and lead_reader points to an entry in the frame
  	 * buffer, we compare their position order.
  	 */
  	if (ntuplestore_acc_tell(level_state->trail_reader, NULL) &&
  		ntuplestore_acc_tell(level_state->lead_reader, NULL) &&
  		ntuplestore_acc_is_before(level_state->lead_reader,
  								  level_state->trail_reader))
  		has_tuples = false;
  
  	/*
  	 * When the trail_reader does not point to any entry in the frame buffer,
  	 * and the trailing edge is x FOLLOWING, it is possible that there are no
  	 * tuples within frame edges for the current_row.
  	 */
  	else if (!ntuplestore_acc_tell(level_state->trail_reader, NULL) &&
  			 EDGE_IS_BOUND_FOLLOWING(level_state->frame->trail))
  	{
  		if (level_state->is_rows)
  		{
  			if (level_state->num_lead_rows < level_state->trail_rows)
  				has_tuples = false;
  		}
  
  		else
  		{
  			if (!ntuplestore_acc_tell(level_state->trail_reader, NULL) &&
  				!exec_eq_exprstate(wstate, level_state->trail_range_eq_expr))
  			{
  				Datum		new_edge_value = 0;
  				bool		new_edge_value_isnull = false;
  				bool		in_trail_edge = false;
  				bool		in_lead_edge = false;
  				MemoryContext oldctx;
  
  				has_tuples = false;
  
  				/* Check if the last row is within the edges */
  
  				oldctx = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
  				new_edge_value =
  					ExecEvalExpr(level_state->trail_range_expr, econtext,
  								 &(new_edge_value_isnull), NULL);
  				MemoryContextSwitchTo(oldctx);
  
  				in_trail_edge =
  					checkLastRowForEdge(level_state, wstate,
  										level_state->frame->trail,
  										level_state->trail_reader,
  										new_edge_value,
  										new_edge_value_isnull,
  										false);
  
  				in_lead_edge = true;
  				if (level_state->lead_range_expr != NULL)
  				{
  					oldctx = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
  					new_edge_value =
  						ExecEvalExpr(level_state->lead_range_expr, econtext,
  									 &(new_edge_value_isnull), NULL);
  					MemoryContextSwitchTo(oldctx);
  
  					in_lead_edge =
  						checkLastRowForEdge(level_state, wstate,
  											level_state->frame->lead,
  											level_state->lead_reader,
  											new_edge_value,
  											new_edge_value_isnull,
  											true);
  				}
  
  				if (in_trail_edge && in_lead_edge)
  					has_tuples = true;
  			}
  		}
  	}
  
  	/*
  	 * When the lead_reader does not point to any entry in the frame buffer,
  	 * and the leading edge is x PRECEDING, it means that there are no tuples
  	 * within the frame edges for the current_row.
  	 */
  	else if (!ntuplestore_acc_tell(level_state->lead_reader, NULL) &&
  			 (EDGE_IS_BOUND_PRECEDING(level_state->frame->lead) &&
  			  ((level_state->is_rows && level_state->lead_rows != 0) ||
  			   (!level_state->is_rows && !exec_eq_exprstate(wstate,
  										 level_state->lead_range_eq_expr)))))
  		has_tuples = false;
  
  	/* Restore trail_reader */
  	if (trail_valid)
  		ntuplestore_acc_seek(level_state->trail_reader, &orig_pos);
  	else
  		ntuplestore_acc_set_invalid(level_state->trail_reader);
  
  	return has_tuples;
  }
  
  /*
   * freeTransValue -- release the space allocated for the given transition value.
   */
  static void
  freeTransValue(Datum *transValue,
  			   bool transTypeByVal,
  			   bool *transValueIsNull,
  			   bool *noTransValue,
  			   bool shouldFree)
  {
  	if (!shouldFree)
  		return;
  
  	if (!transTypeByVal && !(*noTransValue) && !(*transValueIsNull) &&
  		DatumGetPointer(*transValue) != NULL)
  		pfree(DatumGetPointer(*transValue));
  
  	*transValue = 0;
  	*transValueIsNull = true;
  	*noTransValue = true;
  }
  
  /*
   * computeTransValuesThroughScan -- compute transition values
   * for those functions in the given level whose aggregate values
   * can only be computed through scanning through multiple entries
   * in the frame buffer.
   */
  static void
  computeTransValuesThroughScan(WindowStatePerLevel level_state,
  							  WindowState * wstate)
  {
  	/* Indicate if the current frame contains tuples at all. */
  	bool		has_tuples = true;
  	ListCell   *lc;
  	bool		has_curr_entry = false;
  	FrameBufferEntry *curr_entry = level_state->curr_entry_buf;
  	WindowValue *curr_value = NULL;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  	FunctionCallInfoData fcinfo;
  	NTupleStorePos orig_pos;
  
  	has_tuples = hasTuplesInFrame(level_state, wstate);
  
  	/* Remember the original trail_reader position. */
  	ntuplestore_acc_tell(level_state->trail_reader, &orig_pos);
  
  	/*
  	 * Since the trail_reader points to the value before the trailing edge, we
  	 * advance the trail_reader by one.
  	 */
  	if (ntuplestore_acc_tell(level_state->trail_reader, NULL))
  		ntuplestore_acc_advance(level_state->trail_reader, 1);
  	else
  	{
  		if (!ntuplestore_acc_tell(level_state->lead_reader, NULL) &&
  			(EDGE_IS_BOUND_PRECEDING(level_state->frame->lead) &&
  			 ((level_state->is_rows && level_state->lead_rows != 0) ||
  			  (!level_state->is_rows && !exec_eq_exprstate(wstate,
  										 level_state->lead_range_eq_expr)))))
  			has_tuples = false;
  		else
  			ntuplestore_acc_seek_first(level_state->trail_reader);
  	}
  
  	foreach(lc, level_state->level_funcs)
  	{
  		WindowStatePerFunction funcstate = (WindowStatePerFunction)
  		lfirst(lc);
  
  		/*
  		 * Ignore those functions that do not require scanning multiple
  		 * entries in the frame buffer.
  		 */
  		if (funcstate->trivial_frame ||
  			funcstate->winpeercount ||
  			(funcstate->isAgg && OidIsValid(funcstate->invprelimfn_oid)) ||
  			!funcstate->isAgg)
  			continue;
  
  		freeTransValue(&funcstate->final_aggTransValue,
  					   funcstate->aggTranstypeByVal,
  					   &funcstate->final_aggTransValueIsNull,
  					   &funcstate->final_aggNoTransValue,
  					   funcstate->final_aggShouldFree);
  
  		funcstate->final_aggTransValue =
  			datumCopyWithMemManager(0, funcstate->aggInitValue,
  									funcstate->aggTranstypeByVal,
  									funcstate->aggTranstypeLen,
  									&(wstate->mem_manager));
  		funcstate->final_aggTransValueIsNull = funcstate->aggInitValueIsNull;
  		funcstate->final_aggNoTransValue = funcstate->aggInitValueIsNull;
  		funcstate->final_aggShouldFree = !funcstate->aggInitValueIsNull;
  	}
  
  	if (has_tuples)
  	{
  		bool		include_last_agg = false;
  
  		while (ntuplestore_acc_tell(level_state->trail_reader, NULL))
  		{
  			if (ntuplestore_acc_tell(level_state->lead_reader, NULL) &&
  				ntuplestore_acc_is_before(level_state->lead_reader,
  										  level_state->trail_reader))
  				break;
  
  			has_curr_entry = getCurrentValue(level_state->trail_reader,
  											 level_state, curr_entry);
  
  			Assert(has_curr_entry);
  
  			foreach(lc, level_state->level_funcs)
  			{
  				WindowStatePerFunction funcstate = (WindowStatePerFunction)
  				lfirst(lc);
  				FmgrInfo   *fmgr_info;
  
  				/*
  				 * Ignore those functions that do not require scanning
  				 * multiple entries in the frame buffer.
  				 */
  				if (funcstate->trivial_frame ||
  					funcstate->winpeercount ||
  					(funcstate->isAgg &&
  					 OidIsValid(funcstate->invprelimfn_oid)) ||
  					!funcstate->isAgg)
  					continue;
  
  				if (OidIsValid(funcstate->prelimfn_oid))
  				{
  					curr_value =
  						(WindowValue *) list_nth(curr_entry->func_values,
  												 funcstate->serial_index);
  
  					Assert(curr_value);
  
  					fcinfo.arg[1] = curr_value->value;
  					fcinfo.argnull[1] = curr_value->valueIsNull;
  					fmgr_info = &(funcstate->prelimfn);
  				}
  
  				else
  				{
  					int			argno;
  
  					for (argno = 1; argno <= funcstate->numargs; argno++)
  					{
  						curr_value = (WindowValue *)
  							list_nth(curr_entry->func_values,
  									 funcstate->serial_index + (argno - 1));
  						Assert(curr_value);
  
  						fcinfo.arg[argno] = curr_value->value;
  						fcinfo.argnull[argno] = curr_value->valueIsNull;
  					}
  					fmgr_info = &(funcstate->transfn);
  				}
  
  				funcstate->final_aggTransValue =
  					invoke_agg_trans_func(fmgr_info,
  										  fmgr_info->fn_nargs - 1,
  										  funcstate->final_aggTransValue,
  										  &funcstate->final_aggNoTransValue,
  									 &(funcstate->final_aggTransValueIsNull),
  										  funcstate->aggTranstypeByVal,
  										  funcstate->aggTranstypeLen,
  										  &fcinfo, (void *)wstate,
  										  econtext->ecxt_per_tuple_memory,
  										  &(wstate->mem_manager));
  				funcstate->final_aggShouldFree = true;
  			}
  
  			ntuplestore_acc_advance(level_state->trail_reader, 1);
  		}
  
  		/*
  		 * Add the funcstate->aggTransValue if it is in the current frame.
  		 */
  		if (EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->trail, false))
  			include_last_agg = true;
  
  		if (!EDGE_IS_BOUND(level_state->frame->lead) ||
  			EDGE_IS_BOUND_FOLLOWING(level_state->frame->lead) ||
  			EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->lead, true))
  			include_last_agg = true;
  
  		if (!ntuplestore_acc_tell(level_state->lead_reader, NULL) &&
  			level_state->agg_filled && include_last_agg)
  		{
  			foreach(lc, level_state->level_funcs)
  			{
  				WindowStatePerFunction funcstate = (WindowStatePerFunction)
  				lfirst(lc);
  
  				/*
  				 * Ignore those functions that do not require scanning
  				 * multiple entries in the frame buffer.
  				 */
  				if (funcstate->trivial_frame ||
  					funcstate->winpeercount ||
  					(funcstate->isAgg && OidIsValid(funcstate->invprelimfn_oid)) ||
  					!funcstate->isAgg)
  					continue;
  
  				if (OidIsValid(funcstate->prelimfn_oid))
  				{
  					fcinfo.arg[1] = funcstate->aggTransValue;
  					fcinfo.argnull[1] = funcstate->aggTransValueIsNull;
  
  					funcstate->final_aggTransValue =
  						invoke_agg_trans_func(&(funcstate->prelimfn),
  											funcstate->prelimfn.fn_nargs - 1,
  											  funcstate->final_aggTransValue,
  										   &funcstate->final_aggNoTransValue,
  									 &(funcstate->final_aggTransValueIsNull),
  											  funcstate->aggTranstypeByVal,
  											  funcstate->aggTranstypeLen,
  											  &fcinfo, (void *)wstate,
  											  econtext->ecxt_per_tuple_memory,
  											  &(wstate->mem_manager));
  					funcstate->final_aggShouldFree = true;
  				}
  				else
  				{
  					TupleTableSlot *slot = econtext->ecxt_scantuple;
  					bool		found;
  
  					found = ntuplestore_acc_current_tupleslot(wstate->input_buffer->writer,
  															  wstate->spare);
  					Assert(found);
  
  					econtext->ecxt_scantuple = wstate->spare;
  					add_tuple_to_trans(funcstate, wstate, econtext, false);
  
  					/* Reset back to its orginial value */
  					econtext->ecxt_scantuple = slot;
  				}
  			}
  		}
  	}
  
  	/* Reset the trail_reader to its original position. */
  	if (!ntuplestore_acc_seek(level_state->trail_reader, &orig_pos))
  		ntuplestore_acc_set_invalid(level_state->trail_reader);
  }
  
  /*
   * computeFrameValue -- compute transition values for each function
   * in a given key level from the data stored in its frame buffer.
   *
   * The 'trail_reader' and 'lead_reader' point to the positions of
   * the trailing and leading edges.
   */
  static void
  computeFrameValue(WindowStatePerLevel level_state,
  				  WindowState * wstate,
  				  NTupleStoreAccessor * trail_reader,
  				  NTupleStoreAccessor * lead_reader)
  {
  	ListCell   *lc;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  	WindowFrameBuffer buffer = level_state->frame_buffer;
  	FunctionCallInfoData fcinfo;
  
  	FrameBufferEntry *trail_entry = level_state->trail_entry_buf;
  	FrameBufferEntry *lead_entry = level_state->lead_entry_buf;
  	FrameBufferEntry *curr_entry = level_state->curr_entry_buf;
  	bool		has_trail_entry = false;
  	bool		has_lead_entry = false;
  	bool		has_curr_entry = false;
  
  	WindowValue *trail_value = NULL;
  	WindowValue *lead_value = NULL;
  	WindowValue *curr_value = NULL;
  
  	bool		read_value = false;
  	bool		read_edge_value = false;
  
  	bool		require_scanning = false;
  
  	Assert(level_state->is_rows == buffer->is_rows);
  
  	foreach(lc, level_state->level_funcs)
  	{
  		WindowStatePerFunction funcstate = (WindowStatePerFunction) lfirst(lc);
  
  		/*
  		 * Ignore ordinary window functions, except for functions that require
  		 * peer counts.
  		 */
  		if (funcstate->trivial_frame)
  			continue;
  
  		/* Ignore functions that are cumulative aggregate functions. */
  		if (funcstate->cumul_frame)
  			continue;
  
  		/* functions that require peer counts. */
  		else if (funcstate->winpeercount)
  		{
  			/*
  			 * If we didn't read the value pointed by current_row_reader, read
  			 * it now.
  			 */
  			if (!read_value)
  			{
  				read_value = true;
  				has_curr_entry = getCurrentValue(buffer->current_row_reader,
  												 level_state, curr_entry);
  			}
  
  			if (has_curr_entry)
  				curr_value = (WindowValue *) list_nth(curr_entry->func_values,
  													funcstate->serial_index);
  
  			if (curr_value != NULL)
  			{
  				funcstate->win_value = curr_value->value;
  				funcstate->win_value_is_null = curr_value->valueIsNull;
  			}
  			else
  			{
  				TupleTableSlot *slot = econtext->ecxt_scantuple;
  
  				ntuplestore_acc_current_tupleslot(wstate->input_buffer->writer,
  												  wstate->spare);
  				econtext->ecxt_scantuple = wstate->spare;
  				add_tuple_to_trans(funcstate, wstate, econtext, false);
  
  				/* Reset back to its orginial value */
  				econtext->ecxt_scantuple = slot;
  			}
  		}
  
  		/* Aggregate functions with inverse preliminary functions. */
  		else if (funcstate->isAgg && OidIsValid(funcstate->invprelimfn_oid))
  		{
  			uint64		trail_num_not_nulls = 0;
  			uint64		lead_num_not_nulls = 0;
  
  			if (!read_edge_value)
  			{
  				read_edge_value = true;
  
  				if (level_state->is_rows)
  				{
  					if (!EDGE_IS_BOUND_FOLLOWING(level_state->frame->trail) ||
  					level_state->num_lead_rows >= level_state->num_trail_rows)
  					{
  						has_trail_entry =
  							getCurrentValue(level_state->trail_reader,
  											level_state, trail_entry);
  						has_lead_entry =
  							getCurrentValue(level_state->lead_reader,
  											level_state, lead_entry);
  					}
  				}
  				else
  				{
  					has_trail_entry = getCurrentValue(level_state->trail_reader,
  												   level_state, trail_entry);
  
  					/*
  					 * If the leading edge is UNBOUNDED FOLLOWING, the value
  					 * in the leading edge is funcstate->aggTransValue. We
  					 * don't need to read from the buffer.
  					 */
  					if (level_state->frame->lead->kind !=
  						WINDOW_UNBOUND_FOLLOWING)
  					{
  						has_lead_entry =
  							getCurrentValue(level_state->lead_reader,
  											level_state, lead_entry);
  					}
  				}
  			}
  
  			if (!level_state->empty_frame && has_trail_entry)
  			{
  				Assert(list_length(trail_entry->func_values) >
  					   funcstate->serial_index);
  
  				trail_value =
  					(WindowValue *) list_nth(trail_entry->func_values,
  											 funcstate->serial_index);
  			}
  
  			if (!level_state->empty_frame && has_lead_entry)
  			{
  				Assert(list_length(lead_entry->func_values) >
  					   funcstate->serial_index);
  
  				lead_value =
  					(WindowValue *) list_nth(lead_entry->func_values,
  											 funcstate->serial_index);
  			}
  
  			if (trail_value != NULL)
  			{
  				fcinfo.arg[1] = trail_value->value;
  				fcinfo.argnull[1] = trail_value->valueIsNull;
  				trail_num_not_nulls = trail_value->numNotNulls;
  			}
  			else
  			{
  				fcinfo.arg[1] = funcstate->aggInitValue;
  				fcinfo.argnull[1] = funcstate->aggInitValueIsNull;
  				trail_num_not_nulls = 0;
  			}
  
  			freeTransValue(&funcstate->final_aggTransValue,
  						   funcstate->aggTranstypeByVal,
  						   &funcstate->final_aggTransValueIsNull,
  						   &funcstate->final_aggNoTransValue,
  						   funcstate->final_aggShouldFree);
  
  			if (lead_value != NULL || trail_value != NULL)
  			{
  				bool		has_tuples = hasTuplesInFrame(level_state, wstate);
  
  				if (lead_value != NULL)
  				{
  					funcstate->final_aggTransValue =
  						datumCopyWithMemManager(0, lead_value->value,
  												funcstate->aggTranstypeByVal,
  												funcstate->aggTranstypeLen,
  												&(wstate->mem_manager));
  					funcstate->final_aggTransValueIsNull =
  						lead_value->valueIsNull;
  
  					funcstate->final_aggNoTransValue = false;
  					funcstate->final_aggShouldFree = true;
  					lead_num_not_nulls = lead_value->numNotNulls;
  				}
  				else
  				{
  					funcstate->final_aggTransValue =
  						datumCopyWithMemManager(0, funcstate->aggTransValue,
  												funcstate->aggTranstypeByVal,
  												funcstate->aggTranstypeLen,
  												&(wstate->mem_manager));
  					funcstate->final_aggTransValueIsNull =
  						funcstate->aggTransValueIsNull;
  					funcstate->final_aggNoTransValue = false;
  					funcstate->final_aggShouldFree = true;
  					lead_num_not_nulls = funcstate->numNotNulls;
  				}
  
  				if (has_tuples &&
  					(lead_num_not_nulls - trail_num_not_nulls > 0))
  				{
  					funcstate->final_aggTransValue =
  						invoke_agg_trans_func(&(funcstate->invprelimfn),
  										 funcstate->invprelimfn.fn_nargs - 1,
  											  funcstate->final_aggTransValue,
  										   &funcstate->final_aggNoTransValue,
  									 &(funcstate->final_aggTransValueIsNull),
  											  funcstate->aggTranstypeByVal,
  											  funcstate->aggTranstypeLen,
  											  &fcinfo, (void *)wstate,
  											  econtext->ecxt_per_tuple_memory,
  											  &(wstate->mem_manager));
  				}
  
  				else
  				{
  					funcstate->final_aggTransValue = funcstate->aggInitValue;
  					funcstate->final_aggTransValueIsNull =
  						funcstate->aggInitValueIsNull;
  					funcstate->final_aggNoTransValue =
  						funcstate->aggInitValueIsNull;
  					funcstate->final_aggShouldFree = false;
  				}
  			}
  
  			else
  			{
  				/*
  				 * Check if the frame overlaps the current row. This is done
  				 * by checking if trail is identical to the current row when
  				 * the frame is RANGE BETWEEN <following> and <following>, or
  				 * if lead is identical to the current row when the frame is
  				 * RANGE BETWEEN <preceding> and <preceding>.
  				 *
  				 * In the case the frame overlaps, take aggTransValue, otherwise
  				 * aggInitValue, as the final value.
  				 */
  				if ((EDGE_IS_BOUND_FOLLOWING(level_state->frame->trail) &&
  				   ((level_state->is_rows && level_state->trail_rows != 0) ||
  					(!level_state->is_rows &&
  					 !exec_eq_exprstate(wstate, level_state->trail_range_eq_expr)))) ||
  					(EDGE_IS_BOUND_PRECEDING(level_state->frame->lead) &&
  					 ((level_state->is_rows && level_state->lead_rows != 0) ||
  					  (!level_state->is_rows &&
  					   !exec_eq_exprstate(wstate, level_state->lead_range_eq_expr)))))
  				{
  					funcstate->final_aggTransValue = funcstate->aggInitValue;
  					funcstate->final_aggTransValueIsNull = funcstate->aggInitValueIsNull;
  					funcstate->final_aggNoTransValue = funcstate->aggInitValueIsNull;
  					funcstate->final_aggShouldFree = false;
  				}
  				else
  				{
  					funcstate->final_aggTransValue = funcstate->aggTransValue;
  					funcstate->final_aggTransValueIsNull = funcstate->aggTransValueIsNull;
  					funcstate->final_aggNoTransValue = funcstate->aggNoTransValue;
  					funcstate->final_aggShouldFree = false;
  				}
  			}
  		}
  		else if (funcstate->isAgg && OidIsValid(funcstate->prelimfn_oid))
  		{
  			require_scanning = true;
  		}
  		else
  		{
  			/*
  			 * Here are the general cases. The frame buffer stores the input
  			 * arguments. For window aggregate functions, we scan through
  			 * values between the trailing edge and the leading edge, and add
  			 * them into the transition value. For ordinary window functions,
  			 * we pass the frame buffer to its function handler to generate
  			 * the output.
  			 */
  			if (!funcstate->isAgg)
  			{
  				int			argno;
  				FunctionCallInfoData fcinfo;
  
  				argno = initFcinfo(funcstate->wrxstate, &fcinfo, funcstate,
  								   econtext, false);
  
  				InitFunctionCallInfoData(fcinfo,
  										 &funcstate->windowfn,
  										 argno,
  										 (void *)funcstate->wrxstate,
  										 NULL);
  
  				funcstate->win_value = FunctionCallInvoke(&fcinfo);
  				funcstate->win_value_is_null = fcinfo.isnull;
  			}
  			else
  				require_scanning = true;
  		}
  	}
  
  	/*
  	 * Compute transition values for functions whose values have to be
  	 * computed by scanning through multiple entries in the frame buffer. We
  	 * will read each entry once.
  	 */
  	if (require_scanning)
  		computeTransValuesThroughScan(level_state, wstate);
  }
  
  /*
   * Initialize function state for each function in the Window node.
   */
  static void
  initWindowFuncState(WindowState * wstate, Window * node)
  {
  	int			numrefs;
  	int			refno;
  	ListCell   *lc;
  
  	if (wstate->wrxstates == NULL)
  		return;
  
  	wstate->numfuncs = 0;
  	numrefs = list_length(wstate->wrxstates);
  	Insist(numrefs > 0);
  
  	wstate->func_state = palloc0(sizeof(WindowStatePerFunctionData) * numrefs);
  
  	/* Initialize per window ref (both ordinary and aggregate derived). */
  	refno = -1;
  	foreach(lc, wstate->wrxstates)
  	{
  		WindowRefExprState *wrxstate = (WindowRefExprState *) lfirst(lc);
  		WindowRef  *winref = (WindowRef *) wrxstate->xprstate.expr;
  		WindowStatePerFunction funcstate;
  		Oid			inputTypes[FUNC_MAX_ARGS];
  		Form_pg_proc proform;
  		HeapTuple	heap_tuple;
  		ListCell   *lcarg;
  		int			i  ,
  					funcno;
  		Oid			winOwner,
  					winResType;
  		bool		isAgg,
  					isWin ,
  					isSet;
  		AclResult	aclresult;
  		cqContext  *pcqCtx;
  
  		refno++;				/* First one is 0 */
  
  		/* Register the parent window state with the window ref state node. */
  		wrxstate->refno = refno;
  		wrxstate->windowstate = wstate;
  
  		/* Look for a previous duplicate window function */
  		for (funcno = 0; funcno < refno; funcno++)
  		{
  			WindowRef  *w = wstate->func_state[funcno].wref;
  
  			if (equal(winref, w) &&
  				!contain_volatile_functions((Node *) w))
  				break;
  		}
  		if (funcno < refno)
  		{
  			/* Found a match to an existing entry, so just mark it */
  			wrxstate->funcno = funcno;
  			continue;
  		}
  
  		/*
  		 * No match so this window ref represents a new function and we need
  		 * to fill in its function state.
  		 */
  		funcstate = &wstate->func_state[funcno];
  		funcstate->wrxstate = wrxstate;
  		funcstate->wref = winref;
  
  		funcstate->serial_index = -1;
  
  		if (winref->winlevel >= wstate->numlevels)
  		{
  			funcstate->trivial_frame = true;
  			funcstate->wlevel = NULL;
  		}
  
  		else
  		{
  			WindowStatePerLevel level_state =
  			&wstate->level_state[winref->winlevel];
  
  			/*
  			 * When this level has the range frame, this frame may not always
  			 * match with the row_number function. For example, row_number()
  			 * over (order by cn), sum(qty) over (order by cn) Both of these
  			 * functions point to the level "order by cn", which is used the
  			 * default frame "range between unbounded preceding and
  			 * current_row". This is not valid for row_number. We should not
  			 * put row_number and sum in the same level. We can simply ignore
  			 * row_number in this key level.
  			 */
  			if (!level_state->is_rows &&
  				winref->winfnoid == ROW_NUMBER_OID)
  			{
  				funcstate->wlevel = NULL;
  				funcstate->trivial_frame = true;
  			}
  			else
  			{
  				funcstate->wlevel = level_state;
  				level_state->level_funcs = lappend(level_state->level_funcs,
  												   funcstate);
  			}
  		}
  
  		wrxstate->funcno = funcno;
  		wstate->numfuncs = funcno + 1;
  
  		/*
  		 * May we call the function?  Initially all window functions are
  		 * built-in, however, at this point we haven't learned whether we're
  		 * working on an ordinary window function or an aggregate.
  		 */
  		aclresult = pg_proc_aclcheck(winref->winfnoid, GetUserId(),
  									 ACL_EXECUTE);
  		if (aclresult != ACLCHECK_OK)
  			aclcheck_error(aclresult, ACL_KIND_PROC,
  						   get_func_name(winref->winfnoid));
  
  		/* Collect information about the window function's pg_proc entry. */
  		pcqCtx = caql_beginscan(
  								NULL,
  								cql("SELECT * FROM pg_proc "
  									" WHERE oid = :1 ",
  									ObjectIdGetDatum(winref->winfnoid)));
  
  		heap_tuple = caql_getnext(pcqCtx);
  
  		insist_log(HeapTupleIsValid(heap_tuple),
  				   "cache lookup failed for window function proc %u",
  				   winref->winfnoid);
  		proform = (Form_pg_proc)GETSTRUCT(heap_tuple);
  
  		isAgg = proform->proisagg;
  		isWin = proform->proiswin;
  		isSet = proform->proretset;
  		winResType = proform->prorettype;
  		winOwner = proform->proowner;
  
  		caql_endscan(pcqCtx);
  
  		Assert(isAgg != isWin);
  		Assert(!isSet);
  
  
  		/*
  		 * Get actual datatypes of the inputs.	These could be different from
  		 * the declared input types, when the function accepts ANY, ANYARRAY
  		 * or ANYELEMENT.
  		 */
  		i = 0;
  		foreach(lcarg, winref->args)
  		{
  			inputTypes[i++] = exprType((Node *) lfirst(lcarg));
  		}
  		funcstate->numargs = list_length(winref->args);
  
  		funcstate->isAgg = isAgg;
  
  
  		/* The rest depends on the type (agg-derived or ordinary). */
  		if (isAgg)
  		{
  			HeapTuple	agg_tuple;
  			Form_pg_aggregate aggform;
  
  #define NUM_AGG_FNS_TO_CHECK 5
  			int			c;
  			Oid			to_check[NUM_AGG_FNS_TO_CHECK];
  			Oid			aggtranstype;
  			Expr	   *transfnexpr,
  					   *finalfnexpr,
  					   *invtransfnexpr,
  					   *invprelimfnexpr,
  					   *prelimfnexpr;
  			Datum		textInitVal;
  			cqContext  *aggcqCtx;
  
  			Insist(winref->winlevel < wstate->numlevels);
  
  			aggcqCtx = caql_beginscan(
  									  NULL,
  									  cql("SELECT * FROM pg_aggregate "
  										  " WHERE aggfnoid = :1 ",
  										ObjectIdGetDatum(winref->winfnoid)));
  
  			agg_tuple = caql_getnext(aggcqCtx);
  
  			insist_log(HeapTupleIsValid(agg_tuple), "cache lookup failed for aggregate %u",
  					   winref->winfnoid);
  			aggform = (Form_pg_aggregate)GETSTRUCT(agg_tuple);
  
  			/*
  			 * Get the implementation functions and related information we
  			 * need to handle aggregate-derived functions.
  			 *
  			 * TODO - Omit ones we're not interested in.
  			 */
  			to_check[0] = funcstate->transfn_oid = aggform->aggtransfn;
  			to_check[1] = funcstate->finalfn_oid = aggform->aggfinalfn;
  			to_check[2] = funcstate->invtransfn_oid = aggform->agginvtransfn;
  			to_check[3] = funcstate->invprelimfn_oid = aggform->agginvprelimfn;
  			to_check[4] = funcstate->prelimfn_oid = aggform->aggprelimfn;
  
  
  			/* Check that the aggregate owner has needed permissions. */
  			for (c = 0; c < NUM_AGG_FNS_TO_CHECK; c++)
  			{
  				if (!OidIsValid(to_check[c]))
  					continue;
  				aclresult = pg_proc_aclcheck(to_check[c], winOwner, ACL_EXECUTE);
  				if (aclresult != ACLCHECK_OK)
  					aclcheck_error(aclresult, ACL_KIND_PROC,
  								   get_func_name(to_check[c]));
  			}
  
  			aggtranstype = resolve_polymorphic_transtype(aggform->aggtranstype,
  														 winref->winfnoid,
  														 inputTypes);
  
  			/*
  			 * Build expression trees using actual argument & result types.
  			 * c.f., ExecInitAgg(). We should really reduce this to a few
  			 * common functions.
  			 */
  			build_aggregate_fnexprs(inputTypes, funcstate->numargs,
  									aggtranstype, winref->restype,
  									funcstate->transfn_oid,
  									funcstate->finalfn_oid,
  									funcstate->prelimfn_oid,
  									funcstate->invtransfn_oid,
  									funcstate->invprelimfn_oid,
  									&transfnexpr, &finalfnexpr, &prelimfnexpr,
  									&invtransfnexpr, &invprelimfnexpr);
  
  			fmgr_info(funcstate->transfn_oid, &funcstate->transfn);
  			funcstate->transfn.fn_expr = (Node *) transfnexpr;
  
  			if (OidIsValid(funcstate->finalfn_oid))
  			{
  				fmgr_info(funcstate->finalfn_oid, &funcstate->finalfn);
  				funcstate->finalfn.fn_expr = (Node *) finalfnexpr;
  			}
  			if (OidIsValid(funcstate->invtransfn_oid))
  			{
  				fmgr_info(funcstate->invtransfn_oid, &funcstate->invtransfn);
  				funcstate->invtransfn.fn_expr = (Node *) invtransfnexpr;
  			}
  			if (OidIsValid(funcstate->prelimfn_oid))
  			{
  				fmgr_info(funcstate->prelimfn_oid, &funcstate->prelimfn);
  				funcstate->prelimfn.fn_expr = (Node *) prelimfnexpr;
  			}
  			if (OidIsValid(funcstate->invprelimfn_oid))
  			{
  				fmgr_info(funcstate->invprelimfn_oid, &funcstate->invprelimfn);
  				funcstate->invprelimfn.fn_expr = (Node *) invprelimfnexpr;
  			}
  
  			get_typlenbyval(winref->restype,
  							&funcstate->resulttypeLen,
  							&funcstate->resulttypeByVal);
  			get_typlenbyval(aggtranstype,
  							&funcstate->aggTranstypeLen,
  							&funcstate->aggTranstypeByVal);
  
  			/*
  			 * initval is potentially null, so don't try to access it as a
  			 * struct field. Must do it the hard way with caql_getattr
  			 */
  			textInitVal = caql_getattr(aggcqCtx,
  									   Anum_pg_aggregate_agginitval,
  									   &funcstate->aggInitValueIsNull);
  
  			if (funcstate->aggInitValueIsNull)
  				funcstate->aggInitValue = (Datum)0;
  			else
  				funcstate->aggInitValue = GetAggInitVal(textInitVal,
  														aggtranstype);
  
  			/*
  			 * If the transfn is strict and the initval is NULL, make sure
  			 * input type and transtype are the same (or at least
  			 * binary-compatible), so that it's OK to use the first input
  			 * value as the initial aggTransValue.	This should have been
  			 * checked at agg definition time, but just in case...
  			 */
  			if (funcstate->transfn.fn_strict && funcstate->aggInitValueIsNull)
  			{
  				if (funcstate->numargs < 1 ||
  					!IsBinaryCoercible(inputTypes[0], aggtranstype))
  					ereport(ERROR,
  							(errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
  							 errmsg("aggregate %u needs to have compatible "
  									"input type and transition type",
  									winref->winfnoid)));
  			}
  			caql_endscan(aggcqCtx);
  
  			wrxstate->winkind = WINKIND_AGGREGATE;
  		}
  		else if (isWin)
  		{
  			HeapTuple	win_tuple;
  			Form_pg_window winform;
  			Oid			windowfn_oid = InvalidOid;
  			Const	   *refptr;
  			ExprState  *xtrastate;
  			cqContext  *wincqCtx;
  
  			wincqCtx = caql_beginscan(
  									  NULL,
  									  cql("SELECT * FROM pg_window "
  										  " WHERE winfnoid = :1 ",
  										ObjectIdGetDatum(winref->winfnoid)));
  
  			win_tuple = caql_getnext(wincqCtx);
  
  			if (!HeapTupleIsValid(win_tuple))
  				elog(ERROR, "cache lookup failed for window function %u",
  					 winref->winfnoid);
  
  			winform = (Form_pg_window)GETSTRUCT(win_tuple);
  
  			/*
  			 * Lookup ordinary window implementation function info.
  			 */
  			switch (winref->winstage)
  			{
  				case WINSTAGE_IMMEDIATE:
  				case WINSTAGE_ROWKEY:
  					windowfn_oid = winform->winfunc;
  					break;
  				case WINSTAGE_PRELIMINARY:
  					windowfn_oid = winform->winprefunc;
  					break;
  			}
  
  			wrxstate->winkind = winform->winkind;
  
  			Assert(OidIsValid(windowfn_oid));
  
  			/*
  			 * Check that window function owner has permission to call the
  			 * implementation function.
  			 *
  			 * XXX initially all implementation functions are builtin and anyone
  			 * can call them, but good to follow form.
  			 */
  			aclresult = pg_proc_aclcheck(windowfn_oid,
  										 winOwner,
  										 ACL_EXECUTE);
  			if (aclresult != ACLCHECK_OK)
  				aclcheck_error(aclresult, ACL_KIND_PROC,
  							   get_func_name(windowfn_oid));
  
  			funcstate->windowfn_oid = windowfn_oid;
  			funcstate->allowframe = winform->winallowframe;
  
  			/* Inject extra argument. */
  			refptr = makeNode(Const);
  			refptr->consttype = INTERNALOID;
  			refptr->constlen = 4;
  			refptr->constvalue = PointerGetDatum(wrxstate);
  			refptr->constisnull = false;
  			refptr->constbyval = true;
  
  			xtrastate = ExecInitExpr((Expr *)refptr, (PlanState *) wstate);
  
  			wrxstate->args = lcons(xtrastate, wrxstate->args);
  			fmgr_info(funcstate->windowfn_oid, &funcstate->windowfn);
  
  			/*
  			 * Initialize byval and typlen for framed funcs. Of course we're
  			 * creating an entry here for the inserted refptr but we do so for
  			 * consistency.
  			 */
  			if (funcstate->allowframe || IS_LEAD_LAG(wrxstate->winkind) ||
  				IS_FIRST_LAST(wrxstate->winkind))
  			{
  				ListCell   *lc;
  				int			numargs = list_length(wrxstate->args);
  				int			argno = 0;
  
  				wrxstate->argtypbyval = palloc(sizeof(bool) * numargs);
  				wrxstate->argtyplen = palloc(sizeof(int16) * numargs);
  
  				foreach(lc, wrxstate->args)
  				{
  					if (argno == 0)
  					{
  						/* internal arg */
  						wrxstate->argtypbyval[argno] = true;
  						wrxstate->argtyplen[argno] = INTERNALOID;
  					}
  					else
  					{
  						Oid			typid = inputTypes[argno - 1];
  						int16		len;
  						bool		byval;
  
  						get_typlenbyval(typid, &len, &byval);
  						wrxstate->argtypbyval[argno] = byval;
  						wrxstate->argtyplen[argno] = len;
  					}
  					argno++;
  				}
  			}
  
  			/*
  			 * Only set the peer count if we're referencing an actual window
  			 * key level and if the peer count is required. This ensures that
  			 * subsequent iterations do not overwrite the peer count flag.
  			 */
  			if (PointerIsValid(funcstate->wlevel) && winform->winpeercount)
  			{
  				funcstate->wlevel->need_peercount = winform->winpeercount;
  				funcstate->winpeercount = winform->winpeercount;
  				wstate->need_peercount = winform->winpeercount;
  			}
  
  			funcstate->win_value = 0;
  			funcstate->win_value_is_null = true;
  
  			caql_endscan(wincqCtx);
  
  		}
  		else
  		{
  			ereport(ERROR,
  					(errcode(ERRCODE_INTERNAL_ERROR),
  				errmsg("inappropriate use of function as window function")));
  
  		}
  	}
  }
  
  /*
   * Initialize WindowStatePerLevel in the Window node.
   */
  static void
  initWindowStatePerLevel(WindowState * wstate, Window * node)
  {
  	ListCell   *lc;
  	int			level_no = 0;
  	WindowKey  *prev_key = NULL;
  	TupleDesc	desc;
  	int			col_no;
  
  	desc = ExecGetResultType(wstate->ps.lefttree);
  
  	foreach(lc, node->windowKeys)
  	{
  		WindowKey  *key = (WindowKey *) lfirst(lc);
  		WindowFrame *frame;
  		WindowStatePerLevel lvl = &wstate->level_state[level_no++];
  
  		lvl->has_delay_bound = false;
  		lvl->has_only_trans_funcs = false;
  
  		/*
  		 * Copy the sort columns for this level. If the window key contains an
  		 * empty list of sort columns, we copy the previous non-empty list of
  		 * sort columns here. This makes it easier to find keys to be stored
  		 * in the frame buffer later.
  		 */
  		lvl->numSortCols = key->numSortCols;
  		if (lvl->numSortCols == 0 && prev_key != NULL)
  			lvl->numSortCols += prev_key->numSortCols;
  		lvl->sortColIdx = (AttrNumber *)palloc0(lvl->numSortCols * sizeof(AttrNumber));
  		lvl->sortOperators = (Oid *) palloc0(lvl->numSortCols * sizeof(Oid));
  
  		if (lvl->numSortCols > key->numSortCols)
  		{
  			memcpy(lvl->sortColIdx, prev_key->sortColIdx,
  				   prev_key->numSortCols * sizeof(AttrNumber));
  			memcpy(lvl->sortOperators, prev_key->sortOperators,
  				   prev_key->numSortCols * sizeof(Oid));
  		}
  
  		memcpy(lvl->sortColIdx + (lvl->numSortCols - key->numSortCols),
  			   key->sortColIdx,
  			   key->numSortCols * sizeof(AttrNumber));
  		memcpy(lvl->sortOperators + (lvl->numSortCols - key->numSortCols),
  			   key->sortOperators,
  			   key->numSortCols * sizeof(Oid));
  
  		if (key->numSortCols > 0)
  			prev_key = key;
  
  		/* Find comparison functions */
  		lvl->eqfunctions = execTuplesMatchPrepare(desc,
  												  lvl->numSortCols,
  												  lvl->sortColIdx);
  
  		lvl->ltfunctions = get_ltfuncs(desc,
  									   lvl->numSortCols,
  									   lvl->sortColIdx);
  
  		/* Set the frame for this level */
  		lvl->is_rows = false;
  
  		if (!key->frame)
  		{
  			/*
  			 * User didn't specify a frame, so we add the default. We do not
  			 * do this in the parser because otherwise we cannot handle
  			 * situation such as the following query being turned into a view:
  			 *
  			 * SELECT sum(...) OVER (w1), RANK() OVER (w1) ... WINDOW w1 AS
  			 * (ORDER BY foo);
  			 *
  			 * When transformed to a view, we'd add the default framing clause
  			 * and then the parser would barf when it encountered a window
  			 * clause for a RANK function. So, we do it here.
  			 */
  
  			lvl->default_frame = true;
  
  			frame = (WindowFrame *) makeNode(WindowFrame);
  			frame->trail = (WindowFrameEdge *) makeNode(WindowFrameEdge);
  			frame->lead = (WindowFrameEdge *) makeNode(WindowFrameEdge);
  
  			/*
  			 * The default frame is:
  			 *
  			 * RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
  			 */
  			frame->is_rows = false;
  			frame->trail->kind = WINDOW_UNBOUND_PRECEDING;
  			frame->lead->kind = WINDOW_CURRENT_ROW;
  
  			lvl->is_rows = false;
  
  			/*
  			 * Note: we haven't set is_rows or exclude but the use of
  			 * palloc0() inside of makeNode() ensures all fields of the struct
  			 * are zeros so we're golden.
  			 */
  		}
  		else
  		{
  			frame = key->frame;
  			lvl->is_rows = key->frame->is_rows;
  		}
  
  		lvl->frame = frame;
  
  		Assert(PointerIsValid(frame->trail));
  		Assert(PointerIsValid(frame->lead));
  
  		lvl->col_types = palloc(sizeof(Oid) * lvl->numSortCols);
  		lvl->col_typlens = palloc(sizeof(int2) * lvl->numSortCols);
  		lvl->col_typbyvals = palloc(sizeof(bool) * lvl->numSortCols);
  
  		for (col_no = 0; col_no < lvl->numSortCols; col_no++)
  		{
  			AttrNumber	attnum = lvl->sortColIdx[col_no];
  			Form_pg_attribute attr = desc->attrs[attnum - 1];
  
  			lvl->col_types[col_no] = attr->atttypid;
  			lvl->col_typlens[col_no] = attr->attlen;
  			lvl->col_typbyvals[col_no] = attr->attbyval;
  		}
  	}
  }
  
  
  /* WindowState constructor.
   */
  static WindowState *
  makeWindowState(Window * window, EState *estate)
  {
  	int			numlevels;
  	WindowState *wstate = makeNode(WindowState);
  
  	wstate->ps.plan = (Plan *) window;
  	wstate->ps.state = estate;
  
  	wstate->wrxstates = NIL;
  	wstate->eqfunctions = NULL;
  	wstate->numfuncs = 0;
  	wstate->func_state = NULL;
  
  	wstate->row_index = 0;
  	wstate->level_state = NULL;
  
  	numlevels = list_length(window->windowKeys);
  	if (numlevels > 0)
  		wstate->level_state = (WindowStatePerLevel)
  			palloc0(numlevels * sizeof(WindowStatePerLevelData));
  	else
  		wstate->level_state = NULL;
  	wstate->numlevels = numlevels;
  
  	wstate->transcontext = AllocSetContextCreate(CurrentMemoryContext,
  												 "TransContext",
  												 ALLOCSET_DEFAULT_MINSIZE,
  												 ALLOCSET_DEFAULT_INITSIZE,
  												 ALLOCSET_DEFAULT_MAXSIZE);
  	wstate->cmpcontext = AllocSetContextCreate(CurrentMemoryContext,
  											   "CmpContext",
  											   ALLOCSET_DEFAULT_MINSIZE,
  											   ALLOCSET_DEFAULT_INITSIZE,
  											   ALLOCSET_DEFAULT_MAXSIZE);
  
  	/*
  	 * We allocate the buffer to be 1K initially, which should be sufficient
  	 * for most cases.
  	 */
  	wstate->serial_array = palloc0(FRAMEBUFFER_ENTRY_SIZE);
  	wstate->max_size = FRAMEBUFFER_ENTRY_SIZE;
  
  	return wstate;
  }
  
  /* [Re]initialize prior to processing a new partition.
   */
  static void
  initializePartition(WindowState * wstate)
  {
  	int			level;
  
  	/*
  	 * Reset partition primitive values. Note row_index is -1 to flag "before
  	 * first row of partition" and so that it increments to 0.
  	 */
  	wstate->row_index = -1;
  
  	/*
  	 * Reset key-level primitive values.
  	 */
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		level_state->group_index = 0;
  		level_state->prior_non_peer_count = 0;
  		level_state->peer_index = 0;
  		level_state->peer_count = 0;
  		level_state->rank = 1;
  		level_state->dense_rank = 1;
  		level_state->prior_rank = 1;
  		level_state->prior_dense_rank = 1;
  	}
  
  	/* Per-partition input buffer management reinitialzation. */
  	resetInputBuffer(wstate);
  }
  
  /*
   * trimInputBuffer -- trim the old values in the input buffer.
   */
  static void
  trimInputBuffer(WindowInputBuffer buffer)
  {
  	/* Trim all values before the current_row_reader position. */
  	NTupleStorePos pos;
  	bool		found;
  
  	Assert(buffer != NULL);
  
  	found = ntuplestore_acc_tell(buffer->current_row_reader, &pos);
  	if (!found)
  		return;
  
  	ntuplestore_trim(buffer->tuplestore, &pos);
  }
  
  /*
   * Reset an input buffer after a partition boundary has been passed. We
   * must keep the last tuple fetched -- it is the tuple which caused the
   * key break -- and put it in the new buffer.
   *
   * Of course, we could get called to initialise the buffer as well.
   */
  static void
  resetInputBuffer(WindowState * wstate)
  {
  	int			level;
  	bool		save_firsttuple = false;
  
  	if (wstate->input_buffer)
  	{
  		save_firsttuple = wstate->input_buffer->part_break;
  		if (save_firsttuple)
  		{
  #ifdef USE_ASSERT_CHECKING
  			bool		found =
  #endif
  			ntuplestore_acc_current_tupleslot(wstate->input_buffer->writer,
  											  wstate->spare);
  
  			Assert(found);
  			wstate->curslot = ExecCopySlot(wstate->curslot, wstate->spare);
  		}
  		ntuplestore_destroy_accessor(wstate->input_buffer->writer);
  		ntuplestore_destroy_accessor(wstate->input_buffer->current_row_reader);
  		ntuplestore_destroy_accessor(wstate->input_buffer->reader);
  
  		ntuplestore_reset(wstate->input_buffer->tuplestore);
  
  		wstate->input_buffer->part_break = false;
  		wstate->input_buffer->num_tuples = 0;
  
  		wstate->cur_slot_is_new = false;
  		wstate->cur_slot_part_break = false;
  		wstate->cur_slot_key_break = wstate->numlevels;
  
  		resetFrameBuffers(wstate);
  	}
  	else
  	{
  		wstate->input_buffer = (WindowInputBuffer) palloc0(sizeof(WindowInputBufferData));
  		wstate->input_buffer->tuplestore =
  			ntuplestore_create((PlanStateOperatorMemKB((PlanState *) wstate) * 1024L) / 2);		/* correct? */
  		createFrameBuffers(wstate);
  	}
  
  	/*
  	 * Reset transition values.  This has to be done even if we are creating
  	 * new buffer because the trans values might remain in case of rescanning.
  	 */
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		resetTransValues(level_state, wstate);
  	}
  
  	wstate->input_buffer->writer =
  		ntuplestore_create_accessor(wstate->input_buffer->tuplestore, true);
  	wstate->input_buffer->current_row_reader =
  		ntuplestore_create_accessor(wstate->input_buffer->tuplestore, false);
  	wstate->input_buffer->reader =
  		ntuplestore_create_accessor(wstate->input_buffer->tuplestore, false);
  
  	if (save_firsttuple)
  	{
  		ntuplestore_acc_put_tupleslot(wstate->input_buffer->writer,
  									  wstate->curslot);
  		wstate->input_buffer->num_tuples++;
  
  		ntuplestore_acc_seek_first(wstate->input_buffer->current_row_reader);
  		ntuplestore_acc_seek_first(wstate->input_buffer->reader);
  
  		wstate->cur_slot_is_new = true;
  		wstate->row_index++;
  	}
  }
  
  static void
  freeInputBuffer(WindowState * wstate)
  {
  	ntuplestore_destroy_accessor(wstate->input_buffer->writer);
  	ntuplestore_destroy_accessor(wstate->input_buffer->current_row_reader);
  	ntuplestore_destroy(wstate->input_buffer->tuplestore);
  
  	pfree(wstate->input_buffer);
  	wstate->input_buffer = NULL;
  }
  
  /*
   * Advance values related to computing peer count.
   */
  static void
  advancePeerCount(WindowStatePerLevel level_state,
  				 int curr_level, int min_level)
  {
  	if (curr_level >= min_level && min_level != -1)
  	{
  		level_state->prior_non_peer_count += level_state->peer_count + 1;
  		level_state->peer_index = 0;
  		level_state->peer_count = 0;
  	}
  }
  
  
  /* Advance the "primitive" values that drive the calculation of window
   * functions for the window ordering of the specified level.
   *
   * On entry these values are as set by partition initialization or as
   * left by the	previous call to this function. Per-partition state
   * has not yet been advanced by function nextRow(), our caller.
   *
   * The "primitive" values managed here are read-only outside the window
   * framework implemented by this function, initializePartition(), and
   * nextRow().
   *
   * Note that the content of different key levels is independent.  The
   * function nextRow() is responsible for calling this function as needed
   * for related key levels.
   */
  
  static void
  advanceKeyLevelState(WindowState * wstate, int min_level)
  {
  	WindowStatePerLevel level_state;
  	int64		cur_row_index;
  	int			i;
  
  	Assert(-1 <= min_level && min_level < wstate->numlevels);
  
  	cur_row_index = wstate->row_index;
  
  	for (i = 0; i < wstate->numlevels; i++)
  	{
  		level_state = &wstate->level_state[i];
  
  		if (i >= min_level && min_level != -1)
  		{
  			/*
  			 * The level data is still set for the previous row. Remember the
  			 * rank and dense_rank values.
  			 */
  			level_state->prior_rank = level_state->rank;
  			level_state->prior_dense_rank = level_state->dense_rank;
  
  			/* Advance level data for first peer in next group. */
  			level_state->group_index++;
  			level_state->rank = 1 + cur_row_index;
  			level_state->dense_rank = 1 + level_state->group_index;
  		}
  	}
  }
  
  static int
  initFcinfo(WindowRefExprState * wrxstate, FunctionCallInfoData *fcinfo,
  		   WindowStatePerFunction funcstate, ExprContext *econtext,
  		   bool check_nulls)
  {
  	ListCell   *lcarg;
  	int			argno;
  	bool		has_nulls = false;
  	MemoryContext oldctx;
  
  	/* If this is an agg, the first arg will be the trans value */
  	if (funcstate->isAgg)
  		argno = 1;
  	else
  		argno = 0;
  
  	/* Switch memory context just once for all args */
  	oldctx = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
  
  	foreach(lcarg, wrxstate->args)
  	{
  		ExprState  *argstate = (ExprState *)lfirst(lcarg);
  
  		fcinfo->arg[argno] = ExecEvalExpr(argstate, econtext,
  										  fcinfo->argnull + argno, NULL);
  		if (fcinfo->argnull[argno])
  			has_nulls = true;
  		argno++;
  	}
  	MemoryContextSwitchTo(oldctx);
  
  	if (check_nulls && !has_nulls)
  		funcstate->numNotNulls++;
  
  	return argno;
  }
  
  static void
  add_tuple_to_trans(WindowStatePerFunction funcstate, WindowState * wstate,
  				   ExprContext *econtext, bool check_nulls)
  {
  	int			argno;
  	FunctionCallInfoData fcinfo;
  
  	/* Evaluate function arguments, save in fcinfo. */
  	argno = initFcinfo(funcstate->wrxstate, &fcinfo, funcstate, econtext, check_nulls);
  
  	if (funcstate->isAgg)
  	{
  		funcstate->aggTransValue =
  			invoke_agg_trans_func(&(funcstate->transfn),
  								  funcstate->numargs,
  								  funcstate->aggTransValue,
  								  &(funcstate->aggNoTransValue),
  								  &(funcstate->aggTransValueIsNull),
  								  funcstate->aggTranstypeByVal,
  								  funcstate->aggTranstypeLen,
  								  &fcinfo, (void *)wstate,
  								  econtext->ecxt_per_tuple_memory,
  								  &(wstate->mem_manager));
  	}
  	else	/* ordinary window function */
  	{
  		InitFunctionCallInfoData(fcinfo,
  								 &funcstate->windowfn,
  								 argno,
  								 (void *)funcstate->wrxstate,
  								 NULL);
  		if (IS_LEAD_LAG(funcstate->wrxstate->winkind) ||
  			IS_LAST_VALUE(funcstate->wrxstate->winkind) ||
  			(IS_FIRST_VALUE(funcstate->wrxstate->winkind) &&
  			 funcstate->aggNoTransValue))
  		{
  			int			argno = 1;
  
  			funcstate->aggTransValue =
  				datumCopyWithMemManager(0, fcinfo.arg[argno],
  									 funcstate->wrxstate->argtypbyval[argno],
  										funcstate->wrxstate->argtyplen[argno],
  										&(wstate->mem_manager));
  			funcstate->aggTransValueIsNull = fcinfo.argnull[argno];
  			funcstate->aggNoTransValue = false;
  		}
  
  		else
  		{
  			MemoryContext oldctx;
  
  			oldctx = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
  			funcstate->win_value = FunctionCallInvoke(&fcinfo);
  			funcstate->win_value_is_null = fcinfo.isnull;
  			MemoryContextSwitchTo(oldctx);
  		}
  	}
  
  }
  
  static void
  invokeTrivialFuncs(WindowState * wstate, bool *found)
  {
  	int			funcno;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  
  	for (funcno = 0; funcno < wstate->numfuncs; funcno++)
  	{
  		WindowStatePerFunction funcstate = &wstate->func_state[funcno];
  		WindowStatePerLevel level_state = funcstate->wlevel;
  
  		if (funcstate->trivial_frame)
  		{
  			econtext->ecxt_scantuple = wstate->curslot;
  			add_tuple_to_trans(funcstate, wstate, econtext, false);
  
  			if (funcstate->isAgg)
  			{
  				funcstate->final_aggTransValue = funcstate->aggTransValue;
  				funcstate->final_aggTransValueIsNull = funcstate->aggTransValueIsNull;
  				funcstate->final_aggNoTransValue = funcstate->aggNoTransValue;
  				funcstate->final_aggShouldFree = false;
  			}
  		}
  		else if (funcstate->cumul_frame &&
  				 !HAS_ONLY_TRANS_FUNC(funcstate))
  		{
  			WindowFrameBuffer frame_buffer;
  
  			Assert(level_state != NULL);
  			frame_buffer = level_state->frame_buffer;
  			Assert(frame_buffer != NULL);
  
  			freeTransValue(&funcstate->final_aggTransValue,
  						   funcstate->aggTranstypeByVal,
  						   &funcstate->final_aggTransValueIsNull,
  						   &funcstate->final_aggNoTransValue,
  						   funcstate->final_aggShouldFree);
  
  			/*
  			 * When the current_row_reader is set, we retrieve the
  			 * intermediate accumulated aggregate value from the buffer.
  			 */
  			if (ntuplestore_acc_tell(frame_buffer->current_row_reader, NULL))
  			{
  				FrameBufferEntry *entry = level_state->curr_entry_buf;
  				bool		has_entry = false;
  				WindowValue *value = NULL;
  
  				Assert(funcstate->serial_index != -1);
  
  				has_entry = getCurrentValue(frame_buffer->current_row_reader,
  											level_state, entry);
  				Assert(has_entry);
  				value = (WindowValue *) list_nth(entry->func_values,
  												 funcstate->serial_index);
  				Assert(value != NULL);
  				funcstate->final_aggTransValue =
  					datumCopyWithMemManager(0, value->value,
  											funcstate->aggTranstypeByVal,
  											funcstate->aggTranstypeLen,
  											&(wstate->mem_manager));
  				funcstate->final_aggTransValueIsNull = value->valueIsNull;
  				funcstate->final_aggNoTransValue = false;
  				funcstate->final_aggShouldFree = true;
  			}
  
  			/*
  			 * Otherwise, the value for the current_rows is stored in
  			 * aggTransValue.
  			 */
  			else
  			{
  				funcstate->final_aggTransValue = funcstate->aggTransValue;
  				funcstate->final_aggTransValueIsNull = funcstate->aggTransValueIsNull;
  				funcstate->final_aggNoTransValue = funcstate->aggNoTransValue;
  				funcstate->final_aggShouldFree = false;
  			}
  		}
  
  		else
  			*found = true;
  	}
  }
  
  /*
   * cmp_deformed_tuple -- compare two given arrays of Datums.
   *
   * If is_equal is true, this function returns true when these two Datums are
   * equal.
   *
   * If is_equal is false, this function return true if Datum a is ordered
   * before Datum b. The ASC/DESC ordering is defined by 'asc_cols'.
   */
  static bool
  cmp_deformed_tuple(Datum *a, bool *a_nulls, Datum *b, bool *b_nulls,
  				   bool *asc_cols, int ncols, FmgrInfo *ltfuncs,
  				 FmgrInfo *eqfuncs, MemoryContext evalContext, bool is_equal)
  {
  	MemoryContext oldContext;
  	bool		result;
  	int			i	  ,
  				j;
  	FmgrInfo   *funcs;
  
  	/* Reset and switch into the temp context. */
  	MemoryContextReset(evalContext);
  	oldContext = MemoryContextSwitchTo(evalContext);
  
  	if (is_equal)
  	{
  		funcs = eqfuncs;
  		result = true;
  	}
  	else
  	{
  		funcs = ltfuncs;
  		result = true;
  	}
  
  	/*
  	 * For equality testing, we can will most likely find an inequality most
  	 * quickly by working from right to left. But, we might not be doing
  	 * equality testing in which case we must move from left to right.
  	 */
  	if (is_equal)
  		i = ncols - 1;
  	else
  		i = 0;
  
  	for (j = 0; j < ncols; j++, is_equal ? i-- : i++)
  	{
  		Datum		attr1,
  					attr2;
  		bool		isNull1,
  					isNull2;
  		bool		res;
  
  		attr1 = a[i];
  		isNull1 = a_nulls[i];
  
  		attr2 = b[i];
  		isNull2 = b_nulls[i];
  
  		if (isNull1 != isNull2)
  		{
  			if (is_equal)
  			{
  				result = false; /* one null and one not; they aren't equal */
  				break;
  			}
  			else
  			{
  				/*
  				 * Currently, we assume NULLS LAST. In the future, when we add
  				 * support NULLS first to the window order-by clause, we
  				 * should take care of NULLS FIRST as well.
  				 */
  				if (isNull1)
  					result = false;
  				else
  					result = true;
  				break;
  			}
  		}
  
  		else if (isNull1)
  			continue;			/* both are null, treat as equal */
  
  		/*
  		 * As we're not doing equality tests, be aware of user specified
  		 * ordering. If the user specified DESC, swap the order of arguments
  		 * for this test.
  		 */
  		if (!is_equal)
  		{
  			if (!asc_cols[i])
  			{
  				Datum		tmp;
  
  				tmp = attr1;
  				attr1 = attr2;
  				attr2 = tmp;
  			}
  		}
  
  		/* Apply the type-specific function */
  		res = DatumGetBool(FunctionCall2(&funcs[i], attr1, attr2));
  
  		if (is_equal && !res)
  		{
  			/* we wanted equality but didn't get it */
  			result = false;
  			break;
  		}
  		else if (!is_equal)
  		{
  			if (!res)
  			{
  				/*
  				 * We wanted it to be less, but it wasn't. Continue only if
  				 * they're equal.
  				 */
  				res = DatumGetBool(FunctionCall2(&eqfuncs[i], attr1, attr2));
  				if (res)
  				{
  					/*
  					 * Because we found the previous key equal, set the return
  					 * value to false.
  					 */
  					result = false;
  					continue;
  				}
  				else
  					/* LHS is greater than RHS */
  					result = false;
  			}
  			else
  			{
  				/* because all keys are sorted, we return straight away */
  				result = true;
  			}
  			break;
  		}
  	}
  
  	MemoryContextSwitchTo(oldContext);
  
  	return result;
  
  }
  
  /*
   * Like heap_deform_tuple(), but we only extract those fields listed in
   * attnums.
   *
   * values and nulls should be allocated memory by the caller.
   */
  static void
  deform_window_tuple(TupleTableSlot * slot, int nattrs, AttrNumber *attnums,
  					Datum *values, bool *nulls, WindowState * wstate)
  {
  	int			i;
  	MemoryContext oldctx = MemoryContextSwitchTo(wstate->transcontext);
  
  	for (i = 0; i < nattrs; i++)
  	{
  		AttrNumber	attnum = attnums[i];
  		bool		isnull;
  		Form_pg_attribute attr = slot->tts_tupleDescriptor->attrs[attnum - 1];
  		Datum		d = slot_getattr(slot, attnum, &isnull);
  
  		/*
  		 * We must copy this tuple because we may need to keep it around a
  		 * long time. Caller is responsible for freeing these values.
  		 */
  		values[i] = datumCopyWithMemManager(0, d,
  											attr->attbyval, attr->attlen,
  											&(wstate->mem_manager));
  		nulls[i] = isnull;
  	}
  
  	MemoryContextSwitchTo(oldctx);
  }
  
  /*
   * Adjust the trailing and leading edge for ROWS-based framing.
   */
  static void
  adjustEdgesForRows(WindowFrameBuffer buffer,
  				   WindowState * wstate)
  {
  	WindowStatePerLevel level_state = buffer->level_state;
  
  	if (!(level_state->empty_frame &&
  		  EDGE_IS_DELAYED(level_state->frame->trail)))
  	{
  		if (EDGE_IS_BOUND(level_state->frame->trail) ||
  			EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->trail, false))
  		{
  			if (level_state->num_trail_rows <= level_state->trail_rows - 1)
  			{
  				bool		found = false;
  				bool		was_valid = true;
  
  				if (ntuplestore_acc_tell(level_state->trail_reader, NULL))
  					found = ntuplestore_acc_advance(level_state->trail_reader, 1);
  				else
  				{
  					was_valid = false;
  
  					if (!EDGE_IS_BOUND_FOLLOWING(level_state->frame->trail) ||
  						level_state->trail_rows == 0)
  						found = ntuplestore_acc_seek_first(level_state->trail_reader);
  				}
  
  				/* Only increment num_trail_rows when we found the next tuple. */
  				if (found ||
  					(was_valid && level_state->agg_filled &&
  					 EDGE_IS_BOUND_FOLLOWING(level_state->frame->trail)))
  					level_state->num_trail_rows++;
  			}
  		}
  	}
  
  	if (!(level_state->empty_frame &&
  		  EDGE_IS_DELAYED(level_state->frame->lead)))
  	{
  		if (EDGE_IS_BOUND_PRECEDING(level_state->frame->lead) &&
  			level_state->num_lead_rows == level_state->lead_rows)
  		{
  			ntuplestore_acc_seek_first(level_state->lead_reader);
  		}
  
  		else if ((!EDGE_IS_BOUND(level_state->frame->lead) &&
  				  !EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->lead, true)) ||
  				 level_state->num_lead_rows < level_state->lead_rows)
  		{
  			bool		prev_set = ntuplestore_acc_tell(level_state->lead_reader, NULL);
  			bool		found = ntuplestore_acc_advance(level_state->lead_reader, 1);
  
  			if (found || (prev_set && level_state->agg_filled))
  				level_state->num_lead_rows++;
  		}
  	}
  }
  
  /*
   * forwardEdgeForRange -- move the edge forward from its previous
   * position in the frame buffer until it points to the key value
   * whose next value is greater than the current edge value for the
   * leading edge, and whose next value is greater than or equal to
   * the current edge value for the trailing edge.
   */
  static void
  forwardEdgeForRange(WindowStatePerLevel level_state,
  					WindowState * wstate,
  					WindowFrameEdge * edge,
  					ExprState *edge_expr,
  					ExprState *edge_range_expr,
  					NTupleStoreAccessor * edge_reader,
  					bool is_lead_edge)
  {
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  	bool		has_entry = false;
  	FrameBufferEntry *entry = level_state->curr_entry_buf;
  	Datum		new_edge_value = 0;
  	bool		new_edge_value_isnull = true;
  	MemoryContext oldctx;
  
  	bool		is_less = true;
  	bool		is_equal = true;
  	bool		lastrow_edge = false;
  
  	Assert(EDGE_IS_BOUND(edge));
  
  	if (is_lead_edge)
  		level_state->lead_ready = false;
  
  	oldctx = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
  
  	/*
  	 * Compute the new edge value. Note that this value could be NULL.
  	 */
  	new_edge_value = ExecEvalExpr(edge_range_expr, econtext,
  								  &(new_edge_value_isnull), NULL);
  
  	MemoryContextSwitchTo(oldctx);
  
  	if (!ntuplestore_acc_tell(edge_reader, NULL))
  		ntuplestore_acc_seek_first(edge_reader);
  
  	while ((is_equal && is_lead_edge) || is_less)
  	{
  		has_entry = getCurrentValue(edge_reader, level_state, entry);
  
  		if (!has_entry)
  			break;
  
  		is_less = cmp_deformed_tuple(entry->keys, entry->nulls,
  									 &new_edge_value,
  									 &new_edge_value_isnull,
  									 level_state->col_sort_asc,
  									 level_state->numSortCols,
  									 level_state->ltfunctions,
  									 level_state->eqfunctions,
  									 wstate->cmpcontext,
  									 false);
  
  		if (!is_less)
  		{
  			is_equal =
  				cmp_deformed_tuple(entry->keys, entry->nulls,
  								   &new_edge_value,
  								   &new_edge_value_isnull,
  								   level_state->col_sort_asc,
  								   level_state->numSortCols,
  								   level_state->ltfunctions,
  								   level_state->eqfunctions,
  								   wstate->cmpcontext,
  								   true);
  		}
  
  		if ((is_equal && is_lead_edge) || is_less)
  			ntuplestore_acc_advance(edge_reader, 1);
  	}
  
  	if (has_entry)
  	{
  		ntuplestore_acc_advance(edge_reader, -1);
  		if (is_lead_edge)
  			level_state->lead_ready = true;
  	}
  	else
  	{
  		lastrow_edge = checkLastRowForEdge(level_state, wstate,
  										   edge, edge_reader,
  										   new_edge_value,
  										   new_edge_value_isnull,
  										   is_lead_edge);
  		if (!lastrow_edge || !is_lead_edge)
  			ntuplestore_acc_seek_last(edge_reader);
  	}
  }
  
  /*
   * adjustEdgesForRange -- adjust both the trailing and leading edge
   * to the given frame buffer for its level state.
   *
   * This function advances the leading reader until it rests on the position
   * at the frame buffer after which all key values are greater than
   * the leading edge value. The trailing reader rests on the position
   * at the frame buffer which key value is the greatest value in the
   * buffer that is less than the trailing edge value.
   */
  static void
  adjustEdgesForRange(WindowFrameBuffer buffer,
  					WindowState * wstate)
  {
  	WindowStatePerLevel level_state = buffer->level_state;
  
  	if (!(level_state->empty_frame &&
  		  EDGE_IS_DELAYED(level_state->frame->trail)))
  	{
  		if (EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->trail, false))
  		{
  			NTupleStorePos pos;
  
  			if (ntuplestore_acc_tell(buffer->current_row_reader, &pos))
  			{
  				ntuplestore_acc_seek(level_state->trail_reader, &pos);
  				ntuplestore_acc_advance(level_state->trail_reader, -1);
  			}
  			else
  			{
  				ntuplestore_acc_seek_last(level_state->trail_reader);
  			}
  		}
  
  		else if (EDGE_IS_BOUND(level_state->frame->trail))
  		{
  			forwardEdgeForRange(level_state, wstate,
  								level_state->frame->trail,
  								level_state->trail_expr,
  								level_state->trail_range_expr,
  								level_state->trail_reader,
  								false);
  		}
  	}
  
  	if (!(level_state->empty_frame &&
  		  EDGE_IS_DELAYED(level_state->frame->lead)))
  	{
  		if (EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->lead, true))
  		{
  			NTupleStorePos pos;
  
  			if (ntuplestore_acc_tell(buffer->current_row_reader, &pos))
  			{
  				ntuplestore_acc_seek(level_state->lead_reader, &pos);
  				level_state->lead_ready = true;
  			}
  			else
  			{
  				ntuplestore_acc_advance(level_state->lead_reader, 1);
  				level_state->lead_ready = false;
  			}
  		}
  
  		else if (EDGE_IS_BOUND(level_state->frame->lead))
  		{
  			forwardEdgeForRange(level_state, wstate,
  								level_state->frame->lead,
  								level_state->lead_expr,
  								level_state->lead_range_expr,
  								level_state->lead_reader,
  								true);
  		}
  
  		else
  		{
  			ntuplestore_acc_advance(level_state->lead_reader, 1);
  			level_state->lead_ready = false;
  		}
  	}
  }
  
  /*
   * Adjust the trailing and leading edge to the frame buffer for
   * its list of key levels.
   */
  static void
  adjustEdges(WindowFrameBuffer buffer,
  			WindowState * wstate)
  {
  	WindowStatePerLevel level_state = buffer->level_state;
  
  	if (level_state->is_rows)
  		adjustEdgesForRows(buffer, wstate);
  	else
  		adjustEdgesForRange(buffer, wstate);
  }
  
  /*
   * invokeNonTrivialFuncs -- compute transition values for
   * non-trivial window functions.
   */
  static void
  invokeNonTrivialFuncs(WindowState * wstate)
  {
  	int			level;
  
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		computeFrameValue(level_state,
  						  wstate,
  						  level_state->trail_reader,
  						  level_state->lead_reader);
  	}
  }
  
  static void
  finaliseFuncs(WindowState * wstate)
  {
  	int			funcno;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  
  	MemoryContext oldctx = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
  
  	/*
  	 * Put the current transition value into the econtext
  	 */
  	for (funcno = 0; funcno < wstate->numfuncs; funcno++)
  	{
  		WindowStatePerFunction funcstate = &wstate->func_state[funcno];
  		FunctionCallInfoData fcinfo;
  
  		/* Evaluate function arguments, save in fcinfo. */
  
  		if (funcstate->isAgg)
  		{
  			Datum	   *vals = econtext->ecxt_aggvalues;
  			bool	   *isnulls = econtext->ecxt_aggnulls;
  
  			if (OidIsValid(funcstate->finalfn_oid))
  			{
  				InitFunctionCallInfoData(fcinfo, &(funcstate->finalfn), 1,
  										 (void *)wstate, NULL);
  				fcinfo.arg[0] = funcstate->final_aggTransValue;
  				fcinfo.argnull[0] = funcstate->final_aggTransValueIsNull;
  				if (fcinfo.flinfo->fn_strict && funcstate->final_aggTransValueIsNull)
  				{
  					/* don't call a strict function with NULL inputs */
  					vals[funcno] = (Datum)0;
  					isnulls[funcno] = true;
  				}
  				else
  				{
  					vals[funcno] = FunctionCallInvoke(&fcinfo);
  					isnulls[funcno] = fcinfo.isnull;
  				}
  			}
  			else
  			{
  				vals[funcno] = funcstate->final_aggTransValue;
  				isnulls[funcno] = funcstate->final_aggTransValueIsNull;
  			}
  		}
  		else	/* ordinary window function */
  		{
  			Datum	   *vals = econtext->ecxt_aggvalues;
  			bool	   *isnulls = econtext->ecxt_aggnulls;
  
  			vals[funcno] = funcstate->win_value;
  			isnulls[funcno] = funcstate->win_value_is_null;
  		}
  	}
  
  	MemoryContextSwitchTo(oldctx);
  }
  
  /*
   * We must evaluate functions in a 'frame aware' way.
   *
   * UNBOUNDED PRECEDING frame edges are managed by passing the transition value
   * from the last call to the function. In this way, we trivially evaluate
   * all rows in the frame which have already passed. It gets more complex
   * for RANGE UNBOUNDED PRECEDING frames because all peers must emit the same
   * value -- so, we remember this in our state.
   *
   * Bounded PRECEDING and FOLLOWING edges are harder. We have two cases.
   *
   * With ROWS bounded frames, we must detect our position from the current
   * rows. If we are earlier on in the tuple count than the frame edge, those
   * tuples must be removed from the the transition value using the inversion
   * method for the function. If one doesn't exist, we do something even more
   * complex. See below. Once in the frame, we only care about the leading
   * frame edge (it could be the current row or some rows from it). So, we must
   * be careful about that.
   *
   * With RANGE bounded frames, we determine the edge with the equality
   * function(s) of the key level. Before and after the frame edges, we do the
   * same as for ROWS frames but remember, we must emit the same value for
   * all peers.
   *
   * If inversion functions do exist for a function, we must evaluate the frame
   * every single time it moves.
   */
  static void
  invokeWindowFuncs(WindowState * wstate)
  {
  	bool		found_complex_func = false;
  
  	invokeTrivialFuncs(wstate, &found_complex_func);
  
  	/*
  	 * If there are no functions/frames which need non-trivial window
  	 * management, we're done.
  	 */
  	if (found_complex_func)
  		invokeNonTrivialFuncs(wstate);
  
  	finaliseFuncs(wstate);
  }
  
  /*
   * range_is_negative -- return true if the range expression
   * is negative.
   */
  static bool
  range_is_negative(ExprState *edge_expr,
  				  Datum value)
  {
  	/*
  	 * XXX Most likely not good enough to just compare the value with 0 here.
  	 */
  	if (value < 0)
  		return true;
  
  	return false;
  }
  
  /*
   * get_delay_edge -- evaluate the expression from the current row,
   * and return the value for the given DELAY_BOUND edge.
   */
  static Datum
  get_delay_edge(WindowFrameEdge * edge,
  			   ExprState *edge_expr,
  			   bool is_rows,
  			   WindowState * wstate)
  {
  	Datum		edge_param = 0;
  	bool		isnull = true;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  	MemoryContext oldctx;
  
  	Assert(EDGE_IS_DELAYED(edge));
  
  	econtext->ecxt_scantuple = wstate->curslot;
  	if (TupIsNull(wstate->curslot))
  		ereport(ERROR,
  				(errcode(ERROR_INVALID_WINDOW_FRAME_PARAMETER),
  				 errmsg("%s parameter cannot be NULL",
  						(is_rows ? "ROWS" : "RANGE"))));
  
  	oldctx = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
  
  	edge_param = ExecEvalExpr(edge_expr,
  							  econtext,
  							  &isnull,
  							  NULL);
  	MemoryContextSwitchTo(oldctx);
  
  	if (isnull)
  		ereport(ERROR,
  				(errcode(ERROR_INVALID_WINDOW_FRAME_PARAMETER),
  				 errmsg("%s parameter cannot be NULL",
  						(is_rows ? "ROWS" : "RANGE"))));
  
  	if (is_rows && edge_param < 0)
  		ereport(ERROR,
  				(errcode(ERROR_INVALID_WINDOW_FRAME_PARAMETER),
  				 errmsg("%s parameter cannot be negative",
  						(is_rows ? "ROWS" : "RANGE"))));
  
  	/* Check if the range expression is negative. */
  	if (!is_rows)
  		if (range_is_negative(edge_expr, edge_param))
  			ereport(ERROR,
  					(errcode(ERROR_INVALID_WINDOW_FRAME_PARAMETER),
  					 errmsg("%s parameter cannot be negative",
  							(is_rows ? "ROWS" : "RANGE"))));
  
  	if (is_rows && EDGE_IS_BOUND_PRECEDING(edge))
  		edge_param = 0 - edge_param;
  
  	return edge_param;
  }
  
  /*
   * checkLastRowForEdge -- check if the last row in the current partition
   * in the input buffer is within frame edges of the current_row.
   *
   * This function is only used for RANGE-frames.
   *
   * This function returns true if the last row is within frame edges
   * of the current_row.
   */
  static bool
  checkLastRowForEdge(WindowStatePerLevel level_state,
  					WindowState * wstate,
  					WindowFrameEdge * edge,
  					NTupleStoreAccessor * edge_reader,
  					Datum new_edge_value,
  					bool new_edge_value_isnull,
  					bool is_lead_edge)
  {
  	FrameBufferEntry *entry = level_state->curr_entry_buf;
  	bool		is_less = true;
  	bool		is_equal = true;
  
  	Assert(!level_state->is_rows);
  
  	/* Check if the last row is the edge. */
  	if (wstate->input_buffer->part_break)
  	{
  		/*
  		 * If the last tuple breaks the partition key, the tuple to check is
  		 * the last second tuple in the input buffer.
  		 */
  		bool		found = ntuplestore_acc_advance(wstate->input_buffer->writer, -1);
  
  		Assert(found);
  		ntuplestore_acc_current_tupleslot(wstate->input_buffer->writer,
  										  wstate->spare);
  
  		/* Put the writer pointer back to its original place. */
  		found = ntuplestore_acc_advance(wstate->input_buffer->writer, 1);
  		Assert(found);
  	}
  
  	else
  	{
  		ntuplestore_acc_current_tupleslot(wstate->input_buffer->writer,
  										  wstate->spare);
  	}
  
  	deform_window_tuple(wstate->spare,
  						level_state->numSortCols, level_state->sortColIdx,
  						entry->keys, entry->nulls,
  						wstate);
  
  	is_less = cmp_deformed_tuple(entry->keys, entry->nulls,
  								 &new_edge_value, &new_edge_value_isnull,
  								 level_state->col_sort_asc,
  								 level_state->numSortCols,
  								 level_state->ltfunctions,
  								 level_state->eqfunctions,
  								 wstate->cmpcontext,
  								 false);
  
  	if ((EDGE_IS_BOUND_FOLLOWING(edge) && !is_less) ||
  		(EDGE_IS_BOUND_PRECEDING(edge) && is_less))
  	{
  		is_equal =
  			cmp_deformed_tuple(entry->keys, entry->nulls,
  							   &new_edge_value, &new_edge_value_isnull,
  							   level_state->col_sort_asc,
  							   level_state->numSortCols,
  							   level_state->ltfunctions,
  							   level_state->eqfunctions,
  							   wstate->cmpcontext,
  							   true);
  	}
  	else
  		is_equal = false;
  
  	if (is_equal)
  		return true;
  
  	if (is_lead_edge)
  	{
  		if ((EDGE_IS_BOUND_FOLLOWING(edge) && is_less) ||
  			(EDGE_IS_BOUND_PRECEDING(edge) && is_less))
  		{
  			return true;
  		}
  	}
  	else
  	{
  		if ((EDGE_IS_BOUND_FOLLOWING(edge) && !is_less) ||
  			(EDGE_IS_BOUND_PRECEDING(edge) && !is_less))
  			return true;
  	}
  
  	return false;
  }
  
  
  /*
   * advanceEdgeForRange -- advance forward/backward a given RANGE-frame
   * edge from its previous position to its new position in the frame
   * buffer.
   */
  static void
  advanceEdgeForRange(WindowStatePerLevel level_state,
  					WindowState * wstate,
  					WindowFrameEdge * edge,
  					ExprState *edge_expr,
  					ExprState *edge_range_expr,
  					NTupleStoreAccessor * edge_reader,
  					bool is_lead_edge)
  {
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  	bool		has_entry = false;
  	FrameBufferEntry *entry = level_state->curr_entry_buf;
  	Datum		new_edge_value = 0;
  	bool		new_edge_value_isnull = true;
  
  	bool		is_less = false;
  	bool		is_equal = false;
  	bool		lastrow_edge = false;
  	MemoryContext oldctx;
  
  	if (is_lead_edge)
  		level_state->lead_ready = false;
  
  	oldctx = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
  
  	/*
  	 * Compute the new edge value. Note that this value can be NULL.
  	 */
  	new_edge_value = ExecEvalExpr(edge_range_expr, econtext,
  								  &(new_edge_value_isnull), NULL);
  
  	MemoryContextSwitchTo(oldctx);
  
  	if (!ntuplestore_acc_tell(edge_reader, NULL))
  	{
  		lastrow_edge = checkLastRowForEdge(level_state, wstate,
  										   edge, edge_reader,
  										   new_edge_value,
  										   new_edge_value_isnull,
  										   is_lead_edge);
  
  		if ((is_lead_edge || lastrow_edge) && EDGE_IS_BOUND_PRECEDING(edge))
  			ntuplestore_acc_seek_last(edge_reader);
  	}
  
  	if (!EDGE_EQ_CURRENT_ROW(level_state, wstate, edge, is_lead_edge))
  	{
  		bool		prev_is_equal = false;
  
  		do
  		{
  			has_entry = getCurrentValue(edge_reader, level_state, entry);
  
  			if (!has_entry)
  				break;
  
  			is_less = cmp_deformed_tuple(entry->keys, entry->nulls,
  										 &new_edge_value,
  										 &new_edge_value_isnull,
  										 level_state->col_sort_asc,
  										 level_state->numSortCols,
  										 level_state->ltfunctions,
  										 level_state->eqfunctions,
  										 wstate->cmpcontext,
  										 false);
  			prev_is_equal = is_equal;
  			is_equal =
  				cmp_deformed_tuple(entry->keys, entry->nulls,
  								   &new_edge_value,
  								   &new_edge_value_isnull,
  								   level_state->col_sort_asc,
  								   level_state->numSortCols,
  								   level_state->ltfunctions,
  								   level_state->eqfunctions,
  								   wstate->cmpcontext,
  								   true);
  
  			if ((EDGE_IS_BOUND_FOLLOWING(edge) && (is_less || is_equal)) ||
  				(EDGE_IS_BOUND_PRECEDING(edge) && (!is_less || is_equal)))
  			{
  				if (EDGE_IS_BOUND_FOLLOWING(edge))
  					ntuplestore_acc_advance(edge_reader, 1);
  				else
  					ntuplestore_acc_advance(edge_reader, -1);
  			}
  
  		} while ((EDGE_IS_BOUND_FOLLOWING(edge) && (is_less || is_equal)) ||
  				 (EDGE_IS_BOUND_PRECEDING(edge) && (!is_less || is_equal)));
  
  		if (has_entry)
  		{
  			if (EDGE_IS_BOUND_FOLLOWING(edge))
  			{
  				/*
  				 * When the edge is following, edge_reader points to the
  				 * position whose value is after the new edge value. We need
  				 * to back off one position if the value in its previous
  				 * position is equal to the new edge value, or this is the
  				 * leading edge.
  				 */
  				if (prev_is_equal || is_lead_edge)
  					ntuplestore_acc_advance(edge_reader, -1);
  			}
  
  			else
  			{
  				/*
  				 * When the edge is preceding, edge_reader points to the
  				 * position whose value is before the new edge value. We need
  				 * to back off one if the value in its next position is equal
  				 * to the new edge value. We also back off one position if
  				 * this edge is the trailing edge.
  				 */
  				if (prev_is_equal || !is_lead_edge)
  					ntuplestore_acc_advance(edge_reader, 1);
  			}
  
  			if (is_lead_edge)
  				level_state->lead_ready = true;
  		}
  
  		else
  		{
  			ntuplestore_acc_set_invalid(edge_reader);
  
  			if (!EDGE_IS_BOUND_FOLLOWING(edge))
  			{
  				/*
  				 * When the leading edge is "x preceding", check the first
  				 * value in the buffer. If this value is equal to the new edge
  				 * value, point the lead_reader to this value.
  				 */
  				if (is_lead_edge && is_equal)
  					ntuplestore_acc_seek_first(edge_reader);
  
  				return;
  			}
  
  			Assert(EDGE_IS_BOUND_FOLLOWING(edge));
  
  			/*
  			 * If the last value in the frame buffer is equal to the new edge
  			 * value, simple set the reader to the last value. Otherwise, we
  			 * check with the last row in the input buffer.
  			 */
  			if (is_equal)
  				ntuplestore_acc_seek_last(edge_reader);
  			else
  			{
  				lastrow_edge = checkLastRowForEdge(level_state, wstate,
  												   edge, edge_reader,
  												   new_edge_value,
  												   new_edge_value_isnull,
  												   is_lead_edge);
  				if (!lastrow_edge && is_lead_edge)
  					ntuplestore_acc_seek_last(edge_reader);
  			}
  		}
  	}
  
  	/* For the trailing edge, we also back off one. */
  	if (!is_lead_edge)
  	{
  		bool		found = false;
  
  		found = ntuplestore_acc_advance(edge_reader, -1);
  
  		if (!found)
  		{
  			if (EDGE_EQ_CURRENT_ROW(level_state, wstate, edge, is_lead_edge))
  			{
  				if (!ntuplestore_acc_tell(level_state->frame_buffer->current_row_reader,
  										  NULL))
  					ntuplestore_acc_seek_last(edge_reader);
  			}
  			else if (EDGE_IS_BOUND_FOLLOWING(edge))
  			{
  				lastrow_edge = checkLastRowForEdge(level_state, wstate,
  												   edge, edge_reader,
  												   new_edge_value,
  												   new_edge_value_isnull,
  												   is_lead_edge);
  				if (lastrow_edge)
  					ntuplestore_acc_seek_last(edge_reader);
  			}
  		}
  	}
  }
  
  /*
   * adjustDelayBoundEdgeForRange -- evaluate the given DELAY_BOUND edge
   * in RANGE-type frames and adjust the corresponding edge to the frame
   * buffer based on this new value.
   */
  static void
  adjustDelayBoundEdgeForRange(WindowStatePerLevel level_state,
  							 WindowState * wstate,
  							 WindowFrameEdge * edge,
  							 ExprState *edge_expr,
  							 ExprState *edge_range_expr,
  							 NTupleStoreAccessor * edge_reader,
  							 Datum *p_request_value,
  							 bool is_lead_edge)
  {
  	WindowFrameBuffer frame_buffer = level_state->frame_buffer;
  	NTupleStorePos pos;
  	bool		found;
  
  	Assert(level_state->has_delay_bound && !level_state->is_rows);
  	Assert(EDGE_IS_DELAYED(edge));
  	Assert(level_state->numSortCols == 1);
  
  	/* Compute the new edge value */
  	*p_request_value =
  		get_delay_edge(edge, edge_expr, level_state->is_rows, wstate);
  
  	/* Set the edge to current_row */
  	found = ntuplestore_acc_tell(frame_buffer->current_row_reader, &pos);
  	if (found)
  	{
  		found = ntuplestore_acc_seek(edge_reader, &pos);
  		Assert(found);
  	}
  	else
  	{
  		ntuplestore_acc_set_invalid(edge_reader);
  	}
  
  	advanceEdgeForRange(level_state, wstate,
  						edge, edge_expr, edge_range_expr,
  						edge_reader, is_lead_edge);
  }
  
  /*
   * adjustDelayBoundEdgeForRows -- evaluate the given DELAY_BOUND edge in
   * a ROWS-type frame and adjust the corresponding edge to the frame buffer
   * based on this new edge value.
   */
  static void
  adjustDelayBoundEdgeForRows(WindowStatePerLevel level_state,
  							WindowState * wstate,
  							WindowFrameEdge * edge,
  							ExprState *edge_expr,
  							NTupleStoreAccessor * edge_reader,
  							long int *p_request_num_rows,
  							long int *p_num_rows,
  							bool is_lead_edge)
  {
  	WindowFrameBuffer frame_buffer = level_state->frame_buffer;
  	bool		found;
  
  	Assert(level_state->has_delay_bound && level_state->is_rows);
  	Assert(EDGE_IS_DELAYED(edge));
  	Assert(list_length(level_state->level_funcs) >= 1);
  
  	*p_request_num_rows =
  		get_delay_edge(edge, edge_expr, level_state->is_rows, wstate);
  
  	*p_num_rows = 0;
  
  	/* Reset the edge_reader position in the frame buffer. */
  	if (EDGE_IS_BOUND_FOLLOWING(edge))
  	{
  		NTupleStorePos curr_row_pos;
  
  		if (ntuplestore_acc_tell(frame_buffer->current_row_reader, &curr_row_pos))
  		{
  			/*
  			 * Set edge_reader to current_row and advance it to
  			 * *p_request_num_rows.
  			 */
  			found = ntuplestore_acc_seek(edge_reader, &curr_row_pos);
  			Assert(found);
  
  			if (frame_buffer->num_rows_after - 1 >= *p_request_num_rows)
  			{
  				found = ntuplestore_acc_advance(edge_reader, *p_request_num_rows);
  				Assert(found);
  				*p_num_rows = *p_request_num_rows;
  			}
  			else
  			{
  				ntuplestore_acc_set_invalid(edge_reader);
  				*p_num_rows = frame_buffer->num_rows_after - 1;
  				if (level_state->agg_filled)
  					(*p_num_rows)++;
  			}
  		}
  
  		else
  		{
  			ntuplestore_acc_set_invalid(edge_reader);
  			*p_num_rows = 0;
  		}
  	}
  
  	else
  	{
  		NTupleStorePos curr_row_pos;
  		int			ntuples = *p_request_num_rows;
  
  		*p_num_rows = 0;
  
  		if (ntuplestore_acc_tell(frame_buffer->current_row_reader, &curr_row_pos))
  		{
  			found = ntuplestore_acc_seek(edge_reader, &curr_row_pos);
  			Assert(found);
  
  		}
  		else
  		{
  			found = ntuplestore_acc_seek_last(edge_reader);
  			found = ntuplestore_acc_tell(edge_reader, &curr_row_pos);
  			if (level_state->agg_filled)
  				ntuples++;
  		}
  
  		if (found)
  		{
  			if (frame_buffer->num_rows_before >= 0 - *p_request_num_rows)
  			{
  				found = ntuplestore_acc_advance(edge_reader, ntuples);
  				*p_num_rows = *p_request_num_rows;
  			}
  			else
  			{
  				ntuplestore_acc_set_invalid(edge_reader);
  
  				*p_num_rows = 0 - frame_buffer->num_rows_before;
  			}
  		}
  	}
  
  	/* For the trailing edge, we also back off one. */
  	if (!is_lead_edge)
  	{
  		/*
  		 * If the frame is "0 preceding/following" and the current_row_reader
  		 * is pointing to an invalid position, we simple set the trailing edge
  		 * to the last entry in the frame buffer.
  		 */
  		if (*p_request_num_rows == 0 &&
  			!ntuplestore_acc_tell(frame_buffer->current_row_reader, NULL))
  		{
  			ntuplestore_acc_seek_last(edge_reader);
  		}
  
  		/*
  		 * If the trailing edge points to an invalid position and the edge is
  		 * "x following", we want to set the current trailing edge to the end
  		 * of the buffer.
  		 */
  		else if (*p_num_rows > 0 &&
  				 !ntuplestore_acc_tell(edge_reader, NULL))
  		{
  			ntuplestore_acc_seek_last(edge_reader);
  			(*p_num_rows)--;
  		}
  
  		else
  			ntuplestore_acc_advance(edge_reader, -1);
  	}
  }
  
  /*
   * fetchCurrentRow -- retrieve the current_row from the input buffer.
   *
   * The current_row_reader always points to the previous current_row
   * position. When there is no rows in the buffer, this function reads
   * the first tuple from the outer plan.
   */
  static TupleTableSlot *
  fetchCurrentRow(WindowState * wstate)
  {
  	bool		found = false;
  	WindowInputBuffer buffer = wstate->input_buffer;
  	Window	   *window = (Window *) wstate->ps.plan;
  	bool		has_prior_tuple = false;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  	int			level;
  
  	if (buffer != NULL)
  	{
  		/* read the previous slot */
  		ntuplestore_acc_current_tupleslot(buffer->current_row_reader,
  										  wstate->priorslot);
  		has_prior_tuple = true;
  
  		found = ntuplestore_acc_advance(buffer->current_row_reader, 1);
  	}
  
  	if (found)
  		wstate->cur_slot_is_new = false;
  	else
  	{
  		/* Fetch the first tuple from the outer plan */
  		TupleTableSlot *slot = ExecProcNode(outerPlanState(wstate));
  
  		if (TupIsNull(slot))
  			return NULL;
  
  		Gpmon_M_Incr(GpmonPktFromWindowState(wstate), GPMON_QEXEC_M_ROWSIN);
  		CheckSendPlanStateGpmonPkt(&wstate->ps);
  		if (buffer == NULL)
  		{
  			initializePartition(wstate);
  		}
  
  		ntuplestore_acc_put_tupleslot(wstate->input_buffer->writer, slot);
  		wstate->input_buffer->num_tuples++;
  
  		if (buffer == NULL)
  		{
  			/* The first tuple will not break any keys. */
  			wstate->cur_slot_key_break = wstate->numlevels;
  			ntuplestore_acc_seek_first(wstate->input_buffer->current_row_reader);
  			ntuplestore_acc_seek_first(wstate->input_buffer->reader);
  		}
  
  		else
  		{
  			ntuplestore_acc_seek_last(wstate->input_buffer->current_row_reader);
  			ntuplestore_acc_seek_last(wstate->input_buffer->reader);
  		}
  
  		buffer = wstate->input_buffer;
  
  		if (wstate->curslot->tts_tupleDescriptor == NULL)
  			ExecSetSlotDescriptor(wstate->curslot,
  								  slot->tts_tupleDescriptor);
  		if (wstate->priorslot->tts_tupleDescriptor == NULL)
  			ExecSetSlotDescriptor(wstate->priorslot,
  								  slot->tts_tupleDescriptor);
  		if (wstate->spare->tts_tupleDescriptor == NULL)
  			ExecSetSlotDescriptor(wstate->spare,
  								  slot->tts_tupleDescriptor);
  
  		wstate->cur_slot_is_new = true;
  	}
  
  	found = ntuplestore_acc_current_tupleslot(buffer->current_row_reader,
  											  wstate->curslot);
  	Assert(found);
  
  	wstate->cur_slot_part_break = false;
  	wstate->cur_slot_key_break = -1;
  
  	/* Check if the partition key breaks */
  	if (has_prior_tuple)
  	{
  		wstate->cur_slot_part_break =
  			(window->numPartCols &&
  			 (!execTuplesMatch(wstate->curslot, wstate->priorslot,
  							   window->numPartCols, window->partColIdx,
  							   wstate->eqfunctions,
  							   econtext->ecxt_per_tuple_memory)));
  		if (wstate->cur_slot_part_break)
  			wstate->input_buffer->part_break = true;
  		else
  		{
  			int			level = 0;
  			bool		match = true;
  
  			/*
  			 * Process ordering partial key that breaks. We also increment the
  			 * peer_count.
  			 */
  			for (level = 0; level < wstate->numlevels; level++)
  			{
  				WindowStatePerLevel lvl = &wstate->level_state[level];
  
  				match = execTuplesMatch(wstate->curslot, wstate->priorslot,
  										lvl->numSortCols,
  										lvl->sortColIdx,
  										lvl->eqfunctions,
  										econtext->ecxt_per_tuple_memory);
  				if (!match)
  				{
  					wstate->cur_slot_key_break = level;
  					break;
  				}
  				lvl->peer_index++;
  			}
  		}
  	}
  
  	wstate->row_index++;
  	if (wstate->cur_slot_key_break != -1 &&
  		wstate->cur_slot_key_break < wstate->numlevels)
  		advanceKeyLevelState(wstate, wstate->cur_slot_key_break);
  
  	/*
  	 * If this tuple breaks the partition key, re-initialize the state.
  	 */
  	if (wstate->cur_slot_part_break)
  		initializePartition(wstate);
  	else
  		trimInputBuffer(buffer);
  
  	/*
  	 * Evaluate the DELAY_BOUND edges, including LEAD/LAG offset expressions,
  	 * and change the framing clauses accordingly.
  	 *
  	 * We also adjust the trailing edge and the leading edge to the frame buffer
  	 * based on these new edge values.
  	 */
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		if (!level_state->has_delay_bound)
  			continue;
  
  		if (EDGE_IS_DELAYED(level_state->frame->lead))
  		{
  			if (level_state->is_rows)
  				adjustDelayBoundEdgeForRows(level_state, wstate,
  											level_state->frame->lead,
  											level_state->lead_expr,
  											level_state->lead_reader,
  											&(level_state->lead_rows),
  											&(level_state->num_lead_rows),
  											true);
  			else
  				adjustDelayBoundEdgeForRange(level_state, wstate,
  											 level_state->frame->lead,
  											 level_state->lead_expr,
  											 level_state->lead_range_expr,
  											 level_state->lead_reader,
  											 &(level_state->lead_range),
  											 true);
  		}
  
  		if (EDGE_IS_DELAYED(level_state->frame->trail))
  		{
  			if (level_state->is_rows)
  			{
  				if (equal(level_state->frame->lead, level_state->frame->trail))
  				{
  					NTupleStorePos pos;
  					bool		found = false;
  
  					level_state->num_trail_rows = level_state->num_lead_rows;
  					level_state->trail_rows = level_state->lead_rows;
  
  					found = ntuplestore_acc_tell(level_state->lead_reader, &pos);
  					if (found)
  					{
  						ntuplestore_acc_seek(level_state->trail_reader, &pos);
  
  						/* We back off one for the trailing edge. */
  						ntuplestore_acc_advance(level_state->trail_reader, -1);
  					}
  
  					else
  					{
  						if (level_state->num_trail_rows > 0 ||
  							level_state->trail_rows == 0)
  						{
  							ntuplestore_acc_seek_last(level_state->trail_reader);
  							if (level_state->num_trail_rows > 0)
  								level_state->num_trail_rows--;
  						}
  
  						else
  							ntuplestore_acc_set_invalid(level_state->trail_reader);
  					}
  				}
  				else
  					adjustDelayBoundEdgeForRows(level_state, wstate,
  												level_state->frame->trail,
  												level_state->trail_expr,
  												level_state->trail_reader,
  												&(level_state->trail_rows),
  											  &(level_state->num_trail_rows),
  												false);
  			}
  			else
  			{
  				adjustDelayBoundEdgeForRange(level_state, wstate,
  											 level_state->frame->trail,
  											 level_state->trail_expr,
  											 level_state->trail_range_expr,
  											 level_state->trail_reader,
  											 &(level_state->trail_range),
  											 false);
  			}
  		}
  
  		/* set empty_frame flag */
  		setEmptyFrame(level_state, wstate);
  	}
  
  	return wstate->curslot;
  }
  
  /*
   * fetchTupleSlotThroughBuf -- fetch a tuple through the reader of the
   * input buffer.
   */
  static TupleTableSlot *
  fetchTupleSlotThroughBuf(WindowState * wstate)
  {
  	WindowInputBuffer buffer = wstate->input_buffer;
  	bool		found = false;
  	Window	   *window = (Window *) wstate->ps.plan;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  	bool		has_prior_tuple = false;
  
  	/* Read the previous tupleslot if any */
  	if (ntuplestore_acc_tell(buffer->reader, NULL))
  	{
  		found = ntuplestore_acc_current_tupleslot(buffer->reader,
  												  wstate->priorslot);
  		has_prior_tuple = found;
  	}
  
  	if (found)
  	{
  		/* Advance the reader */
  		found = ntuplestore_acc_advance(buffer->reader, 1);
  		Assert(!found);
  	}
  
  
  	if (!found)
  	{
  		TupleTableSlot *slot = ExecProcNode(outerPlanState(wstate));
  
  		if (TupIsNull(slot))
  			return NULL;
  
  		Gpmon_M_Incr(GpmonPktFromWindowState(wstate), GPMON_QEXEC_M_ROWSIN);
  		CheckSendPlanStateGpmonPkt(&wstate->ps);
  		/* Put the new tuple into the input buffer */
  		ntuplestore_acc_put_tupleslot(buffer->writer, slot);
  		buffer->num_tuples++;
  
  		ntuplestore_acc_seek_last(buffer->reader);
  
  		ntuplestore_acc_current_tupleslot(buffer->writer,
  										  wstate->spare);
  		wstate->cur_slot_is_new = true;
  	}
  
  	wstate->cur_slot_part_break = false;
  	wstate->cur_slot_key_break = -1;
  
  	if (!TupIsNull(wstate->priorslot))
  	{
  		wstate->cur_slot_part_break =
  			(window->numPartCols &&
  			 (!execTuplesMatch(wstate->spare, wstate->priorslot,
  							   window->numPartCols, window->partColIdx,
  							   wstate->eqfunctions,
  							   econtext->ecxt_per_tuple_memory)));
  
  		if (wstate->cur_slot_part_break)
  			buffer->part_break = true;
  
  		else
  		{
  			int			level = 0;
  			bool		match;
  
  			/*
  			 * Process ordering partial key that breaks. We also increment the
  			 * peer_index.
  			 */
  			for (level = 0; level < wstate->numlevels; level++)
  			{
  				WindowStatePerLevel lvl = &wstate->level_state[level];
  
  				match = execTuplesMatch(wstate->spare, wstate->priorslot,
  										lvl->numSortCols,
  										lvl->sortColIdx,
  										lvl->eqfunctions,
  										econtext->ecxt_per_tuple_memory);
  				if (!match)
  				{
  					wstate->cur_slot_key_break = level;
  					break;
  				}
  
  				lvl->peer_count++;
  			}
  		}
  	}
  
  	return wstate->spare;
  }
  
  /*
   * ExecWindow
   *
   */
  TupleTableSlot *
  ExecWindow(WindowState * wstate)
  {
  	bool		output_ready;
  	TupleTableSlot *next_tupleslot;
  	TupleTableSlot *resultSlot;
  	ExprContext *econtext;
  	int			level;
  	ExprDoneCond isDone;
  	bool		last_peer = false;
  
  	econtext = wstate->ps.ps_ExprContext;
  
  	/* Fetch the current_row */
  	econtext->ecxt_scantuple = fetchCurrentRow(wstate);
  
  	econtext->ecxt_outertuple =
  		econtext->ecxt_scantuple;		/* XXX really need this? */
  
  	if (TupIsNull(econtext->ecxt_scantuple))
  	{
  		ExecEagerFreeWindow(wstate);
  
  		return NULL;			/* we are done */
  	}
  
  	/* Process the current_row if it has not been processed. */
  	if (wstate->cur_slot_is_new)
  	{
  		processTupleSlot(wstate, wstate->curslot, false);
  		wstate->cur_slot_is_new = false;
  	}
  
  	/*
  	 * Increment the current_row in the frame buffer for each key level.
  	 */
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		if (level_state->empty_frame &&
  			!level_state->has_delay_bound)
  			continue;
  
  		if (!level_state->trivial_frames_only)
  		{
  			if ((wstate->input_buffer->num_tuples > 1) &&
  				(level_state->is_rows ||
  				 (wstate->cur_slot_key_break != -1 &&
  				  level >= wstate->cur_slot_key_break)))
  			{
  				econtext->ecxt_scantuple = wstate->curslot;
  				incrementCurrentRow(level_state->frame_buffer, wstate);
  			}
  		}
  	}
  
  	output_ready = checkOutputReady(wstate);
  
  	while (!output_ready)
  	{
  		next_tupleslot = fetchTupleSlotThroughBuf(wstate);
  
  		if (TupIsNull(next_tupleslot) || wstate->cur_slot_part_break)
  			last_peer = true;
  
  		processTupleSlot(wstate, next_tupleslot, last_peer);
  
  		if (!TupIsNull(next_tupleslot))
  			output_ready = checkOutputReady(wstate);
  		else
  			output_ready = true;
  	}
  
  	ResetExprContext(econtext);
  
  	econtext->ecxt_scantuple = wstate->curslot;
  	econtext->ecxt_outertuple = wstate->curslot;
  
  	invokeWindowFuncs(wstate);
  
  	/*
  	 * Form the result tuple using ExecProject(), and return it.
  	 */
  	resultSlot = ExecProject(wstate->ps.ps_ProjInfo, &isDone);
  
  	if (!TupIsNull(resultSlot))
  	{
  		Gpmon_M_Incr_Rows_Out(GpmonPktFromWindowState(wstate));
  		CheckSendPlanStateGpmonPkt(&wstate->ps);
  	}
  
  	else
  	{
  		ExecEagerFreeWindow(wstate);
  	}
  
  	return resultSlot;
  }
  
  /*
   * resetTransValue -- reset the transition value for
   * a given function.
   */
  static void
  resetTransValue(WindowStatePerFunction funcstate,
  				WindowState * wstate)
  {
  	if (funcstate->isAgg)
  	{
  		freeTransValue(&funcstate->aggTransValue,
  					   funcstate->aggTranstypeByVal,
  					   &funcstate->aggTransValueIsNull,
  					   &funcstate->aggNoTransValue,
  					   true);
  
  		if (!funcstate->aggTranstypeByVal && !funcstate->aggTransValueIsNull &&
  			DatumGetPointer(funcstate->aggTransValue) != NULL)
  			pfree(DatumGetPointer(funcstate->aggTransValue));
  
  		if (funcstate->aggInitValueIsNull)
  			funcstate->aggTransValue = funcstate->aggInitValue;
  		else
  		{
  			funcstate->aggTransValue =
  				datumCopyWithMemManager(0, funcstate->aggInitValue,
  										funcstate->aggTranstypeByVal,
  										funcstate->aggTranstypeLen,
  										&(wstate->mem_manager));
  		}
  
  		funcstate->aggTransValueIsNull = funcstate->aggInitValueIsNull;
  		funcstate->aggNoTransValue = funcstate->aggInitValueIsNull;
  	}
  	else
  	{
  		int			argno = 1;
  		WindowRefExprState *wrxstate = funcstate->wrxstate;
  
  		if (wrxstate->argtypbyval)
  			freeTransValue(&funcstate->aggTransValue,
  						   wrxstate->argtypbyval[argno],
  						   &funcstate->aggTransValueIsNull,
  						   &funcstate->aggNoTransValue,
  						   true);
  
  		funcstate->win_value = 0;
  		funcstate->win_value_is_null = true;
  	}
  }
  
  
  /*
   * resetTransValues -- reset the transition values for
   * each function.
   */
  static void
  resetTransValues(WindowStatePerLevel level_state,
  				 WindowState * wstate)
  {
  	ListCell   *lc;
  
  	foreach(lc, level_state->level_funcs)
  	{
  		WindowStatePerFunction funcstate = (WindowStatePerFunction) lfirst(lc);
  
  		resetTransValue(funcstate, wstate);
  	}
  }
  
  
  /*
   * processTupleSlot -- process a new tuple.
   *
   * This function passes the given tuple to each function in every
   * key level to compute an intermediate result.
   *
   * We only handle non-trivial window functions here.
   *
   * 'last_peer' indicates if the tuple to be inserted is the last tuple from
   * the input. It is used to determine if the trailing edge and leading edge
   * need to be adjusted.
   */
  static void
  processTupleSlot(WindowState * wstate, TupleTableSlot * slot, bool last_peer)
  {
  	int			level;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  		ListCell   *lc;
  		bool		adv_peercount = false;
  
  		/*
  		 * For non-delayed frames, we can simply ignore this level. However,
  		 * for delayed frames, we still need to store the intermediate results
  		 * because they may be needed later.
  		 */
  		if (level_state->empty_frame &&
  			!level_state->has_delay_bound)
  			continue;
  
  		if (!level_state->trivial_frames_only)
  		{
  			bool		write_pre_value = false;
  
  			if (wstate->input_buffer->num_tuples > 1 &&
  				wstate->cur_slot_is_new)
  			{
  				if (level_state->is_rows || level_state->has_only_trans_funcs)
  					write_pre_value = true;
  				else if (wstate->cur_slot_part_break || TupIsNull(slot))
  					write_pre_value = true;
  				else if ((wstate->cur_slot_key_break != -1 &&
  						  level >= wstate->cur_slot_key_break))
  					write_pre_value = true;
  			}
  
  			/*
  			 * If this tuple breaks the order by key in this level, we write
  			 * out previous aggregate values if any.
  			 */
  			if (write_pre_value)
  			{
  				/*
  				 * If there is a function that requires peer counts, we need
  				 * to compute its preliminary value before appending it to the
  				 * frame buffer.
  				 */
  				if (level_state->need_peercount)
  				{
  					foreach(lc, level_state->level_funcs)
  					{
  						WindowStatePerFunction funcstate =
  						(WindowStatePerFunction) lfirst(lc);
  
  						if (funcstate->winpeercount)
  							add_tuple_to_trans(funcstate, wstate, econtext, false);
  					}
  				}
  
  				/*
  				 * Set econtext->ecxt_scantuple because the range frame needs
  				 * this for order keys.
  				 */
  				econtext->ecxt_scantuple = wstate->priorslot;
  				appendToFrameBuffer(level_state, wstate, last_peer);
  
  				/*
  				 * Reset the transition value for those functions which has
  				 * preliminary functions, but not inverse preliminary
  				 * functions.
  				 */
  				foreach(lc, level_state->level_funcs)
  				{
  					WindowStatePerFunction funcstate = (WindowStatePerFunction) lfirst(lc);
  
  					if (funcstate->isAgg &&
  						!funcstate->cumul_frame &&
  						!OidIsValid(funcstate->invprelimfn_oid) &&
  						OidIsValid(funcstate->prelimfn_oid))
  					{
  						resetTransValue(funcstate, wstate);
  					}
  					else if ((IS_LEAD_LAG(funcstate->wrxstate->winkind) ||
  							  IS_FIRST_LAST(funcstate->wrxstate->winkind)) &&
  							 !last_peer)
  					{
  						resetTransValue(funcstate, wstate);
  					}
  				}
  			}
  
  			/*
  			 * If this slot causes the partition key to break, we don't add
  			 * this tuple to the transition value.
  			 */
  			if (wstate->cur_slot_part_break || TupIsNull(slot))
  			{
  				if (write_pre_value)
  					level_state->agg_filled = false;
  
  				continue;
  			}
  
  			foreach(lc, level_state->level_funcs)
  			{
  				WindowStatePerFunction funcstate =
  				(WindowStatePerFunction) lfirst(lc);
  				ExprContext *econtext = wstate->ps.ps_ExprContext;
  
  				if (funcstate->trivial_frame)
  					continue;
  
  				if (HAS_ONLY_TRANS_FUNC(funcstate))
  					continue;
  
  				/*
  				 * Increment the values for peer counts. There may be more
  				 * than one such function in one key level. We should catch
  				 * this in the earlier stage, but in case we didn't, we only
  				 * do this once.
  				 */
  				if (!adv_peercount && funcstate->winpeercount)
  				{
  					if (wstate->cur_slot_is_new &&
  						wstate->cur_slot_key_break != -1 &&
  						wstate->cur_slot_key_break < wstate->numlevels)
  					{
  						advancePeerCount(level_state, level, wstate->cur_slot_key_break);
  					}
  
  					adv_peercount = true;
  				}
  				else
  				{
  					/* Add this tuple to its transition value */
  					econtext->ecxt_scantuple = slot;
  					econtext->ecxt_outertuple = slot;	/* XXX really need this? */
  
  					add_tuple_to_trans(funcstate, wstate, econtext, true);
  
  					level_state->agg_filled = true;
  				}
  			}
  		}
  	}
  }
  
  
  /*
   * checkOutputReady -- check each key level to see if its frame buffer
   * has sufficient data to compute one output rows. If so, return true,
   * othwise, return false.
   */
  static bool
  checkOutputReady(WindowState * wstate)
  {
  	int			level;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  
  	econtext->ecxt_scantuple = wstate->curslot;
  
  	if (wstate->input_buffer->part_break)
  		return true;
  
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		if (level_state->trivial_frames_only)
  			continue;
  
  		if (level_state->is_rows)
  		{
  			if (!hasEnoughDataInRows(level_state->frame_buffer,
  									 level_state,
  									 wstate,
  									 level_state->trail_rows,
  									 level_state->lead_rows))
  				return false;
  		}
  
  		else
  		{
  			if (!hasEnoughDataInRange(level_state->frame_buffer,
  									  level_state,
  									  wstate,
  									  level_state->trail_range,
  									  level_state->lead_range))
  				return false;
  		}
  	}
  
  	return true;
  }
  
  /*
   * coerceType -- coerce the frame parameter expression to a given type.
   */
  static Expr *
  coerceType(Expr *expr, Oid new_type)
  {
  	Oid			expr_type = 0;
  	int32		expr_typmod = 0;
  	Expr	   *new_expr = expr;
  
  	expr_type = exprType((Node *) expr);
  	expr_typmod = exprTypmod((Node *) expr);
  
  	if (expr_type != new_type)
  	{
  		new_expr = (Expr *)coerce_to_target_type(NULL,
  												 (Node *) expr,
  												 expr_type,
  												 new_type, expr_typmod,
  												 COERCION_EXPLICIT,
  												 COERCE_IMPLICIT_CAST,
  												 -1);
  	}
  
  	return new_expr;
  }
  
  static void
  init_bound_frame_edge_expr(WindowFrameEdge * edge, TupleDesc desc,
  						   AttrNumber attnum,
  						   bool is_trail,
  						   WindowStatePerLevel level_state,
  						   WindowState * wstate)
  {
  	ExprState  *n;
  	Expr	   *expr;
  	Expr	   *varexpr;
  	Oid			exprrestype;
  	Oid			ltype ,
  				rtype;
  	HeapTuple	tup;
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  	bool		isNull;
  	char	   *oprname;
  	Form_pg_operator opr;
  	int32		vartypmod = desc->attrs[attnum - 1]->atttypmod;
  	cqContext  *pcqCtx;
  
  	Insist(EDGE_IS_BOUND(edge));
  
  	ltype = desc->attrs[attnum - 1]->atttypid;
  	rtype = exprType(edge->val);
  
  	varexpr = (Expr *)makeVar(0, attnum, ltype, vartypmod, 0);
  
  	expr = (Expr *)edge->val;
  
  	/*
  	 * Set trailing or leading expression state. If one of such expression
  	 * exist, we coerce the other one to the same type.
  	 */
  	if (is_trail)
  	{
  		if (level_state->lead_expr != NULL)
  		{
  			expr = coerceType(expr, exprType((Node *) (level_state->lead_expr->expr)));
  			if (expr == NULL)
  				ereport(ERROR,
  						(errcode(ERRCODE_DATA_EXCEPTION),
  						 errmsg("can't coerce trailing frame bound to type of leading frame bound"),
  						 errhint("specify the leading and trailing frame bounds as the same type")));
  		}
  
  		n = ExecInitExpr(expr, (PlanState *) wstate);
  
  		level_state->trail_expr = n;
  		if (!EDGE_IS_DELAYED(edge))
  			level_state->trail_range = ExecEvalExpr(n, econtext, &isNull, NULL);
  	}
  
  	else
  	{
  		if (level_state->trail_expr != NULL)
  		{
  			expr = coerceType(expr, exprType((Node *) (level_state->trail_expr->expr)));
  			if (expr == NULL)
  				ereport(ERROR,
  						(errcode(ERRCODE_DATA_EXCEPTION),
  						 errmsg("can't coerce leading frame bound to type of trailing frame bound"),
  						 errhint("specify the leading and trailing frame bounds as the same type")));
  		}
  
  		n = ExecInitExpr((Expr *)expr, (PlanState *) wstate);
  
  		level_state->lead_expr = n;
  		if (!EDGE_IS_DELAYED(edge))
  			level_state->lead_range = ExecEvalExpr(n, econtext, &isNull, NULL);
  	}
  
  	if ((EDGE_IS_BOUND_PRECEDING(edge) &&
  		 level_state->col_sort_asc[0]) ||
  		(EDGE_IS_BOUND_FOLLOWING(edge) &&
  		 !level_state->col_sort_asc[0]))
  	{
  		oprname = "-";
  	}
  	else
  	{
  		oprname = "+";
  	}
  
  	pcqCtx = caql_beginscan(
  							NULL,
  							cql("SELECT * FROM pg_operator "
  								" WHERE oprname = :1 "
  								" AND oprleft = :2 "
  								" AND oprright = :3 "
  								" AND oprnamespace = :4 ",
  								CStringGetDatum(oprname),
  								ObjectIdGetDatum(ltype),
  								ObjectIdGetDatum(rtype),
  								ObjectIdGetDatum(PG_CATALOG_NAMESPACE)));
  
  	tup = caql_getnext(pcqCtx);
  
  	/*
  	 * If we didn't find an operator, it's probably because the user has
  	 * specified a RANGE parameter which does not have an implicit cast to the
  	 * sorting column. So, see what operator the parser found via oper() and
  	 * arrange for the LHS to be cast to that for the purpose of the
  	 * expression evaluation.
  	 */
  	if (!HeapTupleIsValid(tup))
  	{
  		HeapTuple	tup2;
  
  		List	   *oprlist = lappend(NIL, makeString(oprname));
  
  		tup2 = oper(NULL, oprlist, ltype, rtype, true, -1);
  
  		list_free_deep(oprlist);
  
  		opr = (Form_pg_operator)GETSTRUCT(tup2);
  
  		/* this is why we're here */
  		Insist(ltype != opr->oprleft);
  
  		varexpr = (Expr *)coerce_to_target_type(NULL, (Node *) varexpr,
  											  ltype, opr->oprleft, vartypmod,
  												COERCION_EXPLICIT,
  												COERCE_IMPLICIT_CAST,
  												-1);
  
  		exprrestype = opr->oprresult;
  		expr = make_opclause(HeapTupleGetOid(tup2), exprrestype,
  							 false, varexpr,
  							 (Expr *)edge->val);
  		((OpExpr *) expr)->opfuncid = opr->oprcode;
  		ReleaseOperator(tup2);
  	}
  	else
  	{
  		opr = ((Form_pg_operator)GETSTRUCT(tup));
  
  		exprrestype = opr->oprresult;
  		expr = make_opclause(HeapTupleGetOid(tup), exprrestype,
  							 false, varexpr,
  							 (Expr *)edge->val);
  		((OpExpr *) expr)->opfuncid = opr->oprcode;
  		caql_endscan(pcqCtx);
  	}
  
  	/*
  	 * If the frame edge operation returns a different type to the input type,
  	 * we must coerce it back.
  	 */
  	if (ltype != exprrestype)
  	{
  		expr =
  			(Expr *)coerce_to_target_type(NULL,
  										  (Node *) expr,
  										  exprrestype,
  										  ltype, vartypmod,
  										  COERCION_EXPLICIT,
  										  COERCE_IMPLICIT_CAST,
  										  -1);
  	}
  
  	Insist(PointerIsValid(expr));
  
  	n = ExecInitExpr(expr, (PlanState *) wstate);
  
  	if (is_trail)
  		level_state->trail_range_expr = n;
  	else
  		level_state->lead_range_expr = n;
  
  	/*
  	 * Construct immediate clause like: var +/- range = var where var is ORDER
  	 * BY value and range is frame range value, so "var +/- range" means frame
  	 * edge created above (expr). Now, if the left hand equals to the right
  	 * hand, the frame edge must be on the current row. The created expression
  	 * is used to determine if the frame edge is on the current row or not.
  	 */
  	n = make_eq_exprstate(wstate, expr, varexpr);
  	if (is_trail)
  		level_state->trail_range_eq_expr = n;
  	else
  		level_state->lead_range_eq_expr = n;
  }
  
  /*
   * make_eq_exprstate
   * Given expr1 and expr2, make clase "expr1 = expr2".
   * In case the result types of expr1 and expr2 are different,
   * the cast is installed on expr2.
   */
  static ExprState *
  make_eq_exprstate(WindowState * wstate, Expr *expr1, Expr *expr2)
  {
  	Oid			restype1,
  				restype2;
  	Expr	   *eq_expr;
  	Operator	eq_optup;
  	Oid			eq_opid;
  
  	restype1 = exprType((Node *) expr1);
  	restype2 = exprType((Node *) expr2);
  	if (restype1 != restype2)
  	{
  		expr2 = (Expr *)coerce_to_target_type(NULL,
  											  (Node *) expr2,
  											  restype2,
  											  restype1,
  											  exprTypmod((Node *) expr2),
  											  COERCION_EXPLICIT,
  											  COERCE_IMPLICIT_CAST,
  											  -1);
  	}
  
  	eq_optup = equality_oper(restype1, false);
  	Assert(exprType((Node *) expr1) == exprType((Node *) expr2));
  	eq_opid = oprid(eq_optup);
  	eq_expr = make_opclause(eq_opid, BOOLOID, false, expr1, expr2);
  	((OpExpr *) eq_expr)->opfuncid = oprfuncid(eq_optup);
  	ReleaseSysCache(eq_optup);
  
  	return ExecInitExpr(eq_expr, (PlanState *) wstate);
  }
  
  /*
   * exec_eq_exprstate
   * Executes eq_exprstate made in make_eq_exprstate() and returns
   * bool result. It sets ecxt_scantuple to the current slot
   * so that Vars contained in eq_exprstate point to the current row.
   */
  static bool
  exec_eq_exprstate(WindowState * wstate, ExprState *eq_exprstate)
  {
  	ExprContext *econtext = wstate->ps.ps_ExprContext;
  	MemoryContext oldctx;
  	bool		isnull,
  				result;
  
  	Assert(IsA(eq_exprstate->expr, OpExpr));
  
  	/* Make sure Var in eq_expr points to the current slot */
  	econtext->ecxt_scantuple = wstate->curslot;
  
  	oldctx = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
  	result = DatumGetBool(ExecEvalExpr(eq_exprstate, econtext, &isnull, NULL));
  	MemoryContextSwitchTo(oldctx);
  
  	return isnull ? false : result;
  }
  
  /*
   * setEmptyFrame -- set empty_frame flag if the framing clause
   * contains no tuples.
   */
  static void
  setEmptyFrame(WindowStatePerLevel level_state,
  			  WindowState * wstate)
  {
  	WindowFrame *frame = level_state->frame;
  
  	level_state->empty_frame = false;
  
  	Assert(!(frame->trail->kind == WINDOW_UNBOUND_FOLLOWING ||
  			 frame->lead->kind == WINDOW_UNBOUND_PRECEDING ||
  			 (EDGE_IS_BOUND_FOLLOWING(frame->trail) &&
  			  EDGE_IS_BOUND_PRECEDING(frame->lead))));
  
  	if (!EDGE_IS_DELAYED_BOUND(frame->trail) &&
  		!EDGE_IS_DELAYED_BOUND(frame->lead) &&
  		((EDGE_IS_BOUND_PRECEDING(frame->trail) &&
  		  EDGE_IS_BOUND_PRECEDING(frame->lead)) ||
  		 (EDGE_IS_BOUND_FOLLOWING(frame->trail) &&
  		  EDGE_IS_BOUND_FOLLOWING(frame->lead))))
  	{
  		bool		is_pre =
  		(EDGE_IS_BOUND_PRECEDING(frame->trail) &&
  		 EDGE_IS_BOUND_PRECEDING(frame->lead));
  
  		if (frame->is_rows)
  		{
  			if (level_state->lead_rows < level_state->trail_rows)
  				level_state->empty_frame = true;
  		}
  		else
  		{
  			Oid			ineq_ordfuncid;
  			Oid			eq_ordfuncid;
  			Datum		ineq_datum;
  			Datum		eq_datum;
  			FmgrInfo	ineq_fcinfo;
  			FmgrInfo	eq_fcinfo;
  			Operator	ineq_optup;
  			Operator	eq_optup;
  
  			ineq_optup = ordering_oper(exprType((Node *) level_state->trail_expr->expr),
  									   false);
  			ineq_ordfuncid = oprfuncid(ineq_optup);
  			ReleaseSysCache(ineq_optup);
  			fmgr_info(ineq_ordfuncid, &ineq_fcinfo);
  
  			eq_optup = equality_oper(exprType((Node *) level_state->trail_expr->expr),
  									 false);
  			eq_ordfuncid = oprfuncid(eq_optup);
  			ReleaseSysCache(eq_optup);
  			fmgr_info(eq_ordfuncid, &eq_fcinfo);
  
  			/* is trail less than or equal to lead */
  			ineq_datum = FunctionCall2(&ineq_fcinfo, level_state->trail_range,
  									   level_state->lead_range);
  			eq_datum = FunctionCall2(&eq_fcinfo, level_state->trail_range,
  									 level_state->lead_range);
  			if (is_pre)
  			{
  				if (DatumGetBool(ineq_datum) && !DatumGetBool(eq_datum))
  					level_state->empty_frame = true;
  			}
  			else
  			{
  				if (!DatumGetBool(ineq_datum) && !DatumGetBool(eq_datum))
  					level_state->empty_frame = true;
  			}
  		}
  	}
  }
  
  /*
   * Initialise frame specific state.
   */
  
  static void
  init_frames(WindowState * wstate)
  {
  	int			level = -1;
  	ListCell   *lc;
  	int			col_no;
  
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  		int			ncols = level_state->numSortCols;
  		WindowFrame *frame = level_state->frame;
  
  		level_state->col_sort_asc = palloc(sizeof(bool) * ncols);
  
  		foreach(lc, level_state->level_funcs)
  		{
  			WindowStatePerFunction funcstate =
  			(WindowStatePerFunction) lfirst(lc);
  
  			if (!frame || (!funcstate->isAgg && !funcstate->allowframe &&
  						   !IS_LEAD_LAG(funcstate->wrxstate->winkind) &&
  						   !funcstate->winpeercount))
  			{
  				funcstate->trivial_frame = true;
  			}
  
  
  			/*
  			 * We say the function is trivial to invoke if its frame is
  			 * defined as ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
  			 * with no exclusion clause OR it is a window function without a
  			 * frame.
  			 */
  			if (funcstate->isAgg && frame && frame->is_rows &&
  				frame->trail->kind == WINDOW_UNBOUND_PRECEDING &&
  				frame->lead->kind == WINDOW_CURRENT_ROW &&
  				frame->exclude == WINDOW_EXCLUSION_NULL)
  			{
  				funcstate->trivial_frame = true;
  			}
  			else if (funcstate->isAgg && frame && !frame->is_rows &&
  					 frame->trail->kind == WINDOW_UNBOUND_PRECEDING &&
  					 frame->lead->kind == WINDOW_CURRENT_ROW &&
  					 frame->exclude == WINDOW_EXCLUSION_NULL)
  			{
  				funcstate->cumul_frame = true;
  			}
  		}
  
  		/* what sort order did the user specify for each key? */
  		for (col_no = 0; col_no < ncols; col_no++)
  		{
  			Oid			sortop = level_state->sortOperators[col_no];
  			char	   *oprname;
  			int			fetchCount;
  
  			oprname = caql_getcstring_plus(
  										   NULL,
  										   &fetchCount,
  										   NULL,
  									   cql("SELECT oprname FROM pg_operator "
  										   " WHERE oid = :1 ",
  										   ObjectIdGetDatum(sortop)));
  
  			Insist(fetchCount);
  
  			if (strcmp(oprname, "<") == 0)
  				level_state->col_sort_asc[col_no] = true;
  			else
  				level_state->col_sort_asc[col_no] = false;
  
  			pfree(oprname);
  		}
  
  		/* now, initialize the actual frame */
  		if (frame)
  		{
  			if (frame->is_rows)
  			{
  				ExprContext *econtext = wstate->ps.ps_ExprContext;
  
  				if (EDGE_IS_BOUND(frame->trail) &&
  					level_state->frame->trail->val != NULL)
  				{
  					long int	rows_param = 0;
  					bool		isnull = true;
  					Expr	   *expr = (Expr *)level_state->frame->trail->val;
  
  					expr = coerceType(expr, INT4OID);
  
  					level_state->trail_expr =
  						ExecInitExpr((Expr *)expr,
  									 (PlanState *) wstate);
  
  
  					if (!EDGE_IS_DELAYED_BOUND(frame->trail))
  					{
  						rows_param = ExecEvalExpr(level_state->trail_expr,
  												  econtext,
  												  &isnull,
  												  NULL);
  
  						Insist(!isnull);
  					}
  
  					if (frame->trail->kind == WINDOW_BOUND_PRECEDING)
  						rows_param = -rows_param;
  
  					level_state->trail_rows = rows_param;
  				}
  				else
  					level_state->trail_rows = 0L;
  
  				if (EDGE_IS_BOUND(frame->lead) &&
  					level_state->frame->lead->val != NULL)
  				{
  					long int	rows_param = 0;
  					bool		isnull = true;
  					Expr	   *expr = (Expr *)level_state->frame->lead->val;
  
  					expr = coerceType(expr, INT4OID);
  
  					level_state->lead_expr =
  						ExecInitExpr((Expr *)expr,
  									 (PlanState *) wstate);
  
  					if (!EDGE_IS_DELAYED_BOUND(frame->lead))
  					{
  						rows_param = ExecEvalExpr(level_state->lead_expr,
  												  econtext,
  												  &isnull,
  												  NULL);
  
  						Insist(!isnull);
  					}
  
  					if (frame->lead->kind == WINDOW_BOUND_PRECEDING)
  						rows_param = -rows_param;
  
  					level_state->lead_rows = rows_param;
  				}
  				else
  					level_state->lead_rows = 0L;
  			}
  			else
  			{
  				TupleDesc	desc = ExecGetResultType(wstate->ps.lefttree);
  
  				/* we only need the subtraction function for bound frames */
  				if (EDGE_IS_BOUND(frame->trail))
  				{
  					init_bound_frame_edge_expr(frame->trail, desc,
  											   level_state->sortColIdx[0],
  											   true, level_state, wstate);
  				}
  
  				if (EDGE_IS_BOUND(frame->lead))
  				{
  					init_bound_frame_edge_expr(frame->lead, desc,
  											   level_state->sortColIdx[0],
  											   false, level_state, wstate);
  				}
  			}
  		}
  
  		/* look for empty frames */
  		setEmptyFrame(level_state, wstate);
  	}
  
  	/*
  	 * Set trivial_frames_only in each level state if its functions all have
  	 * trivial frames.
  	 *
  	 * Also initialize the transition values for functions in this level.
  	 */
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		bool		trivial_frames_only = true;
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		foreach(lc, level_state->level_funcs)
  		{
  			WindowStatePerFunction funcstate = (WindowStatePerFunction)
  			lfirst(lc);
  
  			if (!funcstate->trivial_frame)
  			{
  				trivial_frames_only = false;
  				break;
  			}
  		}
  		level_state->trivial_frames_only = trivial_frames_only;
  
  		resetTransValues(level_state, wstate);
  	}
  
  	/*
  	 * Set has_delay_bound flag in each level state if the frame clause has a
  	 * edge of DELAY_BOUND type.
  	 */
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		level_state->has_delay_bound =
  			(EDGE_IS_DELAYED(level_state->frame->lead) ||
  			 EDGE_IS_DELAYED(level_state->frame->trail));
  	}
  
  	/*
  	 * Set has_only_trans_funcs in each level state if all functions in this
  	 * level are aggregate functions that have no preliminary functions or
  	 * inverse preliminary functions.
  	 */
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  		bool		has_only_trans_funcs = false;
  		int			funcno = 0;
  
  		foreach(lc, level_state->level_funcs)
  		{
  			WindowStatePerFunction funcstate =
  			(WindowStatePerFunction) lfirst(lc);
  
  			Assert(NULL != level_state->level_funcs);
  
  			if (HAS_ONLY_TRANS_FUNC(funcstate))
  			{
  				Assert(funcno == 0 || has_only_trans_funcs);
  				has_only_trans_funcs = true;
  
  				/*
  				 * XXX disable the general case for now. Some work needs to be
  				 * done when adjusting edges after appending values into the
  				 * frame buffer.
  				 */
  				ereport(ERROR,
  						(errcode(ERRCODE_GP_FEATURE_NOT_SUPPORTED),
  						 errmsg("aggregate functions with no prelimfn or "
  					"invprelimfn are not yet supported as window functions")));
  			}
  
  			funcno++;
  		}
  		level_state->has_only_trans_funcs = has_only_trans_funcs;
  	}
  
  	/*
  	 * Allocate one FrameBufferEntry buffer for each level state, so that we
  	 * don't need to do pallocs/pfrees every time we read an entry from the
  	 * frame buffer.
  	 */
  	for (level = 0; level < wstate->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &wstate->level_state[level];
  
  		level_state->curr_entry_buf = createFrameBufferEntry(level_state);
  		level_state->trail_entry_buf = createFrameBufferEntry(level_state);
  		level_state->lead_entry_buf = createFrameBufferEntry(level_state);
  	}
  }
  
  static FmgrInfo *
  get_ltfuncs(TupleDesc tupdesc, int numCols, AttrNumber *matchColIdx)
  {
  	FmgrInfo   *ltfunctions = (FmgrInfo *)palloc(numCols * sizeof(FmgrInfo));
  	int			i;
  
  	for (i = 0; i < numCols; i++)
  	{
  		AttrNumber	att = matchColIdx[i];
  		Oid			typid = tupdesc->attrs[att - 1]->atttypid;
  		Oid			lt_function;
  		Operator	optup;
  
  		optup = ordering_oper(typid, false);
  		lt_function = oprfuncid(optup);
  		fmgr_info(lt_function, &ltfunctions[i]);
  		ReleaseSysCache(optup);
  	}
  
  	return ltfunctions;
  }
  
  /*
   * The buffer cursor abstracts a scan on the frame.  The tricky part is that
   * the underlying tuplestore may or may not have rows while in some circumstances
   * we should look at the input buffer as the last row, which is called "last_agg."
   * The usage of these APIs is straightforward; BeginScan, Next, EndScan.  Don't
   * open multiple cursors at one time on the same window level, as the cursor
   * has a transient tuplestore reader.
   * The forward parameter indicates if this cursor starts at the beginning and scans
   * forward, or if it starts at the end of frame and scans backward.
   */
  static WindowBufferCursor
  windowBufferBeginScan(WindowState * wstate, WindowStatePerLevel level_state,
  					  bool forward)
  {
  	WindowBufferCursor cursor;
  	bool		include_last_agg = false;
  
  	cursor = (WindowBufferCursor) palloc0(sizeof(WindowBufferCursorData));
  
  	/*
  	 * Because it is a short-term use, we don't allocate a new accessor but
  	 * use the existing one to scan the tuplestore.  The other side of reader
  	 * should remain so that detection of frame end can work correctly (i.e.
  	 * if both reader intersects, it means the end of frame.)
  	 *
  	 * Remember the original position of accessor so later we can restore it.
  	 */
  	if (forward)
  	{
  		(void)ntuplestore_acc_tell(level_state->trail_reader,
  								   &cursor->orig_pos);
  		cursor->reader = level_state->trail_reader;
  	}
  	else
  	{
  		(void)ntuplestore_acc_tell(level_state->lead_reader,
  								   &cursor->orig_pos);
  		cursor->reader = level_state->lead_reader;
  	}
  
  	/*
  	 * This is a little complicated condition to tell if the last_agg row
  	 * should be looked at.  After setting use_last_agg, we'll return last_agg
  	 * just after finishing to return all the rows in the tuplestore.
  	 */
  	if (EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->trail, false))
  		include_last_agg = true;
  	if (!EDGE_IS_BOUND(level_state->frame->lead) ||
  		EDGE_IS_BOUND_FOLLOWING(level_state->frame->lead) ||
  	EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->lead, true))
  		include_last_agg = true;
  	cursor->use_last_agg = (!ntuplestore_acc_tell(level_state->lead_reader, NULL) &&
  							level_state->agg_filled && include_last_agg);
  
  	/*
  	 * Also we check if this frame has tuples.	If this is false, entire frame
  	 * is empty and nothing should be returned.
  	 */
  	cursor->has_tuples = hasTuplesInFrame(level_state, wstate);
  
  	if (forward)
  	{
  		/*
  		 * Since the trail_reader points to the value before the trailing
  		 * edge, we advance the trail_reader by one.
  		 */
  		if (ntuplestore_acc_tell(cursor->reader, NULL))
  			ntuplestore_acc_advance(cursor->reader, 1);
  		else
  		{
  			/*
  			 * In forward, if the leader is at invalid position and the
  			 * leading edge is not equal to the current as PRECEDING bound, we
  			 * return nothing.
  			 */
  			if (!ntuplestore_acc_tell(level_state->lead_reader, NULL) &&
  				(EDGE_IS_BOUND_PRECEDING(level_state->frame->lead) &&
  				 ((level_state->is_rows && level_state->lead_rows != 0) ||
  				  (!level_state->is_rows && !exec_eq_exprstate(wstate,
  										 level_state->lead_range_eq_expr)))))
  				cursor->has_tuples = false;
  			else
  				ntuplestore_acc_seek_first(cursor->reader);
  		}
  	}
  	else
  	{
  		/*
  		 * In backward, make sure the reader is on the last.
  		 */
  		if (!ntuplestore_acc_tell(cursor->reader, NULL))
  			ntuplestore_acc_seek_last(cursor->reader);
  	}
  
  	cursor->first = true;
  	cursor->level_state = level_state;
  	cursor->wstate = wstate;
  	cursor->forward = forward;
  
  	return cursor;
  }
  
  /*
   * A helper of windowBufferNext.  Scans and advances in the tuplestore.
   * If it has reached to the end, sets reader_eof to true and returns NULL.
   * Otherwise returns a valid List containing serialized WindowValues.
   */
  static List *
  windowBufferNextTupleStore(WindowBufferCursor cursor)
  {
  	WindowStatePerLevel level_state = cursor->level_state;
  	FrameBufferEntry *entry = level_state->curr_entry_buf;
  
  	if (!cursor->reader_eof)
  	{
  		if (!cursor->first)
  			ntuplestore_acc_advance(cursor->reader, cursor->forward ? 1 : -1);
  		cursor->first = false;
  		if (!ntuplestore_acc_tell(cursor->reader, NULL))
  			cursor->reader_eof = true;
  		else
  		{
  			/*
  			 * If the reader intersects the other side, it means the end of
  			 * frame.
  			 */
  			if (cursor->forward)
  			{
  				if (ntuplestore_acc_tell(level_state->lead_reader, NULL) &&
  					ntuplestore_acc_is_before(level_state->lead_reader,
  											  cursor->reader))
  					cursor->reader_eof = true;
  			}
  			else
  			{
  				if (ntuplestore_acc_tell(level_state->trail_reader, NULL) &&
  					ntuplestore_acc_is_before(cursor->reader,
  											  level_state->trail_reader))
  					cursor->reader_eof = true;
  			}
  		}
  	}
  
  	if (!cursor->reader_eof)
  	{
  		bool		found;
  
  		found = getCurrentValue(cursor->reader, level_state, entry);
  
  		/*
  		 * As we are sure the reader is at a valid position, the entry must be
  		 * found.
  		 */
  		Assert(found);
  
  		return entry->func_values;
  	}
  
  	return NULL;
  }
  
  /*
   * A helper of windowBufferNext.  Scans "last_agg" if we should, and converts the
   * input buffer to a List of WindowValues.	Since last_agg is the last row of
   * the frame if any, we set the use_last_agg to false after scanning and will never
   * get the values again.
   */
  static List *
  windowBufferNextLastAgg(WindowBufferCursor cursor)
  {
  	if (cursor->use_last_agg)
  	{
  		WindowStatePerLevel level_state = cursor->level_state;
  		WindowState *wstate = cursor->wstate;
  		FrameBufferEntry *entry = level_state->curr_entry_buf;
  		bool		found;
  		ExprContext *econtext = wstate->ps.ps_ExprContext;
  		TupleTableSlot *slot = econtext->ecxt_scantuple;
  		Size		len;
  
  		found = ntuplestore_acc_current_tupleslot(wstate->input_buffer->writer,
  												  wstate->spare);
  		Assert(found);
  
  		/*
  		 * Serialize and deserialize the input buffer.	This seems a little
  		 * redundant, but the format of input buffer and WindowValue is
  		 * different, and as we want the same format values from the buffer
  		 * cursor, we do this here. It is actually not so bad, as it's done in
  		 * memory.
  		 */
  		econtext->ecxt_scantuple = wstate->spare;
  		MemSet(wstate->serial_array, 0, wstate->max_size);
  		serializeEntry(level_state, econtext,
  					   &(wstate->serial_array), &(wstate->max_size), &len);
  		entry = deserializeEntry(level_state, entry, wstate->serial_array, len);
  		/* Get back the slot. */
  		econtext->ecxt_scantuple = slot;
  		/* We never use this again as it's the logical last row. */
  		cursor->use_last_agg = false;
  
  		return entry->func_values;
  	}
  
  	return NULL;
  }
  
  /*
   * Scans the frame buffer and returns the current value.  It advances the cursor
   * 1 forward or backward.  If the scan is finished, it returns NULL.
   */
  static List *
  windowBufferNext(WindowBufferCursor cursor)
  {
  	List	   *func_values;
  
  	if (!cursor->has_tuples)
  		return NULL;
  
  	if (cursor->forward)
  	{
  		/*
  		 * In forward, start from tuplestore and use last_agg after that.
  		 */
  		func_values = windowBufferNextTupleStore(cursor);
  		if (func_values)
  			return func_values;
  		return windowBufferNextLastAgg(cursor);
  	}
  	else
  	{
  		/*
  		 * In backward, start from last_agg and use tuplestore after that.
  		 */
  		func_values = windowBufferNextLastAgg(cursor);
  		if (func_values)
  			return func_values;
  		return windowBufferNextTupleStore(cursor);
  	}
  }
  
  /*
   * Ends the cursor.  The cursor will be freed.	The reader will be ok to use again.
   */
  static void
  windowBufferEndScan(WindowBufferCursor cursor)
  {
  	if (!ntuplestore_acc_seek(cursor->reader, &cursor->orig_pos))
  		ntuplestore_acc_set_invalid(cursor->reader);
  	pfree(cursor);
  }
  
  
  /* -----------------
   * ExecInitWindow
   *
   *	Creates the run-time information for the window node produced by the
   *	planner and initializes its outer subtree
   * -----------------
   */
  WindowState *
  ExecInitWindow(Window * node, EState *estate, int eflags)
  {
  	WindowState *wstate;
  	Plan	   *outerPlan;
  	ExprContext *econtext;
  	int			numrefs;
  	TupleDesc	desc;
  
  	/* Check for unsupported flags. */
  	Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
  
  	/* Create state structure. */
  	wstate = makeWindowState(node, estate);
  
  	/* Create expression context. */
  	ExecAssignExprContext(estate, &wstate->ps);
  
  	/* XXX: temporarily high for debugging purposes */
  #define WINDOW_NSLOTS 4
  
  	/* Initialize tuple table. */
  	ExecInitResultTupleSlot(estate, &wstate->ps);
  	wstate->curslot = ExecInitExtraTupleSlot(estate);
  	wstate->priorslot = ExecInitExtraTupleSlot(estate);
  	wstate->spare = ExecInitExtraTupleSlot(estate);
  
  	/*
  	 * Initialize child expressions.
  	 *
  	 * ExecInitExpr adds WindowRefExprState nodes it encounters to the wrxstates
  	 * list.
  	 */
  	wstate->ps.targetlist = (List *)
  		ExecInitExpr((Expr *)node->plan.targetlist,
  					 (PlanState *) wstate);
  	numrefs = list_length(wstate->wrxstates);
  
  	/* Initialize child nodes. */
  	outerPlan = outerPlan(node);
  	outerPlanState(wstate) = ExecInitNode(outerPlan, estate, eflags);
  	Assert(innerPlan(node) == NULL);
  
  	/* Initialize result tuple type and projection info. */
  	ExecAssignResultTypeFromTL(&wstate->ps);
  	ExecAssignProjectionInfo(&wstate->ps, NULL);
  
  	desc = ExecGetResultType(wstate->ps.lefttree);
  
  	/* Precompute fmgr lookup data for partition key equality function. */
  	if (node->numPartCols > 0)
  	{
  		wstate->eqfunctions =
  			execTuplesMatchPrepare(desc,
  								   node->numPartCols,
  								   node->partColIdx);
  	}
  	else
  		wstate->eqfunctions = NULL;
  
  	initWindowStatePerLevel(wstate, node);
  
  	/*
  	 * Allocate result storage and working storage per window ref. (Later we
  	 * may shrink this if we notice duplicate function calls, but allocate for
  	 * worst case.)  Note that we borrow aggregate result storage, since there
  	 * are no non-window aggregates that might use it.
  	 */
  	econtext = wstate->ps.ps_ExprContext;
  	econtext->ecxt_aggvalues = (Datum *)palloc0(sizeof(Datum) * numrefs);
  	econtext->ecxt_aggnulls = (bool *)palloc0(sizeof(bool) * numrefs);
  
  	wstate->need_peercount = false;
  
  	initWindowFuncState(wstate, node);
  
  	wstate->mem_manager.alloc = cxt_alloc;
  	wstate->mem_manager.free = cxt_free;
  	wstate->mem_manager.manager = wstate->transcontext;
  
  	/* Frame initialisation can take place now */
  	init_frames(wstate);
  
  	initGpmonPktForWindow((Plan *) node, &wstate->ps.gpmon_pkt, estate);
  
  	return wstate;
  }
  
  
  int
  ExecCountSlotsWindow(Window * node)
  {
  	return ExecCountSlotsNode(outerPlan(node)) +
  		ExecCountSlotsNode(innerPlan(node)) +
  		WINDOW_NSLOTS;
  }
  
  void
  ExecEndWindow(WindowState * node)
  {
  	int			level = 0;
  
  	for (level = 0; level < node->numlevels; level++)
  	{
  		WindowStatePerLevel level_state = &node->level_state[level];
  
  		freeFrameBufferEntry(level_state->curr_entry_buf);
  		level_state->curr_entry_buf = NULL;
  
  		freeFrameBufferEntry(level_state->trail_entry_buf);
  		level_state->trail_entry_buf = NULL;
  
  		freeFrameBufferEntry(level_state->lead_entry_buf);
  		level_state->lead_entry_buf = NULL;
  	}
  
  	ExecEagerFreeWindow(node);
  
  	/* Free the exprcontext */
  	ExecFreeExprContext(&node->ps);
  
  	/* clean out the tuple table */
  	ExecClearTuple(node->curslot);
  	ExecClearTuple(node->spare);
  	ExecClearTuple(node->priorslot);
  	ExecClearTuple(node->ps.ps_ResultTupleSlot);
  
  	if (node->transcontext != NULL)
  		MemoryContextDelete(node->transcontext);
  
  	if (node->cmpcontext != NULL)
  		MemoryContextDelete(node->cmpcontext);
  
  	pfree(node->serial_array);
  
  	if (node->numlevels > 0)
  		pfree(node->level_state);
  	if (node->func_state != NULL)
  		pfree(node->func_state);
  
  	/* shut down subplans */
  	ExecEndNode(outerPlanState(node));
  
  	EndPlanStateGpmonPkt(&node->ps);
  }
  
  /**
   * ExecReScanWindow
   *
   * Higher-up node is telling window node to perform re-scanning.
   * Note that exprCtxt may be NULL.
   */
  void
  ExecReScanWindow(WindowState * node, ExprContext *exprCtxt)
  {
  	Assert(node);
  
  	resetFrameBuffers(node);
  
  	ExecEagerFreeWindow(node);
  
  	Assert(outerPlanState(node));
  
  	ExecReScan(outerPlanState(node), exprCtxt);
  }
  
  /*
   * window_dummy - dummy execution routine for window functions
   *
   * This function is listed as the implementation (prosrc field) of pg_proc
   * entries for window functions.  Its only purpose is to throw an error
   * if someone mistakenly executes such a function in the normal way.
   */
  Datum
  window_dummy(PG_FUNCTION_ARGS)
  {
  	ereport(ERROR,
  			(errcode(ERRCODE_SYNTAX_ERROR),
  			 errmsg("function %s may only be called as a window function",
  					format_procedure(fcinfo->flinfo->fn_oid)),
  	  errhint("To call a function as a window function use an OVER clause."))
  		);
  
  	return (Datum)0;			/* keep compiler quiet */
  }
  
  /*
   * Implements the gp_execution_segment() function to return the contentid
   * of the current executing segment.
   */
  Datum
  mpp_execution_segment(PG_FUNCTION_ARGS)
  {
  	PG_RETURN_INT32(Gp_segment);
  }
  
  /*
   * Implements the gp_execution_dbid() function to return the dbid of the
   * current executing segment.
   */
  Datum
  gp_execution_dbid(PG_FUNCTION_ARGS)
  {
  	PG_RETURN_INT32(GpIdentity.dbid);
  }
  
  /*
   * ROW_NUMBER() OVER (...) --> BIGINT
   *
   * Implement ROW_NUMBER for the given window state:
   *
   * row_number_immed(internal) --> bigint
   */
  Datum
  row_number_immed(PG_FUNCTION_ARGS)
  {
  	WindowRefExprState *ref_state = (WindowRefExprState *) PG_GETARG_POINTER(0);
  
  	int64		result = 1 + ref_state->windowstate->row_index;
  
  	PG_RETURN_INT64(result);
  }
  
  
  /*
   *	RANK() OVER (... ORDER BY s) --> BIGINT
   *
   * Implement RANK for the given WindowRefExprState.
   *
   * rank_immed(internal) --> bigint
   */
  Datum
  rank_immed(PG_FUNCTION_ARGS)
  {
  	int64		result;
  	WindowRefExprState *ref_state = (WindowRefExprState *) PG_GETARG_POINTER(0);
  	WindowState *window_state = ref_state->windowstate;
  	WindowRef  *ref = (WindowRef *) ref_state->xprstate.expr;
  	WindowStatePerLevel level_state = &window_state->level_state[ref->winlevel];
  
  	/* Don't advance prior_rank here, let the framework do this. */
  
  	result = level_state->rank;
  
  	PG_RETURN_INT64(result);
  }
  
  
  /*
   *	DENSE_RANK() OVER (... ORDER BY s) --> BIGINT
   *
   * Implement DENSE_RANK for the given WindowRefExprState.
   *
   * dense_rank_immed(internal) --> bigint
   */
  Datum
  dense_rank_immed(PG_FUNCTION_ARGS)
  {
  	int64		result;
  	WindowRefExprState *ref_state = (WindowRefExprState *) PG_GETARG_POINTER(0);
  	WindowState *window_state = ref_state->windowstate;
  	WindowRef  *ref = (WindowRef *) ref_state->xprstate.expr;
  	WindowStatePerLevel level_state = &window_state->level_state[ref->winlevel];
  
  	/* Don't advance prior_dense_rank here, let the framework do this. */
  
  	result = level_state->dense_rank;
  
  	PG_RETURN_INT64(result);
  }
  
  
  /*
   * PERCENT_RANK() OVER (... ORDER BY s) -- FLOAT8
   *
   * Implement rank for the given WindowRefExprState.
   *
   * rank_immed(internal) --> bigint
   * percent_rank_final(bigint,bigint) --> float8
   */
  Datum
  percent_rank_final(PG_FUNCTION_ARGS)
  {
  	int64		arg1 = PG_GETARG_INT64(0);		/* rank in partition */
  	int64		arg2 = PG_GETARG_INT64(1);		/* partition row count */
  	double		result;
  
  	if (arg1 < 1 || arg2 < 1)
  	{
  		ereport(ERROR,
  				(errcode(ERRCODE_DATA_EXCEPTION),
  				 errmsg("arguments invalid or inconsistent"),
  			 errhint("inappropriate call to window function implementation")));
  		result = 0.0;			/* quieten GCC */
  	}
  	else if (arg1 == 1 && arg2 == 1)
  	{
  		result = 0.0;
  	}
  	else
  	{
  		/* Do division in double, then check for overflow */
  		result = (double)(arg1 - 1) / (double)(arg2 - 1);
  		if (result > DBL_MAX)
  		{
  			ereport(ERROR,
  					(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
  					 errmsg("value out of range: overflow")));
  		}
  	}
  
  	PG_RETURN_FLOAT8(result);
  }
  
  
  /*
   *	CUME_DIST() OVER (... ORDER BY s) --> FLOAT8
   *
   * Implement CUME_DIST for the given WindowRefExprState.
   *
   * cume_dist_prelim(internal) --> bigint
   * cume_dist_final(bigint,bigint) --> float8
   */
  Datum
  cume_dist_prelim(PG_FUNCTION_ARGS)
  {
  	int64		result;
  	WindowRefExprState *ref_state = (WindowRefExprState *) PG_GETARG_POINTER(0);
  	WindowState *window_state = ref_state->windowstate;
  	WindowRef  *ref = (WindowRef *) ref_state->xprstate.expr;
  	WindowStatePerLevel level_state = &window_state->level_state[ref->winlevel];
  
  	result = level_state->prior_non_peer_count + (level_state->peer_count + 1);
  
  	PG_RETURN_INT64(result);
  }
  
  Datum
  cume_dist_final(PG_FUNCTION_ARGS)
  {
  	int64		arg1 = PG_GETARG_INT64(0);		/* prior_non_peer_count +
  												 * peer_count */
  	int64		arg2 = PG_GETARG_INT64(1);		/* partition row count */
  	double		result;
  
  	if (arg1 < 1 || arg2 < 1)
  	{
  		ereport(ERROR,
  				(errcode(ERRCODE_DATA_EXCEPTION),
  				 errmsg("arguments invalid or inconsistent"),
  		   errhint("inappropriate call to window function implementation")));
  		result = 0.0;			/* quieten GCC */
  	}
  	else if (arg1 == 1 && arg2 == 1)
  	{
  		result = 1.0;			/* or is it 0.0? */
  	}
  	else
  	{
  		/* Do division in double, then check for overflow */
  		result = (double)arg1 / (double)arg2;
  		if (result > DBL_MAX)
  		{
  			ereport(ERROR,
  					(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
  					 errmsg("value out of range: overflow")));
  		}
  	}
  
  	PG_RETURN_FLOAT8((float8)result);
  }
  
  
  
  /*
   *	NTILE(BIGINT) OVER (... ORDER BY s) --> BIGINT
   *
   * Implement NTILE for the given WindowRefExprState.
   *
   * ntile_prelim_int(internal,int) --> bigint[]
   * ntile_prelim_bigint(internal,bigint) --> bigint[]
   * ntile_prelim_numeric(internal,numeric) --> bigint[]
   * ntile_final(bigint[],bigint) --> bigint
   */
  
  /* Helper defined in src/backend/util/adt/numeric.c. */
  extern int64 numeric_to_pos_int8_trunc(Numeric num);
  
  /* Helper. */
  static ArrayType *
  do_ntile_prelim(WindowRefExprState * ref_state, int64 num_tiles)
  {
  	/*
  	 * Pack row_index from state and the argument num_tiles into a two-element
  	 * array of type bigint.
  	 */
  
  	Datum		work  [2];
  
  	WindowState *window_state = ref_state->windowstate;
  
  	if (num_tiles <= 0)
  		ereport(ERROR,
  				(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
  				 errmsg("argument value out of range"),
  				 errhint("NTILE expects a positive integer argument.")));
  
  	work[0] = Int64GetDatumFast(num_tiles);		/* ntile argument */
  	work[1] = Int64GetDatumFast(window_state->row_index);
  
  	return construct_array(work, 2, INT8OID, 8, true, 'd');
  }
  
  Datum
  ntile_prelim_int(PG_FUNCTION_ARGS)
  {
  	WindowRefExprState *rstate = (WindowRefExprState *) PG_GETARG_POINTER(0);
  	int64		ntile_arg = (int64) PG_GETARG_INT32(1);
  
  	PG_RETURN_ARRAYTYPE_P(do_ntile_prelim(rstate, ntile_arg));
  }
  
  Datum
  ntile_prelim_bigint(PG_FUNCTION_ARGS)
  {
  	WindowRefExprState *rstate = (WindowRefExprState *) PG_GETARG_POINTER(0);
  	int64		ntile_arg = PG_GETARG_INT64(1);
  
  	PG_RETURN_ARRAYTYPE_P(do_ntile_prelim(rstate, ntile_arg));
  }
  
  Datum
  ntile_prelim_numeric(PG_FUNCTION_ARGS)
  {
  	int64		num_tiles;
  	WindowRefExprState *rstate = (WindowRefExprState *) PG_GETARG_POINTER(0);
  	Numeric		ntile_arg = PG_GETARG_NUMERIC(1);
  
  	/* Truncate ntile_arg to int8, put in num_tiles. */
  	num_tiles = numeric_to_pos_int8_trunc(ntile_arg);
  	if (num_tiles < 1)
  		num_tiles = 0;			/* Set value out of range to trigger error
  								 * report. */
  
  	PG_RETURN_ARRAYTYPE_P(do_ntile_prelim(rstate, num_tiles));
  }
  
  /*
   * To implement the semantics of NTILE correctly, tuples in the same bucket
   * must be clustered together according to the input ordering --
   * i.e., 1, 1, 1, 2, 2, 2, 3, 3, 4, 4.
   *
   * It's tempting to just do ((N - 1) mod M) + 1 but this will not meet
   * our requirement. Alternatively, we could just divide the partition
   * row count into N buckets and see if the row index fit into an individual
   * bucket. That would give us clustered results but it wouldn't handle the
   * situation where the number of rows in the partition did not divide evenly
   * into the number of buckets (notice that in the above example, these
   * larger-by-one buckets are the leading buckets).
   *
   * So, instead we identify the threshold (prefix_size) to which buckets will
   * have one more row and we divide the rows between them (row_index/max_size).
   * After the threshold, the next bucket will be number 1 + spares and
   * we divide the values evenly between each bucket.
   *
   * Note that row_index counts from 0.
   */
  
  Datum
  ntile_final(PG_FUNCTION_ARGS)
  {
  	int64		result;
  	Datum	   *work;
  	int			len;
  	int64		row_index,
  				num_tiles,
  				min_size ,
  				max_size ,
  				prefix_size,
  				spares;
  
  	ArrayType  *pair = PG_GETARG_ARRAYTYPE_P(0);		/* row_index, num_tiles */
  	int64		partition_row_count = PG_GETARG_INT64(1);		/* partition row count */
  
  	/* we expect the input to be bigint[2] */
  	deconstruct_array(pair, INT8OID, 8, true, 'd', &work, NULL, &len);
  	if (len != 2)
  		elog(ERROR, "expected 2-element int8 array");
  
  	num_tiles = DatumGetInt64(work[0]);
  	row_index = DatumGetInt64(work[1]);
  	min_size = partition_row_count / num_tiles;
  
  	spares = partition_row_count % num_tiles;
  	max_size = min_size + 1;
  	prefix_size = spares * max_size;
  
  	if (row_index < prefix_size)
  		result = 1 + row_index / max_size;
  	else
  		result = 1 + spares +
  			(row_index - prefix_size) / min_size;
  
  	PG_RETURN_INT64(result);
  }
  
  /*
   * The common code for lead and lag.
   */
  static Datum
  lead_lag_internal(PG_FUNCTION_ARGS, bool is_lead, bool *isnull)
  {
  	WindowRefExprState *wrxstate =
  	(WindowRefExprState *) PG_GETARG_POINTER(0);
  	WindowState *wstate = wrxstate->windowstate;
  	WindowStatePerFunction funcstate = &wstate->func_state[wrxstate->funcno];
  	WindowStatePerLevel level_state = funcstate->wlevel;
  	WindowBufferCursor cursor;
  	List	   *func_values;
  
  	Insist(PG_NARGS() >= 2);
  	Assert(level_state->is_rows);
  
  	/*
  	 * Check the offset argument if given.	It is actually not used to
  	 * retrieve the value, as the frame is tracking the offset.
  	 */
  	if (PG_NARGS() > 2)
  	{
  		int64		offset = 1;
  
  		if (PG_ARGISNULL(2))
  		{
  			if (is_lead)
  				elog(ERROR, "LEAD offset cannot be NULL");
  			else
  				elog(ERROR, "LAG offset cannot be NULL");
  		}
  
  		offset = PG_GETARG_INT64(2);
  		if (offset < 0)
  			ereport(ERROR,
  					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
  					 errmsg("%s offset cannot be negative",
  							is_lead ? "LEAD" : "LAG")));
  
  	}
  
  	/*
  	 * The direction doesn't matter, as the frame has only one row.
  	 */
  	cursor = windowBufferBeginScan(wstate, level_state, true);
  	func_values = windowBufferNext(cursor);
  
  	/*
  	 * If this is out of frame, return the default value argument, or NULL if
  	 * not given.
  	 */
  	if (!func_values)
  	{
  		if (PG_NARGS() > 3)
  		{
  			funcstate->win_value_is_null = PG_ARGISNULL(3);
  			if (!funcstate->win_value_is_null)
  				funcstate->win_value = PG_GETARG_DATUM(3);
  			else
  				funcstate->win_value = (Datum)0;
  		}
  		else
  		{
  			funcstate->win_value = (Datum)0;
  			funcstate->win_value_is_null = true;
  		}
  	}
  	else
  	{
  		WindowValue *winvalue;
  
  		winvalue = (WindowValue *)
  			list_nth(func_values, funcstate->serial_index);
  
  		if (winvalue)
  		{
  			funcstate->win_value = winvalue->value;
  			funcstate->win_value_is_null = winvalue->valueIsNull;
  		}
  		else
  		{
  			funcstate->win_value = (Datum)0;
  			funcstate->win_value_is_null = true;
  		}
  	}
  	windowBufferEndScan(cursor);
  
  	*isnull = funcstate->win_value_is_null;
  
  	return funcstate->win_value;
  }
  
  /*
   * Initial implementation of LEAD(col, offset, value_expr)
   */
  Datum
  lead_generic(PG_FUNCTION_ARGS)
  {
  	bool		isnull;
  	Datum		d;
  
  	d = lead_lag_internal(fcinfo, true, &isnull);
  
  	if (isnull)
  		PG_RETURN_NULL();
  
  	PG_RETURN_DATUM(d);
  }
  
  Datum
  lag_generic(PG_FUNCTION_ARGS)
  {
  	bool		isnull;
  	Datum		d;
  
  	d = lead_lag_internal(fcinfo, false, &isnull);
  
  	if (isnull)
  		PG_RETURN_NULL();
  
  	PG_RETURN_DATUM(d);
  }
  
  static Datum
  last_value_internal(WindowRefExprState * wrxstate, bool *isnull)
  {
  	WindowState *wstate = wrxstate->windowstate;
  	WindowStatePerFunction funcstate = &wstate->func_state[wrxstate->funcno];
  	WindowStatePerLevel level_state = funcstate->wlevel;
  	FrameBufferEntry *lead_entry = level_state->lead_entry_buf;
  	bool		has_lead_entry = false;
  	WindowValue *lead_value = NULL;
  	bool		lead_valid = true;
  	bool		has_tuples = true;
  	bool		include_last_agg = false;
  
  	has_tuples = hasTuplesInFrame(level_state, wstate);
  
  	if (has_tuples)
  	{
  		if (EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->trail, false))
  			include_last_agg = true;
  
  		if (!EDGE_IS_BOUND(level_state->frame->lead) ||
  			EDGE_IS_BOUND_FOLLOWING(level_state->frame->lead) ||
  			EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->lead, true))
  			include_last_agg = true;
  
  		if (!level_state->agg_filled &&
  			!ntuplestore_acc_tell(level_state->lead_reader, NULL))
  		{
  			lead_valid = false;
  			ntuplestore_acc_seek_last(level_state->lead_reader);
  		}
  
  		if (level_state->is_rows)
  		{
  			if (!EDGE_IS_BOUND_FOLLOWING(level_state->frame->trail) ||
  				level_state->num_lead_rows >= level_state->trail_rows)
  				has_lead_entry = getCurrentValue(level_state->lead_reader,
  												 level_state, lead_entry);
  		}
  		else
  		{
  			has_lead_entry = getCurrentValue(level_state->lead_reader,
  											 level_state, lead_entry);
  		}
  
  		if (has_lead_entry)
  			lead_value = (WindowValue *) list_nth(lead_entry->func_values,
  												  funcstate->serial_index);
  	}
  
  	if (lead_value != NULL)
  	{
  		funcstate->win_value = lead_value->value;
  		funcstate->win_value_is_null = lead_value->valueIsNull;
  	}
  
  	else
  	{
  		if (has_tuples && include_last_agg && level_state->agg_filled)
  		{
  			funcstate->win_value = funcstate->aggTransValue;
  			funcstate->win_value_is_null = funcstate->aggTransValueIsNull;
  		}
  
  		else
  		{
  			funcstate->win_value = 0;
  			funcstate->win_value_is_null = true;
  		}
  	}
  
  	*isnull = funcstate->win_value_is_null;
  
  	if (!lead_valid)
  		ntuplestore_acc_set_invalid(level_state->lead_reader);
  
  	return funcstate->win_value;
  }
  
  Datum
  last_value_generic(PG_FUNCTION_ARGS)
  {
  	Datum		d;
  	bool		isnull = false;
  
  	d = last_value_internal((WindowRefExprState *) PG_GETARG_POINTER(0),
  							&isnull);
  
  	if (isnull)
  		PG_RETURN_NULL();
  
  	return d;
  }
  
  static Datum
  first_value_internal(WindowRefExprState * wrxstate, bool *isnull)
  {
  	WindowState *wstate = wrxstate->windowstate;
  	WindowStatePerFunction funcstate = &wstate->func_state[wrxstate->funcno];
  	WindowStatePerLevel level_state = funcstate->wlevel;
  	FrameBufferEntry *trail_entry = level_state->trail_entry_buf;
  	bool		has_trail_entry = false;
  	WindowValue *trail_value = NULL;
  	NTupleStorePos orig_pos;
  	bool		trail_valid;
  	bool		has_tuples = true;
  
  	has_tuples = hasTuplesInFrame(level_state, wstate);
  
  	if (!has_tuples)
  	{
  		funcstate->win_value = 0;
  		funcstate->win_value_is_null = true;
  
  		*isnull = true;
  		return funcstate->win_value;
  	}
  
  	/* Save the position for the trail_reader */
  	trail_valid = ntuplestore_acc_tell(level_state->trail_reader, &orig_pos);
  
  	/*
  	 * Since the trail_reader points to the value that is right before the
  	 * trailing edge, we advance the trail_reader by one.
  	 */
  	if (trail_valid)
  		ntuplestore_acc_advance(level_state->trail_reader, 1);
  	else
  	{
  		if (!EDGE_IS_BOUND_FOLLOWING(level_state->frame->trail) ||
  			EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->trail, false))
  			ntuplestore_acc_seek_first(level_state->trail_reader);
  	}
  
  	if (level_state->is_rows)
  	{
  		if (!EDGE_IS_BOUND_FOLLOWING(level_state->frame->trail) ||
  			level_state->num_lead_rows >= level_state->trail_rows)
  			has_trail_entry = getCurrentValue(level_state->trail_reader,
  											  level_state, trail_entry);
  	}
  	else
  	{
  		has_trail_entry = getCurrentValue(level_state->trail_reader,
  										  level_state, trail_entry);
  	}
  
  	if (has_trail_entry)
  		trail_value = (WindowValue *) list_nth(trail_entry->func_values,
  											   funcstate->serial_index);
  
  	if (trail_value != NULL)
  	{
  		funcstate->win_value = trail_value->value;
  		funcstate->win_value_is_null = trail_value->valueIsNull;
  	}
  
  	else
  	{
  		bool		include_last_agg = false;
  
  		if (EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->trail, false))
  			include_last_agg = true;
  
  		if (EDGE_EQ_CURRENT_ROW(level_state, wstate, level_state->frame->lead, true))
  			include_last_agg = true;
  
  		if ((has_tuples || include_last_agg || trail_valid) &&
  			level_state->agg_filled)
  		{
  			funcstate->win_value = funcstate->aggTransValue;
  			funcstate->win_value_is_null = funcstate->aggTransValueIsNull;
  		}
  
  		else
  		{
  			funcstate->win_value = 0;
  			funcstate->win_value_is_null = true;
  		}
  	}
  
  	/* Reset the trail_reader to its original position. */
  	if (!ntuplestore_acc_seek(level_state->trail_reader, &orig_pos))
  		ntuplestore_acc_set_invalid(level_state->trail_reader);
  
  	*isnull = funcstate->win_value_is_null;
  
  	return funcstate->win_value;
  }
  
  Datum
  first_value_generic(PG_FUNCTION_ARGS)
  {
  	Datum		d;
  	bool		isnull = false;
  
  	d = first_value_internal((WindowRefExprState *) PG_GETARG_POINTER(0),
  							 &isnull);
  
  	if (isnull)
  		PG_RETURN_NULL();
  
  	return d;
  }
  
  /*
   * a helper function for lead and lag to make frame.
   */
  Datum
  lead_lag_frame_maker(PG_FUNCTION_ARGS)
  {
  	WindowRef  *wref = (WindowRef *) PG_GETARG_POINTER(0);
  	WindowFrame *frame = makeNode(WindowFrame);
  	Node	   *offset;
  	Oid			fnoid;
  	HeapTuple	tuple;
  	Form_pg_window winform;
  	char		winkind;
  	cqContext  *pcqCtx;
  
  	if (list_length(wref->args) > 1)
  		offset = (Node *) list_nth(wref->args, 1);
  	else
  	{
  		offset = (Node *) makeConst(INT8OID,
  									-1,
  									8,
  									Int64GetDatum(1),
  									false,
  									true);
  	}
  
  	frame->system_generated = true;
  	frame->is_between = true;
  	frame->is_rows = true;
  	frame->trail = makeNode(WindowFrameEdge);
  	frame->lead = makeNode(WindowFrameEdge);
  
  	fnoid = wref->winfnoid;
  
  	/* XXX: select winkind from pg_window */
  	pcqCtx = caql_beginscan(
  							NULL,
  							cql("SELECT * FROM pg_window "
  								" WHERE winfnoid = :1 ",
  								ObjectIdGetDatum(fnoid)));
  
  	tuple = caql_getnext(pcqCtx);
  
  	if (!HeapTupleIsValid(tuple))
  		elog(ERROR, "cache lookup failed for window function %u", fnoid);
  	winform = (Form_pg_window)GETSTRUCT(tuple);
  	winkind = winform->winkind;
  	caql_endscan(pcqCtx);
  
  	switch (winkind)
  	{
  		case WINKIND_LAG:
  			if (IsA(offset, Const))
  				frame->trail->kind = frame->lead->kind =
  					WINDOW_BOUND_PRECEDING;
  			else
  				frame->trail->kind = frame->lead->kind =
  					WINDOW_DELAYED_BOUND_PRECEDING;
  			break;
  		case WINKIND_LEAD:
  			if (IsA(offset, Const))
  				frame->trail->kind = frame->lead->kind =
  					WINDOW_BOUND_FOLLOWING;
  			else
  				frame->trail->kind = frame->lead->kind =
  					WINDOW_DELAYED_BOUND_FOLLOWING;
  			break;
  		default:
  			elog(ERROR, "internal window framing error");
  	}
  
  	frame->trail->val = frame->lead->val = offset;
  
  	PG_RETURN_POINTER(frame);
  }
  
  void
  initGpmonPktForWindow(Plan * planNode, gpmon_packet_t * gpmon_pkt, EState *estate)
  {
  	Assert(planNode != NULL && gpmon_pkt != NULL && IsA(planNode, Window));
  
  	{
  		Assert(GPMON_WINDOW_TOTAL <= (int)GPMON_QEXEC_M_COUNT);
  		InitPlanNodeGpmonPkt(planNode, gpmon_pkt, estate, PMNT_Window,
  							 (int64) planNode->plan_rows, NULL);
  	}
  }
  
  void
  ExecEagerFreeWindow(WindowState * node)
  {
  	if (node->input_buffer != NULL)
  	{
  		freeInputBuffer(node);
  	}
  
  	freeFrameBuffers(node);
  }
  