chore: add deprecate develop branch warning

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

.env.example

@@ -19,5 +19,3 @@ GT_GCS_BUCKET = GCS bucket
 GT_GCS_SCOPE  = GCS scope
 GT_GCS_CREDENTIAL_PATH = GCS credential path 
 GT_GCS_ENDPOINT = GCS end point
-# Settings for kafka wal test
-GT_KAFKA_ENDPOINTS = localhost:9092

.github/ISSUE_TEMPLATE/bug_report.yml

@@ -21,7 +21,6 @@ body:
         - Locking issue
         - Performance issue
         - Unexpected error
-        - User Experience
         - Other
     validations:
       required: true
@@ -34,14 +33,9 @@ body:
       multiple: true
       options:
         - Standalone mode
-        - Distributed Cluster
-        - Storage Engine
-        - Query Engine
-        - Table Engine
-        - Write Protocols
-        - MetaSrv
         - Frontend
         - Datanode
+        - Meta
         - Other
     validations:
       required: true
@@ -83,17 +77,6 @@ body:
     validations:
       required: true
 
-  - type: input
-    id: greptimedb
-    attributes:
-      label: What version of GreptimeDB did you use?
-      description: |
-        Please provide the version of GreptimeDB. For example:
-        0.5.1 etc. You can get it by executing command line `greptime --version`.
-      placeholder: "0.5.1"
-    validations:
-      required: true
-
   - type: textarea
     id: logs
     attributes:

.github/actions/build-greptime-images/action.yml

@@ -53,7 +53,7 @@ runs:
       uses: docker/setup-buildx-action@v2
 
     - name: Download amd64 artifacts
-      uses: actions/download-artifact@v4
+      uses: actions/download-artifact@v3
       with:
         name: ${{ inputs.amd64-artifact-name }}
 
@@ -66,7 +66,7 @@ runs:
         mv ${{ inputs.amd64-artifact-name }} amd64
 
     - name: Download arm64 artifacts
-      uses: actions/download-artifact@v4
+      uses: actions/download-artifact@v3
       if: ${{ inputs.arm64-artifact-name }}
       with:
         name: ${{ inputs.arm64-artifact-name }}

.github/actions/build-windows-artifacts/action.yml

@@ -25,7 +25,7 @@ inputs:
 runs:
   using: composite
   steps:
-    - uses: arduino/setup-protoc@v3
+    - uses: arduino/setup-protoc@v1
 
     - name: Install rust toolchain
       uses: dtolnay/rust-toolchain@master
@@ -38,7 +38,7 @@ runs:
       uses: Swatinem/rust-cache@v2
 
     - name: Install Python
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@v4
       with:
         python-version: '3.10'
 

.github/actions/publish-github-release/action.yml

@@ -15,7 +15,7 @@ runs:
     #   |- greptime-darwin-amd64-v0.5.0.sha256sum/greptime-darwin-amd64-v0.5.0.sha256sum
     #   ...
     - name: Download artifacts
-      uses: actions/download-artifact@v4
+      uses: actions/download-artifact@v3
 
     - name: Create git tag for release
       if: ${{ github.event_name != 'push' }} # Meaning this is a scheduled or manual workflow.

.github/actions/release-cn-artifacts/action.yaml

@@ -73,7 +73,7 @@ runs:
   using: composite
   steps:
     - name: Download artifacts
-      uses: actions/download-artifact@v4
+      uses: actions/download-artifact@v3
       with:
         path: ${{ inputs.artifacts-dir }}
 

.github/actions/upload-artifacts/action.yml

@@ -6,7 +6,7 @@ inputs:
     required: true
   target-file:
     description: The path of the target artifact
-    required: false
+    required: true
   version:
     description: Version of the artifact
     required: true
@@ -18,7 +18,6 @@ runs:
   using: composite
   steps:
     - name: Create artifacts directory
-      if: ${{ inputs.target-file != '' }}
       working-directory: ${{ inputs.working-dir }}
       shell: bash
       run: |
@@ -50,15 +49,15 @@ runs:
       run: Get-FileHash ${{ inputs.artifacts-dir }}.tar.gz -Algorithm SHA256 | select -ExpandProperty Hash > ${{ inputs.artifacts-dir }}.sha256sum
 
     # Note: The artifacts will be double zip compressed(related issue: https://github.com/actions/upload-artifact/issues/39).
-    # However, when we use 'actions/download-artifact' to download the artifacts, it will be automatically unzipped.
+    # However, when we use 'actions/download-artifact@v3' to download the artifacts, it will be automatically unzipped.
     - name: Upload artifacts
-      uses: actions/upload-artifact@v4
+      uses: actions/upload-artifact@v3
       with:
         name: ${{ inputs.artifacts-dir }}
         path: ${{ inputs.working-dir }}/${{ inputs.artifacts-dir }}.tar.gz
 
     - name: Upload checksum
-      uses: actions/upload-artifact@v4
+      uses: actions/upload-artifact@v3
       with:
         name: ${{ inputs.artifacts-dir }}.sha256sum
         path: ${{ inputs.working-dir }}/${{ inputs.artifacts-dir }}.sha256sum

.github/pull_request_template.md

@@ -15,6 +15,6 @@ Please explain IN DETAIL what the changes are in this PR and why they are needed
 
 - [ ]  I have written the necessary rustdoc comments.
 - [ ]  I have added the necessary unit tests and integration tests.
-- [x]  This PR does not require documentation updates.
+- [ ]  This PR does not require documentation updates.
 
 ## Refer to a related PR or issue link (optional)

.github/workflows/apidoc.yml

@@ -1,7 +1,7 @@
 on:
   push:
     branches:
-      - main
+      - develop
     paths-ignore:
       - 'docs/**'
       - 'config/**'
@@ -19,8 +19,8 @@ jobs:
   apidoc:
     runs-on: ubuntu-20.04
     steps:
-    - uses: actions/checkout@v4
-    - uses: arduino/setup-protoc@v3
+    - uses: actions/checkout@v3
+    - uses: arduino/setup-protoc@v1
       with:
         repo-token: ${{ secrets.GITHUB_TOKEN }}
     - uses: dtolnay/rust-toolchain@master

.github/workflows/dev-build.yml

@@ -101,7 +101,7 @@ jobs:
       version: ${{ steps.create-version.outputs.version }}
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -155,12 +155,12 @@ jobs:
     runs-on: ${{ needs.allocate-runners.outputs.linux-amd64-runner }}
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
       - name: Checkout greptimedb
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           repository: ${{ inputs.repository }}
           ref: ${{ inputs.commit }}
@@ -184,12 +184,12 @@ jobs:
     runs-on: ${{ needs.allocate-runners.outputs.linux-arm64-runner }}
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
       - name: Checkout greptimedb
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           repository: ${{ inputs.repository }}
           ref: ${{ inputs.commit }}
@@ -216,7 +216,7 @@ jobs:
     outputs:
       build-result: ${{ steps.set-build-result.outputs.build-result }}
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -247,7 +247,7 @@ jobs:
     runs-on: ubuntu-20.04
     continue-on-error: true
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -281,7 +281,7 @@ jobs:
     ]
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -306,7 +306,7 @@ jobs:
     ]
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -330,14 +330,14 @@ jobs:
     env:
       SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL_DEVELOP_CHANNEL }}
     steps:
-      - name: Notifiy dev build successful result
+      - name: Notifiy nightly build successful result
         uses: slackapi/slack-github-action@v1.23.0
         if: ${{ needs.release-images-to-dockerhub.outputs.build-result == 'success' }}
         with:
           payload: |
             {"text": "GreptimeDB's ${{ env.NEXT_RELEASE_VERSION }} build has completed successfully."}
 
-      - name: Notifiy dev build failed result
+      - name: Notifiy nightly build failed result
         uses: slackapi/slack-github-action@v1.23.0
         if: ${{ needs.release-images-to-dockerhub.outputs.build-result != 'success' }}
         with:

.github/workflows/develop.yml

@@ -9,9 +9,9 @@ on:
       - '.dockerignore'
       - 'docker/**'
       - '.gitignore'
-      - 'grafana/**'
   push:
     branches:
+      - develop
       - main
     paths-ignore:
       - 'docs/**'
@@ -20,7 +20,6 @@ on:
       - '.dockerignore'
       - 'docker/**'
       - '.gitignore'
-      - 'grafana/**'
   workflow_dispatch:
 
 name: CI
@@ -37,19 +36,20 @@ jobs:
     name: Spell Check with Typos
     runs-on: ubuntu-20.04
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
       - uses: crate-ci/typos@v1.13.10
 
   check:
     name: Check
+    if: github.event.pull_request.draft == false
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ windows-latest, ubuntu-20.04 ]
+        os: [ windows-latest-8-cores, ubuntu-20.04 ]
     timeout-minutes: 60
     steps:
-      - uses: actions/checkout@v4
-      - uses: arduino/setup-protoc@v3
+      - uses: actions/checkout@v3
+      - uses: arduino/setup-protoc@v1
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
       - uses: dtolnay/rust-toolchain@master
@@ -57,117 +57,46 @@ jobs:
           toolchain: ${{ env.RUST_TOOLCHAIN }}
       - name: Rust Cache
         uses: Swatinem/rust-cache@v2
-        with: 
-          # Shares across multiple jobs
-          # Shares with `Clippy` job
-          shared-key: "check-lint"
       - name: Run cargo check
         run: cargo check --locked --workspace --all-targets
 
   toml:
     name: Toml Check
+    if: github.event.pull_request.draft == false
     runs-on: ubuntu-20.04
     timeout-minutes: 60
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
       - uses: dtolnay/rust-toolchain@master
         with:
           toolchain: stable
       - name: Rust Cache
         uses: Swatinem/rust-cache@v2
-        with: 
-          # Shares across multiple jobs
-          shared-key: "check-toml"
       - name: Install taplo
-        run: cargo +stable install taplo-cli --version ^0.9 --locked
+        run: cargo +stable install taplo-cli --version ^0.8 --locked
       - name: Run taplo
         run: taplo format --check
 
-  build:
-    name: Build GreptimeDB binaries
+  sqlness:
+    name: Sqlness Test
+    if: github.event.pull_request.draft == false
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ ubuntu-20.04 ]
+        os: [ ubuntu-20.04-8-cores ]
     timeout-minutes: 60
     steps:
-      - uses: actions/checkout@v4
-      - uses: arduino/setup-protoc@v3
+      - uses: actions/checkout@v3
+      - uses: arduino/setup-protoc@v1
+        with:
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
       - uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.RUST_TOOLCHAIN }}
-      - uses: Swatinem/rust-cache@v2
-        with:
-          # Shares across multiple jobs
-          shared-key: "build-binaries"
-      - name: Build greptime binaries
-        shell: bash
-        run: cargo build
-      - name: Pack greptime binaries
-        shell: bash
-        run: |
-          mkdir bins && \
-          mv ./target/debug/greptime bins && \
-          mv ./target/debug/sqlness-runner bins
-      - name: Print greptime binaries info
-        run: ls -lh bins
-      - name: Upload artifacts
-        uses: ./.github/actions/upload-artifacts
-        with:
-          artifacts-dir: bins
-          version: current
-
-  sqlness:
-    name: Sqlness Test
-    needs: build
-    runs-on: ${{ matrix.os }}
-    strategy:
-      matrix:
-        os: [ ubuntu-20.04 ]
-    timeout-minutes: 60
-    steps:
-      - uses: actions/checkout@v4
-      - name: Download pre-built binaries
-        uses: actions/download-artifact@v4
-        with:
-          name: bins
-          path: .
-      - name: Unzip binaries
-        run: tar -xvf ./bins.tar.gz
+      - name: Rust Cache
+        uses: Swatinem/rust-cache@v2
       - name: Run sqlness
-        run: RUST_BACKTRACE=1 ./bins/sqlness-runner -c ./tests/cases --bins-dir ./bins
-      # FIXME: Logs cannot found be on failure (or even success). Need to figure out the cause.
-      - name: Upload sqlness logs
-        if: always()
-        uses: actions/upload-artifact@v3
-        with:
-          name: sqlness-logs
-          path: ${{ runner.temp }}/greptime-*.log
-          retention-days: 3
-
-  sqlness-kafka-wal:
-    name: Sqlness Test with Kafka Wal
-    needs: build
-    runs-on: ${{ matrix.os }}
-    strategy:
-      matrix:
-        os: [ ubuntu-20.04 ]
-    timeout-minutes: 60
-    steps:
-      - uses: actions/checkout@v4
-      - name: Download pre-built binaries
-        uses: actions/download-artifact@v4
-        with:
-          name: bins
-          path: .
-      - name: Unzip binaries
-        run: tar -xvf ./bins.tar.gz
-      - name: Setup kafka server
-        working-directory: tests-integration/fixtures/kafka
-        run: docker compose -f docker-compose-standalone.yml up -d --wait
-      - name: Run sqlness
-        run: RUST_BACKTRACE=1 ./bins/sqlness-runner -w kafka -k 127.0.0.1:9092 -c ./tests/cases --bins-dir ./bins
-      # FIXME: Logs cannot be found on failure (or even success). Need to figure out the cause.
+        run: cargo sqlness
       - name: Upload sqlness logs
         if: always()
         uses: actions/upload-artifact@v3
@@ -178,11 +107,12 @@ jobs:
 
   fmt:
     name: Rustfmt
+    if: github.event.pull_request.draft == false
     runs-on: ubuntu-20.04
     timeout-minutes: 60
     steps:
-      - uses: actions/checkout@v4
-      - uses: arduino/setup-protoc@v3
+      - uses: actions/checkout@v3
+      - uses: arduino/setup-protoc@v1
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
       - uses: dtolnay/rust-toolchain@master
@@ -191,19 +121,17 @@ jobs:
           components: rustfmt
       - name: Rust Cache
         uses: Swatinem/rust-cache@v2
-        with: 
-          # Shares across multiple jobs
-          shared-key: "check-rust-fmt"
       - name: Run cargo fmt
         run: cargo fmt --all -- --check
 
   clippy:
     name: Clippy
+    if: github.event.pull_request.draft == false
     runs-on: ubuntu-20.04
     timeout-minutes: 60
     steps:
-      - uses: actions/checkout@v4
-      - uses: arduino/setup-protoc@v3
+      - uses: actions/checkout@v3
+      - uses: arduino/setup-protoc@v1
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
       - uses: dtolnay/rust-toolchain@master
@@ -212,10 +140,6 @@ jobs:
           components: clippy
       - name: Rust Cache
         uses: Swatinem/rust-cache@v2
-        with: 
-          # Shares across multiple jobs
-          # Shares with `Check` job
-          shared-key: "check-lint"
       - name: Run cargo clippy
         run: cargo clippy --workspace --all-targets -- -D warnings
 
@@ -224,8 +148,8 @@ jobs:
     runs-on: ubuntu-20.04-8-cores
     timeout-minutes: 60
     steps:
-      - uses: actions/checkout@v4
-      - uses: arduino/setup-protoc@v3
+      - uses: actions/checkout@v3
+      - uses: arduino/setup-protoc@v1
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
       - uses: KyleMayes/install-llvm-action@v1
@@ -238,15 +162,12 @@ jobs:
           components: llvm-tools-preview
       - name: Rust Cache
         uses: Swatinem/rust-cache@v2
-        with:
-          # Shares cross multiple jobs
-          shared-key: "coverage-test"
       - name: Install latest nextest release
         uses: taiki-e/install-action@nextest
       - name: Install cargo-llvm-cov
         uses: taiki-e/install-action@cargo-llvm-cov
       - name: Install Python
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@v4
         with:
           python-version: '3.10'
       - name: Install PyArrow Package

.github/workflows/doc-issue.yml

@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-20.04
     steps:
       - name: create an issue in doc repo
-        uses: dacbd/create-issue-action@v1.2.1
+        uses: dacbd/create-issue-action@main
         with:
           owner: GreptimeTeam
           repo: docs
@@ -28,7 +28,7 @@ jobs:
     runs-on: ubuntu-20.04
     steps:
       - name: create an issue in cloud repo
-        uses: dacbd/create-issue-action@v1.2.1
+        uses: dacbd/create-issue-action@main
         with:
           owner: GreptimeTeam
           repo: greptimedb-cloud

.github/workflows/doc-label.yml

@@ -12,25 +12,9 @@ jobs:
     if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     runs-on: ubuntu-latest
     steps:
-    - uses: github/issue-labeler@v3.4
+    - uses: github/issue-labeler@v3.3
       with:
         configuration-path: .github/doc-label-config.yml
         enable-versioned-regex: false
         repo-token: ${{ secrets.GITHUB_TOKEN }}
         sync-labels: 1
-    - name: create an issue in doc repo
-      uses: dacbd/create-issue-action@v1.2.1
-      if: ${{ github.event.action == 'opened' && contains(github.event.pull_request.body, '- [ ]  This PR does not require documentation updates.') }}
-      with:
-        owner: GreptimeTeam
-        repo: docs
-        token: ${{ secrets.DOCS_REPO_TOKEN }}
-        title: Update docs for ${{ github.event.issue.title || github.event.pull_request.title }}
-        body: |
-          A document change request is generated from
-          ${{ github.event.issue.html_url || github.event.pull_request.html_url }}
-    - name: Check doc labels
-      uses: docker://agilepathway/pull-request-label-checker:latest
-      with:
-        one_of: Doc update required,Doc not needed
-        repo_token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/docs.yml

@@ -9,9 +9,9 @@ on:
       - '.dockerignore'
       - 'docker/**'
       - '.gitignore'
-      - 'grafana/**'
   push:
     branches:
+      - develop
       - main
     paths:
       - 'docs/**'
@@ -20,7 +20,6 @@ on:
       - '.dockerignore'
       - 'docker/**'
       - '.gitignore'
-      - 'grafana/**'
   workflow_dispatch:
 
 name: CI
@@ -33,34 +32,39 @@ jobs:
     name: Spell Check with Typos
     runs-on: ubuntu-20.04
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
       - uses: crate-ci/typos@v1.13.10
 
   check:
     name: Check
+    if: github.event.pull_request.draft == false
     runs-on: ubuntu-20.04
     steps:
       - run: 'echo "No action required"'
 
   fmt:
     name: Rustfmt
+    if: github.event.pull_request.draft == false
     runs-on: ubuntu-20.04
     steps:
       - run: 'echo "No action required"'
 
   clippy:
     name: Clippy
+    if: github.event.pull_request.draft == false
     runs-on: ubuntu-20.04
     steps:
       - run: 'echo "No action required"'
 
   coverage:
+    if: github.event.pull_request.draft == false
     runs-on: ubuntu-20.04
     steps:
       - run: 'echo "No action required"'
 
   sqlness:
     name: Sqlness Test
+    if: github.event.pull_request.draft == false
     runs-on: ubuntu-20.04
     steps:
       - run: 'echo "No action required"'

.github/workflows/license.yaml

@@ -3,7 +3,7 @@ name: License checker
 on:
   push:
     branches:
-    - main
+    - develop
   pull_request:
     types: [opened, synchronize, reopened, ready_for_review]
 jobs:
@@ -11,6 +11,6 @@ jobs:
     runs-on: ubuntu-20.04
     name: license-header-check
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v2
     - name: Check License Header
-      uses: korandoru/hawkeye@v4
+      uses: korandoru/hawkeye@v3

.github/workflows/nightly-build.yml

@@ -85,7 +85,7 @@ jobs:
       version: ${{ steps.create-version.outputs.version }}
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -137,7 +137,7 @@ jobs:
     ]
     runs-on: ${{ needs.allocate-runners.outputs.linux-amd64-runner }}
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -156,7 +156,7 @@ jobs:
     ]
     runs-on: ${{ needs.allocate-runners.outputs.linux-arm64-runner }}
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -179,7 +179,7 @@ jobs:
     outputs:
       nightly-build-result: ${{ steps.set-nightly-build-result.outputs.nightly-build-result }}
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -211,7 +211,7 @@ jobs:
     # The ACR have daily sync with DockerHub, so don't worry about the image not being updated.
     continue-on-error: true
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -245,7 +245,7 @@ jobs:
     ]
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -270,7 +270,7 @@ jobs:
     ]
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 

.github/workflows/nightly-ci.yml

@@ -24,8 +24,8 @@ jobs:
         os: [ windows-latest-8-cores ]
     timeout-minutes: 60
     steps:
-      - uses: actions/checkout@v4
-      - uses: arduino/setup-protoc@v3
+      - uses: actions/checkout@v4.1.0
+      - uses: arduino/setup-protoc@v1
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
       - uses: dtolnay/rust-toolchain@master
@@ -57,8 +57,8 @@ jobs:
     timeout-minutes: 60
     steps:
       - run: git config --global core.autocrlf false
-      - uses: actions/checkout@v4
-      - uses: arduino/setup-protoc@v3
+      - uses: actions/checkout@v4.1.0
+      - uses: arduino/setup-protoc@v1
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
       - name: Install Rust toolchain
@@ -71,7 +71,7 @@ jobs:
       - name: Install Cargo Nextest
         uses: taiki-e/install-action@nextest
       - name: Install Python
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@v4
         with:
           python-version: '3.10'
       - name: Install PyArrow Package

.github/workflows/nightly-funtional-tests.yml

@@ -13,7 +13,7 @@ jobs:
     runs-on: ubuntu-22.04
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 

.github/workflows/pr-title-checker.yml

@@ -13,7 +13,7 @@ jobs:
     runs-on: ubuntu-20.04
     timeout-minutes: 10
     steps:
-      - uses: thehanimo/pr-title-checker@v1.4.2
+      - uses: thehanimo/pr-title-checker@v1.3.4
         with:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           pass_on_octokit_error: false
@@ -22,7 +22,7 @@ jobs:
     runs-on: ubuntu-20.04
     timeout-minutes: 10
     steps:
-      - uses: thehanimo/pr-title-checker@v1.4.2
+      - uses: thehanimo/pr-title-checker@v1.3.4
         with:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           pass_on_octokit_error: false

.github/workflows/release-dev-builder-images.yaml

@@ -30,7 +30,7 @@ jobs:
     runs-on: ubuntu-20.04-16-cores
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 

.github/workflows/release.yml

@@ -91,7 +91,7 @@ env:
   # The scheduled version is '${{ env.NEXT_RELEASE_VERSION }}-nightly-YYYYMMDD', like v0.2.0-nigthly-20230313;
   NIGHTLY_RELEASE_PREFIX: nightly
   # Note: The NEXT_RELEASE_VERSION should be modified manually by every formal release.
-  NEXT_RELEASE_VERSION: v0.7.0
+  NEXT_RELEASE_VERSION: v0.6.0
 
 jobs:
   allocate-runners:
@@ -114,7 +114,7 @@ jobs:
       version: ${{ steps.create-version.outputs.version }}
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -168,7 +168,7 @@ jobs:
     ]
     runs-on: ${{ needs.allocate-runners.outputs.linux-amd64-runner }}
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -187,7 +187,7 @@ jobs:
     ]
     runs-on: ${{ needs.allocate-runners.outputs.linux-arm64-runner }}
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -226,7 +226,7 @@ jobs:
     ]
     if: ${{ inputs.build_macos_artifacts || github.event_name == 'push' || github.event_name == 'schedule' }}
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -240,11 +240,6 @@ jobs:
           disable-run-tests: ${{ env.DISABLE_RUN_TESTS }}
           artifacts-dir: ${{ matrix.artifacts-dir-prefix }}-${{ needs.allocate-runners.outputs.version }}
 
-      - name: Set build macos result
-        id: set-build-macos-result
-        run: |
-          echo "build-macos-result=success" >> $GITHUB_OUTPUT    
-
   build-windows-artifacts:
     name: Build Windows artifacts
     strategy:
@@ -267,7 +262,7 @@ jobs:
     steps:
       - run: git config --global core.autocrlf false
 
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -281,11 +276,6 @@ jobs:
           disable-run-tests: ${{ env.DISABLE_RUN_TESTS }}
           artifacts-dir: ${{ matrix.artifacts-dir-prefix }}-${{ needs.allocate-runners.outputs.version }}
 
-      - name: Set build windows result
-        id: set-build-windows-result
-        run: |
-          echo "build-windows-result=success" >> $GITHUB_OUTPUT    
-
   release-images-to-dockerhub:
     name: Build and push images to DockerHub
     if: ${{ inputs.release_images || github.event_name == 'push' || github.event_name == 'schedule' }}
@@ -296,7 +286,7 @@ jobs:
     ]
     runs-on: ubuntu-2004-16-cores
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -309,11 +299,6 @@ jobs:
           image-registry-password: ${{ secrets.DOCKERHUB_TOKEN }}
           version: ${{ needs.allocate-runners.outputs.version }}
 
-      - name: Set build image result
-        id: set-image-build-result
-        run: |
-          echo "build-image-result=success" >> $GITHUB_OUTPUT    
-
   release-cn-artifacts:
     name: Release artifacts to CN region
     if: ${{ inputs.release_images || github.event_name == 'push' || github.event_name == 'schedule' }}
@@ -331,7 +316,7 @@ jobs:
     # The ACR have daily sync with DockerHub, so don't worry about the image not being updated.
     continue-on-error: true
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -367,7 +352,7 @@ jobs:
     ]
     runs-on: ubuntu-20.04
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -390,7 +375,7 @@ jobs:
     ]
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -415,7 +400,7 @@ jobs:
     ]
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
@@ -428,29 +413,3 @@ jobs:
           aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
           aws-region: ${{ vars.EC2_RUNNER_REGION }}
           github-token: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }}
-
-  notification:
-    if: ${{ always() }} # Not requiring successful dependent jobs, always run.
-    name: Send notification to Greptime team
-    needs: [
-      release-images-to-dockerhub,
-      build-macos-artifacts,
-      build-windows-artifacts,
-    ]
-    runs-on: ubuntu-20.04
-    env:
-      SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL_DEVELOP_CHANNEL }}
-    steps:
-      - name: Notifiy release successful result
-        uses: slackapi/slack-github-action@v1.25.0
-        if: ${{ needs.release-images-to-dockerhub.outputs.build-image-result == 'success' && needs.build-windows-artifacts.outputs.build-windows-result == 'success' && needs.build-macos-artifacts.outputs.build-macos-result == 'success' }}
-        with:
-          payload: |
-            {"text": "GreptimeDB's release version has completed successfully."}
-
-      - name: Notifiy release failed result
-        uses: slackapi/slack-github-action@v1.25.0
-        if: ${{ needs.release-images-to-dockerhub.outputs.build-image-result != 'success' || needs.build-windows-artifacts.outputs.build-windows-result != 'success' || needs.build-macos-artifacts.outputs.build-macos-result != 'success' }}
-        with:
-          payload: |
-            {"text": "GreptimeDB's release version has failed, please check 'https://github.com/GreptimeTeam/greptimedb/actions/workflows/release.yml'."}

.github/workflows/size-label.yml

@@ -0,0 +1,25 @@
+name: size-labeler
+
+on: [pull_request_target]
+
+jobs:
+  labeler:
+    runs-on: ubuntu-latest
+    name: Label the PR size
+    permissions:
+      issues: write
+      pull-requests: write
+    steps:
+      - uses: codelytv/pr-size-labeler@v1
+        with:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          s_label: 'Size: S'
+          s_max_size: '100'
+          m_label: 'Size: M'
+          m_max_size: '500'
+          l_label: 'Size: L'
+          l_max_size: '1000'
+          xl_label: 'Size: XL'
+          fail_if_xl: 'false'
+          message_if_xl: ""
+          files_to_ignore: 'Cargo.lock'

.github/workflows/user-doc-label-checker.yml

@@ -0,0 +1,19 @@
+name: Check user doc labels
+on:
+  pull_request:
+    types:
+      - opened
+      - reopened
+      - labeled
+      - unlabeled
+
+jobs:
+
+  check_labels:
+    name: Check doc labels
+    runs-on: ubuntu-latest
+    steps:
+      - uses: docker://agilepathway/pull-request-label-checker:latest
+        with:
+          one_of: Doc update required,Doc not needed
+          repo_token: ${{ secrets.GITHUB_TOKEN }}

CONTRIBUTING.md

@@ -10,7 +10,7 @@ Follow our [README](https://github.com/GreptimeTeam/greptimedb#readme) to get th
 
 It can feel intimidating to contribute to a complex project, but it can also be exciting and fun. These general notes will help everyone participate in this communal activity.
 
-- Follow the [Code of Conduct](https://github.com/GreptimeTeam/greptimedb/blob/main/CODE_OF_CONDUCT.md)
+- Follow the [Code of Conduct](https://github.com/GreptimeTeam/greptimedb/blob/develop/CODE_OF_CONDUCT.md)
 - Small changes make huge differences. We will happily accept a PR making a single character change if it helps move forward. Don't wait to have everything working.
 - Check the closed issues before opening your issue.
 - Try to follow the existing style of the code.
@@ -26,7 +26,7 @@ Pull requests are great, but we accept all kinds of other help if you like. Such
 
 ## Code of Conduct
 
-Also, there are things that we are not looking for because they don't match the goals of the product or benefit the community. Please read [Code of Conduct](https://github.com/GreptimeTeam/greptimedb/blob/main/CODE_OF_CONDUCT.md); we hope everyone can keep good manners and become an honored member.
+Also, there are things that we are not looking for because they don't match the goals of the product or benefit the community. Please read [Code of Conduct](https://github.com/GreptimeTeam/greptimedb/blob/develop/CODE_OF_CONDUCT.md); we hope everyone can keep good manners and become an honored member.
 
 ## License
 

Cargo.toml

@@ -29,11 +29,9 @@ members = [
     "src/common/time",
     "src/common/decimal",
     "src/common/version",
-    "src/common/wal",
     "src/datanode",
     "src/datatypes",
     "src/file-engine",
-    "src/flow",
     "src/frontend",
     "src/log-store",
     "src/meta-client",
@@ -54,14 +52,13 @@ members = [
     "src/store-api",
     "src/table",
     "src/index",
-    "tests-fuzz",
     "tests-integration",
     "tests/runner",
 ]
 resolver = "2"
 
 [workspace.package]
-version = "0.6.0"
+version = "0.5.0"
 edition = "2021"
 license = "Apache-2.0"
 
@@ -71,18 +68,15 @@ aquamarine = "0.3"
 arrow = { version = "47.0" }
 arrow-array = "47.0"
 arrow-flight = "47.0"
-arrow-ipc = "47.0"
 arrow-schema = { version = "47.0", features = ["serde"] }
 async-stream = "0.3"
 async-trait = "0.1"
-axum = { version = "0.6", features = ["headers"] }
 base64 = "0.21"
 bigdecimal = "0.4.2"
 bitflags = "2.4.1"
 bytemuck = "1.12"
 bytes = { version = "1.5", features = ["serde"] }
 chrono = { version = "0.4", features = ["serde"] }
-clap = { version = "4.4", features = ["derive"] }
 dashmap = "5.4"
 datafusion = { git = "https://github.com/apache/arrow-datafusion.git", rev = "26e43acac3a96cec8dd4c8365f22dfb1a84306e9" }
 datafusion-common = { git = "https://github.com/apache/arrow-datafusion.git", rev = "26e43acac3a96cec8dd4c8365f22dfb1a84306e9" }
@@ -96,14 +90,13 @@ etcd-client = "0.12"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "96f1f0404f421ee560a4310c73c5071e49168168" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "a31ea166fc015ea7ff111ac94e26c3a5d64364d2" }
 humantime-serde = "1.1"
 itertools = "0.10"
 lazy_static = "1.4"
 meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev = "abbd357c1e193cd270ea65ee7652334a150b628f" }
 mockall = "0.11.4"
 moka = "0.12"
-num_cpus = "1.16"
 once_cell = "1.18"
 opentelemetry-proto = { git = "https://github.com/waynexia/opentelemetry-rust.git", rev = "33841b38dda79b15f2024952be5f32533325ca02", features = [
     "gen-tonic",
@@ -115,10 +108,10 @@ paste = "1.0"
 pin-project = "1.0"
 prometheus = { version = "0.13.3", features = ["process"] }
 prost = "0.12"
-raft-engine = { version = "0.4.1", default-features = false }
+raft-engine = { git = "https://github.com/tikv/raft-engine.git", rev = "22dfb426cd994602b57725ef080287d3e53db479" }
 rand = "0.8"
 regex = "1.8"
-regex-automata = { version = "0.2", features = ["transducer"] }
+regex-automata = { version = "0.1", features = ["transducer"] }
 reqwest = { version = "0.11", default-features = false, features = [
     "json",
     "rustls-tls-native-roots",
@@ -128,10 +121,8 @@ rskafka = "0.5"
 rust_decimal = "1.33"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-serde_with = "3"
-smallvec = { version = "1", features = ["serde"] }
+smallvec = "1"
 snafu = "0.7"
-sysinfo = "0.30"
 # on branch v0.38.x
 sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "6a93567ae38d42be5c8d08b13c8ff4dde26502ef", features = [
     "visitor",
@@ -164,6 +155,7 @@ common-grpc-expr = { path = "src/common/grpc-expr" }
 common-macro = { path = "src/common/macro" }
 common-mem-prof = { path = "src/common/mem-prof" }
 common-meta = { path = "src/common/meta" }
+common-pprof = { path = "src/common/pprof" }
 common-procedure = { path = "src/common/procedure" }
 common-procedure-test = { path = "src/common/procedure-test" }
 common-query = { path = "src/common/query" }
@@ -173,12 +165,10 @@ common-telemetry = { path = "src/common/telemetry" }
 common-test-util = { path = "src/common/test-util" }
 common-time = { path = "src/common/time" }
 common-version = { path = "src/common/version" }
-common-wal = { path = "src/common/wal" }
 datanode = { path = "src/datanode" }
 datatypes = { path = "src/datatypes" }
 file-engine = { path = "src/file-engine" }
 frontend = { path = "src/frontend" }
-index = { path = "src/index" }
 log-store = { path = "src/log-store" }
 meta-client = { path = "src/meta-client" }
 meta-srv = { path = "src/meta-srv" }
@@ -189,7 +179,6 @@ operator = { path = "src/operator" }
 partition = { path = "src/partition" }
 plugins = { path = "src/plugins" }
 promql = { path = "src/promql" }
-puffin = { path = "src/puffin" }
 query = { path = "src/query" }
 script = { path = "src/script" }
 servers = { path = "src/servers" }

Makefile

@@ -65,7 +65,7 @@ endif
 build: ## Build debug version greptime.
 	cargo ${CARGO_EXTENSION} build ${CARGO_BUILD_OPTS}
 
-.PHONY: build-by-dev-builder
+.POHNY: build-by-dev-builder
 build-by-dev-builder: ## Build greptime by dev-builder.
 	docker run --network=host \
 	-v ${PWD}:/greptimedb -v ${CARGO_REGISTRY_CACHE}:/root/.cargo/registry \
@@ -144,12 +144,11 @@ multi-platform-buildx: ## Create buildx multi-platform builder.
 	docker buildx inspect ${BUILDX_BUILDER_NAME} || docker buildx create --name ${BUILDX_BUILDER_NAME} --driver docker-container --bootstrap --use
 
 ##@ Test
-.PHONY: test
 test: nextest ## Run unit and integration tests.
 	cargo nextest run ${NEXTEST_OPTS}
 
-.PHONY: nextest
-nextest: ## Install nextest tools.
+.PHONY: nextest ## Install nextest tools.
+nextest:
 	cargo --list | grep nextest || cargo install cargo-nextest --locked
 
 .PHONY: sqlness-test

README.md

@@ -1,8 +1,8 @@
 <p align="center">
   <picture>
-    <source media="(prefers-color-scheme: light)" srcset="https://cdn.jsdelivr.net/gh/GreptimeTeam/greptimedb@main/docs/logo-text-padding.png">
-    <source media="(prefers-color-scheme: dark)" srcset="https://cdn.jsdelivr.net/gh/GreptimeTeam/greptimedb@main/docs/logo-text-padding-dark.png">
-    <img alt="GreptimeDB Logo" src="https://cdn.jsdelivr.net/gh/GreptimeTeam/greptimedb@main/docs/logo-text-padding.png" width="400px">
+    <source media="(prefers-color-scheme: light)" srcset="https://cdn.jsdelivr.net/gh/GreptimeTeam/greptimedb@develop/docs/logo-text-padding.png">
+    <source media="(prefers-color-scheme: dark)" srcset="https://cdn.jsdelivr.net/gh/GreptimeTeam/greptimedb@develop/docs/logo-text-padding-dark.png">
+    <img alt="GreptimeDB Logo" src="https://cdn.jsdelivr.net/gh/GreptimeTeam/greptimedb@develop/docs/logo-text-padding.png" width="400px">
   </picture>
 </p>
 
@@ -12,11 +12,11 @@
 </h3>
 
 <p align="center">
-    <a href="https://codecov.io/gh/GrepTimeTeam/greptimedb"><img src="https://codecov.io/gh/GrepTimeTeam/greptimedb/branch/main/graph/badge.svg?token=FITFDI3J3C"></img></a>
+    <a href="https://codecov.io/gh/GrepTimeTeam/greptimedb"><img src="https://codecov.io/gh/GrepTimeTeam/greptimedb/branch/develop/graph/badge.svg?token=FITFDI3J3C"></img></a>
     &nbsp;
     <a href="https://github.com/GreptimeTeam/greptimedb/actions/workflows/develop.yml"><img src="https://github.com/GreptimeTeam/greptimedb/actions/workflows/develop.yml/badge.svg" alt="CI"></img></a>
     &nbsp;
-    <a href="https://github.com/greptimeTeam/greptimedb/blob/main/LICENSE"><img src="https://img.shields.io/github/license/greptimeTeam/greptimedb"></a>
+    <a href="https://github.com/greptimeTeam/greptimedb/blob/develop/LICENSE"><img src="https://img.shields.io/github/license/greptimeTeam/greptimedb"></a>
 </p>
 
 <p align="center">
@@ -27,19 +27,26 @@
     <a href="https://greptime.com/slack"><img src="https://img.shields.io/badge/slack-GreptimeDB-0abd59?logo=slack" alt="slack" /></a>
 </p>
 
+> [!WARNING]  
+> Our default branch has changed from `develop` to `main` (issue [#3025](https://github.com/GreptimeTeam/greptimedb/issues/3025)). Please update your local repository to use the `main` branch.
+
 ## What is GreptimeDB
 
-GreptimeDB is an open-source time-series database focusing on efficiency, scalability, and analytical capabilities.
-It's designed to work on infrastructure of the cloud era, and users benefit from its elasticity and commodity storage.
+GreptimeDB is an open-source time-series database with a special focus on
+scalability, analytical capabilities and efficiency. It's designed to work on
+infrastructure of the cloud era, and users benefit from its elasticity and commodity
+storage.
 
-Our core developers have been building time-series data platforms for years. Based on their best-practices, GreptimeDB is born to give you:
+Our core developers have been building time-series data platform
+for years. Based on their best-practices, GreptimeDB is born to give you:
 
-- Optimized columnar layout for handling time-series data; compacted, compressed, and stored on various storage backends, particularly cloud object storage with 50x cost efficiency.
-- Fully open-source distributed cluster architecture that harnesses the power of cloud-native elastic computing resources.
-- Seamless scalability from a standalone binary at edge to a robust, highly available distributed cluster in cloud, with a transparent experience for both developers and administrators.
-- Native SQL and PromQL for queries, and Python scripting to facilitate complex analytical tasks.
-- Flexible indexing capabilities and distributed, parallel-processing query engine, tackling high cardinality issues down.
-- Widely adopted database protocols and APIs, including MySQL, PostgreSQL, and Prometheus Remote Storage, etc.
+- A standalone binary that scales to highly-available distributed cluster, providing a transparent experience for cluster users
+- Optimized columnar layout for handling time-series data; compacted, compressed, and stored on various storage backends
+- Flexible indexes, tackling high cardinality issues down
+- Distributed, parallel query execution, leveraging elastic computing resource
+- Native SQL, and Python scripting for advanced analytical scenarios
+- Widely adopted database protocols and APIs, native PromQL supports
+- Extensible table engine architecture for extensive workloads
 
 ## Quick Start
 
@@ -127,16 +134,12 @@ To write and query data, GreptimeDB is compatible with multiple [protocols and c
 
 - [GreptimeDB C++ Client](https://github.com/GreptimeTeam/greptimedb-client-cpp)
 - [GreptimeDB Erlang Client](https://github.com/GreptimeTeam/greptimedb-client-erl)
-- [GreptimeDB Go Ingester](https://github.com/GreptimeTeam/greptimedb-ingester-go)
-- [GreptimeDB Java Ingester](https://github.com/GreptimeTeam/greptimedb-ingester-java)
+- [GreptimeDB Go Client](https://github.com/GreptimeTeam/greptimedb-client-go)
+- [GreptimeDB Java Client](https://github.com/GreptimeTeam/greptimedb-client-java)
 - [GreptimeDB Python Client](https://github.com/GreptimeTeam/greptimedb-client-py) (WIP)
 - [GreptimeDB Rust Client](https://github.com/GreptimeTeam/greptimedb-client-rust)
 - [GreptimeDB JavaScript Client](https://github.com/GreptimeTeam/greptime-js-sdk)
 
-### Grafana Dashboard
-
-Our official Grafana dashboard is available at [grafana](./grafana/README.md) directory.
-
 ## Project Status
 
 This project is in its early stage and under heavy development. We move fast and
@@ -165,17 +168,18 @@ In addition, you may:
 
 ## License
 
-GreptimeDB uses the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0.txt) to strike a balance between
+GreptimeDB uses the [Apache 2.0 license][1] to strike a balance between
 open contributions and allowing you to use the software however you want.
 
+[1]: <https://github.com/greptimeTeam/greptimedb/blob/develop/LICENSE>
+
 ## Contributing
 
 Please refer to [contribution guidelines](CONTRIBUTING.md) for more information.
 
 ## Acknowledgement
-
-- GreptimeDB uses [Apache Arrow™](https://arrow.apache.org/) as the memory model and [Apache Parquet™](https://parquet.apache.org/) as the persistent file format.
-- GreptimeDB's query engine is powered by [Apache Arrow DataFusion™](https://arrow.apache.org/datafusion/).
-- [Apache OpenDAL™](https://opendal.apache.org) gives GreptimeDB a very general and elegant data access abstraction layer.
+- GreptimeDB uses [Apache Arrow](https://arrow.apache.org/) as the memory model and [Apache Parquet](https://parquet.apache.org/) as the persistent file format.
+- GreptimeDB's query engine is powered by [Apache Arrow DataFusion](https://github.com/apache/arrow-datafusion).
+- [Apache OpenDAL (incubating)](https://opendal.apache.org) gives GreptimeDB a very general and elegant data access abstraction layer.
 - GreptimeDB's meta service is based on [etcd](https://etcd.io/).
 - GreptimeDB uses [RustPython](https://github.com/RustPython/RustPython) for experimental embedded python scripting.

benchmarks/Cargo.toml

@@ -7,7 +7,7 @@ license.workspace = true
 [dependencies]
 arrow.workspace = true
 chrono.workspace = true
-clap.workspace = true
+clap = { version = "4.0", features = ["derive"] }
 client.workspace = true
 futures-util.workspace = true
 indicatif = "0.17.1"

benchmarks/src/bin/nyc-taxi.rs

@@ -258,7 +258,7 @@ fn create_table_expr(table_name: &str) -> CreateTableExpr {
         catalog_name: CATALOG_NAME.to_string(),
         schema_name: SCHEMA_NAME.to_string(),
         table_name: table_name.to_string(),
-        desc: String::default(),
+        desc: "".to_string(),
         column_defs: vec![
             ColumnDef {
                 name: "VendorID".to_string(),

config/datanode.example.toml

@@ -14,7 +14,7 @@ require_lease_before_startup = false
 
 # Initialize all regions in the background during the startup.
 # By default, it provides services after all regions have been initialized.
-init_regions_in_background = false
+initialize_region_in_background = false
 
 [heartbeat]
 # Interval for sending heartbeat messages to the Metasrv, 3 seconds by default.
@@ -34,7 +34,11 @@ connect_timeout = "1s"
 tcp_nodelay = true
 
 # WAL options.
+# Currently, users are expected to choose the wal through the provider field. 
+# When a wal provider is chose, the user should comment out all other wal config 
+# except those corresponding to the chosen one.
 [wal]
+# WAL data directory
 provider = "raft_engine"
 
 # Raft-engine wal options, see `standalone.example.toml`.
@@ -47,10 +51,9 @@ sync_write = false
 
 # Kafka wal options, see `standalone.example.toml`.
 # broker_endpoints = ["127.0.0.1:9092"]
-# Warning: Kafka has a default limit of 1MB per message in a topic.
-# max_batch_size = "1MB"
+# max_batch_size = "4MB"
 # linger = "200ms"
-# consumer_wait_timeout = "100ms"
+# produce_record_timeout = "100ms"
 # backoff_init = "500ms"
 # backoff_max = "10s"
 # backoff_base = 2
@@ -94,18 +97,15 @@ compress_manifest = false
 max_background_jobs = 4
 # Interval to auto flush a region if it has not flushed yet.
 auto_flush_interval = "1h"
-# Global write buffer size for all regions. If not set, it's default to 1/8 of OS memory with a max limitation of 1GB.
+# Global write buffer size for all regions.
 global_write_buffer_size = "1GB"
-# Global write buffer size threshold to reject write requests. If not set, it's default to 2 times of `global_write_buffer_size`
+# Global write buffer size threshold to reject write requests (default 2G).
 global_write_buffer_reject_size = "2GB"
-# Cache size for SST metadata. Setting it to 0 to disable the cache.
-# If not set, it's default to 1/32 of OS memory with a max limitation of 128MB.
+# Cache size for SST metadata (default 128MB). Setting it to 0 to disable the cache.
 sst_meta_cache_size = "128MB"
-# Cache size for vectors and arrow arrays. Setting it to 0 to disable the cache.
-# If not set, it's default to 1/16 of OS memory with a max limitation of 512MB.
+# Cache size for vectors and arrow arrays (default 512MB). Setting it to 0 to disable the cache.
 vector_cache_size = "512MB"
-# Cache size for pages of SST row groups. Setting it to 0 to disable the cache.
-# If not set, it's default to 1/16 of OS memory with a max limitation of 512MB.
+# Cache size for pages of SST row groups (default 512MB). Setting it to 0 to disable the cache.
 page_cache_size = "512MB"
 # Buffer size for SST writing.
 sst_write_buffer_size = "8MB"
@@ -116,27 +116,6 @@ sst_write_buffer_size = "8MB"
 scan_parallelism = 0
 # Capacity of the channel to send data from parallel scan tasks to the main task (default 32).
 parallel_scan_channel_size = 32
-# Whether to allow stale WAL entries read during replay.
-allow_stale_entries = false
-
-[region_engine.mito.inverted_index]
-# Whether to create the index on flush.
-# - "auto": automatically
-# - "disable": never
-create_on_flush = "auto"
-# Whether to create the index on compaction.
-# - "auto": automatically
-# - "disable": never
-create_on_compaction = "auto"
-# Whether to apply the index on query
-# - "auto": automatically
-# - "disable": never
-apply_on_query = "auto"
-# Memory threshold for performing an external sort during index creation.
-# Setting to empty will disable external sorting, forcing all sorting operations to happen in memory.
-mem_threshold_on_create = "64MB"
-# File system path to store intermediate files for external sorting (default `{data_home}/index_intermediate`).
-intermediate_path = ""
 
 # Log options, see `standalone.example.toml`
 # [logging]
@@ -150,10 +129,11 @@ intermediate_path = ""
 # [export_metrics]
 # whether enable export metrics, default is false
 # enable = false
+# The url of metrics export endpoint, default is `frontend` default HTTP endpoint.
+# endpoint = "127.0.0.1:4000"
+# The database name of exported metrics stores, user needs to specify a valid database
+# db = ""
 # The interval of export metrics
 # write_interval = "30s"
-# [export_metrics.remote_write]
-# The url the metrics send to. The url is empty by default, url example: `http://127.0.0.1:4000/v1/prometheus/write?db=information_schema`
-# url = ""
 # HTTP headers of Prometheus remote-write carry
 # headers = {}

config/frontend.example.toml

@@ -57,9 +57,6 @@ enable = true
 # Prometheus remote storage options, see `standalone.example.toml`.
 [prom_store]
 enable = true
-# Whether to store the data from Prometheus remote write in metric engine.
-# true by default
-with_metric_engine = true
 
 # Metasrv client options, see `datanode.example.toml`.
 [meta_client]
@@ -69,13 +66,6 @@ timeout = "3s"
 ddl_timeout = "10s"
 connect_timeout = "1s"
 tcp_nodelay = true
-# The configuration about the cache of the Metadata.
-# default: 100000
-metadata_cache_max_capacity = 100000
-# default: 10m
-metadata_cache_ttl = "10m"
-# default: 5m
-metadata_cache_tti = "5m"
 
 # Log options, see `standalone.example.toml`
 # [logging]
@@ -97,8 +87,11 @@ tcp_nodelay = true
 # [export_metrics]
 # whether enable export metrics, default is false
 # enable = false
+# The url of metrics export endpoint, default is `frontend` default HTTP endpoint.
+# endpoint = "127.0.0.1:4000"
+# The database name of exported metrics stores, user needs to specify a valid database
+# db = ""
 # The interval of export metrics
 # write_interval = "30s"
-# for `frontend`, `self_import` is recommend to collect metrics generated by itself
-# [export_metrics.self_import]
-# db = "information_schema"
+# HTTP headers of Prometheus remote-write carry
+# headers = {}

config/metasrv.example.toml

@@ -64,6 +64,8 @@ provider = "raft_engine"
 # selector_type = "round_robin"
 # A Kafka topic is constructed by concatenating `topic_name_prefix` and `topic_id`.
 # topic_name_prefix = "greptimedb_wal_topic"
+# Number of partitions per topic.
+# num_partitions = 1
 # Expected number of replicas of each partition.
 # replication_factor = 1
 # Above which a topic creation operation will be cancelled.
@@ -84,10 +86,11 @@ provider = "raft_engine"
 # [export_metrics]
 # whether enable export metrics, default is false
 # enable = false
+# The url of metrics export endpoint, default is `frontend` default HTTP endpoint.
+# endpoint = "127.0.0.1:4000"
+# The database name of exported metrics stores, user needs to specify a valid database
+# db = ""
 # The interval of export metrics
 # write_interval = "30s"
-# [export_metrics.remote_write]
-# The url the metrics send to. The url is empty by default, url example: `http://127.0.0.1:4000/v1/prometheus/write?db=information_schema`
-# url = ""
 # HTTP headers of Prometheus remote-write carry
 # headers = {}

config/standalone.example.toml

@@ -81,9 +81,6 @@ enable = true
 [prom_store]
 # Whether to enable Prometheus remote write and read in HTTP API, true by default.
 enable = true
-# Whether to store the data from Prometheus remote write in metric engine.
-# true by default
-with_metric_engine = true
 
 [wal]
 # Available wal providers:
@@ -91,7 +88,43 @@ with_metric_engine = true
 # - "kafka"
 provider = "raft_engine"
 
-# Raft-engine wal options.
+# There're none raft-engine wal config since meta srv only involves in remote wal currently.
+
+# Kafka wal options.
+# The broker endpoints of the Kafka cluster. ["127.0.0.1:9092"] by default.
+# broker_endpoints = ["127.0.0.1:9092"]
+
+# Number of topics to be created upon start.
+# num_topics = 64
+# Topic selector type.
+# Available selector types: 
+# - "round_robin" (default)
+# selector_type = "round_robin"
+# A Kafka topic is constructed by concatenating `topic_name_prefix` and `topic_id`.
+# topic_name_prefix = "greptimedb_wal_topic"
+# Number of partitions per topic.
+# num_partitions = 1
+# Expected number of replicas of each partition.
+# replication_factor = 1
+
+# The maximum log size a kafka batch producer could buffer.
+# max_batch_size = "4MB"
+# The linger duration of a kafka batch producer.
+# linger = "200ms"
+# The maximum amount of time (in milliseconds) to wait for Kafka records to be returned.
+# produce_record_timeout = "100ms"
+# Above which a topic creation operation will be cancelled.
+# create_topic_timeout = "30s"
+
+# The initial backoff for kafka clients.
+# backoff_init = "500ms"
+# The maximum backoff for kafka clients.
+# backoff_max = "10s"
+# Exponential backoff rate, i.e. next backoff = base * current backoff.
+# backoff_base = 2
+# Stop reconnecting if the total wait time reaches the deadline. If this config is missing, the reconnecting won't terminate.
+# backoff_deadline = "5mins"
+
 # WAL data directory
 # dir = "/tmp/greptimedb/wal"
 # WAL file size in bytes.
@@ -104,47 +137,6 @@ purge_interval = "10m"
 read_batch_size = 128
 # Whether to sync log file after every write.
 sync_write = false
-# Whether to reuse logically truncated log files.
-enable_log_recycle = true
-# Whether to pre-create log files on start up
-prefill_log_files = false
-# Duration for fsyncing log files.
-sync_period = "1000ms"
-
-# Kafka wal options.
-# The broker endpoints of the Kafka cluster. ["127.0.0.1:9092"] by default.
-# broker_endpoints = ["127.0.0.1:9092"]
-
-# Number of topics to be created upon start.
-# num_topics = 64
-# Topic selector type.
-# Available selector types: 
-# - "round_robin" (default)
-# selector_type = "round_robin"
-# The prefix of topic name.
-# topic_name_prefix = "greptimedb_wal_topic"
-# The number of replicas of each partition.
-# Warning: the replication factor must be positive and must not be greater than the number of broker endpoints.
-# replication_factor = 1
-
-# The max size of a single producer batch.
-# Warning: Kafka has a default limit of 1MB per message in a topic.
-# max_batch_size = "1MB"
-# The linger duration.
-# linger = "200ms"
-# The consumer wait timeout.
-# consumer_wait_timeout = "100ms"
-# Create topic timeout.
-# create_topic_timeout = "30s"
-
-# The initial backoff delay.
-# backoff_init = "500ms"
-# The maximum backoff delay.
-# backoff_max = "10s"
-# Exponential backoff rate, i.e. next backoff = base * current backoff.
-# backoff_base = 2
-# The deadline of retries.
-# backoff_deadline = "5mins"
 
 # Metadata storage options.
 [metadata_store]
@@ -196,18 +188,15 @@ compress_manifest = false
 max_background_jobs = 4
 # Interval to auto flush a region if it has not flushed yet.
 auto_flush_interval = "1h"
-# Global write buffer size for all regions. If not set, it's default to 1/8 of OS memory with a max limitation of 1GB.
+# Global write buffer size for all regions.
 global_write_buffer_size = "1GB"
-# Global write buffer size threshold to reject write requests. If not set, it's default to 2 times of `global_write_buffer_size`
+# Global write buffer size threshold to reject write requests (default 2G).
 global_write_buffer_reject_size = "2GB"
-# Cache size for SST metadata. Setting it to 0 to disable the cache.
-# If not set, it's default to 1/32 of OS memory with a max limitation of 128MB.
+# Cache size for SST metadata (default 128MB). Setting it to 0 to disable the cache.
 sst_meta_cache_size = "128MB"
-# Cache size for vectors and arrow arrays. Setting it to 0 to disable the cache.
-# If not set, it's default to 1/16 of OS memory with a max limitation of 512MB.
+# Cache size for vectors and arrow arrays (default 512MB). Setting it to 0 to disable the cache.
 vector_cache_size = "512MB"
-# Cache size for pages of SST row groups. Setting it to 0 to disable the cache.
-# If not set, it's default to 1/16 of OS memory with a max limitation of 512MB.
+# Cache size for pages of SST row groups (default 512MB). Setting it to 0 to disable the cache.
 page_cache_size = "512MB"
 # Buffer size for SST writing.
 sst_write_buffer_size = "8MB"
@@ -218,27 +207,6 @@ sst_write_buffer_size = "8MB"
 scan_parallelism = 0
 # Capacity of the channel to send data from parallel scan tasks to the main task (default 32).
 parallel_scan_channel_size = 32
-# Whether to allow stale WAL entries read during replay.
-allow_stale_entries = false
-
-[region_engine.mito.inverted_index]
-# Whether to create the index on flush.
-# - "auto": automatically
-# - "disable": never
-create_on_flush = "auto"
-# Whether to create the index on compaction.
-# - "auto": automatically
-# - "disable": never
-create_on_compaction = "auto"
-# Whether to apply the index on query
-# - "auto": automatically
-# - "disable": never
-apply_on_query = "auto"
-# Memory threshold for performing an external sort during index creation.
-# Setting to empty will disable external sorting, forcing all sorting operations to happen in memory.
-mem_threshold_on_create = "64M"
-# File system path to store intermediate files for external sorting (default `{data_home}/index_intermediate`).
-intermediate_path = ""
 
 # Log options
 # [logging]
@@ -262,8 +230,11 @@ intermediate_path = ""
 # [export_metrics]
 # whether enable export metrics, default is false
 # enable = false
+# The url of metrics export endpoint, default is `frontend` default HTTP endpoint.
+# endpoint = "127.0.0.1:4000"
+# The database name of exported metrics stores, user needs to specify a valid database
+# db = ""
 # The interval of export metrics
 # write_interval = "30s"
-# for `standalone`, `self_import` is recommend to collect metrics generated by itself
-# [export_metrics.self_import]
-# db = "information_schema"
+# HTTP headers of Prometheus remote-write carry
+# headers = {}

docker/ci/ubuntu/Dockerfile

@@ -1,11 +1,5 @@
 FROM ubuntu:22.04
 
-# The root path under which contains all the dependencies to build this Dockerfile.
-ARG DOCKER_BUILD_ROOT=.
-# The binary name of GreptimeDB executable.
-# Defaults to "greptime", but sometimes in other projects it might be different.
-ARG TARGET_BIN=greptime
-
 RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
     ca-certificates \
     python3.10 \
@@ -13,16 +7,14 @@ RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
     python3-pip \
     curl
 
-COPY $DOCKER_BUILD_ROOT/docker/python/requirements.txt /etc/greptime/requirements.txt
+COPY ./docker/python/requirements.txt /etc/greptime/requirements.txt
 
 RUN python3 -m pip install -r /etc/greptime/requirements.txt
 
 ARG TARGETARCH
 
-ADD $TARGETARCH/$TARGET_BIN /greptime/bin/
+ADD $TARGETARCH/greptime /greptime/bin/
 
 ENV PATH /greptime/bin/:$PATH
 
-ENV TARGET_BIN=$TARGET_BIN
-
-ENTRYPOINT ["sh", "-c", "exec $TARGET_BIN \"$@\"", "--"]
+ENTRYPOINT ["greptime"]

docker/dev-builder/ubuntu/Dockerfile

@@ -1,8 +1,5 @@
 FROM ubuntu:20.04
 
-# The root path under which contains all the dependencies to build this Dockerfile.
-ARG DOCKER_BUILD_ROOT=.
-
 ENV LANG en_US.utf8
 WORKDIR /greptimedb
 
@@ -30,20 +27,10 @@ RUN apt-get -y purge python3.8 && \
     ln -s /usr/bin/python3.10 /usr/bin/python3 && \
     curl -sS https://bootstrap.pypa.io/get-pip.py | python3.10
 
-# Silence all `safe.directory` warnings, to avoid the "detect dubious repository" error when building with submodules.
-# Disabling the safe directory check here won't pose extra security issues, because in our usage for this dev build
-# image, we use it solely on our own environment (that github action's VM, or ECS created dynamically by ourselves),
-# and the repositories are pulled from trusted sources (still us, of course). Doing so does not violate the intention
-# of the Git's addition to the "safe.directory" at the first place (see the commit message here:
-# https://github.com/git/git/commit/8959555cee7ec045958f9b6dd62e541affb7e7d9).
-# There's also another solution to this, that we add the desired submodules to the safe directory, instead of using 
-# wildcard here. However, that requires the git's config files and the submodules all owned by the very same user.
-# It's troublesome to do this since the dev build runs in Docker, which is under user "root"; while outside the Docker,
-# it can be a different user that have prepared the submodules.
-RUN git config --global --add safe.directory *
+RUN git config --global --add safe.directory /greptimedb
 
 # Install Python dependencies.
-COPY $DOCKER_BUILD_ROOT/docker/python/requirements.txt /etc/greptime/requirements.txt
+COPY ./docker/python/requirements.txt /etc/greptime/requirements.txt
 RUN python3 -m pip install -r /etc/greptime/requirements.txt
 
 # Install Rust.

docs/rfcs/2023-11-03-inverted-index.md

@@ -1,6 +1,6 @@
 ---
 Feature Name: Inverted Index for SST File
-Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/2705
+Tracking Issue: TBD
 Date: 2023-11-03
 Author: "Zhong Zhenchi <zhongzc_arch@outlook.com>"
 ---

docs/rfcs/2024-01-17-dataflow-framework.md

@@ -1,97 +0,0 @@
----
-Feature Name: Dataflow Framework
-Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/3187
-Date: 2024-01-17
-Author: "Discord9 <discord9@163.com>"
----
-
-# Summary
-This RFC proposes a Lightweight Module for executing continuous aggregation queries on a stream of data.
-
-# Motivation
-Being able to do continuous aggregation is a very powerful tool. It allows you to do things like:
-1. downsample data from i.e. 1 milliseconds to 1 second
-2. calculate the average of a stream of data
-3. Keeping a sliding window of data in memory
-In order to do those things while maintaining a low memory footprint, you need to be able to manage the data in a smart way. Hence, we only store necessary data in memory, and send/recv data deltas to/from the client.
-
-# Details
-
-## System boundary / What it's and isn't
-- GreptimeFlow provides a way to perform continuous aggregation over time-series data.
-- It's not a complete streaming-processing system. Only a must subset functionalities are provided.
-- Flow can process a configured range of fresh data. Data exceeding this range will be dropped directly. Thus it cannot handle random datasets (random on timestamp).
-- Both sliding windows (e.g., latest 5m from present) and fixed windows (every 5m from some time) are supported. And these two are the major targeting scenarios.
-- Flow can handle most aggregate operators within one table(i.e. Sum, avg, min, max and comparison operators). But others (join, trigger, txn etc.) are not the target feature.
-
-## Framework
-- Greptime Flow's is built on top of [Hydroflow](https://github.com/hydro-project/hydroflow).
-- We have three choices for the Dataflow/Streaming process framework for our simple continuous aggregation feature:
-1. Based on the timely/differential dataflow crate that [materialize](https://github.com/MaterializeInc/materialize) based on. Later, it's proved too obscure for a simple usage, and is hard to customize memory usage control.
-2. Based on a simple dataflow framework that we write from ground up, like what [arroyo](https://www.arroyo.dev/) or [risingwave](https://www.risingwave.dev/) did, for example the core streaming logic of [arroyo](https://github.com/ArroyoSystems/arroyo/blob/master/arroyo-datastream/src/lib.rs) only takes up to 2000 line of codes. However, it means maintaining another layer of dataflow framework, which might seem easy in the beginning, but I fear it might be too burdensome to maintain once we need more features.
-3. Based on a simple and lower level dataflow framework that someone else write, like [hydroflow](https://github.com/hydro-project/hydroflow), this approach combines the best of both worlds. Firstly, it boasts ease of comprehension and customization. Secondly, the dataflow framework offers precisely the necessary features for crafting uncomplicated single-node dataflow programs while delivering decent performance.
-
-Hence, we choose the third option, and use a simple logical plan that's anagonistic to the underlying dataflow framework, as it only describe how the dataflow graph should be doing, not how it do that. And we built operator in hydroflow to execute the plan. And the result hydroflow graph is wrapped in a engine that only support data in/out and tick event to flush and compute the result. This provide a thin middle layer that's easy to maintain and allow switching to other dataflow framework if necessary.
-
-## Deploy mode and protocol
-- Greptime Flow is an independent streaming compute component. It can be used either within a standalone node or as a dedicated node at the same level as frontend in distributed mode.
-- It accepts insert request Rows, which is used between frontend and datanode.
-- New flow job is submitted in the format of modified SQL query like snowflake do, like: `CREATE TASK avg_over_5m WINDOW_SIZE = "5m" AS SELECT avg(value) FROM table WHERE time > now() - 5m GROUP BY time(1m)`. Flow job then got stored in MetaSrv.
-- It also persists results in the format of Rows to frontend.
-- The query plan uses Substrait as codec format. It's the same with GreptimeDB's query engine.
-- Greptime Flow needs a WAL for recovering. It's possible to reuse datanode's.
-
-The workflow is shown in the following diagram
-```mermaid
-graph TB
-subgraph Flownode["Flownode"]
-    subgraph Dataflows
-        df1("Dataflow_1")
-        df2("Dataflow_2")
-    end
-end
-subgraph Frontend["Frontend"]
-    newLines["Mirror Insert
-Create Task From Query
-Write result from flow node"]
-end
-
-subgraph Datanode["Datanode"]
-end
-
-User --> Frontend
-Frontend -->|Register Task| Metasrv
-Metasrv -->|Read Task Metadata| Frontend
-Frontend -->|Create Task| Flownode
-
-Frontend -->|Mirror Insert| Flownode
-Flownode -->|Write back| Frontend
-
-Frontend --> Datanode
-Datanode --> Frontend
-
-```
-
-## Lifecycle of data
-- New data is inserted into frontend like before. Frontend will mirror insert request to Flow node if there is configured flow job.
-- Depending on the timestamp of incoming data, flow will either drop it (outdated data) or process it (fresh data).
-- Greptime Flow will periodically write results back to the result table through frontend.
-- Those result will then be written into a result table stored in datanode.
-- A small table of intermediate state is kept in memory, which is used to calculate the result.
-## Supported operations
-- Greptime Flow accepts a configurable "materialize window", data point exceeds that time window is discarded.
-- Data within that "materialize window" is queryable and updateable.
-- Greptime Flow can handle partitioning, if and only if the input query can be transformed to a fully partitioned plan according to the existing commutative rules. Otherwise the corresponding flow job has to be calculated in a single node.
-- Notice that Greptime Flow has to see all the data belongs to one partition.
-- Deletion and duplicate insertion are not supported at early stage.
-## Miscellaneous 
-- Greptime Flow can translate SQL to it's own plan, however only a selected few aggregate function is supported for now, like min/max/sum/count/avg
-- Greptime Flow's operator is configurable in terms of the size of the materialize window, whether to allow delay of incoming data etc., so simplest operator can choose to not tolerate any delay to save memory.
-
-# Future Work
-- Support UDF that can do one-to-one mapping. Preferably, we can reuse the UDF mechanism in GreptimeDB.
-- Support join operator.
-- Design syntax for config operator for different materialize window and delay tolerance.
-- Support cross partition merge operator that allows complex query plan that not necessary accord with partitioning rule to communicate between nodes and create final materialize result.
-- Duplicate insertion, which can be reverted easily within the current framework, so supporting it could be easy
-- Deletion within "materialize window", this requires operators like min/max to store all inputs within materialize window, which might require further optimization.

grafana/README.md

@@ -1,10 +0,0 @@
-Grafana dashboard for GreptimeDB
---------------------------------
-
-GreptimeDB's official Grafana dashboard.
-
-Status notify: we are still working on this config. It's expected to change frequently in the recent days. Please feel free to submit your feedback and/or contribution to this dashboard 🤗
-
-# How to use
-
-Open Grafana Dashboard page, choose `New` -> `Import`. And upload `greptimedb.json` file.

src/catalog/Cargo.toml

@@ -10,10 +10,10 @@ testing = []
 [dependencies]
 api.workspace = true
 arc-swap = "1.0"
-arrow.workspace = true
 arrow-schema.workspace = true
 async-stream.workspace = true
 async-trait = "0.1"
+build-data = "0.1"
 common-catalog.workspace = true
 common-error.workspace = true
 common-grpc.workspace = true
@@ -24,16 +24,14 @@ common-recordbatch.workspace = true
 common-runtime.workspace = true
 common-telemetry.workspace = true
 common-time.workspace = true
-common-version.workspace = true
 dashmap.workspace = true
 datafusion.workspace = true
 datatypes.workspace = true
 futures = "0.3"
 futures-util.workspace = true
-itertools.workspace = true
 lazy_static.workspace = true
 meta-client.workspace = true
-moka = { workspace = true, features = ["future", "sync"] }
+moka = { workspace = true, features = ["future"] }
 parking_lot = "0.12"
 partition.workspace = true
 paste = "1.0"

src/catalog/src/error.rs

@@ -41,14 +41,6 @@ pub enum Error {
         source: BoxedError,
     },
 
-    #[snafu(display("Failed to list {}.{}'s tables", catalog, schema))]
-    ListTables {
-        location: Location,
-        catalog: String,
-        schema: String,
-        source: BoxedError,
-    },
-
     #[snafu(display("Failed to re-compile script due to internal error"))]
     CompileScriptInternal {
         location: Location,
@@ -164,15 +156,6 @@ pub enum Error {
         location: Location,
     },
 
-    #[snafu(display("Failed to find table partitions: #{table}"))]
-    FindPartitions {
-        source: partition::error::Error,
-        table: String,
-    },
-
-    #[snafu(display("Failed to find region routes"))]
-    FindRegionRoutes { source: partition::error::Error },
-
     #[snafu(display("Failed to read system catalog table records"))]
     ReadSystemCatalog {
         location: Location,
@@ -263,14 +246,11 @@ impl ErrorExt for Error {
         match self {
             Error::InvalidKey { .. }
             | Error::SchemaNotFound { .. }
+            | Error::TableNotFound { .. }
             | Error::CatalogNotFound { .. }
-            | Error::FindPartitions { .. }
-            | Error::FindRegionRoutes { .. }
             | Error::InvalidEntryType { .. }
             | Error::ParallelOpenTable { .. } => StatusCode::Unexpected,
 
-            Error::TableNotFound { .. } => StatusCode::TableNotFound,
-
             Error::SystemCatalog { .. }
             | Error::EmptyValue { .. }
             | Error::ValueDeserialize { .. } => StatusCode::StorageUnavailable,
@@ -290,9 +270,9 @@ impl ErrorExt for Error {
                 StatusCode::InvalidArguments
             }
 
-            Error::ListCatalogs { source, .. }
-            | Error::ListSchemas { source, .. }
-            | Error::ListTables { source, .. } => source.status_code(),
+            Error::ListCatalogs { source, .. } | Error::ListSchemas { source, .. } => {
+                source.status_code()
+            }
 
             Error::OpenSystemCatalog { source, .. }
             | Error::CreateSystemCatalog { source, .. }
@@ -353,7 +333,7 @@ mod tests {
         assert_eq!(
             StatusCode::StorageUnavailable,
             Error::SystemCatalog {
-                msg: String::default(),
+                msg: "".to_string(),
                 location: Location::generate(),
             }
             .status_code()

src/catalog/src/information_schema.rs

@@ -13,27 +13,20 @@
 // limitations under the License.
 
 mod columns;
-mod key_column_usage;
 mod memory_table;
-mod partitions;
-mod predicate;
-mod region_peers;
-mod runtime_metrics;
-mod schemata;
 mod table_names;
 mod tables;
 
 use std::collections::HashMap;
 use std::sync::{Arc, Weak};
 
-use common_catalog::consts::{self, DEFAULT_CATALOG_NAME, INFORMATION_SCHEMA_NAME};
+use common_catalog::consts::{self, INFORMATION_SCHEMA_NAME};
 use common_error::ext::BoxedError;
 use common_recordbatch::{RecordBatchStreamWrapper, SendableRecordBatchStream};
 use datatypes::schema::SchemaRef;
 use futures_util::StreamExt;
 use lazy_static::lazy_static;
 use paste::paste;
-pub(crate) use predicate::Predicates;
 use snafu::ResultExt;
 use store_api::data_source::DataSource;
 use store_api::storage::{ScanRequest, TableId};
@@ -47,12 +40,7 @@ pub use table_names::*;
 
 use self::columns::InformationSchemaColumns;
 use crate::error::Result;
-use crate::information_schema::key_column_usage::InformationSchemaKeyColumnUsage;
 use crate::information_schema::memory_table::{get_schema_columns, MemoryTable};
-use crate::information_schema::partitions::InformationSchemaPartitions;
-use crate::information_schema::region_peers::InformationSchemaRegionPeers;
-use crate::information_schema::runtime_metrics::InformationSchemaMetrics;
-use crate::information_schema::schemata::InformationSchemaSchemata;
 use crate::information_schema::tables::InformationSchemaTables;
 use crate::CatalogManager;
 
@@ -62,23 +50,7 @@ lazy_static! {
         ENGINES,
         COLUMN_PRIVILEGES,
         COLUMN_STATISTICS,
-        CHARACTER_SETS,
-        COLLATIONS,
-        COLLATION_CHARACTER_SET_APPLICABILITY,
-        CHECK_CONSTRAINTS,
-        EVENTS,
-        FILES,
-        OPTIMIZER_TRACE,
-        PARAMETERS,
-        PROFILING,
-        REFERENTIAL_CONSTRAINTS,
-        ROUTINES,
-        SCHEMA_PRIVILEGES,
-        TABLE_PRIVILEGES,
-        TRIGGERS,
-        GLOBAL_STATUS,
-        SESSION_STATUS,
-        PARTITIONS,
+        BUILD_INFO,
     ];
 }
 
@@ -148,36 +120,12 @@ impl InformationSchemaProvider {
 
     fn build_tables(&mut self) {
         let mut tables = HashMap::new();
-
-        // Carefully consider the tables that may expose sensitive cluster configurations,
-        // authentication details, and other critical information.
-        // Only put these tables under `greptime` catalog to prevent info leak.
-        if self.catalog_name == DEFAULT_CATALOG_NAME {
-            tables.insert(
-                RUNTIME_METRICS.to_string(),
-                self.build_table(RUNTIME_METRICS).unwrap(),
-            );
-            tables.insert(
-                BUILD_INFO.to_string(),
-                self.build_table(BUILD_INFO).unwrap(),
-            );
-            tables.insert(
-                REGION_PEERS.to_string(),
-                self.build_table(REGION_PEERS).unwrap(),
-            );
-        }
-
         tables.insert(TABLES.to_string(), self.build_table(TABLES).unwrap());
-        tables.insert(SCHEMATA.to_string(), self.build_table(SCHEMATA).unwrap());
         tables.insert(COLUMNS.to_string(), self.build_table(COLUMNS).unwrap());
-        tables.insert(
-            KEY_COLUMN_USAGE.to_string(),
-            self.build_table(KEY_COLUMN_USAGE).unwrap(),
-        );
 
         // Add memory tables
         for name in MEMORY_TABLES.iter() {
-            tables.insert((*name).to_string(), self.build_table(name).expect(name));
+            tables.insert((*name).to_string(), self.build_table(name).unwrap());
         }
 
         self.tables = tables;
@@ -186,7 +134,7 @@ impl InformationSchemaProvider {
     fn build_table(&self, name: &str) -> Option<TableRef> {
         self.information_table(name).map(|table| {
             let table_info = Self::table_info(self.catalog_name.clone(), &table);
-            let filter_pushdown = FilterPushDownType::Inexact;
+            let filter_pushdown = FilterPushDownType::Unsupported;
             let thin_table = ThinTable::new(table_info, filter_pushdown);
 
             let data_source = Arc::new(InformationTableDataSource::new(table));
@@ -208,41 +156,6 @@ impl InformationSchemaProvider {
             COLUMN_PRIVILEGES => setup_memory_table!(COLUMN_PRIVILEGES),
             COLUMN_STATISTICS => setup_memory_table!(COLUMN_STATISTICS),
             BUILD_INFO => setup_memory_table!(BUILD_INFO),
-            CHARACTER_SETS => setup_memory_table!(CHARACTER_SETS),
-            COLLATIONS => setup_memory_table!(COLLATIONS),
-            COLLATION_CHARACTER_SET_APPLICABILITY => {
-                setup_memory_table!(COLLATION_CHARACTER_SET_APPLICABILITY)
-            }
-            CHECK_CONSTRAINTS => setup_memory_table!(CHECK_CONSTRAINTS),
-            EVENTS => setup_memory_table!(EVENTS),
-            FILES => setup_memory_table!(FILES),
-            OPTIMIZER_TRACE => setup_memory_table!(OPTIMIZER_TRACE),
-            PARAMETERS => setup_memory_table!(PARAMETERS),
-            PROFILING => setup_memory_table!(PROFILING),
-            REFERENTIAL_CONSTRAINTS => setup_memory_table!(REFERENTIAL_CONSTRAINTS),
-            ROUTINES => setup_memory_table!(ROUTINES),
-            SCHEMA_PRIVILEGES => setup_memory_table!(SCHEMA_PRIVILEGES),
-            TABLE_PRIVILEGES => setup_memory_table!(TABLE_PRIVILEGES),
-            TRIGGERS => setup_memory_table!(TRIGGERS),
-            GLOBAL_STATUS => setup_memory_table!(GLOBAL_STATUS),
-            SESSION_STATUS => setup_memory_table!(SESSION_STATUS),
-            KEY_COLUMN_USAGE => Some(Arc::new(InformationSchemaKeyColumnUsage::new(
-                self.catalog_name.clone(),
-                self.catalog_manager.clone(),
-            )) as _),
-            SCHEMATA => Some(Arc::new(InformationSchemaSchemata::new(
-                self.catalog_name.clone(),
-                self.catalog_manager.clone(),
-            )) as _),
-            RUNTIME_METRICS => Some(Arc::new(InformationSchemaMetrics::new())),
-            PARTITIONS => Some(Arc::new(InformationSchemaPartitions::new(
-                self.catalog_name.clone(),
-                self.catalog_manager.clone(),
-            )) as _),
-            REGION_PEERS => Some(Arc::new(InformationSchemaRegionPeers::new(
-                self.catalog_name.clone(),
-                self.catalog_manager.clone(),
-            )) as _),
             _ => None,
         }
     }
@@ -274,7 +187,7 @@ trait InformationTable {
 
     fn schema(&self) -> SchemaRef;
 
-    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream>;
+    fn to_stream(&self) -> Result<SendableRecordBatchStream>;
 
     fn table_type(&self) -> TableType {
         TableType::Temporary
@@ -308,7 +221,7 @@ impl DataSource for InformationTableDataSource {
         &self,
         request: ScanRequest,
     ) -> std::result::Result<SendableRecordBatchStream, BoxedError> {
-        let projection = request.projection.clone();
+        let projection = request.projection;
         let projected_schema = match &projection {
             Some(projection) => self.try_project(projection)?,
             None => self.table.schema(),
@@ -316,7 +229,7 @@ impl DataSource for InformationTableDataSource {
 
         let stream = self
             .table
-            .to_stream(request)
+            .to_stream()
             .map_err(BoxedError::new)
             .context(TablesRecordBatchSnafu)
             .map_err(BoxedError::new)?
@@ -329,7 +242,6 @@ impl DataSource for InformationTableDataSource {
             schema: projected_schema,
             stream: Box::pin(stream),
             output_ordering: None,
-            metrics: Default::default(),
         };
 
         Ok(Box::pin(stream))

src/catalog/src/information_schema/columns.rs

@@ -29,17 +29,14 @@ use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatc
 use datatypes::prelude::{ConcreteDataType, DataType};
 use datatypes::scalars::ScalarVectorBuilder;
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
-use datatypes::value::Value;
 use datatypes::vectors::{StringVectorBuilder, VectorRef};
-use futures::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
-use store_api::storage::{ScanRequest, TableId};
+use store_api::storage::TableId;
 
 use super::{InformationTable, COLUMNS};
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
-use crate::information_schema::Predicates;
 use crate::CatalogManager;
 
 pub(super) struct InformationSchemaColumns {
@@ -54,11 +51,6 @@ const TABLE_NAME: &str = "table_name";
 const COLUMN_NAME: &str = "column_name";
 const DATA_TYPE: &str = "data_type";
 const SEMANTIC_TYPE: &str = "semantic_type";
-const COLUMN_DEFAULT: &str = "column_default";
-const IS_NULLABLE: &str = "is_nullable";
-const COLUMN_TYPE: &str = "column_type";
-const COLUMN_COMMENT: &str = "column_comment";
-const INIT_CAPACITY: usize = 42;
 
 impl InformationSchemaColumns {
     pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
@@ -77,10 +69,6 @@ impl InformationSchemaColumns {
             ColumnSchema::new(COLUMN_NAME, ConcreteDataType::string_datatype(), false),
             ColumnSchema::new(DATA_TYPE, ConcreteDataType::string_datatype(), false),
             ColumnSchema::new(SEMANTIC_TYPE, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(COLUMN_DEFAULT, ConcreteDataType::string_datatype(), true),
-            ColumnSchema::new(IS_NULLABLE, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(COLUMN_TYPE, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(COLUMN_COMMENT, ConcreteDataType::string_datatype(), true),
         ]))
     }
 
@@ -106,14 +94,14 @@ impl InformationTable for InformationSchemaColumns {
         self.schema.clone()
     }
 
-    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
+    fn to_stream(&self) -> Result<SendableRecordBatchStream> {
         let schema = self.schema.arrow_schema().clone();
         let mut builder = self.builder();
         let stream = Box::pin(DfRecordBatchStreamAdapter::new(
             schema,
             futures::stream::once(async move {
                 builder
-                    .make_columns(Some(request))
+                    .make_columns()
                     .await
                     .map(|x| x.into_df_record_batch())
                     .map_err(Into::into)
@@ -138,11 +126,6 @@ struct InformationSchemaColumnsBuilder {
     column_names: StringVectorBuilder,
     data_types: StringVectorBuilder,
     semantic_types: StringVectorBuilder,
-
-    column_defaults: StringVectorBuilder,
-    is_nullables: StringVectorBuilder,
-    column_types: StringVectorBuilder,
-    column_comments: StringVectorBuilder,
 }
 
 impl InformationSchemaColumnsBuilder {
@@ -155,52 +138,62 @@ impl InformationSchemaColumnsBuilder {
             schema,
             catalog_name,
             catalog_manager,
-            catalog_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            schema_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            column_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            data_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            semantic_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            column_defaults: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            is_nullables: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            column_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            column_comments: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            catalog_names: StringVectorBuilder::with_capacity(42),
+            schema_names: StringVectorBuilder::with_capacity(42),
+            table_names: StringVectorBuilder::with_capacity(42),
+            column_names: StringVectorBuilder::with_capacity(42),
+            data_types: StringVectorBuilder::with_capacity(42),
+            semantic_types: StringVectorBuilder::with_capacity(42),
         }
     }
 
     /// Construct the `information_schema.columns` virtual table
-    async fn make_columns(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
+    async fn make_columns(&mut self) -> Result<RecordBatch> {
         let catalog_name = self.catalog_name.clone();
         let catalog_manager = self
             .catalog_manager
             .upgrade()
             .context(UpgradeWeakCatalogManagerRefSnafu)?;
-        let predicates = Predicates::from_scan_request(&request);
 
         for schema_name in catalog_manager.schema_names(&catalog_name).await? {
-            let mut stream = catalog_manager.tables(&catalog_name, &schema_name).await;
+            if !catalog_manager
+                .schema_exists(&catalog_name, &schema_name)
+                .await?
+            {
+                continue;
+            }
 
-            while let Some(table) = stream.try_next().await? {
-                let keys = &table.table_info().meta.primary_key_indices;
-                let schema = table.schema();
+            for table_name in catalog_manager
+                .table_names(&catalog_name, &schema_name)
+                .await?
+            {
+                if let Some(table) = catalog_manager
+                    .table(&catalog_name, &schema_name, &table_name)
+                    .await?
+                {
+                    let keys = &table.table_info().meta.primary_key_indices;
+                    let schema = table.schema();
 
-                for (idx, column) in schema.column_schemas().iter().enumerate() {
-                    let semantic_type = if column.is_time_index() {
-                        SEMANTIC_TYPE_TIME_INDEX
-                    } else if keys.contains(&idx) {
-                        SEMANTIC_TYPE_PRIMARY_KEY
-                    } else {
-                        SEMANTIC_TYPE_FIELD
-                    };
+                    for (idx, column) in schema.column_schemas().iter().enumerate() {
+                        let semantic_type = if column.is_time_index() {
+                            SEMANTIC_TYPE_TIME_INDEX
+                        } else if keys.contains(&idx) {
+                            SEMANTIC_TYPE_PRIMARY_KEY
+                        } else {
+                            SEMANTIC_TYPE_FIELD
+                        };
 
-                    self.add_column(
-                        &predicates,
-                        &catalog_name,
-                        &schema_name,
-                        &table.table_info().name,
-                        semantic_type,
-                        column,
-                    );
+                        self.add_column(
+                            &catalog_name,
+                            &schema_name,
+                            &table_name,
+                            &column.name,
+                            &column.data_type.name(),
+                            semantic_type,
+                        );
+                    }
+                } else {
+                    unreachable!();
                 }
             }
         }
@@ -210,48 +203,19 @@ impl InformationSchemaColumnsBuilder {
 
     fn add_column(
         &mut self,
-        predicates: &Predicates,
         catalog_name: &str,
         schema_name: &str,
         table_name: &str,
+        column_name: &str,
+        data_type: &str,
         semantic_type: &str,
-        column_schema: &ColumnSchema,
     ) {
-        let data_type = &column_schema.data_type.name();
-
-        let row = [
-            (TABLE_CATALOG, &Value::from(catalog_name)),
-            (TABLE_SCHEMA, &Value::from(schema_name)),
-            (TABLE_NAME, &Value::from(table_name)),
-            (COLUMN_NAME, &Value::from(column_schema.name.as_str())),
-            (DATA_TYPE, &Value::from(data_type.as_str())),
-            (SEMANTIC_TYPE, &Value::from(semantic_type)),
-        ];
-
-        if !predicates.eval(&row) {
-            return;
-        }
-
         self.catalog_names.push(Some(catalog_name));
         self.schema_names.push(Some(schema_name));
         self.table_names.push(Some(table_name));
-        self.column_names.push(Some(&column_schema.name));
+        self.column_names.push(Some(column_name));
         self.data_types.push(Some(data_type));
         self.semantic_types.push(Some(semantic_type));
-        self.column_defaults.push(
-            column_schema
-                .default_constraint()
-                .map(|s| format!("{}", s))
-                .as_deref(),
-        );
-        if column_schema.is_nullable() {
-            self.is_nullables.push(Some("Yes"));
-        } else {
-            self.is_nullables.push(Some("No"));
-        }
-        self.column_types.push(Some(data_type));
-        self.column_comments
-            .push(column_schema.column_comment().map(|x| x.as_ref()));
     }
 
     fn finish(&mut self) -> Result<RecordBatch> {
@@ -262,10 +226,6 @@ impl InformationSchemaColumnsBuilder {
             Arc::new(self.column_names.finish()),
             Arc::new(self.data_types.finish()),
             Arc::new(self.semantic_types.finish()),
-            Arc::new(self.column_defaults.finish()),
-            Arc::new(self.is_nullables.finish()),
-            Arc::new(self.column_types.finish()),
-            Arc::new(self.column_comments.finish()),
         ];
 
         RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
@@ -284,7 +244,7 @@ impl DfPartitionStream for InformationSchemaColumns {
             schema,
             futures::stream::once(async move {
                 builder
-                    .make_columns(None)
+                    .make_columns()
                     .await
                     .map(|x| x.into_df_record_batch())
                     .map_err(Into::into)

src/catalog/src/information_schema/key_column_usage.rs

@@ -1,345 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed 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.
-
-use std::sync::{Arc, Weak};
-
-use arrow_schema::SchemaRef as ArrowSchemaRef;
-use common_catalog::consts::INFORMATION_SCHEMA_KEY_COLUMN_USAGE_TABLE_ID;
-use common_error::ext::BoxedError;
-use common_query::physical_plan::TaskContext;
-use common_recordbatch::adapter::RecordBatchStreamAdapter;
-use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
-use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
-use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
-use datatypes::prelude::{ConcreteDataType, MutableVector, ScalarVectorBuilder, VectorRef};
-use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
-use datatypes::value::Value;
-use datatypes::vectors::{ConstantVector, StringVector, StringVectorBuilder, UInt32VectorBuilder};
-use snafu::{OptionExt, ResultExt};
-use store_api::storage::{ScanRequest, TableId};
-
-use super::KEY_COLUMN_USAGE;
-use crate::error::{
-    CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
-};
-use crate::information_schema::{InformationTable, Predicates};
-use crate::CatalogManager;
-
-const CONSTRAINT_SCHEMA: &str = "constraint_schema";
-const CONSTRAINT_NAME: &str = "constraint_name";
-const TABLE_CATALOG: &str = "table_catalog";
-const TABLE_SCHEMA: &str = "table_schema";
-const TABLE_NAME: &str = "table_name";
-const COLUMN_NAME: &str = "column_name";
-const ORDINAL_POSITION: &str = "ordinal_position";
-const INIT_CAPACITY: usize = 42;
-
-/// The virtual table implementation for `information_schema.KEY_COLUMN_USAGE`.
-pub(super) struct InformationSchemaKeyColumnUsage {
-    schema: SchemaRef,
-    catalog_name: String,
-    catalog_manager: Weak<dyn CatalogManager>,
-}
-
-impl InformationSchemaKeyColumnUsage {
-    pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
-        Self {
-            schema: Self::schema(),
-            catalog_name,
-            catalog_manager,
-        }
-    }
-
-    pub(crate) fn schema() -> SchemaRef {
-        Arc::new(Schema::new(vec![
-            ColumnSchema::new(
-                "constraint_catalog",
-                ConcreteDataType::string_datatype(),
-                false,
-            ),
-            ColumnSchema::new(
-                CONSTRAINT_SCHEMA,
-                ConcreteDataType::string_datatype(),
-                false,
-            ),
-            ColumnSchema::new(CONSTRAINT_NAME, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(TABLE_CATALOG, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(TABLE_SCHEMA, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(TABLE_NAME, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(COLUMN_NAME, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(ORDINAL_POSITION, ConcreteDataType::uint32_datatype(), false),
-            ColumnSchema::new(
-                "position_in_unique_constraint",
-                ConcreteDataType::uint32_datatype(),
-                true,
-            ),
-            ColumnSchema::new(
-                "referenced_table_schema",
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            ColumnSchema::new(
-                "referenced_table_name",
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            ColumnSchema::new(
-                "referenced_column_name",
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-        ]))
-    }
-
-    fn builder(&self) -> InformationSchemaKeyColumnUsageBuilder {
-        InformationSchemaKeyColumnUsageBuilder::new(
-            self.schema.clone(),
-            self.catalog_name.clone(),
-            self.catalog_manager.clone(),
-        )
-    }
-}
-
-impl InformationTable for InformationSchemaKeyColumnUsage {
-    fn table_id(&self) -> TableId {
-        INFORMATION_SCHEMA_KEY_COLUMN_USAGE_TABLE_ID
-    }
-
-    fn table_name(&self) -> &'static str {
-        KEY_COLUMN_USAGE
-    }
-
-    fn schema(&self) -> SchemaRef {
-        self.schema.clone()
-    }
-
-    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
-        let schema = self.schema.arrow_schema().clone();
-        let mut builder = self.builder();
-        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
-            schema,
-            futures::stream::once(async move {
-                builder
-                    .make_key_column_usage(Some(request))
-                    .await
-                    .map(|x| x.into_df_record_batch())
-                    .map_err(Into::into)
-            }),
-        ));
-        Ok(Box::pin(
-            RecordBatchStreamAdapter::try_new(stream)
-                .map_err(BoxedError::new)
-                .context(InternalSnafu)?,
-        ))
-    }
-}
-
-/// Builds the `information_schema.KEY_COLUMN_USAGE` table row by row
-///
-/// Columns are based on <https://dev.mysql.com/doc/refman/8.2/en/information-schema-key-column-usage-table.html>
-struct InformationSchemaKeyColumnUsageBuilder {
-    schema: SchemaRef,
-    catalog_name: String,
-    catalog_manager: Weak<dyn CatalogManager>,
-
-    constraint_catalog: StringVectorBuilder,
-    constraint_schema: StringVectorBuilder,
-    constraint_name: StringVectorBuilder,
-    table_catalog: StringVectorBuilder,
-    table_schema: StringVectorBuilder,
-    table_name: StringVectorBuilder,
-    column_name: StringVectorBuilder,
-    ordinal_position: UInt32VectorBuilder,
-    position_in_unique_constraint: UInt32VectorBuilder,
-}
-
-impl InformationSchemaKeyColumnUsageBuilder {
-    fn new(
-        schema: SchemaRef,
-        catalog_name: String,
-        catalog_manager: Weak<dyn CatalogManager>,
-    ) -> Self {
-        Self {
-            schema,
-            catalog_name,
-            catalog_manager,
-            constraint_catalog: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            constraint_schema: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            constraint_name: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            table_catalog: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            table_schema: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            table_name: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            column_name: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            ordinal_position: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
-            position_in_unique_constraint: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
-        }
-    }
-
-    /// Construct the `information_schema.KEY_COLUMN_USAGE` virtual table
-    async fn make_key_column_usage(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
-        let catalog_name = self.catalog_name.clone();
-        let catalog_manager = self
-            .catalog_manager
-            .upgrade()
-            .context(UpgradeWeakCatalogManagerRefSnafu)?;
-        let predicates = Predicates::from_scan_request(&request);
-
-        let mut primary_constraints = vec![];
-
-        for schema_name in catalog_manager.schema_names(&catalog_name).await? {
-            if !catalog_manager
-                .schema_exists(&catalog_name, &schema_name)
-                .await?
-            {
-                continue;
-            }
-
-            for table_name in catalog_manager
-                .table_names(&catalog_name, &schema_name)
-                .await?
-            {
-                if let Some(table) = catalog_manager
-                    .table(&catalog_name, &schema_name, &table_name)
-                    .await?
-                {
-                    let keys = &table.table_info().meta.primary_key_indices;
-                    let schema = table.schema();
-
-                    for (idx, column) in schema.column_schemas().iter().enumerate() {
-                        if column.is_time_index() {
-                            self.add_key_column_usage(
-                                &predicates,
-                                &schema_name,
-                                "TIME INDEX",
-                                &schema_name,
-                                &table_name,
-                                &column.name,
-                                1, //always 1 for time index
-                            );
-                        }
-                        if keys.contains(&idx) {
-                            primary_constraints.push((
-                                schema_name.clone(),
-                                table_name.clone(),
-                                column.name.clone(),
-                            ));
-                        }
-                        // TODO(dimbtp): foreign key constraint not supported yet
-                    }
-                } else {
-                    unreachable!();
-                }
-            }
-        }
-
-        for (i, (schema_name, table_name, column_name)) in
-            primary_constraints.into_iter().enumerate()
-        {
-            self.add_key_column_usage(
-                &predicates,
-                &schema_name,
-                "PRIMARY",
-                &schema_name,
-                &table_name,
-                &column_name,
-                i as u32 + 1,
-            );
-        }
-
-        self.finish()
-    }
-
-    // TODO(dimbtp): Foreign key constraint has not `None` value for last 4
-    // fields, but it is not supported yet.
-    #[allow(clippy::too_many_arguments)]
-    fn add_key_column_usage(
-        &mut self,
-        predicates: &Predicates,
-        constraint_schema: &str,
-        constraint_name: &str,
-        table_schema: &str,
-        table_name: &str,
-        column_name: &str,
-        ordinal_position: u32,
-    ) {
-        let row = [
-            (CONSTRAINT_SCHEMA, &Value::from(constraint_schema)),
-            (CONSTRAINT_NAME, &Value::from(constraint_name)),
-            (TABLE_SCHEMA, &Value::from(table_schema)),
-            (TABLE_NAME, &Value::from(table_name)),
-            (COLUMN_NAME, &Value::from(column_name)),
-            (ORDINAL_POSITION, &Value::from(ordinal_position)),
-        ];
-
-        if !predicates.eval(&row) {
-            return;
-        }
-
-        self.constraint_catalog.push(Some("def"));
-        self.constraint_schema.push(Some(constraint_schema));
-        self.constraint_name.push(Some(constraint_name));
-        self.table_catalog.push(Some("def"));
-        self.table_schema.push(Some(table_schema));
-        self.table_name.push(Some(table_name));
-        self.column_name.push(Some(column_name));
-        self.ordinal_position.push(Some(ordinal_position));
-        self.position_in_unique_constraint.push(None);
-    }
-
-    fn finish(&mut self) -> Result<RecordBatch> {
-        let rows_num = self.table_catalog.len();
-
-        let null_string_vector = Arc::new(ConstantVector::new(
-            Arc::new(StringVector::from(vec![None as Option<&str>])),
-            rows_num,
-        ));
-        let columns: Vec<VectorRef> = vec![
-            Arc::new(self.constraint_catalog.finish()),
-            Arc::new(self.constraint_schema.finish()),
-            Arc::new(self.constraint_name.finish()),
-            Arc::new(self.table_catalog.finish()),
-            Arc::new(self.table_schema.finish()),
-            Arc::new(self.table_name.finish()),
-            Arc::new(self.column_name.finish()),
-            Arc::new(self.ordinal_position.finish()),
-            Arc::new(self.position_in_unique_constraint.finish()),
-            null_string_vector.clone(),
-            null_string_vector.clone(),
-            null_string_vector,
-        ];
-        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
-    }
-}
-
-impl DfPartitionStream for InformationSchemaKeyColumnUsage {
-    fn schema(&self) -> &ArrowSchemaRef {
-        self.schema.arrow_schema()
-    }
-
-    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
-        let schema = self.schema.arrow_schema().clone();
-        let mut builder = self.builder();
-        Box::pin(DfRecordBatchStreamAdapter::new(
-            schema,
-            futures::stream::once(async move {
-                builder
-                    .make_key_column_usage(None)
-                    .await
-                    .map(|x| x.into_df_record_batch())
-                    .map_err(Into::into)
-            }),
-        ))
-    }
-}

src/catalog/src/information_schema/memory_table.rs

@@ -26,7 +26,7 @@ use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatc
 use datatypes::schema::SchemaRef;
 use datatypes::vectors::VectorRef;
 use snafu::ResultExt;
-use store_api::storage::{ScanRequest, TableId};
+use store_api::storage::TableId;
 pub use tables::get_schema_columns;
 
 use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
@@ -74,7 +74,7 @@ impl InformationTable for MemoryTable {
         self.schema.clone()
     }
 
-    fn to_stream(&self, _request: ScanRequest) -> Result<SendableRecordBatchStream> {
+    fn to_stream(&self) -> Result<SendableRecordBatchStream> {
         let schema = self.schema.arrow_schema().clone();
         let mut builder = self.builder();
         let stream = Box::pin(DfRecordBatchStreamAdapter::new(
@@ -169,7 +169,7 @@ mod tests {
         assert_eq!("test", table.table_name());
         assert_eq!(schema, InformationTable::schema(&table));
 
-        let stream = table.to_stream(ScanRequest::default()).unwrap();
+        let stream = table.to_stream().unwrap();
 
         let batches = RecordBatches::try_collect(stream).await.unwrap();
 
@@ -198,7 +198,7 @@ mod tests {
         assert_eq!("test", table.table_name());
         assert_eq!(schema, InformationTable::schema(&table));
 
-        let stream = table.to_stream(ScanRequest::default()).unwrap();
+        let stream = table.to_stream().unwrap();
 
         let batches = RecordBatches::try_collect(stream).await.unwrap();
 

src/catalog/src/information_schema/memory_table/tables.rs

@@ -17,10 +17,12 @@ use std::sync::Arc;
 use common_catalog::consts::MITO_ENGINE;
 use datatypes::prelude::{ConcreteDataType, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
-use datatypes::vectors::{Int64Vector, StringVector};
+use datatypes::vectors::StringVector;
 
 use crate::information_schema::table_names::*;
 
+const UNKNOWN: &str = "unknown";
+
 /// Find the schema and columns by the table_name, only valid for memory tables.
 /// Safety: the user MUST ensure the table schema exists, panic otherwise.
 pub fn get_schema_columns(table_name: &str) -> (SchemaRef, Vec<VectorRef>) {
@@ -70,340 +72,29 @@ pub fn get_schema_columns(table_name: &str) -> (SchemaRef, Vec<VectorRef>) {
             ],
         ),
 
-        BUILD_INFO => {
-            let build_info = common_version::build_info();
-            (
-                string_columns(&[
-                    "GIT_BRANCH",
-                    "GIT_COMMIT",
-                    "GIT_COMMIT_SHORT",
-                    "GIT_DIRTY",
-                    "PKG_VERSION",
-                ]),
-                vec![
-                    Arc::new(StringVector::from(vec![build_info.branch.to_string()])),
-                    Arc::new(StringVector::from(vec![build_info.commit.to_string()])),
-                    Arc::new(StringVector::from(vec![build_info
-                        .commit_short
-                        .to_string()])),
-                    Arc::new(StringVector::from(vec![build_info.dirty.to_string()])),
-                    Arc::new(StringVector::from(vec![build_info.version.to_string()])),
-                ],
-            )
-        }
-
-        CHARACTER_SETS => (
-            vec![
-                string_column("CHARACTER_SET_NAME"),
-                string_column("DEFAULT_COLLATE_NAME"),
-                string_column("DESCRIPTION"),
-                bigint_column("MAXLEN"),
-            ],
-            vec![
-                Arc::new(StringVector::from(vec!["utf8"])),
-                Arc::new(StringVector::from(vec!["utf8_bin"])),
-                Arc::new(StringVector::from(vec!["UTF-8 Unicode"])),
-                Arc::new(Int64Vector::from_slice([4])),
-            ],
-        ),
-
-        COLLATIONS => (
-            vec![
-                string_column("COLLATION_NAME"),
-                string_column("CHARACTER_SET_NAME"),
-                bigint_column("ID"),
-                string_column("IS_DEFAULT"),
-                string_column("IS_COMPILED"),
-                bigint_column("SORTLEN"),
-            ],
-            vec![
-                Arc::new(StringVector::from(vec!["utf8_bin"])),
-                Arc::new(StringVector::from(vec!["utf8"])),
-                Arc::new(Int64Vector::from_slice([1])),
-                Arc::new(StringVector::from(vec!["Yes"])),
-                Arc::new(StringVector::from(vec!["Yes"])),
-                Arc::new(Int64Vector::from_slice([1])),
-            ],
-        ),
-
-        COLLATION_CHARACTER_SET_APPLICABILITY => (
-            vec![
-                string_column("COLLATION_NAME"),
-                string_column("CHARACTER_SET_NAME"),
-            ],
-            vec![
-                Arc::new(StringVector::from(vec!["utf8_bin"])),
-                Arc::new(StringVector::from(vec!["utf8"])),
-            ],
-        ),
-
-        CHECK_CONSTRAINTS => (
+        BUILD_INFO => (
             string_columns(&[
-                "CONSTRAINT_CATALOG",
-                "CONSTRAINT_SCHEMA",
-                "CONSTRAINT_NAME",
-                "CHECK_CLAUSE",
+                "GIT_BRANCH",
+                "GIT_COMMIT",
+                "GIT_COMMIT_SHORT",
+                "GIT_DIRTY",
+                "PKG_VERSION",
             ]),
-            // Not support check constraints yet
-            vec![],
-        ),
-
-        EVENTS => (
             vec![
-                string_column("EVENT_CATALOG"),
-                string_column("EVENT_SCHEMA"),
-                string_column("EVENT_NAME"),
-                string_column("DEFINER"),
-                string_column("TIME_ZONE"),
-                string_column("EVENT_BODY"),
-                string_column("EVENT_DEFINITION"),
-                string_column("EVENT_TYPE"),
-                datetime_column("EXECUTE_AT"),
-                bigint_column("INTERVAL_VALUE"),
-                string_column("INTERVAL_FIELD"),
-                string_column("SQL_MODE"),
-                datetime_column("STARTS"),
-                datetime_column("ENDS"),
-                string_column("STATUS"),
-                string_column("ON_COMPLETION"),
-                datetime_column("CREATED"),
-                datetime_column("LAST_ALTERED"),
-                datetime_column("LAST_EXECUTED"),
-                string_column("EVENT_COMMENT"),
-                bigint_column("ORIGINATOR"),
-                string_column("CHARACTER_SET_CLIENT"),
-                string_column("COLLATION_CONNECTION"),
-                string_column("DATABASE_COLLATION"),
+                Arc::new(StringVector::from(vec![
+                    build_data::get_git_branch().unwrap_or_else(|_| UNKNOWN.to_string())
+                ])),
+                Arc::new(StringVector::from(vec![
+                    build_data::get_git_commit().unwrap_or_else(|_| UNKNOWN.to_string())
+                ])),
+                Arc::new(StringVector::from(vec![
+                    build_data::get_git_commit_short().unwrap_or_else(|_| UNKNOWN.to_string())
+                ])),
+                Arc::new(StringVector::from(vec![
+                    build_data::get_git_dirty().map_or(UNKNOWN.to_string(), |v| v.to_string())
+                ])),
+                Arc::new(StringVector::from(vec![option_env!("CARGO_PKG_VERSION")])),
             ],
-            vec![],
-        ),
-
-        FILES => (
-            vec![
-                bigint_column("FILE_ID"),
-                string_column("FILE_NAME"),
-                string_column("FILE_TYPE"),
-                string_column("TABLESPACE_NAME"),
-                string_column("TABLE_CATALOG"),
-                string_column("TABLE_SCHEMA"),
-                string_column("TABLE_NAME"),
-                string_column("LOGFILE_GROUP_NAME"),
-                bigint_column("LOGFILE_GROUP_NUMBER"),
-                string_column("ENGINE"),
-                string_column("FULLTEXT_KEYS"),
-                bigint_column("DELETED_ROWS"),
-                bigint_column("UPDATE_COUNT"),
-                bigint_column("FREE_EXTENTS"),
-                bigint_column("TOTAL_EXTENTS"),
-                bigint_column("EXTENT_SIZE"),
-                bigint_column("INITIAL_SIZE"),
-                bigint_column("MAXIMUM_SIZE"),
-                bigint_column("AUTOEXTEND_SIZE"),
-                datetime_column("CREATION_TIME"),
-                datetime_column("LAST_UPDATE_TIME"),
-                datetime_column("LAST_ACCESS_TIME"),
-                datetime_column("RECOVER_TIME"),
-                bigint_column("TRANSACTION_COUNTER"),
-                string_column("VERSION"),
-                string_column("ROW_FORMAT"),
-                bigint_column("TABLE_ROWS"),
-                bigint_column("AVG_ROW_LENGTH"),
-                bigint_column("DATA_LENGTH"),
-                bigint_column("MAX_DATA_LENGTH"),
-                bigint_column("INDEX_LENGTH"),
-                bigint_column("DATA_FREE"),
-                datetime_column("CREATE_TIME"),
-                datetime_column("UPDATE_TIME"),
-                datetime_column("CHECK_TIME"),
-                string_column("CHECKSUM"),
-                string_column("STATUS"),
-                string_column("EXTRA"),
-            ],
-            vec![],
-        ),
-
-        OPTIMIZER_TRACE => (
-            vec![
-                string_column("QUERY"),
-                string_column("TRACE"),
-                bigint_column("MISSING_BYTES_BEYOND_MAX_MEM_SIZE"),
-                bigint_column("INSUFFICIENT_PRIVILEGES"),
-            ],
-            vec![],
-        ),
-
-        // MySQL(https://dev.mysql.com/doc/refman/8.2/en/information-schema-parameters-table.html)
-        // has the spec that is different from
-        // PostgreSQL(https://www.postgresql.org/docs/current/infoschema-parameters.html).
-        // Follow `MySQL` spec here.
-        PARAMETERS => (
-            vec![
-                string_column("SPECIFIC_CATALOG"),
-                string_column("SPECIFIC_SCHEMA"),
-                string_column("SPECIFIC_NAME"),
-                bigint_column("ORDINAL_POSITION"),
-                string_column("PARAMETER_MODE"),
-                string_column("PARAMETER_NAME"),
-                string_column("DATA_TYPE"),
-                bigint_column("CHARACTER_MAXIMUM_LENGTH"),
-                bigint_column("CHARACTER_OCTET_LENGTH"),
-                bigint_column("NUMERIC_PRECISION"),
-                bigint_column("NUMERIC_SCALE"),
-                bigint_column("DATETIME_PRECISION"),
-                string_column("CHARACTER_SET_NAME"),
-                string_column("COLLATION_NAME"),
-                string_column("DTD_IDENTIFIER"),
-                string_column("ROUTINE_TYPE"),
-            ],
-            vec![],
-        ),
-
-        PROFILING => (
-            vec![
-                bigint_column("QUERY_ID"),
-                bigint_column("SEQ"),
-                string_column("STATE"),
-                bigint_column("DURATION"),
-                bigint_column("CPU_USER"),
-                bigint_column("CPU_SYSTEM"),
-                bigint_column("CONTEXT_VOLUNTARY"),
-                bigint_column("CONTEXT_INVOLUNTARY"),
-                bigint_column("BLOCK_OPS_IN"),
-                bigint_column("BLOCK_OPS_OUT"),
-                bigint_column("MESSAGES_SENT"),
-                bigint_column("MESSAGES_RECEIVED"),
-                bigint_column("PAGE_FAULTS_MAJOR"),
-                bigint_column("PAGE_FAULTS_MINOR"),
-                bigint_column("SWAPS"),
-                string_column("SOURCE_FUNCTION"),
-                string_column("SOURCE_FILE"),
-                bigint_column("SOURCE_LINE"),
-            ],
-            vec![],
-        ),
-
-        // TODO: _Must_ reimplement this table when foreign key constraint is supported.
-        REFERENTIAL_CONSTRAINTS => (
-            vec![
-                string_column("CONSTRAINT_CATALOG"),
-                string_column("CONSTRAINT_SCHEMA"),
-                string_column("CONSTRAINT_NAME"),
-                string_column("UNIQUE_CONSTRAINT_CATALOG"),
-                string_column("UNIQUE_CONSTRAINT_SCHEMA"),
-                string_column("UNIQUE_CONSTRAINT_NAME"),
-                string_column("MATCH_OPTION"),
-                string_column("UPDATE_RULE"),
-                string_column("DELETE_RULE"),
-                string_column("TABLE_NAME"),
-                string_column("REFERENCED_TABLE_NAME"),
-            ],
-            vec![],
-        ),
-
-        ROUTINES => (
-            vec![
-                string_column("SPECIFIC_NAME"),
-                string_column("ROUTINE_CATALOG"),
-                string_column("ROUTINE_SCHEMA"),
-                string_column("ROUTINE_NAME"),
-                string_column("ROUTINE_TYPE"),
-                string_column("DATA_TYPE"),
-                bigint_column("CHARACTER_MAXIMUM_LENGTH"),
-                bigint_column("CHARACTER_OCTET_LENGTH"),
-                bigint_column("NUMERIC_PRECISION"),
-                bigint_column("NUMERIC_SCALE"),
-                bigint_column("DATETIME_PRECISION"),
-                string_column("CHARACTER_SET_NAME"),
-                string_column("COLLATION_NAME"),
-                string_column("DTD_IDENTIFIER"),
-                string_column("ROUTINE_BODY"),
-                string_column("ROUTINE_DEFINITION"),
-                string_column("EXTERNAL_NAME"),
-                string_column("EXTERNAL_LANGUAGE"),
-                string_column("PARAMETER_STYLE"),
-                string_column("IS_DETERMINISTIC"),
-                string_column("SQL_DATA_ACCESS"),
-                string_column("SQL_PATH"),
-                string_column("SECURITY_TYPE"),
-                datetime_column("CREATED"),
-                datetime_column("LAST_ALTERED"),
-                string_column("SQL_MODE"),
-                string_column("ROUTINE_COMMENT"),
-                string_column("DEFINER"),
-                string_column("CHARACTER_SET_CLIENT"),
-                string_column("COLLATION_CONNECTION"),
-                string_column("DATABASE_COLLATION"),
-            ],
-            vec![],
-        ),
-
-        SCHEMA_PRIVILEGES => (
-            vec![
-                string_column("GRANTEE"),
-                string_column("TABLE_CATALOG"),
-                string_column("TABLE_SCHEMA"),
-                string_column("PRIVILEGE_TYPE"),
-                string_column("IS_GRANTABLE"),
-            ],
-            vec![],
-        ),
-
-        TABLE_PRIVILEGES => (
-            vec![
-                string_column("GRANTEE"),
-                string_column("TABLE_CATALOG"),
-                string_column("TABLE_SCHEMA"),
-                string_column("TABLE_NAME"),
-                string_column("PRIVILEGE_TYPE"),
-                string_column("IS_GRANTABLE"),
-            ],
-            vec![],
-        ),
-
-        TRIGGERS => (
-            vec![
-                string_column("TRIGGER_CATALOG"),
-                string_column("TRIGGER_SCHEMA"),
-                string_column("TRIGGER_NAME"),
-                string_column("EVENT_MANIPULATION"),
-                string_column("EVENT_OBJECT_CATALOG"),
-                string_column("EVENT_OBJECT_SCHEMA"),
-                string_column("EVENT_OBJECT_TABLE"),
-                bigint_column("ACTION_ORDER"),
-                string_column("ACTION_CONDITION"),
-                string_column("ACTION_STATEMENT"),
-                string_column("ACTION_ORIENTATION"),
-                string_column("ACTION_TIMING"),
-                string_column("ACTION_REFERENCE_OLD_TABLE"),
-                string_column("ACTION_REFERENCE_NEW_TABLE"),
-                string_column("ACTION_REFERENCE_OLD_ROW"),
-                string_column("ACTION_REFERENCE_NEW_ROW"),
-                datetime_column("CREATED"),
-                string_column("SQL_MODE"),
-                string_column("DEFINER"),
-                string_column("CHARACTER_SET_CLIENT"),
-                string_column("COLLATION_CONNECTION"),
-                string_column("DATABASE_COLLATION"),
-            ],
-            vec![],
-        ),
-
-        // TODO: Considering store internal metrics in `global_status` and
-        // `session_status` tables.
-        GLOBAL_STATUS => (
-            vec![
-                string_column("VARIABLE_NAME"),
-                string_column("VARIABLE_VALUE"),
-            ],
-            vec![],
-        ),
-
-        SESSION_STATUS => (
-            vec![
-                string_column("VARIABLE_NAME"),
-                string_column("VARIABLE_VALUE"),
-            ],
-            vec![],
         ),
 
         _ => unreachable!("Unknown table in information_schema: {}", table_name),
@@ -424,22 +115,6 @@ fn string_column(name: &str) -> ColumnSchema {
     )
 }
 
-fn bigint_column(name: &str) -> ColumnSchema {
-    ColumnSchema::new(
-        str::to_lowercase(name),
-        ConcreteDataType::int64_datatype(),
-        false,
-    )
-}
-
-fn datetime_column(name: &str) -> ColumnSchema {
-    ColumnSchema::new(
-        str::to_lowercase(name),
-        ConcreteDataType::datetime_datatype(),
-        false,
-    )
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

src/catalog/src/information_schema/partitions.rs

@@ -1,399 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed 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.
-
-use std::sync::{Arc, Weak};
-
-use arrow_schema::SchemaRef as ArrowSchemaRef;
-use common_catalog::consts::INFORMATION_SCHEMA_PARTITIONS_TABLE_ID;
-use common_error::ext::BoxedError;
-use common_query::physical_plan::TaskContext;
-use common_recordbatch::adapter::RecordBatchStreamAdapter;
-use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
-use common_time::datetime::DateTime;
-use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
-use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
-use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
-use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
-use datatypes::value::Value;
-use datatypes::vectors::{
-    ConstantVector, DateTimeVector, DateTimeVectorBuilder, Int64Vector, Int64VectorBuilder,
-    MutableVector, StringVector, StringVectorBuilder, UInt64VectorBuilder,
-};
-use futures::TryStreamExt;
-use partition::manager::PartitionInfo;
-use partition::partition::PartitionDef;
-use snafu::{OptionExt, ResultExt};
-use store_api::storage::{RegionId, ScanRequest, TableId};
-use table::metadata::{TableInfo, TableType};
-
-use super::PARTITIONS;
-use crate::error::{
-    CreateRecordBatchSnafu, FindPartitionsSnafu, InternalSnafu, Result,
-    UpgradeWeakCatalogManagerRefSnafu,
-};
-use crate::information_schema::{InformationTable, Predicates};
-use crate::kvbackend::KvBackendCatalogManager;
-use crate::CatalogManager;
-
-const TABLE_CATALOG: &str = "table_catalog";
-const TABLE_SCHEMA: &str = "table_schema";
-const TABLE_NAME: &str = "table_name";
-const PARTITION_NAME: &str = "partition_name";
-const PARTITION_EXPRESSION: &str = "partition_expression";
-/// The region id
-const GREPTIME_PARTITION_ID: &str = "greptime_partition_id";
-const INIT_CAPACITY: usize = 42;
-
-/// The `PARTITIONS` table provides information about partitioned tables.
-/// See https://dev.mysql.com/doc/refman/8.0/en/information-schema-partitions-table.html
-/// We provide an extral column `greptime_partition_id` for GreptimeDB region id.
-pub(super) struct InformationSchemaPartitions {
-    schema: SchemaRef,
-    catalog_name: String,
-    catalog_manager: Weak<dyn CatalogManager>,
-}
-
-impl InformationSchemaPartitions {
-    pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
-        Self {
-            schema: Self::schema(),
-            catalog_name,
-            catalog_manager,
-        }
-    }
-
-    pub(crate) fn schema() -> SchemaRef {
-        Arc::new(Schema::new(vec![
-            ColumnSchema::new(TABLE_CATALOG, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(TABLE_SCHEMA, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(TABLE_NAME, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(PARTITION_NAME, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(
-                "subpartition_name",
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            ColumnSchema::new(
-                "partition_ordinal_position",
-                ConcreteDataType::int64_datatype(),
-                true,
-            ),
-            ColumnSchema::new(
-                "subpartition_ordinal_position",
-                ConcreteDataType::int64_datatype(),
-                true,
-            ),
-            ColumnSchema::new(
-                "partition_method",
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            ColumnSchema::new(
-                "subpartition_method",
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            ColumnSchema::new(
-                PARTITION_EXPRESSION,
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            ColumnSchema::new(
-                "subpartition_expression",
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            ColumnSchema::new(
-                "partition_description",
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            ColumnSchema::new("table_rows", ConcreteDataType::int64_datatype(), true),
-            ColumnSchema::new("avg_row_length", ConcreteDataType::int64_datatype(), true),
-            ColumnSchema::new("data_length", ConcreteDataType::int64_datatype(), true),
-            ColumnSchema::new("max_data_length", ConcreteDataType::int64_datatype(), true),
-            ColumnSchema::new("index_length", ConcreteDataType::int64_datatype(), true),
-            ColumnSchema::new("data_free", ConcreteDataType::int64_datatype(), true),
-            ColumnSchema::new("create_time", ConcreteDataType::datetime_datatype(), true),
-            ColumnSchema::new("update_time", ConcreteDataType::datetime_datatype(), true),
-            ColumnSchema::new("check_time", ConcreteDataType::datetime_datatype(), true),
-            ColumnSchema::new("checksum", ConcreteDataType::int64_datatype(), true),
-            ColumnSchema::new(
-                "partition_comment",
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            ColumnSchema::new("nodegroup", ConcreteDataType::string_datatype(), true),
-            ColumnSchema::new("tablespace_name", ConcreteDataType::string_datatype(), true),
-            ColumnSchema::new(
-                GREPTIME_PARTITION_ID,
-                ConcreteDataType::uint64_datatype(),
-                true,
-            ),
-        ]))
-    }
-
-    fn builder(&self) -> InformationSchemaPartitionsBuilder {
-        InformationSchemaPartitionsBuilder::new(
-            self.schema.clone(),
-            self.catalog_name.clone(),
-            self.catalog_manager.clone(),
-        )
-    }
-}
-
-impl InformationTable for InformationSchemaPartitions {
-    fn table_id(&self) -> TableId {
-        INFORMATION_SCHEMA_PARTITIONS_TABLE_ID
-    }
-
-    fn table_name(&self) -> &'static str {
-        PARTITIONS
-    }
-
-    fn schema(&self) -> SchemaRef {
-        self.schema.clone()
-    }
-
-    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
-        let schema = self.schema.arrow_schema().clone();
-        let mut builder = self.builder();
-        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
-            schema,
-            futures::stream::once(async move {
-                builder
-                    .make_partitions(Some(request))
-                    .await
-                    .map(|x| x.into_df_record_batch())
-                    .map_err(Into::into)
-            }),
-        ));
-        Ok(Box::pin(
-            RecordBatchStreamAdapter::try_new(stream)
-                .map_err(BoxedError::new)
-                .context(InternalSnafu)?,
-        ))
-    }
-}
-
-struct InformationSchemaPartitionsBuilder {
-    schema: SchemaRef,
-    catalog_name: String,
-    catalog_manager: Weak<dyn CatalogManager>,
-
-    catalog_names: StringVectorBuilder,
-    schema_names: StringVectorBuilder,
-    table_names: StringVectorBuilder,
-    partition_names: StringVectorBuilder,
-    partition_ordinal_positions: Int64VectorBuilder,
-    partition_expressions: StringVectorBuilder,
-    create_times: DateTimeVectorBuilder,
-    partition_ids: UInt64VectorBuilder,
-}
-
-impl InformationSchemaPartitionsBuilder {
-    fn new(
-        schema: SchemaRef,
-        catalog_name: String,
-        catalog_manager: Weak<dyn CatalogManager>,
-    ) -> Self {
-        Self {
-            schema,
-            catalog_name,
-            catalog_manager,
-            catalog_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            schema_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            partition_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            partition_ordinal_positions: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
-            partition_expressions: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            create_times: DateTimeVectorBuilder::with_capacity(INIT_CAPACITY),
-            partition_ids: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
-        }
-    }
-
-    /// Construct the `information_schema.partitions` virtual table
-    async fn make_partitions(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
-        let catalog_name = self.catalog_name.clone();
-        let catalog_manager = self
-            .catalog_manager
-            .upgrade()
-            .context(UpgradeWeakCatalogManagerRefSnafu)?;
-
-        let partition_manager = catalog_manager
-            .as_any()
-            .downcast_ref::<KvBackendCatalogManager>()
-            .map(|catalog_manager| catalog_manager.partition_manager());
-
-        let predicates = Predicates::from_scan_request(&request);
-
-        for schema_name in catalog_manager.schema_names(&catalog_name).await? {
-            let mut stream = catalog_manager.tables(&catalog_name, &schema_name).await;
-
-            while let Some(table) = stream.try_next().await? {
-                let table_info = table.table_info();
-
-                if table_info.table_type == TableType::Temporary {
-                    continue;
-                }
-
-                let table_id = table_info.ident.table_id;
-                let partitions = if let Some(partition_manager) = &partition_manager {
-                    partition_manager
-                        .find_table_partitions(table_id)
-                        .await
-                        .context(FindPartitionsSnafu {
-                            table: &table_info.name,
-                        })?
-                } else {
-                    // Current node must be a standalone instance, contains only one partition by default.
-                    // TODO(dennis): change it when we support multi-regions for standalone.
-                    vec![PartitionInfo {
-                        id: RegionId::new(table_id, 0),
-                        partition: PartitionDef::new(vec![], vec![]),
-                    }]
-                };
-
-                self.add_partitions(
-                    &predicates,
-                    &table_info,
-                    &catalog_name,
-                    &schema_name,
-                    &table_info.name,
-                    &partitions,
-                );
-            }
-        }
-
-        self.finish()
-    }
-
-    #[allow(clippy::too_many_arguments)]
-    fn add_partitions(
-        &mut self,
-        predicates: &Predicates,
-        table_info: &TableInfo,
-        catalog_name: &str,
-        schema_name: &str,
-        table_name: &str,
-        partitions: &[PartitionInfo],
-    ) {
-        let row = [
-            (TABLE_CATALOG, &Value::from(catalog_name)),
-            (TABLE_SCHEMA, &Value::from(schema_name)),
-            (TABLE_NAME, &Value::from(table_name)),
-        ];
-
-        if !predicates.eval(&row) {
-            return;
-        }
-
-        for (index, partition) in partitions.iter().enumerate() {
-            let partition_name = format!("p{index}");
-
-            self.catalog_names.push(Some(catalog_name));
-            self.schema_names.push(Some(schema_name));
-            self.table_names.push(Some(table_name));
-            self.partition_names.push(Some(&partition_name));
-            self.partition_ordinal_positions
-                .push(Some((index + 1) as i64));
-            let expressions = if partition.partition.partition_columns().is_empty() {
-                None
-            } else {
-                Some(partition.partition.to_string())
-            };
-
-            self.partition_expressions.push(expressions.as_deref());
-            self.create_times.push(Some(DateTime::from(
-                table_info.meta.created_on.timestamp_millis(),
-            )));
-            self.partition_ids.push(Some(partition.id.as_u64()));
-        }
-    }
-
-    fn finish(&mut self) -> Result<RecordBatch> {
-        let rows_num = self.catalog_names.len();
-
-        let null_string_vector = Arc::new(ConstantVector::new(
-            Arc::new(StringVector::from(vec![None as Option<&str>])),
-            rows_num,
-        ));
-        let null_i64_vector = Arc::new(ConstantVector::new(
-            Arc::new(Int64Vector::from(vec![None])),
-            rows_num,
-        ));
-        let null_datetime_vector = Arc::new(ConstantVector::new(
-            Arc::new(DateTimeVector::from(vec![None])),
-            rows_num,
-        ));
-        let partition_methods = Arc::new(ConstantVector::new(
-            Arc::new(StringVector::from(vec![Some("RANGE")])),
-            rows_num,
-        ));
-
-        let columns: Vec<VectorRef> = vec![
-            Arc::new(self.catalog_names.finish()),
-            Arc::new(self.schema_names.finish()),
-            Arc::new(self.table_names.finish()),
-            Arc::new(self.partition_names.finish()),
-            null_string_vector.clone(),
-            Arc::new(self.partition_ordinal_positions.finish()),
-            null_i64_vector.clone(),
-            partition_methods,
-            null_string_vector.clone(),
-            Arc::new(self.partition_expressions.finish()),
-            null_string_vector.clone(),
-            null_string_vector.clone(),
-            // TODO(dennis): rows and index statistics info
-            null_i64_vector.clone(),
-            null_i64_vector.clone(),
-            null_i64_vector.clone(),
-            null_i64_vector.clone(),
-            null_i64_vector.clone(),
-            null_i64_vector.clone(),
-            Arc::new(self.create_times.finish()),
-            // TODO(dennis): supports update_time
-            null_datetime_vector.clone(),
-            null_datetime_vector,
-            null_i64_vector,
-            null_string_vector.clone(),
-            null_string_vector.clone(),
-            null_string_vector,
-            Arc::new(self.partition_ids.finish()),
-        ];
-        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
-    }
-}
-
-impl DfPartitionStream for InformationSchemaPartitions {
-    fn schema(&self) -> &ArrowSchemaRef {
-        self.schema.arrow_schema()
-    }
-
-    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
-        let schema = self.schema.arrow_schema().clone();
-        let mut builder = self.builder();
-        Box::pin(DfRecordBatchStreamAdapter::new(
-            schema,
-            futures::stream::once(async move {
-                builder
-                    .make_partitions(None)
-                    .await
-                    .map(|x| x.into_df_record_batch())
-                    .map_err(Into::into)
-            }),
-        ))
-    }
-}

src/catalog/src/information_schema/predicate.rs

@@ -1,609 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed 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.
-
-use arrow::array::StringArray;
-use arrow::compute::kernels::comparison;
-use common_query::logical_plan::DfExpr;
-use datafusion::common::ScalarValue;
-use datafusion::logical_expr::expr::Like;
-use datafusion::logical_expr::Operator;
-use datatypes::value::Value;
-use store_api::storage::ScanRequest;
-
-type ColumnName = String;
-/// Predicate to filter `information_schema` tables stream,
-/// we only support these simple predicates currently.
-/// TODO(dennis): supports more predicate types.
-#[derive(Clone, PartialEq, Eq, Debug)]
-enum Predicate {
-    Eq(ColumnName, Value),
-    Like(ColumnName, String, bool),
-    NotEq(ColumnName, Value),
-    InList(ColumnName, Vec<Value>),
-    And(Box<Predicate>, Box<Predicate>),
-    Or(Box<Predicate>, Box<Predicate>),
-    Not(Box<Predicate>),
-}
-
-impl Predicate {
-    /// Evaluate the predicate with the row, returns:
-    /// - `None` when the predicate can't evaluate with the row.
-    /// - `Some(true)` when the predicate is satisfied,
-    /// - `Some(false)` when the predicate is not satisfied,
-    fn eval(&self, row: &[(&str, &Value)]) -> Option<bool> {
-        match self {
-            Predicate::Eq(c, v) => {
-                for (column, value) in row {
-                    if c != column {
-                        continue;
-                    }
-                    return Some(v == *value);
-                }
-            }
-            Predicate::Like(c, pattern, case_insensitive) => {
-                for (column, value) in row {
-                    if c != column {
-                        continue;
-                    }
-
-                    let Value::String(bs) = value else {
-                        continue;
-                    };
-
-                    return like_utf8(bs.as_utf8(), pattern, case_insensitive);
-                }
-            }
-            Predicate::NotEq(c, v) => {
-                for (column, value) in row {
-                    if c != column {
-                        continue;
-                    }
-                    return Some(v != *value);
-                }
-            }
-            Predicate::InList(c, values) => {
-                for (column, value) in row {
-                    if c != column {
-                        continue;
-                    }
-                    return Some(values.iter().any(|v| v == *value));
-                }
-            }
-            Predicate::And(left, right) => {
-                let left = left.eval(row);
-
-                // short-circuit
-                if matches!(left, Some(false)) {
-                    return Some(false);
-                }
-
-                return match (left, right.eval(row)) {
-                    (Some(left), Some(right)) => Some(left && right),
-                    (None, Some(false)) => Some(false),
-                    _ => None,
-                };
-            }
-            Predicate::Or(left, right) => {
-                let left = left.eval(row);
-
-                // short-circuit
-                if matches!(left, Some(true)) {
-                    return Some(true);
-                }
-
-                return match (left, right.eval(row)) {
-                    (Some(left), Some(right)) => Some(left || right),
-                    (None, Some(true)) => Some(true),
-                    _ => None,
-                };
-            }
-            Predicate::Not(p) => {
-                let Some(b) = p.eval(row) else {
-                    return None;
-                };
-
-                return Some(!b);
-            }
-        }
-
-        // Can't evaluate predicate with the row
-        None
-    }
-
-    /// Try to create a predicate from datafusion [`Expr`], return None if fails.
-    fn from_expr(expr: DfExpr) -> Option<Predicate> {
-        match expr {
-            // NOT expr
-            DfExpr::Not(expr) => {
-                let Some(p) = Self::from_expr(*expr) else {
-                    return None;
-                };
-
-                Some(Predicate::Not(Box::new(p)))
-            }
-            // expr LIKE pattern
-            DfExpr::Like(Like {
-                negated,
-                expr,
-                pattern,
-                case_insensitive,
-                ..
-            }) if is_column(&expr) && is_string_literal(&pattern) => {
-                // Safety: ensured by gurad
-                let DfExpr::Column(c) = *expr else {
-                    unreachable!();
-                };
-                let DfExpr::Literal(ScalarValue::Utf8(Some(pattern))) = *pattern else {
-                    unreachable!();
-                };
-
-                let p = Predicate::Like(c.name, pattern, case_insensitive);
-
-                if negated {
-                    Some(Predicate::Not(Box::new(p)))
-                } else {
-                    Some(p)
-                }
-            }
-            // left OP right
-            DfExpr::BinaryExpr(bin) => match (*bin.left, bin.op, *bin.right) {
-                // left == right
-                (DfExpr::Literal(scalar), Operator::Eq, DfExpr::Column(c))
-                | (DfExpr::Column(c), Operator::Eq, DfExpr::Literal(scalar)) => {
-                    let Ok(v) = Value::try_from(scalar) else {
-                        return None;
-                    };
-
-                    Some(Predicate::Eq(c.name, v))
-                }
-                // left != right
-                (DfExpr::Literal(scalar), Operator::NotEq, DfExpr::Column(c))
-                | (DfExpr::Column(c), Operator::NotEq, DfExpr::Literal(scalar)) => {
-                    let Ok(v) = Value::try_from(scalar) else {
-                        return None;
-                    };
-
-                    Some(Predicate::NotEq(c.name, v))
-                }
-                // left AND right
-                (left, Operator::And, right) => {
-                    let Some(left) = Self::from_expr(left) else {
-                        return None;
-                    };
-
-                    let Some(right) = Self::from_expr(right) else {
-                        return None;
-                    };
-
-                    Some(Predicate::And(Box::new(left), Box::new(right)))
-                }
-                // left OR right
-                (left, Operator::Or, right) => {
-                    let Some(left) = Self::from_expr(left) else {
-                        return None;
-                    };
-
-                    let Some(right) = Self::from_expr(right) else {
-                        return None;
-                    };
-
-                    Some(Predicate::Or(Box::new(left), Box::new(right)))
-                }
-                _ => None,
-            },
-            // [NOT] IN (LIST)
-            DfExpr::InList(list) => {
-                match (*list.expr, list.list, list.negated) {
-                    // column [NOT] IN (v1, v2, v3, ...)
-                    (DfExpr::Column(c), list, negated) if is_all_scalars(&list) => {
-                        let mut values = Vec::with_capacity(list.len());
-                        for scalar in list {
-                            // Safety: checked by `is_all_scalars`
-                            let DfExpr::Literal(scalar) = scalar else {
-                                unreachable!();
-                            };
-
-                            let Ok(value) = Value::try_from(scalar) else {
-                                return None;
-                            };
-
-                            values.push(value);
-                        }
-
-                        let predicate = Predicate::InList(c.name, values);
-
-                        if negated {
-                            Some(Predicate::Not(Box::new(predicate)))
-                        } else {
-                            Some(predicate)
-                        }
-                    }
-                    _ => None,
-                }
-            }
-            _ => None,
-        }
-    }
-}
-
-/// Perform SQL left LIKE right, return `None` if fail to evaluate.
-/// - `s` the target string
-/// - `pattern` the pattern just like '%abc'
-/// - `case_insensitive` whether to perform case-insensitive like or not.
-fn like_utf8(s: &str, pattern: &str, case_insensitive: &bool) -> Option<bool> {
-    let array = StringArray::from(vec![s]);
-    let patterns = StringArray::new_scalar(pattern);
-
-    let Ok(booleans) = (if *case_insensitive {
-        comparison::ilike(&array, &patterns)
-    } else {
-        comparison::like(&array, &patterns)
-    }) else {
-        return None;
-    };
-
-    // Safety: at least one value in result
-    Some(booleans.value(0))
-}
-
-fn is_string_literal(expr: &DfExpr) -> bool {
-    matches!(expr, DfExpr::Literal(ScalarValue::Utf8(Some(_))))
-}
-
-fn is_column(expr: &DfExpr) -> bool {
-    matches!(expr, DfExpr::Column(_))
-}
-
-/// A list of predicate
-pub struct Predicates {
-    predicates: Vec<Predicate>,
-}
-
-impl Predicates {
-    /// Try its best to create predicates from [`ScanRequest`].
-    pub fn from_scan_request(request: &Option<ScanRequest>) -> Predicates {
-        if let Some(request) = request {
-            let mut predicates = Vec::with_capacity(request.filters.len());
-
-            for filter in &request.filters {
-                if let Some(predicate) = Predicate::from_expr(filter.df_expr().clone()) {
-                    predicates.push(predicate);
-                }
-            }
-
-            Self { predicates }
-        } else {
-            Self {
-                predicates: Vec::new(),
-            }
-        }
-    }
-
-    /// Evaluate the predicates with the row.
-    /// returns true when all the predicates are satisfied or can't be evaluated.
-    pub fn eval(&self, row: &[(&str, &Value)]) -> bool {
-        // fast path
-        if self.predicates.is_empty() {
-            return true;
-        }
-
-        self.predicates
-            .iter()
-            .filter_map(|p| p.eval(row))
-            .all(|b| b)
-    }
-}
-
-/// Returns true when the values are all [`DfExpr::Literal`].
-fn is_all_scalars(list: &[DfExpr]) -> bool {
-    list.iter().all(|v| matches!(v, DfExpr::Literal(_)))
-}
-
-#[cfg(test)]
-mod tests {
-    use datafusion::common::{Column, ScalarValue};
-    use datafusion::logical_expr::expr::InList;
-    use datafusion::logical_expr::BinaryExpr;
-
-    use super::*;
-
-    #[test]
-    fn test_predicate_eval() {
-        let a_col = "a".to_string();
-        let b_col = "b".to_string();
-        let a_value = Value::from("a_value");
-        let b_value = Value::from("b_value");
-        let wrong_value = Value::from("wrong_value");
-
-        let a_row = [(a_col.as_str(), &a_value)];
-        let b_row = [("b", &wrong_value)];
-        let wrong_row = [(a_col.as_str(), &wrong_value)];
-
-        // Predicate::Eq
-        let p = Predicate::Eq(a_col.clone(), a_value.clone());
-        assert!(p.eval(&a_row).unwrap());
-        assert!(p.eval(&b_row).is_none());
-        assert!(!p.eval(&wrong_row).unwrap());
-
-        // Predicate::NotEq
-        let p = Predicate::NotEq(a_col.clone(), a_value.clone());
-        assert!(!p.eval(&a_row).unwrap());
-        assert!(p.eval(&b_row).is_none());
-        assert!(p.eval(&wrong_row).unwrap());
-
-        // Predicate::InList
-        let p = Predicate::InList(a_col.clone(), vec![a_value.clone(), b_value.clone()]);
-        assert!(p.eval(&a_row).unwrap());
-        assert!(p.eval(&b_row).is_none());
-        assert!(!p.eval(&wrong_row).unwrap());
-        assert!(p.eval(&[(&a_col, &b_value)]).unwrap());
-
-        let p1 = Predicate::Eq(a_col.clone(), a_value.clone());
-        let p2 = Predicate::Eq(b_col.clone(), b_value.clone());
-        let row = [(a_col.as_str(), &a_value), (b_col.as_str(), &b_value)];
-        let wrong_row = [(a_col.as_str(), &a_value), (b_col.as_str(), &wrong_value)];
-
-        //Predicate::And
-        let p = Predicate::And(Box::new(p1.clone()), Box::new(p2.clone()));
-        assert!(p.eval(&row).unwrap());
-        assert!(!p.eval(&wrong_row).unwrap());
-        assert!(p.eval(&[]).is_none());
-        assert!(p.eval(&[("c", &a_value)]).is_none());
-        assert!(!p
-            .eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &a_value)])
-            .unwrap());
-        assert!(!p
-            .eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &b_value)])
-            .unwrap());
-        assert!(p
-            .eval(&[(a_col.as_ref(), &a_value), ("c", &a_value)])
-            .is_none());
-        assert!(!p
-            .eval(&[(a_col.as_ref(), &b_value), ("c", &a_value)])
-            .unwrap());
-
-        //Predicate::Or
-        let p = Predicate::Or(Box::new(p1), Box::new(p2));
-        assert!(p.eval(&row).unwrap());
-        assert!(p.eval(&wrong_row).unwrap());
-        assert!(p.eval(&[]).is_none());
-        assert!(p.eval(&[("c", &a_value)]).is_none());
-        assert!(!p
-            .eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &a_value)])
-            .unwrap());
-        assert!(p
-            .eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &b_value)])
-            .unwrap());
-        assert!(p
-            .eval(&[(a_col.as_ref(), &a_value), ("c", &a_value)])
-            .unwrap());
-        assert!(p
-            .eval(&[(a_col.as_ref(), &b_value), ("c", &a_value)])
-            .is_none());
-    }
-
-    #[test]
-    fn test_predicate_like() {
-        // case insensitive
-        let expr = DfExpr::Like(Like {
-            negated: false,
-            expr: Box::new(column("a")),
-            pattern: Box::new(string_literal("%abc")),
-            case_insensitive: true,
-            escape_char: None,
-        });
-
-        let p = Predicate::from_expr(expr).unwrap();
-        assert!(
-            matches!(&p, Predicate::Like(c, pattern, case_insensitive) if
-                         c == "a"
-                         && pattern == "%abc"
-                         && *case_insensitive)
-        );
-
-        let match_row = [
-            ("a", &Value::from("hello AbC")),
-            ("b", &Value::from("b value")),
-        ];
-        let unmatch_row = [("a", &Value::from("bca")), ("b", &Value::from("b value"))];
-
-        assert!(p.eval(&match_row).unwrap());
-        assert!(!p.eval(&unmatch_row).unwrap());
-        assert!(p.eval(&[]).is_none());
-
-        // case sensitive
-        let expr = DfExpr::Like(Like {
-            negated: false,
-            expr: Box::new(column("a")),
-            pattern: Box::new(string_literal("%abc")),
-            case_insensitive: false,
-            escape_char: None,
-        });
-
-        let p = Predicate::from_expr(expr).unwrap();
-        assert!(
-            matches!(&p, Predicate::Like(c, pattern, case_insensitive) if
-                         c == "a"
-                         && pattern == "%abc"
-                         && !*case_insensitive)
-        );
-        assert!(!p.eval(&match_row).unwrap());
-        assert!(!p.eval(&unmatch_row).unwrap());
-        assert!(p.eval(&[]).is_none());
-
-        // not like
-        let expr = DfExpr::Like(Like {
-            negated: true,
-            expr: Box::new(column("a")),
-            pattern: Box::new(string_literal("%abc")),
-            case_insensitive: true,
-            escape_char: None,
-        });
-
-        let p = Predicate::from_expr(expr).unwrap();
-        assert!(!p.eval(&match_row).unwrap());
-        assert!(p.eval(&unmatch_row).unwrap());
-        assert!(p.eval(&[]).is_none());
-    }
-
-    fn column(name: &str) -> DfExpr {
-        DfExpr::Column(Column {
-            relation: None,
-            name: name.to_string(),
-        })
-    }
-
-    fn string_literal(v: &str) -> DfExpr {
-        DfExpr::Literal(ScalarValue::Utf8(Some(v.to_string())))
-    }
-
-    fn match_string_value(v: &Value, expected: &str) -> bool {
-        matches!(v, Value::String(bs) if bs.as_utf8() == expected)
-    }
-
-    fn match_string_values(vs: &[Value], expected: &[&str]) -> bool {
-        assert_eq!(vs.len(), expected.len());
-
-        let mut result = true;
-        for (i, v) in vs.iter().enumerate() {
-            result = result && match_string_value(v, expected[i]);
-        }
-
-        result
-    }
-
-    fn mock_exprs() -> (DfExpr, DfExpr) {
-        let expr1 = DfExpr::BinaryExpr(BinaryExpr {
-            left: Box::new(column("a")),
-            op: Operator::Eq,
-            right: Box::new(string_literal("a_value")),
-        });
-
-        let expr2 = DfExpr::BinaryExpr(BinaryExpr {
-            left: Box::new(column("b")),
-            op: Operator::NotEq,
-            right: Box::new(string_literal("b_value")),
-        });
-
-        (expr1, expr2)
-    }
-
-    #[test]
-    fn test_predicate_from_expr() {
-        let (expr1, expr2) = mock_exprs();
-
-        let p1 = Predicate::from_expr(expr1.clone()).unwrap();
-        assert!(matches!(&p1, Predicate::Eq(column, v) if column == "a"
-                         && match_string_value(v, "a_value")));
-
-        let p2 = Predicate::from_expr(expr2.clone()).unwrap();
-        assert!(matches!(&p2, Predicate::NotEq(column, v) if column == "b"
-                         && match_string_value(v, "b_value")));
-
-        let and_expr = DfExpr::BinaryExpr(BinaryExpr {
-            left: Box::new(expr1.clone()),
-            op: Operator::And,
-            right: Box::new(expr2.clone()),
-        });
-        let or_expr = DfExpr::BinaryExpr(BinaryExpr {
-            left: Box::new(expr1.clone()),
-            op: Operator::Or,
-            right: Box::new(expr2.clone()),
-        });
-        let not_expr = DfExpr::Not(Box::new(expr1.clone()));
-
-        let and_p = Predicate::from_expr(and_expr).unwrap();
-        assert!(matches!(and_p, Predicate::And(left, right) if *left == p1 && *right == p2));
-        let or_p = Predicate::from_expr(or_expr).unwrap();
-        assert!(matches!(or_p, Predicate::Or(left, right) if *left == p1 && *right == p2));
-        let not_p = Predicate::from_expr(not_expr).unwrap();
-        assert!(matches!(not_p, Predicate::Not(p) if *p == p1));
-
-        let inlist_expr = DfExpr::InList(InList {
-            expr: Box::new(column("a")),
-            list: vec![string_literal("a1"), string_literal("a2")],
-            negated: false,
-        });
-
-        let inlist_p = Predicate::from_expr(inlist_expr).unwrap();
-        assert!(matches!(&inlist_p, Predicate::InList(c, values) if c == "a"
-                         && match_string_values(values, &["a1", "a2"])));
-
-        let inlist_expr = DfExpr::InList(InList {
-            expr: Box::new(column("a")),
-            list: vec![string_literal("a1"), string_literal("a2")],
-            negated: true,
-        });
-        let inlist_p = Predicate::from_expr(inlist_expr).unwrap();
-        assert!(matches!(inlist_p, Predicate::Not(p) if
-                         matches!(&*p,
-                                  Predicate::InList(c, values) if c == "a"
-                                  && match_string_values(values, &["a1", "a2"]))));
-    }
-
-    #[test]
-    fn test_predicates_from_scan_request() {
-        let predicates = Predicates::from_scan_request(&None);
-        assert!(predicates.predicates.is_empty());
-
-        let (expr1, expr2) = mock_exprs();
-
-        let request = ScanRequest {
-            filters: vec![expr1.into(), expr2.into()],
-            ..Default::default()
-        };
-        let predicates = Predicates::from_scan_request(&Some(request));
-
-        assert_eq!(2, predicates.predicates.len());
-        assert!(
-            matches!(&predicates.predicates[0], Predicate::Eq(column, v) if column == "a"
-                     && match_string_value(v, "a_value"))
-        );
-        assert!(
-            matches!(&predicates.predicates[1], Predicate::NotEq(column, v) if column == "b"
-                     && match_string_value(v, "b_value"))
-        );
-    }
-
-    #[test]
-    fn test_predicates_eval_row() {
-        let wrong_row = [
-            ("a", &Value::from("a_value")),
-            ("b", &Value::from("b_value")),
-            ("c", &Value::from("c_value")),
-        ];
-        let row = [
-            ("a", &Value::from("a_value")),
-            ("b", &Value::from("not_b_value")),
-            ("c", &Value::from("c_value")),
-        ];
-        let c_row = [("c", &Value::from("c_value"))];
-
-        // test empty predicates, always returns true
-        let predicates = Predicates::from_scan_request(&None);
-        assert!(predicates.eval(&row));
-        assert!(predicates.eval(&wrong_row));
-        assert!(predicates.eval(&c_row));
-
-        let (expr1, expr2) = mock_exprs();
-        let request = ScanRequest {
-            filters: vec![expr1.into(), expr2.into()],
-            ..Default::default()
-        };
-        let predicates = Predicates::from_scan_request(&Some(request));
-        assert!(predicates.eval(&row));
-        assert!(!predicates.eval(&wrong_row));
-        assert!(predicates.eval(&c_row));
-    }
-}

src/catalog/src/information_schema/region_peers.rs

@@ -1,279 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed 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.
-
-use core::pin::pin;
-use std::sync::{Arc, Weak};
-
-use arrow_schema::SchemaRef as ArrowSchemaRef;
-use common_catalog::consts::INFORMATION_SCHEMA_REGION_PEERS_TABLE_ID;
-use common_error::ext::BoxedError;
-use common_meta::rpc::router::RegionRoute;
-use common_query::physical_plan::TaskContext;
-use common_recordbatch::adapter::RecordBatchStreamAdapter;
-use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
-use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
-use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
-use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
-use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
-use datatypes::value::Value;
-use datatypes::vectors::{Int64VectorBuilder, StringVectorBuilder, UInt64VectorBuilder};
-use futures::{StreamExt, TryStreamExt};
-use snafu::{OptionExt, ResultExt};
-use store_api::storage::{ScanRequest, TableId};
-use table::metadata::TableType;
-
-use super::REGION_PEERS;
-use crate::error::{
-    CreateRecordBatchSnafu, FindRegionRoutesSnafu, InternalSnafu, Result,
-    UpgradeWeakCatalogManagerRefSnafu,
-};
-use crate::information_schema::{InformationTable, Predicates};
-use crate::kvbackend::KvBackendCatalogManager;
-use crate::CatalogManager;
-
-const REGION_ID: &str = "region_id";
-const PEER_ID: &str = "peer_id";
-const PEER_ADDR: &str = "peer_addr";
-const IS_LEADER: &str = "is_leader";
-const STATUS: &str = "status";
-const DOWN_SECONDS: &str = "down_seconds";
-const INIT_CAPACITY: usize = 42;
-
-/// The `REGION_PEERS` table provides information about the region distribution and routes. Including fields:
-///
-/// - `region_id`: the region id
-/// - `peer_id`: the region storage datanode peer id
-/// - `peer_addr`: the region storage datanode peer address
-/// - `is_leader`: whether the peer is the leader
-/// - `status`: the region status, `ALIVE` or `DOWNGRADED`.
-/// - `down_seconds`: the duration of being offline, in seconds.
-///
-pub(super) struct InformationSchemaRegionPeers {
-    schema: SchemaRef,
-    catalog_name: String,
-    catalog_manager: Weak<dyn CatalogManager>,
-}
-
-impl InformationSchemaRegionPeers {
-    pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
-        Self {
-            schema: Self::schema(),
-            catalog_name,
-            catalog_manager,
-        }
-    }
-
-    pub(crate) fn schema() -> SchemaRef {
-        Arc::new(Schema::new(vec![
-            ColumnSchema::new(REGION_ID, ConcreteDataType::uint64_datatype(), false),
-            ColumnSchema::new(PEER_ID, ConcreteDataType::uint64_datatype(), true),
-            ColumnSchema::new(PEER_ADDR, ConcreteDataType::string_datatype(), true),
-            ColumnSchema::new(IS_LEADER, ConcreteDataType::string_datatype(), true),
-            ColumnSchema::new(STATUS, ConcreteDataType::string_datatype(), true),
-            ColumnSchema::new(DOWN_SECONDS, ConcreteDataType::int64_datatype(), true),
-        ]))
-    }
-
-    fn builder(&self) -> InformationSchemaRegionPeersBuilder {
-        InformationSchemaRegionPeersBuilder::new(
-            self.schema.clone(),
-            self.catalog_name.clone(),
-            self.catalog_manager.clone(),
-        )
-    }
-}
-
-impl InformationTable for InformationSchemaRegionPeers {
-    fn table_id(&self) -> TableId {
-        INFORMATION_SCHEMA_REGION_PEERS_TABLE_ID
-    }
-
-    fn table_name(&self) -> &'static str {
-        REGION_PEERS
-    }
-
-    fn schema(&self) -> SchemaRef {
-        self.schema.clone()
-    }
-
-    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
-        let schema = self.schema.arrow_schema().clone();
-        let mut builder = self.builder();
-        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
-            schema,
-            futures::stream::once(async move {
-                builder
-                    .make_region_peers(Some(request))
-                    .await
-                    .map(|x| x.into_df_record_batch())
-                    .map_err(Into::into)
-            }),
-        ));
-        Ok(Box::pin(
-            RecordBatchStreamAdapter::try_new(stream)
-                .map_err(BoxedError::new)
-                .context(InternalSnafu)?,
-        ))
-    }
-}
-
-struct InformationSchemaRegionPeersBuilder {
-    schema: SchemaRef,
-    catalog_name: String,
-    catalog_manager: Weak<dyn CatalogManager>,
-
-    region_ids: UInt64VectorBuilder,
-    peer_ids: UInt64VectorBuilder,
-    peer_addrs: StringVectorBuilder,
-    is_leaders: StringVectorBuilder,
-    statuses: StringVectorBuilder,
-    down_seconds: Int64VectorBuilder,
-}
-
-impl InformationSchemaRegionPeersBuilder {
-    fn new(
-        schema: SchemaRef,
-        catalog_name: String,
-        catalog_manager: Weak<dyn CatalogManager>,
-    ) -> Self {
-        Self {
-            schema,
-            catalog_name,
-            catalog_manager,
-            region_ids: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
-            peer_ids: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
-            peer_addrs: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            is_leaders: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            statuses: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            down_seconds: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
-        }
-    }
-
-    /// Construct the `information_schema.region_peers` virtual table
-    async fn make_region_peers(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
-        let catalog_name = self.catalog_name.clone();
-        let catalog_manager = self
-            .catalog_manager
-            .upgrade()
-            .context(UpgradeWeakCatalogManagerRefSnafu)?;
-
-        let partition_manager = catalog_manager
-            .as_any()
-            .downcast_ref::<KvBackendCatalogManager>()
-            .map(|catalog_manager| catalog_manager.partition_manager());
-
-        let predicates = Predicates::from_scan_request(&request);
-
-        for schema_name in catalog_manager.schema_names(&catalog_name).await? {
-            let table_id_stream = catalog_manager
-                .tables(&catalog_name, &schema_name)
-                .await
-                .try_filter_map(|t| async move {
-                    let table_info = t.table_info();
-                    if table_info.table_type == TableType::Temporary {
-                        Ok(None)
-                    } else {
-                        Ok(Some(table_info.ident.table_id))
-                    }
-                });
-
-            const BATCH_SIZE: usize = 128;
-
-            // Split table ids into chunks
-            let mut table_id_chunks = pin!(table_id_stream.ready_chunks(BATCH_SIZE));
-
-            while let Some(table_ids) = table_id_chunks.next().await {
-                let table_ids = table_ids.into_iter().collect::<Result<Vec<_>>>()?;
-
-                let table_routes = if let Some(partition_manager) = &partition_manager {
-                    partition_manager
-                        .find_region_routes_batch(&table_ids)
-                        .await
-                        .context(FindRegionRoutesSnafu)?
-                } else {
-                    table_ids.into_iter().map(|id| (id, vec![])).collect()
-                };
-
-                for routes in table_routes.values() {
-                    self.add_region_peers(&predicates, routes);
-                }
-            }
-        }
-
-        self.finish()
-    }
-
-    fn add_region_peers(&mut self, predicates: &Predicates, routes: &[RegionRoute]) {
-        for route in routes {
-            let region_id = route.region.id.as_u64();
-            let peer_id = route.leader_peer.clone().map(|p| p.id);
-            let peer_addr = route.leader_peer.clone().map(|p| p.addr);
-            let status = if let Some(status) = route.leader_status {
-                Some(status.as_ref().to_string())
-            } else {
-                // Alive by default
-                Some("ALIVE".to_string())
-            };
-
-            let row = [(REGION_ID, &Value::from(region_id))];
-
-            if !predicates.eval(&row) {
-                return;
-            }
-
-            // TODO(dennis): adds followers.
-            self.region_ids.push(Some(region_id));
-            self.peer_ids.push(peer_id);
-            self.peer_addrs.push(peer_addr.as_deref());
-            self.is_leaders.push(Some("Yes"));
-            self.statuses.push(status.as_deref());
-            self.down_seconds
-                .push(route.leader_down_millis().map(|m| m / 1000));
-        }
-    }
-
-    fn finish(&mut self) -> Result<RecordBatch> {
-        let columns: Vec<VectorRef> = vec![
-            Arc::new(self.region_ids.finish()),
-            Arc::new(self.peer_ids.finish()),
-            Arc::new(self.peer_addrs.finish()),
-            Arc::new(self.is_leaders.finish()),
-            Arc::new(self.statuses.finish()),
-            Arc::new(self.down_seconds.finish()),
-        ];
-        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
-    }
-}
-
-impl DfPartitionStream for InformationSchemaRegionPeers {
-    fn schema(&self) -> &ArrowSchemaRef {
-        self.schema.arrow_schema()
-    }
-
-    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
-        let schema = self.schema.arrow_schema().clone();
-        let mut builder = self.builder();
-        Box::pin(DfRecordBatchStreamAdapter::new(
-            schema,
-            futures::stream::once(async move {
-                builder
-                    .make_region_peers(None)
-                    .await
-                    .map(|x| x.into_df_record_batch())
-                    .map_err(Into::into)
-            }),
-        ))
-    }
-}

src/catalog/src/information_schema/runtime_metrics.rs

@@ -1,250 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed 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.
-
-use std::sync::Arc;
-
-use arrow_schema::SchemaRef as ArrowSchemaRef;
-use common_catalog::consts::INFORMATION_SCHEMA_RUNTIME_METRICS_TABLE_ID;
-use common_error::ext::BoxedError;
-use common_query::physical_plan::TaskContext;
-use common_recordbatch::adapter::RecordBatchStreamAdapter;
-use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
-use common_time::util::current_time_millis;
-use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
-use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
-use datatypes::prelude::{ConcreteDataType, MutableVector};
-use datatypes::scalars::ScalarVectorBuilder;
-use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
-use datatypes::vectors::{
-    ConstantVector, Float64VectorBuilder, StringVector, StringVectorBuilder,
-    TimestampMillisecondVector, VectorRef,
-};
-use itertools::Itertools;
-use snafu::ResultExt;
-use store_api::storage::{ScanRequest, TableId};
-
-use super::{InformationTable, RUNTIME_METRICS};
-use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
-
-pub(super) struct InformationSchemaMetrics {
-    schema: SchemaRef,
-}
-
-const METRIC_NAME: &str = "metric_name";
-const METRIC_VALUE: &str = "value";
-const METRIC_LABELS: &str = "labels";
-const NODE: &str = "node";
-const NODE_TYPE: &str = "node_type";
-const TIMESTAMP: &str = "timestamp";
-
-/// The `information_schema.runtime_metrics` virtual table.
-/// It provides the GreptimeDB runtime metrics for the users by SQL.
-impl InformationSchemaMetrics {
-    pub(super) fn new() -> Self {
-        Self {
-            schema: Self::schema(),
-        }
-    }
-
-    fn schema() -> SchemaRef {
-        Arc::new(Schema::new(vec![
-            ColumnSchema::new(METRIC_NAME, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(METRIC_VALUE, ConcreteDataType::float64_datatype(), false),
-            ColumnSchema::new(METRIC_LABELS, ConcreteDataType::string_datatype(), true),
-            ColumnSchema::new(NODE, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(NODE_TYPE, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(
-                TIMESTAMP,
-                ConcreteDataType::timestamp_millisecond_datatype(),
-                false,
-            ),
-        ]))
-    }
-
-    fn builder(&self) -> InformationSchemaMetricsBuilder {
-        InformationSchemaMetricsBuilder::new(self.schema.clone())
-    }
-}
-
-impl InformationTable for InformationSchemaMetrics {
-    fn table_id(&self) -> TableId {
-        INFORMATION_SCHEMA_RUNTIME_METRICS_TABLE_ID
-    }
-
-    fn table_name(&self) -> &'static str {
-        RUNTIME_METRICS
-    }
-
-    fn schema(&self) -> SchemaRef {
-        self.schema.clone()
-    }
-
-    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
-        let schema = self.schema.arrow_schema().clone();
-        let mut builder = self.builder();
-        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
-            schema,
-            futures::stream::once(async move {
-                builder
-                    .make_metrics(Some(request))
-                    .await
-                    .map(|x| x.into_df_record_batch())
-                    .map_err(Into::into)
-            }),
-        ));
-        Ok(Box::pin(
-            RecordBatchStreamAdapter::try_new(stream)
-                .map_err(BoxedError::new)
-                .context(InternalSnafu)?,
-        ))
-    }
-}
-
-struct InformationSchemaMetricsBuilder {
-    schema: SchemaRef,
-
-    metric_names: StringVectorBuilder,
-    metric_values: Float64VectorBuilder,
-    metric_labels: StringVectorBuilder,
-}
-
-impl InformationSchemaMetricsBuilder {
-    fn new(schema: SchemaRef) -> Self {
-        Self {
-            schema,
-            metric_names: StringVectorBuilder::with_capacity(42),
-            metric_values: Float64VectorBuilder::with_capacity(42),
-            metric_labels: StringVectorBuilder::with_capacity(42),
-        }
-    }
-
-    fn add_metric(&mut self, metric_name: &str, labels: String, metric_value: f64) {
-        self.metric_names.push(Some(metric_name));
-        self.metric_values.push(Some(metric_value));
-        self.metric_labels.push(Some(&labels));
-    }
-
-    async fn make_metrics(&mut self, _request: Option<ScanRequest>) -> Result<RecordBatch> {
-        let metric_families = prometheus::gather();
-
-        let write_request =
-            common_telemetry::metric::convert_metric_to_write_request(metric_families, None, 0);
-
-        for ts in write_request.timeseries {
-            //Safety: always has `__name__` label
-            let metric_name = ts
-                .labels
-                .iter()
-                .find_map(|label| {
-                    if label.name == "__name__" {
-                        Some(label.value.clone())
-                    } else {
-                        None
-                    }
-                })
-                .unwrap();
-
-            self.add_metric(
-                &metric_name,
-                ts.labels
-                    .into_iter()
-                    .filter_map(|label| {
-                        if label.name == "__name__" {
-                            None
-                        } else {
-                            Some(format!("{}={}", label.name, label.value))
-                        }
-                    })
-                    .join(", "),
-                // Safety: always has a sample
-                ts.samples[0].value,
-            );
-        }
-
-        self.finish()
-    }
-
-    fn finish(&mut self) -> Result<RecordBatch> {
-        let rows_num = self.metric_names.len();
-        let unknowns = Arc::new(ConstantVector::new(
-            Arc::new(StringVector::from(vec!["unknown"])),
-            rows_num,
-        ));
-        let timestamps = Arc::new(ConstantVector::new(
-            Arc::new(TimestampMillisecondVector::from_slice([
-                current_time_millis(),
-            ])),
-            rows_num,
-        ));
-
-        let columns: Vec<VectorRef> = vec![
-            Arc::new(self.metric_names.finish()),
-            Arc::new(self.metric_values.finish()),
-            Arc::new(self.metric_labels.finish()),
-            // TODO(dennis): supports node and node_type for cluster
-            unknowns.clone(),
-            unknowns,
-            timestamps,
-        ];
-
-        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
-    }
-}
-
-impl DfPartitionStream for InformationSchemaMetrics {
-    fn schema(&self) -> &ArrowSchemaRef {
-        self.schema.arrow_schema()
-    }
-
-    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
-        let schema = self.schema.arrow_schema().clone();
-        let mut builder = self.builder();
-        Box::pin(DfRecordBatchStreamAdapter::new(
-            schema,
-            futures::stream::once(async move {
-                builder
-                    .make_metrics(None)
-                    .await
-                    .map(|x| x.into_df_record_batch())
-                    .map_err(Into::into)
-            }),
-        ))
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use common_recordbatch::RecordBatches;
-
-    use super::*;
-
-    #[tokio::test]
-    async fn test_make_metrics() {
-        let metrics = InformationSchemaMetrics::new();
-
-        let stream = metrics.to_stream(ScanRequest::default()).unwrap();
-
-        let batches = RecordBatches::try_collect(stream).await.unwrap();
-
-        let result_literal = batches.pretty_print().unwrap();
-
-        assert!(result_literal.contains(METRIC_NAME));
-        assert!(result_literal.contains(METRIC_VALUE));
-        assert!(result_literal.contains(METRIC_LABELS));
-        assert!(result_literal.contains(NODE));
-        assert!(result_literal.contains(NODE_TYPE));
-        assert!(result_literal.contains(TIMESTAMP));
-    }
-}

src/catalog/src/information_schema/schemata.rs

@@ -1,222 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed 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.
-
-use std::sync::{Arc, Weak};
-
-use arrow_schema::SchemaRef as ArrowSchemaRef;
-use common_catalog::consts::INFORMATION_SCHEMA_SCHEMATA_TABLE_ID;
-use common_error::ext::BoxedError;
-use common_query::physical_plan::TaskContext;
-use common_recordbatch::adapter::RecordBatchStreamAdapter;
-use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
-use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
-use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
-use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
-use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
-use datatypes::value::Value;
-use datatypes::vectors::StringVectorBuilder;
-use snafu::{OptionExt, ResultExt};
-use store_api::storage::{ScanRequest, TableId};
-
-use super::SCHEMATA;
-use crate::error::{
-    CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
-};
-use crate::information_schema::{InformationTable, Predicates};
-use crate::CatalogManager;
-
-const CATALOG_NAME: &str = "catalog_name";
-const SCHEMA_NAME: &str = "schema_name";
-const DEFAULT_CHARACTER_SET_NAME: &str = "default_character_set_name";
-const DEFAULT_COLLATION_NAME: &str = "default_collation_name";
-const INIT_CAPACITY: usize = 42;
-
-/// The `information_schema.schemata` table implementation.
-pub(super) struct InformationSchemaSchemata {
-    schema: SchemaRef,
-    catalog_name: String,
-    catalog_manager: Weak<dyn CatalogManager>,
-}
-
-impl InformationSchemaSchemata {
-    pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
-        Self {
-            schema: Self::schema(),
-            catalog_name,
-            catalog_manager,
-        }
-    }
-
-    pub(crate) fn schema() -> SchemaRef {
-        Arc::new(Schema::new(vec![
-            ColumnSchema::new(CATALOG_NAME, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(SCHEMA_NAME, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(
-                DEFAULT_CHARACTER_SET_NAME,
-                ConcreteDataType::string_datatype(),
-                false,
-            ),
-            ColumnSchema::new(
-                DEFAULT_COLLATION_NAME,
-                ConcreteDataType::string_datatype(),
-                false,
-            ),
-            ColumnSchema::new("sql_path", ConcreteDataType::string_datatype(), true),
-        ]))
-    }
-
-    fn builder(&self) -> InformationSchemaSchemataBuilder {
-        InformationSchemaSchemataBuilder::new(
-            self.schema.clone(),
-            self.catalog_name.clone(),
-            self.catalog_manager.clone(),
-        )
-    }
-}
-
-impl InformationTable for InformationSchemaSchemata {
-    fn table_id(&self) -> TableId {
-        INFORMATION_SCHEMA_SCHEMATA_TABLE_ID
-    }
-
-    fn table_name(&self) -> &'static str {
-        SCHEMATA
-    }
-
-    fn schema(&self) -> SchemaRef {
-        self.schema.clone()
-    }
-
-    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
-        let schema = self.schema.arrow_schema().clone();
-        let mut builder = self.builder();
-        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
-            schema,
-            futures::stream::once(async move {
-                builder
-                    .make_schemata(Some(request))
-                    .await
-                    .map(|x| x.into_df_record_batch())
-                    .map_err(Into::into)
-            }),
-        ));
-        Ok(Box::pin(
-            RecordBatchStreamAdapter::try_new(stream)
-                .map_err(BoxedError::new)
-                .context(InternalSnafu)?,
-        ))
-    }
-}
-
-/// Builds the `information_schema.schemata` table row by row
-///
-/// Columns are based on <https://docs.pingcap.com/tidb/stable/information-schema-schemata>
-struct InformationSchemaSchemataBuilder {
-    schema: SchemaRef,
-    catalog_name: String,
-    catalog_manager: Weak<dyn CatalogManager>,
-
-    catalog_names: StringVectorBuilder,
-    schema_names: StringVectorBuilder,
-    charset_names: StringVectorBuilder,
-    collation_names: StringVectorBuilder,
-    sql_paths: StringVectorBuilder,
-}
-
-impl InformationSchemaSchemataBuilder {
-    fn new(
-        schema: SchemaRef,
-        catalog_name: String,
-        catalog_manager: Weak<dyn CatalogManager>,
-    ) -> Self {
-        Self {
-            schema,
-            catalog_name,
-            catalog_manager,
-            catalog_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            schema_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            charset_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            collation_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            sql_paths: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-        }
-    }
-
-    /// Construct the `information_schema.schemata` virtual table
-    async fn make_schemata(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
-        let catalog_name = self.catalog_name.clone();
-        let catalog_manager = self
-            .catalog_manager
-            .upgrade()
-            .context(UpgradeWeakCatalogManagerRefSnafu)?;
-        let predicates = Predicates::from_scan_request(&request);
-
-        for schema_name in catalog_manager.schema_names(&catalog_name).await? {
-            self.add_schema(&predicates, &catalog_name, &schema_name);
-        }
-
-        self.finish()
-    }
-
-    fn add_schema(&mut self, predicates: &Predicates, catalog_name: &str, schema_name: &str) {
-        let row = [
-            (CATALOG_NAME, &Value::from(catalog_name)),
-            (SCHEMA_NAME, &Value::from(schema_name)),
-            (DEFAULT_CHARACTER_SET_NAME, &Value::from("utf8")),
-            (DEFAULT_COLLATION_NAME, &Value::from("utf8_bin")),
-        ];
-
-        if !predicates.eval(&row) {
-            return;
-        }
-
-        self.catalog_names.push(Some(catalog_name));
-        self.schema_names.push(Some(schema_name));
-        self.charset_names.push(Some("utf8"));
-        self.collation_names.push(Some("utf8_bin"));
-        self.sql_paths.push(None);
-    }
-
-    fn finish(&mut self) -> Result<RecordBatch> {
-        let columns: Vec<VectorRef> = vec![
-            Arc::new(self.catalog_names.finish()),
-            Arc::new(self.schema_names.finish()),
-            Arc::new(self.charset_names.finish()),
-            Arc::new(self.collation_names.finish()),
-            Arc::new(self.sql_paths.finish()),
-        ];
-        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
-    }
-}
-
-impl DfPartitionStream for InformationSchemaSchemata {
-    fn schema(&self) -> &ArrowSchemaRef {
-        self.schema.arrow_schema()
-    }
-
-    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
-        let schema = self.schema.arrow_schema().clone();
-        let mut builder = self.builder();
-        Box::pin(DfRecordBatchStreamAdapter::new(
-            schema,
-            futures::stream::once(async move {
-                builder
-                    .make_schemata(None)
-                    .await
-                    .map(|x| x.into_df_record_batch())
-                    .map_err(Into::into)
-            }),
-        ))
-    }
-}

src/catalog/src/information_schema/table_names.rs

@@ -20,24 +20,3 @@ pub const ENGINES: &str = "engines";
 pub const COLUMN_PRIVILEGES: &str = "column_privileges";
 pub const COLUMN_STATISTICS: &str = "column_statistics";
 pub const BUILD_INFO: &str = "build_info";
-pub const CHARACTER_SETS: &str = "character_sets";
-pub const COLLATIONS: &str = "collations";
-pub const COLLATION_CHARACTER_SET_APPLICABILITY: &str = "collation_character_set_applicability";
-pub const CHECK_CONSTRAINTS: &str = "check_constraints";
-pub const EVENTS: &str = "events";
-pub const FILES: &str = "files";
-pub const SCHEMATA: &str = "schemata";
-pub const KEY_COLUMN_USAGE: &str = "key_column_usage";
-pub const OPTIMIZER_TRACE: &str = "optimizer_trace";
-pub const PARAMETERS: &str = "parameters";
-pub const PROFILING: &str = "profiling";
-pub const REFERENTIAL_CONSTRAINTS: &str = "referential_constraints";
-pub const ROUTINES: &str = "routines";
-pub const SCHEMA_PRIVILEGES: &str = "schema_privileges";
-pub const TABLE_PRIVILEGES: &str = "table_privileges";
-pub const TRIGGERS: &str = "triggers";
-pub const GLOBAL_STATUS: &str = "global_status";
-pub const SESSION_STATUS: &str = "session_status";
-pub const RUNTIME_METRICS: &str = "runtime_metrics";
-pub const PARTITIONS: &str = "partitions";
-pub const REGION_PEERS: &str = "greptime_region_peers";

src/catalog/src/information_schema/tables.rs

@@ -25,28 +25,18 @@ use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
 use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
-use datatypes::value::Value;
 use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder};
-use futures::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
-use store_api::storage::{ScanRequest, TableId};
+use store_api::storage::TableId;
 use table::metadata::TableType;
 
 use super::TABLES;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
-use crate::information_schema::{InformationTable, Predicates};
+use crate::information_schema::InformationTable;
 use crate::CatalogManager;
 
-const TABLE_CATALOG: &str = "table_catalog";
-const TABLE_SCHEMA: &str = "table_schema";
-const TABLE_NAME: &str = "table_name";
-const TABLE_TYPE: &str = "table_type";
-const TABLE_ID: &str = "table_id";
-const ENGINE: &str = "engine";
-const INIT_CAPACITY: usize = 42;
-
 pub(super) struct InformationSchemaTables {
     schema: SchemaRef,
     catalog_name: String,
@@ -64,12 +54,12 @@ impl InformationSchemaTables {
 
     pub(crate) fn schema() -> SchemaRef {
         Arc::new(Schema::new(vec![
-            ColumnSchema::new(TABLE_CATALOG, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(TABLE_SCHEMA, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(TABLE_NAME, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(TABLE_TYPE, ConcreteDataType::string_datatype(), false),
-            ColumnSchema::new(TABLE_ID, ConcreteDataType::uint32_datatype(), true),
-            ColumnSchema::new(ENGINE, ConcreteDataType::string_datatype(), true),
+            ColumnSchema::new("table_catalog", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new("table_schema", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new("table_name", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new("table_type", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new("table_id", ConcreteDataType::uint32_datatype(), true),
+            ColumnSchema::new("engine", ConcreteDataType::string_datatype(), true),
         ]))
     }
 
@@ -95,14 +85,14 @@ impl InformationTable for InformationSchemaTables {
         self.schema.clone()
     }
 
-    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
+    fn to_stream(&self) -> Result<SendableRecordBatchStream> {
         let schema = self.schema.arrow_schema().clone();
         let mut builder = self.builder();
         let stream = Box::pin(DfRecordBatchStreamAdapter::new(
             schema,
             futures::stream::once(async move {
                 builder
-                    .make_tables(Some(request))
+                    .make_tables()
                     .await
                     .map(|x| x.into_df_record_batch())
                     .map_err(Into::into)
@@ -142,48 +132,59 @@ impl InformationSchemaTablesBuilder {
             schema,
             catalog_name,
             catalog_manager,
-            catalog_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            schema_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            table_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            table_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
-            engines: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            catalog_names: StringVectorBuilder::with_capacity(42),
+            schema_names: StringVectorBuilder::with_capacity(42),
+            table_names: StringVectorBuilder::with_capacity(42),
+            table_types: StringVectorBuilder::with_capacity(42),
+            table_ids: UInt32VectorBuilder::with_capacity(42),
+            engines: StringVectorBuilder::with_capacity(42),
         }
     }
 
     /// Construct the `information_schema.tables` virtual table
-    async fn make_tables(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
+    async fn make_tables(&mut self) -> Result<RecordBatch> {
         let catalog_name = self.catalog_name.clone();
         let catalog_manager = self
             .catalog_manager
             .upgrade()
             .context(UpgradeWeakCatalogManagerRefSnafu)?;
-        let predicates = Predicates::from_scan_request(&request);
 
         for schema_name in catalog_manager.schema_names(&catalog_name).await? {
-            let mut stream = catalog_manager.tables(&catalog_name, &schema_name).await;
+            if !catalog_manager
+                .schema_exists(&catalog_name, &schema_name)
+                .await?
+            {
+                continue;
+            }
 
-            while let Some(table) = stream.try_next().await? {
-                let table_info = table.table_info();
-                self.add_table(
-                    &predicates,
-                    &catalog_name,
-                    &schema_name,
-                    &table_info.name,
-                    table.table_type(),
-                    Some(table_info.ident.table_id),
-                    Some(&table_info.meta.engine),
-                );
+            for table_name in catalog_manager
+                .table_names(&catalog_name, &schema_name)
+                .await?
+            {
+                if let Some(table) = catalog_manager
+                    .table(&catalog_name, &schema_name, &table_name)
+                    .await?
+                {
+                    let table_info = table.table_info();
+                    self.add_table(
+                        &catalog_name,
+                        &schema_name,
+                        &table_name,
+                        table.table_type(),
+                        Some(table_info.ident.table_id),
+                        Some(&table_info.meta.engine),
+                    );
+                } else {
+                    unreachable!();
+                }
             }
         }
 
         self.finish()
     }
 
-    #[allow(clippy::too_many_arguments)]
     fn add_table(
         &mut self,
-        predicates: &Predicates,
         catalog_name: &str,
         schema_name: &str,
         table_name: &str,
@@ -191,27 +192,14 @@ impl InformationSchemaTablesBuilder {
         table_id: Option<u32>,
         engine: Option<&str>,
     ) {
-        let table_type = match table_type {
-            TableType::Base => "BASE TABLE",
-            TableType::View => "VIEW",
-            TableType::Temporary => "LOCAL TEMPORARY",
-        };
-
-        let row = [
-            (TABLE_CATALOG, &Value::from(catalog_name)),
-            (TABLE_SCHEMA, &Value::from(schema_name)),
-            (TABLE_NAME, &Value::from(table_name)),
-            (TABLE_TYPE, &Value::from(table_type)),
-        ];
-
-        if !predicates.eval(&row) {
-            return;
-        }
-
         self.catalog_names.push(Some(catalog_name));
         self.schema_names.push(Some(schema_name));
         self.table_names.push(Some(table_name));
-        self.table_types.push(Some(table_type));
+        self.table_types.push(Some(match table_type {
+            TableType::Base => "BASE TABLE",
+            TableType::View => "VIEW",
+            TableType::Temporary => "LOCAL TEMPORARY",
+        }));
         self.table_ids.push(table_id);
         self.engines.push(engine);
     }
@@ -241,7 +229,7 @@ impl DfPartitionStream for InformationSchemaTables {
             schema,
             futures::stream::once(async move {
                 builder
-                    .make_tables(None)
+                    .make_tables()
                     .await
                     .map(|x| x.into_df_record_batch())
                     .map_err(Into::into)

src/catalog/src/kvbackend.rs

@@ -12,9 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub use client::{CachedMetaKvBackend, CachedMetaKvBackendBuilder, MetaKvBackend};
+pub use client::{CachedMetaKvBackend, MetaKvBackend};
 
 mod client;
 mod manager;
 
+#[cfg(feature = "testing")]
+pub mod mock;
 pub use manager::KvBackendCatalogManager;

src/catalog/src/kvbackend/client.rs

@@ -14,10 +14,8 @@
 
 use std::any::Any;
 use std::fmt::Debug;
-use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 use std::time::Duration;
-use std::usize;
 
 use common_error::ext::BoxedError;
 use common_meta::cache_invalidator::KvCacheInvalidator;
@@ -35,91 +33,18 @@ use meta_client::client::MetaClient;
 use moka::future::{Cache, CacheBuilder};
 use snafu::{OptionExt, ResultExt};
 
-use crate::metrics::{
-    METRIC_CATALOG_KV_BATCH_GET, METRIC_CATALOG_KV_GET, METRIC_CATALOG_KV_REMOTE_GET,
-};
+use crate::metrics::{METRIC_CATALOG_KV_GET, METRIC_CATALOG_KV_REMOTE_GET};
 
-const DEFAULT_CACHE_MAX_CAPACITY: u64 = 10000;
-const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(10 * 60);
-const DEFAULT_CACHE_TTI: Duration = Duration::from_secs(5 * 60);
-
-pub struct CachedMetaKvBackendBuilder {
-    cache_max_capacity: Option<u64>,
-    cache_ttl: Option<Duration>,
-    cache_tti: Option<Duration>,
-    meta_client: Arc<MetaClient>,
-}
-
-impl CachedMetaKvBackendBuilder {
-    pub fn new(meta_client: Arc<MetaClient>) -> Self {
-        Self {
-            cache_max_capacity: None,
-            cache_ttl: None,
-            cache_tti: None,
-            meta_client,
-        }
-    }
-
-    pub fn cache_max_capacity(mut self, cache_max_capacity: u64) -> Self {
-        self.cache_max_capacity.replace(cache_max_capacity);
-        self
-    }
-
-    pub fn cache_ttl(mut self, cache_ttl: Duration) -> Self {
-        self.cache_ttl.replace(cache_ttl);
-        self
-    }
-
-    pub fn cache_tti(mut self, cache_tti: Duration) -> Self {
-        self.cache_tti.replace(cache_tti);
-        self
-    }
-
-    pub fn build(self) -> CachedMetaKvBackend {
-        let cache_max_capacity = self
-            .cache_max_capacity
-            .unwrap_or(DEFAULT_CACHE_MAX_CAPACITY);
-        let cache_ttl = self.cache_ttl.unwrap_or(DEFAULT_CACHE_TTL);
-        let cache_tti = self.cache_tti.unwrap_or(DEFAULT_CACHE_TTI);
-
-        let cache = Arc::new(
-            CacheBuilder::new(cache_max_capacity)
-                .time_to_live(cache_ttl)
-                .time_to_idle(cache_tti)
-                .build(),
-        );
-
-        let kv_backend = Arc::new(MetaKvBackend {
-            client: self.meta_client,
-        });
-        let name = format!("CachedKvBackend({})", kv_backend.name());
-        let version = AtomicUsize::new(0);
-
-        CachedMetaKvBackend {
-            kv_backend,
-            cache,
-            name,
-            version,
-        }
-    }
-}
+const CACHE_MAX_CAPACITY: u64 = 10000;
+const CACHE_TTL_SECOND: u64 = 10 * 60;
+const CACHE_TTI_SECOND: u64 = 5 * 60;
 
 pub type CacheBackendRef = Arc<Cache<Vec<u8>, KeyValue>>;
 
-/// A wrapper of `MetaKvBackend` with cache support.
-///
-/// CachedMetaKvBackend is mainly used to read metadata information from Metasrv, and provides
-/// cache for get and batch_get. One way to trigger cache invalidation of CachedMetaKvBackend:
-/// when metadata information changes, Metasrv will broadcast a metadata invalidation request.
-///
-/// Therefore, it is recommended to use CachedMetaKvBackend to only read metadata related
-/// information. Note: If you read other information, you may read expired data, which depends on
-/// TTL and TTI for cache.
 pub struct CachedMetaKvBackend {
     kv_backend: KvBackendRef,
     cache: CacheBackendRef,
     name: String,
-    version: AtomicUsize,
 }
 
 impl TxnService for CachedMetaKvBackend {
@@ -171,38 +96,7 @@ impl KvBackend for CachedMetaKvBackend {
     }
 
     async fn batch_get(&self, req: BatchGetRequest) -> Result<BatchGetResponse> {
-        let _timer = METRIC_CATALOG_KV_BATCH_GET.start_timer();
-
-        let mut kvs = Vec::with_capacity(req.keys.len());
-        let mut miss_keys = Vec::with_capacity(req.keys.len());
-
-        for key in req.keys {
-            if let Some(val) = self.cache.get(&key).await {
-                kvs.push(val);
-            } else {
-                miss_keys.push(key);
-            }
-        }
-
-        let batch_get_req = BatchGetRequest::new().with_keys(miss_keys.clone());
-
-        let pre_version = self.version();
-
-        let unhit_kvs = self.kv_backend.batch_get(batch_get_req).await?.kvs;
-
-        for kv in unhit_kvs.iter() {
-            self.cache.insert(kv.key().to_vec(), kv.clone()).await;
-        }
-
-        if !self.validate_version(pre_version) {
-            for key in miss_keys.iter() {
-                self.cache.invalidate(key).await;
-            }
-        }
-
-        kvs.extend(unhit_kvs);
-
-        Ok(BatchGetResponse { kvs })
+        self.kv_backend.batch_get(req).await
     }
 
     async fn compare_and_put(&self, req: CompareAndPutRequest) -> Result<CompareAndPutResponse> {
@@ -260,14 +154,8 @@ impl KvBackend for CachedMetaKvBackend {
     async fn get(&self, key: &[u8]) -> Result<Option<KeyValue>> {
         let _timer = METRIC_CATALOG_KV_GET.start_timer();
 
-        let pre_version = Arc::new(Mutex::new(None));
-
         let init = async {
-            let version_clone = pre_version.clone();
             let _timer = METRIC_CATALOG_KV_REMOTE_GET.start_timer();
-
-            version_clone.lock().unwrap().replace(self.version());
-
             self.kv_backend.get(key).await.map(|val| {
                 val.with_context(|| CacheNotGetSnafu {
                     key: String::from_utf8_lossy(key),
@@ -278,7 +166,7 @@ impl KvBackend for CachedMetaKvBackend {
         // currently moka doesn't have `optionally_try_get_with_by_ref`
         // TODO(fys): change to moka method when available
         // https://github.com/moka-rs/moka/issues/254
-        let ret = match self.cache.try_get_with_by_ref(key, init).await {
+        match self.cache.try_get_with_by_ref(key, init).await {
             Ok(val) => Ok(Some(val)),
             Err(e) => match e.as_ref() {
                 CacheNotGet { .. } => Ok(None),
@@ -287,40 +175,29 @@ impl KvBackend for CachedMetaKvBackend {
         }
         .map_err(|e| GetKvCache {
             err_msg: e.to_string(),
-        });
-
-        // "cache.invalidate_key" and "cache.try_get_with_by_ref" are not mutually exclusive. So we need
-        // to use the version mechanism to prevent expired data from being put into the cache.
-        if pre_version
-            .lock()
-            .unwrap()
-            .as_ref()
-            .map_or(false, |v| !self.validate_version(*v))
-        {
-            self.cache.invalidate(key).await;
-        }
-
-        ret
+        })
     }
 }
 
 #[async_trait::async_trait]
 impl KvCacheInvalidator for CachedMetaKvBackend {
     async fn invalidate_key(&self, key: &[u8]) {
-        self.create_new_version();
         self.cache.invalidate(key).await;
         debug!("invalidated cache key: {}", String::from_utf8_lossy(key));
     }
 }
 
 impl CachedMetaKvBackend {
-    // only for test
-    #[cfg(test)]
-    fn wrap(kv_backend: KvBackendRef) -> Self {
+    pub fn new(client: Arc<MetaClient>) -> Self {
+        let kv_backend = Arc::new(MetaKvBackend { client });
+        Self::wrap(kv_backend)
+    }
+
+    pub fn wrap(kv_backend: KvBackendRef) -> Self {
         let cache = Arc::new(
-            CacheBuilder::new(DEFAULT_CACHE_MAX_CAPACITY)
-                .time_to_live(DEFAULT_CACHE_TTL)
-                .time_to_idle(DEFAULT_CACHE_TTI)
+            CacheBuilder::new(CACHE_MAX_CAPACITY)
+                .time_to_live(Duration::from_secs(CACHE_TTL_SECOND))
+                .time_to_idle(Duration::from_secs(CACHE_TTI_SECOND))
                 .build(),
         );
 
@@ -329,25 +206,12 @@ impl CachedMetaKvBackend {
             kv_backend,
             cache,
             name,
-            version: AtomicUsize::new(0),
         }
     }
 
     pub fn cache(&self) -> &CacheBackendRef {
         &self.cache
     }
-
-    fn version(&self) -> usize {
-        self.version.load(Ordering::Relaxed)
-    }
-
-    fn validate_version(&self, pre_version: usize) -> bool {
-        self.version() == pre_version
-    }
-
-    fn create_new_version(&self) -> usize {
-        self.version.fetch_add(1, Ordering::Relaxed) + 1
-    }
 }
 
 #[derive(Debug)]
@@ -444,162 +308,3 @@ impl KvBackend for MetaKvBackend {
         self
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::any::Any;
-    use std::sync::atomic::{AtomicU32, Ordering};
-    use std::sync::Arc;
-
-    use async_trait::async_trait;
-    use common_meta::kv_backend::{KvBackend, TxnService};
-    use common_meta::rpc::store::{
-        BatchDeleteRequest, BatchDeleteResponse, BatchGetRequest, BatchGetResponse,
-        BatchPutRequest, BatchPutResponse, CompareAndPutRequest, CompareAndPutResponse,
-        DeleteRangeRequest, DeleteRangeResponse, PutRequest, PutResponse, RangeRequest,
-        RangeResponse,
-    };
-    use common_meta::rpc::KeyValue;
-    use dashmap::DashMap;
-
-    use super::CachedMetaKvBackend;
-
-    #[derive(Default)]
-    pub struct SimpleKvBackend {
-        inner_map: DashMap<Vec<u8>, Vec<u8>>,
-        get_execute_times: Arc<AtomicU32>,
-    }
-
-    impl TxnService for SimpleKvBackend {
-        type Error = common_meta::error::Error;
-    }
-
-    #[async_trait]
-    impl KvBackend for SimpleKvBackend {
-        fn name(&self) -> &str {
-            "SimpleKvBackend"
-        }
-
-        fn as_any(&self) -> &dyn Any {
-            self
-        }
-
-        async fn batch_get(&self, req: BatchGetRequest) -> Result<BatchGetResponse, Self::Error> {
-            let mut kvs = Vec::with_capacity(req.keys.len());
-            for key in req.keys.iter() {
-                if let Some(kv) = self.get(key).await? {
-                    kvs.push(kv);
-                }
-            }
-            Ok(BatchGetResponse { kvs })
-        }
-
-        async fn put(&self, req: PutRequest) -> Result<PutResponse, Self::Error> {
-            self.inner_map.insert(req.key, req.value);
-            // always return None as prev_kv, since we don't use it in this test.
-            Ok(PutResponse { prev_kv: None })
-        }
-
-        async fn get(&self, key: &[u8]) -> Result<Option<KeyValue>, Self::Error> {
-            self.get_execute_times
-                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-            Ok(self.inner_map.get(key).map(|v| KeyValue {
-                key: key.to_vec(),
-                value: v.value().clone(),
-            }))
-        }
-
-        async fn range(&self, _req: RangeRequest) -> Result<RangeResponse, Self::Error> {
-            todo!()
-        }
-
-        async fn batch_put(&self, _req: BatchPutRequest) -> Result<BatchPutResponse, Self::Error> {
-            todo!()
-        }
-
-        async fn compare_and_put(
-            &self,
-            _req: CompareAndPutRequest,
-        ) -> Result<CompareAndPutResponse, Self::Error> {
-            todo!()
-        }
-
-        async fn delete_range(
-            &self,
-            _req: DeleteRangeRequest,
-        ) -> Result<DeleteRangeResponse, Self::Error> {
-            todo!()
-        }
-
-        async fn batch_delete(
-            &self,
-            _req: BatchDeleteRequest,
-        ) -> Result<BatchDeleteResponse, Self::Error> {
-            todo!()
-        }
-    }
-
-    #[tokio::test]
-    async fn test_cached_kv_backend() {
-        let simple_kv = Arc::new(SimpleKvBackend::default());
-        let get_execute_times = simple_kv.get_execute_times.clone();
-        let cached_kv = CachedMetaKvBackend::wrap(simple_kv);
-
-        add_some_vals(&cached_kv).await;
-
-        let batch_get_req = BatchGetRequest {
-            keys: vec![b"k1".to_vec(), b"k2".to_vec()],
-        };
-
-        assert_eq!(get_execute_times.load(Ordering::SeqCst), 0);
-
-        for _ in 0..10 {
-            let _batch_get_resp = cached_kv.batch_get(batch_get_req.clone()).await.unwrap();
-
-            assert_eq!(get_execute_times.load(Ordering::SeqCst), 2);
-        }
-
-        let batch_get_req = BatchGetRequest {
-            keys: vec![b"k1".to_vec(), b"k2".to_vec(), b"k3".to_vec()],
-        };
-
-        let _batch_get_resp = cached_kv.batch_get(batch_get_req.clone()).await.unwrap();
-
-        assert_eq!(get_execute_times.load(Ordering::SeqCst), 3);
-
-        for _ in 0..10 {
-            let _batch_get_resp = cached_kv.batch_get(batch_get_req.clone()).await.unwrap();
-
-            assert_eq!(get_execute_times.load(Ordering::SeqCst), 3);
-        }
-    }
-
-    async fn add_some_vals(kv_backend: &impl KvBackend) {
-        kv_backend
-            .put(PutRequest {
-                key: b"k1".to_vec(),
-                value: b"v1".to_vec(),
-                prev_kv: false,
-            })
-            .await
-            .unwrap();
-
-        kv_backend
-            .put(PutRequest {
-                key: b"k2".to_vec(),
-                value: b"v2".to_vec(),
-                prev_kv: false,
-            })
-            .await
-            .unwrap();
-
-        kv_backend
-            .put(PutRequest {
-                key: b"k3".to_vec(),
-                value: b"v3".to_vec(),
-                prev_kv: false,
-            })
-            .await
-            .unwrap();
-    }
-}

src/catalog/src/kvbackend/manager.rs

@@ -16,21 +16,17 @@ use std::any::Any;
 use std::collections::BTreeSet;
 use std::sync::{Arc, Weak};
 
-use async_stream::try_stream;
 use common_catalog::consts::{DEFAULT_SCHEMA_NAME, INFORMATION_SCHEMA_NAME, NUMBERS_TABLE_ID};
 use common_error::ext::BoxedError;
 use common_meta::cache_invalidator::{CacheInvalidator, CacheInvalidatorRef, Context};
 use common_meta::error::Result as MetaResult;
 use common_meta::key::catalog_name::CatalogNameKey;
 use common_meta::key::schema_name::SchemaNameKey;
-use common_meta::key::table_info::TableInfoValue;
 use common_meta::key::table_name::TableNameKey;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::table_name::TableName;
-use futures_util::stream::BoxStream;
-use futures_util::{StreamExt, TryStreamExt};
-use moka::sync::Cache;
+use futures_util::TryStreamExt;
 use partition::manager::{PartitionRuleManager, PartitionRuleManagerRef};
 use snafu::prelude::*;
 use table::dist_table::DistTable;
@@ -39,8 +35,8 @@ use table::table::numbers::{NumbersTable, NUMBERS_TABLE_NAME};
 use table::TableRef;
 
 use crate::error::{
-    self as catalog_err, ListCatalogsSnafu, ListSchemasSnafu, ListTablesSnafu,
-    Result as CatalogResult, TableMetadataManagerSnafu,
+    self as catalog_err, ListCatalogsSnafu, ListSchemasSnafu, Result as CatalogResult,
+    TableMetadataManagerSnafu,
 };
 use crate::information_schema::InformationSchemaProvider;
 use crate::CatalogManager;
@@ -62,30 +58,20 @@ pub struct KvBackendCatalogManager {
     system_catalog: SystemCatalog,
 }
 
-fn make_table(table_info_value: TableInfoValue) -> CatalogResult<TableRef> {
-    let table_info = table_info_value
-        .table_info
-        .try_into()
-        .context(catalog_err::InvalidTableInfoInCatalogSnafu)?;
-    Ok(DistTable::table(Arc::new(table_info)))
-}
-
 #[async_trait::async_trait]
 impl CacheInvalidator for KvBackendCatalogManager {
-    async fn invalidate_table_id(&self, ctx: &Context, table_id: TableId) -> MetaResult<()> {
-        self.cache_invalidator
-            .invalidate_table_id(ctx, table_id)
-            .await
-    }
-
     async fn invalidate_table_name(&self, ctx: &Context, table_name: TableName) -> MetaResult<()> {
         self.cache_invalidator
             .invalidate_table_name(ctx, table_name)
             .await
     }
-}
 
-const DEFAULT_CACHED_CATALOG: u64 = 128;
+    async fn invalidate_table_id(&self, ctx: &Context, table_id: TableId) -> MetaResult<()> {
+        self.cache_invalidator
+            .invalidate_table_id(ctx, table_id)
+            .await
+    }
+}
 
 impl KvBackendCatalogManager {
     pub fn new(backend: KvBackendRef, cache_invalidator: CacheInvalidatorRef) -> Arc<Self> {
@@ -95,10 +81,9 @@ impl KvBackendCatalogManager {
             cache_invalidator,
             system_catalog: SystemCatalog {
                 catalog_manager: me.clone(),
-                catalog_cache: Cache::new(DEFAULT_CACHED_CATALOG),
                 information_schema_provider: Arc::new(InformationSchemaProvider::new(
                     // The catalog name is not used in system_catalog, so let it empty
-                    String::default(),
+                    "".to_string(),
                     me.clone(),
                 )),
             },
@@ -116,10 +101,6 @@ impl KvBackendCatalogManager {
 
 #[async_trait::async_trait]
 impl CatalogManager for KvBackendCatalogManager {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     async fn catalog_names(&self) -> CatalogResult<Vec<String>> {
         let stream = self
             .table_metadata_manager
@@ -154,22 +135,18 @@ impl CatalogManager for KvBackendCatalogManager {
     }
 
     async fn table_names(&self, catalog: &str, schema: &str) -> CatalogResult<Vec<String>> {
-        let stream = self
+        let mut tables = self
             .table_metadata_manager
             .table_name_manager()
             .tables(catalog, schema)
-            .await;
-        let mut tables = stream
-            .try_collect::<Vec<_>>()
             .await
-            .map_err(BoxedError::new)
-            .context(ListTablesSnafu { catalog, schema })?
+            .context(TableMetadataManagerSnafu)?
             .into_iter()
             .map(|(k, _)| k)
-            .collect::<Vec<_>>();
+            .collect::<Vec<String>>();
         tables.extend_from_slice(&self.system_catalog.table_names(schema));
 
-        Ok(tables.into_iter().collect())
+        Ok(tables)
     }
 
     async fn catalog_exists(&self, catalog: &str) -> CatalogResult<bool> {
@@ -238,56 +215,17 @@ impl CatalogManager for KvBackendCatalogManager {
         else {
             return Ok(None);
         };
-        make_table(table_info_value).map(Some)
+        let table_info = Arc::new(
+            table_info_value
+                .table_info
+                .try_into()
+                .context(catalog_err::InvalidTableInfoInCatalogSnafu)?,
+        );
+        Ok(Some(DistTable::table(table_info)))
     }
 
-    async fn tables<'a>(
-        &'a self,
-        catalog: &'a str,
-        schema: &'a str,
-    ) -> BoxStream<'a, CatalogResult<TableRef>> {
-        let sys_tables = try_stream!({
-            // System tables
-            let sys_table_names = self.system_catalog.table_names(schema);
-            for table_name in sys_table_names {
-                if let Some(table) = self.system_catalog.table(catalog, schema, &table_name) {
-                    yield table;
-                }
-            }
-        });
-
-        let table_id_stream = self
-            .table_metadata_manager
-            .table_name_manager()
-            .tables(catalog, schema)
-            .await
-            .map_ok(|(_, v)| v.table_id());
-        const BATCH_SIZE: usize = 128;
-        let user_tables = try_stream!({
-            // Split table ids into chunks
-            let mut table_id_chunks = table_id_stream.ready_chunks(BATCH_SIZE);
-
-            while let Some(table_ids) = table_id_chunks.next().await {
-                let table_ids = table_ids
-                    .into_iter()
-                    .collect::<Result<Vec<_>, _>>()
-                    .map_err(BoxedError::new)
-                    .context(ListTablesSnafu { catalog, schema })?;
-
-                let table_info_values = self
-                    .table_metadata_manager
-                    .table_info_manager()
-                    .batch_get(&table_ids)
-                    .await
-                    .context(TableMetadataManagerSnafu)?;
-
-                for table_info_value in table_info_values.into_values() {
-                    yield make_table(table_info_value)?;
-                }
-            }
-        });
-
-        Box::pin(sys_tables.chain(user_tables))
+    fn as_any(&self) -> &dyn Any {
+        self
     }
 }
 
@@ -300,7 +238,6 @@ impl CatalogManager for KvBackendCatalogManager {
 #[derive(Clone)]
 struct SystemCatalog {
     catalog_manager: Weak<KvBackendCatalogManager>,
-    catalog_cache: Cache<String, Arc<InformationSchemaProvider>>,
     information_schema_provider: Arc<InformationSchemaProvider>,
 }
 
@@ -336,12 +273,7 @@ impl SystemCatalog {
     fn table(&self, catalog: &str, schema: &str, table_name: &str) -> Option<TableRef> {
         if schema == INFORMATION_SCHEMA_NAME {
             let information_schema_provider =
-                self.catalog_cache.get_with_by_ref(catalog, move || {
-                    Arc::new(InformationSchemaProvider::new(
-                        catalog.to_string(),
-                        self.catalog_manager.clone(),
-                    ))
-                });
+                InformationSchemaProvider::new(catalog.to_string(), self.catalog_manager.clone());
             information_schema_provider.table(table_name)
         } else if schema == DEFAULT_SCHEMA_NAME && table_name == NUMBERS_TABLE_NAME {
             Some(NumbersTable::table(NUMBERS_TABLE_ID))

src/catalog/src/kvbackend/mock.rs

@@ -0,0 +1,128 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed 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.
+
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock as StdRwLock};
+
+use common_recordbatch::RecordBatch;
+use datatypes::data_type::ConcreteDataType;
+use datatypes::schema::{ColumnSchema, Schema};
+use datatypes::vectors::StringVector;
+use table::engine::{CloseTableResult, EngineContext, TableEngine};
+use table::metadata::TableId;
+use table::requests::{
+    AlterTableRequest, CloseTableRequest, CreateTableRequest, DropTableRequest, OpenTableRequest,
+    TruncateTableRequest,
+};
+use table::test_util::MemTable;
+use table::TableRef;
+
+#[derive(Default)]
+pub struct MockTableEngine {
+    tables: StdRwLock<HashMap<TableId, TableRef>>,
+}
+
+#[async_trait::async_trait]
+impl TableEngine for MockTableEngine {
+    fn name(&self) -> &str {
+        "MockTableEngine"
+    }
+
+    /// Create a table with only one column
+    async fn create_table(
+        &self,
+        _ctx: &EngineContext,
+        request: CreateTableRequest,
+    ) -> table::Result<TableRef> {
+        let table_id = request.id;
+
+        let schema = Arc::new(Schema::new(vec![ColumnSchema::new(
+            "name",
+            ConcreteDataType::string_datatype(),
+            true,
+        )]));
+
+        let data = vec![Arc::new(StringVector::from(vec!["a", "b", "c"])) as _];
+        let record_batch = RecordBatch::new(schema, data).unwrap();
+        let table = MemTable::new_with_catalog(
+            &request.table_name,
+            record_batch,
+            table_id,
+            request.catalog_name,
+            request.schema_name,
+            vec![0],
+        );
+
+        let mut tables = self.tables.write().unwrap();
+        let _ = tables.insert(table_id, table.clone() as TableRef);
+        Ok(table)
+    }
+
+    async fn open_table(
+        &self,
+        _ctx: &EngineContext,
+        request: OpenTableRequest,
+    ) -> table::Result<Option<TableRef>> {
+        Ok(self.tables.read().unwrap().get(&request.table_id).cloned())
+    }
+
+    async fn alter_table(
+        &self,
+        _ctx: &EngineContext,
+        _request: AlterTableRequest,
+    ) -> table::Result<TableRef> {
+        unimplemented!()
+    }
+
+    fn get_table(
+        &self,
+        _ctx: &EngineContext,
+        table_id: TableId,
+    ) -> table::Result<Option<TableRef>> {
+        Ok(self.tables.read().unwrap().get(&table_id).cloned())
+    }
+
+    fn table_exists(&self, _ctx: &EngineContext, table_id: TableId) -> bool {
+        self.tables.read().unwrap().contains_key(&table_id)
+    }
+
+    async fn drop_table(
+        &self,
+        _ctx: &EngineContext,
+        _request: DropTableRequest,
+    ) -> table::Result<bool> {
+        unimplemented!()
+    }
+
+    async fn close_table(
+        &self,
+        _ctx: &EngineContext,
+        request: CloseTableRequest,
+    ) -> table::Result<CloseTableResult> {
+        let _ = self.tables.write().unwrap().remove(&request.table_id);
+        Ok(CloseTableResult::Released(vec![]))
+    }
+
+    async fn close(&self) -> table::Result<()> {
+        Ok(())
+    }
+
+    async fn truncate_table(
+        &self,
+        _ctx: &EngineContext,
+        _request: TruncateTableRequest,
+    ) -> table::Result<bool> {
+        Ok(true)
+    }
+}

src/catalog/src/lib.rs

@@ -20,7 +20,6 @@ use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
 use futures::future::BoxFuture;
-use futures_util::stream::BoxStream;
 use table::metadata::TableId;
 use table::requests::CreateTableRequest;
 use table::TableRef;
@@ -57,13 +56,6 @@ pub trait CatalogManager: Send + Sync {
         schema: &str,
         table_name: &str,
     ) -> Result<Option<TableRef>>;
-
-    /// Returns all tables with a stream by catalog and schema.
-    async fn tables<'a>(
-        &'a self,
-        catalog: &'a str,
-        schema: &'a str,
-    ) -> BoxStream<'a, Result<TableRef>>;
 }
 
 pub type CatalogManagerRef = Arc<dyn CatalogManager>;

src/catalog/src/memory/manager.rs

@@ -17,12 +17,10 @@ use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::sync::{Arc, RwLock, Weak};
 
-use async_stream::{stream, try_stream};
 use common_catalog::build_db_string;
 use common_catalog::consts::{
     DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME, DEFAULT_SCHEMA_NAME, INFORMATION_SCHEMA_NAME,
 };
-use futures_util::stream::BoxStream;
 use snafu::OptionExt;
 use table::TableRef;
 
@@ -41,64 +39,10 @@ pub struct MemoryCatalogManager {
 
 #[async_trait::async_trait]
 impl CatalogManager for MemoryCatalogManager {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
-    async fn catalog_names(&self) -> Result<Vec<String>> {
-        Ok(self.catalogs.read().unwrap().keys().cloned().collect())
-    }
-
-    async fn schema_names(&self, catalog: &str) -> Result<Vec<String>> {
-        Ok(self
-            .catalogs
-            .read()
-            .unwrap()
-            .get(catalog)
-            .with_context(|| CatalogNotFoundSnafu {
-                catalog_name: catalog,
-            })?
-            .keys()
-            .cloned()
-            .collect())
-    }
-
-    async fn table_names(&self, catalog: &str, schema: &str) -> Result<Vec<String>> {
-        Ok(self
-            .catalogs
-            .read()
-            .unwrap()
-            .get(catalog)
-            .with_context(|| CatalogNotFoundSnafu {
-                catalog_name: catalog,
-            })?
-            .get(schema)
-            .with_context(|| SchemaNotFoundSnafu { catalog, schema })?
-            .keys()
-            .cloned()
-            .collect())
-    }
-
-    async fn catalog_exists(&self, catalog: &str) -> Result<bool> {
-        self.catalog_exist_sync(catalog)
-    }
-
     async fn schema_exists(&self, catalog: &str, schema: &str) -> Result<bool> {
         self.schema_exist_sync(catalog, schema)
     }
 
-    async fn table_exists(&self, catalog: &str, schema: &str, table: &str) -> Result<bool> {
-        let catalogs = self.catalogs.read().unwrap();
-        Ok(catalogs
-            .get(catalog)
-            .with_context(|| CatalogNotFoundSnafu {
-                catalog_name: catalog,
-            })?
-            .get(schema)
-            .with_context(|| SchemaNotFoundSnafu { catalog, schema })?
-            .contains_key(table))
-    }
-
     async fn table(
         &self,
         catalog: &str,
@@ -117,35 +61,57 @@ impl CatalogManager for MemoryCatalogManager {
         Ok(result)
     }
 
-    async fn tables<'a>(
-        &'a self,
-        catalog: &'a str,
-        schema: &'a str,
-    ) -> BoxStream<'a, Result<TableRef>> {
+    async fn catalog_exists(&self, catalog: &str) -> Result<bool> {
+        self.catalog_exist_sync(catalog)
+    }
+
+    async fn table_exists(&self, catalog: &str, schema: &str, table: &str) -> Result<bool> {
         let catalogs = self.catalogs.read().unwrap();
+        Ok(catalogs
+            .get(catalog)
+            .with_context(|| CatalogNotFoundSnafu {
+                catalog_name: catalog,
+            })?
+            .get(schema)
+            .with_context(|| SchemaNotFoundSnafu { catalog, schema })?
+            .contains_key(table))
+    }
 
-        let Some(schemas) = catalogs.get(catalog) else {
-            return Box::pin(stream!({
-                yield CatalogNotFoundSnafu {
-                    catalog_name: catalog,
-                }
-                .fail();
-            }));
-        };
+    async fn catalog_names(&self) -> Result<Vec<String>> {
+        Ok(self.catalogs.read().unwrap().keys().cloned().collect())
+    }
 
-        let Some(tables) = schemas.get(schema) else {
-            return Box::pin(stream!({
-                yield SchemaNotFoundSnafu { catalog, schema }.fail();
-            }));
-        };
+    async fn schema_names(&self, catalog_name: &str) -> Result<Vec<String>> {
+        Ok(self
+            .catalogs
+            .read()
+            .unwrap()
+            .get(catalog_name)
+            .with_context(|| CatalogNotFoundSnafu { catalog_name })?
+            .keys()
+            .cloned()
+            .collect())
+    }
 
-        let tables = tables.values().cloned().collect::<Vec<_>>();
+    async fn table_names(&self, catalog_name: &str, schema_name: &str) -> Result<Vec<String>> {
+        Ok(self
+            .catalogs
+            .read()
+            .unwrap()
+            .get(catalog_name)
+            .with_context(|| CatalogNotFoundSnafu { catalog_name })?
+            .get(schema_name)
+            .with_context(|| SchemaNotFoundSnafu {
+                catalog: catalog_name,
+                schema: schema_name,
+            })?
+            .keys()
+            .cloned()
+            .collect())
+    }
 
-        return Box::pin(try_stream!({
-            for table in tables {
-                yield table;
-            }
-        }));
+    fn as_any(&self) -> &dyn Any {
+        self
     }
 }
 
@@ -341,7 +307,6 @@ pub fn new_memory_catalog_manager() -> Result<Arc<MemoryCatalogManager>> {
 #[cfg(test)]
 mod tests {
     use common_catalog::consts::*;
-    use futures_util::TryStreamExt;
     use table::table::numbers::{NumbersTable, NUMBERS_TABLE_NAME};
 
     use super::*;
@@ -366,18 +331,8 @@ mod tests {
                 NUMBERS_TABLE_NAME,
             )
             .await
-            .unwrap()
             .unwrap();
-        let stream = catalog_list
-            .tables(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME)
-            .await;
-        let tables = stream.try_collect::<Vec<_>>().await.unwrap();
-        assert_eq!(tables.len(), 1);
-        assert_eq!(
-            table.table_info().table_id(),
-            tables[0].table_info().table_id()
-        );
-
+        let _ = table.unwrap();
         assert!(catalog_list
             .table(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "not_exists")
             .await

src/catalog/src/metrics.rs

@@ -19,19 +19,17 @@ use prometheus::*;
 
 lazy_static! {
     pub static ref METRIC_CATALOG_MANAGER_CATALOG_COUNT: IntGauge =
-        register_int_gauge!("greptime_catalog_catalog_count", "catalog catalog count").unwrap();
+        register_int_gauge!("catalog_catalog_count", "catalog catalog count").unwrap();
     pub static ref METRIC_CATALOG_MANAGER_SCHEMA_COUNT: IntGauge =
-        register_int_gauge!("greptime_catalog_schema_count", "catalog schema count").unwrap();
+        register_int_gauge!("catalog_schema_count", "catalog schema count").unwrap();
     pub static ref METRIC_CATALOG_MANAGER_TABLE_COUNT: IntGaugeVec = register_int_gauge_vec!(
-        "greptime_catalog_table_count",
+        "catalog_table_count",
         "catalog table count",
         &[METRIC_DB_LABEL]
     )
     .unwrap();
     pub static ref METRIC_CATALOG_KV_REMOTE_GET: Histogram =
-        register_histogram!("greptime_catalog_kv_get_remote", "catalog kv get remote").unwrap();
+        register_histogram!("catalog_kv_get_remote", "catalog kv get remote").unwrap();
     pub static ref METRIC_CATALOG_KV_GET: Histogram =
-        register_histogram!("greptime_catalog_kv_get", "catalog kv get").unwrap();
-    pub static ref METRIC_CATALOG_KV_BATCH_GET: Histogram =
-        register_histogram!("greptime_catalog_kv_batch_get", "catalog kv batch get").unwrap();
+        register_histogram!("catalog_kv_get", "catalog kv get").unwrap();
 }

src/catalog/src/table_source.rs

@@ -15,6 +15,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use common_catalog::consts::INFORMATION_SCHEMA_NAME;
 use common_catalog::format_full_table_name;
 use datafusion::common::{ResolvedTableReference, TableReference};
 use datafusion::datasource::provider_as_source;
@@ -29,7 +30,7 @@ use crate::CatalogManagerRef;
 pub struct DfTableSourceProvider {
     catalog_manager: CatalogManagerRef,
     resolved_tables: HashMap<String, Arc<dyn TableSource>>,
-    disallow_cross_catalog_query: bool,
+    disallow_cross_schema_query: bool,
     default_catalog: String,
     default_schema: String,
 }
@@ -37,12 +38,12 @@ pub struct DfTableSourceProvider {
 impl DfTableSourceProvider {
     pub fn new(
         catalog_manager: CatalogManagerRef,
-        disallow_cross_catalog_query: bool,
+        disallow_cross_schema_query: bool,
         query_ctx: &QueryContext,
     ) -> Self {
         Self {
             catalog_manager,
-            disallow_cross_catalog_query,
+            disallow_cross_schema_query,
             resolved_tables: HashMap::new(),
             default_catalog: query_ctx.current_catalog().to_owned(),
             default_schema: query_ctx.current_schema().to_owned(),
@@ -53,18 +54,29 @@ impl DfTableSourceProvider {
         &'a self,
         table_ref: TableReference<'a>,
     ) -> Result<ResolvedTableReference<'a>> {
-        if self.disallow_cross_catalog_query {
+        if self.disallow_cross_schema_query {
             match &table_ref {
                 TableReference::Bare { .. } => (),
-                TableReference::Partial { .. } => {}
+                TableReference::Partial { schema, .. } => {
+                    ensure!(
+                        schema.as_ref() == self.default_schema
+                            || schema.as_ref() == INFORMATION_SCHEMA_NAME,
+                        QueryAccessDeniedSnafu {
+                            catalog: &self.default_catalog,
+                            schema: schema.as_ref(),
+                        }
+                    );
+                }
                 TableReference::Full {
                     catalog, schema, ..
                 } => {
                     ensure!(
-                        catalog.as_ref() == self.default_catalog,
+                        catalog.as_ref() == self.default_catalog
+                            && (schema.as_ref() == self.default_schema
+                                || schema.as_ref() == INFORMATION_SCHEMA_NAME),
                         QueryAccessDeniedSnafu {
                             catalog: catalog.as_ref(),
-                            schema: schema.as_ref(),
+                            schema: schema.as_ref()
                         }
                     );
                 }
@@ -124,21 +136,21 @@ mod tests {
             table: Cow::Borrowed("table_name"),
         };
         let result = table_provider.resolve_table_ref(table_ref);
-        assert!(result.is_ok());
+        let _ = result.unwrap();
 
         let table_ref = TableReference::Partial {
             schema: Cow::Borrowed("public"),
             table: Cow::Borrowed("table_name"),
         };
         let result = table_provider.resolve_table_ref(table_ref);
-        assert!(result.is_ok());
+        let _ = result.unwrap();
 
         let table_ref = TableReference::Partial {
             schema: Cow::Borrowed("wrong_schema"),
             table: Cow::Borrowed("table_name"),
         };
         let result = table_provider.resolve_table_ref(table_ref);
-        assert!(result.is_ok());
+        assert!(result.is_err());
 
         let table_ref = TableReference::Full {
             catalog: Cow::Borrowed("greptime"),
@@ -146,7 +158,7 @@ mod tests {
             table: Cow::Borrowed("table_name"),
         };
         let result = table_provider.resolve_table_ref(table_ref);
-        assert!(result.is_ok());
+        let _ = result.unwrap();
 
         let table_ref = TableReference::Full {
             catalog: Cow::Borrowed("wrong_catalog"),
@@ -160,15 +172,14 @@ mod tests {
             schema: Cow::Borrowed("information_schema"),
             table: Cow::Borrowed("columns"),
         };
-        let result = table_provider.resolve_table_ref(table_ref);
-        assert!(result.is_ok());
+        let _ = table_provider.resolve_table_ref(table_ref).unwrap();
 
         let table_ref = TableReference::Full {
             catalog: Cow::Borrowed("greptime"),
             schema: Cow::Borrowed("information_schema"),
             table: Cow::Borrowed("columns"),
         };
-        assert!(table_provider.resolve_table_ref(table_ref).is_ok());
+        let _ = table_provider.resolve_table_ref(table_ref).unwrap();
 
         let table_ref = TableReference::Full {
             catalog: Cow::Borrowed("dummy"),
@@ -176,12 +187,5 @@ mod tests {
             table: Cow::Borrowed("columns"),
         };
         assert!(table_provider.resolve_table_ref(table_ref).is_err());
-
-        let table_ref = TableReference::Full {
-            catalog: Cow::Borrowed("greptime"),
-            schema: Cow::Borrowed("greptime_private"),
-            table: Cow::Borrowed("columns"),
-        };
-        assert!(table_provider.resolve_table_ref(table_ref).is_ok());
     }
 }

src/client/Cargo.toml

@@ -9,7 +9,6 @@ testing = []
 
 [dependencies]
 api.workspace = true
-arc-swap = "1.6"
 arrow-flight.workspace = true
 async-stream.workspace = true
 async-trait.workspace = true
@@ -36,8 +35,8 @@ prost.workspace = true
 rand.workspace = true
 session.workspace = true
 snafu.workspace = true
-tokio.workspace = true
 tokio-stream = { workspace = true, features = ["net"] }
+tokio.workspace = true
 tonic.workspace = true
 
 [dev-dependencies]

src/client/examples/logical.rs

@@ -37,7 +37,7 @@ async fn run() {
         catalog_name: "greptime".to_string(),
         schema_name: "public".to_string(),
         table_name: "test_logical_dist_exec".to_string(),
-        desc: String::default(),
+        desc: "".to_string(),
         column_defs: vec![
             ColumnDef {
                 name: "timestamp".to_string(),

src/client/src/client.rs

@@ -122,7 +122,7 @@ impl Client {
         self.inner.set_peers(urls);
     }
 
-    pub fn find_channel(&self) -> Result<(String, Channel)> {
+    fn find_channel(&self) -> Result<(String, Channel)> {
         let addr = self
             .inner
             .get_peer()

src/client/src/database.rs

@@ -47,9 +47,6 @@ pub struct Database {
     // The dbname follows naming rule as out mysql, postgres and http
     // protocol. The server treat dbname in priority of catalog/schema.
     dbname: String,
-    // The time zone indicates the time zone where the user is located.
-    // Some queries need to be aware of the user's time zone to perform some specific actions.
-    timezone: String,
 
     client: Client,
     ctx: FlightContext,
@@ -61,8 +58,7 @@ impl Database {
         Self {
             catalog: catalog.into(),
             schema: schema.into(),
-            dbname: String::default(),
-            timezone: String::default(),
+            dbname: "".to_string(),
             client,
             ctx: FlightContext::default(),
         }
@@ -77,9 +73,8 @@ impl Database {
     /// environment
     pub fn new_with_dbname(dbname: impl Into<String>, client: Client) -> Self {
         Self {
-            catalog: String::default(),
-            schema: String::default(),
-            timezone: String::default(),
+            catalog: "".to_string(),
+            schema: "".to_string(),
             dbname: dbname.into(),
             client,
             ctx: FlightContext::default(),
@@ -110,14 +105,6 @@ impl Database {
         self.dbname = dbname.into();
     }
 
-    pub fn timezone(&self) -> &String {
-        &self.timezone
-    }
-
-    pub fn set_timezone(&mut self, timezone: impl Into<String>) {
-        self.timezone = timezone.into();
-    }
-
     pub fn set_auth(&mut self, auth: AuthScheme) {
         self.ctx.auth_header = Some(AuthHeader {
             auth_scheme: Some(auth),
@@ -174,7 +161,6 @@ impl Database {
                 schema: self.schema.clone(),
                 authorization: self.ctx.auth_header.clone(),
                 dbname: self.dbname.clone(),
-                timezone: self.timezone.clone(),
                 // TODO(Taylor-lagrange): add client grpc tracing
                 tracing_context: W3cTrace::new(),
             }),
@@ -309,36 +295,30 @@ impl Database {
                 );
                 Ok(Output::AffectedRows(rows))
             }
-            FlightMessage::Recordbatch(_) | FlightMessage::Metrics(_) => {
-                IllegalFlightMessagesSnafu {
-                    reason: "The first flight message cannot be a RecordBatch or Metrics message",
-                }
-                .fail()
+            FlightMessage::Recordbatch(_) => IllegalFlightMessagesSnafu {
+                reason: "The first flight message cannot be a RecordBatch message",
             }
+            .fail(),
             FlightMessage::Schema(schema) => {
                 let stream = Box::pin(stream!({
                     while let Some(flight_message) = flight_message_stream.next().await {
                         let flight_message = flight_message
                             .map_err(BoxedError::new)
                             .context(ExternalSnafu)?;
-                        match flight_message {
-                            FlightMessage::Recordbatch(record_batch) => yield Ok(record_batch),
-                            FlightMessage::Metrics(_) => {}
-                            FlightMessage::AffectedRows(_) | FlightMessage::Schema(_) => {
-                                yield IllegalFlightMessagesSnafu {reason: format!("A Schema message must be succeeded exclusively by a set of RecordBatch messages, flight_message: {:?}", flight_message)}
+                        let FlightMessage::Recordbatch(record_batch) = flight_message else {
+                            yield IllegalFlightMessagesSnafu {reason: "A Schema message must be succeeded exclusively by a set of RecordBatch messages"}
                                         .fail()
                                         .map_err(BoxedError::new)
                                         .context(ExternalSnafu);
-                                break;
-                            }
-                        }
+                            break;
+                        };
+                        yield Ok(record_batch);
                     }
                 }));
                 let record_batch_stream = RecordBatchStreamWrapper {
                     schema,
                     stream,
                     output_ordering: None,
-                    metrics: Default::default(),
                 };
                 Ok(Output::Stream(Box::pin(record_batch_stream)))
             }

src/client/src/error.rs

@@ -16,7 +16,7 @@ use std::any::Any;
 
 use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
-use common_error::{GREPTIME_DB_HEADER_ERROR_CODE, GREPTIME_DB_HEADER_ERROR_MSG};
+use common_error::{GREPTIME_ERROR_CODE, GREPTIME_ERROR_MSG};
 use common_macro::stack_trace_debug;
 use snafu::{Location, Snafu};
 use tonic::{Code, Status};
@@ -115,7 +115,7 @@ impl From<Status> for Error {
                 .and_then(|v| String::from_utf8(v.as_bytes().to_vec()).ok())
         }
 
-        let code = get_metadata_value(&e, GREPTIME_DB_HEADER_ERROR_CODE)
+        let code = get_metadata_value(&e, GREPTIME_ERROR_CODE)
             .and_then(|s| {
                 if let Ok(code) = s.parse::<u32>() {
                     StatusCode::from_u32(code)
@@ -125,8 +125,8 @@ impl From<Status> for Error {
             })
             .unwrap_or(StatusCode::Unknown);
 
-        let msg = get_metadata_value(&e, GREPTIME_DB_HEADER_ERROR_MSG)
-            .unwrap_or_else(|| e.message().to_string());
+        let msg =
+            get_metadata_value(&e, GREPTIME_ERROR_MSG).unwrap_or_else(|| e.message().to_string());
 
         Self::Server { code, msg }
     }

src/client/src/metrics.rs

@@ -17,30 +17,27 @@ use prometheus::*;
 
 lazy_static! {
     pub static ref METRIC_GRPC_CREATE_TABLE: Histogram =
-        register_histogram!("greptime_grpc_create_table", "grpc create table").unwrap();
-    pub static ref METRIC_GRPC_PROMQL_RANGE_QUERY: Histogram = register_histogram!(
-        "greptime_grpc_promql_range_query",
-        "grpc promql range query"
-    )
-    .unwrap();
+        register_histogram!("grpc_create_table", "grpc create table").unwrap();
+    pub static ref METRIC_GRPC_PROMQL_RANGE_QUERY: Histogram =
+        register_histogram!("grpc_promql_range_query", "grpc promql range query").unwrap();
     pub static ref METRIC_GRPC_INSERT: Histogram =
-        register_histogram!("greptime_grpc_insert", "grpc insert").unwrap();
+        register_histogram!("grpc_insert", "grpc insert").unwrap();
     pub static ref METRIC_GRPC_DELETE: Histogram =
-        register_histogram!("greptime_grpc_delete", "grpc delete").unwrap();
+        register_histogram!("grpc_delete", "grpc delete").unwrap();
     pub static ref METRIC_GRPC_SQL: Histogram =
-        register_histogram!("greptime_grpc_sql", "grpc sql").unwrap();
+        register_histogram!("grpc_sql", "grpc sql").unwrap();
     pub static ref METRIC_GRPC_LOGICAL_PLAN: Histogram =
-        register_histogram!("greptime_grpc_logical_plan", "grpc logical plan").unwrap();
+        register_histogram!("grpc_logical_plan", "grpc logical plan").unwrap();
     pub static ref METRIC_GRPC_ALTER: Histogram =
-        register_histogram!("greptime_grpc_alter", "grpc alter").unwrap();
+        register_histogram!("grpc_alter", "grpc alter").unwrap();
     pub static ref METRIC_GRPC_DROP_TABLE: Histogram =
-        register_histogram!("greptime_grpc_drop_table", "grpc drop table").unwrap();
+        register_histogram!("grpc_drop_table", "grpc drop table").unwrap();
     pub static ref METRIC_GRPC_TRUNCATE_TABLE: Histogram =
-        register_histogram!("greptime_grpc_truncate_table", "grpc truncate table").unwrap();
+        register_histogram!("grpc_truncate_table", "grpc truncate table").unwrap();
     pub static ref METRIC_GRPC_DO_GET: Histogram =
-        register_histogram!("greptime_grpc_do_get", "grpc do get").unwrap();
+        register_histogram!("grpc_do_get", "grpc do get").unwrap();
     pub static ref METRIC_REGION_REQUEST_GRPC: HistogramVec = register_histogram_vec!(
-        "greptime_grpc_region_request",
+        "grpc_region_request",
         "grpc region request",
         &["request_type"]
     )

src/client/src/region.rs

@@ -12,11 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
 use api::v1::region::{QueryRequest, RegionRequest, RegionResponse};
 use api::v1::ResponseHeader;
-use arc_swap::ArcSwapOption;
 use arrow_flight::Ticket;
 use async_stream::stream;
 use async_trait::async_trait;
@@ -28,7 +25,6 @@ use common_meta::error::{self as meta_error, Result as MetaResult};
 use common_recordbatch::error::ExternalSnafu;
 use common_recordbatch::{RecordBatchStreamWrapper, SendableRecordBatchStream};
 use common_telemetry::error;
-use common_telemetry::tracing_context::TracingContext;
 use prost::Message;
 use snafu::{location, Location, OptionExt, ResultExt};
 use tokio_stream::StreamExt;
@@ -123,43 +119,27 @@ impl RegionRequester {
             .fail();
         };
 
-        let metrics_str = Arc::new(ArcSwapOption::from(None));
-        let ref_str = metrics_str.clone();
-
-        let tracing_context = TracingContext::from_current_span();
-
         let stream = Box::pin(stream!({
-            let _span = tracing_context.attach(common_telemetry::tracing::info_span!(
-                "poll_flight_data_stream"
-            ));
             while let Some(flight_message) = flight_message_stream.next().await {
                 let flight_message = flight_message
                     .map_err(BoxedError::new)
                     .context(ExternalSnafu)?;
-
-                match flight_message {
-                    FlightMessage::Recordbatch(record_batch) => yield Ok(record_batch),
-                    FlightMessage::Metrics(s) => {
-                        ref_str.swap(Some(Arc::new(s)));
-                        break;
-                    }
-                    _ => {
-                        yield IllegalFlightMessagesSnafu {
+                let FlightMessage::Recordbatch(record_batch) = flight_message else {
+                    yield IllegalFlightMessagesSnafu {
                             reason: "A Schema message must be succeeded exclusively by a set of RecordBatch messages"
                         }
                         .fail()
                         .map_err(BoxedError::new)
                         .context(ExternalSnafu);
-                        break;
-                    }
-                }
+                    break;
+                };
+                yield Ok(record_batch);
             }
         }));
         let record_batch_stream = RecordBatchStreamWrapper {
             schema,
             stream,
             output_ordering: None,
-            metrics: metrics_str,
         };
         Ok(Box::pin(record_batch_stream))
     }
@@ -250,7 +230,7 @@ mod test {
         let result = check_response_header(Some(ResponseHeader {
             status: Some(PbStatus {
                 status_code: StatusCode::Success as u32,
-                err_msg: String::default(),
+                err_msg: "".to_string(),
             }),
         }));
         assert!(result.is_ok());
@@ -258,7 +238,7 @@ mod test {
         let result = check_response_header(Some(ResponseHeader {
             status: Some(PbStatus {
                 status_code: u32::MAX,
-                err_msg: String::default(),
+                err_msg: "".to_string(),
             }),
         }));
         assert!(matches!(

src/client/src/stream_insert.rs

@@ -39,7 +39,7 @@ use crate::from_grpc_response;
 /// ```
 ///
 /// If you want to see a concrete usage example, please see
-/// [stream_inserter.rs](https://github.com/GreptimeTeam/greptimedb/blob/main/src/client/examples/stream_ingest.rs).
+/// [stream_inserter.rs](https://github.com/GreptimeTeam/greptimedb/blob/develop/src/client/examples/stream_ingest.rs).
 pub struct StreamInserter {
     sender: mpsc::Sender<GreptimeRequest>,
 

src/cmd/Cargo.toml

@@ -18,7 +18,7 @@ async-trait.workspace = true
 auth.workspace = true
 catalog.workspace = true
 chrono.workspace = true
-clap.workspace = true
+clap = { version = "4.4", features = ["derive"] }
 client.workspace = true
 common-base.workspace = true
 common-catalog.workspace = true
@@ -29,12 +29,10 @@ common-meta.workspace = true
 common-procedure.workspace = true
 common-query.workspace = true
 common-recordbatch.workspace = true
-common-runtime.workspace = true
 common-telemetry = { workspace = true, features = [
     "deadlock_detection",
 ] }
 common-time.workspace = true
-common-wal.workspace = true
 config = "0.13"
 datanode.workspace = true
 datatypes.workspace = true

src/cmd/build.rs

@@ -13,5 +13,5 @@
 // limitations under the License.
 
 fn main() {
-    common_version::setup_build_info();
+    common_version::setup_git_versions();
 }

src/cmd/src/cli/bench.rs

@@ -156,7 +156,6 @@ fn create_region_routes(regions: Vec<RegionNumber>) -> Vec<RegionRoute> {
             }),
             follower_peers: vec![],
             leader_status: None,
-            leader_down_since: None,
         });
     }
 

src/cmd/src/cli/export.rs

@@ -58,8 +58,8 @@ pub struct ExportCommand {
     #[clap(long)]
     output_dir: String,
 
-    /// The name of the catalog to export.
-    #[clap(long, default_value = "greptime-*")]
+    /// The name of the catalog to export. Default to "greptime-*"".
+    #[clap(long, default_value = "")]
     database: String,
 
     /// Parallelism of the export.

src/cmd/src/cli/repl.rs

@@ -16,9 +16,7 @@ use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Instant;
 
-use catalog::kvbackend::{
-    CachedMetaKvBackend, CachedMetaKvBackendBuilder, KvBackendCatalogManager,
-};
+use catalog::kvbackend::{CachedMetaKvBackend, KvBackendCatalogManager};
 use client::{Client, Database, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_base::Plugins;
 use common_error::ext::ErrorExt;
@@ -159,20 +157,19 @@ impl Repl {
         let start = Instant::now();
 
         let output = if let Some(query_engine) = &self.query_engine {
-            let query_ctx = QueryContext::with(self.database.catalog(), self.database.schema());
-
-            let stmt = QueryLanguageParser::parse_sql(&sql, &query_ctx)
+            let stmt = QueryLanguageParser::parse_sql(&sql)
                 .with_context(|_| ParseSqlSnafu { sql: sql.clone() })?;
 
+            let query_ctx = QueryContext::with(self.database.catalog(), self.database.schema());
+
             let plan = query_engine
                 .planner()
-                .plan(stmt, query_ctx.clone())
+                .plan(stmt, query_ctx)
                 .await
                 .context(PlanStatementSnafu)?;
 
-            let LogicalPlan::DfPlan(plan) = query_engine
-                .optimize(&query_engine.engine_context(query_ctx), &plan)
-                .context(PlanStatementSnafu)?;
+            let LogicalPlan::DfPlan(plan) =
+                query_engine.optimize(&plan).context(PlanStatementSnafu)?;
 
             let plan = DFLogicalSubstraitConvertor {}
                 .encode(&plan)
@@ -250,8 +247,7 @@ async fn create_query_engine(meta_addr: &str) -> Result<DatafusionQueryEngine> {
         .context(StartMetaClientSnafu)?;
     let meta_client = Arc::new(meta_client);
 
-    let cached_meta_backend =
-        Arc::new(CachedMetaKvBackendBuilder::new(meta_client.clone()).build());
+    let cached_meta_backend = Arc::new(CachedMetaKvBackend::new(meta_client.clone()));
 
     let catalog_list =
         KvBackendCatalogManager::new(cached_meta_backend.clone(), cached_meta_backend);

src/cmd/src/datanode.rs

@@ -18,11 +18,10 @@ use std::time::Duration;
 use async_trait::async_trait;
 use catalog::kvbackend::MetaKvBackend;
 use clap::Parser;
+use common_config::WalConfig;
 use common_telemetry::{info, logging};
-use common_wal::config::DatanodeWalConfig;
 use datanode::config::DatanodeOptions;
 use datanode::datanode::{Datanode, DatanodeBuilder};
-use datanode::service::DatanodeServiceBuilder;
 use meta_client::MetaClientOptions;
 use servers::Mode;
 use snafu::{OptionExt, ResultExt};
@@ -36,13 +35,9 @@ pub struct Instance {
 }
 
 impl Instance {
-    pub fn new(datanode: Datanode) -> Self {
+    fn new(datanode: Datanode) -> Self {
         Self { datanode }
     }
-
-    pub fn datanode_mut(&mut self) -> &mut Datanode {
-        &mut self.datanode
-    }
 }
 
 #[async_trait]
@@ -174,7 +169,7 @@ impl StartCommand {
 
         // `wal_dir` only affects raft-engine config.
         if let Some(wal_dir) = &self.wal_dir
-            && let DatanodeWalConfig::RaftEngine(raft_engine_config) = &mut opts.wal
+            && let WalConfig::RaftEngine(raft_engine_config) = &mut opts.wal
         {
             if raft_engine_config
                 .dir
@@ -224,20 +219,15 @@ impl StartCommand {
             client: Arc::new(meta_client.clone()),
         });
 
-        let mut datanode = DatanodeBuilder::new(opts.clone(), plugins)
+        let datanode = DatanodeBuilder::new(opts, plugins)
             .with_meta_client(meta_client)
             .with_kv_backend(meta_backend)
+            .enable_region_server_service()
+            .enable_http_service()
             .build()
             .await
             .context(StartDatanodeSnafu)?;
 
-        let services = DatanodeServiceBuilder::new(&opts)
-            .with_default_grpc_server(&datanode.region_server())
-            .enable_http_service()
-            .build()
-            .context(StartDatanodeSnafu)?;
-        datanode.setup_services(services);
-
         Ok(Instance::new(datanode))
     }
 }
@@ -316,7 +306,7 @@ mod tests {
         assert_eq!("127.0.0.1:3001".to_string(), options.rpc_addr);
         assert_eq!(Some(42), options.node_id);
 
-        let DatanodeWalConfig::RaftEngine(raft_engine_config) = options.wal else {
+        let WalConfig::RaftEngine(raft_engine_config) = options.wal else {
             unreachable!()
         };
         assert_eq!("/other/wal", raft_engine_config.dir.unwrap());
@@ -504,7 +494,7 @@ mod tests {
                 };
 
                 // Should be read from env, env > default values.
-                let DatanodeWalConfig::RaftEngine(raft_engine_config) = opts.wal else {
+                let WalConfig::RaftEngine(raft_engine_config) = opts.wal else {
                     unreachable!()
                 };
                 assert_eq!(raft_engine_config.read_batch_size, 100);

src/cmd/src/error.rs

@@ -249,12 +249,6 @@ pub enum Error {
         source: BoxedError,
         location: Location,
     },
-
-    #[snafu(display("Failed to build runtime"))]
-    BuildRuntime {
-        location: Location,
-        source: common_runtime::error::Error,
-    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -304,8 +298,6 @@ impl ErrorExt for Error {
             Error::SerdeJson { .. } | Error::FileIo { .. } => StatusCode::Unexpected,
 
             Error::Other { source, .. } => source.status_code(),
-
-            Error::BuildRuntime { source, .. } => source.status_code(),
         }
     }
 

src/cmd/src/frontend.rs

@@ -16,7 +16,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use async_trait::async_trait;
-use catalog::kvbackend::CachedMetaKvBackendBuilder;
+use catalog::kvbackend::CachedMetaKvBackend;
 use clap::Parser;
 use client::client_manager::DatanodeClients;
 use common_meta::heartbeat::handler::parse_mailbox_message::ParseMailboxMessageHandler;
@@ -28,7 +28,6 @@ use frontend::heartbeat::handler::invalidate_table_cache::InvalidateTableCacheHa
 use frontend::heartbeat::HeartbeatTask;
 use frontend::instance::builder::FrontendBuilder;
 use frontend::instance::{FrontendInstance, Instance as FeInstance};
-use frontend::server::Services;
 use meta_client::MetaClientOptions;
 use servers::tls::{TlsMode, TlsOption};
 use servers::Mode;
@@ -46,10 +45,6 @@ impl Instance {
     fn new(frontend: FeInstance) -> Self {
         Self { frontend }
     }
-
-    pub fn mut_inner(&mut self) -> &mut FeInstance {
-        &mut self.frontend
-    }
 }
 
 #[async_trait]
@@ -228,27 +223,15 @@ impl StartCommand {
         let meta_client_options = opts.meta_client.as_ref().context(MissingConfigSnafu {
             msg: "'meta_client'",
         })?;
-
-        let cache_max_capacity = meta_client_options.metadata_cache_max_capacity;
-        let cache_ttl = meta_client_options.metadata_cache_ttl;
-        let cache_tti = meta_client_options.metadata_cache_tti;
-
         let meta_client = FeInstance::create_meta_client(meta_client_options)
             .await
             .context(StartFrontendSnafu)?;
 
-        let cached_meta_backend = CachedMetaKvBackendBuilder::new(meta_client.clone())
-            .cache_max_capacity(cache_max_capacity)
-            .cache_ttl(cache_ttl)
-            .cache_tti(cache_tti)
-            .build();
-        let cached_meta_backend = Arc::new(cached_meta_backend);
+        let meta_backend = Arc::new(CachedMetaKvBackend::new(meta_client.clone()));
 
         let executor = HandlerGroupExecutor::new(vec![
             Arc::new(ParseMailboxMessageHandler),
-            Arc::new(InvalidateTableCacheHandler::new(
-                cached_meta_backend.clone(),
-            )),
+            Arc::new(InvalidateTableCacheHandler::new(meta_backend.clone())),
         ]);
 
         let heartbeat_task = HeartbeatTask::new(
@@ -258,22 +241,24 @@ impl StartCommand {
         );
 
         let mut instance = FrontendBuilder::new(
-            cached_meta_backend.clone(),
+            meta_backend.clone(),
             Arc::new(DatanodeClients::default()),
             meta_client,
         )
-        .with_cache_invalidator(cached_meta_backend)
-        .with_plugin(plugins.clone())
+        .with_cache_invalidator(meta_backend)
+        .with_plugin(plugins)
         .with_heartbeat_task(heartbeat_task)
         .try_build()
         .await
         .context(StartFrontendSnafu)?;
 
-        let servers = Services::new(opts.clone(), Arc::new(instance.clone()), plugins)
-            .build()
-            .context(StartFrontendSnafu)?;
         instance
-            .build_servers(opts, servers)
+            .build_export_metrics_task(&opts.export_metrics)
+            .context(StartFrontendSnafu)?;
+
+        instance
+            .build_servers(opts)
+            .await
             .context(StartFrontendSnafu)?;
 
         Ok(Instance::new(instance))

src/cmd/src/lib.rs

@@ -28,18 +28,13 @@ pub mod standalone;
 
 lazy_static::lazy_static! {
     static ref APP_VERSION: prometheus::IntGaugeVec =
-        prometheus::register_int_gauge_vec!("greptime_app_version", "app version", &["short_version", "version"]).unwrap();
+        prometheus::register_int_gauge_vec!("app_version", "app version", &["short_version", "version"]).unwrap();
 }
 
 #[async_trait]
 pub trait App {
     fn name(&self) -> &str;
 
-    /// A hook for implementor to make something happened before actual startup. Defaults to no-op.
-    fn pre_start(&mut self) -> error::Result<()> {
-        Ok(())
-    }
-
     async fn start(&mut self) -> error::Result<()>;
 
     async fn stop(&self) -> error::Result<()>;
@@ -48,8 +43,6 @@ pub trait App {
 pub async fn start_app(mut app: Box<dyn App>) -> error::Result<()> {
     let name = app.name().to_string();
 
-    app.pre_start()?;
-
     tokio::select! {
         result = app.start() => {
             if let Err(err) = result {

src/cmd/src/metasrv.rs

@@ -128,7 +128,7 @@ impl StartCommand {
         let mut opts: MetaSrvOptions = Options::load_layered_options(
             self.config_file.as_deref(),
             self.env_prefix.as_ref(),
-            MetaSrvOptions::env_list_keys(),
+            None,
         )?;
 
         if let Some(dir) = &cli_options.log_dir {

src/cmd/src/options.rs

@@ -14,8 +14,8 @@
 
 use clap::ArgMatches;
 use common_config::KvBackendConfig;
+use common_meta::wal::WalConfig as MetaSrvWalConfig;
 use common_telemetry::logging::{LoggingOptions, TracingOptions};
-use common_wal::config::MetaSrvWalConfig;
 use config::{Config, Environment, File, FileFormat};
 use datanode::config::{DatanodeOptions, ProcedureConfig};
 use frontend::error::{Result as FeResult, TomlFormatSnafu};
@@ -173,8 +173,8 @@ impl Options {
 mod tests {
     use std::io::Write;
 
+    use common_config::WalConfig;
     use common_test_util::temp_dir::create_named_temp_file;
-    use common_wal::config::DatanodeWalConfig;
     use datanode::config::{DatanodeOptions, ObjectStoreConfig};
 
     use super::*;
@@ -281,7 +281,7 @@ mod tests {
                 );
 
                 // Should be the values from config file, not environment variables.
-                let DatanodeWalConfig::RaftEngine(raft_engine_config) = opts.wal else {
+                let WalConfig::RaftEngine(raft_engine_config) = opts.wal else {
                     unreachable!()
                 };
                 assert_eq!(raft_engine_config.dir.unwrap(), "/tmp/greptimedb/wal");

src/cmd/src/standalone.rs

@@ -18,29 +18,28 @@ use std::{fs, path};
 use async_trait::async_trait;
 use clap::Parser;
 use common_catalog::consts::MIN_USER_TABLE_ID;
+use common_config::wal::StandaloneWalConfig;
 use common_config::{metadata_store_dir, KvBackendConfig};
 use common_meta::cache_invalidator::DummyCacheInvalidator;
 use common_meta::datanode_manager::DatanodeManagerRef;
-use common_meta::ddl::table_meta::TableMetadataAllocator;
-use common_meta::ddl::DdlTaskExecutorRef;
+use common_meta::ddl::{DdlTaskExecutorRef, TableMetadataAllocatorRef};
 use common_meta::ddl_manager::DdlManager;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::region_keeper::MemoryRegionKeeper;
 use common_meta::sequence::SequenceBuilder;
-use common_meta::wal_options_allocator::{WalOptionsAllocator, WalOptionsAllocatorRef};
+use common_meta::wal::{WalOptionsAllocator, WalOptionsAllocatorRef};
 use common_procedure::ProcedureManagerRef;
 use common_telemetry::info;
 use common_telemetry::logging::LoggingOptions;
 use common_time::timezone::set_default_timezone;
-use common_wal::config::StandaloneWalConfig;
 use datanode::config::{DatanodeOptions, ProcedureConfig, RegionEngineConfig, StorageConfig};
 use datanode::datanode::{Datanode, DatanodeBuilder};
 use file_engine::config::EngineConfig as FileEngineConfig;
 use frontend::frontend::FrontendOptions;
 use frontend::instance::builder::FrontendBuilder;
+use frontend::instance::standalone::StandaloneTableMetadataAllocator;
 use frontend::instance::{FrontendInstance, Instance as FeInstance, StandaloneDatanodeManager};
-use frontend::server::Services;
 use frontend::service_config::{
     GrpcOptions, InfluxdbOptions, MysqlOptions, OpentsdbOptions, PostgresOptions, PromStoreOptions,
 };
@@ -119,12 +118,6 @@ pub struct StandaloneOptions {
     pub export_metrics: ExportMetricsOption,
 }
 
-impl StandaloneOptions {
-    pub fn env_list_keys() -> Option<&'static [&'static str]> {
-        Some(&["wal.broker_endpoints"])
-    }
-}
-
 impl Default for StandaloneOptions {
     fn default() -> Self {
         Self {
@@ -213,10 +206,6 @@ impl App for Instance {
             .await
             .context(StartWalOptionsAllocatorSnafu)?;
 
-        plugins::start_frontend_plugins(self.frontend.plugins().clone())
-            .await
-            .context(StartFrontendSnafu)?;
-
         self.frontend.start().await.context(StartFrontendSnafu)?;
         Ok(())
     }
@@ -278,7 +267,7 @@ impl StartCommand {
         let opts: StandaloneOptions = Options::load_layered_options(
             self.config_file.as_deref(),
             self.env_prefix.as_ref(),
-            StandaloneOptions::env_list_keys(),
+            None,
         )?;
 
         self.convert_options(cli_options, opts)
@@ -372,18 +361,20 @@ impl StartCommand {
     #[allow(unused_variables)]
     #[allow(clippy::diverging_sub_expression)]
     async fn build(self, opts: MixOptions) -> Result<Instance> {
-        info!("Standalone start command: {:#?}", self);
-        info!("Building standalone instance with {opts:#?}");
-
-        let mut fe_opts = opts.frontend;
+        let mut fe_opts = opts.frontend.clone();
         #[allow(clippy::unnecessary_mut_passed)]
         let fe_plugins = plugins::setup_frontend_plugins(&mut fe_opts) // mut ref is MUST, DO NOT change it
             .await
             .context(StartFrontendSnafu)?;
 
-        let dn_opts = opts.datanode;
+        let dn_opts = opts.datanode.clone();
 
-        set_default_timezone(fe_opts.default_timezone.as_deref()).context(InitTimezoneSnafu)?;
+        info!("Standalone start command: {:#?}", self);
+
+        info!("Building standalone instance with {opts:#?}");
+
+        set_default_timezone(opts.frontend.default_timezone.as_deref())
+            .context(InitTimezoneSnafu)?;
 
         // Ensure the data_home directory exists.
         fs::create_dir_all(path::Path::new(&opts.data_home)).context(CreateDirSnafu {
@@ -415,18 +406,13 @@ impl StartCommand {
             opts.wal_meta.clone(),
             kv_backend.clone(),
         ));
-
-        let table_metadata_manager =
-            Self::create_table_metadata_manager(kv_backend.clone()).await?;
-
-        let table_meta_allocator = TableMetadataAllocator::new(
+        let table_meta_allocator = Arc::new(StandaloneTableMetadataAllocator::new(
             table_id_sequence,
             wal_options_allocator.clone(),
-            table_metadata_manager.clone(),
-        );
+        ));
 
         let ddl_task_executor = Self::create_ddl_task_executor(
-            table_metadata_manager,
+            kv_backend.clone(),
             procedure_manager.clone(),
             datanode_manager.clone(),
             table_meta_allocator,
@@ -434,16 +420,18 @@ impl StartCommand {
         .await?;
 
         let mut frontend = FrontendBuilder::new(kv_backend, datanode_manager, ddl_task_executor)
-            .with_plugin(fe_plugins.clone())
+            .with_plugin(fe_plugins)
             .try_build()
             .await
             .context(StartFrontendSnafu)?;
 
-        let servers = Services::new(fe_opts.clone(), Arc::new(frontend.clone()), fe_plugins)
-            .build()
-            .context(StartFrontendSnafu)?;
         frontend
-            .build_servers(fe_opts, servers)
+            .build_export_metrics_task(&opts.frontend.export_metrics)
+            .context(StartFrontendSnafu)?;
+
+        frontend
+            .build_servers(opts)
+            .await
             .context(StartFrontendSnafu)?;
 
         Ok(Instance {
@@ -455,11 +443,14 @@ impl StartCommand {
     }
 
     pub async fn create_ddl_task_executor(
-        table_metadata_manager: TableMetadataManagerRef,
+        kv_backend: KvBackendRef,
         procedure_manager: ProcedureManagerRef,
         datanode_manager: DatanodeManagerRef,
-        table_meta_allocator: TableMetadataAllocator,
+        table_meta_allocator: TableMetadataAllocatorRef,
     ) -> Result<DdlTaskExecutorRef> {
+        let table_metadata_manager =
+            Self::create_table_metadata_manager(kv_backend.clone()).await?;
+
         let ddl_task_executor: DdlTaskExecutorRef = Arc::new(
             DdlManager::try_new(
                 procedure_manager,
@@ -475,7 +466,7 @@ impl StartCommand {
         Ok(ddl_task_executor)
     }
 
-    pub async fn create_table_metadata_manager(
+    async fn create_table_metadata_manager(
         kv_backend: KvBackendRef,
     ) -> Result<TableMetadataManagerRef> {
         let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend));
@@ -497,8 +488,8 @@ mod tests {
 
     use auth::{Identity, Password, UserProviderRef};
     use common_base::readable_size::ReadableSize;
+    use common_config::WalConfig;
     use common_test_util::temp_dir::create_named_temp_file;
-    use common_wal::config::DatanodeWalConfig;
     use datanode::config::{FileConfig, GcsConfig};
     use servers::Mode;
 
@@ -605,7 +596,7 @@ mod tests {
         assert_eq!(None, fe_opts.mysql.reject_no_database);
         assert!(fe_opts.influxdb.enable);
 
-        let DatanodeWalConfig::RaftEngine(raft_engine_config) = dn_opts.wal else {
+        let WalConfig::RaftEngine(raft_engine_config) = dn_opts.wal else {
             unreachable!()
         };
         assert_eq!("/tmp/greptimedb/test/wal", raft_engine_config.dir.unwrap());

src/common/base/src/bytes.rs

@@ -216,7 +216,7 @@ mod tests {
         let bytes = StringBytes::from(hello.clone());
         assert_eq!(bytes.len(), hello.len());
 
-        let zero = String::default();
+        let zero = "".to_string();
         let bytes = StringBytes::from(zero);
         assert!(bytes.is_empty());
     }

src/common/base/src/readable_size.rs

@@ -33,7 +33,7 @@ pub const GIB: u64 = MIB * BINARY_DATA_MAGNITUDE;
 pub const TIB: u64 = GIB * BINARY_DATA_MAGNITUDE;
 pub const PIB: u64 = TIB * BINARY_DATA_MAGNITUDE;
 
-#[derive(Clone, Copy, PartialEq, Eq, Ord, PartialOrd)]
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd)]
 pub struct ReadableSize(pub u64);
 
 impl ReadableSize {

src/common/catalog/src/consts.rs

@@ -44,48 +44,6 @@ pub const INFORMATION_SCHEMA_COLUMN_PRIVILEGES_TABLE_ID: u32 = 6;
 pub const INFORMATION_SCHEMA_COLUMN_STATISTICS_TABLE_ID: u32 = 7;
 /// id for information_schema.build_info
 pub const INFORMATION_SCHEMA_BUILD_INFO_TABLE_ID: u32 = 8;
-/// id for information_schema.CHARACTER_SETS
-pub const INFORMATION_SCHEMA_CHARACTER_SETS_TABLE_ID: u32 = 9;
-/// id for information_schema.COLLATIONS
-pub const INFORMATION_SCHEMA_COLLATIONS_TABLE_ID: u32 = 10;
-/// id for information_schema.COLLATIONS
-pub const INFORMATION_SCHEMA_COLLATION_CHARACTER_SET_APPLICABILITY_TABLE_ID: u32 = 11;
-/// id for information_schema.CHECK_CONSTRAINTS
-pub const INFORMATION_SCHEMA_CHECK_CONSTRAINTS_TABLE_ID: u32 = 12;
-/// id for information_schema.EVENTS
-pub const INFORMATION_SCHEMA_EVENTS_TABLE_ID: u32 = 13;
-/// id for information_schema.FILES
-pub const INFORMATION_SCHEMA_FILES_TABLE_ID: u32 = 14;
-/// id for information_schema.SCHEMATA
-pub const INFORMATION_SCHEMA_SCHEMATA_TABLE_ID: u32 = 15;
-/// id for information_schema.KEY_COLUMN_USAGE
-pub const INFORMATION_SCHEMA_KEY_COLUMN_USAGE_TABLE_ID: u32 = 16;
-/// id for information_schema.OPTIMIZER_TRACE
-pub const INFORMATION_SCHEMA_OPTIMIZER_TRACE_TABLE_ID: u32 = 17;
-/// id for information_schema.PARAMETERS
-pub const INFORMATION_SCHEMA_PARAMETERS_TABLE_ID: u32 = 18;
-/// id for information_schema.PROFILING
-pub const INFORMATION_SCHEMA_PROFILING_TABLE_ID: u32 = 19;
-/// id for information_schema.REFERENTIAL_CONSTRAINTS
-pub const INFORMATION_SCHEMA_REFERENTIAL_CONSTRAINTS_TABLE_ID: u32 = 20;
-/// id for information_schema.ROUTINES
-pub const INFORMATION_SCHEMA_ROUTINES_TABLE_ID: u32 = 21;
-/// id for information_schema.SCHEMA_PRIVILEGES
-pub const INFORMATION_SCHEMA_SCHEMA_PRIVILEGES_TABLE_ID: u32 = 22;
-/// id for information_schema.TABLE_PRIVILEGES
-pub const INFORMATION_SCHEMA_TABLE_PRIVILEGES_TABLE_ID: u32 = 23;
-/// id for information_schema.TRIGGERS
-pub const INFORMATION_SCHEMA_TRIGGERS_TABLE_ID: u32 = 24;
-/// id for information_schema.GLOBAL_STATUS
-pub const INFORMATION_SCHEMA_GLOBAL_STATUS_TABLE_ID: u32 = 25;
-/// id for information_schema.SESSION_STATUS
-pub const INFORMATION_SCHEMA_SESSION_STATUS_TABLE_ID: u32 = 26;
-/// id for information_schema.RUNTIME_METRICS
-pub const INFORMATION_SCHEMA_RUNTIME_METRICS_TABLE_ID: u32 = 27;
-/// id for information_schema.PARTITIONS
-pub const INFORMATION_SCHEMA_PARTITIONS_TABLE_ID: u32 = 28;
-/// id for information_schema.REGION_PEERS
-pub const INFORMATION_SCHEMA_REGION_PEERS_TABLE_ID: u32 = 29;
 /// ----- End of information_schema tables -----
 
 pub const MITO_ENGINE: &str = "mito";

src/common/catalog/src/lib.rs

@@ -17,11 +17,6 @@ use consts::DEFAULT_CATALOG_NAME;
 pub mod consts;
 pub mod error;
 
-#[inline]
-pub fn format_schema_name(catalog: &str, schema: &str) -> String {
-    format!("{catalog}.{schema}")
-}
-
 /// Formats table fully-qualified name
 #[inline]
 pub fn format_full_table_name(catalog: &str, schema: &str, table: &str) -> String {
@@ -56,22 +51,11 @@ pub fn build_db_string(catalog: &str, schema: &str) -> String {
 /// - if `[<catalog>-]` is provided, we split database name with `-` and use
 /// `<catalog>` and `<schema>`.
 pub fn parse_catalog_and_schema_from_db_string(db: &str) -> (&str, &str) {
-    match parse_optional_catalog_and_schema_from_db_string(db) {
-        (Some(catalog), schema) => (catalog, schema),
-        (None, schema) => (DEFAULT_CATALOG_NAME, schema),
-    }
-}
-
-/// Attempt to parse catalog and schema from given database name
-///
-/// Similar to [`parse_catalog_and_schema_from_db_string`] but returns an optional
-/// catalog if it's not provided in the database name.
-pub fn parse_optional_catalog_and_schema_from_db_string(db: &str) -> (Option<&str>, &str) {
     let parts = db.splitn(2, '-').collect::<Vec<&str>>();
     if parts.len() == 2 {
-        (Some(parts[0]), parts[1])
+        (parts[0], parts[1])
     } else {
-        (None, db)
+        (DEFAULT_CATALOG_NAME, db)
     }
 }
 
@@ -101,20 +85,5 @@ mod tests {
             ("catalog", "schema1-schema2"),
             parse_catalog_and_schema_from_db_string("catalog-schema1-schema2")
         );
-
-        assert_eq!(
-            (None, "fullschema"),
-            parse_optional_catalog_and_schema_from_db_string("fullschema")
-        );
-
-        assert_eq!(
-            (Some("catalog"), "schema"),
-            parse_optional_catalog_and_schema_from_db_string("catalog-schema")
-        );
-
-        assert_eq!(
-            (Some("catalog"), "schema1-schema2"),
-            parse_optional_catalog_and_schema_from_db_string("catalog-schema1-schema2")
-        );
     }
 }

src/common/config/Cargo.toml

@@ -7,6 +7,8 @@ license.workspace = true
 [dependencies]
 common-base.workspace = true
 humantime-serde.workspace = true
-num_cpus.workspace = true
+rskafka.workspace = true
 serde.workspace = true
-sysinfo.workspace = true
+serde_json.workspace = true
+serde_with = "3"
+toml.workspace = true

src/common/config/src/lib.rs

@@ -12,11 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub mod utils;
+pub mod wal;
 
 use common_base::readable_size::ReadableSize;
 use serde::{Deserialize, Serialize};
 
+pub use crate::wal::{KafkaWalOptions, WalConfig, WalOptions, WAL_OPTIONS_KEY};
+
 pub fn metadata_store_dir(store_dir: &str) -> String {
     format!("{store_dir}/metadata")
 }

src/common/config/src/utils.rs

@@ -1,55 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed 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.
-
-use common_base::readable_size::ReadableSize;
-use sysinfo::System;
-
-/// Get the CPU core number of system, aware of cgroups.
-pub fn get_cpus() -> usize {
-    // This function will check cgroups
-    num_cpus::get()
-}
-
-/// Get the total memory of the system.
-/// If `cgroup_limits` is enabled, it will also check it.
-pub fn get_sys_total_memory() -> Option<ReadableSize> {
-    if sysinfo::IS_SUPPORTED_SYSTEM {
-        let mut sys_info = System::new();
-        sys_info.refresh_memory();
-        let mut total_memory = sys_info.total_memory();
-        // Compare with cgroups memory limit, use smaller values
-        // This method is only implemented for Linux. It always returns None for all other systems.
-        if let Some(cgroup_limits) = sys_info.cgroup_limits() {
-            total_memory = total_memory.min(cgroup_limits.total_memory)
-        }
-        Some(ReadableSize(total_memory))
-    } else {
-        None
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_get_cpus() {
-        assert!(get_cpus() > 0);
-    }
-
-    #[test]
-    fn test_get_sys_total_memory() {
-        assert!(get_sys_total_memory().unwrap() > ReadableSize::mb(0));
-    }
-}

src/common/config/src/wal.rs

@@ -0,0 +1,142 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed 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.
+
+pub mod kafka;
+pub mod raft_engine;
+
+use serde::{Deserialize, Serialize};
+use serde_with::with_prefix;
+
+pub use crate::wal::kafka::{
+    KafkaConfig, KafkaOptions as KafkaWalOptions, StandaloneKafkaConfig, Topic as KafkaWalTopic,
+};
+pub use crate::wal::raft_engine::RaftEngineConfig;
+
+/// An encoded wal options will be wrapped into a (WAL_OPTIONS_KEY, encoded wal options) key-value pair
+/// and inserted into the options of a `RegionCreateRequest`.
+pub const WAL_OPTIONS_KEY: &str = "wal_options";
+
+/// Wal config for datanode.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "provider", rename_all = "snake_case")]
+pub enum WalConfig {
+    RaftEngine(RaftEngineConfig),
+    Kafka(KafkaConfig),
+}
+
+impl From<StandaloneWalConfig> for WalConfig {
+    fn from(value: StandaloneWalConfig) -> Self {
+        match value {
+            StandaloneWalConfig::RaftEngine(config) => WalConfig::RaftEngine(config),
+            StandaloneWalConfig::Kafka(config) => WalConfig::Kafka(config.base),
+        }
+    }
+}
+
+impl Default for WalConfig {
+    fn default() -> Self {
+        WalConfig::RaftEngine(RaftEngineConfig::default())
+    }
+}
+
+/// Wal config for datanode.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "provider", rename_all = "snake_case")]
+pub enum StandaloneWalConfig {
+    RaftEngine(RaftEngineConfig),
+    Kafka(StandaloneKafkaConfig),
+}
+
+impl Default for StandaloneWalConfig {
+    fn default() -> Self {
+        StandaloneWalConfig::RaftEngine(RaftEngineConfig::default())
+    }
+}
+
+/// Wal options allocated to a region.
+/// A wal options is encoded by metasrv with `serde_json::to_string`, and then decoded
+/// by datanode with `serde_json::from_str`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
+#[serde(tag = "wal.provider", rename_all = "snake_case")]
+pub enum WalOptions {
+    #[default]
+    RaftEngine,
+    #[serde(with = "prefix_wal_kafka")]
+    Kafka(KafkaWalOptions),
+}
+
+with_prefix!(prefix_wal_kafka "wal.kafka.");
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use common_base::readable_size::ReadableSize;
+    use rskafka::client::partition::Compression as RsKafkaCompression;
+
+    use crate::wal::kafka::KafkaBackoffConfig;
+    use crate::wal::{KafkaConfig, KafkaWalOptions, WalOptions};
+
+    #[test]
+    fn test_serde_kafka_config() {
+        let toml_str = r#"
+            broker_endpoints = ["127.0.0.1:9092"]
+            max_batch_size = "4MB"
+            linger = "200ms"
+            produce_record_timeout = "100ms"
+            backoff_init = "500ms"
+            backoff_max = "10s"
+            backoff_base = 2
+            backoff_deadline = "5mins"
+        "#;
+        let decoded: KafkaConfig = toml::from_str(toml_str).unwrap();
+        let expected = KafkaConfig {
+            broker_endpoints: vec!["127.0.0.1:9092".to_string()],
+            compression: RsKafkaCompression::default(),
+            max_batch_size: ReadableSize::mb(4),
+            linger: Duration::from_millis(200),
+            produce_record_timeout: Duration::from_millis(100),
+            backoff: KafkaBackoffConfig {
+                init: Duration::from_millis(500),
+                max: Duration::from_secs(10),
+                base: 2,
+                deadline: Some(Duration::from_secs(60 * 5)),
+            },
+        };
+        assert_eq!(decoded, expected);
+    }
+
+    #[test]
+    fn test_serde_wal_options() {
+        // Test serde raft-engine wal options.
+        let wal_options = WalOptions::RaftEngine;
+        let encoded = serde_json::to_string(&wal_options).unwrap();
+        let expected = r#"{"wal.provider":"raft_engine"}"#;
+        assert_eq!(&encoded, expected);
+
+        let decoded: WalOptions = serde_json::from_str(&encoded).unwrap();
+        assert_eq!(decoded, wal_options);
+
+        // Test serde kafka wal options.
+        let wal_options = WalOptions::Kafka(KafkaWalOptions {
+            topic: "test_topic".to_string(),
+        });
+        let encoded = serde_json::to_string(&wal_options).unwrap();
+        let expected = r#"{"wal.provider":"kafka","wal.kafka.topic":"test_topic"}"#;
+        assert_eq!(&encoded, expected);
+
+        let decoded: WalOptions = serde_json::from_str(&encoded).unwrap();
+        assert_eq!(decoded, wal_options);
+    }
+}

src/common/config/src/wal/kafka.rs

@@ -0,0 +1,144 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed 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.
+
+use std::time::Duration;
+
+use common_base::readable_size::ReadableSize;
+use rskafka::client::partition::Compression as RsKafkaCompression;
+use serde::{Deserialize, Serialize};
+use serde_with::with_prefix;
+
+/// Topic name prefix.
+pub const TOPIC_NAME_PREFIX: &str = "greptimedb_wal_topic";
+/// Kafka wal topic.
+pub type Topic = String;
+
+/// The type of the topic selector, i.e. with which strategy to select a topic.
+#[derive(Default, Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum TopicSelectorType {
+    #[default]
+    RoundRobin,
+}
+
+/// Configurations for kafka wal.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(default)]
+pub struct KafkaConfig {
+    /// The broker endpoints of the Kafka cluster.
+    pub broker_endpoints: Vec<String>,
+    /// The compression algorithm used to compress log entries.
+    #[serde(skip)]
+    #[serde(default)]
+    pub compression: RsKafkaCompression,
+    /// The maximum log size a kakfa batch producer could buffer.
+    pub max_batch_size: ReadableSize,
+    /// The linger duration of a kafka batch producer.
+    #[serde(with = "humantime_serde")]
+    pub linger: Duration,
+    /// The maximum amount of time (in milliseconds) to wait for Kafka records to be returned.
+    #[serde(with = "humantime_serde")]
+    pub produce_record_timeout: Duration,
+    /// The backoff config.
+    #[serde(flatten, with = "kafka_backoff")]
+    pub backoff: KafkaBackoffConfig,
+}
+
+impl Default for KafkaConfig {
+    fn default() -> Self {
+        Self {
+            broker_endpoints: vec!["127.0.0.1:9092".to_string()],
+            compression: RsKafkaCompression::NoCompression,
+            max_batch_size: ReadableSize::mb(4),
+            linger: Duration::from_millis(200),
+            produce_record_timeout: Duration::from_millis(100),
+            backoff: KafkaBackoffConfig::default(),
+        }
+    }
+}
+
+with_prefix!(pub kafka_backoff "backoff_");
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(default)]
+pub struct KafkaBackoffConfig {
+    /// The initial backoff for kafka clients.
+    #[serde(with = "humantime_serde")]
+    pub init: Duration,
+    /// The maximum backoff for kafka clients.
+    #[serde(with = "humantime_serde")]
+    pub max: Duration,
+    /// Exponential backoff rate, i.e. next backoff = base * current backoff.
+    // Sets to u32 type since some structs containing the KafkaConfig need to derive the Eq trait.
+    pub base: u32,
+    /// Stop reconnecting if the total wait time reaches the deadline.
+    /// If it's None, the reconnecting won't terminate.
+    #[serde(with = "humantime_serde")]
+    pub deadline: Option<Duration>,
+}
+
+impl Default for KafkaBackoffConfig {
+    fn default() -> Self {
+        Self {
+            init: Duration::from_millis(500),
+            max: Duration::from_secs(10),
+            base: 2,
+            deadline: Some(Duration::from_secs(60 * 5)), // 5 mins
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(default)]
+pub struct StandaloneKafkaConfig {
+    #[serde(flatten)]
+    pub base: KafkaConfig,
+    /// Number of topics to be created upon start.
+    pub num_topics: usize,
+    /// The type of the topic selector with which to select a topic for a region.
+    pub selector_type: TopicSelectorType,
+    /// Topic name prefix.
+    pub topic_name_prefix: String,
+    /// Number of partitions per topic.
+    pub num_partitions: i32,
+    /// The replication factor of each topic.
+    pub replication_factor: i16,
+    /// Above which a topic creation operation will be cancelled.
+    #[serde(with = "humantime_serde")]
+    pub create_topic_timeout: Duration,
+}
+
+impl Default for StandaloneKafkaConfig {
+    fn default() -> Self {
+        let base = KafkaConfig::default();
+        let replication_factor = base.broker_endpoints.len() as i16;
+
+        Self {
+            base,
+            num_topics: 64,
+            selector_type: TopicSelectorType::RoundRobin,
+            topic_name_prefix: "greptimedb_wal_topic".to_string(),
+            num_partitions: 1,
+            replication_factor,
+            create_topic_timeout: Duration::from_secs(30),
+        }
+    }
+}
+
+/// Kafka wal options allocated to a region.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct KafkaOptions {
+    /// Kafka wal topic.
+    pub topic: Topic,
+}

src/common/config/src/wal/raft_engine.rs

@@ -21,26 +21,19 @@ use serde::{Deserialize, Serialize};
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(default)]
 pub struct RaftEngineConfig {
-    /// Wal directory
+    // wal directory
     pub dir: Option<String>,
-    /// Wal file size in bytes
+    // wal file size in bytes
     pub file_size: ReadableSize,
-    /// Wal purge threshold in bytes
+    // wal purge threshold in bytes
     pub purge_threshold: ReadableSize,
-    /// Purge interval in seconds
+    // purge interval in seconds
     #[serde(with = "humantime_serde")]
     pub purge_interval: Duration,
-    /// Read batch size
+    // read batch size
     pub read_batch_size: usize,
-    /// Whether to sync log file after every write
+    // whether to sync log file after every write
     pub sync_write: bool,
-    /// Whether to reuse logically truncated log files.
-    pub enable_log_recycle: bool,
-    /// Whether to pre-create log files on start up
-    pub prefill_log_files: bool,
-    /// Duration for fsyncing log files.
-    #[serde(with = "humantime_serde")]
-    pub sync_period: Option<Duration>,
 }
 
 impl Default for RaftEngineConfig {
@@ -52,9 +45,6 @@ impl Default for RaftEngineConfig {
             purge_interval: Duration::from_secs(600),
             read_batch_size: 128,
             sync_write: false,
-            enable_log_recycle: true,
-            prefill_log_files: false,
-            sync_period: None,
         }
     }
 }

src/common/datasource/Cargo.toml

@@ -5,8 +5,8 @@ edition.workspace = true
 license.workspace = true
 
 [dependencies]
-arrow.workspace = true
 arrow-schema.workspace = true
+arrow.workspace = true
 async-compression = { version = "0.3", features = [
     "bzip2",
     "gzip",
@@ -19,7 +19,6 @@ async-trait.workspace = true
 bytes.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
-common-recordbatch.workspace = true
 common-runtime.workspace = true
 datafusion.workspace = true
 datatypes.workspace = true
@@ -34,8 +33,8 @@ regex = "1.7"
 serde.workspace = true
 snafu.workspace = true
 strum.workspace = true
-tokio.workspace = true
 tokio-util.workspace = true
+tokio.workspace = true
 url = "2.3"
 
 [dev-dependencies]

src/common/datasource/src/buffered_writer.rs

@@ -47,7 +47,7 @@ pub trait ArrowWriterCloser {
 impl<
         T: AsyncWrite + Send + Unpin,
         U: DfRecordBatchEncoder + ArrowWriterCloser,
-        F: Fn(String) -> Fut,
+        F: FnMut(String) -> Fut,
         Fut: Future<Output = Result<T>>,
     > LazyBufferedWriter<T, U, F>
 {
@@ -75,7 +75,7 @@ impl<
 impl<
         T: AsyncWrite + Send + Unpin,
         U: DfRecordBatchEncoder,
-        F: Fn(String) -> Fut,
+        F: FnMut(String) -> Fut,
         Fut: Future<Output = Result<T>>,
     > LazyBufferedWriter<T, U, F>
 {
@@ -149,7 +149,7 @@ impl<
         if let Some(ref mut writer) = self.writer {
             Ok(writer)
         } else {
-            let writer = (self.writer_factory)(self.path.to_string()).await?;
+            let writer = (self.writer_factory)(self.path.clone()).await?;
             Ok(self.writer.insert(writer))
         }
     }

src/common/datasource/src/file_format.rs

@@ -193,15 +193,13 @@ pub async fn stream_to_file<T: DfRecordBatchEncoder, U: Fn(SharedBuffer) -> T>(
     store: ObjectStore,
     path: &str,
     threshold: usize,
-    concurrency: usize,
     encoder_factory: U,
 ) -> Result<usize> {
     let buffer = SharedBuffer::with_capacity(threshold);
     let encoder = encoder_factory(buffer.clone());
     let mut writer = LazyBufferedWriter::new(threshold, buffer, encoder, path, |path| async {
         store
-            .writer_with(&path)
-            .concurrent(concurrency)
+            .writer(&path)
             .await
             .context(error::WriteObjectSnafu { path })
     });

src/common/datasource/src/file_format/csv.rs

@@ -193,9 +193,8 @@ pub async fn stream_to_csv(
     store: ObjectStore,
     path: &str,
     threshold: usize,
-    concurrency: usize,
 ) -> Result<usize> {
-    stream_to_file(stream, store, path, threshold, concurrency, |buffer| {
+    stream_to_file(stream, store, path, threshold, |buffer| {
         csv::Writer::new(buffer)
     })
     .await

src/common/datasource/src/file_format/json.rs

@@ -152,9 +152,8 @@ pub async fn stream_to_json(
     store: ObjectStore,
     path: &str,
     threshold: usize,
-    concurrency: usize,
 ) -> Result<usize> {
-    stream_to_file(stream, store, path, threshold, concurrency, |buffer| {
+    stream_to_file(stream, store, path, threshold, |buffer| {
         json::LineDelimitedWriter::new(buffer)
     })
     .await

src/common/datasource/src/file_format/orc.rs

@@ -12,14 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::pin::Pin;
 use std::sync::Arc;
+use std::task::{Context, Poll};
 
+use arrow::compute::cast;
 use arrow_schema::{ArrowError, Schema, SchemaRef};
 use async_trait::async_trait;
-use common_recordbatch::adapter::RecordBatchStreamTypeAdapter;
+use datafusion::arrow::record_batch::RecordBatch as DfRecordBatch;
 use datafusion::datasource::physical_plan::{FileMeta, FileOpenFuture, FileOpener};
 use datafusion::error::{DataFusionError, Result as DfResult};
-use futures::{StreamExt, TryStreamExt};
+use datafusion::physical_plan::RecordBatchStream;
+use futures::{Stream, StreamExt, TryStreamExt};
 use object_store::ObjectStore;
 use orc_rust::arrow_reader::{create_arrow_schema, Cursor};
 use orc_rust::async_arrow_reader::ArrowStreamReader;
@@ -57,6 +61,73 @@ pub async fn infer_orc_schema<R: AsyncRead + AsyncSeek + Unpin + Send + 'static>
     Ok(create_arrow_schema(&cursor))
 }
 
+pub struct OrcArrowStreamReaderAdapter<T: AsyncRead + AsyncSeek + Unpin + Send + 'static> {
+    output_schema: SchemaRef,
+    projection: Vec<usize>,
+    stream: ArrowStreamReader<T>,
+}
+
+impl<T: AsyncRead + AsyncSeek + Unpin + Send + 'static> OrcArrowStreamReaderAdapter<T> {
+    pub fn new(
+        output_schema: SchemaRef,
+        stream: ArrowStreamReader<T>,
+        projection: Option<Vec<usize>>,
+    ) -> Self {
+        let projection = if let Some(projection) = projection {
+            projection
+        } else {
+            (0..output_schema.fields().len()).collect()
+        };
+
+        Self {
+            output_schema,
+            projection,
+            stream,
+        }
+    }
+}
+
+impl<T: AsyncRead + AsyncSeek + Unpin + Send + 'static> RecordBatchStream
+    for OrcArrowStreamReaderAdapter<T>
+{
+    fn schema(&self) -> SchemaRef {
+        self.output_schema.clone()
+    }
+}
+
+impl<T: AsyncRead + AsyncSeek + Unpin + Send + 'static> Stream for OrcArrowStreamReaderAdapter<T> {
+    type Item = DfResult<DfRecordBatch>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let batch = futures::ready!(Pin::new(&mut self.stream).poll_next(cx))
+            .map(|r| r.map_err(|e| DataFusionError::External(Box::new(e))));
+
+        let projected_schema = self.output_schema.project(&self.projection)?;
+        let batch = batch.map(|b| {
+            b.and_then(|b| {
+                let mut columns = Vec::with_capacity(self.projection.len());
+                for idx in self.projection.iter() {
+                    let column = b.column(*idx);
+                    let field = self.output_schema.field(*idx);
+
+                    if column.data_type() != field.data_type() {
+                        let output = cast(&column, field.data_type())?;
+                        columns.push(output)
+                    } else {
+                        columns.push(column.clone())
+                    }
+                }
+
+                let record_batch = DfRecordBatch::try_new(projected_schema.into(), columns)?;
+
+                Ok(record_batch)
+            })
+        });
+
+        Poll::Ready(batch)
+    }
+}
+
 #[async_trait]
 impl FileFormat for OrcFormat {
     async fn infer_schema(&self, store: &ObjectStore, path: &str) -> Result<Schema> {
@@ -95,15 +166,7 @@ impl OrcOpener {
 impl FileOpener for OrcOpener {
     fn open(&self, meta: FileMeta) -> DfResult<FileOpenFuture> {
         let object_store = self.object_store.clone();
-        let projected_schema = if let Some(projection) = &self.projection {
-            let projected_schema = self
-                .output_schema
-                .project(projection)
-                .map_err(|e| DataFusionError::External(Box::new(e)))?;
-            Arc::new(projected_schema)
-        } else {
-            self.output_schema.clone()
-        };
+        let output_schema = self.output_schema.clone();
         let projection = self.projection.clone();
         Ok(Box::pin(async move {
             let reader = object_store
@@ -115,8 +178,7 @@ impl FileOpener for OrcOpener {
                 .await
                 .map_err(|e| DataFusionError::External(Box::new(e)))?;
 
-            let stream =
-                RecordBatchStreamTypeAdapter::new(projected_schema, stream_reader, projection);
+            let stream = OrcArrowStreamReaderAdapter::new(output_schema, stream_reader, projection);
 
             let adopted = stream.map_err(|e| ArrowError::ExternalError(Box::new(e)));
             Ok(adopted.boxed())

src/common/datasource/src/file_format/parquet.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::future::Future;
+use std::pin::Pin;
 use std::result;
 use std::sync::Arc;
 
@@ -29,7 +31,7 @@ use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion::physical_plan::SendableRecordBatchStream;
 use futures::future::BoxFuture;
 use futures::StreamExt;
-use object_store::{ObjectStore, Reader, Writer};
+use object_store::{ObjectStore, Reader};
 use parquet::basic::{Compression, ZstdLevel};
 use parquet::file::properties::WriterProperties;
 use snafu::ResultExt;
@@ -169,33 +171,22 @@ pub struct BufferedWriter {
 type InnerBufferedWriter = LazyBufferedWriter<
     object_store::Writer,
     ArrowWriter<SharedBuffer>,
-    impl Fn(String) -> BoxFuture<'static, Result<Writer>>,
+    Box<
+        dyn FnMut(
+                String,
+            )
+                -> Pin<Box<dyn Future<Output = error::Result<object_store::Writer>> + Send>>
+            + Send,
+    >,
 >;
 
 impl BufferedWriter {
-    fn make_write_factory(
-        store: ObjectStore,
-        concurrency: usize,
-    ) -> impl Fn(String) -> BoxFuture<'static, Result<Writer>> {
-        move |path| {
-            let store = store.clone();
-            Box::pin(async move {
-                store
-                    .writer_with(&path)
-                    .concurrent(concurrency)
-                    .await
-                    .context(error::WriteObjectSnafu { path })
-            })
-        }
-    }
-
     pub async fn try_new(
         path: String,
         store: ObjectStore,
         arrow_schema: SchemaRef,
         props: Option<WriterProperties>,
         buffer_threshold: usize,
-        concurrency: usize,
     ) -> error::Result<Self> {
         let buffer = SharedBuffer::with_capacity(buffer_threshold);
 
@@ -208,7 +199,15 @@ impl BufferedWriter {
                 buffer,
                 arrow_writer,
                 &path,
-                Self::make_write_factory(store, concurrency),
+                Box::new(move |path| {
+                    let store = store.clone();
+                    Box::pin(async move {
+                        store
+                            .writer(&path)
+                            .await
+                            .context(error::WriteObjectSnafu { path })
+                    })
+                }),
             ),
         })
     }
@@ -237,7 +236,6 @@ pub async fn stream_to_parquet(
     store: ObjectStore,
     path: &str,
     threshold: usize,
-    concurrency: usize,
 ) -> Result<usize> {
     let write_props = WriterProperties::builder()
         .set_compression(Compression::ZSTD(ZstdLevel::default()))
@@ -249,7 +247,6 @@ pub async fn stream_to_parquet(
         schema,
         Some(write_props),
         threshold,
-        concurrency,
     )
     .await?;
     let mut rows_written = 0;

src/common/datasource/src/lib.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 #![feature(assert_matches)]
-#![feature(type_alias_impl_trait)]
 
 pub mod buffered_writer;
 pub mod compression;