You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hawq.apache.org by sh...@apache.org on 2016/04/04 20:52:59 UTC
incubator-hawq git commit: HAWQ-462. Dispatch dfs_address from master
to segment in secure mode
Repository: incubator-hawq
Updated Branches:
refs/heads/master e8fcfb0c0 -> 59ebfa707
HAWQ-462. Dispatch dfs_address from master to segment in secure mode
Project: http://git-wip-us.apache.org/repos/asf/incubator-hawq/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-hawq/commit/59ebfa70
Tree: http://git-wip-us.apache.org/repos/asf/incubator-hawq/tree/59ebfa70
Diff: http://git-wip-us.apache.org/repos/asf/incubator-hawq/diff/59ebfa70
Branch: refs/heads/master
Commit: 59ebfa7072621117827ae3d9464c971a61919672
Parents: e8fcfb0
Author: Shivram Mani <sh...@gmail.com>
Authored: Mon Apr 4 11:52:51 2016 -0700
Committer: Shivram Mani <sh...@gmail.com>
Committed: Mon Apr 4 11:52:51 2016 -0700
----------------------------------------------------------------------
src/backend/cdb/cdbquerycontextdispatching.c | 63 +++++++++++++++++++++--
src/backend/storage/file/fd.c | 2 +
src/bin/gpfusion/gpbridgeapi.c | 17 +++---
src/include/storage/fd.h | 2 +
4 files changed, 74 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/59ebfa70/src/backend/cdb/cdbquerycontextdispatching.c
----------------------------------------------------------------------
diff --git a/src/backend/cdb/cdbquerycontextdispatching.c b/src/backend/cdb/cdbquerycontextdispatching.c
index 52f85f1..b3cf083 100644
--- a/src/backend/cdb/cdbquerycontextdispatching.c
+++ b/src/backend/cdb/cdbquerycontextdispatching.c
@@ -102,7 +102,7 @@ int QueryContextDispatchingSizeMemoryLimit = 100 * 1024; /* KB */
enum QueryContextDispatchingItemType
{
- MasterXid, TablespaceLocation, TupleType, EmptyTable, FileSystemCredential
+ MasterXid, TablespaceLocation, TupleType, EmptyTable, FileSystemCredential, Namespace
};
typedef enum QueryContextDispatchingItemType QueryContextDispatchingItemType;
@@ -183,7 +183,7 @@ static char*
GetExtTableFirstLocation(Datum *array);
static void
-AddFileSystemCredentialForPxfTable(char *uri);
+AddFileSystemCredentialForPxfTable();
/**
* construct the file location for query context dispatching.
@@ -770,6 +770,30 @@ RebuildTupleForRelation(QueryContextInfo *cxt)
}
/*
+ * Deserialize the Namespace data
+ */
+static void
+RebuildNamespace(QueryContextInfo *cxt)
+{
+
+ int len;
+ char buffer[4], *binary;
+ ReadData(cxt, buffer, sizeof(buffer), TRUE);
+
+ len = (int) ntohl(*(uint32 *) buffer);
+ binary = palloc(len);
+ if(ReadData(cxt, binary, len, TRUE))
+ {
+ StringInfoData buffer;
+ initStringInfoOfString(&buffer, binary, len);
+ dfs_address = strdup(buffer.data);
+ } else {
+ elog(ERROR, "Couldn't rebuild Namespace");
+ }
+ pfree(binary);
+}
+
+/*
* rebuild execute context
*/
void
@@ -801,6 +825,9 @@ RebuildQueryContext(QueryContextInfo *cxt, HTAB **currentFilesystemCredentials,
RebuildFilesystemCredentials(cxt, currentFilesystemCredentials,
currentFilesystemCredentialsMemoryContext);
break;
+ case Namespace:
+ RebuildNamespace(cxt);
+ break;
default:
ereport(ERROR,
(errcode(ERRCODE_GP_INTERNAL_ERROR), errmsg( "unrecognized "
@@ -1746,7 +1773,8 @@ prepareDispatchedCatalogExternalTable(QueryContextInfo *cxt,
if (IS_PXF_URI(location))
{
Insist(array_size == 1);
- AddFileSystemCredentialForPxfTable(location);
+ AddFileSystemCredentialForPxfTable();
+ prepareDfsAddressForDispatch(cxt);
}
AddTupleWithToastsToContextInfo(cxt, ExtTableRelationId, "pg_exttable", tuple,
@@ -2880,7 +2908,7 @@ static char* GetExtTableFirstLocation(Datum *array)
* prepareDispatchedCatalogFileSystemCredential will store the token
* using port == 0 in HA case (otherwise the supplied port).
*/
-static void AddFileSystemCredentialForPxfTable(char *uri)
+static void AddFileSystemCredentialForPxfTable()
{
char* dfs_address = NULL;
@@ -2984,3 +3012,30 @@ GetResultRelSegFileInfos(Oid relid, List *segnomaps, List *existing_seginfomaps)
return existing_seginfomaps;
}
+
+/*
+ * prepareDfsAddressForDispatch
+ *
+ * Given the cxt use the currently set value of cxt->sharedPath
+ * and add it to cxt->buffer so that it is dispatched from
+ * master to segment. This is only required in the case of
+ * a secure filesystem.
+ */
+
+void
+prepareDfsAddressForDispatch(QueryContextInfo* cxt)
+{
+ if (!enable_secure_filesystem)
+ return;
+ const char *namespace = cxt->sharedPath;
+ int size = strlen(namespace);
+ StringInfoData buffer;
+ initStringInfo(&buffer);
+
+ pq_sendint(&buffer, (int) Namespace, sizeof(char));
+ pq_sendint(&buffer, size, sizeof(int));
+
+ WriteData(cxt, buffer.data, buffer.len);
+ WriteData(cxt, namespace, size);
+ pfree(buffer.data);
+}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/59ebfa70/src/backend/storage/file/fd.c
----------------------------------------------------------------------
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index e71282a..4ec458e 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -250,6 +250,8 @@ static int RecentRemoteAllocatedDesc = -1;
*/
static long tempFileCounter = 0;
+char* dfs_address;
+
/*--------------------
*
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/59ebfa70/src/bin/gpfusion/gpbridgeapi.c
----------------------------------------------------------------------
diff --git a/src/bin/gpfusion/gpbridgeapi.c b/src/bin/gpfusion/gpbridgeapi.c
index 6d7dfaf..63fc565 100644
--- a/src/bin/gpfusion/gpbridgeapi.c
+++ b/src/bin/gpfusion/gpbridgeapi.c
@@ -241,6 +241,8 @@ void gpbridge_import_start(PG_FUNCTION_ARGS)
/* read some bytes to make sure the connection is established */
churl_read_check_connectivity(context->churl_handle);
+
+ free_dfs_address();
}
void gpbridge_export_start(PG_FUNCTION_ARGS)
@@ -271,6 +273,7 @@ void gpbridge_export_start(PG_FUNCTION_ARGS)
context->churl_handle = churl_init_upload(context->uri.data,
context->churl_headers);
+ free_dfs_address();
}
/*
@@ -507,16 +510,14 @@ size_t fill_buffer(gphadoop_context* context, char* start, size_t size)
void add_delegation_token(PxfInputData *inputData)
{
PxfHdfsTokenData *token = NULL;
- char* dfs_address = NULL;
if (!enable_secure_filesystem)
return;
- token = palloc0(sizeof(PxfHdfsTokenData));
-
- get_hdfs_location_from_filespace(&dfs_address);
+ Assert(dfs_address);
- elog(DEBUG2, "locating token for %s", dfs_address);
+ token = palloc0(sizeof(PxfHdfsTokenData));
+ elog(DEBUG2, "locating token for %s", dfs_address);
token->hdfs_token = find_filesystem_credential_with_uri(dfs_address);
@@ -526,7 +527,6 @@ void add_delegation_token(PxfInputData *inputData)
inputData->token = token;
- pfree(dfs_address);
}
void free_token_resources(PxfInputData *inputData)
@@ -536,3 +536,8 @@ void free_token_resources(PxfInputData *inputData)
pfree(inputData->token);
}
+
+void free_dfs_address()
+{
+ free(dfs_address);
+}
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/59ebfa70/src/include/storage/fd.h
----------------------------------------------------------------------
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index 3e0782d..0c25264 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -73,6 +73,8 @@ typedef int File;
extern int max_files_per_process;
extern bool enable_secure_filesystem;
+/* DFS address needed on segments */
+extern char* dfs_address;
/*
* prototypes for functions in fd.c