Return StructuredDataset which is a field in a dataclass #3071
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Signed-off-by: Nelson Chen <[email protected]>
Signed-off-by: Nelson Chen <[email protected]>
Signed-off-by: Nelson Chen <[email protected]>
arbaobao
requested review from
wild-endeavor,
kumare3,
eapolinario,
pingsutw,
cosmicBboy,
samhita-alla,
thomasjpfan and
Future-Outlier
as code owners
Code Review Agent Run #63793cActionable Suggestions - 2
Review Details
AI Code Review powered by |
Changelist by BitoThis pull request implements the following key changes.
|
flyte-bot
reviewed
Jan 21, 2025
Comment on lines
+738
to
+742
| if isinstance(python_val._literal_sd, StructuredDataset): | ||
| sdt = StructuredDatasetType(format=python_val._literal_sd.file_format) | ||
| metad = literals.StructuredDatasetMetadata(structured_dataset_type=sdt) | ||
| sd_literal = literals.StructuredDataset(uri=python_val._literal_sd.uri, metadata=metad) | ||
| return Literal(scalar=Scalar(structured_dataset=sd_literal)) |
There was a problem hiding this comment.
Private member access needs encapsulation
Accessing private member '_literal_sd'. Consider using a public interface or property to access this data.
Code suggestion
Check the AI-generated fix before applying
- if literal_type.structured_dataset_type is not None and self._literal_sd is not None:
- return self._literal_sd
- if literal_type.structured_dataset_type is not None and self._literal_sd is None:
+ if literal_type.structured_dataset_type is not None and self.literal_sd is not None:
+ return self.literal_sd
+ if literal_type.structured_dataset_type is not None and self.literal_sd is None:
Code Review Run #63793c
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
flyte-bot
reviewed
Jan 21, 2025
Comment on lines
+738
to
+742
| if isinstance(python_val._literal_sd, StructuredDataset): | ||
| sdt = StructuredDatasetType(format=python_val._literal_sd.file_format) | ||
| metad = literals.StructuredDatasetMetadata(structured_dataset_type=sdt) | ||
| sd_literal = literals.StructuredDataset(uri=python_val._literal_sd.uri, metadata=metad) | ||
| return Literal(scalar=Scalar(structured_dataset=sd_literal)) |
There was a problem hiding this comment.
Consider extracting literal creation logic
The code block for handling StructuredDataset passed through dataclass could be simplified by extracting the literal creation logic into a helper method. This would improve code readability and maintainability.
Code suggestion
Check the AI-generated fix before applying
Suggested change
| if isinstance(python_val._literal_sd, StructuredDataset): | |
| sdt = StructuredDatasetType(format=python_val._literal_sd.file_format) | |
| metad = literals.StructuredDatasetMetadata(structured_dataset_type=sdt) | |
| sd_literal = literals.StructuredDataset(uri=python_val._literal_sd.uri, metadata=metad) | |
| return Literal(scalar=Scalar(structured_dataset=sd_literal)) | |
| if isinstance(python_val._literal_sd, StructuredDataset): | |
| return self._create_structured_dataset_literal(python_val._literal_sd.uri, python_val._literal_sd.file_format) | |
| def _create_structured_dataset_literal(self, uri: str, file_format: str) -> Literal: | |
| sdt = StructuredDatasetType(format=file_format) | |
| metad = literals.StructuredDatasetMetadata(structured_dataset_type=sdt) | |
| sd_literal = literals.StructuredDataset(uri=uri, metadata=metad) | |
| return Literal(scalar=Scalar(structured_dataset=sd_literal)) |
Code Review Run #63793c
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
Signed-off-by: Nelson Chen <[email protected]>
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3071 +/- ##
==========================================
+ Coverage 78.20% 79.58% +1.38%
==========================================
Files 292 203 -89
Lines 25401 21599 -3802
Branches 2779 2780 +1
==========================================
- Hits 19864 17190 -2674
+ Misses 4726 3633 -1093
+ Partials 811 776 -35 ☔ View full report in Codecov by Sentry. |
|
When we return a StructuredDataset attribute from a dataclass instance, an error occurs : This issue has been addressed in this PR. |
Signed-off-by: Nelson Chen <[email protected]>
Signed-off-by: Nelson Chen <[email protected]>
Code Review Agent Run #d93af6Actionable Suggestions - 1
Additional Suggestions - 10
Review Details
AI Code Review powered by |
flyte-bot
reviewed
Feb 4, 2025
| @task | ||
| def read_sd(dc: DC) -> StructuredDataset: | ||
| """Read input StructuredDataset.""" | ||
| print("sd:", dc.sd.open(pd.DataFrame).all()) |
There was a problem hiding this comment.
Consider adding error handling for dataset access
Consider adding error handling around open() and all() calls to handle potential exceptions when accessing the structured dataset.
Code suggestion
Check the AI-generated fix before applying
Suggested change
| print("sd:", dc.sd.open(pd.DataFrame).all()) | |
| try: | |
| df = dc.sd.open(pd.DataFrame).all() | |
| print("sd:", df) | |
| except Exception as e: | |
| print(f"Error accessing structured dataset: {e}") | |
| raise |
Code Review Run #d93af6
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
Signed-off-by: Nelson Chen <[email protected]>
Code Review Agent Run #0d1278Actionable Suggestions - 0Additional Suggestions - 10
Review Details
AI Code Review powered by |
Signed-off-by: Nelson Chen <[email protected]>
Code Review Agent Run #7e2d92Actionable Suggestions - 3
Review Details
AI Code Review powered by |
flyte-bot
reviewed
Feb 7, 2025
| file_transfer = SimpleFileTransfer() | ||
| remote_file_path = file_transfer.upload_file(file_type="parquet") | ||
|
|
||
| execution_id = run("attr_access_dc_sd.py", "wf", "--uri", remote_file_path) |
There was a problem hiding this comment.
Consider validating remote file path parameter
Consider adding validation for the remote_file_path parameter before passing it to the run() function. The URI parameter should be validated to ensure it's a valid S3/Minio path.
Code suggestion
Check the AI-generated fix before applying
@@ -893,4 +893,7 @@
file_transfer = SimpleFileTransfer()
remote_file_path = file_transfer.upload_file(file_type="parquet")
+ if not remote_file_path or not remote_file_path.startswith('s3://') or not remote_file_path.endswith('.parquet'):
+ raise ValueError(f'Invalid remote file path: {remote_file_path}')
+
execution_id = run("attr_access_dc_sd.py", "wf", "--uri", remote_file_path)
Code Review Run #7e2d92
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
flyte-bot
reviewed
Feb 7, 2025
| Returns: | ||
| dc: A dataclass with a StructuredDataset attribute. | ||
| """ | ||
| dc = DC(sd=StructuredDataset(uri=uri, file_format="parquet")) |
There was a problem hiding this comment.
Consider validating parquet file format support
Consider adding validation for the file_format parameter. The code assumes parquet format but it may be worth checking if this format is supported by checking against DEFAULT_FORMATS or ENCODERS dictionaries.
Code suggestion
Check the AI-generated fix before applying
-def create_dc(uri: str) -> DC:
- dc = DC(sd=StructuredDataset(uri=uri, file_format="parquet"))
- return dc
+def create_dc(uri: str) -> DC:
+ file_format = "parquet"
+ if file_format not in DEFAULT_FORMATS.values() and file_format not in ENCODERS:
+ raise ValueError(f"File format {file_format} is not supported")
+ dc = DC(sd=StructuredDataset(uri=uri, file_format=file_format))
+ return dc
Code Review Run #7e2d92
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
flyte-bot
reviewed
Feb 7, 2025
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| wf(uri="tests/flytekit/integration/remote/workflows/basic/data/df.parquet") |
There was a problem hiding this comment.
Consider configurable parquet file path
Consider making the parquet file path configurable through environment variables or configuration files instead of hardcoding it. This would make the code more flexible and easier to maintain across different environments.
Code suggestion
Check the AI-generated fix before applying
Suggested change
| wf(uri="tests/flytekit/integration/remote/workflows/basic/data/df.parquet") | |
| import os | |
| default_path = "tests/flytekit/integration/remote/workflows/basic/data/df.parquet" | |
| wf(uri=os.getenv("PARQUET_FILE_PATH", default_path)) |
Code Review Run #7e2d92
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
Signed-off-by: Nelson Chen <[email protected]>
Code Review Agent Run #d69fbdActionable Suggestions - 1
Review Details
AI Code Review powered by |
flyte-bot
reviewed
Feb 7, 2025
| @workflow | ||
| def wf(uri: str) -> None: | ||
| dc = create_dc(uri=uri) | ||
| read_sd(dc=dc) |
There was a problem hiding this comment.
Consider storing read_sd return value
Consider storing the return value of read_sd() since it returns a StructuredDataset. The returned value might be needed later in the workflow.
Code suggestion
Check the AI-generated fix before applying
Suggested change
| read_sd(dc=dc) | |
| sd = read_sd(dc=dc) |
Code Review Run #d69fbd
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
Signed-off-by: Nelson Chen <[email protected]>
Code Review Agent Run #b95e33Actionable Suggestions - 0Additional Suggestions - 10
Review Details
AI Code Review powered by |
|
@Future-Outlier |
Code Review Agent Run #425bc0Actionable Suggestions - 0Additional Suggestions - 6
Review Details
AI Code Review powered by |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Tracking issue
Related to #6117
Why are the changes needed?
If we wrap the StructuredDataset in a dataclass, it will fail during the to_flyte_idl conversion.
What changes were proposed in this pull request?
Before returning
Literals, we check the type ofpython_val._literal_sd. If it is a Python nativeStructuredDataset, we transform it into aLiterals.StructuredDataset.How was this patch tested?
As described in #6117, an error occurs when the
extracttask is executed.Setup process
Screenshots
Check all the applicable boxes
Summary by Bito
This PR implements comprehensive improvements to Flytekit's core functionality, including StructuredDataset handling, caching system enhancements, and a new rate limiting system. The changes transition from in-memory DataFrame to file-based approach, integrate Kubernetes StatefulSet Data Service and Ray plugins, while improving dictionary and annotated type handling. Key improvements include enhanced error handling, better array node handling in remote execution, improved workflow execution tracking, and strengthened input validation across multiple components.Unit tests added: True
Estimated effort to review (1-5, lower is better): 5