Skip to content

Refactor encryptor and decryptor interfaces to enable parallel decrypt in external DBPA #246

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sofia-tekdatum merged 7 commits into from
Feb 26, 2026
Merged

Refactor encryptor and decryptor interfaces to enable parallel decrypt in external DBPA #246

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
52a2b80
First attempt at refactoring classes
sofia-tekdatum Feb 12, 2026
70d955d
Fix linter errors
sofia-tekdatum Feb 12, 2026
c169f27
More Linter errors
sofia-tekdatum Feb 12, 2026
0269fbe
Iterate over interfaces change
sofia-tekdatum Feb 25, 2026
0ef0003
Fixing test and linter
sofia-tekdatum Feb 25, 2026
7b85882
Removed global encoding_properties variables
sofia-tekdatum Feb 26, 2026
58e17ba
Linter fixes
sofia-tekdatum Feb 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 1 addition & 3 deletions cpp/src/parquet/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -506,9 +506,7 @@ add_parquet_test(arrow-reader-writer-test
SOURCES
arrow/arrow_reader_writer_test.cc
arrow/arrow_statistics_test.cc
arrow/variant_test.cc
$<$<BOOL:${PARQUET_REQUIRE_ENCRYPTION}>:encryption/external/test_utils.cc>
)
arrow/variant_test.cc)

add_parquet_test(arrow-internals-test SOURCES arrow/path_internal_test.cc
arrow/reconstruct_internal_test.cc)
Expand Down
67 changes: 0 additions & 67 deletions cpp/src/parquet/arrow/arrow_reader_writer_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,10 +76,6 @@
#include "parquet/properties.h"
#include "parquet/test_util.h"

#ifdef PARQUET_REQUIRE_ENCRYPTION
# include "parquet/encryption/external/test_utils.h"
#endif

using arrow::Array;
using arrow::ArrayData;
using arrow::ArrayFromJSON;
Expand Down Expand Up @@ -1965,69 +1961,6 @@ TEST(TestArrowReadWrite, UseDeprecatedInt96) {
ASSERT_NO_FATAL_FAILURE(::arrow::AssertTablesEqual(*ex_result, *result));
}

#ifdef PARQUET_REQUIRE_ENCRYPTION
TEST(ExternalDbpaConcurrencyTest, FailsWhenUseThreadsTrue) {
std::shared_ptr<::arrow::Array> arr;
::arrow::Int32Builder b;
ASSERT_OK(b.AppendValues({1, 2, 3}));
ASSERT_OK(b.Finish(&arr));
auto schema = ::arrow::schema({::arrow::field("f0", ::arrow::int32())});
auto table = ::arrow::Table::Make(schema, {arr});

::arrow::util::SecureString column_key(std::string("key1234567890123"));
::arrow::util::SecureString footer_key(std::string("footer_key123456"));

std::map<std::string, std::shared_ptr<parquet::ColumnEncryptionProperties>> enc_cols;
parquet::ColumnEncryptionProperties::Builder col_builder("f0");
col_builder.parquet_cipher(parquet::ParquetCipher::EXTERNAL_DBPA_V1);
col_builder.key_id("key1234567890123");
col_builder.key(column_key);
enc_cols["f0"] = col_builder.build();

const std::string lib_path =
parquet::encryption::external::test::TestUtils::GetTestLibraryPath();

parquet::ExternalFileEncryptionProperties::Builder fep_builder(footer_key);
fep_builder.footer_key_metadata("kf")
->encrypted_columns(enc_cols)
->algorithm(parquet::ParquetCipher::AES_GCM_V1)
->configuration_properties(
{{parquet::ParquetCipher::EXTERNAL_DBPA_V1,
{{"agent_library_path", lib_path}, {"file_path", "/tmp/test"}}}});

auto writer_props = parquet::WriterProperties::Builder()
.encryption(fep_builder.build_external())
->build();

ASSERT_OK_AND_ASSIGN(
auto sink, ::arrow::io::BufferOutputStream::Create(1 << 16, default_memory_pool()));
ASSERT_OK(parquet::arrow::WriteTable(*table, default_memory_pool(), sink,
/*chunk_size=*/1024, writer_props));
ASSERT_OK_AND_ASSIGN(auto buffer, sink->Finish());

auto kr = std::make_shared<parquet::StringKeyIdRetriever>();
kr->PutKey("kf", footer_key);
kr->PutKey("key1234567890123", column_key);

parquet::ExternalFileDecryptionProperties::Builder dep_builder;
dep_builder.key_retriever(kr)->app_context("{}")->configuration_properties(
{{parquet::ParquetCipher::EXTERNAL_DBPA_V1,
{{"agent_library_path", lib_path}, {"file_path", "/tmp/test"}}}});
parquet::ReaderProperties rp = parquet::default_reader_properties();
rp.file_decryption_properties(dep_builder.build_external());

ArrowReaderProperties arp;
arp.set_use_threads(true);

parquet::arrow::FileReaderBuilder frb;
ASSERT_OK(frb.Open(std::make_shared<BufferReader>(buffer), rp));
frb.properties(arp);

std::unique_ptr<parquet::arrow::FileReader> fr;
ASSERT_RAISES(Invalid, frb.Build(&fr));
}
#endif

TEST(TestArrowReadWrite, DownsampleDeprecatedInt96) {
using ::arrow::ArrayFromJSON;
using ::arrow::field;
Expand Down
22 changes: 0 additions & 22 deletions cpp/src/parquet/arrow/reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,24 +131,6 @@ std::shared_ptr<std::unordered_set<int>> VectorToSharedSet(
return result;
}

bool IsExternalDBPAEncryptionUsedInColumns(std::shared_ptr<FileMetaData> metadata) {
for (int row_group = 0; row_group < metadata->num_row_groups(); row_group++) {
auto row_group_metadata = metadata->RowGroup(row_group);
for (int column = 0; column < row_group_metadata->num_columns(); column++) {
auto column_metadata = row_group_metadata->ColumnChunk(column);
if (column_metadata->crypto_metadata()) {
auto crypto_metadata = column_metadata->crypto_metadata();
if (crypto_metadata->is_encryption_algorithm_set() &&
crypto_metadata->encryption_algorithm().algorithm ==
ParquetCipher::EXTERNAL_DBPA_V1) {
return true;
}
}
}
}
return false;
}

// Forward declaration
Status GetReader(const SchemaField& field, const std::shared_ptr<ReaderContext>& context,
std::unique_ptr<ColumnReaderImpl>* out);
Expand All @@ -169,10 +151,6 @@ class FileReaderImpl : public FileReader {
// is not safe to use multiple threads to read the file.
if (reader_properties_.use_threads()) {
auto metadata = reader_->metadata();
if (IsExternalDBPAEncryptionUsedInColumns(metadata)) {
return Status::Invalid(
"EXTERNAL_DBPA_V1 encryption does not support multiple threads");
}
}
return SchemaManifest::Make(reader_->metadata()->schema(),
reader_->metadata()->key_value_metadata(),
Expand Down
10 changes: 4 additions & 6 deletions cpp/src/parquet/column_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -571,11 +571,6 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {

// Decrypt it if we need to
if (data_decryptor_ != nullptr) {
std::unique_ptr<EncodingProperties> encoding_properties =
GetEncodingProperties(current_page_header_);

data_decryptor_->UpdateEncodingProperties(std::move(encoding_properties));

std::shared_ptr<ResizableBuffer> decryption_buffer;
if (data_decryptor_->CanCalculateLengths()) {
decryption_buffer = AllocateBuffer(
Expand All @@ -584,9 +579,12 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
data_decryptor_->Decrypt(page_buffer->span_as<uint8_t>(),
decryption_buffer->mutable_span_as<uint8_t>());
} else {
std::unique_ptr<EncodingProperties> encoding_properties =
GetEncodingProperties(current_page_header_);
decryption_buffer = AllocateBuffer(properties_.memory_pool(), 0);
compressed_len = data_decryptor_->DecryptWithManagedBuffer(
page_buffer->span_as<uint8_t>(), decryption_buffer.get());
page_buffer->span_as<uint8_t>(), decryption_buffer.get(),
std::move(encoding_properties));
}

page_buffer = decryption_buffer;
Expand Down
39 changes: 18 additions & 21 deletions cpp/src/parquet/column_writer.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -307,10 +307,6 @@ class SerializedPageWriter : public PageWriter {
if (data_encryptor_.get()) {
UpdateEncryption(encryption::kDictionaryPage);

// Creating an EncodingProperties object from the metadata.
// We're retrieving the column descriptor and writer properties
// from the metadata_ object to simplify the code.
//
// WriterProperties is created using WriterProperties::Builder::build() (in
// parquet/properties.h) via ParquetFileFormat::MakeWriter (in
// arrow/dataset/file_parquet.cc), and passed down to
Expand All @@ -322,21 +318,23 @@ class SerializedPageWriter : public PageWriter {
// where a ColumnDescriptor is extracted from the SchemaDescriptor, and
// passed into ColumnChunkMetaDataBuilder::Make()

std::unique_ptr<EncodingProperties> encoding_properties =
EncodingProperties::MakeFromMetadata(metadata_->descr(),
metadata_->properties(),
static_cast<const DictionaryPage&>(page));
data_encryptor_->UpdateEncodingProperties(std::move(encoding_properties));

if (data_encryptor_->CanCalculateCiphertextLength()) {
PARQUET_THROW_NOT_OK(encryption_buffer_->Resize(
data_encryptor_->CiphertextLength(output_data_len), false));
output_data_len =
data_encryptor_->Encrypt(compressed_data->span_as<uint8_t>(),
encryption_buffer_->mutable_span_as<uint8_t>());
} else {
// Creating an EncodingProperties object from the metadata.
// We're retrieving the column descriptor and writer properties
// from the metadata_ object to simplify the code.
std::unique_ptr<EncodingProperties> encoding_properties =
EncodingProperties::MakeFromMetadata(
metadata_->descr(), metadata_->properties(),
static_cast<const DictionaryPage&>(page));
output_data_len = data_encryptor_->EncryptWithManagedBuffer(
compressed_data->span_as<uint8_t>(), encryption_buffer_.get());
compressed_data->span_as<uint8_t>(), encryption_buffer_.get(),
std::move(encoding_properties));
}

output_data_buffer = encryption_buffer_->data();
Expand Down Expand Up @@ -433,10 +431,6 @@ class SerializedPageWriter : public PageWriter {
if (data_encryptor_.get()) {
UpdateEncryption(encryption::kDataPage);

// Creating an EncodingProperties object from the metadata.
// We're retrieving the column descriptor and writer properties
// from the metadata_ object to simplify the code.
//
// WriterProperties is created using WriterProperties::Builder::build()
// (in parquet/properties.h ) via ParquetFileFormat::MakeWriter
// (in arrow/dataset/file_parquet.cc), and passed down to
Expand All @@ -447,11 +441,6 @@ class SerializedPageWriter : public PageWriter {
// RowGroupMetadataBuilder NextColumnChunk() (parquet/metadata.cc) where a
// ColumnDescriptor is extracted from the SchemaDescriptor, and passed into
// ColumnChunkMetaDataBuilder::Make()
std::unique_ptr<EncodingProperties> encoding_properties =
EncodingProperties::MakeFromMetadata(metadata_->descr(),
metadata_->properties(),
static_cast<const DataPage&>(page));
data_encryptor_->UpdateEncodingProperties(std::move(encoding_properties));

if (data_encryptor_->CanCalculateCiphertextLength()) {
PARQUET_THROW_NOT_OK(encryption_buffer_->Resize(
Expand All @@ -460,8 +449,16 @@ class SerializedPageWriter : public PageWriter {
data_encryptor_->Encrypt(compressed_data->span_as<uint8_t>(),
encryption_buffer_->mutable_span_as<uint8_t>());
} else {
// Creating an EncodingProperties object from the metadata.
// We're retrieving the column descriptor and writer properties
// from the metadata_ object to simplify the code.
std::unique_ptr<EncodingProperties> encoding_properties =
EncodingProperties::MakeFromMetadata(metadata_->descr(),
metadata_->properties(),
static_cast<const DataPage&>(page));
output_data_len = data_encryptor_->EncryptWithManagedBuffer(
compressed_data->span_as<uint8_t>(), encryption_buffer_.get());
compressed_data->span_as<uint8_t>(), encryption_buffer_.get(),
std::move(encoding_properties));
}
output_data_buffer = encryption_buffer_->data();

Expand Down
6 changes: 4 additions & 2 deletions cpp/src/parquet/encryption/aes_encryption.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,7 +159,8 @@ int32_t AesEncryptor::SignedFooterEncrypt(::arrow::util::span<const uint8_t> foo
int32_t AesEncryptor::Encrypt(::arrow::util::span<const uint8_t> plaintext,
::arrow::util::span<const uint8_t> key,
::arrow::util::span<const uint8_t> aad,
::arrow::util::span<uint8_t> ciphertext) {
::arrow::util::span<uint8_t> ciphertext,
std::unique_ptr<EncodingProperties> encoding_properties) {

@argmarco-tkd argmarco-tkd Feb 13, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This LGTM, but thinking performance - is passing a ptr to EncodingProperties, or passing by value more efficient (can be left for a later pass - but decided to ask given that we're changing the interface)

@sofia-tekdatum sofia-tekdatum Feb 25, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd rather keep it this way to explicitly transfer ownership of the pointer. The move here is calling the pointer only, not the deep copy of the EncodingProperties within.

if (static_cast<size_t>(key_length_) != key.size()) {
std::stringstream ss;
ss << "Wrong key length " << key.size() << ". Should be " << key_length_;
Expand Down Expand Up @@ -413,7 +414,8 @@ int32_t AesDecryptor::CiphertextLength(int32_t plaintext_len) const {
int32_t AesDecryptor::Decrypt(::arrow::util::span<const uint8_t> ciphertext,
::arrow::util::span<const uint8_t> key,
::arrow::util::span<const uint8_t> aad,
::arrow::util::span<uint8_t> plaintext) {
::arrow::util::span<uint8_t> plaintext,
std::unique_ptr<EncodingProperties> encoding_properties) {
if (static_cast<size_t>(key_length_) != key.size()) {
std::stringstream ss;
ss << "Wrong key length " << key.size() << ". Should be " << key_length_;
Expand Down
28 changes: 16 additions & 12 deletions cpp/src/parquet/encryption/aes_encryption.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,15 +80,17 @@ class PARQUET_EXPORT AesEncryptor : public AesCryptoContext, public EncryptorInt

/// Encrypts plaintext with the key and aad. Key length is passed only for validation.
/// If different from value in constructor, exception will be thrown.
int32_t Encrypt(::arrow::util::span<const uint8_t> plaintext,
::arrow::util::span<const uint8_t> key,
::arrow::util::span<const uint8_t> aad,
::arrow::util::span<uint8_t> ciphertext) override;
int32_t Encrypt(
::arrow::util::span<const uint8_t> plaintext,
::arrow::util::span<const uint8_t> key, ::arrow::util::span<const uint8_t> aad,
::arrow::util::span<uint8_t> ciphertext,
std::unique_ptr<EncodingProperties> encoding_properties = nullptr) override;

/// Encrypt the plaintext and leave the results in the ciphertext buffer. This method is
/// not supported as we can calculate the ciphertext length before encryption.
int32_t EncryptWithManagedBuffer(::arrow::util::span<const uint8_t> plaintext,
::arrow::ResizableBuffer* ciphertext) override {
int32_t EncryptWithManagedBuffer(
::arrow::util::span<const uint8_t> plaintext, ::arrow::ResizableBuffer* ciphertext,
std::unique_ptr<EncodingProperties> encoding_properties = nullptr) override {
throw ParquetException(
"EncryptWithManagedBuffer is not supported in AesEncryptor, use Encrypt instead");
}
Expand Down Expand Up @@ -168,16 +170,18 @@ class PARQUET_EXPORT AesDecryptor : public AesCryptoContext, public DecryptorInt
/// validation. If different from value in constructor, exception will be thrown.
/// The caller is responsible for ensuring that the plaintext buffer is at least as
/// large as PlaintextLength(ciphertext_len).
int32_t Decrypt(::arrow::util::span<const uint8_t> ciphertext,
::arrow::util::span<const uint8_t> key,
::arrow::util::span<const uint8_t> aad,
::arrow::util::span<uint8_t> plaintext) override;
int32_t Decrypt(
::arrow::util::span<const uint8_t> ciphertext,
::arrow::util::span<const uint8_t> key, ::arrow::util::span<const uint8_t> aad,
::arrow::util::span<uint8_t> plaintext,
std::unique_ptr<EncodingProperties> encoding_properties = nullptr) override;

/// Decrypt the ciphertext and leave the results in the plaintext buffer. This
/// method is not supported as we can calculate the plaintext length before
/// decryption.
int32_t DecryptWithManagedBuffer(::arrow::util::span<const uint8_t> ciphertext,
::arrow::ResizableBuffer* plaintext) override {
int32_t DecryptWithManagedBuffer(
::arrow::util::span<const uint8_t> ciphertext, ::arrow::ResizableBuffer* plaintext,
std::unique_ptr<EncodingProperties> encoding_properties = nullptr) override {
throw ParquetException(
"DecryptWithManagedBuffer is not supported in AesDecryptor, use Decrypt instead");
}
Expand Down
6 changes: 4 additions & 2 deletions cpp/src/parquet/encryption/aes_encryption_nossl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,8 @@ int32_t AesEncryptor::CiphertextLength(int64_t plaintext_len) const {
int32_t AesEncryptor::Encrypt(::arrow::util::span<const uint8_t> plaintext,
::arrow::util::span<const uint8_t> key,
::arrow::util::span<const uint8_t> aad,
::arrow::util::span<uint8_t> ciphertext) {
::arrow::util::span<uint8_t> ciphertext,
std::unique_ptr<EncodingProperties> encoding_properties) {
ThrowOpenSSLRequiredException();
return -1;
}
Expand All @@ -65,7 +66,8 @@ AesEncryptor::AesEncryptor(ParquetCipher::type alg_id, int32_t key_len, bool met
int32_t AesDecryptor::Decrypt(::arrow::util::span<const uint8_t> ciphertext,
::arrow::util::span<const uint8_t> key,
::arrow::util::span<const uint8_t> aad,
::arrow::util::span<uint8_t> plaintext) {
::arrow::util::span<uint8_t> plaintext,
std::unique_ptr<EncodingProperties> encoding_properties) {
ThrowOpenSSLRequiredException();
return -1;
}
Expand Down
20 changes: 8 additions & 12 deletions cpp/src/parquet/encryption/decryptor_interface.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,22 +41,18 @@ class PARQUET_EXPORT DecryptorInterface {
/// Decrypt the ciphertext and leave the results in the plaintext buffer.
/// Most implementations will require the key and aad to be provided, but it is up to
/// each decryptor whether to use them or not.
virtual int32_t Decrypt(::arrow::util::span<const uint8_t> ciphertext,
::arrow::util::span<const uint8_t> key,
::arrow::util::span<const uint8_t> aad,
::arrow::util::span<uint8_t> plaintext) = 0;
virtual int32_t Decrypt(
::arrow::util::span<const uint8_t> ciphertext,
::arrow::util::span<const uint8_t> key, ::arrow::util::span<const uint8_t> aad,
::arrow::util::span<uint8_t> plaintext,
std::unique_ptr<EncodingProperties> encoding_properties = nullptr) = 0;

/// Decrypt the ciphertext and leave the results in the plaintext buffer.
/// The buffer will be resized to the correct size during decryption. This method
/// is used when the decryptor cannot calculate the plaintext length before decryption.
virtual int32_t DecryptWithManagedBuffer(::arrow::util::span<const uint8_t> ciphertext,
::arrow::ResizableBuffer* plaintext) = 0;

// Some Encryptors may need to understand the page encoding before the encryption
/// process. This method will be called from ColumnWriter before invoking the
/// Encrypt method.
virtual void UpdateEncodingProperties(
std::unique_ptr<EncodingProperties> encoding_properties) {}
virtual int32_t DecryptWithManagedBuffer(
::arrow::util::span<const uint8_t> ciphertext, ::arrow::ResizableBuffer* plaintext,
std::unique_ptr<EncodingProperties> encoding_properties = nullptr) = 0;
};

} // namespace parquet::encryption
Loading
Loading