diff --git a/core-plugins/src/e2e-test/features/joiner/JoinerWithFile.feature b/core-plugins/src/e2e-test/features/joiner/JoinerWithFile.feature
index dbb1c701b..bf79ba530 100644
--- a/core-plugins/src/e2e-test/features/joiner/JoinerWithFile.feature
+++ b/core-plugins/src/e2e-test/features/joiner/JoinerWithFile.feature
@@ -250,3 +250,264 @@ Feature: Joiner - Verify File source to File sink data transfer using Joiner ana
     Then Close the pipeline logs
     Then Validate OUT record count of joiner is equal to IN record count of sink
     Then Validate output file generated by file sink plugin "fileSinkTargetBucket" is equal to expected output file "joinerTest4OutputFile"
+
+  @JOINER_TEST1 @JOINER_TEST2 @FILE_SINK_TEST
+  Scenario:To verify data is getting transferred from File to File successfully using Joiner plugin with Advance inner join type
+    Given Open Datafusion Project to configure pipeline
+    When Select plugin: "File" from the plugins list as: "Source"
+    When Select plugin: "File" from the plugins list as: "Source"
+    And Expand Plugin group in the LHS plugins list: "Analytics"
+    When Select plugin: "Joiner" from the plugins list as: "Analytics"
+    Then Connect plugins: "File" and "Joiner" to establish connection
+    Then Connect plugins: "File2" and "Joiner" to establish connection
+    When Expand Plugin group in the LHS plugins list: "Sink"
+    When Select plugin: "File" from the plugins list as: "Sink"
+    Then Connect plugins: "Joiner" and "File3" to establish connection
+    Then Click plugin property: "alignPlugins" button
+    Then Navigate to the properties page of plugin: "File"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "joinerInputTest1"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Click plugin property: "skipHeader"
+    Then Click plugin property: "enableQuotedValues"
+    Then Click on the Get Schema button
+    Then Verify the Output Schema matches the Expected Schema: "joinerCsvFileFirstSchema"
+    Then Validate "File2" plugin properties
+    Then Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "File2"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "joinerInputTest2"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Click plugin property: "skipHeader"
+    Then Click plugin property: "enableQuotedValues"
+    Then Click on the Get Schema button
+    Then Verify the Output Schema matches the Expected Schema: "joinerCsvDataTypeFileSchema"
+    Then Validate "File" plugin properties
+    Then Close the Plugin Properties page
+    When Navigate to the properties page of plugin: "Joiner"
+    Then Select joiner type "Inner"
+    Then Select radio button plugin property: "conditionType" with value: "advanced"
+    Then Enter textarea plugin property: "conditionExpression" with value: "joinConditionSQLExpression"
+    Then Click on the Get Schema button
+    Then Validate "Joiner" plugin properties
+    Then Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "File3"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "fileSinkTargetBucket"
+    Then Replace input plugin property: "pathSuffix" with value: "yyyy-MM-dd-HH-mm-ss"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Validate "File" plugin properties
+    Then Close the Plugin Properties page
+    Then Save the pipeline
+    Then Preview and run the pipeline
+    Then Wait till pipeline preview is in running state
+    Then Open and capture pipeline preview logs
+    Then Verify the preview run status of pipeline in the logs is "succeeded"
+    Then Close the pipeline logs
+    Then Close the preview
+    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 Close the pipeline logs
+    Then Validate OUT record count of joiner is equal to IN record count of sink
+    Then Validate output file generated by file sink plugin "fileSinkTargetBucket" is equal to expected output file "joinerTest1OutputFile"
+
+  @JOINER_TEST1 @JOINER_TEST2 @FILE_SINK_TEST
+  Scenario:To verify data is getting transferred from File to File successfully using Joiner plugin with Advance outer join type
+    Given Open Datafusion Project to configure pipeline
+    When Select plugin: "File" from the plugins list as: "Source"
+    When Select plugin: "File" from the plugins list as: "Source"
+    And Expand Plugin group in the LHS plugins list: "Analytics"
+    When Select plugin: "Joiner" from the plugins list as: "Analytics"
+    Then Connect plugins: "File" and "Joiner" to establish connection
+    Then Connect plugins: "File2" and "Joiner" to establish connection
+    When Expand Plugin group in the LHS plugins list: "Sink"
+    When Select plugin: "File" from the plugins list as: "Sink"
+    Then Connect plugins: "Joiner" and "File3" to establish connection
+    Then Click plugin property: "alignPlugins" button
+    Then Navigate to the properties page of plugin: "File"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "joinerInputTest1"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Click plugin property: "skipHeader"
+    Then Click plugin property: "enableQuotedValues"
+    Then Click on the Get Schema button
+    Then Verify the Output Schema matches the Expected Schema: "joinerCsvFileFirstSchema"
+    Then Validate "File2" plugin properties
+    Then Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "File2"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "joinerInputTest2"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Click plugin property: "skipHeader"
+    Then Click plugin property: "enableQuotedValues"
+    Then Click on the Get Schema button
+    Then Verify the Output Schema matches the Expected Schema: "joinerCsvDataTypeFileSchema"
+    Then Validate "File" plugin properties
+    Then Close the Plugin Properties page
+    When Navigate to the properties page of plugin: "Joiner"
+    Then Select joiner type "Outer"
+    Then Select radio button plugin property: "conditionType" with value: "advanced"
+    Then Enter textarea plugin property: "conditionExpression" with value: "joinConditionSQLExpression"
+    Then Select dropdown plugin property: "inMemoryInputs" with option value: "File2"
+    Then Press Escape Key
+    Then Click on the Get Schema button
+    Then Validate "Joiner" plugin properties
+    Then Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "File3"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "fileSinkTargetBucket"
+    Then Replace input plugin property: "pathSuffix" with value: "yyyy-MM-dd-HH-mm-ss"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Validate "File" plugin properties
+    Then Close the Plugin Properties page
+    Then Save the pipeline
+    Then Preview and run the pipeline
+    Then Wait till pipeline preview is in running state
+    Then Open and capture pipeline preview logs
+    Then Verify the preview run status of pipeline in the logs is "succeeded"
+    Then Close the pipeline logs
+    Then Close the preview
+    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 Close the pipeline logs
+    Then Validate OUT record count of joiner is equal to IN record count of sink
+    Then Validate output file generated by file sink plugin "fileSinkTargetBucket" is equal to expected output file "joinerTest1OutputFile"
+
+  @JOINER_TEST1 @JOINER_TEST2 @FILE_SINK_TEST
+  Scenario:To verify data is getting transferred from File to File successfully using Joiner plugin with outer join type with selected inputs and required input
+    Given Open Datafusion Project to configure pipeline
+    When Select plugin: "File" from the plugins list as: "Source"
+    When Select plugin: "File" from the plugins list as: "Source"
+    And Expand Plugin group in the LHS plugins list: "Analytics"
+    When Select plugin: "Joiner" from the plugins list as: "Analytics"
+    Then Connect plugins: "File" and "Joiner" to establish connection
+    Then Connect plugins: "File2" and "Joiner" to establish connection
+    When Expand Plugin group in the LHS plugins list: "Sink"
+    When Select plugin: "File" from the plugins list as: "Sink"
+    Then Connect plugins: "Joiner" and "File3" to establish connection
+    Then Click plugin property: "alignPlugins" button
+    Then Navigate to the properties page of plugin: "File"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "joinerInputTest1"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Click plugin property: "skipHeader"
+    Then Click plugin property: "enableQuotedValues"
+    Then Click on the Get Schema button
+    Then Verify the Output Schema matches the Expected Schema: "joinerCsvFileFirstSchema"
+    Then Validate "File2" plugin properties
+    Then Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "File2"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "joinerInputTest2"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Click plugin property: "skipHeader"
+    Then Click plugin property: "enableQuotedValues"
+    Then Click on the Get Schema button
+    Then Verify the Output Schema matches the Expected Schema: "joinerCsvDataTypeFileSchema"
+    Then Validate "File" plugin properties
+    Then Close the Plugin Properties page
+    When Navigate to the properties page of plugin: "Joiner"
+    Then Select joiner type "Outer"
+    Then Select radio button plugin property: "conditionType" with value: "basic"
+    Then Expand fields
+    Then Uncheck plugin "File" field "lastname" alias checkbox
+    Then Uncheck plugin "File2" field "item" alias checkbox
+    Then Click on the required input checkbox for schema 1 named "File"
+    Then Click on the Get Schema button
+    Then Validate "Joiner" plugin properties
+    Then Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "File3"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "fileSinkTargetBucket"
+    Then Replace input plugin property: "pathSuffix" with value: "yyyy-MM-dd-HH-mm-ss"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Validate "File" plugin properties
+    Then Close the Plugin Properties page
+    Then Save the pipeline
+    Then Preview and run the pipeline
+    Then Wait till pipeline preview is in running state
+    Then Open and capture pipeline preview logs
+    Then Verify the preview run status of pipeline in the logs is "succeeded"
+    Then Close the pipeline logs
+    Then Close the preview
+    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 Close the pipeline logs
+    Then Validate OUT record count of joiner is equal to IN record count of sink
+    Then Validate output file generated by file sink plugin "fileSinkTargetBucket" is equal to expected output file "joinerTest7OutputFile"
+
+  @JOINER_TEST1 @JOINER_TEST2 @FILE_SINK_TEST
+  Scenario:To verify data is getting transferred from File to File successfully using Joiner plugin with inner join type with selected inputs
+    Given Open Datafusion Project to configure pipeline
+    When Select plugin: "File" from the plugins list as: "Source"
+    When Select plugin: "File" from the plugins list as: "Source"
+    And Expand Plugin group in the LHS plugins list: "Analytics"
+    When Select plugin: "Joiner" from the plugins list as: "Analytics"
+    Then Connect plugins: "File" and "Joiner" to establish connection
+    Then Connect plugins: "File2" and "Joiner" to establish connection
+    When Expand Plugin group in the LHS plugins list: "Sink"
+    When Select plugin: "File" from the plugins list as: "Sink"
+    Then Connect plugins: "Joiner" and "File3" to establish connection
+    Then Click plugin property: "alignPlugins" button
+    Then Navigate to the properties page of plugin: "File"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "joinerInputTest1"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Click plugin property: "skipHeader"
+    Then Click plugin property: "enableQuotedValues"
+    Then Click on the Get Schema button
+    Then Verify the Output Schema matches the Expected Schema: "joinerCsvFileFirstSchema"
+    Then Validate "File2" plugin properties
+    Then Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "File2"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "joinerInputTest2"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Click plugin property: "skipHeader"
+    Then Click plugin property: "enableQuotedValues"
+    Then Click on the Get Schema button
+    Then Verify the Output Schema matches the Expected Schema: "joinerCsvDataTypeFileSchema"
+    Then Validate "File" plugin properties
+    Then Close the Plugin Properties page
+    When Navigate to the properties page of plugin: "Joiner"
+    Then Select joiner type "Inner"
+    Then Select radio button plugin property: "conditionType" with value: "basic"
+    Then Expand fields
+    Then Uncheck plugin "File" field "lastname" alias checkbox
+    Then Uncheck plugin "File" field "state" alias checkbox
+    Then Uncheck plugin "File2" field "item" alias checkbox
+    Then Uncheck plugin "File2" field "price" alias checkbox
+    Then Click on the Get Schema button
+    Then Validate "Joiner" plugin properties
+    Then Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "File3"
+    Then Enter input plugin property: "referenceName" with value: "FileReferenceName"
+    Then Enter input plugin property: "path" with value: "fileSinkTargetBucket"
+    Then Replace input plugin property: "pathSuffix" with value: "yyyy-MM-dd-HH-mm-ss"
+    Then Select dropdown plugin property: "format" with option value: "csv"
+    Then Validate "File" plugin properties
+    Then Close the Plugin Properties page
+    Then Save the pipeline
+    Then Preview and run the pipeline
+    Then Wait till pipeline preview is in running state
+    Then Open and capture pipeline preview logs
+    Then Verify the preview run status of pipeline in the logs is "succeeded"
+    Then Close the pipeline logs
+    Then Close the preview
+    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 Close the pipeline logs
+    Then Validate OUT record count of joiner is equal to IN record count of sink
+    Then Validate output file generated by file sink plugin "fileSinkTargetBucket" is equal to expected output file "joinerTest8OutputFile"
diff --git a/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/actions/JoinerActions.java b/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/actions/JoinerActions.java
index cf7670f33..8e5773955 100644
--- a/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/actions/JoinerActions.java
+++ b/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/actions/JoinerActions.java
@@ -40,6 +40,10 @@ public static void uncheckPluginFieldAliasCheckBox(String plugin, String field)
     ElementHelper.selectCheckbox(JoinerLocators.fieldAliasCheckBox(plugin, field));
   }
 
+  public static void selectRequiredInputCheckbox(int value, String inputSchemaName) {
+    ElementHelper.selectCheckbox(JoinerLocators.requiredInputCheckbox(value - 1, inputSchemaName));
+  }
+
   public static void selectJoinerType(String targetJoinerType) {
     ElementHelper.selectDropdownOption(JoinerLocators.joinerTypeSelectDropdown,
                                        CdfPluginPropertiesLocators.locateDropdownListItem(targetJoinerType));
diff --git a/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/locators/JoinerLocators.java b/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/locators/JoinerLocators.java
index 6c2c29b93..9fc28a26f 100644
--- a/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/locators/JoinerLocators.java
+++ b/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/locators/JoinerLocators.java
@@ -28,7 +28,12 @@ public class JoinerLocators {
 
   public static WebElement fieldAliasCheckBox(String pluginName, String field) {
     String xpath = "//*[@data-cy='" + pluginName + "-stage-expansion-panel']" + "//*[@data-cy='" + field +
-      "-field-selector-name']/..//*[@type='checkbox']";
+      "-field-selector-name']/..//*[@data-cy='" + field + "-field-selector-checkbox']";
+    return SeleniumDriver.getDriver().findElement(By.xpath(xpath));
+  }
+
+  public static WebElement requiredInputCheckbox(int value, String inputSchemaName) {
+    String xpath = "//*[@type='checkbox'][@value='" + value + "-" + inputSchemaName + "']";
     return SeleniumDriver.getDriver().findElement(By.xpath(xpath));
   }
 
diff --git a/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/stepsdesign/Joiner.java b/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/stepsdesign/Joiner.java
index e0dc6366f..cb678a0bb 100644
--- a/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/stepsdesign/Joiner.java
+++ b/core-plugins/src/e2e-test/java/io/cdap/plugin/joiner/stepsdesign/Joiner.java
@@ -36,6 +36,11 @@ public void uncheckPluginFieldAliasCheckBox(String plugin, String field) {
     JoinerActions.uncheckPluginFieldAliasCheckBox(plugin, field);
   }
 
+  @Then("Click on the required input checkbox for schema {int} named {string}")
+  public void clickRequiredInput(int value, String inputSchemaName) {
+    JoinerActions.selectRequiredInputCheckbox(value, inputSchemaName);
+  }
+
   @Then("Enter numPartitions {string}")
   public void openJoinerProperties(String partitions) {
     JoinerActions.enterNumPartitions(PluginPropertyUtils.pluginProp(partitions));
diff --git a/core-plugins/src/e2e-test/resources/pluginParameters.properties b/core-plugins/src/e2e-test/resources/pluginParameters.properties
index f63c6cabb..12a2021b7 100644
--- a/core-plugins/src/e2e-test/resources/pluginParameters.properties
+++ b/core-plugins/src/e2e-test/resources/pluginParameters.properties
@@ -232,7 +232,7 @@ joinerInvalidPartitions=&*^*
 joinerKeys=File.purchase_id = File2.customer_id
 joinerInputMemory=File
 joinerNullKeys=false
-joinConditionSQLExpression=File.customer_name = customers.name
+joinConditionSQLExpression=File.id = File2.customerid
 joinerOutputSchema={ "type": "record", "name": "text", "fields": [ \
   { "name": "purchase_id", "type": "int" }, { "name": "customer_name", "type": "string" }, \
   { "name": "item", "type": "string" }, { "name": "customer_id", "type": "int" }, { "name": "name", "type": "string" } ] }
@@ -242,6 +242,10 @@ joinerTest2OutputFile=e2e-tests/expected_outputs/CSV_JOINER_TEST2_Output.csv
 joinerTest3OutputFile=e2e-tests/expected_outputs/CSV_JOINER_TEST3_Output.csv
 joinerTest4OutputFile=e2e-tests/expected_outputs/CSV_JOINER_TEST4_Output.csv
 joinerMacroOutputFile=e2e-tests/expected_outputs/CSV_JOINER_TEST5_Output.csv
+joinerTest6OutputFile=e2e-tests/expected_outputs/CSV_JOINER_TEST6_Output.csv
+joinerTest7OutputFile=e2e-tests/expected_outputs/CSV_JOINER_TEST7_Output.csv
+joinerTest8OutputFile=e2e-tests/expected_outputs/CSV_JOINER_TEST8_Output.csv
+
 ## JOINER-PLUGIN-PROPERTIES-END
 
 ## ERROR-COLLECTOR-PROPERTIES-START
diff --git a/core-plugins/src/e2e-test/resources/testdata/expected_outputs/CSV_JOINER_TEST6_Output.csv b/core-plugins/src/e2e-test/resources/testdata/expected_outputs/CSV_JOINER_TEST6_Output.csv
new file mode 100644
index 000000000..9e3eb5aaf
--- /dev/null
+++ b/core-plugins/src/e2e-test/resources/testdata/expected_outputs/CSV_JOINER_TEST6_Output.csv
@@ -0,0 +1,6 @@
+1,Douglas,1, Vista Montana,San Jose,95134,408-777-3214,1
+1,Douglas,1, Vista Montana,San Jose,95134,408-777-3214,1
+2,David,3, Baypointe Parkway,Houston,78970,804-777-2341,2
+2,David,3, Baypointe Parkway,Houston,78970,804-777-2341,2
+5,Frank,1609 Far St.,San Diego,29770,201-506-8756,5
+3,Hugh,5, Cool Way,Manhattan,67263,708-234-2168,3
diff --git a/core-plugins/src/e2e-test/resources/testdata/expected_outputs/CSV_JOINER_TEST7_Output.csv b/core-plugins/src/e2e-test/resources/testdata/expected_outputs/CSV_JOINER_TEST7_Output.csv
new file mode 100644
index 000000000..1a8434e77
--- /dev/null
+++ b/core-plugins/src/e2e-test/resources/testdata/expected_outputs/CSV_JOINER_TEST7_Output.csv
@@ -0,0 +1,8 @@
+1,Douglas,1, Vista Montana,San Jose,CA,95134,408-777-3214,1,0.8
+1,Douglas,1, Vista Montana,San Jose,CA,95134,408-777-3214,1,2.05
+6,Serena,123 Far St.,Las Vegas,Nv,45334,888-605-3479,,
+4,Walter,3828, Piermont Dr,Orlando,FL,73498,201-734-7315,,
+2,David,3, Baypointe Parkway,Houston,TX,78970,804-777-2341,2,1.5
+2,David,3, Baypointe Parkway,Houston,TX,78970,804-777-2341,2,0.5
+5,Frank,1609 Far St.,San Diego,CA,29770,201-506-8756,5,0.5
+3,Hugh,5, Cool Way,Manhattan,NY,67263,708-234-2168,3,1.99
diff --git a/core-plugins/src/e2e-test/resources/testdata/expected_outputs/CSV_JOINER_TEST8_Output.csv b/core-plugins/src/e2e-test/resources/testdata/expected_outputs/CSV_JOINER_TEST8_Output.csv
new file mode 100644
index 000000000..9e3eb5aaf
--- /dev/null
+++ b/core-plugins/src/e2e-test/resources/testdata/expected_outputs/CSV_JOINER_TEST8_Output.csv
@@ -0,0 +1,6 @@
+1,Douglas,1, Vista Montana,San Jose,95134,408-777-3214,1
+1,Douglas,1, Vista Montana,San Jose,95134,408-777-3214,1
+2,David,3, Baypointe Parkway,Houston,78970,804-777-2341,2
+2,David,3, Baypointe Parkway,Houston,78970,804-777-2341,2
+5,Frank,1609 Far St.,San Diego,29770,201-506-8756,5
+3,Hugh,5, Cool Way,Manhattan,67263,708-234-2168,3