feat: compute tableDependencies #582
Conversation
WalkthroughThis pull request enhances the Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant Writer as _write_obj
participant Dep as add_dependencies
participant JSON as _write_obj_as_json
Client->>Writer: Call _write_obj(obj)
Writer->>Dep: add_dependencies(obj)
Dep-->>Writer: Dependencies processed
Writer->>JSON: _write_obj_as_json(obj)
JSON-->>Writer: JSON written
Poem
Warning Review ran into problems🔥 ProblemsGitHub Actions and Pipeline Checks: Resource not accessible by integration - https://docs.github.com/rest/actions/workflow-runs#list-workflow-runs-for-a-repository. Please grant the required permissions to the CodeRabbit GitHub App under the organization or repository settings. 📜 Recent review detailsConfiguration used: CodeRabbit UI 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
⏰ Context from checks skipped due to timeout of 90000ms (2)
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
api/python/ai/chronon/group_by.py (1)
343-343: Debug print statement should use logging.Replace print with proper logging to follow best practices.
- print(agg, group_by.metaData.name) + LOGGER.debug(f"Validating aggregation {agg} for group_by {group_by.metaData.name}")api/python/ai/chronon/dag_builder.py (1)
1-1: Unused import.List is imported but never used.
-from typing import List, Union +from typing import Union🧰 Tools
🪛 Ruff (0.8.2)
1-1:
typing.Listimported but unusedRemove unused import:
typing.List(F401)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro (Legacy)
📒 Files selected for processing (3)
api/python/ai/chronon/dag_builder.py(1 hunks)api/python/ai/chronon/group_by.py(4 hunks)api/python/ai/chronon/join.py(3 hunks)
🧰 Additional context used
🧬 Code Definitions (2)
api/python/ai/chronon/group_by.py (1)
api/python/ai/chronon/dag_builder.py (1)
to_dependency(11-54)
api/python/ai/chronon/dag_builder.py (2)
api/python/ai/chronon/group_by.py (3)
Window(245-246)TimeUnit(179-181)GroupBy(431-678)api/python/ai/chronon/join.py (2)
JoinPart(31-109)BootstrapPart(360-395)
🪛 Ruff (0.8.2)
api/python/ai/chronon/dag_builder.py
1-1: typing.List imported but unused
Remove unused import: typing.List
(F401)
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: non_spark_tests
- GitHub Check: non_spark_tests
🔇 Additional comments (13)
api/python/ai/chronon/join.py (4)
26-26: Added import for dependency tracking.Good addition of dag_builder module for table dependency tracking.
515-520: Adds robust default handling for parameters.Properly initializes empty lists when parameters are None, preventing potential NoneType errors.
573-574: Creates table dependencies for tracking.Correctly constructs dependency list by applying to_dependency to left, right_parts, and bootstrap_parts.
582-582: Adds table dependencies to ExecutionInfo.Properly integrates dependency information into the execution context.
api/python/ai/chronon/group_by.py (3)
25-25: Added import for dependency tracking.Good addition for consistency with join.py changes.
635-635: Creates table dependencies from sources.Good implementation of dependency tracking consistent with join.py.
646-646: Adds table dependencies to ExecutionInfo.Properly integrates dependency information into the execution context.
api/python/ai/chronon/dag_builder.py (6)
2-6: Good imports and organization.Properly imports necessary types and utilities.
8-10: Clear docstring.Concise explanation of the function's purpose.
11-15: Handles JoinPart type recursively.Correctly extracts GroupBy dependency from JoinPart.
15-24: Handles GroupBy type.Properly constructs TableDependency with metaData.executionInfo.outputTableInfo.
25-38: Handles BootstrapPart type.Correctly extracts table and partition information to build TableDependency.
39-54: Handles Source type.Uses utility functions to extract table information and builds dependency correctly.
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (3)
api/python/ai/chronon/dag_builder.py (3)
5-5: Remove unused importJoinSource is imported but never used in the code.
-from ai.chronon.api.ttypes import BootstrapPart, GroupBy, JoinPart, Source, JoinSource +from ai.chronon.api.ttypes import BootstrapPart, GroupBy, JoinPart, Source🧰 Tools
🪛 Ruff (0.8.2)
5-5:
ai.chronon.api.ttypes.JoinSourceimported but unusedRemove unused import:
ai.chronon.api.ttypes.JoinSource(F401)
10-10: Unused parameter in function signatureThe lag parameter is defined but never used in function body.
Either utilize the lag parameter or remove it from signature if not needed.
40-57: Source node dependency handling lacks explicit type checkAssumes 'else' case is a Source without verifying type.
- else: # When type of node is a Source + elif isinstance(node, Source):Add an else clause to handle unexpected types:
+ else: + raise TypeError(f"Expected Source, GroupBy, JoinPart, or BootstrapPart, got {type(node).__name__}")
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro (Legacy)
📒 Files selected for processing (3)
api/python/ai/chronon/dag_builder.py(1 hunks)api/python/ai/chronon/group_by.py(3 hunks)api/python/ai/chronon/join.py(3 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- api/python/ai/chronon/group_by.py
🧰 Additional context used
🧬 Code Definitions (2)
api/python/ai/chronon/dag_builder.py (2)
api/python/ai/chronon/group_by.py (3)
TimeUnit(179-181)Window(245-246)GroupBy(430-677)api/python/ai/chronon/join.py (2)
BootstrapPart(360-395)JoinPart(31-109)
api/python/ai/chronon/join.py (1)
api/python/ai/chronon/dag_builder.py (1)
to_dependency(10-57)
🪛 Ruff (0.8.2)
api/python/ai/chronon/dag_builder.py
5-5: ai.chronon.api.ttypes.JoinSource imported but unused
Remove unused import: ai.chronon.api.ttypes.JoinSource
(F401)
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: non_spark_tests
- GitHub Check: non_spark_tests
🔇 Additional comments (8)
api/python/ai/chronon/join.py (4)
24-24: Import module added for dependency generationThe import adds dag_builder module which provides the to_dependency function used for creating table dependencies.
515-520: Good defensive programming with null parameter handlingEnsures None parameters have proper defaults, preventing potential NullPointerExceptions.
574-576: Adds table dependency trackingCreates dependency list from left source, right parts, and bootstrap parts for improved dependency management.
584-584: Dependency propagation to ExecutionInfoAdds table dependencies to the execution info object to propagate dependency information.
api/python/ai/chronon/dag_builder.py (4)
1-9: New module for DAG dependency conversionIntroduces utilities for converting various node types to TableDependency objects.
🧰 Tools
🪛 Ruff (0.8.2)
5-5:
ai.chronon.api.ttypes.JoinSourceimported but unusedRemove unused import:
ai.chronon.api.ttypes.JoinSource(F401)
10-14: JoinPart dependency handlingCorrectly recursively resolves JoinPart dependencies to their underlying GroupBy.
14-23: GroupBy dependency handlingCreates TableDependency with appropriate metadata from the GroupBy node.
24-39: BootstrapPart dependency handlingProperly extracts partition information from bootstrap query when available.
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (4)
api/python/ai/chronon/dag_builder.py (4)
8-11: Expand docstringDocstring needs parameter and return type descriptions.
25-40: Consider refactoring overlapping logicBootstrapPart and Source handling contain similar code patterns.
49-49: Redundant null checkQuery already obtained on line 43, null check unnecessary.
- start_cutoff = query.startPartition if query is not None else None + start_cutoff = query.startPartition if hasattr(query, 'startPartition') else None
50-50: Redundant null checkQuery already obtained on line 43, null check unnecessary.
- end_cutoff = query.endPartition if query is not None else None + end_cutoff = query.endPartition if hasattr(query, 'endPartition') else None
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro (Legacy)
📒 Files selected for processing (3)
api/python/ai/chronon/dag_builder.py(1 hunks)api/python/ai/chronon/group_by.py(3 hunks)api/python/ai/chronon/join.py(3 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- api/python/ai/chronon/join.py
- api/python/ai/chronon/group_by.py
🧰 Additional context used
🧬 Code Definitions (1)
api/python/ai/chronon/dag_builder.py (2)
api/python/ai/chronon/group_by.py (3)
TimeUnit(179-181)Window(245-246)GroupBy(430-676)api/python/ai/chronon/join.py (1)
BootstrapPart(360-395)
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: non_spark_tests
- GitHub Check: non_spark_tests
🔇 Additional comments (1)
api/python/ai/chronon/dag_builder.py (1)
1-58: Add tests for new functionAdd unit tests to verify all node type conversions.
Co-authored-by: Thomas Chow <[email protected]>
tchow-zlai
force-pushed
the
tchow/table-dependencies
branch
from
fbe31ca to
30ca1a4
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (3)
api/python/ai/chronon/dag_builder.py (3)
42-42: Add type verification for SourceThe else block assumes the node is a Source without explicit verification.
- else: # When type of node is a Source + elif isinstance(node, Source): + else: + raise TypeError(f"Unexpected node type: {type(node).__name__}. Expected Source, GroupBy, JoinPart, or BootstrapPart.")
1-60: Missing comprehensive docstringThe function would benefit from a detailed docstring explaining parameters, return values, and behavior.
def to_dependency(node: Union[Source, GroupBy, JoinPart, BootstrapPart], lag: int = 0) -> TableDependency: + """ + Converts a node into a TableDependency object. + + Parameters: + - node: A node of type Source, GroupBy, JoinPart, or BootstrapPart. + - lag: An integer representing the lag (default is 0). + + Returns: + - A TableDependency object representing the dependency of the provided node. + + Raises: + - TypeError: If the node is not one of the expected types. + """
17-25: Consider parameterizing forceComputeThe
forceComputeflag is hardcoded to False. Consider making it a parameter if different use cases need different values.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro (Legacy)
📒 Files selected for processing (3)
api/python/ai/chronon/dag_builder.py(1 hunks)api/python/ai/chronon/group_by.py(3 hunks)api/python/ai/chronon/join.py(3 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- api/python/ai/chronon/group_by.py
🧰 Additional context used
🧬 Code Definitions (2)
api/python/ai/chronon/join.py (1)
api/python/ai/chronon/dag_builder.py (1)
to_dependency(12-59)
api/python/ai/chronon/dag_builder.py (2)
api/python/ai/chronon/group_by.py (3)
TimeUnit(179-181)Window(245-246)GroupBy(430-676)api/python/ai/chronon/join.py (2)
BootstrapPart(360-395)JoinPart(31-109)
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: non_spark_tests
- GitHub Check: non_spark_tests
🔇 Additional comments (6)
api/python/ai/chronon/join.py (4)
24-24: Good addition: Import dag_builder moduleThis import is needed for the new table dependency functionality.
515-520: Defensive initialization of parametersAdding default empty lists for null parameters prevents potential null pointer errors.
573-575: New table dependency tracking implementationProperly collects dependency information from left source, right parts, and bootstrap parts.
583-583: Dependency injection into ExecutionInfoSuccessfully passes table dependencies to the execution info object.
api/python/ai/chronon/dag_builder.py (2)
12-12: Parameter use correctedThe
lagparameter is now properly utilized, addressing a previous review comment.
14-14: Properly propagate lag parameterExcellent - the lag parameter is now correctly passed to the recursive call.
Co-authored-by: Thomas Chow <[email protected]>
Co-authored-by: Thomas Chow <[email protected]>
Co-authored-by: Thomas Chow <[email protected]>
Co-authored-by: Thomas Chow <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
api/python/ai/chronon/repo/compile.py (1)
276-278: Handle unexpected node types safely.
Callingadd_dependencies(obj)might fail ifobjis notGroupBy,Join, orJoinPart. Consider pre-checking object type or catching theValueError.api/python/ai/chronon/repo/dag_builder.py (2)
33-46: Watch for circular references.
add_dependenciesrecurses forJoinPart, which might risk cycles in complex graphs.
48-101: Ensure robust metadata checks.
node.metaData.executionInfois assumed non-null. Add safeguards or validations if that's not guaranteed.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro (Legacy)
📒 Files selected for processing (3)
api/python/ai/chronon/join.py(1 hunks)api/python/ai/chronon/repo/compile.py(2 hunks)api/python/ai/chronon/repo/dag_builder.py(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- api/python/ai/chronon/join.py
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: non_spark_tests
- GitHub Check: non_spark_tests
🔇 Additional comments (1)
api/python/ai/chronon/repo/compile.py (1)
33-33: No concerns with the new import.
| def add_table_info(node: Union[GroupBy, Join, BootstrapPart]): | ||
| table_info = None | ||
| if isinstance(node, GroupBy): | ||
| table_info = TableInfo( | ||
| table=c_utils.group_by_output_table_name(node), | ||
| ) | ||
| elif isinstance(node, Join): | ||
| table_info = TableInfo( | ||
| table=c_utils.get_join_output_table_name(node), | ||
| ) | ||
| elif isinstance(node, BootstrapPart): | ||
| part_col = node.query.partitionColumn | ||
| table_info = TableInfo( | ||
| table=c_utils.get_table, | ||
| partitionColumn=part_col | ||
| ) | ||
| else: | ||
| return | ||
| node.metaData.executionInfo.outputTableInfo = table_info | ||
|
|
There was a problem hiding this comment.
Possible function call omission.
Line 25 references c_utils.get_table instead of calling it. Likely intended:
- table=c_utils.get_table,
+ table=c_utils.get_table(node),📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def add_table_info(node: Union[GroupBy, Join, BootstrapPart]): | |
| table_info = None | |
| if isinstance(node, GroupBy): | |
| table_info = TableInfo( | |
| table=c_utils.group_by_output_table_name(node), | |
| ) | |
| elif isinstance(node, Join): | |
| table_info = TableInfo( | |
| table=c_utils.get_join_output_table_name(node), | |
| ) | |
| elif isinstance(node, BootstrapPart): | |
| part_col = node.query.partitionColumn | |
| table_info = TableInfo( | |
| table=c_utils.get_table, | |
| partitionColumn=part_col | |
| ) | |
| else: | |
| return | |
| node.metaData.executionInfo.outputTableInfo = table_info | |
| def add_table_info(node: Union[GroupBy, Join, BootstrapPart]): | |
| table_info = None | |
| if isinstance(node, GroupBy): | |
| table_info = TableInfo( | |
| table=c_utils.group_by_output_table_name(node), | |
| ) | |
| elif isinstance(node, Join): | |
| table_info = TableInfo( | |
| table=c_utils.get_join_output_table_name(node), | |
| ) | |
| elif isinstance(node, BootstrapPart): | |
| part_col = node.query.partitionColumn | |
| table_info = TableInfo( | |
| table=c_utils.get_table(node), | |
| partitionColumn=part_col | |
| ) | |
| else: | |
| return | |
| node.metaData.executionInfo.outputTableInfo = table_info |
Co-authored-by: Thomas Chow <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 7
♻️ Duplicate comments (1)
api/python/ai/chronon/repo/dag_builder.py (1)
25-25:⚠️ Potential issueFix function reference passed without invocation.
Function reference is passed instead of result.
- table=c_utils.get_table, + table=c_utils.get_table(node),
🧹 Nitpick comments (5)
api/python/ai/chronon/repo/dag_builder.py (5)
45-45: Improve error message for unsupported node types.Current error message shows object instance, not type.
- raise ValueError(f"Obj is of type: {node}") + raise ValueError(f"Unsupported node type: {type(node).__name__}")
23-24: Add null check for node.query.Similar to other nodes, add protection against NoneType errors.
- part_col = node.query.partitionColumn + part_col = node.query.partitionColumn if node.query is not None else None
12-31: Check if metaData and executionInfo exist before assignment.Add initialization if fields don't exist.
def add_table_info(node: Union[GroupBy, Join, BootstrapPart]): table_info = None if isinstance(node, GroupBy): table_info = TableInfo( table=c_utils.group_by_output_table_name(node), ) elif isinstance(node, Join): table_info = TableInfo( table=c_utils.get_join_output_table_name(node), ) elif isinstance(node, BootstrapPart): part_col = node.query.partitionColumn if node.query is not None else None table_info = TableInfo( table=c_utils.get_table(node), partitionColumn=part_col ) else: return + if not hasattr(node, 'metaData'): + node.metaData = type('obj', (object,), {}) + if not hasattr(node.metaData, 'executionInfo'): + node.metaData.executionInfo = type('obj', (object,), {}) node.metaData.executionInfo.outputTableInfo = table_info
33-46: Add similar metaData initialization for add_dependencies.Consistent with previous suggestion.
def add_dependencies(node: Union[GroupBy, Join, JoinPart]): if isinstance(node, GroupBy): raw_deps = node.sources table_deps = [to_dependency(s) for s in raw_deps] + if not hasattr(node, 'metaData'): + node.metaData = type('obj', (object,), {}) + if not hasattr(node.metaData, 'executionInfo'): + node.metaData.executionInfo = type('obj', (object,), {}) node.metaData.executionInfo.tableDependencies=table_deps elif isinstance(node, Join): raw_deps = node.bootstrapParts + node.joinParts + [node.left] table_deps = [to_dependency(up) for up in raw_deps] + if not hasattr(node, 'metaData'): + node.metaData = type('obj', (object,), {}) + if not hasattr(node.metaData, 'executionInfo'): + node.metaData.executionInfo = type('obj', (object,), {}) node.metaData.executionInfo.tableDependencies=table_deps elif isinstance(node, JoinPart): add_dependencies(node.groupBy) else: raise ValueError(f"Unsupported node type: {type(node).__name__}")
7-10: Consider expanding module docstring.Add more details about purpose and usage.
""" -Given a node that represents an upstream, turn it into a TableDependency. +Module for building table dependency graphs. + +Provides functions to extract table information and dependencies +from different node types and convert them to TableDependency objects. """
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro (Legacy)
📒 Files selected for processing (1)
api/python/ai/chronon/repo/dag_builder.py(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: non_spark_tests
- GitHub Check: non_spark_tests
🔇 Additional comments (3)
api/python/ai/chronon/repo/dag_builder.py (3)
1-6: LGTM! Imports look good.Required imports are properly organized.
93-94:⚠️ Potential issueCheck if query exists before accessing properties.
Current code will raise exception if query is None.
- start_cutoff = query.startPartition if query is not None else None - end_cutoff = query.endPartition if query is not None else None + start_cutoff = None + end_cutoff = None + if query is not None: + start_cutoff = query.startPartition + end_cutoff = query.endPartitionLikely an incorrect or invalid review comment.
74-75:⚠️ Potential issueCheck if query exists before accessing properties.
Current code will raise exception if query is None.
- start_cutoff = node.query.startPartition if node.query is not None else None - end_cutoff = node.query.endPartition if node.query is not None else None + start_cutoff = None + end_cutoff = None + if node.query is not None: + start_cutoff = node.query.startPartition + end_cutoff = node.query.endPartitionLikely an incorrect or invalid review comment.
Co-authored-by: Thomas Chow <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (7)
api/python/ai/chronon/repo/dag_builder.py (7)
24-26:⚠️ Potential issueFix function call in BootstrapPart table_info.
c_utils.get_tablereference needs to be called with node parameter.- table=c_utils.get_table, + table=c_utils.get_table(node),
58-60:⚠️ Potential issueFix casing inconsistency - endCutOff.
Variable name should be consistent.
- if source_dep.endCutOff is not None: - if min_end_cutoff is None or min_end_cutoff > source_dep.endCutOff: - min_end_cutoff = source_dep.endCutOff + if source_dep.endCutoff is not None: + if min_end_cutoff is None or min_end_cutoff > source_dep.endCutoff: + min_end_cutoff = source_dep.endCutoff
61-63:⚠️ Potential issueFix casing inconsistency - startCutOff.
Variable name should be consistent.
- if source_dep.startCutOff is not None: - if max_start_cutoff is None or max_start_cutoff < source_dep.startCutOff: - max_start_cutoff = source_dep.startCutOff + if source_dep.startCutoff is not None: + if max_start_cutoff is None or max_start_cutoff < source_dep.startCutoff: + max_start_cutoff = source_dep.startCutoff
69-70:⚠️ Potential issueFix variable casing in constructor.
Use consistent variable naming.
- startCutOff=max_start_cutoff, - endCutOff=min_end_cutoff, + startCutoff=max_start_cutoff, + endCutoff=min_end_cutoff,
81-82:⚠️ Potential issueFix variable casing in constructor.
Use consistent variable naming.
- startCutOff=start_cutoff, - endCutOff=end_cutoff, + startCutoff=start_cutoff, + endCutoff=end_cutoff,
87-88:⚠️ Potential issueAdd null check for query.
Prevent NoneType exception.
query = c_utils.get_query(node) - partition_column = query.partitionColumn + partition_column = query.partitionColumn if query is not None else None
99-100:⚠️ Potential issueFix variable casing in constructor.
Use consistent variable naming.
- startCutOff=start_cutoff, - endCutOff=end_cutoff, + startCutoff=start_cutoff, + endCutoff=end_cutoff,
🧹 Nitpick comments (1)
api/python/ai/chronon/repo/dag_builder.py (1)
7-10: Docstring could be more comprehensive.Consider expanding to describe all three functions.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro (Legacy)
📒 Files selected for processing (1)
api/python/ai/chronon/repo/dag_builder.py(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: non_spark_tests
- GitHub Check: non_spark_tests
🔇 Additional comments (2)
api/python/ai/chronon/repo/dag_builder.py (2)
1-6: Imports look good.Appropriate typing and module imports for the functionality.
33-46: Clean implementation of dependency addition.Good error handling for unsupported types.
Co-authored-by: Thomas Chow <[email protected]>
| raise ValueError(f"Obj is of type: {node}") | ||
|
|
||
|
|
||
| def to_dependency(node: Union[Source, GroupBy, JoinPart, BootstrapPart], lag: int = 0) -> TableDependency: |
There was a problem hiding this comment.
fyi lag not incorporated yet .
Summary
Checklist
Summary by CodeRabbit
New Features
Bug Fixes