support basic json filtering and selection

go.mod

@@ -87,6 +87,7 @@ require (
 	github.com/stretchr/testify v1.4.0 // indirect
 	github.com/syndtr/goleveldb v1.0.0
 	github.com/tidwall/gjson v1.3.2
+	github.com/tidwall/match v1.0.1
 	github.com/uber-go/atomic v1.4.0 // indirect
 	github.com/uber/jaeger-client-go v2.17.0+incompatible // indirect
 	github.com/uber/jaeger-lib v2.0.0+incompatible // indirect

weed/pb/volume_server.proto

@@ -324,9 +324,14 @@ message MemStatus {
 }
 
 message QueryRequest {
-    string select_expression = 1;
+    repeated string selections = 1;
     repeated string from_file_ids = 2;
-    string where_expression = 3;
+    message Filter {
+        string field = 1;
+        string operand = 2;
+        string value = 3;
+    }
+    Filter filter = 3;
 
     message InputSerialization {
         // NONE | GZIP | BZIP2
@@ -372,5 +377,5 @@ message QueryRequest {
     OutputSerialization output_serialization = 5;
 }
 message QueriedStripe {
-    repeated bytes records = 1;
+    bytes records = 1;
 }

weed/query/json/query_json.go

@@ -1,5 +1,107 @@
 package json
 
-func QueryJson(jsonLine string, query string) (jsonOutput string) {
-	return jsonLine
+import (
+	"strconv"
+
+	"github.com/chrislusf/seaweedfs/weed/query/sqltypes"
+	"github.com/tidwall/gjson"
+	"github.com/tidwall/match"
+)
+
+type Query struct {
+	Field string
+	Op    string
+	Value string
+}
+
+func QueryJson(jsonLine string, projections []string, query Query) (passedFilter bool, values []sqltypes.Value) {
+	if filterJson(jsonLine, query) {
+		passedFilter = true
+		fields := gjson.GetMany(jsonLine, projections...)
+		for _, f := range fields {
+			values = append(values, sqltypes.MakeTrusted(sqltypes.Type(f.Type), sqltypes.StringToBytes(f.Raw)))
+		}
+		return
+	}
+	return false, nil
+}
+
+func filterJson(jsonLine string, query Query) bool{
+
+	value := gjson.Get(jsonLine, query.Field)
+
+	// copied from gjson.go queryMatches() function
+	rpv := query.Value
+
+	if !value.Exists() {
+		return false
+	}
+	if query.Op == "" {
+		// the query is only looking for existence, such as:
+		//   friends.#(name)
+		// which makes sure that the array "friends" has an element of
+		// "name" that exists
+		return true
+	}
+	switch value.Type {
+	case gjson.String:
+		switch query.Op {
+		case "=":
+			return value.Str == rpv
+		case "!=":
+			return value.Str != rpv
+		case "<":
+			return value.Str < rpv
+		case "<=":
+			return value.Str <= rpv
+		case ">":
+			return value.Str > rpv
+		case ">=":
+			return value.Str >= rpv
+		case "%":
+			return match.Match(value.Str, rpv)
+		case "!%":
+			return !match.Match(value.Str, rpv)
+		}
+	case gjson.Number:
+		rpvn, _ := strconv.ParseFloat(rpv, 64)
+		switch query.Op {
+		case "=":
+			return value.Num == rpvn
+		case "!=":
+			return value.Num != rpvn
+		case "<":
+			return value.Num < rpvn
+		case "<=":
+			return value.Num <= rpvn
+		case ">":
+			return value.Num > rpvn
+		case ">=":
+			return value.Num >= rpvn
+		}
+	case gjson.True:
+		switch query.Op {
+		case "=":
+			return rpv == "true"
+		case "!=":
+			return rpv != "true"
+		case ">":
+			return rpv == "false"
+		case ">=":
+			return true
+		}
+	case gjson.False:
+		switch query.Op {
+		case "=":
+			return rpv == "false"
+		case "!=":
+			return rpv != "false"
+		case "<":
+			return rpv == "true"
+		case "<=":
+			return true
+		}
+	}
+	return false
+
 }

weed/query/json/query_json_test.go

@@ -58,7 +58,7 @@ func TestGjson(t *testing.T) {
 	projections := []string{"quiz","fruit"}
 
 	gjson.ForEachLine(data, func(line gjson.Result) bool{
-		println(line.String())
+		println(line.Raw)
 		println("+++++++++++")
 		results := gjson.GetMany(line.Raw, projections...)
 		for _, result := range results {
@@ -71,3 +71,66 @@ func TestGjson(t *testing.T) {
 
 
 }
+
+func TestJsonQueryRow(t *testing.T) {
+
+	data := `
+	{
+		"fruit": "Bl\"ue",
+		"size": 6,
+		"quiz": "green"
+	}
+
+`
+	selections := []string{"fruit", "size"}
+
+	isFiltered, values := QueryJson(data, selections, Query{
+		Field: "quiz",
+		Op:    "=",
+		Value: "green",
+	})
+
+	if !isFiltered {
+		t.Errorf("should have been filtered")
+	}
+
+	if values == nil {
+		t.Errorf("values should have been returned")
+	}
+
+	buf := ToJson(nil, selections, values)
+	println(string(buf))
+
+}
+
+func TestJsonQueryNumber(t *testing.T) {
+
+	data := `
+	{
+		"fruit": "Bl\"ue",
+		"size": 6,
+		"quiz": "green"
+	}
+
+`
+	selections := []string{"fruit", "quiz"}
+
+	isFiltered, values := QueryJson(data, selections, Query{
+		Field: "size",
+		Op:    ">=",
+		Value: "6",
+	})
+
+	if !isFiltered {
+		t.Errorf("should have been filtered")
+	}
+
+	if values == nil {
+		t.Errorf("values should have been returned")
+	}
+
+	buf := ToJson(nil, selections, values)
+	println(string(buf))
+
+}
+

weed/query/json/seralize.go

@@ -0,0 +1,17 @@
+package json
+
+import "github.com/chrislusf/seaweedfs/weed/query/sqltypes"
+
+func ToJson(buf []byte, selections []string, values []sqltypes.Value) []byte {
+	buf = append(buf, '{')
+	for i, value := range values {
+		if i > 0 {
+			buf = append(buf, ',')
+		}
+		buf = append(buf, selections[i]...)
+		buf = append(buf, ':')
+		buf = append(buf, value.Raw()...)
+	}
+	buf = append(buf, '}')
+	return buf
+}

weed/server/volume_grpc_query.go

@@ -4,6 +4,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/chrislusf/seaweedfs/weed/query/json"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/tidwall/gjson"
 )
@@ -41,22 +42,30 @@ func (vs *VolumeServer) Query(req *volume_server_pb.QueryRequest, stream volume_
 
 		if req.InputSerialization.JsonInput!=nil{
 
-			err = stream.Send(&volume_server_pb.QueriedStripe{
+			stripe := &volume_server_pb.QueriedStripe{
 				Records:nil,
-			})
-			if err != nil {
-				// println("sending", bytesread, "bytes err", err.Error())
-				return err
+			}
+
+			filter := json.Query{
+				Field: req.Filter.Field,
+				Op: req.Filter.Operand,
+				Value:req.Filter.Value,
 			}
 			gjson.ForEachLine(string(n.Data), func(line gjson.Result) bool{
-				println(line.String())
+				passedFilter, values := json.QueryJson(line.Raw, req.Selections, filter)
+				if !passedFilter {
+					return true
+				}
+				stripe.Records = json.ToJson(stripe.Records, req.Selections, values)
 				return true
 			})
+			err = stream.Send(stripe)
+			if err != nil {
+				return err
+			}
 		}
 
-
-
 	}
 
 	return nil
-}
+}