support create olap property key and dynamic create olap table for ca…

…ssandra

Change-Id: I947c3a96eb7aa67ee9eff5f7fb248e9f4539e91b

hugegraph-api/src/main/java/com/baidu/hugegraph/api/API.java

@@ -57,6 +57,7 @@ public class API {
 
     public static final String ACTION_APPEND = "append";
     public static final String ACTION_ELIMINATE = "eliminate";
+    public static final String ACTION_CLEAR = "clear";
 
     private static final Meter succeedMeter =
                          MetricsUtil.registerMeter(API.class, "commit-succeed");

hugegraph-api/src/main/java/com/baidu/hugegraph/api/graph/VertexAPI.java

@@ -423,8 +423,6 @@ private static class JsonVertex extends JsonElement {
 
         @Override
         public void checkCreate(boolean isBatch) {
-            E.checkArgumentNotNull(this.label,
-                                   "The label of vertex can't be null");
             this.checkUpdate();
         }
 
@@ -446,8 +444,10 @@ public void checkUpdate() {
         public Object[] properties() {
             Object[] props = API.properties(this.properties);
             List<Object> list = new ArrayList<>(Arrays.asList(props));
-            list.add(T.label);
-            list.add(this.label);
+            if (this.label != null) {
+                list.add(T.label);
+                list.add(this.label);
+            }
             if (this.id != null) {
                 list.add(T.id);
                 list.add(this.id);

hugegraph-api/src/main/java/com/baidu/hugegraph/api/schema/PropertyKeyAPI.java

@@ -98,10 +98,20 @@ public String update(@Context GraphManager manager,
         E.checkArgument(name.equals(jsonPropertyKey.name),
                         "The name in url(%s) and body(%s) are different",
                         name, jsonPropertyKey.name);
+
+        HugeGraph g = graph(manager, graph);
+        if (ACTION_CLEAR.equals(action)) {
+            PropertyKey propertyKey = g.propertyKey(name);
+            E.checkArgument(propertyKey.readFrequency().olap(),
+                            "Only olap property key can do action clear, " +
+                            "but got '%s'", propertyKey);
+            g.clearPropertyKey(propertyKey);
+            return "done";
+        }
+
         // Parse action parameter
         boolean append = checkAndParseAction(action);
 
-        HugeGraph g = graph(manager, graph);
         PropertyKey.Builder builder = jsonPropertyKey.convert2Builder(g);
         PropertyKey propertyKey = append ?
                                   builder.append() :

hugegraph-api/src/main/java/com/baidu/hugegraph/auth/HugeGraphAuthProxy.java

@@ -176,6 +176,12 @@ public void removePropertyKey(Id key) {
         this.hugegraph.removePropertyKey(key);
     }
 
+    @Override
+    public void clearPropertyKey(PropertyKey propertyKey) {
+        verifySchemaPermission(HugePermission.DELETE, propertyKey);
+        this.hugegraph.clearPropertyKey(propertyKey);
+    }
+
     @Override
     public Collection<PropertyKey> propertyKeys() {
         Collection<PropertyKey> pkeys = this.hugegraph.propertyKeys();

hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraFeatures.java

@@ -124,6 +124,6 @@ public boolean supportsTtl() {
 
     @Override
     public boolean supportsOlapProperties() {
-        return false;
+        return true;
     }
 }

hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraSerializer.java

@@ -37,7 +37,9 @@
 import com.baidu.hugegraph.schema.PropertyKey;
 import com.baidu.hugegraph.schema.SchemaElement;
 import com.baidu.hugegraph.structure.HugeElement;
+import com.baidu.hugegraph.structure.HugeIndex;
 import com.baidu.hugegraph.structure.HugeProperty;
+import com.baidu.hugegraph.structure.HugeVertex;
 import com.baidu.hugegraph.type.HugeType;
 import com.baidu.hugegraph.type.define.DataType;
 import com.baidu.hugegraph.type.define.HugeKeys;
@@ -58,6 +60,16 @@ protected TableBackendEntry newBackendEntry(TableBackendEntry.Row row) {
         return new CassandraBackendEntry(row);
     }
 
+    @Override
+    protected TableBackendEntry newBackendEntry(HugeIndex index) {
+        TableBackendEntry backendEntry = newBackendEntry(index.type(),
+                                                         index.id());
+        if (index.indexLabel().olap()) {
+            backendEntry.olap(true);
+        }
+        return backendEntry;
+    }
+
     @Override
     protected CassandraBackendEntry convertEntry(BackendEntry backendEntry) {
         if (!(backendEntry instanceof CassandraBackendEntry)) {
@@ -130,6 +142,20 @@ protected void parseProperties(HugeElement element,
         }
     }
 
+    @Override
+    public BackendEntry writeOlapVertex(HugeVertex vertex) {
+        CassandraBackendEntry entry = newBackendEntry(HugeType.OLAP,
+                                                      vertex.id());
+        entry.column(HugeKeys.ID, this.writeId(vertex.id()));
+        HugeProperty<?> prop = vertex.getProperties().values()
+                                     .iterator().next();
+        PropertyKey pk = prop.propertyKey();
+        entry.subId(pk.id());
+        entry.column(HugeKeys.PROPERTY_VALUE,
+                     this.writeProperty(pk, prop.value()));
+        return entry;
+    }
+
     @Override
     protected Object writeProperty(PropertyKey propertyKey, Object value) {
         BytesBuffer buffer = BytesBuffer.allocate(BytesBuffer.BUF_PROPERTY);

hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraStore.java

@@ -25,13 +25,15 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 import org.slf4j.Logger;
 
 import com.baidu.hugegraph.backend.BackendException;
 import com.baidu.hugegraph.backend.id.Id;
 import com.baidu.hugegraph.backend.query.Query;
+import com.baidu.hugegraph.backend.serializer.MergeIterator;
 import com.baidu.hugegraph.backend.store.AbstractBackendStore;
 import com.baidu.hugegraph.backend.store.BackendAction;
 import com.baidu.hugegraph.backend.store.BackendEntry;
@@ -63,9 +65,9 @@ public abstract class CassandraStore
 
     private final BackendStoreProvider provider;
     // TODO: move to parent class
-    private final Map<HugeType, CassandraTable> tables;
+    private final Map<String, CassandraTable> tables;
 
-    private CassandraSessionPool sessions;
+    protected CassandraSessionPool sessions;
     private HugeConfig conf;
 
     public CassandraStore(final BackendStoreProvider provider,
@@ -95,7 +97,11 @@ private void registerMetaHandlers() {
     }
 
     protected void registerTableManager(HugeType type, CassandraTable table) {
-        this.tables.put(type, table);
+        this.registerTableManager(type.string(), table);
+    }
+
+    protected void registerTableManager(String name, CassandraTable table) {
+        this.tables.put(name, table);
     }
 
     @Override
@@ -200,6 +206,12 @@ private void mutate(CassandraSessionPool.Session session,
 
         switch (item.action()) {
             case INSERT:
+                // Insert olap entry
+                if (entry.type() == HugeType.OLAP) {
+                    this.table(this.store + "_" + HugeType.OLAP.string() + "_" + entry.subId().asLong())
+                        .insert(session, entry.row());
+                    break;
+                }
                 // Insert entry
                 if (entry.selfChanged()) {
                     this.table(entry.type()).insert(session, entry.row());
@@ -220,6 +232,10 @@ private void mutate(CassandraSessionPool.Session session,
                 }
                 break;
             case APPEND:
+                if (entry.olap()) {
+                    this.table(this.store + "_" + HugeType.OLAP.string() + "_" + entry.type().string())
+                        .append(session, entry.row());
+                }
                 // Append entry
                 if (entry.selfChanged()) {
                     this.table(entry.type()).append(session, entry.row());
@@ -248,9 +264,27 @@ private void mutate(CassandraSessionPool.Session session,
     @Override
     public Iterator<BackendEntry> query(Query query) {
         this.checkOpened();
-
-        CassandraTable table = this.table(CassandraTable.tableType(query));
-        return table.query(this.sessions.session(), query);
+        HugeType type = CassandraTable.tableType(query);
+
+        // TODO: move to MergeStore
+        CassandraTable table = query.olap() ?
+                this.table(this.store + "_" + HugeType.OLAP.string() + "_" + type.string()) :
+                this.table(type);
+        Iterator<BackendEntry> entrys = table.query(this.sessions.session(),
+                                                    query);
+        Set<Id> olapPks = query.olapPks();
+        if (!olapPks.isEmpty()) {
+            List<Iterator<BackendEntry>> iterators =
+                                         new ArrayList<>(olapPks.size());
+            for (Id pk : olapPks) {
+                Query q = query.copy();
+                table = this.table(this.store + "_" + HugeType.OLAP.string() + "_" + pk.asLong());
+                iterators.add(table.query(this.sessions.session(), q));
+            }
+            entrys = new MergeIterator<>(entrys, iterators,
+                                         BackendEntry::mergable);
+        }
+        return entrys;
     }
 
     @Override
@@ -514,7 +548,12 @@ protected void clearTables() {
     protected void truncateTables() {
         CassandraSessionPool.Session session = this.sessions.session();
         for (CassandraTable table : this.tables()) {
-            table.truncate(session);
+            System.out.println(table);
+            if (table.isOlap()) {
+                table.dropTable(session);
+            } else {
+                table.truncate(session);
+            }
         }
     }
 
@@ -524,10 +563,14 @@ protected Collection<CassandraTable> tables() {
 
     @Override
     protected final CassandraTable table(HugeType type) {
-        assert type != null;
-        CassandraTable table = this.tables.get(type);
+        return this.table(type.string());
+    }
+
+    protected final CassandraTable table(String name) {
+        assert name != null;
+        CassandraTable table = this.tables.get(name);
         if (table == null) {
-            throw new BackendException("Unsupported table type: %s", type);
+            throw new BackendException("Unsupported table: %s", name);
         }
         return table;
     }
@@ -600,6 +643,12 @@ public long getCounter(HugeType type) {
         public boolean isSchemaStore() {
             return true;
         }
+
+        @Override
+        public void createOlapTable(Id pkId) {
+            throw new UnsupportedOperationException(
+                      "CassandraSchemaStore.createOlapTable()");
+        }
     }
 
     public static class CassandraGraphStore extends CassandraStore {
@@ -632,6 +681,17 @@ public CassandraGraphStore(BackendStoreProvider provider,
                                  new CassandraTables.ShardIndex(store));
             registerTableManager(HugeType.UNIQUE_INDEX,
                                  new CassandraTables.UniqueIndex(store));
+
+            registerTableManager(this.store() + "_" + HugeType.OLAP.string() + "_" + HugeType.SECONDARY_INDEX.string(),
+                                 new CassandraTables.OlapSecondaryIndex(store));
+            registerTableManager(this.store() + "_" + HugeType.OLAP.string() + "_" + HugeType.RANGE_INT_INDEX.string(),
+                                 new CassandraTables.OlapRangeIntIndex(store));
+            registerTableManager(this.store() + "_" + HugeType.OLAP.string() + "_" + HugeType.RANGE_LONG_INDEX.string(),
+                                 new CassandraTables.OlapRangeLongIndex(store));
+            registerTableManager(this.store() + "_" + HugeType.OLAP.string() + "_" + HugeType.RANGE_FLOAT_INDEX.string(),
+                                 new CassandraTables.OlapRangeFloatIndex(store));
+            registerTableManager(this.store() + "_" + HugeType.OLAP.string() + "_" + HugeType.RANGE_DOUBLE_INDEX.string(),
+                                 new CassandraTables.OlapRangeDoubleIndex(store));
         }
 
         @Override
@@ -656,5 +716,39 @@ public long getCounter(HugeType type) {
         public boolean isSchemaStore() {
             return false;
         }
+
+        @Override
+        public void createOlapTable(Id pkId) {
+            this.initAndRegisterOlapTable(pkId);
+        }
+
+        @Override
+        public void initAndRegisterOlapTable(Id pkId) {
+            CassandraTable table = new CassandraTables.Olap(this.store(),
+                                                            pkId.asLong());
+            registerTableManager(this.store() + "_" + HugeType.OLAP.string() + "_" + pkId,
+                                 table);
+            this.checkOpened();
+            CassandraSessionPool.Session session = this.sessions.session();
+            table.init(session);
+        }
+
+        @Override
+        public void clearOlapTable(Id pkId) {
+            CassandraTable table = new CassandraTables.Olap(this.store(),
+                                                            pkId.asLong());
+            this.checkOpened();
+            CassandraSessionPool.Session session = this.sessions.session();
+            table.truncate(session);
+        }
+
+        @Override
+        public void removeOlapTable(Id pkId) {
+            CassandraTable table = new CassandraTables.Olap(this.store(),
+                                                            pkId.asLong());
+            this.checkOpened();
+            CassandraSessionPool.Session session = this.sessions.session();
+            table.dropTable(session);
+        }
     }
 }

hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTable.java

@@ -401,7 +401,7 @@ protected Iterator<BackendEntry> results2Entries(Query q, ResultSet r) {
         });
     }
 
-    private static CassandraBackendEntry row2Entry(HugeType type, Row row) {
+    protected static CassandraBackendEntry row2Entry(HugeType type, Row row) {
         CassandraBackendEntry entry = new CassandraBackendEntry(type);
 
         List<Definition> cols = row.getColumnDefinitions().asList();
@@ -656,4 +656,8 @@ public void clear(CassandraSessionPool.Session session) {
     public void truncate(CassandraSessionPool.Session session) {
         this.truncateTable(session);
     }
+
+    public boolean isOlap() {
+        return false;
+    }
 }