Skip to content

Commit

Permalink
test: reproducing test for chenked arrays
Browse files Browse the repository at this point in the history
  • Loading branch information
pacman82 committed Dec 9, 2024
1 parent c6eb577 commit 4841007
Showing 1 changed file with 33 additions and 5 deletions.
38 changes: 33 additions & 5 deletions tests/test_arrow_odbc.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,18 @@ def setup_table(table: str, column_type: str, values: List[Any]):
connection.close()


def empty_table(table, column_type):
"""
Create an empty table as a precondition for a test. The table will have an identy column (id)
and an additional column of the custom type with typename a
"""
connection = pyodbc.connect(MSSQL)
connection.execute(f"DROP TABLE IF EXISTS {table};")
connection.execute(f"CREATE TABLE {table} (id int IDENTITY(1,1), a {column_type});")
connection.commit()
connection.close()


def test_connection_options():
"""
Just a smoke test, that we did not mess up passing the arguments for the connections over the
Expand Down Expand Up @@ -738,11 +750,7 @@ def test_insert_batches():
"""
# Given
table = "InsertBatches"
connection = pyodbc.connect(MSSQL)
connection.execute(f"DROP TABLE IF EXISTS {table};")
connection.execute(f"CREATE TABLE {table} (id int IDENTITY(1,1), a BIGINT);")
connection.commit()
connection.close()
empty_table(table, "BIGINT")
schema = pa.schema([("a", pa.int64())])

def iter_record_batches():
Expand Down Expand Up @@ -893,6 +901,26 @@ def test_into_pyarrow_record_batch_reader_transfers_ownership():
next(iter(arrow_reader))


def test_chunked_arrays_of_variable_length_strings():
"""
See issue: <https://github.com/pacman82/arrow-odbc-py/issues/115>
"""
# Given
table = "ChunkedArraysOfVariableLengthStrings"
empty_table(table, "VARCHAR(3)")

# When
arrow_table = pa.table({"a": pa.chunked_array([["a"], ["bc"]])})
from_table_to_db(arrow_table, target=table, connection_string=MSSQL)

# Then
actual = check_output(
["odbcsv", "fetch", "-c", MSSQL, "-q", f"SELECT a FROM {table} ORDER BY id"]
)
assert "a\na\nbc\n" == actual.decode("utf8")



@pytest.mark.slow
def test_should_not_leak_memory_for_each_batch():
"""
Expand Down

0 comments on commit 4841007

Please sign in to comment.