Just trying daft for the first time. encountered an issue while trying to read data from mysql and writing to parquet.

connectorx_url = f"mysql://{user}:{pwd}@{host}:3306/{db}"
qry = "select * from loan_status_archive"
df = daft.read_sql(qry,conn=connectorx_url, partition_col='date')
df.exclude('followup_date')
#df.explain(show_all=True)
df.write_parquet(f"{s3_root_path}/loan_status_archive/", partition_cols=['date'])

Got this error message, looks like an issue with a date field. Is there a config to ignore and pass it as NULL if parsing fails?

ScanWithTask-WriteFile [Stage:1]:   0%|                                                                                                                    | 0/1 [00:00<?, ?it/s]thread '<unnamed>' panicked at /Users/runner/.cargo/registry/src/index.crates.io-6f17d22bba15001f/mysql_common-0.29.2/src/value/convert/mod.rs:175:23:
Could not retrieve core::option::Option<chrono::naive::date::NaiveDate> from Value
stack backtrace:
   0:        0x289223a00 - <std::sys_common::backtrace::_print::DisplayBacktrace as core::fmt::Display>::fmt::h01b2beffade888b2
   1:        0x289248380 - core::fmt::write::hbadb443a71b75f23
   2:        0x28922049c - std::io::Write::write_fmt::hc09d7755e3ead5f0
   3:        0x289223858 - std::sys_common::backtrace::print::h28349e5c25acbac7
   4:        0x289224d6c - std::panicking::default_hook::{{closure}}::hd24b6196784d991e
   5:        0x289224a50 - std::panicking::default_hook::hfcec80a2720c8c73
   6:        0x289225660 - std::panicking::rust_panic_with_hook::h84760468187ddc85
   7:        0x28922504c - std::panicking::begin_panic_handler::{{closure}}::he666a5eb600a7203
   8:        0x289223e84 - std::sys_common::backtrace::__rust_end_short_backtrace::h592f44d2bf9f843f
   9:        0x289224dc4 - _rust_begin_unwind
  10:        0x2892e7204 - core::panicking::panic_fmt::h98bbf7bdf4994454
  11:        0x2872bc684 - mysql_common::value::convert::from_value::h2ebdd73b2998f593
  12:        0x287250f2c - <connectorx::sources::mysql::MySQLBinarySourceParser as connectorx::sources::Produce<core::option::Option<chrono::naive::date::NaiveDate>>>::produce::hdf90e99c023032d0
  13:        0x287291b48 - core::ops::function::FnOnce::call_once::hb0b54048643d534e
  14:        0x28721d8b0 - core::ops::function::impls::<impl core::ops::function::FnOnce<A> for &mut F>::call_once::h24ea6dcffc98604d
  15:        0x286ff4ad8 - <rayon::iter::map::MapFolder<C,F> as rayon::iter::plumbing::Folder<T>>::consume_iter::h134209a6600690ac
  16:        0x286fec960 - rayon::iter::plumbing::bridge_producer_consumer::helper::h2b4384e8f1933f20
  17:        0x2873a9ab0 - <rayon::vec::IntoIter<T> as rayon::iter::IndexedParallelIterator>::with_producer::hfe665cbc3b600f2c
  18:        0x286fead88 - rayon::iter::try_reduce::try_reduce::h5da963039029df61
  19:        0x287243560 - connectorx::dispatcher::Dispatcher<S,D,TP>::run::hc17ba4a9db68c78e
  20:        0x287211b54 - connectorx::get_arrow::get_arrow::h0bc066a997bd9884
  21:        0x286f0be34 - pyo3::marker::Python::allow_threads::hfe70931472756e74
  22:        0x286f31b64 - connectorx::cx_read_sql::read_sql::hc27af57bcd4b141e
  23:        0x286f1681c - connectorx::__pyfunction_read_sql::hc766ed60ff97d445
  24:        0x286f08ef8 - pyo3::impl_::trampoline::trampoline::hbeffdb339cea3fbc
  25:        0x286f165c0 - connectorx::<impl connectorx::read_sql::MakeDef>::_PYO3_DEF::trampoline::h5a2e63a96480a846
  26:        0x104722c60 - _cfunction_vectorcall_FASTCALL_KEYWORDS
  27:        0x104804e40 - __PyEval_EvalFrameDefault
  28:        0x1046bf83c - __PyFunction_Vectorcall
  29:        0x11a244568 - <pyo3::instance::Bound<pyo3::types::any::PyAny> as pyo3::types::any::PyAnyMethods>::call::inner::h8a371ffc427b70a4
  30:        0x1199c0958 - daft_micropartition::python::read_sql_into_py_table::h72a2c0e1abb94d89
  31:        0x1199bacd0 - daft_micropartition::micropartition::materialize_scan_task::h1cf90f7d1ddb7343
  32:        0x1199dc188 - daft_micropartition::python::PyMicroPartition::__pymethod_from_scan_task__::he498cac3e1010c57
  33:        0x119e45158 - pyo3::impl_::trampoline::trampoline::h04d90781a7e64b09
  34:        0x119758108 - daft_micropartition::python::_::__INVENTORY::trampoline::heaa86a1a83a3553a
  35:        0x104721fb0 - _cfunction_call
  36:        0x104805034 - __PyEval_EvalFrameDefault
  37:        0x1046bf83c - __PyFunction_Vectorcall
  38:        0x1046c54a0 - _method_vectorcall
  39:        0x104809dec - __PyEval_EvalFrameDefault
  40:        0x1046bf83c - __PyFunction_Vectorcall
  41:        0x104809dec - __PyEval_EvalFrameDefault
  42:        0x1046bf83c - __PyFunction_Vectorcall
  43:        0x1046c5580 - _method_vectorcall
  44:        0x1048f074c - _thread_run
  45:        0x10487d9fc - _pythread_wrapper
  46:        0x18b093034 - __pthread_joiner_wake

Hey @sravan kumar Chapala , this looks like an issue with ConnectorX, not daft, so it won’t be a straightforward fix. In the meantime, could you try using a sqlalchemy connection instead? Something like def create_conn(): return sqlalchemy.create_engine(DB_URL).connect() df = daft.read_sql("SELECT * FROM my_table", create_conn)

@Colin Ho Changed the code to use sqlalchemy connection

connectorx_url = f"mysql+pymysql://{user}:{pwd}@{host}:3306/{db}"

def create_conn():
    return sqlalchemy.create_engine(connectorx_url).connect()

qry = "select * from loan_status_archive"
df = daft.read_sql(qry,conn=create_conn, partition_col='date')
#df.exclude('followup_date')
#df.explain(show_all=True)
df.write_parquet(f"{s3_root_path}/loan_status_archive/", partition_cols=['date'])

Got this error:

File "//VSProjects/arca_dagster_dbt/adhoc/try_daft.py", line 37, in <module>
    df.write_parquet(f"{s3_root_path}/loan_status_archive/", partition_cols=['date'])
  File "//miniconda3/lib/python3.11/site-packages/daft/api_annotations.py", line 26, in _wrap
    return timed_method(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/site-packages/daft/analytics.py", line 202, in tracked_method
    result = method(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/site-packages/daft/dataframe/dataframe.py", line 554, in write_parquet
    write_df.collect()
  File "//miniconda3/lib/python3.11/site-packages/daft/api_annotations.py", line 26, in _wrap
    return timed_method(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/site-packages/daft/analytics.py", line 202, in tracked_method
    result = method(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/site-packages/daft/dataframe/dataframe.py", line 2440, in collect
    self._materialize_results()
  File "//miniconda3/lib/python3.11/site-packages/daft/dataframe/dataframe.py", line 2422, in _materialize_results
    self._result_cache = context.runner().run(self._builder)
                         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/site-packages/daft/runners/pyrunner.py", line 265, in run
    results = list(self.run_iter(builder))
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/site-packages/daft/runners/pyrunner.py", line 330, in run_iter
    yield from results_gen
  File "//miniconda3/lib/python3.11/site-packages/daft/runners/pyrunner.py", line 494, in _physical_plan_to_partitions
    materialized_results = done_future.result()
                           ^^^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/concurrent/futures/_base.py", line 449, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "//miniconda3/lib/python3.11/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/site-packages/daft/runners/pyrunner.py", line 570, in build_partitions
    partitions = instruction.run(partitions)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/site-packages/daft/execution/execution_step.py", line 311, in run
    return self._scan(inputs)
           ^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/site-packages/daft/execution/execution_step.py", line 315, in _scan
    table = MicroPartition._from_scan_task(self.scan_task)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "//miniconda3/lib/python3.11/site-packages/daft/table/micropartition.py", line 71, in _from_scan_task
    return MicroPartition._from_pymicropartition(_PyMicroPartition.from_scan_task(scan_task))
                                                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
daft.exceptions.DaftCoreException: DaftError::External PyIOError: RuntimeError: Failed to execute sql: SELECT * FROM (SELECT * FROM loan_status_archive) AS subquery WHERE date >= CAST('2024-04-11' AS DATE) AND date < CAST('2024-04-14' AS DATE) from connection: create_conn, error: object of type <class 'str'> cannot be converted to int

So, this error happens when Daft is reading the results into an Arrow Table via PyArrow, which is trying to cast the string to an int. I think this is happening because there exists some rows in (likely) a date column that are strings or being read as strings. I also think that this is why the original error with connectorX happened. Minimal reproducible example:

pydict = {
    "a": [datetime.date.today(), datetime.date.today(), "2021-01-01"],
}
table = pa.Table.from_pydict(pydict)

pyarrow.lib.ArrowTypeError: object of type <class 'str'> cannot be converted to int

I suggest modifying the original SQL query to filter out these rows or cast them to dates first!