Fix join bug on csv datasources

integration_tests/src/main/python/csv_test.py

@@ -719,4 +719,4 @@ def test_csv_read_gbk_encoded_data(std_input_path):
             .option("header", "true")
             .schema("name string, age int, city string, job string")
             .csv(std_input_path + "/test_gbk.csv"),
-        conf={"spark.sql.legacy.javaCharsets": legacy_charset})
+        conf={"spark.sql.legacy.javaCharsets": legacy_charset})

integration_tests/src/main/python/join_test.py

@@ -485,6 +485,28 @@ def do_join(spark):
         return t0.crossJoin(t1).limit(21)
     assert_gpu_and_cpu_are_equal_collect(do_join)
 
+@allow_non_gpu('CollectLimitExec')
+def test_empty_right_outer_side_with_limit(std_input_path):
+    built_csv_path = std_input_path + '/t1.csv'
+    stream_csv_path = std_input_path + '/t0.csv'
+
+    def create_views(spark):
+        spark.read.csv(built_csv_path, header=True, inferSchema=True).createOrReplaceTempView("built_table")
+        spark.read.csv(stream_csv_path, header=True, inferSchema=True).createOrReplaceTempView("stream_table")
+
+    # create views first on CPU
+    with_cpu_session(lambda spark: create_views(spark))
+
+    # limit to 10 rows to produce `LocalLimitExec` node
+    def do_join(spark):
+        return spark.sql("""
+            SELECT '1', CAST(CAST(stream_table.c0 AS int) as string)
+            FROM built_table
+            RIGHT OUTER JOIN stream_table
+            ON TRUE limit 10
+        """)
+    assert_gpu_and_cpu_are_equal_collect(do_join)
+
 # local sort because of https://github.com/NVIDIA/spark-rapids/issues/84
 # After 3.1.0 is the min spark version we can drop this
 @ignore_order(local=True)

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuTextBasedPartitionReader.scala

@@ -643,7 +643,14 @@ abstract class GpuTextBasedPartitionReader[BUFF <: LineBufferer, FACT <: LineBuf
     //  val cols = (0 until  table.getNumberOfColumns).map(i => table.getColumn(i))
     //  Some(new Table(cols: _*))
     // }
-    Some(table)
+    if (table.getRowCount == 0) {
+      // CSV reader can return empty table, close it and return None
+      // E.g.: CSV file with only header and no data rows, empty table will be returned
+      table.close()
+      None
+    } else {
+      Some(table)
+    }
   }
 
   override def next(): Boolean = {