72 const std::optional<Fragmenter_Namespace::ChunkUpdateStats>&
update_stats) {
77 CHECK(update_stats->chunk);
78 CHECK(update_stats->chunk->getBuffer());
79 CHECK(update_stats->chunk->getBuffer()->getEncoder());
81 auto chunk_metadata = std::make_shared<ChunkMetadata>();
82 update_stats->chunk->getBuffer()->getEncoder()->getMetadata(chunk_metadata);
83 auto cd = update_stats.value().chunk->getColumnDesc();
84 if (cd->columnType.is_fp()) {
86 if (cd->columnType.get_type() ==
kDOUBLE) {
87 min = chunk_metadata->chunkStats.min.doubleval;
88 max = chunk_metadata->chunkStats.max.doubleval;
89 }
else if (cd->columnType.get_type() ==
kFLOAT) {
90 min = chunk_metadata->chunkStats.min.floatval;
91 max = chunk_metadata->chunkStats.max.floatval;
168 bool varlen_update_required)
195 using OffsetVector = std::vector<uint64_t>;
196 using ScalarTargetValueVector = std::vector<ScalarTargetValue>;
197 using RowProcessingFuturesVector = std::vector<std::future<uint64_t>>;
200 auto callback = [
this, &update_parameters](
203 std::vector<const ColumnDescriptor*> columnDescriptors;
204 std::vector<TargetMetaInfo> sourceMetaInfos;
211 columnDescriptors.push_back(target_column);
219 fragmenter->updateColumns(
230 table_update_metadata.fragments_with_deleted_rows[td->tableId].emplace(
231 update_log.getFragmentId());
238 CHECK(rs->didOutputColumnar());
239 CHECK(rs->isDirectColumnarConversionPossible());
241 CHECK_EQ(rs->colCount(), size_t(1));
248 const auto table_lock =
257 auto chunk_metadata =
258 fragment_info.getChunkMetadataMapPhysical().find(cd->columnId);
259 CHECK(chunk_metadata != fragment_info.getChunkMetadataMapPhysical().end());
263 fragment_info.fragmentId};
269 chunk_metadata->second->numBytes,
270 chunk_metadata->second->numElements);
272 auto chunk_buffer = chunk->getBuffer();
275 auto encoder = chunk_buffer->getEncoder();
279 rs.get(), 0, cd->columnType, rs->rowCount());
280 auto buffer =
reinterpret_cast<int8_t*
>(owned_buffer.get());
282 const auto new_chunk_metadata =
283 encoder->appendData(buffer, rs->rowCount(), cd->columnType,
false, 0);
284 CHECK(new_chunk_metadata);
286 auto fragmenter = td->fragmenter.get();
291 auto fragment = fragmenter->getFragmentInfo(fragment_info.fragmentId);
296 fragment->setChunkMetadata(cd->columnId, new_chunk_metadata);
297 fragment->shadowChunkMetadataMap =
298 fragment->getChunkMetadataMapPhysicalCopy();
301 if (data_mgr.gpusPresent()) {
309 auto callback = [
this, &update_parameters](
314 if (rows_per_column == 0) {
318 OffsetVector column_offsets(rows_per_column);
319 ScalarTargetValueVector scalar_target_values(rows_per_column);
325 complete_entry_block_size = entries_per_column;
326 partial_row_block_size = 0;
330 std::atomic<size_t> row_idx{0};
333 [&update_parameters, &column_offsets, &scalar_target_values, &row_idx](
334 auto get_entry_at_func,
335 uint64_t column_index,
336 uint64_t entry_start,
337 uint64_t entry_count) -> uint64_t {
338 uint64_t entries_processed = 0;
339 for (uint64_t entry_index = entry_start;
340 entry_index < (entry_start + entry_count);
342 const auto& row = get_entry_at_func(entry_index);
348 size_t row_index = row_idx.fetch_add(1);
352 auto terminal_column_iter = std::prev(row.end());
353 const auto frag_offset_scalar_tv =
354 boost::get<ScalarTargetValue>(&*terminal_column_iter);
355 CHECK(frag_offset_scalar_tv);
357 column_offsets[row_index] =
358 static_cast<uint64_t
>(*(boost::get<int64_t>(frag_offset_scalar_tv)));
359 scalar_target_values[row_index] =
360 boost::get<ScalarTargetValue>(row[column_index]);
362 return entries_processed;
366 [complete_entry_block_size](uint64_t thread_index) -> uint64_t {
367 return (thread_index * complete_entry_block_size);
370 auto const* table_descriptor =
372 CHECK(table_descriptor);
380 RowProcessingFuturesVector entry_processing_futures;
381 entry_processing_futures.reserve(usable_threads);
383 auto get_entry_at_func = [&update_log,
384 &column_index](
const size_t entry_index) {
392 for (
unsigned i = 0; i < static_cast<unsigned>(usable_threads); i++) {
393 entry_processing_futures.emplace_back(
395 std::forward<decltype(process_rows)>(process_rows),
399 complete_entry_block_size));
401 if (partial_row_block_size) {
402 entry_processing_futures.emplace_back(
404 std::forward<decltype(process_rows)>(process_rows),
407 get_row_index(usable_threads),
408 partial_row_block_size));
411 uint64_t entries_processed(0);
412 for (
auto& t : entry_processing_futures) {
414 entries_processed += t.get();
417 CHECK(row_idx == rows_per_column);
420 const auto fragmenter = table_descriptor->fragmenter;
430 scalar_target_values,
435 table_update_metadata.columns_for_metadata_update[target_column].emplace(
446 using RowProcessingFuturesVector = std::vector<std::future<uint64_t>>;
453 CHECK(rs->didOutputColumnar());
454 CHECK(rs->isDirectColumnarConversionPossible());
455 CHECK_EQ(rs->colCount(), size_t(1));
461 const auto table_lock =
470 auto chunk_metadata =
471 fragment_info.getChunkMetadataMapPhysical().find(cd->columnId);
472 CHECK(chunk_metadata != fragment_info.getChunkMetadataMapPhysical().end());
476 fragment_info.fragmentId};
482 chunk_metadata->second->numBytes,
483 chunk_metadata->second->numElements);
485 auto chunk_buffer = chunk->getBuffer();
488 auto encoder = chunk_buffer->getEncoder();
492 rs.get(), 0, cd->columnType, rs->rowCount());
493 auto buffer =
reinterpret_cast<int8_t*
>(owned_buffer.get());
495 const auto new_chunk_metadata =
496 encoder->appendData(buffer, rs->rowCount(), cd->columnType,
false, 0);
498 auto fragmenter = td->fragmenter.get();
503 auto fragment = fragmenter->getFragmentInfo(fragment_info.fragmentId);
508 fragment->setChunkMetadata(cd->columnId, new_chunk_metadata);
509 fragment->shadowChunkMetadataMap =
510 fragment->getChunkMetadataMapPhysicalCopy();
513 if (data_mgr.gpusPresent()) {
521 auto callback = [
this, &delete_parameters](
526 if (rows_per_column == 0) {
536 complete_row_block_size = rows_per_column;
537 partial_row_block_size = 0;
541 std::atomic<size_t> row_idx{0};
543 auto process_rows = [&update_log, &victim_offsets, &row_idx](
544 uint64_t entry_start, uint64_t entry_count) -> uint64_t {
545 uint64_t entries_processed = 0;
547 for (uint64_t entry_index = entry_start;
548 entry_index < (entry_start + entry_count);
550 auto const row(update_log.
getEntryAt(entry_index));
557 size_t row_index = row_idx.fetch_add(1);
559 auto terminal_column_iter = std::prev(row.end());
560 const auto scalar_tv = boost::get<ScalarTargetValue>(&*terminal_column_iter);
563 uint64_t fragment_offset =
564 static_cast<uint64_t
>(*(boost::get<int64_t>(scalar_tv)));
565 victim_offsets[row_index] = fragment_offset;
567 return entries_processed;
571 [complete_row_block_size](uint64_t thread_index) -> uint64_t {
572 return thread_index * complete_row_block_size;
575 RowProcessingFuturesVector row_processing_futures;
576 row_processing_futures.reserve(usable_threads);
578 for (
unsigned i = 0; i < (unsigned)usable_threads; i++) {
579 row_processing_futures.emplace_back(
581 std::forward<decltype(process_rows)>(process_rows),
583 complete_row_block_size));
585 if (partial_row_block_size) {
586 row_processing_futures.emplace_back(
588 std::forward<decltype(process_rows)>(process_rows),
589 get_row_index(usable_threads),
590 partial_row_block_size));
593 uint64_t rows_processed(0);
594 for (
auto& t : row_processing_futures) {
596 rows_processed += t.get();
599 auto const* table_descriptor =
602 auto* fragmenter = table_descriptor->fragmenter.get();
606 CHECK(deleted_column_desc);
616 table_update_metadata.fragments_with_deleted_rows[table_descriptor->tableId]
630 const auto padded_size = rs->getPaddedSlotWidthBytes(col_idx);
635 auto rs_buffer_size = padded_size * row_count;
636 auto rs_buffer = std::make_unique<int8_t[]>(rs_buffer_size);
637 rs->copyColumnIntoBuffer(col_idx, rs_buffer.get(), rs_buffer_size);
639 if (type_size < padded_size) {
643 auto src_ptr = rs_buffer.get();
644 auto dst_ptr = rs_buffer.get();
645 if (column_type.
is_fp()) {
647 CHECK(padded_size ==
sizeof(
double));
648 for (
size_t i = 0; i < row_count; i++) {
649 const auto old_val = *
reinterpret_cast<double*
>(may_alias_ptr(src_ptr));
650 auto new_val =
static_cast<float>(old_val);
651 std::memcpy(dst_ptr, &new_val, type_size);
652 dst_ptr += type_size;
653 src_ptr += padded_size;
657 for (
size_t i = 0; i < row_count; i++) {
658 std::memcpy(dst_ptr, src_ptr, type_size);
659 dst_ptr += type_size;
660 src_ptr += padded_size;
SQLTypeInfo getColumnType(const size_t col_idx) const
UpdateValuesStats new_values_stats
UpdateTransactionParameters(TableDescriptorType const *table_descriptor, UpdateTargetColumnNamesList const &update_column_names, UpdateTargetTypeList const &target_types, bool varlen_update_required)
auto isVarlenUpdateRequired() const
std::vector< int > ChunkKey
StorageIOFacility::TransactionLog transaction_tracker_
HOST DEVICE int get_size() const
class for a per-database catalog. also includes metadata for the current database and the current use...
void update_stats(Encoder *encoder, const SQLTypeInfo &column_type, DataBlockPtr data_block, const size_t row_count)
const ColumnDescriptor * getDeletedColumn(const TableDescriptor *td) const
std::vector< TargetMetaInfo > UpdateTargetTypeList
StorageIOFacility::UpdateCallback yieldDeleteCallback(DeleteTransactionParameters &delete_parameters)
UpdateTargetColumnNamesList update_column_names_
Data_Namespace::DataMgr & getDataMgr() const
bool should_recompute_metadata(const std::optional< Fragmenter_Namespace::ChunkUpdateStats > &update_stats)
static WriteLock getWriteLockForTable(const Catalog_Namespace::Catalog &cat, const std::string &table_name)
bool is_chunk_min_max_updated(const Fragmenter_Namespace::ChunkUpdateStats &update_stats, int64_t min, int64_t max)
static std::unique_ptr< int8_t[]> getRsBufferNoPadding(const ResultSet *rs, size_t col_idx, const SQLTypeInfo &column_type, size_t row_count)
virtual ~TransactionParameters()=default
bool g_enable_auto_metadata_update
int normalized_cpu_threads() const
HOST DEVICE SQLTypes get_type() const
UpdateTransactionParameters & operator=(UpdateTransactionParameters const &other)=delete
std::function< void(const UpdateLogForFragment &, TableUpdateMetadata &)> Callback
std::vector< std::string > UpdateTargetColumnNamesList
std::vector< uint64_t > UpdateTargetOffsetList
StorageIOFacility(Executor *executor, Catalog_Namespace::Catalog const &catalog)
StorageIOFacility::UpdateCallback yieldUpdateCallback(UpdateTransactionParameters &update_parameters)
future< Result > async(Fn &&fn, Args &&...args)
std::vector< TargetValue > getEntryAt(const size_t index) const override
int get_logical_size() const
decltype(FragmentInfoType::fragmentId) const getFragmentId() const
const DBMetadata & getCurrentDB() const
auto getUpdateColumnCount() const
decltype(FragmentInfoType::physicalTableId) const getPhysicalTableId() const
FragmentInfoType const & getFragmentInfo() const
Catalog_Namespace::Catalog const & catalog_
DeleteTransactionParameters & operator=(DeleteTransactionParameters const &other)=delete
const ColumnDescriptor * getMetadataForColumn(int tableId, const std::string &colName) const
UpdateTargetTypeList const & targets_meta_
std::vector< TargetValue > getTranslatedEntryAt(const size_t index) const override
void checkpointWithAutoRollback(const int logical_table_id) const
DeleteTransactionParameters(const TableDescriptorType *table_descriptor)
std::shared_ptr< Fragmenter_Namespace::AbstractFragmenter > fragmenter
std::unique_ptr< TransactionLog > TransactionLogPtr
std::function< bool(std::string const &)> ColumnValidationFunction
auto const & getUpdateColumnNames() const
void deleteChunksWithPrefix(const ChunkKey &keyPrefix)
bool table_is_temporary(const TableDescriptor *const td)
auto getTargetsMetaInfoSize() const
Data_Namespace::MemoryLevel persistenceLevel
auto const * getTableDescriptor() const
TableDescriptorType const * table_descriptor_
TransactionParameters(const TableDescriptorType *table_descriptor)
UpdateLogForFragment::Callback UpdateCallback
UpdelRoll ModifyTransactionTracker
StorageIOFacility::TransactionLog & getTransactionTracker()
std::vector< uint64_t > DeleteVictimOffsetList
auto tableIsTemporary() const
bool is_dict_encoded_string() const
const TableDescriptor * getMetadataForTable(const std::string &tableName, const bool populateFragmenter=true) const
Returns a pointer to a const TableDescriptor struct matching the provided tableName.
static std::shared_ptr< Chunk > getChunk(const ColumnDescriptor *cd, DataMgr *data_mgr, const ChunkKey &key, const MemoryLevel mem_level, const int deviceId, const size_t num_bytes, const size_t num_elems, const bool pinnable=true)
auto getResultSet() const
auto const & getTargetsMetaInfo() const
void finalizeTransaction(const Catalog_Namespace::Catalog &catalog)
size_t const getRowCount() const override
bool varlen_update_required_
UpdateValuesStats old_values_stats
size_t const getEntryCount() const override
boost::variant< int64_t, double, float, NullableString > ScalarTargetValue