feat/databricks delta table destination connector

[rbiseck3]

Description

Adds in support for writing to databricks delta tables in two different ways:

• Adds in Databricks delta table destination connector, building off of the existing sql foundation.
• Adds support for supplemental table migration in the databricks volumes connector

[micmarty-deepsense]

It seems that destination entry in unstructured_ingest/v2/processes/connectors/sql/__init__.py is missing

[micmarty-deepsense]

I know this line hasn't changed, but I think it's worth double-checking to ensure it's the solution we want to keep. Using SELECT * doesn't seem optimal for just fetching column descriptions (I might be wrong). Perhaps we could opt for something like this:

cursor.execute("DESCRIBE `some_table_name`")

which produces output like this:

[Row(col_name='id', data_type='string', comment=None),
 Row(col_name='record_id', data_type='string', comment=None),
 Row(col_name='element_id', data_type='string', comment=None),
 Row(col_name='text', data_type='string', comment=None),
 Row(col_name='embeddings', data_type='array<float>', comment=None),
 Row(col_name='type', data_type='string', comment=None),
 Row(col_name='url', data_type='string', comment=None),
 Row(col_name='version', data_type='string', comment=None),
 Row(col_name='data_source_date_created', data_type='timestamp', comment=None),
 Row(col_name='data_source_date_modified', data_type='timestamp', comment=None),
 Row(col_name='data_source_date_processed', data_type='timestamp', comment=None),
 Row(col_name='data_source_permissions_data', data_type='string', comment=None),
 Row(col_name='data_source_filesize_bytes', data_type='float', comment=None),
 Row(col_name='data_source_url', data_type='string', comment=None),
 Row(col_name='data_source_version', data_type='string', comment=None),
 Row(col_name='data_source_record_locator', data_type='string', comment=None),
 Row(col_name='category_depth', data_type='int', comment=None),
 Row(col_name='parent_id', data_type='string', comment=None),
 Row(col_name='attached_filename', data_type='string', comment=None),
 Row(col_name='filetype', data_type='string', comment=None),
 Row(col_name='last_modified', data_type='timestamp', comment=None),
 Row(col_name='file_directory', data_type='string', comment=None),
 Row(col_name='filename', data_type='string', comment=None),
 Row(col_name='languages', data_type='array<string>', comment=None),
 Row(col_name='page_number', data_type='string', comment=None),
 Row(col_name='links', data_type='string', comment=None),
 Row(col_name='page_name', data_type='string', comment=None),
 Row(col_name='link_urls', data_type='string', comment=None),
 Row(col_name='link_texts', data_type='string', comment=None),
 Row(col_name='sent_from', data_type='string', comment=None),
 Row(col_name='sent_to', data_type='string', comment=None),
 Row(col_name='subject', data_type='string', comment=None),
 Row(col_name='section', data_type='string', comment=None),
 Row(col_name='header_footer_type', data_type='string', comment=None),
 Row(col_name='emphasized_text_contents', data_type='string', comment=None),
 Row(col_name='emphasized_text_tags', data_type='string', comment=None),
 Row(col_name='text_as_html', data_type='string', comment=None),
 Row(col_name='regex_metadata', data_type='string', comment=None),
 Row(col_name='detection_class_prob', data_type='float', comment=None),
 Row(col_name='is_continuation', data_type='boolean', comment=None),
 Row(col_name='orig_elements', data_type='string', comment=None),
 Row(col_name='coordinates_points', data_type='string', comment=None),
 Row(col_name='coordinates_system', data_type='string', comment=None),
 Row(col_name='coordinates_layout_width', data_type='float', comment=None),
 Row(col_name='coordinates_layout_height', data_type='float', comment=None)]

[rbiseck3]

I don't actually know when this was introduced and I'm more than happy to refactor to something more optimal to fetch column data.

[micmarty-deepsense]

arrays are not correctly inserted; prepare_data converts lists into strings.

[upload] [DATATYPE_MISMATCH.CAST_WITHOUT_SUGGESTION] Cannot resolve "languages" due to data type mismatch: cannot cast "STRING" to "ARRAY<STRING>". SQLSTATE: 42K09; line 1 pos 0

The solution is to either change the SQL statement for insertion so that we can use ARRAY as a keyword, or we change the type to STRING (which worked for me)

[micmarty-deepsense]

actually there's also an issue with embeddings column
https://github.com/databricks/databricks-sql-python/blob/01e998cabe4a4972d5b45f7936219d924706c75c/docs/parameters.md#migrating-to-native-parameters

[micmarty-deepsense]

Unfortunately this SDK seems to not support array parameters:

I don't see any other way that would be safe (not prone to SQL injection) and would allow for having cusor.execute and list of parameters at the same time.

[rbiseck3]

Ya I've reached out to a contact at Databricks. Hopefully we'll hear a work around. Otherwise this entire approach might need to be redone.

[micmarty-deepsense]

(comment not tied to any particular line, but relevant to this file)
I was getting "column does not exist" type of errors, because the schema expects a bunch of columns to have the data_source_ prefix.

We'd need to adjust column names here and there, I guess

[rbiseck3]

@micmarty-deepsense this PR is still in the works as we wait to hear back from databricks on getting a solution for variable binding and using list data.

[micmarty-deepsense]

I've also run into errors saying that the Timestamp is not JSON serializable. To fix it, change this line to:

json.dump(data, f, indent=2, default=str)

... but I believe it's not a good place to fix this 😛

Here's my temporary dev branch that attempts to naively address some of the obstacles I faced when testing this PR