1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
# Copyright (C) 2014-2017 MongoDB, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
module Mongo
class Collection
class View
# Provides behaviour around an aggregation pipeline on a collection view.
#
# @since 2.0.0
class Aggregation
extend Forwardable
include Enumerable
include Immutable
include Iterable
include Explainable
include Loggable
include Retryable
# @return [ View ] view The collection view.
attr_reader :view
# @return [ Array<Hash> ] pipeline The aggregation pipeline.
attr_reader :pipeline
# Delegate necessary operations to the view.
def_delegators :view, :collection, :read, :cluster
# Delegate necessary operations to the collection.
def_delegators :collection, :database, :client
# The reroute message.
#
# @since 2.1.0
REROUTE = 'Rerouting the Aggregation operation to the primary server.'.freeze
# Set to true if disk usage is allowed during the aggregation.
#
# @example Set disk usage flag.
# aggregation.allow_disk_use(true)
#
# @param [ true, false ] value The flag value.
#
# @return [ true, false, Aggregation ] The aggregation if a value was
# set or the value if used as a getter.
#
# @since 2.0.0
def allow_disk_use(value = nil)
configure(:allow_disk_use, value)
end
# Initialize the aggregation for the provided collection view, pipeline
# and options.
#
# @example Create the new aggregation view.
# Aggregation.view.new(view, pipeline)
#
# @param [ Collection::View ] view The collection view.
# @param [ Array<Hash> ] pipeline The pipeline of operations.
# @param [ Hash ] options The aggregation options.
#
# @since 2.0.0
def initialize(view, pipeline, options = {})
@view = view
@pipeline = pipeline.dup
@options = BSON::Document.new(options).freeze
end
# Get the explain plan for the aggregation.
#
# @example Get the explain plan for the aggregation.
# aggregation.explain
#
# @return [ Hash ] The explain plan.
#
# @since 2.0.0
def explain
self.class.new(view, pipeline, options.merge(explain: true)).first
end
private
def server_selector
@view.send(:server_selector)
end
def aggregate_spec(session)
Builder::Aggregation.new(pipeline, view, options.merge(session: session)).specification
end
def new(options)
Aggregation.new(view, pipeline, options)
end
def initial_query_op(session)
Operation::Commands::Aggregate.new(aggregate_spec(session))
end
def valid_server?(server)
server.standalone? || server.mongos? || server.primary? || secondary_ok?
end
def secondary_ok?
pipeline.none? { |op| op.key?('$out') || op.key?(:$out) }
end
def send_initial_query(server, session)
unless valid_server?(server)
log_warn(REROUTE)
server = cluster.next_primary(false)
end
validate_collation!(server)
initial_query_op(session).execute(server)
end
def validate_collation!(server)
if options[:collation] && !server.features.collation_enabled?
raise Error::UnsupportedCollation.new
end
end
end
end
end
end
|
