Merge pull request #295 from marklogic/feature/17084-part-3

MLE-17084 Fix for streaming files with spaces in filename

Author: rjrudin

src/main/java/com/marklogic/spark/reader/file/ArchiveFileReader.java

@@ -87,7 +87,7 @@ public void close() {
     }
 
     private void openNextFile() {
-        this.currentFilePath = fileContext.getDecodedFilePath(filePartition, nextFilePathIndex);
+        this.currentFilePath = fileContext.decodeFilePath(filePartition.getPaths().get(nextFilePathIndex));
         nextFilePathIndex++;
         this.currentZipInputStream = new ZipInputStream(fileContext.openFile(this.currentFilePath));
     }

src/main/java/com/marklogic/spark/reader/file/FileContext.java

@@ -72,8 +72,7 @@ byte[] readBytes(InputStream inputStream) throws IOException {
         return this.encoding != null ? new String(bytes, this.encoding).getBytes() : bytes;
     }
 
-    public String getDecodedFilePath(FilePartition filePartition, int index) {
-        String path = filePartition.getPaths().get(index);
+    public String decodeFilePath(String path) {
         try {
             if (this.encoding != null) {
                 return URLDecoder.decode(path, this.encoding);

src/main/java/com/marklogic/spark/reader/file/GenericFileReader.java

@@ -41,7 +41,11 @@ public boolean next() {
             return false;
         }
 
-        final String path = fileContext.getDecodedFilePath(filePartition, filePathIndex);
+        // If streaming, we want to put the unaltered file path in the row. The writer can then decode it and also use
+        // its original value as the URI, as the PUT v1/documents endpoint does not allow e.g. spaces.
+        final String originalFilePath = filePartition.getPaths().get(filePathIndex);
+        final String path = this.isStreaming ? originalFilePath : fileContext.decodeFilePath(originalFilePath);
+
         filePathIndex++;
         try {
             byte[] content = this.isStreaming ? serializeFileContext() : readFileIntoByteArray(path);

src/main/java/com/marklogic/spark/reader/file/GzipFileReader.java

@@ -38,7 +38,7 @@ public boolean next() {
             return false;
         }
 
-        String currentFilePath = fileContext.getDecodedFilePath(filePartition, nextFilePathIndex);
+        String currentFilePath = fileContext.decodeFilePath(filePartition.getPaths().get(nextFilePathIndex));
         nextFilePathIndex++;
         InputStream gzipInputStream = null;
         try {

src/main/java/com/marklogic/spark/reader/file/MlcpArchiveFileReader.java

@@ -99,7 +99,7 @@ public void close() {
     }
 
     private void openNextFile() {
-        this.currentFilePath = fileContext.getDecodedFilePath(filePartition, nextFilePathIndex);
+        this.currentFilePath = fileContext.decodeFilePath(filePartition.getPaths().get(nextFilePathIndex));
         nextFilePathIndex++;
         this.currentZipInputStream = new ZipInputStream(fileContext.openFile(this.currentFilePath));
     }

src/main/java/com/marklogic/spark/reader/file/RdfFileReader.java

@@ -76,7 +76,7 @@ public void close() throws IOException {
     }
 
     private boolean initializeRdfStreamReader() {
-        this.currentFilePath = fileContext.getDecodedFilePath(filePartition, nextFilePathIndex);
+        this.currentFilePath = fileContext.decodeFilePath(filePartition.getPaths().get(nextFilePathIndex));
         if (logger.isDebugEnabled()) {
             logger.debug("Reading file {}", this.currentFilePath);
         }

src/main/java/com/marklogic/spark/reader/file/RdfZipFileReader.java

@@ -68,7 +68,7 @@ public boolean next() {
             }
 
             // Open up the next zip.
-            this.currentFilePath = fileContext.getDecodedFilePath(filePartition, nextFilePathIndex);
+            this.currentFilePath = fileContext.decodeFilePath(filePartition.getPaths().get(nextFilePathIndex));
             nextFilePathIndex++;
             this.currentZipInputStream = new CustomZipInputStream(fileContext.openFile(currentFilePath));
             return next();

src/main/java/com/marklogic/spark/reader/file/ZipFileReader.java

@@ -71,7 +71,7 @@ public void close() {
     }
 
     private void openNextFile() {
-        this.currentFilePath = fileContext.getDecodedFilePath(filePartition, nextFilePathIndex);
+        this.currentFilePath = fileContext.decodeFilePath(filePartition.getPaths().get(nextFilePathIndex));
         nextFilePathIndex++;
         this.currentZipInputStream = new ZipInputStream(fileContext.openFile(this.currentFilePath));
     }

src/main/java/com/marklogic/spark/reader/file/xml/AggregateXmlFileReader.java

@@ -53,7 +53,7 @@ public boolean next() {
             }
 
             try {
-                String path = fileContext.getDecodedFilePath(filePartition, filePathIndex);
+                String path = fileContext.decodeFilePath(filePartition.getPaths().get(filePathIndex));
                 nextRowToReturn = this.aggregateXMLSplitter.nextRow(path);
                 return true;
             } catch (RuntimeException ex) {
@@ -81,7 +81,7 @@ private boolean initializeAggregateXMLSplitter() {
             return false;
         }
 
-        final String filePath = fileContext.getDecodedFilePath(filePartition, filePathIndex);
+        final String filePath = fileContext.decodeFilePath(filePartition.getPaths().get(filePathIndex));
         try {
             this.inputStream = fileContext.openFile(filePath);
             String identifierForError = "file " + filePath;

src/main/java/com/marklogic/spark/reader/file/xml/ZipAggregateXmlFileReader.java

@@ -83,7 +83,7 @@ public void close() {
     }
 
     private void openNextFile() {
-        this.currentFilePath = fileContext.getDecodedFilePath(filePartition, nextFilePathIndex);
+        this.currentFilePath = fileContext.decodeFilePath(filePartition.getPaths().get(nextFilePathIndex));
         nextFilePathIndex++;
         this.currentZipInputStream = new ZipInputStream(fileContext.openFile(this.currentFilePath));
     }