Skip to content

Commit 6d8f78e

Browse files
LantaoJinLantaoJin
authored
Make composite bucket size configurable (opensearch-project#4544) (opensearch-project#4593)
* Make composite bucket size configurable * update doc --------- (cherry picked from commit 31f81b1) Signed-off-by: Lantao Jin <ltjin@amazon.com>
1 parent 8680ba1 commit 6d8f78e

15 files changed

Lines changed: 147 additions & 18 deletions

File tree

‎common/src/main/java/org/opensearch/sql/common/setting/Settings.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,8 @@ public enum Key {
5252
/** Common Settings for SQL and PPL. */
5353
QUERY_MEMORY_LIMIT("plugins.query.memory_limit"),
5454
QUERY_SIZE_LIMIT("plugins.query.size_limit"),
55+
QUERY_BUCKET_SIZE("plugins.query.buckets"),
56+
SEARCH_MAX_BUCKETS("search.max_buckets"),
5557
ENCYRPTION_MASTER_KEY("plugins.query.datasources.encryption.masterkey"),
5658
DATASOURCES_URI_HOSTS_DENY_LIST("plugins.query.datasources.uri.hosts.denylist"),
5759
DATASOURCES_LIMIT("plugins.query.datasources.limit"),

‎core/src/test/java/org/opensearch/sql/executor/QueryServiceTest.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,7 @@ public Helper() {
106106
lenient().when(analyzer.analyze(any(), any())).thenReturn(logicalPlan);
107107
lenient().when(planner.plan(any())).thenReturn(plan);
108108
lenient().when(settings.getSettingValue(Key.QUERY_SIZE_LIMIT)).thenReturn(200);
109+
lenient().when(settings.getSettingValue(Key.QUERY_BUCKET_SIZE)).thenReturn(1000);
109110
lenient().when(settings.getSettingValue(Key.CALCITE_ENGINE_ENABLED)).thenReturn(false);
110111

111112
queryService = new QueryService(analyzer, executionEngine, planner, null, settings);

‎docs/user/admin/settings.rst‎

Lines changed: 38 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -241,7 +241,7 @@ Result set::
241241
}
242242

243243
plugins.query.size_limit
244-
===========================
244+
========================
245245

246246
Description
247247
-----------
@@ -270,6 +270,43 @@ Result set::
270270

271271
Note: the legacy settings of ``opendistro.query.size_limit`` is deprecated, it will fallback to the new settings if you request an update with the legacy name.
272272

273+
plugins.query.buckets
274+
=====================
275+
276+
Version
277+
-------
278+
3.4.0
279+
280+
Description
281+
-----------
282+
283+
This configuration indicates how many aggregation buckets will return in a single response. The default value equals to ``plugins.query.size_limit``.
284+
You can change the value to any value not greater than the maximum number of aggregation buckets allowed in a single response (`search.max_buckets`), here is an example::
285+
286+
>> curl -H 'Content-Type: application/json' -X PUT localhost:9200/_plugins/_query/settings -d '{
287+
"transient" : {
288+
"plugins.query.buckets" : 1000
289+
}
290+
}'
291+
292+
Result set::
293+
294+
{
295+
"acknowledged" : true,
296+
"persistent" : { },
297+
"transient" : {
298+
"plugins" : {
299+
"query" : {
300+
"buckets" : "1000"
301+
}
302+
}
303+
}
304+
}
305+
306+
Limitations
307+
-----------
308+
The number of aggregation buckets is fixed to ``1000`` in v2. ``plugins.query.buckets`` can only effect the number of aggregation buckets when calcite enabled.
309+
273310
plugins.query.memory_limit
274311
==========================
275312

‎docs/user/ppl/admin/settings.rst‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -133,6 +133,43 @@ Change the size_limit to 1000::
133133

134134
Note: the legacy settings of ``opendistro.query.size_limit`` is deprecated, it will fallback to the new settings if you request an update with the legacy name.
135135

136+
plugins.query.buckets
137+
=====================
138+
139+
Version
140+
-------
141+
3.4.0
142+
143+
Description
144+
-----------
145+
146+
This configuration indicates how many aggregation buckets will return in a single response. The default value equals to ``plugins.query.size_limit``.
147+
You can change the value to any value not greater than the maximum number of aggregation buckets allowed in a single response (`search.max_buckets`), here is an example::
148+
149+
>> curl -H 'Content-Type: application/json' -X PUT localhost:9200/_plugins/_query/settings -d '{
150+
"transient" : {
151+
"plugins.query.buckets" : 1000
152+
}
153+
}'
154+
155+
Result set::
156+
157+
{
158+
"acknowledged" : true,
159+
"persistent" : { },
160+
"transient" : {
161+
"plugins" : {
162+
"query" : {
163+
"buckets" : "1000"
164+
}
165+
}
166+
}
167+
}
168+
169+
Limitations
170+
-----------
171+
The number of aggregation buckets is fixed to ``1000`` in v2. ``plugins.query.buckets`` can only effect the number of aggregation buckets when calcite enabled.
172+
136173
plugins.calcite.all_join_types.allowed
137174
======================================
138175

‎integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ public class CalciteExplainIT extends ExplainIT {
2626
public void init() throws Exception {
2727
super.init();
2828
enableCalcite();
29+
setQueryBucketSize(1000);
2930
loadIndex(Index.BANK_WITH_STRING_VALUES);
3031
loadIndex(Index.NESTED_SIMPLE);
3132
loadIndex(Index.TIME_TEST_DATA);

‎integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalcitePPLIntegTestCase.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,7 @@ private Settings defaultSettings() {
112112
private final Map<Key, Object> defaultSettings =
113113
new ImmutableMap.Builder<Key, Object>()
114114
.put(Key.QUERY_SIZE_LIMIT, 200)
115+
.put(Key.QUERY_BUCKET_SIZE, 1000)
115116
.put(Key.SQL_CURSOR_KEEP_ALIVE, TimeValue.timeValueMinutes(1))
116117
.put(Key.FIELD_TYPE_TOLERANCE, true)
117118
.put(Key.CALCITE_ENGINE_ENABLED, true)

‎integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,8 @@ public abstract class SQLIntegTestCase extends OpenSearchSQLRestTestCase {
4949
public static final String TRANSIENT = "transient";
5050
public static final Integer DEFAULT_QUERY_SIZE_LIMIT =
5151
Integer.parseInt(System.getProperty("defaultQuerySizeLimit", "200"));
52+
public static final Integer DEFAULT_QUERY_BUCKET_SIZE =
53+
Integer.parseInt(System.getProperty("defaultQueryBucketSize", "1000"));
5254
public static final Integer DEFAULT_MAX_RESULT_WINDOW =
5355
Integer.parseInt(System.getProperty("defaultMaxResultWindow", "10000"));
5456

@@ -141,6 +143,20 @@ protected void resetQuerySizeLimit() throws IOException {
141143
DEFAULT_QUERY_SIZE_LIMIT.toString()));
142144
}
143145

146+
protected void setQueryBucketSize(Integer limit) throws IOException {
147+
updateClusterSettings(
148+
new ClusterSetting(
149+
"transient", Settings.Key.QUERY_BUCKET_SIZE.getKeyValue(), limit.toString()));
150+
}
151+
152+
protected void resetQueryBucketSize() throws IOException {
153+
updateClusterSettings(
154+
new ClusterSetting(
155+
"transient",
156+
Settings.Key.QUERY_BUCKET_SIZE.getKeyValue(),
157+
DEFAULT_QUERY_BUCKET_SIZE.toString()));
158+
}
159+
144160
@SneakyThrows
145161
protected void setDataSourcesEnabled(String clusterSettingType, boolean value) {
146162
updateClusterSettings(

‎integ-test/src/test/java/org/opensearch/sql/ppl/StandaloneIT.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,7 @@ private Settings defaultSettings() {
166166
private final Map<Key, Object> defaultSettings =
167167
new ImmutableMap.Builder<Key, Object>()
168168
.put(Key.QUERY_SIZE_LIMIT, 200)
169+
.put(Key.QUERY_BUCKET_SIZE, 1000)
169170
.put(Key.SQL_CURSOR_KEEP_ALIVE, TimeValue.timeValueMinutes(1))
170171
.put(Key.FIELD_TYPE_TOLERANCE, true)
171172
.build();

‎integ-test/src/test/java/org/opensearch/sql/sql/StandalonePaginationIT.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -167,6 +167,7 @@ private Settings defaultSettings() {
167167
private final Map<Key, Object> defaultSettings =
168168
new ImmutableMap.Builder<Key, Object>()
169169
.put(Key.QUERY_SIZE_LIMIT, 200)
170+
.put(Key.QUERY_BUCKET_SIZE, 1000)
170171
.put(Key.SQL_CURSOR_KEEP_ALIVE, TimeValue.timeValueMinutes(1))
171172
.put(Key.SQL_PAGINATION_API_SEARCH_AFTER, true)
172173
.put(Key.FIELD_TYPE_TOLERANCE, true)

‎opensearch/src/main/java/org/opensearch/sql/opensearch/request/AggregateAnalyzer.java‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -98,9 +98,6 @@
9898
*/
9999
public class AggregateAnalyzer {
100100

101-
/** How many composite buckets should be returned. */
102-
public static final int AGGREGATION_BUCKET_SIZE = 1000;
103-
104101
/** metadata field used when there is no argument. Only apply to COUNT. */
105102
private static final String METADATA_FIELD = "_index";
106103

@@ -141,6 +138,7 @@ static class AggregateBuilderHelper {
141138
final Map<String, ExprType> fieldTypes;
142139
final RelOptCluster cluster;
143140
final boolean bucketNullable;
141+
final int bucketSize;
144142

145143
<T extends ValuesSourceAggregationBuilder<T>> T build(RexNode node, T aggBuilder) {
146144
return build(node, aggBuilder::field, aggBuilder::script);
@@ -191,7 +189,8 @@ public static Pair<List<AggregationBuilder>, OpenSearchAggregationResponseParser
191189
RelDataType rowType,
192190
Map<String, ExprType> fieldTypes,
193191
List<String> outputFields,
194-
RelOptCluster cluster)
192+
RelOptCluster cluster,
193+
int bucketSize)
195194
throws ExpressionNotAnalyzableException {
196195
requireNonNull(aggregate, "aggregate");
197196
try {
@@ -204,7 +203,7 @@ public static Pair<List<AggregationBuilder>, OpenSearchAggregationResponseParser
204203
.orElseGet(() -> "true"));
205204
List<Integer> groupList = aggregate.getGroupSet().asList();
206205
AggregateBuilderHelper helper =
207-
new AggregateBuilderHelper(rowType, fieldTypes, cluster, bucketNullable);
206+
new AggregateBuilderHelper(rowType, fieldTypes, cluster, bucketNullable, bucketSize);
208207
List<String> aggFieldNames = outputFields.subList(groupList.size(), outputFields.size());
209208
// Process all aggregate calls
210209
Pair<Builder, List<MetricParser>> builderAndParser =
@@ -245,8 +244,7 @@ && isAutoDateSpan(project.getProjects().get(groupList.get(0)))) {
245244
List<CompositeValuesSourceBuilder<?>> buckets =
246245
createCompositeBuckets(groupList, project, helper);
247246
aggregationBuilder =
248-
AggregationBuilders.composite("composite_buckets", buckets)
249-
.size(AGGREGATION_BUCKET_SIZE);
247+
AggregationBuilders.composite("composite_buckets", buckets).size(bucketSize);
250248
if (newMetricBuilder != null) {
251249
aggregationBuilder.subAggregations(metricBuilder);
252250
}
@@ -644,7 +642,7 @@ private static ValuesSourceAggregationBuilder<?> createTermsAggregationBuilder(
644642
helper.build(
645643
group,
646644
new TermsAggregationBuilder(bucketName)
647-
.size(AGGREGATION_BUCKET_SIZE)
645+
.size(helper.bucketSize)
648646
.order(BucketOrder.key(true)));
649647
// Time types values are converted to LONG in ExpressionAggregationScript::execute
650648
if (List.of(TIMESTAMP, TIME, DATE)

0 commit comments

Comments
 (0)