Add configurable Parquet Data Anonymization feature

presto-delta/src/main/java/com/facebook/presto/delta/DeltaPageSourceProvider.java

@@ -53,6 +53,8 @@
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.security.AccessControlException;
+import org.apache.parquet.anonymization.AnonymizationManager;
+import org.apache.parquet.anonymization.AnonymizationManagerFactory;
 import org.apache.parquet.column.ColumnDescriptor;
 import org.apache.parquet.crypto.InternalFileDecryptor;
 import org.apache.parquet.hadoop.metadata.BlockMetaData;
@@ -86,8 +88,10 @@
 import static com.facebook.presto.delta.DeltaErrorCode.DELTA_PARQUET_SCHEMA_MISMATCH;
 import static com.facebook.presto.delta.DeltaTypeUtils.convertPartitionValue;
 import static com.facebook.presto.hive.CacheQuota.NO_CACHE_CONSTRAINTS;
+import static com.facebook.presto.hive.HiveCommonSessionProperties.getParquetAnonymizationManagerClass;
 import static com.facebook.presto.hive.HiveCommonSessionProperties.getParquetMaxReadBlockSize;
 import static com.facebook.presto.hive.HiveCommonSessionProperties.getReadNullMaskedParquetEncryptedValue;
+import static com.facebook.presto.hive.HiveCommonSessionProperties.isParquetAnonymizationEnabled;
 import static com.facebook.presto.hive.HiveCommonSessionProperties.isParquetBatchReaderVerificationEnabled;
 import static com.facebook.presto.hive.HiveCommonSessionProperties.isParquetBatchReadsEnabled;
 import static com.facebook.presto.hive.parquet.HdfsParquetDataSource.buildHdfsParquetDataSource;
@@ -280,6 +284,14 @@ private static ConnectorPageSource createParquetPageSource(
                 }
             }
             MessageColumnIO messageColumnIO = getColumnIO(fileSchema, requestedSchema);
+            Optional<AnonymizationManager> anonymizationManager = Optional.empty();
+            if (isParquetAnonymizationEnabled(session)) {
+                anonymizationManager = AnonymizationManagerFactory
+                        .getAnonymizationManager(
+                                getParquetAnonymizationManagerClass(session),
+                                configuration,
+                                tableName.toString());
+            }
             ParquetReader parquetReader = new ParquetReader(
                     messageColumnIO,
                     blocks.build(),
@@ -292,7 +304,8 @@ private static ConnectorPageSource createParquetPageSource(
                     parquetPredicate,
                     blockIndexStores,
                     false,
-                    fileDecryptor);
+                    fileDecryptor,
+                    anonymizationManager);
 
             ImmutableList.Builder<String> namesBuilder = ImmutableList.builder();
             ImmutableList.Builder<Type> typesBuilder = ImmutableList.builder();

presto-hive-common/src/main/java/com/facebook/presto/hive/HiveCommonClientConfig.java

@@ -46,6 +46,8 @@ public class HiveCommonClientConfig
     private boolean readNullMaskedParquetEncryptedValueEnabled;
     private boolean useParquetColumnNames;
     private boolean zstdJniDecompressionEnabled;
+    private boolean isParquetAnonymizationEnabled;
+    private String parquetAnonymizationManagerClass;
 
     public NodeSelectionStrategy getNodeSelectionStrategy()
     {
@@ -284,4 +286,30 @@ public HiveCommonClientConfig setZstdJniDecompressionEnabled(boolean zstdJniDeco
         this.zstdJniDecompressionEnabled = zstdJniDecompressionEnabled;
         return this;
     }
+
+    @Config("hive.enable-parquet-anonymization")
+    @ConfigDescription("enable parquet anonymization")
+    public HiveCommonClientConfig setParquetAnonymizationEnabled(boolean isParquetAnonymizationEnabled)
+    {
+        this.isParquetAnonymizationEnabled = isParquetAnonymizationEnabled;
+        return this;
+    }
+
+    public boolean isParquetAnonymizationEnabled()
+    {
+        return this.isParquetAnonymizationEnabled;
+    }
+
+    @Config("hive.parquet-anonymization-manager-class")
+    @ConfigDescription("Anonymization manager class, must be set if parquet anonymization is enabled")
+    public HiveCommonClientConfig setParquetAnonymizationManagerClass(String parquetAnonymizationManagerClass)
+    {
+        this.parquetAnonymizationManagerClass = parquetAnonymizationManagerClass;
+        return this;
+    }
+
+    public String getParquetAnonymizationManagerClass()
+    {
+        return this.parquetAnonymizationManagerClass;
+    }
 }

presto-hive-common/src/main/java/com/facebook/presto/hive/HiveCommonSessionProperties.java

@@ -61,6 +61,8 @@ public class HiveCommonSessionProperties
     private static final String PARQUET_MAX_READ_BLOCK_SIZE = "parquet_max_read_block_size";
     private static final String PARQUET_USE_COLUMN_NAMES = "parquet_use_column_names";
     public static final String READ_MASKED_VALUE_ENABLED = "read_null_masked_parquet_encrypted_value_enabled";
+    public static final String ENABLE_PARQUET_ANONYMIZATION = "enable_parquet_anonymization";
+    public static final String PARQUET_ANONYMIZATION_MANAGER_CLASS = "parquet_anonymization_manager_class";
     private final List<PropertyMetadata<?>> sessionProperties;
 
     @Inject
@@ -177,6 +179,16 @@ public HiveCommonSessionProperties(HiveCommonClientConfig hiveCommonClientConfig
                         READ_MASKED_VALUE_ENABLED,
                         "Return null when access is denied for an encrypted parquet column",
                         hiveCommonClientConfig.getReadNullMaskedParquetEncryptedValue(),
+                        false),
+                booleanProperty(
+                        ENABLE_PARQUET_ANONYMIZATION,
+                        "Is parquet anonymization enabled",
+                        hiveCommonClientConfig.isParquetAnonymizationEnabled(),
+                        false),
+                stringProperty(
+                        PARQUET_ANONYMIZATION_MANAGER_CLASS,
+                        "Parquet anonymization manager class",
+                        hiveCommonClientConfig.getParquetAnonymizationManagerClass(),
                         false));
     }
 
@@ -287,6 +299,16 @@ public static boolean getReadNullMaskedParquetEncryptedValue(ConnectorSession se
         return session.getProperty(READ_MASKED_VALUE_ENABLED, Boolean.class);
     }
 
+    public static boolean isParquetAnonymizationEnabled(ConnectorSession session)
+    {
+        return session.getProperty(ENABLE_PARQUET_ANONYMIZATION, Boolean.class);
+    }
+
+    public static String getParquetAnonymizationManagerClass(ConnectorSession session)
+    {
+        return session.getProperty(PARQUET_ANONYMIZATION_MANAGER_CLASS, String.class);
+    }
+
     public static PropertyMetadata<DataSize> dataSizeSessionProperty(String name, String description, DataSize defaultValue, boolean hidden)
     {
         return new PropertyMetadata<>(

presto-hive-common/src/test/java/com/facebook/presto/hive/TestHiveCommonClientConfig.java

@@ -47,7 +47,9 @@ public void testDefaults()
                 .setZstdJniDecompressionEnabled(false)
                 .setParquetBatchReaderVerificationEnabled(false)
                 .setParquetBatchReadOptimizationEnabled(false)
-                .setReadNullMaskedParquetEncryptedValue(false));
+                .setReadNullMaskedParquetEncryptedValue(false)
+                .setParquetAnonymizationEnabled(false)
+                .setParquetAnonymizationManagerClass(null));
     }
 
     @Test
@@ -72,6 +74,9 @@ public void testExplicitPropertyMappings()
                 .put("hive.enable-parquet-batch-reader-verification", "true")
                 .put("hive.parquet-batch-read-optimization-enabled", "true")
                 .put("hive.read-null-masked-parquet-encrypted-value-enabled", "true")
+                .put("hive.enable-parquet-anonymization", "true")
+                .put("hive.parquet-anonymization-manager-class",
+                        "org.apache.parquet.anonymization.TestAnonymizationManager")
                 .build();
 
         HiveCommonClientConfig expected = new HiveCommonClientConfig()
@@ -92,7 +97,9 @@ public void testExplicitPropertyMappings()
                 .setZstdJniDecompressionEnabled(true)
                 .setParquetBatchReaderVerificationEnabled(true)
                 .setParquetBatchReadOptimizationEnabled(true)
-                .setReadNullMaskedParquetEncryptedValue(true);
+                .setReadNullMaskedParquetEncryptedValue(true)
+                .setParquetAnonymizationEnabled(true)
+                .setParquetAnonymizationManagerClass("org.apache.parquet.anonymization.TestAnonymizationManager");
 
         ConfigAssertions.assertFullMapping(properties, expected);
     }

presto-hive/src/main/java/com/facebook/presto/hive/parquet/ParquetPageSourceFactory.java

@@ -48,6 +48,8 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.Path;
+import org.apache.parquet.anonymization.AnonymizationManager;
+import org.apache.parquet.anonymization.AnonymizationManagerFactory;
 import org.apache.parquet.column.ColumnDescriptor;
 import org.apache.parquet.crypto.DecryptionPropertiesFactory;
 import org.apache.parquet.crypto.FileDecryptionProperties;
@@ -96,8 +98,10 @@
 import static com.facebook.presto.hive.BaseHiveColumnHandle.ColumnType.SYNTHESIZED;
 import static com.facebook.presto.hive.HiveColumnHandle.getPushedDownSubfield;
 import static com.facebook.presto.hive.HiveColumnHandle.isPushedDownSubfield;
+import static com.facebook.presto.hive.HiveCommonSessionProperties.getParquetAnonymizationManagerClass;
 import static com.facebook.presto.hive.HiveCommonSessionProperties.getParquetMaxReadBlockSize;
 import static com.facebook.presto.hive.HiveCommonSessionProperties.getReadNullMaskedParquetEncryptedValue;
+import static com.facebook.presto.hive.HiveCommonSessionProperties.isParquetAnonymizationEnabled;
 import static com.facebook.presto.hive.HiveCommonSessionProperties.isParquetBatchReaderVerificationEnabled;
 import static com.facebook.presto.hive.HiveCommonSessionProperties.isParquetBatchReadsEnabled;
 import static com.facebook.presto.hive.HiveCommonSessionProperties.isUseParquetColumnNames;
@@ -254,6 +258,14 @@ public static ConnectorPageSource createParquetPageSource(
                 nextStart += block.getRowCount();
             }
             MessageColumnIO messageColumnIO = getColumnIO(fileSchema, requestedSchema);
+            Optional<AnonymizationManager> anonymizationManager = Optional.empty();
+            if (isParquetAnonymizationEnabled(session)) {
+                anonymizationManager = AnonymizationManagerFactory
+                        .getAnonymizationManager(
+                                getParquetAnonymizationManagerClass(session),
+                                configuration,
+                                tableName.toString());
+            }
             ParquetReader parquetReader = new ParquetReader(
                     messageColumnIO,
                     blocks.build(),
@@ -266,7 +278,8 @@ public static ConnectorPageSource createParquetPageSource(
                     parquetPredicate,
                     blockIndexStores,
                     columnIndexFilterEnabled,
-                    fileDecryptor);
+                    fileDecryptor,
+                    anonymizationManager);
 
             ImmutableList.Builder<String> namesBuilder = ImmutableList.builder();
             ImmutableList.Builder<Type> typesBuilder = ImmutableList.builder();

presto-hive/src/test/java/com/facebook/presto/hive/parquet/BenchmarkParquetPageSource.java

@@ -291,7 +291,7 @@ ParquetPageSource createParquetPageSource()
                 fields.add(ColumnIOConverter.constructField(getTypeFromTypeSignature(), messageColumnIO.getChild(i)));
             }
 
-            ParquetReader parquetReader = new ParquetReader(messageColumnIO, parquetMetadata.getBlocks(), Optional.empty(), dataSource, newSimpleAggregatedMemoryContext(), new DataSize(16, MEGABYTE), batchReadEnabled, enableVerification, null, null, false, Optional.empty());
+            ParquetReader parquetReader = new ParquetReader(messageColumnIO, parquetMetadata.getBlocks(), Optional.empty(), dataSource, newSimpleAggregatedMemoryContext(), new DataSize(16, MEGABYTE), batchReadEnabled, enableVerification, null, null, false, Optional.empty(), Optional.empty());
             return new ParquetPageSource(parquetReader, Collections.nCopies(channelCount, type), fields, columnNames, new RuntimeStats());
         }
 

presto-hive/src/test/java/com/facebook/presto/hive/parquet/ParquetTester.java

@@ -530,7 +530,7 @@ private static void assertPageSource(List<Type> types, Iterator<?>[] valuesByFie
         assertPageSource(types, valuesByField, pageSource, Optional.empty());
     }
 
-    private static void assertPageSource(List<Type> types, Iterator<?>[] valuesByField, ConnectorPageSource pageSource, Optional<Long> maxReadBlockSize)
+    static void assertPageSource(List<Type> types, Iterator<?>[] valuesByField, ConnectorPageSource pageSource, Optional<Long> maxReadBlockSize)
     {
         Page page;
         while ((page = pageSource.getNextPage()) != null) {
@@ -550,7 +550,7 @@ private static void assertPageSource(List<Type> types, Iterator<?>[] valuesByFie
         }
     }
 
-    private static void assertRecordCursor(List<Type> types, Iterator<?>[] valuesByField, RecordCursor cursor)
+    static void assertRecordCursor(List<Type> types, Iterator<?>[] valuesByField, RecordCursor cursor)
     {
         while (cursor.advanceNextPosition()) {
             for (int field = 0; field < types.size(); field++) {