pipeline/manager/
pipeline_operator.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::HashMap;
16use std::sync::{Arc, RwLock};
17use std::time::{Duration, Instant};
18
19use api::v1::CreateTableExpr;
20use catalog::{CatalogManagerRef, RegisterSystemTableRequest};
21use common_catalog::consts::{DEFAULT_PRIVATE_SCHEMA_NAME, default_engine};
22use common_meta::rpc::ddl::TriggerReason;
23use common_telemetry::info;
24use common_time::FOREVER;
25use datatypes::timestamp::TimestampNanosecond;
26use futures::FutureExt;
27use operator::insert::InserterRef;
28use operator::statement::StatementExecutorRef;
29use query::QueryEngineRef;
30use session::context::QueryContextRef;
31use snafu::{OptionExt, ResultExt};
32use table::TableRef;
33use table::requests::TTL_KEY;
34
35use crate::Pipeline;
36use crate::error::{CatalogSnafu, CreateTableSnafu, PipelineTableNotFoundSnafu, Result};
37use crate::manager::{PipelineInfo, PipelineTableRef, PipelineVersion};
38use crate::metrics::{
39    METRIC_PIPELINE_CREATE_HISTOGRAM, METRIC_PIPELINE_DELETE_HISTOGRAM,
40    METRIC_PIPELINE_RETRIEVE_HISTOGRAM,
41};
42use crate::options::PipelineOptions;
43use crate::table::{PIPELINE_TABLE_NAME, PipelineTable};
44
45/// PipelineOperator is responsible for managing pipelines.
46/// It provides the ability to:
47/// - Create a pipeline table if it does not exist
48/// - Get a pipeline from the pipeline table
49/// - Insert a pipeline into the pipeline table
50/// - Compile a pipeline
51/// - Add a pipeline table to the cache
52/// - Get a pipeline table from the cache
53pub struct PipelineOperator {
54    inserter: InserterRef,
55    statement_executor: StatementExecutorRef,
56    catalog_manager: CatalogManagerRef,
57    query_engine: QueryEngineRef,
58    tables: RwLock<HashMap<String, PipelineTableRef>>,
59    cache_ttl: Duration,
60}
61
62impl PipelineOperator {
63    /// Create a table request for the pipeline table.
64    fn create_table_request(&self, catalog: &str) -> RegisterSystemTableRequest {
65        let (time_index, primary_keys, column_defs) = PipelineTable::build_pipeline_schema();
66
67        let mut table_options = HashMap::new();
68        table_options.insert(TTL_KEY.to_string(), FOREVER.to_string());
69
70        let create_table_expr = CreateTableExpr {
71            catalog_name: catalog.to_string(),
72            schema_name: DEFAULT_PRIVATE_SCHEMA_NAME.to_string(),
73            table_name: PIPELINE_TABLE_NAME.to_string(),
74            desc: "GreptimeDB pipeline table for Log".to_string(),
75            column_defs,
76            time_index,
77            primary_keys,
78            create_if_not_exists: true,
79            table_options,
80            table_id: None, // Should and will be assigned by Meta.
81            engine: default_engine().to_string(),
82        };
83
84        RegisterSystemTableRequest {
85            create_table_expr,
86            open_hook: None,
87        }
88    }
89
90    fn add_pipeline_table_to_cache(&self, catalog: &str, table: TableRef) {
91        let mut tables = self.tables.write().unwrap();
92        if tables.contains_key(catalog) {
93            return;
94        }
95        tables.insert(
96            catalog.to_string(),
97            Arc::new(PipelineTable::new(
98                self.inserter.clone(),
99                self.statement_executor.clone(),
100                table,
101                self.query_engine.clone(),
102                self.cache_ttl,
103            )),
104        );
105    }
106
107    async fn create_pipeline_table_if_not_exists(&self, ctx: QueryContextRef) -> Result<()> {
108        let catalog = ctx.current_catalog();
109
110        // exist in cache
111        if self.get_pipeline_table_from_cache(catalog).is_some() {
112            return Ok(());
113        }
114
115        let RegisterSystemTableRequest {
116            create_table_expr: mut expr,
117            open_hook: _,
118        } = self.create_table_request(catalog);
119
120        // exist in catalog, just open
121        if let Some(table) = self
122            .catalog_manager
123            .table(
124                &expr.catalog_name,
125                &expr.schema_name,
126                &expr.table_name,
127                Some(&ctx),
128            )
129            .await
130            .context(CatalogSnafu)?
131        {
132            self.add_pipeline_table_to_cache(catalog, table);
133            return Ok(());
134        }
135
136        // create table
137        self.statement_executor
138            .create_table_inner(&mut expr, None, ctx.clone(), TriggerReason::AutoCreate)
139            .await
140            .context(CreateTableSnafu)?;
141
142        let schema = &expr.schema_name;
143        let table_name = &expr.table_name;
144
145        // get from catalog
146        let table = self
147            .catalog_manager
148            .table(catalog, schema, table_name, Some(&ctx))
149            .await
150            .context(CatalogSnafu)?
151            .context(PipelineTableNotFoundSnafu)?;
152
153        info!(
154            "Created pipelines table {} with table id {}.",
155            table.table_info().full_table_name(),
156            table.table_info().table_id()
157        );
158
159        // put to cache
160        self.add_pipeline_table_to_cache(catalog, table);
161
162        Ok(())
163    }
164
165    /// Get a pipeline table from the cache.
166    pub fn get_pipeline_table_from_cache(&self, catalog: &str) -> Option<PipelineTableRef> {
167        self.tables.read().unwrap().get(catalog).cloned()
168    }
169}
170
171impl PipelineOperator {
172    /// Create a new PipelineOperator.
173    pub fn new(
174        inserter: InserterRef,
175        statement_executor: StatementExecutorRef,
176        catalog_manager: CatalogManagerRef,
177        query_engine: QueryEngineRef,
178        options: &PipelineOptions,
179    ) -> Self {
180        Self {
181            inserter,
182            statement_executor,
183            catalog_manager,
184            tables: RwLock::new(HashMap::new()),
185            query_engine,
186            cache_ttl: options.cache_ttl,
187        }
188    }
189
190    /// Get a pipeline from the pipeline table.
191    pub async fn get_pipeline(
192        &self,
193        query_ctx: QueryContextRef,
194        name: &str,
195        version: PipelineVersion,
196    ) -> Result<Arc<Pipeline>> {
197        let schema = query_ctx.current_schema();
198        self.create_pipeline_table_if_not_exists(query_ctx.clone())
199            .await?;
200
201        let timer = Instant::now();
202        self.get_pipeline_table_from_cache(query_ctx.current_catalog())
203            .context(PipelineTableNotFoundSnafu)?
204            .get_pipeline(&schema, name, version)
205            .inspect(|re| {
206                METRIC_PIPELINE_RETRIEVE_HISTOGRAM
207                    .with_label_values(&[&re.is_ok().to_string()])
208                    .observe(timer.elapsed().as_secs_f64())
209            })
210            .await
211    }
212
213    /// Get a original pipeline by name.
214    pub async fn get_pipeline_str(
215        &self,
216        name: &str,
217        version: PipelineVersion,
218        query_ctx: QueryContextRef,
219    ) -> Result<(String, TimestampNanosecond)> {
220        let schema = query_ctx.current_schema();
221        self.create_pipeline_table_if_not_exists(query_ctx.clone())
222            .await?;
223
224        let timer = Instant::now();
225        self.get_pipeline_table_from_cache(query_ctx.current_catalog())
226            .context(PipelineTableNotFoundSnafu)?
227            .get_pipeline_str(&schema, name, version)
228            .inspect(|re| {
229                METRIC_PIPELINE_RETRIEVE_HISTOGRAM
230                    .with_label_values(&[&re.is_ok().to_string()])
231                    .observe(timer.elapsed().as_secs_f64())
232            })
233            .await
234            .map(|p| (p.content, p.version))
235    }
236
237    /// Insert a pipeline into the pipeline table.
238    pub async fn insert_pipeline(
239        &self,
240        name: &str,
241        content_type: &str,
242        pipeline: &str,
243        query_ctx: QueryContextRef,
244    ) -> Result<PipelineInfo> {
245        self.create_pipeline_table_if_not_exists(query_ctx.clone())
246            .await?;
247
248        let timer = Instant::now();
249        self.get_pipeline_table_from_cache(query_ctx.current_catalog())
250            .context(PipelineTableNotFoundSnafu)?
251            .insert_and_compile(name, content_type, pipeline)
252            .inspect(|re| {
253                METRIC_PIPELINE_CREATE_HISTOGRAM
254                    .with_label_values(&[&re.is_ok().to_string()])
255                    .observe(timer.elapsed().as_secs_f64())
256            })
257            .await
258    }
259
260    /// Delete a pipeline by name from pipeline table.
261    pub async fn delete_pipeline(
262        &self,
263        name: &str,
264        version: PipelineVersion,
265        query_ctx: QueryContextRef,
266    ) -> Result<Option<()>> {
267        // trigger load pipeline table
268        self.create_pipeline_table_if_not_exists(query_ctx.clone())
269            .await?;
270
271        let timer = Instant::now();
272        self.get_pipeline_table_from_cache(query_ctx.current_catalog())
273            .context(PipelineTableNotFoundSnafu)?
274            .delete_pipeline(name, version)
275            .inspect(|re| {
276                METRIC_PIPELINE_DELETE_HISTOGRAM
277                    .with_label_values(&[&re.is_ok().to_string()])
278                    .observe(timer.elapsed().as_secs_f64())
279            })
280            .await
281    }
282
283    /// Compile a pipeline.
284    pub fn build_pipeline(pipeline: &str) -> Result<Pipeline> {
285        PipelineTable::compile_pipeline(pipeline)
286    }
287}
pipeline/manager/pipeline_operator.rs

pipeline/manager/
pipeline_operator.rs
