Fix join bug on csv datasources

integration_tests/src/main/python/csv_test.py

@@ -19,6 +19,7 @@
 from datetime import datetime, timezone
 from data_gen import *
 from marks import *
+from pyspark.sql import functions as f
 from pyspark.sql.types import *
 from spark_session import *
 
@@ -720,3 +721,22 @@ def test_csv_read_gbk_encoded_data(std_input_path):
             .schema("name string, age int, city string, job string")
             .csv(std_input_path + "/test_gbk.csv"),
         conf={"spark.sql.legacy.javaCharsets": legacy_charset})
+
+
+@allow_non_gpu('CollectLimitExec')
+def test_csv_stream_table_is_empty_when_join(std_input_path):
+    built_csv_path = std_input_path + '/one_row.csv'
+    stream_csv_path = std_input_path + '/empty_with_header.csv'
+
+    def _create_view(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 view first on CPU
+    with_cpu_session(lambda spark: _create_view(spark))
+
+    # then do the join on GPU
+    with_gpu_session(lambda spark:
+                     spark.table("built_table").join(spark.table("stream_table"), f.lit(True), "right_outer").select(
+                         f.col("stream_table.c0")).show(),
+                     conf=_enable_all_types_conf)

integration_tests/src/test/resources/empty_with_header.csv

@@ -0,0 +1 @@
+c0

integration_tests/src/test/resources/one_row.csv

@@ -0,0 +1,2 @@
+c0
+1

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 = {