WIP Metadata database - 560a8cf05d

Modified pom.xml

@@ -32,8 +32,12
<jackson.version>2.9.8</jackson.version>
<logback.version>1.2.3</logback.version>
<lucene.version>7.6.0</lucene.version>
+ <flatbuffers.version>1.10.0.2</flatbuffers.version>
+ <flatbuffers.sources>${basedir}/src/main/flatbuffers</flatbuffers.sources>
+ <flatbuffers.generated.sources>${project.build.directory}/generated-sources/java</flatbuffers.generated.sources>
<junit.jupiter.version>5.3.2</junit.jupiter.version>
<resilience4j.version>0.13.2</resilience4j.version>
+ <lmdb.version>0.6.2</lmdb.version>
</properties>

<dependencyManagement>

@@ -80,6 +84,12
</dependency>

<dependency>
+ <groupId>com.github.davidmoten</groupId>
+ <artifactId>flatbuffers-java</artifactId>
+ <version>${flatbuffers.version}</version>
+ </dependency>
+
+ <dependency>
<groupId>org.immutables</groupId>
<artifactId>value</artifactId>
<version>${immutables.version}</version>

@@ -100,6 +110,12
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>${logback.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.lmdbjava</groupId>
+ <artifactId>lmdbjava</artifactId>
+ <version>${lmdb.version}</version>
</dependency>

<dependency>

@@ -185,6 +201,84
</build>

<profiles>
+ <profile>
+ <id>flatbuffers</id>
+ <build>
+ <plugins>
+ 
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <version>3.1.1</version>
+ <executions>
+ <execution>
+ <id>unpack</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>unpack</goal>
+ </goals>
+ <configuration>
+ <artifactItems>
+ <artifactItem>
+ <groupId>com.github.davidmoten</groupId>
+ <artifactId>flatbuffers-compiler</artifactId>
+ <version>${flatbuffers.version}</version>
+ <type>tar.gz</type>
+ <classifier>distribution-linux</classifier>
+ <overWrite>true</overWrite>
+ <outputDirectory>${project.build.directory}</outputDirectory>
+ </artifactItem>
+ </artifactItems>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>exec-maven-plugin</artifactId>
+ <version>1.6.0</version>
+ <executions>
+ <execution>
+ <goals>
+ <goal>exec</goal>
+ </goals>
+ <phase>generate-sources</phase>
+ </execution>
+ </executions>
+ <configuration>
+ <executable>${project.build.directory}/bin/flatc</executable>
+ <workingDirectory>${flatbuffers.sources}</workingDirectory>
+ <arguments>
+ <argument>--java</argument>
+ <argument>-o</argument>
+ <argument>${flatbuffers.generated.sources}</argument>
+ <argument>recipe.fbs</argument>
+ </arguments>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <version>3.0.0</version>
+ <executions>
+ <execution>
+ <id>add-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <sources>
+ <source>${flatbuffers.generated.sources}</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ 
+ </plugins>
+ </build>
+ </profile>
<profile>
<id>index</id>
<build>

Created src/main/flatbuffers/recipe.fbs

Created src/main/java/co/caio/cerberus/LMDBFlatBuffersExperiment.java

@@ -1,0 +1,154
+package co.caio.cerberus;
+
+import co.caio.cerberus.flatbuffers.FlatRecipe;
+import co.caio.cerberus.model.Recipe;
+import com.google.flatbuffers.FlatBufferBuilder;
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Optional;
+import java.util.stream.Stream;
+import org.lmdbjava.DbiFlags;
+import org.lmdbjava.Env;
+import org.lmdbjava.EnvFlags;
+import reactor.core.publisher.Flux;
+
+public class LMDBFlatBuffersExperiment {
+
+ static class DirectByteBufferFactory implements FlatBufferBuilder.ByteBufferFactory {
+ @Override
+ public ByteBuffer newByteBuffer(int capacity) {
+ return ByteBuffer.allocateDirect(capacity).order(ByteOrder.LITTLE_ENDIAN);
+ }
+ }
+
+ private static final Serializer serializer = new Serializer();
+ private static final String recipesFilename = "tmp/document.jsonlines";
+
+ private static Stream<Recipe> recipeStream() throws Exception {
+ return Files.lines(Path.of(recipesFilename))
+ .map(serializer::readRecipe)
+ .flatMap(Optional::stream);
+ }
+
+ private static void stats(File dest) throws Exception {
+ var env = Env.open(dest, 2_000, EnvFlags.MDB_RDONLY_ENV);
+ var db = env.openDbi("recipe");
+
+ try (var tnx = env.txnRead()) {
+ System.out.println(db.stat(tnx));
+ }
+ }
+
+ private static void readDb(File dest) throws Exception {
+ var env = Env.open(dest, 2_000, EnvFlags.MDB_RDONLY_ENV);
+ var db = env.openDbi("recipe");
+
+ var bbKey = ByteBuffer.allocateDirect(Long.SIZE);
+ recipeStream()
+ .forEach(
+ recipe -> {
+ try (var txn = env.txnRead()) {
+ bbKey.putLong(recipe.recipeId());
+ bbKey.flip();
+
+ var result = db.get(txn, bbKey);
+
+ if (result == null) {
+ System.out.println("Couldn't find recipe " + recipe.recipeId());
+ return;
+ // throw new RuntimeException("Couldn't find recipe " + recipe.recipeId());
+ }
+
+ var flatRecipe = FlatRecipe.getRootAsFlatRecipe(result);
+
+ if (flatRecipe.id() != recipe.recipeId()) {
+ throw new RuntimeException(
+ "Uh, reading recipe " + recipe.recipeId() + " gave me recipe " + flatRecipe);
+ }
+
+ var flatIngSize = flatRecipe.ingredientsLength();
+ if (flatIngSize != recipe.ingredients().size()) {
+ throw new RuntimeException(
+ String.format(
+ "Expected %d ingredients, got %d",
+ recipe.ingredients().size(), flatIngSize));
+ }
+
+ for (int i = 0; i < flatIngSize; i++) {
+ if (! flatRecipe.ingredients(i).equals(recipe.ingredients().get(i))) {
+ throw new RuntimeException(
+ String.format(
+ "Expected ingredient to be '%s' but got '%s'",
+ recipe.ingredients().get(i), flatRecipe.ingredients(i)));
+ }
+ }
+ }
+ });
+ }
+
+ private static ByteBuffer flattenRecipe(Recipe recipe) {
+ var builder = new FlatBufferBuilder(5_000, new DirectByteBufferFactory());
+
+ var nameOffset = builder.createString(recipe.name());
+ var sourceOffset = builder.createString(recipe.crawlUrl());
+
+ var ingredientsOffsets =
+ recipe.ingredients().stream().mapToInt(builder::createString).toArray();
+ var ingredientsVectorOffset = FlatRecipe.createIngredientsVector(builder, ingredientsOffsets);
+
+ var rootTable =
+ FlatRecipe.createFlatRecipe(
+ builder,
+ recipe.recipeId(),
+ nameOffset,
+ sourceOffset,
+ ingredientsVectorOffset,
+ recipe.totalTime().orElse(0),
+ recipe.calories().orElse(0));
+
+ builder.finish(rootTable);
+ return builder.dataBuffer();
+ }
+
+ private static void createDb(File dest) throws Exception {
+ var env = Env.open(dest, 3_000);
+ var db = env.openDbi("recipe", DbiFlags.MDB_CREATE);
+
+ var bbKey = ByteBuffer.allocateDirect(Long.SIZE);
+
+ Flux.fromStream(recipeStream())
+ .buffer(250_000)
+ .subscribe(
+ recipes -> {
+ try (var tnx = env.txnWrite()) {
+ for (Recipe r : recipes) {
+ bbKey.putLong(r.recipeId());
+ bbKey.flip();
+ db.put(tnx, bbKey, flattenRecipe(r));
+ }
+ tnx.commit();
+ System.out.println("Wrote a batch of " + recipes.size() + " recipes");
+ }
+ });
+
+ db.close();
+ env.close();
+ System.out.println("Done!");
+ }
+
+ private static void lmdbLoop() throws Exception {
+ var tmpDir = new File("tmp/lmdb-test");
+ System.out.println("Create DB at " + tmpDir);
+ createDb(tmpDir);
+ System.out.println("Read DB from " + tmpDir);
+ readDb(tmpDir);
+ }
+
+ public static void main(String[] args) throws Exception {
+ //lmdbLoop();
+ stats(new File("tmp/lmdb-test"));
+ }
+}