Rewrite how dag to dataset / dataset alias are stored

[Lee-W]

Why

If someone subclasses a dataset or a dataset alias and makes it unhashable. The dag ref will not be able to be handled.

What

Save the data needed as (type, URI or name) format in dag ref

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[potiuk]

Cautious approve :)

[uranusjr]

Doesn’t Dataset do this automatically? We shouldn’t let an unsanitised URI get into the system in the first place.

airflow/airflow/datasets/__init__.py

Lines 248 to 251 in cb7697f

	uri: str = attr.field(
	converter=_sanitize_uri,
	validator=[attr.validators.min_len(1), attr.validators.max_len(3000)],
	)

[Lee-W]

Doesn’t Dataset do this automatically? We shouldn’t let an unsanitised URI get into the system in the first place.

airflow/airflow/datasets/__init__.py

Lines 248 to 251 in cb7697f

	uri: str = attr.field(
	converter=_sanitize_uri,
	validator=[attr.validators.min_len(1), attr.validators.max_len(3000)],
	)

Yep, but new_datasets are DatasetModels instead of Dataset

[uranusjr]

I was thinking it’s weird we’re using DatasetModel here… (DatasetManager actually converts them back to Dataset for a user hook too)

I was having problems with this function when adding name too. Let me rewrite this entire function a bit.

[Lee-W]

I was thinking it’s weird we’re using DatasetModel here… (DatasetManager actually converts them back to Dataset for a user hook too)

I was having problems with this function when adding name too. Let me rewrite this entire function a bit.

I think the main reason is that we need to check the stored dataset models. Do you want me to continue working on this one? Or wait for your PR? Without it, some edges cases might fail

[uranusjr]

Hmm, wait, this still does not make sense even considering we’re using DatasetModel. d here is a Dataset, so d.uri should have already gone through _sanitize_uri when d is created. Is converter somehow not functioning the way it’s assumed to? Is there a bug somewhere else?

It’s probably best to come up with a test case here first to identify the actual issue. I don’t think calling _sanitize_uri is the correct thing to do, but it’s difficult to tell what the right thing is without an actual bug.

[Lee-W]

Hmm, wait, this still does not make sense even considering we’re using DatasetModel. d here is a Dataset, so d.uri should have already gone through _sanitize_uri when d is created. Is converter somehow not functioning the way it’s assumed to? Is there a bug somewhere else?

It’s probably best to come up with a test case here first to identify the actual issue. I don’t think calling _sanitize_uri is the correct thing to do, but it’s difficult to tell what the right thing is without an actual bug.

https://github.com/astronomer/astro-sdk/blob/main/python-sdk/example_dags/example_datasets.py

This is the DAG that fails and yep, I think the solution is incorrect and that's why I convert it to draft again

[Lee-W]

I think I found the root cause. will push in short

[Lee-W]

Hmm, wait, this still does not make sense even considering we’re using DatasetModel. d here is a Dataset, so d.uri should have already gone through _sanitize_uri when d is created. Is converter somehow not functioning the way it’s assumed to? Is there a bug somewhere else?

It’s probably best to come up with a test case here first to identify the actual issue. I don’t think calling _sanitize_uri is the correct thing to do, but it’s difficult to tell what the right thing is without an actual bug.

This was due to some dataset/dataset alias subclass might not be hashable. I rewrote how the dag ref was handled.

[uranusjr]

If this is already passing, maybe we can also try to merge it first and do the related changes in a different PR. Do you think this is ready to be merged?

[Lee-W]

If this is already passing, maybe we can also try to merge it first and do the related changes in a different PR. Do you think this is ready to be merged?

Yep, I think this one is ready to be merged

[uranusjr]

One nit (actually two but on the same topic), looks good to me.

[uranusjr]

Meh, I’m going to just merge this since I have plans to refactor the entire function anyway. I can change the thing I want when that happens.

[Lee-W]

Thanks! @uranusjr I think this fix is needed for 2.10 so I'm going to backport now