Add client sending dictionary updates to test and remove unnecessary started_ boolean flag

Author: raulcd

cpp/src/arrow/flight/transport_server.cc

@@ -206,17 +206,16 @@ class TransportMessageWriter final : public FlightMessageWriter {
 
   Status Begin(const std::shared_ptr<Schema>& schema,
                const ipc::IpcWriteOptions& options) override {
-    ipc_options_ = options;
     if (batch_writer_) {
       return Status::Invalid("This writer has already been started.");
     }
+    ipc_options_ = options;
     std::unique_ptr<ipc::internal::IpcPayloadWriter> payload_writer(
         new TransportMessagePayloadWriter(stream_, &app_metadata_));
 
     ARROW_ASSIGN_OR_RAISE(batch_writer_,
                           ipc::internal::OpenRecordBatchWriter(std::move(payload_writer),
                                                                schema, ipc_options_));
-    started_ = true;
     return Status::OK();
   }
 
@@ -264,7 +263,7 @@ class TransportMessageWriter final : public FlightMessageWriter {
 
  private:
   Status CheckStarted() {
-    if (!started_) {
+    if (!batch_writer_) {
       return Status::Invalid("This writer is not started. Call Begin() with a schema");
     }
     return Status::OK();
@@ -274,7 +273,6 @@ class TransportMessageWriter final : public FlightMessageWriter {
   std::unique_ptr<ipc::RecordBatchWriter> batch_writer_;
   std::shared_ptr<Buffer> app_metadata_;
   ::arrow::ipc::IpcWriteOptions ipc_options_;
-  bool started_ = false;
   int64_t extra_messages_ = 0;
 };
 

python/pyarrow/tests/test_flight.py

@@ -2542,13 +2542,17 @@ def test_flight_dictionary_deltas_do_exchange():
     class DeltaFlightServer(ConstantFlightServer):
         def do_exchange(self, context, descriptor, reader, writer):
             if descriptor.command == b'dict_deltas':
-                table = simple_dicts_table()
+                expected_table = simple_dicts_table()
+                received_table = reader.read_all()
+                assert received_table.equals(expected_table)
+
                 options = pa.ipc.IpcWriteOptions(emit_dictionary_deltas=True)
-                writer.begin(table.schema, options=options)
-                writer.write_table(table)
+                writer.begin(expected_table.schema, options=options)
+                writer.write_table(expected_table)
 
     with DeltaFlightServer() as server, \
             FlightClient(('localhost', server.port)) as client:
+        expected_table = simple_dicts_table()
 
         descriptor = flight.FlightDescriptor.for_command(b"dict_deltas")
         writer, reader = client.do_exchange(descriptor,
@@ -2557,7 +2561,12 @@ def do_exchange(self, context, descriptor, reader, writer):
                                                     emit_dictionary_deltas=True)
                                             )
                                             )
+        # Send client table with dictionary updates (deltas should be sent)
+        with writer:
+            writer.begin(expected_table.schema, options=pa.ipc.IpcWriteOptions(
+                emit_dictionary_deltas=True))
+            writer.write_table(expected_table)
+            writer.done_writing()
+            received_table = reader.read_all()
 
-        received_table = reader.read_all()
-        expected_table = simple_dicts_table()
         assert received_table.equals(expected_table)