regression-test/suites/compaction/test_admin_compact_table.groovy

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

// End-to-end test for `ADMIN COMPACT TABLE ... WHERE type = 'BASE'|'CUMULATIVE'|'FULL'`
// under the non-cloud (local storage) engine.
//
// Covers:
//   - all three compaction types can be triggered from the FE SQL entry
//   - rowsets are actually merged (not just last_*_success_time updated)
//   - data correctness after compaction
//   - negative cases: unknown type, multi-partition, missing WHERE
suite("test_admin_compact_table", "p0") {
    if (isCloudMode()) {
        return
    }

    def tableName = "test_admin_compact_table"
    def partName = tableName

    sql "DROP TABLE IF EXISTS ${tableName}"
    sql """
        CREATE TABLE ${tableName} (
            k INT,
            v INT
        ) DUPLICATE KEY(k)
        DISTRIBUTED BY HASH(k) BUCKETS 1
        PROPERTIES (
            "replication_num" = "1",
            "disable_auto_compaction" = "true"
        )
    """

    def tablets = sql_return_maparray "SHOW TABLETS FROM ${tableName}"
    assertEquals(1, tablets.size())
    def tabletId = tablets[0].TabletId
    def backendId = tablets[0].BackendId

    def backendId_to_backendIP = [:]
    def backendId_to_backendHttpPort = [:]
    getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort)
    def beHost = backendId_to_backendIP["${backendId}"]
    def bePort = backendId_to_backendHttpPort["${backendId}"]

    // Pull compaction show JSON. Fields of interest:
    //   rowsets: array of "[vL-vR] ... DATA ..." strings
    //   last_<type>_success_time / last_<type>_status
    def showTabletCompaction = {
        def (code, stdout, stderr) = be_show_tablet_status(beHost, bePort, tabletId)
        assertEquals(0, code)
        return parseJson(stdout.trim())
    }

    def countDataRowsets = { json ->
        return json.rowsets.findAll { it.contains(" DATA ") }.size()
    }

    def EPOCH_TIME = "1970-01-01 08:00:00.000"

    // Poll until BOTH `last_<type>_success_time` is updated AND the rowset count
    // has decreased (i.e. the compaction actually merged something). Success time
    // alone can be updated before the new rowset is installed, so we require both
    // to avoid race-y assertions downstream.
    def waitCompaction = { String type, int beforeCount, int timeoutSec = 60 ->
        def deadline = System.currentTimeMillis() + timeoutSec * 1000L
        def last
        while (System.currentTimeMillis() < deadline) {
            last = showTabletCompaction()
            def success = last["last ${type} success time"]
            def timeUpdated = success != null && success != EPOCH_TIME
            def merged = countDataRowsets(last) < beforeCount
            if (timeUpdated && merged) {
                return last
            }
            sleep(500)
        }
        return last
    }

    // ---------- load data to produce multiple rowsets ----------
    // 8 inserts -> 8 additional rowsets on top of [0-1] (initial empty rowset)
    for (int i = 1; i <= 8; i++) {
        sql "INSERT INTO ${tableName} VALUES (${i}, ${i})"
    }

    def beforeCumu = showTabletCompaction()
    def rowsetsBeforeCumu = countDataRowsets(beforeCumu)
    assertTrue(rowsetsBeforeCumu >= 8,
            "expected >= 8 data rowsets before cumulative, got ${rowsetsBeforeCumu}")
    assertEquals(EPOCH_TIME, beforeCumu["last cumulative success time"])

    // ---------- case 1: cumulative ----------
    sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'cumulative'"
    def afterCumu = waitCompaction("cumulative", rowsetsBeforeCumu)
    assertNotEquals(EPOCH_TIME, afterCumu["last cumulative success time"])
    assertEquals("[OK]", afterCumu["last cumulative status"])
    assertTrue(countDataRowsets(afterCumu) < rowsetsBeforeCumu,
            "cumulative did not reduce rowset count: ${afterCumu.rowsets}")

    // ---------- case 2: full ----------
    // Add more rowsets after cumulative so full has work to do.
    for (int i = 9; i <= 12; i++) {
        sql "INSERT INTO ${tableName} VALUES (${i}, ${i})"
    }
    def beforeFull = showTabletCompaction()
    def rowsetsBeforeFull = countDataRowsets(beforeFull)
    assertTrue(rowsetsBeforeFull >= 2,
            "expected >= 2 data rowsets before full, got ${rowsetsBeforeFull}")

    sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'full'"
    def afterFull = waitCompaction("full", rowsetsBeforeFull)
    assertNotEquals(EPOCH_TIME, afterFull["last full success time"])
    assertEquals("[OK]", afterFull["last full status"])
    assertTrue(countDataRowsets(afterFull) < rowsetsBeforeFull,
            "full did not reduce rowset count: ${afterFull.rowsets}")

    // ---------- case 3: base (may be a no-op depending on rowset shape) ----------
    sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'base'"
    // Poll a bit: base may be rejected (initial rowset empty) -> last base status non-empty.
    def afterBase = null
    for (int i = 0; i < 20; i++) {
        afterBase = showTabletCompaction()
        if (afterBase["last base status"] != "") {
            break
        }
        sleep(500)
    }
    assertNotNull(afterBase, "base compaction never scheduled")
    // Either success or a legitimate BE rejection; both prove the FE->BE path
    // accepted the "base" type and dispatched it.
    def baseStatus = afterBase["last base status"]
    assertTrue(baseStatus == "[OK]" || baseStatus.startsWith("[E-"),
            "unexpected base status: ${baseStatus}")

    // ---------- data correctness ----------
    qt_select_count "SELECT COUNT(*) FROM ${tableName}"
    qt_select_all "SELECT * FROM ${tableName} ORDER BY k"

    // ---------- negative: unknown type ----------
    test {
        sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'UNKNOWN'"
        exception "BASE/CUMULATIVE/FULL"
    }

    // ---------- negative: no WHERE ----------
    test {
        sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName})"
        exception "type"
    }
}