[Data] Add Dataset.write_sql

[bveeramani]

Why are these changes needed?

Writing data back to databases is common for many applications like LLMs. For example, you might want to write vector indices back to a database like https://github.com/pgvector/pgvector. To support this use case, this PR adds an API to write Datasets to SQL databases.

Related issue number

Closes #38242

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[c21]

Looks solid to me as a starting point. Can you publish and trigger the wheel build, so we can test it out in real workload?

[pcmoritz]

So this is now working for me, thanks a lot for fixing it!

One thing we should improve before merging it is the error handling. I noticed if the table does not exist, it will just silently fail and write nothing which is not good and can lead to data loss. To fix it I think we should do two things: Improve the error handling and catch exceptions better, and also potentially return some info about how many rows have been written (e.g. in the form of a status / info dict that we could add more stuff to in the future). The former we should do for sure, the latter only if it doesn't involve too many changes for now :)

Here is a repro of the silent failure in sqlite3 -- let's also add a test for it:

In [1]: import sqlite3

In [2]: import ray

In [3]: connection = sqlite3.connect("/tmp/db.sqlite")

In [4]: dataset = ray.data.from_items(
   ...:         [{"string": "spam", "number": 0}, {"string": "ham", "number": 1}]
   ...:     )

In [5]: dataset.write_sql(
   ...:         "INSERT INTO test VALUES(?, ?)", lambda: sqlite3.connect("/tmp/db.sqlite"))

[bveeramani]

@pcmoritz fixed the issue with error messages. Here's what the error looks like now:

import sqlite3
import ray

connection = sqlite3.connect("/tmp/eggs.db")
dataset = ray.data.range(1)

dataset.write_sql(
    "INSERT INTO test VALUES(?)", lambda: sqlite3.connect("/tmp/eggs.db")
)

Traceback (most recent call last):
  ...
  File "/Users/balaji/Documents/GitHub/ray/python/ray/data/datasource/sql_datasource.py", line 55, in write
    cursor.executemany(sql, values)
sqlite3.OperationalError: no such table: test