HADOOP-4494. Allow libhdfs to append to files.

(Pete Wyckoff via dhruba)
svn merge -c 743296 from trunk.



git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.20@743298 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -678,6 +678,9 @@ Release 0.19.1 - Unreleased
 
     HADOOP-5193. Correct calculation of edits modification time. (shv)
 
+    HADOOP-4494. Allow libhdfs to append to files.
+    (Pete Wyckoff via dhruba)
+
 Release 0.19.0 - 2008-11-18
 
   INCOMPATIBLE CHANGES

src/c++/libhdfs/hdfs.c

@@ -410,12 +410,17 @@ hdfsFile hdfsOpenFile(hdfsFS fs, const char* path, int flags,
       return NULL;
     }
 
+    if ((flags & O_CREAT) && (flags & O_EXCL)) {
+      fprintf(stderr, "WARN: hdfs does not truly support O_CREATE && O_EXCL\n");
+    }
 
     /* The hadoop java api/signature */
-    const char* method = ((flags & O_WRONLY) == 0) ? "open" : "create";
+    const char* method = ((flags & O_WRONLY) == 0) ? "open" : (flags & O_APPEND) ? "append" : "create";
     const char* signature = ((flags & O_WRONLY) == 0) ?
         JMETHOD2(JPARAM(HADOOP_PATH), "I", JPARAM(HADOOP_ISTRM)) :
-        JMETHOD2(JPARAM(HADOOP_PATH), "ZISJ", JPARAM(HADOOP_OSTRM));
+      (flags & O_APPEND) ?
+      JMETHOD1(JPARAM(HADOOP_PATH), JPARAM(HADOOP_OSTRM)) :
+      JMETHOD2(JPARAM(HADOOP_PATH), "ZISJ", JPARAM(HADOOP_OSTRM));
 
     /* Return value */
     hdfsFile file = NULL;
@@ -459,7 +464,7 @@ hdfsFile hdfsOpenFile(hdfsFS fs, const char* path, int flags,
         jBufferSize = jVal.i;
     }
 
-    if (flags & O_WRONLY) {
+    if ((flags & O_WRONLY) && (flags & O_APPEND) == 0) {
         //replication
 
         if (!replication) {
@@ -490,15 +495,28 @@ hdfsFile hdfsOpenFile(hdfsFS fs, const char* path, int flags,
     /* Create and return either the FSDataInputStream or
        FSDataOutputStream references jobject jStream */
 
+    // READ?
     if ((flags & O_WRONLY) == 0) {
+      if (invokeMethod(env, &jVal, &jExc, INSTANCE, jFS, HADOOP_FS,
+                       method, signature, jPath, jBufferSize)) {
+        errno = errnoFromException(jExc, env, "org.apache.hadoop.conf."
+                                   "FileSystem::%s(%s)", method,
+                                   signature);
+        goto done;
+      }
+      // WRITE/APPEND?
+      else if ((flags & O_WRONLY) && (flags & O_APPEND)) {
         if (invokeMethod(env, &jVal, &jExc, INSTANCE, jFS, HADOOP_FS,
-                         method, signature, jPath, jBufferSize)) {
-            errno = errnoFromException(jExc, env, "org.apache.hadoop.conf."
-                                       "FileSystem::%s(%s)", method,
-                                       signature);
-            goto done;
+                         method, signature, jPath)) {
+          errno = errnoFromException(jExc, env, "org.apache.hadoop.conf."
+                                     "FileSystem::%s(%s)", method,
+                                     signature);
+          goto done;
         }
+      }
+
     }
+    // WRITE/CREATE
     else {
         jboolean jOverWrite = 1;
         if (invokeMethod(env, &jVal, &jExc, INSTANCE, jFS, HADOOP_FS,

src/c++/libhdfs/hdfs.h

@@ -134,7 +134,8 @@ extern  "C" {
      * hdfsOpenFile - Open a hdfs file in given mode.
      * @param fs The configured filesystem handle.
      * @param path The full path to the file.
-     * @param flags Either O_RDONLY or O_WRONLY, for read-only or write-only.
+     * @param flags - an | of bits/fcntl.h file flags - supported flags are O_RDONLY, O_WRONLY (meaning create or overwrite i.e., implies O_TRUNCAT), 
+     * O_WRONLY|O_APPEND. Other flags are generally ignored other than (O_RDWR || (O_EXCL & O_CREAT)) which return NULL and set errno equal ENOTSUP.
      * @param bufferSize Size of buffer for read/write - pass 0 if you want
      * to use the default configured values.
      * @param replication Block replication - pass 0 if you want to use

src/c++/libhdfs/hdfs_test.c

@@ -63,7 +63,7 @@ int main(int argc, char **argv) {
         fprintf(stderr, "Oops! Failed to connect to 'local' hdfs!\n");
         exit(-1);
     } 
- 
+
         const char* writePath = "/tmp/testfile.txt";
     {
         //Write tests
@@ -317,7 +317,75 @@ int main(int argc, char **argv) {
         totalResult += (result ? 0 : 1);
     }
 
+    {
+      // TEST APPENDS
+      const char *writePath = "/tmp/appends";
+
+      // CREATE
+      hdfsFile writeFile = hdfsOpenFile(fs, writePath, O_WRONLY, 0, 0, 0);
+      if(!writeFile) {
+        fprintf(stderr, "Failed to open %s for writing!\n", writePath);
+        exit(-1);
+      }
+      fprintf(stderr, "Opened %s for writing successfully...\n", writePath);
+
+      char* buffer = "Hello,";
+      tSize num_written_bytes = hdfsWrite(fs, writeFile, (void*)buffer, strlen(buffer));
+      fprintf(stderr, "Wrote %d bytes\n", num_written_bytes);
+
+      if (hdfsFlush(fs, writeFile)) {
+        fprintf(stderr, "Failed to 'flush' %s\n", writePath); 
+        exit(-1);
+        }
+      fprintf(stderr, "Flushed %s successfully!\n", writePath); 
+
+      hdfsCloseFile(fs, writeFile);
+
+      // RE-OPEN
+      writeFile = hdfsOpenFile(fs, writePath, O_WRONLY|O_APPEND, 0, 0, 0);
+      if(!writeFile) {
+        fprintf(stderr, "Failed to open %s for writing!\n", writePath);
+        exit(-1);
+      }
+      fprintf(stderr, "Opened %s for writing successfully...\n", writePath);
 
+      buffer = " World";
+      num_written_bytes = hdfsWrite(fs, writeFile, (void*)buffer, strlen(buffer) + 1);
+      fprintf(stderr, "Wrote %d bytes\n", num_written_bytes);
+
+      if (hdfsFlush(fs, writeFile)) {
+        fprintf(stderr, "Failed to 'flush' %s\n", writePath); 
+        exit(-1);
+      }
+      fprintf(stderr, "Flushed %s successfully!\n", writePath); 
+
+      hdfsCloseFile(fs, writeFile);
+
+      // CHECK size
+      hdfsFileInfo *finfo = hdfsGetPathInfo(fs, writePath);
+      fprintf(stderr, "fileinfo->mSize: == total %s\n", ((result = (finfo->mSize == strlen("Hello, World") + 1)) ? "Success!" : "Failed!"));
+      totalResult += (result ? 0 : 1);
+
+      // READ and check data
+      hdfsFile readFile = hdfsOpenFile(fs, writePath, O_RDONLY, 0, 0, 0);
+      if (!readFile) {
+        fprintf(stderr, "Failed to open %s for reading!\n", writePath);
+        exit(-1);
+      }
+
+      char rdbuffer[32];
+      tSize num_read_bytes = hdfsRead(fs, readFile, (void*)rdbuffer, sizeof(rdbuffer));
+      fprintf(stderr, "Read following %d bytes:\n%s\n", 
+              num_read_bytes, rdbuffer);
+
+      fprintf(stderr, "read == Hello, World %s\n", (result = (strcmp(rdbuffer, "Hello, World") == 0)) ? "Success!" : "Failed!");
+
+      hdfsCloseFile(fs, readFile);
+
+      // DONE test appends
+    }
+      
+      
     totalResult += (hdfsDisconnect(fs) != 0);
 
     {