Merge branch 'main' into import_upper_optimizer_for_merge_into

.github/workflows/release.yml

@@ -1,17 +1,18 @@
 name: Release
 
 on:
-  push:
-    tags:
-      - "v*"
   schedule:
     - cron: "0 22 * * *"
   workflow_dispatch:
     inputs:
-      tags:
+      tag:
         description: The tags to be released
         required: false
         type: string
+      stable:
+        description: Make a stable release
+        required: false
+        type: boolean
 
 permissions:
   id-token: write
@@ -28,61 +29,89 @@ jobs:
     name: create release
     runs-on: ubuntu-latest
     outputs:
-      version: ${{ steps.generated-tag.outputs.tag }}
+      version: ${{ steps.bump.outputs.tag }}
+      sha: ${{ steps.bump.outputs.sha }}
     steps:
       - name: Checkout
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-      - name: Get latest tag
-        id: get-latest-tag
-        env:
-          GH_TOKEN: ${{ github.token }}
-        run: |
-          echo "tag=`gh release list -L 1 | cut -f 1`" >> $GITHUB_OUTPUT
       - name: Bump version
-        id: generated-tag
-        uses: actions/github-script@v6
+        id: bump
+        uses: actions/github-script@v7
         with:
           script: |
             if (context.ref.startsWith("refs/tags/")) {
               let tag = context.ref.replace("refs/tags/", "");
               core.setOutput('tag', tag);
-              console.log(`This event pushed a tag ${tag}, return directly.`)
+              core.info(`Tag event triggered by ${tag}.`);
               return
             }
-
-            if ("${{ github.event.inputs.tags }}") {
-              let tag = "${{ github.event.inputs.tags }}";
-              core.setOutput('tag', tag);
-              console.log(`This event triggered by workflow_dispatch with a tag ${tag}, return directly.`)
-              return
-            }
-
-            let tag = "${{ steps.get-latest-tag.outputs.tag }}";
-            let result = /v(\d+)\.(\d+)\.(\d+)/g.exec(tag);
-            if (result === null) {
-              throw `The previous tag ${{ steps.get-latest-tag.outputs.tag }} is invalid, ignoring`;
+            if ("${{ inputs.stable }}" == "true") {
+              if ("${{ inputs.tag }}") {
+                // trigger stable release by workflow_dispatch with a tag
+                let tag = "${{ inputs.tag }}";
+                let result = /v(\d+)\.(\d+)\.(\d+)-nightly/g.exec(tag);
+                if (result === null) {
+                  core.setFailed(`The tag ${tag} to stablize is invalid, ignoring`);
+                  return
+                }
+                let major = result[1];
+                let minor = result[2];
+                let patch = result[3];
+                let stable_tag = `v${major}.${minor}.${patch}`;
+                core.setOutput('tag', stable_tag);
+                let ref = await github.rest.git.getRef({
+                  owner: context.repo.owner,
+                  repo: context.repo.repo,
+                  ref: `tags/${tag}`,
+                });
+                core.setOutput('sha', ref.data.object.sha);
+                core.info(`Stable release ${stable_tag} from ${tag} (${ref.data.object.sha})`);
+              } else {
+                core.setFailed("Stable release must be triggered with a nightly tag")
+              }
+            } else {
+              if ("${{ inputs.tag }}") {
+                let tag = "${{ inputs.tag }}";
+                core.setOutput('tag', tag);
+                core.info(`Release create manually with tag ${tag}`);
+              } else {
+                let releases = await github.rest.repos.listReleases({
+                  owner: context.repo.owner,
+                  repo: context.repo.repo,
+                  per_page: 1,
+                });
+                let tag = releases.data[0].tag_name;
+                let result = /v(\d+)\.(\d+)\.(\d+)/g.exec(tag);
+                if (result === null) {
+                  core.setFailed(`The previous tag ${tag} is invalid, ignoring`);
+                  return
+                }
+                let major = result[1];
+                let minor = result[2];
+                let patch = (parseInt(result[3]) + 1).toString();
+                let next_tag = `v${major}.${minor}.${patch}-nightly`;
+                core.setOutput('tag', next_tag);
+                core.setOutput('sha', context.sha);
+                core.info(`Nightly release ${next_tag} from ${tag} (${context.sha})`);
+              }
             }
-
-            let major = result[1];
-            let minor = result[2];
-            let patch = (parseInt(result[3]) + 1).toString();
-            let next_tag = `v${major}.${minor}.${patch}-nightly`;
-
-            console.log(`This event is triggered, return generated ${next_tag}.`)
-            core.setOutput('tag', next_tag)
-      - name: Create github release if not exist
-        # Only create release when the tag is not exist
-        if: 'steps.generated-tag.outputs.tag != steps.get-latest-tag.outputs.tag'
-        # Allow this action failure
-        # continue-on-error: true
-        # Reference: https://cli.github.com/manual/gh_release_create
+      - name: Create release
         env:
-          GH_TOKEN: ${{ github.token }}
+          # we need workflow:write permission to create release if there were any workflow changes
+          # which is not possible for github actions token
+          GH_TOKEN: ${{ secrets.DATABEND_BOT_TOKEN }}
         run: |
-          echo "Create a release for ${{ steps.generated-tag.outputs.tag }}"
-          gh release create ${{ steps.generated-tag.outputs.tag }} --generate-notes -p
+          echo "Creating release ${{ steps.bump.outputs.tag }} from ${{ steps.bump.outputs.sha }}"
+          if [ "${{ inputs.stable }}" == "true" ]; then
+            echo "Stable release"
+            gh release create ${{ steps.bump.outputs.tag }} --target ${{ steps.bump.outputs.sha }} --generate-notes --latest
+          else
+            echo "Nightly release"
+            gh release create ${{ steps.bump.outputs.tag }} --target ${{ steps.bump.outputs.sha }} --generate-notes --prerelease
+          fi
+
 
   changelog:
     runs-on: ubuntu-latest
@@ -134,6 +163,7 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
+          ref: ${{ needs.create_release.outputs.version }}
       - name: Get target
         id: target
         run: echo 'target=${{ matrix.arch }}-apple-darwin' >> $GITHUB_OUTPUT
@@ -209,6 +239,7 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
+          ref: ${{ needs.create_release.outputs.version }}
       - name: Get the version
         id: get_version
         run: echo "VERSION=${GITHUB_REF/refs\/tags\//}" >> $GITHUB_OUTPUT
@@ -299,6 +330,7 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
+          ref: ${{ needs.create_release.outputs.version }}
       - name: Get the version
         id: get_version
         run: echo "VERSION=${GITHUB_REF/refs\/tags\//}" >> $GITHUB_OUTPUT
@@ -358,6 +390,8 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v4
+        with:
+          ref: ${{ needs.create_release.outputs.version }}
       - name: Get the version
         id: get_version
         run: echo "VERSION=${GITHUB_REF/refs\/tags\//}" >> $GITHUB_OUTPUT
@@ -425,6 +459,8 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v4
+        with:
+          ref: ${{ needs.create_release.outputs.version }}
       - name: Get the version
         id: get_version
         run: echo "VERSION=${GITHUB_REF/refs\/tags\//}" >> $GITHUB_OUTPUT
@@ -488,6 +524,8 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v4
+        with:
+          ref: ${{ needs.create_release.outputs.version }}
       - name: Install nfpm@latest
         run: |
           curl -sSLo nfpm.tar.gz https://github.com/goreleaser/nfpm/releases/download/v2.26.0/nfpm_2.26.0_Linux_x86_64.tar.gz
@@ -539,6 +577,8 @@ jobs:
     needs: [create_release, distribution]
     steps:
       - uses: actions/checkout@v4
+        with:
+          ref: ${{ needs.create_release.outputs.version }}
       - uses: ./.github/actions/publish_deb
         env:
           GH_TOKEN: ${{ github.token }}
@@ -556,6 +596,8 @@ jobs:
     steps:
       - name: checkout
         uses: actions/checkout@v4
+        with:
+          ref: ${{ needs.create_release.outputs.version }}
       - name: download sha256sums
         uses: actions/download-artifact@v3
         with:
@@ -591,6 +633,8 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
+        with:
+          ref: ${{ needs.create_release.outputs.version }}
       - name: Download artifact for release
         env:
           GH_TOKEN: ${{ github.token }}
@@ -634,6 +678,8 @@ jobs:
       - deb
     steps:
       - uses: actions/checkout@v4
+        with:
+          ref: ${{ needs.create_release.outputs.version }}
       - run: |
           status="${{ (contains(needs.*.result, 'failure') && 'failure') || (contains(needs.*.result, 'cancelled') && 'cancelled') || 'success' }}"
           jq -n -f .github/release-report.jq \

src/common/metrics/src/metrics/transform.rs

@@ -155,3 +155,43 @@ pub fn metrics_inc_exchange_read_count(v: usize) {
 pub fn metrics_inc_exchange_read_bytes(c: usize) {
     EXCHANGE_READ_BYTES.inc_by(c as u64);
 }
+
+// Sort spill metrics
+pub fn metrics_inc_sort_spill_count() {
+    let labels = &vec![("spill", "sort_spill".to_string())];
+    SPILL_COUNT.get_or_create(labels).inc();
+}
+
+pub fn metrics_inc_sort_spill_write_count() {
+    let labels = &vec![("spill", "sort_spill".to_string())];
+    SPILL_WRITE_COUNT.get_or_create(labels).inc();
+}
+
+pub fn metrics_inc_sort_spill_write_bytes(c: u64) {
+    let labels = &vec![("spill", "sort_spill".to_string())];
+    SPILL_WRITE_BYTES.get_or_create(labels).inc_by(c);
+}
+
+pub fn metrics_inc_sort_spill_write_milliseconds(c: u64) {
+    let labels = &vec![("spill", "sort_spill".to_string())];
+    SPILL_WRITE_MILLISECONDS
+        .get_or_create(labels)
+        .observe(c as f64)
+}
+
+pub fn metrics_inc_sort_spill_read_count() {
+    let labels = &vec![("spill", "sort_spill".to_string())];
+    SPILL_READ_COUNT.get_or_create(labels).inc();
+}
+
+pub fn metrics_inc_sort_spill_read_bytes(c: u64) {
+    let labels = &vec![("spill", "sort_spill".to_string())];
+    SPILL_READ_BYTES.get_or_create(labels).inc_by(c);
+}
+
+pub fn metrics_inc_sort_spill_read_milliseconds(c: u64) {
+    let labels = &vec![("spill", "sort_spill".to_string())];
+    SPILL_READ_MILLISECONDS
+        .get_or_create(labels)
+        .observe(c as f64);
+}

src/query/expression/src/block.rs

@@ -383,6 +383,11 @@ impl DataBlock {
         })
     }
 
+    #[inline]
+    pub fn replace_meta(&mut self, meta: BlockMetaInfoPtr) {
+        self.meta.replace(meta);
+    }
+
     #[inline]
     pub fn get_meta(&self) -> Option<&BlockMetaInfoPtr> {
         self.meta.as_ref()

src/query/functions/src/scalars/decimal.rs

@@ -656,7 +656,7 @@ pub(crate) fn register_decimal_compare_op(registry: &mut FunctionRegistry) {
     register_decimal_compare_op!(registry, "gt", is_gt, domain_gt);
     register_decimal_compare_op!(registry, "lte", is_le, domain_lte);
     register_decimal_compare_op!(registry, "gte", is_ge, domain_gte);
-    register_decimal_compare_op!(registry, "ne", is_ne, domain_noteq);
+    register_decimal_compare_op!(registry, "noteq", is_ne, domain_noteq);
 }
 
 pub(crate) fn register_decimal_arithmetic(registry: &mut FunctionRegistry) {

src/query/functions/tests/it/scalars/testdata/comparison.txt

@@ -301,7 +301,7 @@ output         : true
 
 ast            : 1.1!=1.1
 raw expr       : noteq(1.1, 1.1)
-checked expr   : noteq<Float32, Float32>(to_float32<Decimal(2, 1)>(1.1_d128(2,1)), to_float32<Decimal(2, 1)>(1.1_d128(2,1)))
+checked expr   : noteq<Decimal(2, 1), Decimal(2, 1)>(1.1_d128(2,1), 1.1_d128(2,1))
 optimized expr : false
 output type    : Boolean
 output domain  : {FALSE}

src/query/functions/tests/it/scalars/testdata/function_list.txt

@@ -2719,7 +2719,6 @@ Functions overloads:
 198 multiply(Float64 NULL, Float32 NULL) :: Float64 NULL
 199 multiply(Float64, Float64) :: Float64
 200 multiply(Float64 NULL, Float64 NULL) :: Float64 NULL
-0 ne FACTORY
 0 not(Boolean) :: Boolean
 1 not(Boolean NULL) :: Boolean NULL
 0 noteq(Variant, Variant) :: Boolean
@@ -2746,17 +2745,18 @@ Functions overloads:
 21 noteq(UInt64 NULL, UInt64 NULL) :: Boolean NULL
 22 noteq(Int64, Int64) :: Boolean
 23 noteq(Int64 NULL, Int64 NULL) :: Boolean NULL
-24 noteq(Float32, Float32) :: Boolean
-25 noteq(Float32 NULL, Float32 NULL) :: Boolean NULL
-26 noteq(Float64, Float64) :: Boolean
-27 noteq(Float64 NULL, Float64 NULL) :: Boolean NULL
-28 noteq(Boolean, Boolean) :: Boolean
-29 noteq(Boolean NULL, Boolean NULL) :: Boolean NULL
-30 noteq(Array(Nothing), Array(Nothing)) :: Boolean
-31 noteq(Array(Nothing) NULL, Array(Nothing) NULL) :: Boolean NULL
-32 noteq(Array(T0), Array(T0)) :: Boolean
-33 noteq(Array(T0) NULL, Array(T0) NULL) :: Boolean NULL
-34 noteq FACTORY
+24 noteq FACTORY
+25 noteq(Float32, Float32) :: Boolean
+26 noteq(Float32 NULL, Float32 NULL) :: Boolean NULL
+27 noteq(Float64, Float64) :: Boolean
+28 noteq(Float64 NULL, Float64 NULL) :: Boolean NULL
+29 noteq(Boolean, Boolean) :: Boolean
+30 noteq(Boolean NULL, Boolean NULL) :: Boolean NULL
+31 noteq(Array(Nothing), Array(Nothing)) :: Boolean
+32 noteq(Array(Nothing) NULL, Array(Nothing) NULL) :: Boolean NULL
+33 noteq(Array(T0), Array(T0)) :: Boolean
+34 noteq(Array(T0) NULL, Array(T0) NULL) :: Boolean NULL
+35 noteq FACTORY
 0 now() :: Timestamp
 0 oct(Int64) :: String
 1 oct(Int64 NULL) :: String NULL

src/query/pipeline/transforms/Cargo.toml

@@ -19,6 +19,8 @@ async-backtrace = { workspace = true }
 async-trait = { workspace = true }
 jsonb = { workspace = true }
 match-template = { workspace = true }
+serde = { workspace = true }
+typetag = { workspace = true }
 
 [package.metadata.cargo-machete]
 ignored = ["match-template"]

src/query/pipeline/transforms/src/processors/transforms/mod.rs

@@ -40,5 +40,6 @@ pub use transform_dummy::*;
 pub use transform_multi_sort_merge::try_add_multi_sort_merge;
 pub use transform_sort_merge::sort_merge;
 pub use transform_sort_merge::*;
+pub use transform_sort_merge_base::*;
 pub use transform_sort_merge_limit::*;
 pub use transform_sort_partial::*;

src/query/pipeline/transforms/src/processors/transforms/sort/mod.rs

@@ -14,7 +14,9 @@
 
 mod cursor;
 mod rows;
+mod spill;
 pub mod utils;
 
 pub use cursor::*;
 pub use rows::*;
+pub use spill::*;