#134: Added support for HDFS (#136)

.travis.yml

@@ -17,7 +17,7 @@ scala:
 
 env:
   - EXASOL_DOCKER_VERSION="6.2.12-d1"
-  - EXASOL_DOCKER_VERSION="7.0.6"
+  - EXASOL_DOCKER_VERSION="7.0.8"
 
 before_install:
   - git fetch --tags

build.sbt

@@ -18,7 +18,7 @@ lazy val root =
   project
     .in(file("."))
     .settings(moduleName := "cloud-storage-extension")
-    .settings(version := "1.0.0")
+    .settings(version := "1.1.0")
     .settings(orgSettings)
     .settings(buildSettings)
     .settings(Settings.projectSettings(scalaVersion))

doc/changes/changelog.md

@@ -1,5 +1,6 @@
 # Releases
 
+* [1.1.0](changes_1.1.0.md)
 * [1.0.0](changes_1.0.0.md)
 * [0.9.0](changes_0.9.0.md)
 * [0.8.0](changes_0.8.0.md)

doc/changes/changes_1.1.0.md

@@ -0,0 +1,25 @@
+# Cloud Storage Extension 1.1.0, released 2021-MM-DD
+
+## Features
+
+* #134: Added support for Hadoop Distributed Filesystem (HDFS) (PR #136)
+
+## Documentation
+
+* #131: Added access privilege to connection object (PR #132).
+
+## Dependency Updates
+
+### Runtime Dependency Updates
+
+### Test Dependency Updates
+
+* Updated `org.scalatest:scalatest:3.2.3` to `3.2.6`
+* Updated `org.mockito:mockito-core:3.7.7` to `3.8.0`
+* Updated `org.testcontainers:localstack:1.15.1` to `1.15.2`
+* Updated `com.exasol:test-db-builder-java:3.0.0` to `3.1.1`
+* Updated `com.exasol:exasol-testcontainers:3.5.0` to `3.5.1`
+
+### Plugin Updates
+
+* Updated `com.timushev.sbt:sbt-updates:0.5.1` to `0.5.2`

doc/user_guide/user_guide.md

@@ -275,12 +275,12 @@ Run these statements to create export UDF scripts:
 OPEN SCHEMA CLOUD_STORAGE_EXTENSION;
 
 CREATE OR REPLACE JAVA SET SCRIPT EXPORT_PATH(...) EMITS (...) AS
-  %scriptclass com.exasol.cloudetl.scriptclasses.ExportPath;
+  %scriptclass com.exasol.cloudetl.scriptclasses.TableExportQueryGenerator;
   %jar /buckets/bfsdefault/<BUCKET>/exasol-cloud-storage-extension-<VERSION>.jar;
 /
 
 CREATE OR REPLACE JAVA SET SCRIPT EXPORT_TABLE(...) EMITS (ROWS_AFFECTED INT) AS
-  %scriptclass com.exasol.cloudetl.scriptclasses.ExportTable;
+  %scriptclass com.exasol.cloudetl.scriptclasses.TableDataExporter;
   %jar /buckets/bfsdefault/<BUCKET>/exasol-cloud-storage-extension-<VERSION>.jar;
 /
 ```
@@ -291,6 +291,28 @@ UDF and it will call the `EXPORT_TABLE` script internally.
 Make sure you change the `<BUCKET>` name and jar version `<VERSION>`
 accordingly.
 
+#### Setup Export UDF Scripts in Docker
+
+Similar to import, the UDF scripts require slightly different deployment for
+Exasol Docker installations.
+
+```sql
+OPEN SCHEMA CLOUD_STORAGE_EXTENSION;
+
+CREATE OR REPLACE JAVA SET SCRIPT EXPORT_PATH(...) EMITS (...) AS
+  %scriptclass com.exasol.cloudetl.scriptclasses.DockerTableExportQueryGenerator;
+  %jar /buckets/bfsdefault/<BUCKET>/exasol-cloud-storage-extension-<VERSION>.jar;
+/
+
+CREATE OR REPLACE JAVA SET SCRIPT EXPORT_TABLE(...) EMITS (ROWS_AFFECTED INT) AS
+  %scriptclass com.exasol.cloudetl.scriptclasses.DockerTableDataExporter;
+  %jar /buckets/bfsdefault/<BUCKET>/exasol-cloud-storage-extension-<VERSION>.jar;
+/
+```
+
+Please notice that we use different class names for the `%scriptclasses`
+parameter.
+
 ## Prepare an Exasol Table for Import
 
 To store the imported data, you need to create a table inside the Exasol

project/Dependencies.scala

@@ -16,14 +16,14 @@ object Dependencies {
   private val SparkSQLVersion = "3.0.1"
 
   // Test dependencies versions
-  private val ScalaTestVersion = "3.2.3"
+  private val ScalaTestVersion = "3.2.6"
   private val ScalaTestPlusVersion = "1.0.0-M2"
-  private val MockitoCoreVersion = "3.7.7"
+  private val MockitoCoreVersion = "3.8.0"
   private val HamcrestVersion = "2.2"
   private val ExasolHamcrestMatcherVersion = "1.4.0"
-  private val ExasolTestDBBuilderVersion = "3.0.0"
-  private val ExasolTestContainersVersion = "3.5.0"
-  private val TestContainersLocalstackVersion = "1.15.1"
+  private val ExasolTestDBBuilderVersion = "3.1.1"
+  private val ExasolTestContainersVersion = "3.5.1"
+  private val TestContainersLocalstackVersion = "1.15.2"
 
   val Resolvers: Seq[Resolver] = Seq(
     "Exasol Releases" at "https://maven.exasol.com/artifactory/exasol-releases"
@@ -57,6 +57,7 @@ object Dependencies {
         ExclusionRule(organization = "org.apache.curator"),
         ExclusionRule(organization = "org.apache.zookeeper")
     ),
+    "org.apache.hadoop" % "hadoop-hdfs" % HadoopVersion,
     "com.google.cloud.bigdataoss" % "gcs-connector" % GoogleStorageVersion
       exclude ("com.google.guava", "guava")
       exclude ("org.apache.httpcomponents", "httpclient"),

project/plugins.sbt

@@ -27,7 +27,7 @@ addSbtPlugin("org.scoverage" % "sbt-coveralls" % "1.2.7")
 // Adds a `dependencyUpdates` task to check Maven repositories for
 // dependency updates
 // http://github.com/rtimush/sbt-updates
-addSbtPlugin("com.timushev.sbt" % "sbt-updates" % "0.5.1")
+addSbtPlugin("com.timushev.sbt" % "sbt-updates" % "0.5.2")
 
 // Adds a `scalafmt` task for automatic source code formatting
 // https://github.com/lucidsoftware/neo-sbt-scalafmt

project/project/plugins.sbt

@@ -4,4 +4,4 @@ addSbtPlugin("com.lucidchart" % "sbt-scalafmt-coursier" % "1.16")
 
 // Used to get updates for plugins
 // see https://github.com/rtimush/sbt-updates/issues/10
-addSbtPlugin("com.timushev.sbt" % "sbt-updates" % "0.5.0")
+addSbtPlugin("com.timushev.sbt" % "sbt-updates" % "0.5.2")

src/it/scala/com/exasol/cloudetl/BaseIntegrationTest.scala

@@ -24,7 +24,7 @@ trait BaseIntegrationTest extends AnyFunSuite with BeforeAndAfterAll {
   private[this] val JAR_DIRECTORY_PATTERN = "scala-"
   private[this] val JAR_NAME_PATTERN = "cloud-storage-extension-"
 
-  private[this] val DEFAULT_EXASOL_DOCKER_IMAGE = "7.0.6"
+  private[this] val DEFAULT_EXASOL_DOCKER_IMAGE = "7.0.8"
   private[this] val DEFAULT_LOCALSTACK_DOCKER_IMAGE =
     DockerImageName.parse("localstack/localstack:0.12.5")
 

src/main/scala/com/exasol/cloudetl/bucket/Bucket.scala

@@ -142,6 +142,7 @@ object Bucket extends LazyLogging {
       case "abfs" | "abfss" => AzureAbfsBucket(path, storageProperties)
       case "adl"            => AzureAdlsBucket(path, storageProperties)
       case "wasb" | "wasbs" => AzureBlobBucket(path, storageProperties)
+      case "hdfs"           => HDFSBucket(path, storageProperties)
       case "file"           => LocalBucket(path, storageProperties)
       case _ =>
         throw new IllegalArgumentException(s"Unsupported path scheme $scheme!")

src/main/scala/com/exasol/cloudetl/bucket/HDFSBucket.scala

@@ -0,0 +1,31 @@
+package com.exasol.cloudetl.bucket
+
+import com.exasol.cloudetl.storage.StorageProperties
+
+import org.apache.hadoop.conf.Configuration
+
+/** A [[Bucket]] implementation for the HDFS filesystem */
+final case class HDFSBucket(path: String, params: StorageProperties) extends Bucket {
+
+  /** @inheritdoc */
+  override val bucketPath: String = path
+
+  /** @inheritdoc */
+  override val properties: StorageProperties = params
+
+  /** Returns the list of required property keys for HDFS filesystem. */
+  override def getRequiredProperties(): Seq[String] = Seq.empty[String]
+
+  /** @inheritdoc */
+  override def validate(): Unit =
+    validateRequiredProperties()
+
+  /** @inheritdoc */
+  override def getConfiguration(): Configuration = {
+    validate()
+    val conf = new Configuration()
+    conf.set("fs.hdfs.impl", classOf[org.apache.hadoop.hdfs.DistributedFileSystem].getName())
+    conf
+  }
+
+}

src/main/scala/com/exasol/cloudetl/scriptclasses/DockerTableDataExporter.scala

@@ -0,0 +1,17 @@
+package com.exasol.cloudetl.scriptclasses
+
+import com.exasol.ExaIterator
+import com.exasol.ExaMetadata
+
+/**
+ * A table data exporter class to run inside the Exasol docker container.
+ */
+object DockerTableDataExporter {
+
+  def run(metadata: ExaMetadata, iterator: ExaIterator): Unit = {
+    import org.apache.hadoop.security.UserGroupInformation
+    UserGroupInformation.setLoginUser(UserGroupInformation.createRemoteUser("exauser"))
+    TableDataExporter.run(metadata, iterator)
+  }
+
+}

src/main/scala/com/exasol/cloudetl/scriptclasses/DockerTableExportQueryGenerator.scala

@@ -0,0 +1,20 @@
+package com.exasol.cloudetl.scriptclasses
+
+import com.exasol.ExaExportSpecification
+import com.exasol.ExaMetadata
+
+/**
+ * An export query generator class to run inside the Exasol docker container.
+ */
+object DockerTableExportQueryGenerator {
+
+  def generateSqlForExportSpec(
+    metadata: ExaMetadata,
+    exportSpecification: ExaExportSpecification
+  ): String = {
+    import org.apache.hadoop.security.UserGroupInformation
+    UserGroupInformation.setLoginUser(UserGroupInformation.createRemoteUser("exauser"))
+    TableExportQueryGenerator.generateSqlForExportSpec(metadata, exportSpecification)
+  }
+
+}

src/main/scala/com/exasol/cloudetl/scriptclasses/ExportTable.scala → src/main/scala/com/exasol/cloudetl/scriptclasses/TableDataExporter.scala

@@ -13,8 +13,17 @@ import com.exasol.common.data.Row
 
 import com.typesafe.scalalogging.LazyLogging
 
-object ExportTable extends LazyLogging {
+/**
+ * A data exporter class that exports table data into a filesystem.
+ */
+object TableDataExporter extends LazyLogging {
 
+  /**
+   * Reads the table data and saves them to external filesystem.
+   *
+   * @param metadata an Exasol metadata object
+   * @param iterator an Exasol iterator object
+   */
   def run(metadata: ExaMetadata, iterator: ExaIterator): Unit = {
     val storageProperties = StorageProperties(iterator.getString(1), metadata)
     val bucket = Bucket(storageProperties)

src/main/scala/com/exasol/cloudetl/scriptclasses/ExportPath.scala → src/main/scala/com/exasol/cloudetl/scriptclasses/TableExportQueryGenerator.scala

@@ -9,8 +9,17 @@ import com.exasol.cloudetl.storage.StorageProperties
 
 import org.apache.hadoop.fs.Path
 
-object ExportPath {
+/**
+ * A SQL query generator that call the exporter classes.
+ */
+object TableExportQueryGenerator {
 
+  /**
+   * Generates an Exasol SQL for the data export query.
+   *
+   * @param metadata an Exasol metadata object
+   * @param exportSpecification an Exasol export specification object
+   */
   def generateSqlForExportSpec(
     metadata: ExaMetadata,
     exportSpec: ExaExportSpecification
@@ -24,7 +33,6 @@ object ExportPath {
     val parallelism = storageProperties.getParallelism("iproc()")
     val storagePropertiesStr = storageProperties.mkString()
     val scriptSchema = metadata.getScriptSchema
-
     val srcColumns = getSourceColumns(exportSpec)
     val srcColumnsStr = srcColumns.mkString(".")
 

src/test/scala/com/exasol/cloudetl/bucket/HDFSBucketTest.scala

@@ -0,0 +1,17 @@
+package com.exasol.cloudetl.bucket
+
+class HDFSBucketTest extends AbstractBucketTest {
+
+  test("apply sets correct configuration") {
+    val properties = Map(
+      PATH -> "hdfs://dir/path",
+      FORMAT -> "orc"
+    )
+    val exaMetadata = mockConnectionInfo("", "")
+    val bucket = getBucket(properties, exaMetadata)
+    val expectedFileSystemName = classOf[org.apache.hadoop.hdfs.DistributedFileSystem].getName()
+    assert(bucket.isInstanceOf[HDFSBucket])
+    assert(bucket.getConfiguration().get("fs.hdfs.impl") === expectedFileSystemName)
+  }
+
+}

src/test/scala/com/exasol/cloudetl/scriptclasses/ExportTableTest.scala → src/test/scala/com/exasol/cloudetl/scriptclasses/TableDataExporterTest.scala

@@ -15,7 +15,7 @@ import org.mockito.ExtraMockito
 import org.mockito.Mockito._
 import org.scalatest.BeforeAndAfterEach
 
-class ExportTableTest
+class TableDataExporterTest
     extends StorageTest
     with BeforeAndAfterEach
     with DataRecords
@@ -98,7 +98,7 @@ class ExportTableTest
   }
 
   test("run exports table rows") {
-    ExportTable.run(metadata, iterator)
+    TableDataExporter.run(metadata, iterator)
 
     verify(metadata, times(1)).getInputColumnCount
     for { idx <- 3 to 10 } {
@@ -119,7 +119,7 @@ class ExportTableTest
   }
 
   test("imports exported rows from a path") {
-    ExportTable.run(metadata, iterator)
+    TableDataExporter.run(metadata, iterator)
 
     val properties = Map("BUCKET_PATH" -> testResourceDir, "DATA_FORMAT" -> "PARQUET")
     val importIter = mockExasolIterator(properties)
@@ -133,7 +133,7 @@ class ExportTableTest
   }
 
   test("export creates file without compression extension if compression codec is not set") {
-    ExportTable.run(metadata, iterator)
+    TableDataExporter.run(metadata, iterator)
     assert(Files.exists(outputPath) === true)
     assert(Files.list(outputPath).count() === 2)
     checkExportFileExtensions(outputPath, "")
@@ -142,7 +142,7 @@ class ExportTableTest
   test("export creates file with compression extension if compression codec is set") {
     val properties = defaultProperties ++ Map("PARQUET_COMPRESSION_CODEC" -> "SNAPPY")
     iterator = createMockedIterator(outputPath.toUri.toString, properties)
-    ExportTable.run(metadata, iterator)
+    TableDataExporter.run(metadata, iterator)
     assert(Files.exists(outputPath) === true)
     assert(Files.list(outputPath).count() === 2)
     checkExportFileExtensions(outputPath, ".snappy")

src/test/scala/com/exasol/cloudetl/scriptclasses/ExportPathTest.scala → src/test/scala/com/exasol/cloudetl/scriptclasses/TableExportQueryGeneratorTest.scala

@@ -8,7 +8,7 @@ import com.exasol.cloudetl.storage.StorageProperties
 
 import org.mockito.Mockito._
 
-class ExportPathTest extends PathTest {
+class TableExportQueryGeneratorTest extends PathTest {
 
   test("generateSqlForExportSpec returns SQL statement") {
     when(metadata.getScriptSchema()).thenReturn(schema)
@@ -31,7 +31,10 @@ class ExportPathTest extends PathTest {
          |  iproc();
          |""".stripMargin
 
-    assert(ExportPath.generateSqlForExportSpec(metadata, exportSpec) === expectedSQLStatement)
+    assert(
+      TableExportQueryGenerator
+        .generateSqlForExportSpec(metadata, exportSpec) === expectedSQLStatement
+    )
     verify(metadata, atLeastOnce).getScriptSchema
     verify(exportSpec, times(1)).getParameters
     verify(exportSpec, times(1)).getSourceColumnNames
@@ -43,7 +46,7 @@ class ExportPathTest extends PathTest {
     when(exportSpec.getParameters()).thenReturn(newProperties.asJava)
 
     val thrown = intercept[IllegalArgumentException] {
-      ExportPath.generateSqlForExportSpec(metadata, exportSpec)
+      TableExportQueryGenerator.generateSqlForExportSpec(metadata, exportSpec)
     }
     assert(thrown.getMessage === "Please provide a value for the S3_ENDPOINT property!")
     verify(exportSpec, times(1)).getParameters
@@ -57,7 +60,7 @@ class ExportPathTest extends PathTest {
     when(exportSpec.getSourceColumnNames).thenReturn(srcCols.asJava)
 
     val thrown = intercept[RuntimeException] {
-      ExportPath.generateSqlForExportSpec(metadata, exportSpec)
+      TableExportQueryGenerator.generateSqlForExportSpec(metadata, exportSpec)
     }
     assert(thrown.getMessage === "Could not parse the column name from 'tbl.c_int.integer'!")
     verify(metadata, atLeastOnce).getScriptSchema
@@ -79,7 +82,7 @@ class ExportPathTest extends PathTest {
     )
     when(metadata.getScriptSchema()).thenReturn(schema)
     when(exportSpec.getParameters()).thenReturn(newProperties.asJava)
-    ExportPath.generateSqlForExportSpec(metadata, exportSpec)
+    TableExportQueryGenerator.generateSqlForExportSpec(metadata, exportSpec)
     assert(Files.exists(bucketPath) === true)
     assert(Files.list(bucketPath).findAny().isPresent() === true)
     assert(Files.list(bucketPath).count() === 3)
@@ -97,7 +100,7 @@ class ExportPathTest extends PathTest {
     )
     when(metadata.getScriptSchema()).thenReturn(schema)
     when(exportSpec.getParameters()).thenReturn(newProperties.asJava)
-    ExportPath.generateSqlForExportSpec(metadata, exportSpec)
+    TableExportQueryGenerator.generateSqlForExportSpec(metadata, exportSpec)
     assert(Files.exists(bucketPath) === false)
   }