algeon-datafusion 0.1.1

DataFusion physical-plan adapter for native GPU execution.
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
123
124
125
126
127
128

use std::collections::HashMap;

use query_planning::util::StableFnv1a64;

use super::{CanonicalField, DataType, RecordBatch, Schema, collect_canonical_rows};

pub(super) fn fingerprint_schema_only(schema: &Schema) -> String {
    let mut hasher = StableFnv1a64::new();
    hasher.update_field("native-tpcds-parity-v1");
    hash_schema(&mut hasher, schema);
    hasher.update_field("rows=0");
    format!("output-fnv1a64:{:016x}", hasher.finish())
}

pub(super) fn fingerprint_batches(schema: &Schema, batches: &[RecordBatch]) -> String {
    match collect_canonical_rows(schema, batches) {
        Ok(rows) => fingerprint_rows(schema, &rows),
        Err(_) => fingerprint_schema_only(schema),
    }
}

pub(super) fn fingerprint_rows(schema: &Schema, rows: &[Vec<CanonicalField>]) -> String {
    let mut hasher = StableFnv1a64::new();
    hasher.update_field("native-tpcds-parity-v1");
    hash_comparison_schema(&mut hasher, schema);
    hasher.update_field(&format!("rows={}", rows.len()));
    for row in rows {
        for field in row {
            hash_field(&mut hasher, field);
        }
    }
    format!("output-fnv1a64:{:016x}", hasher.finish())
}

pub(super) fn hash_schema(hasher: &mut StableFnv1a64, schema: &Schema) {
    hasher.update_field("schema");
    hasher.update_field(&schema.fields().len().to_string());
    hash_metadata(hasher, schema.metadata());
    for field in schema.fields() {
        hasher.update_field("field");
        hasher.update_field(field.name());
        hasher.update_field(&format!("{:?}", field.data_type()));
        hasher.update_field(if field.is_nullable() { "nullable" } else { "non_nullable" });
        hash_metadata(hasher, field.metadata());
    }
}

pub(super) fn hash_comparison_schema(hasher: &mut StableFnv1a64, schema: &Schema) {
    hasher.update_field("comparison_schema");
    hasher.update_field(&schema.fields().len().to_string());
    for field in schema.fields() {
        hasher.update_field("field");
        hasher.update_field(&comparison_data_type_key(field.data_type()));
    }
}

fn comparison_data_type_key(data_type: &DataType) -> String {
    match data_type {
        DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => "utf8".to_string(),
        DataType::Binary | DataType::LargeBinary | DataType::BinaryView => "binary".to_string(),
        other => format!("{other:?}"),
    }
}

pub(super) fn hash_metadata(hasher: &mut StableFnv1a64, metadata: &HashMap<String, String>) {
    hasher.update_metadata(metadata);
}

pub(super) fn hash_field(hasher: &mut StableFnv1a64, field: &CanonicalField) {
    match field {
        CanonicalField::Null => hasher.update_field("null"),
        CanonicalField::Bool(value) => {
            hasher.update_field("bool");
            hasher.update_field(if *value { "1" } else { "0" });
        },
        CanonicalField::Int { width, value } => {
            hasher.update_field("int");
            hasher.update_field(&width.to_string());
            hasher.update_field(&value.to_string());
        },
        CanonicalField::UInt { width, value } => {
            hasher.update_field("uint");
            hasher.update_field(&width.to_string());
            hasher.update_field(&value.to_string());
        },
        CanonicalField::Float { width, bits } => {
            hasher.update_field("float");
            hasher.update_field(&width.to_string());
            hasher.update_field(&format!("{bits:016x}"));
        },
        CanonicalField::Decimal128 { precision, scale, value } => {
            hasher.update_field("decimal128");
            hasher.update_field(&format!("{precision}:{scale}"));
            hasher.update_field(&value.to_string());
        },
        CanonicalField::Decimal256 { precision, scale, value } => {
            hasher.update_field("decimal256");
            hasher.update_field(&format!("{precision}:{scale}"));
            hasher.update_bytes(value);
        },
        CanonicalField::Date32(value) => {
            hasher.update_field("date32");
            hasher.update_field(&value.to_string());
        },
        CanonicalField::Date64(value) => {
            hasher.update_field("date64");
            hasher.update_field(&value.to_string());
        },
        CanonicalField::Timestamp { unit, tz, value } => {
            hasher.update_field("timestamp");
            hasher.update_field(&format!("{unit:?}"));
            hasher.update_field(tz.as_deref().unwrap_or(""));
            hasher.update_field(&value.to_string());
        },
        CanonicalField::Utf8(value) => {
            hasher.update_field("utf8");
            hasher.update_field(value);
        },
        CanonicalField::Binary(bytes) => {
            hasher.update_field("binary");
            hasher.update_bytes(bytes);
        },
        CanonicalField::Unsupported(text) => {
            hasher.update_field("unsupported");
            hasher.update_field(text);
        },
    }
}