1 files changed, 40 insertions, 0 deletions

diff --git a/core/record.go b/core/record.go
new file mode 100644
index 0000000..e18e58d
--- /dev/null
+++ b/core/record.go
@@ -0,0 +1,40 @@
+package core
+
+type RecordType byte
+
+const (
+        RecordData RecordType = iota
+        RecordWatermark
+        RecordBarrier
+)
+
+type StreamRecord[T any] struct {
+        Value     T
+        Key       string // for keyBy stateful processing
+        Timestamp int64  // epoch ms time
+        BarrierID uint64 // ID checkpoint barrier
+        Type      RecordType
+}
+
+func NewDataRecord[T any](key string, val T, timestamp int64) StreamRecord[T] {
+        return StreamRecord[T]{
+                Value:     val,
+                Key:       key,
+                Timestamp: timestamp,
+                Type:      RecordData,
+        }
+}
+
+func NewWatermarkRecord[T any](timestamp int64) StreamRecord[T] {
+        return StreamRecord[T]{
+                Type:      RecordWatermark,
+                Timestamp: timestamp,
+        }
+}
+
+func NewBarrierRecord[T any](barrierID uint64) StreamRecord[T] {
+        return StreamRecord[T]{
+                Type:      RecordBarrier,
+                BarrierID: barrierID,
+        }
+}