Added allow_missing option to reindex

Andrew Kane
1 parent 13fd0f4e
Showing 9 changed files with 51 additions and 21 deletions Show diff stats
lib/searchkick/bulk_reindex_job.rb
lib/searchkick/index.rb
lib/searchkick/indexer.rb
lib/searchkick/model.rb
lib/searchkick/process_batch_job.rb
lib/searchkick/record_indexer.rb
lib/searchkick/reindex_v2_job.rb
lib/searchkick/relation_indexer.rb
test/partial_reindex_test.rb
@@ -14,7 +14,8 @@ module Searchkick
       relation = Searchkick.load_records(relation, record_ids)
       relation = relation.search_import if relation.respond_to?(:search_import)
  
-      RecordIndexer.new(index).reindex(relation, mode: :inline, method_name: method_name, full: false)
+      # TODO support allow_missing
+      RecordIndexer.new(index).reindex(relation, mode: :inline, method_name: method_name, allow_missing: false, full: false)
       RelationIndexer.new(index).batch_completed(batch_id) if batch_id
     end
   end
@@ -163,11 +163,11 @@ module Searchkick
     end
     alias_method :import, :bulk_index
  
-    def bulk_update(records, method_name)
+    def bulk_update(records, method_name, allow_missing: false)
       return if records.empty?
  
       notify_bulk(records, "Update") do
-        queue_update(records, method_name)
+        queue_update(records, method_name, allow_missing: allow_missing)
       end
     end
  
@@ -211,10 +211,10 @@ module Searchkick
  
     # note: this is designed to be used internally
     # so it does not check object matches index class
-    def reindex(object, method_name: nil, full: false, **options)
+    def reindex(object, method_name: nil, allow_missing: false, full: false, **options)
       if object.is_a?(Array)
         # note: purposefully skip full
-        return reindex_records(object, method_name: method_name, **options)
+        return reindex_records(object, method_name: method_name, allow_missing: allow_missing, **options)
       end
  
       if !object.respond_to?(:searchkick_klass)
@@ -233,7 +233,7 @@ module Searchkick
         raise ArgumentError, "unsupported keywords: #{options.keys.map(&:inspect).join(", ")}" if options.any?
  
         # import only
-        import_scope(relation, method_name: method_name, mode: mode)
+        import_scope(relation, method_name: method_name, mode: mode, allow_missing: allow_missing)
         self.refresh if refresh
         true
       else
@@ -322,8 +322,10 @@ module Searchkick
       Searchkick.indexer.queue(records.reject { |r| r.id.blank? }.map { |r| RecordData.new(self, r).delete_data })
     end
  
-    def queue_update(records, method_name)
-      Searchkick.indexer.queue(records.map { |r| RecordData.new(self, r).update_data(method_name) })
+    def queue_update(records, method_name, allow_missing:)
+      items = records.map { |r| RecordData.new(self, r).update_data(method_name) }
+      items.each { |i| i.instance_variable_set(:@allow_missing, true) } if allow_missing
+      Searchkick.indexer.queue(items)
     end
  
     def relation_indexer
@@ -24,12 +24,20 @@ module Searchkick
         # note: delete does not set error when item not found
         first_with_error = response["items"].map do |item|
           (item["index"] || item["delete"] || item["update"])
-        end.find { |item| item["error"] }
-        raise ImportError, "#{first_with_error["error"]} on item with id '#{first_with_error["_id"]}'"
+        end.find.with_index { |item, i| item["error"] && !allow_missing?(items[i], item["error"]) }
+        if first_with_error
+          raise ImportError, "#{first_with_error["error"]} on item with id '#{first_with_error["_id"]}'"
+        end
       end
  
       # maybe return response in future
       nil
     end
+
+    private
+
+    def allow_missing?(item, error)
+      error["type"] == "document_missing_exception" && item.instance_variable_defined?(:@allow_missing)
+    end
   end
 end
@@ -27,8 +27,8 @@ module Searchkick
       mod = Module.new
       include(mod)
       mod.module_eval do
-        def reindex(method_name = nil, mode: nil, refresh: false)
-          self.class.searchkick_index.reindex([self], method_name: method_name, mode: mode, refresh: refresh, single: true)
+        def reindex(method_name = nil, mode: nil, refresh: false, allow_missing: false)
+          self.class.searchkick_index.reindex([self], method_name: method_name, mode: mode, refresh: refresh, allow_missing: allow_missing, single: true)
         end unless base.method_defined?(:reindex)
  
         def similar(**options)
@@ -14,7 +14,7 @@ module Searchkick
         end
  
       relation = Searchkick.scope(model)
-      RecordIndexer.new(index).reindex_items(relation, items, method_name: nil)
+      RecordIndexer.new(index).reindex_items(relation, items, method_name: nil, allow_missing: false)
     end
   end
 end
@@ -6,7 +6,7 @@ module Searchkick
       @index = index
     end
  
-    def reindex(records, mode:, method_name:, full: false, single: false)
+    def reindex(records, mode:, method_name:, allow_missing:, full: false, single: false)
       # prevents exists? check if records is a relation
       records = records.to_a
       return if records.empty?
@@ -17,6 +17,10 @@ module Searchkick
           raise Error, "Active Job not found"
         end
  
+        if allow_missing
+          raise Error, "Allow missing not supported with async option yet"
+        end
+
         # we could likely combine ReindexV2Job, BulkReindexJob, and ProcessBatchJob
         # but keep them separate for now
         if single
@@ -51,7 +55,7 @@ module Searchkick
         index.reindex_queue.push_records(records)
       when true, :inline
         index_records, other_records = records.partition { |r| index_record?(r) }
-        import_inline(index_records, !full ? other_records : [], method_name: method_name, single: single)
+        import_inline(index_records, !full ? other_records : [], method_name: method_name, allow_missing: allow_missing, single: single)
       else
         raise ArgumentError, "Invalid value for mode"
       end
@@ -60,7 +64,7 @@ module Searchkick
       true
     end
  
-    def reindex_items(klass, items, method_name:, single: false)
+    def reindex_items(klass, items, method_name:, allow_missing:, single: false)
       routing = items.to_h { |r| [r[:id], r[:routing]] }
       record_ids = routing.keys
  
@@ -76,7 +80,7 @@ module Searchkick
           construct_record(klass, id, routing[id])
         end
  
-      import_inline(records, delete_records, method_name: method_name, single: single)
+      import_inline(records, delete_records, method_name: method_name, allow_missing: allow_missing, single: single)
     end
  
     private
@@ -86,13 +90,13 @@ module Searchkick
     end
  
     # import in single request with retries
-    def import_inline(index_records, delete_records, method_name:, single:)
+    def import_inline(index_records, delete_records, method_name:, allow_missing:, single:)
       return if index_records.empty? && delete_records.empty?
  
       maybe_bulk(index_records, delete_records, method_name, single) do
         if index_records.any?
           if method_name
-            index.bulk_update(index_records, method_name)
+            index.bulk_update(index_records, method_name, allow_missing: allow_missing)
           else
             index.bulk_index(index_records)
           end
@@ -11,7 +11,8 @@ module Searchkick
       # but keep for now for backwards compatibility
       model = model.unscoped if model.respond_to?(:unscoped)
       items = [{id: id, routing: routing}]
-      RecordIndexer.new(index).reindex_items(model, items, method_name: method_name, single: true)
+      # TODO support allow_missing
+      RecordIndexer.new(index).reindex_items(model, items, method_name: method_name, allow_missing: false, single: true)
     end
   end
 end
@@ -6,7 +6,7 @@ module Searchkick
       @index = index
     end
  
-    def reindex(relation, mode:, method_name: nil, full: false, resume: false, scope: nil)
+    def reindex(relation, mode:, method_name: nil, allow_missing: false, full: false, resume: false, scope: nil)
       # apply scopes
       if scope
         relation = relation.send(scope)
@@ -32,6 +32,7 @@ module Searchkick
       reindex_options = {
         mode: mode,
         method_name: method_name,
+        allow_missing: allow_missing,
         full: full
       }
       record_indexer = RecordIndexer.new(index)
@@ -71,4 +71,17 @@ class PartialReindexTest &lt; Minitest::Test
     end
     assert_match "document missing", error.message
   end
+
+  def test_allow_missing
+    store [{name: "Hi", color: "Blue"}]
+
+    product = Product.first
+    Product.search_index.remove(product)
+
+    product.reindex(:search_name, allow_missing: true)
+    Product.where(id: product.id).reindex(:search_name, allow_missing: true)
+    Searchkick.callbacks(:bulk) do
+      product.reindex(:search_name, allow_missing: true)
+    end
+  end
 end
...	...	@@ -14,7 +14,8 @@ module Searchkick
14	14	relation = Searchkick.load_records(relation, record_ids)
15	15	relation = relation.search_import if relation.respond_to?(:search_import)
16	16
17		- RecordIndexer.new(index).reindex(relation, mode: :inline, method_name: method_name, full: false)
	17	+ # TODO support allow_missing
	18	+ RecordIndexer.new(index).reindex(relation, mode: :inline, method_name: method_name, allow_missing: false, full: false)
18	19	RelationIndexer.new(index).batch_completed(batch_id) if batch_id
19	20	end
20	21	end
...	...
...	...	@@ -163,11 +163,11 @@ module Searchkick
163	163	end
164	164	alias_method :import, :bulk_index
165	165
166		- def bulk_update(records, method_name)
	166	+ def bulk_update(records, method_name, allow_missing: false)
167	167	return if records.empty?
168	168
169	169	notify_bulk(records, "Update") do
170		- queue_update(records, method_name)
	170	+ queue_update(records, method_name, allow_missing: allow_missing)
171	171	end
172	172	end
173	173
...	...	@@ -211,10 +211,10 @@ module Searchkick
211	211
212	212	# note: this is designed to be used internally
213	213	# so it does not check object matches index class
214		- def reindex(object, method_name: nil, full: false, **options)
	214	+ def reindex(object, method_name: nil, allow_missing: false, full: false, **options)
215	215	if object.is_a?(Array)
216	216	# note: purposefully skip full
217		- return reindex_records(object, method_name: method_name, **options)
	217	+ return reindex_records(object, method_name: method_name, allow_missing: allow_missing, **options)
218	218	end
219	219
220	220	if !object.respond_to?(:searchkick_klass)
...	...	@@ -233,7 +233,7 @@ module Searchkick
233	233	raise ArgumentError, "unsupported keywords: #{options.keys.map(&:inspect).join(", ")}" if options.any?
234	234
235	235	# import only
236		- import_scope(relation, method_name: method_name, mode: mode)
	236	+ import_scope(relation, method_name: method_name, mode: mode, allow_missing: allow_missing)
237	237	self.refresh if refresh
238	238	true
239	239	else
...	...	@@ -322,8 +322,10 @@ module Searchkick
322	322	Searchkick.indexer.queue(records.reject { \|r\| r.id.blank? }.map { \|r\| RecordData.new(self, r).delete_data })
323	323	end
324	324
325		- def queue_update(records, method_name)
326		- Searchkick.indexer.queue(records.map { \|r\| RecordData.new(self, r).update_data(method_name) })
	325	+ def queue_update(records, method_name, allow_missing:)
	326	+ items = records.map { \|r\| RecordData.new(self, r).update_data(method_name) }
	327	+ items.each { \|i\| i.instance_variable_set(:@allow_missing, true) } if allow_missing
	328	+ Searchkick.indexer.queue(items)
327	329	end
328	330
329	331	def relation_indexer
...	...
...	...	@@ -24,12 +24,20 @@ module Searchkick
24	24	# note: delete does not set error when item not found
25	25	first_with_error = response["items"].map do \|item\|
26	26	(item["index"] \|\| item["delete"] \|\| item["update"])
27		- end.find { \|item\| item["error"] }
28		- raise ImportError, "#{first_with_error["error"]} on item with id '#{first_with_error["_id"]}'"
	27	+ end.find.with_index { \|item, i\| item["error"] && !allow_missing?(items[i], item["error"]) }
	28	+ if first_with_error
	29	+ raise ImportError, "#{first_with_error["error"]} on item with id '#{first_with_error["_id"]}'"
	30	+ end
29	31	end
30	32
31	33	# maybe return response in future
32	34	nil
33	35	end
	36	+
	37	+ private
	38	+
	39	+ def allow_missing?(item, error)
	40	+ error["type"] == "document_missing_exception" && item.instance_variable_defined?(:@allow_missing)
	41	+ end
34	42	end
35	43	end
...	...
...	...	@@ -27,8 +27,8 @@ module Searchkick
27	27	mod = Module.new
28	28	include(mod)
29	29	mod.module_eval do
30		- def reindex(method_name = nil, mode: nil, refresh: false)
31		- self.class.searchkick_index.reindex([self], method_name: method_name, mode: mode, refresh: refresh, single: true)
	30	+ def reindex(method_name = nil, mode: nil, refresh: false, allow_missing: false)
	31	+ self.class.searchkick_index.reindex([self], method_name: method_name, mode: mode, refresh: refresh, allow_missing: allow_missing, single: true)
32	32	end unless base.method_defined?(:reindex)
33	33
34	34	def similar(**options)
...	...
...	...	@@ -14,7 +14,7 @@ module Searchkick
14	14	end
15	15
16	16	relation = Searchkick.scope(model)
17		- RecordIndexer.new(index).reindex_items(relation, items, method_name: nil)
	17	+ RecordIndexer.new(index).reindex_items(relation, items, method_name: nil, allow_missing: false)
18	18	end
19	19	end
20	20	end
...	...
...	...	@@ -6,7 +6,7 @@ module Searchkick
6	6	@index = index
7	7	end
8	8
9		- def reindex(records, mode:, method_name:, full: false, single: false)
	9	+ def reindex(records, mode:, method_name:, allow_missing:, full: false, single: false)
10	10	# prevents exists? check if records is a relation
11	11	records = records.to_a
12	12	return if records.empty?
...	...	@@ -17,6 +17,10 @@ module Searchkick
17	17	raise Error, "Active Job not found"
18	18	end
19	19
	20	+ if allow_missing
	21	+ raise Error, "Allow missing not supported with async option yet"
	22	+ end
	23	+
20	24	# we could likely combine ReindexV2Job, BulkReindexJob, and ProcessBatchJob
21	25	# but keep them separate for now
22	26	if single
...	...	@@ -51,7 +55,7 @@ module Searchkick
51	55	index.reindex_queue.push_records(records)
52	56	when true, :inline
53	57	index_records, other_records = records.partition { \|r\| index_record?(r) }
54		- import_inline(index_records, !full ? other_records : [], method_name: method_name, single: single)
	58	+ import_inline(index_records, !full ? other_records : [], method_name: method_name, allow_missing: allow_missing, single: single)
55	59	else
56	60	raise ArgumentError, "Invalid value for mode"
57	61	end
...	...	@@ -60,7 +64,7 @@ module Searchkick
60	64	true
61	65	end
62	66
63		- def reindex_items(klass, items, method_name:, single: false)
	67	+ def reindex_items(klass, items, method_name:, allow_missing:, single: false)
64	68	routing = items.to_h { \|r\| [r[:id], r[:routing]] }
65	69	record_ids = routing.keys
66	70
...	...	@@ -76,7 +80,7 @@ module Searchkick
76	80	construct_record(klass, id, routing[id])
77	81	end
78	82
79		- import_inline(records, delete_records, method_name: method_name, single: single)
	83	+ import_inline(records, delete_records, method_name: method_name, allow_missing: allow_missing, single: single)
80	84	end
81	85
82	86	private
...	...	@@ -86,13 +90,13 @@ module Searchkick
86	90	end
87	91
88	92	# import in single request with retries
89		- def import_inline(index_records, delete_records, method_name:, single:)
	93	+ def import_inline(index_records, delete_records, method_name:, allow_missing:, single:)
90	94	return if index_records.empty? && delete_records.empty?
91	95
92	96	maybe_bulk(index_records, delete_records, method_name, single) do
93	97	if index_records.any?
94	98	if method_name
95		- index.bulk_update(index_records, method_name)
	99	+ index.bulk_update(index_records, method_name, allow_missing: allow_missing)
96	100	else
97	101	index.bulk_index(index_records)
98	102	end
...	...
...	...	@@ -11,7 +11,8 @@ module Searchkick
11	11	# but keep for now for backwards compatibility
12	12	model = model.unscoped if model.respond_to?(:unscoped)
13	13	items = [{id: id, routing: routing}]
14		- RecordIndexer.new(index).reindex_items(model, items, method_name: method_name, single: true)
	14	+ # TODO support allow_missing
	15	+ RecordIndexer.new(index).reindex_items(model, items, method_name: method_name, allow_missing: false, single: true)
15	16	end
16	17	end
17	18	end
...	...
...	...	@@ -6,7 +6,7 @@ module Searchkick
6	6	@index = index
7	7	end
8	8
9		- def reindex(relation, mode:, method_name: nil, full: false, resume: false, scope: nil)
	9	+ def reindex(relation, mode:, method_name: nil, allow_missing: false, full: false, resume: false, scope: nil)
10	10	# apply scopes
11	11	if scope
12	12	relation = relation.send(scope)
...	...	@@ -32,6 +32,7 @@ module Searchkick
32	32	reindex_options = {
33	33	mode: mode,
34	34	method_name: method_name,
	35	+ allow_missing: allow_missing,
35	36	full: full
36	37	}
37	38	record_indexer = RecordIndexer.new(index)
...	...
...	...	@@ -71,4 +71,17 @@ class PartialReindexTest < Minitest::Test
71	71	end
72	72	assert_match "document missing", error.message
73	73	end
	74	+
	75	+ def test_allow_missing
	76	+ store [{name: "Hi", color: "Blue"}]
	77	+
	78	+ product = Product.first
	79	+ Product.search_index.remove(product)
	80	+
	81	+ product.reindex(:search_name, allow_missing: true)
	82	+ Product.where(id: product.id).reindex(:search_name, allow_missing: true)
	83	+ Searchkick.callbacks(:bulk) do
	84	+ product.reindex(:search_name, allow_missing: true)
	85	+ end
	86	+ end
74	87	end
...	...