Storages: Optimize vector search in scenarios with updates (pingcap#316)

Signed-off-by: Lloyd-Pottiger <[email protected]>
Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com>
Co-authored-by: jinhelin <[email protected]>
Co-authored-by: JaySon <[email protected]>

Author: 4 people

Diff for: dbms/src/DataStreams/PushingToViewsBlockOutputStream.cpp

@@ -14,7 +14,6 @@
 
 #include <Common/Exception.h>
 #include <DataStreams/PushingToViewsBlockOutputStream.h>
-#include <DataStreams/SquashingBlockInputStream.h>
 #include <Interpreters/Context.h>
 #include <Interpreters/InterpreterSelectQuery.h>
 #include <Storages/IStorage.h>

Diff for: dbms/src/DataStreams/SquashingBlockInputStream.cpp

@@ -16,6 +16,7 @@
 
 namespace DB
 {
+
 SquashingBlockInputStream::SquashingBlockInputStream(
     const BlockInputStreamPtr & src,
     size_t min_block_size_rows,
@@ -28,7 +29,7 @@ SquashingBlockInputStream::SquashingBlockInputStream(
 }
 
 
-Block SquashingBlockInputStream::readImpl()
+Block SquashingBlockInputStream::read()
 {
     if (all_read)
         return {};

Diff for: dbms/src/DataStreams/SquashingBlockInputStream.h

@@ -14,15 +14,15 @@
 
 #pragma once
 
-#include <DataStreams/IProfilingBlockInputStream.h>
+#include <DataStreams/IBlockInputStream.h>
 #include <DataStreams/SquashingTransform.h>
 
 
 namespace DB
 {
 /** Merging consecutive blocks of stream to specified minimum size.
   */
-class SquashingBlockInputStream : public IProfilingBlockInputStream
+class SquashingBlockInputStream : public IBlockInputStream
 {
     static constexpr auto NAME = "Squashing";
 
@@ -37,8 +37,7 @@ class SquashingBlockInputStream : public IProfilingBlockInputStream
 
     Block getHeader() const override { return children.at(0)->getHeader(); }
 
-protected:
-    Block readImpl() override;
+    Block read() override;
 
 private:
     const LoggerPtr log;

Diff for: dbms/src/Storages/DeltaMerge/BitmapFilter/BitmapFilterView.h

@@ -46,12 +46,15 @@ class BitmapFilterView
         return BitmapFilterView(std::make_shared<BitmapFilter>(size, default_value), 0, size);
     }
 
-    inline bool get(UInt32 n) const
+    BitmapFilterView createSubView(UInt32 offset, UInt32 size) const
     {
-        RUNTIME_CHECK(n < filter_size);
-        return filter->get(filter_offset + n);
+        RUNTIME_CHECK(offset + size <= filter_size, offset, size, filter_size);
+        return BitmapFilterView(filter, filter_offset + offset, size);
     }
 
+    // Caller should ensure n in [0, size).
+    inline bool get(UInt32 n) const { return filter->get(filter_offset + n); }
+
     inline bool operator[](UInt32 n) const { return get(n); }
 
     inline UInt32 size() const { return filter_size; }

Diff for: dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFile.h

@@ -65,9 +65,9 @@ class ColumnFile
         : id(++MAX_COLUMN_FILE_ID)
     {}
 
+public:
     virtual ~ColumnFile() = default;
 
-public:
     enum Type : UInt32
     {
         DELETE_RANGE = 1,
@@ -96,8 +96,8 @@ class ColumnFile
     UInt64 getId() const { return id; }
 
     virtual size_t getRows() const { return 0; }
-    virtual size_t getBytes() const { return 0; };
-    virtual size_t getDeletes() const { return 0; };
+    virtual size_t getBytes() const { return 0; }
+    virtual size_t getDeletes() const { return 0; }
 
     virtual Type getType() const = 0;
 
@@ -106,11 +106,11 @@ class ColumnFile
     /// Is a ColumnFileTiny or not.
     bool isTinyFile() const { return getType() == Type::TINY_FILE; }
     /// Is a ColumnFileDeleteRange or not.
-    bool isDeleteRange() const { return getType() == Type::DELETE_RANGE; };
+    bool isDeleteRange() const { return getType() == Type::DELETE_RANGE; }
     /// Is a ColumnFileBig or not.
-    bool isBigFile() const { return getType() == Type::BIG_FILE; };
+    bool isBigFile() const { return getType() == Type::BIG_FILE; }
     /// Is a ColumnFilePersisted or not
-    bool isPersisted() const { return getType() != Type::INMEMORY_FILE; };
+    bool isPersisted() const { return getType() != Type::INMEMORY_FILE; }
 
     /**
      * Whether this column file SEEMS TO BE flushed from another.
@@ -132,7 +132,8 @@ class ColumnFile
     virtual ColumnFileReaderPtr getReader(
         const DMContext & context,
         const IColumnFileDataProviderPtr & data_provider,
-        const ColumnDefinesPtr & col_defs) const
+        const ColumnDefinesPtr & col_defs,
+        ReadTag read_tag) const
         = 0;
 
     /// Note: Only ColumnFileInMemory can be appendable. Other ColumnFiles (i.e. ColumnFilePersisted) have
@@ -179,9 +180,7 @@ class ColumnFileReader
     virtual size_t skipNextBlock() { throw Exception("Unsupported operation", ErrorCodes::LOGICAL_ERROR); }
 
     /// Create a new reader from current reader with different columns to read.
-    virtual ColumnFileReaderPtr createNewReader(const ColumnDefinesPtr & col_defs) = 0;
-
-    virtual void setReadTag(ReadTag /*read_tag*/) {}
+    virtual ColumnFileReaderPtr createNewReader(const ColumnDefinesPtr & col_defs, ReadTag read_tag) = 0;
 };
 
 std::pair<size_t, size_t> copyColumnsData(

Diff for: dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileBig.cpp

@@ -66,9 +66,10 @@ void ColumnFileBig::removeData(WriteBatches & wbs) const
 ColumnFileReaderPtr ColumnFileBig::getReader(
     const DMContext & dm_context,
     const IColumnFileDataProviderPtr &,
-    const ColumnDefinesPtr & col_defs) const
+    const ColumnDefinesPtr & col_defs,
+    ReadTag read_tag) const
 {
-    return std::make_shared<ColumnFileBigReader>(dm_context, *this, col_defs);
+    return std::make_shared<ColumnFileBigReader>(dm_context, *this, col_defs, read_tag);
 }
 
 void ColumnFileBig::serializeMetadata(WriteBuffer & buf, bool /*save_schema*/) const
@@ -379,17 +380,10 @@ size_t ColumnFileBigReader::skipNextBlock()
     }
 }
 
-ColumnFileReaderPtr ColumnFileBigReader::createNewReader(const ColumnDefinesPtr & new_col_defs)
+ColumnFileReaderPtr ColumnFileBigReader::createNewReader(const ColumnDefinesPtr & new_col_defs, ReadTag read_tag)
 {
     // Currently we don't reuse the cache data.
-    return std::make_shared<ColumnFileBigReader>(dm_context, column_file, new_col_defs);
-}
-
-void ColumnFileBigReader::setReadTag(ReadTag read_tag_)
-{
-    // `read_tag` should be set before `file_stream` is initialized.
-    RUNTIME_CHECK(file_stream == nullptr);
-    read_tag = read_tag_;
+    return std::make_shared<ColumnFileBigReader>(dm_context, column_file, new_col_defs, read_tag);
 }
 
 } // namespace DB::DM

Diff for: dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileBig.h

@@ -81,7 +81,8 @@ class ColumnFileBig : public ColumnFilePersisted
     ColumnFileReaderPtr getReader(
         const DMContext & dm_context,
         const IColumnFileDataProviderPtr & data_provider,
-        const ColumnDefinesPtr & col_defs) const override;
+        const ColumnDefinesPtr & col_defs,
+        ReadTag) const override;
 
     void serializeMetadata(WriteBuffer & buf, bool save_schema) const override;
     void serializeMetadata(dtpb::ColumnFilePersisted * cf_pb, bool save_schema) const override;
@@ -145,7 +146,7 @@ class ColumnFileBigReader : public ColumnFileReader
     Block cur_block;
     Columns cur_block_data; // The references to columns in cur_block, for faster access.
 
-    ReadTag read_tag = ReadTag::Internal;
+    ReadTag read_tag;
 
 private:
     void initStream();
@@ -164,10 +165,12 @@ class ColumnFileBigReader : public ColumnFileReader
     ColumnFileBigReader(
         const DMContext & dm_context_,
         const ColumnFileBig & column_file_,
-        const ColumnDefinesPtr & col_defs_)
+        const ColumnDefinesPtr & col_defs_,
+        ReadTag read_tag_)
         : dm_context(dm_context_)
         , column_file(column_file_)
         , col_defs(col_defs_)
+        , read_tag(read_tag_)
     {
         if (col_defs_->size() == 1)
         {
@@ -199,9 +202,7 @@ class ColumnFileBigReader : public ColumnFileReader
 
     size_t skipNextBlock() override;
 
-    ColumnFileReaderPtr createNewReader(const ColumnDefinesPtr & new_col_defs) override;
-
-    void setReadTag(ReadTag read_tag_) override;
+    ColumnFileReaderPtr createNewReader(const ColumnDefinesPtr & new_col_defs, ReadTag) override;
 };
 
 } // namespace DB::DM

Diff for: dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileDeleteRange.cpp

@@ -16,16 +16,17 @@
 #include <Storages/DeltaMerge/DMContext.h>
 
 
-namespace DB
-{
-namespace DM
+namespace DB::DM
 {
+
 ColumnFileReaderPtr ColumnFileDeleteRange::getReader(
     const DMContext &,
     const IColumnFileDataProviderPtr &,
-    const ColumnDefinesPtr &) const
+    const ColumnDefinesPtr &,
+    ReadTag) const
 {
-    return std::make_shared<ColumnFileEmptyReader>();
+    // ColumnFileDeleteRange is not readable.
+    return nullptr;
 }
 
 void ColumnFileDeleteRange::serializeMetadata(WriteBuffer & buf, bool /*save_schema*/) const
@@ -50,9 +51,4 @@ ColumnFilePersistedPtr ColumnFileDeleteRange::deserializeMetadata(const dtpb::Co
     return std::make_shared<ColumnFileDeleteRange>(RowKeyRange::deserialize(dr_pb.range()));
 }
 
-ColumnFileReaderPtr ColumnFileEmptyReader::createNewReader(const ColumnDefinesPtr &)
-{
-    return std::make_shared<ColumnFileEmptyReader>();
-}
-} // namespace DM
-} // namespace DB
+} // namespace DB::DM

Diff for: dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileDeleteRange.h

@@ -18,10 +18,9 @@
 #include <Storages/DeltaMerge/Remote/Serializer_fwd.h>
 
 
-namespace DB
-{
-namespace DM
+namespace DB::DM
 {
+
 class ColumnFileDeleteRange;
 using ColumnFileDeleteRangePtr = std::shared_ptr<ColumnFileDeleteRange>;
 
@@ -42,8 +41,11 @@ class ColumnFileDeleteRange : public ColumnFilePersisted
     {}
     ColumnFileDeleteRange(const ColumnFileDeleteRange &) = default;
 
-    ColumnFileReaderPtr getReader(const DMContext &, const IColumnFileDataProviderPtr &, const ColumnDefinesPtr &)
-        const override;
+    ColumnFileReaderPtr getReader(
+        const DMContext &,
+        const IColumnFileDataProviderPtr &,
+        const ColumnDefinesPtr &,
+        ReadTag) const override;
 
     const auto & getDeleteRange() { return delete_range; }
 
@@ -74,10 +76,4 @@ class ColumnFileDeleteRange : public ColumnFilePersisted
     String toString() const override { return "{delete_range:" + delete_range.toString() + "}"; }
 };
 
-class ColumnFileEmptyReader : public ColumnFileReader
-{
-public:
-    ColumnFileReaderPtr createNewReader(const ColumnDefinesPtr &) override;
-};
-} // namespace DM
-} // namespace DB
+} // namespace DB::DM

Diff for: dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileInMemory.cpp

@@ -59,7 +59,8 @@ void ColumnFileInMemory::fillColumns(const ColumnDefines & col_defs, size_t col_
 ColumnFileReaderPtr ColumnFileInMemory::getReader(
     const DMContext &,
     const IColumnFileDataProviderPtr &,
-    const ColumnDefinesPtr & col_defs) const
+    const ColumnDefinesPtr & col_defs,
+    ReadTag) const
 {
     return std::make_shared<ColumnFileInMemoryReader>(*this, col_defs);
 }
@@ -155,7 +156,7 @@ size_t ColumnFileInMemoryReader::skipNextBlock()
     return memory_file.getRows();
 }
 
-ColumnFileReaderPtr ColumnFileInMemoryReader::createNewReader(const ColumnDefinesPtr & new_col_defs)
+ColumnFileReaderPtr ColumnFileInMemoryReader::createNewReader(const ColumnDefinesPtr & new_col_defs, ReadTag)
 {
     // Reuse the cache data.
     return std::make_shared<ColumnFileInMemoryReader>(memory_file, new_col_defs, cols_data_cache);

Diff for: dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileInMemory.h

@@ -60,7 +60,7 @@ class ColumnFileInMemory : public ColumnFile
     Type getType() const override { return Type::INMEMORY_FILE; }
 
     size_t getRows() const override { return rows; }
-    size_t getBytes() const override { return bytes; };
+    size_t getBytes() const override { return bytes; }
 
     CachePtr getCache() { return cache; }
 
@@ -72,7 +72,8 @@ class ColumnFileInMemory : public ColumnFile
     ColumnFileReaderPtr getReader(
         const DMContext & context,
         const IColumnFileDataProviderPtr & data_provider,
-        const ColumnDefinesPtr & col_defs) const override;
+        const ColumnDefinesPtr & col_defs,
+        ReadTag) const override;
 
     bool isAppendable() const override { return !disable_append; }
     void disableAppend() override { disable_append = true; }
@@ -133,7 +134,7 @@ class ColumnFileInMemoryReader : public ColumnFileReader
 
     size_t skipNextBlock() override;
 
-    ColumnFileReaderPtr createNewReader(const ColumnDefinesPtr & new_col_defs) override;
+    ColumnFileReaderPtr createNewReader(const ColumnDefinesPtr & new_col_defs, ReadTag) override;
 };
 
 } // namespace DM