fix(mito): avoid shortcut in picking multi window files (#7174) (#7224)

* fix(mito): avoid shortcut in picking multi window files (#7174) * fix/pick-continue: ### Add Tests for TWCS Compaction Logic - **`twcs.rs`**: - Modified the logic in `TwcsPicker` to handle cases with zero runs by using `continue` instead of `return`. - Added two new test cases: `test_build_output_multiple_windows_with_zero_runs` and `test_build_output_single_window_zero_runs` to verify the behavior of the compaction logic when there are zero runs in the windows. - **`memtable_util.rs`**: - Removed unused import `PredicateGroup`. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * fix: clippy Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * fix/pick-continue: * refactor/progressive-compaction: **Enhance Compaction Task Error Handling** - Updated `task.rs` to conditionally execute the removal of expired SST files only when they exist, improving error handling and performance. - Added a check for non-empty `expired_ssts` before initiating the removal process, ensuring unnecessary operations are avoided. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * refactor/progressive-compaction: ### Add Max Background Compaction Tasks Configuration - **`compaction.rs`**: Added `max_background_compactions` to the compaction scheduler to limit background tasks. - **`compaction/compactor.rs`**: Removed immediate manifest update logic after task completion. - **`compaction/picker.rs`**: Introduced `max_background_tasks` parameter in `new_picker` to control task limits. - **`compaction/twcs.rs`**: Updated `TwcsPicker` to include `max_background_tasks` and truncate inputs exceeding this limit. Added related test cases to ensure functionality. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * fix/pick-continue: ### Add Unit Tests for Compaction Task and TWCS Picker - **`twcs.rs`**: Introduced tests for `TwcsPicker` to ensure correct handling of `max_background_tasks` during compaction, including scenarios with and without task truncation. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> --------- Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> Signed-off-by: evenyag <realevenyag@gmail.com> * chore: fix typos Signed-off-by: evenyag <realevenyag@gmail.com> * chore: update bitnami config (#6847) * chore: update bitnami config Signed-off-by: liyang <daviderli614@gmail.com> * update postgresql chart version Signed-off-by: liyang <daviderli614@gmail.com> * fix ci Signed-off-by: liyang <daviderli614@gmail.com> * refactor: add pull-test-deps-images.sh to pull images one by one to avoid rate limit Signed-off-by: zyy17 <zyylsxm@gmail.com> --------- Signed-off-by: liyang <daviderli614@gmail.com> Signed-off-by: zyy17 <zyylsxm@gmail.com> Co-authored-by: zyy17 <zyylsxm@gmail.com> Signed-off-by: evenyag <realevenyag@gmail.com> * chore: use greptime dockerhub image (#6865) Signed-off-by: liyang <daviderli614@gmail.com> Signed-off-by: evenyag <realevenyag@gmail.com> * ci: remove etcd-tls in fixtures Signed-off-by: evenyag <realevenyag@gmail.com> --------- Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> Signed-off-by: evenyag <realevenyag@gmail.com> Signed-off-by: liyang <daviderli614@gmail.com> Signed-off-by: zyy17 <zyylsxm@gmail.com> Co-authored-by: Lei, HUANG <6406592+v0y4g3r@users.noreply.github.com> Co-authored-by: liyang <daviderli614@gmail.com> Co-authored-by: zyy17 <zyylsxm@gmail.com>
2026-05-19 22:40:40 +00:00 · 2025-11-14 15:06:51 +08:00
parent 1048339b06
commit b08bdcb465
18 changed files with 340 additions and 67 deletions
--- a/.github/actions/setup-etcd-cluster/action.yml
+++ b/.github/actions/setup-etcd-cluster/action.yml
@@ -12,7 +12,7 @@ runs:
  steps:
  - name: Install Etcd cluster
    shell: bash
-    run: | 
+    run: |
      helm upgrade \
        --install etcd oci://registry-1.docker.io/bitnamicharts/etcd \
        --set replicaCount=${{ inputs.etcd-replicas }} \
@@ -24,4 +24,9 @@ runs:
        --set auth.rbac.token.enabled=false \
        --set persistence.size=2Gi \
        --create-namespace \
+        --set global.security.allowInsecureImages=true \
+        --set image.registry=docker.io \
+        --set image.repository=greptime/etcd \
+        --set image.tag=3.6.1-debian-12-r3 \
+        --version 12.0.8 \
        -n ${{ inputs.namespace }}
--- a/.github/actions/setup-kafka-cluster/action.yml
+++ b/.github/actions/setup-kafka-cluster/action.yml
@@ -12,7 +12,7 @@ runs:
  steps:
  - name: Install Kafka cluster
    shell: bash
-    run: | 
+    run: |
      helm upgrade \
        --install kafka oci://registry-1.docker.io/bitnamicharts/kafka \
        --set controller.replicaCount=${{ inputs.controller-replicas }} \
@@ -23,4 +23,8 @@ runs:
        --set listeners.controller.protocol=PLAINTEXT \
        --set listeners.client.protocol=PLAINTEXT \
        --create-namespace \
+        --set image.registry=docker.io \
+        --set image.repository=greptime/kafka \
+        --set image.tag=3.9.0-debian-12-r1 \
+        --version 31.0.0 \
        -n ${{ inputs.namespace }}
--- a/.github/actions/setup-postgres-cluster/action.yml
+++ b/.github/actions/setup-postgres-cluster/action.yml
@@ -6,9 +6,7 @@ inputs:
    description: "Number of PostgreSQL replicas"
  namespace:
    default: "postgres-namespace"
-  postgres-version:
-    default: "14.2"
-    description: "PostgreSQL version"
+    description: "The PostgreSQL namespace"
  storage-size:
    default: "1Gi"
    description: "Storage size for PostgreSQL"
@@ -22,7 +20,11 @@ runs:
      helm upgrade \
        --install postgresql oci://registry-1.docker.io/bitnamicharts/postgresql \
        --set replicaCount=${{ inputs.postgres-replicas }} \
-        --set image.tag=${{ inputs.postgres-version }} \
+        --set global.security.allowInsecureImages=true \
+        --set image.registry=docker.io \
+        --set image.repository=greptime/postgresql \
+        --set image.tag=17.5.0-debian-12-r3 \
+        --version 16.7.4 \
        --set persistence.size=${{ inputs.storage-size }} \
        --set postgresql.username=greptimedb \
        --set postgresql.password=admin \
--- a/.github/scripts/pull-test-deps-images.sh
+++ b/.github/scripts/pull-test-deps-images.sh
@@ -0,0 +1,34 @@
+#!/bin/bash
+
+# This script is used to pull the test dependency images that are stored in public ECR one by one to avoid rate limiting.
+
+set -e
+
+MAX_RETRIES=3
+
+IMAGES=(
+  "greptime/zookeeper:3.7"
+  "greptime/kafka:3.9.0-debian-12-r1"
+  "greptime/etcd:3.6.1-debian-12-r3"
+  "greptime/minio:2024"
+  "greptime/mysql:5.7"
+)
+
+for image in "${IMAGES[@]}"; do
+  for ((attempt=1; attempt<=MAX_RETRIES; attempt++)); do
+    if docker pull "$image"; then
+      # Successfully pulled the image.
+      break
+    else
+      # Use some simple exponential backoff to avoid rate limiting.
+      if [ $attempt -lt $MAX_RETRIES ]; then
+        sleep_seconds=$((attempt * 5))
+        echo "Attempt $attempt failed for $image, waiting $sleep_seconds seconds"
+        sleep $sleep_seconds  # 5s, 10s delays
+      else
+        echo "Failed to pull $image after $MAX_RETRIES attempts"
+        exit 1
+      fi
+    fi
+  done
+done
--- a/.github/workflows/develop.yml
+++ b/.github/workflows/develop.yml
@@ -719,6 +719,10 @@ jobs:
          save-if: ${{ github.ref == 'refs/heads/main' }}
      - name: Install latest nextest release
        uses: taiki-e/install-action@nextest
+
+      - name: Pull test dependencies images
+        run: ./.github/scripts/pull-test-deps-images.sh
+
      - name: Setup external services
        working-directory: tests-integration/fixtures
        run: docker compose up -d --wait
--- a/src/mito2/src/compaction.rs
+++ b/src/mito2/src/compaction.rs
@@ -294,6 +294,7 @@ impl CompactionScheduler {
            &options,
            &request.current_version.options.compaction,
            request.current_version.options.append_mode,
+            Some(self.engine_config.max_background_compactions),
        );
        let region_id = request.region_id();
        let CompactionRequest {
--- a/src/mito2/src/compaction/compactor.rs
+++ b/src/mito2/src/compaction/compactor.rs
@@ -451,6 +451,7 @@ impl Compactor for DefaultCompactor {
                &compact_request_options,
                &compaction_region.region_options.compaction,
                compaction_region.region_options.append_mode,
+                None,
            )
            .pick(compaction_region);

--- a/src/mito2/src/compaction/picker.rs
+++ b/src/mito2/src/compaction/picker.rs
@@ -125,6 +125,7 @@ pub fn new_picker(
    compact_request_options: &compact_request::Options,
    compaction_options: &CompactionOptions,
    append_mode: bool,
+    max_background_tasks: Option<usize>,
 ) -> Arc<dyn Picker> {
    if let compact_request::Options::StrictWindow(window) = compact_request_options {
        let window = if window.window_seconds == 0 {
@@ -140,6 +141,7 @@ pub fn new_picker(
                time_window_seconds: twcs_opts.time_window_seconds(),
                max_output_file_size: twcs_opts.max_output_file_size.map(|r| r.as_bytes()),
                append_mode,
+                max_background_tasks,
            }) as Arc<_>,
        }
    }
--- a/src/mito2/src/compaction/twcs.rs
+++ b/src/mito2/src/compaction/twcs.rs
@@ -53,6 +53,8 @@ pub struct TwcsPicker {
    pub max_output_file_size: Option<u64>,
    /// Whether the target region is in append mode.
    pub append_mode: bool,
+    /// Max background compaction tasks.
+    pub max_background_tasks: Option<usize>,
 }

 impl TwcsPicker {
@@ -105,7 +107,7 @@ impl TwcsPicker {
            // because after compaction there will be no overlapping files.
            let filter_deleted = !files.overlapping && found_runs <= 2 && !self.append_mode;
            if found_runs == 0 {
-                return output;
+                continue;
            }

            let mut inputs = if found_runs > 1 {
@@ -155,6 +157,16 @@ impl TwcsPicker {
                    filter_deleted,
                    output_time_range: None, // we do not enforce output time range in twcs compactions.
                });
+
+                if let Some(max_background_tasks) = self.max_background_tasks
+                    && output.len() >= max_background_tasks
+                {
+                    debug!(
+                        "Region ({:?}) compaction task size larger than max background tasks({}), remaining tasks discarded",
+                        region_id, max_background_tasks
+                    );
+                    break;
+                }
            }
        }
        output
@@ -704,6 +716,7 @@ mod tests {
                time_window_seconds: None,
                max_output_file_size: None,
                append_mode: false,
+                max_background_tasks: None,
            }
            .build_output(RegionId::from_u64(0), &mut windows, active_window);

@@ -855,5 +868,185 @@ mod tests {
        }
    }

+    #[test]
+    fn test_build_output_multiple_windows_with_zero_runs() {
+        let file_ids = (0..6).map(|_| FileId::random()).collect::<Vec<_>>();
+
+        let files = [
+            // Window 0: Contains 3 files but not forming any runs (not enough files in sequence to reach trigger_file_num)
+            new_file_handle_with_sequence(file_ids[0], 0, 999, 0, 1),
+            new_file_handle_with_sequence(file_ids[1], 0, 999, 0, 2),
+            new_file_handle_with_sequence(file_ids[2], 0, 999, 0, 3),
+            // Window 3: Contains files that will form 2 runs
+            new_file_handle_with_sequence(file_ids[3], 3000, 3999, 0, 4),
+            new_file_handle_with_sequence(file_ids[4], 3000, 3999, 0, 5),
+            new_file_handle_with_sequence(file_ids[5], 3000, 3999, 0, 6),
+        ];
+
+        let mut windows = assign_to_windows(files.iter(), 3);
+
+        // Create picker with trigger_file_num of 4 so single files won't form runs in first window
+        let picker = TwcsPicker {
+            trigger_file_num: 4, // High enough to prevent runs in first window
+            time_window_seconds: Some(3),
+            max_output_file_size: None,
+            append_mode: false,
+            max_background_tasks: None,
+        };
+
+        let active_window = find_latest_window_in_seconds(files.iter(), 3);
+        let output = picker.build_output(RegionId::from_u64(123), &mut windows, active_window);
+
+        assert!(
+            !output.is_empty(),
+            "Should have output from windows with runs, even when one window has 0 runs"
+        );
+
+        let all_output_files: Vec<_> = output
+            .iter()
+            .flat_map(|o| o.inputs.iter())
+            .map(|f| f.file_id())
+            .collect();
+
+        assert!(
+            all_output_files.contains(&file_ids[3])
+                || all_output_files.contains(&file_ids[4])
+                || all_output_files.contains(&file_ids[5]),
+            "Output should contain files from the window with runs"
+        );
+    }
+
+    #[test]
+    fn test_build_output_single_window_zero_runs() {
+        let file_ids = (0..2).map(|_| FileId::random()).collect::<Vec<_>>();
+
+        let large_file_1 = new_file_handle_with_size_and_sequence(file_ids[0], 0, 999, 0, 1, 2000); // 2000 bytes
+        let large_file_2 = new_file_handle_with_size_and_sequence(file_ids[1], 0, 999, 0, 2, 2500); // 2500 bytes
+
+        let files = [large_file_1, large_file_2];
+
+        let mut windows = assign_to_windows(files.iter(), 3);
+
+        let picker = TwcsPicker {
+            trigger_file_num: 2,
+            time_window_seconds: Some(3),
+            max_output_file_size: Some(1000),
+            append_mode: true,
+            max_background_tasks: None,
+        };
+
+        let active_window = find_latest_window_in_seconds(files.iter(), 3);
+        let output = picker.build_output(RegionId::from_u64(456), &mut windows, active_window);
+
+        // Should return empty output (no compaction needed)
+        assert!(
+            output.is_empty(),
+            "Should return empty output when no runs are found after filtering"
+        );
+    }
+
+    #[test]
+    fn test_max_background_tasks_truncation() {
+        let file_ids = (0..10).map(|_| FileId::random()).collect::<Vec<_>>();
+        let max_background_tasks = 3;
+
+        // Create files across multiple windows that will generate multiple compaction outputs
+        let files = [
+            // Window 0: 4 files that will form a run
+            new_file_handle_with_sequence(file_ids[0], 0, 999, 0, 1),
+            new_file_handle_with_sequence(file_ids[1], 0, 999, 0, 2),
+            new_file_handle_with_sequence(file_ids[2], 0, 999, 0, 3),
+            new_file_handle_with_sequence(file_ids[3], 0, 999, 0, 4),
+            // Window 3: 4 files that will form another run
+            new_file_handle_with_sequence(file_ids[4], 3000, 3999, 0, 5),
+            new_file_handle_with_sequence(file_ids[5], 3000, 3999, 0, 6),
+            new_file_handle_with_sequence(file_ids[6], 3000, 3999, 0, 7),
+            new_file_handle_with_sequence(file_ids[7], 3000, 3999, 0, 8),
+            // Window 6: 4 files that will form another run
+            new_file_handle_with_sequence(file_ids[8], 6000, 6999, 0, 9),
+            new_file_handle_with_sequence(file_ids[9], 6000, 6999, 0, 10),
+        ];
+
+        let mut windows = assign_to_windows(files.iter(), 3);
+
+        let picker = TwcsPicker {
+            trigger_file_num: 4,
+            time_window_seconds: Some(3),
+            max_output_file_size: None,
+            append_mode: false,
+            max_background_tasks: Some(max_background_tasks),
+        };
+
+        let active_window = find_latest_window_in_seconds(files.iter(), 3);
+        let output = picker.build_output(RegionId::from_u64(123), &mut windows, active_window);
+
+        // Should have at most max_background_tasks outputs
+        assert!(
+            output.len() <= max_background_tasks,
+            "Output should be truncated to max_background_tasks: expected <= {}, got {}",
+            max_background_tasks,
+            output.len()
+        );
+
+        // Without max_background_tasks, should have more outputs
+        let picker_no_limit = TwcsPicker {
+            trigger_file_num: 4,
+            time_window_seconds: Some(3),
+            max_output_file_size: None,
+            append_mode: false,
+            max_background_tasks: None,
+        };
+
+        let mut windows_no_limit = assign_to_windows(files.iter(), 3);
+        let output_no_limit = picker_no_limit.build_output(
+            RegionId::from_u64(123),
+            &mut windows_no_limit,
+            active_window,
+        );
+
+        // Without limit, should have more outputs (if there are enough windows)
+        if output_no_limit.len() > max_background_tasks {
+            assert!(
+                output_no_limit.len() > output.len(),
+                "Without limit should have more outputs than with limit"
+            );
+        }
+    }
+
+    #[test]
+    fn test_max_background_tasks_no_truncation_when_under_limit() {
+        let file_ids = (0..4).map(|_| FileId::random()).collect::<Vec<_>>();
+        let max_background_tasks = 10; // Larger than expected outputs
+
+        // Create files in one window that will generate one compaction output
+        let files = [
+            new_file_handle_with_sequence(file_ids[0], 0, 999, 0, 1),
+            new_file_handle_with_sequence(file_ids[1], 0, 999, 0, 2),
+            new_file_handle_with_sequence(file_ids[2], 0, 999, 0, 3),
+            new_file_handle_with_sequence(file_ids[3], 0, 999, 0, 4),
+        ];
+
+        let mut windows = assign_to_windows(files.iter(), 3);
+
+        let picker = TwcsPicker {
+            trigger_file_num: 4,
+            time_window_seconds: Some(3),
+            max_output_file_size: None,
+            append_mode: false,
+            max_background_tasks: Some(max_background_tasks),
+        };
+
+        let active_window = find_latest_window_in_seconds(files.iter(), 3);
+        let output = picker.build_output(RegionId::from_u64(123), &mut windows, active_window);
+
+        // Should have all outputs since we're under the limit
+        assert!(
+            output.len() <= max_background_tasks,
+            "Output should be within limit"
+        );
+        // Should have at least one output
+        assert!(!output.is_empty(), "Should have at least one output");
+    }
+
    // TODO(hl): TTL tester that checks if get_expired_ssts function works as expected.
 }
--- a/src/mito2/src/sst/index/inverted_index/applier/builder.rs
+++ b/src/mito2/src/sst/index/inverted_index/applier/builder.rs
@@ -338,7 +338,7 @@ mod tests {

    #[test]
    fn test_collect_and_basic() {
-        let (_d, facotry) = PuffinManagerFactory::new_for_test_block("test_collect_and_basic_");
+        let (_d, factory) = PuffinManagerFactory::new_for_test_block("test_collect_and_basic_");

        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -346,7 +346,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let expr = Expr::BinaryExpr(BinaryExpr {
--- a/src/mito2/src/sst/index/inverted_index/applier/builder/between.rs
+++ b/src/mito2/src/sst/index/inverted_index/applier/builder/between.rs
@@ -70,14 +70,14 @@ mod tests {

    #[test]
    fn test_collect_between_basic() {
-        let (_d, facotry) = PuffinManagerFactory::new_for_test_block("test_collect_between_basic_");
+        let (_d, factory) = PuffinManagerFactory::new_for_test_block("test_collect_between_basic_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
            "test".to_string(),
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let between = Between {
@@ -110,7 +110,7 @@ mod tests {

    #[test]
    fn test_collect_between_negated() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_between_negated_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -118,7 +118,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let between = Between {
@@ -134,7 +134,7 @@ mod tests {

    #[test]
    fn test_collect_between_field_column() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_between_field_column_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -142,7 +142,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let between = Between {
@@ -175,7 +175,7 @@ mod tests {

    #[test]
    fn test_collect_between_type_mismatch() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_between_type_mismatch_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -183,7 +183,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let between = Between {
@@ -200,7 +200,7 @@ mod tests {

    #[test]
    fn test_collect_between_nonexistent_column() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_between_nonexistent_column_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -208,7 +208,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let between = Between {
--- a/src/mito2/src/sst/index/inverted_index/applier/builder/comparison.rs
+++ b/src/mito2/src/sst/index/inverted_index/applier/builder/comparison.rs
@@ -225,7 +225,7 @@ mod tests {
            ),
        ];

-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_comparison_basic_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -233,7 +233,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        for ((left, op, right), _) in &cases {
@@ -252,7 +252,7 @@ mod tests {

    #[test]
    fn test_collect_comparison_type_mismatch() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_comparison_type_mismatch_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -260,7 +260,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let res = builder.collect_comparison_expr(&tag_column(), &Operator::Lt, &int64_lit(10));
@@ -270,7 +270,7 @@ mod tests {

    #[test]
    fn test_collect_comparison_field_column() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_comparison_field_column_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -278,7 +278,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        builder
@@ -303,7 +303,7 @@ mod tests {

    #[test]
    fn test_collect_comparison_nonexistent_column() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_comparison_nonexistent_column_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -311,7 +311,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let res = builder.collect_comparison_expr(
--- a/src/mito2/src/sst/index/inverted_index/applier/builder/eq_list.rs
+++ b/src/mito2/src/sst/index/inverted_index/applier/builder/eq_list.rs
@@ -134,14 +134,14 @@ mod tests {

    #[test]
    fn test_collect_eq_basic() {
-        let (_d, facotry) = PuffinManagerFactory::new_for_test_block("test_collect_eq_basic_");
+        let (_d, factory) = PuffinManagerFactory::new_for_test_block("test_collect_eq_basic_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
            "test".to_string(),
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        builder
@@ -169,7 +169,7 @@ mod tests {

    #[test]
    fn test_collect_eq_field_column() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_eq_field_column_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -177,7 +177,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        builder
@@ -196,7 +196,7 @@ mod tests {

    #[test]
    fn test_collect_eq_nonexistent_column() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_eq_nonexistent_column_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -204,7 +204,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let res = builder.collect_eq(&nonexistent_column(), &string_lit("abc"));
@@ -214,7 +214,7 @@ mod tests {

    #[test]
    fn test_collect_eq_type_mismatch() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_eq_type_mismatch_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -222,7 +222,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let res = builder.collect_eq(&tag_column(), &int64_lit(1));
@@ -232,7 +232,7 @@ mod tests {

    #[test]
    fn test_collect_or_eq_list_basic() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_or_eq_list_basic_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -240,7 +240,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let eq_expr = DfExpr::BinaryExpr(BinaryExpr {
@@ -289,7 +289,7 @@ mod tests {

    #[test]
    fn test_collect_or_eq_list_invalid_op() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_or_eq_list_invalid_op_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -297,7 +297,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let eq_expr = DfExpr::BinaryExpr(BinaryExpr {
@@ -325,7 +325,7 @@ mod tests {

    #[test]
    fn test_collect_or_eq_list_multiple_columns() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_or_eq_list_multiple_columns_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -333,7 +333,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let eq_expr = DfExpr::BinaryExpr(BinaryExpr {
--- a/src/mito2/src/sst/index/inverted_index/applier/builder/in_list.rs
+++ b/src/mito2/src/sst/index/inverted_index/applier/builder/in_list.rs
@@ -65,14 +65,14 @@ mod tests {

    #[test]
    fn test_collect_in_list_basic() {
-        let (_d, facotry) = PuffinManagerFactory::new_for_test_block("test_collect_in_list_basic_");
+        let (_d, factory) = PuffinManagerFactory::new_for_test_block("test_collect_in_list_basic_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
            "test".to_string(),
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let in_list = InList {
@@ -95,7 +95,7 @@ mod tests {

    #[test]
    fn test_collect_in_list_negated() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_in_list_negated_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -103,7 +103,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let in_list = InList {
@@ -118,7 +118,7 @@ mod tests {

    #[test]
    fn test_collect_in_list_field_column() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_in_list_field_column_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -126,7 +126,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let in_list = InList {
@@ -149,7 +149,7 @@ mod tests {

    #[test]
    fn test_collect_in_list_type_mismatch() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_in_list_type_mismatch_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -157,7 +157,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let in_list = InList {
@@ -173,7 +173,7 @@ mod tests {

    #[test]
    fn test_collect_in_list_nonexistent_column() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_collect_in_list_nonexistent_column_");

        let metadata = test_region_metadata();
@@ -182,7 +182,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let in_list = InList {
--- a/src/mito2/src/sst/index/inverted_index/applier/builder/regex_match.rs
+++ b/src/mito2/src/sst/index/inverted_index/applier/builder/regex_match.rs
@@ -57,14 +57,14 @@ mod tests {

    #[test]
    fn test_regex_match_basic() {
-        let (_d, facotry) = PuffinManagerFactory::new_for_test_block("test_regex_match_basic_");
+        let (_d, factory) = PuffinManagerFactory::new_for_test_block("test_regex_match_basic_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
            "test".to_string(),
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        builder
@@ -83,7 +83,7 @@ mod tests {

    #[test]
    fn test_regex_match_field_column() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_regex_match_field_column_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -91,7 +91,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        builder
@@ -110,7 +110,7 @@ mod tests {

    #[test]
    fn test_regex_match_type_mismatch() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_regex_match_type_mismatch_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -118,7 +118,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        builder
@@ -130,7 +130,7 @@ mod tests {

    #[test]
    fn test_regex_match_type_nonexist_column() {
-        let (_d, facotry) =
+        let (_d, factory) =
            PuffinManagerFactory::new_for_test_block("test_regex_match_type_nonexist_column_");
        let metadata = test_region_metadata();
        let mut builder = InvertedIndexApplierBuilder::new(
@@ -138,7 +138,7 @@ mod tests {
            test_object_store(),
            &metadata,
            HashSet::from_iter([1, 2, 3]),
-            facotry,
+            factory,
        );

        let res = builder.collect_regex_match(&nonexistent_column(), &string_lit("abc"));
--- a/tests-integration/fixtures/docker-compose.yml
+++ b/tests-integration/fixtures/docker-compose.yml
@@ -1,14 +1,14 @@
 services:

  zookeeper:
-    image: docker.io/bitnami/zookeeper:3.7
+    image: greptime/zookeeper:3.7
    ports:
      - '2181:2181'
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes

  kafka:
-    image: docker.io/bitnami/kafka:3.9.0
+    image: greptime/kafka:3.9.0-debian-12-r1
    container_name: kafka
    ports:
      - 9092:9092
@@ -32,7 +32,7 @@ services:
        condition: service_started

  etcd:
-    image: docker.io/bitnami/etcd:3.5
+    image: greptime/etcd:3.6.1-debian-12-r3
    ports:
      - "2379:2379"
      - "2380:2380"
@@ -43,8 +43,35 @@ services:
      ETCD_ADVERTISE_CLIENT_URLS: http://etcd:2379
      ETCD_MAX_REQUEST_BYTES: 10485760

+#  etcd-tls:
+#    image: greptime/etcd:3.6.1-debian-12-r3
+#    ports:
+#      - "2378:2378"
+#      - "2381:2381"
+#    environment:
+#      ALLOW_NONE_AUTHENTICATION: "yes"
+#      ETCD_NAME: etcd-tls
+#      ETCD_LISTEN_CLIENT_URLS: https://0.0.0.0:2378
+#      ETCD_ADVERTISE_CLIENT_URLS: https://etcd-tls:2378
+#      ETCD_LISTEN_PEER_URLS: https://0.0.0.0:2381
+#      ETCD_INITIAL_ADVERTISE_PEER_URLS: https://etcd-tls:2381
+#      ETCD_INITIAL_CLUSTER: etcd-tls=https://etcd-tls:2381
+#      ETCD_INITIAL_CLUSTER_TOKEN: etcd-tls-cluster
+#      ETCD_INITIAL_CLUSTER_STATE: new
+#      ETCD_CERT_FILE: /certs/server.crt
+#      ETCD_KEY_FILE: /certs/server-key.pem
+#      ETCD_TRUSTED_CA_FILE: /certs/ca.crt
+#      ETCD_PEER_CERT_FILE: /certs/server.crt
+#      ETCD_PEER_KEY_FILE: /certs/server-key.pem
+#      ETCD_PEER_TRUSTED_CA_FILE: /certs/ca.crt
+#      ETCD_CLIENT_CERT_AUTH: "true"
+#      ETCD_PEER_CLIENT_CERT_AUTH: "true"
+#      ETCD_MAX_REQUEST_BYTES: 10485760
+#    volumes:
+#      - ./etcd-tls-certs:/certs:ro
+
  minio:
-    image: docker.io/bitnami/minio:2024
+    image: greptime/minio:2024
    ports:
      - '9000:9000'
      - '9001:9001'
@@ -68,7 +95,7 @@ services:
      - POSTGRES_PASSWORD=admin

  mysql:
-    image: bitnami/mysql:5.7
+    image: greptime/mysql:5.7
    ports:
      - 3306:3306
    volumes:
--- a/tests/conf/kafka-cluster.yml
+++ b/tests/conf/kafka-cluster.yml
@@ -2,7 +2,7 @@
 version: '3.8'
 services:
  kafka:
-    image: bitnami/kafka:3.6.0
+    image: greptime/kafka:3.9.0-debian-12-r1
    container_name: kafka
    ports:
      - 9092:9092
--- a/tests/runner/src/util.rs
+++ b/tests/runner/src/util.rs
@@ -363,9 +363,9 @@ pub fn setup_mysql(mysql_port: u16, mysql_version: Option<&str>) {
    }

    let mysql_image = if let Some(mysql_version) = mysql_version {
-        format!("bitnami/mysql:{mysql_version}")
+        format!("greptime/mysql:{mysql_version}")
    } else {
-        "bitnami/mysql:5.7".to_string()
+        "greptime/mysql:5.7".to_string()
    };
    let mysql_password = "admin";
    let mysql_user = "greptimedb";