Class: PostgreSQLCursor::Cursor
- Inherits:
-
Object
- Object
- PostgreSQLCursor::Cursor
- Includes:
- Enumerable
- Defined in:
- lib/postgresql_cursor/cursor.rb
Instance Attribute Summary collapse
-
#connection ⇒ Object
readonly
Returns the value of attribute connection.
-
#count ⇒ Object
readonly
Returns the value of attribute count.
-
#options ⇒ Object
readonly
Returns the value of attribute options.
-
#result ⇒ Object
readonly
Returns the value of attribute result.
-
#sql ⇒ Object
readonly
Returns the value of attribute sql.
Instance Method Summary collapse
- #cast_types(row) ⇒ Object
-
#close ⇒ Object
Public: Closes the cursor.
- #column_types ⇒ Object
-
#each(&block) ⇒ Object
Public: Yields each row of the result set to the passed block.
- #each_array(&block) ⇒ Object
- #each_array_batch(&block) ⇒ Object
-
#each_batch(&block) ⇒ Object
:nodoc:.
- #each_instance(klass = nil, &block) ⇒ Object
- #each_instance_batch(klass = nil, &block) ⇒ Object
- #each_row(&block) ⇒ Object
- #each_row_batch(&block) ⇒ Object
-
#each_tuple(&block) ⇒ Object
:nodoc:.
-
#fetch(options = {}) ⇒ Object
Public: Returns the next row from the cursor, or empty hash if end of results.
-
#fetch_block(block_size = nil) ⇒ Object
Private: Fetches the next block of rows into @block.
-
#initialize(sql, options = {}) ⇒ Cursor
constructor
Public: Start a new PostgreSQL cursor query sql - The SQL statement with interpolated values options - hash of processing controls while: value - Exits loop when block does not return this value.
- #iterate_batched(batched = true) ⇒ Object
-
#iterate_type(type = nil) ⇒ Object
Specify the type to instantiate, or reset to return a Hash.
-
#open ⇒ Object
Public: Opens (actually, “declares”) the cursor.
-
#pluck(*cols) ⇒ Object
Returns an array of columns plucked from the result rows.
-
#set_cursor_tuple_fraction(frac = 1.0) ⇒ Object
Private: Sets the PostgreSQL cursor_tuple_fraction value = 1.0 to assume all rows will be fetched This is a value between 0.1 and 1.0 (PostgreSQL defaults to 0.1, this library defaults to 1.0) used to determine the expected fraction (percent) of result rows returned the the caller.
-
#with_optional_transaction ⇒ Object
Private: Open transaction unless with_hold option, specified.
Constructor Details
#initialize(sql, options = {}) ⇒ Cursor
Public: Start a new PostgreSQL cursor query sql - The SQL statement with interpolated values options - hash of processing controls
while: value - Exits loop when block does not return this value.
until: value - Exits loop when block returns this value.
fraction: 0.1..1.0 - The cursor_tuple_fraction (default 1.0)
block_size: 1..n - The number of rows to fetch per db block fetch
Defaults to 1000
with_hold - Allows the query to remain open across commit points.
Examples
PostgreSQLCursor::Cursor.new("select ....")
Returns the cursor object when called with new.
44 45 46 47 48 49 50 51 |
# File 'lib/postgresql_cursor/cursor.rb', line 44 def initialize(sql, ={}) @sql = sql @options = @connection = @options.fetch(:connection) { ::ActiveRecord::Base.connection } @count = 0 @iterate = [:instances] ? :each_instance : :each_row @batched = false end |
Instance Attribute Details
#connection ⇒ Object (readonly)
Returns the value of attribute connection.
27 28 29 |
# File 'lib/postgresql_cursor/cursor.rb', line 27 def connection @connection end |
#count ⇒ Object (readonly)
Returns the value of attribute count.
27 28 29 |
# File 'lib/postgresql_cursor/cursor.rb', line 27 def count @count end |
#options ⇒ Object (readonly)
Returns the value of attribute options.
27 28 29 |
# File 'lib/postgresql_cursor/cursor.rb', line 27 def @options end |
#result ⇒ Object (readonly)
Returns the value of attribute result.
27 28 29 |
# File 'lib/postgresql_cursor/cursor.rb', line 27 def result @result end |
#sql ⇒ Object (readonly)
Returns the value of attribute sql.
27 28 29 |
# File 'lib/postgresql_cursor/cursor.rb', line 27 def sql @sql end |
Instance Method Details
#cast_types(row) ⇒ Object
222 223 224 |
# File 'lib/postgresql_cursor/cursor.rb', line 222 def cast_types(row) row end |
#close ⇒ Object
Public: Closes the cursor
281 282 283 |
# File 'lib/postgresql_cursor/cursor.rb', line 281 def close @connection.execute("close #{@cursor}") end |
#column_types ⇒ Object
226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 |
# File 'lib/postgresql_cursor/cursor.rb', line 226 def column_types return nil if ::ActiveRecord::VERSION::MAJOR < 4 return @column_types if @column_types types = {} fields = @result.fields fields.each_with_index do |fname, i| ftype = @result.ftype i fmod = @result.fmod i types[fname] = @connection.get_type_map.fetch(ftype, fmod) do |oid, mod| warn "unknown OID: #{fname}(#{oid}) (#{sql})" if ::ActiveRecord::VERSION::MAJOR <= 4 ::ActiveRecord::ConnectionAdapters::PostgreSQLAdapter::OID::Identity.new else ::ActiveRecord::Type::Value.new end end end @column_types = types end |
#each(&block) ⇒ Object
Public: Yields each row of the result set to the passed block
Yields the row to the block. The row is a hash with symbolized keys.
{colname: value, ....}
Returns the count of rows processed
77 78 79 80 81 82 83 84 85 |
# File 'lib/postgresql_cursor/cursor.rb', line 77 def each(&block) if @iterate == :each_row @batched ? self.each_row_batch(&block) : self.each_row(&block) elsif @iterate == :each_array @batched ? self.each_array_batch(&block) : self.each_array(&block) else @batched ? self.each_instance_batch(@type, &block) : self.each_instance(@type, &block) end end |
#each_array(&block) ⇒ Object
94 95 96 97 98 99 100 101 102 103 104 105 |
# File 'lib/postgresql_cursor/cursor.rb', line 94 def each_array(&block) old_iterate = @iterate @iterate = :each_array begin rv = self.each_tuple do |row| block.call(row) end ensure @iterate = old_iterate end rv end |
#each_array_batch(&block) ⇒ Object
127 128 129 130 131 132 133 134 135 136 137 138 |
# File 'lib/postgresql_cursor/cursor.rb', line 127 def each_array_batch(&block) old_iterate = @iterate @iterate = :each_array begin rv = self.each_batch do |batch| block.call(batch) end ensure @iterate = old_iterate end rv end |
#each_batch(&block) ⇒ Object
:nodoc:
200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 |
# File 'lib/postgresql_cursor/cursor.rb', line 200 def each_batch(&block) #:nodoc: has_do_until = @options.key?(:until) has_do_while = @options.key?(:while) @count = 0 @column_types = nil with_optional_transaction do begin open while (batch = fetch_block) break if batch.empty? @count += 1 rc = block.call(batch) break if has_do_until && rc == @options[:until] break if has_do_while && rc != @options[:while] end ensure close if @block end end @count end |
#each_instance(klass = nil, &block) ⇒ Object
107 108 109 110 111 112 113 114 115 116 117 118 |
# File 'lib/postgresql_cursor/cursor.rb', line 107 def each_instance(klass=nil, &block) klass ||= @type self.each_tuple do |row| if ::ActiveRecord::VERSION::MAJOR < 4 model = klass.send(:instantiate,row) else @column_types ||= column_types model = klass.send(:instantiate, row, @column_types) end block.call(model) end end |
#each_instance_batch(klass = nil, &block) ⇒ Object
140 141 142 143 144 145 146 147 148 149 150 151 152 153 |
# File 'lib/postgresql_cursor/cursor.rb', line 140 def each_instance_batch(klass=nil, &block) klass ||= @type self.each_batch do |batch| models = batch.map do |row| if ::ActiveRecord::VERSION::MAJOR < 4 model = klass.send(:instantiate, row) else @column_types ||= column_types model = klass.send(:instantiate, row, @column_types) end end block.call(models) end end |
#each_row(&block) ⇒ Object
87 88 89 90 91 92 |
# File 'lib/postgresql_cursor/cursor.rb', line 87 def each_row(&block) self.each_tuple do |row| row = row.symbolize_keys if @options[:symbolize_keys] block.call(row) end end |
#each_row_batch(&block) ⇒ Object
120 121 122 123 124 125 |
# File 'lib/postgresql_cursor/cursor.rb', line 120 def each_row_batch(&block) self.each_batch do |batch| batch.map!(&:symbolize_keys) if @options[:symbolize_keys] block.call(batch) end end |
#each_tuple(&block) ⇒ Object
:nodoc:
176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 |
# File 'lib/postgresql_cursor/cursor.rb', line 176 def each_tuple(&block) #:nodoc: has_do_until = @options.has_key?(:until) has_do_while = @options.has_key?(:while) @count = 0 @column_types = nil with_optional_transaction do begin open while (row = fetch) do break if row.size==0 @count += 1 rc = block.call(row) break if has_do_until && rc == @options[:until] break if has_do_while && rc != @options[:while] end rescue Exception => e raise e ensure close if @block end end @count end |
#fetch(options = {}) ⇒ Object
Public: Returns the next row from the cursor, or empty hash if end of results
Returns a row as a hash of ‘colname’=>value,…
260 261 262 263 264 265 266 |
# File 'lib/postgresql_cursor/cursor.rb', line 260 def fetch(={}) open unless @block fetch_block if @block.size==0 row = @block.shift row = row.symbolize_keys if row && [:symbolize_keys] row end |
#fetch_block(block_size = nil) ⇒ Object
Private: Fetches the next block of rows into @block
269 270 271 272 273 274 275 276 277 278 |
# File 'lib/postgresql_cursor/cursor.rb', line 269 def fetch_block(block_size=nil) block_size ||= @block_size ||= @options.fetch(:block_size) { 1000 } @result = @connection.execute("fetch #{block_size} from #{@cursor}") if @iterate == :each_array @block = @result.each_row.collect {|row| row } else @block = @result.collect {|row| row } end end |
#iterate_batched(batched = true) ⇒ Object
66 67 68 69 |
# File 'lib/postgresql_cursor/cursor.rb', line 66 def iterate_batched(batched=true) @batched = batched self end |
#iterate_type(type = nil) ⇒ Object
Specify the type to instantiate, or reset to return a Hash
54 55 56 57 58 59 60 61 62 63 64 |
# File 'lib/postgresql_cursor/cursor.rb', line 54 def iterate_type(type=nil) if type.nil? || type == Hash @iterate = :each_row elsif type == Array @iterate = :each_array else @iterate = :each_instance @type = type end self end |
#open ⇒ Object
Public: Opens (actually, “declares”) the cursor. Call this before fetching
249 250 251 252 253 254 255 |
# File 'lib/postgresql_cursor/cursor.rb', line 249 def open set_cursor_tuple_fraction @cursor = @options[:cursor_name] || ("cursor_" + SecureRandom.uuid.gsub("-","")) hold = @options[:with_hold] ? 'with hold ' : '' @result = @connection.execute("declare #{@cursor} no scroll cursor #{hold}for #{@sql}") @block = [] end |
#pluck(*cols) ⇒ Object
Returns an array of columns plucked from the result rows. Experimental function, as this could still use too much memory and negate the purpose of this libarary. Should this return a lazy enumerator instead?
159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 |
# File 'lib/postgresql_cursor/cursor.rb', line 159 def pluck(*cols) = cols.last.is_a?(Hash) ? cols.pop : {} @options.merge!() @options[:symbolize_keys] = true self.iterate_type([:class]) if [:class] cols = cols.map {|c| c.to_sym } result = [] self.each() do |row| row = row.symbolize_keys if row.is_a?(Hash) result << cols.map { |c| row[c] } end result.flatten! if cols.size == 1 result end |
#set_cursor_tuple_fraction(frac = 1.0) ⇒ Object
Private: Sets the PostgreSQL cursor_tuple_fraction value = 1.0 to assume all rows will be fetched This is a value between 0.1 and 1.0 (PostgreSQL defaults to 0.1, this library defaults to 1.0) used to determine the expected fraction (percent) of result rows returned the the caller. This value determines the access path by the query planner.
298 299 300 301 302 303 304 |
# File 'lib/postgresql_cursor/cursor.rb', line 298 def set_cursor_tuple_fraction(frac=1.0) @cursor_tuple_fraction ||= @options.fetch(:fraction) { 1.0 } return @cursor_tuple_fraction if frac == @cursor_tuple_fraction @cursor_tuple_fraction = frac @result = @connection.execute("set cursor_tuple_fraction to #{frac}") frac end |
#with_optional_transaction ⇒ Object
Private: Open transaction unless with_hold option, specified
286 287 288 289 290 291 292 |
# File 'lib/postgresql_cursor/cursor.rb', line 286 def with_optional_transaction if @options[:with_hold] yield else @connection.transaction { yield } end end |