frontend/
events.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::time::Duration;
16
17use async_trait::async_trait;
18use client::inserter::{Context, InsertOptions, Inserter};
19use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME};
20use common_error::ext::BoxedError;
21use common_event_recorder::error::{InsertEventsSnafu, Result};
22use common_event_recorder::{
23    DEFAULT_COMPACTION_TIME_WINDOW, Event, EventHandler, build_row_inserts_request,
24    group_events_by_type,
25};
26use operator::statement::{InserterImpl, StatementExecutorRef};
27use snafu::ResultExt;
28
29/// EventHandlerImpl is the default event handler implementation in frontend.
30pub struct EventHandlerImpl {
31    inserter: Box<dyn Inserter>,
32}
33
34impl EventHandlerImpl {
35    /// Create a new EventHandlerImpl.
36    pub fn new(statement_executor: StatementExecutorRef, ttl: Duration) -> Self {
37        Self {
38            inserter: Box::new(InserterImpl::new(
39                statement_executor,
40                Some(InsertOptions {
41                    ttl,
42                    append_mode: true,
43                    twcs_compaction_time_window: Some(DEFAULT_COMPACTION_TIME_WINDOW),
44                }),
45            )),
46        }
47    }
48}
49
50const DEFAULT_CONTEXT: Context = Context {
51    catalog: DEFAULT_CATALOG_NAME,
52    schema: DEFAULT_PRIVATE_SCHEMA_NAME,
53};
54
55#[async_trait]
56impl EventHandler for EventHandlerImpl {
57    async fn handle(&self, events: &[Box<dyn Event>]) -> Result<()> {
58        let event_groups = group_events_by_type(events);
59
60        for (_, events) in event_groups {
61            let requests = build_row_inserts_request(&events)?;
62
63            self.inserter
64                .insert_rows(&DEFAULT_CONTEXT, requests)
65                .await
66                .map_err(BoxedError::new)
67                .context(InsertEventsSnafu)?;
68        }
69
70        Ok(())
71    }
72}
frontend/events.rs

frontend/
events.rs
