kernel/config/
projection.rs

1//! Projecting a self-describing schema into a fully-typed mapping — without a
2//! database.
3//!
4//! Every gap a thin config once left to the source is now stated in the schema:
5//! a column field carries its [`FlussoType`] and nullability,
6//! an aggregate its result type. So the mapping follows from the schema alone.
7//! The structural rules are unchanged from when the source derived them — a
8//! group is an `object`, a to-many join is a `nested` array, a `count` is a
9//! non-null `long`, a primary key is never null — they just no longer need a
10//! round-trip to ask.
11
12use crate::common::{ColumnName, GenericValue, IndexName};
13
14use super::{
15    Aggregate, AggregateOp, Column, ContentHash, Field, FieldSource, FlussoType, IndexMapping,
16    IndexSchema, Mapping, MappingType, Relation, ResolvedField, Script,
17};
18
19impl IndexSchema {
20    /// Project this schema into its fully-typed [`IndexMapping`].
21    pub fn resolve(&self, index: IndexName) -> IndexMapping {
22        resolve_index(index, self)
23    }
24}
25
26fn resolve_index(index: IndexName, schema: &IndexSchema) -> IndexMapping {
27    IndexMapping {
28        index,
29        // Hash the parsed schema, not the file: structural changes (including a
30        // declared type) flip the hash; cosmetic file changes do not.
31        hash: ContentHash::of(schema),
32        fields: resolve_fields(&schema.fields, schema.primary_key.as_ref()),
33    }
34}
35
36/// Resolve a list of fields. `primary_key` is the root table's key while we are
37/// still on the root row (it passes through groups, which stay on the same row);
38/// it is `None` once we cross into a related table via a join. An input-only
39/// (`emit: false`) field never reaches the document, so it has no mapping.
40fn resolve_fields(fields: &[Field], primary_key: Option<&ColumnName>) -> Vec<ResolvedField> {
41    fields
42        .iter()
43        .filter(|field| field.emit)
44        .map(|field| resolve_field(field, primary_key))
45        .collect()
46}
47
48fn resolve_field(field: &Field, primary_key: Option<&ColumnName>) -> ResolvedField {
49    let (child_fields, child_pk): (&[Field], Option<&ColumnName>) = match &field.source {
50        FieldSource::Relation(Relation::Join(join)) => (&join.fields, Some(&join.primary_key)),
51        FieldSource::Group(fields) => (fields, primary_key),
52        _ => (&[], primary_key),
53    };
54    let children = resolve_fields(child_fields, child_pk);
55
56    let (mapping_type, nullable, array) = type_and_nullability(field, primary_key);
57    let mapping = Mapping {
58        mapping_type,
59        extra: field.options.clone(),
60        map_values: map_value_type(field),
61        decimal: is_decimal(field),
62        enum_order: enum_order(field),
63        subfields: has_subfields(field),
64        indexed: is_indexed(field),
65    };
66
67    ResolvedField {
68        name: field.field.clone(),
69        mapping,
70        nullable,
71        array,
72        children,
73    }
74}
75
76/// The mapping every key of a `map` field gets — the mapping a plain field of
77/// its `values` type resolves to — or `None` for any other field. This is the
78/// only thing distinguishing a `map`'s `object` mapping from a plain one.
79fn map_value_type(field: &Field) -> Option<Box<Mapping>> {
80    match &field.source {
81        FieldSource::Column(Column {
82            ty: FlussoType::Map { values },
83            ..
84        }) => Some(Box::new(Mapping::of_type(values))),
85        _ => None,
86    }
87}
88
89/// Whether a sink may add its automatic subfields: off only for a column that
90/// says `subfields: false`.
91fn has_subfields(field: &Field) -> bool {
92    match &field.source {
93        FieldSource::Column(column) => column.subfields,
94        FieldSource::Script(script) => script.subfields,
95        _ => true,
96    }
97}
98
99/// Whether a sink indexes this field: off only for a `json` column or a join
100/// that says `indexed: false`.
101fn is_indexed(field: &Field) -> bool {
102    match &field.source {
103        FieldSource::Column(column) => column.indexed,
104        FieldSource::Relation(Relation::Join(join)) => join.indexed,
105        _ => true,
106    }
107}
108
109/// The declared variant order of an [`Enum`](FlussoType::Enum) column field, or
110/// `None` for a bare enum (empty order) or any other field. Drives the
111/// order-aware `.sort` subfield the OpenSearch sink emits.
112fn enum_order(field: &Field) -> Option<Vec<String>> {
113    match &field.source {
114        FieldSource::Column(Column {
115            ty: FlussoType::Enum,
116            enum_order,
117            ..
118        })
119        | FieldSource::Script(Script {
120            ty: FlussoType::Enum,
121            enum_order,
122            ..
123        }) if !enum_order.is_empty() => Some(enum_order.clone()),
124        _ => None,
125    }
126}
127
128/// Whether this field's leaf type is [`FlussoType::Decimal`] — a PG
129/// `numeric`/`decimal` (or a decimal-typed `sum`/`min`/`max`/`ids` aggregate, or
130/// a decimal constant). It maps to OpenSearch `double` either way; this is what
131/// lets a typed binding offer a `Decimal` handle instead of `f64`.
132fn is_decimal(field: &Field) -> bool {
133    let ty_is_decimal = |ty: &FlussoType| matches!(ty, FlussoType::Decimal);
134    match &field.source {
135        FieldSource::Column(Column { ty, .. }) | FieldSource::Script(Script { ty, .. }) => {
136            ty_is_decimal(ty)
137        }
138        FieldSource::Constant(GenericValue::Decimal(_)) => true,
139        FieldSource::Relation(Relation::Aggregate(aggregate)) => match &aggregate.op {
140            AggregateOp::Sum(_) | AggregateOp::Min(_) | AggregateOp::Max(_) => {
141                aggregate.value_type.as_ref().is_some_and(ty_is_decimal)
142            }
143            AggregateOp::Ids { element_type } => ty_is_decimal(element_type),
144            AggregateOp::Count | AggregateOp::Avg(_) | AggregateOp::Exists => false,
145        },
146        _ => false,
147    }
148}
149
150/// Returns `(mapping_type, nullable, array)`. `array` is true only for an `ids`
151/// aggregate or an array-returning script, whose `mapping_type` is then the
152/// element type.
153fn type_and_nullability(
154    field: &Field,
155    primary_key: Option<&ColumnName>,
156) -> (MappingType, bool, bool) {
157    match &field.source {
158        FieldSource::Column(Column {
159            column,
160            ty,
161            nullable,
162            default,
163            ..
164        }) => {
165            let forced_non_null = primary_key == Some(column) || default.is_some();
166            (ty.opensearch(), *nullable && !forced_non_null, false)
167        }
168        FieldSource::Group(_) => (MappingType::Object, false, false),
169        FieldSource::Geo(geo) => (
170            MappingType::Other("geo_point".to_owned()),
171            geo.nullable,
172            false,
173        ),
174        FieldSource::Constant(value) => (
175            constant_mapping_type(value),
176            matches!(value, GenericValue::Null),
177            false,
178        ),
179        FieldSource::Relation(Relation::Join(join)) => {
180            let mapping_type = if join.kind.is_to_many() {
181                MappingType::Nested
182            } else {
183                MappingType::Object
184            };
185            (mapping_type, join.nullable, false)
186        }
187        FieldSource::Relation(Relation::Aggregate(aggregate)) => aggregate_type(aggregate),
188        FieldSource::Script(script) => (script.ty.opensearch(), script.nullable, script.array),
189    }
190}
191
192fn aggregate_type(aggregate: &Aggregate) -> (MappingType, bool, bool) {
193    match &aggregate.op {
194        AggregateOp::Count => (MappingType::Long, false, false),
195        AggregateOp::Exists => (MappingType::Boolean, false, false),
196        AggregateOp::Avg(_) => (MappingType::Double, true, false),
197        AggregateOp::Sum(_) | AggregateOp::Min(_) | AggregateOp::Max(_) => {
198            let mapping_type = aggregate
199                .value_type
200                .as_ref()
201                .map(|ty| ty.opensearch())
202                // Conversion requires a `value_type` for these ops; `double` is
203                // a defensive fallback that should never be reached.
204                .unwrap_or(MappingType::Double);
205            (mapping_type, true, false)
206        }
207        AggregateOp::Ids { element_type } => (element_type.opensearch(), false, true),
208    }
209}
210
211#[cfg(test)]
212mod tests;
213
214/// The mapping type a constant value's shape implies.
215fn constant_mapping_type(value: &GenericValue) -> MappingType {
216    match value {
217        GenericValue::Bool(_) => MappingType::Boolean,
218        GenericValue::SmallInt(_) => MappingType::Short,
219        GenericValue::Int(_) => MappingType::Integer,
220        GenericValue::BigInt(_) => MappingType::Long,
221        GenericValue::Float(_) => MappingType::Float,
222        GenericValue::Double(_) | GenericValue::Decimal(_) => MappingType::Double,
223        GenericValue::Date(_)
224        | GenericValue::Time(_)
225        | GenericValue::Timestamp(_)
226        | GenericValue::TimestampTz(_) => MappingType::Date,
227        GenericValue::Bytes(_) => MappingType::Other("binary".to_owned()),
228        GenericValue::Array(items) => items
229            .first()
230            .map(constant_mapping_type)
231            .unwrap_or(MappingType::Keyword),
232        GenericValue::Map(_) => MappingType::Object,
233        GenericValue::String(_) | GenericValue::Uuid(_) | GenericValue::Null => {
234            MappingType::Keyword
235        }
236    }
237}
kernel/config/projection.rs

kernel/config/
projection.rs
