|
| 1 | +// Licensed to the Apache Software Foundation (ASF) under one |
| 2 | +// or more contributor license agreements. See the NOTICE file |
| 3 | +// distributed with this work for additional information |
| 4 | +// regarding copyright ownership. The ASF licenses this file |
| 5 | +// to you under the Apache License, Version 2.0 (the |
| 6 | +// "License"); you may not use this file except in compliance |
| 7 | +// with the License. You may obtain a copy of the License at |
| 8 | +// |
| 9 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | +// |
| 11 | +// Unless required by applicable law or agreed to in writing, |
| 12 | +// software distributed under the License is distributed on an |
| 13 | +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 14 | +// KIND, either express or implied. See the License for the |
| 15 | +// specific language governing permissions and limitations |
| 16 | +// under the License. |
| 17 | + |
| 18 | +#include "arrow/api.h" |
| 19 | +#include "arrow/dataset/file_parquet.h" |
| 20 | +#include "arrow/dataset/parquet_encryption_config.h" |
| 21 | +#include "arrow/filesystem/localfs.h" |
| 22 | +#include "parquet/encryption/crypto_factory.h" |
| 23 | +#include "parquet/encryption/test_in_memory_kms.h" |
| 24 | + |
| 25 | +#include <iostream> |
| 26 | + |
| 27 | +namespace fs = arrow::fs; |
| 28 | + |
| 29 | +namespace ds = arrow::dataset; |
| 30 | + |
| 31 | +arrow::Result<std::shared_ptr<arrow::Table>> GetTable() { |
| 32 | + auto int_builder = arrow::Int32Builder(); |
| 33 | + |
| 34 | + std::shared_ptr<arrow::Array> arr_i; |
| 35 | + ARROW_RETURN_NOT_OK(int_builder.AppendValues({1, 3, 5, 7, 1})); |
| 36 | + ARROW_RETURN_NOT_OK(int_builder.Finish(&arr_i)); |
| 37 | + |
| 38 | + auto struct_type = arrow::struct_({{"a", arrow::int32()}, {"b", arrow::int64()}}); |
| 39 | + auto pool = arrow::default_memory_pool(); |
| 40 | + auto a_builder = std::make_shared<arrow::Int32Builder>(); |
| 41 | + auto b_builder = std::make_shared<arrow::Int64Builder>(); |
| 42 | + auto struct_builder = arrow::StructBuilder(struct_type, pool, {a_builder, b_builder}); |
| 43 | + |
| 44 | + std::shared_ptr<arrow::Array> arr_struct; |
| 45 | + ARROW_RETURN_NOT_OK(struct_builder.Append()); |
| 46 | + ARROW_RETURN_NOT_OK(a_builder->Append(2)); |
| 47 | + ARROW_RETURN_NOT_OK(b_builder->Append(20)); |
| 48 | + ARROW_RETURN_NOT_OK(struct_builder.Append()); |
| 49 | + ARROW_RETURN_NOT_OK(a_builder->Append(4)); |
| 50 | + ARROW_RETURN_NOT_OK(b_builder->Append(40)); |
| 51 | + ARROW_RETURN_NOT_OK(struct_builder.Append()); |
| 52 | + ARROW_RETURN_NOT_OK(a_builder->Append(6)); |
| 53 | + ARROW_RETURN_NOT_OK(b_builder->Append(60)); |
| 54 | + ARROW_RETURN_NOT_OK(struct_builder.Append()); |
| 55 | + ARROW_RETURN_NOT_OK(a_builder->Append(8)); |
| 56 | + ARROW_RETURN_NOT_OK(b_builder->Append(80)); |
| 57 | + ARROW_RETURN_NOT_OK(struct_builder.Append()); |
| 58 | + ARROW_RETURN_NOT_OK(a_builder->Append(10)); |
| 59 | + ARROW_RETURN_NOT_OK(b_builder->Append(100)); |
| 60 | + ARROW_RETURN_NOT_OK(struct_builder.Finish(&arr_struct)); |
| 61 | + |
| 62 | + auto map_type = arrow::map(arrow::int32(), arrow::utf8()); |
| 63 | + auto key_builder = std::make_shared<arrow::Int32Builder>(); |
| 64 | + auto item_builder = std::make_shared<arrow::StringBuilder>(); |
| 65 | + auto map_builder = arrow::MapBuilder(pool, key_builder, item_builder, map_type); |
| 66 | + |
| 67 | + std::shared_ptr<arrow::Array> arr_map; |
| 68 | + ARROW_RETURN_NOT_OK(map_builder.Append()); |
| 69 | + ARROW_RETURN_NOT_OK(key_builder->AppendValues({2, 4})); |
| 70 | + ARROW_RETURN_NOT_OK(item_builder->AppendValues({"2", "4"})); |
| 71 | + ARROW_RETURN_NOT_OK(map_builder.Append()); |
| 72 | + ARROW_RETURN_NOT_OK(key_builder->AppendValues({6})); |
| 73 | + ARROW_RETURN_NOT_OK(item_builder->AppendValues({"6"})); |
| 74 | + ARROW_RETURN_NOT_OK(map_builder.Append()); |
| 75 | + ARROW_RETURN_NOT_OK(map_builder.Append()); |
| 76 | + ARROW_RETURN_NOT_OK(key_builder->AppendValues({8, 10})); |
| 77 | + ARROW_RETURN_NOT_OK(item_builder->AppendValues({"8", "10"})); |
| 78 | + ARROW_RETURN_NOT_OK(map_builder.Append()); |
| 79 | + ARROW_RETURN_NOT_OK(map_builder.Finish(&arr_map)); |
| 80 | + |
| 81 | + auto list_type = arrow::list(arrow::int32()); |
| 82 | + auto value_builder = std::make_shared<arrow::Int32Builder>(); |
| 83 | + auto list_builder = arrow::ListBuilder(pool, value_builder, list_type); |
| 84 | + |
| 85 | + std::shared_ptr<arrow::Array> arr_list; |
| 86 | + ARROW_RETURN_NOT_OK(list_builder.Append()); |
| 87 | + ARROW_RETURN_NOT_OK(value_builder->AppendValues({1, 2, 3})); |
| 88 | + ARROW_RETURN_NOT_OK(list_builder.Append()); |
| 89 | + ARROW_RETURN_NOT_OK(value_builder->AppendValues({4, 5, 6})); |
| 90 | + ARROW_RETURN_NOT_OK(list_builder.Append()); |
| 91 | + ARROW_RETURN_NOT_OK(value_builder->AppendValues({7})); |
| 92 | + ARROW_RETURN_NOT_OK(list_builder.Append()); |
| 93 | + ARROW_RETURN_NOT_OK(value_builder->AppendValues({8})); |
| 94 | + ARROW_RETURN_NOT_OK(list_builder.Append()); |
| 95 | + ARROW_RETURN_NOT_OK(list_builder.Finish(&arr_list)); |
| 96 | + |
| 97 | + auto schema = arrow::schema({ |
| 98 | + arrow::field("i", arrow::int32()), |
| 99 | + arrow::field("s", struct_type), |
| 100 | + arrow::field("m", map_type), |
| 101 | + arrow::field("l", list_type), |
| 102 | + }); |
| 103 | + |
| 104 | + return arrow::Table::Make(schema, {arr_i, arr_struct, arr_map, arr_list}); |
| 105 | +} |
| 106 | + |
| 107 | +std::shared_ptr<parquet::encryption::CryptoFactory> GetCryptoFactory() { |
| 108 | + // Configure KMS. |
| 109 | + std::unordered_map<std::string, std::string> key_map; |
| 110 | + key_map.emplace("footerKeyId", "0123456789012345"); |
| 111 | + key_map.emplace("columnKeyId", "1234567890123456"); |
| 112 | + |
| 113 | + auto crypto_factory = std::make_shared<parquet::encryption::CryptoFactory>(); |
| 114 | + auto kms_client_factory = |
| 115 | + // for testing only, do not use it as an example of KmsClientFactory implementation |
| 116 | + std::make_shared<parquet::encryption::TestOnlyInMemoryKmsClientFactory>( |
| 117 | + /*wrap_locally=*/true, key_map); |
| 118 | + crypto_factory->RegisterKmsClientFactory(std::move(kms_client_factory)); |
| 119 | + return crypto_factory; |
| 120 | +} |
| 121 | + |
| 122 | +arrow::Status WriteEncryptedFile(const std::string& path_to_file) { |
| 123 | + using arrow::internal::checked_pointer_cast; |
| 124 | + |
| 125 | + // Get a configured crypto factory and kms connection conf. |
| 126 | + auto crypto_factory = GetCryptoFactory(); |
| 127 | + auto kms_connection_config = |
| 128 | + std::make_shared<parquet::encryption::KmsConnectionConfig>(); |
| 129 | + |
| 130 | + // Set write options with encryption configuration. |
| 131 | + auto encryption_config = std::make_shared<parquet::encryption::EncryptionConfiguration>( |
| 132 | + std::string("footerKeyId")); |
| 133 | + encryption_config->column_keys = |
| 134 | + "columnKeyId: i, s.a, s.b, m.key_value.key, m.key_value.value, l.list.element"; |
| 135 | + |
| 136 | + auto parquet_encryption_config = std::make_shared<ds::ParquetEncryptionConfig>(); |
| 137 | + // Directly assign shared_ptr objects to ParquetEncryptionConfig members. |
| 138 | + parquet_encryption_config->crypto_factory = crypto_factory; |
| 139 | + parquet_encryption_config->kms_connection_config = kms_connection_config; |
| 140 | + parquet_encryption_config->encryption_config = std::move(encryption_config); |
| 141 | + |
| 142 | + auto file_format = std::make_shared<ds::ParquetFileFormat>(); |
| 143 | + auto parquet_file_write_options = checked_pointer_cast<ds::ParquetFileWriteOptions>( |
| 144 | + file_format->DefaultWriteOptions()); |
| 145 | + parquet_file_write_options->parquet_encryption_config = |
| 146 | + std::move(parquet_encryption_config); |
| 147 | + |
| 148 | + // Write dataset. |
| 149 | + ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Table> table, GetTable()); |
| 150 | + printf("%s", table->ToString().c_str()); |
| 151 | + auto dataset = std::make_shared<ds::InMemoryDataset>(table); |
| 152 | + ARROW_ASSIGN_OR_RAISE(auto scanner_builder, dataset->NewScan()); |
| 153 | + ARROW_ASSIGN_OR_RAISE(auto scanner, scanner_builder->Finish()); |
| 154 | + |
| 155 | + auto file_system = std::make_shared<fs::LocalFileSystem>(); |
| 156 | + auto partitioning = std::make_shared<ds::HivePartitioning>( |
| 157 | + arrow::schema({arrow::field("part", arrow::utf8())})); |
| 158 | + |
| 159 | + ds::FileSystemDatasetWriteOptions write_options; |
| 160 | + write_options.file_write_options = parquet_file_write_options; |
| 161 | + write_options.filesystem = file_system; |
| 162 | + write_options.base_dir = path_to_file; |
| 163 | + write_options.partitioning = partitioning; |
| 164 | + write_options.basename_template = "part{i}.parquet"; |
| 165 | + return ds::FileSystemDataset::Write(write_options, std::move(scanner)); |
| 166 | +} |
| 167 | + |
| 168 | +arrow::Status ReadEncryptedFile(const std::string& path_to_file) { |
| 169 | + // Get a configured crypto factory and kms connection conf |
| 170 | + auto crypto_factory = GetCryptoFactory(); |
| 171 | + auto kms_connection_config = |
| 172 | + std::make_shared<parquet::encryption::KmsConnectionConfig>(); |
| 173 | + |
| 174 | + // Create decryption properties. |
| 175 | + auto decryption_config = |
| 176 | + std::make_shared<parquet::encryption::DecryptionConfiguration>(); |
| 177 | + auto parquet_decryption_config = std::make_shared<ds::ParquetDecryptionConfig>(); |
| 178 | + parquet_decryption_config->crypto_factory = crypto_factory; |
| 179 | + parquet_decryption_config->kms_connection_config = kms_connection_config; |
| 180 | + parquet_decryption_config->decryption_config = std::move(decryption_config); |
| 181 | + |
| 182 | + // Set scan options. |
| 183 | + auto parquet_scan_options = std::make_shared<ds::ParquetFragmentScanOptions>(); |
| 184 | + parquet_scan_options->parquet_decryption_config = std::move(parquet_decryption_config); |
| 185 | + |
| 186 | + // Get configured Parquet file format |
| 187 | + auto file_format = std::make_shared<ds::ParquetFileFormat>(); |
| 188 | + file_format->default_fragment_scan_options = std::move(parquet_scan_options); |
| 189 | + |
| 190 | + // Get the FileSystem. |
| 191 | + auto file_system = std::make_shared<fs::LocalFileSystem>(); |
| 192 | + |
| 193 | + // Get FileInfo objects for all files under the base directory |
| 194 | + fs::FileSelector selector; |
| 195 | + selector.base_dir = path_to_file; |
| 196 | + selector.recursive = true; |
| 197 | + |
| 198 | + // Create the dataset |
| 199 | + ds::FileSystemFactoryOptions factory_options; |
| 200 | + ARROW_ASSIGN_OR_RAISE(auto dataset_factory, |
| 201 | + ds::FileSystemDatasetFactory::Make(file_system, selector, |
| 202 | + file_format, factory_options)); |
| 203 | + ARROW_ASSIGN_OR_RAISE(auto dataset, dataset_factory->Finish()); |
| 204 | + ARROW_ASSIGN_OR_RAISE(auto scanner_builder, dataset->NewScan()); |
| 205 | + ARROW_ASSIGN_OR_RAISE(auto scanner, scanner_builder->Finish()); |
| 206 | + ARROW_ASSIGN_OR_RAISE(auto table, scanner->ToTable()); |
| 207 | + std::cout << "Table size: " << table->num_rows() << "\n"; |
| 208 | + return arrow::Status::OK(); |
| 209 | +} |
| 210 | + |
| 211 | +arrow::Status RunExamples(const std::string& path_to_file) { |
| 212 | + ARROW_RETURN_NOT_OK(WriteEncryptedFile(path_to_file)); |
| 213 | + ARROW_RETURN_NOT_OK(ReadEncryptedFile(path_to_file)); |
| 214 | + return arrow::Status::OK(); |
| 215 | +} |
| 216 | + |
| 217 | +int main(int argc, char** argv) { |
| 218 | + if (argc != 2) { |
| 219 | + // Fake success for CI purposes. |
| 220 | + return EXIT_SUCCESS; |
| 221 | + } |
| 222 | + |
| 223 | + std::string path_to_file = argv[1]; |
| 224 | + arrow::Status status = RunExamples(path_to_file); |
| 225 | + |
| 226 | + if (!status.ok()) { |
| 227 | + std::cerr << "Error occurred: " << status.message() << std::endl; |
| 228 | + return EXIT_FAILURE; |
| 229 | + } |
| 230 | + return EXIT_SUCCESS; |
| 231 | +} |
0 commit comments