Merge pull request #710 from jpmorganchase/issue-705

Fix #705 out of memory error thrown by Files.readAllBytes
Consensys · Apr 23, 2019 · 7510c03 · 7510c03
2 parents f3a2841 + e969c03
commit 7510c03
Show file tree

Hide file tree

Showing 20 changed files with 161 additions and 51 deletions.
diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/BdbDumpFile.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/BdbDumpFile.java
@@ -1,7 +1,9 @@
 package com.quorum.tessera.data.migration;
 
 import java.io.BufferedReader;
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.*;
@@ -18,9 +20,9 @@
 public class BdbDumpFile implements StoreLoader {
 
     @Override
-    public Map<byte[], byte[]> load(Path inputFile) throws IOException {
+    public Map<byte[], InputStream> load(Path inputFile) throws IOException {
 
-        Map<byte[], byte[]> results = new HashMap<>();
+        Map<byte[], InputStream> results = new HashMap<>();
 
         try (BufferedReader reader = Files.newBufferedReader(inputFile)) {
 
@@ -38,8 +40,11 @@ public Map<byte[], byte[]> load(Path inputFile) throws IOException {
 
                 final String value = reader.readLine();
 
-
-                results.put(Base64.getDecoder().decode(Hex.decode(key)), Hex.decode(value));
+                InputStream inputStream = Optional.of(value)
+                    .map(Hex::decode)
+                    .map(ByteArrayInputStream::new)
+                    .get();
+                results.put(Base64.getDecoder().decode(Hex.decode(key)), inputStream);
             }
             return Collections.unmodifiableMap(results);
 

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/CmdLineExecutor.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/CmdLineExecutor.java
@@ -113,7 +113,7 @@ protected static int execute(String... args) throws Exception {
         final StoreLoader storeLoader = StoreLoader.create(storeType);
 
         final Path inputpath = Paths.get(line.getOptionValue("inputpath"));
-        final Map<byte[], byte[]> data = storeLoader.load(inputpath);
+        final Map<byte[], InputStream> data = storeLoader.load(inputpath);
 
         final String username = line.getOptionValue("dbuser");
         final String password = line.getOptionValue("dbpass");

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/DataExporter.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/DataExporter.java
@@ -1,11 +1,12 @@
 package com.quorum.tessera.data.migration;
 
+import java.io.InputStream;
 import java.nio.file.Path;
 import java.sql.SQLException;
 import java.util.Map;
 
 public interface DataExporter {
 
-    void export(Map<byte[], byte[]> data, Path output, String username, String password) throws SQLException;
+    void export(Map<byte[], InputStream> data, Path output, String username, String password) throws SQLException;
 
 }
diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/DirectoryStoreFile.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/DirectoryStoreFile.java
@@ -2,6 +2,7 @@
 
 import com.quorum.tessera.io.FilesDelegate;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Map;
@@ -15,7 +16,7 @@ public class DirectoryStoreFile implements StoreLoader {
     private final FilesDelegate fileDelegate = FilesDelegate.create();
 
     @Override
-    public Map<byte[], byte[]> load(Path directory) throws IOException {
+    public Map<byte[], InputStream> load(Path directory) throws IOException {
 
         Optional.ofNullable(directory)
                 .filter(p -> p.toFile().isDirectory())
@@ -24,7 +25,7 @@ public Map<byte[], byte[]> load(Path directory) throws IOException {
         try (Stream<Path> stream = Files.list(directory)) {
             return stream.collect(Collectors.toMap(
                     p -> new Base32().decode(p.toFile().getName()),
-                    p -> fileDelegate.readAllBytes(p)));
+                    p -> fileDelegate.newInputStream(p)));
         }
     }
 

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/H2DataExporter.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/H2DataExporter.java
@@ -1,5 +1,6 @@
 package com.quorum.tessera.data.migration;
 
+import java.io.InputStream;
 import java.net.URL;
 import java.nio.file.Path;
 import java.sql.SQLException;
@@ -10,7 +11,7 @@ public class H2DataExporter implements DataExporter {
     private static final String INSERT_ROW = "INSERT INTO ENCRYPTED_TRANSACTION (HASH,ENCODED_PAYLOAD) VALUES (?,?)";
 
     @Override
-    public void export(final Map<byte[], byte[]> data,
+    public void export(final Map<byte[], InputStream> data,
                        final Path output,
                        final String username,
                        final String password) throws SQLException {

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/JdbcDataExporter.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/JdbcDataExporter.java
@@ -2,6 +2,7 @@
 
 import com.quorum.tessera.io.IOCallback;
 import com.quorum.tessera.io.UriCallback;
+import java.io.InputStream;
 
 import java.net.URL;
 import java.nio.file.Files;
@@ -29,7 +30,7 @@ public JdbcDataExporter(String jdbcUrl, String insertRow, URL ddl) {
     }
 
     @Override
-    public void export(Map<byte[], byte[]> data, Path output, String username, String password) throws SQLException {
+    public void export(Map<byte[], InputStream> data, Path output, String username, String password) throws SQLException {
 
         try (Connection conn = DriverManager.getConnection(jdbcUrl, username, password)) {
 
@@ -40,9 +41,9 @@ public void export(Map<byte[], byte[]> data, Path output, String username, Strin
             }
 
             try (PreparedStatement insertStatement = conn.prepareStatement(insertRow)) {
-                for (Entry<byte[], byte[]> values : data.entrySet()) {
+                for (Entry<byte[], InputStream> values : data.entrySet()) {
                     insertStatement.setBytes(1, values.getKey());
-                    insertStatement.setBytes(2, values.getValue());
+                    insertStatement.setBinaryStream(2, values.getValue());
                     insertStatement.execute();
                 }
             }

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/SqliteDataExporter.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/SqliteDataExporter.java
@@ -1,25 +1,53 @@
 package com.quorum.tessera.data.migration;
 
-import java.net.URL;
+import com.quorum.tessera.io.IOCallback;
+import com.quorum.tessera.io.UriCallback;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
 import java.util.Map;
 
 public class SqliteDataExporter implements DataExporter {
 
     private static final String INSERT_ROW = "INSERT INTO ENCRYPTED_TRANSACTION (HASH,ENCODED_PAYLOAD) VALUES (?,?)";
 
     @Override
-    public void export(Map<byte[], byte[]> data, Path output, final String username, final String password) throws SQLException {
+    public void export(Map<byte[], InputStream> data, Path output, String username, String password) throws SQLException {
 
         final String connectionString = "jdbc:sqlite:" + output.toString();
 
-        final URL sqlFile = getClass().getResource("/ddls/h2-ddl.sql");
+        final URI sqlFile = UriCallback.execute(() -> getClass().getResource("/ddls/sqlite-ddl.sql").toURI());
 
-        final JdbcDataExporter jdbcDataExporter = new JdbcDataExporter(connectionString, INSERT_ROW, sqlFile);
+        List<String> createTables = IOCallback.execute(() -> Files.readAllLines(Paths.get(sqlFile)));
 
-        jdbcDataExporter.export(data, output, username, password);
+        try (Connection conn = DriverManager.getConnection(connectionString, username, password)) {
 
+            try (Statement stmt = conn.createStatement()) {
+                for (String createTable : createTables) {
+                    stmt.executeUpdate(createTable);
+                }
+            }
+
+            try (PreparedStatement insertStatement = conn.prepareStatement(INSERT_ROW)) {
+                for (Map.Entry<byte[], InputStream> values : data.entrySet()) {
+                    insertStatement.setBytes(1, values.getKey());
+
+                    insertStatement.setBytes(2, Utils.toByteArray(values.getValue()));
+
+                    insertStatement.execute();
+                }
+            }
+
+        }
     }
 
+
 }
diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/SqliteLoader.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/SqliteLoader.java
@@ -1,6 +1,8 @@
 package com.quorum.tessera.data.migration;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.file.Path;
 import java.sql.Connection;
 import java.sql.DriverManager;
@@ -13,7 +15,7 @@
 public class SqliteLoader implements StoreLoader {
 
     @Override
-    public Map<byte[], byte[]> load(Path input) throws IOException {
+    public Map<byte[], InputStream> load(Path input) throws IOException {
 
         final String url = "jdbc:sqlite:" + input.toString();
 
@@ -23,11 +25,11 @@ public Map<byte[], byte[]> load(Path input) throws IOException {
                     Statement statement = conn.createStatement();
                     ResultSet results = statement.executeQuery("SELECT * FROM payload")) {
 
-                Map<byte[], byte[]> loadedData = new HashMap<>();
+                Map<byte[], InputStream> loadedData = new HashMap<>();
                 while (results.next()) {
 
                     byte[] key = results.getBytes("key");
-                    byte[] value = results.getBytes("bytes");
+                    InputStream value = new ByteArrayInputStream(results.getBytes("bytes"));
 
                     loadedData.put(key, value);
 

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/StoreLoader.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/StoreLoader.java
@@ -1,6 +1,7 @@
 package com.quorum.tessera.data.migration;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.file.Path;
 import java.util.Collections;
 import java.util.HashMap;
@@ -9,18 +10,23 @@
 
 public interface StoreLoader {
 
-    Map<byte[], byte[]> load(Path input) throws IOException;
+
+
+    Map<byte[], InputStream> load(Path input) throws IOException;
 
     Map<StoreType, StoreLoader> LOOKUP = Collections.unmodifiableMap(
-        new HashMap<StoreType, StoreLoader>() {{
+        new HashMap<StoreType, StoreLoader>() {
+        {
             put(StoreType.BDB, new BdbDumpFile());
             put(StoreType.DIR, new DirectoryStoreFile());
             put(StoreType.SQLITE, new SqliteLoader());
-        }}
+        }
+    }
     );
 
     static StoreLoader create(StoreType storeType) {
         return Optional.ofNullable(LOOKUP.get(storeType)).get();
     }
 
+
 }
diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/Utils.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/Utils.java
@@ -0,0 +1,22 @@
+package com.quorum.tessera.data.migration;
+
+import com.quorum.tessera.io.IOCallback;
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+
+public interface Utils {
+
+    static byte[] toByteArray(InputStream in) {
+        return IOCallback.execute(() -> {
+            ByteArrayOutputStream os = new ByteArrayOutputStream();
+
+            byte[] buffer = new byte[1024];
+            int len;
+            while ((len = in.read(buffer)) != -1) {
+                os.write(buffer, 0, len);
+            }
+
+            return os.toByteArray();
+        });
+    }
+}
diff --git a/data-migration/src/test/java/com/quorum/tessera/data/migration/BdbDumpFileTest.java b/data-migration/src/test/java/com/quorum/tessera/data/migration/BdbDumpFileTest.java
@@ -1,6 +1,7 @@
 package com.quorum.tessera.data.migration;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.net.URISyntaxException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -16,7 +17,7 @@ public void loadSample() throws URISyntaxException, IOException {
 
         Path inputFile = Paths.get(getClass().getResource("/bdb/bdb-sample.txt").toURI());
 
-        Map<byte[], byte[]> results = new BdbDumpFile().load(inputFile);
+        Map<byte[], InputStream> results = new BdbDumpFile().load(inputFile);
 
         assertThat(results).hasSize(12);
 
@@ -29,7 +30,7 @@ public void loadSimpleEntrySample() throws URISyntaxException, IOException {
 
         Path inputFile = Paths.get(getClass().getResource("/bdb/single-entry.txt").toURI());
 
-        Map<byte[], byte[]> results = new BdbDumpFile().load(inputFile);
+        Map<byte[], InputStream> results = new BdbDumpFile().load(inputFile);
 
         assertThat(results).hasSize(1);
 

diff --git a/data-migration/src/test/java/com/quorum/tessera/data/migration/DirectoryStoreFileTest.java b/data-migration/src/test/java/com/quorum/tessera/data/migration/DirectoryStoreFileTest.java
@@ -1,24 +1,53 @@
 package com.quorum.tessera.data.migration;
 
+import java.io.InputStream;
+import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+
 import static org.assertj.core.api.Assertions.assertThat;
+
 import org.junit.Test;
 
 public class DirectoryStoreFileTest {
-    
+
     @Test
     public void load() throws Exception {
-    
+
         Path directory = Paths.get(getClass().getResource("/dir/").toURI());
-        
+
         DirectoryStoreFile directoryStoreFile = new DirectoryStoreFile();
-        
-        Map<byte[],byte[]> results = directoryStoreFile.load(directory);
-        
+
+        Map<byte[], InputStream> results = directoryStoreFile.load(directory);
+
         assertThat(results).hasSize(22);
+
+    }
+
+    @Test
+    public void loadLarge() throws Exception {
+
+        Path baseDir = Paths.get(getClass().getResource("/").toURI());
 
+        Path directory = baseDir.resolve(UUID.randomUUID().toString());
+
+        Files.createDirectories(directory);
+
+        Path largeFile = Paths.get(directory.toAbsolutePath().toString(), "loadLarge");
+
+        Random random = new Random();
+        byte[] data = new byte[33554432];
+        random.nextBytes(data);
+        Files.write(largeFile, data);
+
+        DirectoryStoreFile directoryStoreFile = new DirectoryStoreFile();
+
+        Map results = directoryStoreFile.load(directory);
+
+        assertThat(results).hasSize(1);
     }
-    
+
 }