HIVE-19089 : Create/Replicate AllocWriteId Event by maheshk114 · Pull Request #329 · apache/hive

maheshk114 · 2018-04-03T03:04:33Z

EVENT_ALLOCATE_WRITE_ID
Source Warehouse:

Create new event type EVENT_ALLOCATE_WRITE_ID with related message format etc.
Capture this event when allocate a table write ID from the sequence table by ACID operation.
Repl dump should read this event from EventNotificationTable and dump the message.
Target Warehouse:

Repl load should read the event from the dump and get the message.
Validate if source txn ID from the event is there in the source-target txn ID map. If not there, just noop the event.
If valid, then Allocate table write ID from sequence table
Extend listener notify event API to add two new parameter , dbconn and sqlgenerator to add the events to notification_log table within the same transaction

sankarh · 2018-04-04T13:44:21Z

        return 0;
+      case REPL_ALLOC_WRITE_ID:
+        List<Long> targetTxnIds = txnManager.replGetTargetTxnIds(replPolicy, work.getTxnIds());
+        txnManager.allocateTableWriteIdsBatch(targetTxnIds, work.getDbName(), work.getTableName());


Instead of 2 metastore calls, it can be just one method replAllocateTableWriteIdsBatch which takes replPolicy, srcTxnsIds and writeIds as input. If input is valid, then take repl flow and get targetTxnId and allocate writeId. In fact AllocateTableWriteIdsRequest can be modified to take optional repl arguments.

remove gettargettxn id and pass replpolicy, txntowriteid list in the request

sankarh · 2018-04-04T13:56:13Z

+    this.tableName = tableName;
+  }
+
+  public String getReplPolicy()  {


getReplPolicy can be a utility function which takes db and table name as input. This can be kept in a separate ql/parse/load/Utils.java class. This makes ReplTxnWork class very clean where we set db/table name only in constructor and replPolicy is just an input.

move the construction of repl policy to a utility class

sankarh · 2018-04-04T14:12:50Z

@@ -250,7 +250,8 @@ private boolean shouldReplayEvent(FileStatus dir, DumpType dumpType) throws Sema
    // repl id stored in the database/table is valid.


Do we really need this method shouldReplayEvent as all txn events are idempotent when it reaches metastore and able to find it from REPL_TO_TXN_MAP table?
Yes, this may help for the case where we repeat OPEN-ABORT/COMMIT events. But, even with this method, we cannot guarantee idempotent behaviour as db/table name can be null here. So, better to push this low level before allocating writeID or copy data files during commit. Also, need to check it on each tables if a particular commit operates on multiple tables.

remove this if check ..let it filer it for all operation

check if some checks can be done at lower layer based on table/partition object specially for write with commit and alloc write id events

sankarh · 2018-04-04T14:17:16Z

      if (txnIds != null && listeners != null && !listeners.isEmpty()) {
-        MetaStoreListenerNotifier.notifyEvent(listeners, EventType.OPEN_TXN,
-            new OpenTxnEvent(txnIds, this));
+        MetaStoreListenerNotifier.notifyTxnEvent(listeners, EventType.OPEN_TXN,


Shall use original notify() method itself if last 2 parameters are null. Also, add implementation for the same.

revert back this call to notifyevent . change notifyTxnEvent to notifyEventDirectSql and change the interface name also

sankarh · 2018-04-04T14:18:51Z

+      AllocateTableWriteIdsResponse response = getTxnHandler().allocateTableWriteIds(rqst);
+      if (listeners != null && !listeners.isEmpty()) {
+        MetaStoreListenerNotifier.notifyTxnEvent(listeners, EventType.ALLOC_WRITE_ID,
+                new AllocWriteIdEvent(rqst.getTxnIds(), rqst.getTableName(), this), null, null);


AllocateWriteIdEvent should take writeIds as well to ensure idempotent behaviour in target when apply. Better to pass TxnToWriteId list.
Also, shouldn't use the regular allocate writeid flow in target. For repl flow, need to check if the latest allocated writeid > requested writeid, then noop. If allocated writeid < requested writeid, then throw exception as it is invalid usage where they skipped some events.

ignore if its greater and fail load if its smaller

sankarh · 2018-04-04T14:21:37Z

+   *         map if no parameters were updated or if no listeners were notified.
+   * @throws MetaException If an error occurred while calling the listeners.
+   */
+  public static Map<String, String> notifyTxnEvent(List<? extends MetaStoreEventListener> listeners,


Shall name is more generic to denote directSQL implementation. I would suggest notifyEventWithDirectSql()

sankarh · 2018-04-04T14:31:34Z

+public class AllocWriteIdEvent extends ListenerEvent {
+
+  private final List<Long> txnIds;
+  private final String tableName;


Even database name should be part of the event, else, how can we filter this event for given database during REPL DUMP? Also, it may end of allocating write id on a table which was not intended to. For reference, Create table event takes DB name as input.

change it to txntowriteid list and add dbname also

sankarh · 2018-04-04T14:36:21Z

            (event.getEventType().equals(MessageFactory.COMMIT_TXN_EVENT)) ||
-            (event.getEventType().equals(MessageFactory.ABORT_TXN_EVENT))
+            (event.getEventType().equals(MessageFactory.ABORT_TXN_EVENT)) ||
+            (event.getEventType().equals(MessageFactory.ALLOC_WRITE_ID_EVENT))


Allocate write ID is a DB/table related event and should be ignored if not lookup on this DB/table.

remove this check for alloc write id

sankarh · 2018-04-09T05:56:20Z

+      tbl = context.db.getTable(dbName, tableName);
+    } catch (InvalidTableException e) {
+      // return a empty list if table does not exist.
+      return new ArrayList<>();


If the current dump have both create table and allocate write id events, then this table object would be missing in metastore. But this is a valid scenario and need to allocate write id. Shall refer to last repl ID if database object to decide this.

During load, by the time alloc write id is replayed ..create table should have been done.

move all these check to replTxntask

sankarh · 2018-04-09T05:59:36Z

+    //context table name is passed to ReplTxnWork , as the repl policy will be created based
+    //on this table name. ReplPolicy is used to extract the target transaction id based on source
+    //transaction id.
+    ReplTxnWork work = new ReplTxnWork(dbName, context.tableName, ReplTxnWork.OperationType.REPL_ALLOC_WRITE_ID,


We shall directly set tableName instead of setting it later. replPolicy can be passed as input.

can be done

sankarh · 2018-04-09T06:02:52Z

+    int numTxn = 50000;
+    List<Long> txnList = openTxnForTest(1, numTxn, "default.*");
+    assert(txnList.size() == numTxn);
+    txnHandler.abortTxns(new AbortTxnsRequest(txnList));


What are we testing here? No verification after abort.

the assert is added in the common function oopenTxnFortest

change the assert ..just count the number of entry added for repltotxnmap

sankarh · 2018-04-09T06:11:37Z

    }
+    allocMsg = new AllocateTableWriteIdsRequest("default1", "tbl1");
+    allocMsg.setReplPolicy("default1.*");
+    allocMsg.setTxnToWriteIdList(txnToWriteId);


Here, the input txnid should be that of source (default). Not target(default1) txn ids.

here the transactions are started as target only ..and then those txn ids are passed to alloc the write id

change the txn list to source and update the asserts

sankarh · 2018-04-09T06:17:41Z

+    allocMsg.setReplPolicy("default1.*");
+    allocMsg.setTxnToWriteIdList(txnToWriteId);
+    txnToWriteId1 = txnHandler.allocateTableWriteIds(allocMsg).getTxnToWriteIds();
+    assert (txnToWriteId.equals(txnToWriteId1));


The result may not match as txn id of source and target doesn't match. However, you can validate if txn id mapped to write id properly or not.

The txns are always target txn ids only

sankarh · 2018-04-09T06:46:45Z

    }
  }

+  private List<Long> getValidTxnIds(AllocateTableWriteIdsRequest rqst, Statement stmt, ResultSet rs)


The method name can be better such as checkAndGetTargetTxnIds.

can be done

sankarh · 2018-04-09T06:52:48Z

+
+    int size = txnToWriteIds.size();
+    for (TxnToWriteId txnToWriteId :  txnToWriteIds) {
+      Long targetTxnId = getTargetTxnId(rqst.getReplPolicy(), txnToWriteId.getTxnId(), stmt);


The implementation of getTargetTxnId can be changed to take list of source txn ids and use IN condition for queries.

in can not guarantee the input to output map ..we may not get the target txn id correctly

sankarh · 2018-04-09T06:53:15Z

 // Request msg to allocate table write ids for the given list of txns
 struct AllocateTableWriteIdsRequest {
-    1: required list<i64> txnIds,
+    1: optional list<i64> txnIds,


Optional members can be moved to bottom and keep repl related members together.

txnIds is not replication related ..it will be used by normal flow ..so have kept it at the starting

sankarh · 2018-04-09T06:54:05Z

    2: required string dbName,
    3: required string tableName,
+    4: optional string replPolicy,
+    5: optional list<TxnToWriteId> txnToWriteIdList,


rename as srcTxnToWriteIdList. Also, add a comment that txn and write ids in this list are assumed to be sorted in ascending order and are in continuous sequence.

sankarh · 2018-04-09T07:00:59Z

+
+    // Get the next write id for the table and check if its in consistent with the input txntowriteids.
+    String s = sqlGenerator.addForUpdateClause(
+            "select nwi_next from NEXT_WRITE_ID where nwi_database = " + quoteString(dbName)


Next_write_id is queried twice in this allocate write id flow. One here and other by caller after this call.
Add a comment, that this query is run under read-committed connection but still it is guaranteed that repeated query on next_write_id within this connection is consistent as this is repl flow and we assume that only repl task would allocate write id.
Also, check if we can refactor and avoid this repeated call.

sankarh · 2018-04-12T05:19:27Z

-      long txnId = txnList.get((int) i);
-      assertEquals(i + startId, txnId);
-    }
+    int numTxnPresentNow = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXNS");


Instead of 2 queries on TXNS table, shall add where clause for range of txnids allocated.

sankarh · 2018-04-12T05:21:31Z

+            " and RTM_REPL_POLICY = \'" + replPolicy + "\'");
+    assertEquals(numReplTxns, targetTxnId.size());
+
+    String[] output = TxnDbUtil.queryToString(conf, "select RTM_TARGET_TXN_ID from REPL_TXN_MAP where " +


Instead of 2 queries, can't we just compare the size of "output" for count?

sankarh · 2018-04-12T05:23:00Z

  public void testReplOpenTxn() throws Exception {
    int numTxn = 50000;
-    List<Long> txnList = openTxnForTest(1, numTxn, "default.*");
+    int numTxnPresentStart = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXNS");


Unused query result. Redundant code.

sankarh · 2018-04-12T05:27:08Z

    allocMsg = new AllocateTableWriteIdsRequest("default", "tbl");
    allocMsg.setReplPolicy("default.*");
-    allocMsg.setTxnToWriteIdList(txnToWriteId);
+    allocMsg.setSrcTxnToWriteIdList(txnToWriteId);


This is an invalid usage where write ids are allocated using normal flow first and then through repl flow. Also, the input txnToWriteId should be that of source but here it is that of target itself.
For idempotent case, we call same method twice with same input parameters.
I think, we shall remove this code here as idempotent case is verified somewhere below this test.

sankarh · 2018-04-12T05:35:22Z


    for (int idx = 0; idx < txnList.size(); idx++) {
-      txnToWriteId.add(new TxnToWriteId(txnList.get(idx), idx+1));
+      txnToWriteId.add(new TxnToWriteId(startTxnId+idx, idx+1));


Better to use appropriate variable names such as srcTxnToWriteId for input and targetTxnToWriteId for output.

sankarh · 2018-04-12T06:21:57Z

        }

-        Collections.sort(txnIds); //easier to read logs
+        Collections.sort(txnIds); //easier to read logs and for assumption done in replication flow


What assumption in replication flow?

sankarh · 2018-04-12T06:23:21Z

+   * @throws LockException
+   */
+  void replAllocateTableWriteIdsBatch(String dbName, String tableName, String replPolicy,
+                                      List<TxnToWriteId> txnToWriteIdList) throws LockException;


rename to srcTxnToWriteIdList.

sankarh · 2018-04-12T06:23:31Z

  }
  @Override
+  public void replAllocateTableWriteIdsBatch(String dbName, String tableName,
+                                         String replPolicy, List<TxnToWriteId> txnToWriteIdList) throws LockException {


rename to srcTxnToWriteIdList.

sankarh · 2018-04-12T06:23:42Z

+  }
+
+  public void replAllocateTableWriteIdsBatch(String dbName, String tableName,
+                                         String replPolicy, List<TxnToWriteId> txnToWriteIdList) throws LockException {


rename to srcTxnToWriteIdList.

sankarh · 2018-04-12T06:24:26Z

+    String tableName = (context.tableName != null && !context.tableName.isEmpty() ? context.tableName : msg
+            .getTableName());
+
+    //context table name is passed to ReplTxnWork , as the repl policy will be created based


Incorrect comment.

sankarh · 2018-04-13T04:40:07Z

-        LOG.debug("Table does not exist so, ignoring the operation");
-        //TODO : need to return if table does not exist. Will be done once acid table replication is implemented.
-        //return 0;
+        LOG.info("Table does not exist so, ignoring the operation");


Edit the comment to include that this could be idempotent case.

sankarh · 2018-04-13T04:41:16Z

+        ReplicationSpec replicationSpec = work.getReplicationSpec();
+        if (replicationSpec != null && !replicationSpec.allowReplacementInto(tbl.getParameters())) {
+          // if the event is already replayed, then no need to replay it again.
+          LOG.debug("ReplTxnTask: Event is skipped as it is already replayed");


Shall log the event type and event id.

sankarh · 2018-04-13T04:46:35Z

+        return true;
      }
      if (tbl != null) {
        Map<String, String> params = tbl.getParameters();


This code snippet to validate repl state is duplicated for database case as well. Can we have a static method in ReplicationSpec which takes current event id and parameters map as arguments?

sankarh · 2018-04-13T05:06:02Z

+  private String server, servicePrincipal, dbName, tableName;
+
+  @JsonProperty
+  private List<Long> txnIdList, writeIdList;


Can't we make list of TxnToWriteId in son format too? instead if 2 more lists.

json does not support TxnToWriteId

sankarh · 2018-04-13T05:06:26Z

+  }
+
+  @Override
+  public String getDbName() {


Why 2 methods to get dbName?

removed the autogenerated one

sankarh · 2018-04-13T05:07:17Z


 import javax.sql.DataSource;

+import com.google.common.collect.Lists;


Unused import as we changed all to Collections.singletonList, right?

sankarh · 2018-04-13T05:11:46Z

+          s = sqlGenerator.addForUpdateClause(
+                  "select nwi_next from NEXT_WRITE_ID where nwi_database = " + quoteString(dbName)
+                          + " and nwi_table = " + quoteString(tblName));
+          LOG.debug("Going to execute query <" + s + ">");


why do we need to move this select from bottom?

code refactoring to have one select per flow

sankarh · 2018-04-13T05:12:00Z

+
+        if (txnIds.size() == 0) {
+          // Idempotent case for replication flow.
+          return new AllocateTableWriteIdsResponse(new ArrayList<>());


txn leak. should rollback.

connection close will rollback the transaction

sankarh · 2018-04-13T05:13:07Z

@@ -1287,27 +1391,16 @@ public AllocateTableWriteIdsResponse allocateTableWriteIds(AllocateTableWriteIds
          return new AllocateTableWriteIdsResponse(txnToWriteIds);


txn leak here as well. need rollback.

connection close will rollback the transaction

sankarh · 2018-04-13T05:14:56Z

-    3: required string tableName,
+    1: required string dbName,
+    2: required string tableName,
+    3: optional list<i64> txnIds,


add a comment saying either txnIds or replPolicy+srcTxnToWriteIdList can exist in a call.

sankarh · 2018-04-13T05:16:58Z

-    return txnIds;
+  public ReplTxnWork(String replPolicy, String dbName, String tableName, Long txnId,
+                     OperationType type, ReplicationSpec replicationSpec) {
+    this(replPolicy, dbName, tableName, Lists.newArrayList(txnId), type, null, replicationSpec);


Singleton list can be used.

…rator parameter

maheshk114 changed the title ~~Create/Replicate AllocWriteId Event~~ HIVE-19089 : Create/Replicate AllocWriteId Event Apr 3, 2018

sankarh requested changes Apr 4, 2018

View reviewed changes

maheshk114 force-pushed the BUG-95522-95521 branch 2 times, most recently from 22b3335 to 09a5a19 Compare April 8, 2018 04:26

sankarh requested changes Apr 9, 2018

View reviewed changes

maheshk114 force-pushed the BUG-95522-95521 branch 2 times, most recently from 627b08c to 6f993ea Compare April 10, 2018 18:23

sankarh requested changes Apr 12, 2018

View reviewed changes

sankarh requested changes Apr 13, 2018

View reviewed changes

maheshk114 force-pushed the BUG-95522-95521 branch 5 times, most recently from 28442c3 to bf712f7 Compare April 13, 2018 08:17

maheshk114 added 6 commits April 14, 2018 12:58

HIVE-19089: Create/Replicate AllocWriteId Event

65befaf

HIVE-19089: Added new API to notify event with connection and sqlgene…

50b13cc

…rator parameter

HIVE-19089: Create/Replicate AllocWriteId Event : Review comment fixes

d51c85e

HIVE-19089: Create/Replicate AllocWriteId Event : Review comment fixes

f8cee95

HIVE-19089: Create/Replicate AllocWriteId Event : Review comment fixes-3

f2eee2b

HIVE-19089: Create/Replicate AllocWriteId Event : Review comment fixes-4

d7f7b75

maheshk114 force-pushed the BUG-95522-95521 branch from bf712f7 to d7f7b75 Compare April 14, 2018 07:29

maheshk114 closed this Apr 25, 2018

		@@ -250,7 +250,8 @@ private boolean shouldReplayEvent(FileStatus dir, DumpType dumpType) throws Sema
		// repl id stored in the database/table is valid.


		import javax.sql.DataSource;

		import com.google.common.collect.Lists;

		@@ -1287,27 +1391,16 @@ public AllocateTableWriteIdsResponse allocateTableWriteIds(AllocateTableWriteIds
		return new AllocateTableWriteIdsResponse(txnToWriteIds);

Conversation

maheshk114 commented Apr 3, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

maheshk114 Apr 5, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sankarh Apr 4, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

maheshk114 Apr 10, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

maheshk114 Apr 5, 2018 •

edited

Loading

sankarh Apr 4, 2018 •

edited

Loading

maheshk114 Apr 10, 2018 •

edited

Loading