Add TableStats to TableFragmentsInfo.

Signed-off-by: ienkovich <[email protected]>

Author: ienkovich

omniscidb/ArrowStorage/ArrowStorage.cpp

@@ -319,6 +319,7 @@ TableFragmentsInfo ArrowStorage::getTableMetadata(int db_id, int table_id) const
       frag_info.setChunkMetadata(columnId(col_idx), frag.metadata[col_idx]);
     }
   }
+  res.setTableStats(table.table_stats);
   return res;
 }
 
@@ -656,10 +657,16 @@ void ArrowStorage::appendArrowTable(std::shared_ptr<arrow::Table> at, int table_
       (static_cast<size_t>(at->num_rows()) + table.fragment_size - 1 - first_frag_size) /
           table.fragment_size +
       1;
+  // Pre-allocate fragment infos and table stats for each column for the following
+  // parallel data import.
   fragments.resize(frag_count);
   for (auto& frag : fragments) {
     frag.metadata.resize(at->columns().size());
   }
+  TableStats table_stats;
+  for (int col_idx = 0; col_idx < static_cast<int>(at->columns().size()); ++col_idx) {
+    table_stats.emplace(columnId(col_idx), ChunkStats{});
+  }
 
   mapd_shared_lock<mapd_shared_mutex> dict_lock(dict_mutex_);
   std::vector<bool> lazy_fetch_cols(at->columns().size(), false);
@@ -788,7 +795,17 @@ void ArrowStorage::appendArrowTable(std::shared_ptr<arrow::Table> at, int table_
                 frag.metadata[col_idx] = meta;
               }
             });  // each fragment
+
+        // Merge fragment stats to the table stats.
+        auto& column_stats = table_stats.at(col_info->column_id);
+        column_stats = fragments[0].metadata[col_idx]->chunkStats();
+        for (size_t frag_idx = 1; frag_idx < frag_count; ++frag_idx) {
+          mergeStats(column_stats,
+                     fragments[frag_idx].metadata[col_idx]->chunkStats(),
+                     col_type);
+        }
       } else {
+        bool has_nulls = false;
         for (size_t frag_idx = 0; frag_idx < frag_count; ++frag_idx) {
           auto& frag = fragments[frag_idx];
           frag.offset =
@@ -804,9 +821,15 @@ void ArrowStorage::appendArrowTable(std::shared_ptr<arrow::Table> at, int table_
               frag.row_count);
           meta->fillStringChunkStats(
               col_arr->Slice(frag.offset, frag.row_count)->null_count());
+          has_nulls = has_nulls || meta->chunkStats().has_nulls;
 
           frag.metadata[col_idx] = meta;
         }
+
+        auto& column_stats = table_stats.at(col_info->column_id);
+        column_stats.has_nulls = has_nulls;
+        column_stats.min.stringval = nullptr;
+        column_stats.max.stringval = nullptr;
       }
     }
   });  // each column
@@ -842,6 +865,13 @@ void ArrowStorage::appendArrowTable(std::shared_ptr<arrow::Table> at, int table_
       start_frag = 1;
     }
 
+    // Merge table stats.
+    for (size_t col_idx = 0; col_idx < at->columns().size(); ++col_idx) {
+      auto col_id = columnId(col_idx);
+      auto col_type = getColumnInfo(db_id_, table_id, col_id)->type;
+      mergeStats(table.table_stats.at(col_id), table_stats.at(col_id), col_type);
+    }
+
     // Copy the rest of fragments adjusting offset.
     table.fragments.reserve(table.fragments.size() + fragments.size() - start_frag);
     for (size_t frag_idx = start_frag; frag_idx < fragments.size(); ++frag_idx) {
@@ -855,7 +885,9 @@ void ArrowStorage::appendArrowTable(std::shared_ptr<arrow::Table> at, int table_
     table.col_data = std::move(col_data);
     table.fragments = std::move(fragments);
     table.row_count = at->num_rows();
+    table.table_stats = std::move(table_stats);
   }
+  CHECK_EQ(table.table_stats.size(), at->columns().size());
 
   auto table_info = getTableInfo(db_id_, table_id);
   table_info->fragments = table.fragments.size();

omniscidb/ArrowStorage/ArrowStorage.h

@@ -163,6 +163,7 @@ class ArrowStorage : public SimpleSchemaProvider, public AbstractDataProvider {
     std::vector<std::shared_ptr<arrow::ChunkedArray>> col_data;
     std::vector<DataFragment> fragments;
     size_t row_count = 0;
+    TableStats table_stats;
   };
 
   struct DictionaryData {

omniscidb/DataMgr/ChunkMetadata.h

@@ -190,6 +190,11 @@ class ChunkMetadata {
     return chunk_stats_;
   }
 
+  // Check if there are pre-computed chunk stats available.
+  // chunkStats can be used in any case but it can trigger
+  // stats computation if there are no pre-computed stats.
+  bool hasComputedChunkStats() const { return !stats_materialize_fn_; }
+
 #ifndef __CUDACC__
   std::string dump() const {
     std::string res = "type: " + type_->toString() +

omniscidb/DataProvider/TableFragmentsInfo.h

@@ -73,6 +73,9 @@ class FragmentInfo {
   mutable ChunkMetadataMap chunkMetadataMap;
 };
 
+using TableStats = std::map<int, ChunkStats>;
+using TableStatsMaterializeFn = std::function<void(TableStats&)>;
+
 class TableFragmentsInfo {
  public:
   TableFragmentsInfo() : numTuples(0) {}
@@ -94,9 +97,37 @@ class TableFragmentsInfo {
     return fragment_num_tupples_upper_bound;
   }
 
+  const TableStats& getTableStats() const {
+    maybeMaterializeTableStats();
+    return table_stats_;
+  }
+
+  // Check if there are pre-computed tables stats available.
+  // getTableStats can be used in any case but it can be costly
+  // if there are no pre-computed stats.
+  bool hasComputedTableStats() const { return !table_stats_materialize_fn_; }
+
+  void setTableStats(const TableStats& stats) const {
+    TableStatsMaterializeFn().swap(table_stats_materialize_fn_);
+    table_stats_ = stats;
+  }
+
+  void setTableStatsMaterializeFn(TableStatsMaterializeFn stats_materialize_fn) const {
+    table_stats_materialize_fn_ = stats_materialize_fn;
+  }
+
   std::vector<int> chunkKeyPrefix;
   std::vector<FragmentInfo> fragments;
 
  private:
+  void maybeMaterializeTableStats() const {
+    if (table_stats_materialize_fn_) {
+      table_stats_materialize_fn_(table_stats_);
+      TableStatsMaterializeFn().swap(table_stats_materialize_fn_);
+    }
+  }
+
   mutable size_t numTuples;
+  mutable TableStats table_stats_;
+  mutable TableStatsMaterializeFn table_stats_materialize_fn_;
 };

omniscidb/QueryEngine/ExpressionRange.cpp

@@ -587,11 +587,29 @@ ExpressionRange getLeafColumnRange(const hdk::ir::ColumnVar* col_expr,
           }
         }
       }
+
+      auto& table_stats = query_info.getTableStats();
+      auto col_stats_it = table_stats.find(col_id);
+      CHECK(col_stats_it != table_stats.end())
+          << query_infos[*ti_idx].db_id << ":" << query_infos[*ti_idx].table_id << ":"
+          << col_id << " " << table_stats.size();
+      if (col_stats_it == table_stats.end()) {
+        return ExpressionRange::makeInvalidRange();
+      }
+
+      auto& col_stats = col_stats_it->second;
+      CHECK_EQ(col_stats.has_nulls || is_outer_join_proj, has_nulls);
+      has_nulls = has_nulls || col_stats.has_nulls;
+
       if (col_type->isFloatingPoint()) {
         const auto min_val =
             extract_min_stat_fp_type(min_it->second->chunkStats(), col_type);
         const auto max_val =
             extract_max_stat_fp_type(max_it->second->chunkStats(), col_type);
+        const auto new_min_val = extract_min_stat_fp_type(col_stats, col_type);
+        const auto new_max_val = extract_max_stat_fp_type(col_stats, col_type);
+        CHECK_EQ(new_min_val, min_val);
+        CHECK_EQ(new_max_val, max_val);
         return col_type->size() == 4
                    ? ExpressionRange::makeFloatRange(min_val, max_val, has_nulls)
                    : ExpressionRange::makeDoubleRange(min_val, max_val, has_nulls);
@@ -600,6 +618,10 @@ ExpressionRange getLeafColumnRange(const hdk::ir::ColumnVar* col_expr,
           extract_min_stat_int_type(min_it->second->chunkStats(), col_type);
       const auto max_val =
           extract_max_stat_int_type(max_it->second->chunkStats(), col_type);
+      const auto new_min_val = extract_min_stat_int_type(col_stats, col_type);
+      const auto new_max_val = extract_max_stat_int_type(col_stats, col_type);
+      CHECK_EQ(new_min_val, min_val);
+      CHECK_EQ(new_max_val, max_val);
       if (max_val < min_val) {
         // The column doesn't contain any non-null values, synthesize an empty range.
         CHECK_GT(min_val, 0);

omniscidb/QueryEngine/InputMetadata.cpp

@@ -22,11 +22,7 @@ InputTableInfoCache::InputTableInfoCache(Executor* executor) : executor_(executo
 namespace {
 
 TableFragmentsInfo copy_table_info(const TableFragmentsInfo& table_info) {
-  TableFragmentsInfo table_info_copy;
-  table_info_copy.chunkKeyPrefix = table_info.chunkKeyPrefix;
-  table_info_copy.fragments = table_info.fragments;
-  table_info_copy.setPhysicalNumTuples(table_info.getPhysicalNumTuples());
-  return table_info_copy;
+  return table_info;
 }
 
 }  // namespace

omniscidb/ResultSetRegistry/ResultSetRegistry.cpp

@@ -292,6 +292,59 @@ ChunkStats ResultSetRegistry::getChunkStats(int table_id,
   return frag.meta.at(columnId(col_idx))->chunkStats();
 }
 
+TableStats ResultSetRegistry::getTableStats(int table_id) const {
+  mapd_shared_lock<mapd_shared_mutex> data_lock(data_mutex_);
+  CHECK_EQ(tables_.count(table_id), (size_t)1);
+  auto& table = *tables_.at(table_id);
+  mapd_shared_lock<mapd_shared_mutex> table_lock(table.mutex);
+  data_lock.unlock();
+
+  if (!table.table_stats.empty()) {
+    return table.table_stats;
+  }
+
+  for (auto& frag : table.fragments) {
+    mapd_shared_lock<mapd_shared_mutex> frag_read_lock(*frag.mutex);
+    if (frag.meta.empty()) {
+      frag_read_lock.unlock();
+      mapd_unique_lock<mapd_shared_mutex> frag_write_lock(*frag.mutex);
+      if (frag.meta.empty()) {
+        frag.meta = synthesizeMetadata(frag.rs.get());
+      }
+    }
+  }
+
+  auto table_stats = buildTableStatsNoLock(table_id);
+  table_lock.unlock();
+  mapd_unique_lock<mapd_shared_mutex> table_write_lock(table.mutex);
+  if (table.table_stats.empty()) {
+    table.table_stats = table_stats;
+  }
+  return table_stats;
+}
+
+TableStats ResultSetRegistry::buildTableStatsNoLock(int table_id) const {
+  // This method is only called when all fragments have computed metadata
+  // and table is read-locked.
+  CHECK(tables_.count(table_id));
+  auto& table = *tables_.at(table_id);
+  TableStats table_stats;
+  {
+    auto& first_frag = table.fragments.front();
+    mapd_shared_lock<mapd_shared_mutex> frag_lock(*first_frag.mutex);
+    for (auto& pr : first_frag.meta) {
+      table_stats.emplace(pr.first, pr.second->chunkStats());
+    }
+  }
+  for (size_t frag_idx = 1; frag_idx < table.fragments.size(); ++frag_idx) {
+    mapd_shared_lock<mapd_shared_mutex> frag_lock(*table.fragments[frag_idx].mutex);
+    for (auto& pr : table.fragments[frag_idx].meta) {
+      mergeStats(table_stats.at(pr.first), pr.second->chunkStats(), pr.second->type());
+    }
+  }
+  return table_stats;
+}
+
 void ResultSetRegistry::fetchBuffer(const ChunkKey& key,
                                     Data_Namespace::AbstractBuffer* dest,
                                     const size_t num_bytes) {
@@ -386,6 +439,7 @@ TableFragmentsInfo ResultSetRegistry::getTableMetadata(int db_id, int table_id)
 
   TableFragmentsInfo res;
   res.setPhysicalNumTuples(table.row_count);
+  bool has_lazy_stats = false;
   for (size_t frag_idx = 0; frag_idx < table.fragments.size(); ++frag_idx) {
     auto& frag = table.fragments[frag_idx];
     auto& frag_info = res.fragments.emplace_back();
@@ -416,13 +470,34 @@ TableFragmentsInfo ResultSetRegistry::getTableMetadata(int db_id, int table_id)
                 stats = this->getChunkStats(table_id, frag_idx, col_idx);
               });
           frag_info.setChunkMetadata(columnId(col_idx), meta);
+          has_lazy_stats = true;
         }
       }
     } else {
       frag_info.setChunkMetadataMap(frag.meta);
     }
   }
 
+  if (table.table_stats.empty()) {
+    if (has_lazy_stats) {
+      res.setTableStatsMaterializeFn(
+          [this, table_id](TableStats& stats) { stats = this->getTableStats(table_id); });
+    } else {
+      // We can get here if all stats were materialized in the loop above.
+      // In this case, build and assigne table stats.
+      TableStats table_stats = buildTableStatsNoLock(table_id);
+      res.setTableStats(table_stats);
+
+      table_lock.unlock();
+      mapd_unique_lock<mapd_shared_mutex> table_write_lock(table.mutex);
+      if (table.table_stats.empty()) {
+        table.table_stats = std::move(table_stats);
+      }
+    }
+  } else {
+    res.setTableStats(table.table_stats);
+  }
+
   return res;
 }
 

omniscidb/ResultSetRegistry/ResultSetRegistry.h

@@ -58,6 +58,8 @@ class ResultSetRegistry : public SimpleSchemaProvider,
 
  private:
   ChunkStats getChunkStats(int table_id, size_t frag_idx, size_t col_idx) const;
+  TableStats getTableStats(int table_id) const;
+  TableStats buildTableStatsNoLock(int table_id) const;
 
   struct DataFragment {
     size_t offset = 0;
@@ -74,6 +76,7 @@ class ResultSetRegistry : public SimpleSchemaProvider,
     size_t row_count;
     bool use_columnar_res;
     bool has_varlen_col;
+    TableStats table_stats;
   };
 
   const int db_id_;

omniscidb/Tests/TestDataProvider.h

@@ -86,6 +86,14 @@ class TestTableData {
 
     auto& frag_info = info_.fragments[data_[col_id - 1].size() - 1];
     frag_info.setChunkMetadata(col_id, chunk_meta);
+
+    if (table_stats_.count(col_id)) {
+      mergeStats(
+          table_stats_.at(col_id), chunk_meta->chunkStats(), col_types_.at(col_id));
+    } else {
+      table_stats_.emplace(col_id, chunk_meta->chunkStats());
+    }
+    info_.setTableStats(table_stats_);
   }
 
   void fetchData(int col_id, int frag_id, int8_t* dst, size_t size) {
@@ -102,6 +110,7 @@ class TestTableData {
   TableRef ref_;
   std::vector<std::vector<std::vector<int8_t>>> data_;
   TableFragmentsInfo info_;
+  TableStats table_stats_;
   std::unordered_map<int, const hdk::ir::Type*> col_types_;
 };