Fix issues with pickling & unpickling the Table object

[utkarsharma2]

Description

What is the current behavior?

Currently, Python-sdk is falling with pickling error.

airflow.exceptions.AirflowException: When using a synchronous executor (e.g. SequentialExecutor and DebugExecutor), the first run of this task will fail on purpose, so the single worker thread is unblocked to execute other tasks. The task is set up for retry and eventually works.
[2022-09-16 22:52:28,847] {backfill_job.py:190} ERROR - Task instance <TaskInstance: example_google_bigquery_gcs_load_and_save.cleanup backfill__2016-01-01T00:00:00+00:00 [failed]> failed
[2022-09-16 22:52:28,851] {backfill_job.py:370} INFO - [backfill progress] | finished run 0 of 1 | tasks waiting: 3 | succeeded: 1 | running: 0 | failed: 1 | skipped: 0 | deadlocked: 0 | not ready: 3
[2022-09-16 22:52:28,865] {base_executor.py:95} INFO - Adding to queue: ['<TaskInstance: example_google_bigquery_gcs_load_and_save.extract_top_5_movies backfill__2016-01-01T00:00:00+00:00 [queued]>']
[2022-09-16 22:52:28,872] {base_executor.py:95} INFO - Adding to queue: ['<TaskInstance: example_google_bigquery_gcs_load_and_save.save_file_to_gcs backfill__2016-01-01T00:00:00+00:00 [queued]>']
[2022-09-16 22:52:28,901] {debug_executor.py:85} ERROR - Failed to execute task: 'utf-8' codec can't decode byte 0x80 in position 0: invalid start byte.
Traceback (most recent call last):
  File "/Users/utkarsharma/sandbox/astronomer/astro/.nox/dev/lib/python3.8/site-packages/airflow/models/xcom.py", line 618, in deserialize_value
    return pickle.loads(result.value)
_pickle.UnpicklingError: state is not a dictionary

What is the new behavior?

To fix the above issue we need to add getstate() to the table class

Does this introduce a breaking change?

Nope

Checklist

• Created tests that fail without the change (if possible)
• Extended the README/documentation, if necessary

[kaxil]

Let's add this test for File object. The reason self.__dict__ is not available for File is because it uses slots.

A class whose instances have no object.dict attribute and define their attributes in a object.slots attribute instead. In attrs, they are created by passing slots=True to @attr.s (and are on by default in attrs.define()/attrs.mutable()/attrs.frozen()).

[utkarsharma2]

Got it

[kaxil]

But atleast this test will verify that you can pickle and unpickle the object.

[kaxil]

Would the following be a better test?

table = Table(name="test")
assert pickle.loads(pickle.dumps(table)) == table

cc @ashb

[utkarsharma2]

@kaxil Ya this seems better.

[kaxil]

Approving so you are not blocked but please address the review comments