#include "postgres.h"
#include "fmgr.h"

#include "funcapi.h"
#include "miscadmin.h"
#include "nodes/primnodes.h"
#include "utils/relcache.h"
#include "catalog/namespace.h"
#include "catalog/pg_am_d.h"
#include "utils/varlena.h"
#include "storage/bufmgr.h"
#include "utils/snapmgr.h"
#include "access/heapam.h"
#include "access/relscan.h"
#include "access/tableam.h"


PG_MODULE_MAGIC;
PG_FUNCTION_INFO_V1(get_ctid);


Datum
get_ctid(PG_FUNCTION_ARGS)
{
    bool		find_max_ctid = 0;
    text 		*relname;
	RangeVar   	*relrv;
	Relation	rel;
	char		ctid[32] = {0};

	if (PG_ARGISNULL(0) || PG_ARGISNULL(1))
		ereport(ERROR,
    			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    			 errmsg("two parameters are required")));

	relname = PG_GETARG_TEXT_PP(0);
	find_max_ctid = PG_GETARG_UINT32(1);

	/* open relation */
	relrv = makeRangeVarFromNameList(textToQualifiedNameList(relname));
	rel = relation_openrv(relrv, AccessShareLock);

	if (RELATION_IS_OTHER_TEMP(rel))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("cannot access temporary tables of other sessions")));

	if (RELKIND_HAS_TABLE_AM(rel->rd_rel->relkind) ||
			rel->rd_rel->relkind == RELKIND_SEQUENCE)
	{
		TableScanDesc scan;
		HeapScanDesc hscan;
		HeapTuple	tuple;
		SnapshotData SnapshotDirty;
		BlockNumber blockNumber;
		OffsetNumber offsetNumber;
		ScanDirection dir;

		if (rel->rd_rel->relam != HEAP_TABLE_AM_OID)
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("only heap AM is supported")));

		/* Disable syncscan because we assume we scan from block zero upwards */
		scan = table_beginscan_strat(rel, SnapshotAny, 0, NULL, true, false);
		hscan = (HeapScanDesc) scan;

		InitDirtySnapshot(SnapshotDirty);

		/* scan the relation for minimum or maximum ctid */
		if (find_max_ctid)
			dir = BackwardScanDirection;
		else
			dir = ForwardScanDirection;

		while ((tuple = heap_getnext(scan, dir)) != NULL)
		{
			CHECK_FOR_INTERRUPTS();

			/* must hold a buffer lock to call HeapTupleSatisfiesVisibility */
			LockBuffer(hscan->rs_cbuf, BUFFER_LOCK_SHARE);

			if (HeapTupleSatisfiesVisibility(tuple, &SnapshotDirty, hscan->rs_cbuf))
			{
				blockNumber = ItemPointerGetBlockNumberNoCheck(&tuple->t_self);
				offsetNumber = ItemPointerGetOffsetNumberNoCheck(&tuple->t_self);

				/* Perhaps someday we should output this as a record. */
				snprintf(ctid, sizeof(ctid), "(%u,%u)", blockNumber, offsetNumber);
				LockBuffer(hscan->rs_cbuf, BUFFER_LOCK_UNLOCK);
				break;
			}
			LockBuffer(hscan->rs_cbuf, BUFFER_LOCK_UNLOCK);
		}

		table_endscan(scan);
		relation_close(rel, AccessShareLock);

		PG_RETURN_CSTRING(pstrdup(ctid));
	}
	else
	{
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("cannot get tuple-level statistics for relation \"%s\"",
						RelationGetRelationName(rel)),
				 errdetail_relkind_not_supported(rel->rd_rel->relkind)));
	}

	PG_RETURN_CSTRING(pstrdup("(0,0)"));	/* should not happen */
}