Repository: risingwavelabs/risingwave
File: src/frontend/src/optimizer/mod.rs (Rust)

Code before:

        if TemporalJoinValidator::exist_dangling_temporal_scan(plan.clone()) {
            return Err(ErrorCode::NotSupported(
                "exist dangling temporal scan".to_owned(),
                "please check your temporal join syntax e.g. consider removing the right outer join if it is being used.".to_owned(),
            ).into());
        }

        if RwTimestampValidator::select_rw_timestamp_in_stream_query(plan.clone()) {
            return Err(ErrorCode::NotSupported(
                "selecting `_rw_timestamp` in a streaming query is not allowed".to_owned(),
                "please run the sql in batch mode or remove the column `_rw_timestamp` from the streaming query".to_owned(),
            ).into());
        }

        let locality_provider_count = LocalityProviderCounter::count(plan.clone());
        if ctx.session_ctx().config().enable_locality_backfill() {
            if locality_provider_count > 5 {
                risingwave_common::license::Feature::LocalityBackfill.check_available()?
            }
        } else if locality_provider_count > 0
            && risingwave_common::license::Feature::LocalityBackfill
                .check_available()
                .is_ok()
        {
            ctx.warn_to_user(
                "This streaming job creation could benefit from the experimental locality backfill. \
                Consider enabling it with `SET enable_locality_backfill = true` for potentially \
                faster backfill performance if there are large data to backfill from upstream(s)."
            );
        }

        Ok(optimized_plan.into_phase(plan))
    }

    /// Generate create index or create materialize view plan.
    fn gen_stream_plan(
        self,
        emit_on_window_close: bool,
        backfill_type: BackfillType,
    ) -> Result<StreamOptimizedLogicalPlanRoot> {
        let ctx = self.plan.ctx();
        let explain_trace = ctx.is_explain_trace();

        let plan = {
            {
                if !ctx
                    .session_ctx()
                    .config()
                    .streaming_allow_jsonb_in_stream_key()
                    && let Some(err) = StreamKeyChecker.visit(self.plan.clone())

Review comment (suggestion):
Actually, locality provider would be created only if `enable_locality_backfill = true`, so when you see `locality_provider_count > 0`, it seems it is enabled.

Code after:

        if TemporalJoinValidator::exist_dangling_temporal_scan(plan.clone()) {
            return Err(ErrorCode::NotSupported(
                "exist dangling temporal scan".to_owned(),
                "please check your temporal join syntax e.g. consider removing the right outer join if it is being used.".to_owned(),
            ).into());
        }

        if RwTimestampValidator::select_rw_timestamp_in_stream_query(plan.clone()) {
            return Err(ErrorCode::NotSupported(
                "selecting `_rw_timestamp` in a streaming query is not allowed".to_owned(),
                "please run the sql in batch mode or remove the column `_rw_timestamp` from the streaming query".to_owned(),
            ).into());
        }

        if LocalityProviderCounter::count(plan.clone()) > 5 {
            // LocalityProviderCounter is non-zero only when locality backfill is enabled.
            assert!(ctx.session_ctx().config().enable_locality_backfill());
            risingwave_common::license::Feature::LocalityBackfill.check_available()?;
        }

        if ctx.missed_locality_providers() > 1
            && risingwave_common::license::Feature::LocalityBackfill
                .check_available()
                .is_ok()
        {
            // missed_locality_providers can only be non-zero when locality backfill is disabled.
            assert!(!ctx.session_ctx().config().enable_locality_backfill());
            ctx.warn_to_user(format!(
                "This streaming job has {} operators that could benefit from locality backfill. \
                Consider enabling it with `SET enable_locality_backfill = true` for potentially \
                faster backfill performance, when existing data volume in upstream(s) is large.",
                ctx.missed_locality_providers()
            ));
        }

        Ok(optimized_plan.into_phase(plan))
    }

    /// Generate create index or create materialize view plan.
    fn gen_stream_plan(
        self,
        emit_on_window_close: bool,
        backfill_type: BackfillType,
    ) -> Result<StreamOptimizedLogicalPlanRoot> {
        let ctx = self.plan.ctx();
        let explain_trace = ctx.is_explain_trace();

        let plan = {
            {
                if !ctx