Creates an attribute from a stream of (key,value) pairs. Applies operators to enforce input semantics, registers the attribute configuration, and installs appropriate indices.
(
&mut self,
name: &str,
config: AttributeConfig,
pairs: &Stream<S, ((Value, Value), T, isize)>,
)
| 105 | /// pairs. Applies operators to enforce input semantics, registers |
| 106 | /// the attribute configuration, and installs appropriate indices. |
| 107 | fn create_attribute<S: Scope + ScopeParent<Timestamp = T>>( |
| 108 | &mut self, |
| 109 | name: &str, |
| 110 | config: AttributeConfig, |
| 111 | pairs: &Stream<S, ((Value, Value), T, isize)>, |
| 112 | ) -> Result<(), Error> { |
| 113 | if self.attributes.contains_key(name) { |
| 114 | Err(Error::conflict(format!( |
| 115 | "An attribute of name {} already exists.", |
| 116 | name |
| 117 | ))) |
| 118 | } else { |
| 119 | let tuples = match config.input_semantics { |
| 120 | InputSemantics::Raw => pairs.as_collection(), |
| 121 | InputSemantics::CardinalityOne => pairs.as_collection().cardinality_one(), |
| 122 | // Ensure that redundant (e,v) pairs don't cause |
| 123 | // misleading proposals during joining. |
| 124 | InputSemantics::CardinalityMany => pairs.as_collection().distinct(), |
| 125 | }; |
| 126 | |
| 127 | // @TODO should only create this if used later |
| 128 | let tuples_reverse = tuples.map(|(e, v)| (v, e)); |
| 129 | |
| 130 | // Propose traces are used in general, whereas the other |
| 131 | // indices are only relevant to Hector. |
| 132 | self.forward_propose.insert( |
| 133 | name.to_string(), |
| 134 | tuples.arrange_named(&format!("->Propose({})", &name)).trace, |
| 135 | ); |
| 136 | |
| 137 | if config.index_direction == IndexDirection::Both { |
| 138 | self.reverse_propose.insert( |
| 139 | name.to_string(), |
| 140 | tuples_reverse |
| 141 | .arrange_named(&format!("->_Propose({})", &name)) |
| 142 | .trace, |
| 143 | ); |
| 144 | } |
| 145 | |
| 146 | // CardinalityOne is a special case, because count, |
| 147 | // propose, and validate are all essentially the same. |
| 148 | if config.input_semantics != InputSemantics::CardinalityOne { |
| 149 | // Count traces are only required for use in |
| 150 | // worst-case optimal joins. |
| 151 | if config.query_support == QuerySupport::AdaptiveWCO { |
| 152 | self.forward_count.insert( |
| 153 | name.to_string(), |
| 154 | tuples |
| 155 | .map(|(k, _v)| (k, ())) |
| 156 | .arrange_named(&format!("->Count({})", name)) |
| 157 | .trace, |
| 158 | ); |
| 159 | |
| 160 | if config.index_direction == IndexDirection::Both { |
| 161 | self.reverse_count.insert( |
| 162 | name.to_string(), |
| 163 | tuples_reverse |
| 164 | .map(|(k, _v)| (k, ())) |
no test coverage detected