fix(sink): handle visibility in remote sink (#12463)

src/common/src/array/arrow.rs

@@ -26,12 +26,14 @@ use crate::types::{Int256, StructType};
 use crate::util::iter_util::ZipEqDebug;
 
 // Implement bi-directional `From` between `DataChunk` and `arrow_array::RecordBatch`.
-
-// note: DataChunk -> arrow RecordBatch will IGNORE the visibilities.
 impl TryFrom<&DataChunk> for arrow_array::RecordBatch {
     type Error = ArrayError;
 
     fn try_from(chunk: &DataChunk) -> Result<Self, Self::Error> {
+        if !chunk.is_compacted() {
+            let c = chunk.clone();
+            return Self::try_from(&c.compact());
+        }
         let columns: Vec<_> = chunk
             .columns()
             .iter()

src/common/src/array/stream_chunk.rs

@@ -189,6 +189,9 @@ impl StreamChunk {
     }
 
     pub fn to_protobuf(&self) -> PbStreamChunk {
+        if !self.is_compacted() {
+            return self.clone().compact().to_protobuf();
+        }
         PbStreamChunk {
             cardinality: self.cardinality() as u32,
             ops: self.ops.iter().map(|op| op.to_protobuf() as i32).collect(),

src/connector/src/sink/clickhouse.rs

@@ -301,6 +301,10 @@ impl ClickHouseSinkWriter {
         )?;
         for (op, row) in chunk.rows() {
             if op != Op::Insert {
+                tracing::warn!(
+                    "append only click house sink receive an {:?} which will be ignored.",
+                    op
+                );
                 continue;
             }
             let mut clickhouse_filed_vec = vec![];

src/connector/src/sink/remote.rs

@@ -350,7 +350,7 @@ where
     async fn write_batch(&mut self, chunk: StreamChunk) -> Result<()> {
         let payload = match self.payload_format {
             SinkPayloadFormat::Json => {
-                let mut row_ops = vec![];
+                let mut row_ops = Vec::with_capacity(chunk.cardinality());
                 let enc = JsonEncoder::new(&self.schema, None, TimestampHandlingMode::String);
                 for (op, row_ref) in chunk.rows() {
                     let map = enc.encode(row_ref)?;

src/connector/src/sink/utils.rs

@@ -21,7 +21,7 @@ use crate::sink::Result;
 
 pub fn chunk_to_json(chunk: StreamChunk, schema: &Schema) -> Result<Vec<String>> {
     let encoder = JsonEncoder::new(schema, None, TimestampHandlingMode::Milli);
-    let mut records: Vec<String> = Vec::with_capacity(chunk.capacity());
+    let mut records: Vec<String> = Vec::with_capacity(chunk.cardinality());
     for (_, row) in chunk.rows() {
         let record = Value::Object(encoder.encode(row)?);
         records.push(record.to_string());