/*
 * Copyright (c) 2011 Teodor Sigaev <teodor@sigaev.ru>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *        notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *        notice, this list of conditions and the following disclaimer in the
 *        documentation and/or other materials provided with the distribution.
 * 3. Neither the name of the author nor the names of any co-contributors
 *        may be used to endorse or promote products derived from this software
 *        without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY CONTRIBUTORS ``AS IS'' AND ANY EXPRESS
 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL CONTRIBUTORS BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE
 * GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER
 * IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN
 * IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

#include "postgres.h"

#include "pgstat.h"
#include "catalog/namespace.h"
#include "commands/vacuum.h"
#include "executor/executor.h"
#include "nodes/nodes.h"
#include "nodes/parsenodes.h"
#include "storage/bufmgr.h"
#include "utils/lsyscache.h"
#include "utils/guc.h"

#ifdef PG_MODULE_MAGIC
PG_MODULE_MAGIC;
#endif

static bool online_analyze_enable = true;
static bool online_analyze_verbose = true;
static double online_analyze_scale_factor = 0.1;
static int online_analyze_threshold = 50;
static double online_analyze_min_interval = 10000;

static ExecutorEnd_hook_type oldhook = NULL;

static void
makeAnalyze(Oid relOid, CmdType operation, uint32 naffected)
{
	PgStat_StatTabEntry		*tabentry;
	TimestampTz 			now = GetCurrentTimestamp();

	if (relOid == InvalidOid)
		return;

	tabentry = pgstat_fetch_stat_tabentry(relOid);

#if PG_VERSION_NUM >= 90000
#define changes_since_analyze(t)	((t)->changes_since_analyze)
#else
#define changes_since_analyze(t)	((t)->n_live_tuples + (t)->n_dead_tuples - (t)->last_anl_tuples)
#endif

	if (    
		tabentry == NULL /* a new table */ ||
		(
			/* do not analyze too often, if both stamps are exceeded the go */
			TimestampDifferenceExceeds(tabentry->analyze_timestamp, now, online_analyze_min_interval) && 
			TimestampDifferenceExceeds(tabentry->autovac_analyze_timestamp, now, online_analyze_min_interval) &&
			/* be in sync with relation_needs_vacanalyze */
			((double)(changes_since_analyze(tabentry) + naffected)) >=
				online_analyze_scale_factor * ((double)(tabentry->n_dead_tuples + tabentry->n_live_tuples)) + 
					(double)online_analyze_threshold
		)
	)
	{
		VacuumStmt				vacstmt;
		TimestampTz				startStamp, endStamp;

		vacstmt.type = T_VacuumStmt;
		vacstmt.freeze_min_age = -1;
		vacstmt.freeze_table_age = -1; /* ??? */
		vacstmt.relation = NULL;
		vacstmt.va_cols = NIL;

#if PG_VERSION_NUM >= 90000
		vacstmt.options = VACOPT_ANALYZE;
		if (online_analyze_verbose)
			vacstmt.options |= VACOPT_VERBOSE;
#else
		vacstmt.vacuum = vacstmt.full = false;
		vacstmt.analyze = true;
		vacstmt.verbose = online_analyze_verbose;
#endif

		if (online_analyze_verbose)
			startStamp = GetCurrentTimestamp();

		analyze_rel(relOid, &vacstmt, GetAccessStrategy(BAS_VACUUM)
#if (PG_VERSION_NUM < 90004) && (PG_VERSION_NUM >= 90000)
			, true
#endif
		);

		if (online_analyze_verbose)
		{
			long	secs;
			int		microsecs;

			endStamp = GetCurrentTimestamp();
			TimestampDifference(startStamp, endStamp, &secs, &microsecs);
			elog(INFO, "analyze \"%s\" took %.02f seconds", get_rel_name(relOid), ((double)secs) + ((double)microsecs)/1.0e6);
		}


		if (tabentry == NULL)
		{
			/* new table */
			pgstat_clear_snapshot();
		}
		else
		{
			/* update last analyze timestamp in local memory of backend */
			tabentry->analyze_timestamp = now;
		}
	}
#if PG_VERSION_NUM >= 90000
	else if (tabentry != NULL)
	{
		tabentry->changes_since_analyze += naffected;
	}
#endif
}

extern PGDLLIMPORT void onlineAnalyzeHooker(QueryDesc *queryDesc);
void
onlineAnalyzeHooker(QueryDesc *queryDesc) 
{
	uint32	naffected = 0;

	if (queryDesc->estate)
		naffected = queryDesc->estate->es_processed;	

	if (online_analyze_enable && queryDesc->plannedstmt &&
			(queryDesc->operation == CMD_INSERT || 
			 queryDesc->operation == CMD_UPDATE ||
			 queryDesc->operation == CMD_DELETE ||
			 (queryDesc->operation == CMD_SELECT && queryDesc->plannedstmt->intoClause)))
	{
		if (queryDesc->plannedstmt->intoClause)
		{
			Oid	relOid = RangeVarGetRelid(queryDesc->plannedstmt->intoClause->rel, true);

			makeAnalyze(relOid, queryDesc->operation, naffected);
		}
		else if (queryDesc->plannedstmt->resultRelations &&
				 queryDesc->plannedstmt->rtable)
		{
			ListCell	*l;

			foreach(l, queryDesc->plannedstmt->resultRelations)
			{
				int 			n = lfirst_int(l);
				RangeTblEntry	*rte = list_nth(queryDesc->plannedstmt->rtable, n-1);
		
				if (rte->rtekind == RTE_RELATION)
					makeAnalyze(rte->relid, queryDesc->operation, naffected);
			}
		}
	}

	if (oldhook)
		(*oldhook)(queryDesc);
	else
		standard_ExecutorEnd(queryDesc);
}

void _PG_init(void);
void
_PG_init(void)
{
	oldhook = ExecutorEnd_hook;

	ExecutorEnd_hook = onlineAnalyzeHooker;

	DefineCustomBoolVariable(
		"online_analyze.enable",
		"Enable on-line analyze",
		"Enables analyze of table directly after insert/update/delete/select into",
		&online_analyze_enable,
#if PG_VERSION_NUM >= 80400
		online_analyze_enable,
#endif
		PGC_USERSET,
#if PG_VERSION_NUM >= 80400
		GUC_NOT_IN_SAMPLE,
#if PG_VERSION_NUM >= 90100
		NULL,
#endif
#endif
		NULL,
		NULL
	);

	DefineCustomBoolVariable(
		"online_analyze.verbose",
		"Verbosity of on-line analyze",
		"Make ANALYZE VERBOSE after table's changes",
		&online_analyze_verbose,
#if PG_VERSION_NUM >= 80400
		online_analyze_verbose,
#endif
		PGC_USERSET,
#if PG_VERSION_NUM >= 80400
		GUC_NOT_IN_SAMPLE,
#if PG_VERSION_NUM >= 90100
		NULL,
#endif
#endif
		NULL,
		NULL
	);

    DefineCustomRealVariable(
		"online_analyze.scale_factor",
		"fraction of table size to start on-line analyze",
		"fraction of table size to start on-line analyze",
		&online_analyze_scale_factor,
#if PG_VERSION_NUM >= 80400
		online_analyze_scale_factor,
#endif
		0.0,
		1.0,
		PGC_USERSET,
#if PG_VERSION_NUM >= 80400
		GUC_NOT_IN_SAMPLE,
#if PG_VERSION_NUM >= 90100
		NULL,
#endif
#endif
		NULL,
		NULL
	);

    DefineCustomIntVariable(
		"online_analyze.threshold",
		"min number of row updates before on-line analyze",
		"min number of row updates before on-line analyze",
		&online_analyze_threshold,
#if PG_VERSION_NUM >= 80400
		online_analyze_threshold,
#endif
		0,
		0x7fffffff,
		PGC_USERSET,
#if PG_VERSION_NUM >= 80400
		GUC_NOT_IN_SAMPLE,
#if PG_VERSION_NUM >= 90100
		NULL,
#endif
#endif
		NULL,
		NULL
	);

    DefineCustomRealVariable(
		"online_analyze.min_interval",
		"minimum time interval between analyze call (in milliseconds)",
		"minimum time interval between analyze call (in milliseconds)",
		&online_analyze_scale_factor,
#if PG_VERSION_NUM >= 80400
		online_analyze_min_interval,
#endif
		0.0,
		1e30,
		PGC_USERSET,
#if PG_VERSION_NUM >= 80400
		GUC_NOT_IN_SAMPLE,
#if PG_VERSION_NUM >= 90100
		NULL,
#endif
#endif
		NULL,
		NULL
	);

}

void _PG_fini(void);
void
_PG_fini(void)
{
	ExecutorEnd_hook = oldhook;
}