fix conflicts

Author: kwannoel

Cargo.lock

@@ -36,7 +36,7 @@ use crate::util::epoch::Epoch;
 
 // This is a hack, &'static str is not allowed as a const generics argument.
 // TODO: refine this using the adt_const_params feature.
-const CONFIG_KEYS: [&str; 38] = [
+const CONFIG_KEYS: [&str; 37] = [
     "RW_IMPLICIT_FLUSH",
     "CREATE_COMPACTION_GROUP_FOR_MV",
     "QUERY_MODE",
@@ -74,7 +74,6 @@ const CONFIG_KEYS: [&str; 38] = [
     "RW_STREAMING_RATE_LIMIT",
     "CDC_BACKFILL",
     "RW_STREAMING_OVER_WINDOW_CACHE_POLICY",
-    "STREAMING_ENABLE_ARRANGEMENT_BACKFILL",
 ];
 
 // MUST HAVE 1v1 relationship to CONFIG_KEYS. e.g. CONFIG_KEYS[IMPLICIT_FLUSH] =
@@ -116,7 +115,6 @@ const STANDARD_CONFORMING_STRINGS: usize = 33;
 const RW_STREAMING_RATE_LIMIT: usize = 34;
 const CDC_BACKFILL: usize = 35;
 const STREAMING_OVER_WINDOW_CACHE_POLICY: usize = 36;
-const STREAMING_ENABLE_ARRANGEMENT_BACKFILL: usize = 37;
 
 trait ConfigEntry: Default + for<'a> TryFrom<&'a [&'a str], Error = RwError> {
     fn entry_name() -> &'static str;
@@ -322,8 +320,6 @@ type Timezone = ConfigString<TIMEZONE>;
 type StreamingParallelism = ConfigU64<STREAMING_PARALLELISM, 0>;
 type StreamingEnableDeltaJoin = ConfigBool<STREAMING_ENABLE_DELTA_JOIN, false>;
 type StreamingEnableBushyJoin = ConfigBool<STREAMING_ENABLE_BUSHY_JOIN, true>;
-// TODO: Revert to false before merging.
-type StreamingEnableArrangementBackfill = ConfigBool<STREAMING_ENABLE_ARRANGEMENT_BACKFILL, true>;
 type EnableTwoPhaseAgg = ConfigBool<ENABLE_TWO_PHASE_AGG, true>;
 type ForceTwoPhaseAgg = ConfigBool<FORCE_TWO_PHASE_AGG, false>;
 type EnableSharePlan = ConfigBool<RW_ENABLE_SHARE_PLAN, true>;
@@ -416,9 +412,6 @@ pub struct ConfigMap {
     /// Enable bushy join for streaming queries. Defaults to true.
     streaming_enable_bushy_join: StreamingEnableBushyJoin,
 
-    /// Enable arrangement backfill for streaming queries. Defaults to false.
-    streaming_enable_arrangement_backfill: StreamingEnableArrangementBackfill,
-
     /// Enable join ordering for streaming and batch queries. Defaults to true.
     enable_join_ordering: EnableJoinOrdering,
 
@@ -546,8 +539,6 @@ impl ConfigMap {
             self.streaming_enable_delta_join = val.as_slice().try_into()?;
         } else if key.eq_ignore_ascii_case(StreamingEnableBushyJoin::entry_name()) {
             self.streaming_enable_bushy_join = val.as_slice().try_into()?;
-        } else if key.eq_ignore_ascii_case(StreamingEnableArrangementBackfill::entry_name()) {
-            self.streaming_enable_arrangement_backfill = val.as_slice().try_into()?;
         } else if key.eq_ignore_ascii_case(EnableJoinOrdering::entry_name()) {
             self.enable_join_ordering = val.as_slice().try_into()?;
         } else if key.eq_ignore_ascii_case(EnableTwoPhaseAgg::entry_name()) {
@@ -654,8 +645,6 @@ impl ConfigMap {
             Ok(self.streaming_enable_delta_join.to_string())
         } else if key.eq_ignore_ascii_case(StreamingEnableBushyJoin::entry_name()) {
             Ok(self.streaming_enable_bushy_join.to_string())
-        } else if key.eq_ignore_ascii_case(StreamingEnableArrangementBackfill::entry_name()) {
-            Ok(self.streaming_enable_arrangement_backfill.to_string())
         } else if key.eq_ignore_ascii_case(EnableJoinOrdering::entry_name()) {
             Ok(self.enable_join_ordering.to_string())
         } else if key.eq_ignore_ascii_case(EnableTwoPhaseAgg::entry_name()) {
@@ -788,11 +777,6 @@ impl ConfigMap {
                 setting : self.streaming_enable_bushy_join.to_string(),
                 description: String::from("Enable bushy join in streaming queries.")
             },
-            VariableInfo{
-                name : StreamingEnableArrangementBackfill::entry_name().to_lowercase(),
-                setting : self.streaming_enable_arrangement_backfill.to_string(),
-                description: String::from("Enable arrangement backfill in streaming queries.")
-            },
             VariableInfo{
                 name : EnableJoinOrdering::entry_name().to_lowercase(),
                 setting : self.enable_join_ordering.to_string(),
@@ -975,10 +959,6 @@ impl ConfigMap {
         *self.streaming_enable_bushy_join
     }
 
-    pub fn get_streaming_enable_arrangement_backfill(&self) -> bool {
-        *self.streaming_enable_arrangement_backfill
-    }
-
     pub fn get_enable_join_ordering(&self) -> bool {
         *self.enable_join_ordering
     }

src/common/src/session_config/mod.rs

@@ -549,7 +549,6 @@ mod tests {
             cardinality: None,
             created_at_epoch: None,
             cleaned_by_watermark: false,
-            output_indices: vec![],
         }
     }
 

src/storage/src/filter_key_extractor.rs

@@ -15,7 +15,6 @@
 use std::future::Future;
 use std::ops::Bound;
 use std::sync::atomic::Ordering as MemOrdering;
-use std::time::Duration;
 
 use bytes::Bytes;
 use itertools::Itertools;
@@ -35,6 +34,7 @@ use crate::hummock::event_handler::HummockEvent;
 use crate::hummock::store::memtable::ImmutableMemtable;
 use crate::hummock::store::state_store::LocalHummockStorage;
 use crate::hummock::store::version::read_filter_for_batch;
+use crate::hummock::utils::wait_for_epoch;
 use crate::hummock::{HummockEpoch, HummockError};
 use crate::monitor::StoreLocalStatistic;
 use crate::store::*;
@@ -186,42 +186,8 @@ impl StateStore for HummockStorage {
             }
             _ => return Ok(()),
         };
-        let mut receiver = self.version_update_notifier_tx.subscribe();
-        // avoid unnecessary check in the loop if the value does not change
-        let max_committed_epoch = *receiver.borrow_and_update();
-        if max_committed_epoch >= wait_epoch {
-            return Ok(());
-        }
-        loop {
-            match tokio::time::timeout(Duration::from_secs(30), receiver.changed()).await {
-                Err(elapsed) => {
-                    // The reason that we need to retry here is batch scan in
-                    // chain/rearrange_chain is waiting for an
-                    // uncommitted epoch carried by the CreateMV barrier, which
-                    // can take unbounded time to become committed and propagate
-                    // to the CN. We should consider removing the retry as well as wait_epoch
-                    // for chain/rearrange_chain if we enforce
-                    // chain/rearrange_chain to be scheduled on the same
-                    // CN with the same distribution as the upstream MV.
-                    // See #3845 for more details.
-                    tracing::warn!(
-                        "wait_epoch {:?} timeout when waiting for version update elapsed {:?}s",
-                        wait_epoch,
-                        elapsed
-                    );
-                    continue;
-                }
-                Ok(Err(_)) => {
-                    return Err(HummockError::wait_epoch("tx dropped").into());
-                }
-                Ok(Ok(_)) => {
-                    let max_committed_epoch = *receiver.borrow();
-                    if max_committed_epoch >= wait_epoch {
-                        return Ok(());
-                    }
-                }
-            }
-        }
+        let receiver = self.version_update_notifier_tx.subscribe();
+        wait_for_epoch(receiver, wait_epoch).await
     }
 
     async fn sync(&self, epoch: u64) -> StorageResult<SyncResult> {

src/storage/src/hummock/state_store.rs

@@ -15,7 +15,6 @@
 use std::future::Future;
 use std::ops::Bound;
 use std::sync::Arc;
-use std::time::Duration;
 
 use await_tree::InstrumentAwait;
 use bytes::Bytes;
@@ -40,10 +39,10 @@ use crate::hummock::shared_buffer::shared_buffer_batch::{
 use crate::hummock::store::version::{read_filter_for_local, HummockVersionReader};
 use crate::hummock::utils::{
     cmp_delete_range_left_bounds, do_delete_sanity_check, do_insert_sanity_check,
-    do_update_sanity_check, filter_with_delete_range, ENABLE_SANITY_CHECK,
+    do_update_sanity_check, filter_with_delete_range, wait_for_epoch, ENABLE_SANITY_CHECK,
 };
 use crate::hummock::write_limiter::WriteLimiterRef;
-use crate::hummock::{HummockError, MemoryLimiter, SstableIterator};
+use crate::hummock::{MemoryLimiter, SstableIterator};
 use crate::mem_table::{merge_stream, KeyOp, MemTable};
 use crate::monitor::{HummockStateStoreMetrics, IterLocalMetricsGuard, StoreLocalStatistic};
 use crate::storage_value::StorageValue;
@@ -119,46 +118,9 @@ impl LocalHummockStorage {
             .await
     }
 
-    pub async fn wait_for_epoch(&self, epoch: u64) -> StorageResult<()> {
-        // TODO(kwannoel): Refactor this copy-paste code.
-        // TODO(kwannoel): Also we should take in read options, because only initial snapshot read
-        // will require this.
-        let wait_epoch = epoch;
-        let mut receiver = self.version_update_notifier_tx.subscribe();
-        let max_committed_epoch = *receiver.borrow_and_update();
-        if max_committed_epoch >= wait_epoch {
-            return Ok(());
-        }
-        loop {
-            match tokio::time::timeout(Duration::from_secs(30), receiver.changed()).await {
-                Err(elapsed) => {
-                    // The reason that we need to retry here is batch scan in
-                    // chain/rearrange_chain is waiting for an
-                    // uncommitted epoch carried by the CreateMV barrier, which
-                    // can take unbounded time to become committed and propagate
-                    // to the CN. We should consider removing the retry as well as wait_epoch
-                    // for chain/rearrange_chain if we enforce
-                    // chain/rearrange_chain to be scheduled on the same
-                    // CN with the same distribution as the upstream MV.
-                    // See #3845 for more details.
-                    tracing::warn!(
-                        "wait_epoch {:?} timeout when waiting for version update elapsed {:?}s",
-                        wait_epoch,
-                        elapsed
-                    );
-                    continue;
-                }
-                Ok(Err(_)) => {
-                    return Err(HummockError::wait_epoch("tx dropped").into());
-                }
-                Ok(Ok(_)) => {
-                    let max_committed_epoch = *receiver.borrow();
-                    if max_committed_epoch >= wait_epoch {
-                        return Ok(());
-                    }
-                }
-            }
-        }
+    pub async fn wait_for_epoch(&self, wait_epoch: u64) -> StorageResult<()> {
+        let receiver = self.version_update_notifier_tx.subscribe();
+        wait_for_epoch(receiver, wait_epoch).await
     }
 
     pub async fn iter_inner(

src/storage/src/hummock/store/state_store.rs

@@ -18,15 +18,17 @@ use std::ops::Bound::{Excluded, Included, Unbounded};
 use std::ops::{Bound, RangeBounds};
 use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
 use std::sync::Arc;
+use std::time::Duration;
 
 use bytes::Bytes;
 use risingwave_common::cache::CachePriority;
 use risingwave_common::catalog::{TableId, TableOption};
-use risingwave_hummock_sdk::can_concat;
 use risingwave_hummock_sdk::key::{
     bound_table_key_range, EmptySliceRef, FullKey, TableKey, UserKey,
 };
+use risingwave_hummock_sdk::{can_concat, HummockEpoch};
 use risingwave_pb::hummock::{HummockVersion, SstableInfo};
+use tokio::sync::watch::Receiver;
 use tokio::sync::Notify;
 
 use super::{HummockError, HummockResult};
@@ -550,6 +552,47 @@ pub(crate) fn filter_with_delete_range<'a>(
     })
 }
 
+pub(crate) async fn wait_for_epoch(
+    mut receiver: Receiver<HummockEpoch>,
+    wait_epoch: u64,
+) -> StorageResult<()> {
+    // avoid unnecessary check in the loop if the value does not change
+    let max_committed_epoch = *receiver.borrow_and_update();
+    if max_committed_epoch >= wait_epoch {
+        return Ok(());
+    }
+    loop {
+        match tokio::time::timeout(Duration::from_secs(30), receiver.changed()).await {
+            Err(elapsed) => {
+                // The reason that we need to retry here is batch scan in
+                // chain/rearrange_chain is waiting for an
+                // uncommitted epoch carried by the CreateMV barrier, which
+                // can take unbounded time to become committed and propagate
+                // to the CN. We should consider removing the retry as well as wait_epoch
+                // for chain/rearrange_chain if we enforce
+                // chain/rearrange_chain to be scheduled on the same
+                // CN with the same distribution as the upstream MV.
+                // See #3845 for more details.
+                tracing::warn!(
+                    "wait_epoch {:?} timeout when waiting for version update elapsed {:?}s",
+                    wait_epoch,
+                    elapsed
+                );
+                continue;
+            }
+            Ok(Err(_)) => {
+                return Err(HummockError::wait_epoch("tx dropped").into());
+            }
+            Ok(Ok(_)) => {
+                let max_committed_epoch = *receiver.borrow();
+                if max_committed_epoch >= wait_epoch {
+                    return Ok(());
+                }
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::future::{poll_fn, Future};