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.arrow.adapter.jdbc;
  
  import java.sql.PreparedStatement;
  import java.sql.SQLException;
  import java.util.HashMap;
  import java.util.Map;
  import org.apache.arrow.adapter.jdbc.binder.ColumnBinder;
  import org.apache.arrow.util.Preconditions;
  import org.apache.arrow.vector.VectorSchemaRoot;
  
  /**
   * A binder binds JDBC prepared statement parameters to rows of Arrow data from a VectorSchemaRoot.
   *
   * <p>Each row of the VectorSchemaRoot will be bound to the configured parameters of the
   * PreparedStatement. One row of data is bound at a time.
   */
  public class JdbcParameterBinder {
    private final PreparedStatement statement;
    private final VectorSchemaRoot root;
    private final ColumnBinder[] binders;
    private final int[] parameterIndices;
    private int nextRowIndex;
  
    /**
     * Create a new parameter binder.
     *
     * @param statement The statement to bind parameters to.
     * @param root The VectorSchemaRoot to pull data from.
     * @param binders Column binders to translate from Arrow data to JDBC parameters, one per
     *     parameter.
     * @param parameterIndices For each binder in <tt>binders</tt>, the index of the parameter to bind
     *     to.
     */
    private JdbcParameterBinder(
        final PreparedStatement statement,
        final VectorSchemaRoot root,
        final ColumnBinder[] binders,
        int[] parameterIndices) {
      Preconditions.checkArgument(
          binders.length == parameterIndices.length,
          "Number of column binders (%s) must equal number of parameter indices (%s)",
          binders.length,
          parameterIndices.length);
      this.statement = statement;
      this.root = root;
      this.binders = binders;
      this.parameterIndices = parameterIndices;
      this.nextRowIndex = 0;
    }
  
    /**
     * Initialize a binder with a builder.
     *
     * @param statement The statement to bind to. The binder does not maintain ownership of the
     *     statement.
     * @param root The {@link VectorSchemaRoot} to pull data from. The binder does not maintain
     *     ownership of the vector schema root.
     */
    public static Builder builder(final PreparedStatement statement, final VectorSchemaRoot root) {
      return new Builder(statement, root);
    }
  
    /** Reset the binder (so the root can be updated with new data). */
    public void reset() {
      nextRowIndex = 0;
    }
  
    /**
     * Bind the next row of data to the parameters of the statement.
     *
     * <p>After this, the application should call the desired method on the prepared statement, such
     * as {@link PreparedStatement#executeUpdate()}, or {@link PreparedStatement#addBatch()}.
     *
     * @return true if a row was bound, false if rows were exhausted
     */
    public boolean next() throws SQLException {
      if (nextRowIndex >= root.getRowCount()) {
        return false;
      }
      for (int i = 0; i < parameterIndices.length; i++) {
        final int parameterIndex = parameterIndices[i];
        binders[i].bind(statement, parameterIndex, nextRowIndex);
      }
      nextRowIndex++;
      return true;
    }
  
    /** A builder for a {@link JdbcParameterBinder}. */
    public static class Builder {
      private final PreparedStatement statement;
      private final VectorSchemaRoot root;
      private final Map<Integer, ColumnBinder> bindings;
  
      Builder(PreparedStatement statement, VectorSchemaRoot root) {
        this.statement = statement;
        this.root = root;
        this.bindings = new HashMap<>();
      }
  
      /** Bind each column to the corresponding parameter in order. */
      public Builder bindAll() {
        for (int i = 0; i < root.getFieldVectors().size(); i++) {
          bind(/* parameterIndex= */ i + 1, /* columnIndex= */ i);
        }
        return this;
      }
  
      /** Bind the given parameter to the given column using the default binder. */
      public Builder bind(int parameterIndex, int columnIndex) {
        return bind(parameterIndex, ColumnBinder.forVector(root.getVector(columnIndex)));
      }
  
      /** Bind the given parameter using the given binder. */
      public Builder bind(int parameterIndex, ColumnBinder binder) {
        Preconditions.checkArgument(
            parameterIndex > 0, "parameterIndex %d must be positive", parameterIndex);
        bindings.put(parameterIndex, binder);
        return this;
      }
  
      /** Build the binder. */
      public JdbcParameterBinder build() {
        ColumnBinder[] binders = new ColumnBinder[bindings.size()];
        int[] parameterIndices = new int[bindings.size()];
        int index = 0;
        for (Map.Entry<Integer, ColumnBinder> entry : bindings.entrySet()) {
          binders[index] = entry.getValue();
          parameterIndices[index] = entry.getKey();
          index++;
        }
        return new JdbcParameterBinder(statement, root, binders, parameterIndices);
      }
    }
  }
  