Potential bug: SIGSEGV in Thin-mode AsyncConnection.fetch_df_batches() for VARCHAR2 + TIMESTAMP result; synchronous dataframe fetch succeeds

[leonkosak]

1. Versions

• python-oracledb: 26.0.1

• Mode: Thin

• Python: 3.14.7

• PyArrow: 25.0.1

• Platform: Linux 7.0.0-29-generic x86_64, glibc 2.41

• 64-bit Python: yes

• OS family used by the application: Debian 13 or newer

• init_oracle_client() is not called

2. Is this an error, hang, or crash?

Crash: the Python interpreter terminates with SIGSEGV, normally with exit

status 139.

No Python exception is raised.

3. Behavior

Thin-mode AsyncConnection.fetch_df_batches() segfaults while fetching a

result containing a VARCHAR2 column followed by a TIMESTAMP(6) column.

The crash occurs while advancing the async dataframe iterator, before the

returned object is converted with pyarrow.table().

The equivalent synchronous Connection.fetch_df_batches() call succeeds and

returns all rows.

Ordinary row fetching with AsyncCursor.fetchmany() also succeeds.

The two source columns are described as:

STATEMENTKEY:

    type_code=DB_TYPE_VARCHAR

    display_size=30

    internal_size=30

    precision=None

    scale=None

    null_ok=False

 

WRITETIME:

    type_code=DB_TYPE_TIMESTAMP

    display_size=23

    internal_size=None

    precision=0

    scale=6

    null_ok=True

For a representative 1,000-row result:

all 1,000 VARCHAR2 values are non-null;
all 1,000 TIMESTAMP values are non-null;
the VARCHAR2 column contains one distinct/repeated value;
the TIMESTAMP column contains 1,000 distinct values.
Isolation results:

Query shape

Async dataframe fetch

VARCHAR2 only

succeeds

NUMBER only

succeeds

TIMESTAMP only

succeeds

VARCHAR2 + NUMBER

succeeds

NUMBER + TIMESTAMP

succeeds

VARCHAR2 + TIMESTAMP

SIGSEGV

VARCHAR2 + CAST(TIMESTAMP AS TIMESTAMP(6))

succeeds in the reduced test

Same VARCHAR2 + TIMESTAMP query using synchronous Connection.fetch_df_batches()

succeeds

A synthetic DUAL CONNECT BY query with a repeated VARCHAR2 value and generated
TIMESTAMP values did not reproduce the crash. It may therefore depend on
additional result metadata or the exact encoded values.

This appears related to, but different from, issue #597:

the VARCHAR2 column has a defined internal_size of 30;
the failure was isolated to AsyncConnection.fetch_df_batches();
synchronous Connection.fetch_df_batches() succeeds.
4. Reproducer using an existing table

The following reproduces consistently against the affected view. The object and
column names below are anonymized, but the reported Oracle data types and
metadata are unchanged.

import asyncio

import faulthandler

 

import oracledb

import pyarrow as pa

 

faulthandler.enable()

 

USER = "..."

PASSWORD = "..."

DSN = "host:1521/service"

 

SQL = """

    select varchar_key, timestamp_value

    from affected_view

    where rownum <= 1000

"""

 

 

async def reproduce_async():

    async with oracledb.connect_async(

        user=USER,

        password=PASSWORD,

        dsn=DSN,

    ) as connection:

        batches = connection.fetch_df_batches(

            statement=SQL,

            size=1000,

            fetch_decimals=True,

        )

 

        iterator = aiter(batches)

 

        print("before async fetch", flush=True)

 

        # The interpreter segfaults during this await. The next print is not

        # reached, and pyarrow.table() is not involved in the crash.

        oracle_dataframe = await anext(iterator)

 

        print("after async fetch", flush=True)

        table = pa.table(oracle_dataframe)

        print(table.schema)

 

 

asyncio.run(reproduce_async())

Equivalent synchronous control:

import oracledb

import pyarrow as pa

 

with oracledb.connect(

    user=USER,

    password=PASSWORD,

    dsn=DSN,

) as connection:

    for oracle_dataframe in connection.fetch_df_batches(

        statement=SQL,

        size=1000,

        fetch_decimals=True,

    ):

        table = pa.table(oracle_dataframe)

        print(table.schema, table.num_rows)

The synchronous control completes successfully.

Ordinary async row-fetch control:

import asyncio

import oracledb

 

 

async def row_fetch_control():

    async with oracledb.connect_async(

        user=USER,

        password=PASSWORD,

        dsn=DSN,

    ) as connection:

        async with connection.cursor() as cursor:

            cursor.arraysize = 1000

            cursor.prefetchrows = 1000

            await cursor.execute(SQL)

            rows = await cursor.fetchmany(1000)

            print(len(rows))

 

 

asyncio.run(row_fetch_control())

This also completes successfully.

5. Requested-schema behavior

The async crash was also observed in the application when passing a PyArrow
requested_schema, for example:

schema = pa.schema([

    ("VARCHAR_KEY", pa.string()),

    ("TIMESTAMP_VALUE", pa.timestamp("us")),

])

 

async for oracle_dataframe in connection.fetch_df_batches(

    statement=SQL,

    size=1000,

    fetch_decimals=True,

    requested_schema=schema,

):

    table = pa.table(oracle_dataframe)

The crash still occurs while fetching, before the pa.table() call.

6. Expected behavior

AsyncConnection.fetch_df_batches() should return the same dataframe values as
the successful synchronous dataframe fetch, or raise a Python exception if the
result cannot be represented.

It should not terminate the interpreter.

[anthony-tuininga]

This seems similar to #608. Are you able to run the test suite successfully? In particular the data frame tests?

pytest tests/create_schema.py
pytest tests/dataframe

I am not able to replicate this issue, either. This is what I used to build the data:

drop table if exists issue_607;

create table issue_607 (
    StatementKey varchar2(30) not null,
    WriteTime timestamp
);

begin
    for i in 1..1000 loop
        insert into issue_607 values
        ('Repeated Value', systimestamp + i / (24 * 60 * 60));
    end loop;
    commit;
end;
/

[leonkosak]

Thanks for investigating this.

I have only read-only access to the affected Oracle database. I cannot create tables, insert rows, modify views, or run the upstream test setup against that database. I also cannot provide database credentials or confidential object names.

I ran a read-only metadata query against the exact 20-column result returned by the import query. I anonymized the object and column names below, but preserved the exact result-column order, Oracle types, lengths, precision, scale, and nullability.

The anonymized result schema is:

A01  VARCHAR2(30)       NOT NULL

A02  VARCHAR2(64)       NOT NULL

A03  VARCHAR2(3)        NOT NULL

A04  VARCHAR2(1)        NULL

A05  NUMBER             NULL

A06  NUMBER             NULL

A07  NUMBER             NULL

A08  NVARCHAR2(900)     NULL

A09  NUMBER(22,4)       NULL

A10  TIMESTAMP(6)       NULL

A11  VARCHAR2(30)       NOT NULL

A12  NVARCHAR2(50)      NULL

A13  NVARCHAR2(150)     NOT NULL

A14  NVARCHAR2(25)      NULL

A15  DATE               NOT NULL

A16  NUMBER             NULL

A17  TIMESTAMP(6)       NULL

A18  TIMESTAMP(6)       NULL

A19  NUMBER(10,0)        NULL

A20  TIMESTAMP(6)       NULL

The equivalent anonymized table definition is:

CREATE TABLE issue_607_source (

    A01 VARCHAR2(30) NOT NULL,

    A02 VARCHAR2(64) NOT NULL,

    A03 VARCHAR2(3) NOT NULL,

    A04 VARCHAR2(1),

    A05 NUMBER,

    A06 NUMBER,

    A07 NUMBER,

    A08 NVARCHAR2(900),

    A09 NUMBER(22,4),

    A10 TIMESTAMP(6),

    A11 VARCHAR2(30) NOT NULL,

    A12 NVARCHAR2(50),

    A13 NVARCHAR2(150) NOT NULL,

    A14 NVARCHAR2(25),

    A15 DATE NOT NULL,

    A16 NUMBER,

    A17 TIMESTAMP(6),

    A18 TIMESTAMP(6),

    A19 NUMBER(10,0),

    A20 TIMESTAMP(6)

);

The following insert creates synthetic data with the relevant characteristics: repeated non-null string values, distinct timestamp values, nullable columns, and 1,000 rows.

INSERT INTO issue_607_source (

    A01, A02, A03, A04, A05, A06, A07, A08, A09, A10,

    A11, A12, A13, A14, A15, A16, A17, A18, A19, A20

)

SELECT

    'REPEATED_VALUE',

    'POSITION_' || MOD(level, 10),

    'EUR',

    'X',

    100 + level,

    200 + level,

    300 + level,

    NULL,

    400 + level,

    SYSTIMESTAMP + NUMTODSINTERVAL(level, 'SECOND'),

    'CUSTOMER_' || MOD(level, 10),

    'REFERENCE_' || MOD(level, 10),

    'Customer name',

    'REGISTER_' || MOD(level, 10),

    TRUNC(SYSDATE) - MOD(level, 365),

    2020 + MOD(level, 5),

    NULL,

    NULL,

    NULL,

    CURRENT_TIMESTAMP

FROM dual

CONNECT BY level <= 1000;

 

COMMIT;

The actual anonymized query shape is:

SELECT

    v.A01,

    v.A02,

    v.A03,

    v.A04,

    v.A05,

    v.A06,

    v.A07,

    v.A08,

    v.A09,

    v.A10,

    b.A11,

    b.A12,

    b.A13,

    b.A14,

    b.A15,

    b.A16,

    CAST(NULL AS TIMESTAMP) AS A17,

    CAST(NULL AS TIMESTAMP) AS A18,

    CAST(NULL AS NUMBER(10)) AS A19,

    CAST(CURRENT_TIMESTAMP AS TIMESTAMP) AS A20

FROM anonymized_source_view v

JOIN (

    SELECT

        c.A11,

        c.A12,

        c.A13,

        c.A14,

        s.A01,

        s.A15,

        EXTRACT(YEAR FROM s.A15) AS A16

    FROM anonymized_table_1 u

    JOIN anonymized_table_2 s

        ON u.A01 = s.A01

    JOIN anonymized_table_3 c

        ON c.A11 = u.A11

    WHERE u.A02 = 0

      AND s.A03 = 2

      AND s.A04 != 'PROJECTION'

      AND c.A12 IS NOT NULL

) b

    ON b.A01 = v.A01

WHERE v.A01 IN (

    SELECT A01

    FROM anonymized_import_control

    WHERE A02 = :chunk_number

);


The complete relevant Oracle-reading implementation is:


import asyncio

import time

from dataclasses import dataclass

from typing import Any, Iterator, Mapping

 

import oracledb

import pyarrow as pa

import pyarrow.parquet as pq

 

 

@dataclass(frozen=True, slots=True)

class OracleArrowBatch:

    table: pa.Table

    fetch_ms: float

 

 

def _validate_requested_schema(

    schema: pa.Schema,

    source_columns: list[str],

) -> pa.Schema:

    fields_by_name = {

        [field.name](http://field.name/).upper(): field

        for field in schema

    }

 

    source_names = [

        name.upper()

        for name in source_columns

    ]

 

    source_name_set = set(source_names)

 

    if len(source_names) != len(source_name_set):

        raise ValueError(

            "Oracle query returned duplicate column names"

        )

 

    not_in_contract = [

        name

        for name, upper_name in zip(

            source_columns,

            source_names,

        )

        if upper_name not in fields_by_name

    ]

 

    not_in_query = [

        field.name

        for field in schema

        if field.name.upper() not in source_name_set

    ]

 

    if not_in_contract or not_in_query:

        raise ValueError(

            "Oracle query columns must match the Arrow contract exactly; "

            f"not in contract={not_in_contract}, "

            f"not in query={not_in_query}"

        )

 

    return pa.schema(

        [

            fields_by_name[name]

            for name in source_names

        ],

        metadata=schema.metadata,

    )

 

 

def _to_contract_table(

    table: pa.Table,

    schema: pa.Schema,

) -> pa.Table:

    if table.schema.equals(

        schema,

        check_metadata=True,

    ):

        return table

 

    columns_by_name = {

        name.upper(): table.column(index)

        for index, name in enumerate(table.column_names)

    }

 

    return pa.Table.from_arrays(

        [

            columns_by_name[field.name.upper()]

            for field in schema

        ],

        schema=schema,

    )

 

 

async def read_oracle_count(

    *,

    user: str,

    password: str,

    dsn: str,

    count_sql: str,

    connect_kwargs: Mapping[str, Any] | None = None,

) -> int | None:

    async with oracledb.connect_async(

        user=user,

        password=password,

        dsn=dsn,

        **dict(connect_kwargs or {}),

    ) as connection:

        row = await connection.fetchone(count_sql)

 

    return (

        int(row[0])

        if row and row[0] is not None

        else None

    )

 

 

def read_oracle_arrow_batches(

    *,

    user: str,

    password: str,

    dsn: str,

    select_sql: str,

    schema: pa.Schema,

    source_columns: list[str],

    chunk_size: int = 10_000,

    connect_kwargs: Mapping[str, Any] | None = None,

    fetch_decimals: bool = False,

) -> Iterator[OracleArrowBatch]:

    if chunk_size <= 0:

        raise ValueError("chunk_size must be > 0")

 

    requested_schema = _validate_requested_schema(

        schema,

        source_columns,

    )

 

    with oracledb.connect(

        user=user,

        password=password,

        dsn=dsn,

        **dict(connect_kwargs or {}),

    ) as connection:

        batches = connection.fetch_df_batches(

            statement=select_sql,

            size=chunk_size,

            fetch_decimals=fetch_decimals,

            requested_schema=requested_schema,

        )

 

        iterator = iter(batches)

 

        while True:

            started_ns = time.perf_counter_ns()

 

            try:

                oracle_dataframe = next(iterator)

            except StopIteration:

                break

 

            table = pa.table(oracle_dataframe)

 

            table = _to_contract_table(

                table,

                schema,

            )

 

            fetch_ms = (

                time.perf_counter_ns() - started_ns

            ) / 1_000_000

 

            yield OracleArrowBatch(

                table=table,

                fetch_ms=fetch_ms,

            )

 

 

def _wrap_select_for_describe(

    select_sql: str,

) -> str:

    statement = select_sql.strip().rstrip(";")

    return (

        f"SELECT * FROM ({statement}) WHERE 1 = 0"

    )

 

 

async def oracle_get_column_names_thin_async(

    *,

    user: str,

    password: str,

    dsn: str,

    select_sql: str,

    connect_kwargs: Mapping[str, Any] | None = None,

    execute_params: Mapping[str, Any] | None = None,

    wrap_for_describe: bool = True,

) -> list[str]:

    statement = (

        _wrap_select_for_describe(select_sql)

        if wrap_for_describe

        else select_sql.strip().rstrip(";")

    )

 

    async with oracledb.connect_async(

        user=user,

        password=password,

        dsn=dsn,

        **dict(connect_kwargs or {}),

    ) as connection:

        async with connection.cursor() as cursor:

            await cursor.execute(

                statement,

                execute_params or {},

            )

 

            return [

                [column.name](http://column.name/)

                for column in cursor.description or ()

            ]

 

 

async def oracle_select_to_parquet(

    *,

    user: str,

    password: str,

    dsn: str,

    select_sql: str,

    out_parquet: str,

    schema: pa.Schema,

    source_columns: list[str],

    chunk_size: int,

    src_total_count: int,

    chunk_total_count: int,

    connect_kwargs: Mapping[str, Any] | None = None,

    compression: str = "zstd",

    row_group_rows: int | None = None,

    fetch_decimals: bool = False,

) -> None:

    await asyncio.to_thread(

        _oracle_select_to_parquet_sync,

        user=user,

        password=password,

        dsn=dsn,

        select_sql=select_sql,

        out_parquet=out_parquet,

        schema=schema,

        source_columns=source_columns,

        chunk_size=chunk_size,

        src_total_count=src_total_count,

        chunk_total_count=chunk_total_count,

        connect_kwargs=connect_kwargs,

        compression=compression,

        row_group_rows=row_group_rows,

        fetch_decimals=fetch_decimals,

    )

 

 

def _oracle_select_to_parquet_sync(

    *,

    user: str,

    password: str,

    dsn: str,

    select_sql: str,

    out_parquet: str,

    schema: pa.Schema,

    source_columns: list[str],

    chunk_size: int,

    src_total_count: int,

    chunk_total_count: int,

    connect_kwargs: Mapping[str, Any] | None,

    compression: str,

    row_group_rows: int | None,

    fetch_decimals: bool,

) -> None:

    writer = pq.ParquetWriter(

        out_parquet,

        schema,

        compression=compression,

    )

 

    rows_written = 0

 

    try:

        for batch in read_oracle_arrow_batches(

            user=user,

            password=password,

            dsn=dsn,

            select_sql=select_sql,

            schema=schema,

            source_columns=source_columns,

            chunk_size=chunk_size,

            connect_kwargs=connect_kwargs,

            fetch_decimals=fetch_decimals,

        ):

            rows_written += batch.table.num_rows

 

            writer.write_table(

                batch.table,

                row_group_size=(

                    row_group_rows or chunk_size

                ),

            )

    finally:

        writer.close()

The failing asynchronous operation is:

async with oracledb.connect_async(

    user=USER,

    password=PASSWORD,

    dsn=DSN,

) as connection:

    batches = connection.fetch_df_batches(

        statement=SQL,

        size=1000,

        fetch_decimals=True,

    )

 

    iterator = aiter(batches)

 

    # The interpreter terminates with SIGSEGV here.

    oracle_dataframe = await anext(iterator)

The equivalent synchronous operation succeeds:

with oracledb.connect(

    user=USER,

    password=PASSWORD,

    dsn=DSN,

) as connection:

    for oracle_dataframe in connection.fetch_df_batches(

        statement=SQL,

        size=1000,

        fetch_decimals=True,

   ):

        table = pa.table(oracle_dataframe)

        print(table.schema, table.num_rows)

Ordinary asynchronous row fetching also succeeds:

async with oracledb.connect_async(

    user=USER,

    password=PASSWORD,

    dsn=DSN,

) as connection:

    async with connection.cursor() as cursor:

        cursor.arraysize = 1000

        cursor.prefetchrows = 1000

 

        await cursor.execute(SQL)

 

        rows = await cursor.fetchmany(1000)

        print(len(rows))



The application workaround is to run the synchronous Connection.fetch_df_batches() implementation inside asyncio.to_thread(). This avoids the crash while retaining the dataframe and PyArrow path.

The environment is:

python-oracledb: 26.0.1

Python: 3.14.7

PyArrow: 25.0.1

Mode: Thin

Platform: Linux x86_64

Because my database access is read-only, I cannot run the upstream table-creation test directly against the affected database. However, the complete anonymized 20-column schema, synthetic table definition, query shape, result metadata, and Oracle helper implementation are included above.

[anthony-tuininga]

The code above is not complete and seems to contain multiple flavors. Can you supply the complete code that crashes? I will try it with the data you supplied and see if I can replicate the issue. Thanks! (And can you remove all of the extra blank lines -- there seems to be one for each line which makes the code harder to follow!).