CASSANALYTICS-42 Add S3-backed Cassandra batch reader - #206
Conversation
jberragan
left a comment
jberragan
left a comment
There was a problem hiding this comment.
Thanks for the patch, this is a great addition! I will take a closer look, but a few high level questions:
- Did you consider putting this in a separate module (e.g.
cassandra-analytics-s3). The Sidecar DataLayer should never had been put incassandra-analytics-coreand should eventually be moved to its own module. - I suppose every user might have a slightly different backup path format, is this made pluggable by the
BackupReaderinterface?
wasn't aware that we wanted the sidecar data layer to live outside of core. happy to discuss the appropriate modularization. A s3 specific one could work, or we can have a batch read specific module? the s3 implementation inherits a good amount of work from the sidecar one.
yes, i didn't include the internal concrete implementation of this interface on this PR because it is tied to a vendor we use. But the goal is to make it pluggable per needs of different organizations, using the |
| */ | ||
| public boolean isMutableMetadata() | ||
| { | ||
| return this == SUMMARY || this == FILTER || this == STATISTICS; |
There was a problem hiding this comment.
Is this code/comment fully accurate? I'm aware Statistics.db can mutate the repair metadata (repairedAt, pendingRepair) to avoid recompaction, and also the compaction level to avoid a full rewrite.
Summary.db appears to only mutate if there is a change in the index sample size (redistributeSummaries).
Filter.db I can't find any place where it mutates. I think it is immutable from the first flush.
There was a problem hiding this comment.
Good catch, i think filter.db is theoretically immutable but there is a code path that re-writes it:
https://github.com/apache/cassandra/blob/45fa31b/src/java/org/apache/cassandra/io/sstable/format/SSTableReaderBuilder.java#L450
It should be very rare, and if I read the code correctly, it happens during live re-opens of an SSTable (startup, import, streaming receipt), and only when its Filter.db is missing or it has no ValidationMetadata.
So it is theoretically possible but most likely can be treated immutable.
Introduce the public backup-reader registration contract and S3 client configuration primitives so S3-backed readers can be wired without a bundled backup implementation.
Implement the S3-backed data layer and SSTable token-index builder behind the backup-reader contract, including Summary.db reader support needed for remote SSTables.
Wire the S3 data layer into the Spark DataSource path with scan statistics, custom task metrics, and prebuilt token-index contexts for efficient planning.
Add focused coverage for S3 config validation, client caching, token-index construction, prebuilt contexts, and Summary.db reader edge cases.
VersionRunner is the parameterized-test base used to iterate Cassandra version bridges. Until now it lived under cassandra-analytics-core's test sources, so external test consumers loading it via the existing testImplementation(testFixtures(project(...))) wiring couldn't see it. This commit relocates it to src/testFixtures and exposes the test fixtures artifact so any downstream test module can depend on it through the standard Gradle testFixtures mechanism. Existing in-module tests continue to resolve VersionRunner because Gradle automatically adds the testFixtures output to the local test compile classpath. - Enable the java-test-fixtures plugin in cassandra-analytics-core. - Declare testFixturesApi dependencies for cassandra-bridge and JUnit Jupiter API so consumers compile against fixtures without redeclaring transitive dependencies. - Move VersionRunner from src/test/java to src/testFixtures/java with no source changes.
Concurrent Spark tasks share a canonical BackupReader instance via ReaderInternCache so that S3 client and manifest state aren't re-allocated per task. With the prior shape, each task installed its own Stats sink through BackupReader.setStats(), which silently raced with sibling tasks on the same executor and attributed S3 GET/HEAD durations, mutable-metadata drift counts, and similar metrics to whichever task wrote setStats() last. Move from per-reader mutable state to per-call argument passing so the correct task-scoped Stats receives every S3 operation measurement. Interface changes (BackupReader): - Drop setStats(Stats); the interface no longer carries mutable per-task state. - Add a trailing Stats parameter to readAsync, readMutableMetadataAsync, getAsync, getMutableMetadataAsync, and exists. Implementations route S3 metrics through the supplied sink rather than a captured field. Config changes (BackupReaderConfig): - Drop the transient stats field and the withStats(Stats) helper. Stats is now exclusively supplied on individual read calls, which removes a sharp edge around closure capture and (de)serialization. - Bump serialVersionUID to reflect the shape change. Wiring changes: - S3CassandraDataLayer passes the task's SparkCustomMetricsStats (context.stats) into every BackupReader call site, including the ranged GET, streaming GET, mutable-metadata variants, and exists(). - The readObject path no longer reapplies setStats on the interned reader; it just recreates the executor-local Spark metrics sink. - SSTableTokenIndexBuilder, whose prebuild path does not flow metrics back to Spark, supplies Stats.DoNothingStats.INSTANCE explicitly. Test changes: - FakeBackupReader is updated to match the new signatures and no longer stores a Stats field.
SSTableTokenBounds.overlaps() previously normalized firstToken > lastToken by swapping the endpoints. For an SSTable whose covered range wraps the ring (i.e. crosses the Murmur3 zero boundary between Long.MAX_VALUE and Long.MIN_VALUE), the swap silently converted the actual covered band [first, MAX] U [MIN, last] into its complement (last, first). The data layer uses overlaps() to filter SSTables in listInstance(), so a query range falling inside the true covered band could be reported as non-overlapping and the SSTable would be skipped, silently dropping rows. This is the same wrap-around convention used elsewhere in the project (e.g. RangeUtils.calculateTokenRanges), where firstToken > lastToken means the range crosses the boundary rather than being inverted. Changes: - SSTableTokenBounds.overlaps(): when firstToken > lastToken, model the bounds as the two segments [first, MAX] and [MIN, last] and report overlap if the query range hits either segment. Non-wrap bounds keep the existing single-segment isConnected() check. - Constrain the new helper constants to the Murmur3 token domain (Long.MIN_VALUE / Long.MAX_VALUE) and document the assumption; the reader path is Murmur3-only today. - SSTableTokenIndexBuilder.toLong(): replace longValue() with longValueExact() so a non-Murmur3 token that ever reaches this path fails loudly instead of silently truncating to the low 64 bits. Tests: - SSTableTokenIndexTest: rename the inverted-bounds test to invertedBoundsModelWrapAround and assert the actual wrap semantics (queries inside either segment overlap, queries in the gap do not, endpoints are inclusive). - Add boundaryAndSingletonBoundsOverlap covering point ranges, shared endpoints, and adjacent-but-disjoint ranges on the well-formed path. - Add extremeTokenBoundsCoverRing exercising Long.MIN_VALUE / Long.MAX_VALUE for both the full-ring and the degenerate MAX..MIN inverted form.
Allow backup readers to provide rack-aware per-range placement when building S3 Cassandra rings, while preserving the existing rack-unaware fallback for readers that still derive placement from token order. Persist the authoritative replica map across JDK and Kryo serialization so driver and executor views rebuild identical range maps.
Mirror the upstream JUnit-to-AssertJ migration (92a9dbf) on the five OSS test files added on this branch that still used JUnit assertion helpers: - S3DataSourceClientConfigBufferTest - S3SSTableLeakTests - SSTableTokenIndexTest - BackupReaderFactorySerializationTest - S3CassandraPrebuiltReadContextTest No behavioural change; assertion semantics are preserved including message wording and exception-type checks. Imports trimmed to the AssertJ entry points and the mockito helpers actually used.
Add a generic primary-hint tie-breaker to replica selection with a no-op default so existing callers, including Sidecar, preserve their behavior. The S3 data layer uses the hook to prefer the token-end owner when availability is otherwise equal, making LOCAL_QUORUM reads less sensitive to set iteration order.
…n CassandraScanBuilder as a standard
The S3 batch reader (lineage apache#206) previously existed only under src/main/spark3, so a Spark 4 build compiled none of it and the shared S3CassandraDataLayer failed to resolve SparkCustomMetricsStats. Bring the reader to src/main/spark4: - Add Spark 4 copies of the version-agnostic classes: S3CassandraDataSource, S3CassandraPrebuiltReadContext(Registry), S3CassandraTokenIndexPrebuilder, CassandraSourceStatistics, SparkCustomMetricsStats, and the 16 sparksql/metrics custom-metric classes. The DataSource V2 connector APIs they use (SessionConfigSupport, NamedReference, Statistics, ColumnStatistics, CustomMetric/CustomSumMetric, CustomTaskMetric, Broadcast) are unchanged between Spark 3.5 and 4.0, so these are straight copies. - Re-apply the S3 integration onto the pristine Spark 4 copies of the shared read-path classes (CassandraScanBuilder, CassandraTable, CassandraTableProvider, CassandraPartitionReaderFactory, SparkRowIterator): token-index broadcast wiring, statistics reporting, and custom task/sum metrics. CassandraTable keeps the Spark 4 capabilities() form (Sets.newHashSet) rather than the spark3 variant. The DataSourceRegister service descriptor lives in src/main/resources and already lists S3CassandraDataSource, so it covers both profiles. Also trim two code-narrating comments (a stale "Spark 3" class note and a "called by Spark" javadoc) in both source trees to satisfy the comment guidelines.
Spark 3.5 stopped pulling commons-lang3 in transitively when it excluded commons-lang3 from commons-compress and avro, so modules that import lang3 APIs need an explicit version. A single global commonsLang3Version pin matched Spark 3.5, but it made Spark 4 profile builds compile and test against an older commons-lang3 than Spark 4 supplies at runtime. Move the pin into the Spark build profiles, mirroring the existing profile-scoped Jackson pins. Spark 3 profiles keep 3.12.0 and the Spark 4 profile uses 3.17.0. The existing consumers continue to read project.commonsLang3Version from the applied profile.
S3ClientCache opted the AWS SDK Netty async client into a non-blocking DNS resolver, gated on a hardcoded probe for the unrelocated class io.netty.resolver.dns.DnsAddressResolverGroup. On the Spark 4 / Core ETL driver bundle this probe is misleading: Spark and Hadoop put the unrelocated class on the classpath, so the probe returns true, but the active AWS SDK comes from iceberg-aws-bundle, whose NettyNioAsyncHttpClient -> AwaitCloseChannelPoolMap -> BootstrapProvider calls DnsResolverLoader, which is relocated to org.apache.iceberg.aws.shaded.io.netty.resolver.dns -- a class the bundle omits. Enabling non-blocking DNS then fails when the first connection pool/bootstrap is created. Remove the setting entirely and always use the SDK default (blocking) resolver. The benefit is negligible for this workload: the S3 backup reader goes through a single (or few) bucket host(s) per job with long-lived cached clients, keep-alive, and connection pooling, so DNS is resolved at connection creation and then cached by the JVM/Netty -- steady-state lookups are rare. Dropping it removes the relocation/ missing-class fragility for no measurable loss and shrinks the surface area to test across the OSS, aggregate, and prod-parity packaging shapes. This deletes the class-name constant, the availability field and probe helpers, the conditional builder branch, and the dns=nonBlocking cache-key suffix, and removes the now-defunct test plus its classpath helper. Netty relocation in the driver bundles remains correct and is retained. Revisit only if a future workload reads across many distinct S3 hosts or runs where DNS is slow.
b03fc5c to
ee61151
Compare
| } | ||
| catch (IOException | RuntimeException exception) | ||
| { | ||
| indexSummary.close(); |
There was a problem hiding this comment.
This is a separate known issue, I think it is already being fixed in #138.
Please don't call indexSummary.close(), iirc it resulted in in seg. faults (SIGSEGV) due to the way Casandra references tracks.
| * <p> | ||
| * Key design decisions: | ||
| * <ul> | ||
| * <li>No JVM shutdown hooks - problematic in Spark executors. Use explicit {@link #closeAll()}</li> |
| } | ||
| } | ||
|
|
||
| // Close async client |
There was a problem hiding this comment.
Is it a problem if two threads call close at the same time? One thread can skip ahead and close asyncClient while S3TransferManager is still shutting down.
| * (minimum) autosnap epoch across all nodes. This ensures conservative TTL behavior: a cell | ||
| * is only expired if it was already expired at every node's snapshot time. | ||
| */ | ||
| public class S3SnapshotTimeProvider implements TimeProvider |
There was a problem hiding this comment.
nit: there's very little difference vs ReaderTimeProvider
| * @param stats per-task stats sink | ||
| * @return {@code true} iff the component exists in the backup | ||
| */ | ||
| boolean exists(String clusterName, |
There was a problem hiding this comment.
Shouldn't this be a CompletableFuture<Boolean>? Checking for existence in S3 would still be blocking, right?
| * @return OptionalLong containing the total size in bytes of all SSTable Data.db files, | ||
| * or empty if size information is not available | ||
| */ | ||
| public OptionalLong calculateTotalSSTableSize() |
There was a problem hiding this comment.
There's an existing interface TableSizeProvider to help with dynamically sizing the spark job, maybe you want to re-use here somehow?
| } | ||
|
|
||
| @Override | ||
| public CustomMetric[] supportedCustomMetrics() |
|
Nice work, interfaces are genuinely pluggable with vendor specific implementations and storage neutral. |
| */ | ||
| public BackupReaderConfig toBackupReaderConfig() | ||
| { | ||
| return BackupReaderConfig.of(s3Config); |
There was a problem hiding this comment.
custom properties are lost here, not sent to BackupReaderConfig
| * | ||
| * @return the S3 client configuration this reader was constructed with | ||
| */ | ||
| S3ClientConfig s3Config(); |
There was a problem hiding this comment.
Organizations may use other types of Object stores as well. Instead of using S3 specific naming, would be helpful if we can use generic ObjectStore* namings.
|
|
||
| private static AwsCredentialsProvider getCredentialsProvider(S3ClientConfig config) | ||
| { | ||
| String accessKeyId = config.s3AccessKeyId(); |
There was a problem hiding this comment.
Would be helpful to have pluggable credentials interfaces, like S3CredentialsProviderFactory, S3CredentialsProviderRegistry etc
| * @param stats per-task stats sink for S3 operation metrics | ||
| * @return future completing with the read bytes | ||
| */ | ||
| CompletableFuture<byte[]> readAsync(String clusterName, |
There was a problem hiding this comment.
There is no sharable Decrypter interface. It would be helpful to have a BackupDecryptor interface so organizations can implement BackupDecryptor only to tune policies while sharing other parts of the logic.
readAsync ─► decryptor.decryptRange(...)
getAsync ─► decryptor.decryptRange(...)
There was a problem hiding this comment.
Also, would be good to have KeyService interface with byte[] unwrapDek(String keyId, byte[] wrappedDek) method.
| { | ||
| // Mutable metadata (Summary.db, Filter.db, Statistics.db) supports | ||
| // size-drift handling for stale autosnap manifests. | ||
| if (fileType.isMutableMetadata()) |
There was a problem hiding this comment.
Under client-side encryption ciphertext length ≠ plaintext length and cannot be inferred from the object size. So a client-side-encrypted Data.db/Index.db whose manifest size is the ciphertext size has no interface to resolve the real plaintext length
It would be helpful if you add a generic, per-component length hook to BackupReader and consult+cache it for all components, not just mutable ones.
| * This class contains only the essential settings needed to create an S3 client. | ||
| * It is shared across batch and streaming configurations. | ||
| */ | ||
| public class S3ClientConfig implements Serializable |
There was a problem hiding this comment.
S3ClientConfig carries no encryption identity, and buildS3Client/buildS3AsyncClient hardcode S3Client.builder() / S3AsyncClient.builder() with no injection point. So an org that wants an S3 Encryption Client cannot supply one.
getCacheKey hashes only region|endpoint|accessKeyId|secretHash — it has no encryption identity — so a plain client and an encrypting client built for the same bucket/credentials would collide on one cache entry: the first one to be created wins and is silently handed to the other caller
| @Override | ||
| public String shortName() | ||
| { | ||
| return "s3CassandraBulkRead"; |
There was a problem hiding this comment.
S3 is in the public API, even though the interfaces (BackupReader/BackupReaderRegistry) are storage-neutral and backupReaderType already selects the backend. Combined with the s3-region / s3-bucket / s3-* option keys, callers of a non-S3 backend still type s3CassandraBulkRead and s3- options.
| { | ||
| s3BackupReader.close(); | ||
| } | ||
| S3ClientCache.closeAll(); |
There was a problem hiding this comment.
A non-S3 reader that owns its own connection pool (e.g., a GCS/Azure client or an Apple Manager-proxy pool) has no hook to release it: the layer only knows how to tear down S3.
A generic call something like below would be helpful
BackupReaderRegistry.factoryFor(backupReaderType).closeSharedResources();
| // placement from the BackupReader; fall back to the naive (rack-unaware) ring when | ||
| // none is available. Exceptions from the reader signal a genuine integrity issue and | ||
| // must surface — see BackupReader#buildRackAwareReplicas for the contract. | ||
| final Partitioner partitioner = Partitioner.Murmur3Partitioner; |
There was a problem hiding this comment.
Murmur3Partitioner is hardcoded here. Can we provide a hook to configure partitioners, something like
Partitioner partitioner = s3BackupReader.partitioner();
@liucao-dd Some of my comments are to make this feature support client encryption use cases or non-s3 object stores etc. I do not expect you to fix all those in this PR. For comments not resolved in this PR, please feel free to create JIRAs if required, so we can implement them later. |
3 participants
Patch information
Jira: https://issues.apache.org/jira/browse/CASSANALYTICS-42
CEP: https://cwiki.apache.org/confluence/display/CASSANDRA/CEP-56%3A+Spark+Bulk+Reading+from+Cassandra+Backup+Uploaded+to+Object+Storage
Summary
Adds an S3-backed Cassandra batch reader so Spark jobs can read SSTables directly from object-storage backups without going through a live cluster. Built on top of the existing
CassandraDataLayer/ bulk-reader abstractions and exposed through Spark SQL via a newS3CassandraDataSource.This PR ships only the generic SPI and the S3 reference implementation. Concrete backup-provider implementations are out of scope and intended to live in downstream repos that plug into the
BackupReaderSPI introduced here.What's included
cassandra-analytics-core/.../spark/data/backup/):BackupReader,BackupReaderConfig,BackupReaderFactory,BackupReaderRegistry— pluggable abstraction so different backup providers can be wired in without changes to core.cassandra-analytics-core):S3CassandraDataLayerwith token-aware partitioning and SSTable selection over S3-resident backups, includingSSTableTokenBoundspruning that correctly handles the Murmur3 wrap-around.S3ClientCache,S3ClientConfig,S3DataSourceClientConfig,S3SizingFactory,S3TableSizeProvider, andS3SnapshotTimeProvider(incassandra-analytics-common).S3CassandraDataSource,S3CassandraPrebuiltReadContext(Registry),S3CassandraTokenIndexPrebuilder, plus refinements toCassandraScanBuilder,CassandraPartitioning,CassandraTable, and a newCassandraSourceStatistics.SparkCustomMetricsStatsplus SparkCustomTaskMetricclasses (TaskTotal*/Total*) for S3 GET/HEAD latency, summary read latency, skipped/corrupt SSTable counts, opened SSTable duration, mutable metadata drift, and head fallback counts. Threaded throughBackupReaderread paths via a per-taskStatsargument.CassandraBridgeImplementationandSummaryDbUtils(both bridges) to thread the per-taskStatsparameter; existing call sites get backwards-compatible overloads.VersionRunneris published viajava-test-fixturesso the bridges can share it in tests.Tests
All new tests live in the modules where the code lives (
cassandra-analytics-core, bridges):S3ClientCacheTest,S3ClientConfigTest,S3DataSourceClientConfigTest,S3DataSourceClientConfigBufferTestS3SSTableLeakTests,SSTableTokenIndexTestBackupReaderFactorySerializationTestplus aFakeBackupReadertest fixture exercising the SPI without a real S3 endpointS3CassandraPrebuiltReadContextTestReaderUtilsTests,SSTableReaderTests,SummaryDbTestsfor the newStats-threaded signaturesThe code is internally powering ~200 spark pipelines in production for ingesting data from s3 backup into data lake, ranging from 10MB to 300TB table size.
Reviewer notes
The diff is large (~9.4k LOC added across ~80 files) because it introduces both the SPI and a complete S3 implementation plus the metrics surface. This is mostly to show the overall idea and actual shipping will likely come with smaller PRs if preferred.
Please also ignore my shadow jar build changes, those are artifacts of our internal build system that we can discuss cleanup later.
Known gap currently deferred