Author: cmccabe
Date: Tue Aug 12 18:58:49 2014
New Revision: 1617563
URL: http://svn.apache.org/r1617563
Log:
HADOOP-10725. Implement listStatus and getFileInfo in the native client
(cmccabe)
Modified:
hadoop/common/branches/HADOOP-10388/hadoop-native-core/src/main/native/ndfs/ndfs.c
hadoop/common/branches/HADOOP-10388/hadoop-native-core/src/main/native/test/fs/test_libhdfs_meta_ops.c
Modified:
hadoop/common/branches/HADOOP-10388/hadoop-native-core/src/main/native/ndfs/ndfs.c
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/HADOOP-10388/hadoop-native-core/src/main/native/ndfs/ndfs.c?rev=1617563&r1=1617562&r2=1617563&view=diff
==============================================================================
---
hadoop/common/branches/HADOOP-10388/hadoop-native-core/src/main/native/ndfs/ndfs.c
(original)
+++
hadoop/common/branches/HADOOP-10388/hadoop-native-core/src/main/native/ndfs/ndfs.c
Tue Aug 12 18:58:49 2014
@@ -139,6 +139,9 @@ static struct hadoop_err *ndfs_connect_s
static void ndfs_free(struct native_fs *fs);
+struct hadoop_err *populate_file_info(struct file_info *info,
+ HdfsFileStatusProto *status, const char *prefix);
+
static void ndfs_nn_proxy_init(struct native_fs *fs, struct hrpc_proxy *proxy)
{
hrpc_proxy_init(proxy, fs->msgr, &fs->nn_addr, CLIENT_NN_PROTOCOL,
@@ -456,8 +459,10 @@ static struct hadoop_err *ndfs_connect_s
static void ndfs_free(struct native_fs *fs)
{
- hrpc_messenger_shutdown(fs->msgr);
- hrpc_messenger_free(fs->msgr);
+ if (fs->msgr) {
+ hrpc_messenger_shutdown(fs->msgr);
+ hrpc_messenger_free(fs->msgr);
+ }
free(fs->url_prefix);
hadoop_uri_free(fs->conn_uri);
if (fs->working_uri) {
@@ -822,19 +827,206 @@ done:
return hadoopfs_errno_and_retcode(err);
}
-static hdfsFileInfo* ndfs_list_directory(hdfsFS bfs __attribute__((unused)),
- const char* uri
__attribute__((unused)),
- int *numEntries
__attribute__((unused)))
+struct hadoop_err *ndfs_list_partial(struct native_fs *fs,
+ const char *path, const char *prev, uint32_t *entries_len,
+ hdfsFileInfo **entries, uint32_t *remaining)
{
- errno = ENOTSUP;
- return NULL;
+ struct hadoop_err *err = NULL;
+ GetListingRequestProto req = GET_LISTING_REQUEST_PROTO__INIT;
+ GetListingResponseProto *resp = NULL;
+ hdfsFileInfo *nentries;
+ struct hrpc_proxy proxy;
+ uint64_t nlen;
+ size_t i;
+ char *prefix = NULL;
+
+ err = dynprintf(&prefix, "%s%s/", fs->url_prefix, path);
+ if (err)
+ goto done;
+ ndfs_nn_proxy_init(fs, &proxy);
+ req.src = (char*)path;
+ req.startafter.data = (unsigned char*)prev;
+ req.startafter.len = strlen(prev);
+ req.needlocation = 0;
+ err = cnn_get_listing(&proxy, &req, &resp);
+ if (err)
+ goto done;
+ if (!resp->dirlist) {
+ err = hadoop_lerr_alloc(ENOENT, "ndfs_list_partial(path=%s, "
+ "prev=%s): No such directory.", path, prev);
+ goto done;
+ }
+ nlen = *entries_len;
+ nlen += resp->dirlist->n_partiallisting;
+ nentries = realloc(*entries, nlen * sizeof(hdfsFileInfo));
+ if (!nentries) {
+ err = hadoop_lerr_alloc(ENOENT, "ndfs_list_partial(path=%s, "
+ "prev=%s): failed to allocate space for %zd new entries.",
+ path, prev, resp->dirlist->n_partiallisting);
+ goto done;
+ }
+ memset(nentries + ((*entries_len) * sizeof(hdfsFileInfo)),
+ 0, (resp->dirlist->n_partiallisting * sizeof(hdfsFileInfo)));
+ *entries = nentries;
+ *entries_len = nlen;
+ *remaining = resp->dirlist->remainingentries;
+ for (i = 0; i < resp->dirlist->n_partiallisting; i++) {
+ err = populate_file_info(&nentries[i],
+ resp->dirlist->partiallisting[i], prefix);
+ if (err)
+ goto done;
+ }
+ err = NULL;
+
+done:
+ free(prefix);
+ if (resp) {
+ get_listing_response_proto__free_unpacked(resp, NULL);
+ }
+ return err;
}
-static hdfsFileInfo *ndfs_get_path_info(hdfsFS bfs __attribute__((unused)),
- const char* uri
__attribute__((unused)))
+struct hadoop_err *populate_file_info(struct file_info *info,
+ HdfsFileStatusProto *status, const char *prefix)
{
- errno = ENOTSUP;
+ if (status->filetype == IS_DIR) {
+ info->mKind = kObjectKindDirectory;
+ } else {
+ // note: we don't support symlinks yet here.
+ info->mKind = kObjectKindFile;
+ }
+ info->mName = malloc(strlen(prefix) + status->path.len + 1);
+ if (!info->mName)
+ goto oom;
+ strcpy(info->mName, prefix);
+ memcpy(info->mName + strlen(prefix), status->path.data, status->path.len);
+ info->mName[strlen(prefix) + status->path.len] = '\0';
+ info->mLastMod = status->modification_time / 1000LL;
+ info->mSize = status->length;
+ if (status->has_block_replication) {
+ info->mReplication = status->block_replication;
+ } else {
+ info->mReplication = 0;
+ }
+ if (status->has_blocksize) {
+ info->mBlockSize = status->blocksize;
+ } else {
+ info->mBlockSize = 0;
+ }
+ info->mOwner = strdup(status->owner);
+ if (!info->mOwner)
+ goto oom;
+ info->mGroup = strdup(status->group);
+ if (!info->mGroup)
+ goto oom;
+ info->mPermissions = status->permission->perm;
+ info->mLastAccess = status->access_time / 1000LL;
return NULL;
+oom:
+ return hadoop_lerr_alloc(ENOMEM, "populate_file_info(%s): OOM",
+ info->mName);
+}
+
+static hdfsFileInfo* ndfs_list_directory(hdfsFS bfs,
+ const char* uri, int *numEntries)
+{
+ struct native_fs *fs = (struct native_fs *)bfs;
+ struct hadoop_err *err = NULL;
+ hdfsFileInfo *entries = NULL;
+ uint32_t entries_len = 0, remaining = 0;
+ char *prev, *path = NULL;
+
+ err = build_path(fs, uri, &path);
+ if (err) {
+ goto done;
+ }
+ // We may need to make multiple RPCs to the Namenode to get all the
+ // entries in this directory. We need to keep making RPCs as long as the
+ // 'remaining' value we get back is more than 0. The actual value of
+ // 'remaining' isn't interesting, because it may have changed by the time
+ // we make the next RPC.
+ do {
+ if (entries_len > 0) {
+ prev = entries[entries_len - 1].mName;
+ } else {
+ prev = "";
+ }
+ err = ndfs_list_partial(fs, path, prev, &entries_len,
+ &entries, &remaining);
+ if (err)
+ goto done;
+ } while (remaining != 0);
+ err = NULL;
+
+done:
+ free(path);
+ if (err) {
+ if (entries) {
+ hdfsFreeFileInfo(entries, entries_len);
+ entries = NULL;
+ }
+ } else {
+ *numEntries = entries_len;
+ }
+ return hadoopfs_errno_and_retptr(err, entries);
+}
+
+static hdfsFileInfo *ndfs_get_path_info(hdfsFS bfs, const char* uri)
+{
+ struct native_fs *fs = (struct native_fs*)bfs;
+ struct hadoop_err *err = NULL;
+ GetFileInfoRequestProto req = GET_FILE_INFO_REQUEST_PROTO__INIT;
+ GetFileInfoResponseProto *resp = NULL;
+ struct hrpc_proxy proxy;
+ char *prefix = NULL, *path = NULL;
+ hdfsFileInfo *info = NULL;
+
+ // The GetFileInfo RPC returns a blank 'path' field.
+ // To maintain 100% compatibility with the JNI client, we need to fill it
+ // in with a URI containing the absolute path to the file.
+ err = dynprintf(&prefix, "%s%s", fs->url_prefix, path);
+ if (err)
+ goto done;
+ ndfs_nn_proxy_init(fs, &proxy);
+ err = build_path(fs, uri, &path);
+ if (err) {
+ goto done;
+ }
+ req.src = path;
+ err = cnn_get_file_info(&proxy, &req, &resp);
+ if (err) {
+ goto done;
+ }
+ if (!resp->fs) {
+ err = hadoop_lerr_alloc(ENOENT, "ndfs_get_path_info(%s): no such "
+ "file or directory.", path);
+ goto done;
+ }
+ info = calloc(1, sizeof(*info));
+ if (!info) {
+ err = hadoop_lerr_alloc(ENOMEM, "ndfs_get_path_info(%s): OOM", path);
+ goto done;
+ }
+ err = populate_file_info(info, resp->fs, prefix);
+ if (err) {
+ err = hadoop_err_prepend(err, 0, "ndfs_get_path_info(%s)", path);
+ goto done;
+ }
+ err = NULL;
+
+done:
+ free(prefix);
+ free(path);
+ if (resp) {
+ get_file_info_response_proto__free_unpacked(resp, NULL);
+ }
+ if (err) {
+ if (info) {
+ hdfsFreeFileInfo(info, 1);
+ info = NULL;
+ }
+ }
+ return hadoopfs_errno_and_retptr(err, info);
}
char***
Modified:
hadoop/common/branches/HADOOP-10388/hadoop-native-core/src/main/native/test/fs/test_libhdfs_meta_ops.c
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/HADOOP-10388/hadoop-native-core/src/main/native/test/fs/test_libhdfs_meta_ops.c?rev=1617563&r1=1617562&r2=1617563&view=diff
==============================================================================
---
hadoop/common/branches/HADOOP-10388/hadoop-native-core/src/main/native/test/fs/test_libhdfs_meta_ops.c
(original)
+++
hadoop/common/branches/HADOOP-10388/hadoop-native-core/src/main/native/test/fs/test_libhdfs_meta_ops.c
Tue Aug 12 18:58:49 2014
@@ -30,8 +30,16 @@
/** Test performing metadata operations via libhdfs. */
+static void print_file_info(const hdfsFileInfo *info)
+{
+ printf("file_info(mName=%s): mOwner=%s, mGroup=%s, "
+ "mPermissions=%04o\n", info->mName, info->mOwner,
+ info->mGroup, info->mPermissions);
+}
+
int main(void)
{
+ hdfsFileInfo *infos, *info;
struct hdfsBuilder *hdfs_bld = NULL;
hdfsFS fs = NULL;
struct NativeMiniDfsCluster* dfs_cluster = NULL;
@@ -39,6 +47,7 @@ int main(void)
.doFormat = 1,
};
const char *nn_uri;
+ int i, num_entries;
nn_uri = getenv("NAMENODE_URI");
if (!nn_uri) {
@@ -65,6 +74,26 @@ int main(void)
EXPECT_INT_ZERO(hdfsCreateDirectory(fs, "/abc/3"));
EXPECT_INT_ZERO(hdfsCreateDirectory(fs, "/abc/alpha"));
EXPECT_INT_ZERO(hdfsDelete(fs, "/abc", 1));
+ hdfsDelete(fs, "/abc", 1);
+ EXPECT_INT_ZERO(hdfsCreateDirectory(fs, "/abc"));
+ EXPECT_INT_ZERO(hdfsCreateDirectory(fs, "/abc/1"));
+ EXPECT_INT_ZERO(hdfsCreateDirectory(fs, "/abc/2"));
+ EXPECT_INT_ZERO(hdfsCreateDirectory(fs, "/abc/3"));
+ EXPECT_INT_ZERO(hdfsCreateDirectory(fs, "/abc/alpha"));
+ infos = hdfsListDirectory(fs, "/abc", &num_entries);
+ EXPECT_NONNULL(infos);
+ EXPECT_INT_EQ(4, num_entries);
+ for (i = 0; i < num_entries; i++) {
+ print_file_info(&infos[i]);
+ }
+ hdfsFreeFileInfo(infos, num_entries);
+ info = hdfsGetPathInfo(fs, "/abc");
+ EXPECT_NONNULL(info);
+ EXPECT_INT_ZERO(info->mReplication);
+ EXPECT_INT_ZERO(info->mBlockSize);
+ EXPECT_INT_EQ(kObjectKindDirectory, info->mKind);
+ print_file_info(info);
+ hdfsFreeFileInfo(info, 1);
EXPECT_INT_ZERO(hdfsDisconnect(fs));
if (dfs_cluster) {
EXPECT_INT_ZERO(nmdShutdown(dfs_cluster));
