Optimize Role model/dataset (#3068)

The Role model is not backed by a dedicated table, but a dataset that is
a UNION of eight individual roles tables. When filtering for roles, this
was done by adding WHERE conditions to the overall dataset, which
resulted in slow query executions. A more efficient approach is to move
the WHERE conditions into the individual datasets per role.

This commit customizes the Role model/dataset. The creation of the
UNIONed dataset is moved into a dedicated class (RoleDataset) that
supports building a dataset with filters (i.e. WHERE conditions) per
role (i.e. individual table). The 'where' method of the Role dataset is
overridden and uses Dataset.from ([1]) to change the dataset's source.
It replaces the source with a new dataset built with filters per role.

As 'where' calls can be chained, the filters need to be kept in the
dataset instance (cache) and the source dataset needs to be built again
on each 'where' invocation with all the filters, i.e. previous and new
WHERE conditions.

Furthermore fully qualified identifiers (table name and column name)
have to be adapted; as the WHERE conditions are applied on the
individual roles tables, no table name prefix is required and thus
removed from the identifiers.

Some filters can now be handled in a more efficient way. When filtering
for a specific role type, the corresponding dataset gets a
'WHERE 1 = 1', whereas all the other datasets get a 'WHERE 1 = 0',
meaning they never return any rows. Something similar is done when
filtering for organizations or spaces; the opposite role types get a
'WHERE 1 = 0' as they don't have the requested 'id' field.

As there might be filters that still should be applied to the overall
dataset, 'where' invocations with a block (aka. virtual row block) are
treated as before, i.e. applied at the end.

Drawbacks:
- The current implementation of the custom Role model/dataset is rather
  strict with regards to allowed filter expressions. This means that if
  some functionality is added (e.g. new or different filter), the custom
  Role model/dataset needs to be adapted. This was done by intention to
  ensure that every code path is tested.
- To limit the number of different filter expressions, invocations of
  'first(cond)' and 'find(cond)' on the custom Role model/dataset have
  been replaced with 'where(cond).first' which is semantically
  identical.

[1] https://sequel.jeremyevans.net/rdoc/classes/Sequel/Dataset.html#method-i-from

app/controllers/v3/roles_controller.rb

@@ -52,14 +52,14 @@ def show
     decorators << IncludeRoleOrganizationDecorator if IncludeRoleOrganizationDecorator.match?(message.include)
     decorators << IncludeRoleSpaceDecorator if IncludeRoleSpaceDecorator.match?(message.include)
 
-    role = readable_roles.first(guid: hashed_params[:guid])
+    role = readable_roles.where(guid: hashed_params[:guid]).first
     resource_not_found!(:role) unless role
 
     render status: :ok, json: Presenters::V3::RolePresenter.new(role, decorators: decorators)
   end
 
   def destroy
-    role = readable_roles.first(guid: hashed_params[:guid])
+    role = readable_roles.where(guid: hashed_params[:guid]).first
     resource_not_found!(:role) unless role
 
     if role.for_space?
@@ -145,7 +145,11 @@ def readable_roles
     if permission_queryer.can_read_globally?
       Role
     else
-      Role.where(space_id: visible_space_ids).or(organization_id: visible_org_ids)
+      # This rather complex filter should be applied to the overall Role dataset and is thus using the (virtual row)
+      # block syntax.
+      readable_by_space = { space_id: visible_space_ids }
+      readable_by_org = { organization_id: visible_org_ids }
+      Role.where { Sequel.|(readable_by_space, readable_by_org) }
     end
   end
 

app/fetchers/base_list_fetcher.rb

@@ -23,16 +23,16 @@ def advanced_filtering(message, dataset, klass)
             values.map do |operator, given_timestamp|
               if operator == RelationalOperators::LESS_THAN_COMPARATOR
                 normalized_timestamp = Time.parse(given_timestamp).utc
-                dataset = dataset.where { Sequel.qualify(klass.table_name, filter) < normalized_timestamp }
+                dataset = dataset.where(Sequel.qualify(klass.table_name, filter) < normalized_timestamp)
               elsif operator == RelationalOperators::LESS_THAN_OR_EQUAL_COMPARATOR
                 normalized_timestamp = (Time.parse(given_timestamp).utc + 0.999999).utc
-                dataset = dataset.where { Sequel.qualify(klass.table_name, filter) <= normalized_timestamp }
+                dataset = dataset.where(Sequel.qualify(klass.table_name, filter) <= normalized_timestamp)
               elsif operator == RelationalOperators::GREATER_THAN_COMPARATOR
                 normalized_timestamp = (Time.parse(given_timestamp).utc + 0.999999).utc
-                dataset = dataset.where { Sequel.qualify(klass.table_name, filter) > normalized_timestamp }
+                dataset = dataset.where(Sequel.qualify(klass.table_name, filter) > normalized_timestamp)
               elsif operator == RelationalOperators::GREATER_THAN_OR_EQUAL_COMPARATOR
                 normalized_timestamp = Time.parse(given_timestamp).utc
-                dataset = dataset.where { Sequel.qualify(klass.table_name, filter) >= normalized_timestamp }
+                dataset = dataset.where(Sequel.qualify(klass.table_name, filter) >= normalized_timestamp)
               end
             end
           else

app/fetchers/role_list_fetcher.rb

@@ -5,8 +5,8 @@
 module VCAP::CloudController
   class RoleListFetcher < BaseListFetcher
     class << self
-      def fetch(message, readable_users_dataset, eager_loaded_associations: [])
-        filter(message, readable_users_dataset).eager(eager_loaded_associations)
+      def fetch(message, readable_roles, eager_loaded_associations: [])
+        filter(message, readable_roles).eager(eager_loaded_associations)
       end
 
       private

app/models/runtime/pollable_job_model.rb

@@ -32,7 +32,7 @@ def resource_exists?
                 Sequel::Model(ActiveSupport::Inflector.pluralize(resource_type).to_sym)
               end
 
-      !!model.find(guid: resource_guid)
+      !!model.where(guid: resource_guid).first
     end
 
     def self.find_by_delayed_job(delayed_job)