risingwavelabs · xxchan · May 7, 2025 · May 7, 2025 · May 7, 2025 · May 7, 2025
diff --git a/Cargo.toml b/Cargo.toml
@@ -165,13 +165,13 @@ otlp-embedded = { git = "https://github.com/risingwavelabs/otlp-embedded", rev =
 prost = { version = "0.13" }
 prost-build = { version = "0.13" }
 # branch dev_rebase_main_20250325
-iceberg = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "9c210b9", features = [
+iceberg = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "53e8240e", features = [
     "storage-s3",
     "storage-gcs",
     "storage-azblob",
 ] }
-iceberg-catalog-rest = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "9c210b9" }
-iceberg-catalog-glue = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "9c210b9" }
+iceberg-catalog-rest = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "53e8240e" }
+iceberg-catalog-glue = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "53e8240e" }
 opendal = "0.49"
 arrow-udf-runtime = "0.7.0"
 clap = { version = "4", features = ["cargo", "derive", "env"] }

diff --git a/ci/scripts/e2e-iceberg-test.sh b/ci/scripts/e2e-iceberg-test.sh
@@ -34,16 +34,20 @@ PGPASSWORD=postgres psql -h db -p 5432 -U postgres -c "DROP DATABASE IF EXISTS m
 risedev ci-start ci-iceberg-test
 
 
+unset BUILDKITE_PARALLEL_JOB
+unset BUILDKITE_PARALLEL_JOB_COUNT
+
+
 echo "--- Running tests"
 cd e2e_test/iceberg
 # Don't remove the `--quiet` option since poetry has a bug when printing output, see
 # https://github.com/python-poetry/poetry/issues/3412
 poetry update --quiet
 poetry run python main.py
 
-echo "--- Running pure slt tests"
-risedev slt './e2e_test/iceberg/test_case/pure_slt/*.slt'
+# echo "--- Running pure slt tests"
+# risedev slt './e2e_test/iceberg/test_case/pure_slt/*.slt'
 
 # Run benchmarks separately (not parallelized)
-echo "--- Running benchmarks"
-poetry run python main.py -t ./benches/predicate_pushdown.toml
+# echo "--- Running benchmarks"
+# poetry run python main.py -t ./benches/predicate_pushdown.toml
diff --git a/ci/workflows/main-cron.yml b/ci/workflows/main-cron.yml
@@ -418,7 +418,7 @@ steps:
           run: iceberg-test-env
       - ./ci/plugins/upload-failure-logs
     timeout_in_minutes: 15
-    parallelism: 2
+    parallelism: 40
     retry: *auto-retry
 
   - label: "e2e java-binding test (release)"

diff --git a/ci/workflows/pull-request.yml b/ci/workflows/pull-request.yml
@@ -246,7 +246,7 @@ steps:
           run: iceberg-test-env
       - ./ci/plugins/upload-failure-logs
     timeout_in_minutes: 22
-    parallelism: 2
+    parallelism: 40
     retry: *auto-retry
 
   - label: "end-to-end pulsar sink test"

diff --git a/e2e_test/iceberg/test_utils/test_result.py b/e2e_test/iceberg/test_utils/test_result.py
@@ -22,11 +22,7 @@ def print_test_summary(results: List[TestResult]):
 
     # Print individual test results
     for result in results:
-        status = (
-            Color.GREEN + "PASSED" + Color.ENDC
-            if result.success
-            else Color.RED + "FAILED" + Color.ENDC
-        )
+        status = "PASSED" if result.success else "FAILED"
         duration_str = f"{result.duration.total_seconds():.2f}s"
         log(
             f"{result.name}: {status} ({duration_str})",

diff --git a/src/common/src/util/deployment.rs b/src/common/src/util/deployment.rs
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::LazyLock;
+
 use enum_as_inner::EnumAsInner;
 
 use super::env_var::env_var_is_true;
@@ -31,12 +33,16 @@ pub enum Deployment {
 impl Deployment {
     /// Returns the deployment environment detected from current environment variables.
     pub fn current() -> Self {
-        if env_var_is_true("RISINGWAVE_CI") {
-            Self::Ci
-        } else if env_var_is_true("RISINGWAVE_CLOUD") {
-            Self::Cloud
-        } else {
-            Self::Other
-        }
+        *CURRENT_DEPLOYMENT
     }
 }
+
+static CURRENT_DEPLOYMENT: LazyLock<Deployment> = LazyLock::new(|| {
+    if env_var_is_true("RISINGWAVE_CI") {
+        Deployment::Ci
+    } else if env_var_is_true("RISINGWAVE_CLOUD") {
+        Deployment::Cloud
+    } else {
+        Deployment::Other
+    }
+});
diff --git a/src/connector/src/source/iceberg/mod.rs b/src/connector/src/source/iceberg/mod.rs
@@ -18,7 +18,7 @@ mod metrics;
 use std::collections::{BinaryHeap, HashMap, HashSet};
 use std::sync::Arc;
 
-use anyhow::anyhow;
+use anyhow::{Context, anyhow};
 use async_trait::async_trait;
 use futures::StreamExt;
 use futures_async_stream::{for_await, try_stream};
@@ -398,7 +398,10 @@ impl IcebergSplitEnumerator {
             .build()
             .map_err(|e| anyhow!(e))?;
 
-        let file_scan_stream = scan.plan_files().await.map_err(|e| anyhow!(e))?;
+        let file_scan_stream = scan
+            .plan_files()
+            .await
+            .context("failed to plan files in list_splits_batch_scan")?;
 
         #[for_await]
         for task in file_scan_stream {
@@ -476,7 +479,10 @@ impl IcebergSplitEnumerator {
             .with_delete_file_processing_enabled(true)
             .build()
             .map_err(|e| anyhow!(e))?;
-        let file_scan_stream = scan.plan_files().await.map_err(|e| anyhow!(e))?;
+        let file_scan_stream = scan
+            .plan_files()
+            .await
+            .context("failed to plan files in list_splits_batch_count_star")?;
 
         #[for_await]
         for task in file_scan_stream {
@@ -504,7 +510,10 @@ impl IcebergSplitEnumerator {
             .with_delete_file_processing_enabled(true)
             .build()
             .map_err(|e| anyhow!(e))?;
-        let file_scan_stream = scan.plan_files().await.map_err(|e| anyhow!(e))?;
+        let file_scan_stream = scan
+            .plan_files()
+            .await
+            .context("failed to plan files in all_delete_parameters")?;
         let schema = scan.snapshot().schema(table.metadata())?;
         let mut equality_ids = vec![];
         let mut have_position_delete = false;

diff --git a/src/utils/pgwire/src/pg_protocol.rs b/src/utils/pgwire/src/pg_protocol.rs
@@ -381,7 +381,19 @@ where
             // Note: all messages will be processed through this code path, making it the
             //       only necessary place to log errors.
             if let Err(error) = &result {
-                tracing::error!(error = %error.as_report(), "error when process message");
+                use risingwave_common::util::deployment::Deployment;
+                if cfg!(debug_assertions) {
+                    // Print backtrace in debug mode.
+                    // Print it here is the last resort.
+                    // It's useful only when:
+                    // - no additional context is added to the error
+                    // - backtrace is captured in the error
+                    // - backtrace is not printed in the middle
+                    tracing::error!(error = ?error.as_report(), "error when process message");
+                    tracing::error!(error = ?error, "Debug error");
+                } else {
+                    tracing::error!(error = %error.as_report(), "error when process message");
+                }
             }
 
             // Log to optionally-enabled target `PGWIRE_QUERY_LOG`.