Aligning joinPart output to always be in line with join left #725
Conversation
WalkthroughThe changes introduce a new Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant DataFrame
participant DataframeOps
User->>DataFrame: call shiftPartition(numDays)
activate DataFrame
DataFrame->>DataframeOps: shiftPartition(numDays, partitionColumn)
DataframeOps->>DataFrame: returns DataFrame with shifted partition column
deactivate DataFrame
sequenceDiagram
participant JoinPartJob
participant GroupBy
participant DataFrame
JoinPartJob->>GroupBy: snapshotEvents/snapshotEntities
GroupBy-->>JoinPartJob: DataFrame
JoinPartJob->>DataFrame: shiftPartition(1)
DataFrame-->>JoinPartJob: shifted DataFrame
Possibly related PRs
Suggested reviewers
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. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. 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 (
|
varant-zlai
force-pushed
the
vz--align_output_date_for_snapshot_joinparts
branch
from
2621fbf to
97aec92
Compare
varant-zlai
force-pushed
the
vz--align_output_date_for_snapshot_joinparts
branch
from
d69041f to
ed9dc23
Compare
![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)
spark/src/main/scala/ai/chronon/spark/JoinBase.scala (2)
110-114: Added debugging output.Shows min/max of ts_ds before join for troubleshooting.
Consider removing or conditionalizing this debug output before production deployment:
- leftDf.selectExpr("min(ts_ds) as min_ds", "max(ts_ds) as max_ds").show() - joinableRightDf.selectExpr("min(ts_ds) as min_ds", "max(ts_ds) as max_ds").show() - //renamedLeftDf.selectExpr("min(ds) as min_ds", "max(ds) as max_ds").show() - // rightDf.selectExpr("min(ds) as min_ds", "max(ds) as max_ds").show() + if (logger.isDebugEnabled()) { + leftDf.selectExpr("min(ts_ds) as min_ds", "max(ts_ds) as max_ds").show() + joinableRightDf.selectExpr("min(ts_ds) as min_ds", "max(ts_ds) as max_ds").show() + }
126-126: Added join result debugging.Shows the joined DataFrame for troubleshooting.
Similar to the earlier debug output, consider conditionalizing:
- joinedDf.show() + if (logger.isDebugEnabled()) { + joinedDf.show() + }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro (Legacy)
📒 Files selected for processing (4)
spark/src/main/scala/ai/chronon/spark/Join.scala(1 hunks)spark/src/main/scala/ai/chronon/spark/JoinBase.scala(2 hunks)spark/src/main/scala/ai/chronon/spark/batch/JoinPartJob.scala(3 hunks)spark/src/main/scala/ai/chronon/spark/batch/MergeJob.scala(2 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- spark/src/main/scala/ai/chronon/spark/batch/MergeJob.scala
- spark/src/main/scala/ai/chronon/spark/batch/JoinPartJob.scala
🧰 Additional context used
🧬 Code Graph Analysis (1)
spark/src/main/scala/ai/chronon/spark/Join.scala (2)
spark/src/main/scala/ai/chronon/spark/catalog/TableUtils.scala (1)
whereClauses(587-591)api/src/main/scala/ai/chronon/api/DataRange.scala (1)
whereClauses(78-80)
⏰ Context from checks skipped due to timeout of 90000ms (31)
- GitHub Check: streaming_tests
- GitHub Check: service_commons_tests
- GitHub Check: service_tests
- GitHub Check: analyzer_tests
- GitHub Check: join_tests
- GitHub Check: cloud_gcp_tests
- GitHub Check: online_tests
- GitHub Check: streaming_tests
- GitHub Check: groupby_tests
- GitHub Check: cloud_aws_tests
- GitHub Check: flink_tests
- GitHub Check: spark_tests
- GitHub Check: service_tests
- GitHub Check: fetcher_tests
- GitHub Check: cloud_gcp_tests
- GitHub Check: batch_tests
- GitHub Check: online_tests
- GitHub Check: analyzer_tests
- GitHub Check: cloud_aws_tests
- GitHub Check: groupby_tests
- GitHub Check: flink_tests
- GitHub Check: spark_tests
- GitHub Check: api_tests
- GitHub Check: join_tests
- GitHub Check: scala_compile_fmt_fix
- GitHub Check: api_tests
- GitHub Check: batch_tests
- GitHub Check: aggregator_tests
- GitHub Check: fetcher_tests
- GitHub Check: aggregator_tests
- GitHub Check: enforce_triggered_workflows
🔇 Additional comments (2)
spark/src/main/scala/ai/chronon/spark/Join.scala (1)
201-201: Simplified partition filtering logic.Now directly using original leftRange for where clauses instead of conditional adjustment.
spark/src/main/scala/ai/chronon/spark/JoinBase.scala (1)
103-103: Removed date increment logic.Now directly using the partition column without transformation, aligning with PR objective.
Summary
Previously, snapshot accuracy would shift the joinPart output to align with the offset. This makes it align to the original left ds.
Checklist
Summary by CodeRabbit
New Features
Bug Fixes
Refactor
Documentation