ksqlDB exposes ROWTIME as a pseudo column containing the timestamp associated with each record. You can use it with TIMESTAMPTOSTRING() to inspect or filter records by date.

Quick answer: Filter a stream using ROWTIME and TIMESTAMPTOSTRING(). If you need to read existing records instead of only new records, set auto.offset.reset to earliest.

Pseudo columnDescription
ROWTIMETimestamp associated with the source record.
ROWPARTITIONPartition of the source record.
ROWOFFSETOffset of the source record.

Query ROWTIME for a stream record

The following query formats ROWTIME so you can inspect the timestamp for matching records.

SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss.SSS')
FROM SOME_STREAM
WHERE COLUMN_NAME = 'some_value'
EMIT CHANGES;

Filter a stream by date

Format ROWTIME as a date and use the result in the filter:

SELECT *
FROM SOME_STREAM
WHERE TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd') = '2026-09-22'
EMIT CHANGES;
Filter a Kafka stream by date using ksqlDB

Read existing records from the beginning

ksqlDB normally reads from the latest offset for a new query. To read available records from the beginning, set:

SET 'auto.offset.reset' = 'earliest';

Pro tips:
ROWTIME is based on the record timestamp used by ksqlDB; by default this comes from the Kafka message timestamp. If your stream is configured with a TIMESTAMP property, that configuration can determine the timestamp used downstream. For time-based processing, make sure the timestamp semantics match your use case.

Available on Microsoft Store

SSRS Reports Migration Wizard

A simple Windows tool for migrating SSRS reports, data sources, and related configurations between report servers.

Kunal Rathi

With over 15 years of experience in data engineering and analytics, I've assisted countless clients in gaining valuable insights from their data. As a dedicated supporter of Data, Cloud and DevOps, I'm excited to connect with individuals who share my passion for this field. If my work resonates with you, we can talk and collaborate.