Add incremental export and cross account export functionality in DynamoDBToS3Operator

[Ghoul-SSZ]

Problem

1. Currently the DynamoDBToS3Operator only supports 2 ways of downloading data: "full export using scan" or "full exports using boto3's export_table_to_point_in_time". When using export_table_to_point_in_time method, you can also use "Incremental Export" instead of "Full Export" if you only wish for the delta of the data changes between the specified period. However this functionality is currently not supported in the DynamoDBToS3Operator.
2. There are times we need to export the data from 1 AWS account (Account A) to another account (Account B). In the full export using scan method, you can specify 2 aws connection to facilitate this need. But for export_table_to_point_in_time, we would need an additional argument (s3_bucket_owner) to allow cross account export.

Proposed Changes

1. Replace the export_time arg with a boolean value make the Operator more visible that it is doing a point in time export.
2. Add export_type, incremental_export_from_time , incremental_export_to_time , incremental_export_view_type arguments to allow Incremental Export.
3. Add s3_bucket_owner arguments to allow cross-account export

closes: #40737
related: #40737

---

^ 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.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[vincbeck]

Very nice work!

[vincbeck]

Static checks are failing. Could you please run them

[Ghoul-SSZ]

I have ran and update the import so that the Static checks should pass.

However, I have noticed that documentation build and spell checks have failed in the GitHub Action for commit 0a195aa but are not related to my changes.

Should I try to fix those now or maybe open a different Pull Request for it?

[vincbeck]

I have ran and update the import so that the Static checks should pass.

However, I have noticed that documentation build and spell checks have failed in the GitHub Action for commit 0a195aa but are not related to my changes.

Should I try to fix those now or maybe open a different Pull Request for it?

I think it has been fixed by #41449, could you please update your fork and update this branch? It should solve the issue

[Ghoul-SSZ]

I have ran and update the import so that the Static checks should pass.
However, I have noticed that documentation build and spell checks have failed in the GitHub Action for commit 0a195aa but are not related to my changes.
Should I try to fix those now or maybe open a different Pull Request for it?

I think it has been fixed by #41449, could you please update your fork and update this branch? It should solve the issue

Looks like it is still failing. 😞

[vincbeck]

I resolved the doc building issue. Now, there is one unit test (Kubernetes provider) that is failing. It is definitely not related to this PR

[vincbeck]

But I am not sure what is going on either ....

[Ghoul-SSZ]

But I am not sure what is going on either ....

yeah it is strange. I will also try to look at it tomorrow when I have time

[vincbeck]

It has been fixed in #41500, please rebase your PR :)

[Ghoul-SSZ]

It has been fixed in #41500, please rebase your PR :)

All checks have passed! 🎉

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

[ferruzzi]

This change has caused the test to start failing. I have found at least three issues; two I have left the fix for in the comments, the third I'm not sure yet. @Ghoul-SSZ - can you have a look and try to make the fixes?

[ferruzzi]

Insert latest_export_time between these two lines

       export_time,
       latest_export_time
       backup_db_to_point_in_time_full_export,

Without it listed in the chain(), it fires off at the very beginning and causes the test to fail because env_id isn't populated yet.

[ferruzzi]

fixed in #41517

[ferruzzi]

this needs to be

self.export_table_to_point_in_time_kwargs = export_table_to_point_in_time_kwargs or {}

otherwise if None is passed in, Line 199 fails because None can't be unpacked.

[ferruzzi]

fixed in #41517

[ferruzzi]

This is an issue I have run out of time for today. with the other two fixes applied locally, the test keeps failing and I've tried a number of solutions. Here's what I've found so far:

error: "from" time can't be equal to "to" time
tried: so I set "from" to export_time - 1_minute
error: "from" can't be less than table creation time
tried: revert the above and set "to" to be latest_export_time + 1_minute
error: Difference between "from" time and "to" is less than 15 minutes
tried: revert the above and set "to" to be latest_export_time + 16_minutes
error: "to" time cannot be greater than the current time
tried: revert the above and set "to" time to utcnow()
error: this gets set at parse time and "to" time is now in the past

We will either need to figure out a way to make this happy or set that task to not run using an unreachable trigger_rule or a branch operator to skip it.

[vincbeck]

"error: Difference between "from" time and "to" is less than 15 minutes" is the most scary. That means we need to wait for 15 minutes so that we can create the incremental export. In that case we dont want to run it as part of the system test, so either we delete it from the system test, or we, somehow, skip it

[ferruzzi]

I agree. I just put up #41546 which skips that task. That way we can still have the snippet in the docs.