Method cloneTables

flow/workflows/snapshot_flow.go:231–292

(
	ctx workflow.Context,
	snapshotType snapshotType,
	slotName string,
	snapshotName string,
	stagingPathOverride string,
	maxParallelClones int,
)

}

func (s *SnapshotFlowExecution) cloneTables(
	ctx workflow.Context,
	snapshotType snapshotType,
	slotName string,
	snapshotName string,
	stagingPathOverride string,
	maxParallelClones int,
) error {
	if snapshotType == SNAPSHOT_TYPE_SLOT {
		s.logger.Info("cloning tables for slot", slog.String("slot", slotName), slog.String("snapshot", snapshotName))
	} else if snapshotType == SNAPSHOT_TYPE_TX {
		s.logger.Info("cloning tables in tx snapshot mode", slog.String("snapshot", snapshotName))
	}

	getParallelLoadKeyForTablesCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
		StartToCloseTimeout: 10 * time.Minute,
		RetryPolicy: &temporal.RetryPolicy{
			InitialInterval: 1 * time.Minute,
		},
	})

	var res *protos.GetDefaultPartitionKeyForTablesOutput
	if err := workflow.ExecuteActivity(getParallelLoadKeyForTablesCtx,
		snapshot.GetDefaultPartitionKeyForTables, s.config).Get(ctx, &res); err != nil {
		return fmt.Errorf("failed to get default partition keys for tables: %w", err)
	}

	boundSelector := shared.NewBoundSelector(ctx, "CloneTablesSelector", maxParallelClones)

	sourcePeerType, err := getPeerType(ctx, s.config.SourceName)
	if err != nil {
		return err
	}
	destinationPeerType, err := getPeerType(ctx, s.config.DestinationName)
	if err != nil {
		return err
	}

	for _, v := range s.config.TableMappings {
		source := v.SourceTableIdentifier
		destination := v.DestinationTableIdentifier
		s.logger.Info(
			fmt.Sprintf("Cloning table with source table %s and destination table name %s", source, destination),
			slog.String("snapshotName", snapshotName),
		)
		if v.PartitionKey == "" {
			v.PartitionKey = res.TableDefaultPartitionKeyMapping[source]
		}
		if err := s.cloneTable(ctx, boundSelector, snapshotName, stagingPathOverride, v, sourcePeerType, destinationPeerType); err != nil {
			s.logger.Error("failed to start clone child workflow", slog.Any("error", err))
			return err
		}
	}

	if err := boundSelector.Wait(ctx); err != nil {
		s.logger.Error("failed to clone some tables", slog.Any("error", err))
		return err
	}