diff --git a/backend/src/main/kotlin/com/opendatamask/application/service/CustomDataMappingService.kt b/backend/src/main/kotlin/com/opendatamask/application/service/CustomDataMappingService.kt
index 581fb4e..34ccaf2 100644
--- a/backend/src/main/kotlin/com/opendatamask/application/service/CustomDataMappingService.kt
+++ b/backend/src/main/kotlin/com/opendatamask/application/service/CustomDataMappingService.kt
@@ -27,7 +27,8 @@ class CustomDataMappingService(
             columnName = request.columnName,
             action = request.action,
             maskingStrategy = if (request.action == MappingAction.MASK) request.maskingStrategy else null,
-            fakeGeneratorType = if (request.action == MappingAction.MASK) request.fakeGeneratorType else null
+            fakeGeneratorType = if (request.action == MappingAction.MASK) request.fakeGeneratorType else null,
+            piiRuleParams = if (request.action == MappingAction.MASK) request.piiRuleParams else null
         )
         return customDataMappingRepository.save(mapping).toResponse()
     }
@@ -64,6 +65,7 @@ class CustomDataMappingService(
         mapping.action = request.action
         mapping.maskingStrategy = if (request.action == MappingAction.MASK) request.maskingStrategy else null
         mapping.fakeGeneratorType = if (request.action == MappingAction.MASK) request.fakeGeneratorType else null
+        mapping.piiRuleParams = if (request.action == MappingAction.MASK) request.piiRuleParams else null
         return customDataMappingRepository.save(mapping).toResponse()
     }
 
@@ -90,7 +92,8 @@ class CustomDataMappingService(
                 columnName = entry.columnName,
                 action = entry.action,
                 maskingStrategy = if (entry.action == MappingAction.MASK) entry.maskingStrategy else null,
-                fakeGeneratorType = if (entry.action == MappingAction.MASK) entry.fakeGeneratorType else null
+                fakeGeneratorType = if (entry.action == MappingAction.MASK) entry.fakeGeneratorType else null,
+                piiRuleParams = if (entry.action == MappingAction.MASK) entry.piiRuleParams else null
             )
         }
         return mappings.map { customDataMappingRepository.save(it).toResponse() }
@@ -129,6 +132,7 @@ class CustomDataMappingService(
         action = action,
         maskingStrategy = maskingStrategy,
         fakeGeneratorType = fakeGeneratorType,
+        piiRuleParams = piiRuleParams,
         createdAt = createdAt,
         updatedAt = updatedAt
     )
diff --git a/backend/src/main/kotlin/com/opendatamask/application/service/DefaultRuleRegistry.kt b/backend/src/main/kotlin/com/opendatamask/application/service/DefaultRuleRegistry.kt
new file mode 100644
index 0000000..bbe2d22
--- /dev/null
+++ b/backend/src/main/kotlin/com/opendatamask/application/service/DefaultRuleRegistry.kt
@@ -0,0 +1,40 @@
+package com.opendatamask.application.service
+
+import com.opendatamask.domain.model.HashRule
+import com.opendatamask.domain.model.PIIMaskingRule
+import com.opendatamask.domain.model.PartialMaskRule
+import com.opendatamask.domain.model.PassThroughRule
+import com.opendatamask.domain.model.RedactRule
+import com.opendatamask.domain.port.output.RuleRegistryPort
+import org.springframework.stereotype.Service
+import java.util.concurrent.ConcurrentHashMap
+
+// Holds all known PIIMaskingRule implementations.
+// Built-in non-parameterized rules are registered at construction time.
+// RegexRule is not registered as a default instance because it requires caller-supplied
+// pattern and replacement parameters; PIIMaskingService constructs RegexRule instances
+// directly from piiRuleParams. To override regex behavior globally, registerCustomRule()
+// can be used with a concrete PIIMaskingRule that has ruleId "regex" or any custom ID.
+// Call registerCustomRule() to inject additional business-specific rules at runtime.
+@Service
+class DefaultRuleRegistry : RuleRegistryPort {
+
+    private val registry = ConcurrentHashMap<String, PIIMaskingRule>()
+
+    init {
+        register(PassThroughRule())
+        register(RedactRule())
+        register(PartialMaskRule())
+        register(HashRule())
+    }
+
+    private fun register(rule: PIIMaskingRule) {
+        registry[rule.ruleId] = rule
+    }
+
+    override fun getRule(ruleId: String): PIIMaskingRule? = registry[ruleId]
+
+    override fun getAllRuleIds(): Set<String> = registry.keys.toSet()
+
+    override fun registerCustomRule(rule: PIIMaskingRule) = register(rule)
+}
diff --git a/backend/src/main/kotlin/com/opendatamask/application/service/JobService.kt b/backend/src/main/kotlin/com/opendatamask/application/service/JobService.kt
index 7d3c5c1..c0ac210 100644
--- a/backend/src/main/kotlin/com/opendatamask/application/service/JobService.kt
+++ b/backend/src/main/kotlin/com/opendatamask/application/service/JobService.kt
@@ -37,7 +37,8 @@ class JobService(
     private val destinationSchemaService: DestinationSchemaService,
     private val postJobActionService: PostJobActionService,
     private val schemaChangeService: SchemaChangeService,
-    private val webhookService: WebhookService
+    private val webhookService: WebhookService,
+    private val piiMaskingService: PIIMaskingService
 ) : JobUseCase {
     @org.springframework.beans.factory.annotation.Autowired(required = false)
     private var subsetExecutionService: SubsetExecutionService? = null
@@ -178,7 +179,7 @@ class JobService(
                     tableConfig.tableName,
                     selectedAttrs
                 )
-                processTable(job.id, tableConfig, sourceConnector, destConnector, subsetRows)
+                processTable(job.id, tableConfig, sourceConnector, destConnector, subsetRows, job.workspaceId, sourceConn.id)
             }
 
             updateJobStatus(job, JobStatus.COMPLETED)
@@ -206,7 +207,9 @@ class JobService(
         tableConfig: TableConfiguration,
         sourceConnector: DatabaseConnector,
         destConnector: DatabaseConnector,
-        preComputedRows: Map<String, List<Map<String, Any?>>> = emptyMap()
+        preComputedRows: Map<String, List<Map<String, Any?>>> = emptyMap(),
+        workspaceId: Long? = null,
+        sourceConnectionId: Long? = null
     ) {
         addLog(jobId, "Processing table: ${tableConfig.tableName} (mode: ${tableConfig.mode})", LogLevel.INFO)
 
@@ -217,7 +220,12 @@ class JobService(
             TableMode.PASSTHROUGH -> {
                 val data = sourceConnector.fetchData(tableConfig.tableName, tableConfig.rowLimit?.toInt(), null, selectedAttrs)
                 addLog(jobId, "Fetched ${data.size} rows from ${tableConfig.tableName}", LogLevel.INFO)
-                val written = destConnector.writeData(tableConfig.tableName, data)
+                val transformed = if (workspaceId != null && sourceConnectionId != null) {
+                    piiMaskingService.applyMappingsToRows(workspaceId, sourceConnectionId, tableConfig.tableName, data)
+                } else {
+                    data
+                }
+                val written = destConnector.writeData(tableConfig.tableName, transformed)
                 addLog(jobId, "Wrote $written rows to destination ${tableConfig.tableName}", LogLevel.INFO)
             }
             TableMode.MASK -> {
diff --git a/backend/src/main/kotlin/com/opendatamask/application/service/PIIMaskingService.kt b/backend/src/main/kotlin/com/opendatamask/application/service/PIIMaskingService.kt
new file mode 100644
index 0000000..08de6c5
--- /dev/null
+++ b/backend/src/main/kotlin/com/opendatamask/application/service/PIIMaskingService.kt
@@ -0,0 +1,136 @@
+package com.opendatamask.application.service
+
+import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
+import com.fasterxml.jackson.module.kotlin.readValue
+import com.opendatamask.domain.model.CustomDataMapping
+import com.opendatamask.domain.model.HashRule
+import com.opendatamask.domain.model.MappingAction
+import com.opendatamask.domain.model.MaskingStrategy
+import com.opendatamask.domain.model.PartialMaskRule
+import com.opendatamask.domain.model.RegexRule
+import com.opendatamask.domain.port.output.CustomDataMappingPort
+import com.opendatamask.domain.port.output.RuleRegistryPort
+import org.slf4j.LoggerFactory
+import org.springframework.stereotype.Service
+
+// Transforms source data rows by applying the PIIMaskingRule selected by each
+// CustomDataMapping entry. Columns without a mapping pass through unchanged.
+@Service
+class PIIMaskingService(
+    private val ruleRegistry: RuleRegistryPort,
+    private val customDataMappingPort: CustomDataMappingPort
+) {
+    private val logger = LoggerFactory.getLogger(PIIMaskingService::class.java)
+    private val mapper = jacksonObjectMapper()
+
+    // Load mappings once and apply to every row in a batch. Prefer this over
+    // the convenience overload (applyMappings with workspaceId/connectionId/tableName/row)
+    // to avoid N+1 persistence queries when processing multiple rows for the same table.
+    fun applyMappingsToRows(
+        workspaceId: Long,
+        connectionId: Long,
+        tableName: String,
+        rows: List<Map<String, Any?>>
+    ): List<Map<String, Any?>> {
+        val mappings = loadMappings(workspaceId, connectionId, tableName)
+        if (mappings.isEmpty()) return rows
+        return rows.map { row -> applyMappings(mappings, row) }
+    }
+
+    // Load the per-column mapping index for a table; result can be reused across rows.
+    fun loadMappings(
+        workspaceId: Long,
+        connectionId: Long,
+        tableName: String
+    ): Map<String, CustomDataMapping> =
+        customDataMappingPort
+            .findByWorkspaceIdAndConnectionIdAndTableName(workspaceId, connectionId, tableName)
+            .associateBy { it.columnName.lowercase() }
+
+    // Apply pre-fetched mappings to a single row. Column name matching is case-insensitive.
+    fun applyMappings(
+        mappings: Map<String, CustomDataMapping>,
+        row: Map<String, Any?>
+    ): Map<String, Any?> {
+        if (mappings.isEmpty()) return row
+        return row.mapValues { (column, value) ->
+            val mapping = mappings[column.lowercase()] ?: return@mapValues value
+            when (mapping.action) {
+                MappingAction.MIGRATE_AS_IS -> value
+                MappingAction.MASK -> applyStrategy(mapping, value)
+            }
+        }
+    }
+
+    // Convenience overload that fetches mappings from persistence on each call.
+    // Use applyMappingsToRows() instead when processing multiple rows for the same table.
+    fun applyMappings(
+        workspaceId: Long,
+        connectionId: Long,
+        tableName: String,
+        row: Map<String, Any?>
+    ): Map<String, Any?> =
+        applyMappings(loadMappings(workspaceId, connectionId, tableName), row)
+
+    private fun applyStrategy(mapping: CustomDataMapping, value: Any?): Any? {
+        val params = parseParams(mapping.piiRuleParams) ?: return value
+
+        // Allow a custom rule to be invoked by specifying its ruleId in params.
+        // This enables runtime-registered rules to be used from any mapping strategy slot.
+        val customRuleId = params["ruleId"]?.takeIf { it.isNotBlank() }
+        if (customRuleId != null) {
+            val customRule = ruleRegistry.getRule(customRuleId)
+            if (customRule != null) return customRule.mask(value)
+            logger.warn("PII rule '{}' not found in registry — passing value through unchanged", customRuleId)
+            return value
+        }
+
+        return when (mapping.maskingStrategy) {
+            MaskingStrategy.NULL -> null
+            MaskingStrategy.REDACT -> ruleRegistry.getRule("redact")?.mask(value) ?: "[REDACTED]"
+            MaskingStrategy.HASH -> {
+                val salt = params["salt"] ?: ""
+                HashRule(salt).mask(value)
+            }
+            MaskingStrategy.PARTIAL_MASK -> {
+                val keepFirst = params["keepFirst"]?.toIntOrNull() ?: 0
+                val keepLast = params["keepLast"]?.toIntOrNull() ?: 4
+                val maskChar = params["maskChar"]?.firstOrNull() ?: '*'
+                PartialMaskRule(keepFirst, keepLast, maskChar).mask(value)
+            }
+            MaskingStrategy.REGEX -> {
+                val pattern = params["pattern"]
+                val replacement = params["replacement"]
+                if (pattern.isNullOrBlank() || replacement == null) {
+                    logger.warn(
+                        "REGEX strategy for column '{}' is missing 'pattern' or 'replacement' in piiRuleParams — passing value through unchanged",
+                        mapping.columnName
+                    )
+                    return value
+                }
+                runCatching { RegexRule(pattern, replacement).mask(value) }
+                    .getOrElse { cause ->
+                        logger.warn(
+                            "REGEX strategy for column '{}' has invalid pattern '{}' — passing value through unchanged: {}",
+                            mapping.columnName, pattern, cause.message
+                        )
+                        value
+                    }
+            }
+            // FAKE strategy is handled upstream by the GeneratorService; return value unchanged.
+            MaskingStrategy.FAKE -> value
+            null -> value
+        }
+    }
+
+    // Returns null (and logs a warning) when piiRuleParams is present but cannot be parsed as JSON.
+    // Returns an empty map when piiRuleParams is absent/blank (treated as "no extra config").
+    private fun parseParams(json: String?): Map<String, String>? {
+        if (json.isNullOrBlank()) return emptyMap()
+        return runCatching { mapper.readValue<Map<String, String>>(json) }
+            .getOrElse { cause ->
+                logger.warn("Failed to parse piiRuleParams JSON '{}': {} — passing value through unchanged", json, cause.message)
+                null
+            }
+    }
+}
diff --git a/backend/src/main/kotlin/com/opendatamask/domain/model/CustomDataMapping.kt b/backend/src/main/kotlin/com/opendatamask/domain/model/CustomDataMapping.kt
index 4100d48..2bbe61a 100644
--- a/backend/src/main/kotlin/com/opendatamask/domain/model/CustomDataMapping.kt
+++ b/backend/src/main/kotlin/com/opendatamask/domain/model/CustomDataMapping.kt
@@ -8,7 +8,7 @@ enum class MappingAction {
 }
 
 enum class MaskingStrategy {
-    FAKE, HASH, NULL
+    FAKE, HASH, NULL, REDACT, PARTIAL_MASK, REGEX
 }
 
 @Entity
@@ -50,6 +50,9 @@ class CustomDataMapping(
     @Column
     var fakeGeneratorType: GeneratorType? = null,
 
+    @Column(name = "pii_rule_params", columnDefinition = "TEXT")
+    var piiRuleParams: String? = null,
+
     @Column(nullable = false)
     var createdAt: LocalDateTime = LocalDateTime.now(),
 
diff --git a/backend/src/main/kotlin/com/opendatamask/domain/model/PIIMaskingRule.kt b/backend/src/main/kotlin/com/opendatamask/domain/model/PIIMaskingRule.kt
new file mode 100644
index 0000000..051357f
--- /dev/null
+++ b/backend/src/main/kotlin/com/opendatamask/domain/model/PIIMaskingRule.kt
@@ -0,0 +1,73 @@
+package com.opendatamask.domain.model
+
+import java.security.MessageDigest
+
+// Core interface every PII masking rule must implement.
+// Each implementation must carry a stable, unique ruleId that identifies it
+// in the registry and in JSON mapping configurations.
+interface PIIMaskingRule {
+    val ruleId: String
+    fun mask(input: Any?): Any?
+}
+
+// Sealed base class for built-in rules so exhaustive when-expressions can be used
+// when dispatch on rule type is needed in the application layer.
+sealed class BuiltInPIIRule(override val ruleId: String) : PIIMaskingRule
+
+// Passes the value through unchanged. Used as the default for unmapped columns.
+class PassThroughRule : BuiltInPIIRule("pass_through") {
+    override fun mask(input: Any?): Any? = input
+}
+
+// Replaces every non-null value with the literal token [REDACTED].
+class RedactRule : BuiltInPIIRule("redact") {
+    override fun mask(input: Any?): Any? = if (input == null) null else "[REDACTED]"
+}
+
+// Masks the middle characters of a string, preserving a configurable number of
+// leading and trailing characters. Useful for partial credit-card or email masking.
+// When the input is shorter than keepFirst + keepLast, the original value is returned unchanged.
+// Negative values for keepFirst or keepLast are treated as zero.
+class PartialMaskRule(
+    val keepFirst: Int = 0,
+    val keepLast: Int = 4,
+    val maskChar: Char = '*'
+) : BuiltInPIIRule("partial_mask") {
+    override fun mask(input: Any?): Any? {
+        if (input == null) return null
+        val str = input.toString()
+        val safeKeepFirst = keepFirst.coerceAtLeast(0)
+        val safeKeepLast = keepLast.coerceAtLeast(0)
+        if (str.length <= safeKeepFirst + safeKeepLast) return str
+        val maskLen = str.length - safeKeepFirst - safeKeepLast
+        return str.take(safeKeepFirst) + maskChar.toString().repeat(maskLen) + str.takeLast(safeKeepLast)
+    }
+}
+
+// Produces a deterministic SHA-256 hex digest of the input, with an optional salt.
+class HashRule(val salt: String = "") : BuiltInPIIRule("hash") {
+    override fun mask(input: Any?): Any? {
+        if (input == null) return null
+        val digest = MessageDigest.getInstance("SHA-256")
+        val bytes = digest.digest(("$salt${input}").toByteArray(Charsets.UTF_8))
+        return bytes.joinToString("") { "%02x".format(it) }
+    }
+}
+
+// Applies a user-supplied regular expression to the string representation of the value,
+// replacing every match with the given replacement string.
+// Throws IllegalArgumentException at construction time if pattern is not a valid regex.
+class RegexRule(val pattern: String, val replacement: String) : BuiltInPIIRule("regex") {
+    private val compiledRegex: Regex = runCatching { Regex(pattern) }
+        .getOrElse { cause ->
+            throw IllegalArgumentException(
+                "Invalid regex pattern: \"$pattern\"",
+                cause
+            )
+        }
+
+    override fun mask(input: Any?): Any? {
+        if (input == null) return null
+        return compiledRegex.replace(input.toString(), replacement)
+    }
+}
diff --git a/backend/src/main/kotlin/com/opendatamask/domain/port/input/dto/CustomDataMappingDto.kt b/backend/src/main/kotlin/com/opendatamask/domain/port/input/dto/CustomDataMappingDto.kt
index fc6cc97..88d1107 100644
--- a/backend/src/main/kotlin/com/opendatamask/domain/port/input/dto/CustomDataMappingDto.kt
+++ b/backend/src/main/kotlin/com/opendatamask/domain/port/input/dto/CustomDataMappingDto.kt
@@ -23,7 +23,13 @@ data class CustomDataMappingRequest(
 
     val maskingStrategy: MaskingStrategy? = null,
 
-    val fakeGeneratorType: GeneratorType? = null
+    val fakeGeneratorType: GeneratorType? = null,
+
+    // JSON string carrying strategy-specific parameters.
+    // HashRule: {"salt":"..."}
+    // PartialMaskRule: {"keepFirst":"2","keepLast":"4","maskChar":"*"}
+    // RegexRule: {"pattern":"\\d","replacement":"#"}
+    val piiRuleParams: String? = null
 )
 
 data class BulkCustomDataMappingRequest(
@@ -46,7 +52,9 @@ data class BulkCustomDataMappingRequest(
 
         val maskingStrategy: MaskingStrategy? = null,
 
-        val fakeGeneratorType: GeneratorType? = null
+        val fakeGeneratorType: GeneratorType? = null,
+
+        val piiRuleParams: String? = null
     )
 }
 
@@ -59,6 +67,7 @@ data class CustomDataMappingResponse(
     val action: MappingAction,
     val maskingStrategy: MaskingStrategy?,
     val fakeGeneratorType: GeneratorType?,
+    val piiRuleParams: String?,
     val createdAt: LocalDateTime,
     val updatedAt: LocalDateTime
 )
diff --git a/backend/src/main/kotlin/com/opendatamask/domain/port/output/RuleRegistryPort.kt b/backend/src/main/kotlin/com/opendatamask/domain/port/output/RuleRegistryPort.kt
new file mode 100644
index 0000000..5cd1e11
--- /dev/null
+++ b/backend/src/main/kotlin/com/opendatamask/domain/port/output/RuleRegistryPort.kt
@@ -0,0 +1,12 @@
+package com.opendatamask.domain.port.output
+
+import com.opendatamask.domain.model.PIIMaskingRule
+
+// Driven port: a registry that maps rule IDs to PIIMaskingRule implementations.
+// The default implementation is provided by DefaultRuleRegistry in the application layer.
+// Custom rules can be registered at runtime via registerCustomRule().
+interface RuleRegistryPort {
+    fun getRule(ruleId: String): PIIMaskingRule?
+    fun getAllRuleIds(): Set<String>
+    fun registerCustomRule(rule: PIIMaskingRule)
+}
diff --git a/backend/src/test/kotlin/com/opendatamask/application/service/DefaultRuleRegistryTest.kt b/backend/src/test/kotlin/com/opendatamask/application/service/DefaultRuleRegistryTest.kt
new file mode 100644
index 0000000..576310e
--- /dev/null
+++ b/backend/src/test/kotlin/com/opendatamask/application/service/DefaultRuleRegistryTest.kt
@@ -0,0 +1,75 @@
+package com.opendatamask.application.service
+
+import com.opendatamask.domain.model.PIIMaskingRule
+import com.opendatamask.domain.model.RedactRule
+import org.junit.jupiter.api.Assertions.*
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+
+class DefaultRuleRegistryTest {
+
+    private lateinit var registry: DefaultRuleRegistry
+
+    @BeforeEach
+    fun setUp() {
+        registry = DefaultRuleRegistry()
+    }
+
+    @Test
+    fun `registry contains all built-in rules after initialisation`() {
+        val ids = registry.getAllRuleIds()
+        assertTrue(ids.contains("pass_through"), "missing pass_through")
+        assertTrue(ids.contains("redact"), "missing redact")
+        assertTrue(ids.contains("partial_mask"), "missing partial_mask")
+        assertTrue(ids.contains("hash"), "missing hash")
+    }
+
+    @Test
+    fun `getRule returns rule for known ID`() {
+        val rule = registry.getRule("redact")
+        assertNotNull(rule)
+        assertEquals("[REDACTED]", rule!!.mask("sensitive"))
+    }
+
+    @Test
+    fun `getRule returns null for unknown ID`() {
+        assertNull(registry.getRule("unknown_rule"))
+    }
+
+    @Test
+    fun `registerCustomRule makes rule available via getRule`() {
+        val customRule = object : PIIMaskingRule {
+            override val ruleId = "eu_gdpr_mask"
+            override fun mask(input: Any?): Any? = if (input == null) null else "EU_MASKED"
+        }
+
+        registry.registerCustomRule(customRule)
+
+        val found = registry.getRule("eu_gdpr_mask")
+        assertNotNull(found)
+        assertEquals("EU_MASKED", found!!.mask("personal_data"))
+    }
+
+    @Test
+    fun `registerCustomRule overrides built-in rule when same ID is used`() {
+        val override = object : PIIMaskingRule {
+            override val ruleId = "redact"
+            override fun mask(input: Any?): Any? = "[CUSTOM_REDACTED]"
+        }
+
+        registry.registerCustomRule(override)
+
+        assertEquals("[CUSTOM_REDACTED]", registry.getRule("redact")!!.mask("x"))
+    }
+
+    @Test
+    fun `getAllRuleIds includes custom rules after registration`() {
+        val customRule = object : PIIMaskingRule {
+            override val ruleId = "custom_biz_rule"
+            override fun mask(input: Any?): Any? = null
+        }
+        registry.registerCustomRule(customRule)
+
+        assertTrue(registry.getAllRuleIds().contains("custom_biz_rule"))
+    }
+}
diff --git a/backend/src/test/kotlin/com/opendatamask/application/service/JobServiceTest.kt b/backend/src/test/kotlin/com/opendatamask/application/service/JobServiceTest.kt
index 118e498..0abdfae 100644
--- a/backend/src/test/kotlin/com/opendatamask/application/service/JobServiceTest.kt
+++ b/backend/src/test/kotlin/com/opendatamask/application/service/JobServiceTest.kt
@@ -4,17 +4,21 @@ import com.opendatamask.domain.port.output.EncryptionPort
 import com.opendatamask.domain.port.output.*
 import com.opendatamask.domain.model.*
 import org.junit.jupiter.api.Assertions.*
+import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
 import org.junit.jupiter.api.extension.ExtendWith
 import org.mockito.InjectMocks
 import org.mockito.Mock
 import org.mockito.junit.jupiter.MockitoExtension
+import org.mockito.junit.jupiter.MockitoSettings
 import org.mockito.kotlin.*
+import org.mockito.quality.Strictness
 import java.time.LocalDateTime
 import java.util.Optional
 
 @ExtendWith(MockitoExtension::class)
+@MockitoSettings(strictness = Strictness.LENIENT)
 class JobServiceTest {
 
     @Mock private lateinit var jobRepository: JobPort
@@ -31,10 +35,20 @@ class JobServiceTest {
     @Mock private lateinit var postJobActionService: PostJobActionService
     @Mock private lateinit var schemaChangeService: SchemaChangeService
     @Mock private lateinit var webhookService: WebhookService
+    @Mock private lateinit var piiMaskingService: PIIMaskingService
 
     @InjectMocks
     private lateinit var jobService: JobService
 
+    @BeforeEach
+    fun setUpDefaults() {
+        // By default piiMaskingService is a pass-through so existing PASSTHROUGH-mode
+        // tests continue to see unmodified rows at the destination.
+        @Suppress("UNCHECKED_CAST")
+        whenever(piiMaskingService.applyMappingsToRows(any(), any(), any(), any<List<Map<String, Any?>>>()))
+            .thenAnswer { it.arguments[3] as List<Map<String, Any?>> }
+    }
+
     private fun makeWorkspace(id: Long = 1L) = Workspace(
         id = id, name = "WS", ownerId = 1L,
         createdAt = LocalDateTime.now(), updatedAt = LocalDateTime.now()
diff --git a/backend/src/test/kotlin/com/opendatamask/application/service/PIIMaskingServiceTest.kt b/backend/src/test/kotlin/com/opendatamask/application/service/PIIMaskingServiceTest.kt
new file mode 100644
index 0000000..30b8cb8
--- /dev/null
+++ b/backend/src/test/kotlin/com/opendatamask/application/service/PIIMaskingServiceTest.kt
@@ -0,0 +1,299 @@
+package com.opendatamask.application.service
+
+import com.opendatamask.domain.model.CustomDataMapping
+import com.opendatamask.domain.model.MappingAction
+import com.opendatamask.domain.model.MaskingStrategy
+import com.opendatamask.domain.model.RedactRule
+import com.opendatamask.domain.port.output.CustomDataMappingPort
+import com.opendatamask.domain.port.output.RuleRegistryPort
+import org.junit.jupiter.api.Assertions.*
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.ExtendWith
+import org.mockito.Mock
+import org.mockito.junit.jupiter.MockitoExtension
+import org.mockito.junit.jupiter.MockitoSettings
+import org.mockito.kotlin.whenever
+import org.mockito.quality.Strictness
+
+@ExtendWith(MockitoExtension::class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class PIIMaskingServiceTest {
+
+    @Mock private lateinit var ruleRegistry: RuleRegistryPort
+    @Mock private lateinit var customDataMappingPort: CustomDataMappingPort
+
+    private lateinit var service: PIIMaskingService
+
+    @BeforeEach
+    fun setUp() {
+        service = PIIMaskingService(ruleRegistry, customDataMappingPort)
+        // Wire the real built-in registry so REDACT tests use the actual rule
+        whenever(ruleRegistry.getRule("redact")).thenReturn(RedactRule())
+    }
+
+    private fun mapping(
+        column: String,
+        action: MappingAction,
+        strategy: MaskingStrategy? = null,
+        params: String? = null
+    ) = CustomDataMapping(
+        id = 1L, workspaceId = 1L, connectionId = 2L,
+        tableName = "users", columnName = column,
+        action = action, maskingStrategy = strategy,
+        fakeGeneratorType = null, piiRuleParams = params
+    )
+
+    // ── no mappings ──────────────────────────────────────────────────────────
+
+    @Test
+    fun `applyMappings returns row unchanged when no mappings exist`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(emptyList())
+
+        val row = mapOf("email" to "john@example.com", "id" to 1L)
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        assertEquals(row, result)
+    }
+
+    // ── MIGRATE_AS_IS ────────────────────────────────────────────────────────
+
+    @Test
+    fun `applyMappings passes through columns with MIGRATE_AS_IS action`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(mapping("email", MappingAction.MIGRATE_AS_IS)))
+
+        val row = mapOf("email" to "john@example.com")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        assertEquals("john@example.com", result["email"])
+    }
+
+    // ── NULL strategy ────────────────────────────────────────────────────────
+
+    @Test
+    fun `applyMappings nullifies column with NULL strategy`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(mapping("ssn", MappingAction.MASK, MaskingStrategy.NULL)))
+
+        val row = mapOf("ssn" to "123-45-6789")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        assertNull(result["ssn"])
+    }
+
+    // ── REDACT strategy ──────────────────────────────────────────────────────
+
+    @Test
+    fun `applyMappings redacts column with REDACT strategy`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(mapping("email", MappingAction.MASK, MaskingStrategy.REDACT)))
+
+        val row = mapOf("email" to "john@example.com")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        assertEquals("[REDACTED]", result["email"])
+    }
+
+    // ── HASH strategy ────────────────────────────────────────────────────────
+
+    @Test
+    fun `applyMappings hashes column with HASH strategy`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(mapping("email", MappingAction.MASK, MaskingStrategy.HASH)))
+
+        val row = mapOf("email" to "hello")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        // SHA-256 of "hello" with no salt
+        assertEquals("2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824", result["email"])
+    }
+
+    @Test
+    fun `applyMappings hashes with salt when piiRuleParams contains salt`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(mapping("email", MappingAction.MASK, MaskingStrategy.HASH, """{"salt":"pepper"}""")))
+
+        val row = mapOf("email" to "hello")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        // Must differ from the unsalted hash
+        assertNotEquals("2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824", result["email"])
+        assertNotNull(result["email"])
+    }
+
+    // ── PARTIAL_MASK strategy ────────────────────────────────────────────────
+
+    @Test
+    fun `applyMappings applies partial mask with default params`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(mapping("credit_card", MappingAction.MASK, MaskingStrategy.PARTIAL_MASK)))
+
+        val row = mapOf("credit_card" to "4111111111111234")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        // Default: keepFirst=0, keepLast=4 → ************1234
+        assertEquals("************1234", result["credit_card"])
+    }
+
+    @Test
+    fun `applyMappings applies partial mask with custom params`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(
+                mapping("name", MappingAction.MASK, MaskingStrategy.PARTIAL_MASK, """{"keepFirst":"1","keepLast":"0"}""")
+            ))
+
+        val row = mapOf("name" to "Alice")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        assertEquals("A****", result["name"])
+    }
+
+    // ── REGEX strategy ───────────────────────────────────────────────────────
+
+    @Test
+    fun `applyMappings applies regex replacement`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(
+                mapping("phone", MappingAction.MASK, MaskingStrategy.REGEX, """{"pattern":"\\d","replacement":"#"}""")
+            ))
+
+        val row = mapOf("phone" to "123-456-7890")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        assertEquals("###-###-####", result["phone"])
+    }
+
+    @Test
+    fun `applyMappings passes value through unchanged when REGEX pattern is missing`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(mapping("phone", MappingAction.MASK, MaskingStrategy.REGEX)))
+
+        val row = mapOf("phone" to "123-456-7890")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        // Missing params → no-op, not data loss
+        assertEquals("123-456-7890", result["phone"])
+    }
+
+    @Test
+    fun `applyMappings passes value through unchanged when piiRuleParams JSON is invalid`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(mapping("phone", MappingAction.MASK, MaskingStrategy.HASH, "not-json")))
+
+        val row = mapOf("phone" to "123-456-7890")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        // Invalid JSON → parseParams returns null → value passes through unchanged
+        assertEquals("123-456-7890", result["phone"])
+    }
+
+    // ── custom ruleId dispatch via piiRuleParams ──────────────────────────────
+
+    @Test
+    fun `applyMappings dispatches to registry when ruleId is set in piiRuleParams`() {
+        val registeredRule = com.opendatamask.domain.model.RedactRule()
+        whenever(ruleRegistry.getRule("my_custom_rule")).thenReturn(registeredRule)
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(
+                mapping("email", MappingAction.MASK, MaskingStrategy.REDACT, """{"ruleId":"my_custom_rule"}""")
+            ))
+
+        val row = mapOf("email" to "john@example.com")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        assertEquals("[REDACTED]", result["email"])
+    }
+
+    @Test
+    fun `applyMappings passes value through unchanged when custom ruleId is not found in registry`() {
+        whenever(ruleRegistry.getRule("unknown_rule")).thenReturn(null)
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(
+                mapping("email", MappingAction.MASK, MaskingStrategy.REDACT, """{"ruleId":"unknown_rule"}""")
+            ))
+
+        val row = mapOf("email" to "john@example.com")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        // Unknown rule → pass through with warning log
+        assertEquals("john@example.com", result["email"])
+    }
+
+    // ── FAKE strategy (handled by GeneratorService, not PIIMaskingService) ───
+
+    @Test
+    fun `applyMappings leaves FAKE strategy columns to GeneratorService`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(mapping("email", MappingAction.MASK, MaskingStrategy.FAKE)))
+
+        val row = mapOf("email" to "original@example.com")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        // FAKE strategy is a no-op in PIIMaskingService; value is left for GeneratorService
+        assertEquals("original@example.com", result["email"])
+    }
+
+    // ── column name case-insensitivity ───────────────────────────────────────
+
+    @Test
+    fun `applyMappings matches columns case-insensitively`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(mapping("EMAIL", MappingAction.MASK, MaskingStrategy.REDACT)))
+
+        val row = mapOf("email" to "john@example.com")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        assertEquals("[REDACTED]", result["email"])
+    }
+
+    // ── mixed row ────────────────────────────────────────────────────────────
+
+    @Test
+    fun `applyMappings handles mixed actions in same row`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(
+                mapping("id", MappingAction.MIGRATE_AS_IS),
+                mapping("email", MappingAction.MASK, MaskingStrategy.REDACT),
+                mapping("ssn", MappingAction.MASK, MaskingStrategy.NULL)
+            ))
+
+        val row = mapOf("id" to 42L, "email" to "jane@example.com", "ssn" to "000-00-0000", "name" to "Jane")
+        val result = service.applyMappings(1L, 2L, "users", row)
+
+        assertEquals(42L, result["id"])
+        assertEquals("[REDACTED]", result["email"])
+        assertNull(result["ssn"])
+        assertEquals("Jane", result["name"]) // unmapped column passes through
+    }
+
+    // ── batch API ─────────────────────────────────────────────────────────────
+
+    @Test
+    fun `applyMappingsToRows fetches mappings once and applies to all rows`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(listOf(mapping("email", MappingAction.MASK, MaskingStrategy.REDACT)))
+
+        val rows = listOf(
+            mapOf("email" to "alice@example.com"),
+            mapOf("email" to "bob@example.com")
+        )
+        val results = service.applyMappingsToRows(1L, 2L, "users", rows)
+
+        assertEquals(2, results.size)
+        assertEquals("[REDACTED]", results[0]["email"])
+        assertEquals("[REDACTED]", results[1]["email"])
+    }
+
+    @Test
+    fun `applyMappingsToRows returns original list unchanged when no mappings exist`() {
+        whenever(customDataMappingPort.findByWorkspaceIdAndConnectionIdAndTableName(1L, 2L, "users"))
+            .thenReturn(emptyList())
+
+        val rows = listOf(mapOf("email" to "alice@example.com"))
+        val results = service.applyMappingsToRows(1L, 2L, "users", rows)
+
+        assertEquals(rows, results)
+    }
+}
diff --git a/backend/src/test/kotlin/com/opendatamask/domain/model/PIIMaskingRuleTest.kt b/backend/src/test/kotlin/com/opendatamask/domain/model/PIIMaskingRuleTest.kt
new file mode 100644
index 0000000..d74d6c4
--- /dev/null
+++ b/backend/src/test/kotlin/com/opendatamask/domain/model/PIIMaskingRuleTest.kt
@@ -0,0 +1,133 @@
+package com.opendatamask.domain.model
+
+import org.junit.jupiter.api.Assertions.*
+import org.junit.jupiter.api.Test
+
+class PIIMaskingRuleTest {
+
+    // ── PassThroughRule ──────────────────────────────────────────────────────
+
+    @Test
+    fun `PassThroughRule returns input unchanged`() {
+        val rule = PassThroughRule()
+        assertEquals("hello", rule.mask("hello"))
+        assertEquals(42, rule.mask(42))
+    }
+
+    @Test
+    fun `PassThroughRule returns null for null input`() {
+        assertNull(PassThroughRule().mask(null))
+    }
+
+    // ── RedactRule ───────────────────────────────────────────────────────────
+
+    @Test
+    fun `RedactRule replaces non-null value with REDACTED token`() {
+        val rule = RedactRule()
+        assertEquals("[REDACTED]", rule.mask("john.doe@example.com"))
+        assertEquals("[REDACTED]", rule.mask(12345))
+        assertEquals("[REDACTED]", rule.mask(""))
+    }
+
+    @Test
+    fun `RedactRule returns null for null input`() {
+        assertNull(RedactRule().mask(null))
+    }
+
+    // ── PartialMaskRule ──────────────────────────────────────────────────────
+
+    @Test
+    fun `PartialMaskRule masks middle characters keeping last 4`() {
+        val rule = PartialMaskRule(keepFirst = 0, keepLast = 4)
+        assertEquals("****1234", rule.mask("12341234"))
+    }
+
+    @Test
+    fun `PartialMaskRule keeps first and last characters`() {
+        val rule = PartialMaskRule(keepFirst = 1, keepLast = 1)
+        assertEquals("j***e", rule.mask("johne"))
+    }
+
+    @Test
+    fun `PartialMaskRule with short input returns input unchanged`() {
+        val rule = PartialMaskRule(keepFirst = 2, keepLast = 3)
+        assertEquals("hello", rule.mask("hello"))
+    }
+
+    @Test
+    fun `PartialMaskRule returns null for null input`() {
+        assertNull(PartialMaskRule().mask(null))
+    }
+
+    @Test
+    fun `PartialMaskRule uses custom mask character`() {
+        val rule = PartialMaskRule(keepFirst = 0, keepLast = 4, maskChar = '#')
+        assertEquals("####5678", rule.mask("12345678"))
+    }
+
+    // ── HashRule ─────────────────────────────────────────────────────────────
+
+    @Test
+    fun `HashRule produces consistent SHA-256 hex digest`() {
+        val rule = HashRule()
+        val result = rule.mask("hello")
+        assertNotNull(result)
+        // SHA-256 of "hello" is well-known
+        assertEquals("2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824", result)
+    }
+
+    @Test
+    fun `HashRule with salt produces different digest`() {
+        val noSalt = HashRule().mask("hello")
+        val withSalt = HashRule(salt = "pepper").mask("hello")
+        assertNotEquals(noSalt, withSalt)
+    }
+
+    @Test
+    fun `HashRule returns null for null input`() {
+        assertNull(HashRule().mask(null))
+    }
+
+    @Test
+    fun `HashRule output is always 64 hex characters`() {
+        val result = HashRule().mask("test value") as String
+        assertEquals(64, result.length)
+        assertTrue(result.all { it in '0'..'9' || it in 'a'..'f' })
+    }
+
+    // ── RegexRule ────────────────────────────────────────────────────────────
+
+    @Test
+    fun `RegexRule replaces all pattern matches`() {
+        val rule = RegexRule(pattern = "\\d", replacement = "#")
+        assertEquals("###-##-####", rule.mask("123-45-6789"))
+    }
+
+    @Test
+    fun `RegexRule with capture group replacement`() {
+        val rule = RegexRule(pattern = "(\\w+)@(\\w+\\.\\w+)", replacement = "***@$2")
+        assertEquals("***@example.com", rule.mask("john@example.com"))
+    }
+
+    @Test
+    fun `RegexRule returns null for null input`() {
+        assertNull(RegexRule(".*", "").mask(null))
+    }
+
+    @Test
+    fun `RegexRule with no match returns original string`() {
+        val rule = RegexRule(pattern = "\\d+", replacement = "")
+        assertEquals("hello", rule.mask("hello"))
+    }
+
+    // ── ruleId contract ──────────────────────────────────────────────────────
+
+    @Test
+    fun `built-in rules have expected stable rule IDs`() {
+        assertEquals("pass_through", PassThroughRule().ruleId)
+        assertEquals("redact", RedactRule().ruleId)
+        assertEquals("partial_mask", PartialMaskRule().ruleId)
+        assertEquals("hash", HashRule().ruleId)
+        assertEquals("regex", RegexRule("", "").ruleId)
+    }
+}
diff --git a/docs/pii-masking-rules-sample.json b/docs/pii-masking-rules-sample.json
new file mode 100644
index 0000000..59c887a
--- /dev/null
+++ b/docs/pii-masking-rules-sample.json
@@ -0,0 +1,37 @@
+{
+  "description": "Sample PII attribute rule mapping for a Customer dataset",
+  "workspaceId": 1,
+  "connectionId": 2,
+  "tableName": "customers",
+  "columnMappings": [
+    {
+      "columnName": "id",
+      "action": "MIGRATE_AS_IS"
+    },
+    {
+      "columnName": "full_name",
+      "action": "MASK",
+      "maskingStrategy": "PARTIAL_MASK",
+      "piiRuleParams": "{\"keepFirst\":\"1\",\"keepLast\":\"0\",\"maskChar\":\"*\"}",
+      "description": "Keeps only the first initial; e.g. 'Alice Smith' → 'A**********'"
+    },
+    {
+      "columnName": "email",
+      "action": "MASK",
+      "maskingStrategy": "REGEX",
+      "piiRuleParams": "{\"pattern\":\"(\\\\w+)@(\\\\w+\\\\.\\\\w+)\",\"replacement\":\"***@$2\"}",
+      "description": "Masks the local part of the email address; e.g. 'alice@example.com' → '***@example.com'"
+    },
+    {
+      "columnName": "ssn",
+      "action": "MASK",
+      "maskingStrategy": "REDACT",
+      "description": "Fully redacts the SSN; e.g. '123-45-6789' → '[REDACTED]'"
+    },
+    {
+      "columnName": "transaction_amount",
+      "action": "MIGRATE_AS_IS",
+      "description": "Transaction amounts are not PII and pass through unchanged"
+    }
+  ]
+}
diff --git a/docs/user-guide.md b/docs/user-guide.md
index 0d2daf1..090292f 100644
--- a/docs/user-guide.md
+++ b/docs/user-guide.md
@@ -12,6 +12,7 @@ Core capabilities:
 - **Job scheduling**: Cron-based automated masking runs
 - **Webhook integration**: Post-job notifications via custom HTTP webhooks or GitHub Actions triggers
 - **REST API + CLI**: Full programmatic access and a Go-based CLI tool
+- **PII attribute rule mapping**: Column-level masking rules (Redact, Partial Mask, Hash, Regex) with a registry pattern for custom business rules
 
 ---
 
@@ -648,6 +649,112 @@ See [verification/README.md](../verification/README.md) for the full reference,
 
 ---
 
+## PII Attribute Rule Mapping
+
+OpenDataMask provides a flexible, configuration-driven system to apply fine-grained masking rules to individual source columns as data moves to the target destination.
+
+### Built-in Masking Strategies
+
+| `maskingStrategy` | Description | Required `piiRuleParams` keys |
+|---|---|---|
+| `FAKE` | Replaces the value with realistic synthetic data using the configured `fakeGeneratorType` | — |
+| `HASH` | Deterministic SHA-256 hex digest of the value | `salt` (optional) |
+| `NULL` | Replaces the value with `null` / `NULL` | — |
+| `REDACT` | Replaces the value with the literal token `[REDACTED]` | — |
+| `PARTIAL_MASK` | Keeps a configurable number of leading/trailing characters and masks the middle | `keepFirst` (default `0`), `keepLast` (default `4`), `maskChar` (default `*`) |
+| `REGEX` | Applies a regular expression replacement to the string representation of the value | `pattern` (required), `replacement` (required) |
+
+### Mapping a column via REST API
+
+```http
+POST /api/workspaces/{workspaceId}/mappings
+Content-Type: application/json
+
+{
+  "connectionId": 2,
+  "tableName": "customers",
+  "columnName": "ssn",
+  "action": "MASK",
+  "maskingStrategy": "REDACT"
+}
+```
+
+### Bulk mapping (all columns of a table at once)
+
+```http
+POST /api/workspaces/{workspaceId}/mappings/bulk
+Content-Type: application/json
+
+{
+  "connectionId": 2,
+  "tableName": "customers",
+  "columnMappings": [
+    { "columnName": "id",                 "action": "MIGRATE_AS_IS" },
+    { "columnName": "full_name",          "action": "MASK", "maskingStrategy": "PARTIAL_MASK",
+      "piiRuleParams": "{\"keepFirst\":\"1\",\"keepLast\":\"0\"}" },
+    { "columnName": "email",              "action": "MASK", "maskingStrategy": "REGEX",
+      "piiRuleParams": "{\"pattern\":\"(\\\\w+)@(\\\\w+\\\\.\\\\w+)\",\"replacement\":\"***@$2\"}" },
+    { "columnName": "ssn",                "action": "MASK", "maskingStrategy": "REDACT" },
+    { "columnName": "transaction_amount", "action": "MIGRATE_AS_IS" }
+  ]
+}
+```
+
+A complete JSON example is available in [`docs/pii-masking-rules-sample.json`](pii-masking-rules-sample.json).
+
+### How rules are applied during a job
+
+Mappings are evaluated during every **PASSTHROUGH** table run: for each row fetched from the source, OpenDataMask looks up any active `CustomDataMapping` entries for that workspace/connection/table combination and applies the configured strategy before writing to the destination. Columns without a mapping pass through unchanged.
+
+### Registering a custom business rule at runtime
+
+The `RuleRegistryPort` / `DefaultRuleRegistry` bean accepts runtime-registered `PIIMaskingRule` implementations. Inject `RuleRegistryPort` into your Spring component and call `registerCustomRule()`:
+
+```kotlin
+@Component
+class EuGdprRuleRegistrar(private val ruleRegistry: RuleRegistryPort) {
+
+    @PostConstruct
+    fun register() {
+        ruleRegistry.registerCustomRule(object : PIIMaskingRule {
+            override val ruleId = "eu_gdpr_conditional"
+            override fun mask(input: Any?): Any? {
+                // Example: if the value looks like an EU phone prefix, redact it fully;
+                // otherwise apply partial masking
+                val str = input?.toString() ?: return null
+                return if (str.startsWith("+3") || str.startsWith("+4"))
+                    "[REDACTED]"
+                else
+                    str.take(0) + "*".repeat((str.length - 4).coerceAtLeast(0)) + str.takeLast(4)
+            }
+        })
+    }
+}
+```
+
+After registering the rule, invoke it via any column mapping by adding `"ruleId": "<your-rule-id>"` to `piiRuleParams`. The engine checks for a `ruleId` key first and, if found, delegates to the registry before falling back to the built-in strategy logic:
+
+```http
+POST /api/workspaces/{workspaceId}/mappings/bulk
+Content-Type: application/json
+
+{
+  "connectionId": 2,
+  "tableName": "customers",
+  "columnMappings": [
+    {
+      "columnName": "phone",
+      "action": "MASK",
+      "maskingStrategy": "REDACT",
+      "piiRuleParams": "{\"ruleId\":\"eu_gdpr_conditional\"}"
+    }
+  ]
+}
+```
+
+The `maskingStrategy` field still controls the UI label and fallback type; the `ruleId` in `piiRuleParams` takes precedence for actual masking logic.
+
+
 ## Contributing
 
 1. Fork the repository
diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts
index 8d98aa8..44b2a32 100644
--- a/frontend/src/types/index.ts
+++ b/frontend/src/types/index.ts
@@ -383,7 +383,10 @@ export enum MappingAction {
 export enum MaskingStrategy {
   FAKE = 'FAKE',
   HASH = 'HASH',
-  NULL = 'NULL'
+  NULL = 'NULL',
+  REDACT = 'REDACT',
+  PARTIAL_MASK = 'PARTIAL_MASK',
+  REGEX = 'REGEX'
 }
 
 export interface CustomDataMapping {
@@ -395,6 +398,7 @@ export interface CustomDataMapping {
   action: MappingAction
   maskingStrategy: MaskingStrategy | null
   fakeGeneratorType: GeneratorType | null
+  piiRuleParams: string | null
   createdAt: string
   updatedAt: string
 }
@@ -406,6 +410,9 @@ export interface CustomDataMappingRequest {
   action: MappingAction
   maskingStrategy?: MaskingStrategy | null
   fakeGeneratorType?: GeneratorType | null
+  // JSON string with strategy-specific params (e.g. {"salt":"..."} for HASH,
+  // {"keepFirst":"0","keepLast":"4"} for PARTIAL_MASK, {"pattern":"...","replacement":"..."} for REGEX)
+  piiRuleParams?: string | null
 }
 
 export interface ColumnMappingEntry {
@@ -413,6 +420,7 @@ export interface ColumnMappingEntry {
   action: MappingAction
   maskingStrategy?: MaskingStrategy | null
   fakeGeneratorType?: GeneratorType | null
+  piiRuleParams?: string | null
 }
 
 export interface BulkCustomDataMappingRequest {
diff --git a/frontend/src/views/__tests__/DataMappingView.test.ts b/frontend/src/views/__tests__/DataMappingView.test.ts
index 7f07ebe..d77a316 100644
--- a/frontend/src/views/__tests__/DataMappingView.test.ts
+++ b/frontend/src/views/__tests__/DataMappingView.test.ts
@@ -19,8 +19,14 @@ describe('MaskingStrategy enum', () => {
     expect(MaskingStrategy.NULL).toBe('NULL')
   })
 
-  it('has exactly 3 values', () => {
-    expect(Object.keys(MaskingStrategy).length).toBe(3)
+  it('has REDACT, PARTIAL_MASK and REGEX values', () => {
+    expect(MaskingStrategy.REDACT).toBe('REDACT')
+    expect(MaskingStrategy.PARTIAL_MASK).toBe('PARTIAL_MASK')
+    expect(MaskingStrategy.REGEX).toBe('REGEX')
+  })
+
+  it('has exactly 6 values', () => {
+    expect(Object.keys(MaskingStrategy).length).toBe(6)
   })
 })