fivetran-destination: Flush after each record is written (#26326)

This PR changes the Fivetran Destination to flush the CSV writer after
each record is written. Previously we wouldn't flush the writer until it
was dropped and for sufficiently large CSV files this would cause a
hang.

I was able to reproduce the hang by hooking up the Fivetran Destination
to testdrive and using the `file-append` command to generate a CSV file
with 1,000,000 records. On `main` this will hang but with this PR we
successfully copy the file.

### Motivation

Fix a hang in the Fivetran Destination

### Checklist

- [ ] This PR has adequate test coverage / QA involvement has been duly
considered.
- [ ] This PR has an associated up-to-date [design
doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md),
is a design doc
([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)),
or is sufficiently small to not require a design.
  <!-- Reference the design in the description. -->
- [ ] If this PR evolves [an existing `$T ⇔ Proto$T`
mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md)
(possibly in a backwards-incompatible way), then it is tagged with a
`T-proto` label.
- [ ] If this PR will require changes to cloud orchestration or tests,
there is a companion cloud PR to account for those changes that is
tagged with the release-blocker label
([example](https://github.com/MaterializeInc/cloud/pull/5021)).
<!-- Ask in #team-cloud on Slack if you need help preparing the cloud
PR. -->
- [x] This PR includes the following [user-facing behavior
changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note):
  - N/a

Author: ParkMyCar

src/fivetran-destination/src/destination/dml.rs

@@ -653,6 +653,8 @@ async fn copy_files(
 
         // Stream the files into the COPY FROM sink.
         for path in files {
+            tracing::info!(?path, "starting copy");
+
             // Open the CSV file, returning an AsyncReader.
             let file = load_file(file_config, path)
                 .await
@@ -666,8 +668,14 @@ async fn copy_files(
             let mut record_stream = adapter.into_stream();
             while let Some(maybe_record) = record_stream.next().await {
                 let record = maybe_record?;
-                csv_sink.write_byte_record(&record).await?;
+                csv_sink
+                    .write_byte_record(&record)
+                    .await
+                    .context("writing record")?;
+                csv_sink.flush().await.context("flushing record")?;
             }
+
+            tracing::info!(?path, "finished copy");
         }
     }