Merge branch 'main' into new-version

e2e_test/batch/functions/session_timezone.slt

@@ -5,13 +5,13 @@ set timezone = "us/pacific";
 query T
 select '2022-01-01'::date::timestamp with time zone;
 ----
-2022-01-01 08:00:00+00:00
+2022-01-01 00:00:00-08:00
 
 # Cast timestamp to timestamptz
 query T
 select '2022-01-01 00:00:00'::timestamp::timestamp with time zone;
 ----
-2022-01-01 08:00:00+00:00
+2022-01-01 00:00:00-08:00
 
 # Cast timestamptz to timestamp
 query T
@@ -47,7 +47,7 @@ t
 query T
 select '2022-01-01 00:00:00'::timestamp with time zone;
 ----
-2022-01-01 08:00:00+00:00
+2022-01-01 00:00:00-08:00
 
 # Cast timestamptz to varchar, should display with timezone information
 query T

e2e_test/ddl/show.slt

@@ -127,6 +127,7 @@ matviewname Varchar
 matviewowner Int32
 definition Varchar
 matviewid Int32
+matviewtimezone Varchar
 matviewgraph Varchar
 
 query TT
@@ -137,4 +138,5 @@ matviewname Varchar
 matviewowner Int32
 definition Varchar
 matviewid Int32
+matviewtimezone Varchar
 matviewgraph Varchar

proto/meta.proto

@@ -84,6 +84,8 @@ message TableFragments {
   map<uint32, Fragment> fragments = 3;
   map<uint32, ActorStatus> actor_status = 4;
   map<uint32, source.ConnectorSplits> actor_splits = 5;
+
+  stream_plan.StreamEnvironment env = 6;
 }
 
 // TODO: remove this when dashboard refactored.
@@ -117,6 +119,7 @@ message ListTableFragmentsResponse {
   }
   message TableFragmentInfo {
     repeated FragmentInfo fragments = 1;
+    stream_plan.StreamEnvironment env = 2;
   }
   map<uint32, TableFragmentInfo> table_fragments = 1;
 }

proto/stream_plan.proto

@@ -575,6 +575,12 @@ enum FragmentTypeFlag {
   CHAIN_NODE = 16;
 }
 
+// The environment associated with a stream plan
+message StreamEnvironment {
+  // The timezone associated with the streaming plan. Only applies to MV for now.
+  string timezone = 1;
+}
+
 message StreamFragmentGraph {
   message StreamFragment {
     // 0-based on frontend, and will be rewritten to global id on meta.
@@ -610,4 +616,5 @@ message StreamFragmentGraph {
 
   repeated uint32 dependent_table_ids = 3;
   uint32 table_ids_cnt = 4;
+  StreamEnvironment env = 5;
 }

src/batch/src/executor/insert.rs

@@ -332,7 +332,6 @@ mod tests {
                 None,
                 ReadOptions {
                     prefix_hint: None,
-                    check_bloom_filter: false,
                     ignore_range_tombstone: false,
                     table_id: Default::default(),
                     retention_seconds: None,

src/batch/src/executor/source.rs

@@ -15,9 +15,10 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use anyhow::anyhow;
 use futures::StreamExt;
 use futures_async_stream::try_stream;
-use risingwave_common::array::DataChunk;
+use risingwave_common::array::{DataChunk, Op, StreamChunk};
 use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, Schema, TableId};
 use risingwave_common::error::ErrorCode::{ConnectorError, ProtocolError};
 use risingwave_common::error::{Result, RwError};
@@ -30,6 +31,7 @@ use risingwave_source::monitor::SourceMetrics;
 use risingwave_source::{SourceColumnDesc, SourceFormat, SourceParserImpl};
 
 use super::Executor;
+use crate::error::BatchError;
 use crate::executor::{BoxedExecutor, BoxedExecutorBuilder, ExecutorBuilder};
 use crate::task::BatchTaskContext;
 
@@ -175,7 +177,7 @@ impl SourceExecutor {
         for chunk in stream {
             match chunk {
                 Ok(chunk) => {
-                    yield chunk.chunk.data_chunk().to_owned();
+                    yield covert_stream_chunk_to_batch_chunk(chunk.chunk)?;
                 }
                 Err(e) => {
                     return Err(e);
@@ -184,3 +186,16 @@ impl SourceExecutor {
         }
     }
 }
+
+fn covert_stream_chunk_to_batch_chunk(chunk: StreamChunk) -> Result<DataChunk> {
+    // chunk read from source must be compact
+    assert!(chunk.data_chunk().visibility().is_none());
+
+    if chunk.ops().iter().any(|op| *op != Op::Insert) {
+        return Err(RwError::from(BatchError::Internal(anyhow!(
+            "Only support insert op in batch source executor"
+        ))));
+    }
+
+    Ok(chunk.data_chunk().clone())
+}

src/common/src/config.rs

@@ -442,7 +442,7 @@ mod default {
         }
 
         pub fn bloom_false_positive() -> f64 {
-            0.01
+            0.001
         }
 
         pub fn share_buffers_sync_parallelism() -> u32 {

src/compute/src/memory_management/memory_manager.rs

@@ -69,9 +69,9 @@ impl GlobalMemoryManager {
         watermark_epoch.store(epoch, Ordering::Relaxed);
     }
 
-    /// Jemalloc is not supported on non-Linux OSs, because tikv-jemalloc is not available.
-    /// See the comments for the macro enable_jemalloc_on_linux!();
     // FIXME: remove such limitation after #7180
+    /// Jemalloc is not supported on Windows, because of tikv-jemalloc's own reasons.
+    /// See the comments for the macro `enable_jemalloc_on_linux!()`
     #[cfg(not(target_os = "linux"))]
     pub async fn run(self: Arc<Self>, _: Arc<BatchManager>, _: Arc<LocalStreamManager>) {}
 

src/config/ci-compaction-test-meta.toml

@@ -13,7 +13,7 @@ checkpoint_frequency = 99999999
 shared_buffer_capacity_mb = 4096
 sstable_size_mb = 256
 block_size_kb = 1024
-bloom_false_positive = 0.01
+bloom_false_positive = 0.001
 data_directory = "hummock_001"
 block_cache_capacity_mb = 4096
 meta_cache_capacity_mb = 1024

src/config/ci-compaction-test.toml

@@ -11,7 +11,7 @@ checkpoint_frequency = 10
 shared_buffer_capacity_mb = 4096
 sstable_size_mb = 256
 block_size_kb = 1024
-bloom_false_positive = 0.01
+bloom_false_positive = 0.001
 data_directory = "hummock_001"
 block_cache_capacity_mb = 4096
 meta_cache_capacity_mb = 1024

src/config/example.toml

@@ -16,7 +16,7 @@ checkpoint_frequency = 10
 shared_buffer_capacity_mb = 4096
 sstable_size_mb = 256
 block_size_kb = 1024
-bloom_false_positive = 0.01
+bloom_false_positive = 0.001
 data_directory = "hummock_001"
 block_cache_capacity_mb = 4096
 meta_cache_capacity_mb = 1024