diff --git a/oracle-plugin/src/e2e-test/features/sink/OracleRunTime.feature b/oracle-plugin/src/e2e-test/features/sink/OracleRunTime.feature
index c2b56e8b7..67293700b 100644
--- a/oracle-plugin/src/e2e-test/features/sink/OracleRunTime.feature
+++ b/oracle-plugin/src/e2e-test/features/sink/OracleRunTime.feature
@@ -117,3 +117,53 @@ Feature: Oracle - Verify data transfer from BigQuery source to Oracle sink
     Then Verify the pipeline status is "Succeeded"
     Then Validate records transferred to target table with record counts of BigQuery table
     Then Validate the values of records transferred to target Oracle table is equal to the values from source BigQuery table
+
+ @BQ_SOURCE_TEST_SMALL_CASE @ORACLE_TEST_TABLE
+  Scenario: To verify data is getting transferred from BigQuery source to Oracle sink successfully when schema is coming in small case
+    Given Open Datafusion Project to configure pipeline
+    When Expand Plugin group in the LHS plugins list: "Source"
+    When Select plugin: "BigQuery" from the plugins list as: "Source"
+    When Expand Plugin group in the LHS plugins list: "Sink"
+    When Select plugin: "Oracle" from the plugins list as: "Sink"
+    Then Connect plugins: "BigQuery" and "Oracle" to establish connection
+    Then Navigate to the properties page of plugin: "BigQuery"
+    Then Replace input plugin property: "project" with value: "projectId"
+    Then Enter input plugin property: "datasetProject" with value: "projectId"
+    Then Enter input plugin property: "referenceName" with value: "BQReferenceName"
+    Then Enter input plugin property: "dataset" with value: "dataset"
+    Then Enter input plugin property: "table" with value: "bqSourceTable"
+    Then Click on the Get Schema button
+    Then Verify the Output Schema matches the Expected Schema: "bqOutputDatatypesSchemaSmallCase"
+    Then Validate "BigQuery" plugin properties
+    Then Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "Oracle"
+    Then Select dropdown plugin property: "select-jdbcPluginName" with option value: "driverName"
+    Then Replace input plugin property: "host" with value: "host" for Credentials and Authorization related fields
+    Then Replace input plugin property: "port" with value: "port" for Credentials and Authorization related fields
+    Then Replace input plugin property: "user" with value: "username" for Credentials and Authorization related fields
+    Then Replace input plugin property: "password" with value: "password" for Credentials and Authorization related fields
+    Then Select radio button plugin property: "connectionType" with value: "service"
+    Then Select radio button plugin property: "role" with value: "normal"
+    Then Enter input plugin property: "referenceName" with value: "sourceRef"
+    Then Replace input plugin property: "database" with value: "databaseName"
+    Then Replace input plugin property: "tableName" with value: "targetTable"
+    Then Replace input plugin property: "dbSchemaName" with value: "schema"
+    Then Replace input plugin property: "user" with value: "username" for Credentials and Authorization related fields
+    Then Replace input plugin property: "password" with value: "password" for Credentials and Authorization related fields
+    Then Enter input plugin property: "referenceName" with value: "targetRef"
+    Then Select radio button plugin property: "connectionType" with value: "service"
+    Then Select radio button plugin property: "role" with value: "normal"
+    Then Validate "Oracle" plugin properties
+    Then Close the Plugin Properties page
+    Then Save the pipeline
+    Then Preview and run the pipeline
+    Then Verify the preview of pipeline is "success"
+    Then Click on preview data for Oracle sink
+    Then Close the preview data
+    Then Deploy the pipeline
+    Then Run the Pipeline in Runtime
+    Then Wait till pipeline is in running state
+    Then Open and capture logs
+    Then Verify the pipeline status is "Succeeded"
+    Then Validate records transferred to target table with record counts of BigQuery table
+    Then Validate the values of records transferred to target Oracle table is equal to the values from source BigQuery table with case
diff --git a/oracle-plugin/src/e2e-test/java/io.cdap.plugin/BQValidation.java b/oracle-plugin/src/e2e-test/java/io.cdap.plugin/BQValidation.java
index 6edfcc8fd..b7d93c80a 100644
--- a/oracle-plugin/src/e2e-test/java/io.cdap.plugin/BQValidation.java
+++ b/oracle-plugin/src/e2e-test/java/io.cdap.plugin/BQValidation.java
@@ -68,11 +68,12 @@ public static boolean validateDBToBQRecordValues(String schema, String sourceTab
                                                      ResultSet.HOLD_CURSORS_OVER_COMMIT);
 
       ResultSet rsSource = statement1.executeQuery(getSourceQuery);
-      return compareResultSetAndJsonData(rsSource, jsonResponse);
+      return compareResultSetAndJsonData(rsSource, jsonResponse, false);
     }
   }
 
-  public static boolean validateBQToDBRecordValues(String schema, String sourceTable, String targetTable)
+  public static boolean validateBQToDBRecordValues(String schema, String sourceTable, String targetTable,
+                                                   boolean isSchemaSmallCase)
     throws SQLException, ClassNotFoundException, ParseException, IOException, InterruptedException {
     List<JsonObject> jsonResponse = new ArrayList<>();
     List<Object> bigQueryRows = new ArrayList<>();
@@ -88,7 +89,7 @@ public static boolean validateBQToDBRecordValues(String schema, String sourceTab
                                                      ResultSet.HOLD_CURSORS_OVER_COMMIT);
 
       ResultSet rsTarget = statement1.executeQuery(getTargetQuery);
-      return compareResultSetAndJsonData(rsTarget, jsonResponse);
+      return compareResultSetAndJsonData(rsTarget, jsonResponse, isSchemaSmallCase);
     }
   }
 
@@ -119,7 +120,8 @@ private static void getBigQueryTableData(String table, List<Object> bigQueryRows
    * @throws ParseException If an error occurs while parsing the data.
    */
 
-  public static boolean compareResultSetAndJsonData(ResultSet rsSource, List<JsonObject> bigQueryData)
+  public static boolean compareResultSetAndJsonData(ResultSet rsSource, List<JsonObject> bigQueryData,
+                                                    boolean isSchemaSmallCase)
     throws SQLException, ParseException {
     ResultSetMetaData mdSource = rsSource.getMetaData();
     boolean result = false;
@@ -146,7 +148,8 @@ public static boolean compareResultSetAndJsonData(ResultSet rsSource, List<JsonO
       while (currentColumnCount <= columnCountSource) {
         String columnTypeName = mdSource.getColumnTypeName(currentColumnCount);
         int columnType = mdSource.getColumnType(currentColumnCount);
-        String columnName = mdSource.getColumnName(currentColumnCount);
+        String columnName = isSchemaSmallCase ? mdSource.getColumnName(currentColumnCount).toLowerCase() :
+          mdSource.getColumnName(currentColumnCount);
         // Perform different comparisons based on column type
         switch (columnType) {
           // Since we skip BFILE in Oracle Sink, we are not comparing the BFILE source and sink values
diff --git a/oracle-plugin/src/e2e-test/java/io.cdap.plugin/common.stepsdesign/TestSetupHooks.java b/oracle-plugin/src/e2e-test/java/io.cdap.plugin/common.stepsdesign/TestSetupHooks.java
index 9a6fb9cdf..c66a001b7 100644
--- a/oracle-plugin/src/e2e-test/java/io.cdap.plugin/common.stepsdesign/TestSetupHooks.java
+++ b/oracle-plugin/src/e2e-test/java/io.cdap.plugin/common.stepsdesign/TestSetupHooks.java
@@ -372,4 +372,18 @@ private static void createSourceBQTableWithQueries(String bqCreateTableQueryFile
     PluginPropertyUtils.addPluginProp("bqSourceTable", bqSourceTable);
     BeforeActions.scenario.write("BQ Source Table " + bqSourceTable + " created successfully");
   }
+
+  @Before(order = 1, value = "@BQ_SOURCE_TEST_SMALL_CASE")
+  public static void createTempSourceBQTableSmallCase() throws IOException, InterruptedException {
+    createSourceBQTableWithQueries(PluginPropertyUtils.pluginProp("CreateBQTableQueryFileSmallCase"),
+                                   PluginPropertyUtils.pluginProp("InsertBQDataQueryFileSmallCase"));
+  }
+
+  @After(order = 1, value = "@BQ_SOURCE_TEST_SMALL_CASE")
+  public static void deleteTempSourceBQTableSmallCase() throws IOException, InterruptedException {
+    String bqSourceTable = PluginPropertyUtils.pluginProp("bqSourceTable");
+    BigQueryClient.dropBqQuery(bqSourceTable);
+    BeforeActions.scenario.write("BQ source Table " + bqSourceTable + " deleted successfully");
+    PluginPropertyUtils.removePluginProp("bqSourceTable");
+  }
 }
diff --git a/oracle-plugin/src/e2e-test/java/io.cdap.plugin/oracle/stepsdesign/Oracle.java b/oracle-plugin/src/e2e-test/java/io.cdap.plugin/oracle/stepsdesign/Oracle.java
index 711c1fad0..2cf9c7e66 100644
--- a/oracle-plugin/src/e2e-test/java/io.cdap.plugin/oracle/stepsdesign/Oracle.java
+++ b/oracle-plugin/src/e2e-test/java/io.cdap.plugin/oracle/stepsdesign/Oracle.java
@@ -94,7 +94,26 @@ public void validateTheValuesOfRecordsTransferredToTargetOracleTableIsEqualToThe
 
     boolean recordsMatched = BQValidation.validateBQToDBRecordValues(PluginPropertyUtils.pluginProp("schema"),
                                                              PluginPropertyUtils.pluginProp("bqSourceTable"),
-                                                             PluginPropertyUtils.pluginProp("targetTable"));
+                                                                     PluginPropertyUtils.pluginProp("targetTable"),
+                                                                     false);
+    Assert.assertTrue("Value of records transferred to the target table should be equal to the value " +
+                        "of the records in the source table", recordsMatched);
+  }
+
+  @Then("Validate the values of records transferred to target Oracle table is equal to the values from source " +
+    "BigQuery table with case")
+  public void
+  validateTheValuesOfRecordsTransferredToTargetOracleTableIsEqualToTheValuesFromSourceBigQueryTableWithCase()
+    throws IOException, InterruptedException, IOException, SQLException, ClassNotFoundException, ParseException {
+    int sourceBQRecordsCount = BigQueryClient.countBqQuery(PluginPropertyUtils.pluginProp("bqSourceTable"));
+    BeforeActions.scenario.write("No of Records from source BigQuery table:" + sourceBQRecordsCount);
+    Assert.assertEquals("Out records should match with target Oracle table records count",
+                        CdfPipelineRunAction.getCountDisplayedOnSourcePluginAsRecordsOut(), sourceBQRecordsCount);
+
+    boolean recordsMatched = BQValidation.validateBQToDBRecordValues(PluginPropertyUtils.pluginProp("schema"),
+                                                                     PluginPropertyUtils.pluginProp("bqSourceTable"),
+                                                                     PluginPropertyUtils.pluginProp("targetTable"),
+                                                                     true);
     Assert.assertTrue("Value of records transferred to the target table should be equal to the value " +
                         "of the records in the source table", recordsMatched);
   }
diff --git a/oracle-plugin/src/e2e-test/resources/pluginParameters.properties b/oracle-plugin/src/e2e-test/resources/pluginParameters.properties
index 88dc3fe34..5362faba6 100644
--- a/oracle-plugin/src/e2e-test/resources/pluginParameters.properties
+++ b/oracle-plugin/src/e2e-test/resources/pluginParameters.properties
@@ -98,6 +98,7 @@ importQuery=where $CONDITIONS
 projectId=cdf-athena
 dataset=test_automation
 bqOutputDatatypesSchema=[{"key":"ID","value":"decimal"},{"key":"LASTNAME","value":"string"}]
+bqOutputDatatypesSchemaSmallCase=[{"key":"id","value":"decimal"},{"key":"lastname","value":"string"}]
 jdbcUrl=jdbc:bigquery://https://www.googleapis.com/bigquery/v2:443;ProjectId=%s;OAuthType=3;
 
 #bq macro properties
@@ -107,6 +108,9 @@ bqUpdateTableSchema=true
 #bq queries file path
 CreateBQTableQueryFile=testdata/BigQuery/BigQueryCreateTableQuery.txt
 InsertBQDataQueryFile=testdata/BigQuery/BigQueryInsertDataQuery.txt
+#bq queries file path for Small Case Schema
+CreateBQTableQueryFileSmallCase=testdata/BigQuery/BigQueryCreateTableQuerySmallCase.txt
+InsertBQDataQueryFileSmallCase=testdata/BigQuery/BigQueryInsertDataQuerySmallCase.txt
 
 #ORACLE Datatypes
 bigQueryColumns=(COL23 FLOAT(4), COL28 TIMESTAMP, COL29 TIMESTAMP(9), COL30 TIMESTAMP WITH TIME ZONE, \
diff --git a/oracle-plugin/src/e2e-test/resources/testdata/BigQuery/BigQueryCreateTableQuerySmallCase.txt b/oracle-plugin/src/e2e-test/resources/testdata/BigQuery/BigQueryCreateTableQuerySmallCase.txt
new file mode 100644
index 000000000..05bdb86b1
--- /dev/null
+++ b/oracle-plugin/src/e2e-test/resources/testdata/BigQuery/BigQueryCreateTableQuerySmallCase.txt
@@ -0,0 +1 @@
+create table `DATASET.TABLE_NAME` (id NUMERIC, lastname STRING)
\ No newline at end of file
diff --git a/oracle-plugin/src/e2e-test/resources/testdata/BigQuery/BigQueryInsertDataQuerySmallCase.txt b/oracle-plugin/src/e2e-test/resources/testdata/BigQuery/BigQueryInsertDataQuerySmallCase.txt
new file mode 100644
index 000000000..6140cc536
--- /dev/null
+++ b/oracle-plugin/src/e2e-test/resources/testdata/BigQuery/BigQueryInsertDataQuerySmallCase.txt
@@ -0,0 +1,6 @@
+insert into `DATASET.TABLE_NAME` (id, lastname) values
+(1,'Shelby'),
+(2,'Simpson'),
+(3,'Williams'),
+(4,'Sherry'),
+(5,'James');
\ No newline at end of file
diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSinkDBRecord.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSinkDBRecord.java
index 7bbd25f22..01b9a8247 100644
--- a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSinkDBRecord.java
+++ b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSinkDBRecord.java
@@ -17,9 +17,12 @@
 package io.cdap.plugin.oracle;
 
 import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.plugin.db.ColumnType;
 import io.cdap.plugin.db.SchemaReader;
 
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
 import java.util.List;
 
 /**
@@ -37,4 +40,14 @@ public OracleSinkDBRecord(StructuredRecord record, List<ColumnType> columnTypes)
   protected SchemaReader getSchemaReader() {
     return new OracleSinkSchemaReader();
   }
+
+  @Override
+  protected void insertOperation(PreparedStatement stmt) throws SQLException {
+    for (int fieldIndex = 0; fieldIndex < columnTypes.size(); fieldIndex++) {
+      ColumnType columnType = columnTypes.get(fieldIndex);
+      // Get the field from the schema using the column name with ignoring case.
+      Schema.Field field = record.getSchema().getField(columnType.getName(), true);
+      writeToDB(stmt, field, fieldIndex);
+    }
+  }
 }