This closes #2754

apache · Apr 28, 2017 · a8a04a2 · a8a04a2
2 parents ccfff5f + c1ceab7
commit a8a04a2
Show file tree

Hide file tree

Showing 6 changed files with 56 additions and 0 deletions.
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/LocalResourceId.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/LocalResourceId.java
@@ -26,6 +26,7 @@
 import java.nio.file.Paths;
 import java.util.Objects;
 import java.util.UUID;
+import javax.annotation.Nullable;
 import org.apache.beam.sdk.io.fs.ResolveOptions;
 import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions;
 import org.apache.beam.sdk.io.fs.ResourceId;
@@ -92,6 +93,12 @@ public LocalResourceId getCurrentDirectory() {
     }
   }
 
+  @Override
+  @Nullable public String getFilename() {
+    Path fileName = getPath().getFileName();
+    return fileName == null ? null : fileName.toString();
+  }
+
   private LocalResourceId resolveLocalPath(String other, ResolveOptions resolveOptions) {
     return new LocalResourceId(
         getPath().resolve(other),

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/fs/ResourceId.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/fs/ResourceId.java
@@ -18,6 +18,7 @@
 package org.apache.beam.sdk.io.fs;
 
 import java.io.Serializable;
+import javax.annotation.Nullable;
 import org.apache.beam.sdk.io.FileSystem;
 import org.apache.beam.sdk.io.FileSystems;
 import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions;
@@ -100,6 +101,16 @@ public interface ResourceId extends Serializable {
    */
   String getScheme();
 
+
+  /**
+   * Returns the name of the file or directory denoted by this {@code ResourceId}. The file name
+   * is the farthest element from the root in the directory hierarchy.
+   *
+   * @return a string representing the name of file or directory, or null if there are zero
+   * components.
+   */
+  @Nullable String getFilename();
+
   /**
    * Returns the string representation of this {@link ResourceId}.
    *

diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/LocalResourceIdTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/LocalResourceIdTest.java
@@ -239,6 +239,17 @@ public void testToString() throws Exception {
     assertThat(dirResource.toString(), startsWith(tmpFolder.getRoot().getAbsolutePath()));
   }
 
+  @Test
+  public void testGetFilename() throws Exception {
+    assertEquals(toResourceIdentifier("/").getFilename(), null);
+    assertEquals(toResourceIdentifier("/root/tmp").getFilename(),
+        "tmp");
+    assertEquals(toResourceIdentifier("/root/tmp/").getFilename(),
+        "tmp");
+    assertEquals(toResourceIdentifier("/root/tmp/xyz.txt").getFilename(),
+        "xyz.txt");
+  }
+
   private LocalResourceId toResourceIdentifier(String str) throws Exception {
     boolean isDirectory;
     if (SystemUtils.IS_OS_WINDOWS) {

diff --git a/...google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/storage/GcsResourceId.java b/...google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/storage/GcsResourceId.java
@@ -21,6 +21,7 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 
+import javax.annotation.Nullable;
 import org.apache.beam.sdk.io.fs.ResolveOptions;
 import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions;
 import org.apache.beam.sdk.io.fs.ResourceId;
@@ -91,6 +92,16 @@ public String getScheme() {
     return "gs";
   }
 
+  @Override
+  @Nullable public String getFilename() {
+    if (gcsPath.getNameCount() <= 1) {
+      return null;
+    } else {
+      GcsPath gcsFilename = gcsPath.getFileName();
+      return gcsFilename == null ? null : gcsFilename.toString();
+    }
+  }
+
   GcsPath getGcsPath() {
     return gcsPath;
   }

diff --git a/...le-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/storage/GcsResourceIdTest.java b/...le-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/storage/GcsResourceIdTest.java
@@ -141,6 +141,17 @@ public void testEquals() throws Exception {
         toResourceIdentifier("gs://my_bucket/tmp/"));
   }
 
+  @Test
+  public void testGetFilename() throws Exception {
+    assertEquals(toResourceIdentifier("gs://my_bucket/").getFilename(), null);
+    assertEquals(toResourceIdentifier("gs://my_bucket/abc").getFilename(),
+        "abc");
+    assertEquals(toResourceIdentifier("gs://my_bucket/abc/").getFilename(),
+        "abc");
+    assertEquals(toResourceIdentifier("gs://my_bucket/abc/xyz.txt").getFilename(),
+        "xyz.txt");
+  }
+
   private GcsResourceId toResourceIdentifier(String str) throws Exception {
     return GcsResourceId.fromGcsPath(GcsPath.fromUri(str));
   }

diff --git a/sdks/java/io/hdfs/src/main/java/org/apache/beam/sdk/io/hdfs/HadoopResourceId.java b/sdks/java/io/hdfs/src/main/java/org/apache/beam/sdk/io/hdfs/HadoopResourceId.java
@@ -39,4 +39,9 @@ public ResourceId getCurrentDirectory() {
   public String getScheme() {
     throw new UnsupportedOperationException();
   }
+
+  @Override
+  public String getFilename() {
+    throw new UnsupportedOperationException();
+  }
 }