Repository: ponder-sh/ponder
Pull request #1623 (2025-03-25 19:46:49)

fix `timestamp` column


Patch:
diff --git a/.changeset/lemon-ads-arrive.md b/.changeset/lemon-ads-arrive.md
new file mode 100644
index 00000000..2a7417e6
--- /dev/null
+++ b/.changeset/lemon-ads-arrive.md
@@ -0,0 +1,5 @@
+---
+"ponder": patch
+---
+
+Fixed `timestamp` column.
diff --git a/packages/core/src/indexing-store/cache.ts b/packages/core/src/indexing-store/cache.ts
index 492b7f6b..88e2c6eb 100644
--- a/packages/core/src/indexing-store/cache.ts
+++ b/packages/core/src/indexing-store/cache.ts
@@ -402,70 +402,72 @@ export const createIndexingCache = ({
         const updateValues = Array.from(updateBuffer.get(table)!.values());
 
         if (insertValues.length > 0) {
-          const text = getCopyText(
-            table,
-            insertValues.map(({ row }) => row),
-          );
-
-          await new Promise(setImmediate);
-
           const endClock = startClock();
 
           // @ts-ignore
           await client.query("SAVEPOINT flush");
-          await copy(table, text)
-            .catch(async (error) => {
-              const result = await recoverBatchError(
-                insertValues,
-                async (values) => {
-                  // @ts-ignore
-                  await client.query("ROLLBACK to flush");
-                  const text = getCopyText(
-                    table,
-                    values.map(({ row }) => row),
-                  );
-                  await copy(table, text);
-                  // @ts-ignore
-                  await client.query("SAVEPOINT flush");
-                },
-              );
 
-              if (result.status === "success") {
-                return;
-              }
+          try {
+            const text = getCopyText(
+              table,
+              insertValues.map(({ row }) => row),
+            );
 
-              // Note: rollback so that the connection is available for other queries
-              // @ts-ignore
-              await client.query("ROLLBACK to flush");
+            await new Promise(setImmediate);
+
+            await copy(table, text);
+          } catch (_error) {
+            let error = _error as Error;
+            const result = await recoverBatchError(
+              insertValues,
+              async (values) => {
+                // @ts-ignore
+                await client.query("ROLLBACK to flush");
+                const text = getCopyText(
+                  table,
+                  values.map(({ row }) => row),
+                );
+                await copy(table, text);
+                // @ts-ignore
+                await client.query("SAVEPOINT flush");
+              },
+            );
+
+            if (result.status === "success") {
+              return;
+            }
+
+            // Note: rollback so that the connection is available for other queries
+            // @ts-ignore
+            await client.query("ROLLBACK to flush");
 
-              error = parseSqlError(result.error);
-              error.stack = undefined;
+            error = parseSqlError(result.error);
+            error.stack = undefined;
 
-              if (result.value.metadata.event) {
-                addErrorMeta(
-                  error,
-                  `db.insert arguments:\n${prettyPrint(result.value.row)}`,
-                );
-                addErrorMeta(error, toErrorMeta(result.value.metadata.event));
-                common.logger.error({
-                  service: "indexing",
-                  msg: `Error while processing ${getTableName(
-                    table,
-                  )}.insert() in event '${result.value.metadata.event.name}'`,
-                  error,
-                });
-              }
-              throw new FlushError(error.message);
-            })
-            .finally(() => {
-              common.metrics.ponder_indexing_cache_query_duration.observe(
-                {
-                  table: getTableName(table),
-                  method: "flush",
-                },
-                endClock(),
+            if (result.value.metadata.event) {
+              addErrorMeta(
+                error,
+                `db.insert arguments:\n${prettyPrint(result.value.row)}`,
               );
-            });
+              addErrorMeta(error, toErrorMeta(result.value.metadata.event));
+              common.logger.error({
+                service: "indexing",
+                msg: `Error while processing ${getTableName(
+                  table,
+                )}.insert() in event '${result.value.metadata.event.name}'`,
+                error,
+              });
+            }
+            throw new FlushError(error.message);
+          } finally {
+            common.metrics.ponder_indexing_cache_query_duration.observe(
+              {
+                table: getTableName(table),
+                method: "flush",
+              },
+              endClock(),
+            );
+          }
 
           for (const [key, entry] of insertBuffer.get(table)!) {
             const bytes = getBytes(entry.row);
@@ -526,59 +528,55 @@ export const createIndexingCache = ({
             `;
           const truncateQuery = `TRUNCATE TABLE "${getTableName(table)}" CASCADE`;
 
-          const text = getCopyText(
-            table,
-            updateValues.map(({ row }) => row),
-          );
-
-          await new Promise(setImmediate);
-
           const endClock = startClock();
 
+          // @ts-ignore
+          await client.query(createTempTableQuery);
+          // @ts-ignore
+          await client.query("SAVEPOINT flush");
+
           try {
-            // @ts-ignore
-            await client.query(createTempTableQuery);
-            // @ts-ignore
-            await client.query("SAVEPOINT flush");
-            await copy(table, text, false).catch(async (error) => {
-              const result = await recoverBatchError(
-                updateValues,
-                async (values) => {
-                  // @ts-ignore
-                  await client.query("ROLLBACK to flush");
-                  const text = getCopyText(
-                    table,
-                    values.map(({ row }) => row),
-                  );
-                  await copy(table, text, false);
-                  // @ts-ignore
-                  await client.query("SAVEPOINT flush");
-                },
-              );
+            const text = getCopyText(
+              table,
+              updateValues.map(({ row }) => row),
+            );
+
+            await new Promise(setImmediate);
+
+            await copy(table, text, false);
+          } catch (_error) {
+            let error = _error as Error;
+            const result = await recoverBatchError(
+              updateValues,
+              async (values) => {
+                // @ts-ignore
+                await client.query("ROLLBACK to flush");
+                const text = getCopyText(
+                  table,
+                  values.map(({ row }) => row),
+                );
+                await copy(table, text, false);
+                // @ts-ignore
+                await client.query("SAVEPOINT flush");
+              },
+            );
 
-              if (result.status === "success") {
-                return;
-              }
+            if (result.status === "success") {
+              return;
+            }
 
-              // Note: rollback so that the connection is available for other queries
-              // @ts-ignore
-              await client.query("ROLLBACK to flush");
+            // Note: rollback so that the connection is available for other queries
+            // @ts-ignore
+            await client.query("ROLLBACK to flush");
 
-              error = parseSqlError(result.error);
-              error.stack = undefined;
+            error = parseSqlError(result.error);
+            error.stack = undefined;
 
-              addErrorMeta(
-                error,
-                `db.update arguments:\n${prettyPrint(result.value.row)}`,
-              );
+            addErrorMeta(
+              error,
+              `db.update arguments:\n${prettyPrint(result.value.row)}`,
+            );
 
-              throw error;
-            });
-            // @ts-ignore
-            await client.query(updateQuery);
-            // @ts-ignore
-            await client.query(truncateQuery);
-          } finally {
             common.metrics.ponder_indexing_cache_query_duration.observe(
               {
                 table: getTableName(table),
@@ -586,7 +584,23 @@ export const createIndexingCache = ({
               },
               endClock(),
             );
+
+            throw error;
           }
+
+          // @ts-ignore
+          await client.query(updateQuery);
+          // @ts-ignore
+          await client.query(truncateQuery);
+
+          common.metrics.ponder_indexing_cache_query_duration.observe(
+            {
+              table: getTableName(table),
+              method: "flush",
+            },
+            endClock(),
+          );
+
           for (const [key, entry] of updateBuffer.get(table)!) {
             const bytes = getBytes(entry.row);
             cacheBytes += bytes;
diff --git a/packages/core/src/indexing-store/utils.ts b/packages/core/src/indexing-store/utils.ts
index 753744a4..a101516f 100644
--- a/packages/core/src/indexing-store/utils.ts
+++ b/packages/core/src/indexing-store/utils.ts
@@ -50,13 +50,22 @@ export const normalizeColumn = (
   }
   if (value === null) return null;
   if (column.mapToDriverValue === undefined) return value;
+
   try {
     if (Array.isArray(value) && column instanceof PgArray) {
-      return value.map((v) =>
-        column.baseColumn.mapFromDriverValue(
+      return value.map((v) => {
+        if (column.baseColumn.columnType === "PgTimestamp") {
+          return v;
+        }
+
+        return column.baseColumn.mapFromDriverValue(
           column.baseColumn.mapToDriverValue(v),
-        ),
-      );
+        );
+      });
+    }
+
+    if (column.columnType === "PgTimestamp") {
+      return value;
     }
 
     return column.mapFromDriverValue(column.mapToDriverValue(value));


Test patch:
diff --git a/packages/core/src/indexing-store/historical.test.ts b/packages/core/src/indexing-store/historical.test.ts
index 4e5c7fbc..8081c510 100644
--- a/packages/core/src/indexing-store/historical.test.ts
+++ b/packages/core/src/indexing-store/historical.test.ts
@@ -1035,3 +1035,249 @@ test("text with null bytes", async (context) => {
     });
   });
 });
+
+test.skip("time", async (context) => {
+  const { database } = await setupDatabaseServices(context);
+
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      time: t.time().notNull(),
+    })),
+  };
+
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      time: "04:05:06",
+    });
+
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      time: "04:05:06",
+    });
+  });
+});
+
+test("timestamp", async (context) => {
+  const { database } = await setupDatabaseServices(context);
+
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      timestamp: t.timestamp().notNull(),
+    })),
+  };
+
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      timestamp: new Date(1742925862000),
+    });
+
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      timestamp: new Date(1742925862000),
+    });
+  });
+});
+
+test.skip("date", async (context) => {
+  const { database } = await setupDatabaseServices(context);
+
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      date: t.date({ mode: "date" }).notNull(),
+    })),
+  };
+
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      date: new Date(1742925862000),
+    });
+
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      date: new Date("2025-03-25T00:00:00.000Z"),
+    });
+  });
+});
+
+test.skip("interval", async (context) => {
+  const { database } = await setupDatabaseServices(context);
+
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      interval: t.interval().notNull(),
+    })),
+  };
+
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      interval: "1 day",
+    });
+
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      interval: "1 day",
+    });
+  });
+});
+
+test("point", async (context) => {
+  const { database } = await setupDatabaseServices(context);
+
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      point: t.point().notNull(),
+    })),
+  };
+
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      point: [1, 2],
+    });
+
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      point: [1, 2],
+    });
+  });
+});
+
+test("line", async (context) => {
+  const { database } = await setupDatabaseServices(context);
+
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      line: t.line().notNull(),
+    })),
+  };
+
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      line: [1, 2, 3],
+    });
+
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      line: [1, 2, 3],
+    });
+  });
+});
diff --git a/packages/core/src/indexing-store/realtime.test.ts b/packages/core/src/indexing-store/realtime.test.ts
index 43b4e3c3..a08e1ac3 100644
--- a/packages/core/src/indexing-store/realtime.test.ts
+++ b/packages/core/src/indexing-store/realtime.test.ts
@@ -791,3 +791,201 @@ test("text with null bytes", async (context) => {
     name: "tencentclub",
   });
 });
+
+test.skip("time", async (context) => {
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      time: t.time().notNull(),
+    })),
+  };
+
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+
+  const indexingStore = createRealtimeIndexingStore({
+    common: context.common,
+    database,
+    schemaBuild: { schema },
+  });
+
+  await indexingStore.insert(schema.account).values({
+    address: zeroAddress,
+    time: "04:05:06 PST",
+  });
+
+  const result = await indexingStore.find(schema.account, {
+    address: zeroAddress,
+  });
+
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    time: "04:05:06",
+  });
+});
+
+test("timestamp", async (context) => {
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      timestamp: t.timestamp().notNull(),
+    })),
+  };
+
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+
+  const indexingStore = createRealtimeIndexingStore({
+    common: context.common,
+    database,
+    schemaBuild: { schema },
+  });
+
+  await indexingStore.insert(schema.account).values({
+    address: zeroAddress,
+    timestamp: new Date(1742925862000),
+  });
+
+  const result = await indexingStore.find(schema.account, {
+    address: zeroAddress,
+  });
+
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    timestamp: new Date(1742925862000),
+  });
+});
+
+test.skip("date", async (context) => {
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      date: t.date({ mode: "date" }).notNull(),
+    })),
+  };
+
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+
+  const indexingStore = createRealtimeIndexingStore({
+    common: context.common,
+    database,
+    schemaBuild: { schema },
+  });
+
+  await indexingStore.insert(schema.account).values({
+    address: zeroAddress,
+    date: new Date(1742925862000),
+  });
+
+  const result = await indexingStore.find(schema.account, {
+    address: zeroAddress,
+  });
+
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    date: new Date("2025-03-25T00:00:00.000Z"),
+  });
+});
+
+test.skip("interval", async (context) => {
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      interval: t.interval().notNull(),
+    })),
+  };
+
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+
+  const indexingStore = createRealtimeIndexingStore({
+    common: context.common,
+    database,
+    schemaBuild: { schema },
+  });
+
+  await indexingStore.insert(schema.account).values({
+    address: zeroAddress,
+    interval: "18 months",
+  });
+
+  const result = await indexingStore.find(schema.account, {
+    address: zeroAddress,
+  });
+
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    interval: "1 year 6 mons",
+  });
+});
+
+test("point", async (context) => {
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      point: t.point().notNull(),
+    })),
+  };
+
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+
+  const indexingStore = createRealtimeIndexingStore({
+    common: context.common,
+    database,
+    schemaBuild: { schema },
+  });
+
+  await indexingStore.insert(schema.account).values({
+    address: zeroAddress,
+    point: [1, 2],
+  });
+
+  const result = await indexingStore.find(schema.account, {
+    address: zeroAddress,
+  });
+
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    point: [1, 2],
+  });
+});
+
+test("line", async (context) => {
+  const schema = {
+    account: onchainTable("account", (t) => ({
+      address: t.hex().primaryKey(),
+      line: t.line().notNull(),
+    })),
+  };
+
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+
+  const indexingStore = createRealtimeIndexingStore({
+    common: context.common,
+    database,
+    schemaBuild: { schema },
+  });
+
+  await indexingStore.insert(schema.account).values({
+    address: zeroAddress,
+    line: [1, 2, 3],
+  });
+
+  const result = await indexingStore.find(schema.account, {
+    address: zeroAddress,
+  });
+
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    line: [1, 2, 3],
+  });
+});