[SPARK-21786][SQL] The 'spark.sql.parquet.compression.codec' and 'spark.sql.orc.compression.codec' configuration doesn't take effect on hive table writing #19218
[SPARK-21786][SQL] The 'spark.sql.parquet.compression.codec' and 'spa…
…rk.sql.orc.compression.codec' configuration doesn't take effect on hive table writing Fix the test case issue
- Loading branch information
commit 676d6a7a64e137ca05981fae6cebea27b2d9d401
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,12 +21,16 @@ import java.io.File | |
|
|
||
| import org.scalatest.BeforeAndAfter | ||
|
|
||
| import scala.collection.JavaConverters._ | ||
|
|
||
| import org.apache.hadoop.fs.Path | ||
| import org.apache.spark.SparkException | ||
| import org.apache.spark.sql.{QueryTest, _} | ||
| import org.apache.spark.sql.catalyst.parser.ParseException | ||
| import org.apache.spark.sql.catalyst.plans.logical.InsertIntoTable | ||
| import org.apache.spark.sql.execution.datasources.parquet.ParquetTest | ||
| import org.apache.spark.sql.hive.orc.OrcFileOperator | ||
| import org.apache.spark.sql.hive.test.TestHiveSingleton | ||
| import org.apache.spark.sql.types._ | ||
| import org.apache.spark.util.Utils | ||
|
|
||
|
|
@@ -35,7 +39,7 @@ case class TestData(key: Int, value: String) | |
| case class ThreeCloumntable(key: Int, value: String, key1: String) | ||
|
|
||
| class InsertSuite extends QueryTest with TestHiveSingleton with BeforeAndAfter | ||
| with ParquetTest { | ||
| import spark.implicits._ | ||
|
|
||
| override lazy val testData = spark.sparkContext.parallelize( | ||
|
|
@@ -730,118 +734,185 @@ class InsertSuite extends QueryTest with TestHiveSingleton with BeforeAndAfter | |
| } | ||
|
|
||
| test("[SPARK-21786] Check 'spark.sql.parquet.compression.codec' " + | ||
|
||
| "and 'spark.sql.orc.compression.codec' taking effect on hive table writing") { | ||
|
|
||
| val hadoopConf = spark.sessionState.newHadoopConf() | ||
|
|
||
| val partitionStr = "p=10000" | ||
|
|
||
| case class TableCompressionConf(name: String, codeC: String) | ||
|
|
||
| case class TableDefine(tableName: String, isPartitioned: Boolean, format: String, | ||
|
||
| compressionConf: Option[TableCompressionConf]) { | ||
| def createTable(rootDir: File): Unit = { | ||
| val compression = compressionConf.map(cf => s"'${cf.name}'='${cf.codeC}'") | ||
| sql( | ||
| s""" | ||
| |CREATE TABLE $tableName(a int) | ||
| |${if (isPartitioned) "PARTITIONED BY (p int)" else ""} | ||
|
||
| |STORED AS $format | ||
| |LOCATION '${rootDir.toURI.toString.stripSuffix("/")}/$tableName' | ||
| |${if (compressionConf.nonEmpty) s"TBLPROPERTIES(${compression.get})" else ""} | ||
| """.stripMargin) | ||
| } | ||
|
|
||
| def insertOverwriteTable(): Unit = { | ||
| sql( | ||
| s""" | ||
| |INSERT OVERWRITE TABLE $tableName | ||
| |${if (isPartitioned) s"partition ($partitionStr)" else ""} | ||
| |SELECT * from table_source | ||
| """.stripMargin) | ||
| } | ||
| } | ||
|
|
||
| def getTableCompressionCodec(path: String, format: String): String = { | ||
| val codecs = format match { | ||
| case "parquet" => for { | ||
| footer <- readAllFootersWithoutSummaryFiles(new Path(path), hadoopConf) | ||
| block <- footer.getParquetMetadata.getBlocks.asScala | ||
| column <- block.getColumns.asScala | ||
| } yield column.getCodec.name() | ||
| case "orc" => new File(path).listFiles() | ||
| .filter(file => file.isFile && !file.getName.endsWith(".crc")).map { | ||
| orcFile => | ||
| OrcFileOperator.getFileReader(orcFile.toPath.toString).get.getCompression.toString | ||
| }.toSeq | ||
| } | ||
|
|
||
| assert(codecs.distinct.length == 1) | ||
| codecs.head | ||
| } | ||
|
|
||
| def checkCompressionCodecForTable(format:String, isPartitioned: Boolean, | ||
| compressionConf: Option[TableCompressionConf])(assertion: String => Boolean): Unit = { | ||
| val table = TableDefine(s"tbl_$format${isPartitioned}", | ||
| isPartitioned, format, compressionConf) | ||
| withTempDir { tmpDir => | ||
| withTable(table.tableName) { | ||
| table.createTable(tmpDir) | ||
| table.insertOverwriteTable() | ||
| val partition = if (table.isPartitioned) partitionStr else "" | ||
| val path = s"${tmpDir.getPath.stripSuffix("/")}/${table.tableName}/$partition" | ||
| assertion(getTableCompressionCodec(path, table.format)) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| def getConvertMetastoreConfName(format: String): String = format match { | ||
| case "parquet" => "spark.sql.hive.convertMetastoreParquet" | ||
| case "orc" => "spark.sql.hive.convertMetastoreOrc" | ||
| } | ||
|
|
||
| def getSparkCompressionConfName(format: String): String = format match { | ||
| case "parquet" => "spark.sql.parquet.compression.codec" | ||
| case "orc" => "spark.sql.orc.compression.codec" | ||
| } | ||
|
|
||
| def checkTableCompressionCodecForCodecs(format: String, isPartitioned: Boolean, | ||
| convertMetastore: Boolean, compressionCodecs: List[String], | ||
| tableCompressionConf: List[TableCompressionConf]) | ||
|
||
| (assertion: (Option[TableCompressionConf], String, String) => Boolean): Unit = { | ||
| withSQLConf(getConvertMetastoreConfName(format) -> convertMetastore.toString) { | ||
| tableCompressionConf.foreach { tableCompression => | ||
| compressionCodecs.foreach { sessionCompressionCodec => | ||
| withSQLConf(getSparkCompressionConfName(format) -> sessionCompressionCodec) { | ||
| val compression = if (tableCompression == null) None else Some(tableCompression) | ||
| checkCompressionCodecForTable(format, isPartitioned, compression) { | ||
| case realCompressionCodec => assertion(compression, | ||
| sessionCompressionCodec, realCompressionCodec) | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| def checkTableCompressionCodec(format: String, compressionCodecs: List[String], | ||
| tableCompressionConf: List[TableCompressionConf]): Unit = { | ||
| // For tables with table-level compression property, when | ||
| // 'spark.sql.hive.convertMetastoreParquet' was set to 'false', partitioned parquet tables | ||
| // and non-partitioned parquet tables will always take the table-level compression | ||
| // configuration first and ignore session compression configuration. | ||
| checkTableCompressionCodecForCodecs(format = format, isPartitioned = true, | ||
| convertMetastore = false, compressionCodecs, tableCompressionConf) { | ||
| case (tableCompressionCodec, sessionCompressionCodec, realCompressionCodec) => | ||
| // table-level take effect | ||
| tableCompressionCodec.get.codeC == realCompressionCodec | ||
| } | ||
|
|
||
| checkTableCompressionCodecForCodecs(format = format, isPartitioned = false, | ||
| convertMetastore = false, compressionCodecs, tableCompressionConf) { | ||
| case (tableCompressionCodec, sessionCompressionCodec, realCompressionCodec) => | ||
| // table-level take effect | ||
| tableCompressionCodec.get.codeC == realCompressionCodec | ||
| } | ||
|
|
||
| // For tables with table-level compression property, when | ||
| // 'spark.sql.hive.convertMetastoreParquet' was set to 'true', partitioned parquet tables | ||
| // will always take the table-level compression configuration first, but non-partitioned tables | ||
| // will take the session-level compression configuration. | ||
| checkTableCompressionCodecForCodecs(format = format, isPartitioned = true, | ||
| convertMetastore = true, compressionCodecs, tableCompressionConf) { | ||
| case (tableCompressionCodec, sessionCompressionCodec, realCompressionCodec) => | ||
| // table-level take effect | ||
| tableCompressionCodec.get.codeC == realCompressionCodec | ||
| } | ||
|
|
||
| checkTableCompressionCodecForCodecs(format = format, isPartitioned = false, | ||
| convertMetastore = true, compressionCodecs, tableCompressionConf) { | ||
| case (tableCompressionCodec, sessionCompressionCodec, realCompressionCodec) => | ||
| // session-level take effect | ||
| sessionCompressionCodec == realCompressionCodec | ||
| } | ||
|
|
||
| // For tables without table-level compression property, session-level compression configuration | ||
| // will take effect. | ||
| checkTableCompressionCodecForCodecs(format = format, isPartitioned = true, | ||
| convertMetastore = true, compressionCodecs, List(null)) { | ||
| case (tableCompressionCodec, sessionCompressionCodec, realCompressionCodec) => | ||
| // session-level take effect | ||
| sessionCompressionCodec == realCompressionCodec | ||
| } | ||
|
|
||
| checkTableCompressionCodecForCodecs(format = format, isPartitioned = false, | ||
| convertMetastore = true, compressionCodecs, List(null)) { | ||
| case (tableCompressionCodec, sessionCompressionCodec, realCompressionCodec) => | ||
| // session-level take effect | ||
| sessionCompressionCodec == realCompressionCodec | ||
| } | ||
|
|
||
| checkTableCompressionCodecForCodecs(format = format, isPartitioned = true, | ||
| convertMetastore = false, compressionCodecs, List(null)) { | ||
| case (tableCompressionCodec, sessionCompressionCodec, realCompressionCodec) => | ||
| // session-level take effect | ||
| sessionCompressionCodec == realCompressionCodec | ||
| } | ||
|
|
||
| checkTableCompressionCodecForCodecs(format = format, isPartitioned = false, | ||
| convertMetastore = false, compressionCodecs, List(null)) { | ||
| case (tableCompressionCodec, sessionCompressionCodec, realCompressionCodec) => | ||
| // session-level take effect | ||
| sessionCompressionCodec == realCompressionCodec | ||
| } | ||
| } | ||
|
|
||
| withTempView("table_source") { | ||
| (0 until 100000).toDF("a").createOrReplaceTempView("table_source") | ||
| val parquetCompressionCodec = List("UNCOMPRESSED", "SNAPPY", "GZIP") | ||
| val tableCompressionConf = parquetCompressionCodec.map { tableCodec => | ||
| TableCompressionConf("parquet.compression", tableCodec) | ||
| } | ||
| checkTableCompressionCodec("parquet", parquetCompressionCodec, tableCompressionConf) | ||
| } | ||
|
|
||
| withTempView("table_source") { | ||
| (0 until 100000).toDF("a").createOrReplaceTempView("table_source") | ||
| val orcCompressionCodec = List("NONE", "SNAPPY", "ZLIB") | ||
| val tableCompressionConf = orcCompressionCodec.map { tableCodec => | ||
| TableCompressionConf("parquet.compression", tableCodec) | ||
| } | ||
| checkTableCompressionCodec("orc", orcCompressionCodec, tableCompressionConf) | ||
| } | ||
| } | ||
| } | ||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is the insert suite. We are unable to do this.
Could you create a separate suite in the current package
org.apache.spark.sql.hive? The suite name can beCompressionCodecSuiteThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please also check whether the compression takes an effect? Compare the size whether is smaller than the original size without compressions?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ok, I will do it.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Seems compressed table does not always be smaller than uncompressed tables.
SNAPPYCompression size may be bigger than non-compression size when the amount of data is not big. So I'd like to check the size not equal when compression are different.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Fine to me. Thanks!