feat: support to copy table from/to CSV and JSON format file (#1475)

* refactor: refactor copy from executor

* feat: support to copy from CSV and JSON format files

* feat: support to copy table to the CSV and JSON format file

* test: add tests copy from/to

* chore: apply suggestions from CR

src/common/recordbatch/src/adapter.rs

@@ -19,6 +19,7 @@ use std::task::{Context, Poll};
 
 use datafusion::arrow::datatypes::SchemaRef as DfSchemaRef;
 use datafusion::error::Result as DfResult;
+use datafusion::parquet::arrow::async_reader::{AsyncFileReader, ParquetRecordBatchStream};
 use datafusion::physical_plan::RecordBatchStream as DfRecordBatchStream;
 use datafusion_common::DataFusionError;
 use datatypes::schema::{Schema, SchemaRef};
@@ -39,6 +40,40 @@ type FutureStream = Pin<
     >,
 >;
 
+/// ParquetRecordBatchStream -> DataFusion RecordBatchStream
+pub struct ParquetRecordBatchStreamAdapter<T> {
+    stream: ParquetRecordBatchStream<T>,
+}
+
+impl<T: Unpin + AsyncFileReader + Send + 'static> ParquetRecordBatchStreamAdapter<T> {
+    pub fn new(stream: ParquetRecordBatchStream<T>) -> Self {
+        Self { stream }
+    }
+}
+
+impl<T: Unpin + AsyncFileReader + Send + 'static> DfRecordBatchStream
+    for ParquetRecordBatchStreamAdapter<T>
+{
+    fn schema(&self) -> DfSchemaRef {
+        self.stream.schema().clone()
+    }
+}
+
+impl<T: Unpin + AsyncFileReader + Send + 'static> Stream for ParquetRecordBatchStreamAdapter<T> {
+    type Item = DfResult<DfRecordBatch>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let batch = futures::ready!(Pin::new(&mut self.stream).poll_next(cx))
+            .map(|r| r.map_err(|e| DataFusionError::External(Box::new(e))));
+        Poll::Ready(batch)
+    }
+
+    #[inline]
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
+}
+
 /// Greptime SendableRecordBatchStream -> DataFusion RecordBatchStream
 pub struct DfRecordBatchStreamAdapter {
     stream: SendableRecordBatchStream,