docs: 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 = CassandraUtil.durationIndexBucket(timestampRange.startMillis * 1000);
      int endBucket = CassandraUtil.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 ("".equals(serviceName)) {
        // 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)
        ));
      }
    }