Bulk Import (Insert and Update) using BULK INSERT and MERGE Statements in SQL Server 2008

In some projects, we may have bulk import functionality to a table which is already exist with our system. SQL Server given lots of facilities for importing data into the database, some are –

1. Using bcp Utility
2. Using BULK INSERT statement
3. Using OPENROWSET
4. Etc.,
   

In this post I am going do bulk import using BULK INSERT and MERGE statements. We use both BULK INSERT and MERGE statements as this implementation not only used for inserting the records, also for updating records if already exist.

So the use case will be as below –

1. The input for the bulk import would be either .csv file or an .xml file.
2. The records from the input file will as per the importing table schema. So the values will be able to insert or update without any issue in the same order and data type.
3. The records from the csv may already present in the table or may not.
4. When no records present, all the records will be inserted (Ex: initial stage).
5. If some records are already present, the records will be updated otherwise inserted.

Note: This implementation will work only from SQL Server 2008 and later versions (as MERGE statement introduced from SQL Server 2008).

The implementation as follows-

Step 1: I am taking Customer table for importing the records (Customer table created from Northwind database schema).

Below is the table schema –

IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[dbo].[Customers]') AND type in (N'U'))
DROP TABLE [dbo].[Customers]
GO

CREATE TABLE [dbo].[Customers](
 [CustomerID] [nchar](5) NOT NULL,
 [CompanyName] [nvarchar](40) NOT NULL,
 [ContactName] [nvarchar](30) NULL,
 [ContactTitle] [nvarchar](30) NULL,
 [Address] [nvarchar](60) NULL,
 [City] [nvarchar](15) NULL,
 [Region] [nvarchar](15) NULL,
 [PostalCode] [nvarchar](10) NULL,
 [Country] [nvarchar](15) NULL,
 [Phone] [nvarchar](24) NULL,
 [Fax] [nvarchar](24) NULL,
 CONSTRAINT [PK_Customers1] PRIMARY KEY CLUSTERED ( [CustomerID] ASC )
)
GO

Implementing with a csv file for importing the data. Some of the records in the csv file would be

ALFKI,Alfreds Futterkiste,Maria Anders1,Sales Representative,Obere Str. 57,Berlin,NULL,12209,Germany,030-0074321,030-0076545
ANATR,Ana Trujillo Emparedados y helados,Ana Trujillo,Owner,Avda. de la Constituci¢n 2222,M‚xico D.F.,NULL,5021,Mexico,(5) 555-4729,(5) 555-3745
ANTON,Antonio Moreno Taquer¡a,Antonio Moreno,Owner,Mataderos  2312,M‚xico D.F.,NULL,5023,Mexico,(5) 555-3932,NULL
AROUT,Around the Horn,Thomas Hardy,Sales Representative,120 Hanover Sq.,London,NULL,WA1 1DP,UK,(171) 555-7788,(171) 555-6750
BERGS,Berglunds snabbk”p,Christina Berglund,Order Administrator,Berguvsv„gen  8,Lule†,NULL,S-958 22,Sweden,0921-12 34 65,0921-12 34 67
BLAUS,Blauer See Delikatessen,Hanna Moos,Sales Representative,Forsterstr. 57,Mannheim,NULL,68306,Germany,0621-08460,0621-08924
BLONP,Blondesddsl pŠre et fils,Fr‚d‚rique Citeaux,Marketing Manager,"24, place Kl‚ber",Strasbourg,NULL,67000,France,88.60.15.31,88.60.15.32
BOTTM,Bottom-Dollar Markets,Elizabeth Lincoln,Accounting Manager,23 Tsawassen Blvd.,Tsawassen,BC,T2F 8M4,Canada,(604) 555-4729,(604) 555-3745
BSBEV,B's Beverages,Victoria Ashworth,Sales Representative,Fauntleroy Circus,London,NULL,EC2 5NT,UK,(171) 555-1212,NULL
CACTU,Cactus Comidas para llevar,Patricio Simpson,Sales Agent,Cerrito 333,Buenos Aires,NULL,1010,Argentina,(1) 135-5555,(1) 135-4892
CENTC,Centro comercial Moctezuma,Francisco Chang,Marketing Manager,Sierras de Granada 9993,M‚xico D.F.,NULL,5022,Mexico,(5) 555-3392,(5) 555-7293

The following store procedure is used for getting the data from csv file and importing to the Customers table (both insert and update).

CREATE PROCEDURE SP_ImportCustomerData
AS
BEGIN
    -- Creating a Temproary Table for importing the data from csv file.
    CREATE TABLE #Customers(
        [CustomerID] [nchar](5) NOT NULL,
        [CompanyName] [nvarchar](40) NOT NULL,
        [ContactName] [nvarchar](30) NULL,
        [ContactTitle] [nvarchar](30) NULL,
        [Address] [nvarchar](60) NULL,
        [City] [nvarchar](15) NULL,
        [Region] [nvarchar](15) NULL,
        [PostalCode] [nvarchar](10) NULL,
        [Country] [nvarchar](15) NULL,
        [Phone] [nvarchar](24) NULL,
        [Fax] [nvarchar](24) NULL,
     CONSTRAINT [PK_Customers1] PRIMARY KEY CLUSTERED ( [CustomerID] ASC )
    );
    
    -- Inserting all the from csv to temproary table using BULK INSERT
    BULK INSERT #Customers
    FROM 'D:\Blog\ImportData\CustomerImport.csv'
    WITH ( FIELDTERMINATOR = ',', ROWTERMINATOR = '\n' );
    
    -- Selecting the records from temproary table. This is just to know the records inserted or not.
    -- SELECT * FROM #Customers;
    
    -- By using MERGE statement, inserting the record if not present and updating if exist.
    MERGE Customers AS TargetTable                            -- Inserting or Updating the table.
    USING #Customers AS SourceTable                           -- Records from the temproary table (records from csv file).
    ON (TargetTable.CustomerID = SourceTable.CustomerID)      -- Defining condition to decide which records are alredy present
    WHEN NOT MATCHED BY TARGET                                -- If the records in the Customer table is not matched?
        THEN INSERT (CustomerID, CompanyName, ContactName, ContactTitle, [Address],    -- then INSERT the record
                     City, Region, PostalCode, Country, Phone, Fax)
            VALUES(SourceTable.CustomerID, SourceTable.CompanyName, SourceTable.ContactName, SourceTable.ContactTitle, SourceTable.[Address],
                    SourceTable.City, SourceTable.Region, SourceTable.PostalCode, SourceTable.Country, SourceTable.Phone, SourceTable.Fax)
    WHEN MATCHED                                              -- If not matched then UPDATE
        THEN UPDATE SET
            TargetTable.CustomerID = SourceTable.CustomerID,
            TargetTable.CompanyName = SourceTable.CompanyName,
            TargetTable.ContactName = SourceTable.ContactName,
            TargetTable.ContactTitle = SourceTable.ContactTitle,
            TargetTable.[Address] = SourceTable.[Address],
            TargetTable.City = SourceTable.City,
            TargetTable.Region = SourceTable.Region,
            TargetTable.PostalCode = SourceTable.PostalCode,
            TargetTable.Country = SourceTable.Country,
            TargetTable.Phone = SourceTable.Phone,
            TargetTable.Fax = SourceTable.Fax;
            
    SELECT * FROM Customers;
END

The store procedure does the following points

1. Create a Temproary Table for importing the data from csv file.
2. Insert all the records from csv to Temproary Table using BULK INSERT
3. Insert the records from Temproary Table to the Customers table when not exist and Update when not exist.

Note: I keeping the CustomerImport.csv file under D:\Blog\ImportData\ path and mentioned the same in the store procedure.

Implementation for XML data:

Let us take the same example and explore using XML file as input.

The XML file looks as below:

<?xml version="1.0" encoding="utf-8" ?>
<Customers>
  <Customer>
    <CustomerID>AEHM1</CustomerID>
    <CompanyName>Around the Horn test</CompanyName>
    <ContactName>Thomas Hardy</ContactName>
    <ContactTitle>Owner</ContactTitle>
    <Address>120 Hanover Sq.</Address>
    <City>London</City>
    <Region>SP</Region>
    <PostalCode>WA1 1DP</PostalCode>
    <Country>Mexico</Country>
    <Phone>(171) 555-7788</Phone>
    <Fax>(171) 555-6750</Fax>
  </Customer>
  <Customer>
    <CustomerID>AFGK</CustomerID>
    <CompanyName>Antonio Moreno Taquer-ía</CompanyName>
    <ContactName>Antonio Moreno</ContactName>
    <ContactTitle>Owner</ContactTitle>
    <Address>Mataderos  2312</Address>
    <City>MGÇÜxico D.F.</City>
    <Region>SP</Region>
    <PostalCode>5023</PostalCode>
    <Country>Mexico</Country>
    <Phone>(5) 555-3932</Phone>
    <Fax>(5) 555-3745</Fax>
  </Customer>
<Customers>

The Store Procedure would be

CREATE PROCEDURE SP_ImportCustomerData_XML
AS
BEGIN
    -- Creating a Temproary Table for importing the data from csv file.
    CREATE TABLE #Customers(
        [CustomerID] [nchar](5) NOT NULL,
        [CompanyName] [nvarchar](40) NOT NULL,
        [ContactName] [nvarchar](30) NULL,
        [ContactTitle] [nvarchar](30) NULL,
        [Address] [nvarchar](60) NULL,
        [City] [nvarchar](15) NULL,
        [Region] [nvarchar](15) NULL,
        [PostalCode] [nvarchar](10) NULL,
        [Country] [nvarchar](15) NULL,
        [Phone] [nvarchar](24) NULL,
        [Fax] [nvarchar](24) NULL,
     CONSTRAINT [PK_Customers1] PRIMARY KEY CLUSTERED ( [CustomerID] ASC )
    );
     
    -- Inserting all the rows from xml to temproary table using OPENROWSET
    -- Thanks to http://pratchev.blogspot.com/2008/11/import-xml-file-to-sql-table.html
    INSERT INTO #Customers (CustomerID, CompanyName, ContactName, ContactTitle, [Address], City, Region, PostalCode, Country, Phone, Fax)
    SELECT X.Customer.query('CustomerID').value('.', 'nchar(5)'),       
   X.Customer.query('CompanyName').value('.', 'nvarchar(40)'),
   X.Customer.query('ContactName').value('.', 'nvarchar(30)'), 
   X.Customer.query('ContactTitle').value('.', 'nvarchar(30)'), 
   X.Customer.query('Address').value('.', 'nvarchar(60)'), 
   X.Customer.query('City').value('.', 'nvarchar(15)'), 
   X.Customer.query('Region').value('.', 'nvarchar(15)'), 
   X.Customer.query('PostalCode').value('.', 'nvarchar(10)'), 
   X.Customer.query('Country').value('.', 'nvarchar(15)'), 
   X.Customer.query('Phone').value('.', 'nvarchar(24)'), 
   X.Customer.query('Fax').value('.', 'nvarchar(24)')
 FROM 
  (SELECT CAST(x AS XML)
   FROM OPENROWSET(BULK 'D:\ImportData\XMLData1.xml',SINGLE_BLOB) AS T(x)
  ) AS T(x)
  CROSS APPLY x.nodes('Customers/Customer') AS X(Customer);
     
    -- Selecting the records from temproary table. This is just to know the records inserted or not.
    -- SELECT * FROM #Customers;
     
    -- By using MERGE statement, inserting the record if not present and updating if exist.
    MERGE Customers AS TargetTable                            -- Inserting or Updating the table.
    USING #Customers AS SourceTable                           -- Records from the temproary table (records from csv file).
    ON (TargetTable.CustomerID = SourceTable.CustomerID)      -- Defining condition to decide which records are alredy present
    WHEN NOT MATCHED BY TARGET                                -- If the records in the Customer table is not matched?
        THEN INSERT (CustomerID, CompanyName, ContactName, ContactTitle, [Address],    -- then INSERT the record
                     City, Region, PostalCode, Country, Phone, Fax)
            VALUES(SourceTable.CustomerID, SourceTable.CompanyName, SourceTable.ContactName, SourceTable.ContactTitle, SourceTable.[Address],
                    SourceTable.City, SourceTable.Region, SourceTable.PostalCode, SourceTable.Country, SourceTable.Phone, SourceTable.Fax)
    WHEN MATCHED                                              -- If not matched then UPDATE
        THEN UPDATE SET
            TargetTable.CustomerID = SourceTable.CustomerID,
            TargetTable.CompanyName = SourceTable.CompanyName,
            TargetTable.ContactName = SourceTable.ContactName,
            TargetTable.ContactTitle = SourceTable.ContactTitle,
            TargetTable.[Address] = SourceTable.[Address],
            TargetTable.City = SourceTable.City,
            TargetTable.Region = SourceTable.Region,
            TargetTable.PostalCode = SourceTable.PostalCode,
            TargetTable.Country = SourceTable.Country,
            TargetTable.Phone = SourceTable.Phone,
            TargetTable.Fax = SourceTable.Fax;
            
    SELECT * FROM Customers;
END

[Update 10-Sep-2012]
As Charles Rice comment, I am updating this post with an example for importing data into a table that has Auto increment column.

Requirement 1: There is a table which contains Country, State, City, Remarks columns. It also contains a column Id which is auto increment values. The requirement is to import these values from csv files.

The table script would be:

CREATE TABLE [dbo].[TempCity](
 [Id] [int] IDENTITY(1,1) NOT NULL,
 [Country] [varchar](50) NULL,
 [State] [varchar](50) NULL,
 [City] [varchar](50) NULL,
 [Remarks] [varchar](200) NULL
) ON [PRIMARY]
GO

The .csv file would be:

1,India,Tamil Nadu,Chennai,
2,India,Tamil Nadu,Coimbatore,
3,India,Tamil Nadu,Madurai,testremarks
4,India,Tamil Nadu,Vellore,
5,India,Karnataka,Bangalore,
6,India,Karnataka,Mangalore,
7,India,Gujarat,Ahmedabad,

Note: Add an empty line at the end.

Here, the autoincrement id act as primary key. So to identity which record to update (if exist) or insert (if not exist) - we need to include that id also in csv file (included as first column). It is important to note is, the autoincrement id will be consider to compare with already inserted records id and not for new records. So, there is no need to include exact id in the csv file. It can contain any number which is not the existing record id would be fine.

The SQL SP would be:

CREATE PROCEDURE SP_ImportCustomerData
AS
BEGIN
 CREATE TABLE #TempCity(
  [Id] [int] IDENTITY(1,1) NOT NULL,
  [Country] [varchar](50) NULL,
  [State] [varchar](50) NULL,
  [City] [varchar](50) NULL,
  [Remarks] [varchar](200) NULL
 ) ON [PRIMARY];

    BULK INSERT #TempCity
    FROM 'C:\Blog\test1.csv'
    WITH ( FIELDTERMINATOR = ',', ROWTERMINATOR = '\n' );
     
    MERGE TempCity AS TargetTable
    USING #TempCity AS SourceTable
    ON (TargetTable.Id = SourceTable.Id)
    WHEN NOT MATCHED BY TARGET
        THEN INSERT (Country, State, City, Remarks)
            VALUES(SourceTable.Country, SourceTable.State, SourceTable.City, SourceTable.City)
    WHEN MATCHED
        THEN UPDATE SET
            TargetTable.Country = SourceTable.Country,
            TargetTable.State = SourceTable.State,
            TargetTable.City = SourceTable.City,
   TargetTable.Remarks = SourceTable.Remarks;
             
    SELECT * FROM TempCity;
END

If we run this SP, this will insert/update the TempCity table from the input csv file.

Requirement 2: In some cases, the requirement is to not include the autoincrement id in the input csv/xml file. The comparision must be decided by other columns (may be more the one) to insert or update the records from csv file.

In our example, let us take the combination of Country, State, City columns are considered as identity column (primary column) for a single record. So the merging process needs to consider by combining these three column. So once a record inserted with these three values, the only possibilities is to update the Remarks column. It is important to note is the auto-increment id is not consider anywhere.

The csv file would be

India,Tamil Nadu,Chennai,
India,Tamil Nadu,Coimbatore,
India,Tamil Nadu,Madurai,
India,Tamil Nadu,Vellore,testremarks
India,Karnataka,Bangalore,
India,Karnataka,Mangalore,
India,Gujarat,Ahmedabad,

Here, I am not passing the value of Id column in the csv file. Note: Add an empty line at the end.

The SQL SP would be :

CREATE PROCEDURE SP_ImportCustomerData
AS
BEGIN
 CREATE TABLE #TempCity(
  [Country] [varchar](50) NULL,
  [State] [varchar](50) NULL,
  [City] [varchar](50) NULL,
  [Remarks] [varchar](200) NULL
 ) ON [PRIMARY];

    BULK INSERT #TempCity
    FROM 'C:\Blog\test2.csv'
    WITH ( FIELDTERMINATOR = ',', ROWTERMINATOR = '\n' );
     
    MERGE TempCity AS TargetTable
    USING #TempCity AS SourceTable
    ON (TargetTable.Country = SourceTable.Country)
 AND (TargetTable.State = SourceTable.State)
 AND (TargetTable.City = SourceTable.City)
    WHEN NOT MATCHED BY TARGET
        THEN INSERT (Country, State, City, Remarks)
            VALUES(SourceTable.Country, SourceTable.State, SourceTable.City, SourceTable.Remarks)
    WHEN MATCHED
        THEN UPDATE SET
   TargetTable.Remarks = SourceTable.Remarks;
    
    SELECT * FROM TempCity;
END

Below are the screen shot of execution job

Table Creation

SP execution at first time (with 76 records in the csv file) - All records are inserted

Adding some more records in the csv file and updating some existing records

Inserted the additional records and updated that 2 records

The updated records are pointed out

The output for XML input will the same as csv input, only the difference is the input file.

When using MERGE statement, there are some points required to keep in mind.

1. Create an index on the join columns in the source table that is unique and covering.
2. Create a unique clustered index on the join columns in the target table.

Refer the below url for more information http://technet.microsoft.com/en-us/library/cc879317.aspx

Database Pagination in GridView using SQL Store Procedure

I have blogged a post on Database pagination in GridView using nHibernate before. But if someone not using nHibernate and using Database Store Procedure, this post will be useful. In this post I am using SQL Server 2008 (This method also will work SQL Server 2005 and higher versions).

I am using Northwind database for this example, so please make sure you installed the same database to test the code.

I have three projects for this example (created with layers as normal business applications have.. )

1. DotNetTwitter.Entities - This project is used for defining the business entity class.
2. DotNetTwitter.DataAccess - This project is used for database operation. Here is where the records are fetched for the required page.
3. DotNetTwitter.DBPagination - Web Application which contains GridView to show the records

The implementation as follows:

Database Script (To create ProductViewview).

Create View [dbo].[ProductView]
As
Select Products.ProductID,
  Products.ProductName,
  Suppliers.CompanyName,
  Categories.CategoryName,
  Products.QuantityPerUnit,
  Products.UnitPrice,
  Products.UnitsInStock,
  Products.UnitsOnOrder,
  Products.ReorderLevel
from Products
Join Suppliers on Suppliers.SupplierID = Products.SupplierID
Join Categories on Categories.CategoryID = Products.CategoryID
GO

Store Procedure

CREATE PROCEDURE [dbo].[SP_GET_PRODUCTS]
@CurrentPageNo AS INT,
@PageRecordsCount AS INT,
@SortBy AS NVARCHAR(250),
@SortType AS NVARCHAR(250),
@TotalRecordCount AS INT OUTPUT
AS

SET NOCOUNT ON

DECLARE @SQL AS NVARCHAR(1000)
DECLARE @RecordsFrom AS INT
DECLARE @RecordsTo AS INT

-- RecordsFrom => For Ex: CurrentPage 2, PageRecordCount = 10, then (((2-1) * 10) + 1) => ((1 * 10) + 1) => 11
SET @RecordsFrom = (((@CurrentPageNo - 1) * @PageRecordsCount) + 1) 

-- @RecordsTo => For Ex: CurrentPage 2, PageRecordCount = 10, then (2 * 10) => 20
SET @RecordsTo = (@CurrentPageNo * @PageRecordsCount)

IF @SortType = '' OR @SortType = NULL
 SET @SortType = 'Asc'

-- Getting the Total Record Count
SELECT @TotalRecordCount = COUNT(*) FROM ProductView

-- Getting the records for a particular page
SET @SQL = 
  'SELECT * FROM ' +
  '( ' +
  ' SELECT ' +
  '  ROW_NUMBER() OVER (ORDER BY ' + @SortBy + ' ' +  @SortType + ') AS RowNumber, *' +
  ' FROM ProductView ' +
  ') AS RECORDS ' +
  'Where RowNumber between ' + CAST(@RecordsFrom AS VARCHAR) + ' and ' + CAST(@RecordsTo AS VARCHAR)

EXECUTE(@SQL)

Note: As I have created a View, I have used View in SP. Incase if you don’t follow creating View, you can use the following Store Procedure (infact both are same).

CREATE PROCEDURE [dbo].[SP_GET_PRODUCTS]
@CurrentPageNo AS INT,
@PageRecordsCount AS INT,
@SortBy AS NVARCHAR(250),
@SortType AS NVARCHAR(250),
@TotalRecordCount AS INT OUTPUT
AS

SET NOCOUNT ON

DECLARE @SQL AS NVARCHAR(1000)
DECLARE @RecordsFrom AS INT
DECLARE @RecordsTo AS INT

-- RecordsFrom => For Ex: CurrentPage 2, PageRecordCount = 10, then (((2-1) * 10) + 1) => ((1 * 10) + 1) => 11
SET @RecordsFrom = (((@CurrentPageNo - 1) * @PageRecordsCount) + 1) 

-- @RecordsTo => For Ex: CurrentPage 2, PageRecordCount = 10, then (2 * 10) => 20
SET @RecordsTo = (@CurrentPageNo * @PageRecordsCount)

IF @SortType = '' OR @SortType = NULL
 SET @SortType = 'Asc'

-- Getting the Total Record Count
-- SELECT @TotalRecordCount = COUNT(*) FROM ProductView

-- Getting the records for a particular page
SET @SQL = 
  'SELECT * FROM ' +
  '( ' +
  ' SELECT ' +
  '  ROW_NUMBER() OVER (ORDER BY ' + @SortBy + ' ' + @SortType + ') AS RowNumber, ' +
  '  Products.ProductID, ' +
  '  Products.ProductName, ' +
  '  Suppliers.CompanyName, ' +
  '  Categories.CategoryName, ' +
  '  Products.QuantityPerUnit, ' +
  '  Products.UnitPrice, ' +
  '  Products.UnitsInStock, ' +
  '  Products.UnitsOnOrder, ' +
  '  Products.ReorderLevel ' +
  ' FROM Products ' +
  ' INNER JOIN Suppliers ON Suppliers.SupplierID = Products.SupplierID ' +
  ' INNER JOIN Categories ON Categories.CategoryID = Products.CategoryID ' +
  ') AS RECORDS ' +
  'Where RowNumber between ' + CAST(@RecordsFrom AS VARCHAR) + ' and ' + CAST(@RecordsTo AS VARCHAR)
  
 EXECUTE(@SQL)
SET NOCOUNT OFF

I have three button and one text box in my gridview to navigate between pages (Next, Previous, Go and a Text box to enter PageNo). The Codebehind would be as

//C# Code behind
/// <summary>
/// Method which binds the data to the Grid
/// </summary>
private void BindGrid()
{
    ProductDAO objProductDAO = new ProductDAO();

    // Defining int variable to get the Total Record Count from Data layer
    int totalRecordCount = 0;

    // Get the SortByExpression and SortType from Hidden Control (by default, that will be updated in script)
    string strSortExpression = ViewState["SortExpression"].ToString().Split(",".ToCharArray())[0];
    SortType sortType = (SortType)Enum.Parse(typeof(SortType), ViewState["SortExpression"].ToString().Split(",".ToCharArray())[1], true);

    // Getting how many records required to show in the Grid per page
    int intPageRecordCount = Convert.ToInt32(ConfigurationManager.AppSettings["PageRecordCount"].ToString());

    IList<ProductView> ProductViewList = objProductDAO.GetProducts(Convert.ToInt32(ViewState["CurrentPage"].ToString()), intPageRecordCount, strSortExpression, sortType, out totalRecordCount);
    //Adding one empty row for just to show the grid
    if (ProductViewList.Count == 0)
        ProductViewList.Add(new ProductView()); 
    
    grdViewProducts.DataSource = ProductViewList;
    grdViewProducts.DataBind();

    grdViewProducts.BottomPagerRow.Visible = true;

    // Assign the Total Record count to 
    ViewState["TotalRecords"] = totalRecordCount.ToString();

    Label lblPageInfo = grdViewProducts.BottomPagerRow.FindControl("lblPageInfo") as Label;
    lblPageInfo.Text = "Page " + Convert.ToInt32(ViewState["CurrentPage"].ToString()).ToString() + " out of " + ((totalRecordCount % intPageRecordCount) > 0 ? (totalRecordCount / intPageRecordCount) + 1 : (totalRecordCount / intPageRecordCount)).ToString();

    // Try to find the sorted column
    for (int intRowIndex = 0; intRowIndex < grdViewProducts.Columns.Count; intRowIndex++)
    {
        if (strSortExpression == grdViewProducts.Columns[intRowIndex].SortExpression)
            ((LinkButton)grdViewProducts.HeaderRow.Cells[intRowIndex].Controls[0])
            .CssClass = (sortType == SortType.Ascending ? "sortup" : "sortdown");
    }
}

/// <summary>
/// Call when clicking Previous button on pagination
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
protected void btnPrevious_Click(object sender, EventArgs e)
{
    if (Convert.ToInt32(ViewState["CurrentPage"].ToString()) > 1)
        ViewState["CurrentPage"] = (Convert.ToInt32(ViewState["CurrentPage"].ToString()) - 1).ToString();
    else
        ViewState["CurrentPage"] = "1";
    BindGrid();
}

/// <summary>
/// Call when clicking Go button on pagination
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
protected void btnGo_Click(object sender, EventArgs e)
{
    int intPageRecordCount = Convert.ToInt32(ConfigurationManager.AppSettings["PageRecordCount"].ToString());
    TextBox txtGoPage = ((TextBox)((ImageButton)sender).Parent.FindControl("txtGoPage"));
    if (txtGoPage.Text.Trim().Length > 0)
    {
        if ((Convert.ToInt32(txtGoPage.Text) > 0) && (Convert.ToInt32(txtGoPage.Text) <= ((Convert.ToInt32(ViewState["TotalRecords"].ToString()) % intPageRecordCount) > 0 ? (Convert.ToInt32(ViewState["TotalRecords"].ToString()) / intPageRecordCount) + 1 : (Convert.ToInt32(ViewState["TotalRecords"].ToString()) / intPageRecordCount))))
        {
            ViewState["CurrentPage"] = Convert.ToInt32(txtGoPage.Text).ToString();
            BindGrid();
        }
    }
}

/// <summary>
/// Call when clicking Next button on pagination
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
protected void btnNext_Click(object sender, EventArgs e)
{
    int intPageRecordCount = Convert.ToInt32(ConfigurationManager.AppSettings["PageRecordCount"].ToString());
    if (Convert.ToInt32(ViewState["CurrentPage"].ToString()) < ((Convert.ToInt32(ViewState["TotalRecords"].ToString()) % intPageRecordCount) > 0 ? (Convert.ToInt32(ViewState["TotalRecords"].ToString()) / intPageRecordCount) + 1 : (Convert.ToInt32(ViewState["TotalRecords"].ToString()) / intPageRecordCount)))
        ViewState["CurrentPage"] = (Convert.ToInt32(ViewState["CurrentPage"].ToString()) + 1).ToString();
    else
        ViewState["CurrentPage"] = (Convert.ToInt32(ViewState["TotalRecords"].ToString()) / intPageRecordCount).ToString();
    BindGrid();
}
/// <summary>
/// Sorts the GridView by clicked column.
/// </summary>
/// <param name="sender">The event sender.</param>
/// <param name="e">The event argument.</param>
protected void grdViewProducts_Sorting(object sender, GridViewSortEventArgs e)
{
    string strSortExpression = ViewState["SortExpression"].ToString().Split(",".ToCharArray())[0];
    SortType sortType = (SortType)Enum.Parse(typeof(SortType), ViewState["SortExpression"].ToString().Split(",".ToCharArray())[1], true);
    ViewState["CurrentPage"] = "1";

    if (strSortExpression == e.SortExpression)
        sortType = sortType == SortType.Ascending ? SortType.Descending : SortType.Ascending;
    else
    {
        strSortExpression = e.SortExpression;
        sortType = SortType.Ascending;
    }
    ViewState["SortExpression"] = e.SortExpression + "," + sortType.ToString();

    BindGrid();
}

'VB Code behind
    ''' <summary>
    ''' Method which binds the data to the Grid
    ''' </summary>
    Private Sub BindGrid()
        Dim objProductDAO As New ProductDAO()

        ' Defining int variable to get the Total Record Count from Data layer
        Dim totalRecordCount As Integer = 0

        ' Get the SortByExpression and SortType from Hidden Control (by default, that will be updated in script)
        Dim strSortExpression As String = ViewState("SortExpression").ToString().Split(",".ToCharArray())(0)
        Dim sortType As SortType = DirectCast([Enum].Parse(GetType(SortType), ViewState("SortExpression").ToString().Split(",".ToCharArray())(1), True), SortType)

        ' Getting how many records required to show in the Grid per page
        Dim intPageRecordCount As Integer = Convert.ToInt32(ConfigurationManager.AppSettings("PageRecordCount").ToString())

        Dim ProductViewList As IList(Of ProductView) = objProductDAO.GetProducts(Convert.ToInt32(ViewState("CurrentPage").ToString()), intPageRecordCount, strSortExpression, sortType, totalRecordCount)
        'Adding one empty row for just to show the grid
        If ProductViewList.Count = 0 Then
            ProductViewList.Add(New ProductView())
        End If

        grdViewProducts.DataSource = ProductViewList
        grdViewProducts.DataBind()

        grdViewProducts.BottomPagerRow.Visible = True

        ' Assign the Total Record count to 
        ViewState("TotalRecords") = totalRecordCount.ToString()

        Dim lblPageInfo As Label = TryCast(grdViewProducts.BottomPagerRow.FindControl("lblPageInfo"), Label)
        lblPageInfo.Text = "Page " & Convert.ToInt32(ViewState("CurrentPage").ToString()).ToString() & " out of " & (If((totalRecordCount Mod intPageRecordCount) > 0, (totalRecordCount \ intPageRecordCount) + 1, (totalRecordCount \ intPageRecordCount))).ToString()

        ' Try to find the sorted column
        For intRowIndex As Integer = 0 To grdViewProducts.Columns.Count - 1
            If strSortExpression = grdViewProducts.Columns(intRowIndex).SortExpression Then
                DirectCast(grdViewProducts.HeaderRow.Cells(intRowIndex).Controls(0), LinkButton).CssClass = (If(sortType = sortType.Ascending, "sortup", "sortdown"))
            End If
        Next
    End Sub

    ''' <summary>
    ''' Call when clicking Previous button on pagination
    ''' </summary>
    ''' <param name="sender"></param>
    ''' <param name="e"></param>
    Protected Sub btnPrevious_Click(ByVal sender As Object, ByVal e As System.Web.UI.ImageClickEventArgs)
        If Convert.ToInt32(ViewState("CurrentPage").ToString()) > 1 Then
            ViewState("CurrentPage") = (Convert.ToInt32(ViewState("CurrentPage").ToString()) - 1).ToString()
        Else
            ViewState("CurrentPage") = "1"
        End If
        BindGrid()
    End Sub

    ''' <summary>
    ''' Call when clicking Go button on pagination
    ''' </summary>
    ''' <param name="sender"></param>
    ''' <param name="e"></param>
    Protected Sub btnGo_Click(ByVal sender As Object, ByVal e As System.Web.UI.ImageClickEventArgs)
        Dim intPageRecordCount As Integer = Convert.ToInt32(ConfigurationManager.AppSettings("PageRecordCount").ToString())
        Dim txtGoPage As TextBox = DirectCast(DirectCast(sender, ImageButton).Parent.FindControl("txtGoPage"), TextBox)
        If txtGoPage.Text.Trim().Length > 0 Then
            If (Convert.ToInt32(txtGoPage.Text) > 0) AndAlso (Convert.ToInt32(txtGoPage.Text) <= (If((Convert.ToInt32(ViewState("TotalRecords").ToString()) Mod intPageRecordCount) > 0, (Convert.ToInt32(ViewState("TotalRecords").ToString()) / intPageRecordCount) + 1, (Convert.ToInt32(ViewState("TotalRecords").ToString()) / intPageRecordCount)))) Then
                ViewState("CurrentPage") = Convert.ToInt32(txtGoPage.Text).ToString()
                BindGrid()
            End If
        End If
    End Sub

    ''' <summary>
    ''' Call when clicking Next button on pagination
    ''' </summary>
    ''' <param name="sender"></param>
    ''' <param name="e"></param>
    Protected Sub btnNext_Click(ByVal sender As Object, ByVal e As System.Web.UI.ImageClickEventArgs)
        Dim intPageRecordCount As Integer = Convert.ToInt32(ConfigurationManager.AppSettings("PageRecordCount").ToString())
        If Convert.ToInt32(ViewState("CurrentPage").ToString()) < (If((Convert.ToInt32(ViewState("TotalRecords").ToString()) Mod intPageRecordCount) > 0, (Convert.ToInt32(ViewState("TotalRecords").ToString()) / intPageRecordCount) + 1, (Convert.ToInt32(ViewState("TotalRecords").ToString()) / intPageRecordCount))) Then
            ViewState("CurrentPage") = (Convert.ToInt32(ViewState("CurrentPage").ToString()) + 1).ToString()
        Else
            ViewState("CurrentPage") = (Convert.ToInt32(ViewState("TotalRecords").ToString()) / intPageRecordCount).ToString()
        End If
        BindGrid()
    End Sub

    ''' <summary>
    ''' Sorts the GridView by clicked column.
    ''' </summary>
    ''' <param name="sender">The event sender.</param>
    ''' <param name="e">The event argument.</param>
    Protected Sub grdViewProducts_Sorting(ByVal sender As Object, ByVal e As GridViewSortEventArgs)
        Dim strSortExpression As String = ViewState("SortExpression").ToString().Split(",".ToCharArray())(0)
        Dim sortType As SortType = DirectCast([Enum].Parse(GetType(SortType), ViewState("SortExpression").ToString().Split(",".ToCharArray())(1), True), SortType)
        ViewState("CurrentPage") = "1"

        If strSortExpression = e.SortExpression Then
            sortType = If(sortType = sortType.Ascending, sortType.Descending, sortType.Ascending)
        Else
            strSortExpression = e.SortExpression
            sortType = sortType.Ascending
        End If
        ViewState("SortExpression") = Convert.ToString(e.SortExpression) & "," & sortType.ToString()

        BindGrid()
    End Sub

Below is the code from data access class, which returns only records for particular page.

//C# Code
public IList<ProductView> GetProducts(int currentPageNo, int pageRecordsCount, string sortBy, SortType sortType, out int totalRecordCount)
{
    SqlConnection connection = null;
    SqlDataReader dataReader = null;

    try
    {
        IList<ProductView> productViewList = new List<ProductView>();
        connection = new SqlConnection(ConfigurationManager.ConnectionStrings["ConnectionString"].ToString());

        SqlDataAdapter dataAdaptor = new SqlDataAdapter("SP_GET_PRODUCTS", connection);
        dataAdaptor.SelectCommand.CommandType = System.Data.CommandType.StoredProcedure;

        dataAdaptor.SelectCommand.Parameters.Add(new SqlParameter("@CurrentPageNo", SqlDbType.Int));
        dataAdaptor.SelectCommand.Parameters["@CurrentPageNo"].Value = currentPageNo;

        dataAdaptor.SelectCommand.Parameters.Add(new SqlParameter("@PageRecordsCount", SqlDbType.Int));
        dataAdaptor.SelectCommand.Parameters["@PageRecordsCount"].Value = pageRecordsCount;

        dataAdaptor.SelectCommand.Parameters.Add(new SqlParameter("@SortBy", SqlDbType.VarChar));
        dataAdaptor.SelectCommand.Parameters["@SortBy"].Value = sortBy;

        dataAdaptor.SelectCommand.Parameters.Add(new SqlParameter("@SortType", SqlDbType.VarChar));
        dataAdaptor.SelectCommand.Parameters["@SortType"].Value = (sortType == SortType.Ascending ? "Asc" : "Desc");

        dataAdaptor.SelectCommand.Parameters.Add(new SqlParameter("@TotalRecordCount", SqlDbType.Int));
        dataAdaptor.SelectCommand.Parameters["@TotalRecordCount"].Direction = ParameterDirection.Output;
        connection.Open();

        DataSet dataSet = new DataSet();
        dataAdaptor.Fill(dataSet);

        foreach(DataRow dataRow in dataSet.Tables[0].Rows)
        {
            ProductView productView = new ProductView();
            productView.ProductID = Convert.ToInt32(dataRow["ProductID"].ToString());
            productView.ProductName = dataRow["ProductName"].ToString();
            productView.CompanyName = dataRow["CompanyName"].ToString();
            productView.CategoryName = dataRow["CategoryName"].ToString();
            productView.QuantityPerUnit = dataRow["QuantityPerUnit"].ToString();
            productView.UnitPrice = Convert.ToDouble(dataRow["UnitPrice"].ToString());
            productView.UnitsInStock = Convert.ToInt32(dataRow["UnitsInStock"].ToString());
            productView.UnitsOnOrder = Convert.ToInt32(dataRow["UnitsOnOrder"].ToString());
            productView.ReorderLevel = Convert.ToInt32(dataRow["ReorderLevel"].ToString());

            productViewList.Add(productView);
        }

        totalRecordCount = Convert.ToInt32(dataAdaptor.SelectCommand.Parameters["@TotalRecordCount"].Value);
        connection.Close();

        //return
        return productViewList;
    }
    catch (Exception ex)
    {
        if (connection != null) connection.Close();
        if (dataReader != null) dataReader.Close();

        // Log
        throw ex;
    }
}

'VB Code
    Public Function GetProducts(ByVal currentPageNo As Integer, ByVal pageRecordsCount As Integer, ByVal sortBy As String, ByVal sortType__1 As SortType, ByRef totalRecordCount As Integer) As IList(Of ProductView)
        Dim connection As SqlConnection = Nothing
        Dim dataReader As SqlDataReader = Nothing

        Try
            Dim productViewList As IList(Of ProductView) = New List(Of ProductView)()
            connection = New SqlConnection(ConfigurationManager.ConnectionStrings("ConnectionString").ToString())

            Dim dataAdaptor As New SqlDataAdapter("SP_GET_PRODUCTS", connection)
            dataAdaptor.SelectCommand.CommandType = System.Data.CommandType.StoredProcedure

            dataAdaptor.SelectCommand.Parameters.Add(New SqlParameter("@CurrentPageNo", SqlDbType.Int))
            dataAdaptor.SelectCommand.Parameters("@CurrentPageNo").Value = currentPageNo

            dataAdaptor.SelectCommand.Parameters.Add(New SqlParameter("@PageRecordsCount", SqlDbType.Int))
            dataAdaptor.SelectCommand.Parameters("@PageRecordsCount").Value = pageRecordsCount

            dataAdaptor.SelectCommand.Parameters.Add(New SqlParameter("@SortBy", SqlDbType.VarChar))
            dataAdaptor.SelectCommand.Parameters("@SortBy").Value = sortBy

            dataAdaptor.SelectCommand.Parameters.Add(New SqlParameter("@SortType", SqlDbType.VarChar))
            dataAdaptor.SelectCommand.Parameters("@SortType").Value = (If(sortType__1 = SortType.Ascending, "Asc", "Desc"))

            dataAdaptor.SelectCommand.Parameters.Add(New SqlParameter("@TotalRecordCount", SqlDbType.Int))
            dataAdaptor.SelectCommand.Parameters("@TotalRecordCount").Direction = ParameterDirection.Output
            connection.Open()

            Dim dataSet As New DataSet()
            dataAdaptor.Fill(dataSet)

            For Each dataRow As DataRow In dataSet.Tables(0).Rows
                Dim productView As New ProductView()
                productView.ProductID = Convert.ToInt32(dataRow("ProductID").ToString())
                productView.ProductName = dataRow("ProductName").ToString()
                productView.CompanyName = dataRow("CompanyName").ToString()
                productView.CategoryName = dataRow("CategoryName").ToString()
                productView.QuantityPerUnit = dataRow("QuantityPerUnit").ToString()
                productView.UnitPrice = Convert.ToDouble(dataRow("UnitPrice").ToString())
                productView.UnitsInStock = Convert.ToInt32(dataRow("UnitsInStock").ToString())
                productView.UnitsOnOrder = Convert.ToInt32(dataRow("UnitsOnOrder").ToString())
                productView.ReorderLevel = Convert.ToInt32(dataRow("ReorderLevel").ToString())

                productViewList.Add(productView)
            Next

            totalRecordCount = Convert.ToInt32(dataAdaptor.SelectCommand.Parameters("@TotalRecordCount").Value)
            connection.Close()

            'return
            Return productViewList
        Catch ex As Exception
            If connection IsNot Nothing Then
                connection.Close()
            End If
            If dataReader IsNot Nothing Then
                dataReader.Close()
            End If

            ' Log
            Throw ex
        End Try
    End Function

This code has been tested with IE 6.0/8.0, Chrome 10.0, Firefox 3.6, Opera 11.01

Here is the output of the example.

Initial Screen (by default Product Name sorted with first page)

Sorted by Category

Sorted in Desc order and showing 3rd page

Skip to 4th page (usage of Go page)

You can see the output in video here

download the working example of the source code in C# here and in VB here