Skip to content

Commit dabb70f

Browse files
MpdreamzMpdreamz
committed
added Reindex() implements #328 
1 parent c946535 commit dabb70f

13 files changed

+393
-21
lines changed

‎src/Nest.Tests.Integration/Index/ReindexTests.cs‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Linq;
4+
using Nest.Tests.MockData;
5+
using Nest.Tests.MockData.Domain;
6+
using NUnit.Framework;
7+
8+
9+
namespace Nest.Tests.Integration.Index
10+
{
11+
[TestFixture]
12+
public class ReindexTests : IntegrationTests
13+
{
14+
15+
16+
[Test]
17+
public void Reindex()
18+
{
19+
var toIndex = ElasticsearchConfiguration.NewUniqueIndexName();
20+
var observable = this._client.Reindex<object>(r => r
21+
.FromIndex(ElasticsearchConfiguration.DefaultIndex)
22+
.ToIndex(toIndex)
23+
.Query(q=>q.MatchAll())
24+
.Scroll("10s")
25+
.CreateIndex(c=>c
26+
.NumberOfReplicas(0)
27+
.NumberOfShards(1)
28+
)
29+
);
30+
var observer = new ReindexObserver<object>(
31+
onNext: (r) =>
32+
{
33+
var scrollResults = r.QueryResponse;
34+
var bulkResults = r.BulkResponse;
35+
36+
Assert.NotNull(scrollResults);
37+
Assert.NotNull(bulkResults);
38+
39+
Assert.True(scrollResults.IsValid);
40+
Assert.True(bulkResults.IsValid);
41+
},
42+
onError: (e) => Assert.Fail(e.Message),
43+
completed: () =>
44+
{
45+
var originalIndexCount = this._client.Count(new[] { ElasticsearchConfiguration.DefaultIndex }, q=>q.MatchAll());
46+
var newIndexCount = this._client.Count(new[] { toIndex }, q=>q.MatchAll());
47+
Assert.AreEqual(originalIndexCount.Count, newIndexCount.Count);
48+
}
49+
);
50+
51+
observable.Subscribe(observer);
52+
}
53+
54+
}
55+
}

‎src/Nest.Tests.Integration/Nest.Tests.Integration.csproj‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@
6060
<Compile Include="ElasticsearchConfiguration.cs" />
6161
<Compile Include="Core\IndexTests.cs" />
6262
<Compile Include="Help\AliasTests.cs" />
63+
<Compile Include="Index\ReindexTests.cs" />
6364
<Compile Include="Index\IndexUsingUrlIdTests.cs" />
6465
<Compile Include="Indices\Analysis\Analyzers\AnalyzerTest.cs" />
6566
<Compile Include="Indices\Analysis\Analyzers\AnalyzerTestResult.cs" />

‎src/Nest/DSL/ReindexDescriptor.cs‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
using System;
2+
3+
namespace Nest
4+
{
5+
public class ReindexDescriptor<T> where T : class
6+
{
7+
internal string _ToIndexName { get; set; }
8+
internal string _FromIndexName { get; set; }
9+
internal string _Scroll { get; set; }
10+
11+
internal Func<QueryDescriptor<T>, BaseQuery> _QuerySelector { get; set; }
12+
13+
internal Func<CreateIndexDescriptor, CreateIndexDescriptor> _CreateIndexSelector { get; set; }
14+
15+
internal RootObjectMappingDescriptor<T> _RootObjectMappingDescriptor { get; set; }
16+
17+
/// <summary>
18+
/// The index into which we're indexing
19+
/// </summary>
20+
public ReindexDescriptor<T> ToIndex(string name)
21+
{
22+
this._ToIndexName = name;
23+
return this;
24+
}
25+
26+
/// <summary>
27+
/// The index from which we're reindexing
28+
/// </summary>
29+
public ReindexDescriptor<T> FromIndex(string name)
30+
{
31+
this._FromIndexName = name;
32+
return this;
33+
}
34+
35+
/// <summary>
36+
/// A search request can be scrolled by specifying the scroll parameter. The scroll parameter is a time value parameter (for example: scroll=5m), indicating for how long the nodes that participate in the search will maintain relevant resources in order to continue and support it. This is very similar in its idea to opening a cursor against a database.
37+
/// </summary>
38+
/// <param name="scrollTime">The scroll parameter is a time value parameter (for example: scroll=5m)</param>
39+
/// <returns></returns>
40+
public ReindexDescriptor<T> Scroll(string scrollTime)
41+
{
42+
scrollTime.ThrowIfNullOrEmpty("scrollTime");
43+
this._Scroll = scrollTime;
44+
return this;
45+
}
46+
47+
/// <summary>
48+
/// A query to optionally limit the documents to use for the reindex operation.
49+
/// </summary>
50+
public ReindexDescriptor<T> Query(Func<QueryDescriptor<T>, BaseQuery> querySelector)
51+
{
52+
querySelector.ThrowIfNull("querySelector");
53+
this._QuerySelector = querySelector;
54+
return this;
55+
}
56+
57+
/// <summary>
58+
/// The new index name to reindex too.
59+
/// </summary>
60+
public ReindexDescriptor<T> NewIndexName(string name)
61+
{
62+
this._ToIndexName = name;
63+
return this;
64+
}
65+
66+
/// <summary>
67+
/// CreateIndex selector, will be passed the a descriptor initialized with the settings from
68+
/// the index we're reindexing from
69+
/// </summary>
70+
public ReindexDescriptor<T> CreateIndex(Func<CreateIndexDescriptor, CreateIndexDescriptor> createIndexSelector)
71+
{
72+
createIndexSelector.ThrowIfNull("createIndexSelector");
73+
this._CreateIndexSelector = createIndexSelector;
74+
return this;
75+
}
76+
77+
}
78+
}

‎src/Nest/Domain/Responses/BaseResponse.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ public BaseResponse()
1414
{
1515
this.IsValid = true;
1616
}
17-
public bool IsValid { get; internal set; }
17+
public virtual bool IsValid { get; internal set; }
1818
public ConnectionStatus ConnectionStatus { get; internal set; }
1919

2020
internal PropertyNameResolver PropertyNameResolver { get; set; }

‎src/Nest/Domain/Responses/IReindexResponse.cs‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
namespace Nest
2+
{
3+
/// <summary>
4+
/// POCO representing the reindex response for a each step
5+
/// </summary>
6+
public interface IReindexResponse<T> where T : class
7+
{
8+
/// <summary>
9+
/// The bulk result indexing the search results into the new index.
10+
/// </summary>
11+
IBulkResponse BulkResponse { get; }
12+
13+
/// <summary>
14+
/// The scroll result
15+
/// </summary>
16+
IQueryResponse<T> QueryResponse { get; }
17+
18+
/// <summary>
19+
/// The no of scroll this result represents
20+
/// </summary>
21+
int Scroll { get; }
22+
23+
/// <summary>
24+
/// Whether both the scroll and reindex result are valid
25+
/// </summary>
26+
bool IsValid { get; }
27+
}
28+
}

‎src/Nest/Domain/Responses/IndexResponse.cs‎

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -2,28 +2,28 @@
22

33
namespace Nest
44
{
5-
public interface IIndexResponse : IResponse
6-
{
7-
string Id { get; }
8-
string Index { get; }
9-
string Type { get; }
10-
string Version { get; }
11-
bool OK { get; }
12-
}
5+
public interface IIndexResponse : IResponse
6+
{
7+
string Id { get; }
8+
string Index { get; }
9+
string Type { get; }
10+
string Version { get; }
11+
bool OK { get; }
12+
}
1313

1414
[JsonObject]
15-
public class IndexResponse : BaseResponse, IIndexResponse
16-
{
17-
[JsonProperty("_index")]
15+
public class IndexResponse : BaseResponse, IIndexResponse
16+
{
17+
[JsonProperty("_index")]
1818
public string Index { get; internal set; }
19-
[JsonProperty("_type")]
20-
public string Type { get; internal set; }
21-
[JsonProperty("_id")]
22-
public string Id { get; internal set; }
23-
[JsonProperty("_version")]
24-
public string Version { get; internal set; }
25-
[JsonProperty("ok")]
26-
public bool OK { get; internal set; }
27-
19+
[JsonProperty("_type")]
20+
public string Type { get; internal set; }
21+
[JsonProperty("_id")]
22+
public string Id { get; internal set; }
23+
[JsonProperty("_version")]
24+
public string Version { get; internal set; }
25+
[JsonProperty("ok")]
26+
public bool OK { get; internal set; }
27+
2828
}
2929
}

‎src/Nest/Domain/Responses/ReindexObservable.cs‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
using System;
2+
3+
namespace Nest
4+
{
5+
public class ReindexObservable<T> : IDisposable, IObservable<IReindexResponse<T>> where T : class
6+
{
7+
private ReindexDescriptor<T> _reindexDescriptor;
8+
internal IElasticClient CurrentClient { get; set; }
9+
internal ReindexDescriptor<T> ReindexDescriptor { get; set; }
10+
11+
public ReindexObservable(IElasticClient client, ReindexDescriptor<T> reindexDescriptor)
12+
{
13+
this._reindexDescriptor = reindexDescriptor;
14+
this.CurrentClient = client;
15+
}
16+
17+
public IDisposable Subscribe(IObserver<IReindexResponse<T>> observer)
18+
{
19+
observer.ThrowIfNull("observer");
20+
try
21+
{
22+
this.Reindex(observer);
23+
}
24+
catch (Exception e)
25+
{
26+
observer.OnError(e);
27+
}
28+
return this;
29+
30+
}
31+
32+
private void Reindex(IObserver<IReindexResponse<T>> observer)
33+
{
34+
var fromIndex = this._reindexDescriptor._FromIndexName;
35+
var toIndex = this._reindexDescriptor._ToIndexName;
36+
var scroll = this._reindexDescriptor._Scroll ?? "2m";
37+
38+
fromIndex.ThrowIfNullOrEmpty("fromIndex");
39+
toIndex.ThrowIfNullOrEmpty("toIndex");
40+
41+
var indexSettings = this.CurrentClient.GetIndexSettings(this._reindexDescriptor._FromIndexName);
42+
var createSettings = new CreateIndexDescriptor(this.CurrentClient.Settings).InitializeUsing(indexSettings.Settings);
43+
var createIndexResponse = this.CurrentClient
44+
.CreateIndex(toIndex, (c) => this._reindexDescriptor._CreateIndexSelector(createSettings));
45+
if (!createIndexResponse.IsValid)
46+
throw new ReindexException(createIndexResponse.ConnectionStatus);
47+
48+
var page = 0;
49+
var searchResult = this.CurrentClient.Search<T>(
50+
s => s
51+
.Index(fromIndex)
52+
.AllTypes()
53+
.From(0)
54+
.Take(100)
55+
.Query(this._reindexDescriptor._QuerySelector)
56+
.SearchType(SearchType.Scan)
57+
.Scroll(scroll)
58+
);
59+
if (searchResult.Total <= 0)
60+
throw new ReindexException(searchResult.ConnectionStatus, "index " + fromIndex + " has no documents!");
61+
IBulkResponse indexResult = null;
62+
do
63+
{
64+
searchResult = this.CurrentClient.Scroll<T>(scroll, searchResult.ScrollId);
65+
if (searchResult.Documents.HasAny())
66+
indexResult = this.IndexSearchResults(searchResult, observer, toIndex, page);
67+
page++;
68+
} while (searchResult.IsValid && indexResult != null && indexResult.IsValid && searchResult.Documents.HasAny());
69+
70+
71+
observer.OnCompleted();
72+
}
73+
74+
public IBulkResponse IndexSearchResults(IQueryResponse<T> searchResult,IObserver<IReindexResponse<T>> observer, string toIndex, int page)
75+
{
76+
if (!searchResult.IsValid)
77+
throw new ReindexException(searchResult.ConnectionStatus, "reindex failed on scroll #" + page);
78+
79+
var indexResult = this.CurrentClient.IndexMany(searchResult.Documents, toIndex);
80+
if (!indexResult.IsValid)
81+
throw new ReindexException(indexResult.ConnectionStatus, "reindex failed when indexing page " + page);
82+
83+
observer.OnNext(new ReindexResponse<T>()
84+
{
85+
BulkResponse = indexResult,
86+
QueryResponse = searchResult,
87+
Scroll = page
88+
});
89+
return indexResult;
90+
}
91+
92+
93+
public void Dispose()
94+
{
95+
96+
}
97+
}
98+
}

‎src/Nest/Domain/Responses/ReindexObserver.cs‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
using System;
2+
3+
namespace Nest
4+
{
5+
public class ReindexObserver<T> : IObserver<IReindexResponse<T>> where T : class
6+
{
7+
private readonly Action<IReindexResponse<T>> _onNext;
8+
private readonly Action<Exception> _onError;
9+
private readonly Action _completed;
10+
11+
public ReindexObserver(Action<IReindexResponse<T>> onNext, Action<Exception> onError, Action completed)
12+
{
13+
this._completed = completed;
14+
this._onError = onError;
15+
this._onNext = onNext;
16+
}
17+
18+
19+
public void OnCompleted()
20+
{
21+
if (this._completed != null)
22+
this._completed();
23+
}
24+
25+
public void OnError(Exception error)
26+
{
27+
if (this._onError != null)
28+
this._onError(error);
29+
}
30+
31+
public void OnNext(IReindexResponse<T> value)
32+
{
33+
if(this._onNext != null)
34+
this._onNext(value);
35+
}
36+
}
37+
}

‎src/Nest/Domain/Responses/ReindexResponse.cs‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
using System.Collections.Generic;
2+
using System.Collections.Concurrent;
3+
using Newtonsoft.Json;
4+
5+
namespace Nest
6+
{
7+
/// <summary>
8+
/// POCO representing the reindex response for a each step
9+
/// </summary>
10+
[JsonObject]
11+
public class ReindexResponse<T> : IReindexResponse<T> where T : class
12+
{
13+
public IBulkResponse BulkResponse { get; internal set; }
14+
public IQueryResponse<T> QueryResponse { get; internal set; }
15+
16+
public int Scroll { get; internal set; }
17+
18+
public bool IsValid
19+
{
20+
get
21+
{
22+
return (this.BulkResponse != null && this.BulkResponse.IsValid
23+
&& this.QueryResponse != null && this.QueryResponse.IsValid);
24+
}
25+
}
26+
}
27+
}

0 commit comments

Comments
 (0)