diff --git a/src/sink.ts b/src/sink.ts
index d767958..22eb58d 100644
--- a/src/sink.ts
+++ b/src/sink.ts
@@ -28,7 +28,7 @@ function handleEntityChange(
 
   switch (change.operation) {
     case "OPERATION_CREATE":
-      return insertEntityChange(change.entity, values, metadata);
+      return insertEntityChange(change.entity, values, { ...metadata, id: change.id });
 
     // case "OPERATION_UPDATE":
     //   return client.update();
@@ -45,7 +45,7 @@ function handleEntityChange(
 async function insertEntityChange(
   table: string,
   values: Record<string, unknown>,
-  metadata: { clock: Clock; manifest: Manifest }
+  metadata: { id: string; clock: Clock; manifest: Manifest }
 ) {
   if (!knownModuleHashes.includes(metadata.manifest.moduleHash)) {
     await client.command({
@@ -56,6 +56,7 @@ async function insertEntityChange(
     knownModuleHashes.push(metadata.manifest.moduleHash);
   }
 
+  values["entity_id"] = metadata.id;
   values["chain"] = metadata.manifest.chain;
   values["block_id"] = metadata.clock.id;
   values["block_number"] = metadata.clock.number;
diff --git a/src/table-initialization.ts b/src/table-initialization.ts
index f5cd59d..589ae77 100644
--- a/src/table-initialization.ts
+++ b/src/table-initialization.ts
@@ -19,6 +19,7 @@ export function initializeManifest(): Promise<unknown> {
 }
 
 const metadataQueries = (tableName: string) => [
+  `ALTER TABLE ${tableName} ADD COLUMN IF NOT EXISTS entity_id    String;`,
   `ALTER TABLE ${tableName} ADD COLUMN IF NOT EXISTS timestamp    DateTime('UTC');`,
   `ALTER TABLE ${tableName} ADD COLUMN IF NOT EXISTS block_number UInt32;`,
   `ALTER TABLE ${tableName} ADD COLUMN IF NOT EXISTS block_id     FixedString(64);`,
@@ -28,9 +29,7 @@ const metadataQueries = (tableName: string) => [
   `ALTER TABLE ${tableName} ADD INDEX  IF NOT EXISTS metadata_index (chain, module_hash) TYPE minmax`,
 ];
 
-export async function handleTableInitialization(
-  schema: TableInitSchema
-): Promise<Response> {
+export async function handleTableInitialization(schema: TableInitSchema): Promise<Response> {
   try {
     const tables = await initializeTables(schema);
     return new Response("OK\nProcessed tables: " + tables);
@@ -43,9 +42,7 @@ export async function initializeTables(schema: string): Promise<string[]> {
   logger.info("Executing schema.");
   const tables = splitSchemaByTableCreation(schema);
   logger.info(
-    `Found ${tables.length} table(s): ${tables
-      .map(({ tableName }) => `'${tableName}'`)
-      .join(", ")}`
+    `Found ${tables.length} table(s): ${tables.map(({ tableName }) => `'${tableName}'`).join(", ")}`
   );
 
   try {