///|
using @morm/engine {type SortDirection, type Sort, type Pageable, type Page}

///|
pub(open) trait Entity: ToJson + @morm/engine.FromParam {
  fn table(Self) -> Table
}

///|
pub fn current_timestamp_utc() -> @time.ZonedDateTime {
  let ms = @env.now()
  let seconds = (ms / 1000UL).reinterpret_as_int64()
  let nanosecond = (ms % 1000UL).to_int() * 1_000_000
  @time.unix(seconds, nanosecond~) catch {
    _ => panic()
  }
}

///|
pub fn current_plain_date_time_utc() -> @time.PlainDateTime {
  current_timestamp_utc().to_plain_date_time()
}

///|
pub fn add_column(table : Table, col : Column) -> Table {
  let new_columns = []
  for column in table.columns {
    new_columns.push(column)
  }
  new_columns.push(col)
  { ..table, columns: FixedArray::from_array(new_columns) }
}

///|
pub fn add_column_engine_option(
  col : Column,
  key : String,
  value : String,
) -> Column {
  let options = col.engine_options.copy()
  let values = match options.get(key) {
    Some(existing) => existing + [value]
    None => [value]
  }
  options.set(key, values)
  { ..col, engine_options: options }
}

///|
pub fn add_index(table : Table, idx : Index) -> Table {
  let new_indexes = []
  for index in table.indexes {
    new_indexes.push(index)
  }
  new_indexes.push(idx)
  { ..table, indexes: FixedArray::from_array(new_indexes) }
}

///|
pub fn add_foreign_key(table : Table, fk : ForeignKey) -> Table {
  let new_fks = []
  for foreign_key in table.foreign_keys {
    new_fks.push(foreign_key)
  }
  new_fks.push(fk)
  { ..table, foreign_keys: FixedArray::from_array(new_fks) }
}

///|
pub fn table_comment(table : Table, text : String) -> Table {
  { ..table, comment: text }
}

///|
pub fn table_schema(table : Table, schema : String) -> Table {
  { ..table, schema: Some(schema) }
}

///|
pub fn table_catalog(table : Table, catalog : String) -> Table {
  { ..table, catalog: Some(catalog) }
}

///|
pub fn table_temporary(table : Table, temporary : Bool) -> Table {
  { ..table, temporary, }
}

///|
pub fn table_if_not_exists(table : Table, if_not_exists : Bool) -> Table {
  { ..table, if_not_exists, }
}

///|
pub fn table_engine(table : Table, engine : String) -> Table {
  { ..table, engine: Some(engine) }
}

///|
pub fn table_tablespace(table : Table, tablespace : String) -> Table {
  { ..table, tablespace: Some(tablespace) }
}

///|
pub fn add_table_engine_option(
  table : Table,
  key : String,
  value : String,
) -> Table {
  let options = table.engine_options.copy()
  let values = match options.get(key) {
    Some(existing) => existing + [value]
    None => [value]
  }
  options.set(key, values)
  { ..table, engine_options: options }
}

///|
pub fn table_create_suffix(table : Table, suffix : String) -> Table {
  add_table_engine_option(table, "create_suffix", suffix)
}

///|
pub fn mysql_table_row_format(table : Table, row_format : String) -> Table {
  add_table_engine_option(table, "mysql.row_format", row_format)
}

///|
pub fn mysql_table_key_block_size(table : Table, key_block_size : Int) -> Table {
  add_table_engine_option(
    table,
    "mysql.key_block_size",
    key_block_size.to_string(),
  )
}

///|
pub fn mysql_table_compression(table : Table, compression : String) -> Table {
  add_table_engine_option(table, "mysql.compression", compression)
}

///|
pub fn pg_table_unlogged(table : Table) -> Table {
  add_table_engine_option(table, "pgsql.unlogged", "true")
}

///|
pub fn pg_table_with(table : Table, with_clause : String) -> Table {
  add_table_engine_option(table, "pgsql.with", with_clause)
}

///|
pub fn pg_table_on_commit(table : Table, on_commit : String) -> Table {
  add_table_engine_option(table, "pgsql.on_commit", on_commit)
}

///|
pub fn pg_schema_authorization(table : Table, role : String) -> Table {
  add_table_engine_option(table, "pgsql.schema.authorization", role)
}

///|
pub fn pg_table_owner(table : Table, role : String) -> Table {
  add_table_engine_option(table, "pgsql.table.owner", role)
}

///|
fn pg_grant_payload(
  role : String,
  privileges : String,
  with_grant_option : Bool,
  for_role : String?,
) -> String {
  let mut payload = "role=\{role};privileges=\{privileges};grant_option=\{with_grant_option}"
  if for_role is Some(owner_role) {
    payload = payload + ";for_role=\{owner_role}"
  }
  payload
}

///|
pub fn pg_schema_grant(
  table : Table,
  role : String,
  privileges : String,
  with_grant_option : Bool,
) -> Table {
  add_table_engine_option(
    table,
    "pgsql.grant.schema",
    pg_grant_payload(role, privileges, with_grant_option, None),
  )
}

///|
pub fn pg_table_grant(
  table : Table,
  role : String,
  privileges : String,
  with_grant_option : Bool,
) -> Table {
  add_table_engine_option(
    table,
    "pgsql.grant.table",
    pg_grant_payload(role, privileges, with_grant_option, None),
  )
}

///|
pub fn pg_sequence_grant(
  table : Table,
  role : String,
  privileges : String,
  with_grant_option : Bool,
) -> Table {
  add_table_engine_option(
    table,
    "pgsql.grant.table_sequence",
    pg_grant_payload(role, privileges, with_grant_option, None),
  )
}

///|
pub fn pg_schema_sequence_grant(
  table : Table,
  role : String,
  privileges : String,
  with_grant_option : Bool,
) -> Table {
  add_table_engine_option(
    table,
    "pgsql.grant.sequence",
    pg_grant_payload(role, privileges, with_grant_option, None),
  )
}

///|
pub fn pg_default_table_grant(
  table : Table,
  role : String,
  privileges : String,
  with_grant_option : Bool,
  for_role : String?,
) -> Table {
  add_table_engine_option(
    table,
    "pgsql.default_grant.table",
    pg_grant_payload(role, privileges, with_grant_option, for_role),
  )
}

///|
pub fn pg_default_sequence_grant(
  table : Table,
  role : String,
  privileges : String,
  with_grant_option : Bool,
  for_role : String?,
) -> Table {
  add_table_engine_option(
    table,
    "pgsql.default_grant.sequence",
    pg_grant_payload(role, privileges, with_grant_option, for_role),
  )
}

///|
pub fn oracle_table_organization(table : Table, organization : String) -> Table {
  add_table_engine_option(table, "oracle.organization", organization)
}

///|
pub fn oracle_table_on_commit(table : Table, on_commit : String) -> Table {
  add_table_engine_option(table, "oracle.on_commit", on_commit)
}

///|
pub fn sqlserver_table_memory_optimized(table : Table) -> Table {
  add_table_engine_option(table, "sqlserver.memory_optimized", "true")
}

///|
pub fn sqlserver_table_durability(table : Table, durability : String) -> Table {
  add_table_engine_option(table, "sqlserver.durability", durability)
}

///|
pub fn sqlserver_table_filegroup(table : Table, filegroup : String) -> Table {
  add_table_engine_option(table, "sqlserver.filegroup", filegroup)
}

///|
pub fn sqlserver_table_textimage_on(table : Table, filegroup : String) -> Table {
  add_table_engine_option(table, "sqlserver.textimage_on", filegroup)
}

///|
pub fn sqlite_table_without_rowid(table : Table) -> Table {
  add_table_engine_option(table, "sqlite.without_rowid", "true")
}

///|
pub fn sqlite_table_strict(table : Table) -> Table {
  add_table_engine_option(table, "sqlite.strict", "true")
}

///|
// ---- 索引构建器函数 ----
pub fn new_index(name : String, columns : FixedArray[String]) -> Index {
  { name, index_type: Index, columns, comment: "" }
}

///|
pub fn new_unique_index(name : String, columns : FixedArray[String]) -> Index {
  { name, index_type: Unique, columns, comment: "" }
}

///|
pub fn new_primary_index(columns : FixedArray[String]) -> Index {
  { name: "PRIMARY", index_type: Primary, columns, comment: "" }
}

///|
// ---- 外键构建器函数 ----
pub fn new_foreign_key(
  name : String,
  column : String,
  referenced_table : String,
  referenced_column : String,
) -> ForeignKey {
  {
    name,
    column,
    referenced_table,
    referenced_column,
    on_delete: "RESTRICT",
    on_update: "RESTRICT",
  }
}

///|
pub fn on_delete_cascade(fk : ForeignKey) -> ForeignKey {
  {
    name: fk.name,
    column: fk.column,
    referenced_table: fk.referenced_table,
    referenced_column: fk.referenced_column,
    on_delete: "CASCADE",
    on_update: fk.on_update,
  }
}

///|
pub fn on_update_cascade(fk : ForeignKey) -> ForeignKey {
  {
    name: fk.name,
    column: fk.column,
    referenced_table: fk.referenced_table,
    referenced_column: fk.referenced_column,
    on_delete: fk.on_delete,
    on_update: "CASCADE",
  }
}

///|
// ---- 类型推断辅助函数 ----
pub fn infer_column_type(value : Json) -> ColumnType {
  match value {
    Number(n, ..) => if n.to_int().to_double() == n { Int } else { Double }
    String(_) => VarChar(255)
    _ => Text
  }
}

///|
fn[T : ToJson] write_json_show(value : T, logger : &Logger) -> Unit {
  logger.write_string(value.to_json().stringify())
}

///|
// ---- DB/Driver error & result types ----
pub(all) struct DbError {
  message : String
} derive(ToJson)

///|
pub impl Show for DbError with fn output(self : DbError, logger : &Logger) {
  write_json_show(self, logger)
}

///|
pub type DbResult[T] = Result[T, DbError]

///|
// ---- Driver 接口定义 ----
pub(open) trait Driver {
  fn open(Self, dsn : String) -> DbResult[Self]
  fn close(Self) -> DbResult[Self]
  fn ping(Self) -> DbResult[Unit]
  fn exec(Self, sql : String, args : FixedArray[@morm/engine.Param]) -> DbResult[
    Unit,
  ]
  fn query(Self, sql : String, args : FixedArray[@morm/engine.Param]) -> DbResult[
    FixedArray[Json],
  ]
}

///|
// ---- 简易查询构建器 ----
pub struct Query {
  stmt : @morm/engine.SelectStatement
} derive(Eq, ToJson, FromJson)

///|
pub impl Show for Query with fn output(self : Query, logger : &Logger) {
  write_json_show(self, logger)
}

///|
pub type OrderBy = @morm/engine.OrderBy

///|
pub type Where = @morm/engine.Where

///|
pub type WhereType = @morm/engine.WhereType

///|
fn json_to_param(value : Json) -> @morm/engine.Param {
  match value {
    Null => Null
    True => Bool(true)
    False => Bool(false)
    Number(n, ..) => Double(n)
    String(s) => String(s)
    _ => Json(value)
  }
}

///|
fn json_storage_for_column(col : Column) -> String {
  match col.engine_options.get("json.storage") {
    Some([value, ..]) => value
    _ => ""
  }
}

///|
fn json_to_param_for_column(value : Json, col : Column) -> @morm/engine.Param {
  match col.column_type {
    Json | JsonB =>
      match json_storage_for_column(col) {
        "bson" => Bytes(@bson.to_bytes(value))
        "blob" => Bytes(@utf8.encode(value.stringify()))
        _ => Json(value)
      }
    _ => json_to_param(value)
  }
}

///|
fn[O : ToJson] collect_owner_key_params(
  owners : FixedArray[O],
  owner_key : String,
) -> FixedArray[@morm/engine.Param] {
  let seen : Map[String, Bool] = Map([])
  let params : Array[@morm/engine.Param] = []
  for owner in owners {
    if owner.to_json() is Object(obj) {
      if obj.get(owner_key) is Some(v) {
        if v is Null {
          continue
        }
        let key = v.stringify()
        if seen.get(key) is None {
          seen.set(key, true)
          params.push(json_to_param(v))
        }
      }
    }
  }
  FixedArray::from_array(params)
}

///|
pub(all) enum AssociationKind {
  BelongsTo
  HasOne
  HasMany
  ManyToMany
} derive(Eq, ToJson, FromJson)

///|
pub impl Show for AssociationKind with fn output(
  self : AssociationKind,
  logger : &Logger,
) {
  write_json_show(self, logger)
}

///|
pub(all) struct Association {
  name : String
  kind : AssociationKind
  owner_key : String
  target_table : String
  target_key : String
  join_table : String?
  join_owner_key : String?
  join_target_key : String?
} derive(Eq, ToJson, FromJson)

///|
pub impl Show for Association with fn output(
  self : Association,
  logger : &Logger,
) {
  write_json_show(self, logger)
}

///|
pub fn belongs_to(
  name : String,
  owner_key : String,
  target_table : String,
  target_key : String,
) -> Association {
  {
    name,
    kind: BelongsTo,
    owner_key,
    target_table,
    target_key,
    join_table: None,
    join_owner_key: None,
    join_target_key: None,
  }
}

///|
pub fn has_one(
  name : String,
  owner_key : String,
  target_table : String,
  target_key : String,
) -> Association {
  {
    name,
    kind: HasOne,
    owner_key,
    target_table,
    target_key,
    join_table: None,
    join_owner_key: None,
    join_target_key: None,
  }
}

///|
pub fn has_many(
  name : String,
  owner_key : String,
  target_table : String,
  target_key : String,
) -> Association {
  {
    name,
    kind: HasMany,
    owner_key,
    target_table,
    target_key,
    join_table: None,
    join_owner_key: None,
    join_target_key: None,
  }
}

///|
pub fn many_to_many(
  name : String,
  owner_key : String,
  target_table : String,
  target_key : String,
  join_table : String,
  join_owner_key : String,
  join_target_key : String,
) -> Association {
  {
    name,
    kind: ManyToMany,
    owner_key,
    target_table,
    target_key,
    join_table: Some(join_table),
    join_owner_key: Some(join_owner_key),
    join_target_key: Some(join_target_key),
  }
}

///|
async fn[E : @morm/engine.Engine] select_by_eq(
  engine : E,
  table : String,
  key : String,
  value : @morm/engine.Param,
) -> FixedArray[Map[String, @morm/engine.Param]] {
  let stmt : @morm/engine.Statement = Select({
    select: "",
    from: table,
    joins: [],
    where_: [{ col: key, value, ty: Eq }],
    order_by: [],
    limit: None,
    offset: None,
  })
  engine.exec(@morm/engine.Statement(stmt)).rows
}

///|
fn row_param(row : Map[String, @morm/engine.Param]) -> @morm/engine.Param {
  let obj : Map[String, @morm/engine.Param] = Map([])
  for key in row.keys() {
    if row.get(key) is Some(value) {
      obj.set(key.to_lower(), value)
    }
  }
  Object(obj)
}

///|
fn page_row_param(row : Map[String, @morm/engine.Param]) -> @morm/engine.Param {
  let mut count = 0
  let mut only : @morm/engine.Param? = None
  for key in row.keys() {
    count = count + 1
    if count == 1 {
      if row.get(key) is Some(value) {
        only = Some(value)
      }
    }
  }
  if count == 1 {
    match only {
      Some(value) => value
      None => Null
    }
  } else {
    row_param(row)
  }
}

///|
fn page_total_elements(
  row : Map[String, @morm/engine.Param],
  fallback : Int,
) -> Int {
  let first_key : String? = None
  let mut key_opt = first_key
  for key in row.keys() {
    key_opt = Some(key)
    break
  }
  match key_opt {
    Some(k) =>
      match row.get(k) {
        Some(Int(n)) => n
        Some(Int16(n)) => n.to_int()
        Some(Int64(n)) => n.to_int()
        Some(BigInt(n)) => n.to_int()
        Some(UInt16(n)) => n.to_int()
        Some(Byte(n)) => n.to_int()
        Some(Json(Number(n, ..))) => n.to_int()
        _ => fallback
      }
    None => fallback
  }
}

///|
fn[T : @morm/engine.FromParam] decode_page_row(
  row : Map[String, @morm/engine.Param],
) -> T {
  try @morm/engine.from_param(row_param(row)) catch {
    _ => try! @morm/engine.from_param(page_row_param(row))
  } noraise {
    value => value
  }
}

///|
pub async fn[E : @morm/engine.Engine, O : ToJson, T : @morm/engine.FromParam] preload_belongs_to(
  engine : E,
  owners : FixedArray[O],
  owner_key : String,
  target_table : String,
  target_key : String,
) -> Map[String, T] {
  let key_params = collect_owner_key_params(owners, owner_key)
  let out : Map[String, T] = Map([])
  for key_param in key_params {
    let key = key_param.to_json().stringify()
    let rows = select_by_eq(engine, target_table, target_key, key_param)
    if rows is [row, ..] {
      try @morm/engine.from_param(row_param(row)) catch {
        _ => ()
      } noraise {
        entity => out.set(key, entity)
      }
    }
  }
  out
}

///|
pub async fn[E : @morm/engine.Engine, O : ToJson, T : @morm/engine.FromParam] preload_has_one(
  engine : E,
  owners : FixedArray[O],
  owner_key : String,
  target_table : String,
  target_key : String,
) -> Map[String, T] {
  let key_params = collect_owner_key_params(owners, owner_key)
  let out : Map[String, T] = Map([])
  for key_param in key_params {
    let key = key_param.to_json().stringify()
    let rows = select_by_eq(engine, target_table, target_key, key_param)
    if rows is [row, ..] {
      try @morm/engine.from_param(row_param(row)) catch {
        _ => ()
      } noraise {
        entity => out.set(key, entity)
      }
    }
  }
  out
}

///|
pub async fn[E : @morm/engine.Engine, O : ToJson, T : @morm/engine.FromParam] preload_has_many(
  engine : E,
  owners : FixedArray[O],
  owner_key : String,
  target_table : String,
  target_key : String,
) -> Map[String, Array[T]] {
  let key_params = collect_owner_key_params(owners, owner_key)
  let out : Map[String, Array[T]] = Map([])
  for key_param in key_params {
    let key = key_param.to_json().stringify()
    let rows = select_by_eq(engine, target_table, target_key, key_param)
    let items : Array[T] = []
    for row in rows {
      try @morm/engine.from_param(row_param(row)) catch {
        _ => ()
      } noraise {
        entity => items.push(entity)
      }
    }
    out.set(key, items)
  }
  out
}

///|
pub async fn[E : @morm/engine.Engine, O : ToJson, T : @morm/engine.FromParam] preload_many_to_many(
  engine : E,
  owners : FixedArray[O],
  owner_key : String,
  target_table : String,
  target_key : String,
  join_table : String,
  join_owner_key : String,
  join_target_key : String,
) -> Map[String, Array[T]] {
  let key_params = collect_owner_key_params(owners, owner_key)
  let out : Map[String, Array[T]] = Map([])
  for owner_param in key_params {
    let key = owner_param.to_json().stringify()
    let join_rows = select_by_eq(
      engine, join_table, join_owner_key, owner_param,
    )
    let items : Array[T] = []
    for join_row in join_rows {
      if join_row.get(join_target_key) is Some(target_value) {
        let target_rows = select_by_eq(
          engine, target_table, target_key, target_value,
        )
        for target_row in target_rows {
          try @morm/engine.from_param(row_param(target_row)) catch {
            _ => ()
          } noraise {
            entity => items.push(entity)
          }
        }
      }
    }
    out.set(key, items)
  }
  out
}

///|
pub(all) enum PreloadResult[T] {
  One(Map[String, T])
  Many(Map[String, Array[T]])
}

///|
pub async fn[E : @morm/engine.Engine, O : ToJson, T : @morm/engine.FromParam] preload(
  engine : E,
  owners : FixedArray[O],
  association : Association,
) -> PreloadResult[T] {
  match association.kind {
    BelongsTo =>
      One(
        preload_belongs_to(
          engine,
          owners,
          association.owner_key,
          association.target_table,
          association.target_key,
        ),
      )
    HasOne =>
      One(
        preload_has_one(
          engine,
          owners,
          association.owner_key,
          association.target_table,
          association.target_key,
        ),
      )
    HasMany =>
      Many(
        preload_has_many(
          engine,
          owners,
          association.owner_key,
          association.target_table,
          association.target_key,
        ),
      )
    ManyToMany =>
      Many(
        preload_many_to_many(
          engine,
          owners,
          association.owner_key,
          association.target_table,
          association.target_key,
          association.join_table.unwrap_or(""),
          association.join_owner_key.unwrap_or(""),
          association.join_target_key.unwrap_or(""),
        ),
      )
  }
}

///|
pub fn select_from(table : String) -> Query {
  {
    stmt: {
      select: "*",
      from: table,
      joins: [],
      where_: [],
      order_by: [],
      limit: None,
      offset: None,
    },
  }
}

///|
pub fn select_raw(table : String, cols_expr : String) -> Query {
  {
    stmt: {
      select: cols_expr,
      from: table,
      joins: [],
      where_: [],
      order_by: [],
      limit: None,
      offset: None,
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] Query::where_eq(
  self : Query,
  col : String,
  value : V,
) -> Query {
  let where_ = self.stmt.where_ + [{ col, value: value.to_param(), ty: Eq }]
  { stmt: { ..self.stmt, where_, } }
}

///|
pub fn[V : @morm/engine.ToParam] Query::where_gt(
  self : Query,
  col : String,
  value : V,
) -> Query {
  let where_ = self.stmt.where_ + [{ col, value: value.to_param(), ty: Gt }]
  { stmt: { ..self.stmt, where_, } }
}

///|
pub fn[V : @morm/engine.ToParam] Query::where_ne(
  self : Query,
  col : String,
  value : V,
) -> Query {
  let where_ = self.stmt.where_ + [{ col, value: value.to_param(), ty: Ne }]
  { stmt: { ..self.stmt, where_, } }
}

///|
pub fn[V : @morm/engine.ToParam] Query::where_lt(
  self : Query,
  col : String,
  value : V,
) -> Query {
  let where_ = self.stmt.where_ + [{ col, value: value.to_param(), ty: Lt }]
  { stmt: { ..self.stmt, where_, } }
}

///|
pub fn[V : @morm/engine.ToParam] Query::where_like(
  self : Query,
  col : String,
  pattern : V,
) -> Query {
  let where_ = self.stmt.where_ + [{ col, value: pattern.to_param(), ty: Like }]
  { stmt: { ..self.stmt, where_, } }
}

///|
pub fn[V : @morm/engine.ToParam] Query::where_gte(
  self : Query,
  col : String,
  value : V,
) -> Query {
  let where_ = self.stmt.where_ + [{ col, value: value.to_param(), ty: Gte }]
  { stmt: { ..self.stmt, where_, } }
}

///|
pub fn[V : @morm/engine.ToParam] Query::where_lte(
  self : Query,
  col : String,
  value : V,
) -> Query {
  let where_ = self.stmt.where_ + [{ col, value: value.to_param(), ty: Lte }]
  { stmt: { ..self.stmt, where_, } }
}

///|
pub fn Query::order_by(self : Query, order_by : OrderBy) -> Query {
  { stmt: { ..self.stmt, order_by: self.stmt.order_by + [order_by] } }
}

///|
pub fn Query::join(self : Query, join_sql : String) -> Query {
  { stmt: { ..self.stmt, joins: self.stmt.joins + [join_sql] } }
}

///|
pub fn Query::limit(self : Query, n : Int) -> Query {
  { stmt: { ..self.stmt, limit: Some(n) } }
}

///|
pub fn Query::offset(self : Query, n : Int) -> Query {
  { stmt: { ..self.stmt, offset: Some(n) } }
}

///|
pub impl @morm/engine.QueryBuilder for Query with fn to_query(self) -> @morm/engine.Query {
  Statement(Select(self.stmt))
}

///|
// ---- INSERT 构建器 ----
pub struct InsertQuery {
  stmt : @morm/engine.InsertStatement
} derive(Eq, ToJson, FromJson)

///|
pub impl Show for InsertQuery with fn output(
  self : InsertQuery,
  logger : &Logger,
) {
  write_json_show(self, logger)
}

///|
pub fn insert_into(table : String) -> InsertQuery {
  { stmt: { into: table, columns: [], values: [], rows: [] } }
}

///|
fn insertable_columns(table : Table) -> FixedArray[Column] {
  let columns : Array[Column] = []
  for col in table.columns {
    if !col.auto_increment {
      columns.push(col)
    }
  }
  FixedArray::from_array(columns)
}

///|
pub fn[E : Entity] InsertQuery::from(
  self : InsertQuery,
  entity : E,
) -> InsertQuery {
  let json = ToJson::to_json(entity)
  let table = entity.table()
  let insert_cols = insertable_columns(table)
  let columns = insert_cols.map(c => c.name)
  let values = if json is Object(obj) {
    let encoded : Array[@morm/engine.Param] = []
    for col in insert_cols {
      match obj.get(col.name) {
        Some(v) => encoded.push(json_to_param_for_column(v, col))
        None => encoded.push(Null)
      }
    }
    FixedArray::from_array(encoded)
  } else {
    []
  }
  { stmt: { ..self.stmt, columns, values, rows: [] } }
}

///|
pub fn[E : Entity] InsertQuery::from_many(
  self : InsertQuery,
  entities : FixedArray[E],
) -> InsertQuery {
  if entities.is_empty() {
    return self
  }
  let first = entities[0]
  let table = first.table()
  let insert_cols = insertable_columns(table)
  let columns = insert_cols.map(c => c.name)
  let rows : Array[FixedArray[@morm/engine.Param]] = []
  for entity in entities {
    let json = ToJson::to_json(entity)
    if json is Object(obj) {
      let encoded : Array[@morm/engine.Param] = []
      for col in insert_cols {
        match obj.get(col.name) {
          Some(v) => encoded.push(json_to_param_for_column(v, col))
          None => encoded.push(Null)
        }
      }
      rows.push(FixedArray::from_array(encoded))
    } else {
      rows.push([])
    }
  }
  {
    stmt: {
      ..self.stmt,
      columns,
      values: [],
      rows: FixedArray::from_array(rows),
    },
  }
}

///|
pub fn InsertQuery::columns(
  self : InsertQuery,
  columns : FixedArray[String],
) -> InsertQuery {
  { stmt: { ..self.stmt, columns, } }
}

///|
pub fn InsertQuery::values(
  self : InsertQuery,
  values : FixedArray[&@morm/engine.ToParam],
) -> InsertQuery {
  { stmt: { ..self.stmt, values: values.map(v => v.to_param()), rows: [] } }
}

///|
pub fn InsertQuery::values_many(
  self : InsertQuery,
  rows : FixedArray[FixedArray[&@morm/engine.ToParam]],
) -> InsertQuery {
  {
    stmt: {
      ..self.stmt,
      rows: rows.map(row => row.map(v => v.to_param())),
      values: [],
    },
  }
}

///|
pub impl @morm/engine.QueryBuilder for InsertQuery with fn to_query(self) -> @morm/engine.Query {
  Statement(Insert(self.stmt))
}

///|
// ---- UPSERT 构建器 ----
pub struct UpsertQuery {
  stmt : @morm/engine.UpsertStatement
} derive(Eq, ToJson, FromJson)

///|
pub impl Show for UpsertQuery with fn output(
  self : UpsertQuery,
  logger : &Logger,
) {
  write_json_show(self, logger)
}

///|
pub fn upsert_into(table : String) -> UpsertQuery {
  { stmt: { table, sets: [], conflict_target: [], update_sets: [] } }
}

///|
pub fn[E : Entity] UpsertQuery::from(
  self : UpsertQuery,
  entity : E,
) -> UpsertQuery {
  let json = ToJson::to_json(entity)
  let table = entity.table()
  let sets = if json is Object(obj) {
    let out : Array[@morm/engine.Set] = []
    for col in table.columns {
      if obj.get(col.name) is Some(value) {
        out.push({ col: col.name, value: json_to_param_for_column(value, col) })
      }
    }
    FixedArray::from_array(out)
  } else {
    []
  }
  let mut conflict_cols : FixedArray[String] = []
  match table_primary_key(table) {
    Some(pk) => conflict_cols = [pk]
    None =>
      for idx in table.indexes {
        match idx.index_type {
          Unique => {
            conflict_cols = idx.columns
            break
          }
          _ => ()
        }
      }
  }
  let update_sets : Array[@morm/engine.Set] = []
  for set_item in sets {
    let mut is_conflict_col = false
    for conflict_col in conflict_cols {
      if conflict_col == set_item.col {
        is_conflict_col = true
        break
      }
    }
    if !is_conflict_col {
      update_sets.push(set_item)
    }
  }
  if update_sets.is_empty() {
    for set_item in sets {
      let mut is_conflict_col = false
      for conflict_col in conflict_cols {
        if conflict_col == set_item.col {
          is_conflict_col = true
          break
        }
      }
      if is_conflict_col {
        update_sets.push(set_item)
        break
      }
    }
  }
  {
    stmt: {
      ..self.stmt,
      sets,
      conflict_target: conflict_cols,
      update_sets: FixedArray::from_array(update_sets),
    },
  }
}

///|
pub fn UpsertQuery::set(
  self : UpsertQuery,
  col : String,
  value : @morm/engine.Param,
) -> UpsertQuery {
  { stmt: { ..self.stmt, sets: self.stmt.sets + [{ col, value }] } }
}

///|
pub fn UpsertQuery::on_conflict(
  self : UpsertQuery,
  columns : FixedArray[String],
) -> UpsertQuery {
  { stmt: { ..self.stmt, conflict_target: columns } }
}

///|
pub fn UpsertQuery::do_update_set(
  self : UpsertQuery,
  col : String,
  value : @morm/engine.Param,
) -> UpsertQuery {
  {
    stmt: { ..self.stmt, update_sets: self.stmt.update_sets + [{ col, value }] },
  }
}

///|
pub impl @morm/engine.QueryBuilder for UpsertQuery with fn to_query(self) -> @morm/engine.Query {
  Statement(Upsert(self.stmt))
}

///|
// ---- UPDATE 构建器 ----
pub type Set = @morm/engine.Set

///|
pub struct UpdateQuery {
  stmt : @morm/engine.UpdateStatement
} derive(Eq, ToJson, FromJson)

///|
pub impl Show for UpdateQuery with fn output(
  self : UpdateQuery,
  logger : &Logger,
) {
  write_json_show(self, logger)
}

///|
pub fn update(table : String) -> UpdateQuery {
  { stmt: { table, sets: [], where_: [] } }
}

///|
pub fn[E : Entity] UpdateQuery::from(
  self : UpdateQuery,
  entity : E,
) -> UpdateQuery {
  let json = ToJson::to_json(entity)
  let table = entity.table()
  let sets = if json is Object(obj) {
    let out : Array[@morm/engine.Set] = []
    for col in table.columns {
      if obj.get(col.name) is Some(value) {
        out.push({ col: col.name, value: json_to_param_for_column(value, col) })
      }
    }
    FixedArray::from_array(out)
  } else {
    []
  }
  { stmt: { ..self.stmt, sets, } }
}

///|
pub fn[V : @morm/engine.ToParam] UpdateQuery::set(
  self : UpdateQuery,
  col : String,
  value : V,
) -> UpdateQuery {
  {
    stmt: {
      ..self.stmt,
      sets: self.stmt.sets + [{ col, value: value.to_param() }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] UpdateQuery::where_eq(
  self : UpdateQuery,
  col : String,
  value : V,
) -> UpdateQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Eq }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] UpdateQuery::where_gt(
  self : UpdateQuery,
  col : String,
  value : V,
) -> UpdateQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Gt }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] UpdateQuery::where_ne(
  self : UpdateQuery,
  col : String,
  value : V,
) -> UpdateQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Ne }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] UpdateQuery::where_lt(
  self : UpdateQuery,
  col : String,
  value : V,
) -> UpdateQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Lt }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] UpdateQuery::where_like(
  self : UpdateQuery,
  col : String,
  pattern : V,
) -> UpdateQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: pattern.to_param(), ty: Like }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] UpdateQuery::where_gte(
  self : UpdateQuery,
  col : String,
  value : V,
) -> UpdateQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Gte }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] UpdateQuery::where_lte(
  self : UpdateQuery,
  col : String,
  value : V,
) -> UpdateQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Lte }],
    },
  }
}

///|
pub impl @morm/engine.QueryBuilder for UpdateQuery with fn to_query(self) -> @morm/engine.Query {
  Statement(Update(self.stmt))
}

///|
// ---- DELETE 构建器 ----
pub struct DeleteQuery {
  stmt : @morm/engine.DeleteStatement
} derive(Eq, ToJson, FromJson)

///|
pub impl Show for DeleteQuery with fn output(
  self : DeleteQuery,
  logger : &Logger,
) {
  write_json_show(self, logger)
}

///|
pub fn delete_from(table : String) -> DeleteQuery {
  { stmt: { from: table, where_: [] } }
}

///|
pub fn[V : @morm/engine.ToParam] DeleteQuery::where_eq(
  self : DeleteQuery,
  col : String,
  value : V,
) -> DeleteQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Eq }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] DeleteQuery::where_gt(
  self : DeleteQuery,
  col : String,
  value : V,
) -> DeleteQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Gt }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] DeleteQuery::where_ne(
  self : DeleteQuery,
  col : String,
  value : V,
) -> DeleteQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Ne }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] DeleteQuery::where_lt(
  self : DeleteQuery,
  col : String,
  value : V,
) -> DeleteQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Lt }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] DeleteQuery::where_like(
  self : DeleteQuery,
  col : String,
  pattern : V,
) -> DeleteQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: pattern.to_param(), ty: Like }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] DeleteQuery::where_gte(
  self : DeleteQuery,
  col : String,
  value : V,
) -> DeleteQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Gte }],
    },
  }
}

///|
pub fn[V : @morm/engine.ToParam] DeleteQuery::where_lte(
  self : DeleteQuery,
  col : String,
  value : V,
) -> DeleteQuery {
  {
    stmt: {
      ..self.stmt,
      where_: self.stmt.where_ + [{ col, value: value.to_param(), ty: Lte }],
    },
  }
}

///|
pub impl @morm/engine.QueryBuilder for DeleteQuery with fn to_query(self) -> @morm/engine.Query {
  Statement(Delete(self.stmt))
}

///|
pub struct TxQuery {
  op : @morm/engine.TxOp
} derive(Eq, ToJson, FromJson)

///|
pub impl Show for TxQuery with fn output(self : TxQuery, logger : &Logger) {
  write_json_show(self, logger)
}

///|
pub fn tx_begin() -> TxQuery {
  { op: Begin }
}

///|
pub fn tx_commit() -> TxQuery {
  { op: Commit }
}

///|
pub fn tx_rollback() -> TxQuery {
  { op: Rollback }
}

///|
pub fn tx_savepoint(name : String) -> TxQuery {
  { op: Savepoint(name) }
}

///|
pub fn tx_rollback_to_savepoint(name : String) -> TxQuery {
  { op: RollbackToSavepoint(name) }
}

///|
pub impl @morm/engine.QueryBuilder for TxQuery with fn to_query(self) -> @morm/engine.Query {
  Tx(self.op)
}

///|
pub struct TxBatch[E] {
  engine : E
}

///|
pub async fn[E : @morm/engine.Engine] tx_block(engine : E) -> TxBatch[E] {
  ignore(engine.exec(tx_begin()))
  { engine, }
}

///|
pub async fn[Q : @morm/engine.QueryBuilder, E : @morm/engine.Engine] TxBatch::exec(
  self : TxBatch[E],
  query : Q,
) -> @morm/engine.QueryResult {
  self.engine.exec(query)
}

///|
pub async fn[E : @morm/engine.Engine] TxBatch::commit(
  self : TxBatch[E],
) -> @morm/engine.QueryResult {
  self.engine.exec(tx_commit())
}

///|
pub async fn[E : @morm/engine.Engine] TxBatch::rollback(
  self : TxBatch[E],
) -> @morm/engine.QueryResult {
  self.engine.exec(tx_rollback())
}

///|
pub async fn[E : @morm/engine.Engine] TxBatch::savepoint(
  self : TxBatch[E],
  name : String,
) -> @morm/engine.QueryResult {
  self.engine.exec(tx_savepoint(name))
}

///|
pub async fn[E : @morm/engine.Engine] TxBatch::rollback_to_savepoint(
  self : TxBatch[E],
  name : String,
) -> @morm/engine.QueryResult {
  self.engine.exec(tx_rollback_to_savepoint(name))
}

///|
pub fn[E : Entity] DeleteQuery::from(
  self : DeleteQuery,
  entity : E,
) -> DeleteQuery {
  let table = entity.table()
  let pk_opt = table_primary_key(table)
  match pk_opt {
    Some(pk) => {
      let obj = if ToJson::to_json(entity) is Object(obj) {
        obj
      } else {
        Map([])
      }
      match obj.get(pk) {
        Some(v) => {
          let pv = json_to_param(v)
          {
            stmt: {
              ..self.stmt,
              where_: self.stmt.where_ + [{ col: pk, value: pv, ty: Eq }],
            },
          }
        }
        None => self
      }
    }
    None => self
  }
}

///|
/// Soft delete default scope (boolean convention): deleted = false
pub fn select_from_scoped(table : String, deleted_col? : String) -> Query {
  let col = deleted_col.unwrap_or("deleted")
  select_from(table).where_eq(col, false)
}

///|
pub fn[V : @morm/engine.ToParam] soft_delete_by_id(
  table : String,
  id_col : String,
  id : V,
  deleted_col? : String,
) -> UpdateQuery {
  let col = deleted_col.unwrap_or("deleted")
  update(table).set(col, true).where_eq(id_col, id)
}

///|
pub fn[V : @morm/engine.ToParam] restore_by_id(
  table : String,
  id_col : String,
  id : V,
  deleted_col? : String,
) -> UpdateQuery {
  let col = deleted_col.unwrap_or("deleted")
  update(table).set(col, false).where_eq(id_col, id)
}

///|
/// Hook callbacks helpers.
pub fn[T] before_create(entity : T, callback : (T) -> T) -> T {
  callback(entity)
}

///|
pub fn[T] after_create(entity : T, callback : (T) -> T) -> T {
  callback(entity)
}

///|
pub fn[T] before_update(entity : T, callback : (T) -> T) -> T {
  callback(entity)
}

///|
pub fn[T] after_update(entity : T, callback : (T) -> T) -> T {
  callback(entity)
}

///|
pub fn[T] before_delete(entity : T, callback : (T) -> T) -> T {
  callback(entity)
}

///|
pub fn[T] after_delete(entity : T, callback : (T) -> T) -> T {
  callback(entity)
}

///|
pub fn[T] after_find(entity : T, callback : (T) -> T) -> T {
  callback(entity)
}

///|
pub(all) struct ColumnDiff {
  added : FixedArray[Column]
  removed : FixedArray[Column]
  changed : FixedArray[(String, Column, Column)]
} derive(ToJson, Eq)

///|
pub impl Show for ColumnDiff with fn output(self : ColumnDiff, logger : &Logger) {
  write_json_show(self, logger)
}

///|
pub(all) struct IndexDiff {
  added : FixedArray[Index]
  removed : FixedArray[Index]
  changed : FixedArray[(String, Index, Index)]
} derive(ToJson, FromJson, Eq)

///|
pub impl Show for IndexDiff with fn output(self : IndexDiff, logger : &Logger) {
  write_json_show(self, logger)
}

///|
pub(all) struct ForeignKeyDiff {
  added : FixedArray[ForeignKey]
  removed : FixedArray[ForeignKey]
  changed : FixedArray[(String, ForeignKey, ForeignKey)]
} derive(ToJson, FromJson, Eq)

///|
pub impl Show for ForeignKeyDiff with fn output(
  self : ForeignKeyDiff,
  logger : &Logger,
) {
  write_json_show(self, logger)
}

///|
pub(all) struct TableDiff {
  columns : ColumnDiff
  indexes : IndexDiff
  foreign_keys : ForeignKeyDiff
} derive(ToJson, Eq)

///|
pub impl Show for TableDiff with fn output(self : TableDiff, logger : &Logger) {
  write_json_show(self, logger)
}

///|
fn diff_columns(old_table : Table, new_table : Table) -> ColumnDiff {
  let old_map : Map[String, Column] = Map([])
  let new_map : Map[String, Column] = Map([])
  for c in old_table.columns {
    old_map.set(c.name, c)
  }
  for c in new_table.columns {
    new_map.set(c.name, c)
  }
  let added : Array[Column] = []
  let removed : Array[Column] = []
  let changed : Array[(String, Column, Column)] = []
  for c in new_table.columns {
    if old_map.get(c.name) is Some(old_col) {
      if old_col != c {
        changed.push((c.name, old_col, c))
      }
    } else {
      added.push(c)
    }
  }
  for c in old_table.columns {
    if new_map.get(c.name) is None {
      removed.push(c)
    }
  }
  {
    added: FixedArray::from_array(added),
    removed: FixedArray::from_array(removed),
    changed: FixedArray::from_array(changed),
  }
}

///|
fn diff_indexes(old_table : Table, new_table : Table) -> IndexDiff {
  let old_map : Map[String, Index] = Map([])
  let new_map : Map[String, Index] = Map([])
  for i in old_table.indexes {
    old_map.set(i.name, i)
  }
  for i in new_table.indexes {
    new_map.set(i.name, i)
  }
  let added : Array[Index] = []
  let removed : Array[Index] = []
  let changed : Array[(String, Index, Index)] = []
  for i in new_table.indexes {
    if old_map.get(i.name) is Some(old_idx) {
      if old_idx != i {
        changed.push((i.name, old_idx, i))
      }
    } else {
      added.push(i)
    }
  }
  for i in old_table.indexes {
    if new_map.get(i.name) is None {
      removed.push(i)
    }
  }
  {
    added: FixedArray::from_array(added),
    removed: FixedArray::from_array(removed),
    changed: FixedArray::from_array(changed),
  }
}

///|
fn diff_foreign_keys(old_table : Table, new_table : Table) -> ForeignKeyDiff {
  let old_map : Map[String, ForeignKey] = Map([])
  let new_map : Map[String, ForeignKey] = Map([])
  for fk in old_table.foreign_keys {
    old_map.set(fk.name, fk)
  }
  for fk in new_table.foreign_keys {
    new_map.set(fk.name, fk)
  }
  let added : Array[ForeignKey] = []
  let removed : Array[ForeignKey] = []
  let changed : Array[(String, ForeignKey, ForeignKey)] = []
  for fk in new_table.foreign_keys {
    if old_map.get(fk.name) is Some(old_fk) {
      if old_fk != fk {
        changed.push((fk.name, old_fk, fk))
      }
    } else {
      added.push(fk)
    }
  }
  for fk in old_table.foreign_keys {
    if new_map.get(fk.name) is None {
      removed.push(fk)
    }
  }
  {
    added: FixedArray::from_array(added),
    removed: FixedArray::from_array(removed),
    changed: FixedArray::from_array(changed),
  }
}

///|
pub fn diff_table(old_table : Table, new_table : Table) -> TableDiff {
  {
    columns: diff_columns(old_table, new_table),
    indexes: diff_indexes(old_table, new_table),
    foreign_keys: diff_foreign_keys(old_table, new_table),
  }
}

///|
fn default_driver_registry() -> Map[String, String] {
  let m : Map[String, String] = Map([])
  m.set("mysql", "mysql")
  m.set("mariadb", "mysql")
  m.set("tidb", "mysql")
  m.set("postgres", "pgsql")
  m.set("postgresql", "pgsql")
  m.set("cockroachdb", "pgsql")
  m.set("gaussdb", "pgsql")
  m.set("sqlserver", "sqlserver")
  m.set("mssql", "sqlserver")
  m.set("oracle", "oracle")
  m.set("sqlite", "sqlite3")
  m
}

///|
let driver_registry : Map[String, String] = default_driver_registry()

///|
fn dsn_scheme(dsn : String) -> String? {
  let core = if dsn.strip_prefix("jdbc:") is Some(rest) {
    rest.to_owned()
  } else {
    dsn
  }
  if core.find("://") is Some(i) {
    let scheme = core[:i].to_owned()
    Some(scheme.to_lower())
  } else {
    None
  }
}

///|
pub fn register_driver_scheme(scheme : String, driver_name : String) -> Unit {
  driver_registry.set(scheme.to_lower(), driver_name)
}

///|
pub fn unregister_driver_scheme(scheme : String) -> Unit {
  ignore(driver_registry.remove(scheme.to_lower()))
}

///|
pub fn resolve_driver_name(dsn : String) -> String? {
  if dsn_scheme(dsn) is Some(scheme) {
    driver_registry.get(scheme)
  } else {
    None
  }
}

///|
pub fn pageable(page : Int, size : Int) -> Pageable {
  { page, size, sort: None }
}

///|
pub fn pageable_with_sort(page : Int, size : Int, sort : Sort) -> Pageable {
  { page, size, sort: Some(sort) }
}

///|
pub fn sort(property : String, direction : SortDirection) -> Sort {
  { property, direction }
}

///|
pub fn asc(property : String) -> Sort {
  { property, direction: Asc }
}

///|
pub fn desc(property : String) -> Sort {
  { property, direction: Desc }
}

///|
pub fn[T] page(
  content : FixedArray[T],
  total_elements : Int,
  number : Int,
  size : Int,
) -> Page[T] {
  let normalized_number = if number < 1 { 1 } else { number }
  let total_pages = if size > 0 {
    (total_elements + size - 1) / size
  } else {
    0
  }
  let first = normalized_number <= 1
  let last = total_pages == 0 || normalized_number >= total_pages
  let empty = content.length() == 0
  {
    content,
    total_elements,
    total_pages,
    number: normalized_number,
    size,
    first,
    last,
    empty,
  }
}

///|
pub fn Query::to_count_sql(self : Query) -> String {
  let sel = self.stmt
  let mut count_sql = "SELECT COUNT(*) FROM " + sel.from
  if !sel.where_.is_empty() {
    let where_sql = sel.where_
      .map(w => {
        match (w.ty, w.value) {
          (Eq, Null | Json(Null)) => "\{w.col} IS NULL"
          (Ne, Null | Json(Null)) => "\{w.col} IS NOT NULL"
          _ => {
            let op = match w.ty {
              Eq => "="
              Ne => "!="
              Gt => ">"
              Lt => "<"
              Gte => ">="
              Lte => "<="
              Like => "LIKE"
            }
            "\{w.col} \{op} ?"
          }
        }
      })
      .join(" AND ")
    count_sql = count_sql + " WHERE " + where_sql
  }
  count_sql
}

///|
pub async fn[E : @morm/engine.Engine, T : @morm/engine.FromParam] paginate(
  engine : E,
  query : &@morm/engine.QueryBuilder,
  pageable : Pageable,
) -> Page[T] {
  let page = if pageable.page < 1 { 1 } else { pageable.page }
  let count_query = match query.to_query() {
    Statement(stmt) =>
      match stmt {
        Select(sel) =>
          @morm/engine.Statement(
            Select({ ..sel, order_by: [], limit: None, offset: None }),
          )
        _ => Statement(stmt)
      }
    other => other
  }
  let (base_sql, _) = @morm/engine.render_query_sql(count_query)
  let upper = base_sql.to_upper()
  let count_sql = if upper.contains(" FROM ") {
    let from_idx = upper.find(" FROM ").unwrap()
    "SELECT COUNT(*) " + base_sql[from_idx:].to_owned()
  } else {
    "SELECT COUNT(*)"
  }
  let data_result = engine.page(query, pageable)
  let items : Array[T] = []
  for row in data_result.rows {
    items.push(decode_page_row(row))
  }
  let count_result = engine.exec_raw(
    count_sql,
    @morm/engine.render_query_sql(count_query).1,
  )
  let total_elements = match count_result.rows {
    [row, ..] => page_total_elements(row, items.length())
    _ => items.length()
  }
  let total_pages = if pageable.size > 0 {
    (total_elements + pageable.size - 1) / pageable.size
  } else {
    0
  }
  {
    content: FixedArray::from_array(items),
    total_elements,
    total_pages,
    number: page,
    size: pageable.size,
    first: page <= 1,
    last: total_pages == 0 || page >= total_pages,
    empty: items.length() == 0,
  }
}

///|
pub async fn[E : @morm/engine.Engine, T : @morm/engine.FromParam] paginate_raw(
  engine : E,
  sql : String,
  params : FixedArray[@morm/engine.Param],
  pageable : Pageable,
) -> Page[T] {
  let page = if pageable.page < 1 { 1 } else { pageable.page }
  let upper = sql.to_upper()
  let count_sql = if upper.contains(" FROM ") {
    let from_idx = upper.find(" FROM ").unwrap()
    "SELECT COUNT(*) " + sql[from_idx:].to_owned()
  } else {
    "SELECT COUNT(*)"
  }
  let data_result = engine.page_raw(sql, params, pageable)
  let items : Array[T] = []
  for row in data_result.rows {
    items.push(decode_page_row(row))
  }
  let count_result = engine.exec_raw(count_sql, params)
  let total_elements = match count_result.rows {
    [row, ..] => {
      let first_key : String? = None
      let mut key_opt = first_key
      for key in row.keys() {
        key_opt = Some(key)
        break
      }
      match key_opt {
        Some(k) =>
          match row.get(k) {
            Some(Int(n)) => n
            _ => items.length()
          }
        None => items.length()
      }
    }
    _ => items.length()
  }
  let total_pages = if pageable.size > 0 {
    (total_elements + pageable.size - 1) / pageable.size
  } else {
    0
  }
  {
    content: FixedArray::from_array(items),
    total_elements,
    total_pages,
    number: page,
    size: pageable.size,
    first: page <= 1,
    last: total_pages == 0 || page >= total_pages,
    empty: items.length() == 0,
  }
}