diff --git a/.asf.yaml b/.asf.yaml
index e4846c164..f27655367 100644
--- a/.asf.yaml
+++ b/.asf.yaml
@@ -37,6 +37,7 @@ github:
   enabled_merge_buttons:
     merge: false
     squash: true
+    squash_commit_message: PR_TITLE
     rebase: false
   rulesets:
     - name: "Branch Protection"
diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md
index 50d87f7f8..045141bcb 100644
--- a/.github/PULL_REQUEST_TEMPLATE.md
+++ b/.github/PULL_REQUEST_TEMPLATE.md
@@ -9,7 +9,10 @@ Linked issue: close #xxx
 
 ### Tests
 
-<!-- List UT and IT cases to verify this change -->
+<!--
+Describe tests added or updated and what they cover; explain if no test changes are needed.
+Omit execution commands, results, and reasons tests were not run locally.
+-->
 
 ### API and Format
 
diff --git a/.github/workflows/build_and_test.yaml b/.github/workflows/build_and_test.yaml
index f9548716d..eeb4fcfb7 100644
--- a/.github/workflows/build_and_test.yaml
+++ b/.github/workflows/build_and_test.yaml
@@ -46,6 +46,8 @@ jobs:
       - name: Run packaging argument tests
         shell: bash
         run: ci/scripts/test_packaging_args.sh
+      - name: Run issue and PR metadata validation tests
+        run: python3 -m unittest discover -s ci/scripts -p 'test_check_*.py' -v
       - name: Run asan_symbolize tests
         shell: bash
         run: ci/scripts/test_asan_symbolize.sh
diff --git a/.github/workflows/issue_template.yaml b/.github/workflows/issue_template.yaml
new file mode 100644
index 000000000..072e7001d
--- /dev/null
+++ b/.github/workflows/issue_template.yaml
@@ -0,0 +1,45 @@
+# 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.
+
+name: Issue Template Check
+
+on:
+  issues:
+    types: [opened, edited, reopened]
+
+concurrency:
+  group: ${{ github.repository }}-issue-template-${{ github.event.issue.number }}
+  cancel-in-progress: false
+
+permissions:
+  contents: read
+  issues: write
+
+jobs:
+  issue-template:
+    name: Check issue template
+    runs-on: ubuntu-24.04
+    timeout-minutes: 5
+    steps:
+      - name: Checkout validation script
+        uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
+        with:
+          persist-credentials: false
+      - name: Check issue and update template reminder
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+        run: python3 ci/scripts/check_issue_template.py
diff --git a/.github/workflows/pr_metadata.yaml b/.github/workflows/pr_metadata.yaml
new file mode 100644
index 000000000..9c804a882
--- /dev/null
+++ b/.github/workflows/pr_metadata.yaml
@@ -0,0 +1,46 @@
+# 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.
+
+name: PR Metadata Check
+
+on:
+  pull_request:
+    types: [opened, edited, reopened, synchronize]
+
+concurrency:
+  group: ${{ github.repository }}-${{ github.event.pull_request.number }}-${{ github.workflow }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  metadata:
+    name: PR metadata
+    runs-on: ubuntu-24.04
+    timeout-minutes: 5
+    steps:
+      - name: Checkout validation script
+        id: checkout
+        uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
+        with:
+          persist-credentials: false
+      - name: Check PR title
+        run: python3 ci/scripts/check_pr_metadata.py title
+      - name: Check PR description
+        if: ${{ !cancelled() && steps.checkout.outcome == 'success' }}
+        run: python3 ci/scripts/check_pr_metadata.py description
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 2355f95af..fa20b1a74 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -37,6 +37,11 @@ Use the [Conventional Commits](https://www.conventionalcommits.org/) format for
 <type>(<optional-scope>): <description>
 ```
 
+Use one of these lowercase types: `build`, `chore`, `ci`, `docs`, `feat`, `fix`, `perf`,
+`refactor`, `revert`, `style`, or `test`. The scope is optional. For a breaking change,
+add `!` immediately before the colon, for example `feat(api)!: remove a deprecated interface`.
+The PR title check enforces this type list and format.
+
 Examples:
 
 ```text
diff --git a/ci/scripts/check_issue_template.py b/ci/scripts/check_issue_template.py
new file mode 100644
index 000000000..1c97fb58f
--- /dev/null
+++ b/ci/scripts/check_issue_template.py
@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Check issue forms and maintain a single bot reminder for missing information."""
+
+import json
+import os
+import re
+import sys
+import urllib.error
+import urllib.request
+from typing import List
+
+from check_pr_metadata import parse_sections
+
+
+SEARCH_SECTION = "Search before asking"
+SEARCH_CONFIRMATION = (
+    "I searched in the [issues](https://github.com/apache/paimon-cpp/issues) "
+    "and found nothing similar."
+)
+BUG_FIELDS = (
+    "Paimon-cpp version", "Minimal reproduce step", "What doesn't meet your expectations?"
+)
+FEATURE_FIELDS = ("Motivation",)
+OPTIONAL_FIELDS = ("Solution", "Anything else?", "Are you willing to submit a PR?")
+ISSUE_SECTIONS = (SEARCH_SECTION,) + BUG_FIELDS + FEATURE_FIELDS + OPTIONAL_FIELDS
+NOTICE_MARKER = "<!-- paimon-cpp-issue-template-check -->"
+BOT_LOGIN = "github-actions[bot]"
+
+
+def check_issue_description(body: str) -> List[str]:
+    sections, headings = parse_sections(body, ISSUE_SECTIONS)
+    search_sections, _ = parse_sections(body, (SEARCH_SECTION,), include_fenced_code=False)
+    search_checked = any(
+        re.fullmatch(r"\s*[-*+]\s+\[[xX]\]\s+" + re.escape(SEARCH_CONFIRMATION) + r"\s*", line)
+        for line in search_sections.get(SEARCH_SECTION, [])
+    )
+    candidates = []
+    for fields in (BUG_FIELDS, FEATURE_FIELDS):
+        if not any(field in sections for field in fields):
+            continue
+        errors = []
+        if headings.count(SEARCH_SECTION) != 1 or not search_checked:
+            errors.append("Complete 'Search before asking' and check the search confirmation box.")
+        for field in fields:
+            if headings.count(field) != 1:
+                errors.append("Include exactly one '### " + field + "' section.")
+                continue
+            content = "\n".join(sections[field]).strip()
+            if not re.search(r"\w", content) or content.strip("*_ .!").lower() in (
+                "no response", "todo", "tbd", "na", "n/a"
+            ):
+                errors.append("Fill in '" + field + "' with the requested information.")
+        candidates.append(errors)
+    if not candidates:
+        return ["Use the Bug report or Feature template and fill in its required fields."]
+    return min(candidates, key=len)
+
+
+def github_request(method, path, payload=None):
+    data = None if payload is None else json.dumps(payload).encode("utf-8")
+    request = urllib.request.Request(
+        os.environ.get("GITHUB_API_URL", "https://api.github.com").rstrip("/") + path,
+        data=data,
+        headers={
+            "Accept": "application/vnd.github+json",
+            "Authorization": "Bearer " + os.environ["GITHUB_TOKEN"],
+            "Content-Type": "application/json",
+            "X-GitHub-Api-Version": "2022-11-28",
+        },
+        method=method,
+    )
+    with urllib.request.urlopen(request, timeout=30) as response:
+        return json.load(response)
+
+
+def find_notice(api, issue_path):
+    page = 1
+    while True:
+        comments = api("GET", issue_path + "/comments?per_page=100&page=" + str(page))
+        for comment in comments:
+            user = comment.get("user") or {}
+            if (
+                user.get("login") == BOT_LOGIN
+                and user.get("type") == "Bot"
+                and (comment.get("body") or "").startswith(NOTICE_MARKER)
+            ):
+                return comment
+        if len(comments) < 100:
+            return None
+        page += 1
+
+
+def update_notice(repository, issue_number, api, server_url="https://github.com"):
+    issue_path = "/repos/" + repository + "/issues/" + str(issue_number)
+    # Read the current body so queued events and reruns cannot restore an outdated reminder.
+    issue = api("GET", issue_path)
+    if "pull_request" in issue or issue.get("state") != "open":
+        return
+    errors = check_issue_description(issue.get("body") or "")
+    notice = find_notice(api, issue_path)
+    if errors:
+        new_issue_url = server_url.rstrip("/") + "/" + repository + "/issues/new?template="
+        message = (
+            NOTICE_MARKER + "\n\n"
+            "Thanks for opening this issue. Please use one of our issue templates:\n\n"
+            "- [Bug report](" + new_issue_url + "bug_report.yaml)\n"
+            "- [Feature request](" + new_issue_url + "feature.yaml)\n\n"
+            "You can edit this issue to include the template sections and required information. "
+            "There is no need to open a duplicate issue.\n\n"
+            "Please address the following:\n\n"
+            + "\n".join("- " + error for error in errors)
+        )
+    elif notice:
+        message = NOTICE_MARKER + "\n\nThanks for updating this issue to follow our issue template."
+    else:
+        return
+    if notice:
+        if notice["body"] != message:
+            api("PATCH", "/repos/" + repository + "/issues/comments/" + str(notice["id"]),
+                {"body": message})
+            print("Updated the issue template reminder.")
+    else:
+        api("POST", issue_path + "/comments", {"body": message})
+        print("Posted the issue template reminder.")
+
+
+def main() -> int:
+    try:
+        with open(os.environ["GITHUB_EVENT_PATH"], encoding="utf-8") as source:
+            event = json.load(source)
+        issue_number = event["issue"]["number"]
+        if not isinstance(issue_number, int) or issue_number <= 0:
+            raise ValueError("Invalid issue number")
+        update_notice(
+            os.environ["GITHUB_REPOSITORY"], issue_number, github_request,
+            os.environ.get("GITHUB_SERVER_URL", "https://github.com"),
+        )
+    except urllib.error.HTTPError as error:
+        print("GitHub API request failed with HTTP " + str(error.code) + ".", file=sys.stderr)
+        return 1
+    except (OSError, KeyError, TypeError, ValueError) as error:
+        print("Cannot check issue template: " + str(error), file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/ci/scripts/check_pr_metadata.py b/ci/scripts/check_pr_metadata.py
new file mode 100644
index 000000000..da41e6635
--- /dev/null
+++ b/ci/scripts/check_pr_metadata.py
@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Validate PR metadata from the GitHub Actions event without invoking a shell."""
+
+import argparse
+import json
+import os
+import re
+import sys
+from typing import Dict, List, Tuple
+
+
+SECTIONS = ("Purpose", "Tests", "API and Format", "Documentation", "Generative AI tooling")
+ALLOWED_TYPES = (
+    "build", "chore", "ci", "docs", "feat", "fix", "perf", "refactor", "revert", "style", "test"
+)
+TITLE_PATTERN = re.compile(
+    r"(?P<type>[^\W\d_][\w-]*)(?:\((?P<scope>[^()\r\n]+)\))?!?: \S[^\r\n]*"
+)
+HEADING_PATTERN = re.compile(r"^ {0,3}(#{1,6})[ \t]+(.+?)[ \t]*$")
+FENCE_PATTERN = re.compile(r"^ {0,3}(`{3,}|~{3,})(.*)$")
+
+
+def check_title(title: str) -> List[str]:
+    match = TITLE_PATTERN.fullmatch(title)
+    if match is None or (match.group("scope") is not None and not match.group("scope").strip()):
+        return [
+            "PR title must follow '<type>[optional scope][!]: <description>', "
+            "for example 'fix(parquet): handle empty input'."
+        ]
+    if match.group("type") not in ALLOWED_TYPES:
+        return ["PR title type must be one of these lowercase types: " + ", ".join(ALLOWED_TYPES) + "."]
+    return []
+
+
+def strip_comment(line: str, in_comment: bool) -> Tuple[str, bool]:
+    visible = ""
+    while line:
+        marker = "-->" if in_comment else "<!--"
+        index = line.find(marker)
+        if index < 0:
+            return visible + ("" if in_comment else line), in_comment
+        if not in_comment:
+            visible += line[:index]
+        line = line[index + len(marker) :]
+        in_comment = not in_comment
+    return visible, in_comment
+
+
+def parse_sections(
+    body: str, section_names: Tuple[str, ...] = SECTIONS, include_fenced_code: bool = True
+) -> Tuple[Dict[str, List[str]], List[str]]:
+    """Read level-three template sections, ignoring headings in comments or fences."""
+    sections: Dict[str, List[str]] = {}
+    headings: List[str] = []
+    current = None
+    fence = ""
+    in_comment = False
+    for line in body.splitlines():
+        if not fence:
+            line, in_comment = strip_comment(line, in_comment)
+        marker = FENCE_PATTERN.match(line)
+        if fence:
+            if (
+                marker
+                and marker.group(1)[0] == fence[0]
+                and len(marker.group(1)) >= len(fence)
+                and not marker.group(2).strip()
+            ):
+                fence = ""
+            elif current is not None and include_fenced_code:
+                sections[current].append(line)
+            continue
+        if marker:
+            fence = marker.group(1)
+            continue
+        heading = HEADING_PATTERN.match(line)
+        if heading and len(heading.group(1)) <= 3:
+            current = None
+            name = re.sub(r"[ \t]+#+$", "", heading.group(2))
+            if len(heading.group(1)) == 3 and name in section_names:
+                headings.append(name)
+                sections.setdefault(name, [])
+                current = name
+        elif current is not None:
+            sections[current].append(line)
+    return sections, headings
+
+
+def check_description(body: str) -> List[str]:
+    sections, headings = parse_sections(body)
+    errors = []
+    if tuple(headings) != SECTIONS:
+        errors.append(
+            "PR description must contain each template heading exactly once, in this order: "
+            + ", ".join("### " + name for name in SECTIONS)
+            + "."
+        )
+    for name in SECTIONS:
+        if name not in sections:
+            continue
+        lines = sections[name]
+        if name == "Purpose":
+            # An issue reference alone does not explain the purpose of the change.
+            lines = [
+                line for line in lines if not re.match(
+                    r"^\s*(?:Linked issue:|(?:close[sd]?|fix(?:e[sd])?|resolve[sd]?)\s+#\d+\s*$)",
+                    line, re.IGNORECASE
+                )
+            ]
+        content = "\n".join(lines).strip()
+        if not re.search(r"\w", content):
+            errors.append("Fill in '" + name + "'; explain why if it is not applicable.")
+        elif re.fullmatch(r"(?:TODO|TBD|N/?A)[.!]?", content, re.IGNORECASE):
+            errors.append("Replace the placeholder in '" + name + "' with an explanation.")
+        if any(
+            re.fullmatch(r"\s*(?:Linked issue:\s*)?close\s+#xxx\s*", line, re.IGNORECASE)
+            for line in sections[name]
+        ):
+            errors.append("Remove or replace the template's 'Linked issue: close #xxx' placeholder.")
+    disclosure = "\n".join(sections.get("Generative AI tooling", [])).strip()
+    if disclosure and not re.fullmatch(r"No\.?", disclosure, re.IGNORECASE):
+        if not re.search(
+            r"^(?:[-*][ \t]+)?Generated-by:",
+            disclosure, re.MULTILINE | re.IGNORECASE
+        ):
+            errors.append(
+                "In 'Generative AI tooling', write 'No' or include 'Generated-by:'."
+            )
+    return errors
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("check", choices=("title", "description"))
+    args = parser.parse_args()
+    try:
+        with open(os.environ["GITHUB_EVENT_PATH"], encoding="utf-8") as source:
+            event = json.load(source)
+        pr = event["pull_request"]
+        value = pr.get("title" if args.check == "title" else "body") or ""
+        if not isinstance(value, str):
+            raise ValueError("PR metadata must be text")
+    except (OSError, KeyError, TypeError, ValueError) as error:
+        print("Cannot read PR metadata: " + str(error), file=sys.stderr)
+        return 1
+    errors = check_title(value) if args.check == "title" else check_description(value)
+    for error in errors:
+        print("::error::" + error)
+    if not errors:
+        print("PR " + args.check + " is valid.")
+    return 1 if errors else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/ci/scripts/test_check_issue_template.py b/ci/scripts/test_check_issue_template.py
new file mode 100644
index 000000000..eda03efb2
--- /dev/null
+++ b/ci/scripts/test_check_issue_template.py
@@ -0,0 +1,284 @@
+#!/usr/bin/env python3
+# 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.
+
+import io
+import json
+import os
+import tempfile
+import unittest
+import urllib.error
+from pathlib import Path
+from unittest import mock
+from urllib.parse import parse_qs, urlsplit
+
+from check_issue_template import (
+    NOTICE_MARKER, check_issue_description, github_request, main, update_notice,
+)
+
+
+SEARCH = """### Search before asking
+
+- [x] I searched in the [issues](https://github.com/apache/paimon-cpp/issues) and found nothing similar.
+"""
+BUG_BODY = SEARCH + """
+### Paimon-cpp version
+
+0.4.0
+
+### Minimal reproduce step
+
+Read an empty table with the example reader.
+
+### What doesn't meet your expectations?
+
+The reader should return an empty result but reports an error.
+
+### Anything else?
+
+_No response_
+
+### Are you willing to submit a PR?
+
+- [ ] I'm willing to submit a PR!
+"""
+FEATURE_BODY = SEARCH + """
+### Motivation
+
+Support reading the new storage format.
+
+### Solution
+
+_No response_
+"""
+REPOSITORY = "apache/paimon-cpp"
+ISSUE_PATH = "/repos/" + REPOSITORY + "/issues/7"
+
+
+def bot_comment(body, comment_id=1):
+    return {
+        "id": comment_id, "body": body,
+        "user": {"login": "github-actions[bot]", "type": "Bot"},
+    }
+
+
+class FakeGitHub:
+    def __init__(self, body, comments=None):
+        self.issue = {"state": "open", "body": body}
+        self.comments = list(comments or [])
+        self.calls = []
+
+    def __call__(self, method, path, payload=None):
+        self.calls.append((method, path, payload))
+        if method == "GET" and path == ISSUE_PATH:
+            return self.issue
+        if method == "GET" and path.startswith(ISSUE_PATH + "/comments?"):
+            page = int(parse_qs(urlsplit(path).query)["page"][0])
+            return self.comments[(page - 1) * 100 : page * 100]
+        if method == "POST" and path == ISSUE_PATH + "/comments":
+            comment = bot_comment(payload["body"], len(self.comments) + 1)
+            self.comments.append(comment)
+            return comment
+        if method == "PATCH" and path.startswith("/repos/" + REPOSITORY + "/issues/comments/"):
+            comment_id = int(path.rsplit("/", 1)[1])
+            comment = next(comment for comment in self.comments if comment["id"] == comment_id)
+            comment["body"] = payload["body"]
+            return comment
+        raise AssertionError("Unexpected API request: " + method + " " + path)
+
+    def writes(self):
+        return [call for call in self.calls if call[0] != "GET"]
+
+
+class IssueTemplateTest(unittest.TestCase):
+    def test_bug_and_feature_forms_allow_empty_optional_fields(self):
+        self.assertEqual(check_issue_description(BUG_BODY), [])
+        self.assertEqual(check_issue_description(FEATURE_BODY), [])
+        self.assertEqual(check_issue_description(FEATURE_BODY.split("### Solution")[0]), [])
+
+    def test_blank_and_unstructured_issues(self):
+        for body in ("", "Please fix this bug.", SEARCH):
+            with self.subTest(body=body):
+                self.assertTrue(check_issue_description(body))
+
+    def test_required_bug_information_and_motivation(self):
+        for value, field in (
+            ("0.4.0", "Paimon-cpp version"),
+            ("Read an empty table with the example reader.", "Minimal reproduce step"),
+            ("The reader should return an empty result but reports an error.",
+             "What doesn't meet your expectations?"),
+        ):
+            for placeholder in ("", "_No response_", "<!-- hint -->", "TODO", "N/A"):
+                with self.subTest(field=field, placeholder=placeholder):
+                    errors = check_issue_description(BUG_BODY.replace(value, placeholder))
+                    self.assertTrue(any(field in error for error in errors))
+        self.assertTrue(check_issue_description(FEATURE_BODY.replace(
+            "Support reading the new storage format.", "_No response_"
+        )))
+
+    def test_missing_and_duplicate_sections(self):
+        for body in (
+            BUG_BODY.replace("### Paimon-cpp version", "### Version"),
+            BUG_BODY + "\n### Paimon-cpp version\n0.3.0\n",
+            FEATURE_BODY + SEARCH,
+            FEATURE_BODY.replace("### Motivation", "## Motivation"),
+        ):
+            with self.subTest(body=body):
+                self.assertTrue(check_issue_description(body))
+
+    def test_required_search_checkbox(self):
+        self.assertEqual(check_issue_description(FEATURE_BODY.replace("[x]", "[X]")), [])
+        for search in (
+            "", SEARCH.replace("[x]", "[ ]"),
+            "### Search before asking\n- [x] A different checkbox.\n",
+            "### Search before asking\n```markdown\n" + SEARCH.splitlines()[2] + "\n```\n",
+        ):
+            with self.subTest(search=search):
+                self.assertTrue(check_issue_description(FEATURE_BODY.replace(SEARCH, search)))
+
+    def test_comment_and_code_headings_do_not_supply_required_sections(self):
+        for begin, end in (("<!--", "-->"), ("```markdown", "```"), ("~~~", "~~~")):
+            body = FEATURE_BODY.replace("### Motivation", "### Other information")
+            body += "\n" + begin + "\n### Motivation\nA reason.\n" + end
+            with self.subTest(begin=begin):
+                self.assertTrue(check_issue_description(body))
+
+    def test_code_only_reproduction_steps_are_valid(self):
+        body = BUG_BODY.replace(
+            "Read an empty table with the example reader.",
+            "```cpp\nReadTable(\"empty-table\");\n```"
+        )
+        self.assertEqual(check_issue_description(body), [])
+
+    def test_template_field_names_match_checked_in_forms(self):
+        template_dir = Path(__file__).resolve().parents[2] / ".github" / "ISSUE_TEMPLATE"
+        for name, fields in (
+            ("bug_report.yaml", ("Search before asking", "Paimon-cpp version",
+                                 "Minimal reproduce step", "What doesn't meet your expectations?")),
+            ("feature.yaml", ("Search before asking", "Motivation")),
+        ):
+            template = (template_dir / name).read_text(encoding="utf-8")
+            for field in fields:
+                with self.subTest(template=name, field=field):
+                    self.assertIn("label: " + field, template)
+
+
+class IssueReminderTest(unittest.TestCase):
+    def test_valid_issue_has_no_comment(self):
+        api = FakeGitHub(BUG_BODY)
+        update_notice(REPOSITORY, 7, api)
+        self.assertEqual(api.writes(), [])
+
+    def test_invalid_issue_gets_one_reminder_with_template_links(self):
+        api = FakeGitHub("Please fix this.")
+        update_notice(REPOSITORY, 7, api)
+        update_notice(REPOSITORY, 7, api)
+        self.assertEqual(len(api.comments), 1)
+        self.assertEqual(len(api.writes()), 1)
+        body = api.comments[0]["body"]
+        self.assertTrue(body.startswith(NOTICE_MARKER))
+        self.assertIn("https://github.com/apache/paimon-cpp/issues/new?template=bug_report.yaml", body)
+        self.assertIn("template=feature.yaml", body)
+        self.assertIn("edit this issue", body)
+
+    def test_changed_errors_update_the_same_comment(self):
+        api = FakeGitHub("Please fix this.")
+        update_notice(REPOSITORY, 7, api)
+        api.issue["body"] = BUG_BODY.replace("0.4.0", "")
+        update_notice(REPOSITORY, 7, api)
+        self.assertEqual(len(api.comments), 1)
+        self.assertEqual([call[0] for call in api.writes()], ["POST", "PATCH"])
+        self.assertIn("Fill in 'Paimon-cpp version'", api.comments[0]["body"])
+
+    def test_corrected_issue_updates_the_existing_reminder_once(self):
+        api = FakeGitHub("Please fix this.")
+        update_notice(REPOSITORY, 7, api)
+        api.issue["body"] = FEATURE_BODY
+        update_notice(REPOSITORY, 7, api)
+        update_notice(REPOSITORY, 7, api)
+        self.assertEqual(len(api.comments), 1)
+        self.assertEqual([call[0] for call in api.writes()], ["POST", "PATCH"])
+        self.assertIn("Thanks for updating", api.comments[0]["body"])
+
+    def test_human_comment_cannot_spoof_bot_reminder(self):
+        comment = bot_comment(NOTICE_MARKER)
+        comment["user"] = {"login": "contributor", "type": "User"}
+        api = FakeGitHub("Please fix this.", [comment])
+        update_notice(REPOSITORY, 7, api)
+        self.assertEqual(len(api.comments), 2)
+        self.assertEqual(api.comments[0]["body"], NOTICE_MARKER)
+        self.assertEqual(api.writes()[0][0], "POST")
+
+    def test_comment_pagination_prevents_duplicate_reminders(self):
+        comments = [{"id": number, "body": "Discussion", "user": {"type": "User"}}
+                    for number in range(1, 101)]
+        api = FakeGitHub("Please fix this.", comments)
+        update_notice(REPOSITORY, 7, api)
+        update_notice(REPOSITORY, 7, api)
+        self.assertEqual(len(api.comments), 101)
+        self.assertEqual(len(api.writes()), 1)
+        self.assertTrue(any("page=2" in call[1] for call in api.calls))
+
+    def test_closed_issues_and_prs_are_ignored(self):
+        for extra in ({"state": "closed"}, {"pull_request": {}}):
+            api = FakeGitHub("Please fix this.")
+            api.issue.update(extra)
+            update_notice(REPOSITORY, 7, api)
+            self.assertEqual(len(api.calls), 1)
+            self.assertEqual(api.writes(), [])
+
+    def test_workflow_entrypoint_reads_latest_issue_instead_of_stale_event_body(self):
+        api = FakeGitHub(FEATURE_BODY)
+        with tempfile.TemporaryDirectory() as directory:
+            event_path = Path(directory) / "event.json"
+            event_path.write_text(json.dumps({"issue": {"number": 7, "body": "stale"}}))
+            with mock.patch.dict(os.environ, {
+                "GITHUB_EVENT_PATH": str(event_path), "GITHUB_REPOSITORY": REPOSITORY,
+            }), mock.patch("check_issue_template.github_request", side_effect=api):
+                self.assertEqual(main(), 0)
+        self.assertEqual(api.writes(), [])
+
+    def test_api_failure_is_reported_without_posting_a_comment(self):
+        failure = urllib.error.HTTPError("https://api.github.com", 403, "Forbidden", {}, None)
+        with tempfile.TemporaryDirectory() as directory:
+            event_path = Path(directory) / "event.json"
+            event_path.write_text(json.dumps({"issue": {"number": 7}}))
+            with mock.patch.dict(os.environ, {
+                "GITHUB_EVENT_PATH": str(event_path), "GITHUB_REPOSITORY": REPOSITORY,
+            }), mock.patch("check_issue_template.github_request", side_effect=failure), \
+                    mock.patch("sys.stderr", new_callable=io.StringIO) as stderr:
+                self.assertEqual(main(), 1)
+                self.assertIn("HTTP 403", stderr.getvalue())
+
+    def test_api_request_sends_structured_json(self):
+        response = io.BytesIO(b'{"id": 7}')
+        with mock.patch.dict(os.environ, {
+            "GITHUB_TOKEN": "test-token", "GITHUB_API_URL": "https://api.github.com",
+        }), mock.patch("urllib.request.urlopen", return_value=response) as opener:
+            self.assertEqual(github_request("POST", ISSUE_PATH + "/comments", {
+                "body": "Please use the template.\n\nMore information."
+            }), {"id": 7})
+        request = opener.call_args[0][0]
+        self.assertEqual(request.get_method(), "POST")
+        self.assertEqual(request.full_url, "https://api.github.com" + ISSUE_PATH + "/comments")
+        self.assertEqual(json.loads(request.data), {
+            "body": "Please use the template.\n\nMore information."
+        })
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/ci/scripts/test_check_pr_metadata.py b/ci/scripts/test_check_pr_metadata.py
new file mode 100644
index 000000000..ab1f386b8
--- /dev/null
+++ b/ci/scripts/test_check_pr_metadata.py
@@ -0,0 +1,215 @@
+#!/usr/bin/env python3
+# 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.
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from check_pr_metadata import check_description, check_title
+
+
+SCRIPT = Path(__file__).with_name("check_pr_metadata.py")
+TEMPLATE = Path(__file__).resolve().parents[2] / ".github" / "PULL_REQUEST_TEMPLATE.md"
+VALID_BODY = """### Purpose
+
+Handle empty input before decoding a batch.
+
+### Tests
+
+Added EmptyInputTest to cover empty and missing batches.
+
+### API and Format
+
+No API or storage format changes.
+
+### Documentation
+
+Not applicable: this fixes existing behavior.
+
+### Generative AI tooling
+
+No
+"""
+
+
+class PrMetadataTest(unittest.TestCase):
+    def test_conventional_titles(self):
+        for title in (
+            "fix: handle empty input", "feat(parquet): support bitmap filtering",
+            "feat(api)!: remove deprecated interface", "fix!: change input handling",
+            "fix(API)!: change input handling",
+            "docs: 更新文档",
+        ):
+            with self.subTest(title=title):
+                self.assertEqual(check_title(title), [])
+
+    def test_all_allowed_types_with_optional_scope_and_breaking_marker(self):
+        for kind in (
+            "build", "chore", "ci", "docs", "feat", "fix", "perf", "refactor",
+            "revert", "style", "test",
+        ):
+            for suffix in ("", "(reader)", "!", "(reader)!"):
+                title = kind + suffix + ": describe the change"
+                with self.subTest(title=title):
+                    self.assertEqual(check_title(title), [])
+
+    def test_unknown_types(self):
+        for kind in ("index", "custom-type", "feature", "hotfix", "deps", "release"):
+            with self.subTest(kind=kind):
+                errors = check_title(kind + ": describe the change")
+                self.assertTrue(errors)
+                self.assertIn("lowercase types", errors[0])
+                self.assertIn("feat", errors[0])
+
+    def test_types_must_be_lowercase(self):
+        for kind in ("Fix", "FIX", "fIx", "Feat", "CI", "Docs", "Revert"):
+            with self.subTest(kind=kind):
+                self.assertTrue(check_title(kind + "(reader)!: describe the change"))
+
+    def test_invalid_titles(self):
+        for title in (
+            "", "Fix empty input", "fix: ", "fix:   ", "fix:missing space",
+            "feat(): empty scope", "feat( ): blank scope", "fix(api: unclosed scope",
+            "feat(api)(reader): nested scopes", "fix: first line\nsecond line",
+        ):
+            with self.subTest(title=title):
+                self.assertTrue(check_title(title))
+
+    def test_complete_description_and_optional_issue(self):
+        self.assertEqual(check_description(VALID_BODY), [])
+        self.assertEqual(check_description(VALID_BODY.replace(
+            "### Purpose", "### Purpose\n\nLinked issue: close #123"
+        )), [])
+        self.assertEqual(check_description(VALID_BODY.replace(
+            "Added EmptyInputTest to cover empty and missing batches.",
+            "Not applicable: documentation-only change."
+        )), [])
+
+    def test_issue_placeholder_with_otherwise_complete_description(self):
+        body = VALID_BODY.replace("### Purpose", "### Purpose\n\nLinked issue: close #xxx")
+        self.assertTrue(any("#xxx" in error for error in check_description(body)))
+
+    def test_unfilled_template_and_empty_body(self):
+        self.assertTrue(check_description(""))
+        errors = check_description(TEMPLATE.read_text(encoding="utf-8"))
+        for section in ("Purpose", "Tests", "API and Format", "Documentation",
+                        "Generative AI tooling"):
+            self.assertTrue(any("'" + section + "'" in error for error in errors))
+        self.assertTrue(any("#xxx" in error for error in errors))
+
+    def test_missing_duplicate_out_of_order_or_wrong_level_headings(self):
+        for body in (
+            VALID_BODY.replace("### Tests", "### Test cases"),
+            VALID_BODY + "\n### Tests\nMore coverage.\n",
+            VALID_BODY.replace("### Tests", "### Documentation", 1),
+            VALID_BODY.replace("### Tests", "### TEMP")
+            .replace("### Documentation", "### Tests").replace("### TEMP", "### Documentation"),
+            VALID_BODY.replace("### Tests", "## Tests"),
+        ):
+            with self.subTest(body=body):
+                self.assertTrue(check_description(body))
+
+    def test_headings_in_comments_and_fences_do_not_fill_missing_sections(self):
+        for opening, closing in (("<!--", "-->"), ("```markdown", "```"), ("~~~", "~~~")):
+            body = VALID_BODY.replace("### Tests", "### Test cases")
+            body += "\n" + opening + "\n### Tests\nFake coverage.\n" + closing
+            with self.subTest(opening=opening):
+                self.assertTrue(check_description(body))
+
+    def test_comments_and_code_examples_in_filled_sections(self):
+        body = VALID_BODY.replace(
+            "Handle empty input before decoding a batch.",
+            "<!-- description hint --> Handle empty input.\n"
+            "````markdown\n### Tests\n<!-- an example, not an HTML comment\n```\n````"
+        )
+        body = body.replace("### Tests\n\nAdded", "### Tests ###\n\nAdded")
+        self.assertEqual(check_description(body), [])
+
+    def test_empty_sections_comments_and_placeholders(self):
+        for replacement in ("", "<!-- List UT and IT cases -->", "<!-- hint\nmore hint -->",
+                            "TODO", "TBD", "N/A", "---"):
+            body = VALID_BODY.replace(
+                "Added EmptyInputTest to cover empty and missing batches.", replacement
+            )
+            with self.subTest(replacement=replacement):
+                self.assertTrue(any("'Tests'" in error for error in check_description(body)))
+
+    def test_issue_link_alone_is_not_a_purpose(self):
+        for reference in ("Linked issue: close #123", "Closes #123", "Fixes #123"):
+            body = VALID_BODY.replace("Handle empty input before decoding a batch.", reference)
+            with self.subTest(reference=reference):
+                self.assertTrue(any("'Purpose'" in error for error in check_description(body)))
+
+    def test_ai_tool_disclosure(self):
+        for disclosure in ("No.", "Generated-by: Codex GPT-6", "- Generated-by: Some Tool 1.0"):
+            with self.subTest(disclosure=disclosure):
+                self.assertEqual(check_description(VALID_BODY.replace("\nNo\n", "\n" + disclosure)), [])
+        for disclosure in ("Yes", "Generated-by", "Generated by: Codex"):
+            with self.subTest(disclosure=disclosure):
+                self.assertTrue(check_description(VALID_BODY.replace("\nNo\n", "\n" + disclosure)))
+
+    def test_generated_by_content_is_not_validated(self):
+        for disclosure in (
+            "Generated-by:", "Generated-by:   ", "Generated-by: Codex",
+            "Generated-by:any content", "- Generated-by:",
+            "Generated-by:\nAdditional details without a tool version.",
+        ):
+            with self.subTest(disclosure=disclosure):
+                body = VALID_BODY.replace("\nNo\n", "\n" + disclosure)
+                self.assertEqual(check_description(body), [])
+
+    def run_check(self, directory, check, event):
+        event_file = Path(directory) / "event.json"
+        event_file.write_text(json.dumps(event), encoding="utf-8")
+        env = dict(os.environ, GITHUB_EVENT_PATH=str(event_file))
+        return subprocess.run(
+            [sys.executable, str(SCRIPT), check], env=env,
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True,
+            check=False,
+        )
+
+    def test_workflow_entrypoint_modes_and_null_body(self):
+        with tempfile.TemporaryDirectory() as directory:
+            event = {"pull_request": {"title": "invalid", "body": VALID_BODY}}
+            result = self.run_check(directory, "description", event)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+            result = self.run_check(directory, "title", event)
+            self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
+            self.assertIn("::error::", result.stdout)
+            result = self.run_check(directory, "description", {"pull_request": {"body": None}})
+            self.assertEqual(result.returncode, 1)
+            result = self.run_check(directory, "title", {})
+            self.assertEqual(result.returncode, 1)
+            self.assertIn("Cannot read PR metadata", result.stderr)
+
+    def test_metadata_is_data_and_not_shell_code(self):
+        with tempfile.TemporaryDirectory() as directory:
+            sentinel = Path(directory) / "must-not-exist"
+            title = "fix: $(touch " + str(sentinel) + ")"
+            event = {"pull_request": {"title": title, "body": VALID_BODY}}
+            result = self.run_check(directory, "title", event)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+            self.assertFalse(sentinel.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()