Fix decode error when "NULL" value in the column with "primary key" flag (#5879)

dbms/src/Columns/ColumnAggregateFunction.cpp

@@ -83,11 +83,20 @@ void ColumnAggregateFunction::insertRangeFrom(const IColumn & from, size_t start
     const ColumnAggregateFunction & from_concrete = static_cast<const ColumnAggregateFunction &>(from);
 
     if (start + length > from_concrete.getData().size())
+<<<<<<< HEAD
         throw Exception("Parameters start = " + toString(start) + ", length = " + toString(length)
                 + " are out of bound in ColumnAggregateFunction::insertRangeFrom method"
                   " (data.size() = "
                 + toString(from_concrete.getData().size())
                 + ").",
+=======
+        throw Exception(
+            fmt::format(
+                "Parameters are out of bound in ColumnAggregateFunction::insertRangeFrom method, start={}, length={}, from.size()={}",
+                start,
+                length,
+                from_concrete.getData().size()),
+>>>>>>> 8e411ae86b (Fix decode error when "NULL" value in the column with "primary key" flag (#5879))
             ErrorCodes::PARAMETER_OUT_OF_BOUND);
 
     if (!empty() && src.get() != &from_concrete)

dbms/src/Columns/ColumnArray.cpp

@@ -383,7 +383,16 @@ void ColumnArray::insertRangeFrom(const IColumn & src, size_t start, size_t leng
     const ColumnArray & src_concrete = static_cast<const ColumnArray &>(src);
 
     if (start + length > src_concrete.getOffsets().size())
+<<<<<<< HEAD
         throw Exception("Parameter out of bound in ColumnArray::insertRangeFrom method.",
+=======
+        throw Exception(
+            fmt::format(
+                "Parameters are out of bound in ColumnArray::insertRangeFrom method, start={}, length={}, src.size()={}",
+                start,
+                length,
+                src_concrete.getOffsets().size()),
+>>>>>>> 8e411ae86b (Fix decode error when "NULL" value in the column with "primary key" flag (#5879))
             ErrorCodes::PARAMETER_OUT_OF_BOUND);
 
     size_t nested_offset = src_concrete.offsetAt(start);

dbms/src/Columns/ColumnDecimal.cpp

@@ -194,8 +194,17 @@ void ColumnDecimal<T>::insertRangeFrom(const IColumn & src, size_t start, size_t
     const ColumnDecimal & src_vec = static_cast<const ColumnDecimal &>(src);
 
     if (start + length > src_vec.data.size())
+<<<<<<< HEAD
         throw Exception("Parameters start = " + toString(start) + ", length = " + toString(length) +
             " are out of bound in ColumnDecimal<T>::insertRangeFrom method (data.size() = " + toString(src_vec.data.size()) + ").",
+=======
+        throw Exception(
+            fmt::format(
+                "Parameters are out of bound in ColumnDecimal<T>::insertRangeFrom method, start={}, length={}, src.size()={}",
+                start,
+                length,
+                src_vec.data.size()),
+>>>>>>> 8e411ae86b (Fix decode error when "NULL" value in the column with "primary key" flag (#5879))
             ErrorCodes::PARAMETER_OUT_OF_BOUND);
 
     size_t old_size = data.size();

dbms/src/Columns/ColumnFixedString.cpp

@@ -150,10 +150,19 @@ void ColumnFixedString::insertRangeFrom(const IColumn & src, size_t start, size_
     const ColumnFixedString & src_concrete = static_cast<const ColumnFixedString &>(src);
 
     if (start + length > src_concrete.size())
+<<<<<<< HEAD
         throw Exception("Parameters start = "
             + toString(start) + ", length = "
             + toString(length) + " are out of bound in ColumnFixedString::insertRangeFrom method"
             " (size() = " + toString(src_concrete.size()) + ").",
+=======
+        throw Exception(
+            fmt::format(
+                "Parameters are out of bound in ColumnFixedString::insertRangeFrom method, start={}, length={}, src.size()={}",
+                start,
+                length,
+                src_concrete.size()),
+>>>>>>> 8e411ae86b (Fix decode error when "NULL" value in the column with "primary key" flag (#5879))
             ErrorCodes::PARAMETER_OUT_OF_BOUND);
 
     size_t old_size = chars.size();

dbms/src/Columns/ColumnString.cpp

@@ -71,7 +71,16 @@ void ColumnString::insertRangeFrom(const IColumn & src, size_t start, size_t len
     const ColumnString & src_concrete = static_cast<const ColumnString &>(src);
 
     if (start + length > src_concrete.offsets.size())
+<<<<<<< HEAD
         throw Exception("Parameter out of bound in IColumnString::insertRangeFrom method.",
+=======
+        throw Exception(
+            fmt::format(
+                "Parameters are out of bound in ColumnString::insertRangeFrom method, start={}, length={}, src.size()={}",
+                start,
+                length,
+                src_concrete.size()),
+>>>>>>> 8e411ae86b (Fix decode error when "NULL" value in the column with "primary key" flag (#5879))
             ErrorCodes::PARAMETER_OUT_OF_BOUND);
 
     size_t nested_offset = src_concrete.offsetAt(start);

dbms/src/Columns/ColumnVector.cpp

@@ -155,10 +155,19 @@ void ColumnVector<T>::insertRangeFrom(const IColumn & src, size_t start, size_t
     const ColumnVector & src_vec = static_cast<const ColumnVector &>(src);
 
     if (start + length > src_vec.data.size())
+<<<<<<< HEAD
         throw Exception("Parameters start = "
             + toString(start) + ", length = "
             + toString(length) + " are out of bound in ColumnVector<T>::insertRangeFrom method"
             " (data.size() = " + toString(src_vec.data.size()) + ").",
+=======
+        throw Exception(
+            fmt::format(
+                "Parameters are out of bound in ColumnVector<T>::insertRangeFrom method, start={}, length={}, src.size()={}",
+                start,
+                length,
+                src_vec.data.size()),
+>>>>>>> 8e411ae86b (Fix decode error when "NULL" value in the column with "primary key" flag (#5879))
             ErrorCodes::PARAMETER_OUT_OF_BOUND);
 
     size_t old_size = data.size();

dbms/src/Storages/DeltaMerge/SSTFilesToBlockInputStream.h

@@ -0,0 +1,149 @@
+// Copyright 2022 PingCAP, Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <DataStreams/IBlockInputStream.h>
+#include <RaftStoreProxyFFI/ColumnFamily.h>
+#include <Storages/DeltaMerge/DMVersionFilterBlockInputStream.h>
+#include <Storages/Transaction/PartitionStreams.h>
+
+#include <memory>
+#include <string_view>
+
+namespace Poco
+{
+class Logger;
+}
+
+namespace DB
+{
+class TMTContext;
+class Region;
+using RegionPtr = std::shared_ptr<Region>;
+
+struct SSTViewVec;
+struct TiFlashRaftProxyHelper;
+struct SSTReader;
+class StorageDeltaMerge;
+
+namespace DM
+{
+struct ColumnDefine;
+using ColumnDefines = std::vector<ColumnDefine>;
+using ColumnDefinesPtr = std::shared_ptr<ColumnDefines>;
+
+// forward declaration
+class SSTFilesToBlockInputStream;
+using SSTFilesToBlockInputStreamPtr = std::shared_ptr<SSTFilesToBlockInputStream>;
+class BoundedSSTFilesToBlockInputStream;
+using BoundedSSTFilesToBlockInputStreamPtr = std::shared_ptr<BoundedSSTFilesToBlockInputStream>;
+
+class SSTFilesToBlockInputStream final : public IBlockInputStream
+{
+public:
+    SSTFilesToBlockInputStream(RegionPtr region_,
+                               const SSTViewVec & snaps_,
+                               const TiFlashRaftProxyHelper * proxy_helper_,
+                               DecodingStorageSchemaSnapshotConstPtr schema_snap_,
+                               Timestamp gc_safepoint_,
+                               bool force_decode_,
+                               TMTContext & tmt_,
+                               size_t expected_size_ = DEFAULT_MERGE_BLOCK_SIZE);
+    ~SSTFilesToBlockInputStream() override;
+
+    String getName() const override { return "SSTFilesToBlockInputStream"; }
+
+    Block getHeader() const override { return toEmptyBlock(*(schema_snap->column_defines)); }
+
+    void readPrefix() override;
+    void readSuffix() override;
+    Block read() override;
+
+public:
+    struct ProcessKeys
+    {
+        size_t default_cf = 0;
+        size_t write_cf = 0;
+        size_t lock_cf = 0;
+
+        inline size_t total() const { return default_cf + write_cf + lock_cf; }
+    };
+
+private:
+    void loadCFDataFromSST(ColumnFamilyType cf, const DecodedTiKVKey * rowkey_to_be_included);
+
+    Block readCommitedBlock();
+
+private:
+    RegionPtr region;
+    const SSTViewVec & snaps;
+    const TiFlashRaftProxyHelper * proxy_helper{nullptr};
+    DecodingStorageSchemaSnapshotConstPtr schema_snap;
+    TMTContext & tmt;
+    const Timestamp gc_safepoint;
+    size_t expected_size;
+    Poco::Logger * log;
+
+    using SSTReaderPtr = std::unique_ptr<SSTReader>;
+    SSTReaderPtr write_cf_reader;
+    SSTReaderPtr default_cf_reader;
+    SSTReaderPtr lock_cf_reader;
+
+    DecodedTiKVKey default_last_loaded_rowkey;
+    DecodedTiKVKey lock_last_loaded_rowkey;
+
+    friend class BoundedSSTFilesToBlockInputStream;
+
+    const bool force_decode;
+    bool is_decode_cancelled = false;
+
+    ProcessKeys process_keys;
+};
+
+// Bound the blocks read from SSTFilesToBlockInputStream by column `_tidb_rowid` and
+// do some calculation for the `DMFileWriter::BlockProperty` of read blocks.
+class BoundedSSTFilesToBlockInputStream final
+{
+public:
+    BoundedSSTFilesToBlockInputStream(SSTFilesToBlockInputStreamPtr child,
+                                      const ColId pk_column_id_,
+                                      const DecodingStorageSchemaSnapshotConstPtr & schema_snap);
+
+    String getName() const { return "BoundedSSTFilesToBlockInputStream"; }
+
+    void readPrefix();
+
+    void readSuffix();
+
+    Block read();
+
+    SSTFilesToBlockInputStream::ProcessKeys getProcessKeys() const;
+
+    RegionPtr getRegion() const;
+
+    // Return values: (effective rows, not clean rows, is delete rows, gc hint version)
+    std::tuple<size_t, size_t, size_t, UInt64> getMvccStatistics() const;
+
+private:
+    const ColId pk_column_id;
+
+    // Note that we only keep _raw_child for getting ingest info / process key, etc. All block should be
+    // read from `mvcc_compact_stream`
+    const SSTFilesToBlockInputStreamPtr _raw_child;
+    std::unique_ptr<DMVersionFilterBlockInputStream<DM_VERSION_FILTER_MODE_COMPACT>> mvcc_compact_stream;
+};
+
+} // namespace DM
+} // namespace DB