Add dedup support to python bindings.

Author: Ori Bernstein

Makefile

@@ -46,7 +46,7 @@ grpc:
 	python -m grpc_tools.protoc -Ibtrdb/grpcinterface --python_out=btrdb/grpcinterface --grpc_python_out=btrdb/grpcinterface btrdb/grpcinterface/btrdb.proto
 	@echo
 	@echo Fixing import statements:
-	sed -i '' 's/btrdb_pb2 as btrdb__pb2/btrdb.grpcinterface.btrdb_pb2 as btrdb__pb2/' btrdb/grpcinterface/btrdb_pb2_grpc.py
+	sed -i'.bak' 's/btrdb_pb2 as btrdb__pb2/btrdb.grpcinterface.btrdb_pb2 as btrdb__pb2/' btrdb/grpcinterface/btrdb_pb2_grpc.py
 
 
 # Build the universal wheel and source distribution

btrdb/endpoint.py

@@ -160,9 +160,15 @@ def changes(self, uu, fromVersion, toVersion, resolution):
             BTrDBError.checkProtoStat(result.stat)
             yield result.ranges, result.versionMajor
 
-    def insert(self, uu, values):
+    def insert(self, uu, values, policy):
+        policy_map = {
+            'never': btrdb_pb2.MergePolicy.NEVER,
+            'equal': btrdb_pb2.MergePolicy.EQUAL,
+            'retain': btrdb_pb2.MergePolicy.RETAIN,
+            'replace': btrdb_pb2.MergePolicy.REPLACE,
+        }
         protoValues = RawPoint.to_proto_list(values)
-        params = btrdb_pb2.InsertParams(uuid = uu.bytes, sync = False, values = protoValues)
+        params = btrdb_pb2.InsertParams(uuid = uu.bytes, sync = False, values = protoValues, merge_policy = policy_map[policy])
         result = self.stub.Insert(params)
         BTrDBError.checkProtoStat(result.stat)
         return result.versionMajor

btrdb/grpcinterface/btrdb.proto

@@ -23,6 +23,8 @@ service BTrDB {
   rpc GetMetadataUsage(MetadataUsageParams) returns (MetadataUsageResponse);
   rpc GenerateCSV(GenerateCSVParams) returns (stream GenerateCSVResponse);
   rpc SQLQuery(SQLQueryParams) returns (stream SQLQueryResponse);
+  //rpc SetCompactionConfig(SetCompactionConfigParams) returns (SetCompactionConfigResponse);
+  //rpc GetCompactionConfig(GetCompactionConfigParams) returns (GetCompactionConfigResponse);
 }
 
 message RawValuesParams {
@@ -167,9 +169,16 @@ message ChangesResponse {
   uint64 versionMinor = 3;
   repeated ChangedRange ranges = 4;
 }
+enum MergePolicy {
+  NEVER = 0;	// Never merge
+  EQUAL = 1;	// Merge identical (key, value) pairs
+  RETAIN = 2;	// When timestamps are equal, keep old value
+  REPLACE = 3;	// When timestamps are equal, keep new value
+}
 message InsertParams {
   bytes uuid = 1;
   bool sync = 2;
+  MergePolicy merge_policy = 4;
   repeated RawPoint values = 3;
 }
 message InsertResponse {
@@ -315,3 +324,54 @@ message SQLQueryResponse {
 message Role {
   string name = 1;
 }
+
+message SetCompactionConfigParams {
+  //Which stream to configure
+  bytes uuid = 1;
+
+  //Accessing versions LESS than this is not allowed
+  uint64 CompactedVersion = 2;
+
+  //For every timestamp >= Start and < End in this list,
+  //we cannot traverse the tree < Resolution.
+  // Ranges cannot overlap, and a range can never have its resolution PW increased
+  // (it is okay to decrease a range PW and it is okay to merge two adjacent ranges
+  // if the new range has <= resolution)
+  repeated ReducedResolutionRange reducedResolutionRanges = 3;
+
+  uint64 unused0 = 4;
+}
+
+message SetCompactionConfigResponse {
+  Status stat = 1;
+}
+
+message GetCompactionConfigParams {
+  //Which stream to query
+  bytes uuid = 1;
+}
+
+message GetCompactionConfigResponse {
+  Status stat = 1;
+
+  //The latest version of the stream, as returned by StreamInfo
+  uint64 LatestMajorVersion = 2;
+
+  //Accessing versions LESS than this is not allowed
+  uint64 CompactedVersion = 3;
+
+  //For every timestamp >= Start and < End in this list,
+  //we cannot traverse the tree < Resolution.
+  // Ranges cannot overlap, and a range can never have its resolution PW increased
+  // (it is okay to decrease a range PW and it is okay to merge two adjacent ranges
+  // if the new range has <= resolution)
+  repeated ReducedResolutionRange reducedResolutionRanges = 4;
+
+  uint64 unused0 = 5;
+}
+
+message ReducedResolutionRange {
+  int64 Start = 1;
+  int64 End = 2;
+  uint32 Resolution = 3;
+}

btrdb/grpcinterface/btrdb_pb2.py

@@ -254,8 +254,9 @@ def GenerateCSV(self, request, context):
     raise NotImplementedError('Method not implemented!')
 
   def SQLQuery(self, request, context):
-    # missing associated documentation comment in .proto file
-    pass
+    """rpc SetCompactionConfig(SetCompactionConfigParams) returns (SetCompactionConfigResponse);
+    rpc GetCompactionConfig(GetCompactionConfigParams) returns (GetCompactionConfigResponse);
+    """
     context.set_code(grpc.StatusCode.UNIMPLEMENTED)
     context.set_details('Method not implemented!')
     raise NotImplementedError('Method not implemented!')

btrdb/grpcinterface/btrdb_pb2_grpc.py

@@ -402,7 +402,7 @@ def version(self):
         """
         return self._btrdb.ep.streamInfo(self._uuid, True, False)[4]
 
-    def insert(self, data):
+    def insert(self, data, merge='never'):
         """
         Insert new data in the form (time, value) into the series.
 
@@ -417,6 +417,12 @@ def insert(self, data):
         data: list[tuple[int, float]]
             A list of tuples in which each tuple contains a time (int) and
             value (float) for insertion to the database
+        merge: str
+            A string describing the merge policy. Valid policies are:
+              - 'never': the default, no points are merged
+              - 'equal': points are deduplicated if the time and value are equal
+              - 'retain': if two points have the same timestamp, the old one is kept
+              - 'replace': if two points have the same timestamp, the new one is kept
 
         Returns
         -------
@@ -428,7 +434,7 @@ def insert(self, data):
         version = 0
         while i < len(data):
             thisBatch = data[i:i + INSERT_BATCH_SIZE]
-            version = self._btrdb.ep.insert(self._uuid, thisBatch)
+            version = self._btrdb.ep.insert(self._uuid, thisBatch, merge)
             i += INSERT_BATCH_SIZE
         return version