/*
 * ExecutorRun hook: all we need do is track nesting depth
 */
static void
pgss_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, long count)
{
	nested_level++;
	PG_TRY();
	{
		if (prev_ExecutorRun)
			prev_ExecutorRun(queryDesc, direction, count);
		else
			standard_ExecutorRun(queryDesc, direction, count);
		nested_level--;
	}
	PG_CATCH();
	{
		nested_level--;
		PG_RE_THROW();
	}
	PG_END_TRY();
}

/*
 * ExecutorFinish hook: all we need do is track nesting depth
 */
static void
pgss_ExecutorFinish(QueryDesc *queryDesc)
{
	nested_level++;
	PG_TRY();
	{
		if (prev_ExecutorFinish)
			prev_ExecutorFinish(queryDesc);
		else
			standard_ExecutorFinish(queryDesc);
		nested_level--;
	}
	PG_CATCH();
	{
		nested_level--;
		PG_RE_THROW();
	}
	PG_END_TRY();
}

/*
 * ExecutorRun hook: all we need do is track nesting depth
 */
static void
histogram_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, uint64 count)
{
	nesting_level++;
	PG_TRY();
	{
		if (prev_ExecutorRun)
			prev_ExecutorRun(queryDesc, direction, count);
		else
			standard_ExecutorRun(queryDesc, direction, count);
		nesting_level--;
	}
	PG_CATCH();
	{
		nesting_level--;
		PG_RE_THROW();
	}
	PG_END_TRY();
}

/*
 * run_pg_parse_json:
 *
 * Wrap pg_parse_json in order to restore InterruptHoldoffCount when parse
 * error occured.
 *
 * Returns true when parse completed. False for unexpected end of string.
 */
bool
run_pg_parse_json(JsonLexContext *lex, JsonSemAction *sem)
{
	MemoryContext ccxt = CurrentMemoryContext;
	uint32 saved_IntrHoldoffCount;

	/*
	 * "ereport(ERROR.." occurs on error in pg_parse_json resets
	 * InterruptHoldoffCount to zero, so we must save the value before calling
	 * json parser to restore it on parse error. See errfinish().
	 */
	saved_IntrHoldoffCount = InterruptHoldoffCount;

	PG_TRY();
	{
		pg_parse_json(lex, sem);
	}
	PG_CATCH();
	{
		ErrorData *errdata;
		MemoryContext ecxt;

		InterruptHoldoffCount = saved_IntrHoldoffCount;

		ecxt = MemoryContextSwitchTo(ccxt);
		errdata = CopyErrorData();
		
		if (errdata->sqlerrcode == ERRCODE_INVALID_TEXT_REPRESENTATION)
		{
			FlushErrorState();
			return false;
		}
		else
		{
			MemoryContextSwitchTo(ecxt);
			PG_RE_THROW();
		}
	}
	PG_END_TRY();

	return true;
}

/*
 * ExecutorRun hook: all we need do is track nesting depth
 */
static void
explain_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction,
					uint64 count, bool execute_once)
{
	nesting_level++;
	PG_TRY();
	{
		if (prev_ExecutorRun)
			prev_ExecutorRun(queryDesc, direction, count, execute_once);
		else
			standard_ExecutorRun(queryDesc, direction, count, execute_once);
		nesting_level--;
	}
	PG_CATCH();
	{
		nesting_level--;
		PG_RE_THROW();
	}
	PG_END_TRY();
}

/*
 * Class:     org_postgresql_pljava_internal_Backend
 * Method:    _log
 * Signature: (ILjava/lang/String;)V
 */
JNIEXPORT void JNICALL
JNICALL Java_org_postgresql_pljava_internal_Backend__1log(JNIEnv* env, jclass cls, jint logLevel, jstring jstr)
{
	BEGIN_NATIVE_NO_ERRCHECK
	char* str = String_createNTS(jstr);
	if(str != 0)
	{
		PG_TRY();
		{
			elog(logLevel, "%s", str);
			pfree(str);
		}
		PG_CATCH();
		{
			Exception_throw_ERROR("ereport");
		}
		PG_END_TRY();
	}
	END_NATIVE
}

static Datum
get_row(FunctionCallInfo fcinfo, PlxResult *plx_result, int nrow)
{
    StringInfoData  buf;
    PlxFn          *plx_fn = plx_result->plx_fn;
    PGresult       *pg_result = plx_result->pg_result;
    Datum           ret;

    if (PQgetisnull(pg_result, nrow, 0))
    {
        fcinfo->isnull = true;
        return (Datum)NULL;
    }

    setFixedStringInfo(&buf,
                       PQgetvalue(pg_result, nrow, 0),
                       PQgetlength(pg_result, nrow, 0));

    PG_TRY();
    {
        if (plx_fn->is_binary)
            ret = ReceiveFunctionCall(&plx_fn->ret_type->receive_fn,
                                      &buf,
                                      plx_fn->ret_type->receive_io_params,
                                      plx_fn->ret_type_mod);
        else
            ret = InputFunctionCall(&plx_fn->ret_type->input_fn,
                                    buf.data,
                                    plx_fn->ret_type->receive_io_params,
                                    plx_fn->ret_type_mod);
    }
    PG_CATCH();
    {
        plx_result_cache_delete(fcinfo);
        PQclear(pg_result);
        pfree(plx_result);
        PG_RE_THROW();
    }
    PG_END_TRY();
    return ret;
}

/*
 * Test that the spill_file->file_info->wfile field is initialized to NULL
 * when creating a workfile throws an exception.
 */
void
test__getSpillFile__Initialize_wfile_exception(void **state)
{
	int alloc_size = 0;
	int file_no = 0;
	int branching_factor = 32;
	ExecWorkFile *ewfile = (ExecWorkFile *) palloc0(sizeof(ExecWorkFile));
	workfile_set *work_set = (workfile_set *) palloc0(sizeof(workfile_set));
	SpillSet *spill_set = (SpillSet *) palloc0(sizeof(SpillSet) + (branching_factor-1) * sizeof (SpillFile));

	volatile SpillFile *spill_file = &spill_set->spill_files[0];
	spill_file->file_info = NULL;

	/* Make workfile_mgr_create_file throw an exception, using the side effect function */
	expect_value(workfile_mgr_create_file, work_set, work_set);
	will_return_with_sideeffect(workfile_mgr_create_file, ewfile, &throw_exception_side_effect, NULL);

	PG_TRY();
	{
		/* This function will throw an exception, and we'll catch it below */
		getSpillFile(work_set, spill_set, file_no, &alloc_size);
	}
	PG_CATCH();
	{
		assert_true(spill_file->file_info != NULL);
		assert_int_equal(spill_file->file_info->total_bytes, 0);
		assert_int_equal(spill_file->file_info->ntuples, 0);
		assert_int_equal(alloc_size, 0);

		/*
		 * This is the main test: We must initialize this pointer to NULL, even
		 * if an exception is thrown
		 */
		assert_true(spill_file->file_info->wfile == NULL);
		return;
	}
	PG_END_TRY();

	/* We shouldn't get here, the getSpillFile should throw an exception */
	assert_true(false);
}

static PyObject *
PLy_spi_execute_query(char *query, long limit)
{
	int			rv;
	volatile MemoryContext oldcontext;
	volatile ResourceOwner oldowner;
	PyObject   *ret = NULL;

	oldcontext = CurrentMemoryContext;
	oldowner = CurrentResourceOwner;

	PLy_spi_subtransaction_begin(oldcontext, oldowner);

	PG_TRY();
	{
		PLyExecutionContext	*exec_ctx = PLy_current_execution_context();

		pg_verifymbstr(query, strlen(query), false);
		rv = SPI_execute(query, exec_ctx->curr_proc->fn_readonly, limit);
		ret = PLy_spi_execute_fetch_result(SPI_tuptable, SPI_processed, rv);

		PLy_spi_subtransaction_commit(oldcontext, oldowner);
	}
	PG_CATCH();
	{
		PLy_spi_subtransaction_abort(oldcontext, oldowner);
		return NULL;
	}
	PG_END_TRY();

	if (rv < 0)
	{
		Py_XDECREF(ret);
		PLy_exception_set(PLy_exc_spi_error,
						  "SPI_execute failed: %s",
						  SPI_result_code_string(rv));
		return NULL;
	}

	return ret;
}

/*
 * Report an error we got from the remote server.
 *
 * elevel: error level to use (typically ERROR, but might be less)
 * res: PGresult containing the error
 * clear: if true, PQclear the result (otherwise caller will handle it)
 * sql: NULL, or text of remote command we tried to execute
 *
 * Note: callers that choose not to throw ERROR for a remote error are
 * responsible for making sure that the associated ConnCacheEntry gets
 * marked with have_error = true.
 */
void
pgfdw_report_error(int elevel, PGresult *res, bool clear, const char *sql)
{
	/* If requested, PGresult must be released before leaving this function. */
	PG_TRY();
	{
		char	   *diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
		char	   *message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
		char	   *message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
		char	   *message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
		char	   *message_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
		int			sqlstate;

		if (diag_sqlstate)
			sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
									 diag_sqlstate[1],
									 diag_sqlstate[2],
									 diag_sqlstate[3],
									 diag_sqlstate[4]);
		else
			sqlstate = ERRCODE_CONNECTION_FAILURE;

		ereport(elevel,
				(errcode(sqlstate),
				 message_primary ? errmsg_internal("%s", message_primary) :
				 errmsg("unknown error"),
			   message_detail ? errdetail_internal("%s", message_detail) : 0,
				 message_hint ? errhint("%s", message_hint) : 0,
				 message_context ? errcontext("%s", message_context) : 0,
				 sql ? errcontext("Remote SQL command: %s", sql) : 0));
	}
	PG_CATCH();
	{
		if (clear)
			PQclear(res);
		PG_RE_THROW();
	}
	PG_END_TRY();
	if (clear)
		PQclear(res);
}

/*
 * PLyNumber_ToJsonbValue(PyObject *obj)
 *
 * Transform python number to JsonbValue.
 */
static JsonbValue *
PLyNumber_ToJsonbValue(PyObject *obj, JsonbValue *jbvNum)
{
	Numeric		num;
	char	   *str = PLyObject_AsString(obj);

	PG_TRY();
	{
		Datum		numd;

		numd = DirectFunctionCall3(numeric_in,
								   CStringGetDatum(str),
								   ObjectIdGetDatum(InvalidOid),
								   Int32GetDatum(-1));
		num = DatumGetNumeric(numd);
	}
	PG_CATCH();
	{
		ereport(ERROR,
				(errcode(ERRCODE_DATATYPE_MISMATCH),
				 (errmsg("could not convert value \"%s\" to jsonb", str))));
	}
	PG_END_TRY();

	pfree(str);

	/*
	 * jsonb doesn't allow NaN (per JSON specification), so we have to prevent
	 * it here explicitly.  (Infinity is also not allowed in jsonb, but
	 * numeric_in above already catches that.)
	 */
	if (numeric_is_nan(num))
		ereport(ERROR,
				(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
				 (errmsg("cannot convert NaN to jsonb"))));

	jbvNum->type = jbvNumeric;
	jbvNum->val.numeric = num;

	return jbvNum;
}

/*
 * Special for sending SET commands that change GUC variables, so they go to all
 * gangs, both reader and writer
 */
void
CdbSetGucOnAllGangs(const char *strCommand,
					bool cancelOnError, bool needTwoPhase)
{
	volatile CdbDispatcherState ds = { NULL, NULL };
	const bool	withSnapshot = true;

	elog((Debug_print_full_dtm ? LOG : DEBUG5),
		 "CdbSetGucOnAllGangs for command = '%s', needTwoPhase = %s",
		 strCommand, (needTwoPhase ? "true" : "false"));

	dtmPreCommand("CdbSetGucOnAllGangs", strCommand, NULL, needTwoPhase,
				  withSnapshot, false /* inCursor */ );

	PG_TRY();
	{
		cdbdisp_dispatchSetCommandToAllGangs(strCommand, NULL, 0, NULL, 0,
											 cancelOnError, needTwoPhase,
											 (struct CdbDispatcherState *)
											 &ds);

		/*
		 * Wait for all QEs to finish. If not all of our QEs were successful,
		 * report the error and throw up.
		 */
		cdbdisp_finishCommand((struct CdbDispatcherState *) &ds, NULL, NULL);
	}
	PG_CATCH();
	{
		/*
		 * Something happend, clean up after ourselves
		 */
		CdbCheckDispatchResult((struct CdbDispatcherState *) &ds,
							   DISPATCH_WAIT_CANCEL);

		cdbdisp_destroyDispatcherState((struct CdbDispatcherState *) &ds);

		PG_RE_THROW();
	}
	PG_END_TRY();
}

/*
 * Clears the contents of the entire Global MDVSN specified.
 *
 * Pinned entries are not deleted, but they are marked for removal as soon
 * as the last user releases them. Entries marked for deletion are not returned
 * as valid results during look-ups either.
 */
void
mdver_glob_mdvsn_nuke(void)
{
	Assert(NULL != mdver_glob_mdvsn);

	LWLockAcquire(MDVerWriteLock, LW_EXCLUSIVE);

	PG_TRY();
	{
		int32 num_deleted = Cache_Clear(mdver_glob_mdvsn);
		elog(gp_mdversioning_loglevel, "Nuke at Global MDVSN deleted %d entries", num_deleted);
	}
	PG_CATCH();
	{
		LWLockRelease(MDVerWriteLock);
		PG_RE_THROW();
	}
	PG_END_TRY();

	LWLockRelease(MDVerWriteLock);
}

void InstallHelper_groundwork()
{
	Invocation ctx;
	Invocation_pushInvocation(&ctx, false);
	ctx.function = Function_INIT_WRITER;
	PG_TRY();
	{
		jstring pljlp = String_createJavaStringFromNTS(pljavaLoadPath);
		JNI_callStaticVoidMethod(
			s_InstallHelper_class, s_InstallHelper_groundwork, pljlp,
			pljavaLoadingAsExtension ? JNI_TRUE : JNI_FALSE,
			extensionExNihilo ? JNI_TRUE : JNI_FALSE);
		Invocation_popInvocation(false);
	}
	PG_CATCH();
	{
		Invocation_popInvocation(true);
		PG_RE_THROW();
	}
	PG_END_TRY();
}

/*
 * Class:     org_postgresql_pljava_internal_Relation
 * Method:    _getName
 * Signature: (J)Ljava/lang/String;
 */
JNIEXPORT jstring JNICALL
Java_org_postgresql_pljava_internal_Relation__1getName(JNIEnv* env, jclass clazz, jlong _this)
{
	jstring result = 0;
	Relation self = Invocation_getWrappedPointer(_this);
	if(self != 0)
	{
		BEGIN_NATIVE
		PG_TRY();
		{
			char* relName = SPI_getrelname(self);
			result = String_createJavaStringFromNTS(relName);
			pfree(relName);
		}
		PG_CATCH();
		{
			Exception_throw_ERROR("SPI_getrelname");
		}
		PG_END_TRY();
		END_NATIVE
	}

/*
 * Tests if we don't throw a version mismatch error if the header versions are identical
 */
static void
test__incompatible_module_error__headerversion_identical(void **state)
{
	/* Module magic is identical to ours */
	Pg_magic_struct module_magic = PG_MODULE_MAGIC_DATA;

	/* We should expect a "default" error */
	snprintf(expectedErrorMsg, 255, "Magic block has unexpected length or padding difference.");

	EXPECT_EREPORT(ERROR);

	PG_TRY();
	{
		incompatible_module_error("test", &module_magic);
		assert_true(false);
	}
	PG_CATCH();
	{
	}
	PG_END_TRY();
}

Datum
plperl_call_handler(PG_FUNCTION_ARGS)
{
	Datum		retval;
	plperl_proc_desc *save_prodesc;
	FunctionCallInfo save_caller_info;
	Tuplestorestate *save_tuple_store;
	TupleDesc	save_tuple_desc;

	plperl_init_all();

	save_prodesc = plperl_current_prodesc;
	save_caller_info = plperl_current_caller_info;
	save_tuple_store = plperl_current_tuple_store;
	save_tuple_desc = plperl_current_tuple_desc;

	PG_TRY();
	{
		if (CALLED_AS_TRIGGER(fcinfo))
			retval = PointerGetDatum(plperl_trigger_handler(fcinfo));
		else
			retval = plperl_func_handler(fcinfo);
	}
	PG_CATCH();
	{
		plperl_current_prodesc = save_prodesc;
		plperl_current_caller_info = save_caller_info;
		plperl_current_tuple_store = save_tuple_store;
		plperl_current_tuple_desc = save_tuple_desc;
		PG_RE_THROW();
	}
	PG_END_TRY();

	plperl_current_prodesc = save_prodesc;
	plperl_current_caller_info = save_caller_info;
	plperl_current_tuple_store = save_tuple_store;
	plperl_current_tuple_desc = save_tuple_desc;

	return retval;
}

/*
 * Convert a Python object to a PostgreSQL bytea datum.  This doesn't
 * go through the generic conversion function to circumvent problems
 * with embedded nulls.  And it's faster this way.
 */
static Datum
PLyObject_ToBytea(PLyObToDatum *arg, PyObject *plrv,
				  bool *isnull, bool inarray)
{
	PyObject   *volatile plrv_so = NULL;
	Datum		rv;

	if (plrv == Py_None)
	{
		*isnull = true;
		return (Datum) 0;
	}
	*isnull = false;

	plrv_so = PyObject_Bytes(plrv);
	if (!plrv_so)
		PLy_elog(ERROR, "could not create bytes representation of Python object");

	PG_TRY();
	{
		char	   *plrv_sc = PyBytes_AsString(plrv_so);
		size_t		len = PyBytes_Size(plrv_so);
		size_t		size = len + VARHDRSZ;
		bytea	   *result = palloc(size);

		SET_VARSIZE(result, size);
		memcpy(VARDATA(result), plrv_sc, len);
		rv = PointerGetDatum(result);
	}
	PG_CATCH();
	{
		Py_XDECREF(plrv_so);
		PG_RE_THROW();
	}
	PG_END_TRY();

	Py_XDECREF(plrv_so);

	return rv;
}

/* execute Python code, propagate Python errors to the backend */
static PyObject *
PLy_procedure_call(PLyProcedure *proc, const char *kargs, PyObject *vargs)
{
	PyObject   *rv;
	int volatile save_subxact_level = list_length(explicit_subtransactions);

	PyDict_SetItemString(proc->globals, kargs, vargs);

	PG_TRY();
	{
#if PY_VERSION_HEX >= 0x03020000
		rv = PyEval_EvalCode(proc->code,
							 proc->globals, proc->globals);
#else
		rv = PyEval_EvalCode((PyCodeObject *) proc->code,
							 proc->globals, proc->globals);
#endif

		/*
		 * Since plpy will only let you close subtransactions that you
		 * started, you cannot *unnest* subtransactions, only *nest* them
		 * without closing.
		 */
		Assert(list_length(explicit_subtransactions) >= save_subxact_level);
	}
	PG_CATCH();
	{
		PLy_abort_open_subtransactions(save_subxact_level);
		PG_RE_THROW();
	}
	PG_END_TRY();

	PLy_abort_open_subtransactions(save_subxact_level);

	/* If the Python code returned an error, propagate it */
	if (rv == NULL)
		PLy_elog(ERROR, NULL);

	return rv;
}

/*
 * ResourceOwnerRelease
 *		Release all resources owned by a ResourceOwner and its descendants,
 *		but don't delete the owner objects themselves.
 *
 * Note that this executes just one phase of release, and so typically
 * must be called three times.	We do it this way because (a) we want to
 * do all the recursion separately for each phase, thereby preserving
 * the needed order of operations; and (b) xact.c may have other operations
 * to do between the phases.
 *
 * phase: release phase to execute
 * isCommit: true for successful completion of a query or transaction,
 *			false for unsuccessful
 * isTopLevel: true if completing a main transaction, else false
 *
 * isCommit is passed because some modules may expect that their resources
 * were all released already if the transaction or portal finished normally.
 * If so it is reasonable to give a warning (NOT an error) should any
 * unreleased resources be present.  When isCommit is false, such warnings
 * are generally inappropriate.
 *
 * isTopLevel is passed when we are releasing TopTransactionResourceOwner
 * at completion of a main transaction.  This generally means that *all*
 * resources will be released, and so we can optimize things a bit.
 */
void
ResourceOwnerRelease(ResourceOwner owner,
					 ResourceReleasePhase phase,
					 bool isCommit,
					 bool isTopLevel)
{
	/* Rather than PG_TRY at every level of recursion, set it up once */
	ResourceOwner save;

	save = CurrentResourceOwner;
	PG_TRY();
	{
		ResourceOwnerReleaseInternal(owner, phase, isCommit, isTopLevel);
	}
	PG_CATCH();
	{
		CurrentResourceOwner = save;
		PG_RE_THROW();
	}
	PG_END_TRY();
	CurrentResourceOwner = save;
}