feat(changefeed): add initial_scan to ydb_table_changefeed

internal/resources/changefeed/changefeed.go

@@ -37,6 +37,7 @@ type changeDataCaptureSettings struct {
 	Format            *string
 	RetentionPeriod   *string
 	VirtualTimestamps *bool
+	InitialScan       bool
 	Entity            *helpers.YDBEntity
 	TableEntity       *helpers.YDBEntity
 	Consumers         []topictypes.Consumer
@@ -128,6 +129,7 @@ func changefeedResourceSchemaToChangefeedResource(d *schema.ResourceData) (*chan
 	if retentionPeriod, ok := d.Get("retention_period").(string); ok && retentionPeriod != "" {
 		settings.RetentionPeriod = &retentionPeriod
 	}
+	settings.InitialScan = d.Get("initial_scan").(bool)
 	settings.Consumers = expandConsumers(d)
 
 	return settings, nil
@@ -167,6 +169,10 @@ func flattenCDCDescription(
 	if err != nil {
 		return
 	}
+	err = d.Set("initial_scan", changefeedResource.InitialScan)
+	if err != nil {
+		return
+	}
 
 	curConsRaw := d.Get("consumer")
 	cons := helpers.ConsumerSort(curConsRaw, consumers)

internal/resources/changefeed/yql.go

@@ -36,6 +36,10 @@ func PrepareCreateRequest(cdc *changeDataCaptureSettings) string {
 		buf = helpers.AppendWithEscape(buf, *cdc.RetentionPeriod)
 		buf = append(buf, '"', ')')
 	}
+	if cdc.InitialScan {
+		buf = append(buf, ',', '\n')
+		buf = append(buf, "INITIAL_SCAN = TRUE"...)
+	}
 	buf = append(buf, '\n', ')')
 	return string(buf)
 }

internal/resources/changefeed/yql_test.go

@@ -0,0 +1,31 @@
+package changefeed
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestPrepareCreateRequest_initialScan(t *testing.T) {
+	format := "JSON"
+	t.Run("omits when false", func(t *testing.T) {
+		q := PrepareCreateRequest(&changeDataCaptureSettings{
+			TablePath:   "my/table",
+			Name:        "cf1",
+			Mode:        "UPDATES",
+			Format:      &format,
+			InitialScan: false,
+		})
+		require.NotContains(t, q, "INITIAL_SCAN")
+	})
+	t.Run("includes when true", func(t *testing.T) {
+		q := PrepareCreateRequest(&changeDataCaptureSettings{
+			TablePath:   "my/table",
+			Name:        "cf1",
+			Mode:        "UPDATES",
+			Format:      &format,
+			InitialScan: true,
+		})
+		require.Contains(t, q, "INITIAL_SCAN = TRUE")
+	})
+}

sdk/terraform/table/changefeed/resource.go

@@ -164,6 +164,14 @@ func ResourceSchema() map[string]*schema.Schema {
 			Optional:    true,
 			ForceNew:    true,
 		},
+		"initial_scan": {
+			Type:     schema.TypeBool,
+			Optional: true,
+			Default:  false,
+			ForceNew: true,
+			Description: "When true, runs an [initial snapshot scan](https://ydb.tech/docs/en/yql/reference/syntax/alter_table/changefeed) of the table when the changefeed is created. " +
+				"Only applies at creation; YDB does not expose this flag on describe, so Terraform keeps the configured value in state.",
+		},
 		"retention_period": {
 			Type:         schema.TypeString,
 			Description:  "Time of data retention in the topic, [ISO 8601](https://ru.wikipedia.org/wiki/ISO_8601) format.",