/*-------------------------------------------------------------------------
 *
 * polar_xlog_idx.c
 *   Implementation of parse xlog records.
 *
 * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 * Portions Copyright (c) 2021, Alibaba Group Holding limited
 *
 * src/backend/access/logindex/polar_xlog_idx.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/bufmask.h"
#include "access/polar_fullpage.h"
#include "access/polar_logindex.h"
#include "access/polar_logindex_internal.h"
#include "catalog/pg_control.h"
#include "miscadmin.h"
#include "storage/buf_internals.h"

static bool
xlog_idx_fpi_redo(XLogReaderState *record, BufferTag *tag, Buffer *buffer)
{
	XLogRedoAction action = BLK_NOTFOUND;
	BufferTag	tag0;

	POLAR_GET_LOG_TAG(record, tag0, 0);

	if (BUFFERTAGS_EQUAL(*tag, tag0))
	{
		/*
		 * Full-page image (FPI) records contain nothing else but a backup
		 * block. The block reference must include a full-page image -
		 * otherwise there would be no point in this record.
		 *
		 * No recovery conflicts are generated by these generic records - if a
		 * resource manager needs to generate conflicts, it has to define a
		 * separate WAL record type and redo routine.
		 *
		 * XLOG_FPI_FOR_HINT records are generated when a page needs to be
		 * WAL- logged because of a hint bit update. They are only generated
		 * when checksums are enabled. There is no difference in handling
		 * XLOG_FPI and XLOG_FPI_FOR_HINT records, they use a different info
		 * code just to distinguish them for statistics purposes.
		 */

		action = POLAR_READ_BUFFER_FOR_REDO(record, 0, buffer);
	}

	return action;
}

/*
 * POLAR: fullpage snapshot image wal redo
 */
static bool
xlog_idx_fpsi_redo(XLogReaderState *record, BufferTag *tag, Buffer *buffer)
{
	XLogRedoAction action = BLK_NOTFOUND;
	BufferTag	tag0;
	Page		page;
	uint64		fullpage_no = 0;

	POLAR_GET_LOG_TAG(record, tag0, 0);

	if (BUFFERTAGS_EQUAL(*tag, tag0))
	{
		page = BufferGetPage(*buffer);
		/* get fullpage_no from record */
		memcpy(&fullpage_no, XLogRecGetData(record), sizeof(uint64));
		/* read fullpage from file */
		polar_read_fullpage(page, fullpage_no);
		action = BLK_RESTORED;
	}

	return action;
}

void
polar_xlog_idx_save(XLogReaderState *record)
{
	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;

	if (info != XLOG_FPI &&
		info != XLOG_FPI_FOR_HINT &&
		info != XLOG_FPSI)
		return;

	switch (info)
	{
		case XLOG_FPI:
		case XLOG_FPI_FOR_HINT:
		case XLOG_FPSI:
			polar_log_index_save_block(record, 0);
			break;

		default:
			break;
	}
}

bool
polar_xlog_idx_parse(XLogReaderState *record)
{
	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;

	if (info == XLOG_FPI || info == XLOG_FPI_FOR_HINT)
	{
		polar_log_index_redo_parse(record, 0);
		return true;
	}
	else if (info == XLOG_FPSI)
	{
		uint64		fullpage_no = 0;

		/* get fullpage_no from record */
		memcpy(&fullpage_no, XLogRecGetData(record), sizeof(uint64));
		/* Update max_fullpage_no */
		polar_update_max_fullpage_no(fullpage_no);
		return true;
	}

	return false;
}

XLogRedoAction
polar_xlog_idx_redo(XLogReaderState *record, BufferTag *tag, Buffer *buffer)
{
	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;

	/*
	 * These operations don't overwrite MVCC data so no conflict processing is
	 * required. The ones in heap2 rmgr do.
	 */

	switch (info)
	{
		case XLOG_FPI:
		case XLOG_FPI_FOR_HINT:
			return xlog_idx_fpi_redo(record, tag, buffer);

		case XLOG_FPSI:
			return xlog_idx_fpsi_redo(record, tag, buffer);

		default:
			elog(PANIC, "polar_xlog_idx_redo: unknown op code %u", info);
			break;
	}

	return BLK_NOTFOUND;
}
