[HUDI-2370] Supports data encryption

hudi-cli/pom.xml

@@ -216,6 +216,11 @@
       </exclusions>
     </dependency>
 
+    <dependency>
+      <groupId>org.apache.parquet</groupId>
+      <artifactId>parquet-column</artifactId>
+    </dependency>
+
     <!-- Spark -->
     <dependency>
       <groupId>org.apache.spark</groupId>
@@ -224,6 +229,16 @@
     <dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-sql_${scala.binary.version}</artifactId>
+        <exclusions>
+            <exclusion>
+                <artifactId>parquet-column</artifactId>
+                <groupId>org.apache.parquet</groupId>
+            </exclusion>
+            <exclusion>
+                <artifactId>parquet-hadoop</artifactId>
+                <groupId>org.apache.parquet</groupId>
+            </exclusion>
+        </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.spark</groupId>

hudi-client/hudi-client-common/pom.xml

@@ -57,6 +57,15 @@
     <dependency>
       <groupId>org.apache.parquet</groupId>
       <artifactId>parquet-avro</artifactId>
+      <version>${parquet.version}</version>
+      <scope>compile</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.parquet</groupId>
+      <artifactId>parquet-column</artifactId>
+      <version>${parquet.version}</version>
+      <scope>compile</scope>
     </dependency>
 
     <!-- Dropwizard Metrics -->

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieFileWriterFactory.java

@@ -31,7 +31,10 @@
 import org.apache.avro.Schema;
 import org.apache.avro.generic.IndexedRecord;
 import org.apache.hadoop.fs.Path;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
 import org.apache.parquet.avro.AvroSchemaConverter;
+import org.apache.parquet.schema.Types;
 
 import java.io.IOException;
 
@@ -41,6 +44,8 @@
 
 public class HoodieFileWriterFactory {
 
+  private static final Logger LOG = LogManager.getLogger(HoodieFileWriterFactory.class);
+
   public static <T extends HoodieRecordPayload, R extends IndexedRecord, I, K, O> HoodieFileWriter<R> getFileWriter(
       String instantTime, Path path, HoodieTable<T, I, K, O> hoodieTable, HoodieWriteConfig config, Schema schema,
       TaskContextSupplier taskContextSupplier) throws IOException {
@@ -67,6 +72,10 @@ private static <T extends HoodieRecordPayload, R extends IndexedRecord> HoodieFi
       String instantTime, Path path, HoodieWriteConfig config, Schema schema, HoodieTable hoodieTable,
       TaskContextSupplier taskContextSupplier, boolean populateMetaFields, boolean enableBloomFilter) throws IOException {
     Option<BloomFilter> filter = enableBloomFilter ? Option.of(createBloomFilter(config)) : Option.empty();
+
+    LOG.error("--123" + SystemUtils.getClassLocation(AvroSchemaConverter.class));
+    LOG.error("--123" + SystemUtils.getClassLocation(Types.class));
+
     HoodieAvroWriteSupport writeSupport = new HoodieAvroWriteSupport(new AvroSchemaConverter(hoodieTable.getHadoopConf()).convert(schema), schema, filter);
 
     HoodieAvroParquetConfig parquetConfig = new HoodieAvroParquetConfig(writeSupport, config.getParquetCompressionCodec(),

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/SystemUtils.java

@@ -0,0 +1,70 @@
+/*
+ * 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.hudi.io.storage;
+
+import java.io.File;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.security.CodeSource;
+import java.security.ProtectionDomain;
+
+/**
+ * system util.
+ */
+public class SystemUtils {
+
+  public static URL getClassLocation(final Class<?> cls) {
+
+    if (cls == null) {
+      throw new IllegalArgumentException("null input: cls");
+    }
+
+    URL result = null;
+    final String clsAsResource = cls.getName().replace('.', '/').concat(".class");
+
+    final ProtectionDomain pd = cls.getProtectionDomain();
+
+    if (pd != null) {
+      final CodeSource cs = pd.getCodeSource();
+      if (cs != null) {
+        result = cs.getLocation();
+      }
+
+      if (result != null) {
+        if ("file".equals(result.getProtocol())) {
+          try {
+            if (result.toExternalForm().endsWith(".jar") || result.toExternalForm().endsWith(".zip")) {
+              result = new URL("jar:".concat(result.toExternalForm()).concat("!/").concat(clsAsResource));
+            } else if (new File(result.getFile()).isDirectory()) {
+              result = new URL(result, clsAsResource);
+            }
+          } catch (MalformedURLException ignore) {
+            System.out.println("-W");
+          }
+        }
+      }
+    }
+
+    if (result == null) {
+      final ClassLoader clsLoader = cls.getClassLoader();
+      result = clsLoader != null ? clsLoader.getResource(clsAsResource) : ClassLoader.getSystemResource(clsAsResource);
+    }
+    return result;
+  }
+}

hudi-client/hudi-java-client/pom.xml

@@ -43,6 +43,11 @@
             <artifactId>parquet-avro</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.parquet</groupId>
+            <artifactId>parquet-column</artifactId>
+        </dependency>
+
         <!-- Hoodie - Test -->
         <dependency>
             <groupId>org.apache.hudi</groupId>

hudi-client/hudi-spark-client/pom.xml

@@ -52,6 +52,16 @@
     <dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-sql_${scala.binary.version}</artifactId>
+      <exclusions>
+        <exclusion>
+          <artifactId>parquet-column</artifactId>
+          <groupId>org.apache.parquet</groupId>
+        </exclusion>
+        <exclusion>
+          <artifactId>parquet-hadoop</artifactId>
+          <groupId>org.apache.parquet</groupId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.spark</groupId>
@@ -65,6 +75,11 @@
       <artifactId>parquet-avro</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>org.apache.parquet</groupId>
+      <artifactId>parquet-column</artifactId>
+    </dependency>
+
     <!-- Hoodie - Test -->
     <dependency>
       <groupId>org.apache.hudi</groupId>

hudi-common/pom.xml

@@ -119,6 +119,11 @@
       <artifactId>parquet-avro</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>org.apache.parquet</groupId>
+      <artifactId>parquet-column</artifactId>
+    </dependency>
+
     <!-- ORC -->
     <dependency>
       <groupId>org.apache.orc</groupId>

hudi-common/src/main/java/org/apache/hudi/crypot/kms/InMemoryKMS.java

@@ -0,0 +1,104 @@
+/*
+ * 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.hudi.crypot.kms;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hudi.common.config.ConfigProperty;
+import org.apache.parquet.crypto.KeyAccessDeniedException;
+import org.apache.parquet.crypto.ParquetCryptoRuntimeException;
+import org.apache.parquet.crypto.keytools.KeyToolkit;
+import org.apache.parquet.crypto.keytools.KmsClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * This is a mock class, built for testing only. Don't use it as an example of KmsClient implementation.
+ * (org.apache.parquet.crypto.keytools.samples.VaultClient is the sample implementation).
+ */
+public class InMemoryKMS implements KmsClient {
+  private static final Logger LOG = LoggerFactory.getLogger(InMemoryKMS.class);
+
+  private static final ConfigProperty<String> KEY_LIST_PROPERTY_NAME = ConfigProperty
+      .key("hoodie.parquet.encryption.key.list")
+      .noDefaultValue()
+      .withDocumentation("hudi parquet encryption key list, example: keyName_1:value_1, keyName_2:value_2");
+
+  private static Map<String,byte[]> masterKeyMap;
+
+  @Override
+  public synchronized void initialize(Configuration configuration, String kmsInstanceID, String kmsInstanceURL, String accessToken) {
+    // Parse master  keys
+    String[] masterKeys = configuration.getTrimmedStrings(KEY_LIST_PROPERTY_NAME.key());
+    if (null == masterKeys || masterKeys.length == 0) {
+      throw new ParquetCryptoRuntimeException("No encryption key list");
+    }
+    masterKeyMap = parseKeyList(masterKeys);
+  }
+
+  private static Map<String, byte[]> parseKeyList(String[] masterKeys) {
+
+    Map<String,byte[]> keyMap = new HashMap<>();
+
+    int nKeys = masterKeys.length;
+    for (int i = 0; i < nKeys; i++) {
+      String[] parts = masterKeys[i].split(":");
+      String keyName = parts[0].trim();
+      if (parts.length != 2) {
+        throw new IllegalArgumentException("Key '" + keyName + "' is not formatted correctly");
+      }
+      String key = parts[1].trim();
+      try {
+        byte[] keyBytes = Base64.getDecoder().decode(key);
+        keyMap.put(keyName, keyBytes);
+      } catch (IllegalArgumentException e) {
+        LOG.warn("Could not decode key '" + keyName + "'!");
+        throw e;
+      }
+    }
+    return keyMap;
+  }
+
+  @Override
+  public synchronized String wrapKey(byte[] keyBytes, String masterKeyIdentifier)
+      throws KeyAccessDeniedException, UnsupportedOperationException {
+
+    byte[] masterKey = masterKeyMap.get(masterKeyIdentifier);
+    if (null == masterKey) {
+      throw new ParquetCryptoRuntimeException("Key not found: " + masterKeyIdentifier);
+    }
+    byte[] add = masterKeyIdentifier.getBytes(StandardCharsets.UTF_8);
+    return KeyToolkit.encryptKeyLocally(keyBytes, masterKey, add);
+  }
+
+  @Override
+  public synchronized byte[] unwrapKey(String wrappedKey, String masterKeyIdentifier)
+      throws KeyAccessDeniedException, UnsupportedOperationException {
+    byte[] masterKey = masterKeyMap.get(masterKeyIdentifier);
+    if (null == masterKey) {
+      throw new ParquetCryptoRuntimeException("Key not found: " + masterKeyIdentifier);
+    }
+    byte[] add = masterKeyIdentifier.getBytes(StandardCharsets.UTF_8);
+    return KeyToolkit.decryptKeyLocally(wrappedKey, masterKey, add);
+  }
+}

hudi-examples/pom.xml

@@ -177,6 +177,16 @@
     <dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-sql_${scala.binary.version}</artifactId>
+        <exclusions>
+            <exclusion>
+                <artifactId>parquet-column</artifactId>
+                <groupId>org.apache.parquet</groupId>
+            </exclusion>
+            <exclusion>
+                <artifactId>parquet-hadoop</artifactId>
+                <groupId>org.apache.parquet</groupId>
+            </exclusion>
+        </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.spark</groupId>
@@ -190,6 +200,12 @@
       <version>${parquet.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>org.apache.parquet</groupId>
+      <artifactId>parquet-column</artifactId>
+      <version>${parquet.version}</version>
+    </dependency>
+
     <!-- Avro -->
     <dependency>
       <groupId>org.apache.avro</groupId>

hudi-flink/pom.xml

@@ -30,7 +30,7 @@
 
   <properties>
     <main.basedir>${project.parent.basedir}</main.basedir>
-    <parquet.version>1.11.1</parquet.version>
+    <parquet.version>1.12.0</parquet.version>
   </properties>
 
   <build>
@@ -155,6 +155,12 @@
       <artifactId>flink-parquet_${scala.binary.version}</artifactId>
       <version>${flink.version}</version>
       <scope>provided</scope>
+      <exclusions>
+        <exclusion>
+          <artifactId>parquet-hadoop</artifactId>
+          <groupId>org.apache.parquet</groupId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.flink</groupId>

hudi-hadoop-mr/pom.xml

@@ -49,6 +49,11 @@
       <artifactId>parquet-avro</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>org.apache.parquet</groupId>
+      <artifactId>parquet-column</artifactId>
+    </dependency>
+
     <!-- Hadoop -->
     <dependency>
       <groupId>org.apache.hadoop</groupId>