Modify uniqness check in MVSecondaryIndex to use upper-bounded cursor

Push upper bound check from MVStoreCursor down to underlying map cursor
Propagate database compatibility mode down to Value comparison operation

h2/src/main/org/h2/mvstore/db/MVPlainTempResult.java

@@ -60,7 +60,7 @@ class MVPlainTempResult extends MVTempResult {
      */
     MVPlainTempResult(Database database, Expression[] expressions, int visibleColumnCount) {
         super(database, expressions.length, visibleColumnCount);
-        ValueDataType valueType = new ValueDataType(database.getCompareMode(), database, new int[columnCount]);
+        ValueDataType valueType = new ValueDataType(database, new int[columnCount]);
         Builder<Long, ValueArray> builder = new MVMap.Builder<Long, ValueArray>()
                                                 .valueType(valueType).singleWriter();
         map = store.openMap("tmp", builder);

h2/src/main/org/h2/mvstore/db/MVPrimaryIndex.java

@@ -51,9 +51,8 @@ public class MVPrimaryIndex extends BaseIndex {
         for (int i = 0; i < columns.length; i++) {
             sortTypes[i] = SortOrder.ASCENDING;
         }
-        ValueDataType keyType = new ValueDataType(null, null, null);
-        ValueDataType valueType = new ValueDataType(db.getCompareMode(), db,
-                sortTypes);
+        ValueDataType keyType = ValueDataType.INSTANCE;
+        ValueDataType valueType = new ValueDataType(db, sortTypes);
         mapName = "table." + getId();
         Transaction t = mvTable.getTransactionBegin();
         dataMap = t.openMap(mapName, keyType, valueType);

h2/src/main/org/h2/mvstore/db/MVSortedTempResult.java

@@ -166,12 +166,12 @@ class MVSortedTempResult extends MVTempResult {
             indexes = null;
         }
         this.indexes = indexes;
-        ValueDataType keyType = new ValueDataType(database.getCompareMode(), database, sortTypes);
+        ValueDataType keyType = new ValueDataType(database, sortTypes);
         Builder<ValueArray, Long> builder = new MVMap.Builder<ValueArray, Long>().keyType(keyType);
         map = store.openMap("tmp", builder);
         if (distinct && length != visibleColumnCount || distinctIndexes != null) {
             int count = distinctIndexes != null ? distinctIndexes.length : visibleColumnCount;
-            ValueDataType distinctType = new ValueDataType(database.getCompareMode(), database, new int[count]);
+            ValueDataType distinctType = new ValueDataType(database, new int[count]);
             Builder<ValueArray, Boolean> indexBuilder = new MVMap.Builder<ValueArray, Boolean>().keyType(distinctType);
             index = store.openMap("idx", indexBuilder);
         }

h2/src/main/org/h2/mvstore/db/MVSpatialIndex.java

@@ -93,7 +93,7 @@ public class MVSpatialIndex extends BaseIndex implements SpatialIndex, MVIndex {
             checkIndexColumnTypes(columns);
         }
         String mapName = "index." + getId();
-        ValueDataType vt = new ValueDataType(null, null, null);
+        ValueDataType vt = new ValueDataType(db, null);
         VersionedValue.Type valueType = new VersionedValue.Type(vt);
         MVRTreeMap.Builder<VersionedValue> mapBuilder =
                 new MVRTreeMap.Builder<VersionedValue>().

h2/src/main/org/h2/mvstore/db/MVTableEngine.java

@@ -160,9 +160,8 @@ public class MVTableEngine implements TableEngine {
                 if (!db.getSettings().reuseSpace) {
                     store.setReuseSpace(false);
                 }
-                this.transactionStore = new TransactionStore(
-                        store,
-                        new ValueDataType(db.getCompareMode(), db, null), db.getLockTimeout());
+                this.transactionStore = new TransactionStore(store,
+                        new ValueDataType(db, null), db.getLockTimeout());
             } catch (IllegalStateException e) {
                 throw convertIllegalStateException(e);
             }

h2/src/main/org/h2/mvstore/db/ValueDataType.java

@@ -13,6 +13,8 @@ import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.util.Arrays;
 import org.h2.api.ErrorCode;
+import org.h2.engine.Database;
+import org.h2.engine.Mode;
 import org.h2.message.DbException;
 import org.h2.mvstore.DataUtils;
 import org.h2.mvstore.WriteBuffer;
@@ -33,6 +35,7 @@ import org.h2.value.ValueBytes;
 import org.h2.value.ValueDate;
 import org.h2.value.ValueDecimal;
 import org.h2.value.ValueDouble;
+import org.h2.value.ValueEnum;
 import org.h2.value.ValueFloat;
 import org.h2.value.ValueGeometry;
 import org.h2.value.ValueInt;
@@ -73,12 +76,20 @@ public class ValueDataType implements DataType {
 
     final DataHandler handler;
     final CompareMode compareMode;
+    protected final Mode mode;
     final int[] sortTypes;
     SpatialDataType spatialType;
 
-    public ValueDataType(CompareMode compareMode, DataHandler handler,
+    public static ValueDataType INSTANCE = new ValueDataType(null, Mode.getRegular(), null, null);
+
+    public ValueDataType(Database database, int[] sortTypes) {
+        this(database.getCompareMode(), database.getMode(), database, sortTypes);
+    }
+
+    private ValueDataType(CompareMode compareMode, Mode mode, DataHandler handler,
             int[] sortTypes) {
         this.compareMode = compareMode;
+        this.mode = mode;
         this.handler = handler;
         this.sortTypes = sortTypes;
     }
@@ -103,7 +114,13 @@ public class ValueDataType implements DataType {
             int len = Math.min(al, bl);
             for (int i = 0; i < len; i++) {
                 int sortType = sortTypes == null ? SortOrder.ASCENDING : sortTypes[i];
-                int comp = compareValues(ax[i], bx[i], sortType);
+                Value one = ax[i];
+                Value two = bx[i];
+                if (one == null || two == null) {
+                    return compareValues(ax[len - 1], bx[len - 1], SortOrder.ASCENDING);
+                }
+
+                int comp = compareValues(one, two, sortType);
                 if (comp != 0) {
                     return comp;
                 }
@@ -136,7 +153,12 @@ public class ValueDataType implements DataType {
         if (aNull || bNull) {
             return SortOrder.compareNull(aNull, sortType);
         }
-        int comp = a.compareTypeSafe(b, compareMode);
+
+        int t2 = Value.getHigherOrder(a.getType(), b.getType());
+        String[] enumerators = ValueEnum.getEnumeratorsForBinaryOperation(a, b);
+        int comp = a.convertTo(t2, -1, mode, null, enumerators)
+                .compareTypeSafe(b.convertTo(t2, -1, mode, null, enumerators), compareMode);
+
         if ((sortType & SortOrder.DESCENDING) != 0) {
             comp = -comp;
         }

h2/src/main/org/h2/tools/Recover.java

@@ -622,12 +622,8 @@ public class Recover extends Tool implements DataHandler {
                 }
                 String tableId = mapName.substring("table.".length());
                 if (Integer.parseInt(tableId) == 0) {
-                    ValueDataType keyType = new ValueDataType(
-                            null, this, null);
-                    ValueDataType valueType = new ValueDataType(
-                            null, this, null);
                     TransactionMap<Value, Value> dataMap = store.begin().openMap(
-                            mapName, keyType, valueType);
+                            mapName, ValueDataType.INSTANCE, ValueDataType.INSTANCE);
                     Iterator<Value> dataIt = dataMap.keyIterator(null);
                     while (dataIt.hasNext()) {
                         Value rowId = dataIt.next();
@@ -660,12 +656,8 @@ public class Recover extends Tool implements DataHandler {
                 if (Integer.parseInt(tableId) == 0) {
                     continue;
                 }
-                ValueDataType keyType = new ValueDataType(
-                        null, this, null);
-                ValueDataType valueType = new ValueDataType(
-                        null, this, null);
                 TransactionMap<Value, Value> dataMap = store.begin().openMap(
-                        mapName, keyType, valueType);
+                        mapName, ValueDataType.INSTANCE, ValueDataType.INSTANCE);
                 Iterator<Value> dataIt = dataMap.keyIterator(null);
                 boolean init = false;
                 while (dataIt.hasNext()) {