/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

#include <unistd.h>

#include "postgres.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "utils/palloc.h"
#include "storage/fd.h"
#include "storage/relfilenode.h"

#include "catalog/catalog.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_tablespace.h"
#include "catalog/pg_database.h"
#include "cdb/cdbsharedoidsearch.h"
#include "cdb/cdbdirectopen.h"
#include "cdb/cdbvars.h"
#include "storage/itemptr.h"
#include "utils/hsearch.h"
#include "storage/shmem.h"
#include "access/genam.h"
#include "access/heapam.h"
#include "access/aomd.h"
#include "access/transam.h"
#include "utils/guc.h"
#include "storage/smgr.h"
#include "storage/ipc.h"
#include "storage/proc.h"
#include "storage/pmsignal.h"           /* PostmasterIsAlive */
#include "storage/sinvaladt.h"
#include "utils/builtins.h"
#include "utils/faultinjector.h"
#include "commands/tablespace.h"
#include "commands/dbcommands.h"
#include "cdb/cdbmetadatacache.h"
#include "cdb/cdbmetadatacache_internal.h"
#include "tcop/tcopprot.h" /* quickdie() */
#include "storage/backendid.h"
#include "postmaster/fork_process.h"
#include "postmaster/postmaster.h"
#include "utils/ps_status.h"
#include "libpq/pqsignal.h"
#include "utils/memutils.h"

#include "funcapi.h"
#include "fmgr.h"
#include "utils/builtins.h"

#define MAX_HDFS_HOST_NUM               1024
#define MAX_BLOCK_INFO_LEN              128
#define BLOCK_INFO_BIT_NUM              16

/*
 *  HDFS Block Info Structure
 */
typedef enum MetadataBlockInfoType
{
    METADATA_BLOCK_INFO_TYPE_HOSTS          = 0,
    METADATA_BLOCK_INFO_TYPE_NAMES,
    METADATA_BLOCK_INFO_TYPE_TOPOLOGYPATHS,
    METADATA_BLOCK_INFO_TYPE_NUM,
} MetadataBlockInfoType;

typedef struct BlockInfoKey
{
    char                block_info[MAX_BLOCK_INFO_LEN];  
} BlockInfoKey;

typedef struct BlockInfoEntry
{
    BlockInfoKey        key;
    uint16_t            index;            
} BlockInfoEntry;

typedef struct RevertBlockInfoKey
{
    uint16_t            index;
} RevertBlockInfoKey;

typedef struct RevertBlockInfoEntry
{
    RevertBlockInfoKey  key;
    char                block_info[MAX_BLOCK_INFO_LEN];
} RevertBlockInfoEntry;

/*
 * Metadata Cache Global Initialization Functions
 */
static bool 
MetadataCacheHashTableInit(void);

static bool 
MetadataBlockInfoTablesInit(void);

static bool
MetadataRevertBlockInfoTablesInit(void);

static bool 
MetadataCacheHdfsBlockArrayInit(void);

/*
 *  Metadata Cache Operation Functions
 */
static void                 
InitMetadataCacheKey(MetadataCacheKey *key, const HdfsFileInfo *file_info);

static MetadataCacheEntry *
MetadataCacheExists(const HdfsFileInfo *file_info);

static MetadataCacheEntry *
MetadataCacheEnter(const HdfsFileInfo *file_info);

static void 
MetadataCacheRemove(const HdfsFileInfo *file_info);

/*
 *  HDFS Block Operation Functions
 */
static void 
AllocMetadataBlock(int block_num, uint32_t *first_block_id, uint32_t *last_block_id);

static void 
ReleaseMetadataBlock(int block_num, uint32_t first_block_id, uint32_t last_block_id);

static uint64_t 
SetMetadataBlockInfo(MetadataBlockInfoType type, char **infos, uint32_t num);

static char *
GetMetadataBlockInfo(MetadataBlockInfoType type, uint64_t infos, int index);

/*
 *  Metadata Cache Global Variables
 */
MetadataCacheSharedData  *MetadataCacheSharedDataInstance = NULL;
HTAB                     *MetadataCache = NULL;
MetadataHdfsBlockInfo    *MetadataBlockArray = NULL;

static HTAB                     *BlockHostsMap = NULL;
static HTAB                     *BlockNamesMap = NULL;
static HTAB                     *BlockTopologyPathsMap = NULL;

static HTAB                     *RevertBlockHostsMap = NULL;
static HTAB                     *RevertBlockNamesMap = NULL;
static HTAB                     *RevertBlockTopologyPathsMap = NULL;

//static BlockLocation *CreateHdfsFileBlockLocations(BlockLocation *hdfs_locations, int block_num);
//static BlockLocation *MergeHdfsFileBlockLocations(BlockLocation *locations1, int block_num1, BlockLocation *locations2, int block_num2);


// create block locations for user
static BlockLocation *GetHdfsFileBlockLocationsNoCache(const HdfsFileInfo *file_info, uint64_t filesize, int *block_num);
static BlockLocation *GetHdfsFileBlockLocationsFromCache(MetadataCacheEntry *entry, uint64_t filesize, int *block_num);
//static BlockLocation *AppendHdfsFileBlockLocationsToCache(const HdfsFileInfo *file_info, MetadataCacheEntry *entry, uint64_t filesize, int *block_num, double *hit_ratio);

/*
 *  Estimate metadata cache shared memory size
 *      - Metadata cache hash size
 *      - Block info (3 types) hash size
 *      - Metadata cache shared structure size
 *      - Metadata hdfs block array size
 */
Size 
MetadataCache_ShmemSize(void)
{
    Size size;
    
    size = hash_estimate_size((Size)metadata_cache_max_hdfs_file_num, sizeof(MetadataCacheEntry));

    size = add_size(size, hash_estimate_size((Size)MAX_HDFS_HOST_NUM, sizeof(BlockInfoEntry)) * METADATA_BLOCK_INFO_TYPE_NUM);

    size = add_size(size, sizeof(MetadataCacheSharedData));

    size = add_size(size, metadata_cache_block_capacity* sizeof(MetadataHdfsBlockInfo));

    return size;
}

/*
 *  Initialize all the data structure in the share memory for metadata cache
 *      - Metadata cache shared data structure
 *      - Metadata cache hash table
 *      - Metadata cache block info hash tables (3 types) 
 *      - Metadata hdfs block array
 */
void 
MetadataCache_ShmemInit(void)
{
    bool found; 

    MetadataCacheSharedDataInstance = (MetadataCacheSharedData *)ShmemInitStruct("Metadata Cache Shared Data", 
                                                        sizeof(MetadataCacheSharedData), &found);

    if (found && MetadataCacheSharedDataInstance)
    {
        return;
    }

    if (NULL == MetadataCacheSharedDataInstance)
    {
        elog(FATAL, "[MetadataCache] fail to allocate share memory for metadata cache shared data");
    }

    MetadataCacheSharedDataInstance->free_block_num = metadata_cache_block_capacity;
    MetadataCacheSharedDataInstance->free_block_head = 0;
    MetadataCacheSharedDataInstance->cur_hosts_idx = 0;
    MetadataCacheSharedDataInstance->cur_names_idx = 0;
    MetadataCacheSharedDataInstance->cur_topologyPaths_idx = 0;

    if (!MetadataCacheHashTableInit())
    {
        elog(FATAL, "[MetadataCache] fail to allocate share memory for metadata cache hash table");
    }
  
    if (!MetadataBlockInfoTablesInit())
    {
        elog(FATAL, "[MetadataCache] fail to allocate share memory for metadata cache block info hash tables");
    }

    if (!MetadataRevertBlockInfoTablesInit())
    {
        elog(FATAL, "[MetadataCache] fail to allocate share memory for metadata cache revert block info hash tables");
    }

    if (!MetadataCacheHdfsBlockArrayInit())
    {
        elog(FATAL, "[MetadataCache] fail to allocate share memory for metadata cache hdfs block array");
    }

    elog(LOG, "[MetadataCache] Metadata cache initialize successfully. block_capacity:%d", metadata_cache_block_capacity);

    return;
}

/*
 * Initialize metadata cache hash table
 */
bool 
MetadataCacheHashTableInit(void)
{
    HASHCTL     info;
    int         hash_flags;
 
    MemSet(&info, 0, sizeof(info));

    info.keysize = sizeof(MetadataCacheKey);
    info.entrysize = sizeof(MetadataCacheEntry);
    info.hash = tag_hash;
    hash_flags = (HASH_ELEM | HASH_FUNCTION);

    MetadataCache = ShmemInitHash("Metadata Cache", metadata_cache_max_hdfs_file_num, metadata_cache_max_hdfs_file_num, &info, hash_flags);
    if (NULL == MetadataCache)
    {
        return false;
    }

    return true;
}

/*
 *  Initialize metadata block info hash tables
 *      - block host ip hash
 *      - block host name hash
 *      - block host topology hash
 */
bool 
MetadataBlockInfoTablesInit(void)
{
    HASHCTL     info;
    int         hash_flags;
 
    MemSet(&info, 0, sizeof(info));

    info.keysize = sizeof(BlockInfoKey);
    info.entrysize = sizeof(BlockInfoEntry);
    hash_flags = HASH_ELEM;

    BlockHostsMap = ShmemInitHash("Metadata Block Hosts Map", MAX_HDFS_HOST_NUM, MAX_HDFS_HOST_NUM, &info, hash_flags);
    if (NULL == BlockHostsMap)
    {
        return false;
    }
 
    BlockNamesMap = ShmemInitHash("Metadata Block Names Map", MAX_HDFS_HOST_NUM, MAX_HDFS_HOST_NUM, &info, hash_flags);
    if (NULL == BlockNamesMap)
    {
        return false;
    }
    
    BlockTopologyPathsMap = ShmemInitHash("Metadata Block TopologyPaths Map", MAX_HDFS_HOST_NUM, MAX_HDFS_HOST_NUM, &info, hash_flags);
    if (NULL == BlockTopologyPathsMap)
    {
        return false;
    }

    return true;
}

/*
 *  Initialize metadata revert block info hash tables
 */
bool 
MetadataRevertBlockInfoTablesInit(void)
{
    HASHCTL     info;
    int         hash_flags;
 
    MemSet(&info, 0, sizeof(info));

    info.keysize = sizeof(RevertBlockInfoKey);
    info.entrysize = sizeof(RevertBlockInfoEntry);
    hash_flags = HASH_ELEM;

    RevertBlockHostsMap = ShmemInitHash("Metadata Revert Block Hosts Map", MAX_HDFS_HOST_NUM, MAX_HDFS_HOST_NUM, &info, hash_flags);
    if (NULL == RevertBlockHostsMap)
    {
        return false;
    }
 
    RevertBlockNamesMap = ShmemInitHash("Metadata Revert Block Names Map", MAX_HDFS_HOST_NUM, MAX_HDFS_HOST_NUM, &info, hash_flags);
    if (NULL == RevertBlockNamesMap)
    {
        return false;
    }
    
    RevertBlockTopologyPathsMap = ShmemInitHash("Metadata Revert Block TopologyPaths Map", MAX_HDFS_HOST_NUM, MAX_HDFS_HOST_NUM, &info, hash_flags);
    if (NULL == RevertBlockTopologyPathsMap)
    {
        return false;
    }

    return true;
}


/*
 *  Initialize metadata hdfs block array
 */
bool 
MetadataCacheHdfsBlockArrayInit(void)
{
    Insist(MetadataCacheSharedDataInstance != NULL);
    
    int i = 0;
    bool found;
    
    MetadataBlockArray = (MetadataHdfsBlockInfo *)ShmemInitStruct("Metadata Cache HDFS Block Array",
                                metadata_cache_block_capacity * sizeof(MetadataHdfsBlockInfo), &found);

    if (NULL == MetadataBlockArray)
    {
        return false;
    }

    FREE_BLOCK_NUM = metadata_cache_block_capacity;
    FREE_BLOCK_HEAD = 0;

    for (i=0;i<FREE_BLOCK_NUM;i++)
    {
        NEXT_BLOCK_ID(i) = i + 1;
    }
    NEXT_BLOCK_ID(FREE_BLOCK_NUM - 1) = END_OF_BLOCK;

    return true;
}

/*
 *  Create HdfsFileInfo structure before calling GetHdfsFileBlockLocations 
 */
HdfsFileInfo *
CreateHdfsFileInfo(RelFileNode rnode, int segno)
{
    char *basepath = NULL;
    int relfile_len = 0;
    
    HdfsFileInfo *file_info = (HdfsFileInfo *)palloc(sizeof(HdfsFileInfo));
    if (NULL == file_info)
    {
        return NULL;
    }

    file_info->tablespace_oid = rnode.spcNode;
    file_info->database_oid = rnode.dbNode;
    file_info->relation_oid = rnode.relNode;
    file_info->segno = segno;

    basepath = relpath(rnode);
    relfile_len = strlen(basepath) + 9;
    file_info->filepath = (char *)palloc0(relfile_len);
    FormatAOSegmentFileName(basepath, file_info->segno, -1, 0, &file_info->segno, file_info->filepath);
    pfree(basepath);

    return file_info;
}

/*
 *  Destroy HdfsFileInfo structure after call GetHdfsFileBlockLocations 
 */
void 
DestroyHdfsFileInfo(HdfsFileInfo *file_info)
{
    if (file_info)
    {
        if (file_info->filepath)
        {
            pfree(file_info->filepath);
        }
        pfree(file_info);
        file_info = NULL;
    }
}

/*
 *  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;
}

/*
 *  Metadata cache exist
 */
MetadataCacheEntry *
MetadataCacheExists(const HdfsFileInfo *file_info)
{
    Insist(file_info != NULL);

    bool found;
    MetadataCacheKey key;
    
    InitMetadataCacheKey(&key, file_info);
    return (MetadataCacheEntry *)hash_search(MetadataCache, (void *)&key, HASH_FIND, &found);
}

/*
 *  Metadata cache insert
 */
MetadataCacheEntry *
MetadataCacheEnter(const HdfsFileInfo *file_info)
{
    Insist(file_info != NULL);
    
    bool found;
    MetadataCacheKey key;

    InitMetadataCacheKey(&key, file_info);
    return (MetadataCacheEntry *)hash_search(MetadataCache, (void *)&key, HASH_ENTER_NULL, &found);
}

/*
 *  Metadata cache remove
 */
void 
MetadataCacheRemove(const HdfsFileInfo *file_info)
{
    Insist(file_info != NULL);
    
    bool found;
    MetadataCacheKey key;

    InitMetadataCacheKey(&key, file_info);
    hash_search(MetadataCache, (void *)&key, HASH_REMOVE, &found);
}


/*
 *  Allocate block in the MetadataBlockArray
 */
void 
AllocMetadataBlock(int block_num, uint32_t *first_block_id, uint32_t *last_block_id)
{
    Insist(block_num > 0);
    Insist(FREE_BLOCK_NUM >= block_num);
    
    int i;
    
    *first_block_id = FREE_BLOCK_HEAD;
    for (i=0;i<block_num;i++)
    {
        *last_block_id = FREE_BLOCK_HEAD; 
        FREE_BLOCK_HEAD = NEXT_BLOCK_ID(*last_block_id);
    }
    NEXT_BLOCK_ID(*last_block_id) = END_OF_BLOCK;

    FREE_BLOCK_NUM -= block_num;
}

/*
 *  Release block in the MetadataBlockArray
 */
void 
ReleaseMetadataBlock(int block_num, uint32_t first_block_id, uint32_t last_block_id)
{
    Insist(block_num > 0);
    Insist(NEXT_BLOCK_ID(last_block_id) == END_OF_BLOCK);

    NEXT_BLOCK_ID(last_block_id) = FREE_BLOCK_HEAD;
    FREE_BLOCK_HEAD = first_block_id; 

    FREE_BLOCK_NUM += block_num; 
}

/*
 *  Transfer hdfs block info to uint64 and put it into related hash table
 */
uint64_t 
SetMetadataBlockInfo(MetadataBlockInfoType type, char **infos, uint32_t num)
{
    Insist(num <= 4);
  
    uint64_t result = 0;
    uint64_t entry_idx = 0;
    bool found; 
    int i;    
    BlockInfoKey key;
    BlockInfoEntry *entry;
    HTAB *htab = NULL;

    RevertBlockInfoKey rkey;
    RevertBlockInfoEntry *rentry;
    HTAB *rhtab = NULL;

    uint32_t *cur_idx = NULL;

    switch (type)
    {
    case METADATA_BLOCK_INFO_TYPE_HOSTS:
        htab = BlockHostsMap;
        rhtab = RevertBlockHostsMap;
        cur_idx = &MetadataCacheSharedDataInstance->cur_hosts_idx;
        break;

    case METADATA_BLOCK_INFO_TYPE_NAMES:
        htab = BlockNamesMap;
        rhtab = RevertBlockNamesMap;
        cur_idx = &MetadataCacheSharedDataInstance->cur_names_idx;
        break;
    
    case METADATA_BLOCK_INFO_TYPE_TOPOLOGYPATHS:
        htab = BlockTopologyPathsMap;
        rhtab = RevertBlockTopologyPathsMap;
        cur_idx = &MetadataCacheSharedDataInstance->cur_topologyPaths_idx;
        break;

    default:
        return 0;
    }

    for (i=0;i<num;i++)
    {
        if (strlen(infos[i]) >= MAX_BLOCK_INFO_LEN)
        {
            elog(ERROR, "The length of hostname must little than %d", MAX_BLOCK_INFO_LEN);
        }

        memset(key.block_info, 0, MAX_BLOCK_INFO_LEN);
        snprintf(key.block_info, MAX_BLOCK_INFO_LEN, "%s", infos[i]);
        entry = (BlockInfoEntry *)hash_search(htab, (void *)&key, HASH_ENTER_NULL, &found);

        if (!found)
        {
            (*cur_idx)++;
            entry->index = (*cur_idx);
            
            rkey.index = (*cur_idx);
            rentry = hash_search(rhtab, (void *)&rkey, HASH_ENTER_NULL, &found);
            memset(rentry->block_info, 0, MAX_BLOCK_INFO_LEN);
            snprintf(rentry->block_info, MAX_BLOCK_INFO_LEN, "%s", infos[i]);
        }
        entry_idx = entry->index;
        result |= (entry_idx << (i * BLOCK_INFO_BIT_NUM));
    }

    return result;

}

/*
 *  Get uint64 hdfs block info by index and transfer it to string
 */
char *
GetMetadataBlockInfo(MetadataBlockInfoType type, uint64_t infos, int index)
{
    Insist(index >= 0 && index < 4);
 
    RevertBlockInfoKey rkey;
    RevertBlockInfoEntry *rentry;
    HTAB *rhtab = NULL;
    uint64_t mask;
    bool found;

    switch (type)
    {
    case METADATA_BLOCK_INFO_TYPE_HOSTS:
        rhtab = RevertBlockHostsMap;
        break;

    case METADATA_BLOCK_INFO_TYPE_NAMES:
        rhtab = RevertBlockNamesMap;
        break;
    
    case METADATA_BLOCK_INFO_TYPE_TOPOLOGYPATHS:
        rhtab = RevertBlockTopologyPathsMap;
        break;

    default:
        return NULL;
    }

    mask = 0x000000000000ffff;
    mask = mask << (index * BLOCK_INFO_BIT_NUM);
    rkey.index = (mask & infos) >> (index * BLOCK_INFO_BIT_NUM);

    rentry = (RevertBlockInfoEntry *)hash_search(rhtab, (void *)&rkey, HASH_FIND, &found);
    if (!found) {
        return NULL;
    }
    return rentry->block_info;
}

/*
 *  Get hdfs file block locations for specfic file size, return hdfs block num and cache hit ratio
 */
BlockLocation *GetHdfsFileBlockLocations(const HdfsFileInfo *file_info, uint64_t filesize, int *block_num, double *hit_ratio)
{
    Insist(file_info != NULL);

    if (0 == filesize)
    {
        // empty file
        *block_num = 0;
        *hit_ratio = 0;
        return NULL;
    }

    MetadataCacheEntry *cache_entry = NULL;
    BlockLocation *locations = NULL;

    LWLockAcquire(MetadataCacheLock, LW_SHARED);

    cache_entry = MetadataCacheExists(file_info);
    if (!cache_entry)
    {
        // Cache Not Hit
        LWLockRelease(MetadataCacheLock); 

        elog(DEBUG1, "[MetadataCache] GetHdfsFileBlockLocations NOT HIT CACHE. filename:%s filesize:"INT64_FORMAT"", 
                                file_info->filepath, 
                                filesize);

        locations = GetHdfsFileBlockLocationsNoCache(file_info, filesize, block_num);
        *hit_ratio = 0;
    }
    else
    {
        elog(DEBUG1, "[MetadataCache] GetHdfsFileBlockLocations HIT CACHE. filename:%s filesize:"INT64_FORMAT", \
                                cache_info[filesize:"INT64_FORMAT" block_num:%u first_block_id:%u last_block_id:%u]",
                                file_info->filepath, 
                                filesize, 
                                cache_entry->file_size, 
                                cache_entry->block_num, 
                                cache_entry->first_block_id, 
                                cache_entry->last_block_id);
        
        if (filesize <= cache_entry->file_size)
        {
            // Cache Hit Fully
            locations = GetHdfsFileBlockLocationsFromCache(cache_entry, filesize, block_num);
            *hit_ratio = 1.0;    

            LWLockRelease(MetadataCacheLock); 
        } 
        else 
        {
            /*
            // Cache Hit Partly
            if (cache_entry->block_num <= 1)
            {
                // only one file, re-fetch 
            */
            
                // re-fetch file's all block locations, because hdfs will get incorrect result when fetch partly
                LWLockRelease(MetadataCacheLock); 
        
                LWLockAcquire(MetadataCacheLock, LW_EXCLUSIVE);
                RemoveHdfsFileBlockLocations(file_info);
                LWLockRelease(MetadataCacheLock); 
                
                locations = GetHdfsFileBlockLocationsNoCache(file_info, filesize, block_num);
                *hit_ratio = 0;
            /*
            }
            else
            {
                LWLockRelease(MetadataCacheLock); 
                
                // fetch extra hdfs block locations and append to cache
                locations = AppendHdfsFileBlockLocationsToCache(file_info, cache_entry, filesize, block_num, hit_ratio);
            }
            */
        }
    }
    
    return locations;
}

/*
 *  Get hdfs file block locations from Hadoop HDFS and put the result into metadata cache
 */
BlockLocation *
GetHdfsFileBlockLocationsNoCache(const HdfsFileInfo *file_info, uint64_t filesize, int *block_num)
{
    BlockLocation *hdfs_locations = NULL; 
    BlockLocation *locations = NULL; 
    MetadataCacheEntry *entry = NULL;

    // 1. fetch hdfs block locations
    hdfs_locations = HdfsGetFileBlockLocations(file_info->filepath, filesize, block_num);
    if ((NULL == hdfs_locations) || (0 == *block_num))
    {
        elog(DEBUG1, "[MetadataCache] GetHdfsFileBlockLocationsNoCache fetch hdfs block locatons fail. filename:%s filesize:"INT64_FORMAT" block_num:%d",
                                file_info->filepath, 
                                filesize, 
                                *block_num);
        goto done;
    }
    
    elog(DEBUG1, "[MetadataCache] GetHdfsFileBlockLocationsNoCache fetch hdfs block locatons successfully. filename:%s filesize:"INT64_FORMAT" block_num:%d",
                                file_info->filepath, 
                                filesize, 
                                *block_num);

    // 2. insert fetch results into cache
    LWLockAcquire(MetadataCacheLock, LW_EXCLUSIVE);

    // 3. generate result block locations
    locations = CreateHdfsFileBlockLocations(hdfs_locations, *block_num);

    entry = MetadataCacheNew(file_info, filesize, hdfs_locations, *block_num); 
    if (NULL == entry)
    {
        LWLockRelease(MetadataCacheLock);
        elog(DEBUG1, "[MetadataCache] GetHdfsFileBlockLocationsNoCache put hdfs block locations info cache fail. filename:%s filesize:"INT64_FORMAT" block_num:%d",
                                file_info->filepath, 
                                filesize, 
                                *block_num);
        goto done;
    }
    entry->last_access_time = entry->create_time = time(NULL);

    LWLockRelease(MetadataCacheLock);

done:
    if (hdfs_locations)
    {
        HdfsFreeFileBlockLocations(hdfs_locations, *block_num);
        hdfs_locations = NULL;
    }

    return locations;
}

/*
 *  Get hdfs file block locations from metadata cache 
 */
BlockLocation *
GetHdfsFileBlockLocationsFromCache(MetadataCacheEntry *entry, uint64_t filesize, int *block_num)
{
    Insist(NULL != entry);
    Insist(entry->file_size >= filesize);

    int i, j, k;
    int last_block_length = 0;
    BlockLocation *locations = NULL;
    char *metadata_block_info = NULL;

    if (entry->file_size == filesize)
    {
        *block_num = entry->block_num;
        last_block_length = GET_BLOCK(entry->last_block_id)->length;
    }
    else
    {
        last_block_length = filesize;
        j = entry->first_block_id;
        *block_num = 0;
        while (true) 
        {
            MetadataHdfsBlockInfo *block_info = GET_BLOCK(j);
            (*block_num)++;

            if (filesize <= block_info->length)
            {
                last_block_length = filesize;
                break; 
            }
            else
            {
                filesize -= block_info->length;
                j = NEXT_BLOCK_ID(j);
            }
        }
    }

    locations = (BlockLocation *)palloc(sizeof(BlockLocation) * (*block_num)); 
    if (NULL == locations)
    {
        return NULL;
    }

    k = entry->first_block_id;
    for (i=0;i<(*block_num);i++)
    {
        MetadataHdfsBlockInfo *block_info = GET_BLOCK(k);

        locations[i].corrupt = 0;
        locations[i].numOfNodes = block_info->node_num; 
        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++)
        {
            metadata_block_info = GetMetadataBlockInfo(METADATA_BLOCK_INFO_TYPE_HOSTS, block_info->hosts, j); 
            if (NULL == metadata_block_info)
            {
                goto err;
            }
            locations[i].hosts[j] = pstrdup(metadata_block_info);

            metadata_block_info = GetMetadataBlockInfo(METADATA_BLOCK_INFO_TYPE_NAMES, block_info->names, j);
            if (NULL == metadata_block_info)
            {
                goto err;
            }
            locations[i].names[j] = pstrdup(metadata_block_info);
            
            metadata_block_info = GetMetadataBlockInfo(METADATA_BLOCK_INFO_TYPE_TOPOLOGYPATHS, block_info->topologyPaths, j);
            if (NULL == metadata_block_info)
            {
                goto err;
            }
            locations[i].topologyPaths[j] = pstrdup(metadata_block_info);
        }
        locations[i].offset = block_info->offset;
        
        if (i == (*block_num) -1)
        {
            // last block
            locations[i].length = last_block_length;
        }
        else
        {
            locations[i].length = block_info->length;
        }
        k = NEXT_BLOCK_ID(k);
    }

    entry->last_access_time = time(NULL);

    return locations;

err:
    FreeHdfsFileBlockLocations(locations, *block_num);

    return NULL;
}

/*
 *  Get hdfs block locations from cache and fetch extra part from hadoop hdfs
 */
/*
BlockLocation *
AppendHdfsFileBlockLocationsToCache(const HdfsFileInfo *file_info, MetadataCacheEntry *entry, uint64_t filesize, int *block_num, double *hit_ratio)
{
    Insist(file_info != NULL);
    Insist(entry != NULL);
    Insist(entry->block_num > 1);
    Insist(entry->file_size < filesize);
    
    int block_size = 0;
    uint64_t entry_file_size = 0;
    BlockLocation *hdfs_locations = NULL; 
    BlockLocation *locations_from_cache = NULL; 
    BlockLocation *locations_from_hdfs = NULL;
    BlockLocation *locations;
    uint32_t extra_first_block_id = 0;
    uint32_t extra_last_block_id = 0;
    uint32_t extra_block_num = 0;
    int i, j;
    int hit_block_num = entry->block_num;

    LWLockAcquire(MetadataCacheLock, LW_SHARED);

    block_size = GET_BLOCK(entry->first_block_id)->length;
    entry_file_size = entry->file_size;

    LWLockRelease(MetadataCacheLock);
   
    // 1. fetch extra hdfs block locations
    hdfs_locations = HdfsGetFileBlockLocations2(file_info->filepath, entry_file_size, filesize - entry_file_size, block_num);
    if (!hdfs_locations)
    {
        elog(DEBUG1, "[MetadataCache] AppendHdfsFileBlockLocationsToCache fetch extra hdfs block locations fail. \
                        filename:%s filesize:"INT64_FORMAT","INT64_FORMAT" block_num:%d",
                            file_info->filepath, 
                            entry_file_size, 
                            filesize, 
                            *block_num);
        goto done;
    }
    
    elog(DEBUG1, "[MetadataCache] AppendHdfsFileBlockLocationsToCache fetch extra hdfs block locations successfully. \
                        filename:%s filesize:"INT64_FORMAT","INT64_FORMAT" block_num:%d",
                            file_info->filepath, 
                            entry_file_size, 
                            filesize, 
                            *block_num);
   
    LWLockAcquire(MetadataCacheLock, LW_EXCLUSIVE);

    locations_from_cache = GetHdfsFileBlockLocationsFromCache(entry, entry->file_size, (int *)&entry->block_num);
    locations_from_hdfs = CreateHdfsFileBlockLocations(hdfs_locations, *block_num);

    if (NULL == locations_from_cache || NULL == locations_from_hdfs)
    {
        goto done;
    }

    if (GET_BLOCK(entry->last_block_id)->length < block_size)
    {
        locations = MergeHdfsFileBlockLocations(locations_from_cache, entry->block_num-1, locations_from_hdfs, *block_num);
    }
    else
    {
        locations = MergeHdfsFileBlockLocations(locations_from_cache, entry->block_num, locations_from_hdfs, *block_num);
    }
    FreeHdfsFileBlockLocations(locations_from_cache, entry->block_num);
    FreeHdfsFileBlockLocations(locations_from_hdfs, *block_num);

    if (*block_num > FREE_BLOCK_NUM)
    {
        elog(DEBUG1, "[MetadataCache] AppendHdfsFileBlockLocationsToCache not enough free block. \
                        filename:%s filesize:"INT64_FORMAT","INT64_FORMAT" block_num:%d",
                            file_info->filepath, 
                            entry->file_size, 
                            filesize, 
                            *block_num);
        goto done;
    }

    if (GET_BLOCK(entry->last_block_id)->length < block_size)
    {
        // merge last block
        extra_block_num = *block_num - 1;
        if (extra_block_num > 0)
        {
            AllocMetadataBlock(extra_block_num, &extra_first_block_id, &extra_last_block_id);
        }
        j = entry->last_block_id;
        hit_block_num--;
    }
    else
    {
        extra_block_num = *block_num;
        AllocMetadataBlock(extra_block_num, &extra_first_block_id, &extra_last_block_id);
        j = extra_first_block_id;
    }

    entry->file_size = filesize;
    entry->block_num += extra_block_num;
    if (extra_block_num > 0)
    {
        NEXT_BLOCK_ID(entry->last_block_id) = extra_first_block_id;
        entry->last_block_id = extra_last_block_id;
    }

    // copy extra blocks into cache
    for (i=0;i<(*block_num);i++)
    {
        MetadataHdfsBlockInfo *block = GET_BLOCK(j);

        block->node_num = hdfs_locations[i].numOfNodes < 4 ? hdfs_locations[i].numOfNodes : 4;
        block->hosts = SetMetadataBlockInfo(METADATA_BLOCK_INFO_TYPE_HOSTS, hdfs_locations[i].hosts, block->node_num);
        block->names = SetMetadataBlockInfo(METADATA_BLOCK_INFO_TYPE_NAMES, hdfs_locations[i].names, block->node_num);
        block->topologyPaths = SetMetadataBlockInfo(METADATA_BLOCK_INFO_TYPE_TOPOLOGYPATHS, hdfs_locations[i].topologyPaths, block->node_num);
        block->length = hdfs_locations[i].length;
        block->offset = hdfs_locations[i].offset;
        
        j = NEXT_BLOCK_ID(j);
    }

done:
    if (hdfs_locations)
    {
        HdfsFreeFileBlockLocations(hdfs_locations, *block_num);
        hdfs_locations = NULL;
    }
    
    *block_num = entry->block_num;
    *hit_ratio = (hit_block_num * 1.0) / (*block_num);
    LWLockRelease(MetadataCacheLock);

    return locations;
}
*/

/*
 * Free hdf file block location which create from GetHdfsFileBlockLocations
 */
void 
FreeHdfsFileBlockLocations(BlockLocation *locations, int block_num)
{
    Insist(NULL != locations);
    Insist(block_num >= 0);

    int i, j;
    
    for (i=0;i<block_num;i++)
    {
        for (j=0;j<locations[i].numOfNodes;j++)
        {
            if (locations[i].hosts[j])
            {
                pfree(locations[i].hosts[j]);
            }

            if (locations[i].names[j])
            {
                pfree(locations[i].names[j]);
            }
            
            if (locations[i].topologyPaths[j])
            {
                pfree(locations[i].topologyPaths[j]);
            }
        }

        if (locations[i].hosts)
        {
            pfree(locations[i].hosts);
        }

        if (locations[i].names)
        {
            pfree(locations[i].names);
        }
        
        if (locations[i].topologyPaths)
        {
            pfree(locations[i].topologyPaths);
        }
    }
    pfree(locations);
    locations = NULL; 
}

/*
 *  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;
}

/*
BlockLocation *
MergeHdfsFileBlockLocations(BlockLocation *locations1, int block_num1, BlockLocation *locations2, int block_num2)
{
    Insist(locations1 != NULL && locations2 != NULL);
    Insist(block_num1 > 0 && block_num2 > 0);

    int i, j, k;
    BlockLocation *locations = NULL;
    BlockLocation *tmp_locations = NULL;

    locations = (BlockLocation *)palloc(sizeof(BlockLocation) * (block_num1 + block_num2));
    if (NULL == locations)
    {
        return NULL;
    }
    for (k=0;k<(block_num1+block_num2);k++)
    {
        if (k < block_num1) 
        {
            i = k;
            tmp_locations = locations1;
        }
        else
        {
            i = k - block_num1;
            tmp_locations = locations2;
        }

        locations[k].corrupt = 0;
        locations[k].numOfNodes = tmp_locations[i].numOfNodes; 

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

    return locations;

}
*/

void 
DumpHdfsFileBlockLocations(BlockLocation *locations, int block_num)
{
    int i, j;
    char msg[1024];

    for (i=0;i<block_num;i++)
    {
        memset(msg, 0, 1024);
        snprintf(msg, MAX_BLOCK_INFO_LEN, "node_num:%d ", locations[i].numOfNodes);
        for (j=0;j<locations[i].numOfNodes;j++)
        {
            snprintf(msg + strlen(msg), MAX_BLOCK_INFO_LEN, "host%d:%s ", j, locations[i].hosts[j]);
            snprintf(msg + strlen(msg), MAX_BLOCK_INFO_LEN, "name%d:%s ", j, locations[i].names[j]);
            snprintf(msg + strlen(msg), MAX_BLOCK_INFO_LEN, "topologyPath%d:%s ", j, locations[i].topologyPaths[j]);
        }
        snprintf(msg + strlen(msg), MAX_BLOCK_INFO_LEN, "offset:"INT64_FORMAT" ", locations[i].offset);
        snprintf(msg + strlen(msg), MAX_BLOCK_INFO_LEN, "length:"INT64_FORMAT" ", locations[i].length);
        elog(DEBUG5, "[MetadataCache] DumpHdfsFileBlockLocations block%d %s", i, msg);
    }
        
    elog(DEBUG5, "[MetadataCache] DumpHdfsFileBlockLocations block_num:%d", block_num);
}

/*
 * Transfer hdfs block locations to cache format and put it into cache
 */
MetadataCacheEntry *
MetadataCacheNew(const HdfsFileInfo *file_info, uint64_t filesize, BlockLocation *hdfs_locations, int block_num)
{
    Insist(file_info != NULL);
    Insist(hdfs_locations != NULL);
    Insist(block_num > 0);

    int i, j;
    MetadataCacheEntry *entry = NULL;

    if (block_num > FREE_BLOCK_NUM)
    {
        elog(DEBUG1, "[Metadata] MetadataCacheNew not enough free block. \
                    filename:%s filesize:"INT64_FORMAT" block_num:%d free_block_num:%d",
                        file_info->filepath, 
                        filesize, 
                        block_num, 
                        FREE_BLOCK_NUM);
        return NULL;
    }
    
    entry = MetadataCacheEnter(file_info);
    if (!entry)
    {
        elog(DEBUG1, "[MetadataCache] MetadataCacheNew enter cache fail. \
                    filename:%s filesize:"INT64_FORMAT" block_num:%d",
                        file_info->filepath, 
                        filesize, 
                        block_num);
        return NULL;
    }

    entry->file_size = filesize;
    entry->block_num = block_num;
    
    AllocMetadataBlock(entry->block_num, &entry->first_block_id, &entry->last_block_id);

    // copy hdfs blocks into cache
    j = entry->first_block_id; 
    for (i=0;i<entry->block_num;i++)
    {
        MetadataHdfsBlockInfo *block = GET_BLOCK(j);
        
        block->node_num = hdfs_locations[i].numOfNodes < 4 ? hdfs_locations[i].numOfNodes : 4;
        block->hosts = SetMetadataBlockInfo(METADATA_BLOCK_INFO_TYPE_HOSTS, hdfs_locations[i].hosts, block->node_num);
        block->names = SetMetadataBlockInfo(METADATA_BLOCK_INFO_TYPE_NAMES, hdfs_locations[i].names, block->node_num);
        block->topologyPaths = SetMetadataBlockInfo(METADATA_BLOCK_INFO_TYPE_TOPOLOGYPATHS, hdfs_locations[i].topologyPaths, block->node_num);
        block->length = hdfs_locations[i].length;
        block->offset = hdfs_locations[i].offset;
       
        j = NEXT_BLOCK_ID(j);
    }
   
    return entry;
}

/*
 *  Remove entry from metadata cache
 */
void 
RemoveHdfsFileBlockLocations(const HdfsFileInfo *file_info)
{
    Insist(NULL != file_info);

    MetadataCacheEntry *entry = MetadataCacheExists(file_info);
    if (NULL == entry)
    {
        elog(DEBUG1, "[MetadataCache] RemoveHdfsFileBlockLocations file not exists. filename:%s", file_info->filepath);
    }
    else
    {
        elog(DEBUG1, "[Metadata] RemoveHdfsFileBlockLocations, filename:%s block_num:%d", 
                    file_info->filepath,
                    entry->block_num); 
        ReleaseMetadataBlock(entry->block_num, entry->first_block_id, entry->last_block_id);
        MetadataCacheRemove(file_info);
    }
}


/*
 *  Metadata Cache UDF
 *
 *  Clear all content in metadata cache
 */
extern Datum gp_metadata_cache_clear(PG_FUNCTION_ARGS)
{
    HASH_SEQ_STATUS hstat;
    MetadataCacheEntry *entry;
    bool found;
    long entry_num = 0;

    LWLockAcquire(MetadataCacheLock, LW_EXCLUSIVE);
    
    entry_num = hash_get_num_entries(MetadataCache);
    if (MetadataCacheRefreshList)
    {
        list_free(MetadataCacheRefreshList);
        MetadataCacheRefreshList = NULL;
    }

    hash_seq_init(&hstat, MetadataCache);
    while ((entry = (MetadataCacheEntry *)hash_seq_search(&hstat)) != NULL)
    {
        ReleaseMetadataBlock(entry->block_num, entry->first_block_id, entry->last_block_id);
        hash_search(MetadataCache, (void *)&entry->key, HASH_REMOVE, &found);
    }
    
    LWLockRelease(MetadataCacheLock);

    char message[1024] = {0};
    snprintf(message, 1024, "Metadata cache clear %ld items", entry_num);
    PG_RETURN_TEXT_P(cstring_to_text(message));    
}

/*
 *  Metadata Cache UDF
 *
 *  Get entry number of metadata cache 
 */
extern Datum gp_metadata_cache_current_num(PG_FUNCTION_ARGS)
{
    LWLockAcquire(MetadataCacheLock, LW_SHARED);
    int64 num = hash_get_num_entries(MetadataCache); 
    LWLockRelease(MetadataCacheLock);

    PG_RETURN_INT64(num);
}
/*
 *  Metadata Cache UDF
 *
 *  Get block number of metadata cache
 */
extern Datum gp_metadata_cache_current_block_num(PG_FUNCTION_ARGS)
{
    PG_RETURN_INT64(metadata_cache_block_capacity - FREE_BLOCK_NUM);
}
/*
 *  Metadata Cache UDF
 *
 *  Check whether an entry in the metadata cache 
 */
extern Datum gp_metadata_cache_exists(PG_FUNCTION_ARGS)
{
    Oid tablespace_oid = PG_GETARG_OID(0);
    Oid database_oid = PG_GETARG_OID(1);
    Oid relation_oid = PG_GETARG_OID(2);
    int4 segno = PG_GETARG_INT32(3);

    RelFileNode rnode;
    rnode.spcNode = tablespace_oid;
    rnode.dbNode = database_oid;
    rnode.relNode = relation_oid;
    MetadataCacheKey key;
    MetadataCacheEntry *entry;

    HdfsFileInfo *file_info = CreateHdfsFileInfo(rnode, segno);
    InitMetadataCacheKey(&key, file_info);

    LWLockAcquire(MetadataCacheLock, LW_SHARED);
    entry = MetadataCacheExists(file_info);
    LWLockRelease(MetadataCacheLock);

    DestroyHdfsFileInfo(file_info);

    PG_RETURN_BOOL(entry != NULL);
}

/*
 *  Metadata Cache UDF
 *
 *  Get entry info in the metadata cache
 */
extern Datum gp_metadata_cache_info(PG_FUNCTION_ARGS)
{
    Oid tablespace_oid = PG_GETARG_OID(0);
    Oid database_oid = PG_GETARG_OID(1);
    Oid relation_oid = PG_GETARG_OID(2);
    int4 segno = PG_GETARG_INT32(3);

    RelFileNode rnode;
    rnode.spcNode = tablespace_oid;
    rnode.dbNode = database_oid;
    rnode.relNode = relation_oid;
    MetadataCacheKey key;
    MetadataCacheEntry *entry;

    HdfsFileInfo *file_info = CreateHdfsFileInfo(rnode, segno);
    InitMetadataCacheKey(&key, file_info);

    char message[1024] = {0};
    LWLockAcquire(MetadataCacheLock, LW_SHARED);
    entry = MetadataCacheExists(file_info);
    if (!entry)
    {
        snprintf(message, 1024, "Metadata cache don't have %s", file_info->filepath);        
    }
    else
    {
        snprintf(message, 1024, "Metadata cache have %s [file_size:"INT64_FORMAT" block_num:%u first_block_id:%u last_block_id:%u create_time:%u ]",
                            file_info->filepath, 
                            entry->file_size, 
                            entry->block_num, 
                            entry->first_block_id, 
                            entry->last_block_id, 
                            entry->create_time); 
    }

    LWLockRelease(MetadataCacheLock);

    DestroyHdfsFileInfo(file_info);
    PG_RETURN_TEXT_P(cstring_to_text(message));    
}

/*
 *  Metadata Cache UDF
 *
 *  Put entry into the metadata cache
 */
extern Datum gp_metadata_cache_put_entry_for_test(PG_FUNCTION_ARGS)
{
    Oid tablespace_oid = PG_GETARG_OID(0);
    Oid database_oid = PG_GETARG_OID(1);
    Oid relation_oid = PG_GETARG_OID(2);
    int4 start = PG_GETARG_INT32(3);
    int4 end = PG_GETARG_INT32(4);

    LWLockAcquire(MetadataCacheLock, LW_EXCLUSIVE);

    int i;
    int4 stop = (start -end) / 10;
    int4 current = 0;
    int4 success = 0;
    for(i=start;i<end;i++)
    {
    	MetadataCacheKey key;
    	key.tablespace_oid = tablespace_oid;
    	key.database_oid = database_oid;
    	key.relation_oid = relation_oid;
    	key.segno = i;

    	bool found;
    	MetadataCacheEntry *entry = (MetadataCacheEntry *)hash_search(MetadataCache, (void *)&key, HASH_ENTER_NULL, &found);
        if(entry == NULL)
        {
            continue;
        }
        entry->file_size = 134217728;
        entry->block_num = 1;

        AllocMetadataBlock(entry->block_num, &entry->first_block_id, &entry->last_block_id);

    	current++;
    	success++;

    	if(current == stop)
    	{
    		current = 0;
    		LWLockRelease(MetadataCacheLock);
    		pg_usleep(1 * USECS_PER_SEC);
    	    LWLockAcquire(MetadataCacheLock, LW_EXCLUSIVE);
    	}
    }
    LWLockRelease(MetadataCacheLock);

    char message[1024] = {0};
    snprintf(message, 1024, "Metadata cache successed putting %d entries. Failed putting %d entries.", success, end - start - success);
    PG_RETURN_TEXT_P(cstring_to_text(message));
}