feat(query): paimon catalog support read and distributed write (#20132)

* feat(meta): add Paimon catalog metadata

Co-authored-by: Cursor <cursoragent@cursor.com>

* feat(query): parse Paimon catalogs

Co-authored-by: Cursor <cursoragent@cursor.com>

* feat(storage): implement read-only Paimon storage

Add the Paimon storage crate with catalog and table scans, split
serialization, predicate pushdown, Arrow sources, and integration tests.

Co-authored-by: Cursor <cursoragent@cursor.com>

* feat(query): register Paimon catalog and engine

Integrate CatalogCreator, the storage factory, query context, and
system.tables display.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(storage): finalize the Paimon core read plan

Fix timestamp and predicate pushdown, reuse the catalog in sources, add
pipeline and stateful tests, and reject CREATE TABLE for read-only catalogs.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(paimon): move read tests to service integration tests

- Load each table only once per source and preserve LIMIT and residual filters.
- Move pipeline and read tests to the databend-query service integration suite.
- Add split round-trip serialization and credential redaction tests.
- Use a local filesystem fixture for stateful tests.
- Give each thread its own TestFixture, remove the paimon_it_lock deadlock,
  and execute DDL through execute_command.
- Verify 8 service, 3 partition, and 2 security tests with UNIT_TEST=TRUE;
  formatting and clippy are clean.

* fix(paimon): repair the local filesystem fixture

The fixture had never completed successfully:
- Configure PaimonSparkSessionExtensions as required by paimon-spark.
- Provide only non-partition columns for the append_t static partition insert.
- Define the pk_t primary key with the primary-key table property because
  Spark cannot parse inline PRIMARY KEY syntax.

* test(paimon): make the stateful read test pass

The suite had not been exercised successfully before this fix:
- Make every shell script executable.
- Redirect fixture output so Spark logs do not pollute stdout.
- Use fully qualified paimon_fs.regression names because USE CATALOG does not
  persist across bendsql sessions.
- Regenerate the 01_read result to match actual output.

The official databend-test runner now verifies 01_0000_append_pk, including
append scans, primary-key deduplication, projection/filter/limit pushdown, and
read-only DDL rejection. The 00_catalog case still depends on a SHOW DATABASES
bug, while 02_cluster requires a multi-node cluster.

* fix(paimon): fix catalog SHOW DATABASES e2e test

* style(paimon): format the catalog adapter

* test(paimon): cover catalog SHOW TABLES

* test(paimon): add stateful SHOW TABLES coverage

* feat(storage): parse Paimon system tables

* feat(storage): read basic Paimon system tables

* feat(storage): read Paimon file metadata system tables

* feat(storage): decode Paimon stats and keys as typed JSON

* fix(storage): scan all partition files through the catalog

* feat(storage): read Paimon size and index system tables

* fix(storage): remove location conversion clippy warning

* test(paimon): validate every system table column count

* fix(storage): enable SQL reads from Paimon system tables

Rebuild PaimonSystemTable from the system table name in get_table_by_info,
lazily load the catalog and table from the TableInfo descriptor, and project
blocks in the source according to the plan to fix ORDER BY column mismatches.

* test(paimon): cover SQL reads from 11 system tables

* style(paimon): format physical_files_size

* test(paimon): fix multi-node cluster scan e2e test

* feat(paimon): add fixed-bucket write routing

Co-authored-by: Cursor <cursoragent@cursor.com>

* feat(paimon): transmit commit messages across nodes

Co-authored-by: Cursor <cursoragent@cursor.com>

* feat(paimon): implement parallel writers and centralized commit

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(paimon): skip centralized commit after failures

Co-authored-by: Cursor <cursoragent@cursor.com>

* feat(paimon): integrate table writes and snapshot commits

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(paimon): improve unsupported write configuration errors

Co-authored-by: Cursor <cursoragent@cursor.com>

* feat(paimon): redistribute writes by partition and bucket

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(paimon): preserve Merge Exchange for centralized commit

Co-authored-by: Cursor <cursoragent@cursor.com>

* test(paimon): cover the preserved SELECT Merge path

Co-authored-by: Cursor <cursoragent@cursor.com>

* test(paimon): cover local filesystem writes

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(paimon): use a portable warehouse path

Co-authored-by: Cursor <cursoragent@cursor.com>

* test(paimon): cover parallel cluster writes

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(paimon): finalize distributed write validation

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(paimon): correct distributed write routing and validation

* fix(paimon): address meta version, REST, and branch review issues

- Add v180 to META_CHANGE_LOG and cover CatalogMeta compatibility.
- Open tables through CatalogFactory to preserve rest_env.
- Resolve system tables against branch metadata paths.

* fix(paimon): strip quotes from dotted CONNECTION keys

Quoted identifiers such as "s3.region" retained their quotes in option keys,
preventing S3 FileIO from reading the region. Match Iceberg by stripping quotes
while building catalogs and tables, and add an offline regression test.

* ci: add Paimon MinIO integration validation

* fix: correct the Paimon S3 warehouse path

* fix: correct local Paimon MinIO validation

* fix(query): resolve Paimon CI failures

* fix(ci): isolate Paimon test dependencies

* fix(ci): stabilize Paimon cluster tests

* fix(query): harden Paimon system tables

---------

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: sundyli <543950155@qq.com>

Author: 3 people

.github/actions/test_sqllogic_paimon/action.yml

@@ -0,0 +1,48 @@
+name: "Test sqllogic Paimon"
+description: "Running Paimon sqllogic tests against MinIO"
+runs:
+  using: "composite"
+  steps:
+    - uses: ./.github/actions/setup_test
+      with:
+        artifacts: sqllogictests,meta,query
+
+    - name: Setup uv
+      uses: astral-sh/setup-uv@v7
+
+    - name: Prepare Paimon on MinIO
+      shell: bash
+      env:
+        AWS_ACCESS_KEY_ID: admin
+        AWS_SECRET_ACCESS_KEY: password
+        AWS_EC2_METADATA_DISABLED: true
+        PAIMON_WAREHOUSE: s3://paimon/warehouse
+        PAIMON_S3_ENDPOINT: http://127.0.0.1:9002
+        PAIMON_S3_ACCESS_KEY: admin
+        PAIMON_S3_SECRET_KEY: password
+      run: |
+        docker compose -f tests/sqllogictests/scripts/docker-compose-paimon.yml up -d
+        for i in $(seq 1 30); do
+          curl -sf http://127.0.0.1:9002/minio/health/live && break
+          echo "Waiting for MinIO... ($i)"
+          sleep 2
+        done
+        curl -sf http://127.0.0.1:9002/minio/health/live
+        aws --endpoint-url http://127.0.0.1:9002 s3 mb s3://paimon
+        uv run --script tests/sqllogictests/scripts/prepare_paimon_fs_data.py
+        # paimon-rust 0.2.0 checks directory paths as exact S3 keys.
+        for path in regression.db regression.db/{write_append,write_append_part,write_pk,write_pk_part}; do
+          aws --endpoint-url http://127.0.0.1:9002 s3api put-object \
+            --bucket paimon --key "warehouse/${path}"
+        done
+
+    - name: Run Paimon sqllogic tests
+      shell: bash
+      env:
+        TEST_HANDLERS: http
+      run: bash ./scripts/ci/ci-run-sqllogic-tests-without-sandbox.sh paimon
+
+    - name: Stop MinIO
+      if: always()
+      shell: bash
+      run: docker compose -f tests/sqllogictests/scripts/docker-compose-paimon.yml down -v

.github/actions/test_stateful_paimon_cluster_linux/action.yml

@@ -0,0 +1,27 @@
+name: "Test Stateful Paimon Cluster Linux"
+description: "Running Paimon stateful tests in cluster mode"
+runs:
+  using: "composite"
+  steps:
+    - uses: ./.github/actions/setup_test
+
+    - uses: actions/setup-java@v5
+      with:
+        distribution: "temurin"
+        java-version: "17"
+
+    - uses: astral-sh/setup-uv@v7
+
+    - name: Test setup
+      shell: bash
+      run: bash ./scripts/setup/dev_setup.sh -yd
+
+    - name: Run Paimon Stateful Tests with Cluster mode
+      shell: bash
+      run: bash ./scripts/ci/ci-run-stateful-paimon-tests-cluster.sh
+
+    - name: Upload failure
+      if: failure()
+      uses: ./.github/actions/artifact_failure
+      with:
+        name: test-stateful-paimon-cluster-linux

.github/workflows/reuse.linux.yml

@@ -349,6 +349,23 @@ jobs:
       - uses: ./.github/actions/test_stateless_cluster_linux
         timeout-minutes: 25
 
+  test_stateful_paimon_cluster:
+    needs: [build, check]
+    runs-on:
+      - self-hosted
+      - "${{ inputs.runner_arch }}"
+      - Linux
+      - 4c
+      - "${{ inputs.runner_provider }}"
+    steps:
+      - uses: actions/checkout@v6
+      - uses: ./.github/actions/setup_license
+        with:
+          runner_provider: ${{ inputs.runner_provider }}
+          type: ${{ inputs.license_type }}
+      - uses: ./.github/actions/test_stateful_paimon_cluster_linux
+        timeout-minutes: 25
+
   test_stateful_standalone:
     needs: [build, check]
     runs-on:

.github/workflows/reuse.sqllogic.yml

@@ -244,6 +244,27 @@ jobs:
         with:
           name: test-sqllogic-standalone-iceberg-tpch
 
+  standalone_paimon:
+    runs-on:
+      - self-hosted
+      - ${{ inputs.runner_arch }}
+      - Linux
+      - 4c
+      - "${{ inputs.runner_provider }}"
+    steps:
+      - uses: actions/checkout@v6
+      - uses: actions/setup-java@v5
+        with:
+          distribution: "temurin"
+          java-version: "17"
+      - uses: ./.github/actions/test_sqllogic_paimon
+        timeout-minutes: 15
+      - name: Upload failure
+        if: failure()
+        uses: ./.github/actions/artifact_failure
+        with:
+          name: test-sqllogic-standalone-paimon
+
   cluster:
     runs-on:
       - self-hosted