Note: you need ODP.Net 9.2 or later with Oracle 9i or above, for this to work.
This article has two main Sections:
The first section
highlights the necessary code for building the BatchInserter object
The second section provides examples of how to use it.
Necessary
comments have been added for clarity. To run the sample, you can either use the Tester() method
in the BatchInserter class, or for abstraction leave it to the Client for implementing it like in the
Section #2.
----------------------------------------------------------------------------------------------------
Section
# 1:
----------------------------------------------------------------------------------------------------
#region Namespaces
using System;
using System.Data;
using System.Text;
using System.Collections.Generic;
using Oracle.DataAccess.Client;
#endregion Namespaces
namespace CommonUtilities
{
/// <summary>
/// Optimized code to mass insert data into oracle table.
/// </summary>
public class BatchInserter : IDisposable
{
internal class Column : DataColumn
{
internal object[] Values;
internal OracleParameter Parameter;
internal Column(string columnName, int rowCount)
:
base(columnName.Trim().ToUpper())
{
Values
= new object[rowCount];
}
}
#region Private Variables
/// <summary>
/// Defines the State
/// </summary>
private enum State
{
Configuration,
Loading
}
private OracleConnection mConnection;
private OracleCommand mCommand;
private string mTablename;
private int mBatchSize;
private int mTotalRowCount;
private int mBatchRowCount;
private List<Column> mColumnList = new List<Column>();
private Column[] mColumnArray;
private State mState = State.Configuration;
/// <summary>
/// Gets the size of the batch.
/// </summary>
/// <value>The size of the batch.</value>
private int BatchSize
{
get
{ return mBatchSize; }
}
/// <summary>
/// Gets the command.
/// </summary>
/// <value>The command.</value>
private OracleCommand Command
{
get
{ return mCommand; }
}
#endregion Private Variables
/// <summary>
/// Constructor
/// </summary>
/// <param name="connection"></param>
/// <param name="tablename"></param>
/// <param name="batchSize"></param>
public BatchInserter(OracleConnection connection, string tablename, int batchSize)
{
if ((mConnection = connection) == null)
ThrowError("BatchInserter does not yet support connection type " + connection.GetType());
mTablename
= tablename;
mCommand
= mConnection.CreateCommand();
mBatchSize
= batchSize;
mCommand.ArrayBindCount
= batchSize;
}
/// <summary>
/// Adds a Column to the Batch Inserter
/// </summary>
/// <param name="columnName"></param>
/// <param name="cSharpType"></param>
/// <param name="oracleType"></param>
/// <returns></returns>
private Column AddColumn(string columnName, Type cSharpType, OracleDbType oracleType)
{
if (InserterState != State.Configuration)
ThrowError("Adding columns not allowed in current state");
Column
column = new Column(columnName, BatchSize);
column.DataType
= cSharpType;
column.Parameter
= Command.Parameters.Add(column.ColumnName, oracleType);
column.Parameter.Value
= column.Values;
mColumnList.Add(column);
return column;
}
/// <summary>
/// Adds Int Column
/// </summary>
/// <param name="columnName"></param>
public void AddIntColumn(string columnName)
{
AddColumn(columnName,
typeof(Int32), OracleDbType.Int32);
}
/// <summary>
/// Adds Double Column
/// </summary>
/// <param name="columnName"></param>
public void AddDoubleColumn(string columnName)
{
AddColumn(columnName,
typeof(Double), OracleDbType.Double);
}
/// <summary>
/// Adds Decimal Column
/// </summary>
/// <param name="columnName"></param>
public void AddDecimalColumn(string columnName)
{
AddColumn(columnName,
typeof(Decimal), OracleDbType.Decimal);
}
/// <summary>
/// Adds Date Column
/// </summary>
/// <param name="columnName"></param>
public void AddDateColumn(string columnName)
{
AddColumn(columnName,
typeof(DateTime), OracleDbType.Date);
}
/// <summary>
/// Adds String Column
/// </summary>
/// <param name="columnName"></param>
public void AddStringColumn(string columnName)
{
AddColumn(columnName,
typeof(string), OracleDbType.Varchar2);
}
/// <summary>
/// Adds String Column with Max Length
/// </summary>
/// <param name="columnName"></param>
/// <param name="maxLength"></param>
public void AddStringColumn(string columnName, int maxLength)
{
Column
column = AddColumn(columnName, typeof(string), OracleDbType.Varchar2);
column.MaxLength
= maxLength;
}
/// <summary>
/// Inserter State
/// </summary>
private State InserterState
{
get
{ return mState; }
set
{ mState = value; }
}
/// <summary>
/// Initializes the Array
/// </summary>
private void Initialize()
{
//Builds the Insert query along with Values clause, which would be replaced with
//the array values.
mColumnArray
= new Column[mColumnList.Count];
mColumnList.CopyTo(mColumnArray);
StringBuilder
insertClause = new StringBuilder();
StringBuilder
valuesClause = new StringBuilder();
insertClause.Append("insert into ");
insertClause.Append(mTablename);
insertClause.Append('(');
valuesClause.Append(" values(");
for (int i = 0; i < mColumnArray.Length; i++)
{
if (i > 0)
{
insertClause.Append(',');
valuesClause.Append(',');
}
insertClause.Append(mColumnArray[i].ColumnName);
valuesClause.Append(':');
valuesClause.Append(mColumnArray[i].ColumnName);
}
insertClause.Append(')');
valuesClause.Append(')');
mCommand.CommandText
= insertClause.ToString() + valuesClause.ToString();
//Set the Inserter State
InserterState
= State.Loading;
}
/// <summary>
/// Adds a Row
/// </summary>
/// <param name="values"></param>
public void AddRow(params object[] values)
{
if (InserterState != State.Loading)
Initialize();
if (values.Length != Command.Parameters.Count)
ThrowError("Number of column values does not match command parameter count");
if (mBatchRowCount >= BatchSize)
ThrowError("Invalid state: batch size exceeded");
for (int i = 0; i < values.Length; i++)
{
mColumnArray[i].Values[mBatchRowCount]
= values[i] == null ? DBNull.Value : values[i];
}
mTotalRowCount++;
if (++mBatchRowCount == BatchSize)
Flush();
}
/// <summary>
/// Flushes the Array to DB
/// </summary>
public void Flush()
{
if (mBatchRowCount == 0)
return;
Command.ArrayBindCount
= mBatchRowCount;
if (Command.Connection.State == ConnectionState.Closed)
Command.Connection.Open();
Command.ExecuteNonQuery();
mBatchRowCount
= 0;
}
/// <summary>
/// Throws an Error
/// </summary>
/// <param name="message"></param>
private void ThrowError(string message)
{
//Should throw a custom exception along with other necessary details
throw ex;
}
/// <summary>
/// Sample Method
/// </summary>
public static void Tester()
{
//Set up your connection here.
OracleConnection
conn = new OracleConnection(OracleDatabaseHelper.ConnectionString);
BatchInserter
inserter = new BatchInserter(conn, "test_batch", 1000);
inserter.AddIntColumn("col1");
inserter.AddDoubleColumn("col2");
inserter.AddStringColumn("col3");
inserter.AddDateColumn("col4");
for (int i = 0; i < 1001; i++)
{
inserter.AddRow(100,
111.03, i.ToString(), DateTime.Today);
}
inserter.Flush();
conn.Close();
}
#region IDisposable Members
/// <summary>
/// Performs application-defined tasks associated with freeing, releasing, or resetting
unmanaged
resources.
/// </summary>
public void Dispose()
{
if (mCommand != null)
{
if (mCommand.Connection.State != ConnectionState.Closed)
mCommand.Connection.Close();
mCommand.Dispose();
}
}
#endregion
}
}
----------------------------------------------------------------------------------------------------
Section
# 2:
----------------------------------------------------------------------------------------------------
Step
1: Create the Columns in the Table for Bacth Insert
private BatchInserter CreateBatchInserterObject(int Rows)
{
//Replace DatabaseName and TableName with appropriate values
BatchInserter BIMF = new BatchInserter("DatabaseName","TableName",Rows);
//Add all the columns of the Table to the BatchInserter Object.
//Specify the lenght too.
BIMF.AddStringColumn("REPORT_DT",50);
BIMF.AddStringColumn("CUSTODIAN_NAME",255);
BIMF.AddStringColumn("CUSTODIAN_NUMBER",16);
BIMF.AddIntColumn("LINE_NO");
return BIMF;
}
Step 2: Extract the rows from
the DataSet and build the rowArray
//Use the CreateBatchInserterObject function above.
//iActualRowCount is the Number of rows in the dataset.
BatchInserter BI = CreateBatchInserterObject(iActualRowCount);
//loop thru the dataset and add the rows to the batchInserter object
for(int i=0;i<=iActualRowCount-1;i++)
{
//Build the object[] from the row values.
object[] param = { };
BI.AddRow(param);
}
BI.Flush(iActualRowCount)
-----------------------------------------------------------------------------------------------------