Call<Map<String, Long>> newBucketedTraceIdCall(
    QueryRequest request, TimestampRange timestampRange, int traceIndexFetchSize) {
    // trace_by_service_span adds special empty-string span name in order to search by all
    String spanName = null != request.spanName() ? request.spanName() : "";
    Long minDuration = request.minDuration(), maxDuration = request.maxDuration();
    int startBucket = durationIndexBucket(timestampRange.startMillis * 1000);
    int endBucket = durationIndexBucket(timestampRange.endMillis * 1000);
    if (startBucket > endBucket) {
      throw new IllegalArgumentException(
        "Start bucket (" + startBucket + ") > end bucket (" + endBucket + ")");
    }

    // "" isn't a real value. it is used to template bucketed calls and replaced later
    String serviceName = null != request.serviceName() ? request.serviceName() : "";

    // TODO: ideally, the buckets are traversed backwards, only spawning queries for older buckets
    // if younger buckets are empty. This will be an async continuation, punted for now.
    List<SelectTraceIdsFromServiceSpan.Input> serviceSpans = new ArrayList<>();
    List<SelectTraceIdsFromServiceRemoteService.Input> serviceRemoteServices = new ArrayList<>();
    String remoteService = request.remoteServiceName();
    for (int bucket = endBucket; bucket >= startBucket; bucket--) {
      boolean addSpanQuery = true;
      if (remoteService != null) {
        if (traceIdsFromServiceRemoteService == null) {
          throw new IllegalArgumentException("remoteService=" + remoteService
            + " unsupported due to missing table " + TABLE_SERVICE_REMOTE_SERVICES);
        }
        serviceRemoteServices.add(
          traceIdsFromServiceRemoteService.newInput(
            serviceName,
            remoteService,
            bucket,
            timestampRange,
            traceIndexFetchSize));
        // If the remote service query can satisfy the request, don't make a redundant span query
        addSpanQuery = !spanName.isEmpty() || minDuration != null;
      }
      if (!addSpanQuery) continue;

      serviceSpans.add(
        traceIdsFromServiceSpan.newInput(
          serviceName,
          spanName,
          bucket,
          minDuration,
          maxDuration,
          timestampRange,
          traceIndexFetchSize));
    }

    if (serviceName.isEmpty()) {
      // If we have no service name, we have to lookup service names before running trace ID queries
      Call<List<String>> serviceNames = getServiceNames();
      if (serviceRemoteServices.isEmpty()) {
        return serviceNames.flatMap(traceIdsFromServiceSpan.newFlatMapper(serviceSpans));
      } else if (serviceSpans.isEmpty()) {
        return serviceNames.flatMap(
          traceIdsFromServiceRemoteService.newFlatMapper(serviceRemoteServices));
      }
      return serviceNames.flatMap(new AggregateFlatMapper<>(
        traceIdsFromServiceSpan.newFlatMapper(serviceSpans),
        traceIdsFromServiceRemoteService.newFlatMapper(serviceRemoteServices)
      ));
    }
    if (serviceRemoteServices.isEmpty()) {
      return traceIdsFromServiceSpan.newCall(serviceSpans);
    } else if (serviceSpans.isEmpty()) {
      return traceIdsFromServiceRemoteService.newCall(serviceRemoteServices);
    } else {
      return new IntersectMaps<>(asList(
        traceIdsFromServiceSpan.newCall(serviceSpans),
        traceIdsFromServiceRemoteService.newCall(serviceRemoteServices)
      ));
    }
  }
Creates a call representing one or more queries against {@link Schema#TABLE_TRACE_BY_SERVICE_SPAN}
and possibly {@link Schema#TABLE_TRACE_BY_SERVICE_REMOTE_SERVICE}.

<p>The result will be an aggregate if the input request serviceName is null, both span name
and remote service name are supplied, or there's more than one day of data in the timestamp
range.

<p>Note that when {@link QueryRequest#serviceName()} is null, the returned query composes over
{@link #getServiceNames()}. This means that if you have 1000 service names, you will end up
with a composition of at least 1000 calls.
zipkin-storage/cassandra/src/main/java/zipkin2/storage/cassandra/CassandraSpanStore.java