algeon-cudf 0.1.2

Safe libcudf execution component for Algeon
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

use std::sync::Arc;

use super::{
    GroupByAggregationOutput,
    errors::{
        groupby_request_agg_count_limit_error, groupby_request_count_limit_error,
        groupby_total_agg_handle_count_limit_error,
    },
    output::{take_groupby_result_columns_for_output, value_counts_result_columns},
};
use crate::{
    DataFrame, Runtime, Series,
    core::{
        Column, Table,
        error::Result,
        types::{DataType, TypeId},
    },
    runtime::execution_domain::ExecutionDomain,
};

fn empty_i32_column_in_domain(output_domain: &Arc<ExecutionDomain>, operation: &'static str) -> Result<Column> {
    Column::empty_with_dtype_in_domain(DataType::new(TypeId::Int32), Arc::clone(output_domain), operation)
        .map_err(|e| e.reanchor(operation))
}

/// Trigger the low-level groupby request-count limit validation path used by
/// integration tests.
pub fn trigger_groupby_request_count_limit_error_for_test() -> Result<()> {
    Err(groupby_request_count_limit_error("groupby::groupby_aggregate", usize::MAX))
}

/// Trigger the low-level groupby per-request aggregation-count limit
/// validation path used by integration tests.
pub fn trigger_groupby_request_agg_count_limit_error_for_test() -> Result<()> {
    Err(groupby_request_agg_count_limit_error("groupby::groupby_aggregate", 0, usize::MAX, 1))
}

/// Trigger the low-level groupby total aggregation-handle count limit
/// validation path used by integration tests.
pub fn trigger_groupby_total_agg_handle_count_limit_error_for_test() -> Result<()> {
    Err(groupby_total_agg_handle_count_limit_error("groupby::groupby_aggregate", usize::MAX, 1))
}

/// Trigger the high-level groupby result-group-index invariant path used by
/// integration tests.
pub fn trigger_groupby_result_group_index_out_of_bounds_for_test() -> Result<()> {
    let runtime = Runtime::new()?;
    let col = Column::from_slice_in_domain(&[1i32], runtime.execution_domain())?;
    let grouped_results = vec![vec![Some(col)]];
    take_groupby_result_columns_for_output(grouped_results, &[(1, 0)], "groupby::GroupBy::agg").map(|_| ())
}

/// Trigger the high-level groupby result-column-index invariant path used by
/// integration tests.
pub fn trigger_groupby_result_column_index_out_of_bounds_for_test() -> Result<()> {
    let runtime = Runtime::new()?;
    let col = Column::from_slice_in_domain(&[1i32], runtime.execution_domain())?;
    let grouped_results = vec![vec![Some(col)]];
    take_groupby_result_columns_for_output(grouped_results, &[(0, 1)], "groupby::GroupBy::agg").map(|_| ())
}

/// Trigger the high-level groupby duplicate result-column consumption
/// invariant path used by integration tests.
pub fn trigger_groupby_duplicate_result_column_consumption_for_test() -> Result<()> {
    let runtime = Runtime::new()?;
    let col = Column::from_slice_in_domain(&[1i32], runtime.execution_domain())?;
    let grouped_results = vec![vec![Some(col)]];
    take_groupby_result_columns_for_output(grouped_results, &[(0, 0), (0, 0)], "groupby::GroupBy::agg").map(|_| ())
}

/// Trigger the `SeriesOps::value_counts` missing-key-columns result-shape path
/// used by integration tests.
pub fn trigger_value_counts_missing_key_columns_for_test() -> Result<DataFrame> {
    let operation = "series::value_counts";
    let runtime = Runtime::new()?;
    let output_domain = Arc::clone(runtime.execution_domain());
    let keys = Table::empty_in_domain(Arc::clone(&output_domain), "testing::value_counts_missing_key_columns")?;

    let (key_col, count_col) =
        value_counts_result_columns(GroupByAggregationOutput { keys, results: vec![Vec::new()] }, operation)?;
    DataFrame::from_series([
        ("value".to_string(), Series::from_column(key_col)?),
        ("count".to_string(), Series::from_column(count_col)?),
    ])
    .map_err(|e| e.reanchor(operation))
}

/// Trigger the `SeriesOps::value_counts` missing-count-results result-shape path
/// used by integration tests.
pub fn trigger_value_counts_missing_count_results_for_test() -> Result<DataFrame> {
    let operation = "series::value_counts";
    let runtime = Runtime::new()?;
    let output_domain = Arc::clone(runtime.execution_domain());
    let key_col = empty_i32_column_in_domain(&output_domain, "testing::value_counts_missing_count_results")?;
    let keys = Table::from_columns(vec![key_col])?;

    let (key_col, count_col) =
        value_counts_result_columns(GroupByAggregationOutput { keys, results: Vec::new() }, operation)?;
    DataFrame::from_series([
        ("value".to_string(), Series::from_column(key_col)?),
        ("count".to_string(), Series::from_column(count_col)?),
    ])
    .map_err(|e| e.reanchor(operation))
}

/// Trigger the `SeriesOps::value_counts` empty-count-result result-shape path
/// used by integration tests.
pub fn trigger_value_counts_empty_count_result_for_test() -> Result<DataFrame> {
    let operation = "series::value_counts";
    let runtime = Runtime::new()?;
    let output_domain = Arc::clone(runtime.execution_domain());
    let key_col = empty_i32_column_in_domain(&output_domain, "testing::value_counts_empty_count_result")?;
    let keys = Table::from_columns(vec![key_col])?;

    let (key_col, count_col) =
        value_counts_result_columns(GroupByAggregationOutput { keys, results: vec![Vec::new()] }, operation)?;
    DataFrame::from_series([
        ("value".to_string(), Series::from_column(key_col)?),
        ("count".to_string(), Series::from_column(count_col)?),
    ])
    .map_err(|e| e.reanchor(operation))
}