/build/source/nativelink-store/src/redis_utils/ft_aggregate.rs

Source
// Copyright 2024-2025 The NativeLink Authors. All rights reserved.
//
// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    See LICENSE file for details
//
// 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 core::fmt::Debug;

use futures::Stream;
use redis::aio::ConnectionLike;
use redis::{Arg, ErrorKind, RedisError, Value};
use tracing::error;

use crate::redis_utils::aggregate_types::RedisCursorData;
use crate::redis_utils::ft_cursor_read::ft_cursor_read;

#[derive(Debug)]
pub(crate) struct FtAggregateCursor {
    pub count: u64,
    pub max_idle: u64,
}

#[derive(Debug)]
pub(crate) struct FtAggregateOptions {
    pub load: Vec<String>,
    pub cursor: FtAggregateCursor,
    pub sort_by: Vec<String>,
    /// Sort direction for every key in `sort_by`.
    pub sort_desc: bool,
}

/// Per-query `FT.AGGREGATE` timeout in milliseconds.
///
/// `RediSearch`'s module default (≈500 ms) is far too tight for the
/// scheduler's awaited-action index under any meaningful load: queries
/// time out, `NativeLink` surfaces them as parse errors, and the dedup
/// lookup fails. When dedup fails the scheduler creates a duplicate
/// operation for an action that is already in flight — observed as
/// "two same actions running on different PRs" with each running the
/// full `max_action_executing_timeout_s` window before completing. Pass an
/// explicit value generous enough to absorb 1M+ document scans on a
/// busy `RediSearch` instance.
const FT_AGGREGATE_TIMEOUT_MS: u64 = 10_000;

/// The bound on a sorted aggregate. `RediSearch` sorts only this many rows
/// and returns no more, and without it the bound is ten.
const FT_AGGREGATE_SORT_MAX: u64 = 1_000_000;

/// Calls `FT.AGGREGATE` in redis. redis-rs does not properly support this command
/// so we have to manually handle it.
pub(crate) async fn ft_aggregate<C>(
    mut connection_manager: C,
    index: String,
    query: String,
    options: FtAggregateOptions,
) -> Result<impl Stream<Item = Result<Value, RedisError>> + Send, RedisError>
where
    C: ConnectionLike + Send,
{
    struct State<C: ConnectionLike> {
        connection_manager: C,
        index: String,
        data: RedisCursorData,
    }

    let mut cmd = redis::cmd("FT.AGGREGATE");
    let mut ft_aggregate_cmd = cmd
        .arg(&index)
        .arg(&query)
        .arg("TIMEOUT")
        .arg(FT_AGGREGATE_TIMEOUT_MS)
        .arg("LOAD")
        .arg(options.load.len())
        .arg(&options.load)
        .arg("WITHCURSOR")
        .arg("COUNT")
        .arg(options.cursor.count)
        .arg("MAXIDLE")
        .arg(options.cursor.max_idle)
        .arg("SORTBY")
        .arg(options.sort_by.len() * 2);
    let direction = if options.sort_desc { "DESC"42 } else { "ASC"99 };
    for key56 in &options.sort_by {
        ft_aggregate_cmd = ft_aggregate_cmd.arg(key).arg(direction);
    }
    if !options.sort_by.is_empty() {
        // A SORTBY without MAX makes RediSearch return ten rows, whatever
        // the cursor's COUNT says: with 300 actions queued the scheduler
        // listed 10, the matching pass took 10 per pass, the abandoned
        // sweep retired 10 a minute, and the provisioner read demand as
        // 10. MAX is the sort's bound, so it has to cover the whole set.
        ft_aggregate_cmd = ft_aggregate_cmd.arg("MAX").arg(FT_AGGREGATE_SORT_MAX);
    }
    let res = ft_aggregate_cmd
        .query_async::<Value>(&mut connection_manager)
        .await;
    let data134 = match res {
        Ok(d) => d,
        Err(e) => {
            let all_args: Vec<_> = ft_aggregate_cmd
                .args_iter()
                .map7(|a| match a {
                    Arg::Simple(bytes) => match str::from_utf8(bytes) {
                        Ok(s) => s.to_string(),
                        Err(_) => format!("{bytes:?}"),
                    },
                    other => {
                        format!("{other:?}")
                    }
                })
                .collect();
            error!(
                ?e,
                index,
                ?query,
                ?options,
                ?all_args,
                "Error calling ft.aggregate"
            );
            return Err(e);
        }
    };

    let state = State {
        connection_manager,
        index,
        data: data.try_into()?0,
    };

    Ok(futures::stream::unfold(
        Some(state),
        move |maybe_state| async move {
            let mut state = maybe_state?0;
            loop {
                if let Some(map147) = state.data.data.pop_front() {
                    return Some((Ok(map), Some(state)));
                }
                if state.data.cursor == 0 {
                    return None;
                }
                let data_res = ft_cursor_read(
                    &mut state.connection_manager,
                    state.index.clone(),
                    state.data.cursor,
                )
                .await;
                state.data = match data_res {
                    Ok(data) => data,
                    Err(err) => return Some((Err(err), None)),
                };
            }
        },
    ))
}

pub(crate) fn resp2_data_parse(
    output: &mut RedisCursorData,
    results_array: &[Value],
) -> Result<(), RedisError> {
    let mut results_iter = results_array.iter();
    match results_iter.next() {
        Some(Value::Int(t)) => {
            output.total = *t;
        }
        Some(other) => {
            error!(?other, "Non-int for first value in ft.aggregate");
            return Err(RedisError::from((
                ErrorKind::Parse,
                "Non int for aggregate total",
                format!("{other:?}"),
            )));
        }
        None => {
            error!("No items in results array for ft.aggregate!");
            return Err(RedisError::from((
                ErrorKind::Parse,
                "No items in results array for ft.aggregate",
            )));
        }
    }

    for item in results_iter133 {
        match item {
            Value::Array(items) if items.len() % 2 == 0 => {}
            other => {
                error!(
                    ?other,
                    "Expected an array with an even number of items, didn't get it for aggregate value"
                );
                return Err(RedisError::from((
                    ErrorKind::Parse,
                    "Expected an array with an even number of items, didn't get it for aggregate value",
                    format!("{other:?}"),
                )));
            }
        }

        output.data.push_back(item.clone());
    }
    Ok(())
}

pub(crate) fn resp3_data_parse(
    output: &mut RedisCursorData,
    results_map: &Vec<(Value, Value)>,
) -> Result<(), RedisError> {
    for (raw_key, value) in results_map3 {
        let Value::SimpleString(key) = raw_key else {
            return Err(RedisError::from((
                ErrorKind::Parse,
                "Expected SimpleString keys",
                format!("{raw_key:?}"),
            )));
        };
        match key.as_str() {
            "attributes" => {
                let Value::Array(attributes) = value else {
                    return Err(RedisError::from((
                        ErrorKind::Parse,
                        "Expected array for attributes",
                        format!("{value:?}"),
                    )));
                };
                if !attributes.is_empty() {
                    return Err(RedisError::from((
                        ErrorKind::Parse,
                        "Expected empty attributes",
                        format!("{attributes:?}"),
                    )));
                }
            }
            "format" => {
                let Value::SimpleString(format) = value else {
                    return Err(RedisError::from((
                        ErrorKind::Parse,
                        "Expected SimpleString for format",
                        format!("{value:?}"),
                    )));
                };
                if format.as_str() != "STRING" {
                    return Err(RedisError::from((
                        ErrorKind::Parse,
                        "Expected STRING format",
                        format.clone(),
                    )));
                }
            }
            "results" => {
                let Value::Array(values) = value else {
                    return Err(RedisError::from((
                        ErrorKind::Parse,
                        "Expected Array for results",
                        format!("{value:?}"),
                    )));
                };
                for raw_value in values3 {
                    let Value::Map(value) = raw_value else {
                        return Err(RedisError::from((
                            ErrorKind::Parse,
                            "Expected list of maps in result",
                            format!("{raw_value:?}"),
                        )));
                    };
                    for (raw_map_key, raw_map_value) in value5 {
                        let Value::SimpleString(map_key) = raw_map_key else {
                            return Err(RedisError::from((
                                ErrorKind::Parse,
                                "Expected SimpleString keys for result maps",
                                format!("{raw_key:?}"),
                            )));
                        };
                        match map_key.as_str() {
                            "extra_attributes" => {
                                let extra_attributes_values4 = match raw_map_value {
                                    Value::Map(extra_attributes_values) => extra_attributes_values,
                                    // A document that expired or was deleted between
                                    // the search phase and the load phase comes back
                                    // as a row with Nil attributes. Under load this is
                                    // routine — completed awaited-action records expire
                                    // constantly — so drop the row instead of failing
                                    // the whole aggregate. Failing here surfaced to
                                    // clients as `INVALID_ARGUMENT`, which Bazel treats
                                    // as permanent, so a single expiry race killed the
                                    // build.
                                    Value::Nil => continue,
                                    other => {
                                        return Err(RedisError::from((
                                            ErrorKind::Parse,
                                            "Expected Map for extra_attributes",
                                            format!("{other:?}"),
                                        )));
                                    }
                                };
                                let mut output_array = vec![];
                                for (e_key, e_value) in extra_attributes_values4 {
                                    output_array.push(e_key.clone());
                                    output_array.push(e_value.clone());
                                }
                                output.data.push_back(Value::Array(output_array));
                            }
                            "values" => {
                                let Value::Array(values_values) = raw_map_value else {
                                    return Err(RedisError::from((
                                        ErrorKind::Parse,
                                        "Expected Array for values",
                                        format!("{raw_map_value:?}"),
                                    )));
                                };
                                if !values_values.is_empty() {
                                    return Err(RedisError::from((
                                        ErrorKind::Parse,
                                        "Expected empty values (all in extra_attributes)",
                                        format!("{values_values:?}"),
                                    )));
                                }
                            }
                            _ => {
                                return Err(RedisError::from((
                                    ErrorKind::Parse,
                                    "Unknown result map key",
                                    format!("{map_key:?}"),
                                )));
                            }
                        }
                    }
                }
            }
            "total_results" => {
                let Value::Int(total) = value else {
                    return Err(RedisError::from((
                        ErrorKind::Parse,
                        "Expected int for total_results",
                        format!("{value:?}"),
                    )));
                };
                output.total = *total;
            }
            "warning" => {
                let Value::Array(warnings) = value else {
                    return Err(RedisError::from((
                        ErrorKind::Parse,
                        "Expected Array for warning",
                        format!("{value:?}"),
                    )));
                };
                if !warnings.is_empty() {
                    return Err(RedisError::from((
                        ErrorKind::Parse,
                        "Expected empty warnings",
                        format!("{warnings:?}"),
                    )));
                }
            }
            _ => {
                return Err(RedisError::from((
                    ErrorKind::Parse,
                    "Unexpected key in ft.aggregate",
                    format!("{key} => {value:?}"),
                )));
            }
        }
    }
    Ok(())
}

impl TryFrom<Value> for RedisCursorData {
    type Error = RedisError;
    fn try_from(raw_value: Value) -> Result<Self, RedisError> {
        let Value::Array(value) = raw_value else {
            error!(
                ?raw_value,
                "Bad data in ft.aggregate, expected array at top-level"
            );
            return Err(RedisError::from((ErrorKind::Parse, "Expected array")));
        };
        if value.len() < 2 {
            return Err(RedisError::from((
                ErrorKind::Parse,
                "Expected at least 2 elements",
            )));
        }
        let mut output = Self::default();
        let mut value = value.into_iter();
        match value.next().unwrap() {
            Value::Array(d) => resp2_data_parse(&mut output, &d)?0,
            Value::Map(d) => resp3_data_parse(&mut output, &d)?0,
            other => {
                error!(
                    ?other,
                    "Bad data in ft.aggregate, expected array for results"
                );
                return Err(RedisError::from((
                    ErrorKind::Parse,
                    "Non map item",
                    format!("{other:?}"),
                )));
            }
        }
        let Value::Int(cursor) = value.next().unwrap() else {
            return Err(RedisError::from((
                ErrorKind::Parse,
                "Expected integer as last element",
            )));
        };
        output.cursor = cursor as u64;
        Ok(output)
    }
}

Coverage Report

Created: 2026-10-01 05:28

Line	Count	Source
1		// Copyright 2024-2025 The NativeLink Authors. All rights reserved.
2		//
3		// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
4		// you may not use this file except in compliance with the License.
5		// You may obtain a copy of the License at
6		//
7		// See LICENSE file for details
8		//
9		// Unless required by applicable law or agreed to in writing, software
10		// distributed under the License is distributed on an "AS IS" BASIS,
11		// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12		// See the License for the specific language governing permissions and
13		// limitations under the License.
14
15		use core::fmt::Debug;
16
17		use futures::Stream;
18		use redis::aio::ConnectionLike;
19		use redis::{Arg, ErrorKind, RedisError, Value};
20		use tracing::error;
21
22		use crate::redis_utils::aggregate_types::RedisCursorData;
23		use crate::redis_utils::ft_cursor_read::ft_cursor_read;
24
25		#[derive(Debug)]
26		pub(crate) struct FtAggregateCursor {
27		pub count: u64,
28		pub max_idle: u64,
29		}
30
31		#[derive(Debug)]
32		pub(crate) struct FtAggregateOptions {
33		pub load: Vec<String>,
34		pub cursor: FtAggregateCursor,
35		pub sort_by: Vec<String>,
36		/// Sort direction for every key in `sort_by`.
37		pub sort_desc: bool,
38		}
39
40		/// Per-query `FT.AGGREGATE` timeout in milliseconds.
41		///
42		/// `RediSearch`'s module default (≈500 ms) is far too tight for the
43		/// scheduler's awaited-action index under any meaningful load: queries
44		/// time out, `NativeLink` surfaces them as parse errors, and the dedup
45		/// lookup fails. When dedup fails the scheduler creates a duplicate
46		/// operation for an action that is already in flight — observed as
47		/// "two same actions running on different PRs" with each running the
48		/// full `max_action_executing_timeout_s` window before completing. Pass an
49		/// explicit value generous enough to absorb 1M+ document scans on a
50		/// busy `RediSearch` instance.
51		const FT_AGGREGATE_TIMEOUT_MS: u64 = 10_000;
52
53		/// The bound on a sorted aggregate. `RediSearch` sorts only this many rows
54		/// and returns no more, and without it the bound is ten.
55		const FT_AGGREGATE_SORT_MAX: u64 = 1_000_000;
56
57		/// Calls `FT.AGGREGATE` in redis. redis-rs does not properly support this command
58		/// so we have to manually handle it.
59	141	pub(crate) async fn ft_aggregate<C>(
60	141	mut connection_manager: C,
61	141	index: String,
62	141	query: String,
63	141	options: FtAggregateOptions,
64	141	) -> Result<impl Stream<Item = Result<Value, RedisError>> + Send, RedisError>
65	141	where
66	141	C: ConnectionLike + Send,
67	141	{
68		struct State<C: ConnectionLike> {
69		connection_manager: C,
70		index: String,
71		data: RedisCursorData,
72		}
73
74	141	let mut cmd = redis::cmd("FT.AGGREGATE");
75	141	let mut ft_aggregate_cmd = cmd
76	141	.arg(&index)
77	141	.arg(&query)
78	141	.arg("TIMEOUT")
79	141	.arg(FT_AGGREGATE_TIMEOUT_MS)
80	141	.arg("LOAD")
81	141	.arg(options.load.len())
82	141	.arg(&options.load)
83	141	.arg("WITHCURSOR")
84	141	.arg("COUNT")
85	141	.arg(options.cursor.count)
86	141	.arg("MAXIDLE")
87	141	.arg(options.cursor.max_idle)
88	141	.arg("SORTBY")
89	141	.arg(options.sort_by.len() * 2);
90	141	let direction = if options.sort_desc { "DESC"42 } else { "ASC"99 };
91	141	for key56 in &options.sort_by {
92	56	ft_aggregate_cmd = ft_aggregate_cmd.arg(key).arg(direction);
93	56	}
94	141	if !options.sort_by.is_empty() {
95	56	// A SORTBY without MAX makes RediSearch return ten rows, whatever
96	56	// the cursor's COUNT says: with 300 actions queued the scheduler
97	56	// listed 10, the matching pass took 10 per pass, the abandoned
98	56	// sweep retired 10 a minute, and the provisioner read demand as
99	56	// 10. MAX is the sort's bound, so it has to cover the whole set.
100	56	ft_aggregate_cmd = ft_aggregate_cmd.arg("MAX").arg(FT_AGGREGATE_SORT_MAX);
101	85	}
102	141	let res = ft_aggregate_cmd
103	141	.query_async::<Value>(&mut connection_manager)
104	141	.await;
105	141	let data134 = match res {
106	134	Ok(d) => d,
107	7	Err(e) => {
108	7	let all_args: Vec<_> = ft_aggregate_cmd
109	7	.args_iter()
110	140	. map7 (\|a\| match a {
111	140	Arg::Simple(bytes) => match str::from_utf8(bytes) {
112	140	Ok(s) => s.to_string(),
113	0	Err(_) => format!("{bytes:?}"),
114		},
115	0	other => {
116	0	format!("{other:?}")
117		}
118	140	})
119	7	.collect();
120	7	error!(
121		?e,
122		index,
123		?query,
124		?options,
125		?all_args,
126		"Error calling ft.aggregate"
127		);
128	7	return Err(e);
129		}
130		};
131
132	134	let state = State {
133	134	connection_manager,
134	134	index,
135	134	data: data.try_into() ?0 ,
136		};
137
138	134	Ok(futures::stream::unfold(
139	134	Some(state),
140	276	move \|maybe_state\| async move {
141	276	let mut state = maybe_state ?0 ;
142		loop {
143	278	if let Some( map147 ) = state.data.data.pop_front() {
144	147	return Some((Ok(map), Some(state)));
145	131	}
146	131	if state.data.cursor == 0 {
147	129	return None;
148	2	}
149	2	let data_res = ft_cursor_read(
150	2	&mut state.connection_manager,
151	2	state.index.clone(),
152	2	state.data.cursor,
153	2	)
154	2	.await;
155	2	state.data = match data_res {
156	2	Ok(data) => data,
157	0	Err(err) => return Some((Err(err), None)),
158		};
159		}
160	552	},
161		))
162	141	}
163
164	133	pub(crate) fn resp2_data_parse(
165	133	output: &mut RedisCursorData,
166	133	results_array: &[Value],
167	133	) -> Result<(), RedisError> {
168	133	let mut results_iter = results_array.iter();
169	133	match results_iter.next() {
170	133	Some(Value::Int(t)) => {
171	133	output.total = *t;
172	133	}
173	0	Some(other) => {
174	0	error!(?other, "Non-int for first value in ft.aggregate");
175	0	return Err(RedisError::from((
176	0	ErrorKind::Parse,
177	0	"Non int for aggregate total",
178	0	format!("{other:?}"),
179	0	)));
180		}
181		None => {
182	0	error!("No items in results array for ft.aggregate!");
183	0	return Err(RedisError::from((
184	0	ErrorKind::Parse,
185	0	"No items in results array for ft.aggregate",
186	0	)));
187		}
188		}
189
190	143	for item in results_iter133 {
191	143	match item {
192	143	Value::Array(items) if items.len() % 2 == 0 => {}
193	0	other => {
194	0	error!(
195		?other,
196		"Expected an array with an even number of items, didn't get it for aggregate value"
197		);
198	0	return Err(RedisError::from((
199	0	ErrorKind::Parse,
200	0	"Expected an array with an even number of items, didn't get it for aggregate value",
201	0	format!("{other:?}"),
202	0	)));
203		}
204		}
205
206	143	output.data.push_back(item.clone());
207		}
208	133	Ok(())
209	133	}
210
211	3	pub(crate) fn resp3_data_parse(
212	3	output: &mut RedisCursorData,
213	3	results_map: &Vec<(Value, Value)>,
214	3	) -> Result<(), RedisError> {
215	11	for (raw_key, value) in results_map3 {
216	11	let Value::SimpleString(key) = raw_key else {
217	0	return Err(RedisError::from((
218	0	ErrorKind::Parse,
219	0	"Expected SimpleString keys",
220	0	format!("{raw_key:?}"),
221	0	)));
222		};
223	11	match key.as_str() {
224	11	"attributes" => {
225	3	let Value::Array(attributes) = value else {
226	0	return Err(RedisError::from((
227	0	ErrorKind::Parse,
228	0	"Expected array for attributes",
229	0	format!("{value:?}"),
230	0	)));
231		};
232	3	if !attributes.is_empty() {
233	0	return Err(RedisError::from((
234	0	ErrorKind::Parse,
235	0	"Expected empty attributes",
236	0	format!("{attributes:?}"),
237	0	)));
238	3	}
239		}
240	8	"format" => {
241	3	let Value::SimpleString(format) = value else {
242	0	return Err(RedisError::from((
243	0	ErrorKind::Parse,
244	0	"Expected SimpleString for format",
245	0	format!("{value:?}"),
246	0	)));
247		};
248	3	if format.as_str() != "STRING" {
249	0	return Err(RedisError::from((
250	0	ErrorKind::Parse,
251	0	"Expected STRING format",
252	0	format.clone(),
253	0	)));
254	3	}
255		}
256	5	"results" => {
257	3	let Value::Array(values) = value else {
258	0	return Err(RedisError::from((
259	0	ErrorKind::Parse,
260	0	"Expected Array for results",
261	0	format!("{value:?}"),
262	0	)));
263		};
264	5	for raw_value in values3 {
265	5	let Value::Map(value) = raw_value else {
266	0	return Err(RedisError::from((
267	0	ErrorKind::Parse,
268	0	"Expected list of maps in result",
269	0	format!("{raw_value:?}"),
270	0	)));
271		};
272	10	for (raw_map_key, raw_map_value) in value5 {
273	10	let Value::SimpleString(map_key) = raw_map_key else {
274	0	return Err(RedisError::from((
275	0	ErrorKind::Parse,
276	0	"Expected SimpleString keys for result maps",
277	0	format!("{raw_key:?}"),
278	0	)));
279		};
280	10	match map_key.as_str() {
281	10	"extra_attributes" => {
282	5	let extra_attributes_values4 = match raw_map_value {
283	4	Value::Map(extra_attributes_values) => extra_attributes_values,
284		// A document that expired or was deleted between
285		// the search phase and the load phase comes back
286		// as a row with Nil attributes. Under load this is
287		// routine — completed awaited-action records expire
288		// constantly — so drop the row instead of failing
289		// the whole aggregate. Failing here surfaced to
290		// clients as `INVALID_ARGUMENT`, which Bazel treats
291		// as permanent, so a single expiry race killed the
292		// build.
293	1	Value::Nil => continue,
294	0	other => {
295	0	return Err(RedisError::from((
296	0	ErrorKind::Parse,
297	0	"Expected Map for extra_attributes",
298	0	format!("{other:?}"),
299	0	)));
300		}
301		};
302	4	let mut output_array = vec![];
303	8	for (e_key, e_value) in extra_attributes_values4 {
304	8	output_array.push(e_key.clone());
305	8	output_array.push(e_value.clone());
306	8	}
307	4	output.data.push_back(Value::Array(output_array));
308		}
309	5	"values" => {
310	5	let Value::Array(values_values) = raw_map_value else {
311	0	return Err(RedisError::from((
312	0	ErrorKind::Parse,
313	0	"Expected Array for values",
314	0	format!("{raw_map_value:?}"),
315	0	)));
316		};
317	5	if !values_values.is_empty() {
318	0	return Err(RedisError::from((
319	0	ErrorKind::Parse,
320	0	"Expected empty values (all in extra_attributes)",
321	0	format!("{values_values:?}"),
322	0	)));
323	5	}
324		}
325		_ => {
326	0	return Err(RedisError::from((
327	0	ErrorKind::Parse,
328	0	"Unknown result map key",
329	0	format!("{map_key:?}"),
330	0	)));
331		}
332		}
333		}
334		}
335		}
336	2	"total_results" => {
337	1	let Value::Int(total) = value else {
338	0	return Err(RedisError::from((
339	0	ErrorKind::Parse,
340	0	"Expected int for total_results",
341	0	format!("{value:?}"),
342	0	)));
343		};
344	1	output.total = *total;
345		}
346	1	"warning" => {
347	1	let Value::Array(warnings) = value else {
348	0	return Err(RedisError::from((
349	0	ErrorKind::Parse,
350	0	"Expected Array for warning",
351	0	format!("{value:?}"),
352	0	)));
353		};
354	1	if !warnings.is_empty() {
355	0	return Err(RedisError::from((
356	0	ErrorKind::Parse,
357	0	"Expected empty warnings",
358	0	format!("{warnings:?}"),
359	0	)));
360	1	}
361		}
362		_ => {
363	0	return Err(RedisError::from((
364	0	ErrorKind::Parse,
365	0	"Unexpected key in ft.aggregate",
366	0	format!("{key} => {value:?}"),
367	0	)));
368		}
369		}
370		}
371	3	Ok(())
372	3	}
373
374		impl TryFrom<Value> for RedisCursorData {
375		type Error = RedisError;
376	134	fn try_from(raw_value: Value) -> Result<Self, RedisError> {
377	134	let Value::Array(value) = raw_value else {
378	0	error!(
379		?raw_value,
380		"Bad data in ft.aggregate, expected array at top-level"
381		);
382	0	return Err(RedisError::from((ErrorKind::Parse, "Expected array")));
383		};
384	134	if value.len() < 2 {
385	0	return Err(RedisError::from((
386	0	ErrorKind::Parse,
387	0	"Expected at least 2 elements",
388	0	)));
389	134	}
390	134	let mut output = Self::default();
391	134	let mut value = value.into_iter();
392	134	match value.next().unwrap() {
393	132	Value::Array(d) => resp2_data_parse(&mut output, &d) ?0 ,
394	2	Value::Map(d) => resp3_data_parse(&mut output, &d) ?0 ,
395	0	other => {
396	0	error!(
397		?other,
398		"Bad data in ft.aggregate, expected array for results"
399		);
400	0	return Err(RedisError::from((
401	0	ErrorKind::Parse,
402	0	"Non map item",
403	0	format!("{other:?}"),
404	0	)));
405		}
406		}
407	134	let Value::Int(cursor) = value.next().unwrap() else {
408	0	return Err(RedisError::from((
409	0	ErrorKind::Parse,
410	0	"Expected integer as last element",
411	0	)));
412		};
413	134	output.cursor = cursor as u64;
414	134	Ok(output)
415	134	}
416		}