// Copyright 2026 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 pretty_xmlish::XmlNode;
use risingwave_connector::sink::catalog::desc::SinkDesc;
use risingwave_pb::stream_plan::IcebergWithPkIndexPositionDeleteMergerNode;
use risingwave_pb::stream_plan::stream_node::NodeBody;

use super::stream::prelude::*;
use crate::optimizer::plan_node::expr_visitable::ExprVisitable;
use crate::optimizer::plan_node::utils::{Distill, childless_record};
use crate::optimizer::plan_node::{
    ExprRewritable, PlanBase, PlanTreeNodeUnary, Stream, StreamNode, StreamPlanRef as PlanRef,
};
use crate::optimizer::property::RequiredDist;
use crate::stream_fragmenter::BuildFragmentGraphState;

/// `StreamIcebergWithPkIndexPositionDeleteMerger` is the stateless singleton executor for the Iceberg
/// with pk index sink. It merges delete-position messages from the upstream `WriterExecutor`
/// with existing position deletes and writes the resulting delete file (a V3 Puffin deletion
/// vector or a V2 file-scoped Parquet position-delete file, depending on the table format version).
///
/// This node shardes the input stream by `file_path` (the first column) to ensure all messages
/// for the same file are processed by the same merger instance.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct StreamIcebergWithPkIndexPositionDeleteMerger {
    pub base: PlanBase<Stream>,
    pub input: PlanRef,
    pub sink_desc: SinkDesc,
}

impl StreamIcebergWithPkIndexPositionDeleteMerger {
    pub fn new(input: PlanRef, sink_desc: SinkDesc) -> Self {
        debug_assert_eq!(input.schema().len(), 2); // Expecting `[file_path, position]` from the Writer.
        let input = RequiredDist::shard_by_key(2, &[0])
            .streaming_enforce_if_not_satisfies(input)
            .expect("distribution enforcement is infallible");

        let base = PlanBase::new_stream(
            input.ctx(),
            input.schema().clone(),
            input.stream_key().map(|keys| keys.to_vec()),
            input.functional_dependency().clone(),
            input.distribution().clone(),
            input.stream_kind(),
            input.emit_on_window_close(),
            input.watermark_columns().clone(),
            input.columns_monotonicity().clone(),
        );
        Self {
            base,
            input,
            sink_desc,
        }
    }
}

impl Distill for StreamIcebergWithPkIndexPositionDeleteMerger {
    fn distill<'a>(&self) -> XmlNode<'a> {
        childless_record("StreamIcebergWithPkIndexPositionDeleteMerger", vec![])
    }
}

impl PlanTreeNodeUnary<Stream> for StreamIcebergWithPkIndexPositionDeleteMerger {
    fn input(&self) -> PlanRef {
        self.input.clone()
    }

    fn clone_with_input(&self, input: PlanRef) -> Self {
        Self::new(input, self.sink_desc.clone())
    }
}

impl_plan_tree_node_for_unary! { Stream, StreamIcebergWithPkIndexPositionDeleteMerger }

impl StreamNode for StreamIcebergWithPkIndexPositionDeleteMerger {
    fn to_stream_prost_body(&self, _state: &mut BuildFragmentGraphState) -> NodeBody {
        NodeBody::IcebergWithPkIndexPositionDeleteMerger(Box::new(
            IcebergWithPkIndexPositionDeleteMergerNode {
                sink_desc: Some(self.sink_desc.to_proto()),
            },
        ))
    }
}

impl ExprRewritable<Stream> for StreamIcebergWithPkIndexPositionDeleteMerger {}

impl ExprVisitable for StreamIcebergWithPkIndexPositionDeleteMerger {}
