Support PostgreSQL and CockroachDB backends

.travis.yml

@@ -80,6 +80,7 @@ env:
     - BACKEND=mysql
     - BACKEND=hbase
     - BACKEND=rocksdb
+    - BACKEND=postgresql
   global:
     - RELEASE_BRANCH=^release-.*$
     - RELEASE_TAG=^v[0-9]\..*$

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/TableDefine.java

@@ -29,20 +29,37 @@
 import com.baidu.hugegraph.type.define.HugeKeys;
 import com.baidu.hugegraph.util.InsertionOrderUtil;
 
+import jersey.repackaged.com.google.common.collect.ImmutableMap;
+
 public class TableDefine {
 
     private final Map<HugeKeys, String> columns;
     private final List<HugeKeys> keys;
+    private final Map<String, String> typesMapping;
 
     public TableDefine() {
         this.columns = InsertionOrderUtil.newMap();
         this.keys = InsertionOrderUtil.newList();
+        this.typesMapping = ImmutableMap.of();
+    }
+
+    public TableDefine(Map<String, String> typesMapping) {
+        this.columns = InsertionOrderUtil.newMap();
+        this.keys = InsertionOrderUtil.newList();
+        this.typesMapping = typesMapping;
     }
 
     public TableDefine column(HugeKeys key, String... desc) {
         StringBuilder sb = new StringBuilder();
         for (int i = 0; i < desc.length; i++) {
-            sb.append(desc[i]);
+            String type = desc[i];
+            // The first element of 'desc' is column data type, which may be
+            // mapped to actual data type supported by backend store
+            if (i == 0 && this.typesMapping.containsKey(type)) {
+                type = this.typesMapping.get(type);
+            }
+            assert type != null;
+            sb.append(type);
             if (i != desc.length - 1) {
                 sb.append(" ");
             }

hugegraph-dist/pom.xml

@@ -59,6 +59,11 @@
             <artifactId>hugegraph-hbase</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.baidu.hugegraph</groupId>
+            <artifactId>hugegraph-postgresql</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.airlift</groupId>
             <artifactId>airline</artifactId>

hugegraph-dist/src/assembly/static/conf/hugegraph.properties

@@ -33,12 +33,13 @@ cassandra.password=
 
 
 # mysql backend config
+#jdbc.driver=com.mysql.jdbc.Driver
 #jdbc.url=jdbc:mysql://127.0.0.1:3306
 #jdbc.username=root
 #jdbc.password=
 #jdbc.reconnect_max_times=3
 #jdbc.reconnect_interval=3
-
+#jdbc.sslmode=disable
 
 # palo backend config
 #palo.host=127.0.0.1

hugegraph-dist/src/assembly/travis/install-backend.sh

@@ -10,4 +10,6 @@ elif [ "$BACKEND" == "hbase" ]; then
     $TRAVIS_DIR/install-hbase.sh
 elif [ "$BACKEND" == "mysql" ]; then
     $TRAVIS_DIR/install-mysql.sh
+elif [ "$BACKEND" == "postgresql" ]; then
+    $TRAVIS_DIR/install-postgresql.sh
 fi

hugegraph-dist/src/assembly/travis/install-postgresql.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+
+set -ev
+
+TRAVIS_DIR=`dirname $0`
+CONF=hugegraph-test/src/main/resources/hugegraph.properties
+
+POSTGRESQL_DRIVER=org.postgresql.Driver
+POSTGRESQL_URL=jdbc:postgresql://localhost:5432/
+POSTGRESQL_USERNAME=postgres
+
+# Set PostgreSQL configurations
+sed -i "s/jdbc.driver=.*/jdbc.driver=$POSTGRESQL_DRIVER/" $CONF
+sed -i "s?jdbc.url=.*?jdbc.url=$POSTGRESQL_URL?" $CONF
+sed -i "s/jdbc.username=.*/jdbc.username=$POSTGRESQL_USERNAME/" $CONF
+
+sudo service postgresql stop 9.2
+sudo service postgresql start 9.5

hugegraph-dist/src/assembly/travis/start-server.sh

@@ -7,15 +7,29 @@ BASE_DIR=hugegraph-$VERSION
 BIN=$BASE_DIR/bin
 CONF=$BASE_DIR/conf/hugegraph.properties
 
+# PostgreSQL configurations
+POSTGRESQL_DRIVER=org.postgresql.Driver
+POSTGRESQL_URL=jdbc:postgresql://localhost:5432/
+POSTGRESQL_USERNAME=postgres
+
 declare -A backend_serializer_map=(["memory"]="text" ["cassandra"]="cassandra" \
                                    ["scylladb"]="scylladb" ["mysql"]="mysql" \
-                                   ["hbase"]="hbase" ["rocksdb"]="binary")
+                                   ["hbase"]="hbase" ["rocksdb"]="binary" \
+                                   ["postgresql"]="postgresql")
 
 SERIALIZER=${backend_serializer_map[$BACKEND]}
 
+# Set backend and serializer
 sed -i "s/backend=.*/backend=$BACKEND/" $CONF
 sed -i "s/serializer=.*/serializer=$SERIALIZER/" $CONF
 
+# Set PostgreSQL configurations if needed
+if [ "$BACKEND" == "postgresql" ]; then
+    sed -i "s/#jdbc.driver=.*/jdbc.driver=$POSTGRESQL_DRIVER/" $CONF
+    sed -i "s?#jdbc.url=.*?jdbc.url=$POSTGRESQL_URL?" $CONF
+    sed -i "s/#jdbc.username=.*/jdbc.username=$POSTGRESQL_USERNAME/" $CONF
+fi
+
 # Append schema.sync_deletion=true to config file
 echo "schema.sync_deletion=true" >> $CONF
 

hugegraph-dist/src/main/java/com/baidu/hugegraph/dist/RegisterUtil.java

@@ -52,7 +52,8 @@ public static void registerBackends() {
         String confFile = "/backend.properties";
         InputStream input = RegisterUtil.class.getClass()
                                         .getResourceAsStream(confFile);
-        E.checkState(input != null, "Can't read file '%s' as stream", confFile);
+        E.checkState(input != null,
+                     "Can't read file '%s' as stream", confFile);
 
         PropertiesConfiguration props = new PropertiesConfiguration();
         props.setDelimiterParsingDisabled(true);
@@ -89,8 +90,12 @@ private static void registerBackend(String backend) {
             case "palo":
                 registerPalo();
                 break;
+            case "postgresql":
+                registerPostgresql();
+                break;
             default:
-                throw new HugeException("Unsupported backend type '%s'", backend);
+                throw new HugeException("Unsupported backend type '%s'",
+                                        backend);
         }
     }
 
@@ -165,6 +170,18 @@ public static void registerPalo() {
                 "com.baidu.hugegraph.backend.store.palo.PaloStoreProvider");
     }
 
+    public static void registerPostgresql() {
+        // Register config
+        OptionSpace.register("postgresql",
+                "com.baidu.hugegraph.backend.store.postgresql.PostgresqlOptions");
+        // Register serializer
+        SerializerFactory.register("postgresql",
+                "com.baidu.hugegraph.backend.store.postgresql.PostgresqlSerializer");
+        // Register backend
+        BackendProviderFactory.register("postgresql",
+                "com.baidu.hugegraph.backend.store.postgresql.PostgresqlStoreProvider");
+    }
+
     public static void registerServer() {
         OptionSpace.register("server", "com.baidu.hugegraph.config.ServerOptions");
     }

hugegraph-dist/src/main/resources/backend.properties

@@ -1 +1 @@
-backends=[cassandra, scylladb, rocksdb, mysql, palo, hbase]
+backends=[cassandra, scylladb, rocksdb, mysql, palo, hbase, postgresql]

hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlOptions.java

@@ -27,7 +27,7 @@
 
 public class MysqlOptions extends OptionHolder {
 
-    private MysqlOptions() {
+    protected MysqlOptions() {
         super();
     }
 
@@ -89,4 +89,12 @@ public static synchronized MysqlOptions instance() {
                     rangeInt(1, 10),
                     3
             );
+
+    public static final ConfigOption<String> SSL_MODE =
+            new ConfigOption<>(
+                    "jdbc.ssl_mode",
+                    "The url of database in JDBC format.",
+                    disallowEmpty(),
+                    "disable"
+            );
 }

hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlSessions.java

@@ -147,8 +147,11 @@ public void createDatabase() {
         try (Connection conn = this.openWithoutDB(0)) {
             conn.createStatement().execute(sql);
         } catch (SQLException e) {
-            throw new BackendException("Failed to create database '%s'",
-                                       this.database);
+            if (!e.getMessage().endsWith("already exists")) {
+                throw new BackendException("Failed to create database '%s'", e,
+                                           this.database);
+            }
+            // Ignore exception if database already exists
         }
     }
 

hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlStore.java

@@ -57,7 +57,7 @@ public abstract class MysqlStore extends AbstractBackendStore<Session> {
 
     private final Map<HugeType, MysqlTable> tables;
 
-    private MysqlSessions sessions;
+    protected MysqlSessions sessions;
 
     public MysqlStore(final BackendStoreProvider provider,
                       final String database, final String store) {
@@ -114,7 +114,8 @@ public synchronized void open(HugeConfig config) {
         try {
             this.sessions.open(config);
         } catch (Exception e) {
-            if (!e.getMessage().startsWith("Unknown database")) {
+            if (!e.getMessage().startsWith("Unknown database") &&
+                !e.getMessage().endsWith("does not exist")) {
                 throw new BackendException("Failed connect with mysql, " +
                                            "please ensure it's ok", e);
             }

hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTable.java

@@ -28,6 +28,7 @@
 import java.util.Map;
 import java.util.Set;
 
+import org.apache.logging.log4j.util.Strings;
 import org.slf4j.Logger;
 
 import com.baidu.hugegraph.backend.BackendException;
@@ -50,6 +51,8 @@ public abstract class MysqlTable
 
     private static final Logger LOG = Log.logger(MysqlStore.class);
 
+    protected String dropTableTemplate = "DROP TABLE IF EXISTS %s;";
+    protected String truncateTableTemplate = "TRUNCATE TABLE %s;";
     // The template for insert and delete statements
     private String insertTemplate;
     private String deleteTemplate;
@@ -91,14 +94,16 @@ protected void createTable(Session session, TableDefine tableDefine) {
         // Specified primary keys
         sql.append(" PRIMARY KEY (");
         int i = 0;
+        int size = tableDefine.keys().size();
         for (HugeKeys key : tableDefine.keys()) {
-            sql.append(key);
-            if (++i != tableDefine.keys().size()) {
+            sql.append(formatKey(key));
+            if (++i != size) {
                 sql.append(", ");
             }
         }
-
-        sql.append(")) ENGINE=InnoDB;");
+        sql.append("))");
+        sql.append(this.engine());
+        sql.append(";");
 
         LOG.debug("Create table: {}", sql);
         try {
@@ -109,9 +114,13 @@ protected void createTable(Session session, TableDefine tableDefine) {
         }
     }
 
+    protected String engine() {
+        return " ENGINE=InnoDB";
+    }
+
     protected void dropTable(Session session) {
         LOG.debug("Drop table: {}", this.table());
-        String sql = String.format("DROP TABLE IF EXISTS %s;", this.table());
+        String sql = this.buildDropTemplate();
         try {
             session.execute(sql);
         } catch (SQLException e) {
@@ -122,7 +131,7 @@ protected void dropTable(Session session) {
 
     protected void truncateTable(Session session) {
         LOG.debug("Truncate table: {}", this.table());
-        String sql = String.format("TRUNCATE TABLE %s;", this.table());
+        String sql = this.buildTruncateTemplate();
         try {
             session.execute(sql);
         } catch (SQLException e) {
@@ -190,6 +199,14 @@ protected String buildDeleteTemplate(List<HugeKeys> idNames) {
         return this.deleteTemplate;
     }
 
+    protected String buildDropTemplate() {
+        return String.format(this.dropTableTemplate, this.table());
+    }
+
+    protected String buildTruncateTemplate() {
+        return String.format(this.truncateTableTemplate, this.table());
+    }
+
     /**
      * Insert an entire row
      */
@@ -202,7 +219,7 @@ public void insert(Session session, MysqlBackendEntry.Row entry) {
             // Create or get insert prepare statement
             insertStmt = session.prepareStatement(template);
             int i = 1;
-            for (Object object : entry.columns().values()) {
+            for (Object object : this.buildInsertObjects(entry)) {
                 insertStmt.setObject(i++, object);
             }
         } catch (SQLException e) {
@@ -413,7 +430,7 @@ protected StringBuilder relation2Sql(Condition.Relation relation) {
         Object value = relation.serialValue();
 
         // Serialize value (TODO: should move to Serializer)
-        value = serializeValue(value);
+        value = this.serializeValue(value);
 
         StringBuilder sql = new StringBuilder(32);
         sql.append(key);
@@ -503,13 +520,19 @@ protected void wrapPage(StringBuilder select, Query query) {
             select.append(where.build());
         }
 
+        select.append(this.orderByKeys());
+
         assert query.limit() != Query.NO_LIMIT;
         // Fetch `limit + 1` records for judging whether reached the last page
         select.append(" limit ");
         select.append(query.limit() + 1);
         select.append(";");
     }
 
+    protected String orderByKeys() {
+        return Strings.EMPTY;
+    }
+
     protected void wrapOffset(StringBuilder select, Query query) {
         assert query.limit() >= 0;
         assert query.offset() >= 0;
@@ -521,7 +544,7 @@ protected void wrapOffset(StringBuilder select, Query query) {
         select.append(";");
     }
 
-    private static Object serializeValue(Object value) {
+    protected Object serializeValue(Object value) {
         if (value instanceof Id) {
             value = ((Id) value).asObject();
         }
@@ -545,6 +568,14 @@ protected void appendPartition(StringBuilder delete) {
         // pass
     }
 
+    protected List<Object> buildInsertObjects(MysqlBackendEntry.Row entry) {
+        List<Object> objects = new ArrayList<>();
+        for (Object key : entry.columns().keySet()) {
+            objects.add(entry.columns().get(key));
+        }
+        return objects;
+    }
+
     public static String formatKey(HugeKeys key) {
         return key.name();
     }