Example.java

package org.apache.beam.sdk.io.gcp.spanner;

import com.google.auto.value.AutoValue;
import com.google.cloud.spanner.Statement;
import com.google.cloud.spanner.Struct;
import java.io.Serializable;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;

public class Example {

  @AutoValue
  static abstract class Range implements Serializable {
    public static Range of(Long start, Long end) {
      return new AutoValue_Example_Range(start, end);
    }

    public abstract Long getStart();
    public abstract Long getEnd();

  }

  public static void main(String[] args) {
    Pipeline p = Pipeline.create(PipelineOptionsFactory.create());

    PCollection<Range> keyRanges = p.apply(Create.of(
        Range.of(0L, 1000L),
        Range.of(1000L, 2000L),
        Range.of(2000L, 3000L)
    ));

    PCollection<ReadOperation> reads = keyRanges.apply(ParDo.of(new DoFn<Range, ReadOperation>() {
      @ProcessElement
      public void processElement(ProcessContext c) throws Exception {
        Range range = c.element();
        Statement st = Statement
            .newBuilder("SELECT * from usertable where id >= @start and id < @end")
            .bind("start").to(range.getStart())
            .bind("end").to(range.getEnd())
        .build();
        c.output(ReadOperation.create().withQuery(st));
      }
    }));

    PCollection<Struct> rows = reads.apply(SpannerIO.readAll());
    
    // Use rows.
  }

}