polars-arrow 0.53.0

Minimal implementation of the Arrow specification forked from arrow2
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

use avro_schema::schema::{
    BytesLogical, Field as AvroField, Fixed, FixedLogical, IntLogical, LongLogical, Record,
    Schema as AvroSchema,
};
use polars_error::{PolarsResult, polars_bail};

use crate::datatypes::*;

/// Converts a [`ArrowSchema`] to an Avro [`Record`].
pub fn to_record(schema: &ArrowSchema, name: String) -> PolarsResult<Record> {
    let mut name_counter: i32 = 0;
    let fields = schema
        .iter_values()
        .map(|f| field_to_field(f, &mut name_counter))
        .collect::<PolarsResult<_>>()?;
    Ok(Record {
        name,
        namespace: None,
        doc: None,
        aliases: vec![],
        fields,
    })
}

fn field_to_field(field: &Field, name_counter: &mut i32) -> PolarsResult<AvroField> {
    let schema = type_to_schema(field.dtype(), field.is_nullable, name_counter)?;
    Ok(AvroField::new(field.name.to_string(), schema))
}

fn type_to_schema(
    dtype: &ArrowDataType,
    is_nullable: bool,
    name_counter: &mut i32,
) -> PolarsResult<AvroSchema> {
    Ok(if is_nullable {
        AvroSchema::Union(vec![
            AvroSchema::Null,
            _type_to_schema(dtype, name_counter)?,
        ])
    } else {
        _type_to_schema(dtype, name_counter)?
    })
}

fn _get_field_name(name_counter: &mut i32) -> String {
    *name_counter += 1;
    format!("r{name_counter}")
}

fn _type_to_schema(dtype: &ArrowDataType, name_counter: &mut i32) -> PolarsResult<AvroSchema> {
    Ok(match dtype.to_storage() {
        ArrowDataType::Null => AvroSchema::Null,
        ArrowDataType::Boolean => AvroSchema::Boolean,
        ArrowDataType::Int32 => AvroSchema::Int(None),
        ArrowDataType::Int64 => AvroSchema::Long(None),
        ArrowDataType::Float32 => AvroSchema::Float,
        ArrowDataType::Float64 => AvroSchema::Double,
        ArrowDataType::Binary => AvroSchema::Bytes(None),
        ArrowDataType::LargeBinary => AvroSchema::Bytes(None),
        ArrowDataType::Utf8 => AvroSchema::String(None),
        ArrowDataType::LargeUtf8 => AvroSchema::String(None),
        ArrowDataType::LargeList(inner) | ArrowDataType::List(inner) => {
            AvroSchema::Array(Box::new(type_to_schema(
                &inner.dtype,
                inner.is_nullable,
                name_counter,
            )?))
        },
        ArrowDataType::Struct(fields) => AvroSchema::Record(Record::new(
            _get_field_name(name_counter),
            fields
                .iter()
                .map(|f| field_to_field(f, name_counter))
                .collect::<PolarsResult<Vec<_>>>()?,
        )),
        ArrowDataType::Date32 => AvroSchema::Int(Some(IntLogical::Date)),
        ArrowDataType::Time32(TimeUnit::Millisecond) => AvroSchema::Int(Some(IntLogical::Time)),
        ArrowDataType::Time64(TimeUnit::Microsecond) => AvroSchema::Long(Some(LongLogical::Time)),
        ArrowDataType::Timestamp(TimeUnit::Millisecond, None) => {
            AvroSchema::Long(Some(LongLogical::LocalTimestampMillis))
        },
        ArrowDataType::Timestamp(TimeUnit::Microsecond, None) => {
            AvroSchema::Long(Some(LongLogical::LocalTimestampMicros))
        },
        ArrowDataType::Interval(IntervalUnit::MonthDayNano) => {
            let mut fixed = Fixed::new("", 12);
            fixed.logical = Some(FixedLogical::Duration);
            AvroSchema::Fixed(fixed)
        },
        ArrowDataType::FixedSizeBinary(size) => AvroSchema::Fixed(Fixed::new("", *size)),
        ArrowDataType::Decimal(p, s) => AvroSchema::Bytes(Some(BytesLogical::Decimal(*p, *s))),
        other => polars_bail!(nyi = "write {other:?} to avro"),
    })
}