How can I use Offset in date_histogram of composite aggregation？

[raidenxu]

There is Offset in date_histogram of composite aggregation since Elasticsearch 7.6

https://www.elastic.co/guide/en/elasticsearch/reference/7.6/search-aggregations-bucket-composite-aggregation.html#_date_histogram

Any support?

[ljluestc]

import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.*;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.composite.*;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;
import java.util.*;
import java.util.function.BiConsumer;

public class ElasticsearchCompositeOffsetExample {
    public static void main(String[] args) throws IOException {
        // Initialize REST client for AWS Elasticsearch
        RestHighLevelClient client = new RestHighLevelClient(
                RestClient.builder(new HttpHost("your-aws-es-endpoint", 443, "https")));

        // Index sample data using Bulk Processor
        indexSampleData(client);

        // Perform composite aggregation with offset workaround for 6.x
        performCompositeAggregationWithOffsetWorkaround(client);

        // Perform standalone date_histogram aggregation with offset for 6.x
        performDateHistogramAggregation(client);

        // Close client
        client.close();
    }

    private static void indexSampleData(RestHighLevelClient client) throws IOException {
        // Bulk Processor with retry handling for 413 errors
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                System.out.println("Executing bulk with " + request.numberOfActions() + " actions");
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                if (response.hasFailures()) {
                    System.out.println("Bulk failed: " + response.buildFailureMessage());
                    for (BulkItemResponse item : response.getItems()) {
                        if (item.isFailed() && item.getFailureMessage().contains("413")) {
                            System.err.println("Detected 413 error for document ID: " + item.getId());
                        }
                    }
                }
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                System.err.println("Bulk failed: " + failure.getMessage());
                if (failure.getMessage().contains("413")) {
                    System.err.println("Non-retryable 413 error. Skipping retry.");
                    request.requests().clear();
                }
            }
        };

        BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer = (request, bulkListener) -> {
            try {
                client.bulkAsync(request, bulkListener);
            } catch (Exception e) {
                if (e.getMessage().contains("413")) {
                    System.err.println("Skipping retry for 413 error: " + e.getMessage());
                    bulkListener.onFailure(e);
                } else {
                    client.bulkAsync(request, bulkListener);
                }
            }
        };

        BulkProcessor bulkProcessor = BulkProcessor.builder(bulkConsumer, listener)
                .setBulkActions(1000)
                .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB))
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                .setConcurrentRequests(1)
                .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
                .build();

        // Add sample documents (small to avoid 413 errors)
        String[] dates = {
                "2015-10-01T05:30:00Z",
                "2015-10-01T06:30:00Z",
                "2015-10-02T05:30:00Z"
        };
        for (int i = 0; i < dates.length; i++) {
            String json = String.format("{\"date\":\"%s\",\"value\":%d}", dates[i], i + 1);
            if (json.length() > 10 * 1024 * 1024) {
                System.err.println("Document too large, skipping: " + i);
                continue;
            }
            bulkProcessor.add(new IndexRequest("my-index-000001", "doc", String.valueOf(i))
                    .source(json, XContentType.JSON));
        }

        try {
            bulkProcessor.awaitClose(10, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            System.err.println("Interrupted: " + e.getMessage());
        }
    }

    private static void performCompositeAggregationWithOffsetWorkaround(RestHighLevelClient client) throws IOException {
        // Workaround: Use a script to adjust date by +6h
        Map<String, String> scriptParams = new HashMap<>();
        scriptParams.put("offset", "21600000"); // 6 hours in milliseconds

        Script script = new Script(
                ScriptType.INLINE,
                "painless",
                "doc['date'].value.millis + params.offset",
                scriptParams
        );

        CompositeAggregationBuilder compositeAgg = AggregationBuilders.composite("my_buckets")
                .sources(Arrays.asList(
                        new CompositeValuesSourceBuilder("date")
                                .script(script)
                                .dateHistogram()
                                .calendarInterval(DateHistogramInterval.DAY)
                                .format("yyyy-MM-dd'T'HH:mm:ss.SSSZ")
                ))
                .size(10); // Paginate up to 10 buckets

        SearchSourceBuilder searchSource = new SearchSourceBuilder()
                .size(0)
                .aggregation(compositeAgg);

        SearchRequest searchRequest = new SearchRequest("my-index-000001")
                .source(searchSource);

        SearchResponse response = client.search(searchRequest);
        System.out.println("Composite Aggregation (6.x Workaround) Results:");
        CompositeAggregation agg = response.getAggregations().get("my_buckets");
        for (CompositeAggregation.Bucket bucket : agg.getBuckets()) {
            System.out.printf("Bucket: %s, Doc Count: %d%n",
                    bucket.getKey().get("date"), bucket.getDocCount());
        }
    }

    private static void performDateHistogramAggregation(RestHighLevelClient client) throws IOException {
        // Standalone date_histogram with offset (supported in 6.x)
        SearchSourceBuilder searchSource = new SearchSourceBuilder()
                .size(0)
                .aggregation(AggregationBuilders.dateHistogram("by_day")
                        .field("date")
                        .calendarInterval(DateHistogramInterval.DAY)
                        .offset("+6h")
                        .format("yyyy-MM-dd'T'HH:mm:ss.SSSZ"));

        SearchRequest searchRequest = new SearchRequest("my-index-000001")
                .source(searchSource);

        SearchResponse response = client.search(searchRequest);
        System.out.println("\nDate Histogram Aggregation (6.x Native Offset) Results:");
        org.elasticsearch.search.aggregations.bucket.histogram.Histogram agg = response.getAggregations().get("by_day");
        for (org.elasticsearch.search.aggregations.bucket.histogram.Histogram.Bucket bucket : agg.get​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​