igluctl: table ownership setup as part of DDL (close #225)

snowplow · Dec 26, 2017 · 47bc622 · 47bc622
1 parent 662fcc4
commit 47bc622
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 8 deletions.
diff --git a/0-common/igluctl/src/main/scala/com.snowplowanalytics.iglu/ctl/Command.scala b/0-common/igluctl/src/main/scala/com.snowplowanalytics.iglu/ctl/Command.scala
@@ -46,6 +46,7 @@ case class Command(
   splitProduct:    Boolean         = false,
   noHeader:        Boolean         = false,
   force:           Boolean         = false,
+  owner:           Option[String]  = None,
 
   // sync
   registryRoot:    Option[HttpUrl] = None,
@@ -66,7 +67,7 @@ case class Command(
 ) {
   def toCommand: Option[Command.CtlCommand] = command match {
     case Some("static generate") => Some(
-      GenerateCommand(input.get, output.getOrElse(new File(".")), db,withJsonPaths, rawMode, schema, varcharSize, splitProduct, noHeader, force))
+      GenerateCommand(input.get, output.getOrElse(new File(".")), db, withJsonPaths, rawMode, schema, varcharSize, splitProduct, noHeader, force, owner))
     case Some("static push") =>
       Some(PushCommand(registryRoot.get, apiKey.get, input.get, isPublic))
     case Some("static s3cp") =>
@@ -144,6 +145,11 @@ object Command {
               valueName "<name>"
               text "Redshift schema name\t\t\t\tDefault: atomic",
 
+            opt[String]("set-owner")
+              action { (x, c) => c.copy(owner = Some(x)) }
+              valueName "<owner>"
+              text "Redshift table owner\t\t\t\tDefault: None",
+
             opt[String]("db")
               action { (x, c) => c.copy(db = x) }
               valueName "<name>"

diff --git a/0-common/igluctl/src/main/scala/com.snowplowanalytics.iglu/ctl/GenerateCommand.scala b/0-common/igluctl/src/main/scala/com.snowplowanalytics.iglu/ctl/GenerateCommand.scala
@@ -49,7 +49,8 @@ case class GenerateCommand(
   varcharSize: Int = 4096,
   splitProduct: Boolean = false,
   noHeader: Boolean = false,
-  force: Boolean = false) extends Command.CtlCommand {
+  force: Boolean = false,
+  owner: Option[String] = None) extends Command.CtlCommand {
 
   import GenerateCommand._
 
@@ -132,7 +133,7 @@ case class GenerateCommand(
   private[ctl] def selfDescSchemaToDdl(schema: IgluSchema, dbSchema: String): Validation[String, TableDefinition] = {
     val ddl = for {
       flatSchema <- FlatSchema.flattenJsonSchema(schema.schema, splitProduct)
-    } yield produceTable(flatSchema, schema.self, dbSchema)
+    } yield produceTable(flatSchema, schema.self, dbSchema, owner)
     ddl match {
       case Failure(fail) => (fail + s" in [${schema.self.toPath}] Schema").failure
       case success => success
@@ -161,13 +162,18 @@ case class GenerateCommand(
    * @param dbSchema DB schema name ("atomic")
    * @return table definition
    */
-  private def produceTable(flatSchema: FlatSchema, schemaMap: SchemaMap, dbSchema: String): TableDefinition = {
+  private def produceTable(flatSchema: FlatSchema, schemaMap: SchemaMap, dbSchema: String, owner: Option[String]): TableDefinition = {
     val (path, filename) = getFileName(schemaMap)
     val tableName = StringUtils.getTableName(schemaMap)
     val schemaCreate = CreateSchema(dbSchema)
     val table = DdlGenerator.generateTableDdl(flatSchema, tableName, Some(dbSchema), varcharSize, rawMode)
     val commentOn = DdlGenerator.getTableComment(tableName, Some(dbSchema), schemaMap)
-    val ddlFile = DdlFile(header ++ List(schemaCreate, Empty, table, Empty, commentOn))
+    val ddlFile = owner match {
+      case Some(ownerStr) =>
+        val owner = AlterTable(dbSchema+"."+tableName, OwnerTo(ownerStr))
+        DdlFile(header ++ List(schemaCreate, Empty, table, Empty, commentOn, Empty, owner))
+      case None => DdlFile(header ++ List(schemaCreate, Empty, table, Empty, commentOn))
+    }
     TableDefinition(path, filename, ddlFile)
   }
 
@@ -200,7 +206,7 @@ case class GenerateCommand(
    */
   private def jsonToRawTable(json: JsonFile): Validation[String, TableDefinition] = {
     val ddl = FlatSchema.flattenJsonSchema(json.content, splitProduct).map { flatSchema =>
-      produceRawTable(flatSchema, json.fileName)
+      produceRawTable(flatSchema, json.fileName, owner)
     }
     ddl match {
       case Failure(fail) => (fail + s" in [${json.fileName}] file").failure
@@ -218,15 +224,23 @@ case class GenerateCommand(
    * @param fileName JSON file, containing filename and content
    * @return DDL File object with all required information to output it
    */
-  private def produceRawTable(flatSchema: FlatSchema, fileName: String): TableDefinition = {
+  private def produceRawTable(flatSchema: FlatSchema, fileName: String, owner: Option[String]): TableDefinition = {
     val name = StringUtils.getTableName(fileName)
     val schemaCreate = dbSchema.map(CreateSchema(_)) match {
       case Some(sc) => List(sc, Empty)
       case None => Nil
     }
     val table = DdlGenerator.generateTableDdl(flatSchema, name, dbSchema, varcharSize, rawMode)
     val comment = DdlGenerator.getTableComment(name, dbSchema, fileName)
-    val ddlFile = DdlFile(header ++ schemaCreate ++ List(table, Empty, comment))
+    val ddlFile = owner match {
+      case Some(ownerStr) =>
+        val owner = dbSchema match {
+          case Some(sc) if sc.length() > 0 => AlterTable(sc+"."+name, OwnerTo(ownerStr))
+          case _ => AlterTable(name, OwnerTo(ownerStr))
+        }
+        DdlFile(header ++ schemaCreate ++ List(table, Empty, comment, Empty, owner))
+      case None => DdlFile(header ++ schemaCreate ++ List(table, Empty, comment))
+    }
     TableDefinition(".", name, ddlFile)
   }