Merge pull request #55 from sekikn/replace-parquet-tools

Replace parquet-tools with parquet-avro

build.gradle

@@ -53,7 +53,8 @@ dependencies {
     }
     testImplementation "org.embulk:embulk-core:0.9.23:tests"
     testImplementation "org.scalatest:scalatest_2.13:3.1.1"
-    testImplementation 'org.apache.parquet:parquet-tools:1.11.0'
+    testImplementation 'org.apache.parquet:parquet-avro:1.11.0'
+    testImplementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-avro:2.14.0'
     testImplementation 'org.apache.hadoop:hadoop-client:2.9.2'
 }
 

src/test/scala/org/embulk/output/s3_parquet/EmbulkPluginTestHelper.scala

@@ -13,12 +13,13 @@ import com.amazonaws.services.s3.transfer.{
   TransferManagerBuilder
 }
 import com.google.inject.{Binder, Guice, Module, Stage}
+import org.apache.avro.generic.GenericRecord
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{Path => HadoopPath}
+import org.apache.parquet.avro.AvroReadSupport
 import org.apache.parquet.hadoop.{ParquetFileReader, ParquetReader}
 import org.apache.parquet.hadoop.util.HadoopInputFile
 import org.apache.parquet.schema.MessageType
-import org.apache.parquet.tools.read.{SimpleReadSupport, SimpleRecord}
 import org.embulk.{TestPluginSourceModule, TestUtilityModule}
 import org.embulk.config.{
   ConfigLoader,
@@ -229,27 +230,16 @@ abstract class EmbulkPluginTestHelper
       )
     ) { reader => messageTypeTest(reader.getFileMetaData.getSchema) }
 
-    val reader: ParquetReader[SimpleRecord] = ParquetReader
+    val reader: ParquetReader[GenericRecord] = ParquetReader
       .builder(
-        new SimpleReadSupport(),
+        new AvroReadSupport[GenericRecord](),
         new HadoopPath(pathString)
       )
       .build()
 
-    def read(
-        reader: ParquetReader[SimpleRecord],
-        records: Seq[Seq[AnyRef]] = Seq()
-    ): Seq[Seq[AnyRef]] = {
-      val simpleRecord: SimpleRecord = reader.read()
-      if (simpleRecord != null) {
-        val r: Seq[AnyRef] = simpleRecord.getValues
-          .map(_.getValue)
-        return read(reader, records :+ r)
-      }
-      records
-    }
-    try read(reader)
-    finally reader.close()
+    Iterator.continually(reader.read()).takeWhile(_ != null).map(
+      record => record.getSchema.getFields.map(f => record.get(f.name()))
+    ).toSeq
   }
 
   def loadConfigSourceFromYamlString(yaml: String): ConfigSource =