around the whole document, but we don't have that. */
+ position: relative;
+}
+
+a.doc-anchor {
+ color: black;
+ display: none;
+ position: absolute;
+ left: -20px;
+ /* We add this padding so that when the cursor moves from the heading's text to the anchor,
+ the anchor doesn't disappear. */
+ padding-right: 5px;
+ /* And this padding is used to make the anchor larger and easier to click on. */
+ padding-left: 3px;
+}
+
+*:hover>.doc-anchor {
+ display: block;
+}
/* Code */
pre, code {
diff --git a/e2e_test/commands/sr_register b/e2e_test/commands/sr_register
new file mode 100755
index 0000000000000..57dc65e50610d
--- /dev/null
+++ b/e2e_test/commands/sr_register
@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+# Register a schema to schema registry
+#
+# Usage: sr_register
+#
+# https://docs.confluent.io/platform/current/schema-registry/develop/api.html#post--subjects-(string-%20subject)-versions
+
+# Validate arguments
+if [[ $# -ne 2 ]]; then
+ echo "Usage: sr_register "
+ exit 1
+fi
+
+subject="$1"
+schema="$2"
+
+
+if [[ -z $subject || -z $schema ]]; then
+ echo "Error: Arguments cannot be empty"
+ exit 1
+fi
+
+echo "$schema" | jq '{"schema": tojson}' \
+| curl -X POST -H 'content-type:application/vnd.schemaregistry.v1+json' -d @- "${RISEDEV_SCHEMA_REGISTRY_URL}/subjects/${subject}/versions"
diff --git a/e2e_test/source_inline/kafka/avro/alter_source.slt b/e2e_test/source_inline/kafka/avro/alter_source.slt
index 446fc6196d32b..57677af57cd92 100644
--- a/e2e_test/source_inline/kafka/avro/alter_source.slt
+++ b/e2e_test/source_inline/kafka/avro/alter_source.slt
@@ -12,8 +12,7 @@ system ok
rpk topic create 'avro_alter_source_test'
system ok
-echo '{"type":"record","name":"Root","fields":[{"name":"foo","type":"string"}]}' | jq '{"schema": tojson}' \
-| curl -X POST -H 'content-type:application/json' -d @- "${RISEDEV_SCHEMA_REGISTRY_URL}/subjects/avro_alter_source_test-value/versions"
+sr_register avro_alter_source_test-value '{"type":"record","name":"Root","fields":[{"name":"foo","type":"string"}]}'
statement ok
create source s
@@ -27,8 +26,7 @@ FORMAT PLAIN ENCODE AVRO (
# create a new version of schema and produce a message
system ok
-echo '{"type":"record","name":"Root","fields":[{"name":"bar","type":"int","default":0},{"name":"foo","type":"string"}]}' | jq '{"schema": tojson}' \
-| curl -X POST -H 'content-type:application/json' -d @- "${RISEDEV_SCHEMA_REGISTRY_URL}/subjects/avro_alter_source_test-value/versions"
+sr_register avro_alter_source_test-value '{"type":"record","name":"Root","fields":[{"name":"bar","type":"int","default":0},{"name":"foo","type":"string"}]}'
system ok
echo '{"foo":"ABC", "bar":1}' | rpk topic produce --schema-id=topic avro_alter_source_test
diff --git a/e2e_test/source_inline/kafka/avro/union.slt b/e2e_test/source_inline/kafka/avro/union.slt
new file mode 100644
index 0000000000000..44e1db659d120
--- /dev/null
+++ b/e2e_test/source_inline/kafka/avro/union.slt
@@ -0,0 +1,175 @@
+control substitution on
+
+system ok
+rpk topic delete 'avro-union' || true; \
+(rpk sr subject delete 'avro-union-value' && rpk sr subject delete 'avro-union-value' --permanent) || true;
+rpk topic create avro-union
+
+system ok
+sr_register avro-union-value '
+{
+ "type": "record",
+ "name": "Root",
+ "fields": [
+ {
+ "name": "unionType",
+ "type": ["int", "string"]
+ },
+ {
+ "name": "unionTypeComplex",
+ "type": [
+ "null",
+ {"type": "record", "name": "Email","fields": [{"name":"inner","type":"string"}]},
+ {"type": "record", "name": "Fax","fields": [{"name":"inner","type":"int"}]},
+ {"type": "record", "name": "Sms","fields": [{"name":"inner","type":"int"}]}
+ ]
+ },
+ {
+ "name": "enumField",
+ "type": ["null", "int", {
+ "type": "enum",
+ "name": "myEnum",
+ "namespace": "my.namespace",
+ "symbols": ["A", "B", "C", "D"]
+ }],
+ "default": null
+ }
+ ]
+}
+'
+
+system ok
+cat<risingwave-sink-deltalake
test
-
- com.risingwave
- risingwave-sink-iceberg
- test
-
com.risingwave
s3-common
diff --git a/java/connector-node/risingwave-sink-iceberg/pom.xml b/java/connector-node/risingwave-sink-iceberg/pom.xml
index d8ef3d6db384a..404a895e7506f 100644
--- a/java/connector-node/risingwave-sink-iceberg/pom.xml
+++ b/java/connector-node/risingwave-sink-iceberg/pom.xml
@@ -74,6 +74,10 @@
org.apache.hive
hive-metastore
+
+ org.apache.hive
+ hive-exec
+
org.apache.parquet
parquet-avro
diff --git a/java/pom.xml b/java/pom.xml
index 644588c9d6b44..2afd5606a39f0 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -387,6 +387,11 @@
hive-metastore
${hive.version}
+
+ org.apache.hive
+ hive-exec
+ ${hive.version}
+
org.apache.hadoop
hadoop-mapreduce-client-core
diff --git a/proto/expr.proto b/proto/expr.proto
index f4016c3912db4..dedfa3f3cd3b7 100644
--- a/proto/expr.proto
+++ b/proto/expr.proto
@@ -307,6 +307,8 @@ message ExprNode {
HAS_TABLE_PRIVILEGE = 2407;
HAS_ANY_COLUMN_PRIVILEGE = 2408;
HAS_SCHEMA_PRIVILEGE = 2409;
+ PG_IS_IN_RECOVERY = 2411;
+ RW_RECOVERY_STATUS = 2412;
// EXTERNAL
ICEBERG_TRANSFORM = 2201;
diff --git a/proto/meta.proto b/proto/meta.proto
index 0d2a1b8832915..3284887590182 100644
--- a/proto/meta.proto
+++ b/proto/meta.proto
@@ -362,12 +362,26 @@ message ListAllNodesResponse {
repeated common.WorkerNode nodes = 2;
}
+message GetClusterRecoveryStatusRequest {}
+
+enum RecoveryStatus {
+ STATUS_UNSPECIFIED = 0;
+ STATUS_STARTING = 1;
+ STATUS_RECOVERING = 2;
+ STATUS_RUNNING = 3;
+}
+
+message GetClusterRecoveryStatusResponse {
+ RecoveryStatus status = 1;
+}
+
service ClusterService {
rpc AddWorkerNode(AddWorkerNodeRequest) returns (AddWorkerNodeResponse);
rpc ActivateWorkerNode(ActivateWorkerNodeRequest) returns (ActivateWorkerNodeResponse);
rpc DeleteWorkerNode(DeleteWorkerNodeRequest) returns (DeleteWorkerNodeResponse);
rpc UpdateWorkerNodeSchedulability(UpdateWorkerNodeSchedulabilityRequest) returns (UpdateWorkerNodeSchedulabilityResponse);
rpc ListAllNodes(ListAllNodesRequest) returns (ListAllNodesResponse);
+ rpc GetClusterRecoveryStatus(GetClusterRecoveryStatusRequest) returns (GetClusterRecoveryStatusResponse);
}
enum SubscribeType {
diff --git a/proto/stream_service.proto b/proto/stream_service.proto
index 85b12d8ed5fa1..fd97bde853487 100644
--- a/proto/stream_service.proto
+++ b/proto/stream_service.proto
@@ -71,7 +71,8 @@ message BarrierCompleteResponse {
common.Status status = 2;
repeated CreateMviewProgress create_mview_progress = 3;
message GroupedSstableInfo {
- uint64 compaction_group_id = 1;
+ reserved 1;
+ reserved "compaction_group_id";
hummock.SstableInfo sst = 2;
map table_stats_map = 3;
}
diff --git a/src/batch/Cargo.toml b/src/batch/Cargo.toml
index c3042346aff85..ec7091ea882c4 100644
--- a/src/batch/Cargo.toml
+++ b/src/batch/Cargo.toml
@@ -32,7 +32,7 @@ hytra = "0.1.2"
iceberg = { workspace = true }
itertools = { workspace = true }
memcomparable = "0.2"
-opendal = "0.47"
+opendal = { workspace = true }
parking_lot = { workspace = true }
parquet = { workspace = true }
paste = "1"
diff --git a/src/cmd/src/lib.rs b/src/cmd/src/lib.rs
index 0e711458d5196..9017f58606f7c 100644
--- a/src/cmd/src/lib.rs
+++ b/src/cmd/src/lib.rs
@@ -44,8 +44,7 @@ pub fn compute(opts: ComputeNodeOpts) -> ! {
pub fn meta(opts: MetaNodeOpts) -> ! {
init_risingwave_logger(LoggerSettings::from_opts(&opts));
- // TODO(shutdown): pass the shutdown token
- main_okk(|_| risingwave_meta_node::start(opts));
+ main_okk(|shutdown| risingwave_meta_node::start(opts, shutdown));
}
pub fn frontend(opts: FrontendOpts) -> ! {
@@ -55,8 +54,7 @@ pub fn frontend(opts: FrontendOpts) -> ! {
pub fn compactor(opts: CompactorOpts) -> ! {
init_risingwave_logger(LoggerSettings::from_opts(&opts));
- // TODO(shutdown): pass the shutdown token
- main_okk(|_| risingwave_compactor::start(opts));
+ main_okk(|shutdown| risingwave_compactor::start(opts, shutdown));
}
pub fn ctl(opts: CtlOpts) -> ! {
diff --git a/src/cmd_all/src/standalone.rs b/src/cmd_all/src/standalone.rs
index 791921fc07d69..325f2f8ff395b 100644
--- a/src/cmd_all/src/standalone.rs
+++ b/src/cmd_all/src/standalone.rs
@@ -194,9 +194,10 @@ pub async fn standalone(
is_in_memory = matches!(opts.backend, Some(MetaBackend::Mem));
tracing::info!("starting meta-node thread with cli args: {:?}", opts);
+ let shutdown = shutdown.clone();
let _meta_handle = tokio::spawn(async move {
let dangerous_max_idle_secs = opts.dangerous_max_idle_secs;
- risingwave_meta_node::start(opts).await;
+ risingwave_meta_node::start(opts, shutdown).await;
tracing::warn!("meta is stopped, shutdown all nodes");
if let Some(idle_exit_secs) = dangerous_max_idle_secs {
eprintln!("{}",
@@ -230,8 +231,9 @@ pub async fn standalone(
}
if let Some(opts) = compactor_opts {
tracing::info!("starting compactor-node thread with cli args: {:?}", opts);
+ let shutdown = shutdown.clone();
let _compactor_handle =
- tokio::spawn(async move { risingwave_compactor::start(opts).await });
+ tokio::spawn(async move { risingwave_compactor::start(opts, shutdown).await });
}
// wait for log messages to be flushed
diff --git a/src/common/common_service/src/lib.rs b/src/common/common_service/src/lib.rs
index c09c84012819b..2cf9a56e076f3 100644
--- a/src/common/common_service/src/lib.rs
+++ b/src/common/common_service/src/lib.rs
@@ -18,9 +18,13 @@
#![feature(impl_trait_in_assoc_type)]
#![feature(error_generic_member_access)]
-pub mod metrics_manager;
-pub mod observer_manager;
+mod metrics_manager;
+mod observer_manager;
+mod tracing;
pub use metrics_manager::MetricsManager;
-
-pub mod tracing;
+pub use observer_manager::{
+ Channel, NotificationClient, ObserverError, ObserverManager, ObserverState,
+ RpcNotificationClient,
+};
+pub use tracing::TracingExtractLayer;
diff --git a/src/common/common_service/src/observer_manager.rs b/src/common/common_service/src/observer_manager.rs
index e760a0e16866c..bf7e457be8b1c 100644
--- a/src/common/common_service/src/observer_manager.rs
+++ b/src/common/common_service/src/observer_manager.rs
@@ -22,42 +22,6 @@ use thiserror_ext::AsReport;
use tokio::task::JoinHandle;
use tonic::{Status, Streaming};
-pub trait SubscribeTypeEnum {
- fn subscribe_type() -> SubscribeType;
-}
-
-pub struct SubscribeFrontend {}
-
-impl SubscribeTypeEnum for SubscribeFrontend {
- fn subscribe_type() -> SubscribeType {
- SubscribeType::Frontend
- }
-}
-
-pub struct SubscribeHummock {}
-
-impl SubscribeTypeEnum for SubscribeHummock {
- fn subscribe_type() -> SubscribeType {
- SubscribeType::Hummock
- }
-}
-
-pub struct SubscribeCompactor {}
-
-impl SubscribeTypeEnum for SubscribeCompactor {
- fn subscribe_type() -> SubscribeType {
- SubscribeType::Compactor
- }
-}
-
-pub struct SubscribeCompute {}
-
-impl SubscribeTypeEnum for SubscribeCompute {
- fn subscribe_type() -> SubscribeType {
- SubscribeType::Compute
- }
-}
-
/// `ObserverManager` is used to update data based on notification from meta.
/// Call `start` to spawn a new asynchronous task
/// We can write the notification logic by implementing `ObserverNodeImpl`.
@@ -68,7 +32,7 @@ pub struct ObserverManager {
}
pub trait ObserverState: Send + 'static {
- type SubscribeType: SubscribeTypeEnum;
+ fn subscribe_type() -> SubscribeType;
/// modify data after receiving notification from meta
fn handle_notification(&mut self, resp: SubscribeResponse);
@@ -109,10 +73,7 @@ where
S: ObserverState,
{
pub async fn new(client: T, observer_states: S) -> Self {
- let rx = client
- .subscribe(S::SubscribeType::subscribe_type())
- .await
- .unwrap();
+ let rx = client.subscribe(S::subscribe_type()).await.unwrap();
Self {
rx,
client,
@@ -214,11 +175,7 @@ where
/// `re_subscribe` is used to re-subscribe to the meta's notification.
async fn re_subscribe(&mut self) {
loop {
- match self
- .client
- .subscribe(S::SubscribeType::subscribe_type())
- .await
- {
+ match self.client.subscribe(S::subscribe_type()).await {
Ok(rx) => {
tracing::debug!("re-subscribe success");
self.rx = rx;
diff --git a/src/common/metrics/src/monitor/connection.rs b/src/common/metrics/src/monitor/connection.rs
index 295fb6399ba4b..e5774a3f16d7d 100644
--- a/src/common/metrics/src/monitor/connection.rs
+++ b/src/common/metrics/src/monitor/connection.rs
@@ -587,28 +587,6 @@ impl tonic::transport::server::Router {
}
}
-#[cfg(not(madsim))]
-pub fn monitored_tcp_incoming(
- listen_addr: std::net::SocketAddr,
- connection_type: impl Into,
- config: TcpConfig,
-) -> Result<
- MonitoredConnection,
- Box,
-> {
- let incoming = tonic::transport::server::TcpIncoming::new(
- listen_addr,
- config.tcp_nodelay,
- config.keepalive_duration,
- )?;
- Ok(MonitoredConnection::new(
- incoming,
- MonitorNewConnectionImpl {
- connection_type: connection_type.into(),
- },
- ))
-}
-
#[derive(Clone)]
pub struct MonitorNewConnectionImpl {
connection_type: String,
diff --git a/src/common/metrics/src/monitor/mod.rs b/src/common/metrics/src/monitor/mod.rs
index 10b5c966e636a..316cac9ea907c 100644
--- a/src/common/metrics/src/monitor/mod.rs
+++ b/src/common/metrics/src/monitor/mod.rs
@@ -12,20 +12,16 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-pub mod connection;
-pub mod my_stats;
-pub mod process;
-pub mod rwlock;
+pub use connection::{monitor_connector, EndpointExt, RouterExt, TcpConfig};
+pub use rwlock::MonitoredRwLock;
-use std::sync::LazyLock;
+mod connection;
+mod process;
+mod rwlock;
-use prometheus::core::{
- AtomicI64, AtomicU64, Collector, GenericCounter, GenericCounterVec, GenericGauge, Metric,
-};
-use prometheus::{Histogram, HistogramVec, Registry};
+use std::sync::LazyLock;
-use crate::monitor::my_stats::MyHistogram;
-use crate::monitor::process::monitor_process;
+use prometheus::Registry;
#[cfg(target_os = "linux")]
static PAGESIZE: std::sync::LazyLock =
@@ -35,59 +31,8 @@ static PAGESIZE: std::sync::LazyLock =
pub static CLOCK_TICK: std::sync::LazyLock =
std::sync::LazyLock::new(|| unsafe { libc::sysconf(libc::_SC_CLK_TCK) as u64 });
-/// Define extension method `print` used in `print_statistics`.
-pub trait Print {
- fn print(&self);
-}
-
-impl Print for GenericCounter {
- fn print(&self) {
- let desc = &self.desc()[0].fq_name;
- let counter = self.metric().get_counter().get_value() as u64;
- println!("{desc} COUNT : {counter}");
- }
-}
-
-impl Print for GenericGauge {
- fn print(&self) {
- let desc = &self.desc()[0].fq_name;
- let counter = self.get();
- println!("{desc} COUNT : {counter}");
- }
-}
-
-impl Print for Histogram {
- fn print(&self) {
- let desc = &self.desc()[0].fq_name;
-
- let histogram = MyHistogram::from_prom_hist(self.metric().get_histogram());
- let p50 = histogram.get_percentile(50.0);
- let p95 = histogram.get_percentile(95.0);
- let p99 = histogram.get_percentile(99.0);
- let p100 = histogram.get_percentile(100.0);
-
- let sample_count = self.get_sample_count();
- let sample_sum = self.get_sample_sum();
- println!("{desc} P50 : {p50} P95 : {p95} P99 : {p99} P100 : {p100} COUNT : {sample_count} SUM : {sample_sum}");
- }
-}
-
-impl Print for HistogramVec {
- fn print(&self) {
- let desc = &self.desc()[0].fq_name;
- println!("{desc} {:?}", self);
- }
-}
-
-impl Print for GenericCounterVec {
- fn print(&self) {
- let desc = &self.desc()[0].fq_name;
- println!("{desc} {:?}", self);
- }
-}
-
pub static GLOBAL_METRICS_REGISTRY: LazyLock = LazyLock::new(|| {
let registry = Registry::new();
- monitor_process(®istry);
+ process::monitor_process(®istry);
registry
});
diff --git a/src/common/metrics/src/monitor/my_stats.rs b/src/common/metrics/src/monitor/my_stats.rs
deleted file mode 100644
index 52c71167f2f97..0000000000000
--- a/src/common/metrics/src/monitor/my_stats.rs
+++ /dev/null
@@ -1,221 +0,0 @@
-// 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::fmt::{Display, Formatter};
-
-use itertools::Itertools;
-use prometheus::proto::Histogram;
-use rw_iter_util::ZipEqFast;
-
-#[derive(Clone, Default, Debug)]
-pub struct MyHistogram {
- pub upper_bound_list: Vec,
- pub count_list: Vec,
- pub total_count: u64,
- pub total_sum: f64,
-}
-
-impl MyHistogram {
- pub fn from_prom_hist(histogram: &Histogram) -> MyHistogram {
- let mut upper_bound_list = Vec::new();
- let mut count_list = Vec::new();
-
- let total_count = histogram.get_sample_count();
- let total_sum = histogram.get_sample_sum();
-
- let buckets = histogram.get_bucket();
- for bucket in buckets {
- let upper_bound = bucket.get_upper_bound();
- let count = bucket.get_cumulative_count();
- upper_bound_list.push(upper_bound);
- count_list.push(count);
- }
-
- MyHistogram {
- upper_bound_list,
- count_list,
- total_count,
- total_sum,
- }
- }
-
- pub fn from_diff(prev: &MyHistogram, cur: &MyHistogram) -> MyHistogram {
- MyHistogram {
- upper_bound_list: cur.upper_bound_list.clone(),
- count_list: match prev.count_list.is_empty() {
- true => cur.count_list.clone(),
- false => prev
- .count_list
- .iter()
- .zip_eq_fast(cur.count_list.iter())
- .map(|(&pb, &cb)| cb - pb)
- .collect_vec(),
- },
- total_sum: cur.total_sum - prev.total_sum,
- total_count: cur.total_count - prev.total_count,
- }
- }
-
- pub fn get_percentile(&self, p: f64) -> f64 {
- let sample_count = self.total_count;
-
- // empty bucket may appear
- if sample_count == 0 {
- return 0.0;
- }
- let threshold = (sample_count as f64 * (p / 100.0_f64)).ceil() as u64;
- let mut last_upper_bound = 0.0;
- let mut last_count = 0;
- for (&upper_bound, &count) in self
- .upper_bound_list
- .iter()
- .zip_eq_fast(self.count_list.iter())
- {
- if count >= threshold {
- // assume scale linearly within this bucket,
- // return a value between last_upper_bound and upper_bound
- let right_left_diff = upper_bound - last_upper_bound;
- return last_upper_bound
- + right_left_diff * (threshold - last_count) as f64
- / (count - last_count) as f64;
- }
- last_upper_bound = upper_bound;
- last_count = count;
- }
-
- 0.0
- }
-}
-
-impl Display for MyHistogram {
- fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- // calculate latencies statistics
- let mean = self.total_sum / self.total_count as f64;
- let p50 = self.get_percentile(50.0);
- let p90 = self.get_percentile(90.0);
- let p99 = self.get_percentile(99.0);
- let p100 = self.get_percentile(100.0);
-
- write!(
- f,
- "latency:
- mean: {},
- p50: {},
- p90: {},
- p99: {},
- p100: {};",
- mean, p50, p90, p99, p100
- )
- }
-}
-
-#[cfg(test)]
-mod tests {
- use prometheus::core::Metric;
- use prometheus::{histogram_opts, register_histogram_with_registry, Registry};
-
- use super::*;
-
- #[test]
- fn test_proc_histogram_basic() {
- fn new_simple_histogram(upper_bound: u64) -> MyHistogram {
- let registry = Registry::new();
- let buckets = (1..=upper_bound).map(|x| x as f64).collect::>();
- let opts = histogram_opts!("test_histogram", "test_histogram", buckets);
-
- let histogram = register_histogram_with_registry!(opts, registry).unwrap();
-
- for value in 1..=upper_bound {
- histogram.observe(value as f64);
- }
-
- MyHistogram::from_prom_hist(histogram.metric().get_histogram())
- }
-
- let histogram = new_simple_histogram(999);
- assert_eq!(histogram.get_percentile(50.0) as u64, 500);
- assert_eq!(histogram.get_percentile(90.0) as u64, 900);
- assert_eq!(histogram.get_percentile(99.0) as u64, 990);
- assert_eq!(histogram.get_percentile(99.9) as u64, 999);
- assert_eq!(histogram.get_percentile(100.0) as u64, 999);
-
- let histogram = new_simple_histogram(1000);
- assert_eq!(histogram.get_percentile(50.0) as u64, 500);
- assert_eq!(histogram.get_percentile(90.0) as u64, 900);
- assert_eq!(histogram.get_percentile(99.0) as u64, 990);
- assert_eq!(histogram.get_percentile(99.9) as u64, 1000);
- assert_eq!(histogram.get_percentile(100.0) as u64, 1000);
-
- let histogram = new_simple_histogram(9999);
- assert_eq!(histogram.get_percentile(50.0) as u64, 5000);
- assert_eq!(histogram.get_percentile(90.0) as u64, 9000);
- assert_eq!(histogram.get_percentile(99.0) as u64, 9900);
- assert_eq!(histogram.get_percentile(99.9) as u64, 9990);
- assert_eq!(histogram.get_percentile(100.0) as u64, 9999);
- }
-
- #[test]
- fn test_proc_histogram_uneven_distributed() {
- let registry = Registry::new();
- let buckets = vec![
- 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
- ];
- let opts = histogram_opts!("test_histogram", "test_histogram", buckets);
- let histogram = register_histogram_with_registry!(opts, registry).unwrap();
-
- let mut i = 0.005;
- while i < 10.0 {
- histogram.observe(i);
- i += 0.005;
- }
-
- let histogram = MyHistogram::from_prom_hist(histogram.metric().get_histogram());
- assert_eq!(histogram.get_percentile(50.0), 5.0);
- assert_eq!(histogram.get_percentile(90.0), 9.004004004004004);
- assert_eq!(histogram.get_percentile(99.0), 9.904904904904905);
- assert_eq!(histogram.get_percentile(99.9), 9.994994994994995);
- assert_eq!(histogram.get_percentile(100.0), 10.0);
- }
-
- #[test]
- fn test_proc_histogram_realistic() {
- let registry = Registry::new();
- let buckets = vec![
- 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
- ];
- let opts = histogram_opts!("test_histogram", "test_histogram", buckets);
- let histogram = register_histogram_with_registry!(opts, registry).unwrap();
-
- histogram.observe(0.0012);
- histogram.observe(0.0013);
- histogram.observe(0.003);
-
- histogram.observe(0.0132);
- histogram.observe(0.0143);
- histogram.observe(0.0146);
- histogram.observe(0.0249);
-
- histogram.observe(0.99);
-
- histogram.observe(6.11);
- histogram.observe(7.833);
-
- let histogram = MyHistogram::from_prom_hist(histogram.metric().get_histogram());
- assert_eq!(histogram.get_percentile(50.0), 0.0175);
- assert_eq!(histogram.get_percentile(90.0), 7.5);
- assert_eq!(histogram.get_percentile(99.0), 10.00);
- assert_eq!(histogram.get_percentile(99.9), 10.00);
- assert_eq!(histogram.get_percentile(100.0), 10.00);
- }
-}
diff --git a/src/common/src/array/stream_chunk.rs b/src/common/src/array/stream_chunk.rs
index 917d484458518..cfee990561aa1 100644
--- a/src/common/src/array/stream_chunk.rs
+++ b/src/common/src/array/stream_chunk.rs
@@ -525,6 +525,10 @@ impl OpRowMutRef<'_> {
}
impl StreamChunkMut {
+ pub fn capacity(&self) -> usize {
+ self.vis.len()
+ }
+
pub fn vis(&self, i: usize) -> bool {
self.vis.is_set(i)
}
diff --git a/src/common/src/types/mod.rs b/src/common/src/types/mod.rs
index 69e727cbde655..3b02b8c38d020 100644
--- a/src/common/src/types/mod.rs
+++ b/src/common/src/types/mod.rs
@@ -573,6 +573,8 @@ pub trait ScalarRef<'a>: ScalarBounds> + 'a + Copy {
macro_rules! scalar_impl_enum {
($( { $variant_name:ident, $suffix_name:ident, $scalar:ty, $scalar_ref:ty } ),*) => {
/// `ScalarImpl` embeds all possible scalars in the evaluation framework.
+ ///
+ /// See `for_all_variants` for the definition.
#[derive(Debug, Clone, PartialEq, Eq, EstimateSize)]
pub enum ScalarImpl {
$( $variant_name($scalar) ),*
diff --git a/src/compute/src/observer/observer_manager.rs b/src/compute/src/observer/observer_manager.rs
index e4192fb0b6f9c..62e2d699668f4 100644
--- a/src/compute/src/observer/observer_manager.rs
+++ b/src/compute/src/observer/observer_manager.rs
@@ -13,7 +13,7 @@
// limitations under the License.
use risingwave_common::system_param::local_manager::LocalSystemParamsManagerRef;
-use risingwave_common_service::observer_manager::{ObserverState, SubscribeCompute};
+use risingwave_common_service::ObserverState;
use risingwave_pb::meta::subscribe_response::Info;
use risingwave_pb::meta::SubscribeResponse;
@@ -22,7 +22,9 @@ pub struct ComputeObserverNode {
}
impl ObserverState for ComputeObserverNode {
- type SubscribeType = SubscribeCompute;
+ fn subscribe_type() -> risingwave_pb::meta::SubscribeType {
+ risingwave_pb::meta::SubscribeType::Compute
+ }
fn handle_notification(&mut self, resp: SubscribeResponse) {
let Some(info) = resp.info.as_ref() else {
diff --git a/src/compute/src/server.rs b/src/compute/src/server.rs
index 3b0466b6cc0d1..d7dcbd5146c31 100644
--- a/src/compute/src/server.rs
+++ b/src/compute/src/server.rs
@@ -28,7 +28,7 @@ use risingwave_common::config::{
MAX_CONNECTION_WINDOW_SIZE, STREAM_WINDOW_SIZE,
};
use risingwave_common::lru::init_global_sequencer_args;
-use risingwave_common::monitor::connection::{RouterExt, TcpConfig};
+use risingwave_common::monitor::{RouterExt, TcpConfig};
use risingwave_common::system_param::local_manager::LocalSystemParamsManager;
use risingwave_common::system_param::reader::SystemParamsRead;
use risingwave_common::telemetry::manager::TelemetryManager;
@@ -38,9 +38,7 @@ use risingwave_common::util::pretty_bytes::convert;
use risingwave_common::util::tokio_util::sync::CancellationToken;
use risingwave_common::{GIT_SHA, RW_VERSION};
use risingwave_common_heap_profiling::HeapProfiler;
-use risingwave_common_service::metrics_manager::MetricsManager;
-use risingwave_common_service::observer_manager::ObserverManager;
-use risingwave_common_service::tracing::TracingExtractLayer;
+use risingwave_common_service::{MetricsManager, ObserverManager, TracingExtractLayer};
use risingwave_connector::source::monitor::GLOBAL_SOURCE_METRICS;
use risingwave_dml::dml_manager::DmlManager;
use risingwave_pb::common::WorkerType;
diff --git a/src/connector/Cargo.toml b/src/connector/Cargo.toml
index 86491ae464a3f..34f8761073454 100644
--- a/src/connector/Cargo.toml
+++ b/src/connector/Cargo.toml
@@ -84,7 +84,7 @@ mysql_common = { version = "0.32", default-features = false, features = [
] }
nexmark = { version = "0.2", features = ["serde"] }
num-bigint = "0.4"
-opendal = { version = "0.47", features = [
+opendal = { workspace = true, features = [
"executors-tokio",
"services-fs",
"services-gcs",
diff --git a/src/connector/codec/src/decoder/avro/mod.rs b/src/connector/codec/src/decoder/avro/mod.rs
index cdd9aea416c8f..2cd6cf5ac77c3 100644
--- a/src/connector/codec/src/decoder/avro/mod.rs
+++ b/src/connector/codec/src/decoder/avro/mod.rs
@@ -15,7 +15,7 @@
mod schema;
use std::sync::LazyLock;
-use apache_avro::schema::{DecimalSchema, RecordSchema};
+use apache_avro::schema::{DecimalSchema, RecordSchema, UnionSchema};
use apache_avro::types::{Value, ValueKind};
use apache_avro::{Decimal as AvroDecimal, Schema};
use chrono::Datelike;
@@ -33,6 +33,7 @@ use risingwave_common::util::iter_util::ZipEqFast;
pub use self::schema::{avro_schema_to_column_descs, MapHandling, ResolvedAvroSchema};
use super::utils::extract_decimal;
use super::{bail_uncategorized, uncategorized, Access, AccessError, AccessResult};
+use crate::decoder::avro::schema::avro_schema_to_struct_field_name;
#[derive(Clone)]
/// Options for parsing an `AvroValue` into Datum, with an optional avro schema.
@@ -107,6 +108,54 @@ impl<'a> AvroParseOptions<'a> {
let v: ScalarImpl = match (type_expected, value) {
(_, Value::Null) => return Ok(DatumCow::NULL),
+ // ---- Union (with >=2 non null variants), and nullable Union ([null, record]) -----
+ (DataType::Struct(struct_type_info), Value::Union(variant, v)) => {
+ let Some(Schema::Union(u)) = self.schema else {
+ // XXX: Is this branch actually unreachable? (if self.schema is correctly used)
+ return Err(create_error());
+ };
+
+ if let Some(inner) = get_nullable_union_inner(u) {
+ // nullable Union ([null, record])
+ return Self {
+ schema: Some(inner),
+ relax_numeric: self.relax_numeric,
+ }
+ .convert_to_datum(v, type_expected);
+ }
+ let variant_schema = &u.variants()[*variant as usize];
+
+ if matches!(variant_schema, &Schema::Null) {
+ return Ok(DatumCow::NULL);
+ }
+
+ // Here we compare the field name, instead of using the variant idx to find the field idx.
+ // The latter approach might also work, but might be more error-prone.
+ // We will need to get the index of the "null" variant, and then re-map the variant index to the field index.
+ // XXX: probably we can unwrap here (if self.schema is correctly used)
+ let expected_field_name = avro_schema_to_struct_field_name(variant_schema)?;
+
+ let mut fields = Vec::with_capacity(struct_type_info.len());
+ for (field_name, field_type) in struct_type_info
+ .names()
+ .zip_eq_fast(struct_type_info.types())
+ {
+ if field_name == expected_field_name {
+ let datum = Self {
+ schema: Some(variant_schema),
+ relax_numeric: self.relax_numeric,
+ }
+ .convert_to_datum(v, field_type)?
+ .to_owned_datum();
+
+ fields.push(datum)
+ } else {
+ fields.push(None)
+ }
+ }
+ StructValue::new(fields).into()
+ }
+ // nullable Union ([null, T])
(_, Value::Union(_, v)) => {
let schema = self.extract_inner_schema(None);
return Self {
@@ -290,6 +339,12 @@ impl Access for AvroAccess<'_> {
let mut value = self.value;
let mut options: AvroParseOptions<'_> = self.options.clone();
+ debug_assert!(
+ path.len() == 1
+ || (path.len() == 2 && matches!(path[0], "before" | "after" | "source")),
+ "unexpected path access: {:?}",
+ path
+ );
let mut i = 0;
while i < path.len() {
let key = path[i];
@@ -299,6 +354,29 @@ impl Access for AvroAccess<'_> {
};
match value {
Value::Union(_, v) => {
+ // The debezium "before" field is a nullable union.
+ // "fields": [
+ // {
+ // "name": "before",
+ // "type": [
+ // "null",
+ // {
+ // "type": "record",
+ // "name": "Value",
+ // "fields": [...],
+ // }
+ // ],
+ // "default": null
+ // },
+ // {
+ // "name": "after",
+ // "type": [
+ // "null",
+ // "Value"
+ // ],
+ // "default": null
+ // },
+ // ...]
value = v;
options.schema = options.extract_inner_schema(None);
continue;
@@ -338,18 +416,30 @@ pub(crate) fn avro_decimal_to_rust_decimal(
))
}
-pub fn avro_schema_skip_union(schema: &Schema) -> anyhow::Result<&Schema> {
+/// If the union schema is `[null, T]` or `[T, null]`, returns `Some(T)`; otherwise returns `None`.
+fn get_nullable_union_inner(union_schema: &UnionSchema) -> Option<&'_ Schema> {
+ let variants = union_schema.variants();
+ // Note: `[null, null] is invalid`, we don't need to worry about that.
+ if variants.len() == 2 && variants.contains(&Schema::Null) {
+ let inner_schema = variants
+ .iter()
+ .find(|s| !matches!(s, &&Schema::Null))
+ .unwrap();
+ Some(inner_schema)
+ } else {
+ None
+ }
+}
+
+pub fn avro_schema_skip_nullable_union(schema: &Schema) -> anyhow::Result<&Schema> {
match schema {
- Schema::Union(union_schema) => {
- let inner_schema = union_schema
- .variants()
- .iter()
- .find(|s| !matches!(s, &&Schema::Null))
- .ok_or_else(|| {
- anyhow::format_err!("illegal avro record schema {:?}", union_schema)
- })?;
- Ok(inner_schema)
- }
+ Schema::Union(union_schema) => match get_nullable_union_inner(union_schema) {
+ Some(s) => Ok(s),
+ None => Err(anyhow::format_err!(
+ "illegal avro union schema, expected [null, T], got {:?}",
+ union_schema
+ )),
+ },
other => Ok(other),
}
}
@@ -372,7 +462,9 @@ pub fn avro_extract_field_schema<'a>(
Ok(&field.schema)
}
Schema::Array(schema) => Ok(schema),
- Schema::Union(_) => avro_schema_skip_union(schema),
+ // Only nullable union should be handled here.
+ // We will not extract inner schema for real union (and it's not extractable).
+ Schema::Union(_) => avro_schema_skip_nullable_union(schema),
Schema::Map(schema) => Ok(schema),
_ => bail!("avro schema does not have inner item, schema: {:?}", schema),
}
@@ -476,11 +568,337 @@ pub(crate) fn avro_to_jsonb(avro: &Value, builder: &mut jsonbb::Builder) -> Acce
mod tests {
use std::str::FromStr;
- use apache_avro::Decimal as AvroDecimal;
+ use apache_avro::{from_avro_datum, Decimal as AvroDecimal};
+ use expect_test::expect;
use risingwave_common::types::{Datum, Decimal};
use super::*;
+ /// Test the behavior of the Rust Avro lib for handling union with logical type.
+ #[test]
+ fn test_avro_lib_union() {
+ // duplicate types
+ let s = Schema::parse_str(r#"["null", "null"]"#);
+ expect![[r#"
+ Err(
+ Unions cannot contain duplicate types,
+ )
+ "#]]
+ .assert_debug_eq(&s);
+ let s = Schema::parse_str(r#"["int", "int"]"#);
+ expect![[r#"
+ Err(
+ Unions cannot contain duplicate types,
+ )
+ "#]]
+ .assert_debug_eq(&s);
+ // multiple map/array are considered as the same type, regardless of the element type!
+ let s = Schema::parse_str(
+ r#"[
+"null",
+{
+ "type": "map",
+ "values" : "long",
+ "default": {}
+},
+{
+ "type": "map",
+ "values" : "int",
+ "default": {}
+}
+]
+"#,
+ );
+ expect![[r#"
+ Err(
+ Unions cannot contain duplicate types,
+ )
+ "#]]
+ .assert_debug_eq(&s);
+ let s = Schema::parse_str(
+ r#"[
+"null",
+{
+ "type": "array",
+ "items" : "long",
+ "default": {}
+},
+{
+ "type": "array",
+ "items" : "int",
+ "default": {}
+}
+]
+"#,
+ );
+ expect![[r#"
+ Err(
+ Unions cannot contain duplicate types,
+ )
+ "#]]
+ .assert_debug_eq(&s);
+ // multiple named types
+ let s = Schema::parse_str(
+ r#"[
+"null",
+{"type":"fixed","name":"a","size":16},
+{"type":"fixed","name":"b","size":32}
+]
+"#,
+ );
+ expect![[r#"
+ Ok(
+ Union(
+ UnionSchema {
+ schemas: [
+ Null,
+ Fixed(
+ FixedSchema {
+ name: Name {
+ name: "a",
+ namespace: None,
+ },
+ aliases: None,
+ doc: None,
+ size: 16,
+ attributes: {},
+ },
+ ),
+ Fixed(
+ FixedSchema {
+ name: Name {
+ name: "b",
+ namespace: None,
+ },
+ aliases: None,
+ doc: None,
+ size: 32,
+ attributes: {},
+ },
+ ),
+ ],
+ variant_index: {
+ Null: 0,
+ },
+ },
+ ),
+ )
+ "#]]
+ .assert_debug_eq(&s);
+
+ // union in union
+ let s = Schema::parse_str(r#"["int", ["null", "int"]]"#);
+ expect![[r#"
+ Err(
+ Unions may not directly contain a union,
+ )
+ "#]]
+ .assert_debug_eq(&s);
+
+ // logical type
+ let s = Schema::parse_str(r#"["null", {"type":"string","logicalType":"uuid"}]"#).unwrap();
+ expect![[r#"
+ Union(
+ UnionSchema {
+ schemas: [
+ Null,
+ Uuid,
+ ],
+ variant_index: {
+ Null: 0,
+ Uuid: 1,
+ },
+ },
+ )
+ "#]]
+ .assert_debug_eq(&s);
+ // Note: Java Avro lib rejects this (logical type unions with its physical type)
+ let s = Schema::parse_str(r#"["string", {"type":"string","logicalType":"uuid"}]"#).unwrap();
+ expect![[r#"
+ Union(
+ UnionSchema {
+ schemas: [
+ String,
+ Uuid,
+ ],
+ variant_index: {
+ String: 0,
+ Uuid: 1,
+ },
+ },
+ )
+ "#]]
+ .assert_debug_eq(&s);
+ // Note: Java Avro lib rejects this (logical type unions with its physical type)
+ let s = Schema::parse_str(r#"["int", {"type":"int", "logicalType": "date"}]"#).unwrap();
+ expect![[r#"
+ Union(
+ UnionSchema {
+ schemas: [
+ Int,
+ Date,
+ ],
+ variant_index: {
+ Int: 0,
+ Date: 1,
+ },
+ },
+ )
+ "#]]
+ .assert_debug_eq(&s);
+ // Note: Java Avro lib allows this (2 decimal with different "name")
+ let s = Schema::parse_str(
+ r#"[
+{"type":"fixed","name":"Decimal128","size":16,"logicalType":"decimal","precision":38,"scale":2},
+{"type":"fixed","name":"Decimal256","size":32,"logicalType":"decimal","precision":50,"scale":2}
+]"#,
+ );
+ expect![[r#"
+ Err(
+ Unions cannot contain duplicate types,
+ )
+ "#]]
+ .assert_debug_eq(&s);
+ }
+
+ #[test]
+ fn test_avro_lib_union_record_bug() {
+ // multiple named types (record)
+ let s = Schema::parse_str(
+ r#"
+ {
+ "type": "record",
+ "name": "Root",
+ "fields": [
+ {
+ "name": "unionTypeComplex",
+ "type": [
+ "null",
+ {"type": "record", "name": "Email","fields": [{"name":"inner","type":"string"}]},
+ {"type": "record", "name": "Fax","fields": [{"name":"inner","type":"int"}]},
+ {"type": "record", "name": "Sms","fields": [{"name":"inner","type":"int"}]}
+ ]
+ }
+ ]
+ }
+ "#,
+ )
+ .unwrap();
+
+ let bytes = hex::decode("060c").unwrap();
+ // Correct should be variant 3 (Sms)
+ let correct_value = from_avro_datum(&s, &mut bytes.as_slice(), None);
+ expect![[r#"
+ Ok(
+ Record(
+ [
+ (
+ "unionTypeComplex",
+ Union(
+ 3,
+ Record(
+ [
+ (
+ "inner",
+ Int(
+ 6,
+ ),
+ ),
+ ],
+ ),
+ ),
+ ),
+ ],
+ ),
+ )
+ "#]]
+ .assert_debug_eq(&correct_value);
+ // Bug: We got variant 2 (Fax) here, if we pass the reader schema.
+ let wrong_value = from_avro_datum(&s, &mut bytes.as_slice(), Some(&s));
+ expect![[r#"
+ Ok(
+ Record(
+ [
+ (
+ "unionTypeComplex",
+ Union(
+ 2,
+ Record(
+ [
+ (
+ "inner",
+ Int(
+ 6,
+ ),
+ ),
+ ],
+ ),
+ ),
+ ),
+ ],
+ ),
+ )
+ "#]]
+ .assert_debug_eq(&wrong_value);
+
+ // The bug below can explain what happened.
+ // The two records below are actually incompatible: https://avro.apache.org/docs/1.11.1/specification/_print/#schema-resolution
+ // > both schemas are records with the _same (unqualified) name_
+ // In from_avro_datum, it first reads the value with the writer schema, and then
+ // it just uses the reader schema to interpret the value.
+ // The value doesn't have record "name" information. So it wrongly passed the conversion.
+ // The correct way is that we need to use both the writer and reader schema in the second step to interpret the value.
+
+ let s = Schema::parse_str(
+ r#"
+ {
+ "type": "record",
+ "name": "Root",
+ "fields": [
+ {
+ "name": "a",
+ "type": "int"
+ }
+ ]
+ }
+ "#,
+ )
+ .unwrap();
+ let s2 = Schema::parse_str(
+ r#"
+{
+ "type": "record",
+ "name": "Root222",
+ "fields": [
+ {
+ "name": "a",
+ "type": "int"
+ }
+ ]
+}
+ "#,
+ )
+ .unwrap();
+
+ let bytes = hex::decode("0c").unwrap();
+ let value = from_avro_datum(&s, &mut bytes.as_slice(), Some(&s2));
+ expect![[r#"
+ Ok(
+ Record(
+ [
+ (
+ "a",
+ Int(
+ 6,
+ ),
+ ),
+ ],
+ ),
+ )
+ "#]]
+ .assert_debug_eq(&value);
+ }
+
#[test]
fn test_convert_decimal() {
// 280
diff --git a/src/connector/codec/src/decoder/avro/schema.rs b/src/connector/codec/src/decoder/avro/schema.rs
index fe96495d089ea..324b7fd426a56 100644
--- a/src/connector/codec/src/decoder/avro/schema.rs
+++ b/src/connector/codec/src/decoder/avro/schema.rs
@@ -14,14 +14,18 @@
use std::sync::{Arc, LazyLock};
+use anyhow::Context;
use apache_avro::schema::{DecimalSchema, RecordSchema, ResolvedSchema, Schema};
use apache_avro::AvroResult;
use itertools::Itertools;
-use risingwave_common::bail;
+use risingwave_common::error::NotImplemented;
use risingwave_common::log::LogSuppresser;
use risingwave_common::types::{DataType, Decimal};
+use risingwave_common::{bail, bail_not_implemented};
use risingwave_pb::plan_common::{AdditionalColumn, ColumnDesc, ColumnDescVersion};
+use super::get_nullable_union_inner;
+
/// Avro schema with `Ref` inlined. The newtype is used to indicate whether the schema is resolved.
///
/// TODO: Actually most of the place should use resolved schema, but currently they just happen to work (Some edge cases are not met yet).
@@ -198,20 +202,46 @@ fn avro_type_mapping(
DataType::List(Box::new(item_type))
}
Schema::Union(union_schema) => {
- // We only support using union to represent nullable fields, not general unions.
- let variants = union_schema.variants();
- if variants.len() != 2 || !variants.contains(&Schema::Null) {
- bail!(
- "unsupported Avro type, only unions like [null, T] is supported: {:?}",
- schema
- );
- }
- let nested_schema = variants
- .iter()
- .find_or_first(|s| !matches!(s, Schema::Null))
- .unwrap();
+ // Note: Unions may not immediately contain other unions. So a `null` must represent a top-level null.
+ // e.g., ["null", ["null", "string"]] is not allowed
+
+ // Note: Unions may not contain more than one schema with the same type, except for the named types record, fixed and enum.
+ // https://avro.apache.org/docs/1.11.1/specification/_print/#unions
+ debug_assert!(
+ union_schema
+ .variants()
+ .iter()
+ .map(Schema::canonical_form) // Schema doesn't implement Eq, but only PartialEq.
+ .duplicates()
+ .next()
+ .is_none(),
+ "Union contains duplicate types: {union_schema:?}",
+ );
+ match get_nullable_union_inner(union_schema) {
+ Some(inner) => avro_type_mapping(inner, map_handling)?,
+ None => {
+ // Convert the union to a struct, each field of the struct represents a variant of the union.
+ // Refer to https://github.com/risingwavelabs/risingwave/issues/16273#issuecomment-2179761345 to see why it's not perfect.
+ // Note: Avro union's variant tag is type name, not field name (unlike Rust enum, or Protobuf oneof).
+
+ // XXX: do we need to introduce union.handling.mode?
+ let (fields, field_names) = union_schema
+ .variants()
+ .iter()
+ // null will mean the whole struct is null
+ .filter(|variant| !matches!(variant, &&Schema::Null))
+ .map(|variant| {
+ avro_type_mapping(variant, map_handling).and_then(|t| {
+ let name = avro_schema_to_struct_field_name(variant)?;
+ Ok((t, name))
+ })
+ })
+ .process_results(|it| it.unzip::<_, _, Vec<_>, Vec<_>>())
+ .context("failed to convert Avro union to struct")?;
- avro_type_mapping(nested_schema, map_handling)?
+ DataType::new_struct(fields, field_names)
+ }
+ }
}
Schema::Ref { name } => {
if name.name == DBZ_VARIABLE_SCALE_DECIMAL_NAME
@@ -219,7 +249,7 @@ fn avro_type_mapping(
{
DataType::Decimal
} else {
- bail!("unsupported Avro type: {:?}", schema);
+ bail_not_implemented!("Avro type: {:?}", schema);
}
}
Schema::Map(value_schema) => {
@@ -229,20 +259,25 @@ fn avro_type_mapping(
if supported_avro_to_json_type(value_schema) {
DataType::Jsonb
} else {
- bail!(
- "unsupported Avro type, cannot convert map to jsonb: {:?}",
+ bail_not_implemented!(
+ issue = 16963,
+ "Avro map type to jsonb: {:?}",
schema
- )
+ );
}
}
None => {
+ // We require it to be specified, because we don't want to have a bad default behavior.
+ // But perhaps changing the default behavior won't be a breaking change,
+ // because it affects only on creation time, what the result ColumnDesc will be, and the ColumnDesc will be persisted.
+ // This is unlike timestamp.handing.mode, which affects parser's behavior on the runtime.
bail!("`map.handling.mode` not specified in ENCODE AVRO (...). Currently supported modes: `jsonb`")
}
}
}
Schema::Uuid => DataType::Varchar,
Schema::Null | Schema::Fixed(_) => {
- bail!("unsupported Avro type: {:?}", schema)
+ bail_not_implemented!("Avro type: {:?}", schema);
}
};
@@ -280,3 +315,71 @@ fn supported_avro_to_json_type(schema: &Schema) -> bool {
| Schema::Union(_) => false,
}
}
+
+/// The field name when converting Avro union type to RisingWave struct type.
+pub(super) fn avro_schema_to_struct_field_name(schema: &Schema) -> Result {
+ Ok(match schema {
+ Schema::Null => unreachable!(),
+ Schema::Union(_) => unreachable!(),
+ // Primitive types
+ Schema::Boolean => "boolean".to_string(),
+ Schema::Int => "int".to_string(),
+ Schema::Long => "long".to_string(),
+ Schema::Float => "float".to_string(),
+ Schema::Double => "double".to_string(),
+ Schema::Bytes => "bytes".to_string(),
+ Schema::String => "string".to_string(),
+ // Unnamed Complex types
+ Schema::Array(_) => "array".to_string(),
+ Schema::Map(_) => "map".to_string(),
+ // Named Complex types
+ Schema::Enum(_) | Schema::Ref { name: _ } | Schema::Fixed(_) | Schema::Record(_) => {
+ // schema.name().unwrap().fullname(None)
+ // See test_avro_lib_union_record_bug
+ // https://github.com/risingwavelabs/risingwave/issues/17632
+ bail_not_implemented!(issue=17632, "Avro named type used in Union type: {:?}", schema)
+
+ }
+
+ // Logical types are currently banned. See https://github.com/risingwavelabs/risingwave/issues/17616
+
+/*
+ Schema::Uuid => "uuid".to_string(),
+ // Decimal is the most tricky. https://avro.apache.org/docs/1.11.1/specification/_print/#decimal
+ // - A decimal logical type annotates Avro bytes _or_ fixed types.
+ // - It has attributes `precision` and `scale`.
+ // "For the purposes of schema resolution, two schemas that are decimal logical types match if their scales and precisions match."
+ // - When the physical type is fixed, it's a named type. And a schema containing 2 decimals is possible:
+ // [
+ // {"type":"fixed","name":"Decimal128","size":16,"logicalType":"decimal","precision":38,"scale":2},
+ // {"type":"fixed","name":"Decimal256","size":32,"logicalType":"decimal","precision":50,"scale":2}
+ // ]
+ // In this case (a logical type's physical type is a named type), perhaps we should use the physical type's `name`.
+ Schema::Decimal(_) => "decimal".to_string(),
+ Schema::Date => "date".to_string(),
+ // Note: in Avro, the name style is "time-millis", etc.
+ // But in RisingWave (Postgres), it will require users to use quotes, i.e.,
+ // select (struct)."time-millis", (struct).time_millies from t;
+ // The latter might be more user-friendly.
+ Schema::TimeMillis => "time_millis".to_string(),
+ Schema::TimeMicros => "time_micros".to_string(),
+ Schema::TimestampMillis => "timestamp_millis".to_string(),
+ Schema::TimestampMicros => "timestamp_micros".to_string(),
+ Schema::LocalTimestampMillis => "local_timestamp_millis".to_string(),
+ Schema::LocalTimestampMicros => "local_timestamp_micros".to_string(),
+ Schema::Duration => "duration".to_string(),
+*/
+ Schema::Uuid
+ | Schema::Decimal(_)
+ | Schema::Date
+ | Schema::TimeMillis
+ | Schema::TimeMicros
+ | Schema::TimestampMillis
+ | Schema::TimestampMicros
+ | Schema::LocalTimestampMillis
+ | Schema::LocalTimestampMicros
+ | Schema::Duration => {
+ bail_not_implemented!(issue=17616, "Avro logicalType used in Union type: {:?}", schema)
+ }
+ })
+}
diff --git a/src/connector/codec/src/decoder/mod.rs b/src/connector/codec/src/decoder/mod.rs
index c7e04ab210a6e..cd7fe14ab74ea 100644
--- a/src/connector/codec/src/decoder/mod.rs
+++ b/src/connector/codec/src/decoder/mod.rs
@@ -16,6 +16,7 @@ pub mod avro;
pub mod json;
pub mod utils;
+use risingwave_common::error::NotImplemented;
use risingwave_common::types::{DataType, Datum, DatumCow, ToOwnedDatum};
use thiserror::Error;
use thiserror_ext::Macro;
@@ -40,23 +41,41 @@ pub enum AccessError {
/// Errors that are not categorized into variants above.
#[error("{message}")]
Uncategorized { message: String },
+
+ #[error(transparent)]
+ NotImplemented(#[from] NotImplemented),
}
pub type AccessResult = std::result::Result;
/// Access to a field in the data structure. Created by `AccessBuilder`.
+///
+/// It's the `ENCODE ...` part in `FORMAT ... ENCODE ...`
pub trait Access {
/// Accesses `path` in the data structure (*parsed* Avro/JSON/Protobuf data),
/// and then converts it to RisingWave `Datum`.
+ ///
/// `type_expected` might or might not be used during the conversion depending on the implementation.
///
/// # Path
///
- /// We usually expect the data is a record (struct), and `path` represents field path.
+ /// We usually expect the data (`Access` instance) is a record (struct), and `path` represents field path.
/// The data (or part of the data) represents the whole row (`Vec`),
/// and we use different `path` to access one column at a time.
///
- /// e.g., for Avro, we access `["col_name"]`; for Debezium Avro, we access `["before", "col_name"]`.
+ /// TODO: the meaning of `path` is a little confusing and maybe over-abstracted.
+ /// `access` does not need to serve arbitrarily deep `path` access, but just "top-level" access.
+ /// The API creates an illusion that arbitrary access is supported, but it's not.
+ /// Perhapts we should separate out another trait like `ToDatum`,
+ /// which only does type mapping, without caring about the path. And `path` itself is only an `enum` instead of `&[&str]`.
+ ///
+ /// What `path` to access is decided by the CDC layer, i.e., the `FORMAT ...` part (`ChangeEvent`).
+ /// e.g.,
+ /// - `DebeziumChangeEvent` accesses `["before", "col_name"]` for value,
+ /// `["source", "db"]`, `["source", "table"]` etc. for additional columns' values,
+ /// `["op"]` for op type.
+ /// - `MaxwellChangeEvent` accesses `["data", "col_name"]` for value, `["type"]` for op type.
+ /// - In the simplest case, for `FORMAT PLAIN/UPSERT` (`KvEvent`), they just access `["col_name"]` for value, and op type is derived.
///
/// # Returns
///
diff --git a/src/connector/codec/tests/integration_tests/avro.rs b/src/connector/codec/tests/integration_tests/avro.rs
index fab143b2bf9e7..11421c151d7a5 100644
--- a/src/connector/codec/tests/integration_tests/avro.rs
+++ b/src/connector/codec/tests/integration_tests/avro.rs
@@ -12,12 +12,14 @@
// See the License for the specific language governing permissions and
// limitations under the License.
+use anyhow::Context;
use apache_avro::from_avro_datum;
use risingwave_connector_codec::decoder::avro::{
avro_schema_to_column_descs, AvroAccess, AvroParseOptions, MapHandling, ResolvedAvroSchema,
};
use risingwave_connector_codec::decoder::Access;
use risingwave_connector_codec::AvroSchema;
+use thiserror_ext::AsReport;
use crate::utils::*;
@@ -44,6 +46,24 @@ struct Config {
data_encoding: TestDataEncoding,
}
+fn avro_schema_str_to_risingwave_schema(
+ avro_schema: &str,
+ config: &Config,
+) -> anyhow::Result<(ResolvedAvroSchema, Vec)> {
+ // manually implement some logic in AvroParserConfig::map_to_columns
+ let avro_schema = AvroSchema::parse_str(avro_schema).context("failed to parse Avro schema")?;
+ let resolved_schema =
+ ResolvedAvroSchema::create(avro_schema.into()).context("failed to resolve Avro schema")?;
+
+ let rw_schema =
+ avro_schema_to_column_descs(&resolved_schema.resolved_schema, config.map_handling)
+ .context("failed to convert Avro schema to RisingWave schema")?
+ .iter()
+ .map(ColumnDesc::from)
+ .collect_vec();
+ Ok((resolved_schema, rw_schema))
+}
+
/// Data driven testing for converting Avro Schema to RisingWave Schema, and then converting Avro data into RisingWave data.
///
/// The expected results can be automatically updated. To run and update the tests:
@@ -79,17 +99,15 @@ fn check(
expected_risingwave_schema: expect_test::Expect,
expected_risingwave_data: expect_test::Expect,
) {
- // manually implement some logic in AvroParserConfig::map_to_columns
- let avro_schema = AvroSchema::parse_str(avro_schema).expect("failed to parse Avro schema");
- let resolved_schema =
- ResolvedAvroSchema::create(avro_schema.into()).expect("failed to resolve Avro schema");
-
- let rw_schema =
- avro_schema_to_column_descs(&resolved_schema.resolved_schema, config.map_handling)
- .expect("failed to convert Avro schema to RisingWave schema")
- .iter()
- .map(ColumnDesc::from)
- .collect_vec();
+ let (resolved_schema, rw_schema) =
+ match avro_schema_str_to_risingwave_schema(avro_schema, &config) {
+ Ok(res) => res,
+ Err(e) => {
+ expected_risingwave_schema.assert_eq(&format!("{}", e.as_report()));
+ expected_risingwave_data.assert_eq("");
+ return;
+ }
+ };
expected_risingwave_schema.assert_eq(&format!(
"{:#?}",
rw_schema.iter().map(ColumnDescTestDisplay).collect_vec()
@@ -554,3 +572,316 @@ fn test_1() {
Owned(Float64(OrderedFloat(NaN)))"#]],
);
}
+
+#[test]
+fn test_union() {
+ // A basic test
+ check(
+ r#"
+{
+ "type": "record",
+ "name": "Root",
+ "fields": [
+ {
+ "name": "unionType",
+ "type": ["int", "string"]
+ },
+ {
+ "name": "unionTypeComplex",
+ "type": [
+ "null",
+ {"type": "record", "name": "Email","fields": [{"name":"inner","type":"string"}]},
+ {"type": "record", "name": "Fax","fields": [{"name":"inner","type":"int"}]},
+ {"type": "record", "name": "Sms","fields": [{"name":"inner","type":"int"}]}
+ ]
+ },
+ {
+ "name": "nullableString",
+ "type": ["null", "string"]
+ }
+ ]
+}
+ "#,
+ &[
+ // {
+ // "unionType": {"int": 114514},
+ // "unionTypeComplex": {"Sms": {"inner":6}},
+ // "nullableString": null
+ // }
+ "00a4fd0d060c00",
+ // {
+ // "unionType": {"int": 114514},
+ // "unionTypeComplex": {"Fax": {"inner":6}},
+ // "nullableString": null
+ // }
+ "00a4fd0d040c00",
+ // {
+ // "unionType": {"string": "oops"},
+ // "unionTypeComplex": null,
+ // "nullableString": {"string": "hello"}
+ // }
+ "02086f6f707300020a68656c6c6f",
+ // {
+ // "unionType": {"string": "oops"},
+ // "unionTypeComplex": {"Email": {"inner":"a@b.c"}},
+ // "nullableString": null
+ // }
+ "02086f6f7073020a6140622e6300",
+ ],
+ Config {
+ map_handling: None,
+ data_encoding: TestDataEncoding::HexBinary,
+ },
+ // FIXME: why the struct type doesn't have field_descs? https://github.com/risingwavelabs/risingwave/issues/17128
+ expect![[r#"
+ failed to convert Avro schema to RisingWave schema: failed to convert Avro union to struct: Feature is not yet implemented: Avro named type used in Union type: Record(RecordSchema { name: Name { name: "Email", namespace: None }, aliases: None, doc: None, fields: [RecordField { name: "inner", doc: None, aliases: None, default: None, schema: String, order: Ascending, position: 0, custom_attributes: {} }], lookup: {"inner": 0}, attributes: {} })
+ Tracking issue: https://github.com/risingwavelabs/risingwave/issues/17632"#]],
+ expect![""],
+ );
+
+ // logicalType is currently rejected
+ // https://github.com/risingwavelabs/risingwave/issues/17616
+ check(
+ r#"
+{
+"type": "record",
+"name": "Root",
+"fields": [
+ {
+ "name": "unionLogical",
+ "type": ["int", {"type":"int", "logicalType": "date"}]
+ }
+]
+}
+ "#,
+ &[],
+ Config {
+ map_handling: None,
+ data_encoding: TestDataEncoding::HexBinary,
+ },
+ expect![[r#"
+ failed to convert Avro schema to RisingWave schema: failed to convert Avro union to struct: Feature is not yet implemented: Avro logicalType used in Union type: Date
+ Tracking issue: https://github.com/risingwavelabs/risingwave/issues/17616"#]],
+ expect![""],
+ );
+
+ // test named type. Consider namespace.
+ // https://avro.apache.org/docs/1.11.1/specification/_print/#names
+ // List of things to take care:
+ // - Record fields and enum symbols DO NOT have namespace.
+ // - If the name specified contains a dot, then it is assumed to be a fullname, and any namespace also specified is IGNORED.
+ // - If a name doesn't have its own namespace, it will look for its most tightly enclosing named schema.
+ check(
+ r#"
+{
+ "type": "record",
+ "name": "Root",
+ "namespace": "RootNamespace",
+ "fields": [
+ {
+ "name": "littleFieldToMakeNestingLooksBetter",
+ "type": ["null","int"], "default": null
+ },
+ {
+ "name": "recordField",
+ "type": ["null", "int", {
+ "type": "record",
+ "name": "my.name.spaced.record",
+ "namespace": "when.name.contains.dot.namespace.is.ignored",
+ "fields": [
+ {"name": "hello", "type": {"type": "int", "default": 1}},
+ {"name": "world", "type": {"type": "double", "default": 1}}
+ ]
+ }],
+ "default": null
+ },
+ {
+ "name": "enumField",
+ "type": ["null", "int", {
+ "type": "enum",
+ "name": "myEnum",
+ "namespace": "my.namespace",
+ "symbols": ["A", "B", "C", "D"]
+ }],
+ "default": null
+ },
+ {
+ "name": "anotherEnumFieldUsingRootNamespace",
+ "type": ["null", "int", {
+ "type": "enum",
+ "name": "myEnum",
+ "symbols": ["A", "B", "C", "D"]
+ }],
+ "default": null
+ }
+ ]
+}
+"#,
+ &[
+ // {
+ // "enumField":{"my.namespace.myEnum":"A"},
+ // "anotherEnumFieldUsingRootNamespace":{"RootNamespace.myEnum": "D"}
+ // }
+ "000004000406",
+ ],
+ Config {
+ map_handling: None,
+ data_encoding: TestDataEncoding::HexBinary,
+ },
+ expect![[r#"
+ failed to convert Avro schema to RisingWave schema: failed to convert Avro union to struct: Feature is not yet implemented: Avro named type used in Union type: Record(RecordSchema { name: Name { name: "record", namespace: Some("my.name.spaced") }, aliases: None, doc: None, fields: [RecordField { name: "hello", doc: None, aliases: None, default: None, schema: Int, order: Ascending, position: 0, custom_attributes: {} }, RecordField { name: "world", doc: None, aliases: None, default: None, schema: Double, order: Ascending, position: 1, custom_attributes: {} }], lookup: {"hello": 0, "world": 1}, attributes: {} })
+ Tracking issue: https://github.com/risingwavelabs/risingwave/issues/17632"#]],
+ expect![""],
+ );
+
+ // This is provided by a user https://github.com/risingwavelabs/risingwave/issues/16273#issuecomment-2051480710
+ check(
+ r#"
+{
+ "namespace": "com.abc.efg.mqtt",
+ "name": "also.DataMessage",
+ "type": "record",
+ "fields": [
+ {
+ "name": "metrics",
+ "type": {
+ "type": "array",
+ "items": {
+ "name": "also_data_metric",
+ "type": "record",
+ "fields": [
+ {
+ "name": "id",
+ "type": "string"
+ },
+ {
+ "name": "name",
+ "type": "string"
+ },
+ {
+ "name": "norm_name",
+ "type": [
+ "null",
+ "string"
+ ],
+ "default": null
+ },
+ {
+ "name": "uom",
+ "type": [
+ "null",
+ "string"
+ ],
+ "default": null
+ },
+ {
+ "name": "data",
+ "type": {
+ "type": "array",
+ "items": {
+ "name": "dataItem",
+ "type": "record",
+ "fields": [
+ {
+ "name": "ts",
+ "type": "string",
+ "doc": "Timestamp of the metric."
+ },
+ {
+ "name": "value",
+ "type": [
+ "null",
+ "boolean",
+ "double",
+ "string"
+ ],
+ "doc": "Value of the metric."
+ }
+ ]
+ }
+ },
+ "doc": "The data message"
+ }
+ ],
+ "doc": "A metric object"
+ }
+ },
+ "doc": "A list of metrics."
+ }
+ ]
+}
+ "#,
+ &[
+ // {
+ // "metrics": [
+ // {"id":"foo", "name":"a", "data": [] }
+ // ]
+ // }
+ "0206666f6f026100000000",
+ // {
+ // "metrics": [
+ // {"id":"foo", "name":"a", "norm_name": null, "uom": {"string":"c"}, "data": [{"ts":"1", "value":null}, {"ts":"2", "value": {"boolean": true }}] }
+ // ]
+ // }
+ "0206666f6f02610002026304023100023202010000",
+ ],
+ Config {
+ map_handling: None,
+ data_encoding: TestDataEncoding::HexBinary,
+ },
+ expect![[r#"
+ [
+ metrics(#1): List(
+ Struct {
+ id: Varchar,
+ name: Varchar,
+ norm_name: Varchar,
+ uom: Varchar,
+ data: List(
+ Struct {
+ ts: Varchar,
+ value: Struct {
+ boolean: Boolean,
+ double: Float64,
+ string: Varchar,
+ },
+ },
+ ),
+ },
+ ),
+ ]"#]],
+ expect![[r#"
+ Owned([
+ StructValue(
+ Utf8("foo"),
+ Utf8("a"),
+ null,
+ null,
+ [],
+ ),
+ ])
+ ----
+ Owned([
+ StructValue(
+ Utf8("foo"),
+ Utf8("a"),
+ null,
+ Utf8("c"),
+ [
+ StructValue(
+ Utf8("1"),
+ null,
+ ),
+ StructValue(
+ Utf8("2"),
+ StructValue(
+ Bool(true),
+ null,
+ null,
+ ),
+ ),
+ ],
+ ),
+ ])"#]],
+ );
+}
diff --git a/src/connector/src/parser/debezium/avro_parser.rs b/src/connector/src/parser/debezium/avro_parser.rs
index 1a40b87c9d498..04f80ebba1ca1 100644
--- a/src/connector/src/parser/debezium/avro_parser.rs
+++ b/src/connector/src/parser/debezium/avro_parser.rs
@@ -19,8 +19,8 @@ use apache_avro::types::Value;
use apache_avro::{from_avro_datum, Schema};
use risingwave_common::try_match_expand;
use risingwave_connector_codec::decoder::avro::{
- avro_extract_field_schema, avro_schema_skip_union, avro_schema_to_column_descs, AvroAccess,
- AvroParseOptions, ResolvedAvroSchema,
+ avro_extract_field_schema, avro_schema_skip_nullable_union, avro_schema_to_column_descs,
+ AvroAccess, AvroParseOptions, ResolvedAvroSchema,
};
use risingwave_pb::catalog::PbSchemaRegistryNameStrategy;
use risingwave_pb::plan_common::ColumnDesc;
@@ -125,8 +125,40 @@ impl DebeziumAvroParserConfig {
}
pub fn map_to_columns(&self) -> ConnectorResult> {
+ // Refer to debezium_avro_msg_schema.avsc for how the schema looks like:
+
+ // "fields": [
+ // {
+ // "name": "before",
+ // "type": [
+ // "null",
+ // {
+ // "type": "record",
+ // "name": "Value",
+ // "fields": [...],
+ // }
+ // ],
+ // "default": null
+ // },
+ // {
+ // "name": "after",
+ // "type": [
+ // "null",
+ // "Value"
+ // ],
+ // "default": null
+ // },
+ // ...]
+
+ // Other fields are:
+ // - source: describes the source metadata for the event
+ // - op
+ // - ts_ms
+ // - transaction
+ // See
+
avro_schema_to_column_descs(
- avro_schema_skip_union(avro_extract_field_schema(
+ avro_schema_skip_nullable_union(avro_extract_field_schema(
// FIXME: use resolved schema here.
// Currently it works because "after" refers to a subtree in "before",
// but in theory, inside "before" there could also be a reference.
@@ -227,7 +259,7 @@ mod tests {
let outer_schema = get_outer_schema();
let expected_inner_schema = Schema::parse_str(inner_shema_str).unwrap();
- let extracted_inner_schema = avro_schema_skip_union(
+ let extracted_inner_schema = avro_schema_skip_nullable_union(
avro_extract_field_schema(&outer_schema, Some("before")).unwrap(),
)
.unwrap();
@@ -318,7 +350,7 @@ mod tests {
fn test_map_to_columns() {
let outer_schema = get_outer_schema();
let columns = avro_schema_to_column_descs(
- avro_schema_skip_union(
+ avro_schema_skip_nullable_union(
avro_extract_field_schema(&outer_schema, Some("before")).unwrap(),
)
.unwrap(),
diff --git a/src/connector/src/parser/unified/json.rs b/src/connector/src/parser/unified/json.rs
index e4a229bb61b98..ca709e2eebc73 100644
--- a/src/connector/src/parser/unified/json.rs
+++ b/src/connector/src/parser/unified/json.rs
@@ -534,7 +534,7 @@ impl JsonParseOptions {
(DataType::Struct(struct_type_info), ValueType::Object) => {
// Collecting into a Result> doesn't reserve the capacity in advance, so we `Vec::with_capacity` instead.
// https://github.com/rust-lang/rust/issues/48994
- let mut fields = Vec::with_capacity(struct_type_info.types().len());
+ let mut fields = Vec::with_capacity(struct_type_info.len());
for (field_name, field_type) in struct_type_info
.names()
.zip_eq_fast(struct_type_info.types())
diff --git a/src/connector/src/sink/mod.rs b/src/connector/src/sink/mod.rs
index 872e7bbfeaf7c..a1a993803568b 100644
--- a/src/connector/src/sink/mod.rs
+++ b/src/connector/src/sink/mod.rs
@@ -38,7 +38,6 @@ pub mod pulsar;
pub mod redis;
pub mod remote;
pub mod snowflake;
-pub mod snowflake_connector;
pub mod sqlserver;
pub mod starrocks;
pub mod test_sink;
diff --git a/src/connector/src/sink/snowflake.rs b/src/connector/src/sink/snowflake.rs
index 0f512933cd0f7..1c9d67352247c 100644
--- a/src/connector/src/sink/snowflake.rs
+++ b/src/connector/src/sink/snowflake.rs
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-use std::collections::{BTreeMap, HashMap};
+use std::collections::BTreeMap;
use std::fmt::Write;
use std::sync::Arc;
@@ -22,7 +22,11 @@ use bytes::{Bytes, BytesMut};
use risingwave_common::array::{Op, StreamChunk};
use risingwave_common::bitmap::Bitmap;
use risingwave_common::catalog::Schema;
-use risingwave_object_store::object::{ObjectStore, OpendalStreamingUploader, StreamingUploader};
+use risingwave_common::config::ObjectStoreConfig;
+use risingwave_object_store::object::object_metrics::GLOBAL_OBJECT_STORE_METRICS;
+use risingwave_object_store::object::{
+ ObjectStore, OpendalObjectStore, OpendalStreamingUploader, StreamingUploader,
+};
use serde::Deserialize;
use serde_json::Value;
use serde_with::serde_as;
@@ -32,70 +36,43 @@ use with_options::WithOptions;
use super::encoder::{
JsonEncoder, RowEncoder, TimeHandlingMode, TimestampHandlingMode, TimestamptzHandlingMode,
};
-use super::snowflake_connector::{generate_s3_file_name, SnowflakeHttpClient, SnowflakeS3Client};
use super::writer::LogSinkerOf;
use super::{SinkError, SinkParam};
use crate::sink::writer::SinkWriterExt;
use crate::sink::{DummySinkCommitCoordinator, Result, Sink, SinkWriter, SinkWriterParam};
pub const SNOWFLAKE_SINK: &str = "snowflake";
+const S3_INTERMEDIATE_FILE_NAME: &str = "RW_SNOWFLAKE_S3_SINK_FILE";
-#[derive(Deserialize, Debug, Clone, WithOptions)]
+#[derive(Debug, Clone, Deserialize, WithOptions)]
pub struct SnowflakeCommon {
- /// The snowflake database used for sinking
- #[serde(rename = "snowflake.database")]
- pub database: String,
-
- /// The corresponding schema where sink table exists
- #[serde(rename = "snowflake.schema")]
- pub schema: String,
-
- /// The created pipe object, will be used as `insertFiles` target
- #[serde(rename = "snowflake.pipe")]
- pub pipe: String,
-
- /// The unique, snowflake provided `account_identifier`
- /// NOTE: please use the form `-`
- /// For detailed guidance, reference:
- #[serde(rename = "snowflake.account_identifier")]
- pub account_identifier: String,
-
- /// The user that owns the table to be sinked
- /// NOTE: the user should've been granted corresponding *role*
- /// reference:
- #[serde(rename = "snowflake.user")]
- pub user: String,
-
- /// The public key fingerprint used when generating custom `jwt_token`
- /// reference:
- #[serde(rename = "snowflake.rsa_public_key_fp")]
- pub rsa_public_key_fp: String,
-
- /// The rsa pem key *without* encryption
- #[serde(rename = "snowflake.private_key")]
- pub private_key: String,
-
/// The s3 bucket where intermediate sink files will be stored
- #[serde(rename = "snowflake.s3_bucket")]
+ #[serde(rename = "snowflake.s3_bucket", alias = "s3.bucket_name")]
pub s3_bucket: String,
/// The optional s3 path to be specified
/// the actual file location would be `s3:////`
/// if this field is specified by user(s)
/// otherwise it would be `s3:///`
- #[serde(rename = "snowflake.s3_path")]
+ #[serde(rename = "snowflake.s3_path", alias = "s3.path")]
pub s3_path: Option,
/// s3 credentials
- #[serde(rename = "snowflake.aws_access_key_id")]
+ #[serde(
+ rename = "snowflake.aws_access_key_id",
+ alias = "s3.credentials.access"
+ )]
pub aws_access_key_id: String,
/// s3 credentials
- #[serde(rename = "snowflake.aws_secret_access_key")]
+ #[serde(
+ rename = "snowflake.aws_secret_access_key",
+ alias = "s3.credentials.secret"
+ )]
pub aws_secret_access_key: String,
/// The s3 region, e.g., us-east-2
- #[serde(rename = "snowflake.aws_region")]
+ #[serde(rename = "snowflake.aws_region", alias = "s3.region_name")]
pub aws_region: String,
}
@@ -173,8 +150,6 @@ pub struct SnowflakeSinkWriter {
pk_indices: Vec,
#[expect(dead_code)]
is_append_only: bool,
- /// the client used to send `insertFiles` post request
- http_client: SnowflakeHttpClient,
/// the client to insert file to external storage (i.e., s3)
s3_client: SnowflakeS3Client,
row_encoder: JsonEncoder,
@@ -187,7 +162,7 @@ pub struct SnowflakeSinkWriter {
/// note: the option here *implicitly* indicates whether we have at
/// least call `streaming_upload` once during this epoch,
/// which is mainly used to prevent uploading empty data.
- streaming_uploader: Option<(OpendalStreamingUploader, String)>,
+ streaming_uploader: Option,
}
impl SnowflakeSinkWriter {
@@ -197,18 +172,6 @@ impl SnowflakeSinkWriter {
pk_indices: Vec,
is_append_only: bool,
) -> Result {
- let http_client = SnowflakeHttpClient::new(
- config.common.account_identifier.clone(),
- config.common.user.clone(),
- config.common.database.clone(),
- config.common.schema.clone(),
- config.common.pipe.clone(),
- config.common.rsa_public_key_fp.clone(),
- config.common.private_key.clone(),
- HashMap::new(),
- config.common.s3_path.clone(),
- );
-
let s3_client = SnowflakeS3Client::new(
config.common.s3_bucket.clone(),
config.common.s3_path.clone(),
@@ -222,7 +185,6 @@ impl SnowflakeSinkWriter {
schema: schema.clone(),
pk_indices,
is_append_only,
- http_client,
s3_client,
row_encoder: JsonEncoder::new(
schema,
@@ -245,7 +207,7 @@ impl SnowflakeSinkWriter {
/// and `streaming_upload` being called the first time.
/// i.e., lazily initialization of the internal `streaming_uploader`.
/// plus, this function is *pure*, the `&mut self` here is to make rustc (and tokio) happy.
- async fn new_streaming_uploader(&mut self) -> Result<(OpendalStreamingUploader, String)> {
+ async fn new_streaming_uploader(&mut self) -> Result {
let file_suffix = self.file_suffix();
let path = generate_s3_file_name(self.s3_client.s3_path(), &file_suffix);
let uploader = self
@@ -260,12 +222,12 @@ impl SnowflakeSinkWriter {
)
})
.map_err(SinkError::Snowflake)?;
- Ok((uploader, file_suffix))
+ Ok(uploader)
}
/// write data to the current streaming uploader for this epoch.
async fn streaming_upload(&mut self, data: Bytes) -> Result<()> {
- let (uploader, _) = match self.streaming_uploader.as_mut() {
+ let uploader = match self.streaming_uploader.as_mut() {
Some(s) => s,
None => {
assert!(
@@ -286,18 +248,18 @@ impl SnowflakeSinkWriter {
/// finalize streaming upload for this epoch.
/// ensure all the data has been properly uploaded to intermediate s3.
- async fn finish_streaming_upload(&mut self) -> Result> {
+ async fn finish_streaming_upload(&mut self) -> Result<()> {
let uploader = std::mem::take(&mut self.streaming_uploader);
- let Some((uploader, file_suffix)) = uploader else {
+ let Some(uploader) = uploader else {
// there is no data to be uploaded for this epoch
- return Ok(None);
+ return Ok(());
};
uploader
.finish()
.await
.context("failed to finish streaming upload to s3")
.map_err(SinkError::Snowflake)?;
- Ok(Some(file_suffix))
+ Ok(())
}
async fn append_only(&mut self, chunk: StreamChunk) -> Result<()> {
@@ -344,13 +306,7 @@ impl SnowflakeSinkWriter {
// note that after `finish_streaming_upload`, do *not* interact with
// `streaming_uploader` until new data comes in at next epoch,
// since the ownership has been taken in this method, and `None` will be left.
- let Some(file_suffix) = self.finish_streaming_upload().await? else {
- // represents there is no data to be uploaded for this epoch
- return Ok(());
- };
- // trigger `insertFiles` post request to snowflake
- self.http_client.send_request(&file_suffix).await?;
- Ok(())
+ self.finish_streaming_upload().await
}
}
@@ -384,3 +340,57 @@ impl SinkWriter for SnowflakeSinkWriter {
Ok(())
}
}
+
+/// The helper function to generate the *global unique* s3 file name.
+pub(crate) fn generate_s3_file_name(s3_path: Option<&str>, suffix: &str) -> String {
+ match s3_path {
+ Some(path) => format!("{}/{}_{}", path, S3_INTERMEDIATE_FILE_NAME, suffix),
+ None => format!("{}_{}", S3_INTERMEDIATE_FILE_NAME, suffix),
+ }
+}
+
+/// todo: refactor this part after s3 sink is available
+pub struct SnowflakeS3Client {
+ #[expect(dead_code)]
+ s3_bucket: String,
+ s3_path: Option,
+ pub opendal_s3_engine: OpendalObjectStore,
+}
+
+impl SnowflakeS3Client {
+ pub fn new(
+ s3_bucket: String,
+ s3_path: Option,
+ aws_access_key_id: String,
+ aws_secret_access_key: String,
+ aws_region: String,
+ ) -> Result {
+ // FIXME: we should use the `ObjectStoreConfig` instead of default
+ // just use default configuration here for opendal s3 engine
+ let config = ObjectStoreConfig::default();
+
+ let metrics = Arc::new(GLOBAL_OBJECT_STORE_METRICS.clone());
+
+ // create the s3 engine for streaming upload to the intermediate s3 bucket
+ let opendal_s3_engine = OpendalObjectStore::new_s3_engine_with_credentials(
+ &s3_bucket,
+ Arc::new(config),
+ metrics,
+ &aws_access_key_id,
+ &aws_secret_access_key,
+ &aws_region,
+ )
+ .context("failed to create opendal s3 engine")
+ .map_err(SinkError::Snowflake)?;
+
+ Ok(Self {
+ s3_bucket,
+ s3_path,
+ opendal_s3_engine,
+ })
+ }
+
+ pub fn s3_path(&self) -> Option<&str> {
+ self.s3_path.as_deref()
+ }
+}
diff --git a/src/connector/src/sink/snowflake_connector.rs b/src/connector/src/sink/snowflake_connector.rs
deleted file mode 100644
index bfd2458900294..0000000000000
--- a/src/connector/src/sink/snowflake_connector.rs
+++ /dev/null
@@ -1,225 +0,0 @@
-// 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::time::{SystemTime, UNIX_EPOCH};
-
-use anyhow::{anyhow, Context};
-use jsonwebtoken::{encode, Algorithm, EncodingKey, Header};
-use object_metrics::GLOBAL_OBJECT_STORE_METRICS;
-use reqwest::{header, Client, RequestBuilder, StatusCode};
-use risingwave_object_store::object::*;
-use serde::{Deserialize, Serialize};
-
-use super::doris_starrocks_connector::POOL_IDLE_TIMEOUT;
-use super::{Result, SinkError};
-
-const SNOWFLAKE_HOST_ADDR: &str = "snowflakecomputing.com";
-const SNOWFLAKE_REQUEST_ID: &str = "RW_SNOWFLAKE_SINK";
-const S3_INTERMEDIATE_FILE_NAME: &str = "RW_SNOWFLAKE_S3_SINK_FILE";
-
-/// The helper function to generate the *global unique* s3 file name.
-pub(crate) fn generate_s3_file_name(s3_path: Option<&str>, suffix: &str) -> String {
- match s3_path {
- Some(path) => format!("{}/{}_{}", path, S3_INTERMEDIATE_FILE_NAME, suffix),
- None => format!("{}_{}", S3_INTERMEDIATE_FILE_NAME, suffix),
- }
-}
-
-/// Claims is used when constructing `jwt_token`
-/// with payload specified.
-/// reference:
-#[derive(Debug, Serialize, Deserialize)]
-struct Claims {
- iss: String,
- sub: String,
- iat: usize,
- exp: usize,
-}
-
-#[derive(Debug)]
-pub struct SnowflakeHttpClient {
- url: String,
- rsa_public_key_fp: String,
- account: String,
- user: String,
- private_key: String,
- #[expect(dead_code)]
- header: HashMap,
- s3_path: Option,
-}
-
-impl SnowflakeHttpClient {
- pub fn new(
- account: String,
- user: String,
- db: String,
- schema: String,
- pipe: String,
- rsa_public_key_fp: String,
- private_key: String,
- header: HashMap,
- s3_path: Option,
- ) -> Self {
- // todo: ensure if we need user to *explicitly* provide the `request_id`
- // currently it seems that this is not important.
- // reference to the snowpipe rest api is as below, i.e.,
- //
- let url = format!(
- "https://{}.{}/v1/data/pipes/{}.{}.{}/insertFiles?requestId={}",
- account.clone(),
- SNOWFLAKE_HOST_ADDR,
- db,
- schema,
- pipe,
- SNOWFLAKE_REQUEST_ID
- );
-
- Self {
- url,
- rsa_public_key_fp,
- account,
- user,
- private_key,
- header,
- s3_path,
- }
- }
-
- /// Generate a 59-minutes valid `jwt_token` for authentication of snowflake side
- /// And please note that we will NOT strictly counting the time interval
- /// of `jwt_token` expiration.
- /// Which essentially means that this method should be called *every time* we want
- /// to send `insertFiles` request to snowflake server.
- fn generate_jwt_token(&self) -> Result {
- let header = Header::new(Algorithm::RS256);
- let now = SystemTime::now()
- .duration_since(UNIX_EPOCH)
- .unwrap()
- .as_secs() as usize;
- let lifetime = 59 * 60;
-
- // Ensure the account and username are uppercase
- let account = self.account.to_uppercase();
- let user = self.user.to_uppercase();
-
- // Construct the fully qualified username
- let qualified_username = format!("{}.{}", account, user);
-
- let claims = Claims {
- iss: format!("{}.{}", qualified_username.clone(), self.rsa_public_key_fp),
- sub: qualified_username,
- iat: now,
- exp: now + lifetime,
- };
-
- let jwt_token = encode(
- &header,
- &claims,
- &EncodingKey::from_rsa_pem(self.private_key.as_ref())
- .context("failed to encode from provided rsa pem key")
- .map_err(SinkError::Snowflake)?,
- )
- .context("failed to encode jwt_token")
- .map_err(SinkError::Snowflake)?;
- Ok(jwt_token)
- }
-
- fn build_request_and_client(&self) -> RequestBuilder {
- let client = Client::builder()
- .pool_idle_timeout(POOL_IDLE_TIMEOUT)
- .build()
- .unwrap();
-
- client.post(&self.url)
- }
-
- /// NOTE: this function should ONLY be called *after*
- /// uploading files to remote external staged storage, i.e., AWS S3
- pub async fn send_request(&self, file_suffix: &str) -> Result<()> {
- let builder = self.build_request_and_client();
-
- // Generate the jwt_token
- let jwt_token = self.generate_jwt_token()?;
- let builder = builder
- .header(header::CONTENT_TYPE, "text/plain")
- .header("Authorization", format!("Bearer {}", jwt_token))
- .header(
- "X-Snowflake-Authorization-Token-Type".to_string(),
- "KEYPAIR_JWT",
- )
- .body(generate_s3_file_name(self.s3_path.as_deref(), file_suffix));
-
- let response = builder
- .send()
- .await
- .map_err(|err| SinkError::Snowflake(anyhow!(err)))?;
-
- if response.status() != StatusCode::OK {
- return Err(SinkError::Snowflake(anyhow!(
- "failed to make http request, error code: {}\ndetailed response: {:#?}",
- response.status(),
- response,
- )));
- }
-
- Ok(())
- }
-}
-
-/// todo: refactor this part after s3 sink is available
-pub struct SnowflakeS3Client {
- #[expect(dead_code)]
- s3_bucket: String,
- s3_path: Option,
- pub opendal_s3_engine: OpendalObjectStore,
-}
-
-impl SnowflakeS3Client {
- pub fn new(
- s3_bucket: String,
- s3_path: Option,
- aws_access_key_id: String,
- aws_secret_access_key: String,
- aws_region: String,
- ) -> Result {
- // FIXME: we should use the `ObjectStoreConfig` instead of default
- // just use default configuration here for opendal s3 engine
- let config = ObjectStoreConfig::default();
- let metrics = Arc::new(GLOBAL_OBJECT_STORE_METRICS.clone());
-
- // create the s3 engine for streaming upload to the intermediate s3 bucket
- let opendal_s3_engine = OpendalObjectStore::new_s3_engine_with_credentials(
- &s3_bucket,
- Arc::new(config),
- metrics,
- &aws_access_key_id,
- &aws_secret_access_key,
- &aws_region,
- )
- .context("failed to create opendal s3 engine")
- .map_err(SinkError::Snowflake)?;
-
- Ok(Self {
- s3_bucket,
- s3_path,
- opendal_s3_engine,
- })
- }
-
- pub fn s3_path(&self) -> Option<&str> {
- self.s3_path.as_deref()
- }
-}
diff --git a/src/connector/with_options_sink.yaml b/src/connector/with_options_sink.yaml
index 27362900dd054..653acaadaaaf1 100644
--- a/src/connector/with_options_sink.yaml
+++ b/src/connector/with_options_sink.yaml
@@ -749,54 +749,36 @@ RedisConfig:
required: true
SnowflakeConfig:
fields:
- - name: snowflake.database
- field_type: String
- comments: The snowflake database used for sinking
- required: true
- - name: snowflake.schema
- field_type: String
- comments: The corresponding schema where sink table exists
- required: true
- - name: snowflake.pipe
- field_type: String
- comments: The created pipe object, will be used as `insertFiles` target
- required: true
- - name: snowflake.account_identifier
- field_type: String
- comments: 'The unique, snowflake provided `account_identifier` NOTE: please use the form `-` For detailed guidance, reference: '
- required: true
- - name: snowflake.user
- field_type: String
- comments: 'The user that owns the table to be sinked NOTE: the user should''ve been granted corresponding *role* reference: '
- required: true
- - name: snowflake.rsa_public_key_fp
- field_type: String
- comments: 'The public key fingerprint used when generating custom `jwt_token` reference: '
- required: true
- - name: snowflake.private_key
- field_type: String
- comments: The rsa pem key *without* encryption
- required: true
- name: snowflake.s3_bucket
field_type: String
comments: The s3 bucket where intermediate sink files will be stored
required: true
+ alias:
+ - s3.bucket_name
- name: snowflake.s3_path
field_type: String
comments: The optional s3 path to be specified the actual file location would be `s3:////` if this field is specified by user(s) otherwise it would be `s3:///`
required: false
+ alias:
+ - s3.path
- name: snowflake.aws_access_key_id
field_type: String
comments: s3 credentials
required: true
+ alias:
+ - s3.credentials.access
- name: snowflake.aws_secret_access_key
field_type: String
comments: s3 credentials
required: true
+ alias:
+ - s3.credentials.secret
- name: snowflake.aws_region
field_type: String
comments: The s3 region, e.g., us-east-2
required: true
+ alias:
+ - s3.region_name
SqlServerConfig:
fields:
- name: sqlserver.host
diff --git a/src/ctl/src/cmd_impl/meta/migration.rs b/src/ctl/src/cmd_impl/meta/migration.rs
index 614fcd6be225c..0f7a5e7e0c428 100644
--- a/src/ctl/src/cmd_impl/meta/migration.rs
+++ b/src/ctl/src/cmd_impl/meta/migration.rs
@@ -748,7 +748,7 @@ pub async fn migrate(from: EtcdBackend, target: String, force_clean: bool) -> an
id: Set(vd.id as _),
prev_id: Set(vd.prev_id as _),
max_committed_epoch: Set(vd.max_committed_epoch as _),
- safe_epoch: Set(vd.safe_epoch as _),
+ safe_epoch: Set(vd.visible_table_safe_epoch() as _),
trivial_move: Set(vd.trivial_move),
full_version_delta: Set((&vd.to_protobuf()).into()),
})
diff --git a/src/expr/macro/src/context.rs b/src/expr/macro/src/context.rs
index afae444e03c45..47b0638284435 100644
--- a/src/expr/macro/src/context.rs
+++ b/src/expr/macro/src/context.rs
@@ -16,7 +16,7 @@ use itertools::Itertools;
use proc_macro2::TokenStream;
use quote::{quote, quote_spanned, ToTokens};
use syn::parse::{Parse, ParseStream};
-use syn::{Error, FnArg, Ident, ItemFn, Result, Token, Type, Visibility};
+use syn::{Error, FnArg, Ident, ItemFn, Pat, PatType, Result, ReturnType, Token, Type, Visibility};
use crate::utils::extend_vis_with_super;
@@ -138,7 +138,47 @@ pub(super) fn generate_captured_function(
mut user_fn: ItemFn,
) -> Result {
let CaptureContextAttr { captures } = attr;
- let orig_user_fn = user_fn.clone();
+ let is_async = user_fn.sig.asyncness.is_some();
+ let mut orig_user_fn = user_fn.clone();
+ if is_async {
+ // Modify the return type to impl Future