some fix

apache · Sep 29, 2024 · 3b07d43 · 3b07d43
1 parent 7646c20
commit 3b07d43
Show file tree

Hide file tree

Showing 11 changed files with 58 additions and 68 deletions.
diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp
@@ -795,6 +795,7 @@ Status CloudMetaMgr::update_tmp_rowset(const RowsetMeta& rs_meta) {
 Status CloudMetaMgr::get_tmp_rowset(TabletSchemaSPtr tablet_schema, int64_t index_id,
                                     int64_t tablet_id, const std::vector<int64_t>& txn_ids,
                                     std::vector<std::shared_ptr<Rowset>>& rowsets) {
+    // see CloudMetaMgr::sync_tablet_rowsets, GetRowsetRequest
     VLOG_DEBUG << "get tmp rowset, tablet_id: " << tablet_id
                << ", txn_ids size: " << txn_ids.size();
     GetTmpRowsetRequest req;

diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp
@@ -108,42 +108,11 @@ Status CloudTablet::capture_rs_readers(const Version& spec_version,
     return capture_rs_readers_unlocked(version_path, rs_splits);
 }
 
-Status CloudTablet::capture_sub_txn_rs_readers(int64_t version,
-                                               const std::vector<int64_t>& sub_txn_ids,
-                                               std::vector<RowSetSplits>* rs_splits) {
-    LOG(INFO) << "capture_sub_txn_rs_readers for partition=" << partition_id()
-              << ", tablet_id=" << tablet_id() << ", version=" << version
-              << ", sub_txn_ids.size=" << sub_txn_ids.size();
-    std::vector<std::shared_ptr<Rowset>> rowsets;
-    RETURN_IF_ERROR(_engine.meta_mgr().get_tmp_rowset(tablet_schema(), index_id(), tablet_id(),
-                                                      sub_txn_ids, rowsets));
-    DCHECK(rowsets.size() == sub_txn_ids.size())
-            << " sub_txn_id size=" << sub_txn_ids.size() << ", rowset size=" << rowsets.size()
-            << ", partition_id=" << partition_id() << ", tablet=" << tablet_id();
-    for (int i = 0; i < rowsets.size(); ++i) {
-        auto& rowset = rowsets[i];
-        DCHECK(rowset != nullptr) << " rowset is nullptr for sub_txn_id=" << sub_txn_ids[i]
-                                  << ", partition_id=" << partition_id()
-                                  << ", tablet=" << tablet_id();
-        // see CloudMetaMgr::sync_tablet_rowsets, GetRowsetRequest
-        int64_t tmp_version = version + i + 1;
-        LOG(INFO) << "sub_txn_id=" << sub_txn_ids[i] << ", partition_id=" << partition_id()
-                  << ", tablet=" << tablet_id() << ", set tmp version=" << tmp_version;
-        rowset->set_version(Version(tmp_version, tmp_version));
-        if (rowset->rowset_meta()->has_delete_predicate()) {
-            rowset->rowset_meta()->mutable_delete_predicate()->set_version(tmp_version);
-        }
-        if (rowset != nullptr) {
-            RowsetReaderSharedPtr rs_reader;
-            auto res = rowset->create_reader(&rs_reader);
-            if (!res.ok()) {
-                return Status::Error<ErrorCode::CAPTURE_ROWSET_READER_ERROR>(
-                        "failed to create reader for rowset:{}", rowset->rowset_id().to_string());
-            }
-            rs_splits->emplace_back(std::move(rs_reader));
-        }
-    }
-    return Status::OK();
+Status CloudTablet::capture_sub_txn_rowsets(int64_t version,
+                                            const std::vector<int64_t>& sub_txn_ids,
+                                            std::vector<RowsetSharedPtr>* rowsets) {
+    return _engine.meta_mgr().get_tmp_rowset(tablet_schema(), index_id(), tablet_id(), sub_txn_ids,
+                                             *rowsets);
 }
 
 // There are only two tablet_states RUNNING and NOT_READY in cloud mode

diff --git a/be/src/cloud/cloud_tablet.h b/be/src/cloud/cloud_tablet.h
@@ -39,8 +39,8 @@ class CloudTablet final : public BaseTablet {
 
     Status capture_rs_readers(const Version& spec_version, std::vector<RowSetSplits>* rs_splits,
                               bool skip_missing_version) override;
-    Status capture_sub_txn_rs_readers(int64_t version, const std::vector<int64_t>& sub_txn_ids,
-                                      std::vector<RowSetSplits>* rs_splits) override;
+    Status capture_sub_txn_rowsets(int64_t version, const std::vector<int64_t>& sub_txn_ids,
+                                   std::vector<RowsetSharedPtr>* rowsets) override;
 
     Status capture_consistent_rowsets_unlocked(
             const Version& spec_version, std::vector<RowsetSharedPtr>* rowsets) const override;

diff --git a/be/src/olap/base_tablet.cpp b/be/src/olap/base_tablet.cpp
@@ -127,6 +127,42 @@ BaseTablet::~BaseTablet() {
     g_total_tablet_num << -1;
 }
 
+Status BaseTablet::capture_sub_txn_rs_readers(int64_t version,
+                                              const std::vector<int64_t>& sub_txn_ids,
+                                              std::vector<RowSetSplits>* rs_splits) {
+    LOG(INFO) << "capture_sub_txn_rs_readers for partition=" << partition_id()
+              << ", tablet_id=" << tablet_id() << ", version=" << version
+              << ", sub_txn_ids.size=" << sub_txn_ids.size();
+    std::vector<RowsetSharedPtr> rowsets;
+    RETURN_IF_ERROR(capture_sub_txn_rowsets(version, sub_txn_ids, &rowsets));
+    DCHECK(rowsets.size() == sub_txn_ids.size())
+            << " sub_txn_id size=" << sub_txn_ids.size() << ", rowset size=" << rowsets.size()
+            << ", partition_id=" << partition_id() << ", tablet=" << tablet_id();
+    for (int i = 0; i < rowsets.size(); ++i) {
+        auto& rowset = rowsets[i];
+        DCHECK(rowset != nullptr) << " rowset is nullptr for sub_txn_id=" << sub_txn_ids[i]
+                                  << ", partition_id=" << partition_id()
+                                  << ", tablet=" << tablet_id();
+        int64_t tmp_version = version + i + 1;
+        LOG(INFO) << "sub_txn_id=" << sub_txn_ids[i] << ", partition_id=" << partition_id()
+                  << ", tablet=" << tablet_id() << ", set tmp version=" << tmp_version;
+        rowset->set_version(Version(tmp_version, tmp_version));
+        if (rowset->rowset_meta()->has_delete_predicate()) {
+            rowset->rowset_meta()->mutable_delete_predicate()->set_version(tmp_version);
+        }
+        if (rowset != nullptr) {
+            RowsetReaderSharedPtr rs_reader;
+            auto res = rowset->create_reader(&rs_reader);
+            if (!res.ok()) {
+                return Status::Error<ErrorCode::CAPTURE_ROWSET_READER_ERROR>(
+                        "failed to create reader for rowset:{}", rowset->rowset_id().to_string());
+            }
+            rs_splits->emplace_back(std::move(rs_reader));
+        }
+    }
+    return Status::OK();
+}
+
 TabletSchemaSPtr BaseTablet::tablet_schema_with_merged_max_schema_version(
         const std::vector<RowsetMetaSharedPtr>& rowset_metas) {
     RowsetMetaSharedPtr max_schema_version_rs = *std::max_element(

diff --git a/be/src/olap/base_tablet.h b/be/src/olap/base_tablet.h
@@ -104,9 +104,11 @@ class BaseTablet {
                                       std::vector<RowSetSplits>* rs_splits,
                                       bool skip_missing_version) = 0;
 
-    virtual Status capture_sub_txn_rs_readers(int64_t version,
-                                              const std::vector<int64_t>& sub_txn_ids,
-                                              std::vector<RowSetSplits>* rs_splits) = 0;
+    Status capture_sub_txn_rs_readers(int64_t version, const std::vector<int64_t>& sub_txn_ids,
+                                      std::vector<RowSetSplits>* rs_splits);
+
+    virtual Status capture_sub_txn_rowsets(int64_t version, const std::vector<int64_t>& sub_txn_ids,
+                                           std::vector<RowsetSharedPtr>* rowsets) = 0;
 
     virtual size_t tablet_footprint() = 0;
 

diff --git a/be/src/olap/tablet.cpp b/be/src/olap/tablet.cpp
@@ -938,34 +938,16 @@ Status Tablet::capture_rs_readers(const Version& spec_version, std::vector<RowSe
     return Status::OK();
 }
 
-Status Tablet::capture_sub_txn_rs_readers(int64_t version, const std::vector<int64_t>& sub_txn_ids,
-                                          std::vector<RowSetSplits>* rs_splits) {
-    LOG(INFO) << "capture_sub_txn_rs_readers for partition=" << partition_id()
-              << ", tablet_id=" << tablet_id() << ", version=" << version
-              << ", sub_txn_ids.size=" << sub_txn_ids.size();
+Status Tablet::capture_sub_txn_rowsets(int64_t version, const std::vector<int64_t>& sub_txn_ids,
+                                       std::vector<RowsetSharedPtr>* rowsets) {
     for (int i = 0; i < sub_txn_ids.size(); ++i) {
         auto sub_txn_id = sub_txn_ids[i];
         auto rowset = _engine.txn_manager()->get_tablet_rowset(tablet_id(), tablet_uid(),
                                                                partition_id(), sub_txn_id);
         DCHECK(rowset != nullptr) << " rowset is nullptr for sub_txn_id=" << sub_txn_ids[i]
                                   << ", partition_id=" << partition_id()
                                   << ", tablet=" << tablet_id();
-        int64_t tmp_version = version + i + 1;
-        LOG(INFO) << "sub_txn_id=" << sub_txn_ids[i] << ", partition_id=" << partition_id()
-                  << ", tablet=" << tablet_id() << ", set tmp version=" << tmp_version;
-        rowset->set_version(Version(tmp_version, tmp_version));
-        if (rowset->rowset_meta()->has_delete_predicate()) {
-            rowset->rowset_meta()->mutable_delete_predicate()->set_version(tmp_version);
-        }
-        if (rowset != nullptr) {
-            RowsetReaderSharedPtr rs_reader;
-            auto res = rowset->create_reader(&rs_reader);
-            if (!res.ok()) {
-                return Status::Error<ErrorCode::CAPTURE_ROWSET_READER_ERROR>(
-                        "failed to create reader for rowset:{}", rowset->rowset_id().to_string());
-            }
-            rs_splits->emplace_back(std::move(rs_reader));
-        }
+        rowsets->push_back(rowset);
     }
     return Status::OK();
 }

diff --git a/be/src/olap/tablet.h b/be/src/olap/tablet.h
@@ -201,8 +201,8 @@ class Tablet final : public BaseTablet {
     // If skip_missing_version is true, skip versions if they are missing.
     Status capture_rs_readers(const Version& spec_version, std::vector<RowSetSplits>* rs_splits,
                               bool skip_missing_version) override;
-    Status capture_sub_txn_rs_readers(int64_t version, const std::vector<int64_t>& sub_txn_ids,
-                                      std::vector<RowSetSplits>* rs_splits) override;
+    Status capture_sub_txn_rowsets(int64_t version, const std::vector<int64_t>& sub_txn_ids,
+                                   std::vector<RowsetSharedPtr>* rowsets) override;
 
     // Find the missed versions until the spec_version.
     //

diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/TxnDeleteJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/TxnDeleteJob.java
@@ -44,7 +44,7 @@ public TxnDeleteJob(long id, long transactionId, String label, Map<Long, Short>
     @Override
     public long beginTxn() throws Exception {
         TransactionEntry txnEntry = ConnectContext.get().getTxnEntry();
-        this.transactionId = txnEntry.beginTransaction(targetTbl, SubTransactionType.DELETE);
+        this.transactionId = txnEntry.beginTransaction(targetTbl);
         this.label = txnEntry.getLabel();
         return this.transactionId;
     }

diff --git a/...main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapTxnInsertExecutor.java b/...main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapTxnInsertExecutor.java
@@ -56,7 +56,7 @@ public void beginTransaction() {
                 throw new AnalysisException("Transaction insert expect label " + txnEntry.getLabel()
                         + ", but got " + this.labelName);
             }
-            this.txnId = txnEntry.beginTransaction(table, SubTransactionType.INSERT);
+            this.txnId = txnEntry.beginTransaction(table);
             this.labelName = txnEntry.getLabel();
         } catch (Exception e) {
             throw new AnalysisException("begin transaction failed. " + e.getMessage(), e);

diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -3716,7 +3716,7 @@ public TQueryOptions toThrift() {
         tResult.setSkipDeletePredicate(skipDeletePredicate);
 
         tResult.setSkipDeleteBitmap(skipDeleteBitmap);
-        if (ConnectContext.get().isTxnModel()) {
+        if (ConnectContext.get() != null && ConnectContext.get().isTxnModel()) {
             // TODO set to true only if the sub txn ids are not empty
             tResult.setSkipDeleteBitmap(true);
         }

diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionEntry.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionEntry.java
@@ -204,7 +204,7 @@ public void setTxnSchemaVersion(int txnSchemaVersion) {
     }
 
     // Used for insert into select, return the sub_txn_id for this insert
-    public long beginTransaction(TableIf table, SubTransactionType subTransactionType) throws Exception {
+    public long beginTransaction(TableIf table) throws Exception {
         if (isInsertValuesTxnBegan()) {
             // FIXME: support mix usage of `insert into values` and `insert into select`
             throw new AnalysisException(