Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

add venv to test #17

Merged
merged 4 commits into from
May 3, 2024
Merged

add venv to test #17

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ad6d41f
add venv to test
aersam May 3, 2024
3341be3
clippy
aersam May 3, 2024
a024f34
try fix warnings
aersam May 3, 2024
4503d42
fix CI
aersam May 3, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions .github/workflows/python-test.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,14 +28,19 @@ jobs:
- name: Install tooling dependencies
run: |
python -m pip install --upgrade pip
pip install maturin
pip install maturin pyright
- name: Create venv
run: python -m venv .venv
- name: Install Dependencies
run: |
pip install pytest polars pyarrow pytest-asyncio pyright python-dotenv docker pyright cffi
source .venv/bin/activate
pip install pytest polars pyarrow pytest-asyncio python-dotenv docker cffi azure-identity
- name: Install Project
run: maturin develop
- name: pytest
shell: bash
run: pytest
run: |
source .venv/bin/activate
pytest
- name: Pyright
run: poetry run pyright .
run: pyright .
4 changes: 2 additions & 2 deletions lakeapi2sql/sql_connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,10 +27,10 @@ async def __aenter__(self) -> "TdsConnection":
async def __aexit__(self, *args, **kwargs) -> None:
pass

async def execute_sql(self, sql: str, arguments: list[str | int | float | bool | None] = None) -> list[int]:
async def execute_sql(self, sql: str, arguments: list[str | int | float | bool | None] | None = None) -> list[int]:
return await lvd.execute_sql(self._connection, sql, arguments or [])

async def execute_sql_with_result(
self, sql: str, arguments: list[str | int | float | bool | None] = None
self, sql: str, arguments: list[str | int | float | bool | None] | None = None
) -> TdsResult:
return await lvd.execute_sql_with_result(self._connection, sql, arguments or [])
63 changes: 18 additions & 45 deletions src/arrow_convert.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,11 @@ use arrow::array::UInt16Array;
use arrow::array::UInt32Array;
use arrow::array::UInt64Array;
use arrow::array::UInt8Array;
use arrow::datatypes::DataType;

use arrow::record_batch::RecordBatch;
use rust_decimal::prelude::*;
use std::borrow::Cow;
use std::fmt::Display;

use std::time::Duration as StdDuration;
use tiberius::numeric::Numeric;
use tiberius::time::time::Date;
Expand Down Expand Up @@ -71,7 +71,7 @@ pub(crate) fn get_token_rows<'a, 'b>(
for (colname, coltype) in colsnames {
let mightcol = batch.column_by_name(colname);

if let None = mightcol {
if mightcol.is_none() {
log::debug!("colname: {}. Not found", colname);
for rowindex in 0..rows {
token_rows[rowindex].push(ColumnData::String(None));
Expand Down Expand Up @@ -111,10 +111,7 @@ pub(crate) fn get_token_rows<'a, 'b>(

let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::String(match val {
Some(vs) => Some(Cow::from(vs)),
None => None,
}));
token_rows[rowindex].push(ColumnData::String(val.map(Cow::from)));
rowindex += 1;
}
}
Expand All @@ -123,10 +120,7 @@ pub(crate) fn get_token_rows<'a, 'b>(

let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::String(match val {
Some(vs) => Some(Cow::from(vs)),
None => None,
}));
token_rows[rowindex].push(ColumnData::String(val.map(Cow::from)));
rowindex += 1;
}
}
Expand Down Expand Up @@ -236,19 +230,13 @@ pub(crate) fn get_token_rows<'a, 'b>(
if coltype == &ColumnType::Int2 {
let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::I16(match val {
Some(v) => Some(v as i16),
None => None,
}));
token_rows[rowindex].push(ColumnData::I16(val.map(|v| v as i16)));
rowindex += 1;
}
} else {
let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::U8(match val {
Some(v) => Some(v as u8),
None => None,
}));
token_rows[rowindex].push(ColumnData::U8(val.map(|v| v as u8)));
rowindex += 1;
}
}
Expand Down Expand Up @@ -276,10 +264,7 @@ pub(crate) fn get_token_rows<'a, 'b>(

let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::I32(match val {
Some(x) => Some(x.clone().into()),
None => None,
}));
token_rows[rowindex].push(ColumnData::I32(val.map(|x| x.into())));
rowindex += 1;
}
}
Expand All @@ -288,10 +273,7 @@ pub(crate) fn get_token_rows<'a, 'b>(

let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::I64(match val {
Some(x) => Some(x.clone().into()),
None => None,
}));
token_rows[rowindex].push(ColumnData::I64(val.map(|x| x.into())));
rowindex += 1;
}
}
Expand All @@ -300,10 +282,7 @@ pub(crate) fn get_token_rows<'a, 'b>(

let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::I64(match val {
Some(x) => Some(x.clone() as i64),
None => None,
}));
token_rows[rowindex].push(ColumnData::I64(val.map(|x| x as i64)));
rowindex += 1;
}
}
Expand All @@ -312,10 +291,7 @@ pub(crate) fn get_token_rows<'a, 'b>(

let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::F32(match val {
Some(x) => Some(x.to_f32()),
None => None,
}));
token_rows[rowindex].push(ColumnData::F32(val.map(|x| x.to_f32())));
rowindex += 1;
}
}
Expand All @@ -333,10 +309,7 @@ pub(crate) fn get_token_rows<'a, 'b>(
if coltype == &ColumnType::Int4 || coltype == &ColumnType::Intn {
let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::I32(match val {
Some(v) => Some(v as i32),
None => None,
}));
token_rows[rowindex].push(ColumnData::I32(val.map(|v| v as i32)));
rowindex += 1;
}
} else {
Expand Down Expand Up @@ -457,7 +430,7 @@ pub(crate) fn get_token_rows<'a, 'b>(

let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::Binary(val.map(|x| Cow::from(x))));
token_rows[rowindex].push(ColumnData::Binary(val.map(Cow::from)));
rowindex += 1;
}
}
Expand All @@ -466,7 +439,7 @@ pub(crate) fn get_token_rows<'a, 'b>(

let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::Binary(val.map(|x| Cow::from(x))));
token_rows[rowindex].push(ColumnData::Binary(val.map(Cow::from)));
rowindex += 1;
}
}
Expand All @@ -475,13 +448,13 @@ pub(crate) fn get_token_rows<'a, 'b>(

let mut rowindex = 0;
for val in ba.iter() {
token_rows[rowindex].push(ColumnData::Binary(val.map(|x| Cow::from(x))));
token_rows[rowindex].push(ColumnData::Binary(val.map(Cow::from)));
rowindex += 1;
}
}
arrow::datatypes::DataType::Decimal128(_, s) => {
let ba = col.as_any().downcast_ref::<Decimal128Array>().unwrap();
let scale: u8 = s.clone().try_into().unwrap();
let scale: u8 = (*s).try_into().unwrap();
let mut rowindex = 0;
match coltype {
ColumnType::Numericn | ColumnType::Decimaln => {
Expand All @@ -505,15 +478,15 @@ pub(crate) fn get_token_rows<'a, 'b>(
_ => {
return Err(LakeApi2SqlError::NotSupported {
dtype: col.data_type().clone(),
column_type: coltype.clone(),
column_type: *coltype,
})
} //other => panic!("Not supported {:?}", other),
}
}
dt => {
return Err(LakeApi2SqlError::NotSupported {
dtype: dt.clone(),
column_type: coltype.clone(),
column_type: *coltype,
})
} //other => panic!("Not supported {:?}", other),
}
Expand Down
17 changes: 6 additions & 11 deletions src/bulk_insert.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use std::{fmt::Display, sync::Arc};
use std::sync::Arc;

use arrow::ffi_stream::ArrowArrayStreamReader;
use arrow::record_batch::RecordBatchReader;
Expand Down Expand Up @@ -51,7 +51,7 @@ pub async fn bulk_insert_batch<'a>(
) -> Result<(), LakeApi2SqlError> {
let nrows = batch.num_rows();
info!("{table_name}: received {nrows}");
let rows = task::block_in_place(|| get_token_rows(batch, &collist))?;
let rows = task::block_in_place(|| get_token_rows(batch, collist))?;
info!("{table_name}: converted {nrows}");
for rowdt in rows {
blk.send(rowdt).await?;
Expand Down Expand Up @@ -112,18 +112,13 @@ pub async fn bulk_insert<'a>(
});
while let Some(v) = rx.recv().await {
let mut blk = db_client
.bulk_insert_with_options(
table_name,
&column_names,
SqlBulkCopyOptions::TableLock,
&[],
)
.bulk_insert_with_options(table_name, column_names, SqlBulkCopyOptions::TableLock, &[])
.await?;
bulk_insert_batch(table_name, &mut blk, &v, &collist).await?;
blk.finalize().await?;
}
let schema = worker.await?;
schema

worker.await?
}

pub async fn bulk_insert_reader(
Expand All @@ -147,7 +142,7 @@ pub async fn bulk_insert_reader(
let mut blk = db_client
.bulk_insert_with_options(
table_name,
&column_names,
column_names,
SqlBulkCopyOptions::TableLock,
&[],
)
Expand Down
Loading
Loading