Class: HBase::Scoped

Inherits:
Object
  • Object
show all
Includes:
Enumerable, Util, Scoped::Aggregation
Defined in:
lib/hbase-jruby/scoped.rb,
lib/hbase-jruby/scoped/aggregation.rb

Overview

Scope of data access

Author:

Defined Under Namespace

Modules: Aggregation

Constant Summary

Constants included from Util

Util::JAVA_BYTE_ARRAY_CLASS, Util::JAVA_BYTE_ARRAY_EMPTY

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Util

append_0, from_bytes, java_bytes?, parse_column_name, to_bytes, to_typed_bytes

Instance Attribute Details

#table ⇒ HBase::Table (readonly)

Returns HBase::Table instance for this scope.

Returns:



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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
 
# File 'lib/hbase-jruby/scoped.rb', line 6

class Scoped
  include Enumerable
  include Scoped::Aggregation
  include HBase::Util

  attr_reader :table

  # A clean HBase::Scoped object for the same table
  # @return [HBase::Scope] A clean HBase::Scoped object for the same table
  def unscope
    Scoped.send(:new, @table, @dcaching)
  end

  # Count the number of rows in the scope
  # @return [Fixnum, Bignum] The number of rows in the scope
  # @param [Hash] options Counting options
  # @option options [Fixnum|nil] :caching The number of rows for caching that will be passed to scanners.
  #   Use higher values for faster scan.
  # @option options [Boolean] :cache_blocks Whether blocks should be cached for this scan
  def count options = {}
    options = { :caching      => nil,
                :cache_blocks => true }.merge(options)

    scan = block_given? ? filtered_scan : filtered_scan_minimum
    scan.cache_blocks = options[:cache_blocks]
    if options[:caching] && (@limit.nil? || options[:caching] < @limit)
      scan.caching = options[:caching]
    end

    cnt = 0
    if block_given?
      iterate(scan) do |result|
        cnt += 1 if yield(Row.send(:new, @table, result))
      end
    else
      iterate(scan) { |r| cnt += 1 }
    end
    cnt
  end

  # Performs GET operations
  # @overload get(rowkey)
  #   Single GET.
  #   Gets a record with the given rowkey. If the record is not found, nil is returned.
  #   @param [Object] rowkey Rowkey
  #   @return [HBase::Row, nil]
  # @overload get(rowkeys)
  #   Batch GET. Gets an array of records with the given rowkeys.
  #   Nonexistent records will be returned as nils.
  #   @param [Array<Object>] *rowkeys Rowkeys
  #   @return [Array<HBase::Row>]
  def get rowkeys
    case rowkeys
    when Array
      htable.get(rowkeys.map { |rk| getify rk }).map { |result|
        result.isEmpty ? nil : Row.send(:new, @table, result)
      }
    else
      result = htable.get(getify rowkeys)
      result.isEmpty ? nil : Row.send(:new, @table, result)
    end
  end

  # Iterate through the scope.
  # @yield [row] Yields each row in the scope
  # @yieldparam [HBase::Row] row
  def each
    return enum_for(:each) unless block_given?

    iterate(filtered_scan) do |result|
      yield Row.send(:new, @table, result)
    end
  end

  # Sets the number of rows for caching that will be passed to scanners.
  # @param [Fixnum] rows The number of rows to cache
  # @return [HBase::Scoped] HBase::Scoped object with the caching option
  def caching rows
    raise ArgumentError, "Invalid caching size. Must be a non-negative integer." unless rows.is_a?(Fixnum) && rows >= 0
    spawn :@caching, rows
  end

  # @overload range(start_key, opts = {})
  #   Returns an HBase::Scoped object with the specified rowkey range
  #   Overrides current range.
  #   @param [Object] start_key Start rowkey
  #   @param [Hash] opts Prefix filter
  #   @option opts [Object, Array<Object>] :prefix Only rows matching any of the given prefixes are returned
  #   @return [HBase::Scoped] HBase::Scoped object with the range
  # @overload range(start_key, stop_key, opts = {})
  #   Returns an HBase::Scoped object with the specified rowkey range
  #   Overrides current range.
  #   @param [Object, nil] start_key Start rowkey. Can be nil.
  #   @param [Object] stop_key Stop rowkey (exclusive)
  #   @param [Hash] opts Prefix filter
  #   @option opts [Object, Array<Object>] :prefix Only rows matching any of the given prefixes are returned
  #   @return [HBase::Scoped] HBase::Scoped object with the range
  # @overload range(start_stop_range, opts = {})
  #   Returns an HBase::Scoped object with the specified rowkey range
  #   Overrides current range.
  #   @param [Range] start_stop_range Rowkey scan range
  #   @param [Hash] opts Prefix filter
  #   @option opts [Object, Array<Object>] :prefix Only rows matching any of the given prefixes are returned
  #   @return [HBase::Scoped] HBase::Scoped object with the range
  # @overload range(opts)
  #   Returns an HBase::Scoped object with the specified rowkey range
  #   Overrides current range.
  #   @param [Hash] opts Prefix filter
  #   @option opts [Object, Array<Object>] :prefix Only rows matching any of the given prefixes are returned
  #   @return [HBase::Scoped] HBase::Scoped object with the range
  #   @example
  #     table.range(:prefix => '2012')
  #     table.range(:prefix => ['2010', '2012'])
  def range *key_range
    if (last = key_range.last).is_a?(Hash)
      prefixes = arrayfy(last[:prefix]).compact
      last = last.reject { |k, v| k == :prefix }

      key_range = key_range[0...-1] # defensive
      key_range << last unless last.empty?
    else
      prefixes = []
    end

    if key_range[0].is_a?(Range)
      raise ArgumentError, "Invalid range" unless key_range.length == 1
    elsif !prefixes.empty?
      raise ArgumentError, "Invalid range" unless [0, 1, 2].include?(key_range.length)
    else
      raise ArgumentError, "Invalid range" unless [1, 2].include?(key_range.length)
    end

    raise ArgumentError, "Invalid range" if !key_range.empty? && key_range.all? { |e| e.nil? }

    spawn :@range,
          key_range[0].is_a?(Range) ?
              key_range[0] :
              (key_range.empty? ? nil : key_range.map { |e| e.nil? ? nil : Util.to_bytes(e) }),
          :@prefixes,
          prefixes
  end

  # Returns an HBase::Scoped object with the filters added
  # @param [Array<Hash, FilterBase, FilterList>] filters
  # @return [HBase::Scoped] HBase::Scoped object also with the specified filters
  def filter *filters
    spawn :@filters, @filters + parse_filter_input(filters)
  end

  # Returns an HBase::Scoped object with the additional filters which will cause early termination of scan
  # @param [Array<Hash, FilterBase, FilterList>] filters
  # @return [HBase::Scoped]HBase::Scoped object with the additional filters which will cause early termination of scan
  def while *filters
    spawn :@filters, @filters + parse_filter_input(filters).map { |filter| WhileMatchFilter.new(filter) }
  end

  # Returns an HBase::Scoped object with the specified row number limit
  # @param [Fixnum|nil] rows Sets the maximum number of rows to return from scan
  # @return [HBase::Scoped] HBase::Scoped object with the specified row number limit
  def limit rows
    unless (rows.is_a?(Fixnum) && rows >= 0) || rows.nil?
      raise ArgumentError, "Invalid limit. Must be a non-negative integer or nil."
    end
    spawn :@limit, rows
  end

  # Returns an HBase::Scoped object with the specified time range
  # @param [Fixnum|Time] min Minimum timestamp (inclusive)
  # @param [Fixnum|Time] max Maximum timestamp (exclusive)
  # @return [HBase::Scoped] HBase::Scoped object with the specified time range
  def time_range min, max
    spawn :@trange, [min, max].map { |e| time_to_long e }
  end

  # Returns an HBase::Scoped object with the specified timestamp
  # @param [Fixnum|Time] ts Timestamp
  # @return [HBase::Scoped] HBase::Scoped object with the specified timestamp
  def at ts
    spawn :@trange, time_to_long(ts)
  end

  # Returns an HBase::Scoped object with the specified projection
  # @param [Array<String>] columns Array of column expressions
  # @return [HBase::Scoped] HBase::Scoped object with the specified projection
  def project *columns
    if columns.first.is_a?(Hash)
      hash = columns.first
      unless (hash.keys - [:prefix, :range, :limit, :offset]).empty?
        raise ArgumentError, "Invalid projection"
      end

      if l = hash[:limit]
        unless l.is_a?(Fixnum) && l >= 0
          raise ArgumentError, ":limit must be a non-negative integer"
        end
      end

      if o = hash[:offset]
        unless o.is_a?(Fixnum) && o >= 0
          raise ArgumentError, ":offset must be a non-negative integer"
        end
      end
    end
    spawn :@project, @project + columns.map { |c|
      cf, cq, type = @table.lookup_schema(c)
      cf ? [cf, cq] : c
    }
  end

  # Returns an HBase::Scoped object with the specified version number limit.
  # If not set, all versions of each value are fetched by default.
  # @param [Fixnum] vs Sets the maximum number of versions
  # @return [HBase::Scoped] HBase::Scoped object with the version number limit
  def versions vs
    raise ArgumentError, "Invalid versions. Must be a positive integer." unless vs.is_a?(Fixnum) && vs > 0
    spawn :@versions, vs
  end

  # Returns an HBase::Scoped object with the specified batch limit
  # @param [Fixnum] b Sets the maximum number of values to fetch each time
  # @return [HBase::Scoped] HBase::Scoped object with the specified batch limit
  def batch b
    raise ArgumentError, "Invalid batch size. Must be a positive integer." unless b.is_a?(Fixnum) && b > 0
    spawn :@batch, b
  end

  # Returns an HBase::Scoped object with the Scan-customization block added.
  # The given block will be evaluated just before an actual scan operation.
  # With method-chaining, multiple blocks can be registered to be evaluated sequentially.
  # @return [HBase::Scoped]
  # @yield [org.apache.hadoop.hbase.client.Scan]
  def with_java_scan &block
    raise ArgumentError, "Block not given" if block.nil?
    raise ArgumentError, "Invalid arity: should be 1" unless block.arity == 1
    spawn :@scan_cbs, @scan_cbs + [block]
  end

  # Returns an HBase::Scoped object with the Get-customization block added
  # The given block will be evaluated just before an actual get operation.
  # With method-chaining, multiple blocks can be registered to be evaluated sequentially.
  # @return [HBase::Scoped]
  # @yield [org.apache.hadoop.hbase.client.Get]
  def with_java_get &block
    raise ArgumentError, "Block not given" if block.nil?
    raise ArgumentError, "Invalid arity: should be 1" unless block.arity == 1
    spawn :@get_cbs, @get_cbs + [block]
  end

private
  # @param [HBase::Table] table
  def initialize table, default_caching
    @table    = table
    @filters  = []
    @project  = []
    @prefixes = []
    @range    = nil
    @versions = nil
    @batch    = nil
    @dcaching = default_caching
    @caching  = nil
    @limit    = nil
    @trange   = nil
    @scan_cbs = []
    @get_cbs  = []
  end

  def spawn *args
    self.dup.tap do |obj|
      args.each_slice(2) do |slice|
        attr, val = slice
        obj.instance_variable_set attr, val
      end
    end
  end

  def htable
    @table.htable
  end

  def process_projection! obj
    limit   = offset = nil
    ranges  = prefixes = []
    filters = []

    @project.each do |col|
      case col
      when Hash
        col.each do |prop, val|
          case prop
          when :prefix
            prefixes += arrayfy(val)
          when :range
            ranges += arrayfy(val)
          when :limit
            limit = val
          when :offset
            offset = val
          else
            # Shouldn't happen
            raise ArgumentError, "Invalid projection: #{prop}"
          end
        end
      else
        cf, cq = Util.parse_column_name col
        if cq
          obj.addColumn cf, cq
        else
          obj.addFamily cf
        end
      end
    end

    if (limit && !offset) || (!limit && offset)
      raise ArgumentError, "Both `limit` and `offset` must be specified"
    end

    # Column prefix filter
    unless prefixes.empty?
      # disjunctive
      filters <<
        MultipleColumnPrefixFilter.new(
          prefixes.map { |pref| Util.to_bytes(pref).to_a }.to_java(Java::byte[]))
    end

    # Column range filter
    unless ranges.empty?
      # disjunctive
      filters <<
        FilterList.new(FilterList::Operator::MUST_PASS_ONE,
          ranges.map { |range|
            raise ArgumentError, "Invalid range type" unless range.is_a? Range

            ColumnRangeFilter.new(
              Util.to_bytes(range.begin), true,
              Util.to_bytes(range.end), !range.exclude_end?) })
    end

    # Column pagniation filter (last)
    if limit && offset
      filters << ColumnPaginationFilter.new(limit, offset)
    end

    filters
  end

  def getify rowkey
    Get.new(Util.to_bytes rowkey).tap { |get|
      if @versions
        get.setMaxVersions @versions
      else
        get.setMaxVersions
      end

      filters = []
      filters += process_projection!(get)

      range = @range || range_for_prefix
      case range
      when Range
        filters <<
          RowFilter.new(
            CompareFilter::CompareOp::GREATER_OR_EQUAL,
            BinaryComparator.new(Util.to_bytes range.begin))

        filters <<
          RowFilter.new(
            (range.exclude_end? ?
              CompareFilter::CompareOp::LESS :
              CompareFilter::CompareOp::LESS_OR_EQUAL),
            BinaryComparator.new(Util.to_bytes range.end))
      when Array
        filters <<
          RowFilter.new(
            CompareFilter::CompareOp::GREATER_OR_EQUAL,
            BinaryComparator.new(range[0])) if range[0]

        filters <<
          RowFilter.new(
            CompareFilter::CompareOp::LESS,
            BinaryComparator.new(range[1])) if range[1]
      else
        raise ArgumentError, "Invalid range"
      end if range

      # Prefix filters
      filters += [*build_prefix_filter].compact

      # RowFilter must precede the others
      filters += @filters

      get.setFilter FilterList.new(filters) unless filters.empty?

      # Timerange / Timestamp
      case @trange
      when Array
        get.setTimeRange(*@trange)
      when Time, Fixnum
        get.setTimeStamp @trange
      end

      # Customization
      @get_cbs.each do |prc|
        prc.call get
      end
    }
  end

  def filter_for cf, cq, type, val
    case val
    when Range
      min, max = [val.begin, val.end].map { |k| Util.to_typed_bytes type, k }
      FilterList.new(FilterList::Operator::MUST_PASS_ALL, [
        SingleColumnValueFilter.new(
          cf, cq,
          CompareFilter::CompareOp::GREATER_OR_EQUAL, min
        ).tap { |f| f.setFilterIfMissing(true) },
        SingleColumnValueFilter.new(
          cf, cq,
          (val.exclude_end? ? CompareFilter::CompareOp::LESS :
                              CompareFilter::CompareOp::LESS_OR_EQUAL), max
        ).tap { |f| f.setFilterIfMissing(true) }
      ])
    when Hash
      FilterList.new(FilterList::Operator::MUST_PASS_ALL,
        val.map { |op, v|
          operator =
            case op
            when :gt, :>
              CompareFilter::CompareOp::GREATER
            when :gte, :>=
              CompareFilter::CompareOp::GREATER_OR_EQUAL
            when :lt, :<
              CompareFilter::CompareOp::LESS
            when :lte, :<=
              CompareFilter::CompareOp::LESS_OR_EQUAL
            when :eq, :==
              CompareFilter::CompareOp::EQUAL
            when :ne # , :!= # Ruby 1.8 compatibility
              CompareFilter::CompareOp::NOT_EQUAL
            else
              if val.length == 1
                return filter_for(cf, cq, nil, Util.to_typed_bytes(type, val))
              else
                raise ArgumentError, "Unknown operator: #{op}"
              end
            end
          case v
          when Array
            # XXX TODO Undocumented feature
            FilterList.new(
              case op
              when :ne # , :!=
                FilterList::Operator::MUST_PASS_ALL
              else
                FilterList::Operator::MUST_PASS_ONE
              end,
              v.map { |vv|
                SingleColumnValueFilter.new(cf, cq, operator, Util.to_typed_bytes(type, vv)).tap { |f|
                  f.setFilterIfMissing( op != :ne )
                }
              }
            )
          else
            SingleColumnValueFilter.new(cf, cq, operator, Util.to_typed_bytes(type, v)).tap { |f|
              f.setFilterIfMissing( op != :ne )
            }
          end
        }
      )
    when Regexp
      SingleColumnValueFilter.new(
        cf, cq,
        CompareFilter::CompareOp::EQUAL,
        RegexStringComparator.new(val.to_s)
      ).tap { |f| f.setFilterIfMissing(true) }
    when nil
      # - has value < '' -> not ok
      # - no value       -> ok
      SingleColumnValueFilter.new(
        cf, cq,
        CompareFilter::CompareOp::LESS,
        HBase::Util::JAVA_BYTE_ARRAY_EMPTY
      )
    else
      SingleColumnValueFilter.new(
        cf, cq,
        CompareFilter::CompareOp::EQUAL,
        Util.to_typed_bytes(type, val)
      ).tap { |f| f.setFilterIfMissing(true) }
    end
  end

  def filtered_scan
    Scan.new.tap { |scan|
      # Range
      range = @range || range_for_prefix
      case range
      when Range
        scan.setStartRow Util.to_bytes range.begin

        if range.exclude_end?
          scan.setStopRow Util.to_bytes range.end
        else
          scan.setStopRow Util.append_0(Util.to_bytes range.end)
        end
      when Array
        scan.setStartRow range[0] if range[0]
        scan.setStopRow  range[1] if range[1]
      else
        # This shouldn't happen though.
        raise ArgumentError, "Invalid range"
      end if range

      # Caching
      scan.caching = @caching if @caching

      # Filters (with projection)
      prefix_filter = [*build_prefix_filter].compact
      filters = prefix_filter + @filters
      filters += process_projection!(scan)

      scan.setFilter FilterList.new(filters) unless filters.empty?

      # Limit
      if @limit
        if [@caching, @dcaching].compact.all? { |c| @limit < c }
          scan.caching = @limit
        end
      end

      # Versions
      if @versions
        scan.setMaxVersions @versions
      else
        scan.setMaxVersions
      end

      # Timerange / Timestamp
      case @trange
      when Array
        scan.setTimeRange(*@trange)
      when Time, Fixnum
        scan.setTimeStamp @trange
      end

      # Batch
      scan.setBatch @batch if @batch

      # Customization
      @scan_cbs.each do |prc|
        prc.call scan
      end
    }
  end

  # Scanner for just counting records
  # @private
  def filtered_scan_minimum
    filtered_scan.tap do |scan|
      scan.setMaxVersions 1

      # FirstKeyOnlyFilter: A filter that will only return the first KV from each row-
      # - Not compatible with SingleColumnValueFilter
      # KeyOnlyFilter: A filter that will only return the key component of each KV
      # - Compatible with SingleColumnValueFilter
      ko = KeyOnlyFilter.new
      if flist = scan.getFilter
        if flist.is_a?(FilterList)
          flist.addFilter ko
        else
          flist = FilterList.new([flist, ko])
        end
      else
        flist = FilterList.new([ko, FirstKeyOnlyFilter.new])
      end
      scan.setFilter flist
    end
  end

  def build_prefix_filter
    return nil if @prefixes.empty?

    filters = @prefixes.map { |prefix|
      PrefixFilter.new(Util.to_bytes prefix)
    }

    if filters.length == 1
      filters.first
    else
      FilterList.new FilterList::Operator::MUST_PASS_ONE, filters
    end
  end

  def range_for_prefix
    return nil if @prefixes.empty?

    [@prefixes.map { |pref| ByteArray.new(pref) }.min.java, nil]
  end

  def parse_filter_input filters
    filters.map { |f|
      case f
      when Hash
        f.map { |col, val|
          cf, cq, type = @table.lookup_and_parse col, true

          case val
          when Array
            FilterList.new(FilterList::Operator::MUST_PASS_ONE,
              val.map { |v| filter_for cf, cq, type, v })
          else
            filter_for cf, cq, type, val
          end
        }.flatten
      when FilterBase, FilterList
        f
      else
        raise ArgumentError, "Unknown filter type"
      end
    }.flatten
  end

  def arrayfy val
    # No range splat
    if Util.java_bytes?(val)
      [val]
    elsif val.is_a?(Array)
      val
    else
      [val]
    end
  end

  def check_closed
    raise RuntimeError, "HBase connection is already closed" if @table.closed?
  end

  def iterate scan
    scanner = htable.getScanner(scan)
    if @limit
      scanner.each_with_index do |result, idx|
        yield result
        break if idx == @limit - 1
      end
    else
      scanner.each do |result|
        yield result
      end
    end
  ensure
    scanner.close if scanner
  end
end

Instance Method Details

#at(ts) ⇒ HBase::Scoped

Returns an HBase::Scoped object with the specified timestamp

Parameters:

  • ts (Fixnum|Time) —

    Timestamp

Returns:

  • (HBase::Scoped) —

    HBase::Scoped object with the specified timestamp



183
184
185
 
# File 'lib/hbase-jruby/scoped.rb', line 183

def at ts
  spawn :@trange, time_to_long(ts)
end

#batch(b) ⇒ HBase::Scoped

Returns an HBase::Scoped object with the specified batch limit

Parameters:

  • b (Fixnum) —

    Sets the maximum number of values to fetch each time

Returns:

  • (HBase::Scoped) —

    HBase::Scoped object with the specified batch limit

Raises:

  • (ArgumentError) 


227
228
229
230
 
# File 'lib/hbase-jruby/scoped.rb', line 227

def batch b
  raise ArgumentError, "Invalid batch size. Must be a positive integer." unless b.is_a?(Fixnum) && b > 0
  spawn :@batch, b
end

#caching(rows) ⇒ HBase::Scoped

Sets the number of rows for caching that will be passed to scanners.

Parameters:

  • rows (Fixnum) —

    The number of rows to cache

Returns:

  • (HBase::Scoped) —

    HBase::Scoped object with the caching option

Raises:

  • (ArgumentError) 


83
84
85
86
 
# File 'lib/hbase-jruby/scoped.rb', line 83

def caching rows
  raise ArgumentError, "Invalid caching size. Must be a non-negative integer." unless rows.is_a?(Fixnum) && rows >= 0
  spawn :@caching, rows
end

#count(options = {}) ⇒ Fixnum, Bignum

Count the number of rows in the scope

Parameters:

  • options (Hash) (defaults to: {}) —

    Counting options

Options Hash (options):

  • :caching (Fixnum|nil) —

    The number of rows for caching that will be passed to scanners. Use higher values for faster scan.

  • :cache_blocks (Boolean) —

    Whether blocks should be cached for this scan

Returns:

  • (Fixnum, Bignum) —

    The number of rows in the scope



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/hbase-jruby/scoped.rb', line 25

def count options = {}
  options = { :caching      => nil,
              :cache_blocks => true }.merge(options)

  scan = block_given? ? filtered_scan : filtered_scan_minimum
  scan.cache_blocks = options[:cache_blocks]
  if options[:caching] && (@limit.nil? || options[:caching] < @limit)
    scan.caching = options[:caching]
  end

  cnt = 0
  if block_given?
    iterate(scan) do |result|
      cnt += 1 if yield(Row.send(:new, @table, result))
    end
  else
    iterate(scan) { |r| cnt += 1 }
  end
  cnt
end

#each {|row| ... } ⇒ Object

Iterate through the scope.

Yields:

  • (row) —

    Yields each row in the scope

Yield Parameters:



72
73
74
75
76
77
78
 
# File 'lib/hbase-jruby/scoped.rb', line 72

def each
  return enum_for(:each) unless block_given?

  iterate(filtered_scan) do |result|
    yield Row.send(:new, @table, result)
  end
end

#filter(*filters) ⇒ HBase::Scoped

Returns an HBase::Scoped object with the filters added

Parameters:

  • filters (Array<Hash, FilterBase, FilterList>)

Returns:

  • (HBase::Scoped) —

    HBase::Scoped object also with the specified filters



151
152
153
 
# File 'lib/hbase-jruby/scoped.rb', line 151

def filter *filters
  spawn :@filters, @filters + parse_filter_input(filters)
end

#get(rowkey) ⇒ HBase::Row? #get(rowkeys) ⇒ Array<HBase::Row>

Performs GET operations

Overloads:

  • #get(rowkey) ⇒ HBase::Row?

    Single GET. Gets a record with the given rowkey. If the record is not found, nil is returned.

    Parameters:

    • rowkey (Object) —

      Rowkey

    Returns:

  • #get(rowkeys) ⇒ Array<HBase::Row>

    Batch GET. Gets an array of records with the given rowkeys. Nonexistent records will be returned as nils.

    Parameters:

    • *rowkeys (Array<Object>) —

      Rowkeys

    Returns:



57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/hbase-jruby/scoped.rb', line 57

def get rowkeys
  case rowkeys
  when Array
    htable.get(rowkeys.map { |rk| getify rk }).map { |result|
      result.isEmpty ? nil : Row.send(:new, @table, result)
    }
  else
    result = htable.get(getify rowkeys)
    result.isEmpty ? nil : Row.send(:new, @table, result)
  end
end

#limit(rows) ⇒ HBase::Scoped

Returns an HBase::Scoped object with the specified row number limit

Parameters:

  • rows (Fixnum|nil) —

    Sets the maximum number of rows to return from scan

Returns:

  • (HBase::Scoped) —

    HBase::Scoped object with the specified row number limit



165
166
167
168
169
170
 
# File 'lib/hbase-jruby/scoped.rb', line 165

def limit rows
  unless (rows.is_a?(Fixnum) && rows >= 0) || rows.nil?
    raise ArgumentError, "Invalid limit. Must be a non-negative integer or nil."
  end
  spawn :@limit, rows
end

#project(*columns) ⇒ HBase::Scoped

Returns an HBase::Scoped object with the specified projection

Parameters:

  • columns (Array<String>) —

    Array of column expressions

Returns:

  • (HBase::Scoped) —

    HBase::Scoped object with the specified projection



190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/hbase-jruby/scoped.rb', line 190

def project *columns
  if columns.first.is_a?(Hash)
    hash = columns.first
    unless (hash.keys - [:prefix, :range, :limit, :offset]).empty?
      raise ArgumentError, "Invalid projection"
    end

    if l = hash[:limit]
      unless l.is_a?(Fixnum) && l >= 0
        raise ArgumentError, ":limit must be a non-negative integer"
      end
    end

    if o = hash[:offset]
      unless o.is_a?(Fixnum) && o >= 0
        raise ArgumentError, ":offset must be a non-negative integer"
      end
    end
  end
  spawn :@project, @project + columns.map { |c|
    cf, cq, type = @table.lookup_schema(c)
    cf ? [cf, cq] : c
  }
end

#range(start_key, opts = {}) ⇒ HBase::Scoped #range(start_key, stop_key, opts = {}) ⇒ HBase::Scoped #range(start_stop_range, opts = {}) ⇒ HBase::Scoped #range(opts) ⇒ HBase::Scoped

Overloads:

  • #range(start_key, opts = {}) ⇒ HBase::Scoped

    Returns an HBase::Scoped object with the specified rowkey range Overrides current range.

    Parameters:

    • start_key (Object) —

      Start rowkey

    • opts (Hash) (defaults to: {}) —

      Prefix filter

    Options Hash (opts):

    • :prefix (Object, Array<Object>) —

      Only rows matching any of the given prefixes are returned

    Returns:

  • #range(start_key, stop_key, opts = {}) ⇒ HBase::Scoped

    Returns an HBase::Scoped object with the specified rowkey range Overrides current range.

    Parameters:

    • start_key (Object, nil) —

      Start rowkey. Can be nil.

    • stop_key (Object) —

      Stop rowkey (exclusive)

    • opts (Hash) (defaults to: {}) —

      Prefix filter

    Options Hash (opts):

    • :prefix (Object, Array<Object>) —

      Only rows matching any of the given prefixes are returned

    Returns:

  • #range(start_stop_range, opts = {}) ⇒ HBase::Scoped

    Returns an HBase::Scoped object with the specified rowkey range Overrides current range.

    Parameters:

    • start_stop_range (Range) —

      Rowkey scan range

    • opts (Hash) (defaults to: {}) —

      Prefix filter

    Options Hash (opts):

    • :prefix (Object, Array<Object>) —

      Only rows matching any of the given prefixes are returned

    Returns:

  • #range(opts) ⇒ HBase::Scoped

    Returns an HBase::Scoped object with the specified rowkey range Overrides current range.

    Examples:

    table.range(:prefix => '2012')
table.range(:prefix => ['2010', '2012'])

    Parameters:

    • opts (Hash) —

      Prefix filter

    Options Hash (opts):

    • :prefix (Object, Array<Object>) —

      Only rows matching any of the given prefixes are returned

    Returns:

Raises:

  • (ArgumentError) 


119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/hbase-jruby/scoped.rb', line 119

def range *key_range
  if (last = key_range.last).is_a?(Hash)
    prefixes = arrayfy(last[:prefix]).compact
    last = last.reject { |k, v| k == :prefix }

    key_range = key_range[0...-1] # defensive
    key_range << last unless last.empty?
  else
    prefixes = []
  end

  if key_range[0].is_a?(Range)
    raise ArgumentError, "Invalid range" unless key_range.length == 1
  elsif !prefixes.empty?
    raise ArgumentError, "Invalid range" unless [0, 1, 2].include?(key_range.length)
  else
    raise ArgumentError, "Invalid range" unless [1, 2].include?(key_range.length)
  end

  raise ArgumentError, "Invalid range" if !key_range.empty? && key_range.all? { |e| e.nil? }

  spawn :@range,
        key_range[0].is_a?(Range) ?
            key_range[0] :
            (key_range.empty? ? nil : key_range.map { |e| e.nil? ? nil : Util.to_bytes(e) }),
        :@prefixes,
        prefixes
end

#time_range(min, max) ⇒ HBase::Scoped

Returns an HBase::Scoped object with the specified time range

Parameters:

  • min (Fixnum|Time) —

    Minimum timestamp (inclusive)

  • max (Fixnum|Time) —

    Maximum timestamp (exclusive)

Returns:

  • (HBase::Scoped) —

    HBase::Scoped object with the specified time range



176
177
178
 
# File 'lib/hbase-jruby/scoped.rb', line 176

def time_range min, max
  spawn :@trange, [min, max].map { |e| time_to_long e }
end

#unscope ⇒ HBase::Scope

A clean HBase::Scoped object for the same table

Returns:

  • (HBase::Scope) —

    A clean HBase::Scoped object for the same table



15
16
17
 
# File 'lib/hbase-jruby/scoped.rb', line 15

def unscope
  Scoped.send(:new, @table, @dcaching)
end

#versions(vs) ⇒ HBase::Scoped

Returns an HBase::Scoped object with the specified version number limit. If not set, all versions of each value are fetched by default.

Parameters:

  • vs (Fixnum) —

    Sets the maximum number of versions

Returns:

  • (HBase::Scoped) —

    HBase::Scoped object with the version number limit

Raises:

  • (ArgumentError) 


219
220
221
222
 
# File 'lib/hbase-jruby/scoped.rb', line 219

def versions vs
  raise ArgumentError, "Invalid versions. Must be a positive integer." unless vs.is_a?(Fixnum) && vs > 0
  spawn :@versions, vs
end

#while(*filters) ⇒ HBase::Scoped

Returns an HBase::Scoped object with the additional filters which will cause early termination of scan

Parameters:

  • filters (Array<Hash, FilterBase, FilterList>)

Returns:

  • (HBase::Scoped) —

    HBase::Scoped object with the additional filters which will cause early termination of scan



158
159
160
 
# File 'lib/hbase-jruby/scoped.rb', line 158

def while *filters
  spawn :@filters, @filters + parse_filter_input(filters).map { |filter| WhileMatchFilter.new(filter) }
end

#with_java_get {|org.apache.hadoop.hbase.client.Get| ... } ⇒ HBase::Scoped

Returns an HBase::Scoped object with the Get-customization block added The given block will be evaluated just before an actual get operation. With method-chaining, multiple blocks can be registered to be evaluated sequentially.

Yields:

  • (org.apache.hadoop.hbase.client.Get)

Returns:

Raises:

  • (ArgumentError) 


248
249
250
251
252
 
# File 'lib/hbase-jruby/scoped.rb', line 248

def with_java_get &block
  raise ArgumentError, "Block not given" if block.nil?
  raise ArgumentError, "Invalid arity: should be 1" unless block.arity == 1
  spawn :@get_cbs, @get_cbs + [block]
end

#with_java_scan {|org.apache.hadoop.hbase.client.Scan| ... } ⇒ HBase::Scoped

Returns an HBase::Scoped object with the Scan-customization block added. The given block will be evaluated just before an actual scan operation. With method-chaining, multiple blocks can be registered to be evaluated sequentially.

Yields:

  • (org.apache.hadoop.hbase.client.Scan)

Returns:

Raises:

  • (ArgumentError) 


237
238
239
240
241
 
# File 'lib/hbase-jruby/scoped.rb', line 237

def with_java_scan &block
  raise ArgumentError, "Block not given" if block.nil?
  raise ArgumentError, "Invalid arity: should be 1" unless block.arity == 1
  spawn :@scan_cbs, @scan_cbs + [block]
end