fi(CDAP-20794): use Spanner null values for Spanner DB when the field…

… value is null, also where clause when the keyRange is not prefixed

cdap-storage-ext-spanner/src/main/java/io/cdap/cdap/storage/spanner/SpannerStructuredTable.java

@@ -170,7 +170,7 @@ public Optional<StructuredRow> read(Collection<Field<?>> keys,
   public CloseableIterator<StructuredRow> scan(Range keyRange, int limit)
       throws InvalidFieldException {
     if (!isRangePrimaryKeys(keyRange)) {
-      // Spanner KeySet is requiring primary keys, instead we use SQL statement
+      // Spanner KeySet is requiring prefixed primary keys, instead we use SQL statement
       return multiScan(Collections.singleton(keyRange), limit);
     }
 
@@ -192,8 +192,8 @@ public CloseableIterator<StructuredRow> scan(Range keyRange, int limit)
 
   private boolean isRangePrimaryKeys(Range range) {
     try {
-      fieldValidator.validatePartialPrimaryKeys(range.getBegin());
-      fieldValidator.validatePartialPrimaryKeys(range.getEnd());
+      fieldValidator.validatePrimaryKeys(range.getBegin(), true);
+      fieldValidator.validatePrimaryKeys(range.getEnd(), true);
       return true;
     } catch (InvalidFieldException ex) {
       LOG.trace("Scanning non-primary key range: {}", range);
@@ -598,28 +598,24 @@ private Key createKey(Collection<Field<?>> fields) {
   /**
    * Converts a {@link Field} into spanner {@link Value}.
    */
-  @Nullable
   private Value getValue(Field<?> field) {
     Object value = field.getValue();
-    if (value == null) {
-      return null;
-    }
 
     switch (field.getFieldType()) {
       case INTEGER:
-        return Value.int64((int) value);
+        return Value.int64(value == null ? null : Long.valueOf((Integer) value));
       case LONG:
-        return Value.int64((long) value);
+        return Value.int64((Long) value);
       case FLOAT:
-        return Value.float64((float) value);
+        return Value.float64(value == null ? null : Double.valueOf((Float) value));
       case DOUBLE:
-        return Value.float64((double) value);
+        return Value.float64((Double) value);
       case STRING:
         return Value.string((String) value);
       case BYTES:
-        return Value.bytes(ByteArray.copyFrom((byte[]) value));
+        return Value.bytes(value == null ? null : (ByteArray.copyFrom((byte[]) value)));
       case BOOLEAN:
-        return Value.bool((boolean) value);
+        return Value.bool((Boolean) value);
     }
 
     // This shouldn't happen

cdap-storage-spi/src/test/java/io/cdap/cdap/spi/data/StructuredTableTest.java

@@ -196,6 +196,42 @@ public void testMultiRead() throws Exception {
     Assert.assertEquals(new HashSet<>(keys), result);
   }
 
+  @Test
+  public void testWritingNullFields() throws Exception {
+    int max = 100;
+
+    // Write rows with null field values
+    for (int i = max - 1; i >= 0; i--) {
+      List<Field<?>> fields = Arrays.asList(Fields.intField(KEY, i),
+          Fields.longField(KEY2, (long) i),
+          Fields.stringField(KEY3, "new key3"),
+          Fields.stringField(STRING_COL, VAL + i),
+          Fields.doubleField(DOUBLE_COL, null),
+          Fields.floatField(FLOAT_COL, null),
+          Fields.bytesField(BYTES_COL, null));
+      getTransactionRunner().run(context -> {
+        StructuredTable table = context.getTable(SIMPLE_TABLE);
+        table.upsert(fields);
+      });
+    }
+
+    // Read all of them back using multiRead
+    Collection<Collection<Field<?>>> keys = new ArrayList<>();
+    for (int i = 0; i < max; i++) {
+      keys.add(Arrays.asList(Fields.intField(KEY, i),
+          Fields.longField(KEY2, (long) i),
+          Fields.stringField(KEY3, "new key3")));
+    }
+    // There is no particular ordering of the result, hence use a set to compare
+    Set<Collection<Field<?>>> result = TransactionRunners.run(getTransactionRunner(), context -> {
+      StructuredTable table = context.getTable(SIMPLE_TABLE);
+      return new HashSet<>(convertRowsToFields(table.multiRead(keys).iterator(), Arrays.asList(KEY, KEY2, KEY3)));
+    });
+
+    Assert.assertEquals(100, result.size());
+    Assert.assertEquals(new HashSet<>(keys), result);
+  }
+
   @Test
   public void testSimpleScan() throws Exception {
     int max = 100;
@@ -244,6 +280,17 @@ public void testSimpleScan() throws Exception {
                      Range.Bound.EXCLUSIVE), max);
     Assert.assertEquals(expected.subList(5, 15), actual);
 
+    // Test partial keys actual range scan with mixed type
+    actual =
+        scanSimpleStructuredRows(
+            Range.create(
+                Collections.singleton(Fields.intField(KEY, 5)),
+                Range.Bound.INCLUSIVE,
+                Arrays.asList(Fields.intField(KEY, 15), Fields.stringField(KEY3, "key3")),
+                Range.Bound.INCLUSIVE),
+            max);
+    Assert.assertEquals(expected.subList(5, 16), actual);
+
     // Test begin only range
     actual =
       scanSimpleStructuredRows(