diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/transform/transforms/SettingsConfig.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/transform/transforms/SettingsConfig.java index 9b0054ac78f9c..06e778ad335da 100644 --- a/client/rest-high-level/src/main/java/org/elasticsearch/client/transform/transforms/SettingsConfig.java +++ b/client/rest-high-level/src/main/java/org/elasticsearch/client/transform/transforms/SettingsConfig.java @@ -25,7 +25,7 @@ public class SettingsConfig implements ToXContentObject { private static final ParseField MAX_PAGE_SEARCH_SIZE = new ParseField("max_page_search_size"); private static final ParseField DOCS_PER_SECOND = new ParseField("docs_per_second"); private static final ParseField DATES_AS_EPOCH_MILLIS = new ParseField("dates_as_epoch_millis"); - private static final ParseField INTERIM_RESULTS = new ParseField("interim_results"); + private static final ParseField ALIGN_CHECKPOINTS = new ParseField("align_checkpoints"); private static final int DEFAULT_MAX_PAGE_SEARCH_SIZE = -1; private static final float DEFAULT_DOCS_PER_SECOND = -1F; @@ -33,12 +33,12 @@ public class SettingsConfig implements ToXContentObject { private static final int DEFAULT_DATES_AS_EPOCH_MILLIS = -1; // use an integer as we need to code 4 states: true, false, null (unchanged), default (defined server side) - private static final int DEFAULT_INTERIM_RESULTS = -1; + private static final int DEFAULT_ALIGN_CHECKPOINTS = -1; private final Integer maxPageSearchSize; private final Float docsPerSecond; private final Integer datesAsEpochMillis; - private final Integer interimResults; + private final Integer alignCheckpoints; private static final ConstructingObjectParser PARSER = new ConstructingObjectParser<>( "settings_config", @@ -59,8 +59,8 @@ public class SettingsConfig implements ToXContentObject { // this boolean requires 4 possible values: true, false, not_specified, default, therefore using a custom parser PARSER.declareField( optionalConstructorArg(), - p -> p.currentToken() == XContentParser.Token.VALUE_NULL ? DEFAULT_INTERIM_RESULTS : p.booleanValue() ? 1 : 0, - INTERIM_RESULTS, + p -> p.currentToken() == XContentParser.Token.VALUE_NULL ? DEFAULT_ALIGN_CHECKPOINTS : p.booleanValue() ? 1 : 0, + ALIGN_CHECKPOINTS, ValueType.BOOLEAN_OR_NULL ); } @@ -69,11 +69,11 @@ public static SettingsConfig fromXContent(final XContentParser parser) { return PARSER.apply(parser, null); } - SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Integer datesAsEpochMillis, Integer interimResults) { + SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Integer datesAsEpochMillis, Integer alignCheckpoints) { this.maxPageSearchSize = maxPageSearchSize; this.docsPerSecond = docsPerSecond; this.datesAsEpochMillis = datesAsEpochMillis; - this.interimResults = interimResults; + this.alignCheckpoints = alignCheckpoints; } @Override @@ -100,11 +100,11 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws builder.field(DATES_AS_EPOCH_MILLIS.getPreferredName(), datesAsEpochMillis > 0 ? true : false); } } - if (interimResults != null) { - if (interimResults.equals(DEFAULT_INTERIM_RESULTS)) { - builder.field(INTERIM_RESULTS.getPreferredName(), (Boolean) null); + if (alignCheckpoints != null) { + if (alignCheckpoints.equals(DEFAULT_ALIGN_CHECKPOINTS)) { + builder.field(ALIGN_CHECKPOINTS.getPreferredName(), (Boolean) null); } else { - builder.field(INTERIM_RESULTS.getPreferredName(), interimResults > 0 ? true : false); + builder.field(ALIGN_CHECKPOINTS.getPreferredName(), alignCheckpoints > 0 ? true : false); } } builder.endObject(); @@ -123,8 +123,8 @@ public Boolean getDatesAsEpochMillis() { return datesAsEpochMillis != null ? datesAsEpochMillis > 0 : null; } - public Boolean getInterimResults() { - return interimResults != null ? interimResults > 0 : null; + public Boolean getAlignCheckpoints() { + return alignCheckpoints != null ? alignCheckpoints > 0 : null; } @Override @@ -140,12 +140,12 @@ public boolean equals(Object other) { return Objects.equals(maxPageSearchSize, that.maxPageSearchSize) && Objects.equals(docsPerSecond, that.docsPerSecond) && Objects.equals(datesAsEpochMillis, that.datesAsEpochMillis) - && Objects.equals(interimResults, that.interimResults); + && Objects.equals(alignCheckpoints, that.alignCheckpoints); } @Override public int hashCode() { - return Objects.hash(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, interimResults); + return Objects.hash(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, alignCheckpoints); } public static Builder builder() { @@ -156,7 +156,7 @@ public static class Builder { private Integer maxPageSearchSize; private Float docsPerSecond; private Integer datesAsEpochMillis; - private Integer interimResults; + private Integer alignCheckpoints; /** * Sets the paging maximum paging maxPageSearchSize that transform can use when @@ -203,20 +203,20 @@ public Builder setDatesAsEpochMillis(Boolean datesAsEpochMillis) { } /** - * Whether to write interim results in transform checkpoints. + * Whether to align transform checkpoint ranges with date histogram interval. * * An explicit `null` resets to default. * - * @param interimResults true if interim results should be written. - * @return the {@link Builder} with interimResults set. + * @param alignCheckpoints true if checkpoint ranges should be aligned with date histogram interval. + * @return the {@link Builder} with alignCheckpoints set. */ - public Builder setInterimResults(Boolean interimResults) { - this.interimResults = interimResults == null ? DEFAULT_INTERIM_RESULTS : interimResults ? 1 : 0; + public Builder setAlignCheckpoints(Boolean alignCheckpoints) { + this.alignCheckpoints = alignCheckpoints == null ? DEFAULT_ALIGN_CHECKPOINTS : alignCheckpoints ? 1 : 0; return this; } public SettingsConfig build() { - return new SettingsConfig(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, interimResults); + return new SettingsConfig(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, alignCheckpoints); } } } diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/transform/transforms/SettingsConfigTests.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/transform/transforms/SettingsConfigTests.java index 9d2776415b432..30e1d63b9e877 100644 --- a/client/rest-high-level/src/test/java/org/elasticsearch/client/transform/transforms/SettingsConfigTests.java +++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/transform/transforms/SettingsConfigTests.java @@ -73,7 +73,7 @@ public void testExplicitNullOnWriteParser() throws IOException { assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set")); assertNull(settingsAsMap.getOrDefault("docs_per_second", "not_set")); assertThat(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"), equalTo("not_set")); - assertThat(settingsAsMap.getOrDefault("interim_results", "not_set"), equalTo("not_set")); + assertThat(settingsAsMap.getOrDefault("align_checkpoints", "not_set"), equalTo("not_set")); config = fromString("{\"dates_as_epoch_millis\" : null}"); assertFalse(config.getDatesAsEpochMillis()); @@ -82,16 +82,16 @@ public void testExplicitNullOnWriteParser() throws IOException { assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set")); assertThat(settingsAsMap.getOrDefault("docs_per_second", "not_set"), equalTo("not_set")); assertNull(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set")); - assertThat(settingsAsMap.getOrDefault("interim_results", "not_set"), equalTo("not_set")); + assertThat(settingsAsMap.getOrDefault("align_checkpoints", "not_set"), equalTo("not_set")); - config = fromString("{\"interim_results\" : null}"); - assertFalse(config.getInterimResults()); + config = fromString("{\"align_checkpoints\" : null}"); + assertFalse(config.getAlignCheckpoints()); settingsAsMap = xContentToMap(config); assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set")); assertThat(settingsAsMap.getOrDefault("docs_per_second", "not_set"), equalTo("not_set")); assertThat(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"), equalTo("not_set")); - assertNull(settingsAsMap.getOrDefault("interim_results", "not_set")); + assertNull(settingsAsMap.getOrDefault("align_checkpoints", "not_set")); } public void testExplicitNullOnWriteBuilder() throws IOException { @@ -103,12 +103,12 @@ public void testExplicitNullOnWriteBuilder() throws IOException { assertNull(settingsAsMap.getOrDefault("max_page_search_size", "not_set")); assertThat(settingsAsMap.getOrDefault("docs_per_second", "not_set"), equalTo("not_set")); assertThat(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"), equalTo("not_set")); - assertThat(settingsAsMap.getOrDefault("interim_results", "not_set"), equalTo("not_set")); + assertThat(settingsAsMap.getOrDefault("align_checkpoints", "not_set"), equalTo("not_set")); SettingsConfig emptyConfig = new SettingsConfig.Builder().build(); assertNull(emptyConfig.getMaxPageSearchSize()); assertNull(emptyConfig.getDatesAsEpochMillis()); - assertNull(emptyConfig.getInterimResults()); + assertNull(emptyConfig.getAlignCheckpoints()); settingsAsMap = xContentToMap(emptyConfig); assertTrue(settingsAsMap.isEmpty()); @@ -120,7 +120,7 @@ public void testExplicitNullOnWriteBuilder() throws IOException { assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set")); assertNull(settingsAsMap.getOrDefault("docs_per_second", "not_set")); assertThat(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"), equalTo("not_set")); - assertThat(settingsAsMap.getOrDefault("interim_results", "not_set"), equalTo("not_set")); + assertThat(settingsAsMap.getOrDefault("align_checkpoints", "not_set"), equalTo("not_set")); config = new SettingsConfig.Builder().setDatesAsEpochMillis(null).build(); // returns false, however it's `null` as in "use default", checked next @@ -130,17 +130,17 @@ public void testExplicitNullOnWriteBuilder() throws IOException { assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set")); assertThat(settingsAsMap.getOrDefault("docs_per_second", "not_set"), equalTo("not_set")); assertNull(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set")); - assertThat(settingsAsMap.getOrDefault("interim_results", "not_set"), equalTo("not_set")); + assertThat(settingsAsMap.getOrDefault("align_checkpoints", "not_set"), equalTo("not_set")); - config = new SettingsConfig.Builder().setInterimResults(null).build(); + config = new SettingsConfig.Builder().setAlignCheckpoints(null).build(); // returns false, however it's `null` as in "use default", checked next - assertFalse(config.getInterimResults()); + assertFalse(config.getAlignCheckpoints()); settingsAsMap = xContentToMap(config); assertThat(settingsAsMap.getOrDefault("max_page_search_size", "not_set"), equalTo("not_set")); assertThat(settingsAsMap.getOrDefault("docs_per_second", "not_set"), equalTo("not_set")); assertThat(settingsAsMap.getOrDefault("dates_as_epoch_millis", "not_set"), equalTo("not_set")); - assertNull(settingsAsMap.getOrDefault("interim_results", "not_set")); + assertNull(settingsAsMap.getOrDefault("align_checkpoints", "not_set")); } private Map xContentToMap(ToXContent xcontent) throws IOException { diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/transform/transforms/hlrc/SettingsConfigTests.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/transform/transforms/hlrc/SettingsConfigTests.java index 6ac3ee81238aa..a42f7e5ca6cea 100644 --- a/client/rest-high-level/src/test/java/org/elasticsearch/client/transform/transforms/hlrc/SettingsConfigTests.java +++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/transform/transforms/hlrc/SettingsConfigTests.java @@ -35,7 +35,7 @@ public static void assertHlrcEquals( assertEquals(serverTestInstance.getMaxPageSearchSize(), clientInstance.getMaxPageSearchSize()); assertEquals(serverTestInstance.getDocsPerSecond(), clientInstance.getDocsPerSecond()); assertEquals(serverTestInstance.getDatesAsEpochMillis(), clientInstance.getDatesAsEpochMillis()); - assertEquals(serverTestInstance.getInterimResults(), clientInstance.getInterimResults()); + assertEquals(serverTestInstance.getAlignCheckpoints(), clientInstance.getAlignCheckpoints()); } @Override diff --git a/docs/reference/rest-api/common-parms.asciidoc b/docs/reference/rest-api/common-parms.asciidoc index f2a80f40f054e..372ae839fb809 100644 --- a/docs/reference/rest-api/common-parms.asciidoc +++ b/docs/reference/rest-api/common-parms.asciidoc @@ -1005,6 +1005,14 @@ throttles the {transform} by adding a wait time between search requests. The default value is `null`, which disables throttling. end::transform-settings-docs-per-second[] +tag::transform-settings-align-checkpoints[] +Specifies whether the transform checkpoint ranges should be optimized for performance. +Such optimization can align checkpoint ranges with date histogram interval when date histogram +is specified as a group source in the transform config. As an effect, less document updates in the +destination index will be performed thus improving overall performance. +The default value is `true`, which means the checkpoint ranges will be optimized if possible. +end::transform-settings-align-checkpoints[] + tag::transform-settings-max-page-search-size[] Defines the initial page size to use for the composite aggregation for each checkpoint. If circuit breaker exceptions occur, the page size is dynamically diff --git a/docs/reference/transform/apis/preview-transform.asciidoc b/docs/reference/transform/apis/preview-transform.asciidoc index 32a174d9460e6..817b7c0849a2b 100644 --- a/docs/reference/transform/apis/preview-transform.asciidoc +++ b/docs/reference/transform/apis/preview-transform.asciidoc @@ -204,6 +204,9 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings] `docs_per_second`::: (Optional, float) include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-docs-per-second] +`align_checkpoints`::: +(Optional, boolean) +include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-align-checkpoints] `max_page_search_size`::: (Optional, integer) include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-max-page-search-size] diff --git a/docs/reference/transform/apis/put-transform.asciidoc b/docs/reference/transform/apis/put-transform.asciidoc index f2c397a6ebee8..3a7f2771b6959 100644 --- a/docs/reference/transform/apis/put-transform.asciidoc +++ b/docs/reference/transform/apis/put-transform.asciidoc @@ -186,6 +186,9 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-dat `docs_per_second`::: (Optional, float) include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-docs-per-second] +`align_checkpoints`::: +(Optional, boolean) +include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-align-checkpoints] `max_page_search_size`::: (Optional, integer) include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-max-page-search-size] diff --git a/docs/reference/transform/apis/update-transform.asciidoc b/docs/reference/transform/apis/update-transform.asciidoc index 8dce1017b3650..c601bd31f47fb 100644 --- a/docs/reference/transform/apis/update-transform.asciidoc +++ b/docs/reference/transform/apis/update-transform.asciidoc @@ -138,6 +138,9 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-dat `docs_per_second`::: (Optional, float) include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-docs-per-second] +`align_checkpoints`::: +(Optional, boolean) +include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-align-checkpoints] `max_page_search_size`::: (Optional, integer) include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-max-page-search-size] diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/TransformField.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/TransformField.java index 90c9e244d0e01..0a616a2c51c8c 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/TransformField.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/TransformField.java @@ -37,7 +37,7 @@ public final class TransformField { public static final ParseField MAX_PAGE_SEARCH_SIZE = new ParseField("max_page_search_size"); public static final ParseField DOCS_PER_SECOND = new ParseField("docs_per_second"); public static final ParseField DATES_AS_EPOCH_MILLIS = new ParseField("dates_as_epoch_millis"); - public static final ParseField INTERIM_RESULTS = new ParseField("interim_results"); + public static final ParseField ALIGN_CHECKPOINTS = new ParseField("align_checkpoints"); public static final ParseField FIELD = new ParseField("field"); public static final ParseField SYNC = new ParseField("sync"); public static final ParseField TIME = new ParseField("time"); diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/SettingsConfig.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/SettingsConfig.java index bef7adc6f9e83..a62a5f67cdf11 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/SettingsConfig.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/SettingsConfig.java @@ -34,7 +34,7 @@ public class SettingsConfig implements Writeable, ToXContentObject { private static final int DEFAULT_MAX_PAGE_SEARCH_SIZE = -1; private static final float DEFAULT_DOCS_PER_SECOND = -1F; private static final int DEFAULT_DATES_AS_EPOCH_MILLIS = -1; - private static final int DEFAULT_INTERIM_RESULTS = -1; + private static final int DEFAULT_ALIGN_CHECKPOINTS = -1; private static ConstructingObjectParser createParser(boolean lenient) { ConstructingObjectParser parser = new ConstructingObjectParser<>( @@ -54,8 +54,8 @@ private static ConstructingObjectParser createParser(boole // this boolean requires 4 possible values: true, false, not_specified, default, therefore using a custom parser parser.declareField( optionalConstructorArg(), - p -> p.currentToken() == XContentParser.Token.VALUE_NULL ? DEFAULT_INTERIM_RESULTS : p.booleanValue() ? 1 : 0, - TransformField.INTERIM_RESULTS, + p -> p.currentToken() == XContentParser.Token.VALUE_NULL ? DEFAULT_ALIGN_CHECKPOINTS : p.booleanValue() ? 1 : 0, + TransformField.ALIGN_CHECKPOINTS, ValueType.BOOLEAN_OR_NULL ); return parser; @@ -64,26 +64,26 @@ private static ConstructingObjectParser createParser(boole private final Integer maxPageSearchSize; private final Float docsPerSecond; private final Integer datesAsEpochMillis; - private final Integer interimResults; + private final Integer alignCheckpoints; public SettingsConfig() { this(null, null, (Integer) null, (Integer) null); } - public SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Boolean datesAsEpochMillis, Boolean interimResults) { + public SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Boolean datesAsEpochMillis, Boolean alignCheckpoints) { this( maxPageSearchSize, docsPerSecond, datesAsEpochMillis == null ? null : datesAsEpochMillis ? 1 : 0, - interimResults == null ? null : interimResults ? 1 : 0 + alignCheckpoints == null ? null : alignCheckpoints ? 1 : 0 ); } - public SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Integer datesAsEpochMillis, Integer interimResults) { + public SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond, Integer datesAsEpochMillis, Integer alignCheckpoints) { this.maxPageSearchSize = maxPageSearchSize; this.docsPerSecond = docsPerSecond; this.datesAsEpochMillis = datesAsEpochMillis; - this.interimResults = interimResults; + this.alignCheckpoints = alignCheckpoints; } public SettingsConfig(final StreamInput in) throws IOException { @@ -95,9 +95,9 @@ public SettingsConfig(final StreamInput in) throws IOException { this.datesAsEpochMillis = DEFAULT_DATES_AS_EPOCH_MILLIS; } if (in.getVersion().onOrAfter(Version.V_7_15_0)) { - this.interimResults = in.readOptionalInt(); + this.alignCheckpoints = in.readOptionalInt(); } else { - this.interimResults = DEFAULT_INTERIM_RESULTS; + this.alignCheckpoints = DEFAULT_ALIGN_CHECKPOINTS; } } @@ -117,12 +117,12 @@ public Integer getDatesAsEpochMillisForUpdate() { return datesAsEpochMillis; } - public Boolean getInterimResults() { - return interimResults != null ? interimResults > 0 : null; + public Boolean getAlignCheckpoints() { + return alignCheckpoints != null ? (alignCheckpoints > 0) || (alignCheckpoints == DEFAULT_ALIGN_CHECKPOINTS) : null; } - public Integer getInterimResultsForUpdate() { - return interimResults; + public Integer getAlignCheckpointsForUpdate() { + return alignCheckpoints; } public ActionRequestValidationException validate(ActionRequestValidationException validationException) { @@ -147,7 +147,7 @@ public void writeTo(StreamOutput out) throws IOException { out.writeOptionalInt(datesAsEpochMillis); } if (out.getVersion().onOrAfter(Version.V_7_15_0)) { - out.writeOptionalInt(interimResults); + out.writeOptionalInt(alignCheckpoints); } } @@ -164,8 +164,8 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws if (datesAsEpochMillis != null && (datesAsEpochMillis.equals(DEFAULT_DATES_AS_EPOCH_MILLIS) == false)) { builder.field(TransformField.DATES_AS_EPOCH_MILLIS.getPreferredName(), datesAsEpochMillis > 0 ? true : false); } - if (interimResults != null && (interimResults.equals(DEFAULT_INTERIM_RESULTS) == false)) { - builder.field(TransformField.INTERIM_RESULTS.getPreferredName(), interimResults > 0 ? true : false); + if (alignCheckpoints != null && (alignCheckpoints.equals(DEFAULT_ALIGN_CHECKPOINTS) == false)) { + builder.field(TransformField.ALIGN_CHECKPOINTS.getPreferredName(), alignCheckpoints > 0 ? true : false); } builder.endObject(); return builder; @@ -184,12 +184,12 @@ public boolean equals(Object other) { return Objects.equals(maxPageSearchSize, that.maxPageSearchSize) && Objects.equals(docsPerSecond, that.docsPerSecond) && Objects.equals(datesAsEpochMillis, that.datesAsEpochMillis) - && Objects.equals(interimResults, that.interimResults); + && Objects.equals(alignCheckpoints, that.alignCheckpoints); } @Override public int hashCode() { - return Objects.hash(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, interimResults); + return Objects.hash(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, alignCheckpoints); } @Override @@ -205,7 +205,7 @@ public static class Builder { private Integer maxPageSearchSize; private Float docsPerSecond; private Integer datesAsEpochMillis; - private Integer interimResults; + private Integer alignCheckpoints; /** * Default builder @@ -221,7 +221,7 @@ public Builder(SettingsConfig base) { this.maxPageSearchSize = base.maxPageSearchSize; this.docsPerSecond = base.docsPerSecond; this.datesAsEpochMillis = base.datesAsEpochMillis; - this.interimResults = base.interimResults; + this.alignCheckpoints = base.alignCheckpoints; } /** @@ -269,15 +269,15 @@ public Builder setDatesAsEpochMillis(Boolean datesAsEpochMillis) { } /** - * Whether to write interim results in transform checkpoints. + * Whether to align transform checkpoint ranges with date histogram interval. * * An explicit `null` resets to default. * - * @param interimResults true if interim results should be written. - * @return the {@link Builder} with interimResults set. + * @param alignCheckpoints true if checkpoint ranges should be aligned with date histogram interval. + * @return the {@link Builder} with alignCheckpoints set. */ - public Builder setInterimResults(Boolean interimResults) { - this.interimResults = interimResults == null ? DEFAULT_INTERIM_RESULTS : interimResults ? 1 : 0; + public Builder setAlignCheckpoints(Boolean alignCheckpoints) { + this.alignCheckpoints = alignCheckpoints == null ? DEFAULT_ALIGN_CHECKPOINTS : alignCheckpoints ? 1 : 0; return this; } @@ -303,17 +303,17 @@ public Builder update(SettingsConfig update) { ? null : update.getDatesAsEpochMillisForUpdate(); } - if (update.getInterimResultsForUpdate() != null) { - this.interimResults = update.getInterimResultsForUpdate().equals(DEFAULT_INTERIM_RESULTS) + if (update.getAlignCheckpointsForUpdate() != null) { + this.alignCheckpoints = update.getAlignCheckpointsForUpdate().equals(DEFAULT_ALIGN_CHECKPOINTS) ? null - : update.getInterimResultsForUpdate(); + : update.getAlignCheckpointsForUpdate(); } return this; } public SettingsConfig build() { - return new SettingsConfig(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, interimResults); + return new SettingsConfig(maxPageSearchSize, docsPerSecond, datesAsEpochMillis, alignCheckpoints); } } } diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfig.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfig.java index ad044cce1c3fc..a365e8908c842 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfig.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfig.java @@ -508,7 +508,7 @@ public static TransformConfig rewriteForUpdate(final TransformConfig transformCo // quick check if a rewrite is required, if none found just return the original // a failing quick check, does not mean a rewrite is necessary if (transformConfig.getVersion() != null - && transformConfig.getVersion().onOrAfter(Version.V_7_11_0) + && transformConfig.getVersion().onOrAfter(Version.V_7_15_0) && (transformConfig.getPivotConfig() == null || transformConfig.getPivotConfig().getMaxPageSearchSize() == null)) { return transformConfig; } @@ -539,7 +539,7 @@ private static TransformConfig applyRewriteForUpdate(Builder builder) { maxPageSearchSize, builder.getSettings().getDocsPerSecond(), builder.getSettings().getDatesAsEpochMillis(), - builder.getSettings().getInterimResults() + builder.getSettings().getAlignCheckpoints() ) ); } @@ -551,18 +551,18 @@ private static TransformConfig applyRewriteForUpdate(Builder builder) { builder.getSettings().getMaxPageSearchSize(), builder.getSettings().getDocsPerSecond(), true, - builder.getSettings().getInterimResults()) + builder.getSettings().getAlignCheckpoints()) ); } - // 3. set interim_results to true for transforms < 7.15 to keep BWC + // 3. set align_checkpoints to false for transforms < 7.15 to keep BWC if (builder.getVersion() != null && builder.getVersion().before(Version.V_7_15_0)) { builder.setSettings( new SettingsConfig( builder.getSettings().getMaxPageSearchSize(), builder.getSettings().getDocsPerSecond(), builder.getSettings().getDatesAsEpochMillis(), - true) + false) ); } diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/SettingsConfigTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/SettingsConfigTests.java index 684c2950ea5f2..94db9471dd9d5 100644 --- a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/SettingsConfigTests.java +++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/SettingsConfigTests.java @@ -80,12 +80,16 @@ public void testExplicitNullParsing() throws IOException { assertThat(fromString("{\"dates_as_epoch_millis\" : null}").getDatesAsEpochMillisForUpdate(), equalTo(-1)); assertNull(fromString("{}").getDatesAsEpochMillisForUpdate()); - assertThat(fromString("{\"interim_results\" : null}").getInterimResultsForUpdate(), equalTo(-1)); - assertNull(fromString("{}").getInterimResultsForUpdate()); + assertThat(fromString("{\"align_checkpoints\" : null}").getAlignCheckpointsForUpdate(), equalTo(-1)); + assertNull(fromString("{}").getAlignCheckpointsForUpdate()); } public void testUpdateUsingBuilder() throws IOException { - SettingsConfig config = fromString("{\"max_page_search_size\" : 10000, \"docs_per_second\" :42, \"dates_as_epoch_millis\": true}"); + SettingsConfig config = + fromString("{\"max_page_search_size\" : 10000, " + + "\"docs_per_second\" :42, " + + "\"dates_as_epoch_millis\": true, " + + "\"align_checkpoints\": false}"); SettingsConfig.Builder builder = new SettingsConfig.Builder(config); builder.update(fromString("{\"max_page_search_size\" : 100}")); @@ -93,16 +97,23 @@ public void testUpdateUsingBuilder() throws IOException { assertThat(builder.build().getMaxPageSearchSize(), equalTo(100)); assertThat(builder.build().getDocsPerSecond(), equalTo(42F)); assertThat(builder.build().getDatesAsEpochMillisForUpdate(), equalTo(1)); + assertThat(builder.build().getAlignCheckpointsForUpdate(), equalTo(0)); builder.update(fromString("{\"max_page_search_size\" : null}")); assertNull(builder.build().getMaxPageSearchSize()); assertThat(builder.build().getDocsPerSecond(), equalTo(42F)); assertThat(builder.build().getDatesAsEpochMillisForUpdate(), equalTo(1)); + assertThat(builder.build().getAlignCheckpointsForUpdate(), equalTo(0)); - builder.update(fromString("{\"max_page_search_size\" : 77, \"docs_per_second\" :null, \"dates_as_epoch_millis\": null}")); + builder.update( + fromString("{\"max_page_search_size\" : 77, " + + "\"docs_per_second\" :null, " + + "\"dates_as_epoch_millis\": null, " + + "\"align_checkpoints\": null}")); assertThat(builder.build().getMaxPageSearchSize(), equalTo(77)); assertNull(builder.build().getDocsPerSecond()); assertNull(builder.build().getDatesAsEpochMillisForUpdate()); + assertNull(builder.build().getAlignCheckpointsForUpdate()); } public void testOmmitDefaultsOnWriteParser() throws IOException { @@ -130,6 +141,12 @@ public void testOmmitDefaultsOnWriteParser() throws IOException { settingsAsMap = xContentToMap(config); assertTrue(settingsAsMap.isEmpty()); + + config = fromString("{\"align_checkpoints\" : null}"); + assertThat(config.getAlignCheckpointsForUpdate(), equalTo(-1)); + + settingsAsMap = xContentToMap(config); + assertTrue(settingsAsMap.isEmpty()); } public void testOmmitDefaultsOnWriteBuilder() throws IOException { @@ -157,6 +174,12 @@ public void testOmmitDefaultsOnWriteBuilder() throws IOException { settingsAsMap = xContentToMap(config); assertTrue(settingsAsMap.isEmpty()); + + config = new SettingsConfig.Builder().setAlignCheckpoints(null).build(); + assertThat(config.getAlignCheckpointsForUpdate(), equalTo(-1)); + + settingsAsMap = xContentToMap(config); + assertTrue(settingsAsMap.isEmpty()); } private Map xContentToMap(ToXContent xcontent) throws IOException { diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfigTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfigTests.java index 20b08479637c1..8ad9f550e8c7c 100644 --- a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfigTests.java +++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfigTests.java @@ -463,11 +463,50 @@ public void testRewriteForUpdate() throws IOException { assertNotNull(transformConfigRewritten.getSettings().getMaxPageSearchSize()); assertEquals(111L, transformConfigRewritten.getSettings().getMaxPageSearchSize().longValue()); assertTrue(transformConfigRewritten.getSettings().getDatesAsEpochMillis()); + assertFalse(transformConfigRewritten.getSettings().getAlignCheckpoints()); assertWarnings("[max_page_search_size] is deprecated inside pivot please use settings instead"); assertEquals(Version.CURRENT, transformConfigRewritten.getVersion()); } + public void testRewriteForUpdateAlignCheckpoints() throws IOException { + String pivotTransform = "{" + + " \"id\" : \"body_id\"," + + " \"source\" : {\"index\":\"src\"}," + + " \"dest\" : {\"index\": \"dest\"}," + + " \"pivot\" : {" + + " \"group_by\": {" + + " \"id\": {" + + " \"terms\": {" + + " \"field\": \"id\"" + + "} } }," + + " \"aggs\": {" + + " \"avg\": {" + + " \"avg\": {" + + " \"field\": \"points\"" + + "} } }" + + "}," + + " \"version\" : \"" + + Version.V_7_12_0.toString() + + "\"" + + "}"; + + TransformConfig transformConfig = createTransformConfigFromString(pivotTransform, "body_id", true); + TransformConfig transformConfigRewritten = TransformConfig.rewriteForUpdate(transformConfig); + assertEquals(Version.CURRENT, transformConfigRewritten.getVersion()); + assertFalse(transformConfigRewritten.getSettings().getAlignCheckpoints()); + + TransformConfig explicitFalseAfter715 = new TransformConfig.Builder(transformConfig) + .setSettings(new SettingsConfig.Builder(transformConfigRewritten.getSettings()).setAlignCheckpoints(false).build()) + .setVersion(Version.V_7_15_0) + .build(); + transformConfigRewritten = TransformConfig.rewriteForUpdate(explicitFalseAfter715); + + assertFalse(transformConfigRewritten.getSettings().getAlignCheckpoints()); + // The config is not rewritten. + assertEquals(Version.V_7_15_0, transformConfigRewritten.getVersion()); + } + public void testRewriteForUpdateMaxPageSizeSearchConflicting() throws IOException { String pivotTransform = "{" + " \"id\" : \"body_id\"," @@ -531,14 +570,25 @@ public void testRewriteForBWCOfDateNormalization() throws IOException { assertTrue(transformConfigRewritten.getSettings().getDatesAsEpochMillis()); assertEquals(Version.CURRENT, transformConfigRewritten.getVersion()); - TransformConfig explicitTrueAfter711 = new TransformConfig.Builder(transformConfig).setSettings( - new SettingsConfig.Builder(transformConfigRewritten.getSettings()).setDatesAsEpochMillis(true).build() - ).setVersion(Version.V_7_11_0).build(); - + TransformConfig explicitTrueAfter711 = new TransformConfig.Builder(transformConfig) + .setSettings(new SettingsConfig.Builder(transformConfigRewritten.getSettings()).setDatesAsEpochMillis(true).build()) + .setVersion(Version.V_7_11_0) + .build(); transformConfigRewritten = TransformConfig.rewriteForUpdate(explicitTrueAfter711); assertTrue(transformConfigRewritten.getSettings().getDatesAsEpochMillis()); - assertEquals(Version.V_7_11_0, transformConfigRewritten.getVersion()); + // The config is still being rewritten due to "settings.align_checkpoints". + assertEquals(Version.CURRENT, transformConfigRewritten.getVersion()); + + TransformConfig explicitTrueAfter715 = new TransformConfig.Builder(transformConfig) + .setSettings(new SettingsConfig.Builder(transformConfigRewritten.getSettings()).setDatesAsEpochMillis(true).build()) + .setVersion(Version.V_7_15_0) + .build(); + transformConfigRewritten = TransformConfig.rewriteForUpdate(explicitTrueAfter715); + + assertTrue(transformConfigRewritten.getSettings().getDatesAsEpochMillis()); + // The config is not rewritten. + assertEquals(Version.V_7_15_0, transformConfigRewritten.getVersion()); } public void testGetAdditionalSourceDestValidations_WithNoRuntimeMappings() throws IOException { diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfigUpdateTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfigUpdateTests.java index 7e4266bd966e7..01edf06271617 100644 --- a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfigUpdateTests.java +++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfigUpdateTests.java @@ -159,14 +159,14 @@ public void testApplySettings() { assertThat(updatedConfig.getSettings().getMaxPageSearchSize(), equalTo(4_000)); assertThat(updatedConfig.getSettings().getDocsPerSecond(), equalTo(config.getSettings().getDocsPerSecond())); assertThat(updatedConfig.getSettings().getDatesAsEpochMillis(), equalTo(config.getSettings().getDatesAsEpochMillis())); - assertThat(updatedConfig.getSettings().getInterimResults(), equalTo(config.getSettings().getInterimResults())); + assertThat(updatedConfig.getSettings().getAlignCheckpoints(), equalTo(config.getSettings().getAlignCheckpoints())); update = new TransformConfigUpdate(null, null, null, null, null, new SettingsConfig(null, 43.244F, (Boolean) null, null), null); updatedConfig = update.apply(updatedConfig); assertThat(updatedConfig.getSettings().getMaxPageSearchSize(), equalTo(4_000)); assertThat(updatedConfig.getSettings().getDocsPerSecond(), equalTo(43.244F)); assertThat(updatedConfig.getSettings().getDatesAsEpochMillis(), equalTo(config.getSettings().getDatesAsEpochMillis())); - assertThat(updatedConfig.getSettings().getInterimResults(), equalTo(config.getSettings().getInterimResults())); + assertThat(updatedConfig.getSettings().getAlignCheckpoints(), equalTo(config.getSettings().getAlignCheckpoints())); // now reset to default using the magic -1 update = new TransformConfigUpdate(null, null, null, null, null, new SettingsConfig(-1, null, (Boolean) null, null), null); @@ -174,14 +174,14 @@ public void testApplySettings() { assertNull(updatedConfig.getSettings().getMaxPageSearchSize()); assertThat(updatedConfig.getSettings().getDocsPerSecond(), equalTo(43.244F)); assertThat(updatedConfig.getSettings().getDatesAsEpochMillis(), equalTo(config.getSettings().getDatesAsEpochMillis())); - assertThat(updatedConfig.getSettings().getInterimResults(), equalTo(config.getSettings().getInterimResults())); + assertThat(updatedConfig.getSettings().getAlignCheckpoints(), equalTo(config.getSettings().getAlignCheckpoints())); update = new TransformConfigUpdate(null, null, null, null, null, new SettingsConfig(-1, -1F, (Boolean) null, null), null); updatedConfig = update.apply(updatedConfig); assertNull(updatedConfig.getSettings().getMaxPageSearchSize()); assertNull(updatedConfig.getSettings().getDocsPerSecond()); assertThat(updatedConfig.getSettings().getDatesAsEpochMillis(), equalTo(config.getSettings().getDatesAsEpochMillis())); - assertThat(updatedConfig.getSettings().getInterimResults(), equalTo(config.getSettings().getInterimResults())); + assertThat(updatedConfig.getSettings().getAlignCheckpoints(), equalTo(config.getSettings().getAlignCheckpoints())); } public void testApplyWithSyncChange() { diff --git a/x-pack/plugin/core/src/test/resources/rest-api-spec/schema/transform_config.schema.json b/x-pack/plugin/core/src/test/resources/rest-api-spec/schema/transform_config.schema.json index 67df081561618..a7cb9fb13bc15 100644 --- a/x-pack/plugin/core/src/test/resources/rest-api-spec/schema/transform_config.schema.json +++ b/x-pack/plugin/core/src/test/resources/rest-api-spec/schema/transform_config.schema.json @@ -180,11 +180,11 @@ "title": "docs per second", "type": "number" }, - "interim_results": { - "$id": "#root/settings/interim_results", - "title": "interim results", + "align_checkpoints": { + "$id": "#root/settings/align_checkpoints", + "title": "align checkpoints", "type": "boolean", - "default": false + "default": true }, "max_page_search_size": { "$id": "#root/settings/max_page_search_size", diff --git a/x-pack/plugin/transform/qa/multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformIT.java b/x-pack/plugin/transform/qa/multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformIT.java index e2f694430152a..c539f98f91a9a 100644 --- a/x-pack/plugin/transform/qa/multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformIT.java +++ b/x-pack/plugin/transform/qa/multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformIT.java @@ -136,6 +136,7 @@ public void testContinuousTransformCrud() throws Exception { createTransformConfigBuilder(transformId, "reviews-by-user-business-day", QueryBuilders.matchAllQuery(), indexName) .setPivotConfig(createPivotConfig(groups, aggs)) .setSyncConfig(TimeSyncConfig.builder().setField("timestamp").setDelay(TimeValue.timeValueSeconds(1)).build()) + .setSettings(SettingsConfig.builder().setAlignCheckpoints(false).build()) .build(); assertTrue(putTransform(config, RequestOptions.DEFAULT).isAcknowledged()); @@ -311,7 +312,8 @@ public void testStopWaitForCheckpoint() throws Exception { TransformStats stateAndStats = getTransformStats(config.getId()).getTransformsStats().get(0); assertThat(stateAndStats.getState(), equalTo(TransformStats.State.STOPPED)); - assertThat(stateAndStats.getIndexerStats().getDocumentsIndexed(), greaterThan(1000L)); + // Despite indexing new documents into the source index, the number of documents in the destination index stays the same. + assertThat(stateAndStats.getIndexerStats().getDocumentsIndexed(), equalTo(1000L)); assertTrue(stopTransform(transformId).isAcknowledged()); deleteTransform(config.getId()); @@ -337,7 +339,7 @@ public void testContinuousTransformRethrottle() throws Exception { .setPivotConfig(createPivotConfig(groups, aggs)) .setSyncConfig(TimeSyncConfig.builder().setField("timestamp").setDelay(TimeValue.timeValueSeconds(1)).build()) // set requests per second and page size low enough to fail the test if update does not succeed, - .setSettings(SettingsConfig.builder().setRequestsPerSecond(1F).setMaxPageSearchSize(10).build()) + .setSettings(SettingsConfig.builder().setRequestsPerSecond(1F).setMaxPageSearchSize(10).setAlignCheckpoints(false).build()) .build(); assertTrue(putTransform(config, RequestOptions.DEFAULT).isAcknowledged()); diff --git a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/checkpoint/TimeBasedCheckpointProvider.java b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/checkpoint/TimeBasedCheckpointProvider.java index 40c63907773f6..f1737b22d15bd 100644 --- a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/checkpoint/TimeBasedCheckpointProvider.java +++ b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/checkpoint/TimeBasedCheckpointProvider.java @@ -120,7 +120,7 @@ public void createNextCheckpoint(final TransformCheckpoint lastCheckpoint, final * @return function aligning the given timestamp with date histogram interval */ private static Function createAlignTimestampFunction(TransformConfig transformConfig) { - if (Boolean.FALSE.equals(transformConfig.getSettings().getInterimResults()) == false) { + if (Boolean.FALSE.equals(transformConfig.getSettings().getAlignCheckpoints())) { return identity(); } if (transformConfig.getPivotConfig() == null) { diff --git a/x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/checkpoint/TimeBasedCheckpointProviderTests.java b/x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/checkpoint/TimeBasedCheckpointProviderTests.java index e2aca8d8fe8aa..132e8cb8e56aa 100644 --- a/x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/checkpoint/TimeBasedCheckpointProviderTests.java +++ b/x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/checkpoint/TimeBasedCheckpointProviderTests.java @@ -289,8 +289,19 @@ public SingleGroupSource get() { AggregationConfigTests.randomAggregationConfig(), null // deprecated ); + SettingsConfig.Builder settingsConfigBuilder = new SettingsConfig.Builder(); + if (randomBoolean()) { + settingsConfigBuilder.setAlignCheckpoints( + randomBoolean() + // Set align_checkpoints setting explicitly to "true". + ? true + // Set align_checkpoints setting explicitly to "null". This will be interpreted as "true". + : null); + } else { + // Leave align_checkpoints setting unset. This will be interpreted as "true". + } return new TransformConfig.Builder(TransformConfigTests.randomTransformConfig(transformId)) - .setSettings(new SettingsConfig.Builder().setInterimResults(false).build()) + .setSettings(settingsConfigBuilder.build()) .setPivotConfig(pivotConfigWithDateHistogramSource) .setSyncConfig(new TimeSyncConfig(TIMESTAMP_FIELD, delay)) .build();