如何实现动态BigQueryIO输入

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.Row;

public class DynamicBigQueryIOInput {
  public static void main(String[] args) {
    // 创建PipelineOptions对象
    PipelineOptions options = PipelineOptionsFactory.create();

    // 创建Pipeline对象
    Pipeline pipeline = Pipeline.create(options);

    // 配置BigQueryIO读取器
    BigQueryIO.Read read = BigQueryIO.read().from("project:dataset.table");

    // 创建动态输入源
    ParDo.SingleOutput<Row, Row> dynamicInput = ParDo.of(new DynamicInputFn());

    // 从BigQuery读取数据
    pipeline.apply(read)
        .apply(dynamicInput)
        .apply(BigQueryIO.writeTableRows().to("project:dataset.table2"));

    // 运行Pipeline
    pipeline.run();
  }

  public static class DynamicInputFn extends DoFn<Row, Row> {
    @ProcessElement
    public void processElement(ProcessContext c) {
      // 根据运行时的条件选择不同的输入源
      if (condition) {
        TableReference tableRef = new TableReference();
        tableRef.setProjectId("project");
        tableRef.setDatasetId("dataset");
        tableRef.setTableId("table1");

        // 读取指定的BigQuery表
        Iterable<Row> rows = c.sideInput(BigQueryIO.readTableRows().from(tableRef));

        // 处理读取到的数据
        for (Row row : rows) {
          // 处理逻辑
          c.output(row);
        }
      } else {
        TableReference tableRef = new TableReference();
        tableRef.setProjectId("project");
        tableRef.setDatasetId("dataset");
        tableRef.setTableId("table2");

        // 读取指定的BigQuery表
        Iterable<Row> rows = c.sideInput(BigQueryIO.readTableRows().from(tableRef));

        // 处理读取到的数据
        for (Row row : rows) {
          // 处理逻辑
          c.output(row);
        }
      }
    }
  }
}