Takes a query plan and turns it into a differential dataflow.
(
name: &str,
scope: &mut S,
context: &mut I,
)
| 915 | |
| 916 | /// Takes a query plan and turns it into a differential dataflow. |
| 917 | pub fn implement<T, I, S>( |
| 918 | name: &str, |
| 919 | scope: &mut S, |
| 920 | context: &mut I, |
| 921 | ) -> Result< |
| 922 | ( |
| 923 | HashMap<String, Collection<S, Vec<Value>, isize>>, |
| 924 | ShutdownHandle, |
| 925 | ), |
| 926 | Error, |
| 927 | > |
| 928 | where |
| 929 | T: Timestamp + Lattice + Default, |
| 930 | I: ImplContext<T>, |
| 931 | S: Scope<Timestamp = T>, |
| 932 | { |
| 933 | scope.iterative::<u64, _, _>(|nested| { |
| 934 | let publish = vec![name]; |
| 935 | let mut rules = collect_dependencies(&*context, &publish[..])?; |
| 936 | |
| 937 | let mut local_arrangements = VariableMap::new(); |
| 938 | let mut result_map = HashMap::new(); |
| 939 | |
| 940 | // Step 0: Canonicalize, check uniqueness of bindings. |
| 941 | if rules.is_empty() { |
| 942 | return Err(Error::not_found(format!( |
| 943 | "Couldn't find any rules for name {}.", |
| 944 | name |
| 945 | ))); |
| 946 | } |
| 947 | |
| 948 | rules.sort_by(|x, y| x.name.cmp(&y.name)); |
| 949 | for index in 1..rules.len() - 1 { |
| 950 | if rules[index].name == rules[index - 1].name { |
| 951 | return Err(Error::conflict(format!( |
| 952 | "Duplicate rule definitions for rule {}", |
| 953 | rules[index].name |
| 954 | ))); |
| 955 | } |
| 956 | } |
| 957 | |
| 958 | // Step 1: Create new recursive variables for each rule. |
| 959 | for rule in rules.iter() { |
| 960 | if context.is_underconstrained(&rule.name) { |
| 961 | local_arrangements.insert( |
| 962 | rule.name.clone(), |
| 963 | Variable::new(nested, Product::new(Default::default(), 1)), |
| 964 | ); |
| 965 | } |
| 966 | } |
| 967 | |
| 968 | // Step 2: Create public arrangements for published relations. |
| 969 | for name in publish.into_iter() { |
| 970 | if let Some(relation) = local_arrangements.get(name) { |
| 971 | result_map.insert(name.to_string(), relation.leave()); |
| 972 | } else { |
| 973 | return Err(Error::not_found(format!( |
| 974 | "Attempted to publish undefined name {}.", |
no test coverage detected