1 files changed, 40 insertions, 0 deletions
diff --git a/core/record.go b/core/record.go
new file mode 100644
index 0000000..cb160b2
--- /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,
+	}
+}