Skip to content

Commit 7948eec

Browse files
rafafrdzrafafrdz
authored
feat: add Conditional transformation, enhanced dry-run, doc generator, lifecycle hooks (#99)
1 parent f104a0e commit 7948eec

14 files changed

Lines changed: 534 additions & 12 deletions

File tree

‎api/src/main/scala/com/eff3ct/teckel/api/DocGen.scala‎

Lines changed: 176 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,176 @@
1+
/*
2+
* MIT License
3+
*
4+
* Copyright (c) 2024 Rafael Fernandez
5+
*
6+
* Permission is hereby granted, free of charge, to any person obtaining a copy
7+
* of this software and associated documentation files (the "Software"), to deal
8+
* in the Software without restriction, including without limitation the rights
9+
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
10+
* copies of the Software, and to permit persons to whom the Software is
11+
* furnished to do so, subject to the following conditions:
12+
*
13+
* The above copyright notice and this permission notice shall be included in all
14+
* copies or substantial portions of the Software.
15+
*
16+
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17+
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18+
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19+
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20+
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21+
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
22+
* SOFTWARE.
23+
*/
24+
25+
package com.eff3ct.teckel.api
26+
27+
import com.eff3ct.teckel.model.{Asset, AssetRef, Context}
28+
import com.eff3ct.teckel.model.Source._
29+
import com.eff3ct.teckel.serializer.Serializer
30+
import com.eff3ct.teckel.serializer.model.etl._
31+
import com.eff3ct.teckel.transform.Rewrite
32+
33+
object DocGen {
34+
35+
def generate(yamlContent: String): Either[Throwable, String] =
36+
Serializer[ETL].decode(yamlContent).map { etl =>
37+
val context = Rewrite.rewrite(etl)
38+
formatDoc(context)
39+
}
40+
41+
private def formatDoc(context: Context[Asset]): String = {
42+
val sb = new StringBuilder
43+
sb.append("# Pipeline Documentation\n\n")
44+
45+
val inputs = context.filter(_._2.source.isInstanceOf[Input])
46+
val outputs = context.filter(_._2.source.isInstanceOf[Output])
47+
val transformations = context.filter(e => e._2.source.isInstanceOf[Transformation])
48+
49+
sb.append("## Data Sources\n\n")
50+
sb.append("| Name | Format | Path |\n")
51+
sb.append("|------|--------|------|\n")
52+
inputs.foreach { case (ref, asset) =>
53+
val s = asset.source.asInstanceOf[Input]
54+
sb.append(s"| $ref | ${s.format} | ${s.sourceRef} |\n")
55+
}
56+
57+
if (transformations.nonEmpty) {
58+
sb.append("\n## Transformations\n\n")
59+
transformations.foreach { case (ref, asset) =>
60+
sb.append(s"### $ref\n\n")
61+
sb.append(describeTransformation(asset.source.asInstanceOf[Transformation]))
62+
sb.append("\n")
63+
}
64+
}
65+
66+
sb.append("## Outputs\n\n")
67+
sb.append("| Name | Format | Mode | Path |\n")
68+
sb.append("|------|--------|------|------|\n")
69+
outputs.foreach { case (ref, asset) =>
70+
val s = asset.source.asInstanceOf[Output]
71+
sb.append(s"| ${s.assetRef} | ${s.format} | ${s.mode} | ${s.sourceRef} |\n")
72+
}
73+
74+
sb.append("\n## Data Flow\n\n")
75+
sb.append("```\n")
76+
context.foreach { case (ref, asset) =>
77+
asset.source match {
78+
case _: Input =>
79+
sb.append(s"[INPUT] $ref\n")
80+
case s: Output =>
81+
sb.append(s" ${s.assetRef} --> [OUTPUT] $ref (${s.format}:${s.sourceRef})\n")
82+
case t: Transformation =>
83+
sb.append(s" ${t.assetRef} --> [$ref]\n")
84+
}
85+
}
86+
sb.append("```\n")
87+
88+
sb.toString
89+
}
90+
91+
private def describeTransformation(t: Transformation): String = t match {
92+
case s: Select =>
93+
s"- **Type**: Select\n- **From**: ${s.assetRef}\n- **Columns**: ${s.columns.toList.mkString(", ")}\n"
94+
case s: Where =>
95+
s"- **Type**: Filter\n- **From**: ${s.assetRef}\n- **Condition**: `${s.condition}`\n"
96+
case s: GroupBy =>
97+
s"- **Type**: Group By\n- **From**: ${s.assetRef}\n- **Group By**: ${s.by.toList.mkString(
98+
", "
99+
)}\n- **Aggregations**: ${s.aggregate.toList.mkString(", ")}\n"
100+
case s: OrderBy =>
101+
s"- **Type**: Order By\n- **From**: ${s.assetRef}\n- **Order By**: ${s.by.toList.mkString(", ")} ${s.order
102+
.getOrElse("asc")}\n"
103+
case s: Join =>
104+
s"- **Type**: Join\n- **From**: ${s.assetRef}\n- **Joins**: ${s.others.toList
105+
.map(r => s"${r.name} (${r.joinType})")
106+
.mkString(", ")}\n"
107+
case s: Distinct =>
108+
s"- **Type**: Distinct\n- **From**: ${s.assetRef}\n${s.columns
109+
.map(c => s"- **Columns**: ${c.toList.mkString(", ")}\n")
110+
.getOrElse("")}"
111+
case s: Limit =>
112+
s"- **Type**: Limit\n- **From**: ${s.assetRef}\n- **Count**: ${s.count}\n"
113+
case s: Union =>
114+
s"- **Type**: Union${if (s.all) " All"
115+
else ""}\n- **From**: ${s.assetRef}\n- **With**: ${s.others.toList.mkString(", ")}\n"
116+
case s: Intersect =>
117+
s"- **Type**: Intersect${if (s.all) " All"
118+
else ""}\n- **From**: ${s.assetRef}\n- **With**: ${s.others.toList.mkString(", ")}\n"
119+
case s: Except =>
120+
s"- **Type**: Except${if (s.all) " All" else ""}\n- **From**: ${s.assetRef}\n- **Other**: ${s.other}\n"
121+
case s: AddColumns =>
122+
s"- **Type**: Add Columns\n- **From**: ${s.assetRef}\n- **Columns**: ${s.columns.toList
123+
.map(c => s"${c.name} = ${c.expression}")
124+
.mkString(", ")}\n"
125+
case s: DropColumns =>
126+
s"- **Type**: Drop Columns\n- **From**: ${s.assetRef}\n- **Columns**: ${s.columns.toList.mkString(", ")}\n"
127+
case s: RenameColumns =>
128+
s"- **Type**: Rename Columns\n- **From**: ${s.assetRef}\n- **Mappings**: ${s.mappings
129+
.map { case (k, v) => s"$k -> $v" }
130+
.mkString(", ")}\n"
131+
case s: CastColumns =>
132+
s"- **Type**: Cast Columns\n- **From**: ${s.assetRef}\n- **Columns**: ${s.columns.toList
133+
.map(c => s"${c.name} as ${c.targetType}")
134+
.mkString(", ")}\n"
135+
case s: Sql =>
136+
s"- **Type**: SQL\n- **From**: ${s.assetRef}\n- **Query**: `${s.query}`\n"
137+
case s: Window =>
138+
s"- **Type**: Window\n- **From**: ${s.assetRef}\n- **Partition By**: ${s.partitionBy.toList
139+
.mkString(", ")}\n${s.orderBy.map(o => s"- **Order By**: ${o.toList.mkString(", ")}\n").getOrElse("")}- **Functions**: ${s.functions.toList
140+
.map(f => s"${f.expression} as ${f.alias}")
141+
.mkString(", ")}\n"
142+
case s: Flatten =>
143+
s"- **Type**: Flatten\n- **From**: ${s.assetRef}\n${s.separator
144+
.map(sep => s"- **Separator**: $sep\n")
145+
.getOrElse("")}${s.explodeArrays.map(e => s"- **Explode Arrays**: $e\n").getOrElse("")}"
146+
case s: Sample =>
147+
s"- **Type**: Sample\n- **From**: ${s.assetRef}\n- **Fraction**: ${s.fraction}\n${s.withReplacement
148+
.map(w => s"- **With Replacement**: $w\n")
149+
.getOrElse("")}${s.seed.map(seed => s"- **Seed**: $seed\n").getOrElse("")}"
150+
case s: Repartition =>
151+
s"- **Type**: Repartition\n- **From**: ${s.assetRef}\n- **Partitions**: ${s.numPartitions}\n${s.columns
152+
.map(c => s"- **Columns**: ${c.toList.mkString(", ")}\n")
153+
.getOrElse("")}"
154+
case s: Coalesce =>
155+
s"- **Type**: Coalesce\n- **From**: ${s.assetRef}\n- **Partitions**: ${s.numPartitions}\n"
156+
case s: Rollup =>
157+
s"- **Type**: Rollup\n- **From**: ${s.assetRef}\n- **By**: ${s.by.toList.mkString(", ")}\n- **Aggregations**: ${s.aggregate.toList
158+
.mkString(", ")}\n"
159+
case s: Cube =>
160+
s"- **Type**: Cube\n- **From**: ${s.assetRef}\n- **By**: ${s.by.toList.mkString(", ")}\n- **Aggregations**: ${s.aggregate.toList
161+
.mkString(", ")}\n"
162+
case s: Pivot =>
163+
s"- **Type**: Pivot\n- **From**: ${s.assetRef}\n- **Group By**: ${s.groupBy.toList.mkString(
164+
", "
165+
)}\n- **Pivot Column**: ${s.pivotColumn}\n${s.values
166+
.map(v => s"- **Values**: ${v.mkString(", ")}\n")
167+
.getOrElse("")}- **Aggregations**: ${s.aggregate.toList.mkString(", ")}\n"
168+
case s: Unpivot =>
169+
s"- **Type**: Unpivot\n- **From**: ${s.assetRef}\n- **IDs**: ${s.ids.toList.mkString(", ")}\n- **Values**: ${s.values.toList
170+
.mkString(", ")}\n- **Variable Column**: ${s.variableColumn}\n- **Value Column**: ${s.valueColumn}\n"
171+
case s: Conditional =>
172+
s"- **Type**: Conditional\n- **From**: ${s.assetRef}\n- **Output Column**: ${s.outputColumn}\n- **Branches**: ${s.branches.toList
173+
.map(b => s"when ${b.condition} then ${b.value}")
174+
.mkString("; ")}\n${s.otherwise.map(o => s"- **Otherwise**: $o\n").getOrElse("")}"
175+
}
176+
}

‎api/src/main/scala/com/eff3ct/teckel/api/DryRun.scala‎

Lines changed: 109 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ object DryRun {
7979
sb.toString
8080
}
8181

82+
// scalastyle:off method.length cyclomatic.complexity
8283
private def planStep(ref: AssetRef, asset: Asset): PlanStep =
8384
asset.source match {
8485
case s: Input =>
@@ -120,12 +121,116 @@ object DryRun {
120121
s"tables=[${s.others.toList.map(_.name).mkString(", ")}]",
121122
s.assetRef :: s.others.toList.map(_.name)
122123
)
123-
case _: Transformation =>
124+
case s: Distinct =>
125+
val cols =
126+
s.columns.map(c => s"columns=[${c.toList.mkString(", ")}]").getOrElse("all columns")
127+
PlanStep(ref, "DISTINCT", cols, List(s.assetRef))
128+
case s: Limit =>
129+
PlanStep(ref, "LIMIT", s"count=${s.count}", List(s.assetRef))
130+
case s: AddColumns =>
131+
val colDefs = s.columns.toList.map(c => s"${c.name}=${c.expression}").mkString(", ")
132+
PlanStep(ref, "ADD_COLUMNS", s"columns=[$colDefs]", List(s.assetRef))
133+
case s: DropColumns =>
124134
PlanStep(
125135
ref,
126-
"TRANSFORMATION",
127-
"custom",
128-
List(asset.source.asInstanceOf[Transformation].assetRef)
136+
"DROP_COLUMNS",
137+
s"columns=[${s.columns.toList.mkString(", ")}]",
138+
List(s.assetRef)
139+
)
140+
case s: RenameColumns =>
141+
val maps = s.mappings.map { case (k, v) => s"$k->$v" }.mkString(", ")
142+
PlanStep(ref, "RENAME_COLUMNS", s"mappings=[$maps]", List(s.assetRef))
143+
case s: CastColumns =>
144+
val cols = s.columns.toList.map(c => s"${c.name}:${c.targetType}").mkString(", ")
145+
PlanStep(ref, "CAST_COLUMNS", s"columns=[$cols]", List(s.assetRef))
146+
case s: Sql =>
147+
PlanStep(ref, "SQL", s"query=${s.query}", List(s.assetRef))
148+
case s: Union =>
149+
PlanStep(
150+
ref,
151+
"UNION",
152+
s"all=${s.all}, others=[${s.others.toList.mkString(", ")}]",
153+
s.assetRef :: s.others.toList
154+
)
155+
case s: Intersect =>
156+
PlanStep(
157+
ref,
158+
"INTERSECT",
159+
s"all=${s.all}, others=[${s.others.toList.mkString(", ")}]",
160+
s.assetRef :: s.others.toList
161+
)
162+
case s: Except =>
163+
PlanStep(
164+
ref,
165+
"EXCEPT",
166+
s"all=${s.all}, other=${s.other}",
167+
List(s.assetRef, s.other)
168+
)
169+
case s: Window =>
170+
val orderStr =
171+
s.orderBy.map(o => s", orderBy=[${o.toList.mkString(", ")}]").getOrElse("")
172+
val funcs = s.functions.toList.map(f => s"${f.expression} as ${f.alias}").mkString(", ")
173+
PlanStep(
174+
ref,
175+
"WINDOW",
176+
s"partitionBy=[${s.partitionBy.toList.mkString(", ")}]$orderStr, functions=[$funcs]",
177+
List(s.assetRef)
178+
)
179+
case s: Flatten =>
180+
val sep = s.separator.map(v => s", separator=$v").getOrElse("")
181+
val exp = s.explodeArrays.map(v => s", explodeArrays=$v").getOrElse("")
182+
PlanStep(ref, "FLATTEN", s"flatten$sep$exp", List(s.assetRef))
183+
case s: Sample =>
184+
val repl = s.withReplacement.map(v => s", withReplacement=$v").getOrElse("")
185+
val sd = s.seed.map(v => s", seed=$v").getOrElse("")
186+
PlanStep(ref, "SAMPLE", s"fraction=${s.fraction}$repl$sd", List(s.assetRef))
187+
case s: Repartition =>
188+
val cols =
189+
s.columns.map(c => s", columns=[${c.toList.mkString(", ")}]").getOrElse("")
190+
PlanStep(ref, "REPARTITION", s"numPartitions=${s.numPartitions}$cols", List(s.assetRef))
191+
case s: Coalesce =>
192+
PlanStep(ref, "COALESCE", s"numPartitions=${s.numPartitions}", List(s.assetRef))
193+
case s: Rollup =>
194+
PlanStep(
195+
ref,
196+
"ROLLUP",
197+
s"by=[${s.by.toList.mkString(", ")}], agg=[${s.aggregate.toList.mkString(", ")}]",
198+
List(s.assetRef)
199+
)
200+
case s: Cube =>
201+
PlanStep(
202+
ref,
203+
"CUBE",
204+
s"by=[${s.by.toList.mkString(", ")}], agg=[${s.aggregate.toList.mkString(", ")}]",
205+
List(s.assetRef)
206+
)
207+
case s: Pivot =>
208+
val vals = s.values.map(v => s", values=[${v.mkString(", ")}]").getOrElse("")
209+
PlanStep(
210+
ref,
211+
"PIVOT",
212+
s"groupBy=[${s.groupBy.toList.mkString(", ")}], pivotColumn=${s.pivotColumn}, agg=[${s.aggregate.toList
213+
.mkString(", ")}]$vals",
214+
List(s.assetRef)
215+
)
216+
case s: Unpivot =>
217+
PlanStep(
218+
ref,
219+
"UNPIVOT",
220+
s"ids=[${s.ids.toList.mkString(", ")}], values=[${s.values.toList
221+
.mkString(", ")}], variableColumn=${s.variableColumn}, valueColumn=${s.valueColumn}",
222+
List(s.assetRef)
223+
)
224+
case s: Conditional =>
225+
val branches =
226+
s.branches.toList.map(b => s"WHEN ${b.condition} THEN ${b.value}").mkString(", ")
227+
val other = s.otherwise.map(v => s", OTHERWISE $v").getOrElse("")
228+
PlanStep(
229+
ref,
230+
"CONDITIONAL",
231+
s"outputColumn=${s.outputColumn}, branches=[$branches$other]",
232+
List(s.assetRef)
129233
)
130234
}
235+
// scalastyle:on method.length cyclomatic.complexity
131236
}

‎api/src/main/scala/com/eff3ct/teckel/api/Hooks.scala‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
/*
2+
* MIT License
3+
*
4+
* Copyright (c) 2024 Rafael Fernandez
5+
*
6+
* Permission is hereby granted, free of charge, to any person obtaining a copy
7+
* of this software and associated documentation files (the "Software"), to deal
8+
* in the Software without restriction, including without limitation the rights
9+
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
10+
* copies of the Software, and to permit persons to whom the Software is
11+
* furnished to do so, subject to the following conditions:
12+
*
13+
* The above copyright notice and this permission notice shall be included in all
14+
* copies or substantial portions of the Software.
15+
*
16+
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17+
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18+
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19+
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20+
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21+
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
22+
* SOFTWARE.
23+
*/
24+
25+
package com.eff3ct.teckel.api
26+
27+
import cats.effect.Sync
28+
import cats.implicits._
29+
import com.eff3ct.teckel.serializer.model.etl.{Hook, Hooks}
30+
31+
object HookRunner {
32+
33+
def runPreHooks[F[_]: Sync](hooks: Option[Hooks]): F[Unit] =
34+
hooks.flatMap(_.preExecution).traverse_ { hookList =>
35+
hookList.traverse_ { hook =>
36+
Sync[F].delay {
37+
println(s"[hook:pre] Running '${hook.name}': ${hook.command}")
38+
val process = Runtime.getRuntime.exec(Array("/bin/sh", "-c", hook.command))
39+
val exitCode = process.waitFor()
40+
if (exitCode != 0)
41+
throw new RuntimeException(
42+
s"Pre-execution hook '${hook.name}' failed with exit code $exitCode"
43+
)
44+
println(s"[hook:pre] '${hook.name}' completed successfully")
45+
}
46+
}
47+
}
48+
49+
def runPostHooks[F[_]: Sync](hooks: Option[Hooks]): F[Unit] =
50+
hooks.flatMap(_.postExecution).traverse_ { hookList =>
51+
hookList.traverse_ { hook =>
52+
Sync[F].delay {
53+
println(s"[hook:post] Running '${hook.name}': ${hook.command}")
54+
val process = Runtime.getRuntime.exec(Array("/bin/sh", "-c", hook.command))
55+
val exitCode = process.waitFor()
56+
if (exitCode != 0)
57+
println(s"[hook:post] Warning: '${hook.name}' failed with exit code $exitCode")
58+
else
59+
println(s"[hook:post] '${hook.name}' completed successfully")
60+
}
61+
}
62+
}
63+
}

0 commit comments

Comments
 (0)