contrib/exthdfs/exthdfs.c - hawq - Git at Google

 /*
  * 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 "common.h"
 #include "access/extprotocol.h"
 #include "cdb/cdbdatalocality.h"
 #include "utils/uri.h"

 #include "storage/cwrapper/hdfs-file-system-c.h"
 #include "cdb/cdbfilesystemcredential.h"
 #include "cdb/cdbvars.h"
 #include "postgres.h"

 PG_MODULE_MAGIC
 ;

 PG_FUNCTION_INFO_V1(hdfsprotocol_blocklocation);
 PG_FUNCTION_INFO_V1(hdfsprotocol_validate);

 Datum hdfsprotocol_blocklocation(PG_FUNCTION_ARGS);
 Datum hdfsprotocol_validate(PG_FUNCTION_ARGS);

 static char * getIpBySocket(const char * socket);
 #define EXPECTED_MAX_HDFS_CONNECTIONS 10
 Datum hdfsprotocol_blocklocation(PG_FUNCTION_ARGS)
 {
 	/* Build the result instance */
 	ExtProtocolBlockLocationData *bldata = palloc0(
 			sizeof(ExtProtocolBlockLocationData));
 	if (bldata == NULL)
 	{
 		elog(ERROR, "hdfsprotocol_blocklocation : "
 		"cannot allocate due to no memory");
 	}
 	bldata->type = T_ExtProtocolBlockLocationData;
 	fcinfo->resultinfo = bldata;

 	ExtProtocolValidatorData *pvalidator_data =
 			(ExtProtocolValidatorData *) (fcinfo->context);

 	/*
 	 * Parse URI of the first location, we expect all locations uses the same
 	 * name node server. This is checked in validation function.
 	 */
 	char *first_uri_str = (char *) strVal(
 			lfirst(list_head(pvalidator_data->url_list)));
 	Uri *uri = ParseExternalTableUri(first_uri_str);

 	elog(DEBUG3, "hdfsprotocol_blocklocation : "
 	"extracted HDFS name node address %s:%d",
 	uri->hostname, uri->port);

 	if (enable_secure_filesystem && Gp_role != GP_ROLE_EXECUTE)
 	{
 		char *ccname = NULL;
 		/*
 		 * refresh kerberos ticket
 		 */
 		if (!login())
 		{
 			errno = EACCES;
 		}
 		ccname = pstrdup(krb5_ccname);
 		SetCcname(ccname);
 		if (ccname)
 		  pfree(ccname);
 	}

 	/* Create file system instance */
 	FscHdfsFileSystemC *fs = FscHdfsNewFileSystem(uri->hostname, uri->port);
 	if (fs == NULL)
 	{
 		elog(ERROR, "hdfsprotocol_blocklocation : "
 		"failed to create HDFS instance to connect to %s:%d",
 		uri->hostname, uri->port);
 	}

 	/* Clean up uri instance as we don't need it any longer */
 	FreeExternalTableUri(uri);

 	/* Check all locations to get files to fetch location. */
 	ListCell *lc = NULL;
 	foreach(lc, pvalidator_data->url_list)
 	{
 		/* Parse current location URI. */
 		char *url = (char *) strVal(lfirst(lc));
 		Uri *uri = ParseExternalTableUri(url);
 		if (uri == NULL)
 		{
 			elog(ERROR, "hdfsprotocol_blocklocation : "
 			"invalid URI encountered %s", url);
 		}

 		/*
 		 * NOTICE: We temporarily support only directories as locations. We plan
 		 *         to extend the logic to specifying single file as one location
 		 *         very soon.
 		 */

 		/* get files contained in the path. */
 		FscHdfsFileInfoArrayC *fiarray = FscHdfsDirPath(fs, uri->path);
 		if (FscHdfsHasErrorRaised(fs))
 		{
 			Assert(fiarray == NULL);
 			CatchedError *ce = FscHdfsGetFileSystemError(fs);
 			elog(ERROR, "hdfsprotocol_blocklocation : "
 			"failed to get files of path %s. %s (%d)",
 			uri->path,
 			ce->errMessage, ce->errCode);
 		}

 		/* Call block location api to get data location for each file */
 		for (int i = 0; true; i++)
 		{
 			FscHdfsFileInfoC *fi = FscHdfsGetFileInfoFromArray(fiarray, i);

 			/* break condition of this for loop */
 			if (fi == NULL)
 			{
 				break;
 			}

 			/* Build file name full path. */
 			const char *fname = FscHdfsGetFileInfoName(fi);
 			char *fullpath = palloc0(strlen(uri->path) + /* path  */
 			1 + /* slash */
 			strlen(fname) + /* name  */
 			1); /* \0    */
 			sprintf(fullpath, "%s/%s", uri->path, fname);

 			elog(DEBUG3, "hdfsprotocol_blocklocation : "
 			"built full path file %s", fullpath);

 			/* Get file full length. */
 			int64_t len = FscHdfsGetFileInfoLength(fi);

 			elog(DEBUG3, "hdfsprotocol_blocklocation : "
 			"got file %s length " INT64_FORMAT,
 			fullpath, len);

 			if (len == 0)
 			{
 				pfree(fullpath);
 				continue;
 			}

 			/* Get block location data for this file */
 			FscHdfsFileBlockLocationArrayC *bla =
 					FscHdfsGetPathFileBlockLocation(fs, fullpath, 0, len);
 			if (FscHdfsHasErrorRaised(fs))
 			{
 				Assert(bla == NULL);
 				CatchedError *ce = FscHdfsGetFileSystemError(fs);
 				elog(ERROR, "hdfsprotocol_blocklocation : "
 				"failed to get block location of path %s. "
 				"It is reported generally due to HDFS service errors or "
 				"another session's ongoing writing.",
 				fullpath);
 			}

 			/* Add file full path and its block number as result. */
 			blocklocation_file *blf = palloc0(sizeof(blocklocation_file));
 			blf->file_uri = pstrdup(fullpath);
 			blf->block_num = FscHdfsGetFileBlockLocationArraySize(bla);
 			blf->locations = palloc0(sizeof(BlockLocation) * blf->block_num);

 			elog(DEBUG3, "hdfsprotocol_blocklocation : file %s has %d blocks",
 			fullpath, blf->block_num);

 			/* We don't need it any longer */
 			pfree(fullpath);

 			/* Add block information as a list. */
 			for (int bidx = 0; bidx < blf->block_num; bidx++)
 			{
 				FscHdfsFileBlockLocationC *blo =
 						FscHdfsGetFileBlockLocationFromArray(bla, bidx);
 				BlockLocation *bl = &(blf->locations[bidx]);
 				bl->numOfNodes = FscHdfsGetFileBlockLocationNNodes(blo);
 				bl->rangeId = -1;
 				bl->replicaGroupId = -1;
 				bl->hosts = (char **) palloc0(sizeof(char *) * bl->numOfNodes);
 				bl->names = (char **) palloc0(sizeof(char *) * bl->numOfNodes);
 				bl->topologyPaths = (char **) palloc0(
 						sizeof(char *) * bl->numOfNodes);
 				bl->offset = FscHdfsGetFileBlockLocationOffset(blo);
 				bl->length = FscHdfsGetFileBlockLocationLength(blo);
 				bl->corrupt = FscHdfsGetFileBlockLocationCorrupt(blo);

 				for (int nidx = 0; nidx < bl->numOfNodes; nidx++)
 				{
 					bl->hosts[nidx] = pstrdup(
 							FscHdfsGetFileBlockLocationNodeHost(blo, nidx));
 					bl->names[nidx] = pstrdup(
 							FscHdfsGetFileBlockLocationNodeName(blo, nidx));
 //					elog (LOG,"the host of bidx %d nidx %d is %s, name is %s.",bidx, nidx, bl->hosts[nidx], bl->names[nidx]);
 					bl->topologyPaths[nidx] = pstrdup(
 							FscHdfsGetFileBlockLocationNodeTopoPath(blo, nidx));
 				}
 			}

 			bldata->files = lappend(bldata->files, (void *) (blf));

 			/* Clean up block location instances created by the lib. */
 			FscHdfsFreeFileBlockLocationArrayC(&bla);
 		}

 		/* Clean up URI instance in loop as we don't need it any longer */
 		FreeExternalTableUri(uri);

 		/* Clean up file info array created by the lib for this location. */
 		FscHdfsFreeFileInfoArrayC(&fiarray);
 	}

 	/* destroy fs instance */
 	FscHdfsFreeFileSystemC(&fs);
 	PG_RETURN_VOID() ;
 }

 Datum hdfsprotocol_validate(PG_FUNCTION_ARGS)
 {
   if(nodeTag(fcinfo->context) == T_ExtProtocolRenameData) {
     if (enable_secure_filesystem && Gp_role != GP_ROLE_EXECUTE)
     {
       char *ccname = NULL;
       /*
        * refresh kerberos ticket
        */
       if (!login())
       {
         errno = EACCES;
       }
       ccname = pstrdup(krb5_ccname);
       SetCcname(ccname);
       if (ccname)
         pfree(ccname);
     }
     ExtProtocolRenameData *renameData= (ExtProtocolRenameData *)fcinfo->context;
     FscHdfsFileSystemC *fs = FscHdfsNewFileSystem(renameData->olduri->hostname,
                                                   renameData->olduri->port);
     if (fs == NULL) {
         elog(ERROR, "RenamePathInHDFS : "
         "failed to create HDFS instance to connect to %s:%d",
         renameData->olduri->hostname, renameData->olduri->port);
       }
     FscHdfsRenamePath(fs, renameData->olduri->path, renameData->newduri->path);
     FscHdfsFreeFileSystemC(&fs);
     PG_RETURN_VOID() ;
   }
 	elog(DEBUG3, "hdfsprotocol_validate() begin");

 	/* Check which action should perform. */
 	ExtProtocolValidatorData *pvalidator_data =
 			(ExtProtocolValidatorData *) (fcinfo->context);

 	if (pvalidator_data->forceCreateDir)
 		Assert(pvalidator_data->url_list && pvalidator_data->url_list->length == 1);

 	if (pvalidator_data->direction == EXT_VALIDATE_WRITE)
 	{
 		/* accept only one directory location */
 		if (list_length(pvalidator_data->url_list) != 1)
 		{
 			ereport(ERROR,
 					(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "only one location url is supported for writable external hdfs")));
 		}
 	}
 	char *first_uri_str = (char *) strVal(
 			lfirst(list_head(pvalidator_data->url_list)));
 	Uri *uri = ParseExternalTableUri(first_uri_str);

 	/* Go through first round to get formatter type */
 	bool isCsv = false;
 	bool isText = false;
 	bool isOrc = false;
 	ListCell *optcell = NULL;

 	foreach(optcell, pvalidator_data->format_opts)
 	{
 		DefElem *de = (DefElem *) lfirst(optcell);
 		if (strcasecmp(de->defname, "formatter") == 0)
 		{
 			char *val = strVal(de->arg);
 			if (strcasecmp(val, "csv") == 0)
 			{
 				isCsv = true;
 			}
 			else if (strcasecmp(val, "text") == 0)
 			{
 				isText = true;
 			}
 			else if (strcasecmp(val, "orc") == 0)
 			{
 				isOrc = true;
 			}
 		}
 	}
 	if (!isCsv && !isText && !isOrc)
 	{
 		ereport(ERROR,
 				(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "only 'csv', 'text' and 'orc' formatter is supported for external hdfs")));
 	}Assert(isCsv || isText || isOrc);

 	/* Validate formatter options */
 	foreach(optcell, pvalidator_data->format_opts)
 	{
 		DefElem *de = (DefElem *) lfirst(optcell);
 		if (strcasecmp(de->defname, "delimiter") == 0)
 		{
 			char *val = strVal(de->arg);
 			/* Validation 1. User can not specify 'OFF' in delimiter */
 			if (strcasecmp(val, "off") == 0)
 			{
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'off' value of 'delimiter' option is not supported")));
 			}
 			/* Validation 2. Can specify multibytes characters */
 			if (strlen(val) < 1)
 			{
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'delimiter' option accepts multibytes characters")));
 			}
 		}

 		if (strcasecmp(de->defname, "escape") == 0)
 		{
 			char *val = strVal(de->arg);
 			/* Validation 3. User can not specify 'OFF' in escape except for TEXT format */
 		  if (strcasecmp(val, "off") == 0 && !isText)
       {
         ereport(ERROR,
             (errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'off' value of 'escape' option is not supported")));
       }
 			/* Validation 4. Can only specify one character */
 			if (strlen(val) != 1 && strcasecmp(val, "off") != 0)
 			{
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'escape' option accepts single character")));
 			}
 		}

 		if (strcasecmp(de->defname, "newline") == 0)
 		{
 			char *val = strVal(de->arg);
 			/* Validation 5. only accept 'lf', 'cr', 'crlf' */
 			if (strcasecmp(val, "lf") != 0 && strcasecmp(val, "cr") != 0
 					&& strcasecmp(val, "crlf") != 0)
 			{
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "the value of 'newline' option can only be " "'lf', 'cr' or 'crlf'")));
 			}
 		}

 		if (strcasecmp(de->defname, "quote") == 0)
 		{
 			/* This is allowed only for csv mode formatter */
 			if (!isCsv)
 			{
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'quote' option is only available in 'csv' formatter")));
 			}

 			char *val = strVal(de->arg);
 			/* Validation 5. Can only specify one character */
 			if (strlen(val) != 1)
 			{
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'quote' option accepts single character")));
 			}
 		}

 		if (strcasecmp(de->defname, "force_notnull") == 0)
 		{
 			/* This is allowed only for csv mode formatter */
 			if (!isCsv)
 			{
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'force_notnull' option is only available in 'csv' formatter")));
 			}
 		}

 		if (strcasecmp(de->defname, "force_quote") == 0)
 		{
 			/* This is allowed only for csv mode formatter */
 			if (!isCsv)
 			{
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'force_quote' option is only available in 'csv' formatter")));
 			}
 		}

 		if (strcasecmp(de->defname, "header") == 0)
 		{
 		  /* this is allowed only for readable table */
 		  if (pvalidator_data->direction != EXT_VALIDATE_READ)
 		  {
 		    ereport(ERROR,
 		        (errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'header' option is only available in readable external table")));
 		  }
 		}
 	}

 	/* All urls should
 	 * 1) have the same protocol name 'hdfs',
 	 * 2) the same hdfs namenode server address
 	 */
 	/* Check all locations to get files to fetch location. */
 	char *nnaddr = NULL;
 	int nnport = -1;
 	ListCell *lc = NULL;
 	foreach(lc, pvalidator_data->url_list)
 	{
 		/* Parse current location URI. */
 		char *url = (char *) strVal(lfirst(lc));
 		Uri *uri = ParseExternalTableUri(url);
 		if (uri == NULL)
 		{
 			elog(ERROR, "hdfsprotocol_validate : "
 			"invalid URI encountered %s", url);
 		}

 		if (uri->protocol != URI_HDFS)
 		{
 			elog(ERROR, "hdfsprotocol_validate : "
 			"invalid URI protocol encountered in %s, "
 			"hdfs:// protocol is required",
 			url);
 		}

 		if(uri->path[1] == '/'){
 		  elog(ERROR, "hdfsprotocol_validate : "
 		      "invalid files path in %s", uri->path);
 		}

 		if (nnaddr == NULL)
 		{
 			nnaddr = pstrdup(uri->hostname);
 			nnport = uri->port;
 		}
 		else
 		{
 			if (strcmp(nnaddr, uri->hostname) != 0)
 			{
 				elog(ERROR, "hdfsprotocol_validate : "
 				"different name server addresses are detected, "
 				"both %s and %s are found",
 				nnaddr, uri->hostname);
 			}
 			if (nnport != uri->port)
 			{
 				elog(ERROR, "hdfsprotocol_validate : "
 				"different name server ports are detected, "
 				"both %d and %d are found",
 				nnport, uri->port);
 			}
 		}

 		/* SHOULD ADD LOGIC HERE TO CREATE UNEXISTING PATH */
 		if (pvalidator_data->forceCreateDir)
 		{

 			elog(LOG, "hdfs_validator() forced creating dir");
 			if (enable_secure_filesystem && Gp_role != GP_ROLE_EXECUTE)
 			{
 				char *ccname = NULL;
 				/*
 				 * refresh kerberos ticket
 				 */
 				if (!login())
 				{
 					errno = EACCES;
 				}
 				ccname = pstrdup(krb5_ccname);
 				SetCcname(ccname);
 				if (ccname)
 				  pfree(ccname);
 			}

 			/* Create file system instance */
 			FscHdfsFileSystemC *fs = FscHdfsNewFileSystem(uri->hostname,
 					uri->port);
 			if (fs == NULL)
 			{
 				elog(ERROR, "hdfsprotocol_validate : "
 				"failed to create HDFS instance to connect to %s:%d",
 				uri->hostname, uri->port);
 			}

 			if (FscHdfsExistPath(fs, uri->path)
 					&& FscHdfsGetFileKind(fs, uri->path) == 'F')
 				elog(ERROR, "hdfsprotocol_validate : "
 				"Location \"%s\" is a file, not supported yet. "
 				"Only support directory now",
 				uri->path);
 			if (pvalidator_data->direction == EXT_VALIDATE_WRITE
 					&& !FscHdfsExistInsertPath(fs, uri->path))
 			{
 				elog(LOG, "hdfs_validator() to create url %s", uri->path);
 				FscHdfsCreateInsertDir(fs, uri->path);
 				if (FscHdfsHasErrorRaised(fs))
 				{
 					CatchedError *ce = FscHdfsGetFileSystemError(fs);
 					elog(ERROR, "hdfsprotocol_validate : "
 					"failed to create directory %s : %s(%d)",
 					uri->path,
 					ce->errMessage, ce->errCode);
 				}
 			}

 			/* destroy fs instance */
 			FscHdfsFreeFileSystemC(&fs);
 		}

 		/* Clean up temporarily created instances */
 		FreeExternalTableUri(uri);
 		if (nnaddr != NULL)
 		{
 			pfree(nnaddr);
 		}
 	}

 	elog(LOG, "passed validating hdfs protocol options");

 	/**************************************************************************
 	 * This is a bad implementation that we check formatter options here. Should
 	 * be moved to call formatter specific validation UDFs.
 	 **************************************************************************/

 	PG_RETURN_VOID() ;
 }

 static char * getIpBySocket(const char * socket) {
   if (socket == NULL) {
     return NULL;
   }
   int len = 0;
   while (socket[len] != ':' && socket[len] != '\0') {
     len++;
   }
   return pnstrdup(socket,len);
 }
	/*
	* 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 "common.h"
	#include "access/extprotocol.h"
	#include "cdb/cdbdatalocality.h"
	#include "utils/uri.h"

	#include "storage/cwrapper/hdfs-file-system-c.h"
	#include "cdb/cdbfilesystemcredential.h"
	#include "cdb/cdbvars.h"
	#include "postgres.h"

	PG_MODULE_MAGIC
	;

	PG_FUNCTION_INFO_V1(hdfsprotocol_blocklocation);
	PG_FUNCTION_INFO_V1(hdfsprotocol_validate);

	Datum hdfsprotocol_blocklocation(PG_FUNCTION_ARGS);
	Datum hdfsprotocol_validate(PG_FUNCTION_ARGS);

	static char * getIpBySocket(const char * socket);
	#define EXPECTED_MAX_HDFS_CONNECTIONS 10
	Datum hdfsprotocol_blocklocation(PG_FUNCTION_ARGS)
	{
	/* Build the result instance */
	ExtProtocolBlockLocationData *bldata = palloc0(
	sizeof(ExtProtocolBlockLocationData));
	if (bldata == NULL)
	{
	elog(ERROR, "hdfsprotocol_blocklocation : "
	"cannot allocate due to no memory");
	}
	bldata->type = T_ExtProtocolBlockLocationData;
	fcinfo->resultinfo = bldata;

	ExtProtocolValidatorData *pvalidator_data =
	(ExtProtocolValidatorData *) (fcinfo->context);

	/*
	* Parse URI of the first location, we expect all locations uses the same
	* name node server. This is checked in validation function.
	*/
	char first_uri_str = (char ) strVal(
	lfirst(list_head(pvalidator_data->url_list)));
	Uri *uri = ParseExternalTableUri(first_uri_str);

	elog(DEBUG3, "hdfsprotocol_blocklocation : "
	"extracted HDFS name node address %s:%d",
	uri->hostname, uri->port);

	if (enable_secure_filesystem && Gp_role != GP_ROLE_EXECUTE)
	{
	char *ccname = NULL;
	/*
	* refresh kerberos ticket
	*/
	if (!login())
	{
	errno = EACCES;
	}
	ccname = pstrdup(krb5_ccname);
	SetCcname(ccname);
	if (ccname)
	pfree(ccname);
	}

	/* Create file system instance */
	FscHdfsFileSystemC *fs = FscHdfsNewFileSystem(uri->hostname, uri->port);
	if (fs == NULL)
	{
	elog(ERROR, "hdfsprotocol_blocklocation : "
	"failed to create HDFS instance to connect to %s:%d",
	uri->hostname, uri->port);
	}

	/* Clean up uri instance as we don't need it any longer */
	FreeExternalTableUri(uri);

	/* Check all locations to get files to fetch location. */
	ListCell *lc = NULL;
	foreach(lc, pvalidator_data->url_list)
	{
	/* Parse current location URI. */
	char url = (char ) strVal(lfirst(lc));
	Uri *uri = ParseExternalTableUri(url);
	if (uri == NULL)
	{
	elog(ERROR, "hdfsprotocol_blocklocation : "
	"invalid URI encountered %s", url);
	}

	/*
	* NOTICE: We temporarily support only directories as locations. We plan
	* to extend the logic to specifying single file as one location
	* very soon.
	*/

	/* get files contained in the path. */
	FscHdfsFileInfoArrayC *fiarray = FscHdfsDirPath(fs, uri->path);
	if (FscHdfsHasErrorRaised(fs))
	{
	Assert(fiarray == NULL);
	CatchedError *ce = FscHdfsGetFileSystemError(fs);
	elog(ERROR, "hdfsprotocol_blocklocation : "
	"failed to get files of path %s. %s (%d)",
	uri->path,
	ce->errMessage, ce->errCode);
	}

	/* Call block location api to get data location for each file */
	for (int i = 0; true; i++)
	{
	FscHdfsFileInfoC *fi = FscHdfsGetFileInfoFromArray(fiarray, i);

	/* break condition of this for loop */
	if (fi == NULL)
	{
	break;
	}

	/* Build file name full path. */
	const char *fname = FscHdfsGetFileInfoName(fi);
	char fullpath = palloc0(strlen(uri->path) + / path */
	1 + /* slash */
	strlen(fname) + /* name */
	1); /* \0 */
	sprintf(fullpath, "%s/%s", uri->path, fname);

	elog(DEBUG3, "hdfsprotocol_blocklocation : "
	"built full path file %s", fullpath);

	/* Get file full length. */
	int64_t len = FscHdfsGetFileInfoLength(fi);

	elog(DEBUG3, "hdfsprotocol_blocklocation : "
	"got file %s length " INT64_FORMAT,
	fullpath, len);

	if (len == 0)
	{
	pfree(fullpath);
	continue;
	}

	/* Get block location data for this file */
	FscHdfsFileBlockLocationArrayC *bla =
	FscHdfsGetPathFileBlockLocation(fs, fullpath, 0, len);
	if (FscHdfsHasErrorRaised(fs))
	{
	Assert(bla == NULL);
	CatchedError *ce = FscHdfsGetFileSystemError(fs);
	elog(ERROR, "hdfsprotocol_blocklocation : "
	"failed to get block location of path %s. "
	"It is reported generally due to HDFS service errors or "
	"another session's ongoing writing.",
	fullpath);
	}

	/* Add file full path and its block number as result. */
	blocklocation_file *blf = palloc0(sizeof(blocklocation_file));
	blf->file_uri = pstrdup(fullpath);
	blf->block_num = FscHdfsGetFileBlockLocationArraySize(bla);
	blf->locations = palloc0(sizeof(BlockLocation) * blf->block_num);

	elog(DEBUG3, "hdfsprotocol_blocklocation : file %s has %d blocks",
	fullpath, blf->block_num);

	/* We don't need it any longer */
	pfree(fullpath);

	/* Add block information as a list. */
	for (int bidx = 0; bidx < blf->block_num; bidx++)
	{
	FscHdfsFileBlockLocationC *blo =
	FscHdfsGetFileBlockLocationFromArray(bla, bidx);
	BlockLocation *bl = &(blf->locations[bidx]);
	bl->numOfNodes = FscHdfsGetFileBlockLocationNNodes(blo);
	bl->rangeId = -1;
	bl->replicaGroupId = -1;
	bl->hosts = (char *) palloc0(sizeof(char ) * bl->numOfNodes);
	bl->names = (char *) palloc0(sizeof(char ) * bl->numOfNodes);
	bl->topologyPaths = (char **) palloc0(
	sizeof(char ) bl->numOfNodes);
	bl->offset = FscHdfsGetFileBlockLocationOffset(blo);
	bl->length = FscHdfsGetFileBlockLocationLength(blo);
	bl->corrupt = FscHdfsGetFileBlockLocationCorrupt(blo);

	for (int nidx = 0; nidx < bl->numOfNodes; nidx++)
	{
	bl->hosts[nidx] = pstrdup(
	FscHdfsGetFileBlockLocationNodeHost(blo, nidx));
	bl->names[nidx] = pstrdup(
	FscHdfsGetFileBlockLocationNodeName(blo, nidx));
	// elog (LOG,"the host of bidx %d nidx %d is %s, name is %s.",bidx, nidx, bl->hosts[nidx], bl->names[nidx]);
	bl->topologyPaths[nidx] = pstrdup(
	FscHdfsGetFileBlockLocationNodeTopoPath(blo, nidx));
	}
	}

	bldata->files = lappend(bldata->files, (void *) (blf));

	/* Clean up block location instances created by the lib. */
	FscHdfsFreeFileBlockLocationArrayC(&bla);
	}

	/* Clean up URI instance in loop as we don't need it any longer */
	FreeExternalTableUri(uri);

	/* Clean up file info array created by the lib for this location. */
	FscHdfsFreeFileInfoArrayC(&fiarray);
	}

	/* destroy fs instance */
	FscHdfsFreeFileSystemC(&fs);
	PG_RETURN_VOID() ;
	}

	Datum hdfsprotocol_validate(PG_FUNCTION_ARGS)
	{
	if(nodeTag(fcinfo->context) == T_ExtProtocolRenameData) {
	if (enable_secure_filesystem && Gp_role != GP_ROLE_EXECUTE)
	{
	char *ccname = NULL;
	/*
	* refresh kerberos ticket
	*/
	if (!login())
	{
	errno = EACCES;
	}
	ccname = pstrdup(krb5_ccname);
	SetCcname(ccname);
	if (ccname)
	pfree(ccname);
	}
	ExtProtocolRenameData renameData= (ExtProtocolRenameData )fcinfo->context;
	FscHdfsFileSystemC *fs = FscHdfsNewFileSystem(renameData->olduri->hostname,
	renameData->olduri->port);
	if (fs == NULL) {
	elog(ERROR, "RenamePathInHDFS : "
	"failed to create HDFS instance to connect to %s:%d",
	renameData->olduri->hostname, renameData->olduri->port);
	}
	FscHdfsRenamePath(fs, renameData->olduri->path, renameData->newduri->path);
	FscHdfsFreeFileSystemC(&fs);
	PG_RETURN_VOID() ;
	}
	elog(DEBUG3, "hdfsprotocol_validate() begin");

	/* Check which action should perform. */
	ExtProtocolValidatorData *pvalidator_data =
	(ExtProtocolValidatorData *) (fcinfo->context);

	if (pvalidator_data->forceCreateDir)
	Assert(pvalidator_data->url_list && pvalidator_data->url_list->length == 1);

	if (pvalidator_data->direction == EXT_VALIDATE_WRITE)
	{
	/* accept only one directory location */
	if (list_length(pvalidator_data->url_list) != 1)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "only one location url is supported for writable external hdfs")));
	}
	}
	char first_uri_str = (char ) strVal(
	lfirst(list_head(pvalidator_data->url_list)));
	Uri *uri = ParseExternalTableUri(first_uri_str);

	/* Go through first round to get formatter type */
	bool isCsv = false;
	bool isText = false;
	bool isOrc = false;
	ListCell *optcell = NULL;

	foreach(optcell, pvalidator_data->format_opts)
	{
	DefElem de = (DefElem ) lfirst(optcell);
	if (strcasecmp(de->defname, "formatter") == 0)
	{
	char *val = strVal(de->arg);
	if (strcasecmp(val, "csv") == 0)
	{
	isCsv = true;
	}
	else if (strcasecmp(val, "text") == 0)
	{
	isText = true;
	}
	else if (strcasecmp(val, "orc") == 0)
	{
	isOrc = true;
	}
	}
	}
	if (!isCsv && !isText && !isOrc)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "only 'csv', 'text' and 'orc' formatter is supported for external hdfs")));
	}Assert(isCsv \|\| isText \|\| isOrc);

	/* Validate formatter options */
	foreach(optcell, pvalidator_data->format_opts)
	{
	DefElem de = (DefElem ) lfirst(optcell);
	if (strcasecmp(de->defname, "delimiter") == 0)
	{
	char *val = strVal(de->arg);
	/* Validation 1. User can not specify 'OFF' in delimiter */
	if (strcasecmp(val, "off") == 0)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'off' value of 'delimiter' option is not supported")));
	}
	/* Validation 2. Can specify multibytes characters */
	if (strlen(val) < 1)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'delimiter' option accepts multibytes characters")));
	}
	}

	if (strcasecmp(de->defname, "escape") == 0)
	{
	char *val = strVal(de->arg);
	/* Validation 3. User can not specify 'OFF' in escape except for TEXT format */
	if (strcasecmp(val, "off") == 0 && !isText)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'off' value of 'escape' option is not supported")));
	}
	/* Validation 4. Can only specify one character */
	if (strlen(val) != 1 && strcasecmp(val, "off") != 0)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'escape' option accepts single character")));
	}
	}

	if (strcasecmp(de->defname, "newline") == 0)
	{
	char *val = strVal(de->arg);
	/* Validation 5. only accept 'lf', 'cr', 'crlf' */
	if (strcasecmp(val, "lf") != 0 && strcasecmp(val, "cr") != 0
	&& strcasecmp(val, "crlf") != 0)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "the value of 'newline' option can only be " "'lf', 'cr' or 'crlf'")));
	}
	}

	if (strcasecmp(de->defname, "quote") == 0)
	{
	/* This is allowed only for csv mode formatter */
	if (!isCsv)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'quote' option is only available in 'csv' formatter")));
	}

	char *val = strVal(de->arg);
	/* Validation 5. Can only specify one character */
	if (strlen(val) != 1)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'quote' option accepts single character")));
	}
	}

	if (strcasecmp(de->defname, "force_notnull") == 0)
	{
	/* This is allowed only for csv mode formatter */
	if (!isCsv)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'force_notnull' option is only available in 'csv' formatter")));
	}
	}

	if (strcasecmp(de->defname, "force_quote") == 0)
	{
	/* This is allowed only for csv mode formatter */
	if (!isCsv)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'force_quote' option is only available in 'csv' formatter")));
	}
	}

	if (strcasecmp(de->defname, "header") == 0)
	{
	/* this is allowed only for readable table */
	if (pvalidator_data->direction != EXT_VALIDATE_READ)
	{
	ereport(ERROR,
	(errcode(ERRCODE_SYNTAX_ERROR), errmsg("hdfsprotocol_validate : " "'header' option is only available in readable external table")));
	}
	}
	}

	/* All urls should
	* 1) have the same protocol name 'hdfs',
	* 2) the same hdfs namenode server address
	*/
	/* Check all locations to get files to fetch location. */
	char *nnaddr = NULL;
	int nnport = -1;
	ListCell *lc = NULL;
	foreach(lc, pvalidator_data->url_list)
	{
	/* Parse current location URI. */
	char url = (char ) strVal(lfirst(lc));
	Uri *uri = ParseExternalTableUri(url);
	if (uri == NULL)
	{
	elog(ERROR, "hdfsprotocol_validate : "
	"invalid URI encountered %s", url);
	}

	if (uri->protocol != URI_HDFS)
	{
	elog(ERROR, "hdfsprotocol_validate : "
	"invalid URI protocol encountered in %s, "
	"hdfs:// protocol is required",
	url);
	}

	if(uri->path[1] == '/'){
	elog(ERROR, "hdfsprotocol_validate : "
	"invalid files path in %s", uri->path);
	}

	if (nnaddr == NULL)
	{
	nnaddr = pstrdup(uri->hostname);
	nnport = uri->port;
	}
	else
	{
	if (strcmp(nnaddr, uri->hostname) != 0)
	{
	elog(ERROR, "hdfsprotocol_validate : "
	"different name server addresses are detected, "
	"both %s and %s are found",
	nnaddr, uri->hostname);
	}
	if (nnport != uri->port)
	{
	elog(ERROR, "hdfsprotocol_validate : "
	"different name server ports are detected, "
	"both %d and %d are found",
	nnport, uri->port);
	}
	}

	/* SHOULD ADD LOGIC HERE TO CREATE UNEXISTING PATH */
	if (pvalidator_data->forceCreateDir)
	{

	elog(LOG, "hdfs_validator() forced creating dir");
	if (enable_secure_filesystem && Gp_role != GP_ROLE_EXECUTE)
	{
	char *ccname = NULL;
	/*
	* refresh kerberos ticket
	*/
	if (!login())
	{
	errno = EACCES;
	}
	ccname = pstrdup(krb5_ccname);
	SetCcname(ccname);
	if (ccname)
	pfree(ccname);
	}

	/* Create file system instance */
	FscHdfsFileSystemC *fs = FscHdfsNewFileSystem(uri->hostname,
	uri->port);
	if (fs == NULL)
	{
	elog(ERROR, "hdfsprotocol_validate : "
	"failed to create HDFS instance to connect to %s:%d",
	uri->hostname, uri->port);
	}

	if (FscHdfsExistPath(fs, uri->path)
	&& FscHdfsGetFileKind(fs, uri->path) == 'F')
	elog(ERROR, "hdfsprotocol_validate : "
	"Location \"%s\" is a file, not supported yet. "
	"Only support directory now",
	uri->path);
	if (pvalidator_data->direction == EXT_VALIDATE_WRITE
	&& !FscHdfsExistInsertPath(fs, uri->path))
	{
	elog(LOG, "hdfs_validator() to create url %s", uri->path);
	FscHdfsCreateInsertDir(fs, uri->path);
	if (FscHdfsHasErrorRaised(fs))
	{
	CatchedError *ce = FscHdfsGetFileSystemError(fs);
	elog(ERROR, "hdfsprotocol_validate : "
	"failed to create directory %s : %s(%d)",
	uri->path,
	ce->errMessage, ce->errCode);
	}
	}

	/* destroy fs instance */
	FscHdfsFreeFileSystemC(&fs);
	}

	/* Clean up temporarily created instances */
	FreeExternalTableUri(uri);
	if (nnaddr != NULL)
	{
	pfree(nnaddr);
	}
	}

	elog(LOG, "passed validating hdfs protocol options");

	/**************************************************************************
	* This is a bad implementation that we check formatter options here. Should
	* be moved to call formatter specific validation UDFs.
	**************************************************************************/

	PG_RETURN_VOID() ;
	}

	static char * getIpBySocket(const char * socket) {
	if (socket == NULL) {
	return NULL;
	}
	int len = 0;
	while (socket[len] != ':' && socket[len] != '\0') {
	len++;
	}
	return pnstrdup(socket,len);
	}