You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@tephra.apache.org by JamesRTaylor <gi...@git.apache.org> on 2018/05/04 22:40:24 UTC
[GitHub] incubator-tephra pull request #75: TEPHRA-287 ActionChange.getChangeKey() no...
GitHub user JamesRTaylor opened a pull request:
https://github.com/apache/incubator-tephra/pull/75
TEPHRA-287 ActionChange.getChangeKey() not implemented correctly
Based on a new config parameter (defaulting to true), send the old change set row key and the new change set row key. This allows a mix of old and new clients to still detect conflicts across them. Once it's know that only new clients are being used, the config parameter can be switched to false in which case only the new change set row key (which correctly embeds separator bytes to prevent false positives) will be sent.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/JamesRTaylor/incubator-tephra master
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-tephra/pull/75.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #75
----
commit 6b6e2303c1a6b7e2392ff63fa17b32879b4be67f
Author: James Taylor <jt...@...>
Date: 2018-05-04T22:33:39Z
TEPHRA-287 ActionChange.getChangeKey() not implemented correctly
----
---
[GitHub] incubator-tephra pull request #75: TEPHRA-287 ActionChange.getChangeKey() no...
Posted by poornachandra <gi...@git.apache.org>.
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/75#discussion_r186321022
--- Diff: tephra-core/src/main/java/org/apache/tephra/AbstractTransactionAwareTable.java ---
@@ -88,21 +93,105 @@ public void updateTx(Transaction tx) {
Collection<byte[]> txChanges = new TreeSet<byte[]>(UnsignedBytes.lexicographicalComparator());
for (Set<ActionChange> changeSet : changeSets.values()) {
for (ActionChange change : changeSet) {
- txChanges.add(getChangeKey(change.getRow(), change.getFamily(), change.getQualifier()));
+ byte[] row = change.getRow();
+ byte[] fam = change.getFamily();
+ byte[] qual = change.getQualifier();
+ txChanges.add(getChangeKey(row, fam, qual));
+ if (pre014ChangeSetKey) {
+ txChanges.add(getChangeKeyWithoutSeparators(row, fam, qual));
+ }
}
}
return txChanges;
}
+ /**
+ * @param vint long to make a vint of.
+ * @return long in vint byte array representation
+ * We could alternatively make this abstract and
+ * implement this method as Bytes.vintToBytes(long) in
+ * every compat module.
+ */
+ protected byte [] getVIntBytes(final long vint) {
+ long i = vint;
+ int size = WritableUtils.getVIntSize(i);
+ byte [] result = new byte[size];
+ int offset = 0;
+ if (i >= -112 && i <= 127) {
+ result[offset] = (byte) i;
+ return result;
+ }
+
+ int len = -112;
+ if (i < 0) {
+ i ^= -1L; // take one's complement'
+ len = -120;
+ }
+
+ long tmp = i;
+ while (tmp != 0) {
+ tmp = tmp >> 8;
+ len--;
+ }
+
+ result[offset++] = (byte) len;
+
+ len = (len < -120) ? -(len + 120) : -(len + 112);
+
+ for (int idx = len; idx != 0; idx--) {
+ int shiftbits = (idx - 1) * 8;
+ long mask = 0xFFL << shiftbits;
+ result[offset++] = (byte) ((i & mask) >> shiftbits);
+ }
+ return result;
+ }
+
+ /**
+ * The unique bytes identifying what is changing. We use the
+ * following structure:
+ * ROW conflict level: <table_name><0 byte separator><row key>
+ * since we know that table_name cannot contain a zero byte.
+ * COLUMN conflict level: <table_name><length of family as vint><family>
--- End diff --
Based on the implementation below this should be `COLUMN conflict level: <table_name>0<length of family as vint><family>...`
---
[GitHub] incubator-tephra pull request #75: TEPHRA-287 ActionChange.getChangeKey() no...
Posted by poornachandra <gi...@git.apache.org>.
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/75#discussion_r186321082
--- Diff: tephra-core/src/main/java/org/apache/tephra/AbstractTransactionAwareTable.java ---
@@ -88,21 +93,105 @@ public void updateTx(Transaction tx) {
Collection<byte[]> txChanges = new TreeSet<byte[]>(UnsignedBytes.lexicographicalComparator());
for (Set<ActionChange> changeSet : changeSets.values()) {
for (ActionChange change : changeSet) {
- txChanges.add(getChangeKey(change.getRow(), change.getFamily(), change.getQualifier()));
+ byte[] row = change.getRow();
+ byte[] fam = change.getFamily();
+ byte[] qual = change.getQualifier();
+ txChanges.add(getChangeKey(row, fam, qual));
+ if (pre014ChangeSetKey) {
+ txChanges.add(getChangeKeyWithoutSeparators(row, fam, qual));
+ }
}
}
return txChanges;
}
+ /**
+ * @param vint long to make a vint of.
+ * @return long in vint byte array representation
+ * We could alternatively make this abstract and
+ * implement this method as Bytes.vintToBytes(long) in
+ * every compat module.
+ */
+ protected byte [] getVIntBytes(final long vint) {
+ long i = vint;
+ int size = WritableUtils.getVIntSize(i);
+ byte [] result = new byte[size];
+ int offset = 0;
+ if (i >= -112 && i <= 127) {
+ result[offset] = (byte) i;
+ return result;
+ }
+
+ int len = -112;
+ if (i < 0) {
+ i ^= -1L; // take one's complement'
+ len = -120;
+ }
+
+ long tmp = i;
+ while (tmp != 0) {
+ tmp = tmp >> 8;
+ len--;
+ }
+
+ result[offset++] = (byte) len;
+
+ len = (len < -120) ? -(len + 120) : -(len + 112);
+
+ for (int idx = len; idx != 0; idx--) {
+ int shiftbits = (idx - 1) * 8;
+ long mask = 0xFFL << shiftbits;
+ result[offset++] = (byte) ((i & mask) >> shiftbits);
+ }
+ return result;
+ }
+
+ /**
+ * The unique bytes identifying what is changing. We use the
+ * following structure:
+ * ROW conflict level: <table_name><0 byte separator><row key>
+ * since we know that table_name cannot contain a zero byte.
+ * COLUMN conflict level: <table_name><length of family as vint><family>
+ * <length of qualifier as vint><qualifier><row>
+ * The last part of the change key does not need the length to be part
+ * of the key since there's nothing after it that may overlap with it.
+ * @param row
+ * @param family
+ * @param qualifier
+ * @return unique change key
+ */
public byte[] getChangeKey(byte[] row, byte[] family, byte[] qualifier) {
+ return getChangeKeyWithSeparators(row, family, qualifier);
+ }
+
+ private byte[] getChangeKeyWithSeparators(byte[] row, byte[] family, byte[] qualifier) {
+ byte[] key;
+ byte[] tableKey = getTableKey();
+ switch (conflictLevel) {
+ case ROW:
+ key = Bytes.concat(tableKey, SEPARATOR_BYTE_ARRAY, row);
+ break;
+ case COLUMN:
+ key = Bytes.concat(tableKey, SEPARATOR_BYTE_ARRAY, getVIntBytes(family.length), family,
+ getVIntBytes(qualifier.length), qualifier, row);
+ break;
+ case NONE:
+ throw new IllegalStateException("NONE conflict detection does not support change keys");
+ default:
+ throw new IllegalStateException("Unknown conflict detection level: " + conflictLevel);
+ }
+ return key;
+ }
+
+ private byte[] getChangeKeyWithoutSeparators(byte[] row, byte[] family, byte[] qualifier) {
byte[] key;
byte[] tableKey = getTableKey();
switch (conflictLevel) {
case ROW:
key = Bytes.concat(tableKey, row);
break;
case COLUMN:
- key = Bytes.concat(tableKey, row, family, qualifier);
+ key = Bytes.concat(tableKey, family, qualifier, row);
--- End diff --
Position of `row` in the change key has changed. Looks like this was unintentional?
---
[GitHub] incubator-tephra pull request #75: TEPHRA-287 ActionChange.getChangeKey() no...
Posted by JamesRTaylor <gi...@git.apache.org>.
Github user JamesRTaylor closed the pull request at:
https://github.com/apache/incubator-tephra/pull/75
---