Skip to content

Commit f9d30ad

Browse files
przemekwitekprzemekwitek
authored
[Transform] Align transform checkpoint range with date_histogram interval for better performance (#74004)
1 parent a388f70 commit f9d30ad

File tree

23 files changed

+679
-115
lines changed

23 files changed

+679
-115
lines changed

‎client/rest-high-level/src/main/java/org/elasticsearch/client/transform/transforms/SettingsConfig.java‎

Lines changed: 44 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,20 +25,25 @@ public class SettingsConfig implements ToXContentObject {
2525
private static final ParseField MAX_PAGE_SEARCH_SIZE = new ParseField("max_page_search_size");
2626
private static final ParseField DOCS_PER_SECOND = new ParseField("docs_per_second");
2727
private static final ParseField DATES_AS_EPOCH_MILLIS = new ParseField("dates_as_epoch_millis");
28+
private static final ParseField INTERIM_RESULTS = new ParseField("interim_results");
2829
private static final int DEFAULT_MAX_PAGE_SEARCH_SIZE = -1;
2930
private static final float DEFAULT_DOCS_PER_SECOND = -1F;
3031

3132
// use an integer as we need to code 4 states: true, false, null (unchanged), default (defined server side)
3233
private static final int DEFAULT_DATES_AS_EPOCH_MILLIS = -1;
3334

35+
// use an integer as we need to code 4 states: true, false, null (unchanged), default (defined server side)
36+
private static final int DEFAULT_INTERIM_RESULTS = -1;
37+
3438
private final Integer maxPageSearchSize;
3539
private final Float docsPerSecond;
3640
private final Integer datesAsEpochMillis;
41+
private final Integer interimResults;
3742

3843
private static final ConstructingObjectParser<SettingsConfig, Void> PARSER = new ConstructingObjectParser<>(
3944
"settings_config",
4045
true,
41-
args -> new SettingsConfig((Integer) args[0], (Float) args[1], (Integer) args[2])
46+
args -> new SettingsConfig((Integer) args[0], (Float) args[1], (Integer) args[2], (Integer) args[3])
4247
);
4348

4449
static {
@@ -51,16 +56,24 @@ public class SettingsConfig implements ToXContentObject {
5156
DATES_AS_EPOCH_MILLIS,
5257
ValueType.BOOLEAN_OR_NULL
5358
);
59+
// this boolean requires 4 possible values: true, false, not_specified, default, therefore using a custom parser
60+
PARSER.declareField(
61+
optionalConstructorArg(),
62+
p -> p.currentToken() == XContentParser.Token.VALUE_NULL ? DEFAULT_INTERIM_RESULTS : p.booleanValue() ? 1 : 0,
63+
INTERIM_RESULTS,
64+
ValueType.BOOLEAN_OR_NULL
65+
);
5466
}
5567

5668
public static SettingsConfig fromXContent(final XContentParser parser) {
5769
return PARSER.apply(parser, null);
5870
}
5971

60-
SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Integer datesAsEpochMillis) {
72+
SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Integer datesAsEpochMillis, Integer interimResults) {
6173
this.maxPageSearchSize = maxPageSearchSize;
6274
this.docsPerSecond = docsPerSecond;
6375
this.datesAsEpochMillis = datesAsEpochMillis;
76+
this.interimResults = interimResults;
6477
}
6578

6679
@Override
@@ -87,6 +100,13 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
87100
builder.field(DATES_AS_EPOCH_MILLIS.getPreferredName(), datesAsEpochMillis > 0 ? true : false);
88101
}
89102
}
103+
if (interimResults != null) {
104+
if (interimResults.equals(DEFAULT_INTERIM_RESULTS)) {
105+
builder.field(INTERIM_RESULTS.getPreferredName(), (Boolean) null);
106+
} else {
107+
builder.field(INTERIM_RESULTS.getPreferredName(), interimResults > 0 ? true : false);
108+
}
109+
}
90110
builder.endObject();
91111
return builder;
92112
}
@@ -103,6 +123,10 @@ public Boolean getDatesAsEpochMillis() {
103123
return datesAsEpochMillis != null ? datesAsEpochMillis > 0 : null;
104124
}
105125

126+
public Boolean getInterimResults() {
127+
return interimResults != null ? interimResults > 0 : null;
128+
}
129+
106130
@Override
107131
public boolean equals(Object other) {
108132
if (other == this) {
@@ -115,12 +139,13 @@ public boolean equals(Object other) {
115139
SettingsConfig that = (SettingsConfig) other;
116140
return Objects.equals(maxPageSearchSize, that.maxPageSearchSize)
117141
&& Objects.equals(docsPerSecond, that.docsPerSecond)
118-
&& Objects.equals(datesAsEpochMillis, that.datesAsEpochMillis);
142+
&& Objects.equals(datesAsEpochMillis, that.datesAsEpochMillis)
143+
&& Objects.equals(interimResults, that.interimResults);
119144
}
120145

121146
@Override
122147
public int hashCode() {
123-
return Objects.hash(maxPageSearchSize, docsPerSecond, datesAsEpochMillis);
148+
return Objects.hash(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, interimResults);
124149
}
125150

126151
public static Builder builder() {
@@ -131,6 +156,7 @@ public static class Builder {
131156
private Integer maxPageSearchSize;
132157
private Float docsPerSecond;
133158
private Integer datesAsEpochMillis;
159+
private Integer interimResults;
134160

135161
/**
136162
* Sets the paging maximum paging maxPageSearchSize that transform can use when
@@ -176,8 +202,21 @@ public Builder setDatesAsEpochMillis(Boolean datesAsEpochMillis) {
176202
return this;
177203
}
178204

205+
/**
206+
* Whether to write interim results in transform checkpoints.
207+
*
208+
* An explicit `null` resets to default.
209+
*
210+
* @param interimResults true if interim results should be written.
211+
* @return the {@link Builder} with interimResults set.
212+
*/
213+
public Builder setInterimResults(Boolean interimResults) {
214+
this.interimResults = interimResults == null ? DEFAULT_INTERIM_RESULTS : interimResults ? 1 : 0;
215+
return this;
216+
}
217+
179218
public SettingsConfig build() {
180-
return new SettingsConfig(maxPageSearchSize, docsPerSecond, datesAsEpochMillis);
219+
return new SettingsConfig(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, interimResults);
181220
}
182221
}
183222
}

‎client/rest-high-level/src/test/java/org/elasticsearch/client/transform/transforms/SettingsConfigTests.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ public static SettingsConfig randomSettingsConfig() {
3030
return new SettingsConfig(
3131
randomBoolean() ? null : randomIntBetween(10, 10_000),
3232
randomBoolean() ? null : randomFloat(),
33+
randomBoolean() ? null : randomIntBetween(-1, 1),
3334
randomBoolean() ? null : randomIntBetween(-1, 1)
3435
);
3536
}
@@ -72,6 +73,7 @@ public void testExplicitNullOnWriteParser() throws IOException {
7273
assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set"));
7374
assertNull(settingsAsMap.getOrDefault("docs_per_second", "not_set"));
7475
assertThat(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"), equalTo("not_set"));
76+
assertThat(settingsAsMap.getOrDefault("interim_results", "not_set"), equalTo("not_set"));
7577

7678
config = fromString("{\"dates_as_epoch_millis\" : null}");
7779
assertFalse(config.getDatesAsEpochMillis());
@@ -80,6 +82,16 @@ public void testExplicitNullOnWriteParser() throws IOException {
8082
assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set"));
8183
assertThat(settingsAsMap.getOrDefault("docs_per_second", "not_set"), equalTo("not_set"));
8284
assertNull(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"));
85+
assertThat(settingsAsMap.getOrDefault("interim_results", "not_set"), equalTo("not_set"));
86+
87+
config = fromString("{\"interim_results\" : null}");
88+
assertFalse(config.getInterimResults());
89+
90+
settingsAsMap = xContentToMap(config);
91+
assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set"));
92+
assertThat(settingsAsMap.getOrDefault("docs_per_second", "not_set"), equalTo("not_set"));
93+
assertThat(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"), equalTo("not_set"));
94+
assertNull(settingsAsMap.getOrDefault("interim_results", "not_set"));
8395
}
8496

8597
public void testExplicitNullOnWriteBuilder() throws IOException {
@@ -91,10 +103,12 @@ public void testExplicitNullOnWriteBuilder() throws IOException {
91103
assertNull(settingsAsMap.getOrDefault("max_page_search_size", "not_set"));
92104
assertThat(settingsAsMap.getOrDefault("docs_per_second", "not_set"), equalTo("not_set"));
93105
assertThat(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"), equalTo("not_set"));
106+
assertThat(settingsAsMap.getOrDefault("interim_results", "not_set"), equalTo("not_set"));
94107

95108
SettingsConfig emptyConfig = new SettingsConfig.Builder().build();
96109
assertNull(emptyConfig.getMaxPageSearchSize());
97110
assertNull(emptyConfig.getDatesAsEpochMillis());
111+
assertNull(emptyConfig.getInterimResults());
98112

99113
settingsAsMap = xContentToMap(emptyConfig);
100114
assertTrue(settingsAsMap.isEmpty());
@@ -106,6 +120,7 @@ public void testExplicitNullOnWriteBuilder() throws IOException {
106120
assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set"));
107121
assertNull(settingsAsMap.getOrDefault("docs_per_second", "not_set"));
108122
assertThat(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"), equalTo("not_set"));
123+
assertThat(settingsAsMap.getOrDefault("interim_results", "not_set"), equalTo("not_set"));
109124

110125
config = new SettingsConfig.Builder().setDatesAsEpochMillis(null).build();
111126
// returns false, however it's `null` as in "use default", checked next
@@ -115,6 +130,17 @@ public void testExplicitNullOnWriteBuilder() throws IOException {
115130
assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set"));
116131
assertThat(settingsAsMap.getOrDefault("docs_per_second", "not_set"), equalTo("not_set"));
117132
assertNull(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"));
133+
assertThat(settingsAsMap.getOrDefault("interim_results", "not_set"), equalTo("not_set"));
134+
135+
config = new SettingsConfig.Builder().setInterimResults(null).build();
136+
// returns false, however it's `null` as in "use default", checked next
137+
assertFalse(config.getInterimResults());
138+
139+
settingsAsMap = xContentToMap(config);
140+
assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set"));
141+
assertThat(settingsAsMap.getOrDefault("docs_per_second", "not_set"), equalTo("not_set"));
142+
assertThat(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"), equalTo("not_set"));
143+
assertNull(settingsAsMap.getOrDefault("interim_results", "not_set"));
118144
}
119145

120146
private Map<String, Object> xContentToMap(ToXContent xcontent) throws IOException {

‎client/rest-high-level/src/test/java/org/elasticsearch/client/transform/transforms/hlrc/SettingsConfigTests.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ public static org.elasticsearch.xpack.core.transform.transforms.SettingsConfig r
2323
return new org.elasticsearch.xpack.core.transform.transforms.SettingsConfig(
2424
randomBoolean() ? null : randomIntBetween(10, 10_000),
2525
randomBoolean() ? null : randomFloat(),
26+
randomBoolean() ? null : randomIntBetween(0, 1),
2627
randomBoolean() ? null : randomIntBetween(0, 1)
2728
);
2829
}
@@ -34,6 +35,7 @@ public static void assertHlrcEquals(
3435
assertEquals(serverTestInstance.getMaxPageSearchSize(), clientInstance.getMaxPageSearchSize());
3536
assertEquals(serverTestInstance.getDocsPerSecond(), clientInstance.getDocsPerSecond());
3637
assertEquals(serverTestInstance.getDatesAsEpochMillis(), clientInstance.getDatesAsEpochMillis());
38+
assertEquals(serverTestInstance.getInterimResults(), clientInstance.getInterimResults());
3739
}
3840

3941
@Override

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/TransformField.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ public final class TransformField {
3737
public static final ParseField MAX_PAGE_SEARCH_SIZE = new ParseField("max_page_search_size");
3838
public static final ParseField DOCS_PER_SECOND = new ParseField("docs_per_second");
3939
public static final ParseField DATES_AS_EPOCH_MILLIS = new ParseField("dates_as_epoch_millis");
40+
public static final ParseField INTERIM_RESULTS = new ParseField("interim_results");
4041
public static final ParseField FIELD = new ParseField("field");
4142
public static final ParseField SYNC = new ParseField("sync");
4243
public static final ParseField TIME = new ParseField("time");

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/SettingsConfig.java‎

Lines changed: 63 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -34,12 +34,13 @@ public class SettingsConfig implements Writeable, ToXContentObject {
3434
private static final int DEFAULT_MAX_PAGE_SEARCH_SIZE = -1;
3535
private static final float DEFAULT_DOCS_PER_SECOND = -1F;
3636
private static final int DEFAULT_DATES_AS_EPOCH_MILLIS = -1;
37+
private static final int DEFAULT_INTERIM_RESULTS = -1;
3738

3839
private static ConstructingObjectParser<SettingsConfig, Void> createParser(boolean lenient) {
3940
ConstructingObjectParser<SettingsConfig, Void> parser = new ConstructingObjectParser<>(
4041
"transform_config_settings",
4142
lenient,
42-
args -> new SettingsConfig((Integer) args[0], (Float) args[1], (Integer) args[2])
43+
args -> new SettingsConfig((Integer) args[0], (Float) args[1], (Integer) args[2], (Integer) args[3])
4344
);
4445
parser.declareIntOrNull(optionalConstructorArg(), DEFAULT_MAX_PAGE_SEARCH_SIZE, TransformField.MAX_PAGE_SEARCH_SIZE);
4546
parser.declareFloatOrNull(optionalConstructorArg(), DEFAULT_DOCS_PER_SECOND, TransformField.DOCS_PER_SECOND);
@@ -50,25 +51,39 @@ private static ConstructingObjectParser<SettingsConfig, Void> createParser(boole
5051
TransformField.DATES_AS_EPOCH_MILLIS,
5152
ValueType.BOOLEAN_OR_NULL
5253
);
54+
// this boolean requires 4 possible values: true, false, not_specified, default, therefore using a custom parser
55+
parser.declareField(
56+
optionalConstructorArg(),
57+
p -> p.currentToken() == XContentParser.Token.VALUE_NULL ? DEFAULT_INTERIM_RESULTS : p.booleanValue() ? 1 : 0,
58+
TransformField.INTERIM_RESULTS,
59+
ValueType.BOOLEAN_OR_NULL
60+
);
5361
return parser;
5462
}
5563

5664
private final Integer maxPageSearchSize;
5765
private final Float docsPerSecond;
5866
private final Integer datesAsEpochMillis;
67+
private final Integer interimResults;
5968

6069
public SettingsConfig() {
61-
this(null, null, (Integer) null);
70+
this(null, null, (Integer) null, (Integer) null);
6271
}
6372

64-
public SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Boolean datesAsEpochMillis) {
65-
this(maxPageSearchSize, docsPerSecond, datesAsEpochMillis == null ? null : datesAsEpochMillis ? 1 : 0);
73+
public SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Boolean datesAsEpochMillis, Boolean interimResults) {
74+
this(
75+
maxPageSearchSize,
76+
docsPerSecond,
77+
datesAsEpochMillis == null ? null : datesAsEpochMillis ? 1 : 0,
78+
interimResults == null ? null : interimResults ? 1 : 0
79+
);
6680
}
6781

68-
public SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Integer datesAsEpochMillis) {
82+
public SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Integer datesAsEpochMillis, Integer interimResults) {
6983
this.maxPageSearchSize = maxPageSearchSize;
7084
this.docsPerSecond = docsPerSecond;
7185
this.datesAsEpochMillis = datesAsEpochMillis;
86+
this.interimResults = interimResults;
7287
}
7388

7489
public SettingsConfig(final StreamInput in) throws IOException {
@@ -79,6 +94,11 @@ public SettingsConfig(final StreamInput in) throws IOException {
7994
} else {
8095
this.datesAsEpochMillis = DEFAULT_DATES_AS_EPOCH_MILLIS;
8196
}
97+
if (in.getVersion().onOrAfter(Version.CURRENT)) { // TODO: 7.15
98+
this.interimResults = in.readOptionalInt();
99+
} else {
100+
this.interimResults = DEFAULT_INTERIM_RESULTS;
101+
}
82102
}
83103

84104
public Integer getMaxPageSearchSize() {
@@ -97,6 +117,14 @@ public Integer getDatesAsEpochMillisForUpdate() {
97117
return datesAsEpochMillis;
98118
}
99119

120+
public Boolean getInterimResults() {
121+
return interimResults != null ? interimResults > 0 : null;
122+
}
123+
124+
public Integer getInterimResultsForUpdate() {
125+
return interimResults;
126+
}
127+
100128
public ActionRequestValidationException validate(ActionRequestValidationException validationException) {
101129
if (maxPageSearchSize != null && (maxPageSearchSize < 10 || maxPageSearchSize > MultiBucketConsumerService.DEFAULT_MAX_BUCKETS)) {
102130
validationException = addValidationError(
@@ -118,6 +146,9 @@ public void writeTo(StreamOutput out) throws IOException {
118146
if (out.getVersion().onOrAfter(Version.V_7_11_0)) {
119147
out.writeOptionalInt(datesAsEpochMillis);
120148
}
149+
if (out.getVersion().onOrAfter(Version.CURRENT)) { // TODO: 7.15
150+
out.writeOptionalInt(interimResults);
151+
}
121152
}
122153

123154
@Override
@@ -133,6 +164,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
133164
if (datesAsEpochMillis != null && (datesAsEpochMillis.equals(DEFAULT_DATES_AS_EPOCH_MILLIS) == false)) {
134165
builder.field(TransformField.DATES_AS_EPOCH_MILLIS.getPreferredName(), datesAsEpochMillis > 0 ? true : false);
135166
}
167+
if (interimResults != null && (interimResults.equals(DEFAULT_INTERIM_RESULTS) == false)) {
168+
builder.field(TransformField.INTERIM_RESULTS.getPreferredName(), interimResults > 0 ? true : false);
169+
}
136170
builder.endObject();
137171
return builder;
138172
}
@@ -149,12 +183,13 @@ public boolean equals(Object other) {
149183
SettingsConfig that = (SettingsConfig) other;
150184
return Objects.equals(maxPageSearchSize, that.maxPageSearchSize)
151185
&& Objects.equals(docsPerSecond, that.docsPerSecond)
152-
&& Objects.equals(datesAsEpochMillis, that.datesAsEpochMillis);
186+
&& Objects.equals(datesAsEpochMillis, that.datesAsEpochMillis)
187+
&& Objects.equals(interimResults, that.interimResults);
153188
}
154189

155190
@Override
156191
public int hashCode() {
157-
return Objects.hash(maxPageSearchSize, docsPerSecond, datesAsEpochMillis);
192+
return Objects.hash(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, interimResults);
158193
}
159194

160195
@Override
@@ -170,6 +205,7 @@ public static class Builder {
170205
private Integer maxPageSearchSize;
171206
private Float docsPerSecond;
172207
private Integer datesAsEpochMillis;
208+
private Integer interimResults;
173209

174210
/**
175211
* Default builder
@@ -185,6 +221,7 @@ public Builder(SettingsConfig base) {
185221
this.maxPageSearchSize = base.maxPageSearchSize;
186222
this.docsPerSecond = base.docsPerSecond;
187223
this.datesAsEpochMillis = base.datesAsEpochMillis;
224+
this.interimResults = base.interimResults;
188225
}
189226

190227
/**
@@ -231,6 +268,19 @@ public Builder setDatesAsEpochMillis(Boolean datesAsEpochMillis) {
231268
return this;
232269
}
233270

271+
/**
272+
* Whether to write interim results in transform checkpoints.
273+
*
274+
* An explicit `null` resets to default.
275+
*
276+
* @param interimResults true if interim results should be written.
277+
* @return the {@link Builder} with interimResults set.
278+
*/
279+
public Builder setInterimResults(Boolean interimResults) {
280+
this.interimResults = interimResults == null ? DEFAULT_INTERIM_RESULTS : interimResults ? 1 : 0;
281+
return this;
282+
}
283+
234284
/**
235285
* Update settings according to given settings config.
236286
*
@@ -253,12 +303,17 @@ public Builder update(SettingsConfig update) {
253303
? null
254304
: update.getDatesAsEpochMillisForUpdate();
255305
}
306+
if (update.getInterimResultsForUpdate() != null) {
307+
this.interimResults = update.getInterimResultsForUpdate().equals(DEFAULT_INTERIM_RESULTS)
308+
? null
309+
: update.getInterimResultsForUpdate();
310+
}
256311

257312
return this;
258313
}
259314

260315
public SettingsConfig build() {
261-
return new SettingsConfig(maxPageSearchSize, docsPerSecond, datesAsEpochMillis);
316+
return new SettingsConfig(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, interimResults);
262317
}
263318
}
264319
}

0 commit comments

Comments
 (0)