@@ -804,52 +804,6 @@ class CoreOptions:
         .with_description("Maximum wait time between commit retries (e.g., '1s', '10s').")
     )
 
-    COMMIT_DEDUP_ON_FIRST_ATTEMPT: ConfigOption[bool] = (
-        ConfigOptions.key("commit.dedup-on-first-attempt")
-        .boolean_type()
-        .default_value(False)
-        .with_description(
-            "Also run the duplicate-commit scan on the first commit attempt, not "
-            "only on optimistic-lock retries. Enable when the same logical commit "
-            "may be re-issued with a stable commit_user (e.g. a Ray committer-actor "
-            "task retried after the actor died mid-commit), so an already-landed "
-            "duplicate is skipped. Off by default: per-worker/two-phase commits use "
-            "a fresh commit_user per builder, so a first-attempt scan would never "
-            "match and would only add a snapshot scan per commit."
-        )
-    )
-
-    COMMIT_DEDUP_BASE_SNAPSHOT: ConfigOption[int] = (
-        ConfigOptions.key("commit.dedup-base-snapshot")
-        .int_type()
-        .no_default_value()
-        .with_description(
-            "Lower bound for the first-attempt duplicate-commit scan: only snapshots "
-            "after this id are checked. Should be the latest snapshot id the writer "
-            "observed before issuing its first commit, so the scan is bounded to "
-            "snapshots produced by this job rather than the table's whole history. "
-            "Only used when commit.dedup-on-first-attempt is enabled; unset means "
-            "scan from the first snapshot."
-        )
-    )
-
-    COMMIT_DEDUP_BASE_FILE_MILLIS: ConfigOption[int] = (
-        ConfigOptions.key("commit.dedup-base-file-millis")
-        .int_type()
-        .no_default_value()
-        .with_description(
-            "Filesystem mtime, in epoch millis, of the snapshot file named by "
-            "commit.dedup-base-snapshot when that bound was taken. Validates the "
-            "bound: snapshot ids are re-used after a rollback, so a bare id can "
-            "point at a different snapshot by the time the scan runs, and the "
-            "scan would then start above this write's own commit and miss the "
-            "duplicate. When the mtime no longer matches, the scan walks back "
-            "from the current latest instead of trusting the id. Only used when "
-            "commit.dedup-on-first-attempt is enabled; unset means the bound "
-            "cannot be validated, so the whole history is scanned."
-        )
-    )
-
     # Keys under this prefix are copied into the committed snapshot's properties
     # (re-prefixed with SNAPSHOT_COMMIT_PREFIX), letting a caller tag one commit
     # with its own metadata without the table schema having to know about it.
@@ -1782,15 +1736,6 @@ def commit_max_retry_wait(self) -> int:
         wait = self.options.get(CoreOptions.COMMIT_MAX_RETRY_WAIT)
         return int(wait.total_seconds() * 1000)
 
-    def commit_dedup_on_first_attempt(self) -> bool:
-        return self.options.get(CoreOptions.COMMIT_DEDUP_ON_FIRST_ATTEMPT)
-
-    def commit_dedup_base_snapshot(self, default=None) -> Optional[int]:
-        return self.options.get(CoreOptions.COMMIT_DEDUP_BASE_SNAPSHOT, default)
-
-    def commit_dedup_base_file_millis(self, default=None) -> Optional[int]:
-        return self.options.get(CoreOptions.COMMIT_DEDUP_BASE_FILE_MILLIS, default)
-
     def global_index_enabled(self, default=None):
         return self.options.get(CoreOptions.GLOBAL_INDEX_ENABLED, default)
 
 
@@ -142,9 +142,6 @@ def setUp(self):
         self.mock_table.table_path = '/test/table/path'
         self.mock_table.file_io = Mock()
         self.mock_table.options.manifest_target_size.return_value = 8 * 1024 * 1024
-        self.mock_table.options.commit_dedup_on_first_attempt.return_value = False
-        self.mock_table.options.commit_dedup_base_snapshot.return_value = None
-        self.mock_table.options.commit_dedup_base_file_millis.return_value = None
         self.mock_table.options.options.to_map.return_value = {}
         self.mock_snapshot_commit = Mock()
 
@@ -372,9 +369,6 @@ def setUp(self):
         self.mock_table.table_path = '/test/table/path'
         self.mock_table.file_io = Mock()
         self.mock_table.options.manifest_target_size.return_value = 8 * 1024 * 1024
-        self.mock_table.options.commit_dedup_on_first_attempt.return_value = False
-        self.mock_table.options.commit_dedup_base_snapshot.return_value = None
-        self.mock_table.options.commit_dedup_base_file_millis.return_value = None
         self.mock_table.options.options.to_map.return_value = {}
 
         # Mock snapshot commit
 
@@ -252,3 +252,45 @@ def test_stream_commit_forwards_snapshot_properties(self):
             commit_identifier=42,
             snapshot_properties={"checkpoint": "42"},
         )
+
+    def test_native_commit_warns_when_first_attempt_dedup_requested(self):
+        """The native committer has no dedup scan, so a replayed commit would
+        publish the same files twice. The commit still goes ahead -- the scan
+        is not implemented on that path -- but the caller is told."""
+        from pypaimon.write.file_store_commit import CommitDedup
+
+        table = Mock()
+        table.identifier = 'default.test_table'
+        table.new_snapshot_commit.return_value = Mock()
+        table.options.native_commit_enabled.return_value = True
+
+        with self.assertLogs(
+                'pypaimon.write.table_commit', level='WARNING') as logs:
+            commit = BatchTableCommit(
+                table, 'stable-user', None,
+                dedup=CommitDedup(on_first_attempt=True))
+
+        self.assertEqual(commit.file_store_commit.table, table)
+        self.assertTrue(any(
+            'commit.native.enabled' in line and 'stable-user' in line
+            for line in logs.output), logs.output)
+
+    def test_native_commit_does_not_warn_without_first_attempt_dedup(self):
+        import logging as _logging
+        from unittest.mock import patch
+        from pypaimon.write.file_store_commit import CommitDedup
+
+        table = Mock()
+        table.identifier = 'default.test_table'
+        table.new_snapshot_commit.return_value = Mock()
+        table.options.native_commit_enabled.return_value = True
+
+        # assertNoLogs is 3.10+; patch warning explicitly so older
+        # interpreters keep working too.
+        logger = _logging.getLogger('pypaimon.write.table_commit')
+        with patch.object(logger, 'warning') as warn:
+            BatchTableCommit(
+                table, 'stable-user', None,
+                dedup=CommitDedup(on_first_attempt=False))
+            BatchTableCommit(table, 'stable-user', None, dedup=None)
+            warn.assert_not_called()
@@ -69,6 +69,38 @@ class DedupEvidenceMissing(RuntimeError):
     """
 
 
+class CommitDedup:
+    """Duplicate-commit scan parameters for one committer.
+
+    Passed to :class:`FileStoreCommit` (via ``WriteBuilder.new_commit``) rather
+    than read from table options: the bound describes *this* writer at *this*
+    moment -- the branch's latest snapshot when the write started -- and would be
+    wrong for any other writer inheriting it from the table schema.
+
+    ``base_snapshot_id`` and ``base_file_millis`` are only meaningful together:
+    the id bounds the scan and the mtime proves the id still names the snapshot
+    it named when the bound was taken. A caller with no bound leaves both None.
+
+    Plain class rather than a dataclass, which pypaimon still supports Python
+    3.6 without.
+    """
+
+    __slots__ = ("on_first_attempt", "base_snapshot_id", "base_file_millis")
+
+    def __init__(self, on_first_attempt: bool = False,
+                 base_snapshot_id: Optional[int] = None,
+                 base_file_millis: Optional[int] = None):
+        self.on_first_attempt = on_first_attempt
+        self.base_snapshot_id = base_snapshot_id
+        self.base_file_millis = base_file_millis
+
+    def __repr__(self) -> str:
+        return ("CommitDedup(on_first_attempt=%r, base_snapshot_id=%r, "
+                "base_file_millis=%r)" % (
+                    self.on_first_attempt, self.base_snapshot_id,
+                    self.base_file_millis))
+
+
 def _row_id_check_from_messages(messages: List[CommitMessage]) -> Optional[int]:
     """Check conflicts from the earliest baseline, matching the native committer."""
     check_from_snapshot = None
@@ -161,13 +193,15 @@ class FileStoreCommit:
     """
 
     def __init__(self, snapshot_commit: SnapshotCommit, table, commit_user: str,
-                 commit_callbacks: Optional[List[CommitCallback]] = None):
+                 commit_callbacks: Optional[List[CommitCallback]] = None,
+                 dedup: Optional[CommitDedup] = None):
         from pypaimon.table.file_store_table import FileStoreTable
 
         self.snapshot_commit = snapshot_commit
         self.table: FileStoreTable = table
         self.commit_user = commit_user
         self.commit_callbacks: List[CommitCallback] = commit_callbacks if commit_callbacks is not None else []
+        self.dedup = dedup if dedup is not None else CommitDedup()
 
         self.snapshot_manager = table.snapshot_manager()
         self.manifest_file_manager = ManifestFileManager(table)
@@ -179,9 +213,6 @@ def __init__(self, snapshot_commit: SnapshotCommit, table, commit_user: str,
         self.commit_timeout = table.options.commit_timeout()
         self.commit_min_retry_wait = table.options.commit_min_retry_wait()
         self.commit_max_retry_wait = table.options.commit_max_retry_wait()
-        self.commit_dedup_on_first_attempt = table.options.commit_dedup_on_first_attempt()
-        self.commit_dedup_base_snapshot = table.options.commit_dedup_base_snapshot()
-        self.commit_dedup_base_file_millis = table.options.commit_dedup_base_file_millis()
 
         self.row_id_conflict_rewrite_max_size = (
             table.options.data_evolution_row_id_conflict_rewrite_max_size()
@@ -874,7 +905,7 @@ def _dedup_start_after_rollback(self, end_id: int) -> Tuple[int, bool]:
         older than the write's start therefore finds the lowest id this write
         could have reached.
         """
-        base_millis = self.commit_dedup_base_file_millis
+        base_millis = self.dedup.base_file_millis
         for snapshot_id in range(end_id, 0, -1):
             file_millis = self.snapshot_manager.snapshot_file_millis(snapshot_id)
             if file_millis is None:
@@ -904,20 +935,20 @@ def _dedup_scan_window(self, retry_result, end_id: int) -> Tuple[int, bool]:
                 return retry_result.latest_snapshot.id + 1, True
             return 1, True  # Snapshot.FIRST_SNAPSHOT_ID
 
-        base_id = self.commit_dedup_base_snapshot
+        base_id = self.dedup.base_snapshot_id
         if base_id is None:
             return 1, True
         if base_id == 0:
             # The branch had no snapshot when this write started, so every
             # snapshot on it is a candidate and none is missing from below.
             return 1, True
-        if self.commit_dedup_base_file_millis is None:
+        if self.dedup.base_file_millis is None:
             # Bound without the mtime that validates it: cannot tell a normal id
             # line from a rolled-back one, so scan everything.
             return 1, base_id <= end_id
 
         base_millis = self.snapshot_manager.snapshot_file_millis(base_id)
-        if base_millis == self.commit_dedup_base_file_millis:
+        if base_millis == self.dedup.base_file_millis:
             # The bound still names the same file, so ids above it were handed
             # out normally and our commit, if any, is among them. Expiry trims
             # the oldest snapshots first, so a surviving base also proves the
@@ -1018,7 +1049,7 @@ def _is_duplicate_commit(
             commit_kind,
             notify_callbacks=False) -> bool:
         if (not isinstance(retry_result, CommitFailRetryResult)
-                and not self.commit_dedup_on_first_attempt):
+                and not self.dedup.on_first_attempt):
             # First attempt of a fresh commit_user commit: it cannot be a
             # duplicate, so skip the scan entirely (per-worker/two-phase
             # default).
 
@@ -35,6 +35,7 @@
 import pyarrow as pa
 
 from pypaimon.write.commit_callback import CommitCallback
+from pypaimon.write.file_store_commit import CommitDedup
 
 if TYPE_CHECKING:
     from pypaimon.table.table import Table
@@ -725,9 +726,9 @@ class PaimonCommitterActor:
     (``{job_id}-{write_uuid}-{task_idx}``) and sends it, with the
     driver-pinned ``base_snapshot_id``, in the RPC args. If the actor dies
     mid-commit, Ray re-issues the *same* actor task with the *same* args, so
-    the replayed commit carries the same ``commit_user``; the injected
-    ``commit.dedup-on-first-attempt`` scan then finds the already-landed
-    snapshot (same ``commit_user`` + ``commit_identifier`` + ``commit_kind``)
+    the replayed commit carries the same ``commit_user``; the first-attempt
+    dedup scan then finds the already-landed snapshot (same ``commit_user`` +
+    ``commit_identifier`` + ``commit_kind``)
     and skips it instead of committing the same files twice. Without this the
     fresh builder's random commit_user would make every replay a duplicate.
     This does not extend across jobs -- see :class:`PaimonActorDatasink`.
@@ -864,18 +865,21 @@ def commit(
             and commit_user is not None
             and commit_user not in self._attempted_users
         )
-        options['commit.dedup-on-first-attempt'] = 'false' if can_vouch else 'true'
-        if dedup_base is not None:
-            base_id = dedup_base.get("snapshot_id")
-            if base_id is not None:
-                options['commit.dedup-base-snapshot'] = str(base_id)
-            file_millis = dedup_base.get("file_millis")
-            if file_millis is not None:
-                options['commit.dedup-base-file-millis'] = str(file_millis)
+        # A runtime argument to new_commit, not an option: it describes this one
+        # commit attempt, whereas options reach the committer through
+        # table.copy() and would be inherited by every other writer built from
+        # that table.
+        dedup = CommitDedup(
+            on_first_attempt=not can_vouch,
+            base_snapshot_id=(
+                dedup_base.get("snapshot_id") if dedup_base else None),
+            base_file_millis=(
+                dedup_base.get("file_millis") if dedup_base else None),
+        )
 
         builder = self._table.new_batch_write_builder()
         # with_options returns a new builder; discarding it silently drops the
-        # options (commit.metadata.* and the dedup options would never land).
+        # options (commit.metadata.* would never land).
         builder = builder.with_options(options)
         if commit_user is not None:
             # with_options rebuilt the builder (regenerating a random
@@ -884,7 +888,7 @@ def commit(
             builder.commit_user = commit_user
 
         # Create commit and pass message via snapshot_properties if provided
-        table_commit = builder.new_commit()
+        table_commit = builder.new_commit(dedup=dedup)
         snapshot_properties = {}
         if message:
             snapshot_properties['commit.message'] = message
 
@@ -18,10 +18,11 @@
 import logging
 from typing import Dict, List, Optional
 
+from pypaimon.common.options.core_options import CoreOptions
 from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
 from pypaimon.write.commit_callback import CommitCallback
 from pypaimon.write.commit_message import CommitMessage
-from pypaimon.write.file_store_commit import FileStoreCommit
+from pypaimon.write.file_store_commit import CommitDedup, FileStoreCommit
 
 logger = logging.getLogger(__name__)
 
@@ -39,7 +40,8 @@ class TableCommit:
       on every call and may be reused for many commits.
     """
 
-    def __init__(self, table, commit_user: str, static_partition: Optional[dict]):
+    def __init__(self, table, commit_user: str, static_partition: Optional[dict],
+                 *, dedup: Optional[CommitDedup] = None):
         from pypaimon.table.file_store_table import FileStoreTable
 
         self.table: FileStoreTable = table
@@ -56,7 +58,21 @@ def __init__(self, table, commit_user: str, static_partition: Optional[dict]):
         self._native_callbacks_registered = False
         self.file_store_commit = FileStoreCommit(
             snapshot_commit, table, commit_user,
-            commit_callbacks=self._commit_callbacks)
+            commit_callbacks=self._commit_callbacks, dedup=dedup)
+        if (dedup is not None and dedup.on_first_attempt
+                and table.options.native_commit_enabled()):
+            # The native committer has no duplicate-commit scan, and _commit
+            # returns as soon as it publishes, so the Python dedup below is
+            # never reached. A caller only asks for first-attempt dedup because
+            # it may re-issue this commit with the same commit_user, and on the
+            # native path that re-issue commits the files a second time.
+            logger.warning(
+                "Table %s has %s enabled while this commit requests "
+                "first-attempt duplicate detection. The native committer does "
+                "not implement that scan, so a re-issued commit with "
+                "commit_user %s can publish the same files twice.",
+                table.identifier, CoreOptions.COMMIT_NATIVE_ENABLED.key(),
+                commit_user)
 
     def add_commit_callback(self, callback: CommitCallback) -> None:
         """Register a callback to be invoked after each successful commit."""
@@ -167,8 +183,9 @@ def close(self):
 class BatchTableCommit(TableCommit):
     """Batch-mode commit; supports at most one commit per instance."""
 
-    def __init__(self, table, commit_user: str, static_partition: Optional[dict]):
-        super().__init__(table, commit_user, static_partition)
+    def __init__(self, table, commit_user: str, static_partition: Optional[dict],
+                 *, dedup: Optional[CommitDedup] = None):
+        super().__init__(table, commit_user, static_partition, dedup=dedup)
         self.batch_committed = False
 
     def commit(
@@ -205,8 +222,9 @@ class StreamTableCommit(TableCommit):
     :meth:`StreamTableWrite.prepare_commit`.
     """
 
-    def __init__(self, table, commit_user: str):
-        super().__init__(table, commit_user, None)
+    def __init__(self, table, commit_user: str,
+                 *, dedup: Optional[CommitDedup] = None):
+        super().__init__(table, commit_user, None, dedup=dedup)
 
     def commit(
             self,