Repository: diennea/herddb (java)

Issue:
Add "PUT" operation - UPSERTS
Is this a question, a feature request or a bug report?

**FEATURE REQUEST**

1. Please describe the feature you are requesting.
In order to ease bulk loads it would be great to have the ability to perform an INSERT operation that does not fail.
In case you are trying to insert a record that is totally equivalent (same logical content) to an existing one no action is written to the WAL.
In case you are updating the value, an UPDATE operation will be written to the WAL.

  
2. Indicate the importance of this issue to you (blocker, must-have, should-have, nice-to-have). Are you currently using any workarounds to address this issue?

should-have


Pull request description:
Implement UPSERT
- implement basic experimental UPSERT support
- use UPSERT in HerdDB Collections framework
- the UPSERT follows exactly the same code path of INSERT, but in case of DuplicatePrimaryKey it turns the INSERT to a full UPDATE, of the full record


Fixes #579 

Patch:
diff --git a/herddb-collections/src/main/java/herddb/collections/TmpMapImpl.java b/herddb-collections/src/main/java/herddb/collections/TmpMapImpl.java
index 1e38d9b0..e745cdac 100644
--- a/herddb-collections/src/main/java/herddb/collections/TmpMapImpl.java
+++ b/herddb-collections/src/main/java/herddb/collections/TmpMapImpl.java
@@ -28,10 +28,8 @@ import herddb.core.stats.TableManagerStats;
 import herddb.index.PrimaryIndexSeek;
 import herddb.model.DataScanner;
 import herddb.model.DataScannerException;
-import herddb.model.DuplicatePrimaryKeyException;
 import herddb.model.GetResult;
 import herddb.model.Predicate;
-import herddb.model.Projection;
 import herddb.model.Record;
 import herddb.model.RecordFunction;
 import herddb.model.StatementEvaluationContext;
@@ -45,7 +43,6 @@ import herddb.model.commands.DropTableStatement;
 import herddb.model.commands.GetStatement;
 import herddb.model.commands.InsertStatement;
 import herddb.model.commands.ScanStatement;
-import herddb.model.commands.UpdateStatement;
 import herddb.utils.Bytes;
 import herddb.utils.DataAccessor;
 import herddb.utils.RawString;
@@ -66,11 +63,9 @@ class TmpMapImpl<K, V> implements TmpMap<K, V> {
     private final TableSpaceManager tableSpaceManager;
     private final TableManagerStats stats;
 
-    private final InsertStatement insert;
+    private final InsertStatement upsert;
     private final DeleteStatement delete;
-    private final UpdateStatement update;
     private final ScanStatement scan;
-    private final ScanStatement scanKeys;
 
     private static class PutStatementEvaluationContext<K, V> extends StatementEvaluationContext {
 
@@ -135,11 +130,9 @@ class TmpMapImpl<K, V> implements TmpMap<K, V> {
                 }
             }
         };
-        insert = new InsertStatement(TableSpace.DEFAULT, tmpTableName, keyFunction, valuesFunction);
-        update = new UpdateStatement(TableSpace.DEFAULT, tmpTableName, keyFunction, valuesFunction, null);
+        upsert = new InsertStatement(TableSpace.DEFAULT, tmpTableName, keyFunction, valuesFunction, true /*upsert*/);
         delete = new DeleteStatement(TableSpace.DEFAULT, tmpTableName, null, new PredicateDeleteImpl(keyFunction));
         scan = new ScanStatement(TableSpace.DEFAULT, table, null);
-        scanKeys = new ScanStatement(TableSpace.DEFAULT, table.name, Projection.PRIMARY_KEY(table), null, null, null);
         stats = tableSpaceManager
                 .getTableManager(tmpTableName)
                 .getStats();
@@ -155,14 +148,9 @@ class TmpMapImpl<K, V> implements TmpMap<K, V> {
     @Override
     public void put(K key, V value) throws CollectionsException {
         StatementEvaluationContext context = new PutStatementEvaluationContext(key, value);
-        // no concurrent access, no need to be super conservative, keep row-level locks..
+        // since 0.15 we are using upsert, that is an atomic operation
         try {
-            try {
-                // our best guess it that the key is not already mapped
-                tableSpaceManager.executeStatement(insert, context, TransactionContext.NO_TRANSACTION);
-            } catch (DuplicatePrimaryKeyException alreadyExists) {
-                tableSpaceManager.executeStatement(update, context, TransactionContext.NO_TRANSACTION);
-            }
+            tableSpaceManager.executeStatement(upsert, context, TransactionContext.NO_TRANSACTION);
         } catch (HerdDBInternalException err) {
             throw new CollectionsException(err);
         }
diff --git a/herddb-core/src/main/java/herddb/core/TableManager.java b/herddb-core/src/main/java/herddb/core/TableManager.java
index a55a42d2..51e284c0 100644
--- a/herddb-core/src/main/java/herddb/core/TableManager.java
+++ b/herddb-core/src/main/java/herddb/core/TableManager.java
@@ -1082,6 +1082,7 @@ public final class TableManager implements AbstractTableManager, Page.Owner {
 
         LockHandle lock = lockForWrite(key, transaction);
         CompletableFuture<StatementExecutionResult> res = null;
+        boolean fallbackToUpsert = false;
         if (transaction != null) {
             if (transaction.recordDeleted(table.name, key)) {
                 // OK, INSERT on a DELETED record inside this transaction
@@ -1089,13 +1090,27 @@ public final class TableManager implements AbstractTableManager, Page.Owner {
                 // ERROR, INSERT on a INSERTED record inside this transaction
                 res = FutureUtils.exception(new DuplicatePrimaryKeyException(key, "key " + key + ", decoded as " + RecordSerializer.deserializePrimaryKey(key, table) + ", already exists in table " + table.name + " inside transaction " + transaction.transactionId));
             } else if (keyToPage.containsKey(key)) {
-                res = FutureUtils.exception(new DuplicatePrimaryKeyException(key, "key " + key + ", decoded as " + RecordSerializer.deserializePrimaryKey(key, table) + ", already exists in table " + table.name + " during transaction " + transaction.transactionId));
+                if (insert.isUpsert()) {
+                    fallbackToUpsert = true;
+                } else {
+                    res = FutureUtils.exception(new DuplicatePrimaryKeyException(key, "key " + key + ", decoded as " + RecordSerializer.deserializePrimaryKey(key, table) + ", already exists in table " + table.name + " during transaction " + transaction.transactionId));
+                }
             }
         } else if (keyToPage.containsKey(key)) {
-            res = FutureUtils.exception(new DuplicatePrimaryKeyException(key, "key " + key + ", decoded as " + RecordSerializer.deserializePrimaryKey(key, table) + ", already exists in table " + table.name));
+            if (insert.isUpsert()) {
+                fallbackToUpsert = true;
+            } else {
+                res = FutureUtils.exception(new DuplicatePrimaryKeyException(key, "key " + key + ", decoded as " + RecordSerializer.deserializePrimaryKey(key, table) + ", already exists in table " + table.name));
+            }
         }
+
         if (res == null) {
-            LogEntry entry = LogEntryFactory.insert(table, key, Bytes.from_array(value), transaction);
+            LogEntry entry;
+            if (fallbackToUpsert) {
+                entry = LogEntryFactory.update(table, key, Bytes.from_array(value), transaction);
+            } else {
+                entry = LogEntryFactory.insert(table, key, Bytes.from_array(value), transaction);
+            }
             CommitLogResult pos = log.log(entry, entry.transactionId <= 0);
             res = pos.logSequenceNumber.thenApplyAsync((lsn) -> {
                 apply(pos, entry, false);
diff --git a/herddb-core/src/main/java/herddb/model/commands/InsertStatement.java b/herddb-core/src/main/java/herddb/model/commands/InsertStatement.java
index ed1e6d8a..190aa4be 100644
--- a/herddb-core/src/main/java/herddb/model/commands/InsertStatement.java
+++ b/herddb-core/src/main/java/herddb/model/commands/InsertStatement.java
@@ -30,21 +30,24 @@ import herddb.model.RecordFunction;
  *
  * @author enrico.olivelli
  */
-public class InsertStatement extends DMLStatement {
+public final class InsertStatement extends DMLStatement {
 
     private final RecordFunction keyFunction;
     private final RecordFunction valuesFunction;
+    private final boolean upsert;
 
     public InsertStatement(String tableSpace, String table, Record record) {
         super(table, tableSpace);
         this.keyFunction = new ConstValueRecordFunction(record.key);
         this.valuesFunction = new ConstValueRecordFunction(record.value);
+        this.upsert = false;
     }
 
-    public InsertStatement(String tableSpace, String table, RecordFunction keyFunction, RecordFunction function) {
+    public InsertStatement(String tableSpace, String table, RecordFunction keyFunction, RecordFunction function, boolean upsert) {
         super(table, tableSpace);
         this.keyFunction = keyFunction;
         this.valuesFunction = function;
+        this.upsert = upsert;
     }
 
     public RecordFunction getKeyFunction() {
@@ -55,4 +58,8 @@ public class InsertStatement extends DMLStatement {
         return valuesFunction;
     }
 
+    public boolean isUpsert() {
+        return upsert;
+    }
+
 }
diff --git a/herddb-core/src/main/java/herddb/model/planner/InsertOp.java b/herddb-core/src/main/java/herddb/model/planner/InsertOp.java
index e01870ab..57dfa71d 100644
--- a/herddb-core/src/main/java/herddb/model/planner/InsertOp.java
+++ b/herddb-core/src/main/java/herddb/model/planner/InsertOp.java
@@ -49,7 +49,6 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
-import java.util.logging.Level;
 import java.util.logging.Logger;
 import org.apache.bookkeeper.common.concurrent.FutureUtils;
 
@@ -59,12 +58,14 @@ public class InsertOp implements PlannerOp {
     private final String tableName;
     private final PlannerOp input;
     private final boolean returnValues;
+    private final boolean upsert;
 
-    public InsertOp(String tableSpace, String tableName, PlannerOp input, boolean returnValues) {
+    public InsertOp(String tableSpace, String tableName, PlannerOp input, boolean returnValues, boolean upsert) {
         this.tableSpace = tableSpace;
         this.tableName = tableName;
         this.input = input.optimize();
         this.returnValues = returnValues;
+        this.upsert = upsert;
     }
 
     @Override
@@ -126,7 +127,7 @@ public class InsertOp implements PlannerOp {
                 }
                 RecordFunction valuesfunction = new SQLRecordFunction(valuesColumns, table, valuesExpressions);
 
-                DMLStatement insertStatement = new InsertStatement(tableSpace, tableName, keyfunction, valuesfunction).setReturnValues(returnValues);
+                DMLStatement insertStatement = new InsertStatement(tableSpace, tableName, keyfunction, valuesfunction, upsert).setReturnValues(returnValues);
                 statements.add(insertStatement);
             }
             if (statements.isEmpty()) {
@@ -166,7 +167,6 @@ public class InsertOp implements PlannerOp {
                     }
                     long newTransactionId = res.transactionId;
                     if (current == statements.size()) {
-                        LOG.severe("multiinsert finished with tx " + newTransactionId + " and " + updateCounts + " recods");
                         DMLStatementExecutionResult finalDMLResult =
                                 new DMLStatementExecutionResult(newTransactionId, updateCounts.get(),
                                         lastKey.get(), lastNewValue.get());
@@ -175,7 +175,6 @@ public class InsertOp implements PlannerOp {
                     }
 
                     DMLStatement nextStatement = statements.get(current);
-                    LOG.log(Level.SEVERE, "executing # " + current + " newTx " + newTransactionId + " -" + nextStatement);
                     TransactionContext transactionContext = new TransactionContext(newTransactionId);
                     CompletableFuture<StatementExecutionResult> nextPromise =
                             tableSpaceManager.executeStatementAsync(nextStatement, context, transactionContext);
@@ -184,7 +183,6 @@ public class InsertOp implements PlannerOp {
             }
 
             DMLStatement firstStatement = statements.get(0);
-            LOG.log(Level.SEVERE, "executing first " + firstStatement);
             tableSpaceManager.executeStatementAsync(firstStatement, context, transactionContext)
                     .whenComplete(new ComputeNext(1));
 
diff --git a/herddb-core/src/main/java/herddb/sql/CalcitePlanner.java b/herddb-core/src/main/java/herddb/sql/CalcitePlanner.java
index 4c0a6bf5..496b730a 100644
--- a/herddb-core/src/main/java/herddb/sql/CalcitePlanner.java
+++ b/herddb-core/src/main/java/herddb/sql/CalcitePlanner.java
@@ -250,7 +250,8 @@ public class CalcitePlanner implements AbstractSQLPlanner {
             if (query.startsWith("EXPLAIN ")) {
                 query = query.substring("EXPLAIN ".length());
                 PlannerResult plan = runPlanner(defaultTableSpace, query);
-                PlannerOp finalPlan = convertRelNode(plan.topNode, plan.originalRowType, returnValues)
+                boolean upsert = detectUpsert(query);
+                PlannerOp finalPlan = convertRelNode(plan.topNode, plan.originalRowType, returnValues, upsert)
                         .optimize();
                 ValuesOp values = new ValuesOp(manager.getNodeId(),
                         new String[]{"name", "value"},
@@ -288,8 +289,9 @@ public class CalcitePlanner implements AbstractSQLPlanner {
             }
 
             PlannerResult plan = runPlanner(defaultTableSpace, query);
+            boolean upsert = detectUpsert(query);
             SQLPlannedOperationStatement sqlPlannedOperationStatement = new SQLPlannedOperationStatement(
-                    convertRelNode(plan.topNode, plan.originalRowType, returnValues)
+                    convertRelNode(plan.topNode, plan.originalRowType, returnValues, upsert)
                             .optimize());
             if (LOG.isLoggable(DUMP_QUERY_LEVEL)) {
                 LOG.log(DUMP_QUERY_LEVEL, "Query: {0} --HerdDB Plan\n{1}",
@@ -347,6 +349,12 @@ public class CalcitePlanner implements AbstractSQLPlanner {
         }
     }
 
+    private static boolean detectUpsert(String query) {
+        // unfortunately Calcite does not retain the UPSERT flag in LogicalTableModify
+        boolean upsert = query.startsWith("UPSERT");
+        return upsert;
+    }
+
 
     private TranslatedQuery calculateShowCreateTable(String query, String defaultTablespace, List<Object> parameters) {
         String[] items = {"SHOW", "CREATE", "TABLE"};
@@ -549,13 +557,13 @@ public class CalcitePlanner implements AbstractSQLPlanner {
 
     private PlannerOp convertRelNode(
             RelNode plan,
-            RelDataType rowType, boolean returnValues
+            RelDataType rowType, boolean returnValues, boolean upsert
     ) throws StatementExecutionException {
         if (plan instanceof EnumerableTableModify) {
             EnumerableTableModify dml = (EnumerableTableModify) plan;
             switch (dml.getOperation()) {
                 case INSERT:
-                    return planInsert(dml, returnValues);
+                    return planInsert(dml, returnValues, upsert);
                 case DELETE:
                     return planDelete(dml);
                 case UPDATE:
@@ -612,7 +620,8 @@ public class CalcitePlanner implements AbstractSQLPlanner {
 
     private PlannerOp planInsert(
             EnumerableTableModify dml,
-            boolean returnValues
+            boolean returnValues,
+            boolean upsert
     ) {
 
         final String tableSpace = dml.getTable().getQualifiedName().get(0);
@@ -669,7 +678,7 @@ public class CalcitePlanner implements AbstractSQLPlanner {
                             keyfunction = new SQLRecordKeyFunction(keyExpressionToColumn, keyValueExpression, table);
                         }
                         RecordFunction valuesfunction = new SQLRecordFunction(valuesColumns, table, valuesExpressions);
-                        statement = new InsertStatement(tableSpace, tableName, keyfunction, valuesfunction).setReturnValues(returnValues);
+                        statement = new InsertStatement(tableSpace, tableName, keyfunction, valuesfunction, upsert).setReturnValues(returnValues);
                     }
                 }
             }
@@ -678,17 +687,17 @@ public class CalcitePlanner implements AbstractSQLPlanner {
             return new SimpleInsertOp(statement);
         }
         PlannerOp input = convertRelNode(dml.getInput(),
-                null, false);
+                null, false, false);
 
         try {
-            return new InsertOp(tableSpace, tableName, input, returnValues);
+            return new InsertOp(tableSpace, tableName, input, returnValues, upsert);
         } catch (IllegalArgumentException err) {
             throw new StatementExecutionException(err);
         }
     }
 
     private PlannerOp planDelete(EnumerableTableModify dml) {
-        PlannerOp input = convertRelNode(dml.getInput(), null, false);
+        PlannerOp input = convertRelNode(dml.getInput(), null, false, false);
 
         final String tableSpace = dml.getTable().getQualifiedName().get(0);
         final String tableName = dml.getTable().getQualifiedName().get(1);
@@ -721,7 +730,7 @@ public class CalcitePlanner implements AbstractSQLPlanner {
     }
 
     private PlannerOp planUpdate(EnumerableTableModify dml, boolean returnValues) {
-        PlannerOp input = convertRelNode(dml.getInput(), null, false);
+        PlannerOp input = convertRelNode(dml.getInput(), null, false, false);
         List<String> updateColumnList = dml.getUpdateColumnList();
         List<RexNode> sourceExpressionList = dml.getSourceExpressionList();
         final String tableSpace = dml.getTable().getQualifiedName().get(0);
@@ -893,7 +902,7 @@ public class CalcitePlanner implements AbstractSQLPlanner {
     }
 
     private PlannerOp planProject(EnumerableProject op, RelDataType rowType) {
-        PlannerOp input = convertRelNode(op.getInput(), null, false);
+        PlannerOp input = convertRelNode(op.getInput(), null, false, false);
 
         final List<RexNode> projects = op.getProjects();
         final RelDataType _rowType = rowType == null ? op.getRowType() : rowType;
@@ -903,8 +912,8 @@ public class CalcitePlanner implements AbstractSQLPlanner {
 
     private PlannerOp planEnumerableHashJoin(EnumerableHashJoin op, RelDataType rowType) {
         // please note that EnumerableSemiJoin has a condition field which actually is not useful
-        PlannerOp left = convertRelNode(op.getLeft(), null, false);
-        PlannerOp right = convertRelNode(op.getRight(), null, false);
+        PlannerOp left = convertRelNode(op.getLeft(), null, false, false);
+        PlannerOp right = convertRelNode(op.getRight(), null, false, false);
         final JoinInfo analyzeCondition = op.analyzeCondition();
         int[] leftKeys = analyzeCondition.leftKeys.toIntArray();
         int[] rightKeys = analyzeCondition.rightKeys.toIntArray();
@@ -945,8 +954,8 @@ public class CalcitePlanner implements AbstractSQLPlanner {
 
     private PlannerOp planEnumerableJoin(EnumerableHashJoin op, RelDataType rowType) {
         // please note that EnumerableJoin has a condition field which actually is not useful
-        PlannerOp left = convertRelNode(op.getLeft(), null, false);
-        PlannerOp right = convertRelNode(op.getRight(), null, false);
+        PlannerOp left = convertRelNode(op.getLeft(), null, false, false);
+        PlannerOp right = convertRelNode(op.getRight(), null, false, false);
         final JoinInfo analyzeCondition = op.analyzeCondition();
         int[] leftKeys = analyzeCondition.leftKeys.toIntArray();
         int[] rightKeys = analyzeCondition.rightKeys.toIntArray();
@@ -971,8 +980,8 @@ public class CalcitePlanner implements AbstractSQLPlanner {
     }
 
     private PlannerOp planEnumerableNestedLoopJoin(EnumerableNestedLoopJoin op, RelDataType rowType) {
-        PlannerOp left = convertRelNode(op.getLeft(), null, false);
-        PlannerOp right = convertRelNode(op.getRight(), null, false);
+        PlannerOp left = convertRelNode(op.getLeft(), null, false, false);
+        PlannerOp right = convertRelNode(op.getRight(), null, false, false);
         CompiledSQLExpression condition = SQLExpressionCompiler.compileExpression(op.getCondition());
         final RelDataType _rowType = rowType == null ? op.getRowType() : rowType;
         List<RelDataTypeField> fieldList = _rowType.getFieldList();
@@ -991,8 +1000,8 @@ public class CalcitePlanner implements AbstractSQLPlanner {
 
     private PlannerOp planEnumerableMergeJoin(EnumerableMergeJoin op, RelDataType rowType) {
         // please note that EnumerableMergeJoin has a condition field which actually is not useful
-        PlannerOp left = convertRelNode(op.getLeft(), null, false);
-        PlannerOp right = convertRelNode(op.getRight(), null, false);
+        PlannerOp left = convertRelNode(op.getLeft(), null, false, false);
+        PlannerOp right = convertRelNode(op.getRight(), null, false, false);
         final JoinInfo analyzeCondition = op.analyzeCondition();
         int[] leftKeys = analyzeCondition.leftKeys.toIntArray();
         int[] rightKeys = analyzeCondition.rightKeys.toIntArray();
@@ -1094,7 +1103,7 @@ public class CalcitePlanner implements AbstractSQLPlanner {
     }
 
     private PlannerOp planSort(EnumerableSort op, RelDataType rowType) {
-        PlannerOp input = convertRelNode(op.getInput(), rowType, false);
+        PlannerOp input = convertRelNode(op.getInput(), rowType, false, false);
         RelCollation collation = op.getCollation();
         List<RelFieldCollation> fieldCollations = collation.getFieldCollations();
         boolean[] directions = new boolean[fieldCollations.size()];
@@ -1113,11 +1122,11 @@ public class CalcitePlanner implements AbstractSQLPlanner {
 
     private PlannerOp planInterpreter(EnumerableInterpreter op, RelDataType rowType, boolean returnValues) {
         // NOOP
-        return convertRelNode(op.getInput(), rowType, returnValues);
+        return convertRelNode(op.getInput(), rowType, returnValues, false);
     }
 
     private PlannerOp planLimit(EnumerableLimit op, RelDataType rowType) {
-        PlannerOp input = convertRelNode(op.getInput(), rowType, false);
+        PlannerOp input = convertRelNode(op.getInput(), rowType, false, false);
         CompiledSQLExpression maxRows = SQLExpressionCompiler.compileExpression(op.fetch);
         CompiledSQLExpression offset = SQLExpressionCompiler.compileExpression(op.offset);
         return new LimitOp(input, maxRows, offset);
@@ -1125,7 +1134,7 @@ public class CalcitePlanner implements AbstractSQLPlanner {
     }
 
     private PlannerOp planFilter(EnumerableFilter op, RelDataType rowType, boolean returnValues) {
-        PlannerOp input = convertRelNode(op.getInput(), rowType, returnValues);
+        PlannerOp input = convertRelNode(op.getInput(), rowType, returnValues, false);
         CompiledSQLExpression condition = SQLExpressionCompiler.compileExpression(op.getCondition());
         return new FilterOp(input, condition);
 
@@ -1137,7 +1146,7 @@ public class CalcitePlanner implements AbstractSQLPlanner {
         }
         List<PlannerOp> inputs = new ArrayList<>(op.getInputs().size());
         for (RelNode input : op.getInputs()) {
-            PlannerOp inputOp = convertRelNode(input, rowType, false).optimize();
+            PlannerOp inputOp = convertRelNode(input, rowType, false, false).optimize();
             inputs.add(inputOp);
         }
         return new UnionAllOp(inputs);
@@ -1168,7 +1177,7 @@ public class CalcitePlanner implements AbstractSQLPlanner {
             aggtypes[idaggcall++] = call.getAggregation().getName();
             argLists.add(call.getArgList());
         }
-        PlannerOp input = convertRelNode(op.getInput(), null, returnValues);
+        PlannerOp input = convertRelNode(op.getInput(), null, returnValues, false);
         return new AggregateOp(input, fieldnames, columns, aggtypes, argLists, groupedFiledsIndexes);
     }
 


Test patch:
diff --git a/herddb-core/src/test/java/herddb/core/SimpleRecoveryTest.java b/herddb-core/src/test/java/herddb/core/SimpleRecoveryTest.java
index 752852de..2d09fe9f 100644
--- a/herddb-core/src/test/java/herddb/core/SimpleRecoveryTest.java
+++ b/herddb-core/src/test/java/herddb/core/SimpleRecoveryTest.java
@@ -790,7 +790,7 @@ public class SimpleRecoveryTest {
             manager.executeStatement(st2, StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
             manager.checkpoint();
 
-            InsertStatement insert = new InsertStatement("tblspace1", "t1", new AutoIncrementPrimaryKeyRecordFunction(), new ConstValueRecordFunction(new byte[0]));
+            InsertStatement insert = new InsertStatement("tblspace1", "t1", new AutoIncrementPrimaryKeyRecordFunction(), new ConstValueRecordFunction(new byte[0]), false);
             DMLStatementExecutionResult insertResult = manager.executeUpdate(insert, StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
             assertEquals(1, insertResult.getUpdateCount());
             int newValue = insertResult.getKey().to_int();
@@ -815,7 +815,7 @@ public class SimpleRecoveryTest {
 
             manager.waitForTablespace("tblspace1", 10000);
 
-            InsertStatement insert = new InsertStatement("tblspace1", "t1", new AutoIncrementPrimaryKeyRecordFunction(), new ConstValueRecordFunction(new byte[0]));
+            InsertStatement insert = new InsertStatement("tblspace1", "t1", new AutoIncrementPrimaryKeyRecordFunction(), new ConstValueRecordFunction(new byte[0]), false);
             DMLStatementExecutionResult insertResult = manager.executeUpdate(insert, StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
             assertEquals(1, insertResult.getUpdateCount());
             int newValue = insertResult.getKey().to_int();
@@ -863,7 +863,7 @@ public class SimpleRecoveryTest {
             manager.executeStatement(st2, StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
             manager.checkpoint();
 
-            InsertStatement insert = new InsertStatement("tblspace1", "t1", new AutoIncrementPrimaryKeyRecordFunction(), new ConstValueRecordFunction(new byte[0]));
+            InsertStatement insert = new InsertStatement("tblspace1", "t1", new AutoIncrementPrimaryKeyRecordFunction(), new ConstValueRecordFunction(new byte[0]), false);
             DMLStatementExecutionResult insertResult = manager.executeUpdate(insert, StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
             assertEquals(1, insertResult.getUpdateCount());
             int newValue = insertResult.getKey().to_int();
@@ -886,7 +886,7 @@ public class SimpleRecoveryTest {
 
             manager.waitForTablespace("tblspace1", 10000);
 
-            InsertStatement insert = new InsertStatement("tblspace1", "t1", new AutoIncrementPrimaryKeyRecordFunction(), new ConstValueRecordFunction(new byte[0]));
+            InsertStatement insert = new InsertStatement("tblspace1", "t1", new AutoIncrementPrimaryKeyRecordFunction(), new ConstValueRecordFunction(new byte[0]), false);
             DMLStatementExecutionResult insertResult = manager.executeUpdate(insert, StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
             assertEquals(1, insertResult.getUpdateCount());
             int newValue = insertResult.getKey().to_int();
diff --git a/herddb-core/src/test/java/herddb/core/UpdateTest.java b/herddb-core/src/test/java/herddb/core/UpdateTest.java
index 9328b0cd..0de12527 100644
--- a/herddb-core/src/test/java/herddb/core/UpdateTest.java
+++ b/herddb-core/src/test/java/herddb/core/UpdateTest.java
@@ -27,15 +27,19 @@ import static herddb.core.TestUtils.executeUpdate;
 import static herddb.core.TestUtils.scan;
 import static herddb.model.TransactionContext.NO_TRANSACTION;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import herddb.mem.MemoryCommitLogManager;
 import herddb.mem.MemoryDataStorageManager;
 import herddb.mem.MemoryMetadataStorageManager;
+import herddb.model.DataScanner;
 import herddb.model.StatementEvaluationContext;
 import herddb.model.StatementExecutionException;
 import herddb.model.TransactionContext;
 import herddb.model.commands.CreateTableSpaceStatement;
+import herddb.utils.DataAccessor;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import org.junit.Test;
 
 /**
@@ -183,4 +187,71 @@ public class UpdateTest {
         }
 
     }
+
+     @Test
+    public void upsertTest() throws Exception {
+        String nodeId = "localhost";
+        try (DBManager manager = new DBManager("localhost", new MemoryMetadataStorageManager(), new MemoryDataStorageManager(), new MemoryCommitLogManager(), null, null)) {
+            manager.start();
+            CreateTableSpaceStatement st1 = new CreateTableSpaceStatement("tblspace1", Collections.singleton(nodeId), nodeId, 1, 0, 0);
+            manager.executeStatement(st1, StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
+            manager.waitForTablespace("tblspace1", 10000);
+
+            execute(manager, "CREATE TABLE tblspace1.tsql (k1 string primary key,n1 int,s1 string)", Collections.emptyList());
+
+            assertEquals(1, executeUpdate(manager, "INSERT INTO tblspace1.tsql(k1,n1) values(?,?)", Arrays.asList("mykey", Integer.valueOf(1234))).getUpdateCount());
+            try (DataScanner scan = scan(manager, "SELECT n1 from tblspace1.tsql where k1=?", Arrays.asList("mykey"))) {
+                List<DataAccessor> recordSet = scan.consumeAndClose();
+                assertEquals(1, recordSet.size());
+                assertEquals(1234, recordSet.get(0).get(0));
+            }
+            assertEquals(1, executeUpdate(manager, "UPSERT INTO tblspace1.tsql(k1,n1) values(?,?)", Arrays.asList("mykey", Integer.valueOf(1235))).getUpdateCount());
+            try (DataScanner scan = scan(manager, "SELECT n1 from tblspace1.tsql where k1=?", Arrays.asList("mykey"))) {
+                List<DataAccessor> recordSet = scan.consumeAndClose();
+                assertEquals(1, recordSet.size());
+                assertEquals(1235, recordSet.get(0).get(0));
+            }
+            assertEquals(4, executeUpdate(manager, "UPSERT INTO tblspace1.tsql(k1,n1)"
+                    + "values(?,?),(?,?),(?,?),(?,?)", Arrays.asList(
+                            "mykey", Integer.valueOf(1235),
+                            "mykey", Integer.valueOf(1236),
+                            "mykey", Integer.valueOf(1237),
+                            "mykey", Integer.valueOf(1238)
+                            )).getUpdateCount());
+            try (DataScanner scan = scan(manager, "SELECT n1 from tblspace1.tsql where k1=?", Arrays.asList("mykey"))) {
+                List<DataAccessor> recordSet = scan.consumeAndClose();
+                assertEquals(1, recordSet.size());
+                assertEquals(1238, recordSet.get(0).get(0));
+            }
+
+            assertEquals(1, executeUpdate(manager, "DELETE FROM tblspace1.tsql", Collections.emptyList()).getUpdateCount());
+            execute(manager, "ALTER TABLE tblspace1.tsql MODIFY s1 string not null", Collections.emptyList());
+
+            // assert that UPSERT fails
+            StatementExecutionException error =
+                    herddb.utils.TestUtils.expectThrows(StatementExecutionException.class, () ->  {
+                        executeUpdate(manager, "UPSERT INTO tblspace1.tsql(k1,n1) values(?,?)", Arrays.asList("mykey", Integer.valueOf(1235)));
+                    });
+            assertEquals("From line 1, column 13 to line 1, column 26: Column 's1' has no default value and does not allow NULLs", error.getMessage());
+
+            // insert a value, n1 has a value
+            assertEquals(1, executeUpdate(manager, "UPSERT INTO tblspace1.tsql(k1,n1,s1) values(?,?,'non-empty')", Arrays.asList("mykey", Integer.valueOf(1235))).getUpdateCount());
+
+            try (DataScanner scan = scan(manager, "SELECT n1 from tblspace1.tsql where k1=?", Arrays.asList("mykey"))) {
+                List<DataAccessor> recordSet = scan.consumeAndClose();
+                assertEquals(1, recordSet.size());
+                assertEquals(1235, recordSet.get(0).get(0));
+            }
+
+            // upsert, making n1 null now, because it has not been named in the INSERT clause
+            assertEquals(1, executeUpdate(manager, "UPSERT INTO tblspace1.tsql(k1,s1) values(?,'non-empty')", Arrays.asList("mykey")).getUpdateCount());
+
+            try (DataScanner scan = scan(manager, "SELECT n1 from tblspace1.tsql where k1=?", Arrays.asList("mykey"))) {
+                List<DataAccessor> recordSet = scan.consumeAndClose();
+                assertEquals(1, recordSet.size());
+                assertNull(recordSet.get(0).get(0));
+            }
+        }
+    }
+
 }