This is an automated email from the ASF dual-hosted git repository.
anishshri-db pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 3bf13f1fecde [SPARK-57269][SS] Enforce read-only access in the
StateDateSource/StateMetadataSource
3bf13f1fecde is described below
commit 3bf13f1fecdebeaba58cee1f2432a1e8c4ce2bb4
Author: liviazhu <[email protected]>
AuthorDate: Fri Jul 24 11:26:47 2026 -0700
[SPARK-57269][SS] Enforce read-only access in the
StateDateSource/StateMetadataSource
### What changes were proposed in this pull request?
Keep the state data source read path from writing to the checkpoint, so
reads work on
read-only storage.
- `StatePartitionReader` / `StatePartitionAllColumnFamiliesReader`: use
`getReadStore` +
`release()` instead of `getStore` + `abort()`.
- `StreamStreamJoinStatePartitionReader`: build
`SymmetricHashJoinStateManager` with
`readOnly = true`. The handler exposes a writable `stateStore` (guarded
by `require(!readOnly)`)
and a mode-aware `readStateStore`; reads use `readStateStore`,
`abortIfNeeded` calls `release()`.
- `HDFSBackedStateStoreProvider`: defer `baseDir` mkdirs to the first
write; guard `release()`;
reject `readOnly = true` in `replayStateFromSnapshot` (use
`replayReadStateFromSnapshot`).
- `ReadStateStore.registerColFamily`: read-safe column-family registration
(in-memory id +
encoders, no forced snapshot), split from the writable
`StateStore.createColFamilyIfAbsent`.
Tests use `WriteProtectedLocalFileSystem` /
`WriteProtectedAbstractFileSystem` (covering the
`FileSystem` and `FileContext` write paths) via
`WriteProtectedCheckpointTestMixin`, which
auto-protects every `withTempDir`; `testStream` / `withWritableCheckpoint`
lift protection for
real writes. Mixed into `StateDataSourceTestBase` and
`OperatorStateMetadataSuite`.
### Why are the changes needed?
Reads previously issued writes (mkdirs, changelog files) to the checkpoint,
failing on read-only
storage.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
`StateDataSource{Read,ChangeDataRead,TransformWithState}Suite`,
`StatePartitionAllColumnFamilies{Reader,Writer}Suite`,
`StateDataSourceNoEmptyDirCreationSuite`,
`OfflineStateRepartitionIntegrationSuite`, `OperatorStateMetadataSuite`,
`RocksDBStateStoreSuite`,
`StateStoreSuite`, `StreamingJoinV4Suite`. Added framework-sanity and
read-only enforcement
tests; verified the framework catches an injected `mkdir` on the read path.
### Was this patch authored or co-authored using generative AI tooling?
Yes, with Claude (Claude Code).
Closes #56332 from
liviazhu/liviazhu-db/SPARK-57225-state-datasource-read-only.
Authored-by: liviazhu <[email protected]>
Signed-off-by: Anish Shrigondekar <[email protected]>
(cherry picked from commit 3c2f9ef1cf40de7d037a3c6776e06272a01c5e1d)
Signed-off-by: Anish Shrigondekar <[email protected]>
---
.../v2/state/StatePartitionReader.scala | 27 +-
.../StreamStreamJoinStatePartitionReader.scala | 7 +-
.../join/SymmetricHashJoinStateManager.scala | 248 +++++++++++----
.../state/HDFSBackedStateStoreProvider.scala | 38 ++-
.../sql/execution/streaming/state/RocksDB.scala | 10 +-
.../state/RocksDBStateStoreProvider.scala | 27 +-
.../sql/execution/streaming/state/StateStore.scala | 32 ++
.../state/StateDataSourceChangeDataReadSuite.scala | 42 ++-
.../v2/state/StateDataSourceReadSuite.scala | 135 ++++++--
.../v2/state/StateDataSourceTestBase.scala | 9 +-
.../StateDataSourceTransformWithStateSuite.scala | 78 ++++-
.../v2/state/WriteProtectedLocalFileSystem.scala | 348 +++++++++++++++++++++
.../OfflineStateRepartitionIntegrationSuite.scala | 6 +-
.../state/OperatorStateMetadataSuite.scala | 60 ++--
...tatePartitionAllColumnFamiliesWriterSuite.scala | 41 +--
.../streaming/state/StateStoreSuite.scala | 25 ++
16 files changed, 965 insertions(+), 168 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StatePartitionReader.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StatePartitionReader.scala
index 2f7d9ab2d8c7..268d03c6b805 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StatePartitionReader.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StatePartitionReader.scala
@@ -97,6 +97,11 @@ class StatePartitionReaderFactory(
/**
* An implementation of [[PartitionReader]] for State data source. This is
used to support
* general read from a state store instance, rather than specific to the
operator.
+ *
+ * NOTE: The state data source is strictly read-only. Any new reader added
here must open the
+ * state store in read-only mode (via [[StateStoreProvider.getReadStore]] or
+ * [[SupportsFineGrainedReplay.replayReadStateFromSnapshot]]) so the read path
never writes to
+ * the checkpoint and does not require write access to it.
*/
abstract class StatePartitionReaderBase(
storeConf: StateStoreConf,
@@ -167,7 +172,9 @@ abstract class StatePartitionReaderBase(
useMultipleValuesPerKey = useMultipleValuesPerKey,
stateSchemaProviderOpt)
if (useColFamilies) {
- val store = provider.getStore(
+ // Register the column family on a read-only store so the read path does
not write
+ // to the checkpoint (column-family registration is an in-memory
operation).
+ val readStore = provider.getReadStore(
partition.sourceOptions.batchId + 1,
getEndStoreUniqueId)
require(stateStoreColFamilySchemaOpt.isDefined)
@@ -176,14 +183,14 @@ abstract class StatePartitionReaderBase(
StateStoreColumnFamilySchemaUtils.isTestingInternalColFamily(
stateStoreColFamilySchema.colFamilyName)
require(stateStoreColFamilySchema.keyStateEncoderSpec.isDefined)
- store.createColFamilyIfAbsent(
+ readStore.registerColFamily(
stateStoreColFamilySchema.colFamilyName,
stateStoreColFamilySchema.keySchema,
stateStoreColFamilySchema.valueSchema,
stateStoreColFamilySchema.keyStateEncoderSpec.get,
useMultipleValuesPerKey = useMultipleValuesPerKey,
isInternal = isInternal)
- store.abort()
+ readStore.release()
}
provider
}
@@ -364,7 +371,7 @@ class StatePartitionAllColumnFamiliesReader(
private def checkAllColFamiliesExist(
colFamilyNames: List[String],
- stateStore: StateStore
+ stateStore: ReadStateStore
): Unit = {
// Filter out DEFAULT column family from validation for two reasons:
// 1. Some operators (e.g., stream-stream join v3) don't include DEFAULT
in their schema
@@ -385,13 +392,11 @@ class StatePartitionAllColumnFamiliesReader(
s"Column families in state store but not in metadata:
${expectedCFs.diff(actualCFs)}")
}
- // Use a single store instance for both registering column families and
iteration.
- // We cannot abort and then get a read store because abort() invalidates the
loaded version,
- // causing getReadStore() to reload from checkpoint and clear the column
family registrations.
- private lazy val store: StateStore = {
+ // Use a single read-mode store instance for both registering column
families and iteration.
+ private lazy val store: ReadStateStore = {
assert(getStartStoreUniqueId == getEndStoreUniqueId,
"Start and end store unique IDs must be the same when reading all column
families")
- val stateStore = provider.getStore(
+ val stateStore = provider.getReadStore(
partition.sourceOptions.batchId + 1,
getStartStoreUniqueId
)
@@ -408,7 +413,7 @@ class StatePartitionAllColumnFamiliesReader(
val useMultipleValuesPerKey =
isMultiValuedCF(cfSchema.colFamilyName)
require(cfSchema.keyStateEncoderSpec.isDefined,
s"keyStateEncoderSpec must be defined for column family
${cfSchema.colFamilyName}")
- stateStore.createColFamilyIfAbsent(
+ stateStore.registerColFamily(
cfSchema.colFamilyName,
cfSchema.keySchema,
cfSchema.valueSchema,
@@ -444,7 +449,7 @@ class StatePartitionAllColumnFamiliesReader(
}
override def close(): Unit = {
- store.abort()
+ store.release()
super.close()
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StreamStreamJoinStatePartitionReader.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StreamStreamJoinStatePartitionReader.scala
index 471ebc6fa8a6..9ed1460e2dd3 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StreamStreamJoinStatePartitionReader.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StreamStreamJoinStatePartitionReader.scala
@@ -50,6 +50,10 @@ class StreamStreamJoinStatePartitionReaderFactory(
/**
* An implementation of [[PartitionReader]] for State Store data source,
specifically to read
* the partition for the state from stream-stream join.
+ *
+ * NOTE: The state data source is strictly read-only. Any new reader added
here must open the
+ * state store in read-only mode. For the join reader, this means constructing
the
+ * [[SymmetricHashJoinStateManager]] with readOnly = true.
*/
class StreamStreamJoinStatePartitionReader(
storeConf: StateStoreConf,
@@ -184,7 +188,8 @@ class StreamStreamJoinStatePartitionReader(
startKeyWithIndexToValueStateStoreCkptId =
startKeyWithIndexToValueStateStoreCkptId,
endKeyToNumValuesStateStoreCkptId =
endKeyToNumValuesStateStoreCkptId,
endKeyWithIndexToValueStateStoreCkptId =
endKeyWithIndexToValueStateStoreCkptId)),
- joinStoreGenerator = new JoinStateManagerStoreGenerator()
+ joinStoreGenerator = new JoinStateManagerStoreGenerator(),
+ readOnly = true
)
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/SymmetricHashJoinStateManager.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/SymmetricHashJoinStateManager.scala
index ea8fc861b104..97878b33c778 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/SymmetricHashJoinStateManager.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/SymmetricHashJoinStateManager.scala
@@ -32,7 +32,7 @@ import
org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
import org.apache.spark.sql.execution.metric.SQLMetric
import
org.apache.spark.sql.execution.streaming.operators.stateful.{StatefulOperatorStateInfo,
StatefulOpStateStoreCheckpointInfo, WatermarkSupport}
import
org.apache.spark.sql.execution.streaming.operators.stateful.join.StreamingSymmetricHashJoinHelper._
-import
org.apache.spark.sql.execution.streaming.state.{DropLastNFieldsStatePartitionKeyExtractor,
KeyStateEncoderSpec, NoopStatePartitionKeyExtractor,
NoPrefixKeyStateEncoderSpec, RangeScanBoundaryUtils,
StatePartitionKeyExtractor, StateSchemaBroadcast, StateStore,
StateStoreCheckpointInfo, StateStoreColFamilySchema, StateStoreConf,
StateStoreErrors, StateStoreId, StateStoreMetrics, StateStoreProvider,
StateStoreProviderId, SupportsFineGrainedReplay,
TimestampAsPostfixKeyStateEncoderSpec [...]
+import
org.apache.spark.sql.execution.streaming.state.{DropLastNFieldsStatePartitionKeyExtractor,
KeyStateEncoderSpec, NoopStatePartitionKeyExtractor,
NoPrefixKeyStateEncoderSpec, RangeScanBoundaryUtils, ReadStateStore,
StatePartitionKeyExtractor, StateSchemaBroadcast, StateStore,
StateStoreCheckpointInfo, StateStoreColFamilySchema, StateStoreConf,
StateStoreErrors, StateStoreId, StateStoreMetrics, StateStoreProvider,
StateStoreProviderId, SupportsFineGrainedReplay, TimestampAsPostfixKey [...]
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{BooleanType, DataType, LongType, NullType,
StructField, StructType}
import org.apache.spark.util.NextIterator
@@ -239,7 +239,8 @@ class SymmetricHashJoinStateManagerV4(
useStateStoreCoordinator: Boolean = true,
snapshotOptions: Option[SnapshotOptions] = None,
joinStoreGenerator: JoinStateManagerStoreGenerator,
- joinKeyOrdinalForWatermark: Option[Int] = None)
+ joinKeyOrdinalForWatermark: Option[Int] = None,
+ readOnly: Boolean = false)
extends SymmetricHashJoinStateManager with SupportsEvictByTimestamp with
Logging {
// TODO: [SPARK-55729] Once the new state manager is integrated to
stream-stream join operator,
@@ -319,23 +320,69 @@ class SymmetricHashJoinStateManagerV4(
private var stateStoreProvider: StateStoreProvider = _
+ private var _stateStore: StateStore = _
+ private var _readStateStore: ReadStateStore = _
+
+ /**
+ * Writable store. Access fails fast when in read-only mode.
+ */
+ private def stateStore: StateStore = {
+ require(!readOnly,
+ "Writable stateStore is not available on a read-only V4 join state
manager")
+ _stateStore
+ }
+
+ /** Read view of the loaded store. Safe to use in either mode. */
+ private def readStateStore: ReadStateStore =
+ if (readOnly) _readStateStore else _stateStore
+
+ /**
+ * Register a column family on the loaded store. In read-only mode this is
the read-safe
+ * in-memory registration ([[ReadStateStore.registerColFamily]]); in write
mode it is the
+ * writable [[StateStore.createColFamilyIfAbsent]] (which forces a snapshot
for a new family).
+ */
+ private def registerColFamily(
+ colFamilyName: String,
+ keySchema: StructType,
+ valueSchema: StructType,
+ keyStateEncoderSpec: KeyStateEncoderSpec,
+ useMultipleValuesPerKey: Boolean = false,
+ isInternal: Boolean = false): Unit = {
+ if (readOnly) {
+ readStateStore.registerColFamily(colFamilyName, keySchema, valueSchema,
+ keyStateEncoderSpec, useMultipleValuesPerKey, isInternal)
+ } else {
+ stateStore.createColFamilyIfAbsent(colFamilyName, keySchema, valueSchema,
+ keyStateEncoderSpec, useMultipleValuesPerKey, isInternal)
+ }
+ }
+
// We will use the dummy schema for the default CF since we will register CF
separately.
- private val stateStore = getStateStore(
+ initializeStateStore(
dummySchema, dummySchema, useVirtualColumnFamilies = true,
NoPrefixKeyStateEncoderSpec(dummySchema), useMultipleValuesPerKey = false
)
- private def getStateStore(
+ /**
+ * Load the state store and populate `_stateStore` (when !readOnly) or
`_readStateStore`
+ * (when readOnly). Subclasses invoke this for the side effect during
construction;
+ * afterwards, the [[stateStore]] / [[readStateStore]] accessors return the
right view.
+ */
+ private def initializeStateStore(
keySchema: StructType,
valueSchema: StructType,
useVirtualColumnFamilies: Boolean,
keyStateEncoderSpec: KeyStateEncoderSpec,
- useMultipleValuesPerKey: Boolean): StateStore = {
+ useMultipleValuesPerKey: Boolean): Unit = {
val storeName = StateStoreId.DEFAULT_STORE_NAME
val storeProviderId = StateStoreProviderId(stateInfo.get, partitionId,
storeName)
- val store = if (useStateStoreCoordinator) {
+ val handle: ReadStateStore = if (useStateStoreCoordinator) {
assert(handlerSnapshotOptions.isEmpty, "Should not use state store
coordinator " +
"when reading state as data source.")
+ // JoinStateManagerStoreGenerator only exposes a writable getStore.
Read-only mode is
+ // only ever used with useStateStoreCoordinator=false (data source
reads).
+ require(!readOnly,
+ "readOnly mode is not supported with useStateStoreCoordinator=true")
joinStoreGenerator.getStore(
storeProviderId, keySchema, valueSchema, keyStateEncoderSpec,
stateInfo.get.storeVersion, stateStoreCkptId, None,
useVirtualColumnFamilies,
@@ -348,24 +395,43 @@ class SymmetricHashJoinStateManagerV4(
storeConf, hadoopConf, useMultipleValuesPerKey =
useMultipleValuesPerKey,
stateSchemaProvider = None)
if (handlerSnapshotOptions.isDefined) {
+ // Fine-grained snapshot replay (state data source with
snapshotStartBatchId).
if (!stateStoreProvider.isInstanceOf[SupportsFineGrainedReplay]) {
throw
StateStoreErrors.stateStoreProviderDoesNotSupportFineGrainedReplay(
stateStoreProvider.getClass.toString)
}
val opts = handlerSnapshotOptions.get
- stateStoreProvider.asInstanceOf[SupportsFineGrainedReplay]
- .replayStateFromSnapshot(
+ val replayer =
stateStoreProvider.asInstanceOf[SupportsFineGrainedReplay]
+ if (readOnly) {
+ replayer.replayReadStateFromSnapshot(
+ opts.snapshotVersion,
+ opts.endVersion,
+ opts.startStateStoreCkptId,
+ opts.endStateStoreCkptId)
+ } else {
+ replayer.replayStateFromSnapshot(
opts.snapshotVersion,
opts.endVersion,
- readOnly = true,
+ readOnly = false,
opts.startStateStoreCkptId,
opts.endStateStoreCkptId)
+ }
} else {
- stateStoreProvider.getStore(stateInfo.get.storeVersion,
stateStoreCkptId)
+ // No snapshot options: open the store at its committed version
normally.
+ if (readOnly) {
+ stateStoreProvider.getReadStore(stateInfo.get.storeVersion,
stateStoreCkptId)
+ } else {
+ stateStoreProvider.getStore(stateInfo.get.storeVersion,
stateStoreCkptId)
+ }
}
}
- logInfo(log"Loaded store ${MDC(STATE_STORE_ID, store.id)}")
- store
+ if (readOnly) {
+ _readStateStore = handle
+ logInfo(log"Loaded read-only store ${MDC(STATE_STORE_ID, handle.id)}")
+ } else {
+ _stateStore = handle.asInstanceOf[StateStore]
+ logInfo(log"Loaded store ${MDC(STATE_STORE_ID, handle.id)}")
+ }
}
private val keyWithTsToValues = new KeyWithTsToValuesStore
@@ -577,7 +643,9 @@ class SymmetricHashJoinStateManagerV4(
}
override def abortIfNeeded(): Unit = {
- if (!stateStore.hasCommitted) {
+ if (readOnly) {
+ readStateStore.release()
+ } else if (!stateStore.hasCommitted) {
logInfo(log"Aborted store ${MDC(STATE_STORE_ID, stateStore.id)}")
stateStore.abort()
}
@@ -626,8 +694,7 @@ class SymmetricHashJoinStateManagerV4(
private val attachTimestampProjection: UnsafeProjection =
TimestampKeyStateEncoder.getAttachTimestampProjection(keySchema)
- // Create the specific column family in the store for this join side's
KeyWithTsToValuesStore.
- stateStore.createColFamilyIfAbsent(
+ registerColFamily(
colFamilyName,
keySchema,
valueRowConverter.valueAttributes.toStructType,
@@ -659,7 +726,7 @@ class SymmetricHashJoinStateManagerV4(
}
def get(key: UnsafeRow, timestamp: Long): Iterator[ValueAndMatchPair] = {
- stateStore.valuesIterator(createKeyRow(key, timestamp),
colFamilyName).map { valueRow =>
+ readStateStore.valuesIterator(createKeyRow(key, timestamp),
colFamilyName).map { valueRow =>
valueRowConverter.convertValue(valueRow)
}
}
@@ -700,9 +767,9 @@ class SymmetricHashJoinStateManagerV4(
val startKey = createKeyRow(key, minTs).copy()
// rangeScanWithMultiValues endKey is exclusive, so use maxTs + 1
val endKey = Some(createKeyRow(key, maxTs + 1))
- stateStore.rangeScanWithMultiValues(Some(startKey), endKey,
colFamilyName)
+ readStateStore.rangeScanWithMultiValues(Some(startKey), endKey,
colFamilyName)
} else {
- stateStore.prefixScanWithMultiValues(key, colFamilyName)
+ readStateStore.prefixScanWithMultiValues(key, colFamilyName)
}
private var currentTs = -1L
@@ -763,7 +830,7 @@ class SymmetricHashJoinStateManagerV4(
}
def iterator(): Iterator[KeyAndTsToValuePair] = {
- val iter = stateStore.iteratorWithMultiValues(colFamilyName)
+ val iter = readStateStore.iteratorWithMultiValues(colFamilyName)
val reusableKeyAndTsToValuePair = KeyAndTsToValuePair()
iter.map { kv =>
val keyRow = detachTimestampProjection(kv.key)
@@ -800,9 +867,8 @@ class SymmetricHashJoinStateManagerV4(
private val attachTimestampProjection: UnsafeProjection =
TimestampKeyStateEncoder.getAttachTimestampProjection(keySchema)
- // Create the specific column family in the store for this join side's
TsWithKeyStore.
// Mark as internal so that numKeys counts only primary data, not the
secondary index.
- stateStore.createColFamilyIfAbsent(
+ registerColFamily(
colFamilyName,
keySchema,
valueStructType,
@@ -881,7 +947,8 @@ class SymmetricHashJoinStateManagerV4(
} else {
None
}
- val evictIterator = stateStore.rangeScanWithMultiValues(startKeyRow,
endKeyRow, colFamilyName)
+ val evictIterator =
+ readStateStore.rangeScanWithMultiValues(startKeyRow, endKeyRow,
colFamilyName)
new NextIterator[EvictedKeysResult]() {
var currentKeyRow: UnsafeRow = null
var currentEventTime: Long = -1L
@@ -1025,7 +1092,8 @@ abstract class SymmetricHashJoinStateManagerBase(
skippedNullValueCount: Option[SQLMetric] = None,
useStateStoreCoordinator: Boolean = true,
snapshotOptions: Option[SnapshotOptions] = None,
- joinStoreGenerator: JoinStateManagerStoreGenerator)
+ joinStoreGenerator: JoinStateManagerStoreGenerator,
+ readOnly: Boolean = false)
extends SymmetricHashJoinStateManager
with SupportsEvictByCondition
with SupportsIndexedKeys
@@ -1522,8 +1590,43 @@ abstract class SymmetricHashJoinStateManagerBase(
handlerSnapshotOptions: Option[HandlerSnapshotOptions] = None) extends
Logging {
private var stateStoreProvider: StateStoreProvider = _
- /** StateStore that the subclasses of this class is going to operate on */
- protected def stateStore: StateStore
+ // Exactly one is populated by initializeStateStore -- readOnly decides
which.
+ private var _stateStore: StateStore = _
+ private var _readStateStore: ReadStateStore = _
+
+ /**
+ * Writable store. Access fails fast when in read-only mode.
+ */
+ protected def stateStore: StateStore = {
+ require(!readOnly,
+ "Writable stateStore is not available on a read-only state store
handler")
+ _stateStore
+ }
+
+ /** Read view of the loaded store. Safe to use in either mode. */
+ protected def readStateStore: ReadStateStore =
+ if (readOnly) _readStateStore else _stateStore
+
+ /**
+ * Register a column family on the loaded store. In read-only mode this is
the read-safe
+ * in-memory registration ([[ReadStateStore.registerColFamily]]); in write
mode it is the
+ * writable [[StateStore.createColFamilyIfAbsent]] (which forces a
snapshot for a new family).
+ */
+ protected def registerColFamily(
+ colFamilyName: String,
+ keySchema: StructType,
+ valueSchema: StructType,
+ keyStateEncoderSpec: KeyStateEncoderSpec,
+ useMultipleValuesPerKey: Boolean = false,
+ isInternal: Boolean = false): Unit = {
+ if (readOnly) {
+ readStateStore.registerColFamily(colFamilyName, keySchema, valueSchema,
+ keyStateEncoderSpec, useMultipleValuesPerKey, isInternal)
+ } else {
+ stateStore.createColFamilyIfAbsent(colFamilyName, keySchema,
valueSchema,
+ keyStateEncoderSpec, useMultipleValuesPerKey, isInternal)
+ }
+ }
def commit(): Unit = {
stateStore.commit()
@@ -1531,7 +1634,9 @@ abstract class SymmetricHashJoinStateManagerBase(
}
def abortIfNeeded(): Unit = {
- if (!stateStore.hasCommitted) {
+ if (readOnly) {
+ readStateStore.release()
+ } else if (!stateStore.hasCommitted) {
logInfo(log"Aborted store ${MDC(STATE_STORE_ID, stateStore.id)}")
stateStore.abort()
}
@@ -1544,24 +1649,33 @@ abstract class SymmetricHashJoinStateManagerBase(
def metrics: StateStoreMetrics = stateStore.metrics
- def getLatestCheckpointInfo(): StateStoreCheckpointInfo = {
+ def getLatestCheckpointInfo(): StateStoreCheckpointInfo =
stateStore.getStateStoreCheckpointInfo()
- }
- /** Get the StateStore with the given schema */
- protected def getStateStore(
+ /**
+ * Load the state store for this handler and populate `_stateStore` (when
!readOnly)
+ * or `_readStateStore` (when readOnly). Subclasses invoke this for the
side effect
+ * during construction; afterwards, the [[stateStore]] /
[[readStateStore]] accessors
+ * return the right view based on mode.
+ */
+ protected def initializeStateStore(
keySchema: StructType,
valueSchema: StructType,
- useVirtualColumnFamilies: Boolean): StateStore = {
+ useVirtualColumnFamilies: Boolean): Unit = {
val storeName = if (useVirtualColumnFamilies) {
StateStoreId.DEFAULT_STORE_NAME
} else {
getStateStoreName(joinSide, stateStoreType)
}
val storeProviderId = StateStoreProviderId(stateInfo.get, partitionId,
storeName)
- val store = if (useStateStoreCoordinator) {
+ val handle: ReadStateStore = if (useStateStoreCoordinator) {
assert(handlerSnapshotOptions.isEmpty, "Should not use state store
coordinator " +
"when reading state as data source.")
+ // JoinStateManagerStoreGenerator only exposes a writable getStore.
Read-only
+ // mode is only ever used with useStateStoreCoordinator=false (data
source reads),
+ // so this branch must not be reached with readOnly=true.
+ require(!readOnly,
+ "readOnly mode is not supported with useStateStoreCoordinator=true")
joinStoreGenerator.getStore(
storeProviderId, keySchema, valueSchema,
NoPrefixKeyStateEncoderSpec(keySchema),
stateInfo.get.storeVersion, stateStoreCkptId, None,
useVirtualColumnFamilies,
@@ -1573,24 +1687,44 @@ abstract class SymmetricHashJoinStateManagerBase(
useColumnFamilies = useVirtualColumnFamilies, storeConf, hadoopConf,
useMultipleValuesPerKey = false, stateSchemaProvider = None)
if (handlerSnapshotOptions.isDefined) {
+ // Fine-grained snapshot replay (state data source with
snapshotStartBatchId).
if (!stateStoreProvider.isInstanceOf[SupportsFineGrainedReplay]) {
throw
StateStoreErrors.stateStoreProviderDoesNotSupportFineGrainedReplay(
stateStoreProvider.getClass.toString)
}
val opts = handlerSnapshotOptions.get
- stateStoreProvider.asInstanceOf[SupportsFineGrainedReplay]
- .replayStateFromSnapshot(
+ val replayer =
stateStoreProvider.asInstanceOf[SupportsFineGrainedReplay]
+ if (readOnly) {
+ replayer.replayReadStateFromSnapshot(
opts.snapshotVersion,
opts.endVersion,
- readOnly = true,
opts.startStateStoreCkptId,
opts.endStateStoreCkptId)
+ } else {
+ replayer.replayStateFromSnapshot(
+ opts.snapshotVersion,
+ opts.endVersion,
+ readOnly = false,
+ opts.startStateStoreCkptId,
+ opts.endStateStoreCkptId)
+ }
} else {
- stateStoreProvider.getStore(stateInfo.get.storeVersion,
stateStoreCkptId)
+ // No snapshot options: open the store at its committed version
normally.
+ // readOnly mode must not write to the checkpoint path; use the
read-only store.
+ if (readOnly) {
+ stateStoreProvider.getReadStore(stateInfo.get.storeVersion,
stateStoreCkptId)
+ } else {
+ stateStoreProvider.getStore(stateInfo.get.storeVersion,
stateStoreCkptId)
+ }
}
}
- logInfo(log"Loaded store ${MDC(STATE_STORE_ID, store.id)}")
- store
+ if (readOnly) {
+ _readStateStore = handle
+ logInfo(log"Loaded read-only store ${MDC(STATE_STORE_ID, handle.id)}")
+ } else {
+ _stateStore = handle.asInstanceOf[StateStore]
+ logInfo(log"Loaded store ${MDC(STATE_STORE_ID, handle.id)}")
+ }
}
}
@@ -1617,8 +1751,7 @@ abstract class SymmetricHashJoinStateManagerBase(
private val longValueSchema = new StructType().add("value", "long")
private val longToUnsafeRow = UnsafeProjection.create(longValueSchema)
private val valueRow = longToUnsafeRow(new
SpecificInternalRow(longValueSchema))
- protected val stateStore: StateStore =
- getStateStore(keySchema, longValueSchema, useVirtualColumnFamilies)
+ initializeStateStore(keySchema, longValueSchema, useVirtualColumnFamilies)
// Set up virtual column family name in the store if it is being used
private val colFamilyName = if (useVirtualColumnFamilies) {
@@ -1629,7 +1762,7 @@ abstract class SymmetricHashJoinStateManagerBase(
// Create the specific column family in the store for this join side's
KeyToNumValuesStore
if (useVirtualColumnFamilies) {
- stateStore.createColFamilyIfAbsent(
+ registerColFamily(
colFamilyName,
keySchema,
longValueSchema,
@@ -1640,7 +1773,7 @@ abstract class SymmetricHashJoinStateManagerBase(
/** Get the number of values the key has */
def get(key: UnsafeRow): Long = {
- val longValueRow = stateStore.get(key, colFamilyName)
+ val longValueRow = readStateStore.get(key, colFamilyName)
if (longValueRow != null) longValueRow.getLong(0) else 0L
}
@@ -1657,7 +1790,7 @@ abstract class SymmetricHashJoinStateManagerBase(
def iterator: Iterator[KeyAndNumValues] = {
val keyAndNumValues = new KeyAndNumValues()
- stateStore.iterator(colFamilyName).map { pair =>
+ readStateStore.iterator(colFamilyName).map { pair =>
keyAndNumValues.withNew(pair.key, pair.value.getLong(0))
}
}
@@ -1727,7 +1860,7 @@ abstract class SymmetricHashJoinStateManagerBase(
private val valueRowConverter = StreamingSymmetricHashJoinValueRowConverter
.create(inputValueAttributes, stateFormatVersion)
- protected val stateStore = getStateStore(keyWithIndexSchema,
+ initializeStateStore(keyWithIndexSchema,
valueRowConverter.valueAttributes.toStructType, useVirtualColumnFamilies)
// Set up virtual column family name in the store if it is being used
@@ -1739,7 +1872,7 @@ abstract class SymmetricHashJoinStateManagerBase(
// Create the specific column family in the store for this join side's
KeyWithIndexToValueStore
if (useVirtualColumnFamilies) {
- stateStore.createColFamilyIfAbsent(
+ registerColFamily(
colFamilyName,
keyWithIndexSchema,
valueRowConverter.valueAttributes.toStructType,
@@ -1749,7 +1882,7 @@ abstract class SymmetricHashJoinStateManagerBase(
def get(key: UnsafeRow, valueIndex: Long): ValueAndMatchPair = {
valueRowConverter.convertValue(
- stateStore.get(keyWithIndexRow(key, valueIndex), colFamilyName)
+ readStateStore.get(keyWithIndexRow(key, valueIndex), colFamilyName)
)
}
@@ -1767,7 +1900,7 @@ abstract class SymmetricHashJoinStateManagerBase(
while (hasMoreValues) {
val keyWithIndex = keyWithIndexRow(key, index)
val valuePair =
- valueRowConverter.convertValue(stateStore.get(keyWithIndex,
colFamilyName))
+ valueRowConverter.convertValue(readStateStore.get(keyWithIndex,
colFamilyName))
if (valuePair == null) {
handleNullValuePair(key, index, numValues)
skippedNullValueCount.foreach(_ += 1L)
@@ -1813,7 +1946,7 @@ abstract class SymmetricHashJoinStateManagerBase(
def iterator: Iterator[KeyWithIndexAndValue] = {
val keyWithIndexAndValue = new KeyWithIndexAndValue()
- stateStore.iterator(colFamilyName).map { pair =>
+ readStateStore.iterator(colFamilyName).map { pair =>
val valuePair = valueRowConverter.convertValue(pair.value)
keyWithIndexAndValue.withNew(
keyRowGenerator(pair.key),
pair.key.getLong(indexOrdinalInKeyWithIndexRow), valuePair)
@@ -1851,12 +1984,13 @@ class SymmetricHashJoinStateManagerV1(
skippedNullValueCount: Option[SQLMetric] = None,
useStateStoreCoordinator: Boolean = true,
snapshotOptions: Option[SnapshotOptions] = None,
- joinStoreGenerator: JoinStateManagerStoreGenerator)
+ joinStoreGenerator: JoinStateManagerStoreGenerator,
+ readOnly: Boolean = false)
extends SymmetricHashJoinStateManagerBase(
joinSide, inputValueAttributes, joinKeys, stateInfo, storeConf, hadoopConf,
partitionId, keyToNumValuesStateStoreCkptId,
keyWithIndexToValueStateStoreCkptId,
stateFormatVersion, skippedNullValueCount, useStateStoreCoordinator,
snapshotOptions,
- joinStoreGenerator) {
+ joinStoreGenerator, readOnly) {
/** Commit all the changes to all the state stores */
override def commit(): Unit = {
@@ -1931,12 +2065,13 @@ class SymmetricHashJoinStateManagerV2(
skippedNullValueCount: Option[SQLMetric] = None,
useStateStoreCoordinator: Boolean = true,
snapshotOptions: Option[SnapshotOptions] = None,
- joinStoreGenerator: JoinStateManagerStoreGenerator)
+ joinStoreGenerator: JoinStateManagerStoreGenerator,
+ readOnly: Boolean = false)
extends SymmetricHashJoinStateManagerBase(
joinSide, inputValueAttributes, joinKeys, stateInfo, storeConf, hadoopConf,
partitionId, keyToNumValuesStateStoreCkptId,
keyWithIndexToValueStateStoreCkptId,
stateFormatVersion, skippedNullValueCount, useStateStoreCoordinator,
snapshotOptions,
- joinStoreGenerator) {
+ joinStoreGenerator, readOnly) {
/** Commit all the changes to the state store */
override def commit(): Unit = {
@@ -2043,27 +2178,28 @@ object SymmetricHashJoinStateManager {
useStateStoreCoordinator: Boolean = true,
snapshotOptions: Option[SnapshotOptions] = None,
joinStoreGenerator: JoinStateManagerStoreGenerator,
- joinKeyOrdinalForWatermark: Option[Int] = None):
SymmetricHashJoinStateManager = {
+ joinKeyOrdinalForWatermark: Option[Int] = None,
+ readOnly: Boolean = false): SymmetricHashJoinStateManager = {
if (stateFormatVersion == 4) {
new SymmetricHashJoinStateManagerV4(
joinSide, inputValueAttributes, joinKeys, stateInfo, storeConf,
hadoopConf,
partitionId, keyToNumValuesStateStoreCkptId,
keyWithIndexToValueStateStoreCkptId,
stateFormatVersion, skippedNullValueCount, useStateStoreCoordinator,
snapshotOptions,
- joinStoreGenerator, joinKeyOrdinalForWatermark
+ joinStoreGenerator, joinKeyOrdinalForWatermark, readOnly
)
} else if (stateFormatVersion == 3) {
new SymmetricHashJoinStateManagerV2(
joinSide, inputValueAttributes, joinKeys, stateInfo, storeConf,
hadoopConf,
partitionId, keyToNumValuesStateStoreCkptId,
keyWithIndexToValueStateStoreCkptId,
stateFormatVersion, skippedNullValueCount, useStateStoreCoordinator,
snapshotOptions,
- joinStoreGenerator
+ joinStoreGenerator, readOnly
)
} else {
new SymmetricHashJoinStateManagerV1(
joinSide, inputValueAttributes, joinKeys, stateInfo, storeConf,
hadoopConf,
partitionId, keyToNumValuesStateStoreCkptId,
keyWithIndexToValueStateStoreCkptId,
stateFormatVersion, skippedNullValueCount, useStateStoreCoordinator,
snapshotOptions,
- joinStoreGenerator
+ joinStoreGenerator, readOnly
)
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
index 2562f1ff3304..e64e790bd21f 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
@@ -149,7 +149,10 @@ private[sql] class HDFSBackedStateStoreProvider extends
StateStoreProvider with
private val newVersion = version + 1
@volatile private var state: STATE = UPDATING
private val finalDeltaFile: Path = deltaFile(newVersion)
- private lazy val deltaFileStream = fm.createAtomic(finalDeltaFile,
overwriteIfPossible = true)
+ private lazy val deltaFileStream = {
+ createBaseDirIfNotExists()
+ fm.createAtomic(finalDeltaFile, overwriteIfPossible = true)
+ }
private lazy val compressedStream = compressStream(deltaFileStream)
override def id: StateStoreId =
HDFSBackedStateStoreProvider.this.stateStoreId
@@ -241,6 +244,12 @@ private[sql] class HDFSBackedStateStoreProvider extends
StateStoreProvider with
}
}
+ override def release(): Unit = {
+ verify(state == UPDATING || state == RELEASED,
+ s"Cannot release after state $state")
+ state = RELEASED
+ }
+
/** Abort all the updates made on this store. This store will not be
usable any more. */
override def abort(): Unit = {
// This if statement is to ensure that files are deleted only once: if
either commit or abort
@@ -469,7 +478,8 @@ private[sql] class HDFSBackedStateStoreProvider extends
StateStoreProvider with
this.numColsPrefixKey = getNumColsPrefixKey(keyStateEncoderSpec)
- fm.mkdirs(baseDir)
+ // baseDir is created lazily on the first write -- see
createBaseDirIfNotExists().
+ // This keeps read-only callers from issuing an mkdirs on the checkpoint
path.
}
override def stateStoreId: StateStoreId = stateStoreId_
@@ -525,6 +535,23 @@ private[sql] class HDFSBackedStateStoreProvider extends
StateStoreProvider with
private lazy val loadedMaps = new util.TreeMap[Long,
HDFSBackedStateStoreMap](
Ordering[Long].reverse)
private lazy val baseDir = stateStoreId.storeCheckpointLocation()
+
+ // baseDir is created lazily on the first write, so read-only entry points
never trigger a
+ // mkdirs on the checkpoint path. compareAndSet ensures only one task issues
the mkdirs even
+ // if several tasks for this partition race here; the flag is reset on
failure so a transient
+ // mkdirs error does not permanently skip creation.
+ private val baseDirChecked = new AtomicBoolean(false)
+ private def createBaseDirIfNotExists(): Unit = {
+ if (baseDirChecked.compareAndSet(false, true)) {
+ try {
+ if (!fm.exists(baseDir)) fm.mkdirs(baseDir)
+ } catch {
+ case e: Throwable =>
+ baseDirChecked.set(false)
+ throw e
+ }
+ }
+ }
// Visible to state pkg for testing.
private[state] lazy val fm = {
val mgr = CheckpointFileManager.create(baseDir, hadoopConf)
@@ -890,6 +917,7 @@ private[sql] class HDFSBackedStateStoreProvider extends
StateStoreProvider with
var rawOutput: CancellableFSDataOutputStream = null
var output: DataOutputStream = null
try {
+ createBaseDirIfNotExists()
rawOutput = fm.createAtomic(targetFile, overwriteIfPossible = true)
output = compressStream(rawOutput)
// Entry iterator doesn't do verification, we will do it ourselves
@@ -1232,6 +1260,12 @@ private[sql] class HDFSBackedStateStoreProvider extends
StateStoreProvider with
"HDFSBackedStateStoreProvider does not support checkpointFormatVersion
> 1 " +
"but a state store checkpointID is passed in")
}
+ // The trait's return type is StateStore (writable), so we cannot return
+ // HDFSBackedReadStateStore here. Callers that need a read-only store must
call
+ // replayReadStateFromSnapshot instead.
+ require(!readOnly,
+ "HDFSBackedStateStoreProvider.replayStateFromSnapshot cannot return a
read-only " +
+ "store; use replayReadStateFromSnapshot when readOnly = true.")
val newMap = replayLoadedMapFromSnapshot(snapshotVersion, endVersion)
logInfo(log"Retrieved snapshot at version " +
log"${MDC(LogKeys.STATE_STORE_VERSION, snapshotVersion)} and apply delta
files to version " +
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
index bd479ffc6d2a..8b23c9628450 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
@@ -309,13 +309,19 @@ class RocksDB(
*
* @param colFamilyName - column family name
* @param isInternal - whether the column family is for internal use or not
+ * @param readOnly - when true (read-only callers), do not force a snapshot
for a newly created
+ * family; read stores never commit, so registering a
family carries no write
+ * intent.
* @return - virtual column family id
*/
- def createColFamilyIfAbsent(colFamilyName: String, isInternal: Boolean):
Short = {
+ def createColFamilyIfAbsent(
+ colFamilyName: String,
+ isInternal: Boolean,
+ readOnly: Boolean = false): Short = {
if (!checkColFamilyExists(colFamilyName)) {
val newColumnFamilyId = maxColumnFamilyId.incrementAndGet().toShort
addToColFamilyMaps(colFamilyName, newColumnFamilyId, isInternal)
- shouldForceSnapshot.set(true)
+ if (!readOnly) shouldForceSnapshot.set(true)
newColumnFamilyId
} else {
colFamilyNameToInfoMap.get(colFamilyName).cfId
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
index c181130eec94..52e7ce296567 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
@@ -201,6 +201,20 @@ private[sql] class RocksDBStateStoreProvider
}
}
+ // Read-only registration: sets up the virtual column-family id and the
key/value encoders
+ // needed to decode this family. Purely in-memory -- it does not write to
the checkpoint and
+ // carries no write intent (no forced snapshot). Safe to call on a
read-only store.
+ override def registerColFamily(
+ colFamilyName: String,
+ keySchema: StructType,
+ valueSchema: StructType,
+ keyStateEncoderSpec: KeyStateEncoderSpec,
+ useMultipleValuesPerKey: Boolean = false,
+ isInternal: Boolean = false): Unit = {
+ registerColFamilyInternal(colFamilyName, keySchema, valueSchema,
keyStateEncoderSpec,
+ useMultipleValuesPerKey, isInternal)
+ }
+
override def createColFamilyIfAbsent(
colFamilyName: String,
keySchema: StructType,
@@ -208,9 +222,20 @@ private[sql] class RocksDBStateStoreProvider
keyStateEncoderSpec: KeyStateEncoderSpec,
useMultipleValuesPerKey: Boolean = false,
isInternal: Boolean = false): Unit = {
+ registerColFamilyInternal(colFamilyName, keySchema, valueSchema,
keyStateEncoderSpec,
+ useMultipleValuesPerKey, isInternal)
+ }
+
+ private def registerColFamilyInternal(
+ colFamilyName: String,
+ keySchema: StructType,
+ valueSchema: StructType,
+ keyStateEncoderSpec: KeyStateEncoderSpec,
+ useMultipleValuesPerKey: Boolean,
+ isInternal: Boolean): Unit = {
validateAndTransitionState(UPDATE)
verifyColFamilyCreationOrDeletion("create_col_family", colFamilyName,
isInternal)
- val cfId = rocksDB.createColFamilyIfAbsent(colFamilyName, isInternal)
+ val cfId = rocksDB.createColFamilyIfAbsent(colFamilyName, isInternal,
readOnly)
val dataEncoderCacheKey = StateRowEncoderCacheKey(
queryRunId = StateStoreProvider.getRunId(hadoopConf),
operatorId = stateStoreId.operatorId,
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala
index ad067b8edcc3..961d2f963ce8 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala
@@ -270,6 +270,22 @@ trait ReadStateStore {
* @return Set of all column family names
*/
def allColumnFamilyNames: Set[String]
+
+ /**
+ * Read-only counterpart of [[StateStore.createColFamilyIfAbsent]]: a purely
in-memory
+ * registration of the column family's id and encoders so reads can decode
it. Must not write
+ * to the checkpoint or carry write intent (e.g. forcing a snapshot).
+ */
+ def registerColFamily(
+ colFamilyName: String,
+ keySchema: StructType,
+ valueSchema: StructType,
+ keyStateEncoderSpec: KeyStateEncoderSpec,
+ useMultipleValuesPerKey: Boolean = false,
+ isInternal: Boolean = false): Unit = {
+ throw new UnsupportedOperationException(
+ "registerColFamily is not supported by this ReadStateStore")
+ }
}
/**
@@ -474,6 +490,22 @@ class WrappedReadStateStore(store: StateStore) extends
ReadStateStore {
colFamilyName: String): StateStoreIterator[UnsafeRowPair] = {
store.iteratorWithMultiValues(colFamilyName)
}
+
+ // Registering a column family is a purely in-memory operation (virtual
column family id and
+ // key/value encoders); it does not touch the native store or write to the
checkpoint. Reads
+ // cannot decode a column family that has not been registered, so this is
delegated to the
+ // underlying store via its read-safe registerColFamily rather than rejected.
+ override def registerColFamily(
+ colFamilyName: String,
+ keySchema: StructType,
+ valueSchema: StructType,
+ keyStateEncoderSpec: KeyStateEncoderSpec,
+ useMultipleValuesPerKey: Boolean = false,
+ isInternal: Boolean = false): Unit = {
+ store.registerColFamily(
+ colFamilyName, keySchema, valueSchema, keyStateEncoderSpec,
+ useMultipleValuesPerKey, isInternal)
+ }
}
/**
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceChangeDataReadSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceChangeDataReadSuite.scala
index 4e9f6cca2ffc..bb4b00e74d7a 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceChangeDataReadSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceChangeDataReadSuite.scala
@@ -233,17 +233,20 @@ abstract class StateDataSourceChangeDataReaderSuite
extends StateDataSourceTestB
test("ERROR: mixed checkpoint format versions not supported") {
withTempDir { tempDir =>
- val commitLog = new CommitLog(spark,
- new File(tempDir.getAbsolutePath, "commits").getAbsolutePath)
-
- // Start version: treated as v1 (no operator unique ids)
- val startMetadata = CommitMetadata(0)
- assert(commitLog.add(0, startMetadata))
-
- // End version: treated as v2 (operator 0 has unique ids)
- val endMetadata = CommitMetadataV2(0,
- Some(Map[Long, Array[Array[String]]](0L -> Array(Array("uid")))))
- assert(commitLog.add(1, endMetadata))
+ // Test setup writes commit log entries.
+ withWritableCheckpoint {
+ val commitLog = new CommitLog(spark,
+ new File(tempDir.getAbsolutePath, "commits").getAbsolutePath)
+
+ // Start version: treated as v1 (no operator unique ids)
+ val startMetadata = CommitMetadata(0)
+ assert(commitLog.add(0, startMetadata))
+
+ // End version: treated as v2 (operator 0 has unique ids)
+ val endMetadata = CommitMetadataV2(0,
+ Some(Map[Long, Array[Array[String]]](0L -> Array(Array("uid")))))
+ assert(commitLog.add(1, endMetadata))
+ }
val exc =
intercept[StateDataSourceMixedCheckpointFormatVersionsNotSupported] {
spark.read.format("statestore")
@@ -417,6 +420,23 @@ abstract class StateDataSourceChangeDataReaderSuite
extends StateDataSourceTestB
}
}
+ test("read flatMapGroupsWithState state change feed") {
+ withTempDir { tempDir =>
+ runFlatMapGroupsWithStateQuery(tempDir.getAbsolutePath)
+ val df = spark.read.format("statestore")
+ .option(StateSourceOptions.PATH, tempDir.getAbsolutePath)
+ .option(StateSourceOptions.READ_CHANGE_FEED, value = true)
+ .option(StateSourceOptions.CHANGE_START_BATCH_ID, 0)
+ .option(StateSourceOptions.CHANGE_END_BATCH_ID, 1)
+ .load()
+ assert(df.collect().nonEmpty,
+ "expected flatMapGroupsWithState state change feed to return change
records")
+ val keys = df.selectExpr("key.value AS
k").collect().map(_.getString(0)).toSet
+ assert(keys == Set("hello", "world", "scala"),
+ s"unexpected change-feed keys: $keys")
+ }
+ }
+
test("read stream-stream join state change feed") {
withTempDir { tempDir =>
val inputData = MemoryStream[(Int, Long)]
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala
index 4a2a454077a7..a81c586af433 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala
@@ -65,13 +65,15 @@ class StateDataSourceNegativeTestSuite extends
StateDataSourceTestBase {
withTempDir { tempDir =>
runLargeDataStreamingAggregationQuery(tempDir.getAbsolutePath)
- val offsetLog = new OffsetSeqLog(spark,
- new File(tempDir.getAbsolutePath, "offsets").getAbsolutePath)
- val commitLog = new CommitLog(spark,
- new File(tempDir.getAbsolutePath, "commits").getAbsolutePath)
-
- offsetLog.purgeAfter(0)
- commitLog.purgeAfter(-1)
+ // Purge logs to create the error condition. This writes to checkpoint
files.
+ withWritableCheckpoint {
+ val offsetLog = new OffsetSeqLog(spark,
+ new File(tempDir.getAbsolutePath, "offsets").getAbsolutePath)
+ val commitLog = new CommitLog(spark,
+ new File(tempDir.getAbsolutePath, "commits").getAbsolutePath)
+ offsetLog.purgeAfter(0)
+ commitLog.purgeAfter(-1)
+ }
intercept[StataDataSourceCommittedBatchUnavailable] {
spark.read.format("statestore").load(tempDir.getAbsolutePath)
@@ -477,6 +479,63 @@ class HDFSBackedStateDataSourceReadSuite extends
StateDataSourceReadSuite {
testSnapshotOnJoinState("hdfs", 1)
testSnapshotOnJoinState("hdfs", 2)
}
+
+ test("read-only checkpoint: HDFS-backed join snapshot replay honors
readOnly") {
+ // Use the pre-generated golden join1 state so this test does not race the
maintenance
+ // thread. Copy it into a withTempDir (auto-protected) so any unintended
write on the
+ // read path is caught by WriteProtectedLocalFileSystem.
+ val resourceUri = this.getClass.getResource(
+ "/structured-streaming/checkpoint-version-4.0.0/hdfs/join1/").toURI
+ withTempDir { tempDir =>
+ withWritableCheckpoint {
+ Utils.copyDirectory(new File(resourceUri), tempDir)
+ }
+ val df = spark.read
+ .format("statestore")
+ .option(StateSourceOptions.PATH, tempDir.getAbsolutePath)
+ .option(StateSourceOptions.JOIN_SIDE, "left")
+ .option(StateSourceOptions.SNAPSHOT_START_BATCH_ID, 1)
+ .option(StateSourceOptions.SNAPSHOT_PARTITION_ID, 2)
+ .load()
+ assert(df.collect().nonEmpty)
+ }
+ }
+
+ // HDFSBackedStateStoreProvider.init() defers baseDir mkdirs to the first
write
+ // (createBaseDirIfNotExists). Pin this behavior: deleting a per-store
baseDir under a
+ // partition lets partition discovery and schema discovery succeed
(PARTITION_ID_TO_CHECK_SCHEMA=0
+ // is intact, all partition dirs still exist) but exposes the task-side
init() in a way
+ // that previously would have triggered mkdirs -- and now must NOT.
+ test("read-only checkpoint: HDFS provider does not mkdirs baseDir on read
path") {
+ withTempDir { tempDir =>
+ runStreamStreamJoinQuery(tempDir.getAbsolutePath)
+
+ val namedStoreDir = new java.io.File(tempDir,
"state/0/2/left-keyToNumValues")
+ assert(namedStoreDir.exists(),
+ "Expected state/0/2/left-keyToNumValues to exist after running the
join query")
+ Utils.deleteRecursively(namedStoreDir)
+ assert(!namedStoreDir.exists())
+
+ val exc = intercept[SparkException] {
+ spark.read
+ .format("statestore")
+ .option(StateSourceOptions.PATH, tempDir.getAbsolutePath)
+ .option(StateSourceOptions.JOIN_SIDE, "left")
+ .load()
+ .collect()
+ }
+ // Read fails with a data-load error from the missing state files -- NOT
by
+ // attempting to (re)create the directory.
+ checkError(
+ exception = exc,
+ condition =
"CANNOT_LOAD_STATE_STORE.CANNOT_READ_DELTA_FILE_NOT_EXISTS",
+ sqlState = Some("58030"),
+ parameters = Map(
+ "clazz" ->
("HDFSStateStoreProvider\\[.*storeName=left-keyToNumValues.*\\]"),
+ "fileToRead" -> ".*state/0/2/left-keyToNumValues/1\\.delta"),
+ matchPVals = true)
+ }
+ }
}
class RocksDBStateDataSourceReadSuite extends StateDataSourceReadSuite {
@@ -717,6 +776,19 @@ abstract class StateDataSourceReadSuite extends
StateDataSourceTestBase with Ass
protected val keySchema: StructType = StateStoreTestsHelper.keySchema
protected val valueSchema: StructType = StateStoreTestsHelper.valueSchema
+ test("framework sanity: write to protected path inside withTempDir is
rejected") {
+ val e = intercept[Exception] {
+ withTempDir { tempDir =>
+ // No withWritableCheckpoint / testStream: tempDir is protected.
+ val out = new java.io.File(tempDir, "forbidden.txt")
+ val fs = new org.apache.hadoop.fs.Path(out.getAbsolutePath)
+ .getFileSystem(spark.sessionState.newHadoopConf())
+ fs.create(new org.apache.hadoop.fs.Path(out.getAbsolutePath)).close()
+ }
+ }
+ assertWriteProtectionFailure(e)
+ }
+
protected def newStateStoreProvider(): StateStoreProvider
/**
@@ -1164,15 +1236,12 @@ abstract class StateDataSourceReadSuite extends
StateDataSourceTestBase with Ass
.groupBy("value")
.count()
- stream.addData(1 to 10000: _*)
-
- val query = df.writeStream.format("noop")
- .option("checkpointLocation", tempDir.getAbsolutePath)
- .outputMode(OutputMode.Update())
- .start()
-
- query.processAllAvailable()
- query.stop()
+ testStream(df, OutputMode.Update())(
+ StartStream(checkpointLocation = tempDir.getAbsolutePath),
+ AddData(stream, 1 to 10000: _*),
+ ProcessAllAvailable(),
+ StopStream
+ )
val stateReadDf = spark.read
.format("statestore")
@@ -1267,12 +1336,15 @@ abstract class StateDataSourceReadSuite extends
StateDataSourceTestBase with Ass
protected def testSnapshotNotFound(): Unit = {
withTempDir { tempDir =>
- val provider = getNewStateStoreProvider(tempDir.getAbsolutePath)
- for (i <- 1 to 4) {
- val store = provider.getStore(i - 1)
- put(store, "a", i, i)
- store.commit()
- provider.doMaintenance() // create a snapshot every other delta file
+ val provider = withWritableCheckpoint {
+ val p = getNewStateStoreProvider(tempDir.getAbsolutePath)
+ for (i <- 1 to 4) {
+ val store = p.getStore(i - 1)
+ put(store, "a", i, i)
+ store.commit()
+ p.doMaintenance() // create a snapshot every other delta file
+ }
+ p
}
val exc = intercept[SparkException] {
@@ -1286,16 +1358,19 @@ abstract class StateDataSourceReadSuite extends
StateDataSourceTestBase with Ass
protected def testGetReadStoreWithStartVersion(): Unit = {
withTempDir { tempDir =>
val versionToCkptId = scala.collection.mutable.Map[Long,
Option[String]]()
- val provider = getNewStateStoreProvider(tempDir.getAbsolutePath)
- for (i <- 1 to 4) {
- val store = provider.getStore(i - 1, versionToCkptId.getOrElse(i - 1,
None))
- put(store, "a", i, i)
- store.commit()
+ val provider = withWritableCheckpoint {
+ val p = getNewStateStoreProvider(tempDir.getAbsolutePath)
+ for (i <- 1 to 4) {
+ val store = p.getStore(i - 1, versionToCkptId.getOrElse(i - 1, None))
+ put(store, "a", i, i)
+ store.commit()
- val ssInfo = store.getStateStoreCheckpointInfo()
- versionToCkptId(ssInfo.batchVersion) = ssInfo.stateStoreCkptId
+ val ssInfo = store.getStateStoreCheckpointInfo()
+ versionToCkptId(ssInfo.batchVersion) = ssInfo.stateStoreCkptId
- provider.doMaintenance()
+ p.doMaintenance()
+ }
+ p
}
val result =
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTestBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTestBase.scala
index aed484470e47..c5262833b1eb 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTestBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTestBase.scala
@@ -29,7 +29,10 @@ import org.apache.spark.sql.streaming._
import org.apache.spark.sql.streaming.util.{ColumnFamilyMetadata,
StreamManualClock}
import org.apache.spark.sql.types.{BooleanType, IntegerType, LongType,
NullType, StringType, StructField, StructType, TimestampType}
-trait StateDataSourceTestBase extends StreamTest with StateStoreMetricsTest {
+trait StateDataSourceTestBase
+ extends StreamTest
+ with WriteProtectedCheckpointTestMixin
+ with StateStoreMetricsTest {
import testImplicits._
override def beforeEach(): Unit = {
@@ -39,8 +42,10 @@ trait StateDataSourceTestBase extends StreamTest with
StateStoreMetricsTest {
override def afterEach(): Unit = {
// Stop maintenance tasks because they may access already deleted
checkpoint.
- StateStore.stop()
+ // The WriteProtectedCheckpointTestMixin's afterEach (called via super)
clears
+ // WriteProtectedPaths first, so teardown writes are not falsely flagged.
super.afterEach()
+ StateStore.stop()
}
protected def runCompositeKeyStreamingAggregationQuery(checkpointRoot:
String): Unit = {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala
index 41f66807bef3..78e394b7cf75 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala
@@ -28,7 +28,7 @@ import
org.apache.spark.sql.execution.streaming.runtime.{MemoryStream, Streaming
import
org.apache.spark.sql.execution.streaming.state.{AlsoTestWithEncodingTypes,
AlsoTestWithRocksDBFeatures, EnableStateStoreRowChecksum, RocksDBFileManager,
RocksDBStateStoreProvider, TestClass}
import org.apache.spark.sql.functions.{col, explode, timestamp_seconds}
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.streaming.{InputMapRow, ListState, MapInputEvent,
MapOutputEvent, MapStateTTLProcessor, MaxEventTimeStatefulProcessor,
OutputMode, RunningCountStatefulProcessor,
RunningCountStatefulProcessorWithProcTimeTimerUpdates, StatefulProcessor,
StateStoreMetricsTest, TestMapStateProcessor, TimeMode, TimerValues,
TransformWithStateSuiteUtils, Trigger, TTLConfig, ValueState}
+import org.apache.spark.sql.streaming.{InputMapRow, ListState, MapInputEvent,
MapOutputEvent, MapStateTTLProcessor, MaxEventTimeStatefulProcessor,
OutputMode, RunningCountStatefulProcessor,
RunningCountStatefulProcessorWithProcTimeTimerUpdates, StatefulProcessor,
TestMapStateProcessor, TimeMode, TimerValues, TransformWithStateSuiteUtils,
Trigger, TTLConfig, ValueState}
import org.apache.spark.sql.streaming.util.StreamManualClock
import org.apache.spark.tags.{ExtendedSQLTest, SlowSQLTest}
import org.apache.spark.util.Utils
@@ -128,7 +128,7 @@ class SessionGroupsStatefulProcessorWithTTL extends
* Test suite to verify integration of state data source reader with the
transformWithState operator
*/
@SlowSQLTest
-class StateDataSourceTransformWithStateSuite extends StateStoreMetricsTest
+class StateDataSourceTransformWithStateSuite extends StateDataSourceTestBase
with AlsoTestWithEncodingTypes with AlsoTestWithRocksDBFeatures {
import testImplicits._
@@ -1134,12 +1134,15 @@ class StateDataSourceTransformWithStateSuite extends
StateStoreMetricsTest
} else {
dfsRootDir.getAbsolutePath + "/3.changelog"
}
- Utils.deleteRecursively(new File(changelogFilePath))
-
- // Write the retained entry back to the changelog
- val changelogWriter = fileManager.getChangeLogWriter(3, false,
checkpointUniqueId, lineage)
- changelogWriter.put(retainEntry._2, retainEntry._3)
- changelogWriter.commit()
+ withWritableCheckpoint {
+ Utils.deleteRecursively(new File(changelogFilePath))
+
+ // Write the retained entry back to the changelog
+ val changelogWriter = fileManager.getChangeLogWriter(
+ 3, false, checkpointUniqueId, lineage)
+ changelogWriter.put(retainEntry._2, retainEntry._3)
+ changelogWriter.commit()
+ }
// Ensure that we have only one entry in the changelog for version 3
// For this test - key 9 is retained and key 12 is deleted
@@ -1199,6 +1202,65 @@ class StateDataSourceTransformWithStateSuite extends
StateStoreMetricsTest
}
}
}
+
+ /**
+ * Recursively removes write permission from a directory tree, simulating a
+ * read-only filesystem (e.g. a read-only cloud object store with only READ
permission).
+ */
+ private def makeReadOnly(dir: java.io.File): Unit = {
+ if (dir.isDirectory) {
+ dir.listFiles().foreach(makeReadOnly)
+ }
+ dir.setWritable(false)
+ }
+
+ /** Recursively restores write permission to a directory tree. */
+ private def makeWritable(dir: java.io.File): Unit = {
+ dir.setWritable(true)
+ if (dir.isDirectory) {
+ dir.listFiles().foreach(makeWritable)
+ }
+ }
+
+ test("SPARK-57269: TWS state data source read succeeds on read-only
checkpoint") {
+ withTempDir { tempDir =>
+ withSQLConf(SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
+ classOf[RocksDBStateStoreProvider].getName,
+ SQLConf.SHUFFLE_PARTITIONS.key ->
+ TransformWithStateSuiteUtils.NUM_SHUFFLE_PARTITIONS.toString) {
+ val checkpointPath = tempDir.getAbsolutePath
+ val inputData = MemoryStream[String]
+ val result = inputData.toDS()
+ .groupByKey(x => x)
+ .transformWithState(new RunningCountStatefulProcessor(),
+ TimeMode.None(),
+ OutputMode.Update())
+
+ testStream(result, OutputMode.Update())(
+ StartStream(checkpointLocation = checkpointPath),
+ AddData(inputData, "a"),
+ AddData(inputData, "b"),
+ ProcessAllAvailable(),
+ StopStream
+ )
+
+ val stateDir = new java.io.File(tempDir, "state")
+ assert(stateDir.exists(), "State directory should exist after running
the query")
+ makeReadOnly(stateDir)
+
+ try {
+ val df = spark.read
+ .format("statestore")
+ .option(StateSourceOptions.PATH, checkpointPath)
+ .option(StateSourceOptions.STATE_VAR_NAME, "countState")
+ .load()
+ assert(df.collect().length > 0)
+ } finally {
+ makeWritable(stateDir)
+ }
+ }
+ }
+ }
}
@ExtendedSQLTest
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/WriteProtectedLocalFileSystem.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/WriteProtectedLocalFileSystem.scala
new file mode 100644
index 000000000000..2b712d14af70
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/WriteProtectedLocalFileSystem.scala
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.v2.state
+
+import java.net.URI
+import java.nio.file.AccessDeniedException
+import java.util
+import java.util.concurrent.atomic.AtomicReference
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{CreateFlag, DelegateToFileSystem,
FSDataOutputStream, FsServerDefaults, Options, Path}
+import org.apache.hadoop.fs.local.LocalConfigKeys
+import org.apache.hadoop.fs.permission.FsPermission
+import org.apache.hadoop.util.Progressable
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.streaming.StreamTest
+
+/**
+ * Write protection for the Hadoop FileSystem API (`fs.file.impl`).
+ *
+ * Extends [[org.apache.spark.DebugFilesystem]] so it replaces DebugFilesystem
+ * without losing open-connection tracking. Every write entry point is
overridden to deny
+ * the operation when the target path is protected; reads are left untouched.
+ */
+class WriteProtectedLocalFileSystem extends org.apache.spark.DebugFilesystem {
+
+ override def create(
+ f: Path,
+ permission: FsPermission,
+ overwrite: Boolean,
+ bufferSize: Int,
+ replication: Short,
+ blockSize: Long,
+ progress: Progressable): FSDataOutputStream = {
+ if (WriteProtectedPaths.isProtected(f)) WriteProtectedPaths.deny("create",
f)
+ super.create(f, permission, overwrite, bufferSize, replication, blockSize,
progress)
+ }
+
+ override def create(
+ f: Path,
+ permission: FsPermission,
+ flags: util.EnumSet[CreateFlag],
+ bufferSize: Int,
+ replication: Short,
+ blockSize: Long,
+ progress: Progressable,
+ checksumOpt: Options.ChecksumOpt): FSDataOutputStream = {
+ if (WriteProtectedPaths.isProtected(f)) WriteProtectedPaths.deny("create",
f)
+ super.create(
+ f, permission, flags, bufferSize, replication, blockSize, progress,
checksumOpt)
+ }
+
+ override def createNonRecursive(
+ f: Path,
+ permission: FsPermission,
+ flags: util.EnumSet[CreateFlag],
+ bufferSize: Int,
+ replication: Short,
+ blockSize: Long,
+ progress: Progressable): FSDataOutputStream = {
+ if (WriteProtectedPaths.isProtected(f))
WriteProtectedPaths.deny("createNonRecursive", f)
+ super.createNonRecursive(f, permission, flags, bufferSize, replication,
blockSize, progress)
+ }
+
+ override def append(f: Path, bufferSize: Int, progress: Progressable):
FSDataOutputStream = {
+ if (WriteProtectedPaths.isProtected(f)) WriteProtectedPaths.deny("append",
f)
+ super.append(f, bufferSize, progress)
+ }
+
+ override def rename(src: Path, dst: Path): Boolean = {
+ // A rename out of a protected path is also a write to that path -- check
both ends.
+ if (WriteProtectedPaths.isProtected(src)) WriteProtectedPaths.deny("rename
from", src)
+ if (WriteProtectedPaths.isProtected(dst)) WriteProtectedPaths.deny("rename
to", dst)
+ super.rename(src, dst)
+ }
+
+ override def delete(f: Path, recursive: Boolean): Boolean = {
+ if (WriteProtectedPaths.isProtected(f)) WriteProtectedPaths.deny("delete",
f)
+ super.delete(f, recursive)
+ }
+
+ override def mkdirs(f: Path, permission: FsPermission): Boolean = {
+ if (WriteProtectedPaths.isProtected(f)) WriteProtectedPaths.deny("mkdirs",
f)
+ super.mkdirs(f, permission)
+ }
+
+ override def truncate(f: Path, newLength: Long): Boolean = {
+ if (WriteProtectedPaths.isProtected(f))
WriteProtectedPaths.deny("truncate", f)
+ super.truncate(f, newLength)
+ }
+
+ override def setTimes(f: Path, mtime: Long, atime: Long): Unit = {
+ if (WriteProtectedPaths.isProtected(f))
WriteProtectedPaths.deny("setTimes", f)
+ super.setTimes(f, mtime, atime)
+ }
+}
+
+/**
+ * Write protection for the Hadoop FileContext/AbstractFileSystem API
+ * (`fs.AbstractFileSystem.file.impl`).
+ *
+ * Needed because FileContextBasedCheckpointFileManager goes through
FileContext, which
+ * bypasses the FileSystem-level write methods overridden in
[[WriteProtectedLocalFileSystem]].
+ */
+class WriteProtectedAbstractFileSystem(uri: URI, conf: Configuration)
+ extends DelegateToFileSystem(
+ uri,
+ new WriteProtectedLocalFileSystem,
+ conf,
+ "file",
+ false) {
+
+ override def getUriDefaultPort(): Int = -1
+ override def getServerDefaults(): FsServerDefaults =
LocalConfigKeys.getServerDefaults
+ override def isValidName(src: String): Boolean = true
+
+ override def createInternal(
+ f: Path,
+ flag: util.EnumSet[CreateFlag],
+ absolutePermission: FsPermission,
+ bufferSize: Int,
+ replication: Short,
+ blockSize: Long,
+ progress: Progressable,
+ checksumOpt: Options.ChecksumOpt,
+ createParent: Boolean): FSDataOutputStream = {
+ if (WriteProtectedPaths.isProtected(f)) WriteProtectedPaths.deny("create",
f)
+ super.createInternal(
+ f, flag, absolutePermission, bufferSize, replication, blockSize,
+ progress, checksumOpt, createParent)
+ }
+
+ override def mkdir(
+ dir: Path, permission: FsPermission, createParent: Boolean): Unit = {
+ if (WriteProtectedPaths.isProtected(dir))
WriteProtectedPaths.deny("mkdir", dir)
+ super.mkdir(dir, permission, createParent)
+ }
+
+ override def delete(f: Path, recursive: Boolean): Boolean = {
+ if (WriteProtectedPaths.isProtected(f)) WriteProtectedPaths.deny("delete",
f)
+ super.delete(f, recursive)
+ }
+
+ override def renameInternal(src: Path, dst: Path): Unit = {
+ // A rename out of a protected path is also a write to that path -- check
both ends.
+ if (WriteProtectedPaths.isProtected(src)) WriteProtectedPaths.deny("rename
from", src)
+ if (WriteProtectedPaths.isProtected(dst)) WriteProtectedPaths.deny("rename
to", dst)
+ super.renameInternal(src, dst)
+ }
+
+ override def truncate(f: Path, newLength: Long): Boolean = {
+ if (WriteProtectedPaths.isProtected(f))
WriteProtectedPaths.deny("truncate", f)
+ super.truncate(f, newLength)
+ }
+
+ override def setTimes(f: Path, mtime: Long, atime: Long): Unit = {
+ if (WriteProtectedPaths.isProtected(f))
WriteProtectedPaths.deny("setTimes", f)
+ super.setTimes(f, mtime, atime)
+ }
+}
+
+/**
+ * Shared state for write-protected paths. Used by both
[[WriteProtectedLocalFileSystem]]
+ * (FileSystem API) and [[WriteProtectedAbstractFileSystem]] (FileContext API).
+ *
+ * State is JVM-global: parallel suites in the same JVM would interfere. Tests
in this
+ * package run serially via [[StateDataSourceTestBase]], which clears state in
`afterEach`.
+ *
+ * Path normalization: prefixes are stored with a trailing "/", and matches use
+ * `pathStr.startsWith(prefix)`. Operations on the protected directory itself
+ * (without trailing "/") are NOT matched -- only operations on its
descendants.
+ */
+object WriteProtectedPaths {
+ // Single AtomicReference so snapshot/restore are atomic; readers do one
volatile load.
+ private val protectedPrefixes =
+ new AtomicReference[Set[String]](Set.empty[String])
+
+ private def normalize(path: String): String =
+ if (path.endsWith("/")) path else path + "/"
+
+ private[state] def deny(op: String, p: Path): Nothing = {
+ throw new AccessDeniedException(
+ s"Write-protected path: $op attempted on path: $p. " +
+ "This path is protected to simulate a read-only cloud object store
checkpoint. " +
+ "If you are seeing this in a test, it means the data source code path " +
+ "is attempting a write that would fail on a read-only cloud object store
path.")
+ }
+
+ def protectPath(path: String): Unit = {
+ val normalized = normalize(path)
+ protectedPrefixes.updateAndGet(_ + normalized)
+ }
+
+ def unprotectPath(path: String): Unit = {
+ val normalized = normalize(path)
+ protectedPrefixes.updateAndGet(_ - normalized)
+ }
+
+ def isProtected(p: Path): Boolean = {
+ val current = protectedPrefixes.get()
+ if (current.isEmpty) return false
+ val pathStr = p.toUri.getPath
+ current.exists(pathStr.startsWith)
+ }
+
+ def clearAll(): Unit = protectedPrefixes.set(Set.empty)
+
+ /** Snapshot current prefixes and atomically clear -- single store update. */
+ def snapshotAndClear(): java.util.Set[String] =
+ new util.HashSet(protectedPrefixes.getAndSet(Set.empty).asJava)
+
+ /**
+ * Re-add the snapshot's entries. Non-destructive: prefixes added
concurrently by other
+ * code while the snapshot was inactive are preserved.
+ */
+ def restore(snapshot: java.util.Set[String]): Unit = {
+ val toAdd = snapshot.asScala.toSet
+ protectedPrefixes.updateAndGet(_ ++ toAdd)
+ }
+}
+
+/**
+ * Mix-in that installs [[WriteProtectedLocalFileSystem]] and auto-protects
every
+ * [[withTempDir]] from writes. Tests that perform writes outside of
[[testStream]]
+ * (e.g. provider API calls, log purging, direct file manipulation) must wrap
them in
+ * [[withWritableCheckpoint]] to temporarily suspend protection.
+ *
+ * Mix this into any test base whose reads should never write to the
checkpoint path
+ * (state data source, ...). The framework is JVM-global and assumes a single
suite runs
+ * at a time within a JVM.
+ */
+trait WriteProtectedCheckpointTestMixin extends StreamTest with
BeforeAndAfterEach {
+
+ // Set the write-protected filesystem via sparkConf so it is applied before
the
+ // SparkContext is created. Setting it later (e.g., on the session) is too
late --
+ // the Hadoop FileSystem cache may already hold a DebugFilesystem from
startup.
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set("spark.hadoop.fs.file.impl",
+ classOf[WriteProtectedLocalFileSystem].getCanonicalName)
+ .set("spark.hadoop.fs.file.impl.disable.cache", "true")
+ .set("spark.hadoop.fs.AbstractFileSystem.file.impl",
+ classOf[WriteProtectedAbstractFileSystem].getCanonicalName)
+ .set("spark.hadoop.fs.AbstractFileSystem.file.impl.disable.cache",
"true")
+ }
+
+ override def afterEach(): Unit = {
+ // Clear protections so teardown writes (state-store maintenance, temp-dir
cleanup)
+ // are not falsely flagged.
+ WriteProtectedPaths.clearAll()
+ super.afterEach()
+ }
+
+ /**
+ * Override withTempDir to automatically protect the directory from writes.
+ * [[testStream]] temporarily suspends protections for streaming query
writes.
+ * Everything else (data source reads) is blocked from writing.
+ */
+ override def withTempDir(f: java.io.File => Unit): Unit = {
+ super.withTempDir { dir =>
+ // Protect both absolute and canonical paths since data sources resolve
paths
+ // via makeQualified which may canonicalize.
+ WriteProtectedPaths.protectPath(dir.getAbsolutePath)
+ WriteProtectedPaths.protectPath(dir.getCanonicalPath)
+ try {
+ f(dir)
+ } finally {
+ WriteProtectedPaths.unprotectPath(dir.getAbsolutePath)
+ WriteProtectedPaths.unprotectPath(dir.getCanonicalPath)
+ }
+ }
+ }
+
+ /**
+ * Suspends write protection for a block. Use this for test setup code that
writes
+ * to the checkpoint outside of [[testStream]] (e.g. provider API calls, log
purging,
+ * writeStream.start(), legacy behavior tests).
+ */
+ protected def withWritableCheckpoint[T](body: => T): T = {
+ val saved = WriteProtectedPaths.snapshotAndClear()
+ try {
+ body
+ } finally {
+ WriteProtectedPaths.restore(saved)
+ }
+ }
+
+ /**
+ * Assert that a write-protection-related exception exists in the cause
chain.
+ *
+ * Accepted shapes:
+ * - AccessDeniedException: thrown directly by
[[WriteProtectedLocalFileSystem]] /
+ * [[WriteProtectedAbstractFileSystem]].
+ * - "FAILED_TO_GET_CHANGELOG_WRITER" error class (CANNOT_LOAD_STATE_STORE
in
+ * [[org.apache.spark.sql.errors.QueryExecutionErrors]]): RocksDB wraps
the
+ * underlying AccessDeniedException when changelog-writer creation fails.
+ */
+ protected def assertWriteProtectionFailure(e: Throwable): Unit = {
+ def matches(t: Throwable): Boolean = {
+ val msg = Option(t.getMessage).getOrElse("")
+ t.isInstanceOf[java.nio.file.AccessDeniedException] ||
+ msg.contains("Write-protected") ||
+ msg.contains("FAILED_TO_GET_CHANGELOG_WRITER")
+ }
+ val found = Iterator.iterate[Throwable](e)(_.getCause).takeWhile(_ !=
null).exists(matches)
+ assert(found,
+ s"Expected write-protection failure in cause chain, got: ${e.getClass}:
${e.getMessage}")
+ }
+
+ /**
+ * Override testStream to temporarily suspend all protections during
streaming query
+ * execution. Streaming queries write checkpoints, state, and metadata.
After the
+ * query completes, protections are restored so subsequent reads are checked.
+ */
+ override def testStream(
+ _stream: org.apache.spark.sql.Dataset[_],
+ outputMode: org.apache.spark.sql.streaming.OutputMode =
+ org.apache.spark.sql.streaming.OutputMode.Append,
+ extraOptions: Map[String, String] = Map.empty,
+ sink: org.apache.spark.sql.execution.streaming.sources.MemorySink =
+ new org.apache.spark.sql.execution.streaming.sources.MemorySink())(
+ actions: StreamAction*): Unit = {
+ val saved = WriteProtectedPaths.snapshotAndClear()
+ try {
+ super.testStream(_stream, outputMode, extraOptions, sink)(actions: _*)
+ } finally {
+ WriteProtectedPaths.restore(saved)
+ }
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionIntegrationSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionIntegrationSuite.scala
index caf60d51ed62..26e94102a5c0 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionIntegrationSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionIntegrationSuite.scala
@@ -148,8 +148,10 @@ abstract class OfflineStateRepartitionIntegrationSuiteBase
extends StateDataSour
}
// Step 3: Run repartition
- spark.streamingCheckpointManager.repartition(
- checkpointDir.getAbsolutePath, newPartitions)
+ withWritableCheckpoint {
+ spark.streamingCheckpointManager.repartition(
+ checkpointDir.getAbsolutePath, newPartitions)
+ }
val repartitionBatchId = lastBatchId + 1
val hadoopConf = spark.sessionState.newHadoopConf()
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OperatorStateMetadataSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OperatorStateMetadataSuite.scala
index 5fa8d86c21af..d7ae82e97e04 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OperatorStateMetadataSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OperatorStateMetadataSuite.scala
@@ -21,7 +21,7 @@ import org.apache.hadoop.fs.Path
import org.apache.spark.SparkRuntimeException
import org.apache.spark.sql.{Column, Row}
-import
org.apache.spark.sql.execution.datasources.v2.state.{StateDataSourceUnspecifiedRequiredOption,
StateSourceOptions}
+import
org.apache.spark.sql.execution.datasources.v2.state.{StateDataSourceUnspecifiedRequiredOption,
StateSourceOptions, WriteProtectedCheckpointTestMixin}
import
org.apache.spark.sql.execution.streaming.checkpointing.{CheckpointFileManager,
OffsetSeq, OffsetSeqLog}
import org.apache.spark.sql.execution.streaming.runtime.{LongOffset,
MemoryStream}
import
org.apache.spark.sql.execution.streaming.runtime.StreamingCheckpointConstants.DIR_NAME_OFFSETS
@@ -30,7 +30,7 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.streaming.{OutputMode,
RunningCountStatefulProcessor, StreamTest, TimeMode}
import org.apache.spark.sql.streaming.OutputMode.{Complete, Update}
-class OperatorStateMetadataSuite extends StreamTest {
+class OperatorStateMetadataSuite extends StreamTest with
WriteProtectedCheckpointTestMixin {
import testImplicits._
private lazy val hadoopConf = spark.sessionState.newHadoopConf()
@@ -70,15 +70,19 @@ class OperatorStateMetadataSuite extends StreamTest {
test("Serialize and deserialize stateful operator metadata") {
withTempDir { checkpointDir =>
- val offsetLog = new OffsetSeqLog(spark,
- new Path(checkpointDir.toString, DIR_NAME_OFFSETS).toString)
- val batch0 = OffsetSeq.fill(LongOffset(0), LongOffset(1), LongOffset(2))
- offsetLog.add(0, batch0)
- val statePath = new Path(checkpointDir.toString, "state/0")
- val stateStoreInfo = (1 to 4).map(i => StateStoreMetadataV1(s"store$i",
1, 200))
- val operatorInfo = OperatorInfoV1(1, "Join")
- val operatorMetadata = OperatorStateMetadataV1(operatorInfo,
stateStoreInfo.toArray)
- new OperatorStateMetadataV1Writer(statePath,
hadoopConf).write(operatorMetadata)
+ // Test setup writes the offset log and operator metadata directly to
the checkpoint.
+ val operatorMetadata = withWritableCheckpoint {
+ val offsetLog = new OffsetSeqLog(spark,
+ new Path(checkpointDir.toString, DIR_NAME_OFFSETS).toString)
+ val batch0 = OffsetSeq.fill(LongOffset(0), LongOffset(1),
LongOffset(2))
+ offsetLog.add(0, batch0)
+ val statePath = new Path(checkpointDir.toString, "state/0")
+ val stateStoreInfo = (1 to 4).map(i =>
StateStoreMetadataV1(s"store$i", 1, 200))
+ val operatorInfo = OperatorInfoV1(1, "Join")
+ val metadata = OperatorStateMetadataV1(operatorInfo,
stateStoreInfo.toArray)
+ new OperatorStateMetadataV1Writer(statePath,
hadoopConf).write(metadata)
+ metadata
+ }
checkOperatorStateMetadata(checkpointDir.toString, 0, operatorMetadata)
val df = spark.read.format("state-metadata").load(checkpointDir.toString)
// Commit log is empty, there is no available batch id.
@@ -413,10 +417,12 @@ class OperatorStateMetadataSuite extends StreamTest {
StopStream
)
- // Delete operator metadata path
- val metadataPath = new Path(checkpointDir.toString,
s"state/0/_metadata/metadata")
- val fm = CheckpointFileManager.create(new
Path(checkpointDir.getCanonicalPath), hadoopConf)
- fm.delete(metadataPath)
+ // Delete operator metadata path (test setup mutates the checkpoint).
+ withWritableCheckpoint {
+ val metadataPath = new Path(checkpointDir.toString,
s"state/0/_metadata/metadata")
+ val fm = CheckpointFileManager.create(new
Path(checkpointDir.getCanonicalPath), hadoopConf)
+ fm.delete(metadataPath)
+ }
// Restart the query
testStream(aggregated, Complete)(
@@ -484,10 +490,13 @@ class OperatorStateMetadataSuite extends StreamTest {
ProcessAllAvailable(),
StopStream)
- // Delete the state directory to simulate deleted state files
- val stateDir = new Path(checkpointDir.toString, "state")
- val fileManager = CheckpointFileManager.create(stateDir, hadoopConf)
- fileManager.delete(stateDir)
+ // Delete the state directory to simulate deleted state files (test
setup mutates
+ // the checkpoint).
+ withWritableCheckpoint {
+ val stateDir = new Path(checkpointDir.toString, "state")
+ val fileManager = CheckpointFileManager.create(stateDir, hadoopConf)
+ fileManager.delete(stateDir)
+ }
// Restart the query - should fail with empty state directory error
testStream(stream.dropDuplicates())(
@@ -520,11 +529,14 @@ class OperatorStateMetadataSuite extends StreamTest {
ProcessAllAvailable(),
StopStream)
- // Delete the state directory if it exists (it shouldn't for stateless
query)
- val stateDir = new Path(checkpointDir.toString, "state")
- val fileManager = CheckpointFileManager.create(stateDir, hadoopConf)
- if (fileManager.exists(stateDir)) {
- fileManager.delete(stateDir)
+ // Delete the state directory if it exists (it shouldn't for stateless
query).
+ // Wrapped so the test-setup delete bypasses write protection.
+ withWritableCheckpoint {
+ val stateDir = new Path(checkpointDir.toString, "state")
+ val fileManager = CheckpointFileManager.create(stateDir, hadoopConf)
+ if (fileManager.exists(stateDir)) {
+ fileManager.delete(stateDir)
+ }
}
// Restart with a stateful operator added - should fail
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala
index 22d0af0a77fd..66617a4d0908 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala
@@ -83,24 +83,29 @@ class StatePartitionAllColumnFamiliesWriterSuite extends
StateDataSourceTestBase
val lastBatch = targetCheckpointMetadata.commitLog.getLatestBatchId().get
val targetOffsetSeq = targetCheckpointMetadata.offsetLog.get(lastBatch).get
val writeBatchId = lastBatch + 1
- targetCheckpointMetadata.offsetLog.add(writeBatchId, targetOffsetSeq)
-
- val rewriter = new StateRewriter(
- spark,
- readBatchId,
- writeBatchId,
- targetCpLocation,
- hadoopConf,
- readResolvedCheckpointLocation = Some(sourceCpLocation),
- transformFunc = None,
- writeCheckpointMetadata = Some(targetCheckpointMetadata)
- )
- val checkpointInfos = rewriter.run()
-
- // Commit to commitLog with checkpoint IDs
- val latestCommit = targetCheckpointMetadata.commitLog.get(lastBatch).get
- val commitMetadata = latestCommit.withStateUniqueIds(checkpointInfos)
- targetCheckpointMetadata.commitLog.add(writeBatchId, commitMetadata)
+ // Wrap the offset/commit-log writes and StateRewriter.run() (which writes
state files
+ // and metadata into the target checkpoint) so they bypass write
protection on the
+ // target temp dir. Subsequent state-data-source reads stay protected.
+ withWritableCheckpoint {
+ targetCheckpointMetadata.offsetLog.add(writeBatchId, targetOffsetSeq)
+
+ val rewriter = new StateRewriter(
+ spark,
+ readBatchId,
+ writeBatchId,
+ targetCpLocation,
+ hadoopConf,
+ readResolvedCheckpointLocation = Some(sourceCpLocation),
+ transformFunc = None,
+ writeCheckpointMetadata = Some(targetCheckpointMetadata)
+ )
+ val checkpointInfos = rewriter.run()
+
+ // Commit to commitLog with checkpoint IDs
+ val latestCommit = targetCheckpointMetadata.commitLog.get(lastBatch).get
+ val commitMetadata = latestCommit.withStateUniqueIds(checkpointInfos)
+ targetCheckpointMetadata.commitLog.add(writeBatchId, commitMetadata)
+ }
val versionToCheck = writeBatchId + 1
storeToColumnFamilies.foreach { case (storeName, columnFamilies) =>
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala
index 65009a2c1c6b..b3d85c855e2f 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala
@@ -1707,6 +1707,31 @@ class StateStoreSuite extends
StateStoreSuiteBase[HDFSBackedStateStoreProvider]
filePath.createNewFile()
}
+ // The trait return type of replayStateFromSnapshot is the writable
StateStore. On
+ // HDFS, the writable and read-only stores are separate classes, so a
readOnly=true
+ // call cannot honor the flag through this method. The provider rejects it
and directs
+ // callers to replayReadStateFromSnapshot. (RocksDB's StateStore class is
unified and
+ // respects the flag internally, so RocksDB still implements both paths via
this entry
+ // point.)
+ test("replayStateFromSnapshot rejects readOnly=true (use
replayReadStateFromSnapshot)") {
+ tryWithProviderResource(newStoreProvider(opId = Random.nextInt(),
partition = 0,
+ minDeltasForSnapshot = 1)) { provider =>
+ val s1 = provider.getStore(0)
+ put(s1, "a", 1, 1)
+ s1.commit()
+ val s2 = provider.getStore(1)
+ put(s2, "a", 2, 2)
+ s2.commit()
+ provider.doMaintenance()
+
+ val ex = intercept[IllegalArgumentException] {
+ provider.asInstanceOf[SupportsFineGrainedReplay]
+ .replayStateFromSnapshot(1, 2, readOnly = true)
+ }
+ assert(ex.getMessage.contains("replayReadStateFromSnapshot"))
+ }
+ }
+
override protected def testQuietly(name: String)(f: => Unit): Unit = {
// Use the implementation from StateStoreSuiteBase.
// There is another in QueryTest. Doing this to avoid conflict error.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
