/*
 * Find the compression implementation (in pg_compression) for a particular
 * compression type.
 *
 * Comparison is case insensitive.
 */
PGFunction *
GetCompressionImplementation(char *comptype)
{
	HeapTuple		tuple;
	NameData		compname;
	PGFunction	   *funcs;
	Form_pg_compression ctup;
	FmgrInfo		finfo;
	Relation	comprel;
	ScanKeyData	scankey;
	SysScanDesc scan;

	comprel = heap_open(CompressionRelationId, AccessShareLock);

	compname = comptype_to_name(comptype);

	/* SELECT * FROM pg_compression WHERE compname = :1 */
	ScanKeyInit(&scankey,
				Anum_pg_compression_compname,
				BTEqualStrategyNumber, F_NAMEEQ,
				NameGetDatum(&compname));

	scan = systable_beginscan(comprel, CompressionCompnameIndexId, true,
							  SnapshotNow, 1, &scankey);
	tuple = systable_getnext(scan);
	if (!HeapTupleIsValid(tuple))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("unknown compress type \"%s\"",
						comptype)));

	funcs = palloc0(sizeof(PGFunction) * NUM_COMPRESS_FUNCS);

	ctup = (Form_pg_compression)GETSTRUCT(tuple);

	Insist(OidIsValid(ctup->compconstructor));
	fmgr_info(ctup->compconstructor, &finfo);
	funcs[COMPRESSION_CONSTRUCTOR] = finfo.fn_addr;

	Insist(OidIsValid(ctup->compdestructor));
	fmgr_info(ctup->compdestructor, &finfo);
	funcs[COMPRESSION_DESTRUCTOR] = finfo.fn_addr;

	Insist(OidIsValid(ctup->compcompressor));
	fmgr_info(ctup->compcompressor, &finfo);
	funcs[COMPRESSION_COMPRESS] = finfo.fn_addr;

	Insist(OidIsValid(ctup->compdecompressor));
	fmgr_info(ctup->compdecompressor, &finfo);
	funcs[COMPRESSION_DECOMPRESS] = finfo.fn_addr;

	Insist(OidIsValid(ctup->compvalidator));
	fmgr_info(ctup->compvalidator, &finfo);
	funcs[COMPRESSION_VALIDATOR] = finfo.fn_addr;

	systable_endscan(scan);
	heap_close(comprel, AccessShareLock);

	return funcs;
}

/*
 * Add a single attribute encoding entry.
 */
static void
add_attribute_encoding_entry(Oid relid, AttrNumber attnum, Datum attoptions)
{
	Datum values[Natts_pg_attribute_encoding];
	bool nulls[Natts_pg_attribute_encoding];
	HeapTuple tuple;
	cqContext	   *pcqCtx;
	
	Insist(!gp_upgrade_mode);
	Insist(attnum != InvalidAttrNumber);
	
	pcqCtx = caql_beginscan(
			NULL,
			cql("INSERT INTO pg_attribute_encoding",
				NULL));

	MemSet(nulls, 0, sizeof(nulls));
	values[Anum_pg_attribute_encoding_attrelid - 1] = ObjectIdGetDatum(relid);
	values[Anum_pg_attribute_encoding_attnum - 1] = Int16GetDatum(attnum);
	values[Anum_pg_attribute_encoding_attoptions - 1] = attoptions;

	tuple = caql_form_tuple(pcqCtx, values, nulls);

	/* insert a new tuple */
	caql_insert(pcqCtx, tuple); /* implicit update of index as well */

	heap_freetuple(tuple);

	caql_endscan(pcqCtx);
}

postDTMRecv_dbTblSpc_Hash_Entry *
Persistent_PostDTMRecv_InsertHashEntry(Oid dbId, postDTMRecv_dbTblSpc_Hash_Entry *values, bool *exists)
{
	bool	foundPtr;
	postDTMRecv_dbTblSpc_Hash_Entry *entry;
	Insist(PT_PostDTMRecv_Info);
	Insist(PT_PostDTMRecv_Info->postDTMRecv_dbTblSpc_Hash != NULL);

	entry = (postDTMRecv_dbTblSpc_Hash_Entry *) hash_search(
										PT_PostDTMRecv_Info->postDTMRecv_dbTblSpc_Hash,
										(void *) &dbId,
										HASH_ENTER,
										&foundPtr);
	if (entry == NULL)
	{
		ereport(ERROR, (errcode(ERRCODE_OUT_OF_MEMORY),
				(errmsg("Not enough shared memory"))));
	}

	if (foundPtr)
		*exists = TRUE;
	else
	{
		*exists = FALSE;
		entry->database = values->database;
		entry->tablespace = values->tablespace;
		elog(LOG,"Added %d database %d tablespace to Hash",entry->database, entry->tablespace);
	}

	return entry;
}

Datum
zlib_constructor(PG_FUNCTION_ARGS)
{
	TupleDesc		 td	   = PG_GETARG_POINTER(0);
	StorageAttributes *sa = PG_GETARG_POINTER(1);
	CompressionState *cs	   = palloc0(sizeof(CompressionState));
	zlib_state	   *state	= palloc0(sizeof(zlib_state));
	bool			  compress = PG_GETARG_BOOL(2);

	cs->opaque = (void *) state;
	cs->desired_sz = NULL;

	Insist(PointerIsValid(td));
	Insist(PointerIsValid(sa->comptype));

	if (sa->complevel == 0)
		sa->complevel = 1;

	state->level = sa->complevel;
	state->compress = compress;
	state->compress_fn = compress2;
	state->decompress_fn = uncompress;

	PG_RETURN_POINTER(cs);

}

/* ---------------------
 * addProcCallback() - Add a new callback to pg_proc_callback
 *
 * Parameters:
 *    profnoid    - oid of the function that has a callback
 *    procallback - oid of the callback function
 *    promethod   - role the callback function plays
 *
 * Notes:
 *    This function does not maintain dependencies in pg_depend, that behavior
 *    is currently controlled in pg_proc.c
 * ---------------------
 */
void 
addProcCallback(Oid profnoid, Oid procallback, char promethod)
{
	Relation	rel;
	bool		nulls[Natts_pg_proc_callback];
	Datum		values[Natts_pg_proc_callback];
	HeapTuple   tup;
	
	Insist(OidIsValid(profnoid));
	Insist(OidIsValid(procallback));

	/* open pg_proc_callback */
	rel = heap_open(ProcCallbackRelationId, RowExclusiveLock);

	/* Build the tuple and insert it */
	nulls[Anum_pg_proc_callback_profnoid - 1]	  = false;
	nulls[Anum_pg_proc_callback_procallback - 1]  = false;
	nulls[Anum_pg_proc_callback_promethod - 1]	  = false;
	values[Anum_pg_proc_callback_profnoid - 1]	  = ObjectIdGetDatum(profnoid);
	values[Anum_pg_proc_callback_procallback - 1] = ObjectIdGetDatum(procallback);
	values[Anum_pg_proc_callback_promethod - 1]	  = CharGetDatum(promethod);

	tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
	
	/* Insert tuple into the relation */
	simple_heap_insert(rel, tup);
	CatalogUpdateIndexes(rel, tup);

	heap_close(rel, RowExclusiveLock);
}

Datum
snappy_decompress_internal(PG_FUNCTION_ARGS)
{
	const char		*src	= PG_GETARG_POINTER(0);
	size_t			src_sz = PG_GETARG_INT32(1);
	char			*dst	= PG_GETARG_POINTER(2);
	int32			dst_sz = PG_GETARG_INT32(3);
	int32			*dst_used = PG_GETARG_POINTER(4);
	size_t			uncompressed_length;
	snappy_status	retval;

	Insist(src_sz > 0 && dst_sz > 0);

	retval = snappy_uncompressed_length((char *) src, (size_t) src_sz,
										&uncompressed_length);
	if (retval != SNAPPY_OK)
		elog_snappy_error(retval, "snappy_uncompressed_length",
						  src_sz, dst_sz, *dst_used);

	Insist(dst_sz >= uncompressed_length);

	retval = snappy_uncompress((char *) src, src_sz, (char *) dst,
							   &uncompressed_length);
	*dst_used = uncompressed_length;

	if (retval != SNAPPY_OK)
		elog_snappy_error(retval, "snappy_uncompressed",
						  src_sz, dst_sz, *dst_used);

	PG_RETURN_VOID();
}

static void checkExpectations(void)
{
  Insist(!CallbackNew.shouldBeCalled);
  Insist(!CallbackDelete.shouldBeCalled);
  Insist(!CallbackGrow.shouldBeCalled);
  Insist(!CallbackShrink.shouldBeCalled);
} 

/*
 *  Create metadata hdfs file block locations from original hdfs block locations
 */
BlockLocation *
CreateHdfsFileBlockLocations(BlockLocation *hdfs_locations, int block_num)
{
    Insist(hdfs_locations != NULL);
    Insist(block_num > 0);

    int i, j;
    BlockLocation *locations = NULL;

    locations = (BlockLocation *)palloc(sizeof(BlockLocation) * block_num);
    if (NULL == locations)
    {
        return NULL;
    }
    for (i=0;i<block_num;i++)
    {
        locations[i].corrupt = 0;
        locations[i].numOfNodes = hdfs_locations[i].numOfNodes; 

        locations[i].hosts = (char **)palloc(sizeof(char *) * locations[i].numOfNodes);
        locations[i].names= (char **)palloc(sizeof(char *) * locations[i].numOfNodes);
        locations[i].topologyPaths = (char **)palloc(sizeof(char *) * locations[i].numOfNodes);
        for (j=0;j<locations[i].numOfNodes;j++)
        {
            locations[i].hosts[j] = pstrdup(hdfs_locations[i].hosts[j]);
            locations[i].names[j] = pstrdup(hdfs_locations[i].names[j]);
            locations[i].topologyPaths[j] = pstrdup(hdfs_locations[i].topologyPaths[j]);
        }
        locations[i].length = hdfs_locations[i].length;
        locations[i].offset = hdfs_locations[i].offset;
    }

    return locations;
}

static void check(FBMState state)
{
  CheckFBMClosureStruct closure;

  closure.state = state;
  closure.limit = addrOfIndex(state, ArraySize);
  closure.oldLimit = state->block;

  switch (state->type) {
  case FBMTypeCBS:
    CBSIterate(state->the.cbs, checkCBSCallback, (void *)&closure, 0);
    break;
  case FBMTypeFreelist:
    FreelistIterate(state->the.fl, checkFLCallback, (void *)&closure, 0);
    break;
  default:
    fail();
    return;
  }

  if (closure.oldLimit == state->block)
    Insist(BTIsSetRange(state->allocTable, 0,
                        indexOfAddr(state, closure.limit)));
  else if (closure.limit > closure.oldLimit)
    Insist(BTIsSetRange(state->allocTable,
                        indexOfAddr(state, closure.oldLimit),
                        indexOfAddr(state, closure.limit)));
  else
    Insist(closure.oldLimit == closure.limit);
}

static Bool checkCBSAction(CBS cbs, CBSBlock cbsBlock, void *p)
{
  Addr base, limit;
  CheckCBSClosure closure = (CheckCBSClosure)p;

  /* Don't need to check cbs every time */
  UNUSED(cbs);
  Insist(closure != NULL);

  base = CBSBlockBase(cbsBlock);
  limit = CBSBlockLimit(cbsBlock);

  if (base > closure->oldLimit) {
    Insist(BTIsSetRange(closure->allocTable,
                      indexOfAddr(closure->base, closure->oldLimit),
                      indexOfAddr(closure->base, base)));
  } else { /* must be at start of table */
    Insist(base == closure->oldLimit);
    Insist(closure->oldLimit == closure->base);
  }
 
  Insist(BTIsResRange(closure->allocTable,
                      indexOfAddr(closure->base, base),
                      indexOfAddr(closure->base, limit)));


  closure->oldLimit = limit;

  return TRUE;
}

char *
serialize_filesystem_credentials(int *size)
{
	HASH_SEQ_STATUS status;
	struct FileSystemCredential *entry;
	StringInfoData buffer;

    HTAB * currentFilesystemCredentials;
    MemoryContext currentFilesystemCredentialsMemoryContext;

    get_current_credential_cache_and_memcxt(&currentFilesystemCredentials,
            &currentFilesystemCredentialsMemoryContext);

    Insist(NULL != currentFilesystemCredentials);
    Insist(NULL != currentFilesystemCredentialsMemoryContext);


	initStringInfo(&buffer);
	hash_seq_init(&status, currentFilesystemCredentials);

	while (NULL != (entry = hash_seq_search(&status)))
		serialize_filesystem_credential(&buffer, entry);

	*size = buffer.len;
	return buffer.data;
}

static Bool checkCallback(Range range, void *closureP, Size closureS)
{
  Addr base, limit;
  CheckFBMClosure cl = (CheckFBMClosure)closureP;

  UNUSED(closureS);
  Insist(cl != NULL);

  base = RangeBase(range);
  limit = RangeLimit(range);

  if (base > cl->oldLimit) {
    Insist(BTIsSetRange(cl->state->allocTable,
                        indexOfAddr(cl->state, cl->oldLimit),
                        indexOfAddr(cl->state, base)));
  } else { /* must be at start of table */
    Insist(base == cl->oldLimit);
    Insist(cl->oldLimit == cl->state->block);
  }
 
  Insist(BTIsResRange(cl->state->allocTable,
                      indexOfAddr(cl->state, base),
                      indexOfAddr(cl->state, limit)));

  cl->oldLimit = limit;

  return TRUE;
}

/*
 * 
 * FileRepPrimary_RunHeartBeat()
 *
 *
 */
static void 
FileRepPrimary_RunHeartBeat(void)
{	
	int retry = 0;
	
	Insist(fileRepRole == FileRepPrimaryRole);
	
	Insist(dataState == DataStateInSync ||
		   dataState == DataStateInResync);	
	
	while (1) 
	{
		FileRepSubProcess_ProcessSignals();
		
		while (FileRepSubProcess_GetState() == FileRepStateFault ||
			   
			   (fileRepShmemArray[0]->state == FileRepStateNotInitialized &&
				FileRepSubProcess_GetState() != FileRepStateShutdownBackends &&
			    FileRepSubProcess_GetState() != FileRepStateShutdown)) {
			
			FileRepSubProcess_ProcessSignals();
			pg_usleep(50000L); /* 50 ms */	
		}
		
		if (FileRepSubProcess_GetState() == FileRepStateShutdown ||
			FileRepSubProcess_GetState() == FileRepStateShutdownBackends) {
			
			break;
		}

		/* verify if flow from primary to mirror and back is alive once per minute */
		pg_usleep(50000L); /* 50 ms */
		
		if (FileRepSubProcess_ProcessSignals() == true ||
			FileRepSubProcess_GetState() == FileRepStateFault)
		{
			continue;
		}
		
		retry++;
		if (retry == 1200) /* 1200 * 50 ms = 60 sec */
		{
			FileRepPrimary_MirrorHeartBeat(FileRepMessageTypeXLog);
			continue;
		}
		
		if (retry == 1201) /* 1200 * 50 ms = 60 sec */
		{
			FileRepPrimary_MirrorHeartBeat(FileRepMessageTypeWriter);
			continue;
		}
		
		if (retry == 1202) /* 1200 * 50 ms = 60 sec */
		{
			FileRepPrimary_MirrorHeartBeat(FileRepMessageTypeAO01);
			retry = 0;
		}
	} // while(1)	
}

bool
Persistent_PostDTMRecv_IsHashFull(void)
{
	Insist(PT_PostDTMRecv_Info);
	Insist(PT_PostDTMRecv_Info->postDTMRecv_dbTblSpc_Hash);

	return (hash_get_num_entries(PT_PostDTMRecv_Info->postDTMRecv_dbTblSpc_Hash) == PT_MAX_NUM_POSTDTMRECV_DB);
}

static void roots_stepper(mps_addr_t *ref, mps_root_t root, void *p, size_t s)
{
  roots_stepper_data_t data = p;
  Insist(ref != NULL);
  Insist(p != NULL);
  Insist(s == sizeof *data);
  Insist(root == data->exactRoot);
  ++ data->count;
}

static void cbsNewCallback(CBS cbs, CBSBlock cbsBlock,
                           Size oldSize, Size newSize)
{
  testCallback(cbs, cbsBlock, oldSize, newSize, &CallbackNew);
  Insist(oldSize < cbs->minSize);
  Insist(newSize >= cbs->minSize);

  NNewBlocks++;
}

static void cbsDeleteCallback(CBS cbs, CBSBlock cbsBlock,
                              Size oldSize, Size newSize)
{
  testCallback(cbs, cbsBlock, oldSize, newSize, &CallbackDelete);
  Insist(oldSize >= cbs->minSize);
  Insist(newSize < cbs->minSize);

  NDeleteBlocks++;
}

Datum
zlib_decompress(PG_FUNCTION_ARGS)
{
	const char	   *src	= PG_GETARG_POINTER(0);
	int32			src_sz = PG_GETARG_INT32(1);
	void		   *dst	= PG_GETARG_POINTER(2);
	int32			dst_sz = PG_GETARG_INT32(3);
	int32		   *dst_used = PG_GETARG_POINTER(4);
	CompressionState *cs = (CompressionState *) PG_GETARG_POINTER(5);
	zlib_state	   *state = (zlib_state *) cs->opaque;
	int				last_error;
	unsigned long amount_available_used = dst_sz;

	Insist(src_sz > 0 && dst_sz > 0);


	last_error = state->decompress_fn(dst, &amount_available_used,
									  (const Bytef *) src, src_sz);

	*dst_used = amount_available_used;

	if (last_error != Z_OK)
	{
		switch (last_error)
		{
			case Z_MEM_ERROR:
				elog(ERROR, "out of memory");
				break;

			case Z_BUF_ERROR:

				/* 
				 * This would be a bug. We should have given a buffer big
				 * enough in the decompress case.
				 */
				elog(ERROR, "buffer size %d insufficient for compressed data",
					 dst_sz);
				break;

			case Z_DATA_ERROR:
				/*
				 * zlib data structures corrupted.
				 *
				 * Check out the error message: kind of like 'catalog
				 * convergence' for data corruption :-).
				 */
				elog(ERROR, "zlib encountered data in an unexpected format");

			default:
				/* shouldn't get here */
				Insist(false);
				break;
		}
	}

	PG_RETURN_VOID();
}

static void cbsShrinkCallback(CBS cbs, CBSBlock cbsBlock,
                              Size oldSize, Size newSize)
{
  testCallback(cbs, cbsBlock, oldSize, newSize, &CallbackShrink);
  Insist(oldSize >= cbs->minSize);
  Insist(newSize >= cbs->minSize);
  Insist(oldSize > newSize);

  NShrinkBlocks++;
}

/*
 *  Initialize metadata cache key
 */
void 
InitMetadataCacheKey(MetadataCacheKey *key, const HdfsFileInfo *file_info)
{
    Insist(NULL != key);
    Insist(NULL != file_info);
    
    key->tablespace_oid = file_info->tablespace_oid;
    key->database_oid = file_info->database_oid;
    key->relation_oid = file_info->relation_oid;
    key->segno = file_info->segno;
}