feat(compactor): allow header timestamp override

Author: corylanou

compactor.go

@@ -36,7 +36,8 @@ type Compactor struct {
 	total atomic.Uint32 // total pages (from last input's Commit)
 
 	// These flags will be set when encoding the header.
-	HeaderFlags uint32
+	HeaderFlags     uint32
+	HeaderTimestamp *int64
 
 	// If true, the compactor will not validate that input files have contiguous
 	// transaction IDs. This is false by default but can be enabled when
@@ -106,6 +107,10 @@ func (c *Compactor) Compact(ctx context.Context) (retErr error) {
 	// Fetch the first and last headers from the sorted readers.
 	minHdr := c.inputs[0].dec.Header()
 	maxHdr := c.inputs[len(c.inputs)-1].dec.Header()
+	timestamp := maxHdr.Timestamp
+	if c.HeaderTimestamp != nil {
+		timestamp = *c.HeaderTimestamp
+	}
 
 	// Generate output header. Skip NodeID as it's not meaningful after compaction.
 	if err := c.enc.EncodeHeader(Header{
@@ -115,7 +120,7 @@ func (c *Compactor) Compact(ctx context.Context) (retErr error) {
 		Commit:           maxHdr.Commit,
 		MinTXID:          minHdr.MinTXID,
 		MaxTXID:          maxHdr.MaxTXID,
-		Timestamp:        maxHdr.Timestamp,
+		Timestamp:        timestamp,
 		PreApplyChecksum: minHdr.PreApplyChecksum,
 	}); err != nil {
 		return fmt.Errorf("write header: %w", err)

compactor_test.go

@@ -120,6 +120,67 @@ func TestCompactor_Compact(t *testing.T) {
 			},
 		})
 	})
+	t.Run("HeaderTimestampOverride", func(t *testing.T) {
+		input0 := &ltx.FileSpec{
+			Header: ltx.Header{
+				Version:          ltx.Version,
+				PageSize:         512,
+				Commit:           1,
+				MinTXID:          2,
+				MaxTXID:          2,
+				Timestamp:        1000,
+				PreApplyChecksum: ltx.ChecksumFlag | 2,
+			},
+			Pages: []ltx.PageSpec{
+				{
+					Header: ltx.PageHeader{Pgno: 1},
+					Data:   bytes.Repeat([]byte{0x81}, 512),
+				},
+			},
+			Trailer: ltx.Trailer{
+				PostApplyChecksum: ltx.ChecksumFlag | 2,
+			},
+		}
+		input1 := &ltx.FileSpec{
+			Header: ltx.Header{
+				Version:          ltx.Version,
+				PageSize:         512,
+				Commit:           1,
+				MinTXID:          3,
+				MaxTXID:          3,
+				Timestamp:        2000,
+				PreApplyChecksum: ltx.ChecksumFlag | 3,
+			},
+			Pages: []ltx.PageSpec{
+				{
+					Header: ltx.PageHeader{Pgno: 1},
+					Data:   bytes.Repeat([]byte{0x91}, 512),
+				},
+			},
+			Trailer: ltx.Trailer{
+				PostApplyChecksum: ltx.ChecksumFlag | 3,
+			},
+		}
+		var buf0, buf1 bytes.Buffer
+		writeFileSpec(t, &buf0, input0)
+		writeFileSpec(t, &buf1, input1)
+
+		var output bytes.Buffer
+		c, err := ltx.NewCompactor(&output, []io.Reader{&buf0, &buf1})
+		if err != nil {
+			t.Fatal(err)
+		}
+		timestamp := int64(1000)
+		c.HeaderTimestamp = &timestamp
+		if err := c.Compact(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		spec := readFileSpec(t, &output)
+		if got, want := spec.Header.Timestamp, int64(1000); got != want {
+			t.Fatalf("Timestamp=%d, want %d", got, want)
+		}
+	})
 	t.Run("NonSnapshotPageDataOnly", func(t *testing.T) {
 		spec, err := compactFileSpecs(t,
 			&ltx.FileSpec{