MLE-17084 Fix for streaming files with spaces in filename

We can't encode the URI due to the PUT bug, but this at least fixes streaming so that it can read and write a file with spaces in its 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));
     }