Add SAP HANA support (#1511)

src/main/java/io/ebean/config/dbplatform/DatabasePlatform.java

@@ -176,9 +176,17 @@ public class DatabasePlatform {
    * findIterate() and findVisit().
    */
   protected boolean forwardOnlyHintOnFindIterate;
+  
+  /**
+   * If set then use the CONCUR_UPDATABLE hint when creating ResultSets.
+   * 
+   * This is {@code false} for HANA
+   */
+  protected boolean supportsResultSetConcurrencyModeUpdatable = true;
+
 
   /**
-   * By default we use JDBC batch when cascading (except for SQL Server).
+   * By default we use JDBC batch when cascading (except for SQL Server and HANA).
    */
   protected PersistBatch persistBatchOnCascade = PersistBatch.ALL;
 
@@ -516,6 +524,24 @@ public class DatabasePlatform {
   public void setForwardOnlyHintOnFindIterate(boolean forwardOnlyHintOnFindIterate) {
     this.forwardOnlyHintOnFindIterate = forwardOnlyHintOnFindIterate;
   }
+  
+  /**
+   * Return true if the ResultSet CONCUR_UPDATABLE Hint should be used on
+   * createNativeSqlTree() PreparedStatements.
+   * <p>
+   * This specifically is required for Hana which doesn't support CONCUR_UPDATABLE
+   * </p>
+   */
+  public boolean isSupportsResultSetConcurrencyModeUpdatable() {
+    return supportsResultSetConcurrencyModeUpdatable;
+  }
+  
+  /**
+   * Set to true if the ResultSet CONCUR_UPDATABLE Hint should be used by default on createNativeSqlTree() PreparedStatements.
+   */
+  public void setSupportsResultSetConcurrencyModeUpdatable(boolean supportsResultSetConcurrencyModeUpdatable) {
+    this.supportsResultSetConcurrencyModeUpdatable = supportsResultSetConcurrencyModeUpdatable;
+  }
 
   /**
    * Normally not needed - overridden in CockroachPlatform.

src/main/java/io/ebean/config/dbplatform/hana/HanaBasicSqlLimiter.java

@@ -0,0 +1,24 @@
+package io.ebean.config.dbplatform.hana;
+
+import io.ebean.config.dbplatform.BasicSqlLimiter;
+
+public class HanaBasicSqlLimiter implements BasicSqlLimiter {
+  @Override
+  public String limit(String dbSql, int firstRow, int maxRows) {
+    StringBuilder sb = new StringBuilder(50 + dbSql.length());
+
+    sb.append(dbSql);
+
+    if (maxRows > 0) {
+      sb.append(" ").append("limit");
+      sb.append(" ").append(maxRows);
+
+      if (firstRow > 0) {
+        sb.append(" ").append("offset").append(" ");
+        sb.append(firstRow);
+      }
+    }
+
+    return sb.toString();
+  }
+}

src/main/java/io/ebean/config/dbplatform/hana/HanaHistorySupport.java

@@ -0,0 +1,27 @@
+package io.ebean.config.dbplatform.hana;
+
+import io.ebean.config.dbplatform.DbStandardHistorySupport;
+
+public class HanaHistorySupport extends DbStandardHistorySupport {
+
+  @Override
+  public String getAsOfViewSuffix(String asOfViewSuffix) {
+    return " for system_time as of ?";
+  }
+
+  @Override
+  public String getVersionsBetweenSuffix(String asOfViewSuffix) {
+    return " for system_time between ? and ?";
+  }
+
+  @Override
+  public String getSysPeriodLower(String tableAlias, String sysPeriod) {
+    return tableAlias + "." + sysPeriod + "_start";
+  }
+
+  @Override
+  public String getSysPeriodUpper(String tableAlias, String sysPeriod) {
+    return tableAlias + "." + sysPeriod + "_end";
+  }
+  
+}

src/main/java/io/ebean/config/dbplatform/hana/HanaPlatform.java

@@ -0,0 +1,101 @@
+package io.ebean.config.dbplatform.hana;
+
+import io.ebean.Query.ForUpdate;
+import io.ebean.annotation.PersistBatch;
+import io.ebean.annotation.Platform;
+import io.ebean.config.PlatformConfig;
+import io.ebean.config.dbplatform.DatabasePlatform;
+import io.ebean.config.dbplatform.DbPlatformType;
+import io.ebean.config.dbplatform.DbType;
+import io.ebean.config.dbplatform.IdType;
+import io.ebean.config.dbplatform.SqlErrorCodes;
+
+public class HanaPlatform extends DatabasePlatform {
+  public HanaPlatform() {
+    this.basicSqlLimiter = new HanaBasicSqlLimiter();
+    
+    this.columnAliasPrefix = null;
+
+    this.dbDefaultValue.setNow("current_timestamp");
+
+    this.dbIdentity.setIdType(IdType.IDENTITY);
+    this.dbIdentity.setSelectLastInsertedIdTemplate("select current_identity_value() from sys.dummy");
+    this.dbIdentity.setSupportsGetGeneratedKeys(false);
+    this.dbIdentity.setSupportsIdentity(true);
+
+    this.dbTypeMap.put(DbType.BIGINT, new DbPlatformType("bigint", false));
+    this.dbTypeMap.put(DbType.BINARY, new DbPlatformType("varbinary", 255));
+    this.dbTypeMap.put(DbType.BIT, new DbPlatformType("smallint", false));
+    this.dbTypeMap.put(DbType.BLOB, new DbPlatformType("blob", false));
+    this.dbTypeMap.put(DbType.CHAR, new DbPlatformType("nvarchar", 255));
+    this.dbTypeMap.put(DbType.CLOB, new DbPlatformType("nclob", false));
+    this.dbTypeMap.put(DbType.INTEGER, new DbPlatformType("integer", false));
+    this.dbTypeMap.put(DbType.JSONVARCHAR, new DbPlatformType("nvarchar", 255));
+    this.dbTypeMap.put(DbType.LINESTRING, new DbPlatformType("st_geometry"));
+    this.dbTypeMap.put(DbType.LONGVARBINARY, new DbPlatformType("blob", false));
+    this.dbTypeMap.put(DbType.LONGVARCHAR, new DbPlatformType("nclob", false));
+    this.dbTypeMap.put(DbType.MULTILINESTRING, new DbPlatformType("st_geometry"));
+    this.dbTypeMap.put(DbType.MULTIPOINT, new DbPlatformType("st_geometry"));
+    this.dbTypeMap.put(DbType.MULTIPOLYGON, new DbPlatformType("st_geometry"));
+    this.dbTypeMap.put(DbType.POINT, new DbPlatformType("st_point"));
+    this.dbTypeMap.put(DbType.POLYGON, new DbPlatformType("st_geometry"));
+    this.dbTypeMap.put(DbType.SMALLINT, new DbPlatformType("smallint", false));
+    this.dbTypeMap.put(DbType.TINYINT, new DbPlatformType("smallint", false));
+    this.dbTypeMap.put(DbType.UUID, new DbPlatformType("varchar", 40));
+    this.dbTypeMap.put(DbType.VARBINARY, new DbPlatformType("varbinary", 255));
+    this.dbTypeMap.put(DbType.VARCHAR, new DbPlatformType("nvarchar", 255));
+
+    this.exceptionTranslator = new SqlErrorCodes().addAcquireLock("131", "133", "146")
+        .addDataIntegrity("130", "429", "461", "462").addDuplicateKey("144", "301", "349").build();
+
+    this.historySupport = new HanaHistorySupport();
+    
+    this.likeClauseRaw = "like ?";
+
+    this.maxConstraintNameLength = 127;
+    this.maxTableNameLength = 127;
+
+    this.persistBatchOnCascade = PersistBatch.NONE;
+    
+    this.platform = Platform.HANA;
+
+    this.sqlLimiter = new HanaSqlLimiter();
+    
+    this.supportsResultSetConcurrencyModeUpdatable = false;
+  }
+
+  @Override
+  protected void addGeoTypes(int srid) {
+    this.dbTypeMap.put(DbType.LINESTRING, new DbPlatformType("st_geometry(" + srid + ")", false));
+    this.dbTypeMap.put(DbType.MULTILINESTRING, new DbPlatformType("st_geometry(" + srid + ")", false));
+    this.dbTypeMap.put(DbType.MULTIPOINT, new DbPlatformType("st_geometry(" + srid + ")", false));
+    this.dbTypeMap.put(DbType.MULTIPOLYGON, new DbPlatformType("st_geometry(" + srid + ")", false));
+    this.dbTypeMap.put(DbType.POINT, new DbPlatformType("st_point(" + srid + ")", false));
+    this.dbTypeMap.put(DbType.POLYGON, new DbPlatformType("st_geometry(" + srid + ")", false));
+  }
+
+  @Override
+  protected String withForUpdate(String sql, ForUpdate forUpdateMode) {
+    switch (forUpdateMode) {
+    case BASE:
+      return sql + " for update";
+    case NOWAIT:
+      return sql + " for update nowait";
+    case SKIPLOCKED:
+      return sql + " for update ignore locked";
+    default:
+      throw new IllegalArgumentException("Unknown update mode: " + forUpdateMode.name());
+    }
+  }
+
+  @Override
+  protected void configure(PlatformConfig config, boolean allQuotedIdentifiers) {
+    super.configure(config, allQuotedIdentifiers);
+    if (config.getDbUuid().useBinary()) {
+      this.dbTypeMap.put(DbType.UUID, new DbPlatformType("varbinary", 16));
+    } else {
+      this.dbTypeMap.put(DbType.UUID, new DbPlatformType("varchar", 40));
+    }
+  }
+
+}

src/main/java/io/ebean/config/dbplatform/hana/HanaSqlLimiter.java

@@ -0,0 +1,35 @@
+package io.ebean.config.dbplatform.hana;
+
+import io.ebean.config.dbplatform.SqlLimitRequest;
+import io.ebean.config.dbplatform.SqlLimitResponse;
+import io.ebean.config.dbplatform.SqlLimiter;
+
+public class HanaSqlLimiter implements SqlLimiter {
+  @Override
+  public SqlLimitResponse limit(SqlLimitRequest request) {
+    String dbSql = request.getDbSql();
+
+    StringBuilder sb = new StringBuilder(50 + dbSql.length());
+    sb.append("select ");
+    if (request.isDistinct()) {
+      sb.append("distinct ");
+    }
+
+    sb.append(dbSql);
+
+    int firstRow = request.getFirstRow();
+    int maxRows = request.getMaxRows();
+
+    if (maxRows > 0) {
+      sb.append(" ").append("limit ").append(maxRows);
+      if (firstRow > 0) {
+        sb.append(" ").append("offset ");
+        sb.append(firstRow);
+      }
+    }
+
+    String sql = request.getDbPlatform().completeSql(sb.toString(), request.getOrmQuery());
+
+    return new SqlLimitResponse(sql, false);
+  }
+}

src/main/java/io/ebean/config/dbplatform/hana/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * SAP HANA specific support.
+ */
+package io.ebean.config.dbplatform.hana;

src/main/java/io/ebeaninternal/dbmigration/DefaultDbMigration.java

@@ -10,6 +10,7 @@ import io.ebean.config.ServerConfig;
 import io.ebean.config.dbplatform.DatabasePlatform;
 import io.ebean.config.dbplatform.db2.DB2Platform;
 import io.ebean.config.dbplatform.h2.H2Platform;
+import io.ebean.config.dbplatform.hana.HanaPlatform;
 import io.ebean.config.dbplatform.hsqldb.HsqldbPlatform;
 import io.ebean.config.dbplatform.mysql.MySqlPlatform;
 import io.ebean.config.dbplatform.oracle.OraclePlatform;
@@ -679,6 +680,8 @@ public class DefaultDbMigration implements DbMigration {
         return new DB2Platform();
       case SQLITE:
         return new SQLitePlatform();
+      case HANA:
+    	return new HanaPlatform();
       case GENERIC:
         return new DatabasePlatform();
 

src/main/java/io/ebeaninternal/dbmigration/ddlgeneration/BaseDdlHandler.java

@@ -28,7 +28,11 @@ public class BaseDdlHandler implements DdlHandler {
   protected final TableDdl tableDdl;
 
   public BaseDdlHandler(ServerConfig serverConfig, PlatformDdl platformDdl) {
-    this.tableDdl = new BaseTableDdl(serverConfig, platformDdl);
+    this(serverConfig, platformDdl, new BaseTableDdl(serverConfig, platformDdl));
+  }
+  
+  protected BaseDdlHandler(ServerConfig serverConfig, PlatformDdl platformDdl, TableDdl tableDdl) {
+    this.tableDdl = tableDdl;
   }
 
   @Override

src/main/java/io/ebeaninternal/dbmigration/ddlgeneration/platform/AbstractHanaDdl.java

@@ -0,0 +1,226 @@
+package io.ebeaninternal.dbmigration.ddlgeneration.platform;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import io.ebean.config.ServerConfig;
+import io.ebean.config.dbplatform.DatabasePlatform;
+import io.ebean.config.dbplatform.DbPlatformType;
+import io.ebeaninternal.dbmigration.ddlgeneration.DdlBuffer;
+import io.ebeaninternal.dbmigration.ddlgeneration.DdlHandler;
+import io.ebeaninternal.dbmigration.migration.AlterColumn;
+
+public abstract class AbstractHanaDdl extends PlatformDdl {
+
+  private static final Pattern ARRAY_PATTERN = Pattern.compile("(\\w+)\\s*\\[\\s*\\]\\s*(\\(\\d+\\))?",
+      Pattern.CASE_INSENSITIVE);
+
+  public AbstractHanaDdl(DatabasePlatform platform) {
+    super(platform);
+    this.addColumn = "add (";
+    this.addColumnSuffix = ")";
+    this.alterColumn = "alter (";
+    this.alterColumnSuffix = ")";
+    this.columnDropDefault = " default null";
+    this.columnSetDefault = " default";
+    this.columnSetNotnull = " not null";
+    this.columnSetNull = " null";
+    this.dropColumn = "drop (";
+    this.dropColumnSuffix = ")";
+    this.dropConstraintIfExists = "drop constraint ";
+    this.dropIndexIfExists = "drop index ";
+    this.dropSequenceIfExists = "drop sequence ";
+    this.dropTableCascade = " cascade";
+    this.dropTableIfExists = "drop table ";
+    this.fallbackArrayType = "nvarchar(1000)";
+    this.historyDdl = new HanaHistoryDdl();
+    this.identitySuffix = " generated by default as identity";
+  }
+
+  @Override
+  public String alterColumnBaseAttributes(AlterColumn alter) {
+    String tableName = alter.getTableName();
+    String columnName = alter.getColumnName();
+    String currentType = alter.getCurrentType();
+    String type = alter.getType() != null ? alter.getType() : currentType;
+    type = convert(type, false);
+    currentType = convert(currentType, false);
+    boolean notnull = (alter.isNotnull() != null) ? alter.isNotnull() : Boolean.TRUE.equals(alter.isCurrentNotnull());
+    String notnullClause = notnull ? " not null" : "";
+    String defaultValue = DdlHelp.isDropDefault(alter.getDefaultValue()) ? "null"
+        : (alter.getDefaultValue() != null ? alter.getDefaultValue() : alter.getCurrentDefaultValue());
+    String defaultValueClause = (defaultValue == null || defaultValue.isEmpty()) ? "" : " default " + defaultValue;
+
+    try {
+      DdlBuffer buffer = new BaseDdlBuffer(null);
+      if (!isConvertible(currentType, type)) {
+        // add an intermediate conversion if possible
+        if (isNumberType(currentType)) {
+          // numbers can always be converted to decimal
+          buffer.append("alter table ").append(tableName).append(" ").append(alterColumn).append(" ").append(columnName)
+              .append(" decimal ").append(defaultValueClause).append(notnullClause).append(alterColumnSuffix)
+              .endOfStatement();
+        } else if (isStringType(currentType)) {
+          // strings can always be converted to nclob
+          buffer.append("alter table ").append(tableName).append(" ").append(alterColumn).append(" ").append(columnName)
+              .append(" nclob ").append(defaultValueClause).append(notnullClause).append(alterColumnSuffix)
+              .endOfStatement();
+        }
+      }
+
+      buffer.append("alter table ").append(tableName).append(" ").append(alterColumn).append(" ").append(columnName)
+          .append(" ").append(type).append(defaultValueClause).append(notnullClause).append(alterColumnSuffix);
+
+      return buffer.getBuffer();
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public String alterColumnDefaultValue(String tableName, String columnName, String defaultValue) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public String alterColumnNotnull(String tableName, String columnName, boolean notnull) {
+    return null;
+  }
+
+  @Override
+  public DdlHandler createDdlHandler(ServerConfig serverConfig) {
+    return new HanaDdlHandler(serverConfig, this);
+  }
+
+  @Override
+  public String alterColumnType(String tableName, String columnName, String type) {
+    return null;
+  }
+
+  @Override
+  protected String convertArrayType(String logicalArrayType) {
+    Matcher matcher = ARRAY_PATTERN.matcher(logicalArrayType);
+    if (matcher.matches()) {
+      return convert(matcher.group(1), false) + " array" + (matcher.group(2) == null ? "" : matcher.group(2));
+    } else {
+      return fallbackArrayType;
+    }
+  }
+
+  @Override
+  public String alterTableAddUniqueConstraint(String tableName, String uqName, String[] columns,
+      String[] nullableColumns) {
+    if (nullableColumns == null || nullableColumns.length == 0) {
+      return super.alterTableAddUniqueConstraint(tableName, uqName, columns, nullableColumns);
+    } else {
+      return "-- cannot create unique index \"" + uqName + "\" on table \"" + tableName + "\" with nullable columns";
+    }
+  }
+
+  @Override
+  public String alterTableDropUniqueConstraint(String tableName, String uniqueConstraintName) {
+    DdlBuffer buffer = new BaseDdlBuffer(null);
+    try {
+      buffer.append("delimiter $$").newLine();
+      buffer.append("do").newLine();
+      buffer.append("begin").newLine();
+      buffer.append("declare exit handler for sql_error_code 397 begin end").endOfStatement();
+      buffer.append("exec 'alter table ").append(tableName).append(" ").append(dropUniqueConstraint).append(" ")
+          .append(maxConstraintName(uniqueConstraintName)).append("'").endOfStatement();
+      buffer.append("end").endOfStatement();
+      buffer.append("$$");
+      return buffer.getBuffer();
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+  
+  @Override
+  public String alterTableDropConstraint(String tableName, String constraintName) {
+    return alterTableDropUniqueConstraint(tableName, constraintName);
+  }
+
+  /**
+   * It is rather complex to delete a column on HANA as there must not exist any
+   * foreign keys. That's why we call a user stored procedure here
+   */
+  @Override
+  public void alterTableDropColumn(DdlBuffer buffer, String tableName, String columnName) throws IOException {
+    buffer.append("CALL usp_ebean_drop_column('").append(tableName).append("', '").append(columnName).append("')")
+        .endOfStatement();
+  }
+
+  /**
+   * Check if a data type can be converted to another data type. Data types can't
+   * be converted if the target type has a lower precision than the source type.
+   * 
+   * @param sourceType The source data type
+   * @param targetType the target data type
+   * @return {@code true} if the type can be converted, {@code false} otherwise
+   */
+  private boolean isConvertible(String sourceType, String targetType) {
+    if (Objects.equals(sourceType, targetType)) {
+      return true;
+    }
+
+    if (sourceType == null || targetType == null) {
+      return true;
+    }
+
+    if ("bigint".equals(sourceType)) {
+      if ("integer".equals(targetType) || "smallint".equals(targetType) || "tinyint".equals(targetType)) {
+        return false;
+      }
+    } else if ("integer".equals(sourceType)) {
+      if ("smallint".equals(targetType) || "tinyint".equals(targetType)) {
+        return false;
+      }
+    } else if ("smallint".equals(sourceType)) {
+      if ("tinyint".equals(targetType)) {
+        return false;
+      }
+    } else if ("double".equals(sourceType)) {
+      if ("real".equals(targetType)) {
+        return false;
+      }
+    }
+
+    DbPlatformType dbPlatformSourceType = DbPlatformType.parse(sourceType);
+
+    if ("float".equals(dbPlatformSourceType.getName())) {
+      if ("real".equals(targetType)) {
+        return false;
+      }
+    } else if ("varchar".equals(dbPlatformSourceType.getName()) || "nvarchar".equals(dbPlatformSourceType.getName())) {
+      DbPlatformType dbPlatformTargetType = DbPlatformType.parse(targetType);
+      if ("varchar".equals(dbPlatformTargetType.getName()) || "nvarchar".equals(dbPlatformTargetType.getName())) {
+        if (dbPlatformSourceType.getDefaultLength() > dbPlatformTargetType.getDefaultLength()) {
+          return false;
+        }
+      }
+    } else if ("decimal".equals(dbPlatformSourceType.getName())) {
+      DbPlatformType dbPlatformTargetType = DbPlatformType.parse(targetType);
+      if ("decimal".equals(dbPlatformTargetType.getName())) {
+        if (dbPlatformSourceType.getDefaultLength() > dbPlatformTargetType.getDefaultLength()
+            || dbPlatformSourceType.getDefaultScale() > dbPlatformTargetType.getDefaultScale()) {
+          return false;
+        }
+      }
+    }
+
+    return true;
+  }
+
+  private boolean isNumberType(String type) {
+    return type != null
+        && ("bigint".equals(type) || "integer".equals(type) || "smallint".equals(type) || "tinyint".equals(type)
+            || type.startsWith("float") || "real".equals(type) || "double".equals(type) || type.startsWith("decimal"));
+  }
+
+  private boolean isStringType(String type) {
+    return type != null
+        && (type.startsWith("varchar") || type.startsWith("nvarchar") || "clob".equals(type) || "nclob".equals(type));
+  }
+}

src/main/java/io/ebeaninternal/dbmigration/ddlgeneration/platform/BaseTableDdl.java

@@ -260,7 +260,7 @@ public class BaseTableDdl implements TableDdl {
     String partitionMode = createTable.getPartitionMode();
 
     DdlBuffer apply = writer.apply();
-    apply.append("create table ").append(tableName).append(" (");
+    apply.append(platformDdl.getCreateTableCommandPrefix()).append(" ").append(tableName).append(" (");
     writeTableColumns(apply, columns, useIdentity);
     writeCheckConstraints(apply, createTable);
     writeUniqueConstraints(apply, createTable);

src/main/java/io/ebeaninternal/dbmigration/ddlgeneration/platform/DbTriggerBasedHistoryDdl.java

@@ -175,7 +175,7 @@ public abstract class DbTriggerBasedHistoryDdl implements PlatformHistoryDdl {
 
   protected void createHistoryTable(DdlBuffer apply, MTable table) throws IOException {
 
-    apply.append("create table ").append(table.getName()).append(historySuffix).append("(").newLine();
+    apply.append(platformDdl.getCreateTableCommandPrefix()).append(" ").append(table.getName()).append(historySuffix).append("(").newLine();
 
     Collection<MColumn> cols = table.allColumns();
     for (MColumn column : cols) {
@@ -238,8 +238,8 @@ public abstract class DbTriggerBasedHistoryDdl implements PlatformHistoryDdl {
   }
 
   protected void dropSysPeriodColumns(DdlBuffer buffer, String baseTableName) throws IOException {
-    buffer.append("alter table ").append(baseTableName).append(" drop column ").append(sysPeriodStart).endOfStatement();
-    buffer.append("alter table ").append(baseTableName).append(" drop column ").append(sysPeriodEnd).endOfStatement();
+    platformDdl.alterTableDropColumn(buffer, baseTableName, sysPeriodStart);
+    platformDdl.alterTableDropColumn(buffer, baseTableName, sysPeriodEnd);
   }
 
   protected void appendInsertIntoHistory(DdlBuffer buffer, String historyTable, List<String> columns) throws IOException {

src/main/java/io/ebeaninternal/dbmigration/ddlgeneration/platform/HanaColumnStoreDdl.java

@@ -0,0 +1,51 @@
+package io.ebeaninternal.dbmigration.ddlgeneration.platform;
+
+import java.io.IOException;
+
+import io.ebean.config.dbplatform.DatabasePlatform;
+import io.ebeaninternal.dbmigration.ddlgeneration.DdlBuffer;
+
+public class HanaColumnStoreDdl extends AbstractHanaDdl {
+
+  public HanaColumnStoreDdl(DatabasePlatform platform) {
+    super(platform);
+    this.createTable = "create column table";
+  }
+
+  @Override
+  public String createIndex(String indexName, String tableName, String[] columns) {
+    if (columns == null || columns.length == 0) {
+      return "-- cannot create index: no columns given";
+    }
+
+    if (columns.length == 1) {
+      return "-- explicit index \"" + indexName + "\" for single column \"" + columns[0] + "\" of table \"" + tableName
+          + "\" is not necessary";
+    }
+
+    StringBuilder buffer = new StringBuilder();
+    buffer.append("create inverted hash index ").append(maxConstraintName(indexName)).append(" on ").append(tableName);
+    appendColumns(columns, buffer);
+
+    return buffer.toString();
+  }
+
+  @Override
+  public String dropIndex(String indexName, String tableName) {
+    DdlBuffer buffer = new BaseDdlBuffer(null);
+    try {
+      buffer.append("delimiter $$").newLine();
+      buffer.append("do").newLine();
+      buffer.append("begin").newLine();
+      buffer.append("declare exit handler for sql_error_code 261 begin end").endOfStatement();
+      buffer.append("exec '").append(dropIndexIfExists).append(maxConstraintName(indexName)).append("'")
+          .endOfStatement();
+      buffer.append("end").endOfStatement();
+      buffer.append("$$");
+      return buffer.getBuffer();
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+}

src/main/java/io/ebeaninternal/dbmigration/ddlgeneration/platform/HanaDdlHandler.java

@@ -0,0 +1,11 @@
+package io.ebeaninternal.dbmigration.ddlgeneration.platform;
+
+import io.ebean.config.ServerConfig;
+import io.ebeaninternal.dbmigration.ddlgeneration.BaseDdlHandler;
+
+public class HanaDdlHandler extends BaseDdlHandler {
+
+  public HanaDdlHandler(ServerConfig serverConfig, PlatformDdl platformDdl) {
+    super(serverConfig, platformDdl, new HanaTableDdl(serverConfig, platformDdl));
+  }
+}

src/main/java/io/ebeaninternal/dbmigration/ddlgeneration/platform/HanaHistoryDdl.java

@@ -0,0 +1,160 @@
+package io.ebeaninternal.dbmigration.ddlgeneration.platform;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.ebean.config.ServerConfig;
+import io.ebeaninternal.dbmigration.ddlgeneration.DdlBuffer;
+import io.ebeaninternal.dbmigration.ddlgeneration.DdlWrite;
+import io.ebeaninternal.dbmigration.migration.AddHistoryTable;
+import io.ebeaninternal.dbmigration.migration.DropHistoryTable;
+import io.ebeaninternal.dbmigration.model.MColumn;
+import io.ebeaninternal.dbmigration.model.MTable;
+
+public class HanaHistoryDdl implements PlatformHistoryDdl {
+
+  private String systemPeriodStart;
+  private String systemPeriodEnd;
+  private PlatformDdl platformDdl;
+  private String historySuffix;
+  private final AtomicInteger counter = new AtomicInteger(0);
+  private Map<String, String> createdHistoryTables = new ConcurrentHashMap<>();
+
+  @Override
+  public void configure(ServerConfig serverConfig, PlatformDdl platformDdl) {
+    this.systemPeriodStart = serverConfig.getAsOfSysPeriod() + "_start";
+    this.systemPeriodEnd = serverConfig.getAsOfSysPeriod() + "_end";
+    this.platformDdl = platformDdl;
+    this.historySuffix = serverConfig.getHistoryTableSuffix();
+  }
+
+  @Override
+  public void createWithHistory(DdlWrite writer, MTable table) throws IOException {
+    String tableName = table.getName();
+    String historyTableName = tableName + historySuffix;
+    DdlBuffer apply = writer.applyHistoryView();
+    if (apply.isEmpty()) {
+      createdHistoryTables.clear();
+    }
+
+    apply.append(platformDdl.getCreateTableCommandPrefix()).append(" ").append(historyTableName).append(" (").newLine();
+
+    // create history table
+    Collection<MColumn> cols = table.allColumns();
+    for (MColumn column : cols) {
+      if (!column.isDraftOnly()) {
+        writeColumnDefinition(apply, column.getName(), column.getType(), column.getDefaultValue(), column.isNotnull(),
+            column.isIdentity() ? platformDdl.identitySuffix : null);
+        apply.append(",").newLine();
+      }
+    }
+    writeColumnDefinition(apply, systemPeriodStart, "TIMESTAMP", null, false, null);
+    apply.append(",").newLine();
+    writeColumnDefinition(apply, systemPeriodEnd, "TIMESTAMP", null, false, null);
+    apply.newLine().append(")").endOfStatement();
+
+    // enable system versioning
+    apply.append("alter table ").append(tableName).append(" add (").newLine();
+    apply.append("    ").append(systemPeriodStart).append(" TIMESTAMP NOT NULL GENERATED ALWAYS AS ROW START, ")
+        .newLine();
+    apply.append("    ").append(systemPeriodEnd).append(" TIMESTAMP NOT NULL GENERATED ALWAYS AS ROW END").newLine();
+    apply.append(")").endOfStatement();
+
+    apply.append("alter table ").append(tableName).append(" add period for system_time(").append(systemPeriodStart)
+        .append(",").append(systemPeriodEnd).append(")").endOfStatement();
+
+    enableSystemVersioning(apply, tableName, historyTableName, true, false);
+
+    createdHistoryTables.put(tableName, historyTableName);
+
+    dropHistoryTable(writer.dropAll(), tableName, historyTableName);
+  }
+
+  @Override
+  public void dropHistoryTable(DdlWrite writer, DropHistoryTable dropHistoryTable) throws IOException {
+    dropHistoryTable(writer.applyDropDependencies(), dropHistoryTable.getBaseTable(),
+        dropHistoryTable.getBaseTable() + historySuffix);
+  }
+
+  protected void dropHistoryTable(DdlBuffer apply, String baseTable, String historyTable) throws IOException {
+    // disable system versioning
+    disableSystemVersioning(apply, baseTable);
+
+    apply.append("alter table ").append(baseTable).append(" drop period for system_time").endOfStatement();
+
+    // drop the period columns
+    apply.append("alter table ").append(baseTable).append(" drop (").append(systemPeriodStart).append(",")
+        .append(systemPeriodEnd).append(")").endOfStatement();
+
+    // drop the history table
+    apply.append("drop table ").append(historyTable).append(" cascade").endOfStatement();
+  }
+
+  @Override
+  public void addHistoryTable(DdlWrite writer, AddHistoryTable addHistoryTable) throws IOException {
+    MTable table = writer.getTable(addHistoryTable.getBaseTable());
+    if (table == null) {
+      throw new IllegalStateException(
+          "MTable " + addHistoryTable.getBaseTable() + " not found in writer? (required for history DDL)");
+    }
+    createWithHistory(writer, table);
+  }
+
+  @Override
+  public void updateTriggers(DdlWrite write, HistoryTableUpdate baseTable) throws IOException {
+    // nothing to do
+  }
+
+  protected void writeColumnDefinition(DdlBuffer buffer, String columnName, String type, String defaultValue,
+      boolean isNotNull, String generated) throws IOException {
+
+    String platformType = platformDdl.convert(type, false);
+    buffer.append(" ").append(platformDdl.lowerColumnName(columnName));
+    buffer.append(" ").append(platformType);
+    if (defaultValue != null) {
+      buffer.append(" default ").append(defaultValue);
+    }
+    if (isNotNull) {
+      buffer.append(" not null");
+    }
+    if (generated != null) {
+      buffer.append(" ").append(generated);
+    }
+  }
+
+  public void disableSystemVersioning(DdlBuffer apply, String tableName) throws IOException {
+    disableSystemVersioning(apply, tableName, false);
+  }
+
+  public void disableSystemVersioning(DdlBuffer apply, String tableName, boolean uniqueStatement) throws IOException {
+    apply.append("alter table ").append(tableName).append(" drop system versioning");
+    if (uniqueStatement) {
+      // needed for the DB migration test to prevent the statement from being filtered
+      // out as a duplicate
+      apply.append(" /* ").append(String.valueOf(counter.getAndIncrement())).append(" */");
+    }
+    apply.endOfStatement();
+  }
+
+  public void enableSystemVersioning(DdlBuffer apply, String tableName, String historyTableName, boolean validated,
+      boolean uniqueStatement) throws IOException {
+    apply.append("alter table ").append(tableName).append(" add system versioning history table ")
+        .append(historyTableName);
+    if (!validated) {
+      apply.append(" not validated");
+    }
+    if (uniqueStatement) {
+      // needed for the DB migration test to prevent the statement from being filtered
+      // out as a duplicate
+      apply.append(" /* ").append(String.valueOf(counter.getAndIncrement())).append(" */");
+    }
+    apply.endOfStatement();
+  }
+
+  public boolean isSystemVersioningEnabled(String tableName) {
+    return !createdHistoryTables.containsKey(tableName);
+  }
+}

src/main/java/io/ebeaninternal/dbmigration/ddlgeneration/platform/HanaRowStoreDdl.java

@@ -0,0 +1,11 @@
+package io.ebeaninternal.dbmigration.ddlgeneration.platform;
+
+import io.ebean.config.dbplatform.DatabasePlatform;
+
+public class HanaRowStoreDdl extends AbstractHanaDdl {
+
+  public HanaRowStoreDdl(DatabasePlatform platform) {
+    super(platform);
+    this.createTable = "create row table";
+  }
+}

src/main/java/io/ebeaninternal/dbmigration/ddlgeneration/platform/HanaTableDdl.java

@@ -0,0 +1,136 @@
+package io.ebeaninternal.dbmigration.ddlgeneration.platform;
+
+import java.io.IOException;
+import java.util.List;
+
+import io.ebean.config.PropertiesWrapper;
+import io.ebean.config.ServerConfig;
+import io.ebeaninternal.dbmigration.ddlgeneration.DdlWrite;
+import io.ebeaninternal.dbmigration.migration.AddColumn;
+import io.ebeaninternal.dbmigration.migration.AlterColumn;
+import io.ebeaninternal.dbmigration.migration.Column;
+import io.ebeaninternal.dbmigration.migration.DropColumn;
+import io.ebeaninternal.dbmigration.model.MTable;
+
+public class HanaTableDdl extends BaseTableDdl {
+
+  private final HanaHistoryDdl historyDdl;
+  private final boolean generateUniqueDdl;
+
+  public HanaTableDdl(ServerConfig serverConfig, PlatformDdl platformDdl) {
+    super(serverConfig, platformDdl);
+    this.historyDdl = (HanaHistoryDdl) platformDdl.historyDdl;
+    if (serverConfig.getProperties() != null) {
+      PropertiesWrapper wrapper = new PropertiesWrapper("ebean", "hana", serverConfig.getProperties(),
+          serverConfig.getClassLoadConfig());
+      this.generateUniqueDdl = wrapper.getBoolean("generateUniqueDdl", false);
+    } else {
+      this.generateUniqueDdl = false;
+    }
+  }
+
+  @Override
+  protected void alterColumnDefaultValue(DdlWrite writer, AlterColumn alter) throws IOException {
+    String ddl = platformDdl.alterColumnBaseAttributes(alter);
+    if (hasValue(ddl)) {
+      writer.apply().append(ddl).endOfStatement();
+    }
+  }
+
+  @Override
+  public void generate(DdlWrite writer, AddColumn addColumn) throws IOException {
+    String tableName = addColumn.getTableName();
+    MTable table = writer.getTable(tableName);
+    if (table == null) {
+      super.generate(writer, addColumn);
+      return;
+    }
+
+    boolean manageSystemVersioning = isTrue(table.isWithHistory()) && historyDdl.isSystemVersioningEnabled(tableName);
+
+    if (manageSystemVersioning) {
+      historyDdl.disableSystemVersioning(writer.apply(), table.getName(), this.generateUniqueDdl);
+    }
+
+    super.generate(writer, addColumn);
+
+    if (manageSystemVersioning) {
+      // make same changes to the history table
+      String historyTable = historyTable(tableName);
+      List<Column> columns = addColumn.getColumn();
+      for (Column column : columns) {
+        alterTableAddColumn(writer.apply(), historyTable, column, true, true);
+      }
+
+      historyDdl.enableSystemVersioning(writer.apply(), table.getName(), historyTable, false, this.generateUniqueDdl);
+    }
+  }
+
+  @Override
+  public void generate(DdlWrite writer, AlterColumn alterColumn) throws IOException {
+    String tableName = alterColumn.getTableName();
+    MTable table = writer.getTable(tableName);
+    if (table == null) {
+      super.generate(writer, alterColumn);
+      return;
+    }
+
+    boolean manageSystemVersioning = isTrue(table.isWithHistory()) && historyDdl.isSystemVersioningEnabled(tableName);
+
+    if (manageSystemVersioning) {
+      historyDdl.disableSystemVersioning(writer.apply(), tableName, this.generateUniqueDdl);
+    }
+
+    super.generate(writer, alterColumn);
+
+    if (manageSystemVersioning) {
+      // make same changes to the history table
+      String historyTable = historyTable(tableName);
+      if (hasValue(alterColumn.getType()) || hasValue(alterColumn.getDefaultValue())
+          || alterColumn.isNotnull() != null) {
+        AlterColumn alterHistoryColumn = new AlterColumn();
+        alterHistoryColumn.setTableName(historyTable);
+        alterHistoryColumn.setColumnName(alterColumn.getColumnName());
+        alterHistoryColumn.setType(alterColumn.getType());
+        alterHistoryColumn.setDefaultValue(alterColumn.getDefaultValue());
+        alterHistoryColumn.setNotnull(alterColumn.isNotnull());
+        alterHistoryColumn.setCurrentType(alterColumn.getCurrentType());
+        alterHistoryColumn.setCurrentDefaultValue(alterColumn.getCurrentDefaultValue());
+        alterHistoryColumn.setCurrentNotnull(alterColumn.isCurrentNotnull());
+        String histColumnDdl = platformDdl.alterColumnBaseAttributes(alterHistoryColumn);
+
+        // write the apply to history table
+        writer.apply().append(histColumnDdl).endOfStatement();
+      }
+
+      historyDdl.enableSystemVersioning(writer.apply(), tableName, historyTable, false, this.generateUniqueDdl);
+    }
+  }
+
+  @Override
+  public void generate(DdlWrite writer, DropColumn dropColumn) throws IOException {
+    String tableName = dropColumn.getTableName();
+    MTable table = writer.getTable(tableName);
+    if (table == null) {
+      super.generate(writer, dropColumn);
+      return;
+    }
+
+    boolean manageSystemVersioning = isTrue(table.isWithHistory()) && historyDdl.isSystemVersioningEnabled(tableName);
+
+    if (manageSystemVersioning) {
+      historyDdl.disableSystemVersioning(writer.apply(), tableName, this.generateUniqueDdl);
+    }
+
+    super.generate(writer, dropColumn);
+
+    if (manageSystemVersioning) {
+      // also drop from the history table
+      String historyTable = historyTable(tableName);
+      alterTableDropColumn(writer.apply(), historyTable, dropColumn.getColumnName());
+
+      historyDdl.enableSystemVersioning(writer.apply(), tableName, historyTable, false, this.generateUniqueDdl);
+    }
+  }
+
+}

src/main/java/io/ebeaninternal/dbmigration/ddlgeneration/platform/PlatformDdl.java

@@ -78,12 +78,16 @@ public class PlatformDdl {
   protected String dropIndexIfExists = "drop index if exists ";
 
   protected String alterColumn = "alter column";
+  
+  protected String alterColumnSuffix = "";
 
   protected String dropUniqueConstraint = "drop constraint";
 
   protected String addConstraint = "add constraint";
 
   protected String addColumn = "add column";
+  
+  protected String addColumnSuffix = "";
 
   protected String columnSetType = "";
 
@@ -96,6 +100,12 @@ public class PlatformDdl {
   protected String columnSetNull = "set null";
 
   protected String updateNullWithDefault = "update ${table} set ${column} = ${default} where ${column} is null";
+  
+  protected String createTable = "create table";
+  
+  protected String dropColumn = "drop column";
+  
+  protected String dropColumnSuffix = "";
 
   /**
    * Set false for MsSqlServer to allow multiple nulls for OneToOne mapping.
@@ -458,6 +468,7 @@ public class PlatformDdl {
       if (isTrue(column.isNotnull())) {
         buffer.append(" not null");
       }
+      buffer.append(addColumnSuffix);
       buffer.endOfStatement();
 
       // check constraints cannot be added in one statement for h2
@@ -467,14 +478,15 @@ public class PlatformDdl {
         buffer.append(ddl).endOfStatement();
       }
     } else {
+      buffer.append(addColumnSuffix);
       buffer.endOfStatement();
     }
 
   }
 
   public void alterTableDropColumn(DdlBuffer buffer, String tableName, String columnName) throws IOException {
-    buffer.append("alter table ").append(tableName).append(" drop column ").append(columnName)
-    .endOfStatement();
+    buffer.append("alter table ").append(tableName).append(" ").append(dropColumn).append(" ").append(columnName)
+    .append(dropColumnSuffix).endOfStatement();
   }
 
   /**
@@ -494,19 +506,19 @@ public class PlatformDdl {
    */
   public String alterColumnType(String tableName, String columnName, String type) {
 
-    return "alter table " + tableName + " " + alterColumn + " " + columnName + " " + columnSetType + convert(type, false);
+    return "alter table " + tableName + " " + alterColumn + " " + columnName + " " + columnSetType + convert(type, false) + alterColumnSuffix;
   }
 
   /**
    * Alter a column adding or removing the not null constraint.
    * <p>
-   * Note that that MySql and SQL Server instead use alterColumnBaseAttributes()
+   * Note that that MySql, SQL Server, and HANA instead use alterColumnBaseAttributes()
    * </p>
    */
   public String alterColumnNotnull(String tableName, String columnName, boolean notnull) {
 
     String suffix = notnull ? columnSetNotnull : columnSetNull;
-    return "alter table " + tableName + " " + alterColumn + " " + columnName + " " + suffix;
+    return "alter table " + tableName + " " + alterColumn + " " + columnName + " " + suffix + alterColumnSuffix;
   }
 
   /**
@@ -522,17 +534,17 @@ public class PlatformDdl {
    */
   public String alterColumnDefaultValue(String tableName, String columnName, String defaultValue) {
     String suffix = DdlHelp.isDropDefault(defaultValue) ? columnDropDefault : columnSetDefault + " " + convertDefaultValue(defaultValue);
-    return "alter table " + tableName + " " + alterColumn + " " + columnName + " " + suffix;
+    return "alter table " + tableName + " " + alterColumn + " " + columnName + " " + suffix + alterColumnSuffix;
   }
 
   /**
    * Alter column setting both the type and not null constraint.
    * <p>
-   * Used by MySql and SQL Server as these require both column attributes to be set together.
+   * Used by MySql, SQL Server, and HANA as these require both column attributes to be set together.
    * </p>
    */
   public String alterColumnBaseAttributes(AlterColumn alter) {
-    // by default do nothing, only used by mysql and sql server as they can only
+    // by default do nothing, only used by mysql, sql server, and HANA as they can only
     // modify the column with the full column definition
     return null;
   }
@@ -662,6 +674,17 @@ public class PlatformDdl {
   public void unlockTables(DdlBuffer buffer, Collection<String> tables) throws IOException {
 
   }
+  
+  /**
+   * Returns the database-specific "create table" command prefix. For HANA this is
+   * either "create column table" or "create row table", for all other databases
+   * it is "create table".
+   * 
+   * @return The "create table" command prefix
+   */
+  public String getCreateTableCommandPrefix() {
+    return createTable;
+  }
 
   public boolean suppressPrimaryKeyOnPartition() {
     return false;

src/main/java/io/ebeaninternal/server/core/DatabasePlatformFactory.java

@@ -5,6 +5,7 @@ import io.ebean.config.dbplatform.DatabasePlatform;
 import io.ebean.config.dbplatform.cockroach.CockroachPlatform;
 import io.ebean.config.dbplatform.db2.DB2Platform;
 import io.ebean.config.dbplatform.h2.H2Platform;
+import io.ebean.config.dbplatform.hana.HanaPlatform;
 import io.ebean.config.dbplatform.hsqldb.HsqldbPlatform;
 import io.ebean.config.dbplatform.mysql.MySqlPlatform;
 import io.ebean.config.dbplatform.oracle.OraclePlatform;
@@ -106,6 +107,9 @@ public class DatabasePlatformFactory {
     if (dbName.equals("sqlite")) {
       return new SQLitePlatform();
     }
+    if (dbName.equals("hana")) {
+      return new HanaPlatform();
+    }
 
     throw new RuntimeException("database platform " + dbName + " is not known?");
   }
@@ -156,6 +160,8 @@ public class DatabasePlatformFactory {
       return new DB2Platform();
     } else if (dbProductName.contains("sql anywhere")) {
       return new SqlAnywherePlatform();
+    }else if (dbProductName.contains("hdb")) {
+      return new HanaPlatform();
     }
 
     // use the standard one

src/main/java/io/ebeaninternal/server/core/PlatformDdlBuilder.java

@@ -4,6 +4,7 @@ import io.ebean.config.dbplatform.DatabasePlatform;
 import io.ebeaninternal.dbmigration.ddlgeneration.platform.CockroachDdl;
 import io.ebeaninternal.dbmigration.ddlgeneration.platform.DB2Ddl;
 import io.ebeaninternal.dbmigration.ddlgeneration.platform.H2Ddl;
+import io.ebeaninternal.dbmigration.ddlgeneration.platform.HanaColumnStoreDdl;
 import io.ebeaninternal.dbmigration.ddlgeneration.platform.HsqldbDdl;
 import io.ebeaninternal.dbmigration.ddlgeneration.platform.MySqlDdl;
 import io.ebeaninternal.dbmigration.ddlgeneration.platform.Oracle10Ddl;
@@ -47,6 +48,8 @@ public class PlatformDdlBuilder {
       case SQLSERVER17:
       case SQLSERVER:
         return new SqlServerDdl(platform);
+      case HANA:
+        return new HanaColumnStoreDdl(platform);
       default:
         return new PlatformDdl(platform);
     }

src/main/java/io/ebeaninternal/server/expression/platform/DbExpressionHandlerFactory.java

@@ -25,6 +25,8 @@ public class DbExpressionHandlerFactory {
       case SQLSERVER17:
       case SQLSERVER:
         return new SqlServerDbExpression();
+      case HANA:
+        return new HanaDbExpression();
       default:
         return new BasicDbExpression();
     }

src/main/java/io/ebeaninternal/server/expression/platform/HanaDbExpression.java

@@ -0,0 +1,60 @@
+package io.ebeaninternal.server.expression.platform;
+
+import io.ebeaninternal.api.SpiExpressionRequest;
+import io.ebeaninternal.server.expression.BitwiseOp;
+import io.ebeaninternal.server.expression.Op;
+
+/**
+ * HANA handling of platform specific expressions.
+ */
+public class HanaDbExpression extends BaseDbExpression {
+
+  @Override
+  public void bitwise(SpiExpressionRequest request, String propName, BitwiseOp operator, long flags, String compare,
+      long match) {
+    bitwiseFunction(request, propName, operator, compare);
+  }
+
+  @Override
+  public void json(SpiExpressionRequest request, String propName, String path, Op operator, Object value) {
+    request.append("json_value(").append(propName).append(", '$.").append(path).append("')");
+    request.append(operator.bind());
+  }
+
+  @Override
+  public void arrayIsEmpty(SpiExpressionRequest request, String propName, boolean empty) {
+    request.append("cardinality(").append(propName).append(")");
+    if (empty) {
+      request.append(" = 0");
+    } else {
+      request.append(" <> 0");
+    }
+  }
+
+  @Override
+  public String concat(String property0, String separator, String property1, String suffix) {
+    StringBuilder sb = new StringBuilder();
+    sb.append("concat(").append(property0).append(", '").append(separator).append("'||").append(property1);
+    if (suffix != null && !suffix.isEmpty()) {
+      sb.append("||'").append(suffix).append('\'');
+    }
+    sb.append(")");
+    return sb.toString();
+  }
+
+  @Override
+  public void arrayContains(SpiExpressionRequest request, String propName, boolean contains, Object... values) {
+    for (int i = 0; i < values.length; i++) {
+      if (i > 0) {
+        request.append(" and ");
+      }
+      request.append("(?");
+      if (!contains) {
+        request.append(" not ");
+      }
+      request.append(" member of ").append(propName).append(")");
+
+    }
+  }
+
+}

src/main/java/io/ebeaninternal/server/query/CQueryBuilder.java

@@ -426,7 +426,7 @@ class CQueryBuilder {
     try {
       // For SqlServer we need either "selectMethod=cursor" in the connection string or fetch explicitly a cursorable
       // statement here by specifying ResultSet.CONCUR_UPDATABLE
-      PreparedStatement statement = connection.prepareStatement(sql,ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_UPDATABLE);
+      PreparedStatement statement = connection.prepareStatement(sql,ResultSet.TYPE_FORWARD_ONLY, dbPlatform.isSupportsResultSetConcurrencyModeUpdatable() ? ResultSet.CONCUR_UPDATABLE : ResultSet.CONCUR_READ_ONLY);
       predicates.bind(statement, connection);
 
       ResultSet resultSet = statement.executeQuery();