diff --git a/hadoop-common-project/hadoop-common/CHANGES.txt b/hadoop-common-project/hadoop-common/CHANGES.txt
index d8c03470139..9967dc9afe8 100644
--- a/hadoop-common-project/hadoop-common/CHANGES.txt
+++ b/hadoop-common-project/hadoop-common/CHANGES.txt
@@ -809,6 +809,9 @@ Release 2.1.0-beta - UNRELEASED
     HADOOP-9599. hadoop-config.cmd doesn't set JAVA_LIBRARY_PATH correctly.
     (Mostafa Elhemali via ivanmi)
 
+    HADOOP-9637. Adding Native Fstat for Windows as needed by YARN. (Chuan Liu
+    via cnauroth)
+
 Release 2.0.5-alpha - 06/06/2013
 
   INCOMPATIBLE CHANGES
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/nativeio/NativeIO.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/nativeio/NativeIO.java
index 7c2d62d3215..2c49a5de436 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/nativeio/NativeIO.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/nativeio/NativeIO.java
@@ -247,7 +247,21 @@ public static class Stat {
         this.groupId = groupId;
         this.mode = mode;
       }
-
+      
+      Stat(String owner, String group, int mode) {
+        if (!Shell.WINDOWS) {
+          this.owner = owner;
+        } else {
+          this.owner = stripDomain(owner);
+        }
+        if (!Shell.WINDOWS) {
+          this.group = group;
+        } else {
+          this.group = stripDomain(group);
+        }
+        this.mode = mode;
+      }
+      
       @Override
       public String toString() {
         return "Stat(owner='" + owner + "', group='" + group + "'" +
@@ -273,9 +287,25 @@ public int getMode() {
      * @throws IOException thrown if there was an IO error while obtaining the file stat.
      */
     public static Stat getFstat(FileDescriptor fd) throws IOException {
-      Stat stat = fstat(fd);
-      stat.owner = getName(IdCache.USER, stat.ownerId);
-      stat.group = getName(IdCache.GROUP, stat.groupId);
+      Stat stat = null;
+      if (!Shell.WINDOWS) {
+        stat = fstat(fd); 
+        stat.owner = getName(IdCache.USER, stat.ownerId);
+        stat.group = getName(IdCache.GROUP, stat.groupId);
+      } else {
+        try {
+          stat = fstat(fd);
+        } catch (NativeIOException nioe) {
+          if (nioe.getErrorCode() == 6) {
+            throw new NativeIOException("The handle is invalid.",
+                Errno.EBADF);
+          } else {
+            LOG.warn(String.format("NativeIO.getFstat error (%d): %s",
+                nioe.getErrorCode(), nioe.getMessage()));
+            throw new NativeIOException("Unknown error", Errno.UNKNOWN);
+          }
+        }
+      }
       return stat;
     }
 
@@ -448,14 +478,27 @@ public CachedUid(String username, long timestamp) {
       new ConcurrentHashMap<Long, CachedUid>();
   private static long cacheTimeout;
   private static boolean initialized = false;
+  
+  /**
+   * The Windows logon name has two part, NetBIOS domain name and
+   * user account name, of the format DOMAIN\UserName. This method
+   * will remove the domain part of the full logon name.
+   *
+   * @param the full principal name containing the domain
+   * @return name with domain removed
+   */
+  private static String stripDomain(String name) {
+    int i = name.indexOf('\\');
+    if (i != -1)
+      name = name.substring(i + 1);
+    return name;
+  }
 
   public static String getOwner(FileDescriptor fd) throws IOException {
     ensureInitialized();
     if (Shell.WINDOWS) {
       String owner = Windows.getOwner(fd);
-      int i = owner.indexOf('\\');
-      if (i != -1)
-        owner = owner.substring(i + 1);
+      owner = stripDomain(owner);
       return owner;
     } else {
       long uid = POSIX.getUIDforFDOwnerforOwner(fd);
diff --git a/hadoop-common-project/hadoop-common/src/main/native/src/org/apache/hadoop/io/nativeio/NativeIO.c b/hadoop-common-project/hadoop-common/src/main/native/src/org/apache/hadoop/io/nativeio/NativeIO.c
index f9181cabcff..d12abe036af 100644
--- a/hadoop-common-project/hadoop-common/src/main/native/src/org/apache/hadoop/io/nativeio/NativeIO.c
+++ b/hadoop-common-project/hadoop-common/src/main/native/src/org/apache/hadoop/io/nativeio/NativeIO.c
@@ -50,6 +50,7 @@
 // the NativeIO$POSIX$Stat inner class and its constructor
 static jclass stat_clazz;
 static jmethodID stat_ctor;
+static jmethodID stat_ctor2;
 
 // the NativeIOException class and its constructor
 static jclass nioe_clazz;
@@ -84,10 +85,12 @@ static int workaround_non_threadsafe_calls(JNIEnv *env, jclass clazz) {
   return result;
 }
 
-#ifdef UNIX
 static void stat_init(JNIEnv *env, jclass nativeio_class) {
+  jclass clazz = NULL;
+  jclass obj_class = NULL;
+  jmethodID  obj_ctor = NULL;
   // Init Stat
-  jclass clazz = (*env)->FindClass(env, "org/apache/hadoop/io/nativeio/NativeIO$POSIX$Stat");
+  clazz = (*env)->FindClass(env, "org/apache/hadoop/io/nativeio/NativeIO$POSIX$Stat");
   if (!clazz) {
     return; // exception has been raised
   }
@@ -100,12 +103,16 @@ static void stat_init(JNIEnv *env, jclass nativeio_class) {
   if (!stat_ctor) {
     return; // exception has been raised
   }
-
-  jclass obj_class = (*env)->FindClass(env, "java/lang/Object");
+  stat_ctor2 = (*env)->GetMethodID(env, stat_clazz, "<init>",
+    "(Ljava/lang/String;Ljava/lang/String;I)V");
+  if (!stat_ctor2) {
+    return; // exception has been raised
+  }
+  obj_class = (*env)->FindClass(env, "java/lang/Object");
   if (!obj_class) {
     return; // exception has been raised
   }
-  jmethodID  obj_ctor = (*env)->GetMethodID(env, obj_class,
+  obj_ctor = (*env)->GetMethodID(env, obj_class,
     "<init>", "()V");
   if (!obj_ctor) {
     return; // exception has been raised
@@ -130,7 +137,6 @@ static void stat_deinit(JNIEnv *env) {
     pw_lock_object = NULL;
   }
 }
-#endif
 
 static void nioe_init(JNIEnv *env) {
   // Init NativeIOException
@@ -168,10 +174,8 @@ static void nioe_deinit(JNIEnv *env) {
 JNIEXPORT void JNICALL
 Java_org_apache_hadoop_io_nativeio_NativeIO_initNative(
 	JNIEnv *env, jclass clazz) {
-#ifdef UNIX
   stat_init(env, clazz);
   PASS_EXCEPTIONS_GOTO(env, error);
-#endif
   nioe_init(env);
   PASS_EXCEPTIONS_GOTO(env, error);
   fd_init(env);
@@ -229,9 +233,44 @@ cleanup:
 #endif
 
 #ifdef WINDOWS
-  THROW(env, "java/io/IOException",
-    "The function POSIX.fstat() is not supported on Windows");
-  return NULL;
+  LPWSTR owner = NULL;
+  LPWSTR group = NULL;
+  int mode;
+  jstring jstr_owner = NULL;
+  jstring jstr_group = NULL;
+  int rc;
+  jobject ret = NULL;
+  HANDLE hFile = (HANDLE) fd_get(env, fd_object);
+  PASS_EXCEPTIONS_GOTO(env, cleanup);
+
+  rc = FindFileOwnerAndPermissionByHandle(hFile, &owner, &group, &mode);
+  if (rc != ERROR_SUCCESS) {
+    throw_ioe(env, rc);
+    goto cleanup;
+  }
+
+  jstr_owner = (*env)->NewString(env, owner, (jsize) wcslen(owner));
+  if (jstr_owner == NULL) goto cleanup;
+
+  jstr_group = (*env)->NewString(env, group, (jsize) wcslen(group));;
+  if (jstr_group == NULL) goto cleanup;
+
+  ret = (*env)->NewObject(env, stat_clazz, stat_ctor2,
+    jstr_owner, jstr_group, (jint)mode);
+
+cleanup:
+  if (ret == NULL) {
+    if (jstr_owner != NULL)
+      (*env)->ReleaseStringChars(env, jstr_owner, owner);
+
+    if (jstr_group != NULL)
+      (*env)->ReleaseStringChars(env, jstr_group, group);
+  }
+
+  LocalFree(owner);
+  LocalFree(group);
+
+  return ret;
 #endif
 }
 
diff --git a/hadoop-common-project/hadoop-common/src/main/winutils/chmod.c b/hadoop-common-project/hadoop-common/src/main/winutils/chmod.c
index c82e18e73e7..c48f7347f6d 100644
--- a/hadoop-common-project/hadoop-common/src/main/winutils/chmod.c
+++ b/hadoop-common-project/hadoop-common/src/main/winutils/chmod.c
@@ -561,22 +561,11 @@ static BOOL ConvertActionsToMask(__in LPCWSTR path,
 {
   MODE_CHANGE_ACTION const *curr = NULL;
 
-  BY_HANDLE_FILE_INFORMATION fileInformation;
   DWORD dwErrorCode = ERROR_SUCCESS;
 
   INT mode = 0;
 
-  dwErrorCode = GetFileInformationByName(path, FALSE, &fileInformation);
-  if (dwErrorCode != ERROR_SUCCESS)
-  {
-    ReportErrorCode(L"GetFileInformationByName", dwErrorCode);
-    return FALSE;
-  }
-  if (IsDirFileInfo(&fileInformation))
-  {
-    mode |= UX_DIRECTORY;
-  }
-  dwErrorCode = FindFileOwnerAndPermission(path, NULL, NULL, &mode);
+  dwErrorCode = FindFileOwnerAndPermission(path, FALSE, NULL, NULL, &mode);
   if (dwErrorCode != ERROR_SUCCESS)
   {
     ReportErrorCode(L"FindFileOwnerAndPermission", dwErrorCode);
diff --git a/hadoop-common-project/hadoop-common/src/main/winutils/chown.c b/hadoop-common-project/hadoop-common/src/main/winutils/chown.c
index 958699679a3..bc2aefc79ee 100644
--- a/hadoop-common-project/hadoop-common/src/main/winutils/chown.c
+++ b/hadoop-common-project/hadoop-common/src/main/winutils/chown.c
@@ -52,7 +52,7 @@ static DWORD ChangeFileOwnerBySid(__in LPCWSTR path,
 
   // Get a pointer to the existing owner information and DACL
   //
-  dwRtnCode = FindFileOwnerAndPermission(longPathName, NULL, NULL, &oldMode);
+  dwRtnCode = FindFileOwnerAndPermission(longPathName, FALSE, NULL, NULL, &oldMode);
   if (dwRtnCode != ERROR_SUCCESS)
   {
     goto ChangeFileOwnerByNameEnd;
diff --git a/hadoop-common-project/hadoop-common/src/main/winutils/include/winutils.h b/hadoop-common-project/hadoop-common/src/main/winutils/include/winutils.h
index 76859475edb..753fb849db0 100644
--- a/hadoop-common-project/hadoop-common/src/main/winutils/include/winutils.h
+++ b/hadoop-common-project/hadoop-common/src/main/winutils/include/winutils.h
@@ -63,6 +63,7 @@ enum UnixAclMask
   UX_U_WRITE   = 00200, // S_IWUSR
   UX_U_READ    = 00400, // S_IRUSR
   UX_DIRECTORY = 0040000, // S_IFDIR
+  UX_REGULAR   = 0100000, // S_IFREG
   UX_SYMLINK   = 0120000, // S_IFLNK
 };
 
@@ -130,6 +131,13 @@ BOOL IsDirFileInfo(const BY_HANDLE_FILE_INFORMATION *fileInformation);
 
 DWORD FindFileOwnerAndPermission(
   __in LPCWSTR pathName,
+  __in BOOL followLink,
+  __out_opt LPWSTR *pOwnerName,
+  __out_opt LPWSTR *pGroupName,
+  __out_opt PINT pMask);
+
+DWORD FindFileOwnerAndPermissionByHandle(
+  __in HANDLE fileHandle,
   __out_opt LPWSTR *pOwnerName,
   __out_opt LPWSTR *pGroupName,
   __out_opt PINT pMask);
diff --git a/hadoop-common-project/hadoop-common/src/main/winutils/libwinutils.c b/hadoop-common-project/hadoop-common/src/main/winutils/libwinutils.c
index 666158cfa4c..3e0768a4b17 100644
--- a/hadoop-common-project/hadoop-common/src/main/winutils/libwinutils.c
+++ b/hadoop-common-project/hadoop-common/src/main/winutils/libwinutils.c
@@ -707,6 +707,71 @@ CheckAccessEnd:
   return dwRtnCode;
 }
 
+
+//----------------------------------------------------------------------------
+// Function: FindFileOwnerAndPermissionByHandle
+//
+// Description:
+//	Find the owner, primary group and permissions of a file object given the
+//  the file object handle. The function will always follow symbolic links.
+//
+// Returns:
+//	ERROR_SUCCESS: on success
+//  Error code otherwise
+//
+// Notes:
+//  - Caller needs to destroy the memeory of owner and group names by calling
+//    LocalFree() function.
+//
+//  - If the user or group name does not exist, the user or group SID will be
+//    returned as the name.
+//
+DWORD FindFileOwnerAndPermissionByHandle(
+  __in HANDLE fileHandle,
+  __out_opt LPWSTR *pOwnerName,
+  __out_opt LPWSTR *pGroupName,
+  __out_opt PINT pMask)
+{
+  LPWSTR path = NULL;
+  DWORD cchPathLen = 0;
+  DWORD dwRtnCode = ERROR_SUCCESS;
+
+  DWORD ret = ERROR_SUCCESS;
+
+  dwRtnCode = GetFinalPathNameByHandle(fileHandle, path, cchPathLen, 0);
+  if (dwRtnCode == 0)
+  {
+    ret = GetLastError();
+    goto FindFileOwnerAndPermissionByHandleEnd;
+  }
+  cchPathLen = dwRtnCode;
+  path = (LPWSTR) LocalAlloc(LPTR, cchPathLen * sizeof(WCHAR));
+  if (path == NULL)
+  {
+    ret = GetLastError();
+    goto FindFileOwnerAndPermissionByHandleEnd;
+  }
+
+  dwRtnCode = GetFinalPathNameByHandle(fileHandle, path, cchPathLen, 0);
+  if (dwRtnCode != cchPathLen - 1)
+  {
+    ret = GetLastError();
+    goto FindFileOwnerAndPermissionByHandleEnd;
+  }
+
+  dwRtnCode = FindFileOwnerAndPermission(path, TRUE, pOwnerName, pGroupName, pMask);
+  if (dwRtnCode != ERROR_SUCCESS)
+  {
+    ret = dwRtnCode;
+    goto FindFileOwnerAndPermissionByHandleEnd;
+  }
+
+FindFileOwnerAndPermissionByHandleEnd:
+  LocalFree(path);
+  return ret;
+}
+
+
 //----------------------------------------------------------------------------
 // Function: FindFileOwnerAndPermission
 //
@@ -726,6 +791,7 @@ CheckAccessEnd:
 //
 DWORD FindFileOwnerAndPermission(
   __in LPCWSTR pathName,
+  __in BOOL followLink,
   __out_opt LPWSTR *pOwnerName,
   __out_opt LPWSTR *pGroupName,
   __out_opt PINT pMask)
@@ -740,6 +806,9 @@ DWORD FindFileOwnerAndPermission(
   DWORD cbSid = SECURITY_MAX_SID_SIZE;
   PACL pDacl = NULL;
 
+  BOOL isSymlink;
+  BY_HANDLE_FILE_INFORMATION fileInformation;
+
   ACCESS_MASK ownerAccessRights = 0;
   ACCESS_MASK groupAccessRights = 0;
   ACCESS_MASK worldAccessRights = 0;
@@ -801,6 +870,28 @@ DWORD FindFileOwnerAndPermission(
 
   if (pMask == NULL) goto FindFileOwnerAndPermissionEnd;
 
+  dwRtnCode = GetFileInformationByName(pathName,
+    followLink, &fileInformation);
+  if (dwRtnCode != ERROR_SUCCESS)
+  {
+    ret = dwRtnCode;
+    goto FindFileOwnerAndPermissionEnd;
+  }
+
+  dwRtnCode = SymbolicLinkCheck(pathName, &isSymlink);
+  if (dwRtnCode != ERROR_SUCCESS)
+  {
+    ret = dwRtnCode;
+    goto FindFileOwnerAndPermissionEnd;
+  }
+
+  if (isSymlink)
+    *pMask |= UX_SYMLINK;
+  else if (IsDirFileInfo(&fileInformation))
+    *pMask |= UX_DIRECTORY;
+  else
+    *pMask |= UX_REGULAR;
+
   if ((dwRtnCode = GetEffectiveRightsForSid(pSd,
     psidOwner, &ownerAccessRights)) != ERROR_SUCCESS)
   {
diff --git a/hadoop-common-project/hadoop-common/src/main/winutils/ls.c b/hadoop-common-project/hadoop-common/src/main/winutils/ls.c
index df94e3a59d5..8e71226f03f 100644
--- a/hadoop-common-project/hadoop-common/src/main/winutils/ls.c
+++ b/hadoop-common-project/hadoop-common/src/main/winutils/ls.c
@@ -253,8 +253,6 @@ int Ls(__in int argc, __in_ecount(argc) wchar_t *argv[])
 
   LARGE_INTEGER fileSize;
 
-  BOOL isSymlink = FALSE;
-
   int ret = EXIT_FAILURE;
   int optionsMask = 0;
 
@@ -290,19 +288,8 @@ int Ls(__in int argc, __in_ecount(argc) wchar_t *argv[])
     goto LsEnd;
   }
 
-  dwErrorCode = SymbolicLinkCheck(longPathName, &isSymlink);
-  if (dwErrorCode != ERROR_SUCCESS)
-  {
-     ReportErrorCode(L"IsSymbolicLink", dwErrorCode);
-     goto LsEnd;
-  }
-
-  if (isSymlink)
-    unixAccessMode |= UX_SYMLINK;
-  else if (IsDirFileInfo(&fileInformation))
-    unixAccessMode |= UX_DIRECTORY;
-
   dwErrorCode = FindFileOwnerAndPermission(longPathName,
+    optionsMask & CmdLineOptionFollowSymlink,
     &ownerName, &groupName, &unixAccessMode);
   if (dwErrorCode != ERROR_SUCCESS)
   {
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/nativeio/TestNativeIO.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/nativeio/TestNativeIO.java
index 165fd202620..4d71e15c4b3 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/nativeio/TestNativeIO.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/nativeio/TestNativeIO.java
@@ -26,6 +26,7 @@
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import org.junit.Assert;
@@ -42,6 +43,7 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.NativeCodeLoader;
 import org.apache.hadoop.util.Time;
 
@@ -64,17 +66,23 @@ public void setupTestDir() {
 
   @Test (timeout = 30000)
   public void testFstat() throws Exception {
-    if (Path.WINDOWS) {
-      return;
-    }
-
     FileOutputStream fos = new FileOutputStream(
       new File(TEST_DIR, "testfstat"));
     NativeIO.POSIX.Stat stat = NativeIO.POSIX.getFstat(fos.getFD());
     fos.close();
     LOG.info("Stat: " + String.valueOf(stat));
 
-    assertEquals(System.getProperty("user.name"), stat.getOwner());
+    String owner = stat.getOwner();
+    String expectedOwner = System.getProperty("user.name");
+    if (Path.WINDOWS) {
+      UserGroupInformation ugi =
+          UserGroupInformation.createRemoteUser(expectedOwner);
+      final String adminsGroupString = "Administrators";
+      if (Arrays.asList(ugi.getGroupNames()).contains(adminsGroupString)) {
+        expectedOwner = adminsGroupString;
+      }
+    }
+    assertEquals(expectedOwner, owner);
     assertNotNull(stat.getGroup());
     assertTrue(!stat.getGroup().isEmpty());
     assertEquals("Stat mode field should indicate a regular file",
@@ -136,10 +144,6 @@ public void run() {
 
   @Test (timeout = 30000)
   public void testFstatClosedFd() throws Exception {
-    if (Path.WINDOWS) {
-      return;
-    }
-
     FileOutputStream fos = new FileOutputStream(
       new File(TEST_DIR, "testfstat2"));
     fos.close();