metric_engine/engine/
bulk_insert.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::HashSet;
16
17use api::v1::{ArrowIpc, SemanticType};
18use bytes::Bytes;
19use common_grpc::flight::{FlightEncoder, FlightMessage};
20use datatypes::arrow::record_batch::RecordBatch;
21use snafu::{OptionExt, ensure};
22use store_api::codec::PrimaryKeyEncoding;
23use store_api::metadata::RegionMetadataRef;
24use store_api::region_request::{AffectedRows, RegionBulkInsertsRequest, RegionRequest};
25use store_api::storage::RegionId;
26
27use crate::batch_modifier::{TagColumnInfo, modify_batch_sparse};
28use crate::engine::MetricEngineInner;
29use crate::error;
30use crate::error::Result;
31use crate::metrics::MITO_OPERATION_ELAPSED;
32
33impl MetricEngineInner {
34    /// Bulk-inserts rows into a metric region.
35    ///
36    /// **Logical region path:** The request payload is a logical `RecordBatch`
37    /// (timestamp, value and tag columns). It is transformed to physical format
38    /// via `modify_batch_sparse`, encoded to Arrow IPC, and forwarded as a
39    /// `BulkInserts` request to the data region.
40    ///
41    /// **Physical region path:** The request payload is already in physical format
42    /// (produced by the batcher's `flush_batch_physical`). It is forwarded directly
43    /// to the data region with no transformation.
44    ///
45    /// Returns the number of affected rows, or `0` if the input batch is empty.
46    pub async fn bulk_insert_region(
47        &self,
48        region_id: RegionId,
49        request: RegionBulkInsertsRequest,
50    ) -> Result<AffectedRows> {
51        if request.payload.num_rows() == 0 {
52            return Ok(0);
53        }
54        if self.is_physical_region(region_id) {
55            let _timer = MITO_OPERATION_ELAPSED
56                .with_label_values(&["bulk_insert_physical"])
57                .start_timer();
58            return self.bulk_insert_physical_region(region_id, request).await;
59        }
60
61        let _timer = MITO_OPERATION_ELAPSED
62            .with_label_values(&["bulk_insert_logical"])
63            .start_timer();
64        self.bulk_insert_logical_region(region_id, request).await
65    }
66
67    /// Passthrough for bulk inserts targeting a physical data region.
68    ///
69    /// The batch is already in physical format (with `__primary_key`, timestamp,
70    /// value columns), so no logical-to-physical transformation is needed.
71    async fn bulk_insert_physical_region(
72        &self,
73        region_id: RegionId,
74        mut request: RegionBulkInsertsRequest,
75    ) -> Result<AffectedRows> {
76        // Other logical tables can add fields to the shared physical schema.
77        // Let Mito fill fields absent from this batch before writing it.
78        request.aligned_schema_version = None;
79        self.data_region
80            .write_data(region_id, RegionRequest::BulkInserts(request))
81            .await
82    }
83
84    /// Bulk-inserts logical rows, transforming them to physical format first.
85    async fn bulk_insert_logical_region(
86        &self,
87        region_id: RegionId,
88        request: RegionBulkInsertsRequest,
89    ) -> Result<AffectedRows> {
90        let (physical_region_id, data_region_id, primary_key_encoding) =
91            self.find_data_region_meta(region_id)?;
92
93        if primary_key_encoding != PrimaryKeyEncoding::Sparse {
94            return error::UnsupportedRegionRequestSnafu {
95                request: RegionRequest::BulkInserts(request),
96            }
97            .fail();
98        }
99
100        let batch = request.payload;
101        if batch.num_rows() == 0 {
102            return Ok(0);
103        }
104
105        let logical_metadata = self
106            .logical_region_metadata(physical_region_id, region_id)
107            .await?;
108        let (tag_columns, non_tag_indices) = self.resolve_tag_columns_from_metadata(
109            region_id,
110            data_region_id,
111            &batch,
112            &logical_metadata,
113        )?;
114        let modified_batch = modify_batch_sparse(
115            batch.clone(),
116            region_id.table_id(),
117            &tag_columns,
118            &non_tag_indices,
119        )?;
120        let (schema, data_header, payload) = record_batch_to_ipc(&modified_batch)?;
121
122        let partition_expr_version = request.partition_expr_version;
123
124        let request = RegionBulkInsertsRequest {
125            skip_wal: request.skip_wal,
126            region_id: data_region_id,
127            payload: modified_batch,
128            raw_data: ArrowIpc {
129                schema,
130                data_header,
131                payload,
132            },
133            partition_expr_version,
134            aligned_schema_version: None,
135        };
136        self.data_region
137            .write_data(data_region_id, RegionRequest::BulkInserts(request))
138            .await
139    }
140
141    fn resolve_tag_columns_from_metadata(
142        &self,
143        logical_region_id: RegionId,
144        data_region_id: RegionId,
145        batch: &RecordBatch,
146        logical_metadata: &RegionMetadataRef,
147    ) -> Result<(Vec<TagColumnInfo>, Vec<usize>)> {
148        let tag_names: HashSet<&str> = logical_metadata
149            .column_metadatas
150            .iter()
151            .filter_map(|column| {
152                if column.semantic_type == SemanticType::Tag {
153                    Some(column.column_schema.name.as_str())
154                } else {
155                    None
156                }
157            })
158            .collect();
159
160        let mut tag_columns = Vec::new();
161        let mut non_tag_indices = Vec::new();
162        {
163            let state = self.state.read().unwrap();
164            let physical_columns = state
165                .physical_region_states()
166                .get(&data_region_id)
167                .context(error::PhysicalRegionNotFoundSnafu {
168                    region_id: data_region_id,
169                })?
170                .physical_columns();
171
172            for (index, field) in batch.schema().fields().iter().enumerate() {
173                let name = field.name();
174                let column_id = physical_columns
175                    .get(name)
176                    .map(|info| info.column_id)
177                    .with_context(|| error::ColumnNotFoundSnafu {
178                        name: name.clone(),
179                        region_id: logical_region_id,
180                    })?;
181                if tag_names.contains(name.as_str()) {
182                    tag_columns.push(TagColumnInfo {
183                        name: name.clone(),
184                        index,
185                        column_id,
186                    });
187                } else {
188                    non_tag_indices.push(index);
189                }
190            }
191        }
192
193        tag_columns.sort_by(|a, b| a.name.cmp(&b.name));
194        Ok((tag_columns, non_tag_indices))
195    }
196}
197
198fn record_batch_to_ipc(record_batch: &RecordBatch) -> Result<(Bytes, Bytes, Bytes)> {
199    let mut encoder = FlightEncoder::default();
200    let schema = encoder.encode_schema(record_batch.schema().as_ref());
201    let mut iter = encoder
202        .encode(FlightMessage::RecordBatch(record_batch.clone()))
203        .into_iter();
204
205    let Some(flight_data) = iter.next() else {
206        return error::UnexpectedRequestSnafu {
207            reason: "Failed to encode empty flight data",
208        }
209        .fail();
210    };
211    ensure!(
212        iter.next().is_none(),
213        error::UnexpectedRequestSnafu {
214            reason: "Bulk insert RecordBatch with dictionary arrays is unsupported".to_string(),
215        }
216    );
217
218    Ok((
219        schema.data_header,
220        flight_data.data_header,
221        flight_data.data_body,
222    ))
223}
224
225#[cfg(test)]
226mod tests {
227    use std::assert_matches;
228    use std::sync::Arc;
229
230    use api::v1::ArrowIpc;
231    use common_error::ext::ErrorExt;
232    use common_query::prelude::{greptime_timestamp, greptime_value};
233    use common_recordbatch::RecordBatches;
234    use datafusion::parquet::basic::Encoding;
235    use datatypes::arrow::array::{Float64Array, StringArray, TimestampMillisecondArray};
236    use datatypes::arrow::datatypes::{DataType, Field, Schema as ArrowSchema, TimeUnit};
237    use datatypes::arrow::record_batch::RecordBatch;
238    use mito2::config::MitoConfig;
239    use mito2::sst::parquet::metadata::MetadataLoader;
240    use mito2::sst::parquet::reader::MetadataCacheMetrics;
241    use store_api::metric_engine_consts::{
242        METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY, PRIMARY_KEY_ENCODING,
243    };
244    use store_api::mito_engine_options::EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING;
245    use store_api::path_utils::table_dir;
246    use store_api::region_engine::RegionEngine;
247    use store_api::region_request::{
248        PathType, RegionBulkInsertsRequest, RegionCloseRequest, RegionFlushRequest,
249        RegionOpenRequest, RegionPutRequest, RegionRequest,
250    };
251    use store_api::storage::{RegionId, ScanRequest};
252
253    use super::record_batch_to_ipc;
254    use crate::batch_modifier::{TagColumnInfo, modify_batch_sparse};
255    use crate::engine::MetricEngine;
256    use crate::error::Error;
257    use crate::test_util::{self, TestEnv};
258
259    fn build_logical_batch(start: usize, rows: usize) -> RecordBatch {
260        let schema = Arc::new(ArrowSchema::new(vec![
261            Field::new(
262                greptime_timestamp(),
263                DataType::Timestamp(TimeUnit::Millisecond, None),
264                false,
265            ),
266            Field::new(greptime_value(), DataType::Float64, true),
267            Field::new("job", DataType::Utf8, true),
268        ]));
269
270        let mut ts = Vec::with_capacity(rows);
271        let mut values = Vec::with_capacity(rows);
272        let mut tags = Vec::with_capacity(rows);
273        for i in start..start + rows {
274            ts.push(i as i64);
275            values.push(i as f64);
276            tags.push("tag_0".to_string());
277        }
278
279        RecordBatch::try_new(
280            schema,
281            vec![
282                Arc::new(TimestampMillisecondArray::from(ts)),
283                Arc::new(Float64Array::from(values)),
284                Arc::new(StringArray::from(tags)),
285            ],
286        )
287        .unwrap()
288    }
289
290    fn build_bulk_request(
291        logical_region_id: RegionId,
292        batch: RecordBatch,
293        skip_wal: bool,
294    ) -> RegionRequest {
295        let (schema, data_header, payload) = record_batch_to_ipc(&batch).unwrap();
296        RegionRequest::BulkInserts(RegionBulkInsertsRequest {
297            skip_wal,
298            region_id: logical_region_id,
299            payload: batch,
300            raw_data: ArrowIpc {
301                schema,
302                data_header,
303                payload,
304            },
305            partition_expr_version: None,
306            aligned_schema_version: None,
307        })
308    }
309
310    async fn init_dense_metric_region(env: &TestEnv) -> RegionId {
311        let physical_region_id = env.default_physical_region_id();
312        env.create_physical_region(
313            physical_region_id,
314            &TestEnv::default_table_dir(),
315            vec![(PRIMARY_KEY_ENCODING.to_string(), "dense".to_string())],
316        )
317        .await;
318
319        let logical_region_id = env.default_logical_region_id();
320        let request = test_util::create_logical_region_request(
321            &["job"],
322            physical_region_id,
323            &table_dir("test", logical_region_id.table_id()),
324        );
325        env.metric()
326            .handle_request(logical_region_id, RegionRequest::Create(request))
327            .await
328            .unwrap();
329        logical_region_id
330    }
331
332    #[tokio::test]
333    async fn test_bulk_insert_empty_batch_returns_zero() {
334        let env = TestEnv::new().await;
335        env.init_metric_region().await;
336        let logical_region_id = env.default_logical_region_id();
337
338        let batch = build_logical_batch(0, 0);
339        let request = RegionRequest::BulkInserts(RegionBulkInsertsRequest {
340            skip_wal: false,
341            region_id: logical_region_id,
342            payload: batch,
343            raw_data: ArrowIpc::default(),
344            partition_expr_version: None,
345            aligned_schema_version: None,
346        });
347        let response = env
348            .metric()
349            .handle_request(logical_region_id, request)
350            .await
351            .unwrap();
352        assert_eq!(response.affected_rows, 0);
353    }
354
355    #[tokio::test]
356    async fn test_bulk_insert_physical_region_passthrough() {
357        check_bulk_insert_physical_region_passthrough(false).await;
358        check_bulk_insert_physical_region_passthrough(true).await;
359    }
360
361    async fn check_bulk_insert_physical_region_passthrough(skip_wal: bool) {
362        // Use flat format so that BulkMemtable is used (supports write_bulk).
363        let mito_config = MitoConfig {
364            default_flat_format: true,
365            ..Default::default()
366        };
367        let env = TestEnv::with_mito_config("", mito_config, Default::default()).await;
368        env.init_metric_region().await;
369        env.metric().inner.flush_task.stop().await.unwrap();
370        let physical_region_id = env.default_physical_region_id();
371        let logical_region_id = env.default_logical_region_id();
372
373        // First, do a normal logical bulk insert so we can compare results.
374        let logical_batch = build_logical_batch(0, 3);
375        let logical_request = build_bulk_request(logical_region_id, logical_batch, skip_wal);
376        let response = env
377            .metric()
378            .handle_request(logical_region_id, logical_request)
379            .await
380            .unwrap();
381        assert_eq!(response.affected_rows, 3);
382
383        // Now build a physical-format batch using modify_batch_sparse (simulating
384        // what the batcher's flush_batch_physical does) and send it directly to
385        // the physical region.
386        let tag_columns = vec![TagColumnInfo {
387            name: "job".to_string(),
388            index: 2,
389            column_id: 2, // column_id for "job" in the physical table
390        }];
391        let non_tag_indices = vec![0, 1]; // timestamp, value
392        let second_batch = build_logical_batch(3, 3);
393        let physical_batch = modify_batch_sparse(
394            second_batch,
395            logical_region_id.table_id(),
396            &tag_columns,
397            &non_tag_indices,
398        )
399        .unwrap();
400        let request = build_bulk_request(physical_region_id, physical_batch, skip_wal);
401        let response = env
402            .metric()
403            .handle_request(physical_region_id, request)
404            .await
405            .unwrap();
406        assert_eq!(response.affected_rows, 3);
407
408        // Verify all 6 rows are readable from the logical region.
409        let stream = env
410            .metric()
411            .scan_to_stream(logical_region_id, ScanRequest::default())
412            .await
413            .unwrap();
414        let batches = RecordBatches::try_collect(stream).await.unwrap();
415        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 6);
416
417        // Closing without a flush must recover only WAL-backed data. Recreate
418        // the wrapper too, so its metadata cache cannot hide missing metadata.
419        let stat = env.mito().region_statistic(physical_region_id).unwrap();
420        assert_eq!(stat.sst_num, 0);
421        env.metric()
422            .handle_request(
423                physical_region_id,
424                RegionRequest::Close(RegionCloseRequest {
425                    flush_on_close: false,
426                }),
427            )
428            .await
429            .unwrap();
430        let reopened = MetricEngine::try_new(env.mito(), Default::default()).unwrap();
431        reopened.inner.flush_task.stop().await.unwrap();
432        reopened
433            .handle_request(
434                physical_region_id,
435                RegionRequest::Open(RegionOpenRequest {
436                    engine: METRIC_ENGINE_NAME.to_string(),
437                    table_dir: TestEnv::default_table_dir(),
438                    path_type: PathType::Bare,
439                    options: [(PHYSICAL_TABLE_METADATA_KEY.to_string(), String::new())]
440                        .into_iter()
441                        .collect(),
442                    skip_wal_replay: false,
443                    checkpoint: None,
444                    requirements: Default::default(),
445                }),
446            )
447            .await
448            .unwrap();
449        let stream = reopened
450            .scan_to_stream(logical_region_id, ScanRequest::default())
451            .await
452            .unwrap();
453        let batches = RecordBatches::try_collect(stream).await.unwrap();
454        assert_eq!(
455            batches.iter().map(|b| b.num_rows()).sum::<usize>(),
456            if skip_wal { 0 } else { 6 },
457        );
458    }
459
460    #[tokio::test]
461    async fn test_bulk_insert_physical_region_empty_batch() {
462        // Use flat format so that BulkMemtable is used (supports write_bulk).
463        let mito_config = MitoConfig {
464            default_flat_format: true,
465            ..Default::default()
466        };
467        let env = TestEnv::with_mito_config("", mito_config, Default::default()).await;
468        env.init_metric_region().await;
469        let physical_region_id = env.default_physical_region_id();
470
471        let batch = build_logical_batch(0, 0);
472        let request = build_bulk_request(physical_region_id, batch, false);
473        let response = env
474            .metric()
475            .handle_request(physical_region_id, request)
476            .await
477            .unwrap();
478        assert_eq!(response.affected_rows, 0);
479    }
480
481    #[tokio::test]
482    async fn test_bulk_insert_unknown_column_errors() {
483        let env = TestEnv::new().await;
484        env.init_metric_region().await;
485        let logical_region_id = env.default_logical_region_id();
486
487        let schema = Arc::new(ArrowSchema::new(vec![
488            Field::new(
489                greptime_timestamp(),
490                DataType::Timestamp(TimeUnit::Millisecond, None),
491                false,
492            ),
493            Field::new(greptime_value(), DataType::Float64, true),
494            Field::new("nonexistent_column", DataType::Utf8, true),
495        ]));
496        let batch = RecordBatch::try_new(
497            schema,
498            vec![
499                Arc::new(TimestampMillisecondArray::from(vec![0i64])),
500                Arc::new(Float64Array::from(vec![1.0])),
501                Arc::new(StringArray::from(vec!["val"])),
502            ],
503        )
504        .unwrap();
505
506        let request = build_bulk_request(logical_region_id, batch, false);
507        let err = env
508            .metric()
509            .handle_request(logical_region_id, request)
510            .await
511            .unwrap_err();
512        let Some(err) = err.as_any().downcast_ref::<Error>() else {
513            panic!("unexpected error type");
514        };
515        assert_matches!(err, Error::ColumnNotFound { .. });
516    }
517
518    #[tokio::test]
519    async fn test_bulk_insert_multiple_tag_columns() {
520        let env = TestEnv::new().await;
521        let physical_region_id = env.default_physical_region_id();
522        env.create_physical_region(physical_region_id, &TestEnv::default_table_dir(), vec![])
523            .await;
524        let logical_region_id = env.default_logical_region_id();
525        let request = test_util::create_logical_region_request(
526            &["host", "region"],
527            physical_region_id,
528            &table_dir("test", logical_region_id.table_id()),
529        );
530        env.metric()
531            .handle_request(logical_region_id, RegionRequest::Create(request))
532            .await
533            .unwrap();
534
535        let schema = Arc::new(ArrowSchema::new(vec![
536            Field::new(
537                greptime_timestamp(),
538                DataType::Timestamp(TimeUnit::Millisecond, None),
539                false,
540            ),
541            Field::new(greptime_value(), DataType::Float64, true),
542            Field::new("host", DataType::Utf8, true),
543            Field::new("region", DataType::Utf8, true),
544        ]));
545        let batch = RecordBatch::try_new(
546            schema,
547            vec![
548                Arc::new(TimestampMillisecondArray::from(vec![0i64, 1, 2])),
549                Arc::new(Float64Array::from(vec![10.0, 20.0, 30.0])),
550                Arc::new(StringArray::from(vec!["h1", "h2", "h1"])),
551                Arc::new(StringArray::from(vec!["us-east", "us-west", "eu-west"])),
552            ],
553        )
554        .unwrap();
555
556        let request = build_bulk_request(logical_region_id, batch, false);
557        let response = env
558            .metric()
559            .handle_request(logical_region_id, request)
560            .await
561            .unwrap();
562        assert_eq!(response.affected_rows, 3);
563
564        let stream = env
565            .metric()
566            .scan_to_stream(logical_region_id, ScanRequest::default())
567            .await
568            .unwrap();
569        let batches = RecordBatches::try_collect(stream).await.unwrap();
570        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 3);
571    }
572
573    #[tokio::test]
574    async fn test_bulk_insert_accumulates_rows() {
575        let env = TestEnv::new().await;
576        env.init_metric_region().await;
577        let logical_region_id = env.default_logical_region_id();
578
579        let request = build_bulk_request(logical_region_id, build_logical_batch(0, 3), false);
580        let response = env
581            .metric()
582            .handle_request(logical_region_id, request)
583            .await
584            .unwrap();
585        assert_eq!(response.affected_rows, 3);
586
587        let request = build_bulk_request(logical_region_id, build_logical_batch(3, 5), false);
588        let response = env
589            .metric()
590            .handle_request(logical_region_id, request)
591            .await
592            .unwrap();
593        assert_eq!(response.affected_rows, 5);
594
595        let stream = env
596            .metric()
597            .scan_to_stream(logical_region_id, ScanRequest::default())
598            .await
599            .unwrap();
600        let batches = RecordBatches::try_collect(stream).await.unwrap();
601        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 8);
602    }
603
604    #[tokio::test]
605    async fn test_bulk_insert_sparse_encoding() {
606        let env = TestEnv::new().await;
607        let physical_region_id = env.default_physical_region_id();
608        env.create_physical_region(
609            physical_region_id,
610            &TestEnv::default_table_dir(),
611            vec![(
612                EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING.to_string(),
613                "byte_stream_split".to_string(),
614            )],
615        )
616        .await;
617        let logical_region_id = env.default_logical_region_id();
618        env.create_logical_region(physical_region_id, logical_region_id)
619            .await;
620
621        let rows = 4;
622        let request = build_bulk_request(logical_region_id, build_logical_batch(0, rows), false);
623        let response = env
624            .metric()
625            .handle_request(logical_region_id, request)
626            .await
627            .unwrap();
628        assert_eq!(response.affected_rows, rows);
629
630        let data_region_id = crate::utils::to_data_region_id(physical_region_id);
631        env.mito()
632            .handle_request(
633                data_region_id,
634                RegionRequest::Flush(RegionFlushRequest::default()),
635            )
636            .await
637            .unwrap();
638
639        let region = env.mito().find_region(data_region_id).unwrap();
640        let entry = region
641            .manifest_sst_entries()
642            .await
643            .into_iter()
644            .find(|entry| entry.visible && entry.file_path.ends_with(".parquet"))
645            .unwrap();
646        let mut cache_metrics = MetadataCacheMetrics::default();
647        let footer = MetadataLoader::new(
648            region.access_layer().object_store().clone(),
649            &entry.file_path,
650            entry.file_size,
651        )
652        .load(&mut cache_metrics)
653        .await
654        .unwrap();
655        let field_column_index = footer
656            .file_metadata()
657            .schema_descr()
658            .columns()
659            .iter()
660            .position(|column| column.name() == greptime_value())
661            .unwrap();
662        assert!(!footer.row_groups().is_empty());
663        assert!(footer.row_groups().iter().all(|row_group| {
664            row_group
665                .column(field_column_index)
666                .encodings()
667                .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT)
668        }));
669
670        let stream = env
671            .metric()
672            .scan_to_stream(logical_region_id, ScanRequest::default())
673            .await
674            .unwrap();
675        let batches = RecordBatches::try_collect(stream).await.unwrap();
676        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), rows);
677    }
678
679    #[tokio::test]
680    async fn test_bulk_insert_dense_encoding_rejected() {
681        let env = TestEnv::new().await;
682        let logical_region_id = init_dense_metric_region(&env).await;
683
684        let request = build_bulk_request(logical_region_id, build_logical_batch(0, 2), false);
685        let err = env
686            .metric()
687            .handle_request(logical_region_id, request)
688            .await
689            .unwrap_err();
690        let Some(err) = err.as_any().downcast_ref::<Error>() else {
691            panic!("unexpected error type");
692        };
693        assert_matches!(err, Error::UnsupportedRegionRequest { .. });
694    }
695
696    #[tokio::test]
697    async fn test_bulk_insert_matches_put() {
698        let env_put = TestEnv::new().await;
699        env_put.init_metric_region().await;
700        let logical_region_id = env_put.default_logical_region_id();
701        let schema = test_util::row_schema_with_tags(&["job"]);
702        let rows = test_util::build_rows(1, 5);
703        env_put
704            .metric()
705            .handle_request(
706                logical_region_id,
707                RegionRequest::Put(RegionPutRequest {
708                    skip_wal: false,
709                    rows: api::v1::Rows { schema, rows },
710                    hint: None,
711                    partition_expr_version: None,
712                }),
713            )
714            .await
715            .unwrap();
716        let put_stream = env_put
717            .metric()
718            .scan_to_stream(logical_region_id, ScanRequest::default())
719            .await
720            .unwrap();
721        let put_batches = RecordBatches::try_collect(put_stream).await.unwrap();
722        let put_output = put_batches.pretty_print().unwrap();
723
724        let env_bulk = TestEnv::new().await;
725        env_bulk.init_metric_region().await;
726        let request = build_bulk_request(logical_region_id, build_logical_batch(0, 5), false);
727        env_bulk
728            .metric()
729            .handle_request(logical_region_id, request)
730            .await
731            .unwrap();
732        let bulk_stream = env_bulk
733            .metric()
734            .scan_to_stream(logical_region_id, ScanRequest::default())
735            .await
736            .unwrap();
737        let bulk_batches = RecordBatches::try_collect(bulk_stream).await.unwrap();
738        let bulk_output = bulk_batches.pretty_print().unwrap();
739
740        assert_eq!(put_output, bulk_output);
741    }
742}
metric_engine/engine/bulk_insert.rs

metric_engine/engine/
bulk_insert.rs
