(
- vec: GenericCounterVec,
-) -> MetricVec> {
- vec
-}
-
-pub fn __extract_gauge_builder(
- vec: GenericGaugeVec,
-) -> MetricVec> {
- vec
-}
-
-pub fn __extract_histogram_builder(vec: HistogramVec) -> MetricVec {
- vec
-}
-
#[macro_export]
macro_rules! register_guarded_histogram_vec_with_registry {
($NAME:expr, $HELP:expr, $LABELS_NAMES:expr, $REGISTRY:expr $(,)?) => {{
@@ -105,9 +89,35 @@ macro_rules! register_guarded_int_counter_vec_with_registry {
}};
}
-pub type VecBuilderOfCounter = impl MetricVecBuilder>;
-pub type VecBuilderOfGauge = impl MetricVecBuilder>;
-pub type VecBuilderOfHistogram = impl MetricVecBuilder;
+// put TAITs in a separate module to avoid "non-defining opaque type use in defining scope"
+mod tait {
+ use prometheus::core::{
+ Atomic, GenericCounter, GenericCounterVec, GenericGauge, GenericGaugeVec, MetricVec,
+ MetricVecBuilder,
+ };
+ use prometheus::{Histogram, HistogramVec};
+
+ pub type VecBuilderOfCounter = impl MetricVecBuilder>;
+ pub type VecBuilderOfGauge = impl MetricVecBuilder>;
+ pub type VecBuilderOfHistogram = impl MetricVecBuilder;
+
+ pub fn __extract_counter_builder(
+ vec: GenericCounterVec,
+ ) -> MetricVec> {
+ vec
+ }
+
+ pub fn __extract_gauge_builder(
+ vec: GenericGaugeVec,
+ ) -> MetricVec> {
+ vec
+ }
+
+ pub fn __extract_histogram_builder(vec: HistogramVec) -> MetricVec {
+ vec
+ }
+}
+pub use tait::*;
pub type LabelGuardedHistogramVec = LabelGuardedMetricVec;
pub type LabelGuardedIntCounterVec =
diff --git a/src/common/metrics/src/lib.rs b/src/common/metrics/src/lib.rs
index a2e4156b525d1..574e684c6703b 100644
--- a/src/common/metrics/src/lib.rs
+++ b/src/common/metrics/src/lib.rs
@@ -15,7 +15,6 @@
#![feature(lazy_cell)]
#![feature(type_alias_impl_trait)]
#![feature(impl_trait_in_assoc_type)]
-#![feature(array_methods)]
use std::ops::Deref;
use std::sync::LazyLock;
diff --git a/src/common/src/acl/mod.rs b/src/common/src/acl/mod.rs
index 91f69fea51310..b1c41bc8ec6ad 100644
--- a/src/common/src/acl/mod.rs
+++ b/src/common/src/acl/mod.rs
@@ -14,7 +14,6 @@
//! `Acl` defines all grantable privileges.
-use std::convert::Into;
use std::fmt::Formatter;
use std::sync::LazyLock;
@@ -100,11 +99,11 @@ pub static ALL_AVAILABLE_DATABASE_MODES: LazyLock =
LazyLock::new(|| make_bitflags!(AclMode::{Create | Connect}).into());
pub static ALL_AVAILABLE_SCHEMA_MODES: LazyLock =
LazyLock::new(|| make_bitflags!(AclMode::{Create | Usage}).into());
+// Including TABLES and VIEWS
pub static ALL_AVAILABLE_TABLE_MODES: LazyLock =
LazyLock::new(|| make_bitflags!(AclMode::{Select | Insert | Update | Delete}).into());
pub static ALL_AVAILABLE_SOURCE_MODES: LazyLock = LazyLock::new(AclModeSet::readonly);
pub static ALL_AVAILABLE_MVIEW_MODES: LazyLock = LazyLock::new(AclModeSet::readonly);
-pub static ALL_AVAILABLE_VIEW_MODES: LazyLock = LazyLock::new(AclModeSet::readonly);
pub static ALL_AVAILABLE_SINK_MODES: LazyLock = LazyLock::new(AclModeSet::empty);
pub static ALL_AVAILABLE_SUBSCRIPTION_MODES: LazyLock =
LazyLock::new(AclModeSet::empty);
diff --git a/src/common/src/array/arrow/arrow_impl.rs b/src/common/src/array/arrow/arrow_impl.rs
index 514d3b299769c..773efdd088e3d 100644
--- a/src/common/src/array/arrow/arrow_impl.rs
+++ b/src/common/src/array/arrow/arrow_impl.rs
@@ -36,8 +36,11 @@
//! mod arrow_impl;
//! ```
+// Is this a bug? Why do we have these lints?
+#![allow(unused_imports)]
+#![allow(dead_code)]
+
use std::fmt::Write;
-use std::sync::Arc;
use arrow_buffer::OffsetBuffer;
use chrono::{NaiveDateTime, NaiveTime};
@@ -47,7 +50,6 @@ use itertools::Itertools;
use super::{arrow_array, arrow_buffer, arrow_cast, arrow_schema};
// Other import should always use the absolute path.
use crate::array::*;
-use crate::buffer::Bitmap;
use crate::types::*;
use crate::util::iter_util::ZipEqFast;
diff --git a/src/common/src/array/arrow/arrow_udf.rs b/src/common/src/array/arrow/arrow_udf.rs
index e2f9e39ad385a..83383044f506d 100644
--- a/src/common/src/array/arrow/arrow_udf.rs
+++ b/src/common/src/array/arrow/arrow_udf.rs
@@ -29,82 +29,110 @@ use crate::array::{ArrayError, ArrayImpl, DataType, DecimalArray, JsonbArray};
#[path = "./arrow_impl.rs"]
mod arrow_impl;
-/// Arrow conversion for the current version of UDF. This is in use but will be deprecated soon.
-///
-/// In the current version of UDF protocol, decimal and jsonb types are mapped to Arrow `LargeBinary` and `LargeUtf8` types.
-pub struct UdfArrowConvert;
+/// Arrow conversion for UDF.
+#[derive(Default, Debug)]
+pub struct UdfArrowConvert {
+ /// Whether the UDF talks in legacy mode.
+ ///
+ /// If true, decimal and jsonb types are mapped to Arrow `LargeBinary` and `LargeUtf8` types.
+ /// Otherwise, they are mapped to Arrow extension types.
+ /// See .
+ pub legacy: bool,
+}
impl ToArrow for UdfArrowConvert {
- // Decimal values are stored as ASCII text representation in a large binary array.
fn decimal_to_arrow(
&self,
_data_type: &arrow_schema::DataType,
array: &DecimalArray,
) -> Result {
- Ok(Arc::new(arrow_array::LargeBinaryArray::from(array)))
+ if self.legacy {
+ // Decimal values are stored as ASCII text representation in a large binary array.
+ Ok(Arc::new(arrow_array::LargeBinaryArray::from(array)))
+ } else {
+ Ok(Arc::new(arrow_array::StringArray::from(array)))
+ }
}
- // JSON values are stored as text representation in a large string array.
fn jsonb_to_arrow(&self, array: &JsonbArray) -> Result {
- Ok(Arc::new(arrow_array::LargeStringArray::from(array)))
+ if self.legacy {
+ // JSON values are stored as text representation in a large string array.
+ Ok(Arc::new(arrow_array::LargeStringArray::from(array)))
+ } else {
+ Ok(Arc::new(arrow_array::StringArray::from(array)))
+ }
}
fn jsonb_type_to_arrow(&self, name: &str) -> arrow_schema::Field {
- arrow_schema::Field::new(name, arrow_schema::DataType::LargeUtf8, true)
+ if self.legacy {
+ arrow_schema::Field::new(name, arrow_schema::DataType::LargeUtf8, true)
+ } else {
+ arrow_schema::Field::new(name, arrow_schema::DataType::Utf8, true)
+ .with_metadata([("ARROW:extension:name".into(), "arrowudf.json".into())].into())
+ }
}
fn decimal_type_to_arrow(&self, name: &str) -> arrow_schema::Field {
- arrow_schema::Field::new(name, arrow_schema::DataType::LargeBinary, true)
+ if self.legacy {
+ arrow_schema::Field::new(name, arrow_schema::DataType::LargeBinary, true)
+ } else {
+ arrow_schema::Field::new(name, arrow_schema::DataType::Utf8, true)
+ .with_metadata([("ARROW:extension:name".into(), "arrowudf.decimal".into())].into())
+ }
}
}
impl FromArrow for UdfArrowConvert {
fn from_large_utf8(&self) -> Result {
- Ok(DataType::Jsonb)
+ if self.legacy {
+ Ok(DataType::Jsonb)
+ } else {
+ Ok(DataType::Varchar)
+ }
}
fn from_large_binary(&self) -> Result {
- Ok(DataType::Decimal)
+ if self.legacy {
+ Ok(DataType::Decimal)
+ } else {
+ Ok(DataType::Bytea)
+ }
}
fn from_large_utf8_array(
&self,
array: &arrow_array::LargeStringArray,
) -> Result {
- Ok(ArrayImpl::Jsonb(array.try_into()?))
+ if self.legacy {
+ Ok(ArrayImpl::Jsonb(array.try_into()?))
+ } else {
+ Ok(ArrayImpl::Utf8(array.into()))
+ }
}
fn from_large_binary_array(
&self,
array: &arrow_array::LargeBinaryArray,
) -> Result {
- Ok(ArrayImpl::Decimal(array.try_into()?))
+ if self.legacy {
+ Ok(ArrayImpl::Decimal(array.try_into()?))
+ } else {
+ Ok(ArrayImpl::Bytea(array.into()))
+ }
}
}
-/// Arrow conversion for the next version of UDF. This is unused for now.
-///
-/// In the next version of UDF protocol, decimal and jsonb types will be mapped to Arrow extension types.
-/// See .
-pub struct NewUdfArrowConvert;
-
-impl ToArrow for NewUdfArrowConvert {}
-impl FromArrow for NewUdfArrowConvert {}
-
#[cfg(test)]
mod tests {
- use std::sync::Arc;
-
use super::*;
use crate::array::*;
- use crate::buffer::Bitmap;
#[test]
fn struct_array() {
// Empty array - risingwave to arrow conversion.
let test_arr = StructArray::new(StructType::empty(), vec![], Bitmap::ones(0));
assert_eq!(
- UdfArrowConvert
+ UdfArrowConvert::default()
.struct_to_arrow(
&arrow_schema::DataType::Struct(arrow_schema::Fields::empty()),
&test_arr
@@ -117,7 +145,7 @@ mod tests {
// Empty array - arrow to risingwave conversion.
let test_arr_2 = arrow_array::StructArray::from(vec![]);
assert_eq!(
- UdfArrowConvert
+ UdfArrowConvert::default()
.from_struct_array(&test_arr_2)
.unwrap()
.len(),
@@ -146,7 +174,7 @@ mod tests {
),
])
.unwrap();
- let actual_risingwave_struct_array = UdfArrowConvert
+ let actual_risingwave_struct_array = UdfArrowConvert::default()
.from_struct_array(&test_arrow_struct_array)
.unwrap()
.into_struct();
@@ -168,8 +196,10 @@ mod tests {
fn list() {
let array = ListArray::from_iter([None, Some(vec![0, -127, 127, 50]), Some(vec![0; 0])]);
let data_type = arrow_schema::DataType::new_list(arrow_schema::DataType::Int32, true);
- let arrow = UdfArrowConvert.list_to_arrow(&data_type, &array).unwrap();
- let rw_array = UdfArrowConvert
+ let arrow = UdfArrowConvert::default()
+ .list_to_arrow(&data_type, &array)
+ .unwrap();
+ let rw_array = UdfArrowConvert::default()
.from_list_array(arrow.as_any().downcast_ref().unwrap())
.unwrap();
assert_eq!(rw_array.as_list(), &array);
diff --git a/src/common/src/array/data_chunk.rs b/src/common/src/array/data_chunk.rs
index cdb012a3185cb..7fb908ecc138f 100644
--- a/src/common/src/array/data_chunk.rs
+++ b/src/common/src/array/data_chunk.rs
@@ -13,10 +13,10 @@
// limitations under the License.
use std::borrow::Cow;
+use std::fmt;
use std::fmt::Display;
use std::hash::BuildHasher;
use std::sync::Arc;
-use std::{fmt, usize};
use bytes::Bytes;
use either::Either;
@@ -249,6 +249,7 @@ impl DataChunk {
Self::new(columns, Bitmap::ones(cardinality))
}
+ /// Scatter a compacted chunk to a new chunk with the given visibility.
pub fn uncompact(self, vis: Bitmap) -> Self {
let mut uncompact_builders: Vec<_> = self
.columns
diff --git a/src/common/src/array/primitive_array.rs b/src/common/src/array/primitive_array.rs
index 9840ba68135c5..29c7bfb49289a 100644
--- a/src/common/src/array/primitive_array.rs
+++ b/src/common/src/array/primitive_array.rs
@@ -315,7 +315,6 @@ impl ArrayBuilder for PrimitiveArrayBuilder {
#[cfg(test)]
mod tests {
use super::*;
- use crate::types::{F32, F64};
fn helper_test_builder(data: Vec>) -> PrimitiveArray {
let mut builder = PrimitiveArrayBuilder::::new(data.len());
diff --git a/src/common/src/array/proto_reader.rs b/src/common/src/array/proto_reader.rs
index b0b133d1b2df0..073ad0b3de7ba 100644
--- a/src/common/src/array/proto_reader.rs
+++ b/src/common/src/array/proto_reader.rs
@@ -17,16 +17,10 @@ use std::io::{Cursor, Read};
use anyhow::Context;
use byteorder::{BigEndian, ReadBytesExt};
use paste::paste;
-use risingwave_pb::data::{PbArray, PbArrayType};
+use risingwave_pb::data::PbArrayType;
use super::*;
use crate::array::value_reader::{PrimitiveValueReader, VarSizedValueReader};
-use crate::array::{
- Array, ArrayBuilder, ArrayImpl, ArrayResult, BoolArray, DateArrayBuilder, IntervalArrayBuilder,
- PrimitiveArrayBuilder, PrimitiveArrayItemType, TimeArrayBuilder, TimestampArrayBuilder,
-};
-use crate::buffer::Bitmap;
-use crate::types::{Date, Interval, Time, Timestamp};
impl ArrayImpl {
pub fn from_protobuf(array: &PbArray, cardinality: usize) -> ArrayResult {
@@ -252,12 +246,6 @@ fn read_string_array>(
#[cfg(test)]
mod tests {
use super::*;
- use crate::array::{
- Array, ArrayBuilder, BoolArray, BoolArrayBuilder, DateArray, DateArrayBuilder,
- DecimalArray, DecimalArrayBuilder, I32Array, I32ArrayBuilder, TimeArray, TimeArrayBuilder,
- TimestampArray, TimestampArrayBuilder, Utf8Array, Utf8ArrayBuilder,
- };
- use crate::types::{Date, Decimal, Time, Timestamp};
// Convert a column to protobuf, then convert it back to column, and ensures the two are
// identical.
diff --git a/src/common/src/cache.rs b/src/common/src/cache.rs
index 99a373d6a94a8..bb4af1d615eb9 100644
--- a/src/common/src/cache.rs
+++ b/src/common/src/cache.rs
@@ -1019,14 +1019,12 @@ impl Clone for CacheableEntry {
#[cfg(test)]
mod tests {
use std::collections::hash_map::DefaultHasher;
- use std::hash::{Hash, Hasher};
+ use std::hash::Hasher;
use std::pin::Pin;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering::Relaxed;
- use std::sync::Arc;
use std::task::{Context, Poll};
- use futures::FutureExt;
use rand::rngs::SmallRng;
use rand::{RngCore, SeedableRng};
use tokio::sync::oneshot::error::TryRecvError;
diff --git a/src/common/src/catalog/mod.rs b/src/common/src/catalog/mod.rs
index 1909ca1635f7c..cef5b021b9dbd 100644
--- a/src/common/src/catalog/mod.rs
+++ b/src/common/src/catalog/mod.rs
@@ -33,6 +33,7 @@ use risingwave_pb::catalog::{
};
use risingwave_pb::plan_common::ColumnDescVersion;
pub use schema::{test_utils as schema_test_utils, Field, FieldDisplay, Schema};
+use serde::{Deserialize, Serialize};
use crate::array::DataChunk;
pub use crate::constants::hummock;
@@ -228,7 +229,20 @@ impl From for u32 {
}
}
-#[derive(Clone, Copy, Debug, Display, Default, Hash, PartialOrd, PartialEq, Eq, Ord)]
+#[derive(
+ Clone,
+ Copy,
+ Debug,
+ Display,
+ Default,
+ Hash,
+ PartialOrd,
+ PartialEq,
+ Eq,
+ Ord,
+ Serialize,
+ Deserialize,
+)]
#[display("{table_id}")]
pub struct TableId {
pub table_id: u32,
diff --git a/src/common/src/config.rs b/src/common/src/config.rs
index a5cb55445c070..41a4e1415b301 100644
--- a/src/common/src/config.rs
+++ b/src/common/src/config.rs
@@ -24,7 +24,7 @@ use std::num::NonZeroUsize;
use anyhow::Context;
use clap::ValueEnum;
use educe::Educe;
-use foyer::memory::{LfuConfig, LruConfig, S3FifoConfig};
+use foyer::{LfuConfig, LruConfig, S3FifoConfig};
use risingwave_common_proc_macro::ConfigDoc;
pub use risingwave_common_proc_macro::OverrideConfig;
use risingwave_pb::meta::SystemParams;
@@ -583,6 +583,8 @@ pub enum CacheEvictionConfig {
},
S3Fifo {
small_queue_capacity_ratio_in_percent: Option,
+ ghost_queue_capacity_ratio_in_percent: Option,
+ small_to_main_freq_threshold: Option,
},
}
@@ -1347,6 +1349,14 @@ pub mod default {
10
}
+ pub fn ghost_queue_capacity_ratio_in_percent() -> usize {
+ 1000
+ }
+
+ pub fn small_to_main_freq_threshold() -> u8 {
+ 1
+ }
+
pub fn meta_cache_capacity_mb() -> usize {
128
}
@@ -1889,6 +1899,16 @@ impl EvictionConfig {
}
}
+impl From for foyer::EvictionConfig {
+ fn from(value: EvictionConfig) -> Self {
+ match value {
+ EvictionConfig::Lru(lru) => foyer::EvictionConfig::Lru(lru),
+ EvictionConfig::Lfu(lfu) => foyer::EvictionConfig::Lfu(lfu),
+ EvictionConfig::S3Fifo(s3fifo) => foyer::EvictionConfig::S3Fifo(s3fifo),
+ }
+ }
+}
+
pub struct StorageMemoryConfig {
pub block_cache_capacity_mb: usize,
pub block_cache_shard_num: usize,
@@ -1975,11 +1995,19 @@ pub fn extract_storage_memory_config(s: &RwConfig) -> StorageMemoryConfig {
}),
CacheEvictionConfig::S3Fifo {
small_queue_capacity_ratio_in_percent,
+ ghost_queue_capacity_ratio_in_percent,
+ small_to_main_freq_threshold,
} => EvictionConfig::S3Fifo(S3FifoConfig {
small_queue_capacity_ratio: small_queue_capacity_ratio_in_percent
.unwrap_or(default::storage::small_queue_capacity_ratio_in_percent())
as f64
/ 100.0,
+ ghost_queue_capacity_ratio: ghost_queue_capacity_ratio_in_percent
+ .unwrap_or(default::storage::ghost_queue_capacity_ratio_in_percent())
+ as f64
+ / 100.0,
+ small_to_main_freq_threshold: small_to_main_freq_threshold
+ .unwrap_or(default::storage::small_to_main_freq_threshold()),
}),
}
};
@@ -2053,8 +2081,6 @@ pub struct CompactionConfig {
#[cfg(test)]
mod tests {
- use std::collections::BTreeMap;
-
use super::*;
/// This test ensures that `config/example.toml` is up-to-date with the default values specified
diff --git a/src/common/src/field_generator/varchar.rs b/src/common/src/field_generator/varchar.rs
index c4af1499620ca..9f68bcb287c46 100644
--- a/src/common/src/field_generator/varchar.rs
+++ b/src/common/src/field_generator/varchar.rs
@@ -12,8 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-use std::string::ToString;
-
use rand::distributions::Alphanumeric;
use rand::rngs::StdRng;
use rand::{Rng, SeedableRng};
diff --git a/src/common/src/hash/consistent_hash/mapping.rs b/src/common/src/hash/consistent_hash/mapping.rs
index c542ab2050cf1..07e62b7eac275 100644
--- a/src/common/src/hash/consistent_hash/mapping.rs
+++ b/src/common/src/hash/consistent_hash/mapping.rs
@@ -335,7 +335,6 @@ mod tests {
use rand::Rng;
use super::*;
- use crate::util::iter_util::ZipEqDebug;
struct Test;
impl VnodeMappingItem for Test {
diff --git a/src/common/src/hash/consistent_hash/vnode.rs b/src/common/src/hash/consistent_hash/vnode.rs
index 9bc49a9372ac0..fc6dac0978adf 100644
--- a/src/common/src/hash/consistent_hash/vnode.rs
+++ b/src/common/src/hash/consistent_hash/vnode.rs
@@ -169,7 +169,6 @@ mod tests {
use super::*;
use crate::array::DataChunkTestExt;
use crate::row::OwnedRow;
- use crate::types::ScalarImpl;
use crate::util::row_id::RowIdGenerator;
#[test]
diff --git a/src/common/src/hash/key.rs b/src/common/src/hash/key.rs
index c136710edb23f..4911e041370a9 100644
--- a/src/common/src/hash/key.rs
+++ b/src/common/src/hash/key.rs
@@ -21,7 +21,6 @@
//! are encoded from both `t.b` and `t.c`. If `t.b="abc"` and `t.c=1`, the hashkey may be
//! encoded in certain format of `("abc", 1)`.
-use std::convert::TryInto;
use std::default::Default;
use std::fmt::Debug;
use std::hash::{BuildHasher, Hasher};
@@ -643,11 +642,9 @@ mod tests {
DateArray, DecimalArray, F32Array, F64Array, I16Array, I32Array, I32ArrayBuilder, I64Array,
TimeArray, TimestampArray, Utf8Array,
};
- use crate::hash::{
- HashKey, Key128, Key16, Key256, Key32, Key64, KeySerialized, PrecomputedBuildHasher,
- };
+ use crate::hash::{HashKey, Key128, Key16, Key256, Key32, Key64, KeySerialized};
use crate::test_utils::rand_array::seed_rand_array_ref;
- use crate::types::{DataType, Datum};
+ use crate::types::Datum;
#[derive(Hash, PartialEq, Eq)]
struct Row(Vec);
diff --git a/src/common/src/lib.rs b/src/common/src/lib.rs
index ea3cd4a4f7102..c30b12ba4cf2d 100644
--- a/src/common/src/lib.rs
+++ b/src/common/src/lib.rs
@@ -34,16 +34,14 @@
#![feature(inline_const_pat)]
#![allow(incomplete_features)]
#![feature(iterator_try_collect)]
-#![feature(round_ties_even)]
#![feature(iter_order_by)]
#![feature(exclusive_range_pattern)]
#![feature(binary_heap_into_iter_sorted)]
#![feature(impl_trait_in_assoc_type)]
#![feature(map_entry_replace)]
#![feature(negative_impls)]
-#![feature(bound_map)]
-#![feature(array_methods)]
#![feature(register_tool)]
+#![feature(btree_cursors)]
#![register_tool(rw)]
#[cfg_attr(not(test), allow(unused_extern_crates))]
diff --git a/src/common/src/row/owned_row.rs b/src/common/src/row/owned_row.rs
index 8ea3803538d84..15a01d192a64c 100644
--- a/src/common/src/row/owned_row.rs
+++ b/src/common/src/row/owned_row.rs
@@ -182,7 +182,7 @@ mod tests {
use super::*;
use crate::row::RowExt;
- use crate::types::{DataType as Ty, Interval, ScalarImpl};
+ use crate::types::DataType as Ty;
use crate::util::hash_util::Crc32FastBuilder;
#[test]
diff --git a/src/common/src/system_param/mod.rs b/src/common/src/system_param/mod.rs
index 06d4cce2e4e6b..f71324cb4e55d 100644
--- a/src/common/src/system_param/mod.rs
+++ b/src/common/src/system_param/mod.rs
@@ -77,10 +77,10 @@ macro_rules! for_all_params {
{ barrier_interval_ms, u32, Some(1000_u32), true, "The interval of periodic barrier.", },
{ checkpoint_frequency, u64, Some(1_u64), true, "There will be a checkpoint for every n barriers.", },
{ sstable_size_mb, u32, Some(256_u32), false, "Target size of the Sstable.", },
- { parallel_compact_size_mb, u32, Some(512_u32), false, "", },
+ { parallel_compact_size_mb, u32, Some(512_u32), false, "The size of parallel task for one compact/flush job.", },
{ block_size_kb, u32, Some(64_u32), false, "Size of each block in bytes in SST.", },
{ bloom_false_positive, f64, Some(0.001_f64), false, "False positive probability of bloom filter.", },
- { state_store, String, None, false, "", },
+ { state_store, String, None, false, "URL for the state store", },
{ data_directory, String, None, false, "Remote directory for storing data and metadata objects.", },
{ backup_storage_url, String, None, true, "Remote storage url for storing snapshots.", },
{ backup_storage_directory, String, None, true, "Remote directory for storing snapshots.", },
diff --git a/src/common/src/types/decimal.rs b/src/common/src/types/decimal.rs
index b6031a1665540..b38dbb4822e68 100644
--- a/src/common/src/types/decimal.rs
+++ b/src/common/src/types/decimal.rs
@@ -13,14 +13,15 @@
// limitations under the License.
use std::fmt::Debug;
-use std::io::{Read, Write};
+use std::io::{Cursor, Read, Write};
use std::ops::{Add, Div, Mul, Neg, Rem, Sub};
+use byteorder::{BigEndian, ReadBytesExt};
use bytes::{BufMut, Bytes, BytesMut};
use num_traits::{
CheckedAdd, CheckedDiv, CheckedMul, CheckedNeg, CheckedRem, CheckedSub, Num, One, Zero,
};
-use postgres_types::{accepts, to_sql_checked, IsNull, ToSql, Type};
+use postgres_types::{accepts, to_sql_checked, FromSql, IsNull, ToSql, Type};
use risingwave_common_estimate_size::ZeroHeapSize;
use rust_decimal::prelude::FromStr;
use rust_decimal::{Decimal as RustDecimal, Error, MathematicalOps as _, RoundingStrategy};
@@ -145,6 +146,28 @@ impl ToSql for Decimal {
}
}
+impl<'a> FromSql<'a> for Decimal {
+ fn from_sql(
+ ty: &Type,
+ raw: &'a [u8],
+ ) -> Result> {
+ let mut rdr = Cursor::new(raw);
+ let _n_digits = rdr.read_u16::()?;
+ let _weight = rdr.read_i16::()?;
+ let sign = rdr.read_u16::()?;
+ match sign {
+ 0xC000 => Ok(Self::NaN),
+ 0xD000 => Ok(Self::PositiveInf),
+ 0xF000 => Ok(Self::NegativeInf),
+ _ => RustDecimal::from_sql(ty, raw).map(Self::Normalized),
+ }
+ }
+
+ fn accepts(ty: &Type) -> bool {
+ matches!(*ty, Type::NUMERIC)
+ }
+}
+
macro_rules! impl_convert_int {
($T:ty) => {
impl core::convert::From<$T> for Decimal {
diff --git a/src/common/src/types/interval.rs b/src/common/src/types/interval.rs
index 4436181059d78..495561561e937 100644
--- a/src/common/src/types/interval.rs
+++ b/src/common/src/types/interval.rs
@@ -22,15 +22,12 @@ use std::sync::LazyLock;
use byteorder::{BigEndian, NetworkEndian, ReadBytesExt, WriteBytesExt};
use bytes::BytesMut;
-use chrono::Timelike;
use num_traits::{CheckedAdd, CheckedNeg, CheckedSub, Zero};
-use postgres_types::{to_sql_checked, FromSql};
+use postgres_types::to_sql_checked;
use regex::Regex;
-use risingwave_common_estimate_size::ZeroHeapSize;
use risingwave_pb::data::PbInterval;
use rust_decimal::prelude::Decimal;
-use super::to_binary::ToBinary;
use super::*;
/// Every interval can be represented by a `Interval`.
diff --git a/src/common/src/types/mod.rs b/src/common/src/types/mod.rs
index 9364f438cfacf..bb41e75c8a4f8 100644
--- a/src/common/src/types/mod.rs
+++ b/src/common/src/types/mod.rs
@@ -17,7 +17,6 @@
// NOTE: When adding or modifying data types, remember to update the type matrix in
// src/expr/macro/src/types.rs
-use std::convert::TryFrom;
use std::fmt::Debug;
use std::hash::Hash;
use std::str::FromStr;
diff --git a/src/common/src/types/num256.rs b/src/common/src/types/num256.rs
index 43c97fd8b155a..2acbcf636a037 100644
--- a/src/common/src/types/num256.rs
+++ b/src/common/src/types/num256.rs
@@ -335,7 +335,6 @@ impl EstimateSize for Int256 {
#[cfg(test)]
mod tests {
use super::*;
- use crate::types::F64;
macro_rules! check_op {
($t:ty, $lhs:expr, $rhs:expr, [$($op:tt),+]) => {
diff --git a/src/common/src/types/ordered.rs b/src/common/src/types/ordered.rs
index 77a0588e06672..bb2b0c9f316c1 100644
--- a/src/common/src/types/ordered.rs
+++ b/src/common/src/types/ordered.rs
@@ -14,7 +14,7 @@
//! `ScalarImpl` and `Datum` wrappers that implement `PartialOrd` and `Ord` with default order type.
-use std::cmp::{Ord, Ordering};
+use std::cmp::Ordering;
use std::ops::Deref;
use risingwave_common_estimate_size::EstimateSize;
diff --git a/src/common/src/types/scalar_impl.rs b/src/common/src/types/scalar_impl.rs
index db2732ca67c48..d222473fa6bbb 100644
--- a/src/common/src/types/scalar_impl.rs
+++ b/src/common/src/types/scalar_impl.rs
@@ -15,8 +15,6 @@
use std::hash::Hasher;
use super::*;
-use crate::array::list_array::{ListRef, ListValue};
-use crate::array::struct_array::{StructRef, StructValue};
use crate::{dispatch_scalar_ref_variants, dispatch_scalar_variants, for_all_native_types};
/// `ScalarPartialOrd` allows comparison between `Scalar` and `ScalarRef`.
diff --git a/src/common/src/util/memcmp_encoding.rs b/src/common/src/util/memcmp_encoding.rs
index 5709d42d226f8..7f262d0ff8d2c 100644
--- a/src/common/src/util/memcmp_encoding.rs
+++ b/src/common/src/util/memcmp_encoding.rs
@@ -333,15 +333,12 @@ pub fn decode_row(
mod tests {
use std::ops::Neg;
- use itertools::Itertools;
use rand::thread_rng;
use super::*;
- use crate::array::{DataChunk, ListValue, StructValue};
- use crate::row::{OwnedRow, RowExt};
- use crate::types::{DataType, FloatExt, ScalarImpl, F32};
- use crate::util::iter_util::ZipEqFast;
- use crate::util::sort_util::{ColumnOrder, OrderType};
+ use crate::array::{ListValue, StructValue};
+ use crate::row::RowExt;
+ use crate::types::FloatExt;
#[test]
fn test_memcomparable() {
diff --git a/src/common/src/util/panic.rs b/src/common/src/util/panic.rs
index 43db76df4f768..517a040e03b09 100644
--- a/src/common/src/util/panic.rs
+++ b/src/common/src/util/panic.rs
@@ -71,8 +71,8 @@ pub fn is_catching_unwind() -> bool {
}
#[cfg(all(test, not(madsim)))]
+#[expect(clippy::disallowed_methods)]
mod tests {
- #![allow(clippy::disallowed_methods)]
use rusty_fork::rusty_fork_test;
diff --git a/src/common/src/util/sort_util.rs b/src/common/src/util/sort_util.rs
index 015222136d382..ff4bde652496d 100644
--- a/src/common/src/util/sort_util.rs
+++ b/src/common/src/util/sort_util.rs
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-use std::cmp::{Ord, Ordering};
+use std::cmp::Ordering;
use std::fmt;
use std::sync::Arc;
@@ -574,13 +574,11 @@ pub fn cmp_rows(lhs: impl Row, rhs: impl Row, order_types: &[OrderType]) -> Orde
#[cfg(test)]
mod tests {
- use std::cmp::Ordering;
-
use itertools::Itertools;
use super::*;
- use crate::array::{DataChunk, ListValue, StructValue};
- use crate::row::{OwnedRow, Row};
+ use crate::array::{ListValue, StructValue};
+ use crate::row::OwnedRow;
use crate::types::{DataType, Datum, ScalarImpl};
#[test]
diff --git a/src/common/src/util/value_encoding/column_aware_row_encoding.rs b/src/common/src/util/value_encoding/column_aware_row_encoding.rs
index 28fcb65c4d8db..1bfbc0c926410 100644
--- a/src/common/src/util/value_encoding/column_aware_row_encoding.rs
+++ b/src/common/src/util/value_encoding/column_aware_row_encoding.rs
@@ -26,7 +26,6 @@ use bitflags::bitflags;
use super::*;
use crate::catalog::ColumnId;
-use crate::row::Row;
// deprecated design of have a Width to represent number of datum
// may be considered should `ColumnId` representation be optimized
diff --git a/src/common/src/util/value_encoding/mod.rs b/src/common/src/util/value_encoding/mod.rs
index 57e21c3782aa0..aa80c15617048 100644
--- a/src/common/src/util/value_encoding/mod.rs
+++ b/src/common/src/util/value_encoding/mod.rs
@@ -20,7 +20,7 @@ use either::{for_both, Either};
use enum_as_inner::EnumAsInner;
use risingwave_pb::data::PbDatum;
-use crate::array::{ArrayImpl, ListRef, ListValue, StructRef, StructValue};
+use crate::array::ArrayImpl;
use crate::row::{Row, RowDeserializer as BasicDeserializer};
use crate::types::*;
diff --git a/src/compute/src/lib.rs b/src/compute/src/lib.rs
index 344e0c781eb5e..87f052470ea7d 100644
--- a/src/compute/src/lib.rs
+++ b/src/compute/src/lib.rs
@@ -76,10 +76,6 @@ pub struct ComputeNodeOpts {
#[clap(long, env = "RW_META_ADDR", default_value = "http://127.0.0.1:5690")]
pub meta_address: MetaAddressStrategy,
- /// Payload format of connector sink rpc
- #[clap(long, env = "RW_CONNECTOR_RPC_SINK_PAYLOAD_FORMAT")]
- pub connector_rpc_sink_payload_format: Option,
-
/// The path of `risingwave.toml` configuration file.
///
/// If empty, default configuration values will be used.
@@ -91,7 +87,7 @@ pub struct ComputeNodeOpts {
pub total_memory_bytes: usize,
/// Reserved memory for the compute node in bytes.
- /// If not set, a portion (default to 30%) for the total_memory_bytes will be used as the reserved memory.
+ /// If not set, a portion (default to 30%) for the `total_memory_bytes` will be used as the reserved memory.
///
/// The total memory compute and storage can use is `total_memory_bytes` - `reserved_memory_bytes`.
#[clap(long, env = "RW_RESERVED_MEMORY_BYTES")]
diff --git a/src/compute/src/memory/config.rs b/src/compute/src/memory/config.rs
index dffc950429a91..29afbf6feba82 100644
--- a/src/compute/src/memory/config.rs
+++ b/src/compute/src/memory/config.rs
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-use foyer::memory::{LfuConfig, LruConfig, S3FifoConfig};
+use foyer::{LfuConfig, LruConfig, S3FifoConfig};
use risingwave_common::config::{
CacheEvictionConfig, EvictionConfig, StorageConfig, StorageMemoryConfig,
MAX_BLOCK_CACHE_SHARD_BITS, MAX_META_CACHE_SHARD_BITS, MIN_BUFFER_SIZE_PER_SHARD,
@@ -211,12 +211,22 @@ pub fn storage_memory_config(
}),
CacheEvictionConfig::S3Fifo {
small_queue_capacity_ratio_in_percent,
+ ghost_queue_capacity_ratio_in_percent,
+ small_to_main_freq_threshold,
} => EvictionConfig::S3Fifo(S3FifoConfig {
small_queue_capacity_ratio: small_queue_capacity_ratio_in_percent.unwrap_or(
risingwave_common::config::default::storage::small_queue_capacity_ratio_in_percent(
),
) as f64
/ 100.0,
+ ghost_queue_capacity_ratio: ghost_queue_capacity_ratio_in_percent.unwrap_or(
+ risingwave_common::config::default::storage::ghost_queue_capacity_ratio_in_percent(
+ ),
+ ) as f64
+ / 100.0,
+ small_to_main_freq_threshold: small_to_main_freq_threshold.unwrap_or(
+ risingwave_common::config::default::storage::small_to_main_freq_threshold(),
+ ),
}),
};
diff --git a/src/compute/src/rpc/service/config_service.rs b/src/compute/src/rpc/service/config_service.rs
index 97793f71722c9..e827ef176413e 100644
--- a/src/compute/src/rpc/service/config_service.rs
+++ b/src/compute/src/rpc/service/config_service.rs
@@ -18,7 +18,6 @@ use risingwave_common::error::tonic::ToTonicStatus;
use risingwave_pb::compute::config_service_server::ConfigService;
use risingwave_pb::compute::{ShowConfigRequest, ShowConfigResponse};
use risingwave_stream::task::LocalStreamManager;
-use serde_json;
use tonic::{Code, Request, Response, Status};
pub struct ConfigServiceImpl {
diff --git a/src/compute/src/server.rs b/src/compute/src/server.rs
index 07dee5dfa4c8c..cb459c1368a8d 100644
--- a/src/compute/src/server.rs
+++ b/src/compute/src/server.rs
@@ -13,7 +13,6 @@
// limitations under the License.
use std::net::SocketAddr;
-use std::sync::atomic::AtomicU32;
use std::sync::Arc;
use std::time::Duration;
@@ -42,7 +41,6 @@ use risingwave_connector::source::monitor::GLOBAL_SOURCE_METRICS;
use risingwave_dml::dml_manager::DmlManager;
use risingwave_pb::common::WorkerType;
use risingwave_pb::compute::config_service_server::ConfigServiceServer;
-use risingwave_pb::connector_service::SinkPayloadFormat;
use risingwave_pb::health::health_server::HealthServer;
use risingwave_pb::meta::add_worker_node_request::Property;
use risingwave_pb::monitor_service::monitor_service_server::MonitorServiceServer;
@@ -217,12 +215,6 @@ pub async fn compute_node_serve(
));
let compaction_executor = Arc::new(CompactionExecutor::new(Some(1)));
- let max_task_parallelism = Arc::new(AtomicU32::new(
- (compaction_executor.worker_num() as f32
- * storage_opts.compactor_max_task_multiplier)
- .ceil() as u32,
- ));
-
let compactor_context = CompactorContext {
storage_opts,
sstable_store: storage.sstable_store(),
@@ -235,8 +227,6 @@ pub async fn compute_node_serve(
await_tree_reg: await_tree_config
.clone()
.map(new_compaction_await_tree_reg_ref),
- running_task_parallelism: Arc::new(AtomicU32::new(0)),
- max_task_parallelism,
};
let (handle, shutdown_sender) = start_compactor(
@@ -334,28 +324,9 @@ pub async fn compute_node_serve(
config.server.metrics_level,
);
- info!(
- "connector param: payload_format={:?}",
- opts.connector_rpc_sink_payload_format
- );
-
- let connector_params = risingwave_connector::ConnectorParams {
- sink_payload_format: match opts.connector_rpc_sink_payload_format.as_deref() {
- None | Some("stream_chunk") => SinkPayloadFormat::StreamChunk,
- Some("json") => SinkPayloadFormat::Json,
- _ => {
- unreachable!(
- "invalid sink payload format: {:?}. Should be either json or stream_chunk",
- opts.connector_rpc_sink_payload_format
- )
- }
- },
- };
-
// Initialize the streaming environment.
let stream_env = StreamEnvironment::new(
advertise_addr.clone(),
- connector_params,
stream_config,
worker_id,
state_store,
diff --git a/src/compute/src/telemetry.rs b/src/compute/src/telemetry.rs
index 143ec51c5bc39..c35a6063ae267 100644
--- a/src/compute/src/telemetry.rs
+++ b/src/compute/src/telemetry.rs
@@ -33,6 +33,7 @@ impl ComputeTelemetryCreator {
#[async_trait::async_trait]
impl TelemetryReportCreator for ComputeTelemetryCreator {
+ #[allow(refining_impl_trait)]
async fn create_report(
&self,
tracking_id: String,
diff --git a/src/config/docs.md b/src/config/docs.md
index aa5cab015c7ee..d8f551f96fdf4 100644
--- a/src/config/docs.md
+++ b/src/config/docs.md
@@ -155,7 +155,7 @@ This page is automatically generated by `./risedev generate-example-config`
| data_directory | Remote directory for storing data and metadata objects. | |
| enable_tracing | Whether to enable distributed tracing. | false |
| max_concurrent_creating_streaming_jobs | Max number of concurrent creating streaming jobs. | 1 |
-| parallel_compact_size_mb | | 512 |
+| parallel_compact_size_mb | The size of parallel task for one compact/flush job. | 512 |
| pause_on_next_bootstrap | Whether to pause all data sources on next bootstrap. | false |
| sstable_size_mb | Target size of the Sstable. | 256 |
-| state_store | | |
+| state_store | URL for the state store | |
diff --git a/src/connector/Cargo.toml b/src/connector/Cargo.toml
index 8e0a16c95f8bb..c8953e1c46d84 100644
--- a/src/connector/Cargo.toml
+++ b/src/connector/Cargo.toml
@@ -33,6 +33,7 @@ auto_impl = "1"
await-tree = { workspace = true }
aws-config = { workspace = true }
aws-credential-types = { workspace = true }
+aws-msk-iam-sasl-signer = "1.0.0"
aws-sdk-kinesis = { workspace = true }
aws-sdk-s3 = { workspace = true }
aws-smithy-http = { workspace = true }
@@ -43,6 +44,7 @@ aws-types = { workspace = true }
base64 = "0.22"
byteorder = "1"
bytes = { version = "1", features = ["serde"] }
+cfg-or-panic = "0.2"
chrono = { version = "0.4", default-features = false, features = [
"clock",
"std",
diff --git a/src/connector/src/connector_common/common.rs b/src/connector/src/connector_common/common.rs
index 302b68dd664a1..e5aa7529e3e97 100644
--- a/src/connector/src/connector_common/common.rs
+++ b/src/connector/src/connector_common/common.rs
@@ -44,6 +44,8 @@ use crate::source::nats::source::NatsOffset;
pub const PRIVATE_LINK_BROKER_REWRITE_MAP_KEY: &str = "broker.rewrite.endpoints";
pub const PRIVATE_LINK_TARGETS_KEY: &str = "privatelink.targets";
+const AWS_MSK_IAM_AUTH: &str = "AWS_MSK_IAM";
+
#[derive(Debug, Clone, Deserialize)]
pub struct AwsPrivateLinkItem {
pub az_id: Option,
@@ -59,16 +61,28 @@ use aws_types::SdkConfig;
/// A flatten config map for aws auth.
#[derive(Deserialize, Debug, Clone, WithOptions)]
pub struct AwsAuthProps {
+ #[serde(rename = "aws.region", alias = "region")]
pub region: Option,
- #[serde(alias = "endpoint_url")]
+
+ #[serde(
+ rename = "aws.endpoint_url",
+ alias = "endpoint_url",
+ alias = "endpoint"
+ )]
pub endpoint: Option,
+ #[serde(rename = "aws.credentials.access_key_id", alias = "access_key")]
pub access_key: Option,
+ #[serde(rename = "aws.credentials.secret_access_key", alias = "secret_key")]
pub secret_key: Option,
+ #[serde(rename = "aws.credentials.session_token", alias = "session_token")]
pub session_token: Option,
+ /// IAM role
+ #[serde(rename = "aws.credentials.role.arn", alias = "arn")]
pub arn: Option,
- /// This field was added for kinesis. Not sure if it's useful for other connectors.
- /// Please ignore it in the documentation for now.
+ /// external ID in IAM role trust policy
+ #[serde(rename = "aws.credentials.role.external_id", alias = "external_id")]
pub external_id: Option,
+ #[serde(rename = "aws.profile", alias = "profile")]
pub profile: Option,
}
@@ -181,7 +195,7 @@ pub struct KafkaCommon {
#[serde(rename = "properties.ssl.key.password")]
ssl_key_password: Option,
- /// SASL mechanism if SASL is enabled. Currently support PLAIN, SCRAM and GSSAPI.
+ /// SASL mechanism if SASL is enabled. Currently support PLAIN, SCRAM, GSSAPI, and AWS_MSK_IAM.
#[serde(rename = "properties.sasl.mechanism")]
sasl_mechanism: Option,
@@ -286,6 +300,13 @@ impl RdKafkaPropertiesCommon {
impl KafkaCommon {
pub(crate) fn set_security_properties(&self, config: &mut ClientConfig) {
+ // AWS_MSK_IAM
+ if self.is_aws_msk_iam() {
+ config.set("security.protocol", "SASL_SSL");
+ config.set("sasl.mechanism", "OAUTHBEARER");
+ return;
+ }
+
// Security protocol
if let Some(security_protocol) = self.security_protocol.as_ref() {
config.set("security.protocol", security_protocol);
@@ -356,6 +377,16 @@ impl KafkaCommon {
// Currently, we only support unsecured OAUTH.
config.set("enable.sasl.oauthbearer.unsecure.jwt", "true");
}
+
+ pub(crate) fn is_aws_msk_iam(&self) -> bool {
+ if let Some(sasl_mechanism) = self.sasl_mechanism.as_ref()
+ && sasl_mechanism == AWS_MSK_IAM_AUTH
+ {
+ true
+ } else {
+ false
+ }
+ }
}
#[derive(Clone, Debug, Deserialize, WithOptions)]
diff --git a/src/connector/src/lib.rs b/src/connector/src/lib.rs
index 70de9f8561a76..c866a68b298d6 100644
--- a/src/connector/src/lib.rs
+++ b/src/connector/src/lib.rs
@@ -41,7 +41,6 @@
use std::time::Duration;
use duration_str::parse_std;
-use risingwave_pb::connector_service::SinkPayloadFormat;
use serde::de;
pub mod aws_utils;
@@ -64,19 +63,6 @@ pub use with_options::WithPropertiesExt;
#[cfg(test)]
mod with_options_test;
-#[derive(Clone, Debug, Default)]
-pub struct ConnectorParams {
- pub sink_payload_format: SinkPayloadFormat,
-}
-
-impl ConnectorParams {
- pub fn new(sink_payload_format: SinkPayloadFormat) -> Self {
- Self {
- sink_payload_format,
- }
- }
-}
-
pub(crate) fn deserialize_u32_from_string<'de, D>(deserializer: D) -> Result
where
D: de::Deserializer<'de>,
diff --git a/src/connector/src/parser/avro/parser.rs b/src/connector/src/parser/avro/parser.rs
index 65601d9c3a3e6..c1950313b5e5c 100644
--- a/src/connector/src/parser/avro/parser.rs
+++ b/src/connector/src/parser/avro/parser.rs
@@ -182,8 +182,8 @@ mod test {
use std::path::PathBuf;
use apache_avro::schema::RecordSchema;
- use apache_avro::types::{Record, Value};
- use apache_avro::{Codec, Days, Duration, Millis, Months, Reader, Schema, Writer};
+ use apache_avro::types::Record;
+ use apache_avro::{Codec, Days, Duration, Millis, Months, Writer};
use itertools::Itertools;
use risingwave_common::array::Op;
use risingwave_common::catalog::ColumnId;
@@ -195,12 +195,9 @@ mod test {
use super::*;
use crate::connector_common::AwsAuthProps;
- use crate::error::ConnectorResult;
use crate::parser::plain_parser::PlainParser;
use crate::parser::unified::avro::unix_epoch_days;
- use crate::parser::{
- AccessBuilderImpl, EncodingType, SourceStreamChunkBuilder, SpecificParserConfig,
- };
+ use crate::parser::{AccessBuilderImpl, SourceStreamChunkBuilder, SpecificParserConfig};
use crate::source::{SourceColumnDesc, SourceContext};
fn test_data_path(file_name: &str) -> String {
diff --git a/src/connector/src/parser/canal/simd_json_parser.rs b/src/connector/src/parser/canal/simd_json_parser.rs
index 9bc236392eb48..ac8a63a6457e7 100644
--- a/src/connector/src/parser/canal/simd_json_parser.rs
+++ b/src/connector/src/parser/canal/simd_json_parser.rs
@@ -141,7 +141,6 @@ mod tests {
use super::*;
use crate::parser::SourceStreamChunkBuilder;
- use crate::source::SourceColumnDesc;
#[tokio::test]
async fn test_data_types() {
diff --git a/src/connector/src/parser/debezium/avro_parser.rs b/src/connector/src/parser/debezium/avro_parser.rs
index 29d9139d221cf..3b11a1926f107 100644
--- a/src/connector/src/parser/debezium/avro_parser.rs
+++ b/src/connector/src/parser/debezium/avro_parser.rs
@@ -139,7 +139,6 @@ mod tests {
use std::io::Read;
use std::path::PathBuf;
- use apache_avro::Schema;
use itertools::Itertools;
use maplit::{convert_args, hashmap};
use risingwave_common::array::Op;
@@ -151,9 +150,7 @@ mod tests {
use risingwave_pb::plan_common::{PbEncodeType, PbFormatType};
use super::*;
- use crate::parser::{
- DebeziumAvroParserConfig, DebeziumParser, SourceStreamChunkBuilder, SpecificParserConfig,
- };
+ use crate::parser::{DebeziumParser, SourceStreamChunkBuilder, SpecificParserConfig};
use crate::source::{SourceColumnDesc, SourceContext};
const DEBEZIUM_AVRO_DATA: &[u8] = b"\x00\x00\x00\x00\x06\x00\x02\xd2\x0f\x0a\x53\x61\x6c\x6c\x79\x0c\x54\x68\x6f\x6d\x61\x73\x2a\x73\x61\x6c\x6c\x79\x2e\x74\x68\x6f\x6d\x61\x73\x40\x61\x63\x6d\x65\x2e\x63\x6f\x6d\x16\x32\x2e\x31\x2e\x32\x2e\x46\x69\x6e\x61\x6c\x0a\x6d\x79\x73\x71\x6c\x12\x64\x62\x73\x65\x72\x76\x65\x72\x31\xc0\xb4\xe8\xb7\xc9\x61\x00\x30\x66\x69\x72\x73\x74\x5f\x69\x6e\x5f\x64\x61\x74\x61\x5f\x63\x6f\x6c\x6c\x65\x63\x74\x69\x6f\x6e\x12\x69\x6e\x76\x65\x6e\x74\x6f\x72\x79\x00\x02\x12\x63\x75\x73\x74\x6f\x6d\x65\x72\x73\x00\x00\x20\x6d\x79\x73\x71\x6c\x2d\x62\x69\x6e\x2e\x30\x30\x30\x30\x30\x33\x8c\x06\x00\x00\x00\x02\x72\x02\x92\xc3\xe8\xb7\xc9\x61\x00";
diff --git a/src/connector/src/parser/debezium/simd_json_parser.rs b/src/connector/src/parser/debezium/simd_json_parser.rs
index 63cb939d81238..dff681687bc47 100644
--- a/src/connector/src/parser/debezium/simd_json_parser.rs
+++ b/src/connector/src/parser/debezium/simd_json_parser.rs
@@ -99,8 +99,6 @@ impl AccessBuilder for DebeziumMongoJsonAccessBuilder {
#[cfg(test)]
mod tests {
- use std::convert::TryInto;
-
use chrono::{NaiveDate, NaiveTime};
use risingwave_common::array::{Op, StructValue};
use risingwave_common::catalog::ColumnId;
diff --git a/src/connector/src/parser/mod.rs b/src/connector/src/parser/mod.rs
index afdd07716634e..08e8bd3dd46ea 100644
--- a/src/connector/src/parser/mod.rs
+++ b/src/connector/src/parser/mod.rs
@@ -43,7 +43,7 @@ use self::avro::AvroAccessBuilder;
use self::bytes_parser::BytesAccessBuilder;
pub use self::mysql::mysql_row_to_owned_row;
use self::plain_parser::PlainParser;
-pub use self::postgres::{postgres_row_to_owned_row, EnumString};
+pub use self::postgres::postgres_row_to_owned_row;
use self::simd_json_parser::DebeziumJsonAccessBuilder;
pub use self::unified::json::TimestamptzHandling;
use self::unified::AccessImpl;
@@ -75,7 +75,9 @@ mod maxwell;
mod mysql;
pub mod plain_parser;
mod postgres;
+
mod protobuf;
+pub mod scalar_adapter;
mod unified;
mod upsert_parser;
mod util;
diff --git a/src/connector/src/parser/postgres.rs b/src/connector/src/parser/postgres.rs
index 61d7e0b588875..f01484ad18fb2 100644
--- a/src/connector/src/parser/postgres.rs
+++ b/src/connector/src/parser/postgres.rs
@@ -12,23 +12,20 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-use std::str::FromStr;
use std::sync::LazyLock;
-use bytes::BytesMut;
use chrono::{NaiveDate, Utc};
-use pg_bigdecimal::PgNumeric;
use risingwave_common::catalog::Schema;
use risingwave_common::log::LogSuppresser;
use risingwave_common::row::OwnedRow;
use risingwave_common::types::{
- DataType, Date, Decimal, Int256, Interval, JsonbVal, ListValue, ScalarImpl, Time, Timestamp,
+ DataType, Date, Decimal, Interval, JsonbVal, ListValue, ScalarImpl, Time, Timestamp,
Timestamptz,
};
-use rust_decimal::Decimal as RustDecimal;
use thiserror_ext::AsReport;
-use tokio_postgres::types::{to_sql_checked, FromSql, IsNull, Kind, ToSql, Type};
+use tokio_postgres::types::{Kind, Type};
+use crate::parser::scalar_adapter::ScalarAdapter;
use crate::parser::util::log_error;
static LOG_SUPPERSSER: LazyLock = LazyLock::new(LogSuppresser::default);
@@ -93,284 +90,225 @@ pub fn postgres_row_to_owned_row(row: tokio_postgres::Row, schema: &Schema) -> O
for i in 0..schema.fields.len() {
let rw_field = &schema.fields[i];
let name = rw_field.name.as_str();
- let datum = {
- match &rw_field.data_type {
- DataType::Boolean => {
- handle_data_type!(row, i, name, bool)
- }
- DataType::Int16 => {
- handle_data_type!(row, i, name, i16)
- }
- DataType::Int32 => {
- handle_data_type!(row, i, name, i32)
- }
- DataType::Int64 => {
- handle_data_type!(row, i, name, i64)
- }
- DataType::Float32 => {
- handle_data_type!(row, i, name, f32)
- }
- DataType::Float64 => {
- handle_data_type!(row, i, name, f64)
- }
- DataType::Decimal => {
- handle_data_type!(row, i, name, RustDecimal, Decimal)
+ let datum = postgres_cell_to_scalar_impl(&row, &rw_field.data_type, i, name);
+ datums.push(datum);
+ }
+ OwnedRow::new(datums)
+}
+
+fn postgres_cell_to_scalar_impl(
+ row: &tokio_postgres::Row,
+ data_type: &DataType,
+ i: usize,
+ name: &str,
+) -> Option {
+ // We observe several incompatibility issue in Debezium's Postgres connector. We summarize them here:
+ // Issue #1. The null of enum list is not supported in Debezium. An enum list contains `NULL` will fallback to `NULL`.
+ // Issue #2. In our parser, when there's inf, -inf, nan or invalid item in a list, the whole list will fallback null.
+ match data_type {
+ DataType::Boolean => {
+ handle_data_type!(row, i, name, bool)
+ }
+ DataType::Int16 => {
+ handle_data_type!(row, i, name, i16)
+ }
+ DataType::Int32 => {
+ handle_data_type!(row, i, name, i32)
+ }
+ DataType::Int64 => {
+ handle_data_type!(row, i, name, i64)
+ }
+ DataType::Float32 => {
+ handle_data_type!(row, i, name, f32)
+ }
+ DataType::Float64 => {
+ handle_data_type!(row, i, name, f64)
+ }
+ DataType::Decimal => {
+ handle_data_type!(row, i, name, Decimal)
+ }
+ DataType::Int256 => {
+ // Currently in order to handle the decimal beyond RustDecimal,
+ // we use the PgNumeric type to convert the decimal to a string.
+ // Then we convert the string to Int256.
+ // Note: It's only used to map the numeric type in upstream Postgres to RisingWave's rw_int256.
+ let res = row.try_get::<_, Option>>(i);
+ match res {
+ Ok(val) => val.and_then(|v| v.into_scalar(DataType::Int256)),
+ Err(err) => {
+ log_error!(name, err, "parse numeric column as pg_numeric failed");
+ None
}
- DataType::Int256 => {
- // Currently in order to handle the decimal beyond RustDecimal,
- // we use the PgNumeric type to convert the decimal to a string.
- // Then we convert the string to Int256.
- // Note: It's only used to map the numeric type in upstream Postgres to RisingWave's rw_int256.
- let res = row.try_get::<_, Option>(i);
- match res {
- Ok(val) => pg_numeric_to_rw_int256(val, name),
- Err(err) => {
- log_error!(name, err, "parse numeric column as pg_numeric failed");
- None
- }
+ }
+ }
+ DataType::Varchar => {
+ if let Kind::Enum(_) = row.columns()[i].type_().kind() {
+ // enum type needs to be handled separately
+ let res = row.try_get::<_, Option>>(i);
+ match res {
+ Ok(val) => val.and_then(|v| v.into_scalar(DataType::Varchar)),
+ Err(err) => {
+ log_error!(name, err, "parse enum column failed");
+ None
}
}
- DataType::Varchar => {
- if let Kind::Enum(_) = row.columns()[i].type_().kind() {
- // enum type needs to be handled separately
- let res = row.try_get::<_, Option>(i);
+ } else {
+ match *row.columns()[i].type_() {
+ // Since we don't support UUID natively, adapt it to a VARCHAR column
+ Type::UUID => {
+ let res = row.try_get::<_, Option>>(i);
match res {
- Ok(val) => val.map(|v| ScalarImpl::from(v.0)),
+ Ok(val) => val.and_then(|v| v.into_scalar(DataType::Varchar)),
Err(err) => {
- log_error!(name, err, "parse enum column failed");
+ log_error!(name, err, "parse uuid column failed");
None
}
}
- } else {
- match *row.columns()[i].type_() {
- // Since we don't support UUID natively, adapt it to a VARCHAR column
- Type::UUID => {
- let res = row.try_get::<_, Option>(i);
- match res {
- Ok(val) => val.map(|v| ScalarImpl::from(v.to_string())),
- Err(err) => {
- log_error!(name, err, "parse uuid column failed");
- None
- }
- }
- }
- // we support converting NUMERIC to VARCHAR implicitly
- Type::NUMERIC => {
- // Currently in order to handle the decimal beyond RustDecimal,
- // we use the PgNumeric type to convert the decimal to a string.
- // Note: It's only used to map the numeric type in upstream Postgres to RisingWave's varchar.
- let res = row.try_get::<_, Option>(i);
- match res {
- Ok(val) => pg_numeric_to_varchar(val),
- Err(err) => {
- log_error!(
- name,
- err,
- "parse numeric column as pg_numeric failed"
- );
- None
- }
- }
- }
- _ => {
- handle_data_type!(row, i, name, String)
+ }
+ // we support converting NUMERIC to VARCHAR implicitly
+ Type::NUMERIC => {
+ // Currently in order to handle the decimal beyond RustDecimal,
+ // we use the PgNumeric type to convert the decimal to a string.
+ // Note: It's only used to map the numeric type in upstream Postgres to RisingWave's varchar.
+ let res = row.try_get::<_, Option>>(i);
+ match res {
+ Ok(val) => val.and_then(|v| v.into_scalar(DataType::Varchar)),
+ Err(err) => {
+ log_error!(name, err, "parse numeric column as pg_numeric failed");
+ None
}
}
}
+ _ => {
+ handle_data_type!(row, i, name, String)
+ }
}
- DataType::Date => {
- handle_data_type!(row, i, name, NaiveDate, Date)
- }
- DataType::Time => {
- handle_data_type!(row, i, name, chrono::NaiveTime, Time)
- }
- DataType::Timestamp => {
- handle_data_type!(row, i, name, chrono::NaiveDateTime, Timestamp)
- }
- DataType::Timestamptz => {
- handle_data_type!(row, i, name, chrono::DateTime, Timestamptz)
+ }
+ }
+ DataType::Date => {
+ handle_data_type!(row, i, name, NaiveDate, Date)
+ }
+ DataType::Time => {
+ handle_data_type!(row, i, name, chrono::NaiveTime, Time)
+ }
+ DataType::Timestamp => {
+ handle_data_type!(row, i, name, chrono::NaiveDateTime, Timestamp)
+ }
+ DataType::Timestamptz => {
+ handle_data_type!(row, i, name, chrono::DateTime, Timestamptz)
+ }
+ DataType::Bytea => {
+ let res = row.try_get::<_, Option>>(i);
+ match res {
+ Ok(val) => val.map(|v| ScalarImpl::from(v.into_boxed_slice())),
+ Err(err) => {
+ log_error!(name, err, "parse column failed");
+ None
}
- DataType::Bytea => {
- let res = row.try_get::<_, Option>>(i);
- match res {
- Ok(val) => val.map(|v| ScalarImpl::from(v.into_boxed_slice())),
- Err(err) => {
- log_error!(name, err, "parse column failed");
- None
+ }
+ }
+ DataType::Jsonb => {
+ handle_data_type!(row, i, name, serde_json::Value, JsonbVal)
+ }
+ DataType::Interval => {
+ handle_data_type!(row, i, name, Interval)
+ }
+ DataType::List(dtype) => {
+ let mut builder = dtype.create_array_builder(0);
+ // enum list needs to be handled separately
+ if let Kind::Array(item_type) = row.columns()[i].type_().kind()
+ && let Kind::Enum(_) = item_type.kind()
+ {
+ // Issue #1, we use ScalarAdapter instead of Option
+ let res = row.try_get::<_, Option>>>(i);
+ match res {
+ Ok(val) => {
+ if let Some(vec) = val {
+ for val in vec {
+ builder.append(val.into_scalar(DataType::Varchar))
+ }
}
+ Some(ScalarImpl::from(ListValue::new(builder.finish())))
+ }
+ Err(err) => {
+ log_error!(name, err, "parse enum column failed");
+ None
}
}
- DataType::Jsonb => {
- handle_data_type!(row, i, name, serde_json::Value, JsonbVal)
- }
- DataType::Interval => {
- handle_data_type!(row, i, name, Interval)
- }
- DataType::List(dtype) => {
- let mut builder = dtype.create_array_builder(0);
- // enum list needs to be handled separately
- if let Kind::Array(item_type) = row.columns()[i].type_().kind()
- && let Kind::Enum(_) = item_type.kind()
- {
- // FIXME(Kexiang): The null of enum list is not supported in Debezium.
- // As `NULL` in enum list is not supported in Debezium, we use `EnumString`
- // instead of `Option` to handle enum to keep the behaviors aligned.
- // An enum list contains `NULL` will be converted to `NULL`.
- let res = row.try_get::<_, Option>>(i);
+ } else {
+ match **dtype {
+ DataType::Boolean => {
+ handle_list_data_type!(row, i, name, bool, builder);
+ }
+ DataType::Int16 => {
+ handle_list_data_type!(row, i, name, i16, builder);
+ }
+ DataType::Int32 => {
+ handle_list_data_type!(row, i, name, i32, builder);
+ }
+ DataType::Int64 => {
+ handle_list_data_type!(row, i, name, i64, builder);
+ }
+ DataType::Float32 => {
+ handle_list_data_type!(row, i, name, f32, builder);
+ }
+ DataType::Float64 => {
+ handle_list_data_type!(row, i, name, f64, builder);
+ }
+ DataType::Decimal => {
+ let res = row.try_get::<_, Option>>>>(i);
match res {
Ok(val) => {
- if let Some(v) = val {
- v.into_iter().for_each(|val| {
- builder.append(Some(ScalarImpl::from(val.0)));
- });
+ if let Some(vec) = val {
+ builder = ScalarAdapter::build_scalar_in_list(
+ vec,
+ DataType::Decimal,
+ builder,
+ )?;
}
- Some(ScalarImpl::from(ListValue::new(builder.finish())))
}
Err(err) => {
- log_error!(name, err, "parse enum column failed");
- None
- }
- }
- } else {
- match **dtype {
- DataType::Boolean => {
- handle_list_data_type!(row, i, name, bool, builder);
- }
- DataType::Int16 => {
- handle_list_data_type!(row, i, name, i16, builder);
- }
- DataType::Int32 => {
- handle_list_data_type!(row, i, name, i32, builder);
- }
- DataType::Int64 => {
- handle_list_data_type!(row, i, name, i64, builder);
- }
- DataType::Float32 => {
- handle_list_data_type!(row, i, name, f32, builder);
- }
- DataType::Float64 => {
- handle_list_data_type!(row, i, name, f64, builder);
- }
- DataType::Decimal => {
- handle_list_data_type!(row, i, name, RustDecimal, builder, Decimal);
- }
- DataType::Date => {
- handle_list_data_type!(row, i, name, NaiveDate, builder, Date);
- }
- DataType::Varchar => {
- match *row.columns()[i].type_() {
- // Since we don't support UUID natively, adapt it to a VARCHAR column
- Type::UUID_ARRAY => {
- let res =
- row.try_get::<_, Option>>>(i);
- match res {
- Ok(val) => {
- if let Some(v) = val {
- v.into_iter().for_each(|val| {
- builder.append(val.map(|v| {
- ScalarImpl::from(v.to_string())
- }))
- });
- }
- }
- Err(err) => {
- log_error!(name, err, "parse uuid column failed");
- }
- };
- }
- Type::NUMERIC_ARRAY => {
- let res =
- row.try_get::<_, Option>>>(i);
- match res {
- Ok(val) => {
- if let Some(v) = val {
- v.into_iter().for_each(|val| {
- builder.append(pg_numeric_to_varchar(val))
- });
- }
- }
- Err(err) => {
- log_error!(
- name,
- err,
- "parse numeric list column as pg_numeric list failed"
- );
- }
- };
- }
- _ => {
- handle_list_data_type!(row, i, name, String, builder);
- }
- }
- }
- DataType::Time => {
- handle_list_data_type!(
- row,
- i,
- name,
- chrono::NaiveTime,
- builder,
- Time
- );
- }
- DataType::Timestamp => {
- handle_list_data_type!(
- row,
- i,
- name,
- chrono::NaiveDateTime,
- builder,
- Timestamp
- );
- }
- DataType::Timestamptz => {
- handle_list_data_type!(
- row,
- i,
- name,
- chrono::DateTime,
- builder,
- Timestamptz
- );
- }
- DataType::Interval => {
- handle_list_data_type!(row, i, name, Interval, builder);
+ log_error!(name, err, "parse uuid column failed");
}
- DataType::Jsonb => {
- handle_list_data_type!(
- row,
- i,
- name,
- serde_json::Value,
- builder,
- JsonbVal
- );
- }
- DataType::Bytea => {
- let res = row.try_get::<_, Option>>>>(i);
+ };
+ }
+ DataType::Date => {
+ handle_list_data_type!(row, i, name, NaiveDate, builder, Date);
+ }
+ DataType::Varchar => {
+ match *row.columns()[i].type_() {
+ // Since we don't support UUID natively, adapt it to a VARCHAR column
+ Type::UUID_ARRAY => {
+ let res =
+ row.try_get::<_, Option>>>>(i);
match res {
Ok(val) => {
- if let Some(v) = val {
- v.into_iter().for_each(|val| {
- builder.append(val.map(|v| {
- ScalarImpl::from(v.into_boxed_slice())
- }))
- })
+ if let Some(vec) = val {
+ for val in vec {
+ builder.append(
+ val.and_then(|v| {
+ v.into_scalar(DataType::Varchar)
+ }),
+ )
+ }
}
}
Err(err) => {
- log_error!(name, err, "parse column failed");
+ log_error!(name, err, "parse uuid column failed");
}
- }
+ };
}
- DataType::Int256 => {
- let res = row.try_get::<_, Option>>>(i);
+ Type::NUMERIC_ARRAY => {
+ let res =
+ row.try_get::<_, Option>>>>(i);
match res {
Ok(val) => {
- if let Some(v) = val {
- v.into_iter().for_each(|val| {
- builder.append(pg_numeric_to_rw_int256(val, name))
- });
+ if let Some(vec) = val {
+ builder = ScalarAdapter::build_scalar_in_list(
+ vec,
+ DataType::Varchar,
+ builder,
+ )?;
}
}
Err(err) => {
@@ -382,104 +320,92 @@ pub fn postgres_row_to_owned_row(row: tokio_postgres::Row, schema: &Schema) -> O
}
};
}
- DataType::Struct(_) | DataType::List(_) | DataType::Serial => {
- tracing::warn!(
- "unsupported List data type {:?}, set the List to empty",
- **dtype
+ _ => {
+ handle_list_data_type!(row, i, name, String, builder);
+ }
+ }
+ }
+ DataType::Time => {
+ handle_list_data_type!(row, i, name, chrono::NaiveTime, builder, Time);
+ }
+ DataType::Timestamp => {
+ handle_list_data_type!(
+ row,
+ i,
+ name,
+ chrono::NaiveDateTime,
+ builder,
+ Timestamp
+ );
+ }
+ DataType::Timestamptz => {
+ handle_list_data_type!(
+ row,
+ i,
+ name,
+ chrono::DateTime,
+ builder,
+ Timestamptz
+ );
+ }
+ DataType::Interval => {
+ handle_list_data_type!(row, i, name, Interval, builder);
+ }
+ DataType::Jsonb => {
+ handle_list_data_type!(row, i, name, serde_json::Value, builder, JsonbVal);
+ }
+ DataType::Bytea => {
+ let res = row.try_get::<_, Option>>>>(i);
+ match res {
+ Ok(val) => {
+ if let Some(v) = val {
+ v.into_iter().for_each(|val| {
+ builder.append(
+ val.map(|v| ScalarImpl::from(v.into_boxed_slice())),
+ )
+ })
+ }
+ }
+ Err(err) => {
+ log_error!(name, err, "parse column failed");
+ }
+ }
+ }
+ DataType::Int256 => {
+ let res = row.try_get::<_, Option>>>>(i);
+ match res {
+ Ok(val) => {
+ if let Some(vec) = val {
+ builder = ScalarAdapter::build_scalar_in_list(
+ vec,
+ DataType::Int256,
+ builder,
+ )?;
+ }
+ }
+ Err(err) => {
+ log_error!(
+ name,
+ err,
+ "parse numeric list column as pg_numeric list failed"
);
}
};
- Some(ScalarImpl::from(ListValue::new(builder.finish())))
}
- }
- DataType::Struct(_) | DataType::Serial => {
- // Interval, Struct, List are not supported
- tracing::warn!(rw_field.name, ?rw_field.data_type, "unsupported data type, set to null");
- None
- }
+ DataType::Struct(_) | DataType::List(_) | DataType::Serial => {
+ tracing::warn!(
+ "unsupported List data type {:?}, set the List to empty",
+ **dtype
+ );
+ }
+ };
+ Some(ScalarImpl::from(ListValue::new(builder.finish())))
}
- };
- datums.push(datum);
- }
- OwnedRow::new(datums)
-}
-
-fn pg_numeric_to_rw_int256(val: Option, name: &str) -> Option {
- let string = pg_numeric_to_string(val)?;
- match Int256::from_str(string.as_str()) {
- Ok(num) => Some(ScalarImpl::from(num)),
- Err(err) => {
- log_error!(name, err, "parse numeric string as rw_int256 failed");
- None
}
- }
-}
-
-fn pg_numeric_to_varchar(val: Option) -> Option {
- pg_numeric_to_string(val).map(ScalarImpl::from)
-}
-
-fn pg_numeric_to_string(val: Option) -> Option {
- if let Some(pg_numeric) = val {
- // TODO(kexiang): NEGATIVE_INFINITY -> -Infinity, POSITIVE_INFINITY -> Infinity, NAN -> NaN
- // The current implementation is to ensure consistency with the behavior of cdc event parsor.
- match pg_numeric {
- PgNumeric::NegativeInf => Some(String::from("NEGATIVE_INFINITY")),
- PgNumeric::Normalized(big_decimal) => Some(big_decimal.to_string()),
- PgNumeric::PositiveInf => Some(String::from("POSITIVE_INFINITY")),
- PgNumeric::NaN => Some(String::from("NAN")),
- }
- } else {
- // NULL
- None
- }
-}
-
-#[derive(Clone, Debug)]
-pub struct EnumString(pub String);
-
-impl<'a> FromSql<'a> for EnumString {
- fn from_sql(
- _ty: &Type,
- raw: &'a [u8],
- ) -> Result> {
- Ok(EnumString(String::from_utf8_lossy(raw).into_owned()))
- }
-
- fn accepts(ty: &Type) -> bool {
- matches!(ty.kind(), Kind::Enum(_))
- }
-}
-
-impl ToSql for EnumString {
- to_sql_checked!();
-
- fn accepts(ty: &Type) -> bool {
- matches!(ty.kind(), Kind::Enum(_))
- }
-
- fn to_sql(
- &self,
- ty: &Type,
- out: &mut BytesMut,
- ) -> Result>
- where
- Self: Sized,
- {
- match ty.kind() {
- Kind::Enum(e) => {
- if e.contains(&self.0) {
- out.extend_from_slice(self.0.as_bytes());
- Ok(IsNull::No)
- } else {
- Err(format!(
- "EnumString value {} is not in the enum type {:?}",
- self.0, e
- )
- .into())
- }
- }
- _ => Err("EnumString can only be used with ENUM types".into()),
+ DataType::Struct(_) | DataType::Serial => {
+ // Interval and Struct are not supported
+ tracing::warn!(name, ?data_type, "unsupported data type, set to null");
+ None
}
}
}
@@ -488,7 +414,7 @@ impl ToSql for EnumString {
mod tests {
use tokio_postgres::NoTls;
- use crate::parser::postgres::EnumString;
+ use crate::parser::scalar_adapter::EnumString;
const DB: &str = "postgres";
const USER: &str = "kexiang";
diff --git a/src/connector/src/parser/protobuf/parser.rs b/src/connector/src/parser/protobuf/parser.rs
index e9ae317ca5ebc..7dcb502b1f674 100644
--- a/src/connector/src/parser/protobuf/parser.rs
+++ b/src/connector/src/parser/protobuf/parser.rs
@@ -583,7 +583,7 @@ mod test {
use std::path::PathBuf;
use prost::Message;
- use risingwave_common::types::{DataType, ListValue, StructType};
+ use risingwave_common::types::StructType;
use risingwave_pb::catalog::StreamSourceInfo;
use risingwave_pb::data::data_type::PbTypeName;
use risingwave_pb::plan_common::{PbEncodeType, PbFormatType};
diff --git a/src/connector/src/parser/scalar_adapter.rs b/src/connector/src/parser/scalar_adapter.rs
new file mode 100644
index 0000000000000..6a6546a8d7600
--- /dev/null
+++ b/src/connector/src/parser/scalar_adapter.rs
@@ -0,0 +1,272 @@
+// Copyright 2024 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::str::FromStr;
+
+use anyhow::anyhow;
+use bytes::BytesMut;
+use pg_bigdecimal::PgNumeric;
+use risingwave_common::types::{DataType, Decimal, Int256, ScalarImpl, ScalarRefImpl};
+use thiserror_ext::AsReport;
+use tokio_postgres::types::{to_sql_checked, FromSql, IsNull, Kind, ToSql, Type};
+
+use crate::error::ConnectorResult;
+
+#[derive(Clone, Debug)]
+pub struct EnumString(pub String);
+
+impl<'a> FromSql<'a> for EnumString {
+ fn from_sql(
+ _ty: &Type,
+ raw: &'a [u8],
+ ) -> Result> {
+ Ok(EnumString(String::from_utf8_lossy(raw).into_owned()))
+ }
+
+ fn accepts(ty: &Type) -> bool {
+ matches!(ty.kind(), Kind::Enum(_))
+ }
+}
+
+impl ToSql for EnumString {
+ to_sql_checked!();
+
+ fn to_sql(
+ &self,
+ ty: &Type,
+ out: &mut BytesMut,
+ ) -> Result>
+ where
+ Self: Sized,
+ {
+ match ty.kind() {
+ Kind::Enum(e) => {
+ if e.contains(&self.0) {
+ out.extend_from_slice(self.0.as_bytes());
+ Ok(IsNull::No)
+ } else {
+ Err(format!(
+ "EnumString value {} is not in the enum type {:?}",
+ self.0, e
+ )
+ .into())
+ }
+ }
+ _ => Err("EnumString can only be used with ENUM types".into()),
+ }
+ }
+
+ fn accepts(ty: &Type) -> bool {
+ matches!(ty.kind(), Kind::Enum(_))
+ }
+}
+
+/// Adapter for `ScalarImpl` to Postgres data type,
+/// which can be used to encode/decode to/from Postgres value.
+#[derive(Debug)]
+pub(crate) enum ScalarAdapter<'a> {
+ Builtin(ScalarRefImpl<'a>),
+ Uuid(uuid::Uuid),
+ Numeric(PgNumeric),
+ Enum(EnumString),
+}
+
+impl ToSql for ScalarAdapter<'_> {
+ to_sql_checked!();
+
+ fn to_sql(
+ &self,
+ ty: &Type,
+ out: &mut bytes::BytesMut,
+ ) -> Result> {
+ match self {
+ ScalarAdapter::Builtin(v) => v.to_sql(ty, out),
+ ScalarAdapter::Uuid(v) => v.to_sql(ty, out),
+ ScalarAdapter::Numeric(v) => v.to_sql(ty, out),
+ ScalarAdapter::Enum(v) => v.to_sql(ty, out),
+ }
+ }
+
+ fn accepts(_ty: &Type) -> bool {
+ true
+ }
+}
+
+/// convert from Postgres uuid, numeric and enum to `ScalarAdapter`
+impl<'a> FromSql<'a> for ScalarAdapter<'_> {
+ fn from_sql(
+ ty: &Type,
+ raw: &'a [u8],
+ ) -> Result> {
+ match ty.kind() {
+ Kind::Simple => match *ty {
+ Type::UUID => {
+ let uuid = uuid::Uuid::from_sql(ty, raw)?;
+ Ok(ScalarAdapter::Uuid(uuid))
+ }
+ Type::NUMERIC => {
+ let numeric = PgNumeric::from_sql(ty, raw)?;
+ Ok(ScalarAdapter::Numeric(numeric))
+ }
+ _ => Err(anyhow!("failed to convert type {:?} to ScalarAdapter", ty).into()),
+ },
+ Kind::Enum(_) => Ok(ScalarAdapter::Enum(EnumString::from_sql(ty, raw)?)),
+ _ => Err(anyhow!("failed to convert type {:?} to ScalarAdapter", ty).into()),
+ }
+ }
+
+ fn accepts(ty: &Type) -> bool {
+ matches!(ty, &Type::UUID | &Type::NUMERIC) || ::accepts(ty)
+ }
+}
+
+impl ScalarAdapter<'_> {
+ pub fn name(&self) -> &'static str {
+ match self {
+ ScalarAdapter::Builtin(_) => "Builtin",
+ ScalarAdapter::Uuid(_) => "Uuid",
+ ScalarAdapter::Numeric(_) => "Numeric",
+ ScalarAdapter::Enum(_) => "Enum",
+ }
+ }
+
+ /// convert `ScalarRefImpl` to `ScalarAdapter` so that we can correctly encode to postgres value
+ pub(crate) fn from_scalar<'a>(
+ scalar: ScalarRefImpl<'a>,
+ ty: &Type,
+ ) -> ConnectorResult> {
+ Ok(match (scalar, ty, ty.kind()) {
+ (ScalarRefImpl::Utf8(s), &Type::UUID, _) => ScalarAdapter::Uuid(s.parse()?),
+ (ScalarRefImpl::Utf8(s), &Type::NUMERIC, _) => {
+ ScalarAdapter::Numeric(string_to_pg_numeric(s)?)
+ }
+ (ScalarRefImpl::Int256(s), &Type::NUMERIC, _) => {
+ ScalarAdapter::Numeric(string_to_pg_numeric(&s.to_string())?)
+ }
+ (ScalarRefImpl::Utf8(s), _, Kind::Enum(_)) => {
+ ScalarAdapter::Enum(EnumString(s.to_owned()))
+ }
+ _ => ScalarAdapter::Builtin(scalar),
+ })
+ }
+
+ pub fn into_scalar(self, ty: DataType) -> Option {
+ match (&self, &ty) {
+ (ScalarAdapter::Builtin(scalar), _) => Some(scalar.into_scalar_impl()),
+ (ScalarAdapter::Uuid(uuid), &DataType::Varchar) => {
+ Some(ScalarImpl::from(uuid.to_string()))
+ }
+ (ScalarAdapter::Numeric(numeric), &DataType::Varchar) => {
+ Some(ScalarImpl::from(pg_numeric_to_string(numeric)))
+ }
+ (ScalarAdapter::Numeric(numeric), &DataType::Int256) => {
+ pg_numeric_to_rw_int256(numeric)
+ }
+ (ScalarAdapter::Enum(EnumString(s)), &DataType::Varchar) => Some(ScalarImpl::from(s)),
+ _ => {
+ tracing::error!(
+ adapter = self.name(),
+ rw_type = ty.pg_name(),
+ "failed to convert from ScalarAdapter: invalid conversion"
+ );
+ None
+ }
+ }
+ }
+
+ pub fn build_scalar_in_list(
+ vec: Vec>>,
+ ty: DataType,
+ mut builder: risingwave_common::array::ArrayBuilderImpl,
+ ) -> Option {
+ for val in vec {
+ let scalar = match (val, &ty) {
+ (Some(ScalarAdapter::Numeric(numeric)), &DataType::Varchar) => {
+ if pg_numeric_is_special(&numeric) {
+ return None;
+ } else {
+ Some(ScalarImpl::from(pg_numeric_to_string(&numeric)))
+ }
+ }
+ (Some(ScalarAdapter::Numeric(numeric)), &DataType::Int256) => match numeric {
+ PgNumeric::Normalized(big_decimal) => {
+ match Int256::from_str(big_decimal.to_string().as_str()) {
+ Ok(num) => Some(ScalarImpl::from(num)),
+ Err(err) => {
+ tracing::error!(error = %err.as_report(), "parse pg-numeric as rw_int256 failed");
+ return None;
+ }
+ }
+ }
+ _ => return None,
+ },
+ (Some(ScalarAdapter::Numeric(numeric)), &DataType::Decimal) => match numeric {
+ PgNumeric::Normalized(big_decimal) => {
+ match Decimal::from_str(big_decimal.to_string().as_str()) {
+ Ok(num) => Some(ScalarImpl::from(num)),
+ Err(err) => {
+ tracing::error!(error = %err.as_report(), "parse pg-numeric as rw-numeric failed (likely out-of-range");
+ return None;
+ }
+ }
+ }
+ _ => return None,
+ },
+ (Some(_), _) => unreachable!(
+ "into_scalar_in_list should only be called with ScalarAdapter::Numeric types"
+ ),
+ (None, _) => None,
+ };
+ builder.append(scalar);
+ }
+ Some(builder)
+ }
+}
+
+fn pg_numeric_is_special(val: &PgNumeric) -> bool {
+ matches!(
+ val,
+ PgNumeric::NegativeInf | PgNumeric::PositiveInf | PgNumeric::NaN
+ )
+}
+
+fn pg_numeric_to_rw_int256(val: &PgNumeric) -> Option {
+ match Int256::from_str(pg_numeric_to_string(val).as_str()) {
+ Ok(num) => Some(ScalarImpl::from(num)),
+ Err(err) => {
+ tracing::error!(error = %err.as_report(), "failed to convert PgNumeric to Int256");
+ None
+ }
+ }
+}
+
+fn pg_numeric_to_string(val: &PgNumeric) -> String {
+ // TODO(kexiang): NEGATIVE_INFINITY -> -Infinity, POSITIVE_INFINITY -> Infinity, NAN -> NaN
+ // The current implementation is to ensure consistency with the behavior of cdc event parsor.
+ match val {
+ PgNumeric::NegativeInf => String::from("NEGATIVE_INFINITY"),
+ PgNumeric::Normalized(big_decimal) => big_decimal.to_string(),
+ PgNumeric::PositiveInf => String::from("POSITIVE_INFINITY"),
+ PgNumeric::NaN => String::from("NAN"),
+ }
+}
+
+fn string_to_pg_numeric(s: &str) -> super::ConnectorResult {
+ Ok(match s {
+ "NEGATIVE_INFINITY" => PgNumeric::NegativeInf,
+ "POSITIVE_INFINITY" => PgNumeric::PositiveInf,
+ "NAN" => PgNumeric::NaN,
+ _ => PgNumeric::Normalized(s.parse().unwrap()),
+ })
+}
diff --git a/src/connector/src/parser/unified/avro.rs b/src/connector/src/parser/unified/avro.rs
index 4e7d134f42554..257ca2926c342 100644
--- a/src/connector/src/parser/unified/avro.rs
+++ b/src/connector/src/parser/unified/avro.rs
@@ -431,7 +431,7 @@ pub(crate) fn unix_epoch_days() -> i32 {
#[cfg(test)]
mod tests {
use apache_avro::Decimal as AvroDecimal;
- use risingwave_common::types::{Decimal, Timestamptz};
+ use risingwave_common::types::Decimal;
use super::*;
diff --git a/src/connector/src/parser/unified/json.rs b/src/connector/src/parser/unified/json.rs
index d9deb73d582d9..11c569832268e 100644
--- a/src/connector/src/parser/unified/json.rs
+++ b/src/connector/src/parser/unified/json.rs
@@ -385,10 +385,16 @@ impl JsonParseOptions {
.map_err(|_| create_error())?
.into()
}
-
- (Some(DataType::Decimal), ValueType::String) => ScalarImpl::Decimal(
- Decimal::from_str(value.as_str().unwrap()).map_err(|_err| create_error())?,
- ),
+ (Some(DataType::Decimal), ValueType::String) => {
+ let str = value.as_str().unwrap();
+ // the following values are special string generated by Debezium and should be handled separately
+ match str {
+ "NAN" => ScalarImpl::Decimal(Decimal::NaN),
+ "POSITIVE_INFINITY" => ScalarImpl::Decimal(Decimal::PositiveInf),
+ "NEGATIVE_INFINITY" => ScalarImpl::Decimal(Decimal::NegativeInf),
+ _ => ScalarImpl::Decimal(Decimal::from_str(str).map_err(|_err| create_error())?),
+ }
+ }
(Some(DataType::Decimal), ValueType::Object) => {
// ref https://github.com/risingwavelabs/risingwave/issues/10628
// handle debezium json (variable scale): {"scale": int, "value": bytes}
diff --git a/src/connector/src/sink/encoder/avro.rs b/src/connector/src/sink/encoder/avro.rs
index 924beb281eda7..20d8a0997a7ff 100644
--- a/src/connector/src/sink/encoder/avro.rs
+++ b/src/connector/src/sink/encoder/avro.rs
@@ -526,7 +526,7 @@ mod tests {
test_ok(
&DataType::Int64,
- Some(ScalarImpl::Int64(std::i64::MAX)),
+ Some(ScalarImpl::Int64(i64::MAX)),
r#""long""#,
Value::Long(i64::MAX),
);
diff --git a/src/connector/src/sink/encoder/json.rs b/src/connector/src/sink/encoder/json.rs
index 6eb7ca5cd1b50..2ee8bfcc0d27c 100644
--- a/src/connector/src/sink/encoder/json.rs
+++ b/src/connector/src/sink/encoder/json.rs
@@ -436,8 +436,8 @@ fn type_as_json_schema(rw_type: &DataType) -> Map {
mod tests {
use risingwave_common::types::{
- DataType, Date, Decimal, Interval, Scalar, ScalarImpl, StructRef, StructType, StructValue,
- Time, Timestamp,
+ Date, Decimal, Interval, Scalar, ScalarImpl, StructRef, StructType, StructValue, Time,
+ Timestamp,
};
use super::*;
@@ -485,7 +485,7 @@ mod tests {
data_type: DataType::Int64,
..mock_field.clone()
},
- Some(ScalarImpl::Int64(std::i64::MAX).as_scalar_ref_impl()),
+ Some(ScalarImpl::Int64(i64::MAX).as_scalar_ref_impl()),
DateHandlingMode::FromCe,
TimestampHandlingMode::String,
TimestamptzHandlingMode::UtcString,
@@ -495,7 +495,7 @@ mod tests {
.unwrap();
assert_eq!(
serde_json::to_string(&int64_value).unwrap(),
- std::i64::MAX.to_string()
+ i64::MAX.to_string()
);
// https://github.com/debezium/debezium/blob/main/debezium-core/src/main/java/io/debezium/time/ZonedTimestamp.java
diff --git a/src/connector/src/sink/kafka.rs b/src/connector/src/sink/kafka.rs
index f15173a4aabf6..8313ea20dafa7 100644
--- a/src/connector/src/sink/kafka.rs
+++ b/src/connector/src/sink/kafka.rs
@@ -35,7 +35,9 @@ use with_options::WithOptions;
use super::catalog::{SinkFormat, SinkFormatDesc};
use super::{Sink, SinkError, SinkParam};
-use crate::connector_common::{KafkaCommon, KafkaPrivateLinkCommon, RdKafkaPropertiesCommon};
+use crate::connector_common::{
+ AwsAuthProps, KafkaCommon, KafkaPrivateLinkCommon, RdKafkaPropertiesCommon,
+};
use crate::sink::catalog::desc::SinkDesc;
use crate::sink::formatter::SinkFormatterImpl;
use crate::sink::log_store::DeliveryFutureManagerAddFuture;
@@ -43,7 +45,9 @@ use crate::sink::writer::{
AsyncTruncateLogSinkerOf, AsyncTruncateSinkWriter, AsyncTruncateSinkWriterExt, FormattedSink,
};
use crate::sink::{DummySinkCommitCoordinator, Result, SinkWriterParam};
-use crate::source::kafka::{KafkaProperties, KafkaSplitEnumerator, PrivateLinkProducerContext};
+use crate::source::kafka::{
+ KafkaContextCommon, KafkaProperties, KafkaSplitEnumerator, RwProducerContext,
+};
use crate::source::{SourceEnumeratorContext, SplitEnumerator};
use crate::{
deserialize_duration_from_string, deserialize_u32_from_string, dispatch_sink_formatter_impl,
@@ -244,6 +248,9 @@ pub struct KafkaConfig {
#[serde(flatten)]
pub privatelink_common: KafkaPrivateLinkCommon,
+
+ #[serde(flatten)]
+ pub aws_auth_props: AwsAuthProps,
}
impl KafkaConfig {
@@ -274,6 +281,7 @@ impl From for KafkaProperties {
rdkafka_properties_common: val.rdkafka_properties_common,
rdkafka_properties_consumer: Default::default(),
privatelink_common: val.privatelink_common,
+ aws_auth_props: val.aws_auth_props,
unknown_fields: Default::default(),
}
}
@@ -393,7 +401,7 @@ const KAFKA_WRITER_MAX_QUEUE_SIZE_RATIO: f32 = 1.2;
const KAFKA_WRITER_MAX_QUEUE_SIZE: usize = 100000;
struct KafkaPayloadWriter<'a> {
- inner: &'a FutureProducer,
+ inner: &'a FutureProducer,
add_future: DeliveryFutureManagerAddFuture<'a, KafkaSinkDeliveryFuture>,
config: &'a KafkaConfig,
}
@@ -402,13 +410,13 @@ pub type KafkaSinkDeliveryFuture = impl TryFuture +
pub struct KafkaSinkWriter {
formatter: SinkFormatterImpl,
- inner: FutureProducer,
+ inner: FutureProducer,
config: KafkaConfig,
}
impl KafkaSinkWriter {
async fn new(config: KafkaConfig, formatter: SinkFormatterImpl) -> Result {
- let inner: FutureProducer = {
+ let inner: FutureProducer = {
let mut c = ClientConfig::new();
// KafkaConfig configuration
@@ -419,13 +427,16 @@ impl KafkaSinkWriter {
c.set("bootstrap.servers", &config.common.brokers);
// Create the producer context, will be used to create the producer
- let producer_ctx = PrivateLinkProducerContext::new(
- config.privatelink_common.broker_rewrite_map.clone(),
- // fixme: enable kafka native metrics for sink
+ let broker_rewrite_map = config.privatelink_common.broker_rewrite_map.clone();
+ let ctx_common = KafkaContextCommon::new(
+ broker_rewrite_map,
None,
None,
- )?;
-
+ config.aws_auth_props.clone(),
+ config.common.is_aws_msk_iam(),
+ )
+ .await?;
+ let producer_ctx = RwProducerContext::new(ctx_common);
// Generate the producer
c.create_with_context(producer_ctx).await?
};
diff --git a/src/connector/src/sink/mod.rs b/src/connector/src/sink/mod.rs
index d281a97ef6c26..2ef4bb953b67e 100644
--- a/src/connector/src/sink/mod.rs
+++ b/src/connector/src/sink/mod.rs
@@ -71,7 +71,6 @@ use crate::sink::catalog::desc::SinkDesc;
use crate::sink::catalog::{SinkCatalog, SinkId};
use crate::sink::log_store::{LogReader, LogStoreReadItem, LogStoreResult, TruncateOffset};
use crate::sink::writer::SinkWriter;
-use crate::ConnectorParams;
const BOUNDED_CHANNEL_SIZE: usize = 16;
#[macro_export]
@@ -288,7 +287,6 @@ impl SinkMetrics {
#[derive(Clone)]
pub struct SinkWriterParam {
- pub connector_params: ConnectorParams,
pub executor_id: u64,
pub vnode_bitmap: Option,
pub meta_client: Option,
@@ -326,7 +324,6 @@ impl SinkMetaClient {
impl SinkWriterParam {
pub fn for_test() -> Self {
SinkWriterParam {
- connector_params: Default::default(),
executor_id: Default::default(),
vnode_bitmap: Default::default(),
meta_client: Default::default(),
diff --git a/src/connector/src/sink/redis.rs b/src/connector/src/sink/redis.rs
index 9da1d8b4d3f03..6baeccdd32ed8 100644
--- a/src/connector/src/sink/redis.rs
+++ b/src/connector/src/sink/redis.rs
@@ -378,8 +378,8 @@ mod test {
use std::collections::BTreeMap;
use rdkafka::message::FromBytes;
- use risingwave_common::array::{Array, I32Array, Op, StreamChunk, Utf8Array};
- use risingwave_common::catalog::{Field, Schema};
+ use risingwave_common::array::{Array, I32Array, Op, Utf8Array};
+ use risingwave_common::catalog::Field;
use risingwave_common::types::DataType;
use risingwave_common::util::iter_util::ZipEqDebug;
diff --git a/src/connector/src/sink/remote.rs b/src/connector/src/sink/remote.rs
index 3ab7e90a69367..63fc60ecc4510 100644
--- a/src/connector/src/sink/remote.rs
+++ b/src/connector/src/sink/remote.rs
@@ -42,8 +42,8 @@ use risingwave_pb::connector_service::sink_writer_stream_request::{
use risingwave_pb::connector_service::{
sink_coordinator_stream_request, sink_coordinator_stream_response, sink_writer_stream_response,
PbSinkParam, SinkCoordinatorStreamRequest, SinkCoordinatorStreamResponse, SinkMetadata,
- SinkPayloadFormat, SinkWriterStreamRequest, SinkWriterStreamResponse, TableSchema,
- ValidateSinkRequest, ValidateSinkResponse,
+ SinkWriterStreamRequest, SinkWriterStreamResponse, TableSchema, ValidateSinkRequest,
+ ValidateSinkResponse,
};
use risingwave_rpc_client::error::RpcError;
use risingwave_rpc_client::{
@@ -68,7 +68,6 @@ use crate::sink::{
DummySinkCommitCoordinator, LogSinker, Result, Sink, SinkCommitCoordinator, SinkError,
SinkLogReader, SinkMetrics, SinkParam, SinkWriterParam,
};
-use crate::ConnectorParams;
macro_rules! def_remote_sink {
() => {
@@ -82,7 +81,6 @@ macro_rules! def_remote_sink {
{ HttpJava, HttpJavaSink, "http" }
}
};
- () => {};
({ $variant_name:ident, $sink_type_name:ident, $sink_name:expr }) => {
#[derive(Debug)]
pub struct $variant_name;
@@ -168,7 +166,7 @@ impl Sink for RemoteSink {
async fn validate_remote_sink(param: &SinkParam, sink_name: &str) -> ConnectorResult<()> {
if sink_name == ElasticSearchSink::SINK_NAME
&& param.downstream_pk.len() > 1
- && param.properties.get(ES_OPTION_DELIMITER).is_none()
+ && !param.properties.contains_key(ES_OPTION_DELIMITER)
{
bail!("Es sink only support single pk or pk with delimiter option");
}
@@ -283,7 +281,7 @@ impl RemoteLogSinker {
request_sender,
response_stream,
} = EmbeddedConnectorClient::new()?
- .start_sink_writer_stream(payload_schema, sink_proto, SinkPayloadFormat::StreamChunk)
+ .start_sink_writer_stream(payload_schema, sink_proto)
.await?;
let sink_metrics = writer_param.sink_metrics;
@@ -547,12 +545,8 @@ impl Sink for CoordinatedRemoteSink {
"sink needs coordination should not have singleton input"
))
})?,
- CoordinatedRemoteSinkWriter::new(
- self.param.clone(),
- writer_param.connector_params,
- writer_param.sink_metrics.clone(),
- )
- .await?,
+ CoordinatedRemoteSinkWriter::new(self.param.clone(), writer_param.sink_metrics.clone())
+ .await?,
)
.await?
.into_log_sinker(writer_param.sink_metrics))
@@ -572,18 +566,10 @@ pub struct CoordinatedRemoteSinkWriter {
}
impl CoordinatedRemoteSinkWriter {
- pub async fn new(
- param: SinkParam,
- connector_params: ConnectorParams,
- sink_metrics: SinkMetrics,
- ) -> Result {
+ pub async fn new(param: SinkParam, sink_metrics: SinkMetrics) -> Result {
let sink_proto = param.to_proto();
let stream_handle = EmbeddedConnectorClient::new()?
- .start_sink_writer_stream(
- sink_proto.table_schema.clone(),
- sink_proto,
- connector_params.sink_payload_format,
- )
+ .start_sink_writer_stream(sink_proto.table_schema.clone(), sink_proto)
.await?;
Ok(Self {
@@ -717,13 +703,11 @@ impl EmbeddedConnectorClient {
&self,
payload_schema: Option,
sink_proto: PbSinkParam,
- sink_payload_format: SinkPayloadFormat,
) -> Result> {
let (handle, first_rsp) = SinkWriterStreamHandle::initialize(
SinkWriterStreamRequest {
request: Some(SinkRequest::Start(StartSink {
sink_param: Some(sink_proto),
- format: sink_payload_format as i32,
payload_schema,
})),
},
diff --git a/src/connector/src/sink/snowflake_connector.rs b/src/connector/src/sink/snowflake_connector.rs
index eee4af2258b09..ee1cb90b70a52 100644
--- a/src/connector/src/sink/snowflake_connector.rs
+++ b/src/connector/src/sink/snowflake_connector.rs
@@ -19,7 +19,6 @@ use std::time::{SystemTime, UNIX_EPOCH};
use anyhow::{anyhow, Context};
use jsonwebtoken::{encode, Algorithm, EncodingKey, Header};
use reqwest::{header, Client, RequestBuilder, StatusCode};
-use risingwave_common::config::ObjectStoreConfig;
use risingwave_object_store::object::*;
use serde::{Deserialize, Serialize};
diff --git a/src/connector/src/source/cdc/external/mock_external_table.rs b/src/connector/src/source/cdc/external/mock_external_table.rs
index fcfb1505225ea..7242f5614d409 100644
--- a/src/connector/src/source/cdc/external/mock_external_table.rs
+++ b/src/connector/src/source/cdc/external/mock_external_table.rs
@@ -86,7 +86,7 @@ impl MockExternalTableReader {
]),
];
- let snapshots = vec![snap0];
+ let snapshots = [snap0];
if snap_idx >= snapshots.len() {
return Ok(());
}
diff --git a/src/connector/src/source/cdc/external/postgres.rs b/src/connector/src/source/cdc/external/postgres.rs
index 049fedd1485af..a4e3117b49519 100644
--- a/src/connector/src/source/cdc/external/postgres.rs
+++ b/src/connector/src/source/cdc/external/postgres.rs
@@ -32,6 +32,7 @@ use tokio_postgres::{NoTls, Statement};
use crate::error::{ConnectorError, ConnectorResult};
use crate::parser::postgres_row_to_owned_row;
+use crate::parser::scalar_adapter::ScalarAdapter;
#[cfg(not(madsim))]
use crate::source::cdc::external::maybe_tls_connector::MaybeMakeTlsConnector;
use crate::source::cdc::external::{
@@ -240,12 +241,12 @@ impl PostgresExternalTableReader {
let stream = match start_pk_row {
Some(ref pk_row) => {
- let params: Vec>> = pk_row
+ let params: Vec>> = pk_row
.iter()
.zip_eq_fast(self.prepared_scan_stmt.params())
.map(|(datum, ty)| {
datum
- .map(|scalar| scalar_adapter::to_extra(scalar, ty))
+ .map(|scalar| ScalarAdapter::from_scalar(scalar, ty))
.transpose()
})
.try_collect()?;
@@ -259,7 +260,7 @@ impl PostgresExternalTableReader {
Self::get_normalized_table_name(&table_name),
order_key,
);
- let params: Vec>> = vec![];
+ let params: Vec>> = vec![];
client.query_raw(&sql, ¶ms).await?
}
};
@@ -297,72 +298,6 @@ impl PostgresExternalTableReader {
}
}
-mod scalar_adapter {
- use pg_bigdecimal::PgNumeric;
- use risingwave_common::types::ScalarRefImpl;
- use tokio_postgres::types::{to_sql_checked, IsNull, Kind, ToSql, Type};
-
- use crate::parser::EnumString;
-
- #[derive(Debug)]
- pub(super) enum ScalarAdapter<'a> {
- Builtin(ScalarRefImpl<'a>),
- Uuid(uuid::Uuid),
- Numeric(PgNumeric),
- Enum(EnumString),
- }
-
- impl ToSql for ScalarAdapter<'_> {
- to_sql_checked!();
-
- fn to_sql(
- &self,
- ty: &Type,
- out: &mut bytes::BytesMut,
- ) -> Result> {
- match self {
- ScalarAdapter::Builtin(v) => v.to_sql(ty, out),
- ScalarAdapter::Uuid(v) => v.to_sql(ty, out),
- ScalarAdapter::Numeric(v) => v.to_sql(ty, out),
- ScalarAdapter::Enum(v) => v.to_sql(ty, out),
- }
- }
-
- fn accepts(_ty: &Type) -> bool {
- true
- }
- }
-
- /// convert `ScalarRefImpl` to `ScalarAdapter` so that we can correctly encode to postgres value
- pub(super) fn to_extra<'a>(
- scalar: ScalarRefImpl<'a>,
- ty: &Type,
- ) -> super::ConnectorResult> {
- Ok(match (scalar, ty, ty.kind()) {
- (ScalarRefImpl::Utf8(s), &Type::UUID, _) => ScalarAdapter::Uuid(s.parse()?),
- (ScalarRefImpl::Utf8(s), &Type::NUMERIC, _) => {
- ScalarAdapter::Numeric(string_to_pg_numeric(s)?)
- }
- (ScalarRefImpl::Int256(s), &Type::NUMERIC, _) => {
- ScalarAdapter::Numeric(string_to_pg_numeric(&s.to_string())?)
- }
- (ScalarRefImpl::Utf8(s), _, Kind::Enum(_)) => {
- ScalarAdapter::Enum(EnumString(s.to_owned()))
- }
- _ => ScalarAdapter::Builtin(scalar),
- })
- }
-
- fn string_to_pg_numeric(s: &str) -> super::ConnectorResult {
- Ok(match s {
- "NEGATIVE_INFINITY" => PgNumeric::NegativeInf,
- "POSITIVE_INFINITY" => PgNumeric::PositiveInf,
- "NAN" => PgNumeric::NaN,
- _ => PgNumeric::Normalized(s.parse().unwrap()),
- })
- }
-}
-
#[cfg(test)]
mod tests {
use futures::pin_mut;
diff --git a/src/connector/src/source/filesystem/opendal_source/opendal_enumerator.rs b/src/connector/src/source/filesystem/opendal_source/opendal_enumerator.rs
index 96646ade0e1df..f65eacd9a9696 100644
--- a/src/connector/src/source/filesystem/opendal_source/opendal_enumerator.rs
+++ b/src/connector/src/source/filesystem/opendal_source/opendal_enumerator.rs
@@ -56,10 +56,7 @@ impl SplitEnumerator for OpendalEnumerator {
impl OpendalEnumerator {
pub async fn list(&self) -> ConnectorResult {
- let prefix = match &self.prefix {
- Some(prefix) => prefix,
- None => "",
- };
+ let prefix = self.prefix.as_deref().unwrap_or("");
let object_lister = self
.op
diff --git a/src/connector/src/source/kafka/client_context.rs b/src/connector/src/source/kafka/client_context.rs
new file mode 100644
index 0000000000000..6a01da356ff51
--- /dev/null
+++ b/src/connector/src/source/kafka/client_context.rs
@@ -0,0 +1,227 @@
+// Copyright 2024 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::thread;
+
+use anyhow::anyhow;
+use aws_config::Region;
+use aws_sdk_s3::config::SharedCredentialsProvider;
+use rdkafka::client::{BrokerAddr, OAuthToken};
+use rdkafka::consumer::ConsumerContext;
+use rdkafka::message::DeliveryResult;
+use rdkafka::producer::ProducerContext;
+use rdkafka::{ClientContext, Statistics};
+
+use super::private_link::{BrokerAddrRewriter, PrivateLinkContextRole};
+use super::stats::RdKafkaStats;
+use crate::connector_common::AwsAuthProps;
+use crate::error::ConnectorResult;
+
+struct IamAuthEnv {
+ credentials_provider: SharedCredentialsProvider,
+ region: Region,
+ // XXX(runji): madsim does not support `Handle` for now
+ #[cfg(not(madsim))]
+ rt: tokio::runtime::Handle,
+}
+
+pub struct KafkaContextCommon {
+ // For VPC PrivateLink support
+ addr_rewriter: BrokerAddrRewriter,
+
+ // identifier is required when reporting metrics as a label, usually it is compose by connector
+ // format (source or sink) and corresponding id (source_id or sink_id)
+ // identifier and metrics should be set at the same time
+ identifier: Option,
+ metrics: Option>,
+
+ /// Credential and region for AWS MSK
+ auth: Option,
+}
+
+impl KafkaContextCommon {
+ pub async fn new(
+ broker_rewrite_map: Option>,
+ identifier: Option,
+ metrics: Option>,
+ auth: AwsAuthProps,
+ is_aws_msk_iam: bool,
+ ) -> ConnectorResult {
+ let addr_rewriter =
+ BrokerAddrRewriter::new(PrivateLinkContextRole::Consumer, broker_rewrite_map)?;
+ let auth = if is_aws_msk_iam {
+ let config = auth.build_config().await?;
+ let credentials_provider = config
+ .credentials_provider()
+ .ok_or_else(|| anyhow!("missing aws credentials_provider"))?;
+ let region = config
+ .region()
+ .ok_or_else(|| anyhow!("missing aws region"))?
+ .clone();
+ Some(IamAuthEnv {
+ credentials_provider,
+ region,
+ #[cfg(not(madsim))]
+ rt: tokio::runtime::Handle::current(),
+ })
+ } else {
+ None
+ };
+ Ok(Self {
+ addr_rewriter,
+ identifier,
+ metrics,
+ auth,
+ })
+ }
+}
+
+impl KafkaContextCommon {
+ fn stats(&self, statistics: Statistics) {
+ if let Some(metrics) = &self.metrics
+ && let Some(id) = &self.identifier
+ {
+ metrics.report(id.as_str(), &statistics);
+ }
+ }
+
+ fn rewrite_broker_addr(&self, addr: BrokerAddr) -> BrokerAddr {
+ self.addr_rewriter.rewrite_broker_addr(addr)
+ }
+
+ // XXX(runji): oauth is ignored in simulation
+ #[cfg_or_panic::cfg_or_panic(not(madsim))]
+ fn generate_oauth_token(
+ &self,
+ _oauthbearer_config: Option<&str>,
+ ) -> Result> {
+ use aws_msk_iam_sasl_signer::generate_auth_token_from_credentials_provider;
+ use tokio::time::{timeout, Duration};
+
+ if let Some(IamAuthEnv {
+ credentials_provider,
+ region,
+ rt,
+ }) = &self.auth
+ {
+ let region = region.clone();
+ let credentials_provider = credentials_provider.clone();
+ let rt = rt.clone();
+ let (token, expiration_time_ms) = {
+ let handle = thread::spawn(move || {
+ rt.block_on(async {
+ timeout(
+ Duration::from_secs(10),
+ generate_auth_token_from_credentials_provider(
+ region,
+ credentials_provider,
+ ),
+ )
+ .await
+ })
+ });
+ handle.join().unwrap()??
+ };
+ Ok(OAuthToken {
+ token,
+ principal_name: "".to_string(),
+ lifetime_ms: expiration_time_ms,
+ })
+ } else {
+ Err("must provide AWS IAM credential".into())
+ }
+ }
+
+ fn enable_refresh_oauth_token(&self) -> bool {
+ self.auth.is_some()
+ }
+}
+
+pub type BoxConsumerContext = Box;
+
+/// Kafka consumer context used for private link, IAM auth, and metrics
+pub struct RwConsumerContext {
+ common: KafkaContextCommon,
+}
+
+impl RwConsumerContext {
+ pub fn new(common: KafkaContextCommon) -> Self {
+ Self { common }
+ }
+}
+
+impl ClientContext for RwConsumerContext {
+ /// this func serves as a callback when `poll` is completed.
+ fn stats(&self, statistics: Statistics) {
+ self.common.stats(statistics);
+ }
+
+ fn rewrite_broker_addr(&self, addr: BrokerAddr) -> BrokerAddr {
+ self.common.rewrite_broker_addr(addr)
+ }
+
+ fn generate_oauth_token(
+ &self,
+ oauthbearer_config: Option<&str>,
+ ) -> Result> {
+ self.common.generate_oauth_token(oauthbearer_config)
+ }
+
+ fn enable_refresh_oauth_token(&self) -> bool {
+ self.common.enable_refresh_oauth_token()
+ }
+}
+
+// required by the trait bound of BaseConsumer
+impl ConsumerContext for RwConsumerContext {}
+
+/// Kafka producer context used for private link, IAM auth, and metrics
+pub struct RwProducerContext {
+ common: KafkaContextCommon,
+}
+
+impl RwProducerContext {
+ pub fn new(common: KafkaContextCommon) -> Self {
+ Self { common }
+ }
+}
+
+impl ClientContext for RwProducerContext {
+ fn stats(&self, statistics: Statistics) {
+ self.common.stats(statistics);
+ }
+
+ fn rewrite_broker_addr(&self, addr: BrokerAddr) -> BrokerAddr {
+ self.common.rewrite_broker_addr(addr)
+ }
+
+ fn generate_oauth_token(
+ &self,
+ oauthbearer_config: Option<&str>,
+ ) -> Result> {
+ self.common.generate_oauth_token(oauthbearer_config)
+ }
+
+ fn enable_refresh_oauth_token(&self) -> bool {
+ self.common.enable_refresh_oauth_token()
+ }
+}
+
+impl ProducerContext for RwProducerContext {
+ type DeliveryOpaque = ();
+
+ fn delivery(&self, _: &DeliveryResult<'_>, _: Self::DeliveryOpaque) {}
+}
diff --git a/src/connector/src/source/kafka/enumerator/client.rs b/src/connector/src/source/kafka/enumerator/client.rs
index 697fe421a5fb2..d4dddad67bfdf 100644
--- a/src/connector/src/source/kafka/enumerator/client.rs
+++ b/src/connector/src/source/kafka/enumerator/client.rs
@@ -25,7 +25,9 @@ use risingwave_common::bail;
use crate::error::ConnectorResult;
use crate::source::base::SplitEnumerator;
use crate::source::kafka::split::KafkaSplit;
-use crate::source::kafka::{KafkaProperties, PrivateLinkConsumerContext, KAFKA_ISOLATION_LEVEL};
+use crate::source::kafka::{
+ KafkaContextCommon, KafkaProperties, RwConsumerContext, KAFKA_ISOLATION_LEVEL,
+};
use crate::source::SourceEnumeratorContextRef;
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
@@ -40,7 +42,7 @@ pub struct KafkaSplitEnumerator {
context: SourceEnumeratorContextRef,
broker_address: String,
topic: String,
- client: BaseConsumer,
+ client: BaseConsumer,
start_offset: KafkaEnumeratorOffset,
// maybe used in the future for batch processing
@@ -90,10 +92,30 @@ impl SplitEnumerator for KafkaSplitEnumerator {
}
// don't need kafka metrics from enumerator
- let client_ctx = PrivateLinkConsumerContext::new(broker_rewrite_map, None, None)?;
- let client: BaseConsumer =
+ let ctx_common = KafkaContextCommon::new(
+ broker_rewrite_map,
+ None,
+ None,
+ properties.aws_auth_props,
+ common_props.is_aws_msk_iam(),
+ )
+ .await?;
+ let client_ctx = RwConsumerContext::new(ctx_common);
+ let client: BaseConsumer =
config.create_with_context(client_ctx).await?;
+ // Note that before any SASL/OAUTHBEARER broker connection can succeed the application must call
+ // rd_kafka_oauthbearer_set_token() once – either directly or, more typically, by invoking either
+ // rd_kafka_poll(), rd_kafka_consumer_poll(), rd_kafka_queue_poll(), etc, in order to cause retrieval
+ // of an initial token to occur.
+ // https://docs.confluent.io/platform/current/clients/librdkafka/html/rdkafka_8h.html#a988395722598f63396d7a1bedb22adaf
+ if common_props.is_aws_msk_iam() {
+ #[cfg(not(madsim))]
+ client.poll(Duration::from_secs(10)); // note: this is a blocking call
+ #[cfg(madsim)]
+ client.poll(Duration::from_secs(10)).await;
+ }
+
Ok(Self {
context,
broker_address,
@@ -129,6 +151,7 @@ impl SplitEnumerator for KafkaSplitEnumerator {
partition,
start_offset: start_offsets.remove(&partition).unwrap(),
stop_offset: stop_offsets.remove(&partition).unwrap(),
+ hack_seek_to_latest: false,
})
.collect();
@@ -232,6 +255,7 @@ impl KafkaSplitEnumerator {
partition: *partition,
start_offset: Some(start_offset),
stop_offset: Some(stop_offset),
+ hack_seek_to_latest:false
}
})
.collect::