Change connection_config from a string:string map to a map from Parqu…

cpp/src/parquet/encryption/crypto_factory.h

@@ -129,13 +129,14 @@ struct PARQUET_EXPORT ExternalEncryptionConfiguration : public EncryptionConfigu
   /// Format: "{\"user_id\": \"abc123\", \"location\": {\"lat\": 9.7489, \"lon\": -83.7534}}"
   std::string app_context;
 
-  /// Key/value map of the location of configuration files needed by the external
-  /// encryptors. This may include location of a dynamically-linked library, or the
-  /// location of a file where the external encryptor can find urls, certificates, and parameters
-  /// needed to make a remote call. 
+  /// Map of the encryption algorithms to the key/value map of the location of configuration files
+  /// needed by the external encryptors. This may include location of a dynamically-linked
+  /// library, or the location of a file where the external encryptor can find urls, certificates,
+  /// and parameters needed to make a remote call. 
   /// For security, these values should never be sent in this config, only the locations of 
   /// the files that the external encryptor will know how to access.
-  std::unordered_map<std::string, std::string> connection_config;
+  std::unordered_map<ParquetCipher::type, std::unordered_map<std::string, std::string>>
+      connection_config;
 };
 
 struct PARQUET_EXPORT DecryptionConfiguration {

cpp/src/parquet/encryption/encryption.cc

@@ -413,7 +413,7 @@ ExternalFileEncryptionProperties::Builder* ExternalFileEncryptionProperties::Bui
 
 ExternalFileEncryptionProperties::Builder*
 ExternalFileEncryptionProperties::Builder::connection_config(
-    const std::map<std::string, std::string>& config) {
+    const std::map<ParquetCipher::type, std::map<std::string, std::string>>& config) {
   if (connection_config_.size() != 0) {
     throw ParquetException("Connection config already set");
   }
@@ -439,7 +439,7 @@ ExternalFileEncryptionProperties::ExternalFileEncryptionProperties(
     const std::string& aad_prefix, bool store_aad_prefix_in_file,
     const ColumnPathToEncryptionPropertiesMap& encrypted_columns,
     const std::string& app_context,
-    const std::map<std::string, std::string>& connection_config)
+    const std::map<ParquetCipher::type, std::map<std::string, std::string>>& connection_config)
     : FileEncryptionProperties(cipher, footer_key, footer_key_metadata, encrypted_footer,
                                aad_prefix, store_aad_prefix_in_file, encrypted_columns),
       app_context_(app_context),

cpp/src/parquet/encryption/encryption.h

@@ -563,11 +563,12 @@ class PARQUET_EXPORT ExternalFileEncryptionProperties : public FileEncryptionPro
     /// Valid JSON string with additional application context needed for security checks. 
     Builder* app_context(const std::string& context);
 
-    /// Key/value map of the location of configuration files needed by the external
-    /// encryption service, including location of a dynamically-linked library, or config files
-    /// where the external service can find urls, certificates, and parameters needed to make a
-    /// remote service call. 
-    Builder* connection_config(const std::map<std::string, std::string>& config);
+    /// Map of the encryption algorithms to the key/value map of the location of configuration files
+    /// needed by the external encryptors, including location of a dynamically-linked library,
+    /// or config files where the external encryptors can find urls, certificates, and parameters
+    /// needed to make a remote call. 
+    Builder* connection_config(
+      const std::map<ParquetCipher::type, std::map<std::string, std::string>>& config);
 
     /// Forward all base class property methods to the base class Builder so we can return the
     /// correct Builder type.
@@ -610,27 +611,28 @@ class PARQUET_EXPORT ExternalFileEncryptionProperties : public FileEncryptionPro
 
    private:
     std::string app_context_;
-    std::map<std::string, std::string> connection_config_;
+    std::map<ParquetCipher::type, std::map<std::string, std::string>> connection_config_;
   };
 
   const std::string& app_context() const {
     return app_context_;
   }
 
-  const std::map<std::string, std::string>& connection_config() const {
+  const std::map<ParquetCipher::type, std::map<std::string, std::string>>&
+     connection_config() const {
     return connection_config_;
   }
 
  private:
   std::string app_context_;
-  std::map<std::string, std::string> connection_config_;
+  std::map<ParquetCipher::type, std::map<std::string, std::string>> connection_config_;
 
   ExternalFileEncryptionProperties(ParquetCipher::type cipher, const std::string& footer_key,
-                                   const std::string& footer_key_metadata, bool encrypted_footer,
-                                   const std::string& aad_prefix, bool store_aad_prefix_in_file,
-                                   const ColumnPathToEncryptionPropertiesMap& encrypted_columns,
-                                   const std::string& app_context,
-                                   const std::map<std::string, std::string>& connection_config);
+        const std::string& footer_key_metadata, bool encrypted_footer,
+        const std::string& aad_prefix, bool store_aad_prefix_in_file,
+        const ColumnPathToEncryptionPropertiesMap& encrypted_columns,
+        const std::string& app_context,
+        const std::map<ParquetCipher::type, std::map<std::string, std::string>>& connection_config);
 };
 
 }  // namespace parquet

cpp/src/parquet/encryption/properties_test.cc

@@ -310,9 +310,9 @@ TEST(TestExternalFileEncryptionProperties, SetExternalContextAndConfig) {
                    "    \"lon\": -84.0\n"
                    "  }\n"
                    "}";
-  std::map<std::string, std::string> connection_config;
-  connection_config["lib_location"] = "path/to/lib.so";
-  connection_config["config_file"] = "path/to/config/file";
+  std::map<ParquetCipher::type, std::map<std::string, std::string>> connection_config;
+  connection_config[ParquetCipher::AES_GCM_V1]["lib_location"] = "path/to/lib.so";
+  connection_config[ParquetCipher::AES_GCM_V1]["config_file"] = "path/to/config/file";
 
   ExternalFileEncryptionProperties::Builder builder(kFooterEncryptionKey);
   builder.app_context(app_context);

python/pyarrow/_parquet_encryption.pyx

@@ -236,10 +236,17 @@ cdef class ExternalEncryptionConfiguration(EncryptionConfiguration):
     @property
     def connection_config(self):
         """Get the connection configuration as a Python dictionary."""
+
+        cdef pair[ParquetCipher, unordered_map[c_string, c_string]] outer_pair
+        cdef pair[c_string, c_string] inner_pair
         result = {}
 
-        for pair in self.external_configuration.get().connection_config:
-            result[frombytes(pair.first)] = frombytes(pair.second)
+        for outer_pair in self.external_configuration.get().connection_config:
+            cipher_name = cipher_to_name(outer_pair.first)
+            inner_map = {}
+            for inner_pair in outer_pair.second:
+                inner_map[frombytes(inner_pair.first)] = frombytes(inner_pair.second)
+            result[cipher_name] = inner_map
 
         return result
 
@@ -249,14 +256,23 @@ cdef class ExternalEncryptionConfiguration(EncryptionConfiguration):
         if value is None:
             raise ValueError("Connection config value cannot be None")
 
-        cdef unordered_map[c_string, c_string] cpp_map
-        for k, v in value.items():
-            if not isinstance(k, str):
-                raise TypeError(f"Connection config key must be str, got {type(k).__name__}")
-            if not isinstance(v, str):
-                raise TypeError(f"Connection config value must be str, got {type(v).__name__}")
-            cpp_map[tobytes(k)] = tobytes(v)
+        cdef unordered_map[ParquetCipher, unordered_map[c_string, c_string]] cpp_map
+        cdef unordered_map[c_string, c_string] inner_cpp_map
+        cdef ParquetCipher cipher_enum
 
+        for cipher_name, inner_dict in value.items():
+            cipher_enum = cipher_from_name(cipher_name)
+            if not isinstance(inner_dict, dict):
+                raise TypeError(f"Inner value for cipher {cipher_name} must be a dict")
+            # Clear the map from the values of the previous iteration
+            inner_cpp_map.clear()
+
+            for k, v in inner_dict.items():
+                if not isinstance(k, str) or not isinstance(v, str):
+                    raise TypeError("All inner config keys/values must be str")
+                inner_cpp_map[tobytes(k)] = tobytes(v)
+            cpp_map[cipher_enum] = inner_cpp_map
+
         self.external_configuration.get().connection_config = cpp_map
 
     @property

python/pyarrow/includes/libparquet_encryption.pxd

@@ -100,7 +100,7 @@ cdef extern from "parquet/encryption/crypto_factory.h" \
         int32_t data_key_length_bits
         unordered_map[c_string, CColumnEncryptionAttributes] per_column_encryption
         c_string app_context
-        unordered_map[c_string, c_string] connection_config
+        unordered_map[ParquetCipher, unordered_map[c_string, c_string]] connection_config 
 
     cdef cppclass CDecryptionConfiguration\
             " parquet::encryption::DecryptionConfiguration":

python/pyarrow/tests/parquet/test_external_encryption.py

@@ -14,6 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+
 import pytest
 from datetime import  timedelta
 import pyarrow.parquet as pq
@@ -49,8 +50,10 @@ def external_encryption_config():
             "location": "Presidio"
         },
         connection_config={
-            "config_file": "path/to/config/file",
-            "config_file_decryption_key": "some_key"
+            "EXTERNAL_DBPA_V1": {
+                "config_file": "path/to/config/file",
+                "config_file_decryption_key": "some_key"
+            }
         }
     )
 
@@ -123,8 +126,10 @@ def test_external_encryption_configuration_properties(external_encryption_config
     }
 
     assert external_encryption_config.connection_config == {
-        "config_file": "path/to/config/file",
-        "config_file_decryption_key": "some_key"
+        "EXTERNAL_DBPA_V1": {
+            "config_file": "path/to/config/file",
+            "config_file_decryption_key": "some_key"
+        }
     }
 
     assert external_encryption_config.per_column_encryption == {
@@ -175,21 +180,25 @@ def test_external_encryption_per_column_encryption_new_algorithm():
 
 def test_external_encryption_connection_config_invalid_types():
     """Ensure connection_config rejects non-string keys or values."""
-    with pytest.raises(TypeError, match="Connection config key must be str, got int"):
+    with pytest.raises(TypeError, match="All inner config keys/values must be str"):
         config=pe.ExternalEncryptionConfiguration(
             footer_key="key"
         )
         config.connection_config={
-                "config_file": "path/to/file",
-                123: "should-fail"  # Invalid: key is not a string
+                "EXTERNAL_DBPA_V1": {
+                    "config_file": "path/to/file",
+                    123: "should-fail"  # Invalid: key is not a string
+                }
             }
 
-    with pytest.raises(TypeError, match="Connection config value must be str, got list"):
+    with pytest.raises(TypeError, match="All inner config keys/values must be str"):
         config = pe.ExternalEncryptionConfiguration(
             footer_key="key"
         )
         config.connection_config={
-                "config_file": ["not", "a", "string"]  # Invalid: value is not a string
+                "EXTERNAL_DBPA_V1": {
+                    "config_file": ["not", "a", "string"]  # Invalid: value is not a string
+                }
             }
 
 def test_external_encryption_rejects_none_values():