fix for relative local file urls causing npe

ClusterlessHQ · Jun 24, 2023 · 338a01a · 338a01a
1 parent db6eed0
commit 338a01a
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 8 deletions.
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/factory/ManifestReader.java b/tessellate-main/src/main/java/io/clusterless/tessellate/factory/ManifestReader.java
@@ -48,19 +48,19 @@ public static ManifestReader from(Dataset dataset) {
 
     public ManifestReader(Source source) {
         this.manifestURI = source.manifest();
-        this.uris = source.uris();
+        this.uris = clean(source.uris());
         this.numPartitions = source.partitions().size();
     }
 
     public ManifestReader(List<URI> uris) {
-        this.uris = uris;
+        this.uris = clean(uris);
         this.manifestURI = null;
         this.numPartitions = 0;
     }
 
     public List<URI> uris(Properties conf) throws IOException {
         if (manifestURI == null) {
-            return dedupe(uris);
+            return uris;
         }
 
         if (manifestUris != null) {
@@ -88,7 +88,7 @@ public List<URI> uris(Properties conf) throws IOException {
             throw new IllegalStateException("manifest: " + manifestURI + ", is empty");
         }
 
-        manifestUris = dedupe(found);
+        manifestUris = clean(found);
 
         return manifestUris;
     }
@@ -119,9 +119,10 @@ public URI findCommonRoot(Properties conf) throws IOException {
         return URI.create(commonPrefix);
     }
 
-    protected List<URI> dedupe(List<URI> uris) {
+    protected List<URI> clean(List<URI> uris) {
         List<URI> distinct = uris.stream()
                 .map(URIs::copyWithoutQuery)
+                .map(URIs::makeAbsolute)
                 .distinct()
                 .collect(Collectors.toList());
 

diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/factory/ManifestWriter.java b/tessellate-main/src/main/java/io/clusterless/tessellate/factory/ManifestWriter.java
@@ -19,6 +19,7 @@
 import io.clusterless.tessellate.model.Dataset;
 import io.clusterless.tessellate.model.Sink;
 import io.clusterless.tessellate.util.JSONUtil;
+import io.clusterless.tessellate.util.URIs;
 
 import java.io.IOException;
 import java.net.URI;
@@ -69,7 +70,7 @@ public ManifestWriter(Sink dataset, URI uriPrefix) {
     public ManifestWriter(URI manifestURI, String lot, URI uriPrefix) {
         this.manifestURI = manifestURI;
         this.lot = lot;
-        this.uriPrefix = uriPrefix;
+        this.uriPrefix = URIs.makeAbsolute(uriPrefix);
 
         if (this.lot == null) {
             throw new IllegalArgumentException("lot is required when manifest is set");

diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/factory/hdfs/FSFactory.java b/tessellate-main/src/main/java/io/clusterless/tessellate/factory/hdfs/FSFactory.java
@@ -62,7 +62,7 @@ public int openWritesThreshold() {
     }
 
     @Override
-    protected Tap createTap(PipelineOptions pipelineOptions, Dataset dataset, Fields currentFields) throws IOException {
+    protected Tap<Properties, ?, ?> createTap(PipelineOptions pipelineOptions, Dataset dataset, Fields currentFields) throws IOException {
         boolean isSink = isSink(dataset);
 
         Fields declaredFields = declaredFields(dataset, currentFields);

diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/util/URIs.java b/tessellate-main/src/main/java/io/clusterless/tessellate/util/URIs.java
@@ -88,6 +88,14 @@ public static URI trim(URI uri, int trim) {
     }
 
     public static URI cleanFileUrls(URI uri) {
-        return uri.getScheme().equals("file") ? URIs.copyWithHost(uri, "") : uri.normalize();
+        return "file".equals(uri.getScheme()) ? URIs.copyWithHost(uri, "") : uri.normalize();
+    }
+
+    public static URI makeAbsolute(URI uri) {
+        if (uri.isAbsolute()) {
+            return uri;
+        }
+
+        return Paths.get(uri.getPath()).toAbsolutePath().toUri();
     }
 }