[CARMEL-5873] Upgrade Parquet to 1.12.2 (#901)

* [CARMEL-5873] Upgrade Parquet to 1.12.2 (#896)

* [SPARK-36726] Upgrade Parquet to 1.12.1

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

Upgrade Apache Parquet to 1.12.1

### Why are the changes needed?

Parquet 1.12.1 contains the following bug fixes:
- PARQUET-2064: Make Range public accessible in RowRanges
- PARQUET-2022: ZstdDecompressorStream should close `zstdInputStream`
- PARQUET-2052: Integer overflow when writing huge binary using dictionary encoding
- PARQUET-1633: Fix integer overflow
- PARQUET-2054: fix TCP leaking when calling ParquetFileWriter.appendFile
- PARQUET-2072: Do Not Determine Both Min/Max for Binary Stats
- PARQUET-2073: Fix estimate remaining row count in ColumnWriteStoreBase
- PARQUET-2078: Failed to read parquet file after writing with the same

In particular PARQUET-2078 is a blocker for the upcoming Apache Spark 3.2.0 release.

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

No

### How was this patch tested?

Existing tests + a new test for the issue in SPARK-36696

Closes #33969 from sunchao/upgrade-parquet-12.1.

Authored-by: Chao Sun <[email protected]>
Signed-off-by: DB Tsai <[email protected]>

(cherry picked from commit a927b08)

* [SPARK-34542][BUILD] Upgrade Parquet to 1.12.0

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

Parquet 1.12.0 New Feature
- PARQUET-41 - Add bloom filters to parquet statistics
- PARQUET-1373 - Encryption key management tools
- PARQUET-1396 - Example of using EncryptionPropertiesFactory and DecryptionPropertiesFactory
- PARQUET-1622 - Add BYTE_STREAM_SPLIT encoding
- PARQUET-1784 - Column-wise configuration
- PARQUET-1817 - Crypto Properties Factory
- PARQUET-1854 - Properties-Driven Interface to Parquet Encryption

Parquet 1.12.0 release notes:
https://github.com/apache/parquet-mr/blob/apache-parquet-1.12.0/CHANGES.md

### Why are the changes needed?

- Bloom filters to improve filter performance
- ZSTD enhancement

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

No.

### How was this patch tested?

Existing unit test.

Closes #31649 from wangyum/SPARK-34542.

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

(cherry picked from commit cbffc12)

Co-authored-by: Chao Sun <[email protected]>

* [HADP-44647] Parquet file based kms client for encryption keys (#897)

* [HADP-44647] Parquet file based kms client for encryption keys (#82)

create/write parquet encryption table.
```
set spark.sql.parquet.encryption.key.file=/path/to/key/file;

create table parquet_encryption(a int, b int, c int)
using parquet
options (
'parquet.encryption.column.keys' 'columnKey1: a, b; columnKey2: c',
'parquet.encryption.footer.key' 'footerKey');
```

read parquet encryption table;

```
set spark.sql.parquet.encryption.key.file=/path/to/key/file;

select ... from parquet_encryption ...
```

Will raise another pr for default footerKey.

* [HADP-44647][FOLLOWUP] Reuse the kms instance for same key file (#84)

* Fix

Co-authored-by: fwang12 <[email protected]>

Co-authored-by: Chao Sun <[email protected]>
Co-authored-by: fwang12 <[email protected]>

Author: 3 people

dev/deps/spark-deps-hadoop-2.7-hive-1.2

@@ -61,7 +61,7 @@ curator-recipes/2.7.1//curator-recipes-2.7.1.jar
 datanucleus-api-jdo/3.2.6//datanucleus-api-jdo-3.2.6.jar
 datanucleus-core/3.2.10//datanucleus-core-3.2.10.jar
 datanucleus-rdbms/3.2.9//datanucleus-rdbms-3.2.9.jar
-delta-core_2.12/0.8.0.carmel0.1-SNAPSHOT//delta-core_2.12-0.8.0.carmel0.1-SNAPSHOT.jar
+delta-core_2.12/0.9.0.carmel0.2-SNAPSHOT//delta-core_2.12-0.9.0.carmel0.2-SNAPSHOT.jar
 derby/10.12.1.1//derby-10.12.1.1.jar
 flatbuffers-java/1.9.0//flatbuffers-java-1.9.0.jar
 generex/1.0.2//generex-1.0.2.jar
@@ -148,7 +148,9 @@ jta/1.1//jta-1.1.jar
 jul-to-slf4j/1.7.30//jul-to-slf4j-1.7.30.jar
 kafka-clients/1.1.1//kafka-clients-1.1.1.jar
 kafka-schema-registry-client/3.0.0//kafka-schema-registry-client-3.0.0.jar
+kryo-serializers/0.43//kryo-serializers-0.43.jar
 kryo-shaded/4.0.2//kryo-shaded-4.0.2.jar
+kryo/4.0.2//kryo-4.0.2.jar
 kubernetes-client/4.9.2//kubernetes-client-4.9.2.jar
 kubernetes-model-common/4.9.2//kubernetes-model-common-4.9.2.jar
 kubernetes-model/4.9.2//kubernetes-model-4.9.2.jar
@@ -167,7 +169,7 @@ metrics-jmx/4.1.1//metrics-jmx-4.1.1.jar
 metrics-json/4.1.1//metrics-json-4.1.1.jar
 metrics-jvm/4.1.1//metrics-jvm-4.1.1.jar
 minlog/1.3.0//minlog-1.3.0.jar
-mysql-connector-java/5.1.38//mysql-connector-java-5.1.38.jar
+mysql-connector-java/8.0.27//mysql-connector-java-8.0.27.jar
 netty-all/4.1.47.Final//netty-all-4.1.47.Final.jar
 objenesis/2.5.1//objenesis-2.5.1.jar
 okhttp/3.12.6//okhttp-3.12.6.jar
@@ -180,16 +182,17 @@ org.abego.treelayout.core/1.0.3//org.abego.treelayout.core-1.0.3.jar
 oro/2.0.8//oro-2.0.8.jar
 osgi-resource-locator/1.0.3//osgi-resource-locator-1.0.3.jar
 paranamer/2.8//paranamer-2.8.jar
-parquet-column/1.11.1-ebay-carmel1//parquet-column-1.11.1-ebay-carmel1.jar
-parquet-common/1.11.1-ebay-carmel1//parquet-common-1.11.1-ebay-carmel1.jar
-parquet-encoding/1.11.1-ebay-carmel1//parquet-encoding-1.11.1-ebay-carmel1.jar
-parquet-format-structures/1.11.1-ebay-carmel1//parquet-format-structures-1.11.1-ebay-carmel1.jar
+parquet-column/1.12.2.0.1.0//parquet-column-1.12.2.0.1.0.jar
+parquet-common/1.12.2.0.1.0//parquet-common-1.12.2.0.1.0.jar
+parquet-encoding/1.12.2.0.1.0//parquet-encoding-1.12.2.0.1.0.jar
+parquet-format-structures/1.12.2.0.1.0//parquet-format-structures-1.12.2.0.1.0.jar
 parquet-hadoop-bundle/1.6.0//parquet-hadoop-bundle-1.6.0.jar
-parquet-hadoop/1.11.1-ebay-carmel1//parquet-hadoop-1.11.1-ebay-carmel1.jar
-parquet-jackson/1.11.1-ebay-carmel1//parquet-jackson-1.11.1-ebay-carmel1.jar
+parquet-hadoop/1.12.2.0.1.0//parquet-hadoop-1.12.2.0.1.0.jar
+parquet-jackson/1.12.2.0.1.0//parquet-jackson-1.12.2.0.1.0.jar
 protobuf-java/2.5.0//protobuf-java-2.5.0.jar
 py4j/0.10.9//py4j-0.10.9.jar
 pyrolite/4.30//pyrolite-4.30.jar
+reflectasm/1.11.3//reflectasm-1.11.3.jar
 rheos-client_JDK-1.8/2.0.2//rheos-client_JDK-1.8-2.0.2.jar
 rheos-common_JDK-1.8/2.0.2//rheos-common_JDK-1.8-2.0.2.jar
 rheos-kafka-security_JDK-1.8/2.0.2//rheos-kafka-security_JDK-1.8-2.0.2.jar

dev/deps/spark-deps-hadoop-2.7-hive-2.3

@@ -60,7 +60,7 @@ curator-recipes/2.7.1//curator-recipes-2.7.1.jar
 datanucleus-api-jdo/4.2.4//datanucleus-api-jdo-4.2.4.jar
 datanucleus-core/4.1.17//datanucleus-core-4.1.17.jar
 datanucleus-rdbms/4.1.19//datanucleus-rdbms-4.1.19.jar
-delta-core_2.12/0.8.0.carmel0.1-SNAPSHOT//delta-core_2.12-0.8.0.carmel0.1-SNAPSHOT.jar
+delta-core_2.12/0.9.0.carmel0.2-SNAPSHOT//delta-core_2.12-0.9.0.carmel0.2-SNAPSHOT.jar
 derby/10.12.1.1//derby-10.12.1.1.jar
 dropwizard-metrics-hadoop-metrics2-reporter/0.1.2//dropwizard-metrics-hadoop-metrics2-reporter-0.1.2.jar
 flatbuffers-java/1.9.0//flatbuffers-java-1.9.0.jar
@@ -164,7 +164,9 @@ jta/1.1//jta-1.1.jar
 jul-to-slf4j/1.7.30//jul-to-slf4j-1.7.30.jar
 kafka-clients/1.1.1//kafka-clients-1.1.1.jar
 kafka-schema-registry-client/3.0.0//kafka-schema-registry-client-3.0.0.jar
+kryo-serializers/0.43//kryo-serializers-0.43.jar
 kryo-shaded/4.0.2//kryo-shaded-4.0.2.jar
+kryo/4.0.2//kryo-4.0.2.jar
 kubernetes-client/4.9.2//kubernetes-client-4.9.2.jar
 kubernetes-model-common/4.9.2//kubernetes-model-common-4.9.2.jar
 kubernetes-model/4.9.2//kubernetes-model-4.9.2.jar
@@ -183,7 +185,7 @@ metrics-jmx/4.1.1//metrics-jmx-4.1.1.jar
 metrics-json/4.1.1//metrics-json-4.1.1.jar
 metrics-jvm/4.1.1//metrics-jvm-4.1.1.jar
 minlog/1.3.0//minlog-1.3.0.jar
-mysql-connector-java/5.1.38//mysql-connector-java-5.1.38.jar
+mysql-connector-java/8.0.27//mysql-connector-java-8.0.27.jar
 netty-all/4.1.47.Final//netty-all-4.1.47.Final.jar
 objenesis/2.5.1//objenesis-2.5.1.jar
 okhttp/3.12.6//okhttp-3.12.6.jar
@@ -196,15 +198,16 @@ org.abego.treelayout.core/1.0.3//org.abego.treelayout.core-1.0.3.jar
 oro/2.0.8//oro-2.0.8.jar
 osgi-resource-locator/1.0.3//osgi-resource-locator-1.0.3.jar
 paranamer/2.8//paranamer-2.8.jar
-parquet-column/1.11.1-ebay-carmel1//parquet-column-1.11.1-ebay-carmel1.jar
-parquet-common/1.11.1-ebay-carmel1//parquet-common-1.11.1-ebay-carmel1.jar
-parquet-encoding/1.11.1-ebay-carmel1//parquet-encoding-1.11.1-ebay-carmel1.jar
-parquet-format-structures/1.11.1-ebay-carmel1//parquet-format-structures-1.11.1-ebay-carmel1.jar
-parquet-hadoop/1.11.1-ebay-carmel1//parquet-hadoop-1.11.1-ebay-carmel1.jar
-parquet-jackson/1.11.1-ebay-carmel1//parquet-jackson-1.11.1-ebay-carmel1.jar
+parquet-column/1.12.2.0.1.0//parquet-column-1.12.2.0.1.0.jar
+parquet-common/1.12.2.0.1.0//parquet-common-1.12.2.0.1.0.jar
+parquet-encoding/1.12.2.0.1.0//parquet-encoding-1.12.2.0.1.0.jar
+parquet-format-structures/1.12.2.0.1.0//parquet-format-structures-1.12.2.0.1.0.jar
+parquet-hadoop/1.12.2.0.1.0//parquet-hadoop-1.12.2.0.1.0.jar
+parquet-jackson/1.12.2.0.1.0//parquet-jackson-1.12.2.0.1.0.jar
 protobuf-java/2.5.0//protobuf-java-2.5.0.jar
 py4j/0.10.9//py4j-0.10.9.jar
 pyrolite/4.30//pyrolite-4.30.jar
+reflectasm/1.11.3//reflectasm-1.11.3.jar
 rheos-client_JDK-1.8/2.0.2//rheos-client_JDK-1.8-2.0.2.jar
 rheos-common_JDK-1.8/2.0.2//rheos-common_JDK-1.8-2.0.2.jar
 rheos-kafka-security_JDK-1.8/2.0.2//rheos-kafka-security_JDK-1.8-2.0.2.jar

dev/deps/spark-deps-hadoop-3.2-hive-2.3

@@ -56,7 +56,7 @@ curator-recipes/2.13.0//curator-recipes-2.13.0.jar
 datanucleus-api-jdo/4.2.4//datanucleus-api-jdo-4.2.4.jar
 datanucleus-core/4.1.17//datanucleus-core-4.1.17.jar
 datanucleus-rdbms/4.1.19//datanucleus-rdbms-4.1.19.jar
-delta-core_2.12/0.8.0.carmel0.1-SNAPSHOT//delta-core_2.12-0.8.0.carmel0.1-SNAPSHOT.jar
+delta-core_2.12/0.9.0.carmel0.2-SNAPSHOT//delta-core_2.12-0.9.0.carmel0.2-SNAPSHOT.jar
 derby/10.12.1.1//derby-10.12.1.1.jar
 dnsjava/2.1.7//dnsjava-2.1.7.jar
 dropwizard-metrics-hadoop-metrics2-reporter/0.1.2//dropwizard-metrics-hadoop-metrics2-reporter-0.1.2.jar
@@ -174,7 +174,9 @@ kerby-config/1.0.1//kerby-config-1.0.1.jar
 kerby-pkix/1.0.1//kerby-pkix-1.0.1.jar
 kerby-util/1.0.1//kerby-util-1.0.1.jar
 kerby-xdr/1.0.1//kerby-xdr-1.0.1.jar
+kryo-serializers/0.43//kryo-serializers-0.43.jar
 kryo-shaded/4.0.2//kryo-shaded-4.0.2.jar
+kryo/4.0.2//kryo-4.0.2.jar
 kubernetes-client/4.9.2//kubernetes-client-4.9.2.jar
 kubernetes-model-common/4.9.2//kubernetes-model-common-4.9.2.jar
 kubernetes-model/4.9.2//kubernetes-model-4.9.2.jar
@@ -193,7 +195,7 @@ metrics-jmx/4.1.1//metrics-jmx-4.1.1.jar
 metrics-json/4.1.1//metrics-json-4.1.1.jar
 metrics-jvm/4.1.1//metrics-jvm-4.1.1.jar
 minlog/1.3.0//minlog-1.3.0.jar
-mysql-connector-java/5.1.38//mysql-connector-java-5.1.38.jar
+mysql-connector-java/8.0.27//mysql-connector-java-8.0.27.jar
 netty-all/4.1.47.Final//netty-all-4.1.47.Final.jar
 nimbus-jose-jwt/4.41.1//nimbus-jose-jwt-4.41.1.jar
 objenesis/2.5.1//objenesis-2.5.1.jar
@@ -208,16 +210,17 @@ org.abego.treelayout.core/1.0.3//org.abego.treelayout.core-1.0.3.jar
 oro/2.0.8//oro-2.0.8.jar
 osgi-resource-locator/1.0.3//osgi-resource-locator-1.0.3.jar
 paranamer/2.8//paranamer-2.8.jar
-parquet-column/1.11.1-ebay-carmel1//parquet-column-1.11.1-ebay-carmel1.jar
-parquet-common/1.11.1-ebay-carmel1//parquet-common-1.11.1-ebay-carmel1.jar
-parquet-encoding/1.11.1-ebay-carmel1//parquet-encoding-1.11.1-ebay-carmel1.jar
-parquet-format-structures/1.11.1-ebay-carmel1//parquet-format-structures-1.11.1-ebay-carmel1.jar
-parquet-hadoop/1.11.1-ebay-carmel1//parquet-hadoop-1.11.1-ebay-carmel1.jar
-parquet-jackson/1.11.1-ebay-carmel1//parquet-jackson-1.11.1-ebay-carmel1.jar
+parquet-column/1.12.2.0.1.0//parquet-column-1.12.2.0.1.0.jar
+parquet-common/1.12.2.0.1.0//parquet-common-1.12.2.0.1.0.jar
+parquet-encoding/1.12.2.0.1.0//parquet-encoding-1.12.2.0.1.0.jar
+parquet-format-structures/1.12.2.0.1.0//parquet-format-structures-1.12.2.0.1.0.jar
+parquet-hadoop/1.12.2.0.1.0//parquet-hadoop-1.12.2.0.1.0.jar
+parquet-jackson/1.12.2.0.1.0//parquet-jackson-1.12.2.0.1.0.jar
 protobuf-java/2.5.0//protobuf-java-2.5.0.jar
 py4j/0.10.9//py4j-0.10.9.jar
 pyrolite/4.30//pyrolite-4.30.jar
 re2j/1.1//re2j-1.1.jar
+reflectasm/1.11.3//reflectasm-1.11.3.jar
 rheos-client_JDK-1.8/2.0.2//rheos-client_JDK-1.8-2.0.2.jar
 rheos-common_JDK-1.8/2.0.2//rheos-common_JDK-1.8-2.0.2.jar
 rheos-kafka-security_JDK-1.8/2.0.2//rheos-kafka-security_JDK-1.8-2.0.2.jar

pom.xml

@@ -149,7 +149,7 @@
     <!-- note that this should be compatible with Kafka brokers version 0.10 and up -->
     <kafka.version>2.4.1</kafka.version>
     <derby.version>10.12.1.1</derby.version>
-    <parquet.version>1.11.1-ebay-carmel1</parquet.version>
+    <parquet.version>1.12.2.0.1.0</parquet.version>
     <orc.version>1.5.10</orc.version>
     <orc.classifier></orc.classifier>
     <hive.parquet.group>com.twitter</hive.parquet.group>
@@ -1987,7 +1987,7 @@
             <groupId>${hive.group}</groupId>
             <artifactId>hive-service-rpc</artifactId>
           </exclusion>
-          <!-- parquet-hadoop-bundle:1.8.1 conflict with 1.10.1 -->
+          <!-- parquet-hadoop-bundle:1.8.1 conflict with 1.12.0 -->
           <exclusion>
             <groupId>org.apache.parquet</groupId>
             <artifactId>parquet-hadoop-bundle</artifactId>

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -891,6 +891,12 @@ object SQLConf {
     .stringConf
     .createWithDefault("org.apache.parquet.hadoop.ParquetOutputCommitter")
 
+  val PARQUET_ENCRYPTION_KEY_FILE = buildConf("spark.sql.parquet.encryption.key.file")
+    .doc("The key file for parquet encryption.")
+    .internal()
+    .stringConf
+    .createOptional
+
   val PARQUET_VECTORIZED_READER_ENABLED =
     buildConf("spark.sql.parquet.enableVectorizedReader")
       .doc("Enables vectorized parquet decoding.")

sql/core/pom.xml

@@ -168,6 +168,11 @@
       <artifactId>selenium-htmlunit-driver</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.codehaus.jackson</groupId>
+      <artifactId>jackson-mapper-asl</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
   <build>
     <outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetEncryptionFileBasedKMS.scala

@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.io.File
+import java.nio.charset.StandardCharsets
+import java.util.{Base64, UUID}
+import java.util.concurrent.ConcurrentHashMap
+
+import scala.io.Source
+import scala.util.control.NonFatal
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.crypto.{EncryptionPropertiesFactory, ParquetCryptoRuntimeException}
+import org.apache.parquet.crypto.keytools.{KeyToolkit, KmsClient, PropertiesDrivenCryptoFactory}
+
+import org.apache.spark.SparkFiles
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+
+class ParquetEncryptionFileBasedKMS extends KmsClient {
+  import ParquetEncryptionFileBasedKMS.KEY_FILE_PROPERTY_NAME
+
+  var masterKeyMap: Map[String, Array[Byte]] = _
+
+  override def initialize(
+      configuration: Configuration,
+      kmsInstanceID: String,
+      kmsInstanceURL: String,
+      accessToken: String): Unit = {
+    val keyFile = configuration.get(KEY_FILE_PROPERTY_NAME)
+    if (null == keyFile || keyFile.isEmpty) {
+      throw new ParquetCryptoRuntimeException("No encryption key file")
+    }
+
+    val localizedKeyFile = {
+      try {
+        val keyFileName = new Path(keyFile).getName
+        new File(SparkFiles.get(keyFileName))
+      } catch {
+        case e: Throwable =>
+          throw new ParquetCryptoRuntimeException(
+            s"Failed to get the localized encryption key file $keyFile", e)
+      }
+    }
+    masterKeyMap = parseKeyList(localizedKeyFile)
+  }
+
+  def parseKeyList(keyFile: File): Map[String, Array[Byte]] = {
+    val lines = Source.fromFile(keyFile).getLines().filter(_.trim.nonEmpty)
+    lines.map { line =>
+      val parts = line.split(":")
+      val keyName = parts.head.trim
+      if (parts.length != 2) {
+        throw new IllegalArgumentException(s"Key '$keyName' is not formatted correctly")
+      }
+      val key = parts.last.trim
+      try {
+        val keyBytes = Base64.getDecoder.decode(key)
+        keyName -> keyBytes
+      } catch {
+        case e: Throwable =>
+          throw new IllegalArgumentException(s"Could not decode key '$keyName'!", e)
+      }
+    }.toMap
+  }
+
+  override def wrapKey(keyBytes: Array[Byte], masterKeyIdentifier: String): String = {
+    val masterKey = getMasterKey(masterKeyIdentifier)
+    val AAD = masterKeyIdentifier.getBytes(StandardCharsets.UTF_8)
+    KeyToolkit.encryptKeyLocally(keyBytes, masterKey, AAD)
+  }
+
+  override def unwrapKey(wrappedKey: String, masterKeyIdentifier: String): Array[Byte] = {
+    val masterKey = getMasterKey(masterKeyIdentifier)
+    val AAD = masterKeyIdentifier.getBytes(StandardCharsets.UTF_8)
+    KeyToolkit.decryptKeyLocally(wrappedKey, masterKey, AAD)
+  }
+
+  private def getMasterKey(masterKeyIdentifier: String): Array[Byte] = {
+    masterKeyMap.get(masterKeyIdentifier).getOrElse(
+      throw new ParquetCryptoRuntimeException(s"Key not found:$masterKeyIdentifier"))
+  }
+}
+
+object ParquetEncryptionFileBasedKMS extends Logging {
+  val KEY_FILE_PROPERTY_NAME = "parquet.encryption.key.file"
+
+  // assume that, if the key file path is the same, its content is not change
+  private[parquet] val keyFileKmsInstanceIdMap = new ConcurrentHashMap[String, String]()
+
+  def getEncryptionConf(sparkSession: SparkSession, keyFile: String): Map[String, String] = {
+    try {
+      val kmsInstanceId = keyFileKmsInstanceIdMap.computeIfAbsent(
+        keyFile,
+        keyFile => {
+          sparkSession.sparkContext.addFile(keyFile)
+          UUID.randomUUID().toString
+        }
+      )
+      Map(KEY_FILE_PROPERTY_NAME -> keyFile,
+        KeyToolkit.KMS_CLIENT_CLASS_PROPERTY_NAME ->
+          classOf[ParquetEncryptionFileBasedKMS].getName,
+        EncryptionPropertiesFactory.CRYPTO_FACTORY_CLASS_PROPERTY_NAME ->
+          classOf[PropertiesDrivenCryptoFactory].getName,
+        KeyToolkit.KMS_INSTANCE_ID_PROPERTY_NAME -> kmsInstanceId)
+    } catch {
+      case NonFatal(e) =>
+        logError("Failed when getting parquet encryption conf", e)
+        Map.empty[String, String]
+    }
+  }
+}