From afa2c768cdf2191b5e418db8e949f268c89cc48a Mon Sep 17 00:00:00 2001
From: Mahmoud Ismail <maism@hops.io>
Date: Mon, 26 Jun 2017 17:13:37 +0200
Subject: [PATCH] recovery bug fix

---
 include/TableTailer.h          | 3 +--
 src/FsMutationsTableTailer.cpp | 2 +-
 src/HopsworksOpsLogTailer.cpp  | 2 +-
 src/MetadataLogTailer.cpp      | 2 +-
 src/Notifier.cpp               | 4 +++-
 src/TableTailer.cpp            | 5 ++---
 6 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/include/TableTailer.h b/include/TableTailer.h
index fe6f2360..f66fda69 100644
--- a/include/TableTailer.h
+++ b/include/TableTailer.h
@@ -34,7 +34,6 @@ struct WatchTable{
     const NdbDictionary::Event::TableEvent* mWatchEvents;
     const int mNoEvents;
     const string mRecoveryIndex;
-    const string mRecoveryColumn;
 };
 
 class TableTailer {
@@ -50,7 +49,7 @@ class TableTailer {
     Ndb* mNdbConnection;
     
 private:
-    void recover(int recoverFromId);
+    void recover();
     void createListenerEvent();
     void removeListenerEvent();
     void waitForEvents();
diff --git a/src/FsMutationsTableTailer.cpp b/src/FsMutationsTableTailer.cpp
index 0e8b3352..8709f98a 100644
--- a/src/FsMutationsTableTailer.cpp
+++ b/src/FsMutationsTableTailer.cpp
@@ -42,7 +42,7 @@ const string _mutation_cols[_mutation_noCols]=
 const int _mutation_noEvents = 1; 
 const NdbDictionary::Event::TableEvent _mutation_events[_mutation_noEvents] = { NdbDictionary::Event::TE_INSERT };
 
-const WatchTable FsMutationsTableTailer::TABLE = {_mutation_table, _mutation_cols, _mutation_noCols , _mutation_events, _mutation_noEvents, _mutation_cols[2], _mutation_cols[2]};
+const WatchTable FsMutationsTableTailer::TABLE = {_mutation_table, _mutation_cols, _mutation_noCols , _mutation_events, _mutation_noEvents, _mutation_cols[2]};
 
 //const static ptime EPOCH_TIME(boost::gregorian::date(1970,1,1)); 
 
diff --git a/src/HopsworksOpsLogTailer.cpp b/src/HopsworksOpsLogTailer.cpp
index 69e44a8c..62b25949 100644
--- a/src/HopsworksOpsLogTailer.cpp
+++ b/src/HopsworksOpsLogTailer.cpp
@@ -44,7 +44,7 @@ const string _opslog_cols[_opslog_noCols]=
 const int _opslog_noEvents = 1; 
 const NdbDictionary::Event::TableEvent _opslog_events[_opslog_noEvents] = { NdbDictionary::Event::TE_INSERT};
 
-const WatchTable HopsworksOpsLogTailer::TABLE = {_opslog_table, _opslog_cols, _opslog_noCols , _opslog_events, _opslog_noEvents, "PRIMARY", "id"};
+const WatchTable HopsworksOpsLogTailer::TABLE = {_opslog_table, _opslog_cols, _opslog_noCols , _opslog_events, _opslog_noEvents, "PRIMARY"};
 
 const int OPS_ID_PK = 0;
 const int OPS_OP_ID = 1;
diff --git a/src/MetadataLogTailer.cpp b/src/MetadataLogTailer.cpp
index 7ea64741..42575605 100644
--- a/src/MetadataLogTailer.cpp
+++ b/src/MetadataLogTailer.cpp
@@ -40,7 +40,7 @@ const string _metalog_cols[_metalog_noCols]=
 const int _metalog_noEvents = 1; 
 const NdbDictionary::Event::TableEvent _metalog_events[_metalog_noEvents] = { NdbDictionary::Event::TE_INSERT };
 
-const WatchTable MetadataLogTailer::TABLE = {_metalog_table, _metalog_cols, _metalog_noCols , _metalog_events, _metalog_noEvents, "PRIMARY", _metalog_cols[0]};
+const WatchTable MetadataLogTailer::TABLE = {_metalog_table, _metalog_cols, _metalog_noCols , _metalog_events, _metalog_noEvents, "PRIMARY"};
 
 
 //Common
diff --git a/src/Notifier.cpp b/src/Notifier.cpp
index 7bcb42c4..4455e56f 100644
--- a/src/Notifier.cpp
+++ b/src/Notifier.cpp
@@ -59,7 +59,9 @@ void Notifier::start() {
     mFsMutationsBatcher->start();
     mFsMutationsTableTailer->start(mRecovery);
     
-    mMetadataLogTailer->start(mRecovery);
+    if(mMetadataType == Schemabased || mMetadataType == Schemaless || mMetadataType == Both){
+        mMetadataLogTailer->start(mRecovery);
+    }
     
     if (mMetadataType == Schemabased || mMetadataType == Both) {
         mSchemabasedMetadataBatcher->start();
diff --git a/src/TableTailer.cpp b/src/TableTailer.cpp
index 9ea21387..dfac5cb5 100644
--- a/src/TableTailer.cpp
+++ b/src/TableTailer.cpp
@@ -38,7 +38,7 @@ void TableTailer::start(bool recovery) {
     
     if(recovery){
         LOG_INFO("start with recovery for " << mTable.mTableName);
-        recover(0);
+        recover();
     }
     
     createListenerEvent();
@@ -46,7 +46,7 @@ void TableTailer::start(bool recovery) {
     mStarted = true;
 }
 
-void TableTailer::recover(int recoverFromId) {
+void TableTailer::recover() {
     const NdbDictionary::Dictionary* database = getDatabase(mNdbConnection);
     const NdbDictionary::Index* index = getIndex(database, mTable.mTableName, mTable.mRecoveryIndex);
     
@@ -54,7 +54,6 @@ void TableTailer::recover(int recoverFromId) {
     NdbIndexScanOperation* scanOp = getNdbIndexScanOperation(transaction, index);
     
     scanOp->readTuples(NdbOperation::LM_CommittedRead, NdbScanOperation::SF_OrderBy);
-    scanOp->setBound(mTable.mRecoveryColumn.c_str(), NdbIndexScanOperation::BoundLT, (char*) & recoverFromId);
     
     NdbRecAttr * row[mTable.mNoColumns];