| 1270 | } |
| 1271 | |
| 1272 | Status WriteDictionaries(const RecordBatch& batch) { |
| 1273 | ARROW_ASSIGN_OR_RAISE(const auto dictionaries, CollectDictionaries(batch, mapper_)); |
| 1274 | const auto equal_options = EqualOptions().nans_equal(true); |
| 1275 | |
| 1276 | for (const auto& pair : dictionaries) { |
| 1277 | int64_t dictionary_id = pair.first; |
| 1278 | const auto& dictionary = pair.second; |
| 1279 | |
| 1280 | // If a dictionary with this id was already emitted, check if it was the same. |
| 1281 | auto* last_dictionary = &last_dictionaries_[dictionary_id]; |
| 1282 | const bool dictionary_exists = (*last_dictionary != nullptr); |
| 1283 | int64_t delta_start = 0; |
| 1284 | if (dictionary_exists) { |
| 1285 | if ((*last_dictionary)->data() == dictionary->data()) { |
| 1286 | // Fast shortcut for a common case. |
| 1287 | // Same dictionary data by pointer => no need to emit it again |
| 1288 | continue; |
| 1289 | } |
| 1290 | const int64_t last_length = (*last_dictionary)->length(); |
| 1291 | const int64_t new_length = dictionary->length(); |
| 1292 | if (new_length == last_length && |
| 1293 | ((*last_dictionary)->Equals(dictionary, equal_options))) { |
| 1294 | // Same dictionary by value => no need to emit it again |
| 1295 | // (while this can have a CPU cost, this code path is required |
| 1296 | // for the IPC file format) |
| 1297 | continue; |
| 1298 | } |
| 1299 | |
| 1300 | // (the read path doesn't support outer dictionary deltas, don't emit them) |
| 1301 | if (new_length > last_length && options_.emit_dictionary_deltas && |
| 1302 | !HasNestedDict(*dictionary->data()) && |
| 1303 | ((*last_dictionary) |
| 1304 | ->RangeEquals(dictionary, 0, last_length, 0, equal_options))) { |
| 1305 | // New dictionary starts with the current dictionary |
| 1306 | delta_start = last_length; |
| 1307 | } |
| 1308 | |
| 1309 | if (is_file_format_ && !delta_start) { |
| 1310 | return Status::Invalid( |
| 1311 | "Dictionary replacement detected when writing IPC file format. " |
| 1312 | "Arrow IPC files only support a single non-delta dictionary for " |
| 1313 | "a given field across all batches."); |
| 1314 | } |
| 1315 | } |
| 1316 | |
| 1317 | IpcPayload payload; |
| 1318 | if (delta_start) { |
| 1319 | RETURN_NOT_OK(GetDictionaryPayload(dictionary_id, /*is_delta=*/true, |
| 1320 | dictionary->Slice(delta_start), options_, |
| 1321 | &payload)); |
| 1322 | } else { |
| 1323 | RETURN_NOT_OK( |
| 1324 | GetDictionaryPayload(dictionary_id, dictionary, options_, &payload)); |
| 1325 | } |
| 1326 | RETURN_NOT_OK(WritePayload(payload)); |
| 1327 | ++stats_.num_dictionary_batches; |
| 1328 | if (dictionary_exists) { |
| 1329 | if (delta_start) { |
no test coverage detected