risingwave_connector/sink/elasticsearch_opensearch/
opensearch.rs

1// Copyright 2024 RisingWave Labs
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 anyhow::anyhow;
16use risingwave_common::catalog::Schema;
17use risingwave_common::session_config::sink_decouple::SinkDecouple;
18use tonic::async_trait;
19
20use super::super::writer::{AsyncTruncateLogSinkerOf, AsyncTruncateSinkWriterExt};
21use super::super::{Sink, SinkError, SinkParam, SinkWriterParam};
22use super::elasticsearch_opensearch_client::ElasticSearchOpenSearchSinkWriter;
23use super::elasticsearch_opensearch_config::{ElasticSearchOpenSearchConfig, OpenSearchConfig};
24use crate::enforce_secret::EnforceSecret;
25use crate::sink::Result;
26
27pub const OPENSEARCH_SINK: &str = "opensearch";
28
29#[derive(Debug)]
30pub struct OpenSearchSink {
31    config: ElasticSearchOpenSearchConfig,
32    schema: Schema,
33    pk_indices: Vec<usize>,
34    is_append_only: bool,
35}
36
37impl EnforceSecret for OpenSearchSink {
38    fn enforce_secret<'a>(
39        prop_iter: impl Iterator<Item = &'a str>,
40    ) -> crate::error::ConnectorResult<()> {
41        for prop in prop_iter {
42            ElasticSearchOpenSearchConfig::enforce_one(prop)?;
43        }
44        Ok(())
45    }
46}
47#[async_trait]
48impl TryFrom<SinkParam> for OpenSearchSink {
49    type Error = SinkError;
50
51    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {
52        let schema = param.schema();
53        let pk_indices = param.downstream_pk_or_empty();
54        let config = OpenSearchConfig::from_btreemap(param.properties)?.inner;
55        Ok(Self {
56            config,
57            schema,
58            pk_indices,
59            is_append_only: param.sink_type.is_append_only(),
60        })
61    }
62}
63
64impl Sink for OpenSearchSink {
65    type LogSinker = AsyncTruncateLogSinkerOf<ElasticSearchOpenSearchSinkWriter>;
66
67    const SINK_NAME: &'static str = OPENSEARCH_SINK;
68
69    fn validate_unknown_fields(&self) -> Result<()> {
70        crate::sink::validate_sink_unknown_fields(&self.config)
71    }
72
73    async fn validate(&self) -> Result<()> {
74        risingwave_common::license::Feature::OpenSearchSink
75            .check_available()
76            .map_err(|e| anyhow::anyhow!(e))?;
77        self.config.validate_config(&self.schema)?;
78        let client = self.config.build_client(Self::SINK_NAME)?;
79        client.ping().await?;
80        Ok(())
81    }
82
83    async fn new_log_sinker(&self, _writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
84        Ok(ElasticSearchOpenSearchSinkWriter::new(
85            self.config.clone(),
86            self.schema.clone(),
87            self.pk_indices.clone(),
88            Self::SINK_NAME,
89            self.is_append_only,
90        )?
91        .into_log_sinker(self.config.concurrent_requests))
92    }
93
94    fn set_default_commit_checkpoint_interval(
95        desc: &mut crate::sink::catalog::desc::SinkDesc,
96        user_specified: &risingwave_common::session_config::sink_decouple::SinkDecouple,
97    ) -> Result<()> {
98        if crate::sink::is_sink_support_commit_checkpoint_interval(Self::SINK_NAME) {
99            match desc
100                .properties
101                .get(crate::sink::decouple_checkpoint_log_sink::COMMIT_CHECKPOINT_INTERVAL)
102            {
103                Some(commit_checkpoint_interval) => {
104                    let commit_checkpoint_interval = commit_checkpoint_interval
105                        .parse::<u64>()
106                        .map_err(|e| SinkError::Config(anyhow!(e)))?;
107                    if std::matches!(user_specified, SinkDecouple::Disable)
108                        && commit_checkpoint_interval > 1
109                    {
110                        return Err(SinkError::Config(anyhow!(
111                            "config conflict: `commit_checkpoint_interval` larger than 1 means that sink decouple must be enabled, but session config sink_decouple is disabled"
112                        )));
113                    }
114                }
115                None => match user_specified {
116                    risingwave_common::session_config::sink_decouple::SinkDecouple::Default
117                    | risingwave_common::session_config::sink_decouple::SinkDecouple::Enable => {
118                        desc.properties.insert(
119                            crate::sink::decouple_checkpoint_log_sink::COMMIT_CHECKPOINT_INTERVAL.to_owned(),
120                            crate::sink::decouple_checkpoint_log_sink::DEFAULT_COMMIT_CHECKPOINT_INTERVAL_WITH_SINK_DECOUPLE.to_string(),
121                        );
122                    }
123                    risingwave_common::session_config::sink_decouple::SinkDecouple::Disable => {
124                        desc.properties.insert(
125                            crate::sink::decouple_checkpoint_log_sink::COMMIT_CHECKPOINT_INTERVAL.to_owned(),
126                            crate::sink::decouple_checkpoint_log_sink::DEFAULT_COMMIT_CHECKPOINT_INTERVAL_WITHOUT_SINK_DECOUPLE.to_string(),
127                        );
128                    }
129                },
130            }
131        }
132        Ok(())
133    }
134
135    fn is_sink_decouple(
136        user_specified: &risingwave_common::session_config::sink_decouple::SinkDecouple,
137    ) -> Result<bool> {
138        match user_specified {
139            risingwave_common::session_config::sink_decouple::SinkDecouple::Default
140            | risingwave_common::session_config::sink_decouple::SinkDecouple::Enable => Ok(true),
141            risingwave_common::session_config::sink_decouple::SinkDecouple::Disable => Ok(false),
142        }
143    }
144}
risingwave_connector/sink/elasticsearch_opensearch/opensearch.rs

risingwave_connector/sink/elasticsearch_opensearch/
opensearch.rs
