[SPARK-52574][SQL][TESTS] Ensure compression codec is correctly applied in Hive tables and dirs

### What changes were proposed in this pull request?

This PR adds a test to verify that the compression codec specified in `spark.sql.parquet.compression.codec` is correctly applied to Hive tables and directories during table creation and data insertion.

### Why are the changes needed?

We add compression codec through `setupHadoopConfForCompression` and keep compress codec in hadoopConf:
https://github.com/apache/spark/blob/3e0808c33f185c13808ce2d547ce9ba0057d31a6/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/V1WritesHiveUtils.scala#L114-L141

However, we might accidentally use `sparkSession.sessionState.newHadoopConf()` instead of the old hadoopConf. This can cause the compression codec to be lost. So, we add a test to make sure this does not happen.

Keep compression   codec | Lost compression codec
-- | --
<img width="747" alt="image" src="https://github.com/user-attachments/assets/67c5c3c6-a265-48c8-a77b-5d2db71af1a0" /> | <img width="768" alt="image" src="https://github.com/user-attachments/assets/71dcab6f-4237-4515-b942-e3531fb08899" />

</body>

</html>

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Manual test.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #51279 from wangyum/SPARK-52574.

Lead-authored-by: Yuming Wang <[email protected]>
Co-authored-by: Yuming Wang <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: wangyum

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveParquetSuite.scala

@@ -17,10 +17,15 @@
 
 package org.apache.spark.sql.hive
 
+import java.io.File
 import java.time.{Duration, Period}
 import java.time.temporal.ChronoUnit
 
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.hadoop.ParquetFileReader
+
 import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.execution.datasources.parquet.{ParquetCompressionCodec, ParquetTest}
 import org.apache.spark.sql.hive.test.TestHiveSingleton
 import org.apache.spark.sql.internal.SQLConf
@@ -179,4 +184,41 @@ class HiveParquetSuite extends QueryTest
       }
     }
   }
+
+  test("SPARK-52574: Ensure compression codec is correctly applied in Hive tables and dirs") {
+    withSQLConf(
+      HiveUtils.CONVERT_METASTORE_PARQUET.key -> "false",
+      HiveUtils.CONVERT_METASTORE_INSERT_DIR.key -> "false",
+      SQLConf.PARQUET_COMPRESSION.key -> ParquetCompressionCodec.SNAPPY.lowerCaseName()) {
+      withTable("tbl") {
+        sql("CREATE TABLE tbl(id int) STORED AS PARQUET")
+        sql("INSERT INTO tbl SELECT id AS part FROM range(10)")
+        val tblMata = spark.sessionState.catalog.getTableMetadata(TableIdentifier("tbl"))
+        checkCompressionCodec(new File(tblMata.storage.locationUri.get))
+      }
+
+      withTempPath { dir =>
+        sql(
+          s"""
+             |INSERT OVERWRITE LOCAL DIRECTORY '${dir.getCanonicalPath}'
+             |STORED AS parquet
+             |SELECT id FROM range(10)
+             |""".stripMargin)
+        checkCompressionCodec(dir)
+      }
+    }
+
+    def checkCompressionCodec(dir: File): Unit = {
+      val parquetFiles = dir.listFiles().filter(_.getName.startsWith("part-"))
+      assert(parquetFiles.nonEmpty, "No Parquet files found")
+
+      val conf = spark.sessionState.newHadoopConf()
+      val file = parquetFiles.head
+      val footer = ParquetFileReader.readFooter(conf, new Path(file.getAbsolutePath))
+
+      val codec = footer.getBlocks.get(0).getColumns.get(0).getCodec.name()
+      assert(codec.equalsIgnoreCase(ParquetCompressionCodec.SNAPPY.lowerCaseName()),
+        s"Expected ${ParquetCompressionCodec.SNAPPY.lowerCaseName()} compression but found $codec")
+    }
+  }
 }