IGNITE-28556 Calcite. Support wrap_key and wrap_value flags

modules/calcite/src/main/codegen/config.fmpp

@@ -53,6 +53,8 @@ data: {
       "DATA_REGION"
 #     "KEY_TYPE" // already presented in Calcite
       "VALUE_TYPE"
+      "WRAP_KEY"
+      "WRAP_VALUE"
       "ENCRYPTED"
       "INDEX"
       "PARALLEL"
@@ -88,6 +90,8 @@ data: {
       "DATA_REGION"
 #     "KEY_TYPE" // already presented in Calcite
       "VALUE_TYPE"
+      "WRAP_KEY"
+      "WRAP_VALUE"
       "ENCRYPTED"
       "PARALLEL"
       "INLINE_SIZE"

modules/calcite/src/main/codegen/includes/parserImpls.ftl

@@ -78,6 +78,10 @@ SqlLiteral CreateTableOptionKey() :
     <VALUE_TYPE> { return SqlLiteral.createSymbol(IgniteSqlCreateTableOptionEnum.VALUE_TYPE, getPos()); }
 |
     <ENCRYPTED> { return SqlLiteral.createSymbol(IgniteSqlCreateTableOptionEnum.ENCRYPTED, getPos()); }
+|
+    <WRAP_KEY> { return SqlLiteral.createSymbol(IgniteSqlCreateTableOptionEnum.WRAP_KEY, getPos()); }
+|
+    <WRAP_VALUE> { return SqlLiteral.createSymbol(IgniteSqlCreateTableOptionEnum.WRAP_VALUE, getPos()); }
 }
 
 void CreateTableOption(List<SqlNode> list) :

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ddl/DdlCommandHandler.java

@@ -68,6 +68,10 @@
 import static org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal.SAVEPOINTS_EXPLICIT_TX_ONLY;
 import static org.apache.ignite.internal.processors.query.QueryUtils.convert;
 import static org.apache.ignite.internal.processors.query.QueryUtils.isDdlOnSchemaSupported;
+import static org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlCreateTableOptionEnum.KEY_TYPE;
+import static org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlCreateTableOptionEnum.VALUE_TYPE;
+import static org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlCreateTableOptionEnum.WRAP_KEY;
+import static org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlCreateTableOptionEnum.WRAP_VALUE;
 
 /** */
 public class DdlCommandHandler {
@@ -172,6 +176,8 @@
             throw new SchemaOperationException(SchemaOperationException.CODE_TABLE_EXISTS, cmd.tableName());
         }
 
+        checkKVWrappedParam(cmd);
+
         CacheConfiguration<?, ?> ccfg = new CacheConfiguration<>(cmd.tableName());
 
         QueryEntity e = toQueryEntity(cmd);
@@ -213,6 +219,42 @@
         }
     }
 
+    /** */
+    private void checkKVWrappedParam(CreateTableCommand cmd) {
+        Boolean wrapKey = cmd.wrapKey();
+
+        if (wrapKey != null) {
+            if (!wrapKey) {
+                if (cmd.primaryKeyColumns().size() > 1) {
+                    throw new IgniteSQLException(WRAP_KEY + " parameter cannot be \"false\" when composite primary key exists.",
+                        IgniteQueryErrorCode.PARSING);
+                }
+
+                if (!F.isEmpty(cmd.keyTypeName())) {
+                    throw new IgniteSQLException(WRAP_KEY + " parameter cannot be \"false\" when " + KEY_TYPE + " is defined.",
+                        IgniteQueryErrorCode.PARSING);
+                }
+            }
+        }
+        else
+            cmd.wrapKey(!F.isEmpty(cmd.keyTypeName()) || cmd.primaryKeyColumns().size() > 1);
+
+        Boolean wrapVal = cmd.wrapValue();
+
+        if (wrapVal != null && !wrapVal) {
+            if (!cmd.primaryKeyColumns().isEmpty() && cmd.columns().size() - cmd.primaryKeyColumns().size() > 1) {
+                throw new IgniteSQLException(WRAP_VALUE + " parameter cannot be \"false\" with multiple columns.",
+                    IgniteQueryErrorCode.PARSING);
+            }
+
+            if (!F.isEmpty(cmd.valueTypeName())) {
+                throw new IgniteSQLException(WRAP_VALUE + " parameter cannot be \"false\" when " + VALUE_TYPE + " is defined.",
+                    IgniteQueryErrorCode.PARSING);
+            }
+        }
+        // By default, value is always wrapped to allow for ALTER TABLE ADD COLUMN commands.
+    }
+
     /** */
     private void handle0(DropTableCommand cmd) throws IgniteCheckedException {
         isDdlOnSchemaSupported(cmd.schemaName());
@@ -250,7 +292,7 @@
 
             if (QueryUtils.isSqlType(typeDesc.valueClass())) {
                 throw new SchemaOperationException("Cannot add column(s) because table was created " +
-                    "with WRAP_VALUE=false option.");
+                    "based on cache configured with built-in types or with " + WRAP_VALUE + "=false option.");
             }
 
             List<QueryField> cols = new ArrayList<>(cmd.columns().size());
@@ -412,9 +454,18 @@
         else if (!F.isEmpty(cmd.primaryKeyColumns()) && cmd.primaryKeyColumns().size() == 1) {
             String pkFieldName = cmd.primaryKeyColumns().get(0);
 
-            keyTypeName = res.getFields().get(pkFieldName);
+            if (cmd.wrapKey()) {
+                res.setKeyFields(Set.copyOf(cmd.primaryKeyColumns()));
 
-            res.setKeyFieldName(pkFieldName);
+                keyTypeName = QueryUtils.createTableKeyTypeName(valTypeName);
+
+                res.setPreserveKeysOrder(true);
+            }
+            else {
+                keyTypeName = res.getFields().get(pkFieldName);
+
+                res.setKeyFieldName(pkFieldName);
+            }
         }
         else {
             // if pk is not explicitly set, we create it ourselves
@@ -423,6 +474,22 @@
             res.implicitPk(true);
         }
 
+        if (cmd.wrapValue() != null && !cmd.wrapValue()) {
+            ColumnDefinition valCol = null;
+
+            for (ColumnDefinition col : cmd.columns()) {
+                if (!cmd.primaryKeyColumns().contains(col.name())) {
+                    valCol = col;
+                    break;
+                }
+            }
+
+            if (valCol != null) {
+                valTypeName = Commons.typeFactory().getResultClass(valCol.type()).getTypeName();
+                res.setValueFieldName(valCol.name());
+            }
+        }
+
         res.setValueType(F.isEmpty(cmd.valueTypeName()) ? valTypeName : cmd.valueTypeName());
         res.setKeyType(keyTypeName);
 

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/ddl/CreateTableCommand.java

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.processors.query.calcite.prepare.ddl;
 
+import java.util.Collections;
 import java.util.List;
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.ignite.cache.CacheAtomicityMode;
@@ -45,6 +46,12 @@ public class CreateTableCommand implements DdlCommand {
     /** Name of cache key type. */
     private String keyTypeName;
 
+    /** Wrap key flag. */
+    private Boolean wrapKey;
+
+    /** Wrap value flag. */
+    private Boolean wrapValue;
+
     /** Name of cache value type. */
     private String valTypeName;
 
@@ -67,7 +74,7 @@ public class CreateTableCommand implements DdlCommand {
     private List<ColumnDefinition> cols;
 
     /** Primary key columns. */
-    private List<String> pkCols;
+    private List<String> pkCols = Collections.emptyList();
 
     /** Name of the column that represents affinity key. */
     private String affinityKey;
@@ -109,6 +116,34 @@ public void cacheName(String cacheName) {
         this.cacheName = cacheName;
     }
 
+    /**
+     * @return wrap_key flag.
+     */
+    @Nullable public Boolean wrapKey() {
+        return wrapKey;
+    }
+
+    /**
+     * @param wrapKey wrap_key flag.
+     */
+    public void wrapKey(boolean wrapKey) {
+        this.wrapKey = wrapKey;
+    }
+
+    /**
+     * @return wrap_value flag.
+     */
+    @Nullable public Boolean wrapValue() {
+        return wrapValue;
+    }
+
+    /**
+     * @param wrapValue wrap_value flag.
+     */
+    public void wrapValue(boolean wrapValue) {
+        this.wrapValue = wrapValue;
+    }
+
     /**
      * @return Name of cache key type.
      */

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/ddl/DdlSqlToCommandConverter.java

@@ -76,19 +76,49 @@
 import static org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlCreateTableOptionEnum.KEY_TYPE;
 import static org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlCreateTableOptionEnum.TEMPLATE;
 import static org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlCreateTableOptionEnum.VALUE_TYPE;
+import static org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlCreateTableOptionEnum.WRAP_KEY;
+import static org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlCreateTableOptionEnum.WRAP_VALUE;
 import static org.apache.ignite.internal.processors.query.calcite.sql.IgniteSqlCreateTableOptionEnum.WRITE_SYNCHRONIZATION_MODE;
 import static org.apache.ignite.internal.processors.query.calcite.util.PlanUtils.deriveObjectName;
 import static org.apache.ignite.internal.processors.query.calcite.util.PlanUtils.deriveSchemaName;
 
 /** */
 public class DdlSqlToCommandConverter {
+    /** */
+    private static final String SIMPLE_PREDICATE = "a simple identifier";
+
     /** Processor that validates a value is a Sql Identifier. */
-    private static final BiFunction<IgniteSqlCreateTableOption, PlanningContext, String> VALUE_IS_IDENTIFIER_VALIDATOR = (opt, ctx) -> {
+    private static final BiFunction<IgniteSqlCreateTableOption, PlanningContext, String> VALUE_IS_IDENTIFIER_VALIDATOR =
+        DdlSqlToCommandConverter::paramAsSimpleSqlIdentifier;
+
+    /** Processor that validates that value can be parsed as boolean. */
+    private static final BiFunction<IgniteSqlCreateTableOption, PlanningContext, Boolean> VALUE_IS_BOOL_IDENTIFIER_VALIDATOR =
+        (opt, ctx) -> {
+            SqlNode val = opt.value();
+            if (val instanceof SqlLiteral)
+                return ((SqlLiteral)val).booleanValue();
+
+            String simple = paramAsSimpleSqlIdentifier(opt, ctx);
+
+            if (!"true".equalsIgnoreCase(simple) && !"false".equalsIgnoreCase(simple))
+                throwOptionParsingException(opt, "Unexpected identifier: " + simple, ctx.query());
+
+            return Boolean.valueOf(((SqlIdentifier)val).getSimple());
+        };
+
+    /**
+     * Shortcut for validating that option value is a simple identifier.
+     *
+     * @param opt An option to validate.
+     * @param ctx Planning context.
+     * @throws IgniteSQLException In case the param is not a simple identifier.
+     */
+    private static String paramAsSimpleSqlIdentifier(IgniteSqlCreateTableOption opt, PlanningContext ctx) {
         if (!(opt.value() instanceof SqlIdentifier) || !((SqlIdentifier)opt.value()).isSimple())
-            throwOptionParsingException(opt, "a simple identifier", ctx.query());
+            throwOptionParsingException(opt, SIMPLE_PREDICATE, ctx.query());
 
         return ((SqlIdentifier)opt.value()).getSimple();
-    };
+    }
 
     /** Processor that unconditionally throws an AssertionException. */
     private static final TableOptionProcessor<Void> UNSUPPORTED_OPTION_PROCESSOR = new TableOptionProcessor<>(
@@ -107,6 +137,8 @@
         new TableOptionProcessor<>(DATA_REGION, VALUE_IS_IDENTIFIER_VALIDATOR, CreateTableCommand::dataRegionName),
         new TableOptionProcessor<>(KEY_TYPE, VALUE_IS_IDENTIFIER_VALIDATOR, CreateTableCommand::keyTypeName),
         new TableOptionProcessor<>(VALUE_TYPE, VALUE_IS_IDENTIFIER_VALIDATOR, CreateTableCommand::valueTypeName),
+        new TableOptionProcessor<>(WRAP_KEY, VALUE_IS_BOOL_IDENTIFIER_VALIDATOR, CreateTableCommand::wrapKey),
+        new TableOptionProcessor<>(WRAP_VALUE, VALUE_IS_BOOL_IDENTIFIER_VALIDATOR, CreateTableCommand::wrapValue),
         new TableOptionProcessor<>(ATOMICITY, validatorForEnumValue(CacheAtomicityMode.class), CreateTableCommand::atomicityMode),
         new TableOptionProcessor<>(WRITE_SYNCHRONIZATION_MODE, validatorForEnumValue(CacheWriteSynchronizationMode.class),
             CreateTableCommand::writeSynchronizationMode),
@@ -455,20 +487,6 @@
         return alterTblCmd;
     }
 
-    /**
-     * Short cut for validating that option value is a simple identifier.
-     *
-     * @param opt An option to validate.
-     * @param ctx Planning context.
-     * @throws IgniteSQLException In case the validation was failed.
-     */
-    private String paramIsSqlIdentifierValidator(IgniteSqlCreateTableOption opt, PlanningContext ctx) {
-        if (!(opt.value() instanceof SqlIdentifier) || !((SqlIdentifier)opt.value()).isSimple())
-            throwOptionParsingException(opt, "a simple identifier", ctx.query());
-
-        return ((SqlIdentifier)opt.value()).getSimple();
-    }
-
     /**
      * Creates a validator for an option which value should be value of given enumeration.
      *

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/sql/IgniteSqlCreateTableOptionEnum.java

@@ -52,4 +52,10 @@ public enum IgniteSqlCreateTableOptionEnum {
 
     /** This flag specified whether the encryption should be enabled for the underlying cache. */
     ENCRYPTED,
+
+    /** Flag controls whether a single column PRIMARY KEY should be wrapped in the BinaryObjects format. */
+    WRAP_KEY,
+
+    /** Flag controls whether a single column value of a primitive type should be wrapped in the BinaryObjects format. */
+    WRAP_VALUE,
 }