Table ownership setup as part of DDL (close snowplow/iglu#225)

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>"

src/main/scala/com.snowplowanalytics.iglu/ctl/GenerateCommand.scala

@@ -52,7 +52,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._
 
@@ -211,7 +212,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
@@ -236,13 +237,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)
   }
 
@@ -275,7 +281,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
@@ -293,15 +299,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)
   }
 

src/test/scala/com/snowplowanalytics/iglu/ctl/GenerateCommandSpec.scala

@@ -35,6 +35,7 @@ class GenerateCommandSpec extends Specification { def is = s2"""
     output correct warnings for DDL-generation process $e5
     warn about missing schema versions (addition) $e6
     warn about missing 1-0-0 schema version $e7
+    correctly setup table ownership $e8
   """
 
   def e1 = {
@@ -761,4 +762,45 @@ class GenerateCommandSpec extends Specification { def is = s2"""
         s" Use --force to switch off schema version check."
     ))
   }
+
+  def e8 = {
+    val sourceSchema = parse(
+      """
+        |{
+        |      "$schema": "http://iglucentral.com/schemas/com.snowplowanalytics.self-desc/schema/jsonschema/1-0-0#",
+        |      "description": "Schema for custom contexts",
+        |      "type": "object",
+        |      "properties": {
+        |                      "name": {
+        |                      "type": "string"
+        |                      },
+        |                      "age": {
+        |                      "type": "number"
+        |                      }
+        |      },
+        |      "required":["name"]
+        |}
+      """.stripMargin)
+
+    val resultContent =
+      """|CREATE TABLE IF NOT EXISTS 1_0_0 (
+         |    "name" VARCHAR(4096)    ENCODE ZSTD NOT NULL,
+         |    "age"  DOUBLE PRECISION ENCODE RAW
+         |);
+         |
+         |COMMENT ON TABLE 1_0_0 IS 'Source: 1-0-0';
+         |
+         |ALTER TABLE 1_0_0 OWNER TO storageloader;""".stripMargin
+
+    val jsonFile = JsonFile(sourceSchema, new File("1-0-0"))
+    val stubFile: File = new File(".")
+    val command = GenerateCommand(stubFile, stubFile, rawMode = true, noHeader = true, owner = Some("storageloader"))
+    val ddl = command.transformRaw(List(jsonFile))
+
+    val expected = GenerateCommand.DdlOutput(
+      List(TextFile(new File("./1_0_0.sql"), resultContent))
+    )
+
+    ddl must beEqualTo(expected)
+  }
 }