Address review feedback: rename to FlushToSink and consolidate Python tests

- Rename WriteAndClearBuffer to FlushToSink (shorter, clearer intent)
- Consolidate Python tests into a single parameterized test with 3 cases:
  empty batch at beginning, middle, and end

Author: rynewang

cpp/src/arrow/csv/writer.cc

@@ -541,7 +541,7 @@ class CSVWriterImpl : public ipc::RecordBatchWriter {
     for (auto maybe_slice : iterator) {
       ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> slice, maybe_slice);
       RETURN_NOT_OK(TranslateMinimalBatch(*slice));
-      RETURN_NOT_OK(WriteAndClearBuffer());
+      RETURN_NOT_OK(FlushToSink());
       stats_.num_record_batches++;
     }
     return Status::OK();
@@ -554,7 +554,7 @@ class CSVWriterImpl : public ipc::RecordBatchWriter {
     RETURN_NOT_OK(reader.ReadNext(&batch));
     while (batch != nullptr) {
       RETURN_NOT_OK(TranslateMinimalBatch(*batch));
-      RETURN_NOT_OK(WriteAndClearBuffer());
+      RETURN_NOT_OK(FlushToSink());
       RETURN_NOT_OK(reader.ReadNext(&batch));
       stats_.num_record_batches++;
     }
@@ -590,9 +590,9 @@ class CSVWriterImpl : public ipc::RecordBatchWriter {
     return Status::OK();
   }
 
-  // GH-36889: Write buffer to sink and clear it to avoid stale content
+  // GH-36889: Flush buffer to sink and clear it to avoid stale content
   // being written again if the next batch is empty.
-  Status WriteAndClearBuffer() {
+  Status FlushToSink() {
     RETURN_NOT_OK(sink_->Write(data_buffer_));
     return data_buffer_->Resize(0, /*shrink_to_fit=*/false);
   }
@@ -661,7 +661,7 @@ class CSVWriterImpl : public ipc::RecordBatchWriter {
     next += options_.eol.size();
     DCHECK_EQ(reinterpret_cast<uint8_t*>(next),
               data_buffer_->data() + data_buffer_->size());
-    return WriteAndClearBuffer();
+    return FlushToSink();
   }
 
   Status TranslateMinimalBatch(const RecordBatch& batch) {

python/pyarrow/tests/test_csv.py

@@ -2067,34 +2067,35 @@ def readinto(self, *args):
             read_csv(MyBytesIO(data))
 
 
-def test_write_csv_empty_batch_no_duplicate_header():
-    # GH-36889: Empty batches at the start should not cause duplicate headers
-    table = pa.table({"col1": ["a", "b", "c"]})
-
-    # Concatenate empty table with data table
-    empty_table = table.schema.empty_table()
-    combined = pa.concat_tables([empty_table, table])
-
-    buf = io.BytesIO()
-    write_csv(combined, buf)
-    buf.seek(0)
-    result = buf.read()
-
-    # Should have exactly one header, not two
-    assert result == b'"col1"\n"a"\n"b"\n"c"\n'
-
-
-def test_write_csv_empty_batch_in_middle():
-    # GH-36889: Empty batches in the middle should not cause issues
-    table1 = pa.table({"col1": ["a"]})
-    table2 = pa.table({"col1": ["b"]})
-    empty_table = table1.schema.empty_table()
-
-    combined = pa.concat_tables([table1, empty_table, table2])
+@pytest.mark.parametrize("tables,expected", [
+    # GH-36889: Empty batch at the beginning
+    (
+        lambda: [pa.table({"col1": []}).cast(pa.schema([("col1", pa.string())])),
+                 pa.table({"col1": ["a"]}),
+                 pa.table({"col1": ["b"]})],
+        b'"col1"\n"a"\n"b"\n'
+    ),
+    # GH-36889: Empty batch in the middle
+    (
+        lambda: [pa.table({"col1": ["a"]}),
+                 pa.table({"col1": []}).cast(pa.schema([("col1", pa.string())])),
+                 pa.table({"col1": ["b"]})],
+        b'"col1"\n"a"\n"b"\n'
+    ),
+    # GH-36889: Empty batch at the end
+    (
+        lambda: [pa.table({"col1": ["a"]}),
+                 pa.table({"col1": ["b"]}),
+                 pa.table({"col1": []}).cast(pa.schema([("col1", pa.string())]))],
+        b'"col1"\n"a"\n"b"\n'
+    ),
+])
+def test_write_csv_empty_batch_should_not_pollute_output(tables, expected):
+    combined = pa.concat_tables(tables())
 
     buf = io.BytesIO()
     write_csv(combined, buf)
     buf.seek(0)
     result = buf.read()
 
-    assert result == b'"col1"\n"a"\n"b"\n'
+    assert result == expected