IGNITE-25016 Relax Mapper requirements and allow unmapped columns (#5564)

Author: AMashenkov

modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientCompute.java

@@ -364,7 +364,8 @@ private static <K, T, R> CompletableFuture<SubmitResult> executeColocatedObjectK
     ) {
         return executeColocatedInternal(
                 t,
-                (outputChannel, schema) -> ClientRecordSerializer.writeRecRaw(key, keyMapper, schema, outputChannel.out(), TuplePart.KEY),
+                (outputChannel, schema) ->
+                        ClientRecordSerializer.writeRecRaw(key, keyMapper, schema, outputChannel.out(), TuplePart.KEY, true),
                 ClientTupleSerializer.getPartitionAwarenessProvider(keyMapper, key),
                 descriptor,
                 arg

modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java

@@ -651,8 +651,8 @@ private HashMap<K, V> readGetAllResponse(ClientSchema schema, PayloadInputChanne
 
         var res = new LinkedHashMap<K, V>(cnt);
 
-        Marshaller keyMarsh = schema.getMarshaller(keySer.mapper(), TuplePart.KEY);
-        Marshaller valMarsh = schema.getMarshaller(valSer.mapper(), TuplePart.VAL);
+        Marshaller keyMarsh = schema.getMarshaller(keySer.mapper(), TuplePart.KEY, false);
+        Marshaller valMarsh = schema.getMarshaller(valSer.mapper(), TuplePart.VAL, false);
 
         for (int i = 0; i < cnt; i++) {
             // TODO: Optimize (IGNITE-16022).

modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java

@@ -85,9 +85,11 @@ Mapper<R> mapper() {
      * @param out Packer.
      * @param part Tuple part.
      * @param <R> Record type.
+     * @param allowUnmappedFields Allow unmapped fields.
      */
-    public static <R> void writeRecRaw(@Nullable R rec, Mapper<R> mapper, ClientSchema schema, ClientMessagePacker out, TuplePart part) {
-        writeRecRaw(rec, out, schema.getMarshaller(mapper, part), columnCount(schema, part));
+    public static <R> void writeRecRaw(@Nullable R rec, Mapper<R> mapper, ClientSchema schema, ClientMessagePacker out, TuplePart part,
+            boolean allowUnmappedFields) {
+        writeRecRaw(rec, out, schema.getMarshaller(mapper, part, allowUnmappedFields), columnCount(schema, part));
     }
 
     /**
@@ -99,7 +101,7 @@ public static <R> void writeRecRaw(@Nullable R rec, Mapper<R> mapper, ClientSche
      * @param columnCount Column count.
      * @param <R> Record type.
      */
-    static <R> void writeRecRaw(@Nullable R rec, ClientMessagePacker out, Marshaller marshaller, int columnCount) {
+    private static <R> void writeRecRaw(@Nullable R rec, ClientMessagePacker out, Marshaller marshaller, int columnCount) {
         var builder = new BinaryTupleBuilder(columnCount);
         var noValueSet = new BitSet();
 
@@ -109,8 +111,8 @@ static <R> void writeRecRaw(@Nullable R rec, ClientMessagePacker out, Marshaller
         out.packBinaryTuple(builder, noValueSet);
     }
 
-    void writeRecRaw(@Nullable R rec, ClientSchema schema, ClientMessagePacker out, TuplePart part) {
-        writeRecRaw(rec, mapper, schema, out, part);
+    private void writeRecRaw(@Nullable R rec, ClientSchema schema, ClientMessagePacker out, TuplePart part, boolean allowUnmappedFields) {
+        writeRecRaw(rec, mapper, schema, out, part, allowUnmappedFields);
     }
 
     void writeRec(
@@ -120,12 +122,24 @@ void writeRec(
             PayloadOutputChannel out,
             WriteContext ctx,
             TuplePart part
+    ) {
+        writeRec(tx, rec, schema, out, ctx, part, false);
+    }
+
+    void writeRec(
+            @Nullable Transaction tx,
+            @Nullable R rec,
+            ClientSchema schema,
+            PayloadOutputChannel out,
+            WriteContext ctx,
+            TuplePart part,
+            boolean allowUnmappedFields
     ) {
         out.out().packInt(tableId);
         writeTx(tx, out, ctx);
         out.out().packInt(schema.version());
 
-        writeRecRaw(rec, schema, out.out(), part);
+        writeRecRaw(rec, schema, out.out(), part, allowUnmappedFields);
     }
 
     void writeRecs(
@@ -141,7 +155,7 @@ void writeRecs(
         writeTx(tx, out, ctx);
         out.out().packInt(schema.version());
 
-        Marshaller marshaller = schema.getMarshaller(mapper, part);
+        Marshaller marshaller = schema.getMarshaller(mapper, part, false);
         int columnCount = columnCount(schema, part);
 
         writeRecRaw(rec, out.out(), marshaller, columnCount);
@@ -155,13 +169,25 @@ void writeRecs(
             PayloadOutputChannel out,
             WriteContext ctx,
             TuplePart part
+    ) {
+        writeRecs(tx, recs, schema, out, ctx, part, false);
+    }
+
+    void writeRecs(
+            @Nullable Transaction tx,
+            Collection<R> recs,
+            ClientSchema schema,
+            PayloadOutputChannel out,
+            WriteContext ctx,
+            TuplePart part,
+            boolean allowUnmappedFields
     ) {
         out.out().packInt(tableId);
         writeTx(tx, out, ctx);
         out.out().packInt(schema.version());
         out.out().packInt(recs.size());
 
-        Marshaller marshaller = schema.getMarshaller(mapper, part);
+        Marshaller marshaller = schema.getMarshaller(mapper, part, allowUnmappedFields);
         int columnCount = columnCount(schema, part);
 
         for (R rec : recs) {

modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java

@@ -83,7 +83,7 @@ public CompletableFuture<R> getAsync(@Nullable Transaction tx, R keyRec) {
 
         return tbl.doSchemaOutInOpAsync(
                 ClientOp.TUPLE_GET,
-                (s, w, n) -> ser.writeRec(tx, keyRec, s, w, n, TuplePart.KEY),
+                (s, w, n) -> ser.writeRec(tx, keyRec, s, w, n, TuplePart.KEY, true),
                 (s, r) -> ser.readValRec(keyRec, s, r.in()),
                 null,
                 ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keyRec),
@@ -105,7 +105,7 @@ public CompletableFuture<List<R>> getAllAsync(@Nullable Transaction tx, Collecti
 
         return tbl.doSchemaOutInOpAsync(
                 ClientOp.TUPLE_GET_ALL,
-                (s, w, n) -> ser.writeRecs(tx, keyRecs, s, w, n, TuplePart.KEY),
+                (s, w, n) -> ser.writeRecs(tx, keyRecs, s, w, n, TuplePart.KEY, true),
                 (s, r) -> ser.readRecs(s, r.in(), true, TuplePart.KEY_AND_VAL),
                 Collections.emptyList(),
                 ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keyRecs),
@@ -126,7 +126,7 @@ public CompletableFuture<Boolean> containsAsync(@Nullable Transaction tx, R key)
 
         return tbl.doSchemaOutOpAsync(
                 ClientOp.TUPLE_CONTAINS_KEY,
-                (s, w, n) -> ser.writeRec(tx, key, s, w, n, TuplePart.KEY),
+                (s, w, n) -> ser.writeRec(tx, key, s, w, n, TuplePart.KEY, true),
                 r -> r.in().unpackBoolean(),
                 ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), key),
                 tx);
@@ -149,7 +149,7 @@ public CompletableFuture<Boolean> containsAllAsync(@Nullable Transaction tx, Col
 
         return tbl.doSchemaOutOpAsync(
                 ClientOp.TUPLE_CONTAINS_ALL_KEYS,
-                (s, w, n) -> ser.writeRecs(tx, keys, s, w, n, TuplePart.KEY),
+                (s, w, n) -> ser.writeRecs(tx, keys, s, w, n, TuplePart.KEY, true),
                 r -> r.in().unpackBoolean(),
                 ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keys),
                 tx);
@@ -332,7 +332,7 @@ public CompletableFuture<Boolean> deleteAsync(@Nullable Transaction tx, R keyRec
 
         return tbl.doSchemaOutOpAsync(
                 ClientOp.TUPLE_DELETE,
-                (s, w, n) -> ser.writeRec(tx, keyRec, s, w, n, TuplePart.KEY),
+                (s, w, n) -> ser.writeRec(tx, keyRec, s, w, n, TuplePart.KEY, true),
                 r -> r.in().unpackBoolean(),
                 ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keyRec),
                 tx);
@@ -370,7 +370,7 @@ public CompletableFuture<R> getAndDeleteAsync(@Nullable Transaction tx, R keyRec
 
         return tbl.doSchemaOutInOpAsync(
                 ClientOp.TUPLE_GET_AND_DELETE,
-                (s, w, n) -> ser.writeRec(tx, keyRec, s, w, n, TuplePart.KEY),
+                (s, w, n) -> ser.writeRec(tx, keyRec, s, w, n, TuplePart.KEY, true),
                 (s, r) -> ser.readValRec(keyRec, s, r.in()),
                 null,
                 ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keyRec),
@@ -394,7 +394,7 @@ public CompletableFuture<List<R>> deleteAllAsync(@Nullable Transaction tx, Colle
 
         return tbl.doSchemaOutInOpAsync(
                 ClientOp.TUPLE_DELETE_ALL,
-                (s, w, n) -> ser.writeRecs(tx, keyRecs, s, w, n, TuplePart.KEY),
+                (s, w, n) -> ser.writeRecs(tx, keyRecs, s, w, n, TuplePart.KEY, true),
                 (s, r) -> ser.readRecs(s, r.in(), false, TuplePart.KEY),
                 Collections.emptyList(),
                 ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keyRecs),

modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientSchema.java

@@ -235,9 +235,9 @@ <T> Marshaller getMarshaller(Mapper mapper, TuplePart part, boolean allowUnmappe
             case KEY:
                 return marshallers.getKeysMarshaller(marshallerSchema(), mapper, true, allowUnmappedFields);
             case VAL:
-                return marshallers.getValuesMarshaller(marshallerSchema(), mapper, true, allowUnmappedFields);
+                return marshallers.getValuesMarshaller(marshallerSchema(), mapper, false, allowUnmappedFields);
             case KEY_AND_VAL:
-                return marshallers.getRowMarshaller(marshallerSchema(), mapper, true, allowUnmappedFields);
+                return marshallers.getRowMarshaller(marshallerSchema(), mapper, false, allowUnmappedFields);
             default:
                 throw new AssertionError("Unexpected tuple part: " + part);
         }

modules/client/src/test/java/org/apache/ignite/client/AbstractClientTableTest.java

@@ -219,6 +219,13 @@ protected static class CompositeKeyPojo {
         public long gid;
     }
 
+    /** Partial column set. */
+    protected static class IncompleteValPojo {
+        public byte zbyte;
+        public String zstring;
+        public byte[] zbytes;
+    }
+
     /** Columns of all types. */
     protected static class AllColumnsPojo {
         public long gid;

modules/client/src/test/java/org/apache/ignite/client/ClientKeyValueViewTest.java

@@ -93,35 +93,52 @@ public void testPrimitivePutBinaryGet() {
     }
 
     @Test
-    public void testMissingValueColumnsThrowException() {
+    public void testMissingValueColumns() {
         Table table = fullTable();
         KeyValueView<Tuple, Tuple> kvView = table.keyValueView();
-        KeyValueView<IncompletePojo, IncompletePojo> pojoView = table.keyValueView(IncompletePojo.class, IncompletePojo.class);
+        KeyValueView<CompositeKeyPojo, IncompleteValPojo> pojoView = table.keyValueView(CompositeKeyPojo.class, IncompleteValPojo.class);
 
         kvView.put(null, allColumnsTableKey(1), allColumnsTableVal("x", true));
 
-        var key = new IncompletePojo();
+        var key = new CompositeKeyPojo();
         key.id = "1";
         key.gid = 1;
 
-        Throwable e = assertThrowsWithCause(
-                () -> pojoView.get(null, key),
-                IgniteException.class,
-                "Failed to deserialize server response: No mapped object field found for column 'ZBOOLEAN'"
-        );
-        assertThat(Arrays.asList(e.getStackTrace()), anyOf(hasToString(containsString("ClientKeyValueView"))));
+        IncompleteValPojo res = pojoView.get(null, key);
+
+        assertEquals(11, res.zbyte);
+        assertEquals("x", res.zstring);
+        assertArrayEquals(new byte[]{1, 2}, res.zbytes);
+
+        pojoView.put(null, key, res);
+        Tuple val = kvView.get(null, allColumnsTableKey(1));
+
+        assertEquals(
+                Tuple.create().set("zbyte", (byte) 11).set("zstring", "x").set("zbytes", new byte[]{1, 2})
+                        .set("zboolean", null)
+                        .set("zshort", null)
+                        .set("zint", null)
+                        .set("zlong", null)
+                        .set("zfloat", null)
+                        .set("zdouble", null)
+                        .set("zdate", null)
+                        .set("ztime", null)
+                        .set("ztimestamp", null)
+                        .set("zdecimal", null)
+                        .set("zuuid", null),
+                val);
     }
 
     @Test
     public void testAllColumnsBinaryPutPojoGet() {
         Table table = fullTable();
-        KeyValueView<IncompletePojo, AllColumnsValPojo> pojoView = table.keyValueView(
-                Mapper.of(IncompletePojo.class),
+        KeyValueView<CompositeKeyPojo, AllColumnsValPojo> pojoView = table.keyValueView(
+                Mapper.of(CompositeKeyPojo.class),
                 Mapper.of(AllColumnsValPojo.class));
 
         table.recordView().upsert(null, allColumnsTableVal("foo", false));
 
-        var key = new IncompletePojo();
+        var key = new CompositeKeyPojo();
         key.gid = (int) (long) DEFAULT_ID;
         key.id = String.valueOf(DEFAULT_ID);
 
@@ -209,6 +226,28 @@ public void testMissingKeyColumnThrowsException() {
         assertThat(Arrays.asList(e.getStackTrace()), anyOf(hasToString(containsString("ClientKeyValueView"))));
     }
 
+    @Test
+    public void testExtraKeyColumnThrowsException() {
+        KeyValueView<IncompletePojo, AllColumnsValPojoNullable> pojoView = fullTable()
+                .keyValueView(IncompletePojo.class, AllColumnsValPojoNullable.class);
+
+        Throwable e = assertThrowsWithCause(
+                () -> pojoView.get(null, new IncompletePojo()),
+                IgniteException.class,
+                "Fields [zbyte, zbytes, zstring] of type org.apache.ignite.client.AbstractClientTableTest$IncompletePojo are not mapped"
+                        + " to columns"
+        );
+        assertThat(Arrays.asList(e.getStackTrace()), anyOf(hasToString(containsString("ClientKeyValueView"))));
+
+        e = assertThrowsWithCause(
+                () -> pojoView.put(null, new IncompletePojo(), new AllColumnsValPojoNullable()),
+                IgniteException.class,
+                "Fields [zbyte, zbytes, zstring] of type org.apache.ignite.client.AbstractClientTableTest$IncompletePojo are not mapped"
+                        + " to columns"
+        );
+        assertThat(Arrays.asList(e.getStackTrace()), anyOf(hasToString(containsString("ClientKeyValueView"))));
+    }
+
     @Test
     public void testNullablePrimitiveFields() {
         KeyValueView<CompositeKeyPojo, AllColumnsValPojoNullable> pojoView = fullTable().keyValueView(

modules/client/src/test/java/org/apache/ignite/client/ClientRecordViewTest.java

@@ -91,7 +91,7 @@ public void testPrimitivePutBinaryGet() {
     }
 
     @Test
-    public void testMissingValueColumnsThrowException() {
+    public void testMissingValueColumns() {
         Table table = fullTable();
         KeyValueView<Tuple, Tuple> kvView = table.keyValueView();
         RecordView<IncompletePojo> pojoView = table.recordView(IncompletePojo.class);
@@ -102,10 +102,11 @@ public void testMissingValueColumnsThrowException() {
         key.id = "1";
         key.gid = 1;
 
-        // This POJO does not have fields for all table columns, which is not allowed (to avoid unexpected data loss).
-        IgniteException ex = assertThrows(IgniteException.class, () -> pojoView.get(null, key));
-        assertEquals("Failed to deserialize server response: No mapped object field found for column 'ZBOOLEAN'", ex.getMessage());
-        assertThat(Arrays.asList(ex.getStackTrace()), anyOf(hasToString(containsString("ClientRecordView"))));
+        IncompletePojo res = pojoView.get(null, key);
+
+        assertEquals(11, res.zbyte);
+        assertEquals("x", res.zstring);
+        assertArrayEquals(new byte[]{1, 2}, res.zbytes);
     }
 
     @Test

modules/marshaller-common/src/main/java/org/apache/ignite/internal/marshaller/Marshaller.java

@@ -151,6 +151,10 @@ private static PojoMarshaller pojoMarshaller(
                 Set<String> fieldSet = new TreeSet<>(fields);
                 for (MarshallerColumn col : cols) {
                     String fieldName = mapper.fieldForColumn(col.name());
+                    if (fieldName == null) {
+                        assert !requireAllFields;
+                        continue;
+                    }
                     fieldSet.remove(fieldName);
                 }
 

modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientMarshallingTest.java

@@ -144,11 +144,16 @@ public void testKvMissingKeyPojoFields() {
 
     @Test
     public void testKvMissingValPojoFields() {
-        Table table = ignite().tables().table(TABLE_NAME);
-        var kvPojoView = table.keyValueView(Integer.class, MissingFieldPojo.class);
+        String tableName = "tableWithExtraField";
+        ignite().sql().execute(null, "CREATE TABLE " + tableName + " (KEY INT PRIMARY KEY, VAL VARCHAR, EXTRA VARCHAR)");
+        Table table = ignite().tables().table(tableName);
+
+        var kvPojoView = table.keyValueView(Integer.class, MissingFieldPojo2.class);
 
-        Throwable ex = assertThrowsWithCause(() -> kvPojoView.put(null, 1, new MissingFieldPojo()), IllegalArgumentException.class);
-        assertEquals("No mapped object field found for column 'VAL'", ex.getMessage());
+        kvPojoView.put(null, 1, new MissingFieldPojo2("x"));
+        MissingFieldPojo2 val = kvPojoView.get(null, 1);
+
+        assertEquals("x", val.val);
     }
 
     @Test
@@ -392,6 +397,18 @@ private static class MissingFieldPojo {
         public int unknown;
     }
 
+    private static class MissingFieldPojo2 {
+        public String val;
+
+        MissingFieldPojo2() {
+            // No-op.
+        }
+
+        public MissingFieldPojo2(String val) {
+            this.val = val;
+        }
+    }
+
     private static class IncompatibleFieldPojo {
         public String key; // Must be int.
         public BigDecimal val;

modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/reflection/KvMarshallerImpl.java

@@ -74,7 +74,7 @@ public KvMarshallerImpl(SchemaDescriptor schema, MarshallersProvider marshallers
 
         MarshallerSchema marshallerSchema = schema.marshallerSchema();
         keyMarsh = marshallers.getKeysMarshaller(marshallerSchema, keyMapper, true, false);
-        valMarsh = marshallers.getValuesMarshaller(marshallerSchema, valueMapper, true, false);
+        valMarsh = marshallers.getValuesMarshaller(marshallerSchema, valueMapper, false, false);
         keyPositions = schema.keyColumns().stream().mapToInt(Column::positionInRow).toArray();
         valPositions = schema.valueColumns().stream().mapToInt(Column::positionInRow).toArray();
     }