hunk.code_window:         boundary_indices = [
               pac.sum(comp_fn(table[col], b)).as_py() for b in boundaries
           ]
           ret = []
           prev_i = 0
           for i in boundary_indices:
  
hunk.labels: keep keep add keep keep keep
hunk.after_edit:         ### Compute the boundary indices in O(n) time via scan.  # noqa
           # boundary_indices = []
           # remaining = boundaries.copy()
           # values = table[col]
           # for i, x in enumerate(values):
           #     while remaining and not comp_fn(x, remaining[0]).as_py():
           #         remaining.pop(0)
           #         boundary_indices.append(i)
           # for _ in remaining:
           #     boundary_indices.append(len(values))
  
hunk.file_path: python/ray/data/impl/arrow_block.py
hunk.type: add
file: package io.ray.streaming.api.function.impl;
  
  import io.ray.streaming.api.function.Function;
  
  /**
   * Interface of Source functions.
   *
   * @param <T> Type of the data output by the source.
   */
  public interface SourceFunction<T> extends Function {
  
    void init(int parallelism, int index);
  
    void fetch(SourceContext<T> ctx) throws Exception;
  
    void close();
  
    interface SourceContext<T> {
  
      void collect(T element) throws Exception;
    }
  }
  