diff --git a/datastore/datastore_test.go b/datastore/datastore_test.go
index ae42e1c09764..5fde4e718861 100644
--- a/datastore/datastore_test.go
+++ b/datastore/datastore_test.go
@@ -46,10 +46,10 @@ func TestQueryConstruction(t *testing.T) {
 			q: NewQuery("Foo").Filter("foo >", 7),
 			exp: &Query{
 				kind: "Foo",
-				filter: []filter{
-					{
+				filter: []EntityFilter{
+					PropertyFilter{
 						FieldName: "foo",
-						Op:        greaterThan,
+						Operator:  string(greaterThan),
 						Value:     7,
 					},
 				},
@@ -61,10 +61,10 @@ func TestQueryConstruction(t *testing.T) {
 			q: NewQuery("Foo").Filter("foo=", 6),
 			exp: &Query{
 				kind: "Foo",
-				filter: []filter{
-					{
+				filter: []EntityFilter{
+					PropertyFilter{
 						FieldName: "foo",
-						Op:        equal,
+						Operator:  string(equal),
 						Value:     6,
 					},
 				},
@@ -76,10 +76,10 @@ func TestQueryConstruction(t *testing.T) {
 			q: NewQuery("Foo").Filter(" foo< ", 8),
 			exp: &Query{
 				kind: "Foo",
-				filter: []filter{
-					{
+				filter: []EntityFilter{
+					PropertyFilter{
 						FieldName: "foo",
-						Op:        lessThan,
+						Operator:  string(lessThan),
 						Value:     8,
 					},
 				},
@@ -91,10 +91,10 @@ func TestQueryConstruction(t *testing.T) {
 			q: NewQuery("Foo").Filter("foo >=", 9),
 			exp: &Query{
 				kind: "Foo",
-				filter: []filter{
-					{
+				filter: []EntityFilter{
+					PropertyFilter{
 						FieldName: "foo",
-						Op:        greaterEq,
+						Operator:  string(greaterEq),
 						Value:     9,
 					},
 				},
diff --git a/datastore/integration_test.go b/datastore/integration_test.go
index 1f9d9612f600..5af0a780cf8d 100644
--- a/datastore/integration_test.go
+++ b/datastore/integration_test.go
@@ -485,6 +485,70 @@ func testSmallQueries(ctx context.Context, t *testing.T, client *Client, parent
 	}
 }
 
+func TestIntegration_FilterEntity(t *testing.T) {
+	ctx := context.Background()
+	client := newTestClient(ctx, t)
+	defer client.Close()
+
+	parent := NameKey("SQParent", "TestIntegration_Filters"+suffix, nil)
+	now := timeNow.Truncate(time.Millisecond).Unix()
+	tomorrow := timeNow.Truncate(time.Millisecond).AddDate(0, 0, 1).Unix()
+	children := []*SQChild{
+		{I: 0, J: 99, T: tomorrow, U: now},
+		{I: 1, J: 98, T: tomorrow, U: now},
+		{I: 2, J: 97, T: tomorrow, U: now},
+		{I: 3, J: 96, T: now, U: now},
+		{I: 4, J: 95, T: now, U: now},
+		{I: 5, J: 94, T: now, U: now},
+		{I: 6, J: 93, T: now, U: now},
+		{I: 7, J: 92, T: now, U: now},
+	}
+	baseQuery := NewQuery("SQChild").Ancestor(parent)
+	testSmallQueries(ctx, t, client, parent, children, []SQTestCase{
+		{
+			desc: "I>1",
+			q: baseQuery.Filter("T=", now).FilterEntity(
+				PropertyFilter{FieldName: "I", Operator: ">", Value: 1}),
+			wantCount: 5,
+			wantSum:   3 + 4 + 5 + 6 + 7 + 96 + 95 + 94 + 93 + 92,
+		},
+		{
+			desc: "I<=1 or I >= 6",
+			q: baseQuery.Filter("T=", now).FilterEntity(OrFilter{
+				[]EntityFilter{
+					PropertyFilter{FieldName: "I", Operator: "<", Value: 4},
+					PropertyFilter{FieldName: "I", Operator: ">=", Value: 6},
+				},
+			}),
+			wantCount: 3,
+			wantSum:   3 + 6 + 7 + 92 + 93 + 96,
+		},
+		{
+			desc: "(T = now) and (((J > 97) and (T = tomorrow)) or (J < 94))",
+			q: baseQuery.FilterEntity(
+				AndFilter{
+					Filters: []EntityFilter{
+						OrFilter{
+							Filters: []EntityFilter{
+								AndFilter{
+									[]EntityFilter{
+										PropertyFilter{FieldName: "J", Operator: ">", Value: 97},
+										PropertyFilter{FieldName: "T", Operator: "=", Value: tomorrow},
+									},
+								},
+								PropertyFilter{FieldName: "J", Operator: "<", Value: 94},
+							},
+						},
+						PropertyFilter{FieldName: "T", Operator: "=", Value: now},
+					},
+				},
+			),
+			wantCount: 2,
+			wantSum:   6 + 7 + 92 + 93,
+		},
+	})
+}
+
 func TestIntegration_Filters(t *testing.T) {
 	ctx := context.Background()
 	client := newTestClient(ctx, t)
diff --git a/datastore/query.go b/datastore/query.go
index 195ae102bbf5..fb83f1c2cfea 100644
--- a/datastore/query.go
+++ b/datastore/query.go
@@ -30,21 +30,31 @@ import (
 	pb "google.golang.org/genproto/googleapis/datastore/v1"
 )
 
-type operator int
+type operator string
 
 const (
-	lessThan operator = iota + 1
-	lessEq
-	equal
-	greaterEq
-	greaterThan
-	in
-	notIn
-	notEqual
+	lessThan    operator = "<"
+	lessEq      operator = "<="
+	equal       operator = "="
+	greaterEq   operator = ">="
+	greaterThan operator = ">"
+	in          operator = "in"
+	notIn       operator = "not-in"
+	notEqual    operator = "!="
 
 	keyFieldName = "__key__"
 )
 
+var stringToOperator = createStringToOperator()
+
+func createStringToOperator() map[string]operator {
+	strToOp := make(map[string]operator)
+	for op := range operatorToProto {
+		strToOp[string(op)] = op
+	}
+	return strToOp
+}
+
 var operatorToProto = map[operator]pb.PropertyFilter_Operator{
 	lessThan:    pb.PropertyFilter_LESS_THAN,
 	lessEq:      pb.PropertyFilter_LESS_THAN_OR_EQUAL,
@@ -56,13 +66,6 @@ var operatorToProto = map[operator]pb.PropertyFilter_Operator{
 	notEqual:    pb.PropertyFilter_NOT_EQUAL,
 }
 
-// filter is a conditional filter on query results.
-type filter struct {
-	FieldName string
-	Op        operator
-	Value     interface{}
-}
-
 type sortDirection bool
 
 const (
@@ -81,6 +84,150 @@ type order struct {
 	Direction sortDirection
 }
 
+// EntityFilter represents a datastore filter.
+type EntityFilter interface {
+	toValidFilter() (EntityFilter, error)
+	toProto() (*pb.Filter, error)
+}
+
+// PropertyFilter represents field based filter.
+//
+// The operator parameter takes the following strings: ">", "<", ">=", "<=",
+// "=", "!=", "in", and "not-in".
+// Fields are compared against the provided value using the operator.
+// Field names which contain spaces, quote marks, or operator characters
+// should be passed as quoted Go string literals as returned by strconv.Quote
+// or the fmt package's %q verb.
+type PropertyFilter struct {
+	FieldName string
+	Operator  string
+	Value     interface{}
+}
+
+func (pf PropertyFilter) toProto() (*pb.Filter, error) {
+
+	if pf.FieldName == "" {
+		return nil, errors.New("datastore: empty query filter field name")
+	}
+	v, err := interfaceToProto(reflect.ValueOf(pf.Value).Interface(), false)
+	if err != nil {
+		return nil, fmt.Errorf("datastore: bad query filter value type: %w", err)
+	}
+
+	op, isOp := stringToOperator[pf.Operator]
+	if !isOp {
+		return nil, fmt.Errorf("datastore: invalid operator %q in filter", pf.Operator)
+	}
+
+	opProto, ok := operatorToProto[op]
+	if !ok {
+		return nil, errors.New("datastore: unknown query filter operator")
+	}
+	xf := &pb.PropertyFilter{
+		Op:       opProto,
+		Property: &pb.PropertyReference{Name: pf.FieldName},
+		Value:    v,
+	}
+	return &pb.Filter{
+		FilterType: &pb.Filter_PropertyFilter{PropertyFilter: xf},
+	}, nil
+}
+
+func (pf PropertyFilter) toValidFilter() (EntityFilter, error) {
+	op := strings.TrimSpace(pf.Operator)
+	_, isOp := stringToOperator[op]
+	if !isOp {
+		return nil, fmt.Errorf("datastore: invalid operator %q in filter", pf.Operator)
+	}
+
+	unquotedFieldName, err := unquote(pf.FieldName)
+	if err != nil {
+		return nil, fmt.Errorf("datastore: invalid syntax for quoted field name %q", pf.FieldName)
+	}
+
+	return PropertyFilter{Operator: op, FieldName: unquotedFieldName, Value: pf.Value}, nil
+}
+
+// CompositeFilter represents datastore composite filters.
+type CompositeFilter interface {
+	EntityFilter
+	isCompositeFilter()
+}
+
+// OrFilter represents a union of two or more filters.
+type OrFilter struct {
+	Filters []EntityFilter
+}
+
+func (OrFilter) isCompositeFilter() {}
+
+func (of OrFilter) toProto() (*pb.Filter, error) {
+
+	var pbFilters []*pb.Filter
+
+	for _, filter := range of.Filters {
+		pbFilter, err := filter.toProto()
+		if err != nil {
+			return nil, err
+		}
+		pbFilters = append(pbFilters, pbFilter)
+	}
+	return &pb.Filter{FilterType: &pb.Filter_CompositeFilter{CompositeFilter: &pb.CompositeFilter{
+		Op:      pb.CompositeFilter_OR,
+		Filters: pbFilters,
+	}}}, nil
+}
+
+func (of OrFilter) toValidFilter() (EntityFilter, error) {
+	var validFilters []EntityFilter
+	for _, filter := range of.Filters {
+		validFilter, err := filter.toValidFilter()
+		if err != nil {
+			return nil, err
+		}
+		validFilters = append(validFilters, validFilter)
+	}
+	of.Filters = validFilters
+	return of, nil
+}
+
+// AndFilter represents the intersection of two or more filters.
+type AndFilter struct {
+	Filters []EntityFilter
+}
+
+func (AndFilter) isCompositeFilter() {}
+
+func (af AndFilter) toProto() (*pb.Filter, error) {
+
+	var pbFilters []*pb.Filter
+
+	for _, filter := range af.Filters {
+		pbFilter, err := filter.toProto()
+		if err != nil {
+			return nil, err
+		}
+		pbFilters = append(pbFilters, pbFilter)
+	}
+	return &pb.Filter{FilterType: &pb.Filter_CompositeFilter{CompositeFilter: &pb.CompositeFilter{
+		Op:      pb.CompositeFilter_AND,
+		Filters: pbFilters,
+	}}}, nil
+}
+
+func (af AndFilter) toValidFilter() (EntityFilter, error) {
+	var validFilters []EntityFilter
+	for _, filter := range af.Filters {
+		validFilter, err := filter.toValidFilter()
+		if err != nil {
+			return nil, err
+		}
+		validFilters = append(validFilters, validFilter)
+	}
+	af.Filters = validFilters
+	return af, nil
+}
+
 // NewQuery creates a new Query for a specific entity kind.
 //
 // An empty kind means to return all entities, including entities created and
@@ -97,7 +244,7 @@ func NewQuery(kind string) *Query {
 type Query struct {
 	kind       string
 	ancestor   *Key
-	filter     []filter
+	filter     []EntityFilter
 	order      []order
 	projection []string
 
@@ -121,7 +268,7 @@ func (q *Query) clone() *Query {
 	x := *q
 	// Copy the contents of the slice-typed fields to a new backing store.
 	if len(q.filter) > 0 {
-		x.filter = make([]filter, len(q.filter))
+		x.filter = make([]EntityFilter, len(q.filter))
 		copy(x.filter, q.filter)
 	}
 	if len(q.order) > 0 {
@@ -176,6 +323,22 @@ func (q *Query) Transaction(t *Transaction) *Query {
 	return q
 }
 
+// FilterEntity returns a query with provided filter.
+//
+// Filter can be a single field comparison or a composite filter
+// AndFilter and OrFilter are supported composite filters
+// Filters in multiple calls are joined together by AND
+func (q *Query) FilterEntity(ef EntityFilter) *Query {
+	q = q.clone()
+	vf, err := ef.toValidFilter()
+	if err != nil {
+		q.err = err
+		return q
+	}
+	q.filter = append(q.filter, vf)
+	return q
+}
+
 // Filter returns a derivative query with a field-based filter.
 //
 // Deprecated: Use the FilterField method instead, which supports the same
@@ -209,42 +372,11 @@ func (q *Query) Filter(filterStr string, value interface{}) *Query {
 // should be passed as quoted Go string literals as returned by strconv.Quote
 // or the fmt package's %q verb.
 func (q *Query) FilterField(fieldName, operator string, value interface{}) *Query {
-	q = q.clone()
-
-	f := filter{
+	return q.FilterEntity(PropertyFilter{
 		FieldName: fieldName,
+		Operator:  operator,
 		Value:     value,
-	}
-
-	switch o := strings.TrimSpace(operator); o {
-	case "<=":
-		f.Op = lessEq
-	case ">=":
-		f.Op = greaterEq
-	case "<":
-		f.Op = lessThan
-	case ">":
-		f.Op = greaterThan
-	case "=":
-		f.Op = equal
-	case "in":
-		f.Op = in
-	case "not-in":
-		f.Op = notIn
-	case "!=":
-		f.Op = notEqual
-	default:
-		q.err = fmt.Errorf("datastore: invalid operator %q in filter", operator)
-		return q
-	}
-	var err error
-	f.FieldName, err = unquote(f.FieldName)
-	if err != nil {
-		q.err = fmt.Errorf("datastore: invalid syntax for quoted field name %q", f.FieldName)
-		return q
-	}
-	q.filter = append(q.filter, f)
-	return q
+	})
 }
 
 // Order returns a derivative query with a field-based sort order. Orders are
@@ -411,25 +543,11 @@ func (q *Query) toProto() (*pb.Query, error) {
 	}
 	var filters []*pb.Filter
 	for _, qf := range q.filter {
-		if qf.FieldName == "" {
-			return nil, errors.New("datastore: empty query filter field name")
-		}
-		v, err := interfaceToProto(reflect.ValueOf(qf.Value).Interface(), false)
+		pbFilter, err := qf.toProto()
 		if err != nil {
-			return nil, fmt.Errorf("datastore: bad query filter value type: %v", err)
-		}
-		op, ok := operatorToProto[qf.Op]
-		if !ok {
-			return nil, errors.New("datastore: unknown query filter operator")
+			return nil, err
 		}
-		xf := &pb.PropertyFilter{
-			Op:       op,
-			Property: &pb.PropertyReference{Name: qf.FieldName},
-			Value:    v,
-		}
-		filters = append(filters, &pb.Filter{
-			FilterType: &pb.Filter_PropertyFilter{PropertyFilter: xf},
-		})
+		filters = append(filters, pbFilter)
 	}
 
 	if q.ancestor != nil {
diff --git a/datastore/query_test.go b/datastore/query_test.go
index 824c439e97a0..22d13e1776ab 100644
--- a/datastore/query_test.go
+++ b/datastore/query_test.go
@@ -482,24 +482,136 @@ var (
 	}
 	// Operators not supported in either filter method
 	filterUnsupported = []testFilterCase{
-		{"x IN", "x", "IN", 0, ""},
-		{"x NOT-IN", "x", "NOT-IN", 0, ""},
-		{"x EQ", "x", "EQ", 0, ""},
-		{"x lt", "x", "lt", 0, ""},
-		{"x <>", "x", "<>", 0, ""},
-		{"x >>", "x", ">>", 0, ""},
-		{"x ==", "x", "==", 0, ""},
-		{"x =<", "x", "=<", 0, ""},
-		{"x =>", "x", "=>", 0, ""},
-		{"x !", "x", "!", 0, ""},
-		{"x ", "x", "", 0, ""},
-		{"x", "x", "", 0, ""},
-		{`" x =`, `" x`, "=", 0, ""},
-		{`" x ="`, `" x `, `="`, 0, ""},
-		{"` x \" =", "` x \"", "=", 0, ""},
+		{filterStr: "x IN", fieldName: "x", operator: "IN", wantOp: "", wantFieldName: ""},
+		{filterStr: "x NOT-IN", fieldName: "x", operator: "NOT-IN", wantOp: "", wantFieldName: ""},
+		{filterStr: "x EQ", fieldName: "x", operator: "EQ", wantOp: "", wantFieldName: ""},
+		{filterStr: "x lt", fieldName: "x", operator: "lt", wantOp: "", wantFieldName: ""},
+		{filterStr: "x <>", fieldName: "x", operator: "<>", wantOp: "", wantFieldName: ""},
+		{filterStr: "x >>", fieldName: "x", operator: ">>", wantOp: "", wantFieldName: ""},
+		{filterStr: "x ==", fieldName: "x", operator: "==", wantOp: "", wantFieldName: ""},
+		{filterStr: "x =<", fieldName: "x", operator: "=<", wantOp: "", wantFieldName: ""},
+		{filterStr: "x =>", fieldName: "x", operator: "=>", wantOp: "", wantFieldName: ""},
+		{filterStr: "x !", fieldName: "x", operator: "!", wantOp: "", wantFieldName: ""},
+		{filterStr: "x ", fieldName: "x", operator: "", wantOp: "", wantFieldName: ""},
+		{filterStr: "x", fieldName: "x", operator: "", wantOp: "", wantFieldName: ""},
+		{filterStr: `" x =`, fieldName: `" x`, operator: "=", wantOp: "", wantFieldName: ""},
+		{filterStr: `" x ="`, fieldName: `" x `, operator: `="`, wantOp: "", wantFieldName: ""},
+		{filterStr: "` x \" =", fieldName: "` x \"", operator: "=", wantOp: "", wantFieldName: ""},
 	}
 )
 
+type pfToProtoTestCase struct {
+	pf         PropertyFilter
+	wantErrMsg string
+}
+
+func TestPropertyFilterToProto(t *testing.T) {
+
+	testCases := []pfToProtoTestCase{
+		{PropertyFilter{FieldName: "x", Operator: "=", Value: 4}, ""},
+		{PropertyFilter{FieldName: "x ", Operator: "=", Value: 4}, ""},
+		{PropertyFilter{FieldName: "", Operator: "=", Value: 4}, "datastore: empty query filter field name"},
+		{PropertyFilter{FieldName: "x", Operator: "==", Value: 4}, "datastore: invalid operator \"==\" in filter"},
+		{PropertyFilter{FieldName: "x", Operator: "==", Value: struct{ x string }{x: "sample"}}, "datastore: bad query filter value type: invalid Value type struct { x string }"},
+	}
+
+	successFilterFieldTestCases := append(filterTestCases, filterFieldTestCases...)
+	for _, sfftc := range successFilterFieldTestCases {
+		testCases = append(testCases, pfToProtoTestCase{
+			PropertyFilter{FieldName: sfftc.fieldName, Operator: sfftc.operator, Value: 4}, "",
+		})
+	}
+
+	for _, tc := range testCases {
+		_, err := tc.pf.toProto()
+		gotErrMsg := ""
+		if err != nil {
+			gotErrMsg = err.Error()
+		}
+
+		if gotErrMsg != tc.wantErrMsg {
+			t.Errorf("PropertyFilter proto conversion error: \nwant %v,\ngot %v", tc.wantErrMsg, gotErrMsg)
+		}
+	}
+}
+
+func TestCompositeFilterToProto(t *testing.T) {
+	testCases := []struct {
+		cf         CompositeFilter
+		wantErrMsg string
+	}{
+		{AndFilter{
+			[]EntityFilter{
+				PropertyFilter{FieldName: "x", Operator: "=", Value: 4},
+				PropertyFilter{FieldName: "y", Operator: "<", Value: 3},
+			},
+		}, ""},
+		{OrFilter{
+			[]EntityFilter{
+				PropertyFilter{FieldName: "x", Operator: "=", Value: 4},
+				PropertyFilter{FieldName: "y", Operator: "<", Value: 3},
+			},
+		}, ""},
+
+		// Fail when inner filter is malformed
+		{AndFilter{
+			[]EntityFilter{
+				PropertyFilter{FieldName: "x", Operator: "==", Value: 4},
+				PropertyFilter{FieldName: "y", Operator: "<", Value: 3},
+			},
+		}, "datastore: invalid operator \"==\" in filter"},
+		{OrFilter{
+			[]EntityFilter{
+				PropertyFilter{FieldName: "x", Operator: "==", Value: 4},
+				PropertyFilter{FieldName: "y", Operator: "<", Value: 3},
+			},
+		}, "datastore: invalid operator \"==\" in filter"},
+	}
+	for _, tc := range testCases {
+		_, gotErr := tc.cf.toProto()
+		gotErrMsg := ""
+		if gotErr != nil {
+			gotErrMsg = gotErr.Error()
+		}
+
+		if gotErrMsg != tc.wantErrMsg {
+			t.Errorf("CompositeFilter proto conversion error: \nwant %v,\ngot %v", tc.wantErrMsg, gotErrMsg)
+		}
+
+	}
+}
+
+func TestFilterEntity(t *testing.T) {
+	testCases := []struct {
+		ef         EntityFilter
+		wantErrMsg string
+	}{
+		{AndFilter{
+			[]EntityFilter{
+				PropertyFilter{FieldName: "x", Operator: "==", Value: 4},
+				PropertyFilter{FieldName: "y", Operator: "<", Value: 3},
+			},
+		}, "datastore: invalid operator \"==\" in filter"},
+		{AndFilter{
+			[]EntityFilter{
+				PropertyFilter{FieldName: "`x", Operator: "=", Value: 4},
+				PropertyFilter{FieldName: "y", Operator: "<", Value: 3},
+			},
+		}, "datastore: invalid syntax for quoted field name \"`x\""},
+	}
+
+	for _, tc := range testCases {
+		q := NewQuery("foo").FilterEntity(tc.ef)
+		gotErrMsg := ""
+		if q.err != nil {
+			gotErrMsg = q.err.Error()
+		}
+		if gotErrMsg != tc.wantErrMsg {
+			t.Errorf("FilterEntity error: \nwant %v,\ngot %v", tc.wantErrMsg, gotErrMsg)
+		}
+	}
+}
+
 func TestFilterParser(t *testing.T) {
 	// Success cases
 	for _, tc := range filterTestCases {
@@ -512,7 +624,7 @@ func TestFilterParser(t *testing.T) {
 			t.Errorf("%q: len=%d, want %d", tc.filterStr, len(q.filter), 1)
 			continue
 		}
-		got, want := q.filter[0], filter{tc.wantFieldName, tc.wantOp, 42}
+		got, want := q.filter[0], PropertyFilter{tc.wantFieldName, string(tc.wantOp), 42}
 		if got != want {
 			t.Errorf("%q: got %v, want %v", tc.filterStr, got, want)
 			continue
@@ -540,7 +652,7 @@ func TestFilterField(t *testing.T) {
 			t.Errorf("%q: len=%d, want %d", tc.fieldName, len(q.filter), 1)
 			continue
 		}
-		got, want := q.filter[0], filter{tc.fieldName, tc.wantOp, 42}
+		got, want := q.filter[0], PropertyFilter{tc.fieldName, string(tc.wantOp), 42}
 		if got != want {
 			t.Errorf("%q %q: got %v, want %v", tc.fieldName, tc.operator, got, want)
 			continue