feat: support multi-threaded writing of Parquet files with modular encryption

[rok]

Which issue does this PR close?

• Closes Support multi-threaded writing of encrypted Parquet files #16737.

Rationale for this change

#16351 added modular encryption reading and writing. This builds on top of #16351 and uses apache/arrow-rs#7818 to enable multi threaded encrypted writing.

What changes are included in this PR?

This uses a lower level ArrowRowGroupWriterFactory API to create encryption-aware column writers that are then run multi threaded.

Are these changes tested?

Yes.

Are there any user-facing changes?

Previously parquet_opts.global.allow_single_file_parallelism == true would be ignored and encrypted write would always be single threaded. Now it will run multithreaded.

[rok]

Note: this includes some unrelated changes so as to be able to use changes in apache/arrow-rs#7818.
Also note: row_group_index at write time is not handled correctly yet. Do not merge.

@adamreeve @corwinjoy thought?

[adamreeve]

This approach looks good to me! Do the existing tests hit the parallel write code path? We probably want to make sure there are encryption tests for both the parallel and serial code paths.

[corwinjoy]

Via copilot:
This pull request introduces several updates across multiple components of the codebase, focusing on dependency management, feature enhancements, and code cleanup. The most significant changes include switching Arrow-related dependencies to a custom fork, deprecating the use of max_statistics_size in Parquet options, and extending support for additional decimal data types.

Dependency Updates:

• Updated Arrow-related dependencies (arrow, arrow-buffer, arrow-flight, etc.) in Cargo.toml to use a custom fork hosted on GitHub (https://github.com/rok/arrow-rs.git) with the multi-threaded_encrypted_writing branch. This change enables multi-threaded encrypted writing features. [1] [2]

Feature Enhancements:

• Added support for Decimal32 and Decimal64 data types in various parts of the codebase, including scalar value creation (datafusion/common/src/scalar/mod.rs), native type conversion (datafusion/common/src/types/native.rs), and hashing utilities (datafusion/expr/src/utils.rs). [1] [2] [3]
• Modified the Parquet writer to allow single-file parallelism explicitly in tests and serialization tasks, while introducing a new ArrowRowGroupWriterFactory for parallel row group writing. [1] [2] [3]

Code Cleanup:

• Deprecated the use of max_statistics_size in Parquet writer options and removed related code, including references to the deprecated constant and associated logic. [1] [2] [3]
• Removed unused example files (flight_sql_server, flight_server, flight_client) from datafusion-examples/Cargo.toml.

Protobuf Updates:

• Extended the protobuf conversion logic to handle Decimal32 and Decimal64 types in datafusion/proto-common/src/to_proto/mod.rs.

[corwinjoy]

@adamreeve Definitely something to note. We will want to resolve this before the final PR.

[corwinjoy]

Looks good to me, with the exception of multi-row group writing being missing. When you go to rebase to the latest datafusion the diff should get a lot simpler since they have upgraded to support arrow-55.3 and added the missing statitistics and types. I also like @adamreeve suggestion of having a test for the parallel write path.

[corwinjoy]

Copilot comments:

• Row Group Indexing: The TODO comment indicates that the row group writer currently assumes a single row group index (always 0). If multiple row groups are written in parallel, this might cause incorrect output or data corruption. Consider addressing this before merging.
• Encryption Parallelism: The removal of the explicit check for parallelism when encryption is enabled relies on arrow-rs’ internal handling. Double-check upstream to ensure there’s a clear error or fallback if parallelism is not supported with encryption.
• Testing: These changes affect core serialization logic. Ensure thorough integration and performance tests, especially for edge cases (encryption, large datasets, multiple row groups).
• Documentation: Consider updating related docs or comments, as the code flow and APIs used have changed significantly.

[rok]

I feel these were mostly addressed or the code was significantly refactored so they might not apply anymore.

[rok]

We have an open PR upstream apache/arrow-rs#8162 that should make this proposal possible. I'd like to ask for a round of reviews mostly to have an early reality check in case we're missing something.

[adamreeve]

👍 This approach looks good to me thanks Rok, I've just left one question.

[rok]

With arrow/parquet bumped to 56.2.0 multi-threaded encryption appears to work fine. In fact the whole test suite seems to pass. So once datafusion upgrades (#17751) arrow/parquet this can be merged too.

[rok]

cc @corwinjoy

[rok]

@alamb with arrow/parquet bumped (#17751) this would be ready for review. Since you've recently reviewed #8162 it might make sense to look at this while you still remember it :).

[alamb]

I'll review this shortly. Thanks @rok and @adamreeve