reifydb_engine/flow/compiler/operator/
distinct.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_core::{
5	interface::{
6		catalog::flow::FlowNodeId,
7		identifier::{ColumnIdentifier, ColumnShape},
8		resolved::{ResolvedColumn, ResolvedShape},
9	},
10	row::Ttl,
11};
12use reifydb_rql::{
13	expression::{ColumnExpression, Expression},
14	flow::node::FlowNodeType::Distinct,
15	nodes::DistinctNode,
16	query::QueryPlan,
17};
18use reifydb_transaction::transaction::Transaction;
19use reifydb_value::{Result, fragment::Fragment};
20
21use crate::flow::compiler::{CompileOperator, FlowCompiler};
22
23pub(crate) struct DistinctCompiler {
24	pub input: Box<QueryPlan>,
25	pub columns: Vec<ResolvedColumn>,
26	pub ttl: Option<Ttl>,
27}
28
29impl From<DistinctNode> for DistinctCompiler {
30	fn from(node: DistinctNode) -> Self {
31		Self {
32			input: node.input,
33			columns: node.columns.into_iter().collect(),
34			ttl: node.ttl,
35		}
36	}
37}
38
39fn resolved_to_column_identifier(resolved: ResolvedColumn) -> ColumnIdentifier {
40	let shape = match resolved.shape() {
41		ResolvedShape::Table(t) => ColumnShape::Qualified {
42			namespace: Fragment::internal(t.namespace().name()),
43			name: Fragment::internal(t.name()),
44		},
45		ResolvedShape::View(v) => ColumnShape::Qualified {
46			namespace: Fragment::internal(v.namespace().name()),
47			name: Fragment::internal(v.name()),
48		},
49		ResolvedShape::RingBuffer(r) => ColumnShape::Qualified {
50			namespace: Fragment::internal(r.namespace().name()),
51			name: Fragment::internal(r.name()),
52		},
53		_ => ColumnShape::Alias(Fragment::internal("_unknown")),
54	};
55
56	ColumnIdentifier {
57		shape,
58		name: Fragment::internal(resolved.name()),
59	}
60}
61
62impl CompileOperator for DistinctCompiler {
63	fn compile(self, compiler: &mut FlowCompiler, txn: &mut Transaction<'_>) -> Result<FlowNodeId> {
64		let input_node = compiler.compile_plan(txn, *self.input)?;
65
66		let expressions: Vec<Expression> = self
67			.columns
68			.into_iter()
69			.map(|col| Expression::Column(ColumnExpression(resolved_to_column_identifier(col))))
70			.collect();
71
72		let node_id = compiler.add_node(
73			txn,
74			Distinct {
75				expressions,
76			},
77		)?;
78
79		compiler.write_operator_settings(txn, node_id, self.ttl)?;
80
81		compiler.add_edge(txn, &input_node, &node_id)?;
82		Ok(node_id)
83	}
84}
reifydb_engine/flow/compiler/operator/distinct.rs

reifydb_engine/flow/compiler/operator/
distinct.rs
