[FEA][Java] ArrowIPCTableWriter writes en empty batch in the case of an empty table.

[firestarman]

Is your feature request related to a problem? Please describe.
Currently the ArrowIPCTableWriter will write no batches into the steam if giving an empty table, which PySpark cannot handle correctly, complaining of the error as below.

E                     File "/usr/local/spark/python/lib/pyspark.zip/pyspark/sql/pandas/serializers.py", line 297, in load_stream
E                       [self.arrow_to_pandas(c) for c in pa.Table.from_batches(batch2).itercolumns()]
E                     File "pyarrow/table.pxi", line 1609, in pyarrow.lib.Table.from_batches
E                   ValueError: Must pass schema, or at least one RecordBatch

Pyspark is calling the pyarrow.Table.from_batches without specifying a schema, then it expects at least one batch (even an empty one) to be received to infer the batch schema.

I have made an unit test to reproduce this case.

  @Test
  void testArrowIPCWriteEmptyToBufferChunked() {
    try (Table emptyTable = new Table.TestBuilder().timestampDayColumn().build();
         MyBufferConsumer consumer = new MyBufferConsumer()) {
      ArrowIPCWriterOptions options = ArrowIPCWriterOptions.builder()
              .withColumnNames("day")
              .build();
      try (TableWriter writer = Table.writeArrowIPCChunked(options, consumer)) {
        writer.write(emptyTable);
      }
      try (StreamedTableReader reader = Table.readArrowIPCChunked(new MyBufferProvider(consumer))) {
        boolean done = false;
        int count = 0;
        while (!done) {
          try (Table t = reader.getNextIfAvailable()) {
            if (t == null) {
              done = true;
            } else {
              assertTablesAreEqual(emptyTable, t);
              count++;
            }
          }
        }
        // Expect one empty batch for the empty table.
        assertEquals(1, count);
      }
    }
  }

Describe the solution you'd like
ArrowIPCTableWriter writes en empty batch explicitly in the case of an empty table. We can do this in the JNI layer easily, e.g.

diff --git a/java/src/main/native/src/TableJni.cpp b/java/src/main/native/src/TableJni.cpp
index ad280cad5f..c23c5a3ccb 100644
--- a/java/src/main/native/src/TableJni.cpp
+++ b/java/src/main/native/src/TableJni.cpp
@@ -258,7 +258,15 @@ public:
       writer = *tmp_writer;
       initialized = true;
     }
-    writer->WriteTable(*arrow_tab, max_chunk);
+    if (arrow_tab->num_rows() == 0) {
+      // Arrow C++ IPC writer will not write an empty batch in the case of an
+      // empty table, so need to write an empty batch explicitly.
+      // For more please see https://issues.apache.org/jira/browse/ARROW-17912.
+      auto empty_batch = arrow::RecordBatch::MakeEmpty(arrow_tab->schema());
+      writer->WriteRecordBatch(*(*empty_batch));
+    } else {
+      writer->WriteTable(*arrow_tab, max_chunk);
+    }
   }

Describe alternatives you've considered
Let the Arrow C++ IPC writer used by the cuDF JNI support writing en empty batch for an empty table, I tried but failed.
For more details, please refer to https://issues.apache.org/jira/browse/ARROW-17912. BTW, Arrow Java IPC writer can do this, I mean, sending out an empty batch implicitly.

We can also update the Pyspark to specify the schema, but not sure how long it will take to have the change done.

So in a short term, we can do it in the cuDF JNI.