diff --git a/document/doc_id.go b/document/doc_id.go
index 18a0bb5..b97b6cb 100644
--- a/document/doc_id.go
+++ b/document/doc_id.go
@@ -61,7 +61,7 @@ type fullDocIDSet struct {
 	numTotalDocs int32
 }
 
-func (s *fullDocIDSet) NumDocuments() int32 { return int32(s.numTotalDocs) }
+func (s *fullDocIDSet) NumDocuments() int32 { return s.numTotalDocs }
 
 func (s *fullDocIDSet) Iter() DocIDSetIterator { return NewFullDocIDSetIterator(s.numTotalDocs) }
 
diff --git a/event/field/field.go b/event/field/field.go
index 02ca2ef..5b18a8a 100644
--- a/event/field/field.go
+++ b/event/field/field.go
@@ -1,11 +1,5 @@
 package field
 
-import (
-	"fmt"
-
-	"github.com/xichen2020/eventdb/event/field"
-)
-
 // ValueType is the type of a field value.
 type ValueType int
 
@@ -39,30 +33,33 @@ func (t ValueType) String() string {
 	}
 }
 
-// ConvertibleFrom returns a list of value types that can be converted into type `t`.
-func (t ValueType) ConvertibleFrom() ([]ValueType, error) {
-	vts, exists := comparableTypeMap[t]
-	if exists {
-		return vts, nil
-	}
-	return nil, fmt.Errorf("unknown value type %v does not have comparable types", t)
-}
-
 // ValueTypeSet is a set of value types.
-type ValueTypeSet map[field.ValueType]struct{}
+type ValueTypeSet map[ValueType]struct{}
 
 // Clone clones a value type set.
 func (m ValueTypeSet) Clone() ValueTypeSet {
 	if len(m) == 0 {
 		return nil
 	}
-	cloned := make(field.ValueTypeSet, len(m))
+	cloned := make(ValueTypeSet, len(m))
 	for k := range m {
 		cloned[k] = struct{}{}
 	}
 	return cloned
 }
 
+var (
+	// OrderableTypes is a list of value types eligible for ordering.
+	OrderableTypes = ValueTypeSet{
+		NullType:   struct{}{},
+		BoolType:   struct{}{},
+		IntType:    struct{}{},
+		DoubleType: struct{}{},
+		StringType: struct{}{},
+		TimeType:   struct{}{},
+	}
+)
+
 // ValueUnion is a value union.
 type ValueUnion struct {
 	Type         ValueType
@@ -73,46 +70,6 @@ type ValueUnion struct {
 	TimeNanosVal int64
 }
 
-// ValueExtractionMode determines how to extract values from the value union.
-type ValueExtractionMode int
-
-// A list of supported value extraction modes.
-const (
-	DontConvertTypes ValueExtractionMode = iota
-	ConvertTypes
-)
-
-// BoolValue attempts to extract a bool value from the union.
-func (v *ValueUnion) BoolValue(mode ValueExtractionMode) (bool, error) {
-	if v.Type == BoolType {
-		return v.BoolVal, nil
-	}
-	if mode == DontConvertTypes {
-		return false, fmt.Errorf("unable to extract a bool value from the union %v", v)
-	}
-
-}
-
-var (
-	// OrderableTypes is a list of value types eligible for ordering.
-	OrderableTypes = []ValueType{
-		NullType,
-		BoolType,
-		IntType,
-		DoubleType,
-		StringType,
-		TimeType,
-	}
-	convertibleFromTypesMap = map[ValueType][]ValueType{
-		NullType:   []ValueType{NullType},
-		BoolType:   []ValueType{BoolType},
-		IntType:    []ValueType{IntType, DoubleType},
-		DoubleType: []ValueType{IntType, DoubleType},
-		StringType: []ValueType{StringType},
-		TimeType:   []ValueType{TimeType},
-	}
-)
-
 // Field is an event field.
 type Field struct {
 	Path  []string
diff --git a/filter/filter_op.go b/filter/filter_op.go
index dea8e8d..aeead74 100644
--- a/filter/filter_op.go
+++ b/filter/filter_op.go
@@ -98,21 +98,17 @@ func (f Op) BoolFilter(v *field.ValueUnion) (BoolFilter, error) {
 		return nil, fmt.Errorf("operator %v is not a value filter", f)
 	}
 	if v == nil {
-		return nil, fmt.Errorf("operator %v has nil operand", f)
+		return nil, fmt.Errorf("operator %v has nil RHS operand", f)
 	}
-	bv, err := v.BoolValue()
-	if err != nil {
-		return nil, fmt.Errorf("unable to extract bool operand for operator %v: %v", f, err)
-	}
-	switch f {
-	case Equals:
-	case NotEquals:
-	case LargerThan:
-	case LargerThanOrEqual:
-	case SmallerThan:
-	case SmallerThanOrEqual:
+	switch v.Type {
+	case field.BoolType:
+		filterGen, exists := boolToBoolFilterOps[f]
+		if !exists {
+			return nil, fmt.Errorf("operator %v does not have a bool filter generator for bool RHS value", f)
+		}
+		return filterGen(v.BoolVal), nil
 	default:
-		return nil, fmt.Errorf("operator %v does not apply to bool values", f)
+		return nil, fmt.Errorf("operator %v has an invalid RHS operand type %v", f, v.Type)
 	}
 }
 
@@ -122,21 +118,23 @@ func (f Op) IntFilter(v *field.ValueUnion) (IntFilter, error) {
 		return nil, fmt.Errorf("operator %v is not a value filter", f)
 	}
 	if v == nil {
-		return nil, fmt.Errorf("operator %v has nil operand", f)
-	}
-	iv, err := v.IntValue()
-	if err != nil {
-		return nil, fmt.Errorf("unable to extract int operand for operator %v: %v", f, err)
+		return nil, fmt.Errorf("operator %v has nil RHS operand", f)
 	}
-	switch f {
-	case Equals:
-	case NotEquals:
-	case LargerThan:
-	case LargerThanOrEqual:
-	case SmallerThan:
-	case SmallerThanOrEqual:
+	switch v.Type {
+	case field.IntType:
+		filterGen, exists := intToIntFilterOps[f]
+		if !exists {
+			return nil, fmt.Errorf("operator %v does not have an int filter generator for int RHS value", f)
+		}
+		return filterGen(v.IntVal), nil
+	case field.DoubleType:
+		filterGen, exists := doubleToIntFilterOps[f]
+		if !exists {
+			return nil, fmt.Errorf("operator %v does not have an int filter generator for double RHS value", f)
+		}
+		return filterGen(v.DoubleVal), nil
 	default:
-		return nil, fmt.Errorf("operator %v does not apply to int values", f)
+		return nil, fmt.Errorf("operator %v has an invalid RHS operand type %v", f, v.Type)
 	}
 }
 
@@ -146,21 +144,23 @@ func (f Op) DoubleFilter(v *field.ValueUnion) (DoubleFilter, error) {
 		return nil, fmt.Errorf("operator %v is not a value filter", f)
 	}
 	if v == nil {
-		return nil, fmt.Errorf("operator %v has nil operand", f)
+		return nil, fmt.Errorf("operator %v has nil RHS operand", f)
 	}
-	dv, err := v.DoubleValue()
-	if err != nil {
-		return nil, fmt.Errorf("unable to extract double operand for operator %v: %v", f, err)
-	}
-	switch f {
-	case Equals:
-	case NotEquals:
-	case LargerThan:
-	case LargerThanOrEqual:
-	case SmallerThan:
-	case SmallerThanOrEqual:
+	switch v.Type {
+	case field.IntType:
+		filterGen, exists := intToDoubleFilterOps[f]
+		if !exists {
+			return nil, fmt.Errorf("operator %v does not have a double filter generator for int RHS value", f)
+		}
+		return filterGen(v.IntVal), nil
+	case field.DoubleType:
+		filterGen, exists := doubleToDoubleFilterOps[f]
+		if !exists {
+			return nil, fmt.Errorf("operator %v does not have a double filter generator for double RHS value", f)
+		}
+		return filterGen(v.DoubleVal), nil
 	default:
-		return nil, fmt.Errorf("operator %v does not apply to double values", f)
+		return nil, fmt.Errorf("operator %v has an invalid RHS operand type %v", f, v.Type)
 	}
 }
 
@@ -170,27 +170,17 @@ func (f Op) StringFilter(v *field.ValueUnion) (StringFilter, error) {
 		return nil, fmt.Errorf("operator %v is not a value filter", f)
 	}
 	if v == nil {
-		return nil, fmt.Errorf("operator %v has nil operand", f)
-	}
-	sv, err := v.StringValue()
-	if err != nil {
-		return nil, fmt.Errorf("unable to extract string operand for operator %v: %v", f, err)
+		return nil, fmt.Errorf("operator %v has nil RHS operand", f)
 	}
-	switch f {
-	case Equals:
-	case NotEquals:
-	case LargerThan:
-	case LargerThanOrEqual:
-	case SmallerThan:
-	case SmallerThanOrEqual:
-	case StartsWith:
-	case DoesNotStartWith:
-	case EndsWith:
-	case DoesNotEndWith:
-	case Contains:
-	case DoesNotContain:
+	switch v.Type {
+	case field.StringType:
+		filterGen, exists := stringToStringFilterOps[f]
+		if !exists {
+			return nil, fmt.Errorf("operator %v does not have a string filter generator for string RHS value", f)
+		}
+		return filterGen(v.StringVal), nil
 	default:
-		return nil, fmt.Errorf("operator %v does not apply to string values", f)
+		return nil, fmt.Errorf("operator %v has an invalid RHS operand type %v", f, v.Type)
 	}
 }
 
@@ -200,21 +190,17 @@ func (f Op) TimeFilter(v *field.ValueUnion) (TimeFilter, error) {
 		return nil, fmt.Errorf("operator %v is not a value filter", f)
 	}
 	if v == nil {
-		return nil, fmt.Errorf("operator %v has nil operand", f)
+		return nil, fmt.Errorf("operator %v has nil RHS operand", f)
 	}
-	tv, err := v.TimeValue()
-	if err != nil {
-		return nil, fmt.Errorf("unable to extract time operand for operator %v: %v", f, err)
-	}
-	switch f {
-	case Equals:
-	case NotEquals:
-	case LargerThan:
-	case LargerThanOrEqual:
-	case SmallerThan:
-	case SmallerThanOrEqual:
+	switch v.Type {
+	case field.TimeType:
+		filterGen, exists := timeToTimeFilterOps[f]
+		if !exists {
+			return nil, fmt.Errorf("operator %v does not have a time filter generator for time RHS value", f)
+		}
+		return filterGen(v.TimeNanosVal), nil
 	default:
-		return nil, fmt.Errorf("operator %v does not apply to time values", f)
+		return nil, fmt.Errorf("operator %v has an invalid RHS operand type %v", f, v.Type)
 	}
 }
 
@@ -230,19 +216,19 @@ func (f Op) AllowedTypes(v *field.ValueUnion) (field.ValueTypeSet, error) {
 		if !exists {
 			return nil, fmt.Errorf("doc ID set filter op %v does not have allowed types", f)
 		}
-		return allowed.Clone()
+		return allowed.Clone(), nil
 	}
 
 	// Value filter.
 	if v == nil {
-		return nil, fmt.Errorf("value filter op %v does not have a RHS operand", f)
+		return nil, fmt.Errorf("value filter op %v has a nil RHS operand", f)
 	}
 	allowedByRHSType, exists := allowedTypesByValueFilterOpAndRHSType[f]
 	if !exists {
 		return nil, fmt.Errorf("value filter op %v does not have allowed types", f)
 	}
 	allowed := allowedByRHSType[v.Type]
-	return allowed.Clone()
+	return allowed.Clone(), nil
 }
 
 // MultiTypeCombinator returns the filter combinator to join the filter result for
@@ -351,6 +337,12 @@ var (
 		LargerThanOrEqual:  largerThanOrEqualStringString,
 		SmallerThan:        smallerThanStringString,
 		SmallerThanOrEqual: smallerThanOrEqualStringString,
+		StartsWith:         startsWithStringString,
+		DoesNotStartWith:   doesNotStartWithStringString,
+		EndsWith:           endsWithStringString,
+		DoesNotEndWith:     doesNotEndWithStringString,
+		Contains:           containsStringString,
+		DoesNotContain:     doesNotContainStringString,
 	}
 	timeToTimeFilterOps = map[Op]timeToTimeFilterFn{
 		Equals:             equalsTimeTime,
@@ -410,97 +402,6 @@ var (
 		DoesNotExist:       "notExists",
 	}
 	stringToOps map[string]Op
-
-	/*
-		opCompatibleValueTypes = map[Op]map[field.ValueType]struct{}{
-			Equals: map[field.ValueType]struct{}{
-				field.BoolType:   struct{}{},
-				field.IntType:    struct{}{},
-				field.DoubleType: struct{}{},
-				field.StringType: struct{}{},
-				field.TimeType:   struct{}{},
-			},
-			NotEquals: map[field.ValueType]struct{}{
-				field.BoolType:   struct{}{},
-				field.IntType:    struct{}{},
-				field.DoubleType: struct{}{},
-				field.StringType: struct{}{},
-				field.TimeType:   struct{}{},
-			},
-			LargerThan: map[field.ValueType]struct{}{
-				field.BoolType:   struct{}{},
-				field.IntType:    struct{}{},
-				field.DoubleType: struct{}{},
-				field.StringType: struct{}{},
-				field.TimeType:   struct{}{},
-			},
-			LargerThanOrEqual: map[field.ValueType]struct{}{
-				field.BoolType:   struct{}{},
-				field.IntType:    struct{}{},
-				field.DoubleType: struct{}{},
-				field.StringType: struct{}{},
-				field.TimeType:   struct{}{},
-			},
-			SmallerThan: map[field.ValueType]struct{}{
-				field.BoolType:   struct{}{},
-				field.IntType:    struct{}{},
-				field.DoubleType: struct{}{},
-				field.StringType: struct{}{},
-				field.TimeType:   struct{}{},
-			},
-			SmallerThanOrEqual: map[field.ValueType]struct{}{
-				field.BoolType:   struct{}{},
-				field.IntType:    struct{}{},
-				field.DoubleType: struct{}{},
-				field.StringType: struct{}{},
-				field.TimeType:   struct{}{},
-			},
-			StartsWith: map[field.ValueType]struct{}{
-				field.StringType: struct{}{},
-			},
-			DoesNotStartWith: map[field.ValueType]struct{}{
-				field.StringType: struct{}{},
-			},
-			EndsWith: map[field.ValueType]struct{}{
-				field.StringType: struct{}{},
-			},
-			DoesNotEndWith: map[field.ValueType]struct{}{
-				field.StringType: struct{}{},
-			},
-			Contains: map[field.ValueType]struct{}{
-				field.StringType: struct{}{},
-			},
-			DoesNotContain: map[field.ValueType]struct{}{
-				field.StringType: struct{}{},
-			},
-			IsNull: map[field.ValueType]struct{}{
-				field.NullType: struct{}{},
-			},
-			IsNotNull: map[field.ValueType]struct{}{
-				field.BoolType:   struct{}{},
-				field.IntType:    struct{}{},
-				field.DoubleType: struct{}{},
-				field.StringType: struct{}{},
-				field.TimeType:   struct{}{},
-			},
-			Exists: map[field.ValueType]struct{}{
-				field.NullType:   struct{}{},
-				field.BoolType:   struct{}{},
-				field.IntType:    struct{}{},
-				field.DoubleType: struct{}{},
-				field.StringType: struct{}{},
-				field.TimeType:   struct{}{},
-			},
-			DoesNotExist: map[field.ValueType]struct{}{
-				field.NullType:   struct{}{},
-				field.BoolType:   struct{}{},
-				field.IntType:    struct{}{},
-				field.DoubleType: struct{}{},
-				field.StringType: struct{}{},
-				field.TimeType:   struct{}{},
-			},
-		}
-	*/
 )
 
 func addAllowedTypes(op Op, lhsType, rhsType field.ValueType) {
diff --git a/query/filter.go b/query/filter.go
index 6efa480..f29af27 100644
--- a/query/filter.go
+++ b/query/filter.go
@@ -18,13 +18,13 @@ type RawFilter struct {
 	Value interface{} `json:"value"`
 }
 
-// FilterList is a list of sanitized filters.
+// FilterList is a list of parsed filters.
 type FilterList struct {
 	Filters          []Filter
 	FilterCombinator filter.Combinator
 }
 
-// Filter is a sanitized filter.
+// Filter is a parsed filter.
 // TODO(xichen): Handle range query.
 type Filter struct {
 	FieldPath []string
@@ -32,34 +32,10 @@ type Filter struct {
 	Value     *field.ValueUnion
 }
 
-// AllowedFieldTypes returns the set of field types eligible for
-// matching against the filter.
-func (f Filter) AllowedFieldTypes() (map[field.ValueType]struct{}, error) {
-	opTypes, err := f.Op.CompatibleTypes()
-	if err != nil {
-		return nil, err
-	}
-	res := make(map[field.ValueType]struct{}, len(opTypes))
-	if f.Value == nil {
-		for t := range opTypes {
-			res[t] = struct{}{}
-		}
-		return res, nil
-	}
-	// Find the set of types that are compatible with the filter operator
-	// and comparable with the filter value.
-	valueTypes, err := f.Value.Type.ComparableTypes()
-	if err != nil {
-		return nil, err
-	}
-	for _, t := range valueTypes {
-		_, exists := opTypes[t]
-		if !exists {
-			continue
-		}
-		res[t] = struct{}{}
-	}
-	return res, nil
+// AllowedFieldTypes returns a list of allowed field types given the filter
+// operator and the RHS value if applicable.
+func (f Filter) AllowedFieldTypes() (field.ValueTypeSet, error) {
+	return f.Op.AllowedTypes(f.Value)
 }
 
 // BoolFilter returns a bool filter if applicable, or an error otherwise.
diff --git a/storage/immutable_segment.go b/storage/immutable_segment.go
index 91ed83f..8d7ac1a 100644
--- a/storage/immutable_segment.go
+++ b/storage/immutable_segment.go
@@ -18,7 +18,7 @@ import (
 
 type retrieveFieldOptions struct {
 	fieldPath            []string
-	fieldTypesToRetrieve map[field.ValueType]struct{}
+	fieldTypesToRetrieve field.ValueTypeSet
 }
 
 // immutableSegment is an immutable segment.
@@ -58,10 +58,6 @@ const (
 	segmentUnloaded
 )
 
-var (
-	errNoEligibleFieldTypesForQuery = errors.New("no eligible field types for query")
-)
-
 type immutableSegmentOptions struct {
 	timestampFieldPath    []string
 	rawDocSourcePath      []string
@@ -238,7 +234,7 @@ func (s *immutableSeg) collectFieldsForRawQuery(
 	filters []query.FilterList,
 	orderBy []query.OrderBy,
 ) (
-	allowedFieldTypes []map[field.ValueType]struct{},
+	allowedFieldTypes []field.ValueTypeSet,
 	fieldIndexMap []int,
 	fieldsToRetrieve []retrieveFieldOptions,
 	err error,
@@ -257,8 +253,8 @@ func (s *immutableSeg) collectFieldsForRawQuery(
 	currIndex := 0
 	s.addQueryFieldToMap(fieldMap, queryFieldMeta{
 		fieldPath: s.timestampFieldPath,
-		allowedTypesBySourceIdx: map[int]map[field.ValueType]struct{}{
-			currIndex: map[field.ValueType]struct{}{
+		allowedTypesBySourceIdx: map[int]field.ValueTypeSet{
+			currIndex: field.ValueTypeSet{
 				field.TimeType: struct{}{},
 			},
 		},
@@ -268,8 +264,8 @@ func (s *immutableSeg) collectFieldsForRawQuery(
 	currIndex++
 	s.addQueryFieldToMap(fieldMap, queryFieldMeta{
 		fieldPath: s.rawDocSourcePath,
-		allowedTypesBySourceIdx: map[int]map[field.ValueType]struct{}{
-			currIndex: map[field.ValueType]struct{}{
+		allowedTypesBySourceIdx: map[int]field.ValueTypeSet{
+			currIndex: field.ValueTypeSet{
 				field.StringType: struct{}{},
 			},
 		},
@@ -285,7 +281,7 @@ func (s *immutableSeg) collectFieldsForRawQuery(
 			}
 			s.addQueryFieldToMap(fieldMap, queryFieldMeta{
 				fieldPath: f.FieldPath,
-				allowedTypesBySourceIdx: map[int]map[field.ValueType]struct{}{
+				allowedTypesBySourceIdx: map[int]field.ValueTypeSet{
 					currIndex: allowedFieldTypes,
 				},
 			})
@@ -297,8 +293,8 @@ func (s *immutableSeg) collectFieldsForRawQuery(
 	for _, ob := range orderBy {
 		s.addQueryFieldToMap(fieldMap, queryFieldMeta{
 			fieldPath: ob.FieldPath,
-			allowedTypesBySourceIdx: map[int]map[field.ValueType]struct{}{
-				currIndex: toFieldTypeMap(field.OrderableTypes),
+			allowedTypesBySourceIdx: map[int]field.ValueTypeSet{
+				currIndex: field.OrderableTypes.Clone(),
 			},
 		})
 		currIndex++
@@ -309,12 +305,12 @@ func (s *immutableSeg) collectFieldsForRawQuery(
 	s.intersectWithAvailableTypes(fieldMap)
 
 	// Flatten the list of fields.
-	allowedFieldTypes = make([]map[field.ValueType]struct{}, numFieldsForQuery)
+	allowedFieldTypes = make([]field.ValueTypeSet, numFieldsForQuery)
 	fieldIndexMap = make([]int, numFieldsForQuery)
 	fieldsToRetrieve = make([]retrieveFieldOptions, 0, len(fieldMap))
 	fieldIndex := 0
 	for _, f := range fieldMap {
-		allAllowedTypes := make(map[field.ValueType]struct{})
+		allAllowedTypes := make(field.ValueTypeSet)
 		for sourceIdx, types := range f.allowedTypesBySourceIdx {
 			allowedFieldTypes[sourceIdx] = types
 			fieldIndexMap[sourceIdx] = fieldIndex
@@ -524,7 +520,7 @@ func (s *immutableSeg) insertFields(
 func applyFilters(
 	startNanosInclusive, endNanosExclusive int64,
 	filters []query.FilterList,
-	allowedFieldTypes []map[field.ValueType]struct{},
+	allowedFieldTypes []field.ValueTypeSet,
 	fieldIndexMap []int,
 	queryFields []document.ReadOnlyDocsField,
 	numTotalDocs int32,
@@ -596,7 +592,7 @@ func applyFilters(
 func applyFilter(
 	flt query.Filter,
 	fld document.ReadOnlyDocsField,
-	fieldTypes map[field.ValueType]struct{},
+	fieldTypes field.ValueTypeSet,
 	numTotalDocs int32,
 ) (document.DocIDSetIterator, error) {
 	// Determine whether the filter operator is a doc ID set filter.
@@ -745,22 +741,14 @@ func applyFilterForType(
 	}
 }
 
-func toFieldTypeMap(fieldTypes []field.ValueType) map[field.ValueType]struct{} {
-	m := make(map[field.ValueType]struct{}, len(fieldTypes))
-	for _, ft := range fieldTypes {
-		m[ft] = struct{}{}
-	}
-	return m
-}
-
 func intersectFieldTypes(
 	first []field.ValueType,
-	second map[field.ValueType]struct{},
-) map[field.ValueType]struct{} {
+	second field.ValueTypeSet,
+) field.ValueTypeSet {
 	if len(first) == 0 || len(second) == 0 {
 		return nil
 	}
-	res := make(map[field.ValueType]struct{}, len(first))
+	res := make(field.ValueTypeSet, len(first))
 	for _, t := range first {
 		_, exists := second[t]
 		if !exists {
@@ -779,7 +767,7 @@ type loadFieldMetadata struct {
 
 type queryFieldMeta struct {
 	fieldPath               []string
-	allowedTypesBySourceIdx map[int]map[field.ValueType]struct{}
+	allowedTypesBySourceIdx map[int]field.ValueTypeSet
 }
 
 // Precondition: m.fieldPath == other.fieldPath.
@@ -798,8 +786,8 @@ func (m *queryFieldMeta) SourceIndices() []int {
 	return res
 }
 
-func (m *queryFieldMeta) AllAllowedTypes() map[field.ValueType]struct{} {
-	res := make(map[field.ValueType]struct{})
+func (m *queryFieldMeta) AllAllowedTypes() field.ValueTypeSet {
+	res := make(field.ValueTypeSet)
 	for _, tm := range m.allowedTypesBySourceIdx {
 		for t := range tm {
 			res[t] = struct{}{}