TSDB delete by range (#377)

* enhance TSDB delete

* update tsdb to point to talIguaz - temp

* update v3io-go + v3io-tsdb

* python linter

* update TSDB version

backends/kv/backend.go

@@ -104,7 +104,8 @@ func (b *Backend) updateItem(request *frames.ExecRequest) error {
 	}
 
 	b.logger.DebugWith("update item", "path", path, "key", key, "expr", expr, "condition", condition)
-	return container.UpdateItemSync(&v3io.UpdateItemInput{Path: path + key, Expression: &expr, Condition: condition})
+	_, err = container.UpdateItemSync(&v3io.UpdateItemInput{Path: path + key, Expression: &expr, Condition: condition})
+	return err
 }
 
 /*func (b *Backend) newContainer(session *frames.Session) (v3io.Container, error) {

backends/tsdb/backend.go

@@ -29,12 +29,11 @@ import (
 	"time"
 	"unsafe"
 
-	"github.com/v3io/frames"
-	"github.com/v3io/frames/backends"
-
 	"github.com/golang/groupcache/lru"
 	"github.com/nuclio/logger"
 	"github.com/pkg/errors"
+	"github.com/v3io/frames"
+	"github.com/v3io/frames/backends"
 	"github.com/v3io/frames/v3ioutils"
 	v3io "github.com/v3io/v3io-go/pkg/dataplane"
 	"github.com/v3io/v3io-tsdb/pkg/config"
@@ -264,11 +263,12 @@ func (b *Backend) Delete(request *frames.DeleteRequest) error {
 		}
 	}
 
-	err = adapter.DeleteDB(delAll, false, start, end)
-	if err == nil {
-		return err
-	}
-
+	params := tsdb.DeleteParams{DeleteAll: delAll,
+		From:    start,
+		To:      end,
+		Filter:  request.Proto.Filter,
+		Metrics: request.Proto.Metrics}
+	err = adapter.DeleteDB(params)
 	return err
 
 }

clients/py/v3io_frames/client.py

@@ -12,9 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import pandas as pd
 from os import environ
 
+import pandas as pd
+
 from . import frames_pb2 as fpb
 from .errors import (CreateError, DeleteError, ExecuteError, ReadError,
                      WriteError)
@@ -177,7 +178,7 @@ def create(self, backend, table, attrs=None, schema=None, if_exists=FAIL):
         return self._create(backend, table, attrs, schema, if_exists)
 
     def delete(self, backend, table, filter='', start='', end='',
-               if_missing=FAIL):
+               if_missing=FAIL, metrics=[]):
         """Deletes a table or stream or specific table items
 
         Parameters
@@ -202,14 +203,17 @@ def delete(self, backend, table, filter='', start='', end='',
              days), or 0 for the earliest time; the default is "now"
         if_missing : int
             One of IGNORE or FAIL
+        metrics : string
+             (`tsdb` backend only) List of specific metric names to delete.
 
         Raises
         ------
         DeleteError
             On request error or backend error
         """
         self._validate_request(backend, table, DeleteError)
-        return self._delete(backend, table, filter, start, end, if_missing)
+        return self._delete(backend, table, filter, start, end,
+                            if_missing, metrics)
 
     def execute(self, backend, table, command='', args=None, expression=''):
         """Executes a backend-specific command on a table or stream

clients/py/v3io_frames/grpc.py

@@ -150,7 +150,7 @@ def _create(self, backend, table, attrs, schema, if_exists):
         stub.Create(request)
 
     @grpc_raise(DeleteError)
-    def _delete(self, backend, table, filter, start, end, if_missing):
+    def _delete(self, backend, table, filter, start, end, if_missing, metrics):
         start, end = time2str(start), time2str(end)
         stub = fgrpc.FramesStub(self._channel)
         request = fpb.DeleteRequest(
@@ -161,6 +161,7 @@ def _delete(self, backend, table, filter, start, end, if_missing):
             start=start,
             end=end,
             if_missing=if_missing,
+            metrics=metrics,
         )
         stub.Delete(request)
 

clients/py/v3io_frames/http.py

@@ -13,12 +13,13 @@
 # limitations under the License.
 
 import json
-import requests
 import struct
 from base64 import b64decode
 from datetime import datetime
 from functools import partial, wraps
 from itertools import chain
+
+import requests
 from requests.exceptions import RequestException
 from urllib3.exceptions import HTTPError
 
@@ -146,7 +147,7 @@ def _create(self, backend, table, attrs, schema, if_exists):
             raise CreateError(resp.text)
 
     @connection_error(DeleteError)
-    def _delete(self, backend, table, filter, start, end, if_missing):
+    def _delete(self, backend, table, filter, start, end, if_missing, metrics):
         request = {
             'session': pb2py(self.session),
             'backend': backend,
@@ -155,6 +156,7 @@ def _delete(self, backend, table, filter, start, end, if_missing):
             'start': start,
             'end': end,
             'if_missing': if_missing,
+            'metrics': metrics,
         }
 
         convert_go_times(request, ('start', 'end'))

frames.proto

@@ -194,6 +194,7 @@ message DeleteRequest {
     // TSDB and Stream specific fields
     string start = 6;
     string end = 7;
+    repeated string metrics = 8;
 }
 
 message DeleteResponse {}

go.mod

@@ -7,10 +7,11 @@ require (
 	github.com/nuclio/errors v0.0.1
 	github.com/nuclio/logger v0.0.1
 	github.com/nuclio/zap v0.0.2
+	github.com/philhofer/fwd v1.0.0 // indirect
 	github.com/pkg/errors v0.8.1
 	github.com/stretchr/testify v1.4.0
-	github.com/v3io/v3io-go v0.0.5-0.20191205125653-9003ae83f0b6
-	github.com/v3io/v3io-tsdb v0.9.12-4
+	github.com/v3io/v3io-go v0.0.7-0.20200216132233-3b52a325296d
+	github.com/v3io/v3io-tsdb v0.9.12-5
 	github.com/valyala/fasthttp v1.2.0
 	github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2
 	golang.org/x/net v0.0.0-20181114220301-adae6a3d119a