code: public Partition getGlobalPartition(long previousWatermark) {
      ExtractType extractType =
          ExtractType.valueOf(state.getProp(ConfigurationKeys.SOURCE_QUERYBASED_EXTRACT_TYPE).toUpperCase());
      WatermarkType watermarkType = WatermarkType.valueOf(
          state.getProp(ConfigurationKeys.SOURCE_QUERYBASED_WATERMARK_TYPE, ConfigurationKeys.DEFAULT_WATERMARK_TYPE)
              .toUpperCase());
  
      WatermarkPredicate watermark = new WatermarkPredicate(null, watermarkType);
      int deltaForNextWatermark = watermark.getDeltaNumForNextWatermark();
  
      long lowWatermark = getLowWatermark(extractType, watermarkType, previousWatermark, deltaForNextWatermark);
      long highWatermark = getHighWatermark(extractType, watermarkType);
      return new Partition(lowWatermark, highWatermark, true, hasUserSpecifiedHighWatermark);
    }
docs: Get the global partition of the whole data set, which has the global low and high watermarks
  
  @param previousWatermark previous watermark for computing the low watermark of current run
  @return a Partition instance