/*-------------------------------------------------------------------------

 *

 * nodeTidrangescan.c

 *    Routines to support TID range scans of relations

 *

 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group

 * Portions Copyright (c) 1994, Regents of the University of California

 *

 *

 * IDENTIFICATION

 *    src/backend/executor/nodeTidrangescan.c

 *

 *-------------------------------------------------------------------------

 */

#include "postgres.h"

#include "access/relscan.h"

#include "access/sysattr.h"

#include "access/tableam.h"

#include "catalog/pg_operator.h"

#include "executor/executor.h"

#include "executor/nodeTidrangescan.h"

#include "nodes/nodeFuncs.h"

#include "utils/rel.h"

/*

 * It's sufficient to check varattno to identify the CTID variable, as any

 * Var in the relation scan qual must be for our table.  (Even if it's a

 * parameterized scan referencing some other table's CTID, the other table's

 * Var would have become a Param by the time it gets here.)

 */

#define IsCTIDVar(node)  \

  ((node) != NULL && \

   IsA((node), Var) && \

   ((Var *) (node))->varattno == SelfItemPointerAttributeNumber)

typedef enum

{

  TIDEXPR_UPPER_BOUND,

  TIDEXPR_LOWER_BOUND,

} TidExprType;

/* Upper or lower range bound for scan */

typedef struct TidOpExpr

{

  TidExprType exprtype;   /* type of op; lower or upper */

  ExprState  *exprstate;    /* ExprState for a TID-yielding subexpr */

  bool    inclusive;    /* whether op is inclusive */

} TidOpExpr;

/*

 * For the given 'expr', build and return an appropriate TidOpExpr taking into

 * account the expr's operator and operand order.

 */

static TidOpExpr *

MakeTidOpExpr(OpExpr *expr, TidRangeScanState *tidstate)

{

  Node     *arg1 = get_leftop((Expr *) expr);

  Node     *arg2 = get_rightop((Expr *) expr);

  ExprState  *exprstate = NULL;

  bool    invert = false;

  TidOpExpr  *tidopexpr;

  if (IsCTIDVar(arg1))

    exprstate = ExecInitExpr((Expr *) arg2, &tidstate->ss.ps);

  else if (IsCTIDVar(arg2))

  {

    exprstate = ExecInitExpr((Expr *) arg1, &tidstate->ss.ps);

    invert = true;

  }

  else

    elog(ERROR, "could not identify CTID variable");

  tidopexpr = (TidOpExpr *) palloc(sizeof(TidOpExpr));

  tidopexpr->inclusive = false;  /* for now */

  switch (expr->opno)

  {

    case TIDLessEqOperator:

      tidopexpr->inclusive = true;

      /* fall through */

    case TIDLessOperator:

      tidopexpr->exprtype = invert ? TIDEXPR_LOWER_BOUND : TIDEXPR_UPPER_BOUND;

      break;

    case TIDGreaterEqOperator:

      tidopexpr->inclusive = true;

      /* fall through */

    case TIDGreaterOperator:

      tidopexpr->exprtype = invert ? TIDEXPR_UPPER_BOUND : TIDEXPR_LOWER_BOUND;

      break;

    default:

      elog(ERROR, "could not identify CTID operator");

  }

  tidopexpr->exprstate = exprstate;

  return tidopexpr;

}

/*

 * Extract the qual subexpressions that yield TIDs to search for,

 * and compile them into ExprStates if they're ordinary expressions.

 */

static void

TidExprListCreate(TidRangeScanState *tidrangestate)

{

  TidRangeScan *node = (TidRangeScan *) tidrangestate->ss.ps.plan;

  List     *tidexprs = NIL;

  ListCell   *l;

  foreach(l, node->tidrangequals)

  {

    OpExpr     *opexpr = lfirst(l);

    TidOpExpr  *tidopexpr;

    if (!IsA(opexpr, OpExpr))

      elog(ERROR, "could not identify CTID expression");

    tidopexpr = MakeTidOpExpr(opexpr, tidrangestate);

    tidexprs = lappend(tidexprs, tidopexpr);

  }

  tidrangestate->trss_tidexprs = tidexprs;

}

/* ----------------------------------------------------------------

 *    TidRangeEval

 *

 *    Compute and set node's block and offset range to scan by evaluating

 *    node->trss_tidexprs.  Returns false if we detect the range cannot

 *    contain any tuples.  Returns true if it's possible for the range to

 *    contain tuples.  We don't bother validating that trss_mintid is less

 *    than or equal to trss_maxtid, as the scan_set_tidrange() table AM

 *    function will handle that.

 * ----------------------------------------------------------------

 */

static bool

TidRangeEval(TidRangeScanState *node)

{

  ExprContext *econtext = node->ss.ps.ps_ExprContext;

  ItemPointerData lowerBound;

  ItemPointerData upperBound;

  ListCell   *l;

  /*

   * Set the upper and lower bounds to the absolute limits of the range of

   * the ItemPointer type.  Below we'll try to narrow this range on either

   * side by looking at the TidOpExprs.

   */

  ItemPointerSet(&lowerBound, 0, 0);

  ItemPointerSet(&upperBound, InvalidBlockNumber, PG_UINT16_MAX);

  foreach(l, node->trss_tidexprs)

  {

    TidOpExpr  *tidopexpr = (TidOpExpr *) lfirst(l);

    ItemPointer itemptr;

    bool    isNull;

    /* Evaluate this bound. */

    itemptr = (ItemPointer)

      DatumGetPointer(ExecEvalExprSwitchContext(tidopexpr->exprstate,

                            econtext,

                            &isNull));

    /* If the bound is NULL, *nothing* matches the qual. */

    if (isNull)

      return false;

    if (tidopexpr->exprtype == TIDEXPR_LOWER_BOUND)

    {

      ItemPointerData lb;

      ItemPointerCopy(itemptr, &lb);

      /*

       * Normalize non-inclusive ranges to become inclusive.  The

       * resulting ItemPointer here may not be a valid item pointer.

       */

      if (!tidopexpr->inclusive)

        ItemPointerInc(&lb);

      /* Check if we can narrow the range using this qual */

      if (ItemPointerCompare(&lb, &lowerBound) > 0)

        ItemPointerCopy(&lb, &lowerBound);

    }

    else if (tidopexpr->exprtype == TIDEXPR_UPPER_BOUND)

    {

      ItemPointerData ub;

      ItemPointerCopy(itemptr, &ub);

      /*

       * Normalize non-inclusive ranges to become inclusive.  The

       * resulting ItemPointer here may not be a valid item pointer.

       */

      if (!tidopexpr->inclusive)

        ItemPointerDec(&ub);

      /* Check if we can narrow the range using this qual */

      if (ItemPointerCompare(&ub, &upperBound) < 0)

        ItemPointerCopy(&ub, &upperBound);

    }

  }

  ItemPointerCopy(&lowerBound, &node->trss_mintid);

  ItemPointerCopy(&upperBound, &node->trss_maxtid);

  return true;

}

/* ----------------------------------------------------------------

 *    TidRangeNext

 *

 *    Retrieve a tuple from the TidRangeScan node's currentRelation

 *    using the TIDs in the TidRangeScanState information.

 *

 * ----------------------------------------------------------------

 */

static TupleTableSlot *

TidRangeNext(TidRangeScanState *node)

{

  TableScanDesc scandesc;

  EState     *estate;

  ScanDirection direction;

  TupleTableSlot *slot;

  /*

   * extract necessary information from TID scan node

   */

  scandesc = node->ss.ss_currentScanDesc;

  estate = node->ss.ps.state;

  slot = node->ss.ss_ScanTupleSlot;

  direction = estate->es_direction;

  if (!node->trss_inScan)

  {

    /* First time through, compute TID range to scan */

    if (!TidRangeEval(node))

      return NULL;

    if (scandesc == NULL)

    {

      scandesc = table_beginscan_tidrange(node->ss.ss_currentRelation,

                        estate->es_snapshot,

                        &node->trss_mintid,

                        &node->trss_maxtid);

      node->ss.ss_currentScanDesc = scandesc;

    }

    else

    {

      /* rescan with the updated TID range */

      table_rescan_tidrange(scandesc, &node->trss_mintid,

                  &node->trss_maxtid);

    }

    node->trss_inScan = true;

  }

  /* Fetch the next tuple. */

  if (!table_scan_getnextslot_tidrange(scandesc, direction, slot))

  {

    node->trss_inScan = false;

    ExecClearTuple(slot);

  }

  return slot;

}

/*

 * TidRangeRecheck -- access method routine to recheck a tuple in EvalPlanQual

 */

static bool

TidRangeRecheck(TidRangeScanState *node, TupleTableSlot *slot)

{

  if (!TidRangeEval(node))

    return false;

  Assert(ItemPointerIsValid(&slot->tts_tid));

  /* Recheck the ctid is still within range */

  if (ItemPointerCompare(&slot->tts_tid, &node->trss_mintid) < 0 ||

    ItemPointerCompare(&slot->tts_tid, &node->trss_maxtid) > 0)

    return false;

  return true;

}

/* ----------------------------------------------------------------

 *    ExecTidRangeScan(node)

 *

 *    Scans the relation using tids and returns the next qualifying tuple.

 *    We call the ExecScan() routine and pass it the appropriate

 *    access method functions.

 *

 *    Conditions:

 *      -- the "cursor" maintained by the AMI is positioned at the tuple

 *       returned previously.

 *

 *    Initial States:

 *      -- the relation indicated is opened for TID range scanning.

 * ----------------------------------------------------------------

 */

static TupleTableSlot *

ExecTidRangeScan(PlanState *pstate)

{

  TidRangeScanState *node = castNode(TidRangeScanState, pstate);

  return ExecScan(&node->ss,

          (ExecScanAccessMtd) TidRangeNext,

          (ExecScanRecheckMtd) TidRangeRecheck);

}

/* ----------------------------------------------------------------

 *    ExecReScanTidRangeScan(node)

 * ----------------------------------------------------------------

 */

void

ExecReScanTidRangeScan(TidRangeScanState *node)

{

  /* mark scan as not in progress, and tid range list as not computed yet */

  node->trss_inScan = false;

  /*

   * We must wait until TidRangeNext before calling table_rescan_tidrange.

   */

  ExecScanReScan(&node->ss);

}

/* ----------------------------------------------------------------

 *    ExecEndTidRangeScan

 *

 *    Releases any storage allocated through C routines.

 *    Returns nothing.

 * ----------------------------------------------------------------

 */

void

ExecEndTidRangeScan(TidRangeScanState *node)

{

  TableScanDesc scan = node->ss.ss_currentScanDesc;

  if (scan != NULL)

    table_endscan(scan);

}

/* ----------------------------------------------------------------

 *    ExecInitTidRangeScan

 *

 *    Initializes the tid range scan's state information, creates

 *    scan keys, and opens the scan relation.

 *

 *    Parameters:

 *      node: TidRangeScan node produced by the planner.

 *      estate: the execution state initialized in InitPlan.

 * ----------------------------------------------------------------

 */

TidRangeScanState *

ExecInitTidRangeScan(TidRangeScan *node, EState *estate, int eflags)

{

  TidRangeScanState *tidrangestate;

  Relation  currentRelation;

  /*

   * create state structure

   */

  tidrangestate = makeNode(TidRangeScanState);

  tidrangestate->ss.ps.plan = (Plan *) node;

  tidrangestate->ss.ps.state = estate;

  tidrangestate->ss.ps.ExecProcNode = ExecTidRangeScan;

  /*

   * Miscellaneous initialization

   *

   * create expression context for node

   */

  ExecAssignExprContext(estate, &tidrangestate->ss.ps);

  /*

   * mark scan as not in progress, and TID range as not computed yet

   */

  tidrangestate->trss_inScan = false;

  /*

   * open the scan relation

   */

  currentRelation = ExecOpenScanRelation(estate, node->scan.scanrelid, eflags);

  tidrangestate->ss.ss_currentRelation = currentRelation;

  tidrangestate->ss.ss_currentScanDesc = NULL; /* no table scan here */

  /*

   * get the scan type from the relation descriptor.

   */

  ExecInitScanTupleSlot(estate, &tidrangestate->ss,

              RelationGetDescr(currentRelation),

              table_slot_callbacks(currentRelation));

  /*

   * Initialize result type and projection.

   */

  ExecInitResultTypeTL(&tidrangestate->ss.ps);

  ExecAssignScanProjectionInfo(&tidrangestate->ss);

  /*

   * initialize child expressions

   */

  tidrangestate->ss.ps.qual =

    ExecInitQual(node->scan.plan.qual, (PlanState *) tidrangestate);

  TidExprListCreate(tidrangestate);

  /*

   * all done.

   */

  return tidrangestate;

}