使用行时间戳测量管道中的延迟

行时间戳提供了表中每一行上次更新时间的精确按时间顺序排列的记录。在同一事务中修改的行共享完全相同的时间戳，而在不同事务中修改的行按其提交时间排序。

主要用例包括：

  • 管道可观察性： 以高于客户端时间戳的准确度，测量流式引入、CDC 和 ETL 工作负载的端到端延迟及数据新鲜度。

  • 可靠的增量处理： 通过使用确定的提交时间，捕获事件时间戳可能会跳过的延迟或回填记录。

  • 确定的审计轨迹： 为监管合规或 SCD2 风格的里程碑建立事件的时间顺序。

要在表上设置行时间戳，请选择以下选项之一：

  • 在表上设置行时间戳： 使用对该表拥有 OWNERSHIP 权限的角色，在执行 CREATE TABLE 或 ALTER TABLE 命令时将 ROW_TIMESTAMP 属性设置为 TRUE。

    例如，CREATE TABLE … ROW_TIMESTAMP = TRUE 或 ALTER TABLE … SET ROW_TIMESTAMP = TRUE。

  • 默认为容器中的新表设置行时间戳： 在容器上将 ROW_TIMESTAMP_DEFAULT 属性设置为 TRUE。

    例如，ALTER SCHEMA … SET ROW_TIMESTAMP_DEFAULT = TRUE 表示在设置参数后，该架构中创建的每个新表都将默认启用行时间戳。

  • 批量为现有表启用行时间戳： 使用系统函数 SELECT SYSTEM$SET_ROW_TIMESTAMP_ON_ALL_SUPPORTED_TABLES。

    例如 SELECT SYSTEM$SET_ROW_TIMESTAMP_ON_ALL_SUPPORTED_TABLES('schema', '{my_db}.my_schema')。

    • 第一个实参是级别：schema、database 或 account 之一。

    • 第二个实参是容器的完全限定名称。

    此函数将向容器内所有符合条件的现有表添加行时间戳列，并确保新创建的表自动启用行时间戳。

    要成功执行该函数，您需要对所调用的容器拥有 MODIFY 权限。

启用行时间戳后，表将公开 METADATA$ROW_LAST_COMMIT_TIME 列，该列返回每一行上次修改时的时间戳。这实现了基于行修改时间的变更跟踪、增量处理和 Time Travel 查询。

备注

在数据共享场景中，即使生产者表启用了行时间戳，消费者也无法选择 METADATA$ROW_LAST_COMMIT_TIME。如果生产者想要与使用者共享行时间戳，则必须创建一个选择 METADATA$ROW_LAST_COMMIT_TIME 的视图，然后共享该视图。

以下语句演示了如何创建一个支持行时间戳的表。这些语句将数据插入表中并检索每一行的时间戳。

CREATE OR REPLACE TABLE table1(value1 STRING)
  ROW_TIMESTAMP = TRUE;

INSERT INTO table1 VALUES('some-value-a');

INSERT INTO table1 VALUES('some-value-b');

SELECT METADATA$ROW_LAST_COMMIT_TIME AS row_timestamp, *
  FROM table1
  ORDER BY 1;
Copy

主要用例

METADATA$ROW_LAST_COMMIT_TIME 元数据列有助于跟踪延迟。例如，如果您的目标是 5 秒的总延迟，此列可帮助您确定 Snowflake 对该延迟的贡献。

关键用例包括：

  • 测量引入延迟：跟踪从客户端创建行到在 Snowflake 中可见之间的时间，允许用户计算数据引入时间。

  • 测量端到端延迟：结合引入延迟和管道延迟，测量从数据生成到最终状态的总时间。

  • 测量管道延迟：跟踪数据在管道中移动时的时间戳。通过比较初始表和最终表的时间戳，用户可以衡量管道处理数据所需的时间。

    • 支持基于流、动态表和任务的管道。

示例：测量引入延迟

要使用 METADATA$ROW_LAST_COMMIT_TIME 元数据列测量引入延迟，请执行以下操作：

  1. 使用以下方法之一创建将数据发送到 Snowflake 的引入管道：

    • Snowpipe Streaming 引入 SDK。有关如何使用客户端 SDK 构建 Snowpipe Streaming 应用程序的简单示例，请参阅 此 Java 文件 (https://github.com/snowflakedb/snowflake-ingest-java/blob/master/src/main/java/net/snowflake/ingest/streaming/example/SnowflakeStreamingIngestExample.java) (GitHub)。

    • Snowpipe

    • COPY INTO <table> 命令

  2. 执行以下命令：

    ALTER SESSION SET TIMESTAMP_TZ_OUTPUT_FORMAT = 'YYYY-MM-DDTHH:MI:SS.FF3 TZH';

ALTER SESSION SET TIMEZONE = 'UTC';

CREATE OR REPLACE DATABASE mydb;

CREATE OR ALTER SCHEMA myschema;

CREATE OR REPLACE TABLE table1(record_id STRING, client_timestamp TIMESTAMP_LTZ);

-- The rows inserted from server-side-insert-1 up to this point will not have a valid METADATA$ROW_LAST_COMMIT_TIME timestamp.
INSERT INTO table1 VALUES('server-side-insert-1', current_timestamp());
    Copy

  3. 修改表以启用 METADATA$ROW_LAST_COMMIT_TIME 功能。

    ALTER TABLE table1 SET ROW_TIMESTAMP = TRUE;
    Copy

  4. 使用第 1 步中定义的引入管道，将包含 record_id 和 client_timestamp 列的数据引入您的 Snowflake 表。

  5. 如果不使用引入管道，可以插入新行作为即时示例。与第 2 步中的插入不同，由于已启用表属性，此次插入将拥有有效的 METADATA$ROW_LAST_COMMIT_TIME 时间戳。

    INSERT INTO table1 VALUES('server-side-insert-2', current_timestamp());
    Copy

  6. 再次运行您的客户端程序，然后执行以下操作：

    SELECT *, METADATA$ROW_LAST_COMMIT_TIME AS ROW_TIMESTAMP, TIMESTAMPDIFF(ms, CLIENT_TIMESTAMP, ROW_TIMESTAMP)
  AS INGEST_LATENCY FROM table1 ORDER BY 2;
    Copy

次要用例

行时间戳也可用于以下情况：

  • 数据保留：行时间戳可帮助删除旧记录，以节省存储成本。

  • 事件排序和变更跟踪：您可以使用行时间戳来跟踪变更。时间戳最大的行代表最新的变更。

  • 仅追加数据：如果行仅为追加模式，行时间戳可帮助筛选特定时间点的表状态，使您能够无视数据保留策略而使用 Time Travel。

限制和注意事项

  • 行时间戳仅保证在同一表内维持时间顺序，但发生故障转移时除外，此时不保证顺序。不保证跨表、跨区域或其他时间源的顺序。您不应跨表或其他来源比较行时间戳，因为这样做可能导致不一致。

  • 行时间戳反映的是上次更新时间，而非创建时间。例如，如果数据行在提交后被更新，行时间戳将反映最后更新时间，而不是数据的创建时间。

  • 在表启用行时间戳之前创建的行，其时间戳将被设置为 NULL。

  • 只要行被存储，行时间戳就会被存储。

  • 将 ROW_TIMESTAMP 属性设置为 FALSE 会永久删除所有已存储的 METADATA$ROW_LAST_COMMIT_TIME 值。重新启用该属性不会恢复这些值，且 Time Travel 查询将返回空结果。

  • 不支持为 Apache Iceberg™ 表、外部表、混合表、流或视图设置行时间戳。

  • 元数据列 METADATA$ROW_LAST_COMMIT_TIME 不能在以下内容中被引用：

  • 行时间戳无法通过存档表还原进行恢复。作为替代方案，您可以将 METADATA$ROW_LAST_COMMIT_TIME 实例化为另一个表的持久化列，以便在存档还原中使用。

行时间戳的克隆注意事项

克隆表会完整保留行时间戳。CREATE TABLE AS SELECT (CTAS) 和 INSERT INTO ... SELECT 等创建数据物理副本的操作将分配新的行时间戳，反映副本创建的时间。源表中的原始行时间戳将不会被保留。如果您想保留这些记录，请显式地将它们选择并存入持久化列，如以下示例所示：

CREATE TABLE my_archive AS
 SELECT *, METADATA$ROW_LAST_COMMIT_TIME AS original_commit_time
 FROM my_source_table;
Copy