func shouldSkipLiveBatchWithFilter()

in query/aql_processor.go [1455:1526]
63 lines of code
22 McCabe index (conditional complexity)

func shouldSkipLiveBatchWithFilter(b *memstore.LiveBatch, filter expr.Expr) bool {
	if filter == nil {
		return false
	}

	if binExpr, ok := filter.(*expr.BinaryExpr); ok {
		var columnExpr *expr.VarRef
		var numExpr *expr.NumberLiteral
		op := binExpr.Op
		switch op {
		case expr.GTE, expr.GT, expr.LT, expr.LTE, expr.EQ:
		default:
			return false
		}
		// First try lhs VarRef, rhs Num.
		lhsVarRef, lhsOK := binExpr.LHS.(*expr.VarRef)
		rhsNum, rhsOK := binExpr.RHS.(*expr.NumberLiteral)
		if lhsOK && rhsOK {
			columnExpr = lhsVarRef
			numExpr = rhsNum
		} else {
			// Then try rhs VarRef, lhs Num.
			lhsNum, lhsOK := binExpr.LHS.(*expr.NumberLiteral)
			rhsVarRef, rhsOK := binExpr.RHS.(*expr.VarRef)
			if lhsOK && rhsOK {
				// Swap column to the left and number to right.
				columnExpr = rhsVarRef
				numExpr = lhsNum
				// Invert the OP.
				switch op {
				case expr.GTE:
					op = expr.LTE
				case expr.GT:
					op = expr.LT
				case expr.LTE:
					op = expr.GTE
				case expr.LT:
					op = expr.GT
				}
			}
		}

		if columnExpr != nil && numExpr != nil {
			// Time filters and main table filters are guaranteed to be on main table.
			vp := b.GetVectorParty(columnExpr.ColumnID)
			if vp == nil {
				return true
			}

			if columnExpr.DataType != memCom.Uint32 {
				return false
			}

			num := int64(numExpr.Int)
			minUint32, maxUint32 := vp.(memCom.LiveVectorParty).GetMinMaxValue()
			min, max := int64(minUint32), int64(maxUint32)
			switch op {
			case expr.GTE:
				return max < num
			case expr.GT:
				return max <= num
			case expr.LTE:
				return min > num
			case expr.LT:
				return min >= num
			case expr.EQ:
				return min > num || max < num
			}
		}
	}
	return false
}