Merge pull request #296 from marklogic/feature/17095-swap-order-of-archive

MLE-17059 Changing order of entries in archive

Author: rjrudin

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

@@ -29,6 +29,9 @@ class ArchiveFileReader implements PartitionReader<InternalRow> {
     private int nextFilePathIndex;
     private InternalRow nextRowToReturn;
 
+    // Legacy = content first, then metadata.
+    private Boolean isLegacyFormat;
+
     ArchiveFileReader(FilePartition filePartition, FileContext fileContext) {
         this.filePartition = filePartition;
         this.fileContext = fileContext;
@@ -45,15 +48,23 @@ class ArchiveFileReader implements PartitionReader<InternalRow> {
     @Override
     public boolean next() {
         try {
-            ZipEntry contentZipEntry = FileUtil.findNextFileEntry(currentZipInputStream);
-            if (contentZipEntry == null) {
+            ZipEntry nextZipEntry = FileUtil.findNextFileEntry(currentZipInputStream);
+            if (nextZipEntry == null) {
                 return openNextFileAndReadNextEntry();
             }
+
+            if (isLegacyFormat == null) {
+                isLegacyFormat = !nextZipEntry.getName().endsWith(".metadata");
+            }
+            if (!isLegacyFormat) {
+                return readMetadataFollowedByContentEntry();
+            }
+
             byte[] content = fileContext.readBytes(currentZipInputStream);
             if (content == null || content.length == 0) {
                 return openNextFileAndReadNextEntry();
             }
-            final String zipEntryName = contentZipEntry.getName();
+            final String zipEntryName = nextZipEntry.getName();
 
             byte[] metadataBytes = readMetadataEntry(zipEntryName);
             if (metadataBytes == null || metadataBytes.length == 0) {
@@ -86,6 +97,24 @@ public void close() {
         IOUtils.closeQuietly(this.currentZipInputStream);
     }
 
+    private boolean readMetadataFollowedByContentEntry() throws IOException {
+        byte[] metadataBytes = fileContext.readBytes(currentZipInputStream);
+        if (metadataBytes == null || metadataBytes.length == 0) {
+            return openNextFileAndReadNextEntry();
+        }
+
+        ZipEntry contentZipEntry = FileUtil.findNextFileEntry(currentZipInputStream);
+        byte[] content = fileContext.readBytes(currentZipInputStream);
+
+        DocumentMetadataHandle metadata = new DocumentMetadataHandle();
+        metadata.fromBuffer(metadataBytes);
+        this.nextRowToReturn = new DocumentRowBuilder(this.metadataCategories)
+            .withUri(contentZipEntry.getName())
+            .withContent(content).withMetadata(metadata)
+            .buildRow();
+        return true;
+    }
+
     private void openNextFile() {
         this.currentFilePath = fileContext.decodeFilePath(filePartition.getPaths().get(nextFilePathIndex));
         nextFilePathIndex++;

src/main/java/com/marklogic/spark/writer/file/ZipFileWriter.java

@@ -58,14 +58,16 @@ public void write(InternalRow row) throws IOException {
         }
         final String uri = row.getString(0);
         final String entryName = FileUtil.makePathFromDocumentURI(uri);
-        zipOutputStream.putNextEntry(new ZipEntry(entryName));
-        this.contentWriter.writeContent(row, zipOutputStream);
-        zipEntryCounter++;
+
         if (hasMetadata(row)) {
             zipOutputStream.putNextEntry(new ZipEntry(entryName + ".metadata"));
             this.contentWriter.writeMetadata(row, zipOutputStream);
             zipEntryCounter++;
         }
+
+        zipOutputStream.putNextEntry(new ZipEntry(entryName));
+        this.contentWriter.writeContent(row, zipOutputStream);
+        zipEntryCounter++;
     }
 
     @Override

src/test/java/com/marklogic/spark/writer/file/WriteArchiveTest.java

@@ -64,6 +64,12 @@ private void verifyMetadataFiles(Path tempDir, String metadataValue) {
 
         assertEquals(4, rows.size(), "Expecting the 2 author JSON entries and 2 entries for metadata.");
 
+        for (int i = 0; i < 4; i += 2) {
+            String entryName = rows.get(i).getString(0);
+            assertTrue(entryName.endsWith(".metadata"), "The metadata should come before the content entry. " +
+                "This allows for the content to later be streamed back into MarkLogic. Entry name: " + entryName);
+        }
+
         final String expectedUriPrefix = "file://" + tempDir.toFile().getAbsolutePath();
         for (Row row : rows) {
             String uri = row.getString(0);