Merge pull request #32 from civitaspo/develop

v0.1.1

CHANGELOG.md

@@ -1,3 +1,8 @@
+0.1.1 (2018-12-07)
+==================
+
+* [Fix] `athena.ctas>` Skip removing objects if removable targets do not exist.
+
 0.1.0 (2018-10-20)
 ==================
 

README.md

@@ -15,7 +15,7 @@ _export:
     repositories:
       - https://jitpack.io
     dependencies:
-      - pro.civitaspo:digdag-operator-athena:0.1.0
+      - pro.civitaspo:digdag-operator-athena:0.1.1
   athena:
     auth_method: profile
 
@@ -145,7 +145,7 @@ Define the below options on properties (which is indicated by `-c`, `--config`).
 - **table_mode**: Specify the expected behavior of CTAS results. Available values are `"default"`, `"empty"`, `"data_only"`. See the below explanation of the behaviour. (string, default: `"default"`)
   - `"default"`: Do not do any care. This option require the least IAM privileges for digdag, but the behaviour depends on Athena.
   - `"empty_table"`: Create a new empty table with the same schema as the select query results.
-  - `"data_only"`:
+  - `"data_only"`: Create a new table with data by CTAS, but drop this after CTAS execution. The table created by CTAS is an external table, so the data is left even if the table is dropped.
 - **save_mode**: Specify the expected behavior of CTAS. Available values are `"none"`, `"error_if_exists"`, `"ignore"`, `"overwrite"`. See the below explanation of the behaviour. (string, default: `"overwrite"`)
   - `"none"`: Do not do any care. This option require the least IAM privileges for digdag, but the behaviour depends on Athena.
   - `"error_if_exists"`: Raise error if the distination table or location exists.

build.gradle

@@ -6,7 +6,7 @@ plugins {
 }
 
 group = 'pro.civitaspo'
-version = '0.1.0'
+version = '0.1.1'
 
 def digdagVersion = '0.9.27'
 def awsSdkVersion = "1.11.372"

example/example.dig

@@ -4,7 +4,7 @@ _export:
       - file://${repos}
       # - https://jitpack.io
     dependencies:
-      - pro.civitaspo:digdag-operator-athena:0.1.0
+      - pro.civitaspo:digdag-operator-athena:0.1.1
   athena:
     auth_method: profile
     value: 5

src/main/scala/pro/civitaspo/digdag/plugin/athena/operator/AthenaCtasOperator.scala

@@ -111,6 +111,7 @@ class AthenaCtasOperator(operatorName: String, context: OperatorContext, systemC
   protected def rmObjects(location: String): Unit = {
     val uri: AmazonS3URI = AmazonS3URI(location)
     val keys: Seq[String] = withS3(_.listObjectsV2(uri.getBucket, uri.getKey)).getObjectSummaries.asScala.map(_.getKey)
+    if (keys.isEmpty) return
     val r: DeleteObjectsResult = withS3(_.deleteObjects(new DeleteObjectsRequest(uri.getBucket).withKeys(keys: _*)))
     r.getDeletedObjects.asScala.foreach(o => logger.info(s"Deleted: s3://${uri.getBucket}/${o.getKey}"))
   }