text: /*-------------------------------------------------------------------------
   *
   * conversioncmds.c
   *	  conversion creation command support code
   *
   * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
   * Portions Copyright (c) 1994, Regents of the University of California
   *
   *
   * IDENTIFICATION
   *	  src/backend/commands/conversioncmds.c
   *
   *-------------------------------------------------------------------------
   */
  #include "postgres.h"
  
  #include "access/heapam.h"
  #include "access/htup_details.h"
  #include "catalog/dependency.h"
  #include "catalog/indexing.h"
  #include "catalog/oid_dispatch.h"
  #include "catalog/pg_conversion.h"
  #include "catalog/pg_conversion_fn.h"
  #include "catalog/pg_type.h"
  #include "commands/alter.h"
  #include "commands/conversioncmds.h"
  #include "mb/pg_wchar.h"
  #include "miscadmin.h"
  #include "parser/parse_func.h"
  #include "utils/builtins.h"
  #include "utils/lsyscache.h"
  #include "utils/rel.h"
  #include "utils/syscache.h"
  
  #include "cdb/cdbvars.h"
  #include "cdb/cdbdisp_query.h"
  
  /*
   * CREATE CONVERSION
   */
  ObjectAddress
  CreateConversionCommand(CreateConversionStmt *stmt)
  {
  	Oid			namespaceId;
  	char	   *conversion_name;
  	AclResult	aclresult;
  	int			from_encoding;
  	int			to_encoding;
  	Oid			funcoid;
  	const char *from_encoding_name = stmt->for_encoding_name;
  	const char *to_encoding_name = stmt->to_encoding_name;
  	List	   *func_name = stmt->func_name;
  	static const Oid funcargs[] = {INT4OID, INT4OID, CSTRINGOID, INTERNALOID, INT4OID};
  	char		result[1];
  
  	/* Convert list of names to a name and namespace */
  	namespaceId = QualifiedNameGetCreationNamespace(stmt->conversion_name,
  													&conversion_name);
  
  	/* Check we have creation rights in target namespace */
  	aclresult = pg_namespace_aclcheck(namespaceId, GetUserId(), ACL_CREATE);
  	if (aclresult != ACLCHECK_OK)
  		aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
  					   get_namespace_name(namespaceId));
  
  	/* Check the encoding names */
  	from_encoding = pg_char_to_encoding(from_encoding_name);
  	if (from_encoding < 0)
  		ereport(ERROR,
  				(errcode(ERRCODE_UNDEFINED_OBJECT),
  				 errmsg("source encoding \"%s\" does not exist",
  						from_encoding_name)));
  
  	to_encoding = pg_char_to_encoding(to_encoding_name);
  	if (to_encoding < 0)
  		ereport(ERROR,
  				(errcode(ERRCODE_UNDEFINED_OBJECT),
  				 errmsg("destination encoding \"%s\" does not exist",
  						to_encoding_name)));
  
  	/*
  	 * Check the existence of the conversion function. Function name could be
  	 * a qualified name.
  	 */
  	funcoid = LookupFuncName(func_name, sizeof(funcargs) / sizeof(Oid),
  							 funcargs, false);
  
  	/* Check it returns VOID, else it's probably the wrong function */
  	if (get_func_rettype(funcoid) != VOIDOID)
  		ereport(ERROR,
  				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
  		  errmsg("encoding conversion function %s must return type \"void\"",
  				 NameListToString(func_name))));
  
  	/* Check we have EXECUTE rights for the function */
  	aclresult = pg_proc_aclcheck(funcoid, GetUserId(), ACL_EXECUTE);
  	if (aclresult != ACLCHECK_OK)
  		aclcheck_error(aclresult, ACL_KIND_PROC,
  					   NameListToString(func_name));
  
  	/*
  	 * Check that the conversion function is suitable for the requested source
  	 * and target encodings. We do that by calling the function with an empty
  	 * string; the conversion function should throw an error if it can't
  	 * perform the requested conversion.
  	 */
  	OidFunctionCall5(funcoid,
  					 Int32GetDatum(from_encoding),
  					 Int32GetDatum(to_encoding),
  					 CStringGetDatum(""),
  					 CStringGetDatum(result),
  					 Int32GetDatum(0));
  
  	/*
  	 * All seem ok, go ahead (possible failure would be a duplicate conversion
  	 * name)
  	 */
  	ObjectAddress objAddr;
  	objAddr = ConversionCreate(conversion_name, namespaceId, GetUserId(),
  							   from_encoding, to_encoding, funcoid, stmt->def);
  					 
  	if (Gp_role == GP_ROLE_DISPATCH)
  	{
  		CdbDispatchUtilityStatement((Node *) stmt,
  									DF_CANCEL_ON_ERROR|
  									DF_WITH_SNAPSHOT|
  									DF_NEED_TWO_PHASE,
  									GetAssignedOidsForDispatch(),
  									NULL);
  	}
  
  	return objAddr;
  }
  