common_function/admin/
flush_compact_table.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (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//     http://www.apache.org/licenses/LICENSE-2.0
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
15use std::str::FromStr;
16
17use api::v1::region::{StrictWindow, compact_request};
18use arrow::datatypes::DataType as ArrowDataType;
19use common_error::ext::BoxedError;
20use common_macro::admin_fn;
21use common_query::error::{
22    InvalidFuncArgsSnafu, MissingTableMutationHandlerSnafu, Result, TableMutationSnafu,
23    UnsupportedInputDataTypeSnafu,
24};
25use common_telemetry::info;
26use common_time::range::TimestampRange;
27use common_time::{Timestamp, Timezone};
28use datafusion_expr::{Signature, Volatility};
29use datatypes::prelude::*;
30use session::context::QueryContextRef;
31use session::table_name::table_name_to_full_name;
32use snafu::{ResultExt, ensure};
33use table::requests::{CompactTableRequest, FlushTableRequest};
34
35use crate::handlers::TableMutationHandlerRef;
36
37/// Compact type: strict window.
38const COMPACT_TYPE_STRICT_WINDOW: &str = "strict_window";
39/// Compact type: strict window (short name).
40const COMPACT_TYPE_STRICT_WINDOW_SHORT: &str = "swcs";
41
42const DEFAULT_COMPACTION_PARALLELISM: u32 = 1;
43
44#[admin_fn(
45    name = FlushTableFunction,
46    display_name = flush_table,
47    sig_fn = flush_signature,
48    ret = uint64
49)]
50pub(crate) async fn flush_table(
51    table_mutation_handler: &TableMutationHandlerRef,
52    query_ctx: &QueryContextRef,
53    params: &[ValueRef<'_>],
54) -> Result<Value> {
55    ensure!(
56        params.len() == 1,
57        InvalidFuncArgsSnafu {
58            err_msg: format!(
59                "The length of the args is not correct, expect 1, have: {}",
60                params.len()
61            ),
62        }
63    );
64
65    let ValueRef::String(table_name) = params[0] else {
66        return UnsupportedInputDataTypeSnafu {
67            function: "flush_table",
68            datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
69        }
70        .fail();
71    };
72
73    let (catalog_name, schema_name, table_name) = table_name_to_full_name(table_name, query_ctx)
74        .map_err(BoxedError::new)
75        .context(TableMutationSnafu)?;
76
77    let affected_rows = table_mutation_handler
78        .flush(
79            FlushTableRequest {
80                catalog_name,
81                schema_name,
82                table_name,
83            },
84            query_ctx.clone(),
85        )
86        .await?;
87
88    Ok(Value::from(affected_rows as u64))
89}
90
91#[admin_fn(
92    name = CompactTableFunction,
93    display_name = compact_table,
94    sig_fn = compact_signature,
95    ret = uint64
96)]
97pub(crate) async fn compact_table(
98    table_mutation_handler: &TableMutationHandlerRef,
99    query_ctx: &QueryContextRef,
100    params: &[ValueRef<'_>],
101) -> Result<Value> {
102    let request = parse_compact_request(params, query_ctx)?;
103    info!("Compact table request: {:?}", request);
104
105    let affected_rows = table_mutation_handler
106        .compact(request, query_ctx.clone())
107        .await?;
108
109    Ok(Value::from(affected_rows as u64))
110}
111
112fn flush_signature() -> Signature {
113    Signature::uniform(1, vec![ArrowDataType::Utf8], Volatility::Immutable)
114}
115
116fn compact_signature() -> Signature {
117    Signature::variadic(vec![ArrowDataType::Utf8], Volatility::Immutable)
118}
119
120/// Parses `compact_table` UDF parameters. This function accepts following combinations:
121/// - `[<table_name>]`: only tables name provided, using default compaction type: regular
122/// - `[<table_name>, <type>]`: specify table name and compaction type. The compaction options will be default.
123/// - `[<table_name>, <type>, <options>]`: provides both type and type-specific options.
124///   - For `twcs`, it accepts `parallelism=[N]` where N is an unsigned 32 bits number
125///   - For `swcs`, it accepts two numeric parameter: `parallelism` and `window`.
126///   - Both types accept `start_time` and `end_time` to constrain compaction windows.
127fn parse_compact_request(
128    params: &[ValueRef<'_>],
129    query_ctx: &QueryContextRef,
130) -> Result<CompactTableRequest> {
131    ensure!(
132        !params.is_empty() && params.len() <= 3,
133        InvalidFuncArgsSnafu {
134            err_msg: format!(
135                "The length of the args is not correct, expect 1-3, have: {}",
136                params.len()
137            ),
138        }
139    );
140
141    let timezone = query_ctx.timezone();
142    let (table_name, compact_type, parallelism, time_range) = match params {
143        // 1. Only table name, strategy defaults to twcs and default parallelism.
144        [ValueRef::String(table_name)] => (
145            table_name,
146            compact_request::Options::Regular(Default::default()),
147            DEFAULT_COMPACTION_PARALLELISM,
148            None,
149        ),
150        // 2. Both table name and strategy are provided.
151        [
152            ValueRef::String(table_name),
153            ValueRef::String(compact_ty_str),
154        ] => {
155            let (compact_type, parallelism, time_range) =
156                parse_compact_options(compact_ty_str, None, &timezone)?;
157            (table_name, compact_type, parallelism, time_range)
158        }
159        // 3. Table name, strategy and strategy specific options
160        [
161            ValueRef::String(table_name),
162            ValueRef::String(compact_ty_str),
163            ValueRef::String(options_str),
164        ] => {
165            let (compact_type, parallelism, time_range) =
166                parse_compact_options(compact_ty_str, Some(options_str), &timezone)?;
167            (table_name, compact_type, parallelism, time_range)
168        }
169        _ => {
170            return UnsupportedInputDataTypeSnafu {
171                function: "compact_table",
172                datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
173            }
174            .fail();
175        }
176    };
177
178    let (catalog_name, schema_name, table_name) = table_name_to_full_name(table_name, query_ctx)
179        .map_err(BoxedError::new)
180        .context(TableMutationSnafu)?;
181
182    Ok(CompactTableRequest {
183        catalog_name,
184        schema_name,
185        table_name,
186        compact_options: compact_type,
187        parallelism,
188        time_range,
189    })
190}
191
192/// Parses compaction strategy type. For `strict_window` or `swcs` strict window compaction is chosen,
193/// otherwise choose regular (TWCS) compaction.
194fn parse_compact_options(
195    type_str: &str,
196    option: Option<&str>,
197    timezone: &Timezone,
198) -> Result<(compact_request::Options, u32, Option<TimestampRange>)> {
199    let strict_window = type_str.eq_ignore_ascii_case(COMPACT_TYPE_STRICT_WINDOW)
200        || type_str.eq_ignore_ascii_case(COMPACT_TYPE_STRICT_WINDOW_SHORT);
201    let Some(option_str) = option else {
202        let options = if strict_window {
203            compact_request::Options::StrictWindow(StrictWindow { window_seconds: 0 })
204        } else {
205            compact_request::Options::Regular(Default::default())
206        };
207        return Ok((options, DEFAULT_COMPACTION_PARALLELISM, None));
208    };
209
210    // For compatibility, strict-window compaction accepts a single number as window size.
211    if strict_window && let Ok(window_seconds) = i64::from_str(option_str) {
212        return Ok((
213            compact_request::Options::StrictWindow(StrictWindow { window_seconds }),
214            DEFAULT_COMPACTION_PARALLELISM,
215            None,
216        ));
217    }
218
219    let mut window_seconds = 0i64;
220    let mut parallelism = DEFAULT_COMPACTION_PARALLELISM;
221    let mut start_time = None;
222    let mut end_time = None;
223
224    for pair in option_str.split(',') {
225        let Some((key, value)) = pair.trim().split_once('=') else {
226            return InvalidFuncArgsSnafu {
227                err_msg: format!("Invalid key-value pair: {}", pair.trim()),
228            }
229            .fail();
230        };
231        let key = key.trim();
232        let value = value.trim();
233
234        match key {
235            "window" | "window_seconds" if strict_window => {
236                window_seconds = i64::from_str(value).map_err(|_| {
237                    InvalidFuncArgsSnafu {
238                        err_msg: format!("Invalid value for window: {}", value),
239                    }
240                    .build()
241                })?;
242            }
243            "parallelism" => {
244                parallelism = value.parse::<u32>().map_err(|_| {
245                    InvalidFuncArgsSnafu {
246                        err_msg: format!("Invalid value for parallelism: {}", value),
247                    }
248                    .build()
249                })?;
250            }
251            "start_time" => {
252                start_time = Some(Timestamp::from_str(value, Some(timezone)).map_err(|_| {
253                    InvalidFuncArgsSnafu {
254                        err_msg: format!("Invalid value for start_time: {}", value),
255                    }
256                    .build()
257                })?);
258            }
259            "end_time" => {
260                end_time = Some(Timestamp::from_str(value, Some(timezone)).map_err(|_| {
261                    InvalidFuncArgsSnafu {
262                        err_msg: format!("Invalid value for end_time: {}", value),
263                    }
264                    .build()
265                })?);
266            }
267            _ => {
268                return InvalidFuncArgsSnafu {
269                    err_msg: format!("Unknown parameter: {}", key),
270                }
271                .fail();
272            }
273        }
274    }
275
276    let time_range = match (start_time, end_time) {
277        (None, None) => None,
278        (Some(start), Some(end)) if start < end => {
279            Some(TimestampRange::new(start, end).ok_or_else(|| {
280                InvalidFuncArgsSnafu {
281                    err_msg: "invalid compaction time range".to_string(),
282                }
283                .build()
284            })?)
285        }
286        (Some(_), Some(_)) => {
287            return InvalidFuncArgsSnafu {
288                err_msg: "start_time must be earlier than end_time".to_string(),
289            }
290            .fail();
291        }
292        _ => {
293            return InvalidFuncArgsSnafu {
294                err_msg: "start_time and end_time must be specified together".to_string(),
295            }
296            .fail();
297        }
298    };
299
300    let options = if strict_window {
301        compact_request::Options::StrictWindow(StrictWindow { window_seconds })
302    } else {
303        compact_request::Options::Regular(Default::default())
304    };
305    Ok((options, parallelism, time_range))
306}
307
308#[cfg(test)]
309mod tests {
310    use std::sync::Arc;
311
312    use api::v1::region::compact_request::Options;
313    use arrow::array::StringArray;
314    use arrow::datatypes::{DataType, Field};
315    use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
316    use common_time::Timestamp;
317    use common_time::range::TimestampRange;
318    use datafusion_expr::ColumnarValue;
319    use session::context::QueryContext;
320
321    use super::*;
322    use crate::function::FunctionContext;
323    use crate::function_factory::ScalarFunctionFactory;
324
325    macro_rules! define_table_function_test {
326        ($name: ident, $func: ident) => {
327            paste::paste!{
328                #[test]
329                fn [<test_ $name _misc>]() {
330                    let factory: ScalarFunctionFactory = $func::factory().into();
331                    let f = factory.provide(FunctionContext::mock());
332                    assert_eq!(stringify!($name), f.name());
333                    assert_eq!(
334                        DataType::UInt64,
335                        f.return_type(&[]).unwrap()
336                    );
337                    assert!(matches!(f.signature(),
338                                     datafusion_expr::Signature {
339                                         type_signature: datafusion_expr::TypeSignature::Uniform(1, valid_types),
340                                         volatility: datafusion_expr::Volatility::Immutable,
341                                         ..
342                                     } if valid_types == &vec![ArrowDataType::Utf8]));
343                }
344
345                #[tokio::test]
346                async fn [<test_ $name _missing_table_mutation>]() {
347                    let factory: ScalarFunctionFactory = $func::factory().into();
348                    let provider = factory.provide(FunctionContext::default());
349                    let f = provider.as_async().unwrap();
350
351                    let func_args = datafusion::logical_expr::ScalarFunctionArgs {
352                        args: vec![
353                            ColumnarValue::Array(Arc::new(StringArray::from(vec!["test"]))),
354                        ],
355                        arg_fields: vec![
356                            Arc::new(Field::new("arg_0", DataType::Utf8, false)),
357                        ],
358                        return_field: Arc::new(Field::new("result", DataType::UInt64, true)),
359                        number_rows: 1,
360                        config_options: Arc::new(datafusion_common::config::ConfigOptions::default()),
361                    };
362                    let result = f.invoke_async_with_args(func_args).await.unwrap_err();
363                    assert_eq!(
364                        "Execution error: Missing TableMutationHandler, not expected",
365                        result.to_string()
366                    );
367                }
368
369                #[tokio::test]
370                async fn [<test_ $name>]() {
371                    let factory: ScalarFunctionFactory = $func::factory().into();
372                    let provider = factory.provide(FunctionContext::mock());
373                    let f = provider.as_async().unwrap();
374
375                    let func_args = datafusion::logical_expr::ScalarFunctionArgs {
376                        args: vec![
377                            ColumnarValue::Array(Arc::new(StringArray::from(vec!["test"]))),
378                        ],
379                        arg_fields: vec![
380                            Arc::new(Field::new("arg_0", DataType::Utf8, false)),
381                        ],
382                        return_field: Arc::new(Field::new("result", DataType::UInt64, true)),
383                        number_rows: 1,
384                        config_options: Arc::new(datafusion_common::config::ConfigOptions::default()),
385                    };
386                    let result = f.invoke_async_with_args(func_args).await.unwrap();
387
388                    match result {
389                        ColumnarValue::Array(array) => {
390                            let result_array = array.as_any().downcast_ref::<arrow::array::UInt64Array>().unwrap();
391                            assert_eq!(result_array.value(0), 42u64);
392                        }
393                        ColumnarValue::Scalar(scalar) => {
394                            assert_eq!(scalar, datafusion_common::ScalarValue::UInt64(Some(42)));
395                        }
396                    }
397                }
398            }
399        }
400    }
401
402    define_table_function_test!(flush_table, FlushTableFunction);
403
404    fn check_parse_compact_params(cases: &[(&[&str], CompactTableRequest)]) {
405        for (params, expected) in cases {
406            let params = params
407                .iter()
408                .map(|s| ValueRef::String(s))
409                .collect::<Vec<_>>();
410
411            assert_eq!(
412                expected,
413                &parse_compact_request(&params, &QueryContext::arc()).unwrap()
414            );
415        }
416    }
417
418    #[test]
419    fn test_parse_compact_time_range() {
420        let params = [
421            "table",
422            "regular",
423            "start_time=2026-01-01T00:00:00Z,end_time=2026-02-01T00:00:00Z",
424        ]
425        .into_iter()
426        .map(ValueRef::String)
427        .collect::<Vec<_>>();
428
429        let request = parse_compact_request(&params, &QueryContext::arc()).unwrap();
430        assert_eq!(
431            Some(
432                TimestampRange::new(
433                    Timestamp::from_str_utc("2026-01-01T00:00:00Z").unwrap(),
434                    Timestamp::from_str_utc("2026-02-01T00:00:00Z").unwrap(),
435                )
436                .unwrap()
437            ),
438            request.time_range
439        );
440
441        let query_ctx = QueryContext::arc();
442        query_ctx.set_timezone(Timezone::from_tz_string("Asia/Shanghai").unwrap());
443        let params = [
444            "table",
445            "regular",
446            "start_time=2026-01-01T00:00:00,end_time=2026-02-01T00:00:00",
447        ]
448        .into_iter()
449        .map(ValueRef::String)
450        .collect::<Vec<_>>();
451        let request = parse_compact_request(&params, &query_ctx).unwrap();
452        assert_eq!(
453            Some(
454                TimestampRange::new(
455                    Timestamp::from_str_utc("2025-12-31T16:00:00Z").unwrap(),
456                    Timestamp::from_str_utc("2026-01-31T16:00:00Z").unwrap(),
457                )
458                .unwrap()
459            ),
460            request.time_range
461        );
462    }
463
464    #[test]
465    fn test_parse_strict_window_compact_time_range() {
466        let params = [
467            "table",
468            "strict_window",
469            "window=3600,parallelism=2,start_time=2026-01-01T00:00:00Z,end_time=2026-02-01T00:00:00Z",
470        ]
471        .into_iter()
472        .map(ValueRef::String)
473        .collect::<Vec<_>>();
474
475        let request = parse_compact_request(&params, &QueryContext::arc()).unwrap();
476        assert_eq!(
477            Options::StrictWindow(StrictWindow {
478                window_seconds: 3600,
479            }),
480            request.compact_options
481        );
482        assert_eq!(2, request.parallelism);
483        assert!(request.time_range.is_some());
484    }
485
486    #[test]
487    fn test_parse_compact_time_range_requires_valid_bounds() {
488        for options in [
489            "start_time=2026-01-01T00:00:00Z",
490            "end_time=2026-02-01T00:00:00Z",
491            "start_time=2026-02-01T00:00:00Z,end_time=2026-01-01T00:00:00Z",
492            "start_time=2026-01-01T00:00:00Z,end_time=2026-01-01T00:00:00Z",
493        ] {
494            let params = ["table", "regular", options]
495                .into_iter()
496                .map(ValueRef::String)
497                .collect::<Vec<_>>();
498            assert!(parse_compact_request(&params, &QueryContext::arc()).is_err());
499        }
500    }
501
502    #[test]
503    fn test_parse_compact_params() {
504        check_parse_compact_params(&[
505            (
506                &["table"],
507                CompactTableRequest {
508                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
509                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
510                    table_name: "table".to_string(),
511                    compact_options: Options::Regular(Default::default()),
512                    parallelism: 1,
513                    time_range: None,
514                },
515            ),
516            (
517                &[&format!("{}.table", DEFAULT_SCHEMA_NAME)],
518                CompactTableRequest {
519                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
520                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
521                    table_name: "table".to_string(),
522                    compact_options: Options::Regular(Default::default()),
523                    parallelism: 1,
524                    time_range: None,
525                },
526            ),
527            (
528                &[&format!(
529                    "{}.{}.table",
530                    DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME
531                )],
532                CompactTableRequest {
533                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
534                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
535                    table_name: "table".to_string(),
536                    compact_options: Options::Regular(Default::default()),
537                    parallelism: 1,
538                    time_range: None,
539                },
540            ),
541            (
542                &["table", "regular"],
543                CompactTableRequest {
544                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
545                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
546                    table_name: "table".to_string(),
547                    compact_options: Options::Regular(Default::default()),
548                    parallelism: 1,
549                    time_range: None,
550                },
551            ),
552            (
553                &["table", "strict_window"],
554                CompactTableRequest {
555                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
556                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
557                    table_name: "table".to_string(),
558                    compact_options: Options::StrictWindow(StrictWindow { window_seconds: 0 }),
559                    parallelism: 1,
560                    time_range: None,
561                },
562            ),
563            (
564                &["table", "strict_window", "3600"],
565                CompactTableRequest {
566                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
567                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
568                    table_name: "table".to_string(),
569                    compact_options: Options::StrictWindow(StrictWindow {
570                        window_seconds: 3600,
571                    }),
572                    parallelism: 1,
573                    time_range: None,
574                },
575            ),
576            (
577                &["table", "swcs", "120"],
578                CompactTableRequest {
579                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
580                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
581                    table_name: "table".to_string(),
582                    compact_options: Options::StrictWindow(StrictWindow {
583                        window_seconds: 120,
584                    }),
585                    parallelism: 1,
586                    time_range: None,
587                },
588            ),
589            // Test with parallelism parameter
590            (
591                &["table", "regular", "parallelism=4"],
592                CompactTableRequest {
593                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
594                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
595                    table_name: "table".to_string(),
596                    compact_options: Options::Regular(Default::default()),
597                    parallelism: 4,
598                    time_range: None,
599                },
600            ),
601            (
602                &["table", "strict_window", "window=3600,parallelism=2"],
603                CompactTableRequest {
604                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
605                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
606                    table_name: "table".to_string(),
607                    compact_options: Options::StrictWindow(StrictWindow {
608                        window_seconds: 3600,
609                    }),
610                    parallelism: 2,
611                    time_range: None,
612                },
613            ),
614            (
615                &["table", "strict_window", "window=3600"],
616                CompactTableRequest {
617                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
618                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
619                    table_name: "table".to_string(),
620                    compact_options: Options::StrictWindow(StrictWindow {
621                        window_seconds: 3600,
622                    }),
623                    parallelism: 1,
624                    time_range: None,
625                },
626            ),
627            (
628                &["table", "strict_window", "window_seconds=7200"],
629                CompactTableRequest {
630                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
631                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
632                    table_name: "table".to_string(),
633                    compact_options: Options::StrictWindow(StrictWindow {
634                        window_seconds: 7200,
635                    }),
636                    parallelism: 1,
637                    time_range: None,
638                },
639            ),
640            (
641                &["table", "strict_window", "window=1800"],
642                CompactTableRequest {
643                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
644                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
645                    table_name: "table".to_string(),
646                    compact_options: Options::StrictWindow(StrictWindow {
647                        window_seconds: 1800,
648                    }),
649                    parallelism: 1,
650                    time_range: None,
651                },
652            ),
653            (
654                &["table", "regular", "parallelism=8"],
655                CompactTableRequest {
656                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
657                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
658                    table_name: "table".to_string(),
659                    compact_options: Options::Regular(Default::default()),
660                    parallelism: 8,
661                    time_range: None,
662                },
663            ),
664        ]);
665
666        assert!(
667            parse_compact_request(
668                &["table", "strict_window", "abc"]
669                    .into_iter()
670                    .map(ValueRef::String)
671                    .collect::<Vec<_>>(),
672                &QueryContext::arc(),
673            )
674            .is_err()
675        );
676
677        assert!(
678            parse_compact_request(
679                &["a.b.table", "strict_window", "abc"]
680                    .into_iter()
681                    .map(ValueRef::String)
682                    .collect::<Vec<_>>(),
683                &QueryContext::arc(),
684            )
685            .is_err()
686        );
687
688        // Test invalid parallelism
689        assert!(
690            parse_compact_request(
691                &["table", "regular", "options", "invalid"]
692                    .into_iter()
693                    .map(ValueRef::String)
694                    .collect::<Vec<_>>(),
695                &QueryContext::arc(),
696            )
697            .is_err()
698        );
699
700        // Test too many parameters
701        assert!(
702            parse_compact_request(
703                &["table", "regular", "options", "4", "extra"]
704                    .into_iter()
705                    .map(ValueRef::String)
706                    .collect::<Vec<_>>(),
707                &QueryContext::arc(),
708            )
709            .is_err()
710        );
711
712        // Test invalid keyword argument format
713        assert!(
714            parse_compact_request(
715                &["table", "strict_window", "window"]
716                    .into_iter()
717                    .map(ValueRef::String)
718                    .collect::<Vec<_>>(),
719                &QueryContext::arc(),
720            )
721            .is_err()
722        );
723
724        // Test invalid keyword
725        assert!(
726            parse_compact_request(
727                &["table", "strict_window", "invalid_key=123"]
728                    .into_iter()
729                    .map(ValueRef::String)
730                    .collect::<Vec<_>>(),
731                &QueryContext::arc(),
732            )
733            .is_err()
734        );
735
736        assert!(
737            parse_compact_request(
738                &["table", "regular", "abcd"]
739                    .into_iter()
740                    .map(ValueRef::String)
741                    .collect::<Vec<_>>(),
742                &QueryContext::arc(),
743            )
744            .is_err()
745        );
746
747        // Test invalid window value
748        assert!(
749            parse_compact_request(
750                &["table", "strict_window", "window=abc"]
751                    .into_iter()
752                    .map(ValueRef::String)
753                    .collect::<Vec<_>>(),
754                &QueryContext::arc(),
755            )
756            .is_err()
757        );
758
759        // Test invalid parallelism in options string
760        assert!(
761            parse_compact_request(
762                &["table", "strict_window", "parallelism=abc"]
763                    .into_iter()
764                    .map(ValueRef::String)
765                    .collect::<Vec<_>>(),
766                &QueryContext::arc(),
767            )
768            .is_err()
769        );
770    }
771}
common_function/admin/flush_compact_table.rs

common_function/admin/
flush_compact_table.rs
