Fixes bug resolving lock w/ empty col fam. Reported in #660

modules/api/src/main/java/org/apache/fluo/api/data/Column.java

@@ -28,26 +28,42 @@
 public final class Column implements Comparable<Column>, Serializable {
 
   private static final long serialVersionUID = 1L;
-  private static final Bytes UNSET = Bytes.of(new byte[0]);
 
-  private Bytes family = UNSET;
-  private Bytes qualifier = UNSET;
-  private Bytes visibility = UNSET;
+  private final Bytes family;
+  private final Bytes qualifier;
+  private final Bytes visibility;
+
+  private final boolean isFamilySet;
+  private final boolean isQualifierSet;
+  private final boolean isVisibilitySet;
+
   private int hashCode = 0;
 
   public static final Column EMPTY = new Column();
 
   /**
    * Creates an empty Column where family, qualifier and visibility are not set
    */
-  public Column() {}
+  public Column() {
+    this.family = Bytes.EMPTY;
+    this.isFamilySet = false;
+    this.qualifier = Bytes.EMPTY;
+    this.isQualifierSet = false;
+    this.visibility = Bytes.EMPTY;
+    this.isVisibilitySet = false;
+  }
 
   /**
    * Creates Column with only a family.
    */
   public Column(Bytes family) {
     Objects.requireNonNull(family, "Family must not be null");
     this.family = family;
+    this.isFamilySet = true;
+    this.qualifier = Bytes.EMPTY;
+    this.isQualifierSet = false;
+    this.visibility = Bytes.EMPTY;
+    this.isVisibilitySet = false;
   }
 
   /**
@@ -64,7 +80,11 @@ public Column(Bytes family, Bytes qualifier) {
     Objects.requireNonNull(family, "Family must not be null");
     Objects.requireNonNull(qualifier, "Qualifier must not be null");
     this.family = family;
+    this.isFamilySet = true;
     this.qualifier = qualifier;
+    this.isQualifierSet = true;
+    this.visibility = Bytes.EMPTY;
+    this.isVisibilitySet = false;
   }
 
   /**
@@ -82,8 +102,11 @@ public Column(Bytes family, Bytes qualifier, Bytes visibility) {
     Objects.requireNonNull(qualifier, "Qualifier must not be null");
     Objects.requireNonNull(visibility, "Visibility must not be null");
     this.family = family;
+    this.isFamilySet = true;
     this.qualifier = qualifier;
+    this.isQualifierSet = true;
     this.visibility = visibility;
+    this.isVisibilitySet = true;
   }
 
   /**
@@ -98,16 +121,13 @@ public Column(CharSequence family, CharSequence qualifier, CharSequence visibili
    * Returns true if family is set
    */
   public boolean isFamilySet() {
-    return family != UNSET;
+    return isFamilySet;
   }
 
   /**
    * Retrieves Column Family (in Bytes). Returns Bytes.EMPTY if not set.
    */
   public Bytes getFamily() {
-    if (!isFamilySet()) {
-      return Bytes.EMPTY;
-    }
     return family;
   }
 
@@ -122,16 +142,13 @@ public String getsFamily() {
    * Returns true if qualifier is set
    */
   public boolean isQualifierSet() {
-    return qualifier != UNSET;
+    return isQualifierSet;
   }
 
   /**
    * Retrieves Column Qualifier (in Bytes). Returns Bytes.EMPTY if not set.
    */
   public Bytes getQualifier() {
-    if (!isQualifierSet()) {
-      return Bytes.EMPTY;
-    }
     return qualifier;
   }
 
@@ -146,16 +163,13 @@ public String getsQualifier() {
    * Returns true if visibility is set.
    */
   public boolean isVisibilitySet() {
-    return visibility != UNSET;
+    return isVisibilitySet;
   }
 
   /**
    * Retrieves Column Visibility (in Bytes). Returns Bytes.EMPTY if not set.
    */
   public Bytes getVisibility() {
-    if (!isVisibilitySet()) {
-      return Bytes.EMPTY;
-    }
     return visibility;
   }
 

modules/api/src/test/java/org/apache/fluo/api/data/ColumnTest.java

@@ -15,6 +15,9 @@
 
 package org.apache.fluo.api.data;
 
+import java.util.Arrays;
+import java.util.List;
+
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -73,6 +76,23 @@ public void testCreation() {
     Assert.assertEquals(Bytes.of("cq3"), col.getQualifier());
     Assert.assertEquals(Bytes.of("cv3"), col.getVisibility());
     Assert.assertEquals(new Column("cf3", "cq3", "cv3"), col);
+
+    // an empty string should always be considered as set, try diff combos of empty and non empty
+    // strings.
+    for (String fam : Arrays.asList("", "f")) {
+      for (String qual : Arrays.asList("", "q")) {
+        for (String vis : Arrays.asList("", "v")) {
+          col = new Column(fam, qual, vis);
+          Assert.assertTrue(col.isFamilySet());
+          Assert.assertTrue(col.isQualifierSet());
+          Assert.assertTrue(col.isVisibilitySet());
+          Assert.assertEquals(Bytes.of(fam), col.getFamily());
+          Assert.assertEquals(Bytes.of(qual), col.getQualifier());
+          Assert.assertEquals(Bytes.of(vis), col.getVisibility());
+          Assert.assertEquals(new Column(fam, qual, vis), col);
+        }
+      }
+    }
   }
 
   @Test

modules/integration-tests/src/main/java/org/apache/fluo/integration/impl/FailureIT.java

@@ -20,10 +20,12 @@
 
 import org.apache.accumulo.core.client.Scanner;
 import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.CompactionConfig;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.security.Authorizations;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.fluo.accumulo.format.FluoFormatter;
 import org.apache.fluo.accumulo.util.ColumnConstants;
 import org.apache.fluo.accumulo.util.ColumnType;
 import org.apache.fluo.accumulo.util.LongUtil;
@@ -32,6 +34,7 @@
 import org.apache.fluo.api.client.TransactionBase;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumn;
 import org.apache.fluo.api.exceptions.CommitException;
 import org.apache.fluo.api.exceptions.FluoException;
 import org.apache.fluo.api.observer.Observer;
@@ -652,4 +655,40 @@ private boolean wasRolledBackPrimary(long startTs, String rolledBackRow)
     return sawExpected;
   }
 
+  /*
+   * There was a bug where a locked column with an empty family could not be recovered.
+   */
+  @Test
+  public void testRecoverEmptyColumn() {
+    Column ecol = new Column("", "bal");
+
+    TestTransaction tx1 = new TestTransaction(env);
+
+    tx1.set("bob", ecol, "10");
+    tx1.set("joe", ecol, "20");
+    tx1.set("jill", ecol, "60");
+
+    tx1.done();
+
+    // partially commit a transaction, leaving the row 'joe' with a lock.
+    TestTransaction tx2 = new TestTransaction(env);
+    TestUtil.increment(tx2, "bob", ecol, 5);
+    TestUtil.increment(tx2, "joe", ecol, -5);
+
+    CommitData cd = tx2.createCommitData();
+    RowColumn primary = new RowColumn("bob", ecol);
+    Assert.assertTrue(tx2.preCommit(cd, primary));
+    Stamp commitTs = env.getSharedResources().getOracleClient().getStamp();
+    tx2.commitPrimaryColumn(cd, commitTs);
+    tx2.close();
+
+    // this transaction should roll forward the above transaction
+    TestTransaction tx3 = new TestTransaction(env);
+    Assert.assertEquals("15", tx3.gets("bob", ecol));
+    Assert.assertEquals("15", tx3.gets("joe", ecol));
+    Assert.assertEquals("60", tx3.gets("jill", ecol));
+    tx3.close();
+
+
+  }
 }