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

risingwave_connector/sink/elasticsearch_opensearch/
opensearch.rs
