Add savepoints for parquet migration

[pizzaeueu]

No description provided.

[Copilot]

Pull Request Overview

This PR adds comprehensive savepoints support for Parquet-to-Scylla migrations, enabling resumable data migrations by tracking processed Parquet files at a granular level. The implementation handles both single-partition and multi-partition files, supports parallel processing, and provides detailed progress tracking through Spark listeners.

Key changes:

• Implemented file-level savepoints tracking with partition awareness to support resumable Parquet migrations
• Refactored ScyllaMigrator into a trait-based architecture for extensibility with specialized Parquet migration logic
• Added FileCompletionListener to track Spark task completion and aggregate partition-level events into file-level completion

Reviewed Changes

Copilot reviewed 29 out of 29 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
ParquetToScyllaBasicMigrationTest.scala	Refactored to use new ParquetMigratorSuite base class with fixture-based resource management
ParquetSavepointsTest.scala	Unit tests for Parquet file listing, filtering, and savepoints manager state tracking
ParquetSavepointsIntegrationTest.scala	Integration test verifying savepoints include all processed files after migration
ParquetResumeIntegrationTest.scala	Tests for migration resumption after interruption and idempotency with pre-processed files
ParquetParallelModeTest.scala	Tests parallel Parquet file processing with savepoints in both simple and concurrent scenarios
ParquetMultiPartitionTest.scala	Tests handling of large Parquet files split across multiple Spark partitions
ParquetMigratorSuite.scala	New base test suite providing fixtures for tables, Parquet directories, and savepoints
PartitionMetadataReaderTest.scala	Tests for reading Spark partition metadata and building file-to-partition mappings
FileCompletionListenerTest.scala	Tests for Spark listener tracking partition completion and file-level aggregation
ParquetConfigSerializationTest.scala	Tests YAML serialization/deserialization of skipParquetFiles configuration
StringSetAccumulatorTest.scala	Tests thread-safe string set accumulator for tracking processed files
parquet-to-scylla-*.yaml	Configuration files for various test scenarios including savepoints, resume, and parallel modes
.gitignore	Updated to track spark-master directory for savepoints storage
ScyllaMigrator.scala	Refactored into trait-based architecture with specialized ScyllaParquetMigrator implementation
PartitionMetadataReader.scala	Reads Spark partition metadata to build file-partition mappings for tracking
ParquetSavepointsManager.scala	Manages Parquet savepoints state using string set accumulator for processed files
Parquet.scala	Refactored to implement parallel file processing with savepoints and file completion tracking
FileCompletionListener.scala	Spark listener aggregating partition completions into file-level tracking
MigratorConfig.scala	Added skipParquetFiles field and helper method for configuration
StringSetAccumulator.scala	Thread-safe accumulator for collecting processed file paths across Spark tasks
Migrator.scala	Updated Parquet-to-Scylla migration path to use new migrateToScylla method

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[tarzanek]

why false when you support savepoints now?

[tarzanek]

I think this needs a better look,
since basically savepoint manager needs to dump his state to savepoint file periodically (and in case of failure of course)
and I am wondering if this periodic write to savepoint file will happen if this is false ?!?!?

[tarzanek]

(or the manager is self contained and will dump savepoints automatically once it's instantiated?)

[pizzaeueu]

why false when you support savepoints now?

Basically, this false flag will not create savepoint manager for basic ScyllaMigrator, and will not be used at all in ParquetMigrator at all
it's necessary to not create additional SavepointManager since we have external one created in Parquet object

Actually, I see current architecture / naming is confusing.

(or the manager is self contained and will dump savepoints automatically once it's instantiated?)

Yeah, it is 😌

[tarzanek]

below has nothing to do with parquet, but OK, I will just close one eye, but PR clean wise below just confuses people, so non related changes should be in separate PRs/commits

[pizzaeueu]

Yes, thank you
Looks like the leftover after the approach with both parallel / sequential migrations

[tarzanek]

why do we need this?

I mean manager will be closed anyways and should be closed anyways
that's why it periodically dumps its state and of course it should close cleanly when any failure happens

[tarzanek]

what will happen with parquet manager ?
who will close it then? is this a leak? :-D ?

[pizzaeueu]

Yeah, probably the problem here is over-unification of different approaches for savepoints:

• Common ScyllaMigrator creates CQLSavepointsManager on its own
• Parquet File Migrator uses external SavepointsManager created in Parquet.scala ( because it should be used inside our custom Spark listener )

The idea is:

• Parquet object is responsible to open SavepointsManager => same object should be responsible to close it ( and it's done safely with Using statement )

However, it makes sense to reconsider an approach to avoid unnecessary mess 🤔

[tarzanek]

please check my comments

[Copilot]

Pull request overview

Copilot reviewed 29 out of 29 changed files in this pull request and generated 1 comment.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[tarzanek]

I have no other comment from my end, so let's merge

[tarzanek]

looks ok, we still have a real life test in flight for this, so let's merge if no problems will be seen

[tarzanek]

shuffle files before passing them over?

[tarzanek]

or is spark shuffling this by itself?

[pizzaeueu]

Spark handles partition sizing on its own using the spark.sql.files.maxPartitionBytes, so

• small files will be combined into single partition
• big files will be split between several partitions

However, I can add scala.util.Random.shuffle to be more safe, overhead is negligible

[tarzanek]

how does this behave when you have lots of 100-200MB files and few files that are around 5G (how is splitting)

[tarzanek]

before merging we are checking the logic of partition to file mapping - per tests with a case with lots of small (~200MB) and few big (~5G) files (with large partitions!) we saw delays and problems - if the mapping isn't correct, job might not finish

[tarzanek]

alternator test might be impacted due to 2025.4 released https://forum.scylladb.com/t/release-scylladb-2025-4-0/5222

[tarzanek]

Caused by: software.amazon.awssdk.services.dynamodb.model.DynamoDbException: GlobalSecondaryIndexes and LocalSecondaryIndexes with tablets require the rf_rack_valid_keyspaces option to be enabled. (Service: DynamoDb, Status Code: 400, Request ID: null)

• so we need to pass this rf_rack_valid_keyspaces option for scylladb creation to scylla.yaml to true
  OR in other case we need to disable tablets
  on line https://github.com/scylladb/scylla-migrator/blob/master/docker-compose-tests.yml#L39
  either enable_tablets to false

or better do tablets_mode_for_new_keyspaces set to disabled on the cmd line for scylla start in docker
(until scylladb/spark-scylladb-connector#19 is merged we don't support tablets properly anyways)

[tarzanek]

Note: https://spark.apache.org/docs/latest/sql-performance-tuning.html#adaptive-query-execution
might need to be disabled for supporting parquet resume, is spark.sql.adaptive.enabled setting to false needed?

[Copilot]

Pull request overview

Copilot reviewed 40 out of 41 changed files in this pull request and generated 7 comments.

Comments suppressed due to low confidence (1)

migrator/src/main/scala/com/scylladb/migrator/scylla/ScyllaMigrator.scala:95

• The error handling in the catch block swallows the exception after logging it. This means that even if the migration fails with an error, the method will not propagate the failure to the caller. The application will appear to complete successfully despite data not being fully migrated. The exception should be re-thrown after dumping the savepoint state to ensure proper error propagation and to prevent silent failures.

    } catch {
      case NonFatal(e) => // Catching everything on purpose to try and dump the accumulator state
        log.error(
          "Caught error while writing the DataFrame. Will create a savepoint before exiting",
          e)

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Race condition in concurrent file completion checking. Between the check at line 58 (if (completedFiles.contains(filename))) and the putIfAbsent at line 67, multiple threads could pass the first check and all attempt to mark the file as complete. While putIfAbsent prevents duplicate entries in the map, this could result in savepointsManager.markFileAsProcessed(filename) being called multiple times for the same file. Consider moving the completion check inside a synchronized block or using completedFiles.putIfAbsent earlier to gate the entire completion logic.

[Copilot]

Resource management concern: The ParquetSavepointsManager is closed by the Using.resource block, but it's passed to ScyllaParquetMigrator.migrate which internally marks it with shouldCloseManager = false to prevent double-closing. However, this creates a brittle design where the lifecycle management is split between two places. If an exception occurs during migration (line 80), the savepointsManager.dumpMigrationState("completed") at line 82 will never be called, and only the "final" dump in ScyllaMigrator will execute. Consider consolidating the resource management logic or adding explicit error handling to ensure proper state dumping in all scenarios.

[Copilot]

Potential inconsistency in DataFrame creation. When skipFiles.isEmpty is true (line 52), the code reads from source.path which may include newly added files that weren't in the original allFiles list. However, when skipFiles is not empty, it reads from the specific filesToProcess list. This could lead to different behavior: in the first case, files added after the listParquetFiles call would be processed, while in the second case they would not. Consider always using the explicit file list for consistency.

Suggested change

	val df = if (skipFiles.isEmpty) {
	spark.read.parquet(source.path)
	} else {
	spark.read.parquet(filesToProcess: _*)
	}
	val df = spark.read.parquet(filesToProcess: _*)

[Copilot]

Missing documentation for the new enableParquetFileTracking parameter. The Savepoints case class should include Scaladoc comments explaining what this parameter does, when it should be set to true vs false, and its default value. This is especially important since it controls a significant behavior change in how Parquet migrations work.

Suggested change

	/**
	* Configuration for periodic savepoints written during a migration run.
	*
	* @param intervalSeconds
	* How often, in seconds, a savepoint should be written.
	* @param path
	* Filesystem path (directory or prefix) where savepoint data will be stored.
	* @param enableParquetFileTracking
	* When `true` (the default), enables tracking of already-processed Parquet files
	* as part of the savepoint state. This prevents the same Parquet file from
	* being migrated more than once if the job is restarted or savepoints are
	* resumed.
	*
	* Set this to `false` to keep the legacy behavior where Parquet files are not
	* tracked in savepoints. Disabling tracking may be useful for backwards
	* compatibility with older savepoints or when file tracking is handled by an
	* external mechanism, but it means repeated runs may reprocess the same
	* Parquet files.
	*/

[Copilot]

Missing documentation for the new skipParquetFiles configuration field. This field should have Scaladoc comments explaining its purpose, format (Set of file URIs), and how it's used in the Parquet migration process. Without documentation, users won't understand what format the file paths should be in or when this field is populated.

[Copilot]

Inconsistent indentation detected. Lines 31-32 and 35-36 use tabs instead of spaces. The codebase should use consistent spacing (2 spaces based on other files in the project). Please replace tabs with spaces to maintain consistency.

Suggested change

	val message = "Several FileSourceScanExec were found in plan"
	logger.error(s"$message. Plan: ${plan.treeString}")
	throw new IllegalArgumentException(message)
	case list if list.isEmpty =>
	val message = "DataFrame is not based on file source (FileSourceScanExec not found in plan)"
	logger.error(s"$message. Plan: ${plan.treeString}")
	throw new IllegalArgumentException(message)
	val message = "Several FileSourceScanExec were found in plan"
	logger.error(s"$message. Plan: ${plan.treeString}")
	throw new IllegalArgumentException(message)
	case list if list.isEmpty =>
	val message = "DataFrame is not based on file source (FileSourceScanExec not found in plan)"
	logger.error(s"$message. Plan: ${plan.treeString}")
	throw new IllegalArgumentException(message)

[Copilot]

The StringSetAccumulator class is placed in the alternator package, but it's a general-purpose utility used for Parquet migration tracking. This creates an incorrect package coupling. Consider moving this class to a more appropriate location such as com.scylladb.migrator.util or com.scylladb.migrator.accumulator to better reflect its purpose and avoid confusion.