Skip to main content

Migrating Read Models and Events

Code changes after data exists. In an event-sourced application, two kinds of stored data can fall out of step with the code:

  • the rows of a read model, the A of a Projection[A], kept in a projection store;
  • the events, kept in an event store.

They are handled in opposite ways. The rows are derived from the events, so they can be thrown away and rebuilt by replaying the events. The events are the source of truth and are never rewritten, so new code has to read old events as they were stored.

Find your change:

This page is about the data a projection keeps. For the Migration type itself, see Migration; for migrating a relational database while it runs, see Data Migration.

Changing the Read Model​

The rows of a projection are kept in a table named after the row class, in snake_case (UserProfile becomes user_profile), with one column for each field. On every start, the engine checks whether that table still matches the row type, so a change is handled when the new version of the application starts.

How a Change Is Detected — SchemaHash​

object SchemaHash {
  def compute[A: Schema]: String
}

compute returns a fingerprint of the shape of A: a SHA-256 hash, as 64 hex characters. The store keeps the fingerprint of the rows it holds next to the watermark, the sequence number of the last event applied, in a table called _projection_meta. On start, the engine compares the stored fingerprint with the current one:

  • No fingerprint is stored, as in a new store: it stores the current one.
  • The fingerprints are the same: nothing happens.
  • They differ: the rows are out of date. The engine takes the shortcut if one is registered and fits, and otherwise rebuilds the rows. With lazyRebuild on, it only marks the projection for a later rebuild.

The fingerprint changes when you:

  • rename the row class;
  • add, remove, rename or reorder a field;
  • change the type of a field, for example from Int to Long;
  • change the fields of a case class used directly as the type of a field.

It does not change when you:

  • change the fields of a case class held in an Option, a List or another collection;
  • add a case to a sealed trait used as the type of a field;
  • move @Modifier.id to another field;
  • change a handler.

For those, see Rebuilding on Demand.

Full Rebuild​

A full rebuild throws the rows away and computes them again from the stored events:

  1. The table is dropped and created again, with the columns of the current row type.
  2. The watermark is set back to 0.
  3. Every event of the projection's event store is read, from the first one, and applied through the handlers.
  4. The watermark and the new fingerprint are saved.
  5. The engine then goes on as usual: it catches up on newer events and follows new ones.

start waits for the rebuild, so with a long history the application takes longer to start. In return, old rows get real values for new fields, computed from the events. An event whose change cannot be applied to the table is logged and skipped. An event that can no longer be decoded stops the rebuild; see Changing the Fields of an Event.

The example below runs two versions of an application on the same files. The first keeps UserProfile(id, name); the second adds email, so its start rebuilds the table, and the old users get their email from the stored events:

projection-examples/src/main/scala/projection/examples/Example8SchemaEvolution.scala
/*
 * Example 8 of 13: schema evolution.
 *
 * Builds on Example2PerEntity. Next: Example9MigrationShortcut.
 *
 * When the read model's class changes, such as `UserProfile` gaining an `email`
 * field, the data already stored no longer matches. The engine notices this when
 * it starts, by comparing a hash of the schema with the one saved in the store,
 * and rebuilds the store by replaying the events.
 *
 * This example runs on real SQLite files and uses two engines in a row, like an
 * application that is stopped and started again with a newer version. The events
 * stay in `.projection-examples/example8/events/users.db`, and the read model in
 * `.projection-examples/example8/profiles/userProfiles.db`.
 *   1. The first engine runs the projection with `V1.UserProfile` (id, name).
 *   2. The second engine runs it with `V2.UserProfile` (id, name, email). Both
 *      classes are called `UserProfile`, so they use the same table, and the
 *      projection keeps its name.
 *
 * It shows:
 *   - `SchemaHash.compute`, and that two versions of a type hash differently
 *   - the second engine detecting the changed schema in `start` and rebuilding
 *     the table from the stored events, so the old rows gain their `email`
 *   - that the rebuilt projection keeps processing new events
 *
 * Run it from the repository root with:
 *   sbt "projection-examples/runMain projection.examples.Example8SchemaEvolution"
 */

package projection.examples

import zio.*
import zio.blocks.projection.*
import zio.blocks.schema.{Modifier, Schema}

import Polling.pollUntil

object Example8SchemaEvolution extends ZIOAppDefault {

  private val profilesFolder: String = Sqlite.path("example8/profiles")

  object V1 {
    case class UserProfile(@Modifier.id id: String, name: String)
    object UserProfile {
      implicit val schema: Schema[UserProfile]         = Schema.derived[UserProfile]
      implicit val entityPath: EntityPath[UserProfile] = EntityPath.derived[UserProfile](profilesFolder)
    }
  }

  object V2 {
    case class UserProfile(@Modifier.id id: String, name: String, email: String)
    object UserProfile {
      implicit val schema: Schema[UserProfile]         = Schema.derived[UserProfile]
      implicit val entityPath: EntityPath[UserProfile] = EntityPath.derived[UserProfile](profilesFolder)
    }
  }

  private def show(profile: Option[V2.UserProfile]): String =
    profile.map(u => s"${u.name} <${u.email}>").getOrElse("not found")

  def demo: ZIO[Any, Throwable, Unit] =
    Console.printLine("\n--- Example 8: Schema Evolution (UserProfile V1 -> V2) ---").orDie *>
      ZIO.scoped {
        for {
          _ <- Sqlite.folders(profilesFolder, Sqlite.path("example8/events"))

          hashV1 = SchemaHash.compute[V1.UserProfile]
          hashV2 = SchemaHash.compute[V2.UserProfile]
          _     <- Console.printLine(s"  SchemaHash V1: ${hashV1.take(16)}...").orDie
          _     <- Console.printLine(s"  SchemaHash V2: ${hashV2.take(16)}...").orDie
          _     <- Console.printLine(s"  Hashes differ: ${hashV1 != hashV2}").orDie

          projectionV1 = Projection[V1.UserProfile]("userProfiles")
                           .from("users")
                           .routeToSelf
                           .on[UserCreated]
                           .insert((e, ctx) => V1.UserProfile(ctx.entityId, e.name))

          projectionV2 = Projection[V2.UserProfile]("userProfiles")
                           .from("users")
                           .routeToSelf
                           .on[UserCreated]
                           .insert((e, ctx) => V2.UserProfile(ctx.entityId, e.name, e.email))

          // 1. The first version of the application.
          _ <- ZIO.scoped {
                 for {
                   cache  <- TransactorCache.make()
                   users  <- Sqlite.eventStore[UserCreated](cache, Sqlite.path("example8/events/users.db"))
                   engine <- ProjectionEngine.makeWithStoresVarargs(
                               cache,
                               Map[String, EventStore[?]]("users" -> users)
                             )(projectionV1)
                   _ <- engine.start

                   _ <- users.append("user-1", UserCreated("Alice", "alice@example.com"))
                   _ <- users.append("user-2", UserCreated("Bob", "bob@example.com"))

                   u2 <- pollUntil(engine.query(projectionV1, "user-2"))
                   u1 <- engine.query(projectionV1, "user-1")
                   _  <- Console.printLine(s"  V1 user-1: ${u1.map(_.name).getOrElse("not found")}").orDie
                   _  <- Console.printLine(s"  V1 user-2: ${u2.map(_.name).getOrElse("not found")}").orDie
                 } yield ()
               }

          _ <- Console.printLine("  --- restart with V2 ---").orDie

          // 2. The second version of the application, on the same files.
          _ <- ZIO.scoped {
                 for {
                   cache  <- TransactorCache.make()
                   users  <- Sqlite.eventStore[UserCreated](cache, Sqlite.path("example8/events/users.db"))
                   engine <- ProjectionEngine.makeWithStoresVarargs(
                               cache,
                               Map[String, EventStore[?]]("users" -> users)
                             )(projectionV2)
                   // `start` finds the changed schema hash and rebuilds the table from the stored events.
                   _ <- engine.start

                   u1 <- engine.query(projectionV2, "user-1")
                   u2 <- engine.query(projectionV2, "user-2")
                   _  <- Console.printLine(s"  After rebuild, user-1: ${show(u1)}").orDie
                   _  <- Console.printLine(s"  After rebuild, user-2: ${show(u2)}").orDie

                   _  <- users.append("user-3", UserCreated("Carol", "carol@example.com"))
                   u3 <- pollUntil(engine.query(projectionV2, "user-3"))
                   _  <- Console.printLine(s"  New event, user-3:    ${show(u3)}").orDie
                 } yield ()
               }

          _ <- Console.printLine("  ✓ Schema Evolution: hash check + rebuild from stored events keeps the data").orDie
        } yield ()
      }

  def run: ZIO[Any, Throwable, Unit] = demo
}

It prints After rebuild, user-1: Alice <alice@example.com>.

Adding Fields Without a Rebuild — registerMigration​

When the only change is new fields, the engine can add the columns instead of replaying the events. Describe the change with a Migration from the old row type to the new one, and register it on the engine:

final class ProjectionEngine {
  def registerMigration[Old, New](spec: Projection[New], migration: Migration[Old, New]): Task[Unit]
}

Build the migration with Migration.newBuilder[Old, New], one addField for each new field, and build; see Migration. The rules:

  • Register it before start, which is when the fingerprints are compared.
  • The engine keeps it under the projection's name, so registering another one replaces it.
  • It is used only when every step of the migration adds a top-level field. Any other step, such as a rename, means a full rebuild.
  • Each new field becomes a column added with ALTER TABLE ... ADD COLUMN. The fingerprint is saved, the watermark is kept, and no event is replayed.

The old rows get no value in the new columns: the default given to addField is not written. They read back as null for a String, None for an Option and 0 for an Int, whatever the default was. So use the shortcut for fields that old rows can do without, such as an Option, and let the engine rebuild otherwise.

caution

The engine does not check that a registered migration describes the change it finds. While a migration made only of addField steps is registered, the engine uses it for any change of the fingerprint: it adds the migration's columns, saves the new fingerprint and does not rebuild, even when other fields changed too. The table is then left with missing columns. Register a migration only in the version of the application that makes that change, and remove it afterwards.

The example below runs the two versions of the full rebuild example again, with a migration registered for the second version. The old users come back with an empty email:

projection-examples/src/main/scala/projection/examples/Example9MigrationShortcut.scala
/*
 * Example 9 of 13: the migration shortcut.
 *
 * Read Example8SchemaEvolution first. It is the same story, with one change: the
 * second engine is told how the read model changed. Next: Example10RebuildOnDemand.
 *
 * When the read model gains a field, such as `UserProfile` gaining an `email`,
 * the engine normally rebuilds the whole table by replaying every event. If the
 * change is only added fields, a `Migration` registered with
 * `engine.registerMigration` lets the engine add the columns with
 * `ALTER TABLE ADD COLUMN` instead, and no event is replayed.
 *
 * This example runs on real SQLite files and uses two engines in a row. The
 * events stay in `.projection-examples/example9/events/users.db`, and the read
 * model in `.projection-examples/example9/profiles/userProfiles.db`.
 *   1. The first engine runs the projection with `V1.UserProfile` (id, name).
 *   2. The second engine runs it with `V2.UserProfile` (id, name, email), after
 *      registering a migration that adds `email` with the default `""`.
 *
 * It shows:
 *   - registering the migration before `start`, so the engine can use it
 *   - the old rows being kept without a rebuild. Their new `email` column is
 *     empty (`null`), not `""` and not the real address: the shortcut only adds
 *     the column and does not replay the events. A rebuild, as in
 *     Example8SchemaEvolution, would fill it from the stored events.
 *   - that the migrated projection keeps processing new events
 *
 * Run it from the repository root with:
 *   sbt "projection-examples/runMain projection.examples.Example9MigrationShortcut"
 */

package projection.examples

import zio.*
import zio.blocks.projection.*
import zio.blocks.schema.{Modifier, Schema, SchemaExpr}
import zio.blocks.schema.migration.Migration

import Polling.pollUntil

object Example9MigrationShortcut extends ZIOAppDefault {

  private val profilesFolder: String = Sqlite.path("example9/profiles")

  object V1 {
    case class UserProfile(@Modifier.id id: String, name: String)
    object UserProfile {
      implicit val schema: Schema[UserProfile]         = Schema.derived[UserProfile]
      implicit val entityPath: EntityPath[UserProfile] = EntityPath.derived[UserProfile](profilesFolder)
    }
  }

  object V2 {
    case class UserProfile(@Modifier.id id: String, name: String, email: String)
    object UserProfile {
      implicit val schema: Schema[UserProfile]         = Schema.derived[UserProfile]
      implicit val entityPath: EntityPath[UserProfile] = EntityPath.derived[UserProfile](profilesFolder)
    }
  }

  private def show(profile: Option[V2.UserProfile]): String =
    profile.map(u => s"${u.name} <${u.email}>").getOrElse("not found")

  def demo: ZIO[Any, Throwable, Unit] =
    Console.printLine("\n--- Example 9: Migration Shortcut (UserProfile V1 -> V2) ---").orDie *>
      ZIO.scoped {
        for {
          _ <- Sqlite.folders(profilesFolder, Sqlite.path("example9/events"))

          hashV1 = SchemaHash.compute[V1.UserProfile]
          hashV2 = SchemaHash.compute[V2.UserProfile]
          _     <- Console.printLine(s"  SchemaHash V1: ${hashV1.take(16)}...").orDie
          _     <- Console.printLine(s"  SchemaHash V2: ${hashV2.take(16)}...").orDie
          _     <- Console.printLine(s"  Hashes differ: ${hashV1 != hashV2}").orDie

          projectionV1 = Projection[V1.UserProfile]("userProfiles")
                           .from("users")
                           .routeToSelf
                           .on[UserCreated]
                           .insert((e, ctx) => V1.UserProfile(ctx.entityId, e.name))

          projectionV2 = Projection[V2.UserProfile]("userProfiles")
                           .from("users")
                           .routeToSelf
                           .on[UserCreated]
                           .insert((e, ctx) => V2.UserProfile(ctx.entityId, e.name, e.email))

          // Describes the change from V1 to V2: the new field `email` starts as "".
          migration = Migration
                        .newBuilder[V1.UserProfile, V2.UserProfile]
                        .addField(_.email, SchemaExpr.literal(""))
                        .build

          // 1. The first version of the application.
          _ <- ZIO.scoped {
                 for {
                   cache  <- TransactorCache.make()
                   users  <- Sqlite.eventStore[UserCreated](cache, Sqlite.path("example9/events/users.db"))
                   engine <- ProjectionEngine.makeWithStoresVarargs(
                               cache,
                               Map[String, EventStore[?]]("users" -> users)
                             )(projectionV1)
                   _ <- engine.start

                   _ <- users.append("user-1", UserCreated("Alice", "alice@example.com"))
                   _ <- users.append("user-2", UserCreated("Bob", "bob@example.com"))

                   u2 <- pollUntil(engine.query(projectionV1, "user-2"))
                   u1 <- engine.query(projectionV1, "user-1")
                   _  <- Console.printLine(s"  V1 user-1: ${u1.map(_.name).getOrElse("not found")}").orDie
                   _  <- Console.printLine(s"  V1 user-2: ${u2.map(_.name).getOrElse("not found")}").orDie
                 } yield ()
               }

          _ <- Console.printLine("  --- restart with V2 ---").orDie

          // 2. The second version of the application, on the same files.
          _ <- ZIO.scoped {
                 for {
                   cache  <- TransactorCache.make()
                   users  <- Sqlite.eventStore[UserCreated](cache, Sqlite.path("example9/events/users.db"))
                   engine <- ProjectionEngine.makeWithStoresVarargs(
                               cache,
                               Map[String, EventStore[?]]("users" -> users)
                             )(projectionV2)
                   // The migration is registered before `start`, which is where the changed schema hash is found.
                   // Because it only adds a field, the engine adds the column and does not replay the events.
                   _ <- engine.registerMigration(projectionV2, migration)
                   _ <- engine.start

                   u1 <- engine.query(projectionV2, "user-1")
                   u2 <- engine.query(projectionV2, "user-2")
                   _  <- Console.printLine(s"  After migration, user-1: ${show(u1)}").orDie
                   _  <- Console.printLine(s"  After migration, user-2: ${show(u2)}").orDie

                   _  <- users.append("user-3", UserCreated("Carol", "carol@example.com"))
                   u3 <- pollUntil(engine.query(projectionV2, "user-3"))
                   _  <- Console.printLine(s"  New event, user-3:    ${show(u3)}").orDie
                 } yield ()
               }

          _ <- Console
                 .printLine("  ✓ Migration shortcut: the column was added, old rows kept, their new field is empty")
                 .orDie
        } yield ()
      }

  def run: ZIO[Any, Throwable, Unit] = demo
}

It prints After migration, user-1: Alice <null>.

Rebuilding Later — lazyRebuild​

final case class ProjectionEngineConfig(
  // ...
  rebuildParallelism: Int = 4,
  lazyRebuild: Boolean = false,
  evolution: SchemaEvolutionConfig = SchemaEvolutionConfig.default
)

By default, start rebuilds a changed projection before it returns. With lazyRebuild = true, it only marks the projection as pending, so the application starts at once:

  • The first query of the projection waits while it is rebuilt, or while the shortcut runs.
  • A background fiber rebuilds the pending projections one after the other, in groups of rebuildParallelism.
  • queryByName does not trigger the rebuild.

Until then, the engine keeps applying events to the old table, and the events that no longer fit it fail and are logged. The rebuild applies them again from the stored events. The evolution settings are not read by the engine; see SchemaEvolution. The other settings are described under Configuration.

Rebuilding on Demand​

Some changes need a rebuild that the engine cannot detect: a handler that now computes something else, or a change the fingerprint does not see. There are two ways to rebuild:

  • Delete the projection's store file while the application is stopped: <folder>/<name>.db, with its -wal and -shm files. For a cross-entity projection, delete its <name>_shard_*.db files too, and for a global projection the file is global/<name>.db. When the application starts, the empty store has no fingerprint and a watermark of 0, so the engine applies every stored event again as it catches up.
  • In code, call SchemaEvolution.rebuildWithHash on the store before the engine starts. It empties the store and applies every stored event again, with the current handlers.

The example below runs three versions of an application on the same files. The second changes the handler to store the names in upper case, and its old rows keep their names, because nothing triggers a rebuild. The third rebuilds on demand, before the engine starts:

projection-examples/src/main/scala/projection/examples/Example10RebuildOnDemand.scala
/*
 * Example 10 of 13: rebuilding on demand.
 *
 * Builds on Example8SchemaEvolution. Read it first. Next: Example11TagRename.
 *
 * The engine rebuilds a projection by itself only when the shape of its row type
 * changes. A handler that now computes something else does not change that
 * shape, so the rows already stored keep the old results. To recompute them,
 * rebuild the projection yourself: `SchemaEvolution.rebuildWithHash` empties the
 * projection's store and applies every stored event again through the current
 * handlers.
 *
 * This example runs on real SQLite files: the events are kept in
 * `.projection-examples/example10/events/users.db`, and the read model in
 * `.projection-examples/example10/profiles/user_profiles.db`. It runs three
 * versions of an application in a row, on the same files:
 *   1. The first version stores the users' names as they are.
 *   2. The second changes the handler to store the names in upper case. The old
 *      rows keep their names, because nothing triggers a rebuild.
 *   3. The third has the same handler as the second, and calls
 *      `rebuildWithHash` before the engine starts. Now the old rows are in upper
 *      case too.
 *
 * It shows:
 *   - that changing a handler does not rebuild the stored rows
 *   - rebuilding on demand with `SchemaEvolution.rebuildWithHash`, before
 *     `start`, with the stores created for `makeWithStores`
 *
 * Run it from the repository root with:
 *   sbt "projection-examples/runMain projection.examples.Example10RebuildOnDemand"
 */

package projection.examples

import zio.*
import zio.blocks.projection.*
import zio.blocks.schema.{Modifier, Schema}

import Polling.pollUntil

object Example10RebuildOnDemand extends ZIOAppDefault {

  private val profilesFolder: String = Sqlite.path("example10/profiles")
  private val profilesFile: String   = s"$profilesFolder/user_profiles.db"
  private val eventsFile: String     = Sqlite.path("example10/events/users.db")

  case class UserProfile(@Modifier.id userId: String, name: String)
  object UserProfile {
    implicit val schema: Schema[UserProfile]         = Schema.derived[UserProfile]
    implicit val entityPath: EntityPath[UserProfile] = EntityPath.derived[UserProfile](profilesFolder)
  }

  // The first version stores the name as it is.
  val asIs: Projection[UserProfile] = Projection[UserProfile]("user_profiles")
    .from("users")
    .routeToSelf
    .on[UserCreated]
    .insert((e, ctx) => UserProfile(ctx.entityId, e.name))

  // A later version stores the name in upper case. The row type is the same, so its fingerprint is the same too.
  val upperCase: Projection[UserProfile] = Projection[UserProfile]("user_profiles")
    .from("users")
    .routeToSelf
    .on[UserCreated]
    .insert((e, ctx) => UserProfile(ctx.entityId, e.name.toUpperCase))

  private def name(profile: Option[UserProfile]): String = profile.map(_.name).getOrElse("not found")

  def demo: ZIO[Any, Throwable, Unit] =
    Console.printLine("\n--- Example 10: Rebuilding on Demand (a changed handler) ---").orDie *>
      ZIO.scoped {
        for {
          _ <- Sqlite.folders(profilesFolder, Sqlite.path("example10/events"))

          // 1. The first version of the application.
          _ <- ZIO.scoped {
                 for {
                   cache  <- TransactorCache.make()
                   users  <- Sqlite.eventStore[UserCreated](cache, eventsFile)
                   engine <-
                     ProjectionEngine.makeWithStoresVarargs(cache, Map[String, EventStore[?]]("users" -> users))(asIs)
                   _  <- engine.start
                   _  <- users.append("user-1", UserCreated("Alice", "alice@example.com"))
                   _  <- users.append("user-2", UserCreated("Bob", "bob@example.com"))
                   u2 <- pollUntil(engine.query(asIs, "user-2"))
                   u1 <- engine.query(asIs, "user-1")
                   _  <- Console.printLine(s"  V1 user-1: ${name(u1)}").orDie
                   _  <- Console.printLine(s"  V1 user-2: ${name(u2)}").orDie
                 } yield ()
               }

          _ <- Console.printLine("  --- restart with the upper-case handler ---").orDie

          // 2. The handler changed but the row type did not, so `start` finds the same fingerprint and does not rebuild.
          _ <- ZIO.scoped {
                 for {
                   cache  <- TransactorCache.make()
                   users  <- Sqlite.eventStore[UserCreated](cache, eventsFile)
                   engine <- ProjectionEngine.makeWithStoresVarargs(
                               cache,
                               Map[String, EventStore[?]]("users" -> users)
                             )(upperCase)
                   _  <- engine.start
                   u1 <- engine.query(upperCase, "user-1")
                   _  <- Console.printLine(s"  V2 user-1: ${name(u1)} (not rebuilt)").orDie
                   _  <- users.append("user-3", UserCreated("Carol", "carol@example.com"))
                   u3 <- pollUntil(engine.query(upperCase, "user-3"))
                   _  <- Console.printLine(s"  V2 new event, user-3: ${name(u3)}").orDie
                 } yield ()
               }

          _ <- Console.printLine("  --- restart and rebuild on demand ---").orDie

          // 3. The same handler, with the rows rebuilt before the engine starts.
          _ <- ZIO.scoped {
                 for {
                   cache    <- TransactorCache.make()
                   tx       <- cache.get(eventsFile)
                   users    <- SQLiteEventStore.make[UserCreated](tx)
                   profiles <- SQLiteProjectionStore.make[UserProfile](profilesFile, cache)
                   // Empties the store and applies every stored event again, with the current handler.
                   _      <- SchemaEvolution.rebuildWithHash(profiles, users, upperCase)
                   engine <- ProjectionEngine.makeWithStores(
                               specs = List(upperCase),
                               stores = Map("user_profiles" -> profiles),
                               eventStores = Map("users" -> users),
                               cache = cache
                             )
                   _    <- engine.start
                   rows <-
                     ZIO.foreach(List("user-1", "user-2", "user-3"))(id => engine.query(upperCase, id).map(id -> _))
                   _ <- ZIO.foreachDiscard(rows) { case (id, row) =>
                          Console.printLine(s"  After rebuildWithHash, $id: ${name(row)}").orDie
                        }
                 } yield ()
               }

          _ <-
            Console
              .printLine("  ✓ Rebuild on demand: rebuildWithHash applied every stored event again with the new handler")
              .orDie
        } yield ()
      }

  def run: ZIO[Any, Throwable, Unit] = demo
}

It prints V2 user-1: Alice (not rebuilt), then After rebuildWithHash, user-1: ALICE.

Neither way works for a projection that reads several sources; see Limitations.

Renaming the Row Class or the Projection​

The name of the row class is also the name of its table. When you rename the class, the engine finds no table and no fingerprint under the new name: it creates an empty table and fills it by applying every stored event, as for a new projection. The old table stays in the file, unused.

To keep the table, give the renamed class its old table name with @Modifier.config("sql.table_name", ...). The class name is part of the fingerprint, so the engine then rebuilds that table as for any other change:

import zio.blocks.projection.*
import zio.blocks.schema.{Modifier, Schema}

// Renamed from UserProfile; the annotation keeps the table `user_profile`
@Modifier.config("sql.table_name", "user_profile")
case class Profile(@Modifier.id userId: String, name: String)
object Profile {
  implicit val schema: Schema[Profile]         = Schema.derived[Profile]
  implicit val entityPath: EntityPath[Profile] = EntityPath.derived[Profile]
}

Renaming the projection, or moving it to another folder with its EntityPath, gives it a new file: the engine builds it from scratch from the stored events, and the old file stays. A migration registered with registerMigration is looked up under the new name.

Changing Events​

Events are never rewritten. Each stored event is a row of the event store's events table, with a sequence number, a tag and a payload. The tag says which case of the event type the event is, and the payload is the event as JSON, keyed by the case name:

seq | tag          | payload
1   | UserLoggedIn | {"UserLoggedIn":{"userId":"user-1"}}

When events are read back, to catch up or to rebuild, the store keeps only the rows whose tag it knows, and decodes each payload with the current event type.

Which Stored Events Are Read​

The store knows the tags the event type has now, and the old names of renamed cases that you declared with a migration. Every read keeps only the events with those tags, including catching up and rebuilding. An event under any other tag, for example of a case that was removed, or renamed without a migration, is skipped without an error. When the store is created, it logs a warning that lists the tags it found in the database and cannot explain:

Unknown event tags detected: Closed, Opened not in known tags Activated

So removing a case from the event type hides its stored events from every read, including a full rebuild.

Renaming an Event Case​

A renamed case keeps its old name in the stored events, in their tag and in their JSON. To go on reading them, describe the rename with a Migration of the event type to itself, and create the event store with it:

object SQLiteEventStore {
  def make[E: Schema](transactor: Transactor, migration: Migration[E, E]): Task[SQLiteEventStore[E]]
}

The other event store factories take a migration the same way; see EventStore.

import zio.*
import zio.blocks.projection.*
import zio.blocks.schema.Schema
import zio.blocks.schema.migration.Migration

sealed trait LoginEvent
object LoginEvent {
  case class UserAuthenticated(userId: String) extends LoginEvent // was UserLoggedIn
  implicit val schema: Schema[LoginEvent] = Schema.derived[LoginEvent]
}

val renames: Migration[LoginEvent, LoginEvent] =
  Migration.newBuilder[LoginEvent, LoginEvent].renameCase("UserLoggedIn", "UserAuthenticated").build

def openEvents(cache: TransactorCache): Task[SQLiteEventStore[LoginEvent]] =
  cache.get("events/logins.db").flatMap(tx => SQLiteEventStore.make[LoginEvent](tx, renames))

The store then:

  • reads the events saved under the old name too, also when a read asks only for the new tag;
  • decodes them as the renamed case;
  • reports them under the new tag, so the handlers only need on[UserAuthenticated];
  • saves new events under the new name.

The stored events themselves are not changed. The rules:

  • Only renameCase steps make old names readable. Do not rely on other steps of the migration to change stored events.
  • If a case was renamed more than once, list every rename: UserLoggedIn to UserSignedIn, then UserSignedIn to UserAuthenticated, makes both old names read as UserAuthenticated.
  • Keep the renames for as long as events with the old names exist.

The example below appends an event with the old version of the event type, then reads it back through a store created with the rename. It also prints the aliases the store works out:

projection-examples/src/main/scala/projection/examples/Example11TagRename.scala
/*
 * Example 11 of 13: renaming an event.
 *
 * Builds on Example8SchemaEvolution. Next: Example12EventFieldChanges.
 *
 * Event tags are stored as strings, so renaming a case, here `UserLoggedIn` to
 * `UserAuthenticated`, leaves old events in the store under the old tag. A
 * `Migration` that renames the case lets `TagResolver` work out the alias, so old
 * events can still be recognised and read.
 *
 * This example runs on a real SQLite event store,
 * `.projection-examples/example11/events/logins.db`, with two versions of the event type in
 * a row. Both are called `LoginEvent`, so they use the same table.
 *   1. `V1.LoginEvent` has the case `UserLoggedIn`. An event is appended with it.
 *   2. `V2.LoginEvent` has the case `UserAuthenticated`. The store is opened again
 *      with a migration that renames the case, and the old event is read back
 *      under its new name.
 *
 * It shows:
 *   - `TagResolver.resolve` building a `TagInfo` from a migration, and its
 *     helpers: `currentTags`, `aliases`, `allTags`, `expandRequested`, `isOldTag`,
 *     `currentTagFor` and `normalize`
 *   - a chain of renames (A to B to C), which resolves straight to the final name
 *   - `SQLiteEventStore.make` with a migration, so that `readAll` returns an old
 *     event with the current tag and decoded as the new case
 *
 * Run it from the repository root with:
 *   sbt "projection-examples/runMain projection.examples.Example11TagRename"
 */

package projection.examples

import zio.*
import zio.blocks.projection.*
import zio.blocks.schema.Schema
import zio.blocks.schema.migration.Migration

object Example11TagRename extends ZIOAppDefault {

  object V1 {
    sealed trait LoginEvent
    object LoginEvent {
      case class UserLoggedIn(userId: String, timestamp: String) extends LoginEvent
      object UserLoggedIn {
        implicit val schema: Schema[UserLoggedIn] = Schema.derived[UserLoggedIn]
      }
      implicit val schema: Schema[LoginEvent] = Schema.derived[LoginEvent]
    }
  }

  object V2 {
    sealed trait LoginEvent
    object LoginEvent {
      case class UserAuthenticated(userId: String, timestamp: String) extends LoginEvent
      object UserAuthenticated {
        implicit val schema: Schema[UserAuthenticated] = Schema.derived[UserAuthenticated]
      }
      implicit val schema: Schema[LoginEvent] = Schema.derived[LoginEvent]
    }
  }

  def demo: ZIO[Any, Throwable, Unit] =
    Console.printLine("\n--- Example 11: Tag Rename (UserLoggedIn -> UserAuthenticated) ---").orDie *>
      ZIO.scoped {
        for {
          _ <- Sqlite.folders(Sqlite.path("example11/events"))

          migration = Migration
                        .newBuilder[V2.LoginEvent, V2.LoginEvent]
                        .renameCase("UserLoggedIn", "UserAuthenticated")
                        .build

          tagInfo = TagResolver.resolve[V2.LoginEvent](migration)

          _ <- Console.printLine(s"  Current tags: ${tagInfo.currentTags.mkString(", ")}").orDie
          _ <- Console.printLine(s"  Aliases:      ${tagInfo.aliases}").orDie
          _ <- Console.printLine(s"  All tags:     ${tagInfo.allTags.mkString(", ")}").orDie

          expanded = tagInfo.expandRequested(Set("UserAuthenticated"))
          _       <- Console.printLine(s"  expandRequested(UserAuthenticated) -> ${expanded.mkString(", ")}").orDie

          _ <- Console.printLine(s"  isOldTag('UserLoggedIn'):      ${tagInfo.isOldTag("UserLoggedIn")}").orDie
          _ <- Console.printLine(s"  isOldTag('UserAuthenticated'): ${tagInfo.isOldTag("UserAuthenticated")}").orDie
          _ <- Console.printLine(s"  currentTagFor('UserLoggedIn'): ${tagInfo.currentTagFor("UserLoggedIn")}").orDie
          _ <- Console.printLine(s"  normalize('UserLoggedIn'):      ${tagInfo.normalize("UserLoggedIn")}").orDie
          _ <- Console.printLine(s"  normalize('UserAuthenticated'): ${tagInfo.normalize("UserAuthenticated")}").orDie

          migration2 = Migration
                         .newBuilder[V2.LoginEvent, V2.LoginEvent]
                         .renameCase("UserLoggedIn", "UserSignedIn")
                         .renameCase("UserSignedIn", "UserAuthenticated")
                         .build
          tagInfo2 = TagResolver.resolve[V2.LoginEvent](migration2)
          _       <- Console.printLine(s"  Transitive chain: ${tagInfo2.aliases}").orDie

          // 1. The first version writes an event under the tag `UserLoggedIn`.
          _ <- ZIO.scoped {
                 for {
                   cache  <- TransactorCache.make()
                   tx     <- cache.get(Sqlite.path("example11/events/logins.db"))
                   events <- SQLiteEventStore.make[V1.LoginEvent](tx)
                   _      <- events.append("user-1", V1.LoginEvent.UserLoggedIn("user-1", "10:00"))
                 } yield ()
               }

          // 2. The second version opens the same file with the migration, and also appends an event.
          _ <- ZIO.scoped {
                 for {
                   cache  <- TransactorCache.make()
                   tx     <- cache.get(Sqlite.path("example11/events/logins.db"))
                   events <- SQLiteEventStore.make[V2.LoginEvent](tx, migration)
                   _      <- events.append("user-2", V2.LoginEvent.UserAuthenticated("user-2", "11:00"))
                   all    <- events.readAll().runCollect
                   _      <- ZIO.foreachDiscard(all) { env =>
                          Console.printLine(s"  seq ${env.seq}: tag ${env.tag}, ${env.entityId}, ${env.event}").orDie
                        }
                   byNewTag <- events.readAll(Set("UserAuthenticated")).runCollect
                   _        <- Console.printLine(s"  readAll(UserAuthenticated) returns ${byNewTag.size} events").orDie
                 } yield ()
               }

          _ <- Console.printLine("  ✓ Tag Rename: migration aliases ensure old events are readable").orDie
        } yield ()
      }

  def run: ZIO[Any, Throwable, Unit] = demo
}

It prints seq 1: tag UserAuthenticated, user-1, UserAuthenticated(user-1,10:00) for the event stored as UserLoggedIn.

Numeric Tags — @eventTag​

final class eventTag(val value: Int) extends StaticAnnotation

By default, the tag of an event is the name of its case. @eventTag(n) on a case makes the tag the number n instead. Only the tag changes: the payload stays keyed by the case name. The numbers are not checked for uniqueness.

Renaming a numbered case needs care. The number still matches, but the stored payloads carry the old case name, so they no longer decode, and the read fails. renameCase does not help, because it works on tags, not on payloads. Instead, keep the old name readable with @Modifier.alias on the renamed case:

import zio.blocks.projection.eventTag
import zio.blocks.schema.{Modifier, Schema}

sealed trait LoginEvent
object LoginEvent {
  // Renamed from UserLoggedIn, which was stored with the tag 1
  @eventTag(1)
  @Modifier.alias("UserLoggedIn")
  case class UserAuthenticated(userId: String) extends LoginEvent

  @eventTag(2)
  case class UserLoggedOut(userId: String) extends LoginEvent

  implicit val schema: Schema[LoginEvent] = Schema.derived[LoginEvent]
}

Removing @eventTag from a case changes its tag back to the case name, so the events stored under the number are no longer read; see Which Stored Events Are Read.

Changing the Fields of an Event​

Stored payloads are not migrated. An old event is decoded with the current case class, and the JSON codec decides what it becomes:

  • Added field: it takes the field's default value, None for an Option, or an empty collection. A field with none of these makes the event unreadable, with missing required field.
  • Removed field: its stored value is ignored.
  • Renamed field: the new name counts as missing, unless the field keeps its old name as an alias: @Modifier.alias("owner") holder: String.
  • Changed type: the event is unreadable unless the stored value still fits the new type, as a String fits an Option[String].
caution

A field that holds its default value is not written to the payload. If you change the default later, the old events that had the default read back with the new one.

So add a field as an Option or with a default, and never change a default or a type: add a new field or a new case instead.

The example below stores two events with the first version of an event type, then reads them back with two newer versions:

projection-examples/src/main/scala/projection/examples/Example12EventFieldChanges.scala
/*
 * Example 12 of 13: changing the fields of an event.
 *
 * Builds on Example11TagRename. Read it first. Next: Example13EngineCatchUp.
 *
 * Stored events are never rewritten. When a case of the event type gains, loses
 * or changes a field, the events already stored keep their old JSON, and the
 * event store decodes that JSON with the current case class. No migration is
 * applied: the JSON codec decides what an old event becomes.
 *   - a field that was removed is ignored
 *   - a field that was added takes its default value, `None` for an `Option`, or
 *     an empty collection
 *   - a field that was added with none of these makes the event fail to decode
 *
 * This example stores two `AccountOpened` events with the first version of the
 * event type, in `.projection-examples/example12/events/accounts.db`, then reads
 * them back with two newer versions:
 *   1. V1 has `owner` and `plan`.
 *   2. V2 drops `plan` and adds `currency` with a default, `referrer` as an
 *      `Option` and `tags` as a `List`. The old events read back with `EUR`,
 *      `None` and an empty list.
 *   3. V3 adds `country` with no default. Reading fails with
 *      `missing required field "country"`.
 *
 * It shows:
 *   - that old events are decoded with the current case class, without a
 *     migration
 *   - which added fields are filled in, and which make an old event unreadable
 *
 * Run it from the repository root with:
 *   sbt "projection-examples/runMain projection.examples.Example12EventFieldChanges"
 */

package projection.examples

import zio.*
import zio.blocks.projection.*
import zio.blocks.schema.Schema

object Example12EventFieldChanges extends ZIOAppDefault {

  object V1 {
    sealed trait AccountEvent
    object AccountEvent {
      case class AccountOpened(owner: String, plan: String) extends AccountEvent
      implicit val schema: Schema[AccountEvent] = Schema.derived[AccountEvent]
    }
  }

  object V2 {
    sealed trait AccountEvent
    object AccountEvent {
      case class AccountOpened(owner: String, currency: String = "EUR", referrer: Option[String], tags: List[String])
          extends AccountEvent
      implicit val schema: Schema[AccountEvent] = Schema.derived[AccountEvent]
    }
  }

  object V3 {
    sealed trait AccountEvent
    object AccountEvent {
      case class AccountOpened(owner: String, country: String) extends AccountEvent
      implicit val schema: Schema[AccountEvent] = Schema.derived[AccountEvent]
    }
  }

  private val eventsFile: String = Sqlite.path("example12/events/accounts.db")

  // Opens the event store with the given version of the event type and reads every event.
  private def readAll[E: Schema]: Task[Chunk[EventEnvelope[E]]] =
    ZIO.scoped {
      for {
        cache  <- TransactorCache.make()
        events <- Sqlite.eventStore[E](cache, eventsFile)
        all    <- events.readAll().runCollect
      } yield all
    }

  def demo: ZIO[Any, Throwable, Unit] =
    Console.printLine("\n--- Example 12: Changing the Fields of an Event (AccountOpened V1 -> V2 -> V3) ---").orDie *>
      ZIO.scoped {
        for {
          _ <- Sqlite.folders(Sqlite.path("example12/events"))

          // 1. The first version stores two events.
          _ <- ZIO.scoped {
                 for {
                   cache <- TransactorCache.make()
                   v1    <- Sqlite.eventStore[V1.AccountEvent](cache, eventsFile)
                   _     <- v1.append("acc-1", V1.AccountEvent.AccountOpened("alice", "basic"))
                   _     <- v1.append("acc-2", V1.AccountEvent.AccountOpened("bob", "pro"))
                 } yield ()
               }

          // 2. A newer version reads them: `plan` is ignored, and the new fields are filled in.
          v2 <- readAll[V2.AccountEvent]
          _  <- ZIO.foreachDiscard(v2)(e => Console.printLine(s"  V2 reads seq ${e.seq}: ${e.event}").orDie)

          // 3. A version with a new required field cannot read them.
          v3 <- readAll[V3.AccountEvent].either
          _  <- v3 match {
                 case Right(events) => Console.printLine(s"  V3 reads ${events.size} events").orDie
                 case Left(error)   => Console.printLine(s"  V3 fails: ${error.getMessage}").orDie
               }

          _ <- Console
                 .printLine("  ✓ Event fields: old events are read with the current case class, without a migration")
                 .orDie
        } yield ()
      }

  def run: ZIO[Any, Throwable, Unit] = demo
}

It prints V2 reads seq 1: AccountOpened(alice,EUR,None,List()), and for the third version missing required field "country".

An event that cannot be decoded fails the whole read:

  • While catching up, the engine stops reading that source and logs bind failed. It does not follow the source's new events either, until the application is restarted with code that can read the event.
  • During a full rebuild, the replay stops at that event and is logged. The rows hold only the events before it, yet the new fingerprint is saved and the watermark stays at 0. When the application next starts with fixed code, the engine catches up from the first event on top of those rows, so the events the rebuild had applied are applied twice. After fixing the code, rebuild on demand instead.

Limitations​

These cases are not handled yet:

  • Several sources. A projection that reads several event stores keeps one watermark for all of them, although each store numbers its events from 1. An event appended to one source while the application is stopped can fall below the watermark set by another source, and is then never applied. A full rebuild replays only the first source, and the events of the others below the watermark are lost; deleting the store file does not help.
  • Cross-entity projections. A full rebuild writes every row to the main file and leaves the shard files as they were. Later events for those rows fail with no column named ..., and queries return the stale rows of the main file. To rebuild such a projection, stop the application and delete the main file and its shard files.
  • Engines without event stores. An engine created with make or makeWithConfig has no event stores, so it cannot rebuild: it logs no EventStore for source ..., keeps the old fingerprint, and queries fail on the changed table.
  • TestEngine. It starts the engine inside make, on new in-memory stores, so a test can neither store an old fingerprint first nor register a migration before start. Test a migration with two engines in a row on SQLite files, as the examples on this page do.

Advanced: Running the Steps Yourself​

Advanced

You do not need this section to migrate a projection: the engine runs these steps for you. It documents the building blocks, for tests and for custom runners.

SchemaEvolution​

object SchemaEvolution {
  def needsRebuild[A: Schema](store: ProjectionStore[A]): Task[Boolean]
  def rebuild[A](store: ProjectionStore[A], eventStore: EventStore[_], spec: Projection[A], currentHash: String): Task[Unit]
  def rebuildWithHash[A: Schema: EntityPath](store: ProjectionStore[A], eventStore: EventStore[_], spec: Projection[A]): Task[Unit]
  def isSimpleAddFieldMigration(migration: Migration[?, ?]): Boolean
  def tryMigrationShortcut[A: Schema: EntityPath](
    store: ProjectionStore[A],
    migrationOpt: Option[Migration[?, A]],
    currentHash: String
  ): Task[Boolean]
  def checkAndRebuild[A: Schema: EntityPath](
    store: ProjectionStore[A],
    eventStore: EventStore[_],
    spec: Projection[A],
    config: SchemaEvolutionConfig = SchemaEvolutionConfig.default,
    migrationOpt: Option[Migration[?, A]] = None
  ): Task[Boolean]
}
  • needsRebuild is true when a fingerprint is stored and differs from the current one. It is false for a new store.
  • rebuild does the full rebuild and saves the fingerprint you pass; rebuildWithHash computes the fingerprint from A.
  • isSimpleAddFieldMigration says whether a migration qualifies for the shortcut: it is not empty, and every step adds a top-level field. It is useful in a test.
  • tryMigrationShortcut runs the shortcut and returns true, or returns false when there is no migration or it does not qualify.
  • checkAndRebuild compares the fingerprints, then runs the shortcut or a rebuild, like the engine does on start. It returns true when it changed the store.

SchemaEvolutionConfig configures these helpers, not the engine:

final case class SchemaEvolutionConfig(
  rebuildParallelism: Int = 4,
  lazyRebuild: Boolean = false,
  enableMigrationShortcut: Boolean = true
)
  • enableMigrationShortcut = false makes checkAndRebuild always rebuild.
  • rebuildParallelism is how many projections bulkRebuildPending rebuilds at the same time.
  • lazyRebuild is not read.

The other members of SchemaEvolution, tryMigrationShortcutForSpec, registerMigration, lazyRebuildIfNeeded and bulkRebuildPending, are the engine's own building blocks and are not meant to be called directly. A custom store takes part in a rebuild and in the shortcut through recreateTable and addColumn; see ProjectionStore.

TagInfo and TagResolver​

The event store gathers what it knows about tags in a TagInfo, built by TagResolver from the event type and the migration:

final case class TagInfo(
  aliases: Map[String, String],
  allTags: Set[String],
  currentTags: Set[String],
  migration: DynamicMigration
) {
  def isOldTag(tag: String): Boolean
  def currentTagFor(oldTag: String): Option[String]
  def normalize(tag: String): String
  def migrateValue(dv: DynamicValue): Either[SchemaError, DynamicValue]
  def expandRequested(requested: Set[String]): Set[String]
  def unknownTags(distinctDbTags: Set[String]): Set[String]
}
  • aliases maps each old tag to the current tag it became, with chains of renames followed.
  • currentTags are the tags of the event type now, and allTags adds the old ones.
  • migration is the migration the store was given.
  • isOldTag and currentTagFor look a tag up in aliases. normalize returns the current tag for an old one, and any other tag unchanged.
  • expandRequested widens the tags that a read asks for with their old names. An empty request means all the known tags.
  • unknownTags returns the tags found in the database that are neither current nor old; they make the startup warning.
  • migrateValue runs the migration on a decoded value. The store uses it only as a fallback, when a renamed event does not decode after its old case name was replaced.
object TagResolver {
  def resolve[E: Schema]: TagInfo
  def resolve[E: Schema](migration: Migration[E, E]): TagInfo
  def resolveOpt[E](opt: Option[Migration[E, E]])(using schema: Schema[E]): TagInfo
  def resolveImplicit[E: Schema](implicit migration: Migration[E, E] = null): TagInfo
  inline def resolveAuto[E](using schema: Schema[E]): TagInfo
  def currentTags[E](using schema: Schema[E]): Set[String]
  def allTags[E: Schema](migration: Migration[E, E] = null): Set[String]
  def aliases[E: Schema](migration: Migration[E, E] = null): Map[String, String]
  def aliasMap[E](migration: Migration[E, E]): Map[String, String]
  def aliasMapFromDynamic(dm: DynamicMigration): Map[String, String]
  def unknownTags(tagInfo: TagInfo, distinctDbTags: Set[String]): Set[String]
}
  • resolve builds the TagInfo of an event type, with the aliases of a migration when you give one; SQLiteEventStore.make uses it. resolveOpt takes the migration as an Option, and resolveImplicit and resolveAuto use a migration found in implicit scope, if there is one.
  • currentTags, allTags and aliases each return one part of the TagInfo. aliasMap gives the aliases of a migration with chains followed, and aliasMapFromDynamic the renames of a DynamicMigration, without following chains.
  • unknownTags is the same as TagInfo#unknownTags.

The TagResolver[E] trait is a view of one TagInfo: implement tagInfo, and aliases, allTags and currentTags come from it. For how the event store uses its TagInfo, see EventStore.

See Also​