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

force overwrite metadata in table properties #610

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
Open

force overwrite metadata in table properties #610

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5d353d0
force overwrite metadata
better365 Nov 9, 2023
b63af69
Merge branch 'master' of github.com:airbnb/chronon into pengyu--force…
better365 Nov 10, 2023
14f5adc
unset table property
better365 Nov 10, 2023
49f80a0
clean up
better365 Nov 10, 2023
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
4 changes: 4 additions & 0 deletions spark/src/main/scala/ai/chronon/spark/Driver.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,6 +213,10 @@ object Driver {
opt[String](required = false,
descr =
"Start date to compute join backfill, this start date will override start partition in conf.")
val forceOverwriteMetadata: ScallopOption[Boolean] =
opt[Boolean](required = false,
default = Some(false),
descr = "Force overwrite metadata in the table properties if it already exists.")
lazy val joinConf: api.Join = parseConf[api.Join](confPath())
override def subcommandName() = s"join_${joinConf.metaData.name}"
}
Expand Down
5 changes: 3 additions & 2 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 @@ -21,7 +21,8 @@ abstract class JoinBase(joinConf: api.Join,
tableUtils: TableUtils,
skipFirstHole: Boolean,
mutationScan: Boolean = true,
showDf: Boolean = false) {
showDf: Boolean = false,
forceOverwriteMetadata: Boolean = false) {
assert(Option(joinConf.metaData.outputNamespace).nonEmpty, s"output namespace could not be empty or null")
val metrics: Metrics.Context = Metrics.Context(Metrics.Environment.JoinOffline, joinConf)
private val outputTable = joinConf.metaData.outputTable
Expand Down Expand Up @@ -301,7 +302,7 @@ abstract class JoinBase(joinConf: api.Join,

// First run command to archive tables that have changed semantically since the last run
val archivedAtTs = Instant.now()
tablesToRecompute(joinConf, outputTable, tableUtils).foreach(
tablesToRecompute(joinConf, outputTable, tableUtils, forceOverwriteMetadata).foreach(
tableUtils.archiveOrDropTableIfExists(_, Some(archivedAtTs)))

// detect holes and chunks to fill
Expand Down
6 changes: 5 additions & 1 deletion spark/src/main/scala/ai/chronon/spark/JoinUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,7 +319,11 @@ object JoinUtils {

def tablesToRecompute(joinConf: ai.chronon.api.Join,
outputTable: String,
tableUtils: TableUtils): collection.Seq[String] = {
tableUtils: TableUtils,
forceOverwriteMetadata: Boolean = false): collection.Seq[String] = {
if (forceOverwriteMetadata) {
return collection.Seq.empty
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This makes sure it won't recompute, but will the next run recompute?
I think when we save we store the table properties so it should probably work? Else may be take the opportunity to set the table properties here (hence overwriting the metadata as the flag implies)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good point. This will only overwrite the metadata when we do a new run

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed comments.

}
val gson = new Gson()
(for (
props <- tableUtils.getTableProperties(outputTable);
Expand Down
9 changes: 8 additions & 1 deletion spark/src/test/scala/ai/chronon/spark/test/JoinTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -791,6 +791,12 @@ class JoinTest {
JoinUtils.tablesToRecompute(leftChangeJoinConf, leftChangeJoinConf.metaData.outputTable, tableUtils)
println(leftChangeRecompute)
assertEquals(leftChangeRecompute.size, 3)

val leftChangeNoRecompute =
JoinUtils.tablesToRecompute(leftChangeJoinConf, leftChangeJoinConf.metaData.outputTable, tableUtils, true)
println(leftChangeNoRecompute)
assertEquals(leftChangeNoRecompute.size, 0)

val partTable = s"${leftChangeJoinConf.metaData.outputTable}_user_unit_test_item_views"
assertEquals(leftChangeRecompute,
Seq(partTable, leftChangeJoinConf.metaData.bootstrapTable, leftChangeJoinConf.metaData.outputTable))
Expand All @@ -804,6 +810,7 @@ class JoinTest {
val addPartRecompute =
JoinUtils.tablesToRecompute(addPartJoinConf, addPartJoinConf.metaData.outputTable, tableUtils)
assertEquals(addPartRecompute.size, 1)
assertEquals(JoinUtils.tablesToRecompute(addPartJoinConf, addPartJoinConf.metaData.outputTable, tableUtils,true).size, 0)
assertEquals(addPartRecompute, Seq(addPartJoinConf.metaData.outputTable))
// Compute to ensure that it works and to set the stage for the next assertion
addPartJoin.computeJoin(Some(100))
Expand All @@ -815,6 +822,7 @@ class JoinTest {
val rightModRecompute =
JoinUtils.tablesToRecompute(rightModJoinConf, rightModJoinConf.metaData.outputTable, tableUtils)
assertEquals(rightModRecompute.size, 2)
assertEquals(JoinUtils.tablesToRecompute(rightModJoinConf, rightModJoinConf.metaData.outputTable, tableUtils,true).size, 0)
val rightModPartTable = s"${addPartJoinConf.metaData.outputTable}_user_2_unit_test_item_views"
assertEquals(rightModRecompute, Seq(rightModPartTable, addPartJoinConf.metaData.outputTable))
// Modify both
Expand Down Expand Up @@ -1055,7 +1063,6 @@ class JoinTest {

@Test
def testMigration(): Unit = {

// Left
val itemQueriesTable = s"$namespace.item_queries"
val ds = "2023-01-01"
Expand Down