code: private Histogram getHistogram(String entity, String watermarkColumn, SourceState state,
        Partition partition) {
      SalesforceConnector connector = getConnector(state);
  
      try {
        if (!connector.connect()) {
          throw new RuntimeException("Failed to connect.");
        }
      } catch (RestApiConnectionException e) {
        throw new RuntimeException("Failed to connect.", e);
      }
  
      Histogram histogram = getHistogramByDayBucketing(connector, entity, watermarkColumn, partition);
  
      // exchange the first histogram group key with the global low watermark to ensure that the low watermark is captured
      // in the range of generated partitions
      HistogramGroup firstGroup = histogram.get(0);
      Date lwmDate = Utils.toDate(partition.getLowWatermark(), Partitioner.WATERMARKTIMEFORMAT);
      histogram.getGroups().set(0, new HistogramGroup(Utils.epochToDate(lwmDate.getTime(), SECONDS_FORMAT),
          firstGroup.getCount()));
  
      // refine the histogram
      if (state.getPropAsBoolean(ENABLE_DYNAMIC_PROBING)) {
        histogram = getRefinedHistogram(connector, entity, watermarkColumn, state, partition, histogram);
      }
  
      return histogram;
    }
docs: Generate the histogram