diff --git a/api/v2/model.go b/api/v2/model.go index d19c31d6f4..b0f06591e4 100644 --- a/api/v2/model.go +++ b/api/v2/model.go @@ -1133,6 +1133,7 @@ type Table struct { // SinkConfig represents sink config for a changefeed // This is a duplicate of config.SinkConfig type SinkConfig struct { +<<<<<<< HEAD Protocol *string `json:"protocol,omitempty"` SchemaRegistry *string `json:"schema_registry,omitempty"` CSVConfig *CSVConfig `json:"csv,omitempty"` @@ -1144,6 +1145,19 @@ type SinkConfig struct { DateSeparator *string `json:"date_separator,omitempty"` EnablePartitionSeparator *bool `json:"enable_partition_separator,omitempty"` FileIndexWidth *int `json:"file_index_width,omitempty"` +======= + Protocol *string `json:"protocol,omitempty" toml:"protocol,omitempty"` + SchemaRegistry *string `json:"schema_registry,omitempty" toml:"schema-registry,omitempty"` + CSVConfig *CSVConfig `json:"csv,omitempty" toml:"csv,omitempty"` + DispatchRules []*DispatchRule `json:"dispatchers,omitempty" toml:"dispatchers,omitempty"` + ColumnSelectors []*ColumnSelector `json:"column_selectors,omitempty" toml:"column-selectors,omitempty"` + TxnAtomicity *string `json:"transaction_atomicity,omitempty" toml:"transaction-atomicity,omitempty"` + EncoderConcurrency *int `json:"encoder_concurrency,omitempty" toml:"encoder-concurrency,omitempty"` + Terminator *string `json:"terminator,omitempty" toml:"terminator,omitempty"` + DateSeparator *config.DateSeparator `json:"date_separator,omitempty" toml:"date-separator,omitempty"` + EnablePartitionSeparator *bool `json:"enable_partition_separator,omitempty" toml:"enable-partition-separator,omitempty"` + FileIndexWidth *int `json:"file_index_width,omitempty" toml:"file-index-digit,omitempty"` +>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078)) // deprecated: it's become useless since v9.0.0 EnableKafkaSinkV2 *bool `json:"enable_kafka_sink_v2,omitempty"` OnlyOutputUpdatedColumns *bool `json:"only_output_updated_columns,omitempty"` diff --git a/api/v2/model_test.go b/api/v2/model_test.go index fe0afaee00..d93e71d7fb 100644 --- a/api/v2/model_test.go +++ b/api/v2/model_test.go @@ -13,6 +13,7 @@ package v2 import ( + "encoding/json" "testing" "github.com/pingcap/ticdc/pkg/config" @@ -20,6 +21,68 @@ import ( "github.com/stretchr/testify/require" ) +<<<<<<< HEAD +======= +func TestSinkConfigDateSeparator(t *testing.T) { + t.Parallel() + + var sinkConfig SinkConfig + require.NoError(t, json.Unmarshal([]byte(`{"date_separator":"DAY"}`), &sinkConfig)) + require.Equal(t, config.DateSeparatorDay, util.GetOrZero(sinkConfig.DateSeparator)) + + err := json.Unmarshal([]byte(`{"date_separator":"week"}`), &SinkConfig{}) + require.Error(t, err) + require.ErrorContains(t, err, "CDC:ErrStorageSinkInvalidConfig") +} + +func TestChangeFeedInfoCloneWithMaskedSensitiveData(t *testing.T) { + info := &ChangeFeedInfo{ + ID: "test", + SinkURI: "kafka://user:sink-password-sentinel@127.0.0.1:9092/topic?secret=uri-secret-sentinel", + Config: &ReplicaConfig{ + Sink: &SinkConfig{ + SchemaRegistry: util.AddressOf("https://registry.example.com?access-key=registry-secret-sentinel"), + KafkaConfig: &KafkaConfig{ + KafkaClientID: util.AddressOf("visible-client-id"), + SASLPassword: util.AddressOf("plain-password-sentinel"), + SASLGssAPIPassword: util.AddressOf("gssapi-password-sentinel"), + SASLOAuthClientSecret: util.AddressOf("oauth-secret-sentinel"), + SASLOAuthTokenURL: util.AddressOf("https://oauth.example.com/token?client_secret=token-url-secret-sentinel"), + LargeMessageHandle: &LargeMessageHandleConfig{ClaimCheckStorageURI: "s3://bucket/prefix?access-key=claim-check-secret-sentinel"}, + GlueSchemaRegistryConfig: &GlueSchemaRegistryConfig{ + AccessKey: "glue-access-sentinel", + SecretAccessKey: "glue-secret-sentinel", + Token: "glue-token-sentinel", + }, + }, + PulsarConfig: &PulsarConfig{ + AuthenticationToken: util.AddressOf("pulsar-token-sentinel"), + BasicPassword: util.AddressOf("pulsar-password-sentinel"), + OAuth2: &PulsarOAuth2{OAuth2PrivateKey: "pulsar-private-key-sentinel"}, + }, + }, + Consistent: &ConsistentConfig{Storage: util.AddressOf("s3://bucket/prefix?access-key=consistent-secret-sentinel")}, + }, + } + original, err := info.Marshal() + require.NoError(t, err) + + masked, err := info.CloneWithMaskedSensitiveData() + require.NoError(t, err) + output, err := masked.Marshal() + require.NoError(t, err) + require.NotContains(t, output, "sentinel") + require.NotContains(t, output, "memory_quota") + require.Contains(t, output, "visible-client-id") + require.Nil(t, masked.Config.Sink.KafkaConfig.Key) + after, err := info.Marshal() + require.NoError(t, err) + require.Equal(t, original, after) +} + +// TestReplicaConfigConversion verifies API/internal replica config conversion, +// including round-tripping the optional event collector batch overrides. +>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078)) func TestReplicaConfigConversion(t *testing.T) { t.Parallel() diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index 42081fbc9d..70f33a3ac4 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -60,6 +60,7 @@ type indexRange struct { type consumer struct { replicationCfg *config.ReplicaConfig + dateSeparator config.DateSeparator codecCfg *common.Config columnSelectors *columnselector.ColumnSelectors externalStorage storeapi.Storage @@ -103,6 +104,7 @@ func newConsumer(ctx context.Context) (*consumer, error) { log.Error("failed to validate replica config", zap.Error(err)) return nil, err } + dateSeparator := putil.GetOrZero(replicaConfig.Sink.DateSeparator) switch putil.GetOrZero(replicaConfig.Sink.Protocol) { case config.ProtocolCsv.String(): @@ -157,6 +159,7 @@ func newConsumer(ctx context.Context) (*consumer, error) { return &consumer{ replicationCfg: replicaConfig, + dateSeparator: dateSeparator, codecCfg: codecConfig, columnSelectors: columnSelectors, externalStorage: storage, @@ -223,12 +226,24 @@ func (c *consumer) getNewFiles( origDMLIdxMap[k] = m } +<<<<<<< HEAD err := c.externalStorage.WalkDir(ctx, opt, func(path string, size int64) error { if cloudstorage.IsSchemaFile(path) { err := c.parseSchemaFilePath(ctx, path) if err != nil { log.Error("failed to parse schema file path", zap.Error(err)) // skip handling this file +======= + err := c.externalStorage.WalkDir(ctx, opt, func(path string, _ int64) error { + if cloudstorage.IsSchemaFile(path) { + c.parseSchemaFilePath(ctx, path) + return nil + } + if strings.HasSuffix(path, ".index") { + var dmlkey cloudstorage.DMLPathKey + if err := dmlkey.ParseIndexFilePath(c.dateSeparator, path); err != nil { + log.Debug("ignore handling unsupported dml index file", zap.String("path", path)) +>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078)) return nil } } else if strings.HasSuffix(path, ".index") { diff --git a/downstreamadapter/sink/cloudstorage/dml_writers_test.go b/downstreamadapter/sink/cloudstorage/dml_writers_test.go index 488b798ff4..4fdf1b34aa 100644 --- a/downstreamadapter/sink/cloudstorage/dml_writers_test.go +++ b/downstreamadapter/sink/cloudstorage/dml_writers_test.go @@ -161,7 +161,7 @@ func TestCloudStorageWriteEventsWithoutDateSeparator(t *testing.T) { err = replicaConfig.ValidateAndAdjust(sinkURI) require.NoError(t, err) - replicaConfig.Sink.DateSeparator = putil.AddressOf(config.DateSeparatorNone.String()) + replicaConfig.Sink.DateSeparator = putil.AddressOf(config.DateSeparatorNone) replicaConfig.Sink.FileIndexWidth = putil.AddressOf(6) ctx, cancel := context.WithCancel(context.Background()) @@ -279,7 +279,7 @@ func TestCloudStorageWriteEventsWithDateSeparator(t *testing.T) { err = replicaConfig.ValidateAndAdjust(sinkURI) require.NoError(t, err) - replicaConfig.Sink.DateSeparator = putil.AddressOf(config.DateSeparatorDay.String()) + replicaConfig.Sink.DateSeparator = putil.AddressOf(config.DateSeparatorDay) replicaConfig.Sink.FileIndexWidth = putil.AddressOf(6) mockClock := pclock.NewMock() diff --git a/downstreamadapter/sink/cloudstorage/sink.go b/downstreamadapter/sink/cloudstorage/sink.go index 223136f53a..3324171057 100644 --- a/downstreamadapter/sink/cloudstorage/sink.go +++ b/downstreamadapter/sink/cloudstorage/sink.go @@ -405,11 +405,11 @@ func (s *sink) initCron( } func (s *sink) bgCleanup(ctx context.Context) { - if s.cfg.DateSeparator != config.DateSeparatorDay.String() || s.cfg.FileExpirationDays <= 0 { + if s.cfg.DateSeparator != config.DateSeparatorDay || s.cfg.FileExpirationDays <= 0 { log.Info("skip cleanup expired files for storage sink", zap.String("keyspace", s.changefeedID.Keyspace()), zap.String("changefeedID", s.changefeedID.Name()), - zap.String("dateSeparator", s.cfg.DateSeparator), + zap.Stringer("dateSeparator", s.cfg.DateSeparator), zap.Int("expiredFileTTL", s.cfg.FileExpirationDays)) return } @@ -419,7 +419,7 @@ func (s *sink) bgCleanup(ctx context.Context) { log.Info("start schedule cleanup expired files for storage sink", zap.String("keyspace", s.changefeedID.Keyspace()), zap.String("changefeedID", s.changefeedID.Name()), - zap.String("dateSeparator", s.cfg.DateSeparator), + zap.Stringer("dateSeparator", s.cfg.DateSeparator), zap.Int("expiredFileTTL", s.cfg.FileExpirationDays)) // wait for the context done diff --git a/downstreamadapter/sink/cloudstorage/sink_test.go b/downstreamadapter/sink/cloudstorage/sink_test.go index f04891ced2..055b592af7 100644 --- a/downstreamadapter/sink/cloudstorage/sink_test.go +++ b/downstreamadapter/sink/cloudstorage/sink_test.go @@ -148,7 +148,7 @@ func TestCloudStorageSinkWithColumnSelector(t *testing.T) { } err = replicaConfig.ValidateAndAdjust(sinkURI) require.NoError(t, err) - replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String()) + replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone) ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -646,8 +646,20 @@ func TestCleanupExpiredFiles(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() +<<<<<<< HEAD mockPDClock := pdutil.NewClock4Test() appcontext.SetService(appcontext.DefaultPDClock, mockPDClock) +======= + cloudStorageSink := &sink{ + changefeedID: common.NewChangefeedID4Test("test", "test"), + cfg: &cloudstorage.Config{ + DateSeparator: config.DateSeparatorDay, + FileExpirationDays: 1, + FileCleanupCronSpec: util.GetOrZero(replicaConfig.Sink.CloudStorageConfig.FileCleanupCronSpec), + }, + } + require.NoError(t, cloudStorageSink.initCron(ctx, sinkURI, cleanupJobs)) +>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078)) cloudStorageSink, err := newSinkForTest(ctx, replicaConfig, sinkURI, cleanupJobs) go cloudStorageSink.Run(ctx) diff --git a/downstreamadapter/sink/cloudstorage/writer_test.go b/downstreamadapter/sink/cloudstorage/writer_test.go index 8b8214154a..c9251a6182 100644 --- a/downstreamadapter/sink/cloudstorage/writer_test.go +++ b/downstreamadapter/sink/cloudstorage/writer_test.go @@ -55,7 +55,7 @@ func testWriter(ctx context.Context, t *testing.T, dir string) *writer { require.NoError(t, err) cfg := cloudstorage.NewConfig() replicaConfig := config.GetDefaultReplicaConfig() - replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String()) + replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone) err = cfg.Apply(context.TODO(), sinkURI, replicaConfig.Sink, true) cfg.FileIndexWidth = 6 require.NoError(t, err) @@ -470,7 +470,7 @@ func TestWriterStoresPendingMessagesInSpoolBeforeFlush(t *testing.T) { cfg := cloudstorage.NewConfig() replicaConfig := config.GetDefaultReplicaConfig() - replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String()) + replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone) replicaConfig.Sink.CloudStorageConfig = &config.CloudStorageConfig{ // Keep the quota larger than this encoded batch so the controller still // spills it to local spool files instead of taking the oversized in-memory fast path. @@ -645,7 +645,7 @@ func TestWriterIndexWriteError(t *testing.T) { require.NoError(t, err) cfg := cloudstorage.NewConfig() replicaConfig := config.GetDefaultReplicaConfig() - replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String()) + replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone) err = cfg.Apply(context.TODO(), sinkURI, replicaConfig.Sink, true) require.NoError(t, err) cfg.FileIndexWidth = 6 @@ -711,7 +711,7 @@ func TestWriterDataFileCloseError(t *testing.T) { require.NoError(t, err) cfg := cloudstorage.NewConfig() replicaConfig := config.GetDefaultReplicaConfig() - replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String()) + replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone) err = cfg.Apply(context.TODO(), sinkURI, replicaConfig.Sink, true) require.NoError(t, err) cfg.FileIndexWidth = 6 diff --git a/pkg/cloudstorage/path_key.go b/pkg/cloudstorage/path_key.go new file mode 100644 index 0000000000..f0709e12f6 --- /dev/null +++ b/pkg/cloudstorage/path_key.go @@ -0,0 +1,344 @@ +// Copyright 2023 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package cloudstorage + +import ( + "cmp" + "path" + "regexp" + "strconv" + "strings" + + "github.com/pingcap/log" + "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/config" + "github.com/pingcap/ticdc/pkg/errors" + "go.uber.org/zap" +) + +const schemaFilePartitionNum int64 = -1 + +// SchemaPathKey identifies the schema path scope parsed from or used to build +// cloud storage paths. +type SchemaPathKey struct { + // Schema is the first directory level in storage sink paths. + // Example: ///... + Schema string + // Table is the second directory level for table-level schema file and data paths. + // For database-level schema files, this field is empty and the path is + // /meta/schema_{tableVersion}_{checksum}.json. + Table string + // TableVersion is the schema version encoded in the path. + // In CDC it is carried by tableInfoVersion, and for DDL-related versions it + // is typically equal to the DDL finishedTs. + TableVersion uint64 +} + +// GetKey returns the quoted schema/table key used by consumer maps. +// For database-level schema files, Table is empty. +func (s *SchemaPathKey) GetKey() string { + return common.QuoteSchema(s.Schema, s.Table) +} + +// Parse fills SchemaPathKey from a schema file path. +// Input: +// - /meta/schema__.json +// - /
/meta/schema__.json +// +// Output fields are Schema, Table, and TableVersion. Table is empty for +// database-level schema files. Invalid paths panic. +func (s *SchemaPathKey) Parse(path string) { + // For /
/meta/schema_{tableVersion}_{checksum}.json, the parts + // should be ["", "
", "meta", "schema_{tableVersion}_{checksum}.json"]. + matches := strings.Split(path, "/") + + var schema, table string + schema = matches[0] + switch len(matches) { + case 3: + table = "" + case 4: + table = matches[1] + default: + log.Panic("cannot match schema path pattern", zap.String("path", path)) + } + + if matches[len(matches)-2] != "meta" { + log.Panic("cannot match schema path pattern", zap.String("path", path)) + } + + schemaFileName := matches[len(matches)-1] + version, _ := mustParseSchemaFileName(schemaFileName) + + *s = SchemaPathKey{ + Schema: schema, + Table: table, + TableVersion: version, + } +} + +type FileIndexKey struct { + // DispatcherID is used in file name only when table-across-nodes is enabled. + // File pattern: CDC_{dispatcherID}_{index}.{ext} + DispatcherID string + // EnableTableAcrossNodes controls whether dispatcher ID is embedded in + // data/index file names to avoid collisions across captures. + EnableTableAcrossNodes bool +} + +type FileIndex struct { + FileIndexKey + // Idx is the monotonically increasing file sequence number in one + // directory scope (schema/table/version[/partition][/date] or + // tableID/version[/date]). + Idx uint64 +} + +// DMLPathKey is the key of dml path. +type DMLPathKey struct { + SchemaPathKey + // UseTableIDAsPath controls whether TableID is used as the first path + // element instead of Schema/Table. + UseTableIDAsPath bool + // TableID is set when UseTableIDAsPath is true. + TableID int64 + // PartitionNum is an optional path level for partition table output. + // It is present only when partition-separator is enabled. + PartitionNum int64 + // Date is an optional path level controlled by date-separator + // (year/month/day/none). + Date string +} + +// NewSchemaFileDMLPathKey returns the synthetic DML path key used to order a +// schema file before data files with the same schema version. +func NewSchemaFileDMLPathKey(schemaKey SchemaPathKey) DMLPathKey { + return DMLPathKey{ + SchemaPathKey: schemaKey, + PartitionNum: schemaFilePartitionNum, + } +} + +// IsSchemaFileDMLPathKey checks whether the key represents a schema file marker. +func (d DMLPathKey) IsSchemaFileDMLPathKey() bool { + return d.PartitionNum == schemaFilePartitionNum && d.Date == "" +} + +// CompareDMLPathKey compares DML path keys in cloud storage replay order. +func CompareDMLPathKey(x, y DMLPathKey) int { + if r := cmp.Compare(x.TableVersion, y.TableVersion); r != 0 { + return r + } + if r := cmp.Compare(x.PartitionNum, y.PartitionNum); r != 0 { + return r + } + if r := cmp.Compare(x.Date, y.Date); r != 0 { + return r + } + if x.UseTableIDAsPath != y.UseTableIDAsPath { + if x.UseTableIDAsPath { + return 1 + } + return -1 + } + if r := cmp.Compare(x.TableID, y.TableID); r != 0 { + return r + } + if r := cmp.Compare(x.Schema, y.Schema); r != 0 { + return r + } + return cmp.Compare(x.Table, y.Table) +} + +// GenerateDMLFilePath returns the full data file path. +// The receiver supplies the data directory fields. fileIndex supplies the +// dispatcher ID and sequence number used in the file name. extension is the +// file suffix, for example ".json" or ".csv". +func (d *DMLPathKey) GenerateDMLFilePath( + fileIndex *FileIndex, extension string, fileIndexWidth int, +) string { + fileName := generateDataFileName( + fileIndex.EnableTableAcrossNodes, fileIndex.DispatcherID, + fileIndex.Idx, extension, fileIndexWidth) + return path.Join(d.generateDMLDataDirPath(), fileName) +} + +// GenerateIndexFilePath returns the index file path for this data directory. +// Output is /meta/CDC.index or /meta/CDC_.index. +func (d *DMLPathKey) GenerateIndexFilePath(fileIndexKey FileIndexKey) string { + return path.Join( + d.generateDMLDataDirPath(), + generateIndexFileName(fileIndexKey.EnableTableAcrossNodes, fileIndexKey.DispatcherID), + ) +} + +func isDMLIndexFileName(fileName string) bool { + if fileName == "CDC.index" { + return true + } + return strings.HasPrefix(fileName, "CDC_") && + strings.HasSuffix(fileName, ".index") && + len(fileName) > len("CDC_.index") +} + +// generateDMLDataDirPath returns the canonical data directory path. +// Output is either /[/date] or +// /
/[/partition][/date]. +func (d DMLPathKey) generateDMLDataDirPath() string { + elems := make([]string, 0, 5) + if d.UseTableIDAsPath { + elems = append(elems, strconv.FormatInt(d.TableID, 10)) + elems = append(elems, strconv.FormatUint(d.TableVersion, 10)) + if d.Date != "" { + elems = append(elems, d.Date) + } + return path.Join(elems...) + } + elems = append(elems, d.Schema, d.Table) + elems = append(elems, strconv.FormatUint(d.TableVersion, 10)) + if d.PartitionNum != 0 { + elems = append(elems, strconv.FormatInt(d.PartitionNum, 10)) + } + if d.Date != "" { + elems = append(elems, d.Date) + } + return path.Join(elems...) +} + +func (d *DMLPathKey) parseDMLDataDir( + dateSeparator config.DateSeparator, parts []string, filePath string, +) error { + var ( + key DMLPathKey + version string + tableID string + partition string + hasDate bool + dateRE string + ) + switch dateSeparator { + case config.DateSeparatorNone: + case config.DateSeparatorYear: + hasDate = true + dateRE = config.DateSeparatorYear.GetPattern() + case config.DateSeparatorMonth: + hasDate = true + dateRE = config.DateSeparatorMonth.GetPattern() + case config.DateSeparatorDay: + hasDate = true + dateRE = config.DateSeparatorDay.GetPattern() + default: + return errors.ErrStorageSinkInvalidDateSeparator.GenWithStackByArgs(dateSeparator) + } + + switch { + case !hasDate && len(parts) == 2: + key.Schema = parts[0] + tableID = parts[0] + version = parts[1] + case !hasDate && len(parts) == 3: + key.Schema, key.Table = parts[0], parts[1] + version = parts[2] + case !hasDate && len(parts) == 4: + key.Schema, key.Table = parts[0], parts[1] + partition = parts[3] + version = parts[2] + case hasDate && len(parts) == 3: + key.Schema = parts[0] + tableID = parts[0] + key.Date = parts[2] + version = parts[1] + case hasDate && len(parts) == 4: + key.Schema, key.Table = parts[0], parts[1] + key.Date = parts[3] + version = parts[2] + case hasDate && len(parts) == 5: + key.Schema, key.Table = parts[0], parts[1] + partition = parts[3] + key.Date = parts[4] + version = parts[2] + default: + return invalidDMLPathError(filePath) + } + + if hasDate && !regexp.MustCompile("^"+dateRE+"$").MatchString(key.Date) { + return invalidDMLPathError(filePath) + } + if tableID != "" { + tableIDNum, err := strconv.ParseInt(tableID, 10, 64) + if err != nil { + return invalidDMLPathWrapError(err, filePath) + } + key.UseTableIDAsPath = true + key.TableID = tableIDNum + } + if partition != "" { + partitionNum, err := strconv.ParseInt(partition, 10, 64) + if err != nil { + return invalidDMLPathWrapError(err, filePath) + } + key.PartitionNum = partitionNum + } + tableVersion, err := strconv.ParseUint(version, 10, 64) + if err != nil { + return invalidDMLPathWrapError(err, filePath) + } + key.TableVersion = tableVersion + *d = key + return nil +} + +// ParseIndexFilePath fills DMLPathKey from an index file path. +// Input is /meta/CDC.index or +// /meta/CDC_.index. Only the data directory portion is +// parsed here; the file index itself is stored in the index file content and is +// read by the caller. +func (d *DMLPathKey) ParseIndexFilePath(dateSeparator config.DateSeparator, path string) error { + parts := strings.Split(path, "/") + if len(parts) < 4 || parts[len(parts)-2] != "meta" || !isDMLIndexFileName(parts[len(parts)-1]) { + return invalidDMLPathError(path) + } + return d.parseDMLDataDir(dateSeparator, parts[:len(parts)-2], path) +} + +// ParseDMLFilePath fills DMLPathKey from a data file path and returns the file +// index encoded in the file name. +// Input is /CDC or +// /CDC__. Invalid paths panic. +func (d *DMLPathKey) ParseDMLFilePath( + dateSeparator config.DateSeparator, filePath, extension string, +) FileIndex { + parts := strings.Split(filePath, "/") + fileIndex, err := ParseFileIndexFromFileName(parts[len(parts)-1], extension) + if err != nil { + log.Panic("parse file index from file name failed", + zap.String("path", filePath), zap.Error(err)) + } + if err := d.parseDMLDataDir(dateSeparator, parts[:len(parts)-1], filePath); err != nil { + log.Panic("parse dml data dir failed", + zap.String("path", filePath), zap.Error(err)) + } + return fileIndex +} + +func invalidDMLPathError(filePath string) error { + return errors.ErrStorageSinkInvalidFileName.GenWithStack( + "cannot match dml path pattern for %q", filePath) +} + +func invalidDMLPathWrapError(err error, filePath string) error { + return errors.WrapError( + errors.ErrStorageSinkInvalidFileName, err, + "cannot match dml path pattern for %q", filePath) +} diff --git a/pkg/cloudstorage/path_key_test.go b/pkg/cloudstorage/path_key_test.go new file mode 100644 index 0000000000..10e948ca2a --- /dev/null +++ b/pkg/cloudstorage/path_key_test.go @@ -0,0 +1,195 @@ +// Copyright 2023 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package cloudstorage + +import ( + "fmt" + "testing" + + "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/config" + "github.com/stretchr/testify/require" +) + +func TestSchemaPathKey(t *testing.T) { + t.Parallel() + + testCases := []struct { + path string + schemakey SchemaPathKey + }{ + // Test for database schema path: /meta/schema_{tableVersion}_{checksum}.json + { + path: "test_schema/meta/schema_1_2.json", + schemakey: SchemaPathKey{ + Schema: "test_schema", + Table: "", + TableVersion: 1, + }, + }, + // Test for table-level schema file path: /
/meta/schema_{tableVersion}_{checksum}.json + { + path: "test_schema/test_table/meta/schema_11_22.json", + schemakey: SchemaPathKey{ + Schema: "test_schema", + Table: "test_table", + TableVersion: 11, + }, + }, + } + for _, tc := range testCases { + var schemaKey SchemaPathKey + schemaKey.Parse(tc.path) + require.Equal(t, tc.schemakey, schemaKey) + } +} + +func TestGenerateDMLFilePath(t *testing.T) { + t.Parallel() + + dispatcherID := common.NewDispatcherID() + testCases := []struct { + index uint64 + fileIndexWidth int + extension string + dateSeparator config.DateSeparator + path string + dmlkey DMLPathKey + }{ + { + index: 10, + fileIndexWidth: 20, + extension: ".csv", + dateSeparator: config.DateSeparatorDay, + path: fmt.Sprintf("schema1/table1/123456/2023-05-09/CDC_%s_00000000000000000010.csv", dispatcherID.String()), + dmlkey: DMLPathKey{ + SchemaPathKey: SchemaPathKey{ + Schema: "schema1", + Table: "table1", + TableVersion: 123456, + }, + PartitionNum: 0, + Date: "2023-05-09", + }, + }, + { + index: 10, + fileIndexWidth: 20, + extension: ".csv", + dateSeparator: config.DateSeparatorNone, + path: fmt.Sprintf("12345/123456/CDC_%s_00000000000000000010.csv", dispatcherID.String()), + dmlkey: DMLPathKey{ + SchemaPathKey: SchemaPathKey{ + Schema: "12345", + TableVersion: 123456, + }, + UseTableIDAsPath: true, + TableID: 12345, + }, + }, + { + index: 10, + fileIndexWidth: 20, + extension: ".csv", + dateSeparator: config.DateSeparatorDay, + path: fmt.Sprintf("schema1/table1/123456/55/2023-05-09/CDC_%s_00000000000000000010.csv", dispatcherID.String()), + dmlkey: DMLPathKey{ + SchemaPathKey: SchemaPathKey{ + Schema: "schema1", + Table: "table1", + TableVersion: 123456, + }, + PartitionNum: 55, + Date: "2023-05-09", + }, + }, + } + + for _, tc := range testCases { + fileIndex := &FileIndex{ + FileIndexKey: FileIndexKey{ + DispatcherID: dispatcherID.String(), + EnableTableAcrossNodes: true, + }, + Idx: tc.index, + } + fileName := tc.dmlkey.GenerateDMLFilePath(fileIndex, tc.extension, tc.fileIndexWidth) + require.Equal(t, tc.path, fileName) + var pathKey DMLPathKey + gotFileIndex := pathKey.ParseDMLFilePath(tc.dateSeparator, fileName, tc.extension) + require.Equal(t, tc.dmlkey, pathKey) + require.Equal(t, *fileIndex, gotFileIndex) + } +} + +func TestSchemaFileDMLPathKeyOrder(t *testing.T) { + t.Parallel() + + schemaKey := SchemaPathKey{ + Schema: "schema1", + Table: "table1", + TableVersion: 123456, + } + schemaDMLKey := NewSchemaFileDMLPathKey(schemaKey) + require.True(t, schemaDMLKey.IsSchemaFileDMLPathKey()) + + dataDMLKey := DMLPathKey{ + SchemaPathKey: schemaKey, + Date: "2023-05-09", + } + require.Less(t, CompareDMLPathKey(schemaDMLKey, dataDMLKey), 0) + require.Greater(t, CompareDMLPathKey(dataDMLKey, schemaDMLKey), 0) + require.Zero(t, CompareDMLPathKey(schemaDMLKey, NewSchemaFileDMLPathKey(schemaKey))) + + tableIDPathKey := DMLPathKey{ + SchemaPathKey: SchemaPathKey{ + Schema: "12345", + TableVersion: schemaKey.TableVersion, + }, + UseTableIDAsPath: true, + TableID: 12345, + } + require.NotZero(t, CompareDMLPathKey(dataDMLKey, tableIDPathKey)) +} + +func TestParseIndexFilePathRejectsUnsupportedPath(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + dateSeparator config.DateSeparator + path string + }{ + { + name: "legacy schema table date index", + dateSeparator: config.DateSeparatorDay, + path: "test/binary_columns_dummy/2026-06-23/meta/CDC.index", + }, + { + name: "invalid index file name", + dateSeparator: config.DateSeparatorNone, + path: "schema1/table1/123456/meta/notCDC.index", + }, + { + name: "date does not match separator", + dateSeparator: config.DateSeparatorMonth, + path: "schema1/table1/123456/2023-05-09/meta/CDC.index", + }, + } + + for _, tc := range testCases { + var pathKey DMLPathKey + require.Error(t, pathKey.ParseIndexFilePath(tc.dateSeparator, tc.path), tc.name) + } +} diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go index b41783619e..b01c019468 100644 --- a/pkg/config/replica_config.go +++ b/pkg/config/replica_config.go @@ -67,7 +67,7 @@ var defaultReplicaConfig = &ReplicaConfig{ }, EncoderConcurrency: util.AddressOf(DefaultEncoderGroupConcurrency), Terminator: util.AddressOf(CRLF), - DateSeparator: util.AddressOf(DateSeparatorDay.String()), + DateSeparator: util.AddressOf(DateSeparatorDay), EnablePartitionSeparator: util.AddressOf(true), OnlyOutputUpdatedColumns: util.AddressOf(false), DeleteOnlyOutputHandleKeyColumns: util.AddressOf(false), diff --git a/pkg/config/sink.go b/pkg/config/sink.go index f7599235a1..7010da2520 100644 --- a/pkg/config/sink.go +++ b/pkg/config/sink.go @@ -152,7 +152,7 @@ type SinkConfig struct { // Terminator is NOT available when the downstream is DB. Terminator *string `toml:"terminator" json:"terminator,omitempty"` // DateSeparator is only available when the downstream is Storage. - DateSeparator *string `toml:"date-separator" json:"date-separator,omitempty"` + DateSeparator *DateSeparator `toml:"date-separator" json:"date-separator,omitempty"` // EnablePartitionSeparator is only available when the downstream is Storage. EnablePartitionSeparator *bool `toml:"enable-partition-separator" json:"enable-partition-separator,omitempty"` // FileIndexWidth is only available when the downstream is Storage @@ -356,6 +356,20 @@ func (d *DateSeparator) FromString(separator string) error { return nil } +// MarshalText implements encoding.TextMarshaler. +func (d DateSeparator) MarshalText() ([]byte, error) { + return []byte(d.String()), nil +} + +// UnmarshalText implements encoding.TextUnmarshaler. +func (d *DateSeparator) UnmarshalText(text []byte) error { + if err := d.FromString(string(text)); err != nil { + return cerror.ErrStorageSinkInvalidConfig.GenWithStack( + "invalid date separator %q", text) + } + return nil +} + // GetPattern returns the pattern of the date separator. func (d DateSeparator) GetPattern() string { switch d { @@ -832,14 +846,6 @@ func (s *SinkConfig) validateAndAdjust(sinkURI *url.URL) error { // validate storage sink related config if sinkURI != nil && IsStorageScheme(sinkURI.Scheme) { - // validate date separator - if len(util.GetOrZero(s.DateSeparator)) > 0 { - var separator DateSeparator - if err := separator.FromString(util.GetOrZero(s.DateSeparator)); err != nil { - return cerror.WrapError(cerror.ErrSinkInvalidConfig, err) - } - } - // File index width should be in [minFileIndexWidth, maxFileIndexWidth]. // In most scenarios, the user does not need to change this configuration, // so the default value of this parameter is not set and just make silent diff --git a/pkg/config/sink_test.go b/pkg/config/sink_test.go index 20c0d9075c..e802c26830 100644 --- a/pkg/config/sink_test.go +++ b/pkg/config/sink_test.go @@ -14,13 +14,44 @@ package config import ( + "encoding/json" "net/url" "testing" +<<<<<<< HEAD +======= + "github.com/BurntSushi/toml" + "github.com/pingcap/ticdc/pkg/errors" +>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078)) "github.com/pingcap/ticdc/pkg/util" "github.com/stretchr/testify/require" ) +func TestDateSeparatorSerialization(t *testing.T) { + t.Parallel() + + var jsonConfig SinkConfig + require.NoError(t, json.Unmarshal([]byte(`{"date-separator":"DAY"}`), &jsonConfig)) + require.Equal(t, DateSeparatorDay, util.GetOrZero(jsonConfig.DateSeparator)) + + encoded, err := json.Marshal(jsonConfig) + require.NoError(t, err) + require.JSONEq(t, `{"date-separator":"day","integrity":null}`, string(encoded)) + + var tomlConfig SinkConfig + _, err = toml.Decode(`date-separator = "Month"`, &tomlConfig) + require.NoError(t, err) + require.Equal(t, DateSeparatorMonth, util.GetOrZero(tomlConfig.DateSeparator)) + + err = json.Unmarshal([]byte(`{"date-separator":"week"}`), &SinkConfig{}) + require.Error(t, err) + require.True(t, errors.ErrStorageSinkInvalidConfig.Equal(err), "%+v", err) + + _, err = toml.Decode(`date-separator = "week"`, &SinkConfig{}) + require.Error(t, err) + require.ErrorContains(t, err, "CDC:ErrStorageSinkInvalidConfig") +} + func TestValidateTxnAtomicity(t *testing.T) { t.Parallel() testCases := []struct { diff --git a/pkg/sink/cloudstorage/config.go b/pkg/sink/cloudstorage/config.go index 1acf3dbf9f..034385df1d 100644 --- a/pkg/sink/cloudstorage/config.go +++ b/pkg/sink/cloudstorage/config.go @@ -80,7 +80,7 @@ type Config struct { FlushInterval time.Duration FileSize int FileIndexWidth int - DateSeparator string + DateSeparator config.DateSeparator FileExpirationDays int FileCleanupCronSpec string EnablePartitionSeparator bool diff --git a/pkg/sink/cloudstorage/config_test.go b/pkg/sink/cloudstorage/config_test.go index 7fb96a85e2..92463ed188 100644 --- a/pkg/sink/cloudstorage/config_test.go +++ b/pkg/sink/cloudstorage/config_test.go @@ -22,6 +22,7 @@ import ( "github.com/aws/aws-sdk-go-v2/aws" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/util" "github.com/stretchr/testify/require" ) @@ -31,7 +32,7 @@ func TestConfigApply(t *testing.T) { expected.FlushInterval = 10 * time.Second expected.FileSize = 16 * 1024 * 1024 expected.FileIndexWidth = config.DefaultFileIndexWidth - expected.DateSeparator = config.DateSeparatorDay.String() + expected.DateSeparator = config.DateSeparatorDay expected.EnablePartitionSeparator = true expected.FlushConcurrency = 1 expected.SpoolDiskQuota = 10 * 1024 * 1024 * 1024 @@ -41,6 +42,7 @@ func TestConfigApply(t *testing.T) { require.NoError(t, err) replicaConfig := config.GetDefaultReplicaConfig() + replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorDay) err = replicaConfig.ValidateAndAdjust(sinkURI) require.NoError(t, err) cfg := NewConfig() diff --git a/pkg/sink/cloudstorage/path.go b/pkg/sink/cloudstorage/path.go index 6939229c97..46be61a0b2 100644 --- a/pkg/sink/cloudstorage/path.go +++ b/pkg/sink/cloudstorage/path.go @@ -348,11 +348,11 @@ func (f *FilePathGenerator) GenerateDateStr() string { currTime := f.pdClock.CurrentTime() // Note: `dateStr` is formatted using local TZ. switch f.config.DateSeparator { - case config.DateSeparatorYear.String(): + case config.DateSeparatorYear: dateStr = currTime.Format("2006") - case config.DateSeparatorMonth.String(): + case config.DateSeparatorMonth: dateStr = currTime.Format("2006-01") - case config.DateSeparatorDay.String(): + case config.DateSeparatorDay: dateStr = currTime.Format("2006-01-02") default: } @@ -537,9 +537,15 @@ func RemoveExpiredFiles( storage storeapi.Storage, cfg *Config, checkpointTs uint64, +<<<<<<< HEAD:pkg/sink/cloudstorage/path.go ) (uint64, error) { if cfg.DateSeparator != config.DateSeparatorDay.String() { return 0, nil +======= +) error { + if cfg.DateSeparator != config.DateSeparatorDay { + return nil +>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078)):pkg/cloudstorage/generator.go } if dateSeparatorDayRegexp == nil { dateSeparatorDayRegexp = regexp.MustCompile(config.DateSeparatorDay.GetPattern()) diff --git a/pkg/sink/cloudstorage/path_test.go b/pkg/sink/cloudstorage/path_test.go index 547969eb4d..613ff7a9cf 100644 --- a/pkg/sink/cloudstorage/path_test.go +++ b/pkg/sink/cloudstorage/path_test.go @@ -56,7 +56,7 @@ func testFilePathGenerator(ctx context.Context, t *testing.T, dir string) *FileP sinkURI, err := url.Parse(uri) require.NoError(t, err) replicaConfig := config.GetDefaultReplicaConfig() - replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String()) + replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone) replicaConfig.Sink.Protocol = util.AddressOf(config.ProtocolOpen.String()) replicaConfig.Sink.FileIndexWidth = util.AddressOf(6) cfg := NewConfig() @@ -95,8 +95,12 @@ func TestGenerateDataFilePath(t *testing.T) { // date-separator: year mockClock := clock.NewMock() f = testFilePathGenerator(ctx, t, dir) +<<<<<<< HEAD:pkg/sink/cloudstorage/path_test.go f.versionMap[table] = table.TableInfoVersion f.config.DateSeparator = config.DateSeparatorYear.String() +======= + f.config.DateSeparator = config.DateSeparatorYear +>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078)):pkg/cloudstorage/path_test.go f.SetClock(pdutil.NewMonotonicClock(mockClock)) mockClock.Set(time.Date(2022, 12, 31, 23, 59, 59, 0, time.UTC)) date = f.GenerateDateStr() @@ -113,8 +117,12 @@ func TestGenerateDataFilePath(t *testing.T) { // date-separator: month mockClock = clock.NewMock() f = testFilePathGenerator(ctx, t, dir) +<<<<<<< HEAD:pkg/sink/cloudstorage/path_test.go f.versionMap[table] = table.TableInfoVersion f.config.DateSeparator = config.DateSeparatorMonth.String() +======= + f.config.DateSeparator = config.DateSeparatorMonth +>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078)):pkg/cloudstorage/path_test.go f.SetClock(pdutil.NewMonotonicClock(mockClock)) mockClock.Set(time.Date(2022, 12, 31, 23, 59, 59, 0, time.UTC)) @@ -132,8 +140,12 @@ func TestGenerateDataFilePath(t *testing.T) { // date-separator: day mockClock = clock.NewMock() f = testFilePathGenerator(ctx, t, dir) +<<<<<<< HEAD:pkg/sink/cloudstorage/path_test.go f.versionMap[table] = table.TableInfoVersion f.config.DateSeparator = config.DateSeparatorDay.String() +======= + f.config.DateSeparator = config.DateSeparatorDay +>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078)):pkg/cloudstorage/path_test.go f.SetClock(pdutil.NewMonotonicClock(mockClock)) mockClock.Set(time.Date(2022, 12, 31, 23, 59, 59, 0, time.UTC)) @@ -179,8 +191,119 @@ func TestGenerateDataFilePathWithTableIDAsPath(t *testing.T) { func TestFetchIndexFromFileName(t *testing.T) { t.Parallel() +<<<<<<< HEAD:pkg/sink/cloudstorage/path_test.go ctx, cancel := context.WithCancel(context.TODO()) defer cancel() +======= + ctx := t.Context() + + testCases := []struct { + name string + dateSeparator config.DateSeparator + date string + useTableIDAsPath bool + enablePartition bool + enableTableAcross bool + table VersionedTableName + expectedDMLPathKey DMLPathKey + }{ + { + name: "schema table with date", + dateSeparator: config.DateSeparatorDay, + date: "2023-05-09", + enableTableAcross: true, + table: VersionedTableName{ + TableNameWithPhysicTableID: commonType.TableName{ + Schema: "test", + Table: "table1", + }, + TableInfoVersion: 5, + DispatcherID: commonType.NewDispatcherID(), + }, + expectedDMLPathKey: DMLPathKey{ + SchemaPathKey: SchemaPathKey{ + Schema: "test", + Table: "table1", + TableVersion: 5, + }, + Date: "2023-05-09", + }, + }, + { + name: "table id path", + dateSeparator: config.DateSeparatorNone, + useTableIDAsPath: true, + table: VersionedTableName{ + TableNameWithPhysicTableID: commonType.TableName{ + Schema: "test", + Table: "table1", + TableID: 12345, + }, + TableInfoVersion: 5, + DispatcherID: commonType.NewDispatcherID(), + }, + expectedDMLPathKey: DMLPathKey{ + SchemaPathKey: SchemaPathKey{ + Schema: "12345", + TableVersion: 5, + }, + UseTableIDAsPath: true, + TableID: 12345, + }, + }, + { + name: "partition with date", + dateSeparator: config.DateSeparatorDay, + date: "2023-05-09", + enablePartition: true, + table: VersionedTableName{ + TableNameWithPhysicTableID: commonType.TableName{ + Schema: "test", + Table: "table1", + TableID: 55, + IsPartition: true, + }, + TableInfoVersion: 5, + DispatcherID: commonType.NewDispatcherID(), + }, + expectedDMLPathKey: DMLPathKey{ + SchemaPathKey: SchemaPathKey{ + Schema: "test", + Table: "table1", + TableVersion: 5, + }, + PartitionNum: 55, + Date: "2023-05-09", + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + f := testFilePathGenerator(ctx, t, t.TempDir()) + f.config.DateSeparator = tc.dateSeparator + f.config.UseTableIDAsPath = tc.useTableIDAsPath + f.config.EnablePartitionSeparator = tc.enablePartition + f.config.EnableTableAcrossNodes = tc.enableTableAcross + + indexPath := f.GenerateIndexFilePath(tc.table, tc.date) + indexKey := FileIndexKey{ + DispatcherID: tc.table.DispatcherID.String(), + EnableTableAcrossNodes: tc.enableTableAcross, + } + require.Equal(t, indexPath, tc.expectedDMLPathKey.GenerateIndexFilePath(indexKey)) + var pathKey DMLPathKey + require.NoError(t, pathKey.ParseIndexFilePath(tc.dateSeparator, indexPath)) + require.Equal(t, tc.expectedDMLPathKey, pathKey) + }) + } +} + +func TestParseFileIndexFromFileName(t *testing.T) { + t.Parallel() + + ctx := t.Context() +>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078)):pkg/cloudstorage/path_test.go dir := t.TempDir() f := testFilePathGenerator(ctx, t, dir) @@ -233,7 +356,7 @@ func TestGenerateDataFilePathWithIndexFile(t *testing.T) { dir := t.TempDir() f := testFilePathGenerator(ctx, t, dir) mockClock := clock.NewMock() - f.config.DateSeparator = config.DateSeparatorDay.String() + f.config.DateSeparator = config.DateSeparatorDay f.SetClock(pdutil.NewMonotonicClock(mockClock)) mockClock.Set(time.Date(2023, 3, 9, 23, 59, 59, 0, time.UTC)) @@ -466,7 +589,7 @@ func TestRemoveExpiredFilesWithoutPartition(t *testing.T) { sinkURI, err := url.Parse(uri) require.NoError(t, err) replicaConfig := config.GetDefaultReplicaConfig() - replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorDay.String()) + replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorDay) replicaConfig.Sink.Protocol = util.AddressOf(config.ProtocolCsv.String()) replicaConfig.Sink.FileIndexWidth = util.AddressOf(6) replicaConfig.Sink.CloudStorageConfig = &config.CloudStorageConfig{