Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: left table hash mismatch #761

Merged
merged 1 commit into from
May 13, 2024
Merged

fix: left table hash mismatch #761

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
22 changes: 12 additions & 10 deletions spark/src/main/scala/ai/chronon/spark/JoinBase.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -410,16 +410,6 @@ abstract class JoinBase(joinConf: api.Join,
s"groupBy.metaData.team needs to be set for joinPart ${jp.groupBy.metaData.name}")
}

val source = joinConf.left
if (useBootstrapForLeft) {
logger.info("Overwriting left side to use saved Bootstrap table...")
source.overwriteTable(bootstrapTable)
val query = source.query
// sets map and where clauses already applied to bootstrap transformation
query.setSelects(null)
query.setWheres(null)
}

// Run validations before starting the job
val today = tableUtils.partitionSpec.at(System.currentTimeMillis())
val analyzer = new Analyzer(tableUtils, joinConf, today, today, silenceMode = true)
Expand All @@ -439,9 +429,21 @@ abstract class JoinBase(joinConf: api.Join,

// First run command to archive tables that have changed semantically since the last run
val archivedAtTs = Instant.now()
// TODO: We should not archive the output table in the case of selected join parts mode
tablesToRecompute(joinConf, outputTable, tableUtils).foreach(
tableUtils.archiveOrDropTableIfExists(_, Some(archivedAtTs)))

// Check semantic hash before overwriting left side
val source = joinConf.left
if (useBootstrapForLeft) {
logger.info("Overwriting left side to use saved Bootstrap table...")
source.overwriteTable(bootstrapTable)
val query = source.query
// sets map and where clauses already applied to bootstrap transformation
query.setSelects(null)
query.setWheres(null)
}

// detect holes and chunks to fill
// OverrideStartPartition is used to replace the start partition of the join config. This is useful when
// 1 - User would like to test run with different start partition
Expand Down