content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.beam.sdk.io.range;
  
  /**
   * A {@code RangeTracker} is a thread-safe helper object for implementing dynamic work rebalancing
   * in position-based {@link org.apache.beam.sdk.io.BoundedSource.BoundedReader} subclasses.
   *
   * <h3>Usage of the RangeTracker class hierarchy</h3>
   *
   * The abstract {@code RangeTracker} interface should not be used per se - all users should use its
   * subclasses directly. We declare it here because all subclasses have roughly the same interface
   * and the same properties, to centralize the documentation. Currently we provide one implementation
   * - {@link OffsetRangeTracker}.
   *
   * <h3>Position-based sources</h3>
   *
   * A position-based source is one where the source can be described by a range of positions of an
   * ordered type and the records returned by the reader can be described by positions of the same
   * type.
   *
   * <p>In case a record occupies a range of positions in the source, the most important thing about
   * the record is the position where it starts.
   *
   * <p>Defining the semantics of positions for a source is entirely up to the source class, however
   * the chosen definitions have to obey certain properties in order to make it possible to correctly
   * split the source into parts, including dynamic splitting. Two main aspects need to be defined:
   *
   * <ul>
   *   <li>How to assign starting positions to records.
   *   <li>Which records should be read by a source with a range {@code [A, B)}.
   * </ul>
   *
   * Moreover, reading a range must be <i>efficient</i>, i.e., the performance of reading a range
   * should not significantly depend on the location of the range. For example, reading the range
   * {@code [A, B)} should not require reading all data before {@code A}.
   *
   * <p>The sections below explain exactly what properties these definitions must satisfy, and how to
   * use a {@code RangeTracker} with a properly defined source.
   *
   * <h3>Properties of position-based sources</h3>
   *
   * The main requirement for position-based sources is <i>associativity</i>: reading records from
   * {@code [A, B)} and records from {@code [B, C)} should give the same records as reading from
   * {@code [A, C)}, where {@code A <= B <= C}. This property ensures that no matter how a range of
   * positions is split into arbitrarily many sub-ranges, the total set of records described by them
   * stays the same.
   *
   * <p>The other important property is how the source's range relates to positions of records in the
   * source. In many sources each record can be identified by a unique starting position. In this
   * case:
   *
   * <ul>
   *   <li>All records returned by a source {@code [A, B)} must have starting positions in this range.
   *   <li>All but the last record should end within this range. The last record may or may not extend
   *       past the end of the range.
   *   <li>Records should not overlap.
   * </ul>
   *
   * Such sources should define "read {@code [A, B)}" as "read from the first record starting at or
   * after A, up to but not including the first record starting at or after B".
   *
   * <p>Some examples of such sources include reading lines or CSV from a text file, reading keys and
   * values from a BigTable, etc.
   *
   * <p>The concept of <i>split points</i> allows to extend the definitions for dealing with sources
   * where some records cannot be identified by a unique starting position.
   *
   * <p>In all cases, all records returned by a source {@code [A, B)} must <i>start</i> at or after
   * {@code A}.
   *
   * <h3>Split points</h3>
   *
   * <p>Some sources may have records that are not directly addressable. For example, imagine a file
   * format consisting of a sequence of compressed blocks. Each block can be assigned an offset, but
   * records within the block cannot be directly addressed without decompressing the block. Let us
   * refer to this hypothetical format as <i>CBF (Compressed Blocks Format)</i>.
   *
   * <p>Many such formats can still satisfy the associativity property. For example, in CBF, reading
   * {@code [A, B)} can mean "read all the records in all blocks whose starting offset is in {@code
   * [A, B)}".
   *
   * <p>To support such complex formats, we introduce the notion of <i>split points</i>. We say that a
   * record is a split point if there exists a position {@code A} such that the record is the first
   * one to be returned when reading the range {@code [A, infinity)}. In CBF, the only split points
   * would be the first records in each block.
   *
   * <p>Split points allow us to define the meaning of a record's position and a source's range in all
   * cases:
   *
   * <ul>
   *   <li>For a record that is at a split point, its position is defined to be the largest {@code A}
   *       such that reading a source with the range {@code [A, infinity)} returns this record;
   *   <li>Positions of other records are only required to be non-decreasing;
   *   <li>Reading the source {@code [A, B)} must return records starting from the first split point
   *       at or after {@code A}, up to but not including the first split point at or after {@code B}.
   *       In particular, this means that the first record returned by a source MUST always be a split
   *       point.
   *   <li>Positions of split points must be unique.
   * </ul>
   *
   * As a result, for any decomposition of the full range of the source into position ranges, the
   * total set of records will be the full set of records in the source, and each record will be read
   * exactly once.
   *
   * <h3>Consumed positions</h3>
   *
   * As the source is being read, and records read from it are being passed to the downstream
   * transforms in the pipeline, we say that positions in the source are being <i>consumed</i>. When a
   * reader has read a record (or promised to a caller that a record will be returned), positions up
   * to and including the record's start position are considered <i>consumed</i>.
   *
   * <p>Dynamic splitting can happen only at <i>unconsumed</i> positions. If the reader just returned
   * a record at offset 42 in a file, dynamic splitting can happen only at offset 43 or beyond, as
   * otherwise that record could be read twice (by the current reader and by a reader of the task
   * starting at 43).
   *
   * <h3>Example</h3>
   *
   * The following example uses an {@link OffsetRangeTracker} to support dynamically splitting a
   * source with integer positions (offsets).
   *
   * <pre>{@code
   * class MyReader implements BoundedReader<Foo> {
   *   private MySource currentSource;
   *   private final OffsetRangeTracker tracker = new OffsetRangeTracker();
   *   ...
   *   MyReader(MySource source) {
   *     this.currentSource = source;
   *     this.tracker = new MyRangeTracker<>(source.getStartOffset(), source.getEndOffset())
   *   }
   *   ...
   *   boolean start() {
   *     ... (general logic for locating the first record) ...
   *     if (!tracker.tryReturnRecordAt(true, recordStartOffset)) return false;
   *     ... (any logic that depends on the record being returned, e.g. counting returned records)
   *     return true;
   *   }
   *   boolean advance() {
   *     ... (general logic for locating the next record) ...
   *     if (!tracker.tryReturnRecordAt(isAtSplitPoint, recordStartOffset)) return false;
   *     ... (any logic that depends on the record being returned, e.g. counting returned records)
   *     return true;
   *   }
   *
   *   double getFractionConsumed() {
   *     return tracker.getFractionConsumed();
   *   }
   * }
   * }</pre>
   *
   * <h3>Usage with different models of iteration</h3>
   *
   * When using this class to protect a {@link org.apache.beam.sdk.io.BoundedSource.BoundedReader},
   * follow the pattern described above.
   *
   * <p>When using this class to protect iteration in the {@code hasNext()/next()} model, consider the
   * record consumed when {@code hasNext()} is about to return true, rather than when {@code next()}
   * is called, because {@code hasNext()} returning true is promising the caller that {@code next()}
   * will have an element to return - so {@link #trySplitAtPosition} must not split the range in a way
   * that would make the record promised by {@code hasNext()} belong to a different range.
   *
   * <p>Also note that implementations of {@code hasNext()} need to ensure that they call {@link
   * #tryReturnRecordAt} only once even if {@code hasNext()} is called repeatedly, due to the
   * requirement on uniqueness of split point positions.
   *
   * @param <PositionT> Type of positions used by the source to define ranges and identify records.
   */
  public interface RangeTracker<PositionT> {
    /** Returns the starting position of the current range, inclusive. */
    PositionT getStartPosition();
  
    /** Returns the ending position of the current range, exclusive. */
    PositionT getStopPosition();
  
    /**
     * Atomically determines whether a record at the given position can be returned and updates
     * internal state. In particular:
     *
     * <ul>
     *   <li>If {@code isAtSplitPoint} is {@code true}, and {@code recordStart} is outside the current
     *       range, returns {@code false};
     *   <li>Otherwise, updates the last-consumed position to {@code recordStart} and returns {@code
     *       true}.
     * </ul>
     *
     * <p>This method MUST be called on all split point records. It may be called on every record.
     */
    boolean tryReturnRecordAt(boolean isAtSplitPoint, PositionT recordStart);
  
    /**
     * Atomically splits the current range [{@link #getStartPosition}, {@link #getStopPosition}) into
     * a "primary" part [{@link #getStartPosition}, {@code splitPosition}) and a "residual" part
     * [{@code splitPosition}, {@link #getStopPosition}), assuming the current last-consumed position
     * is within [{@link #getStartPosition}, splitPosition) (i.e., {@code splitPosition} has not been
     * consumed yet).
     *
     * <p>Updates the current range to be the primary and returns {@code true}. This means that all
     * further calls on the current object will interpret their arguments relative to the primary
     * range.
     *
     * <p>If the split position has already been consumed, or if no {@link #tryReturnRecordAt} call
     * was made yet, returns {@code false}. The second condition is to prevent dynamic splitting
     * during reader start-up.
     */
    boolean trySplitAtPosition(PositionT splitPosition);
  
    /**
     * Returns the approximate fraction of positions in the source that have been consumed by
     * successful {@link #tryReturnRecordAt} calls, or 0.0 if no such calls have happened.
     */
    double getFractionConsumed();
  }
  