test: add integration test for compaction offload (#1573)

## Rationale
Close #1571 

## Detailed Changes
- Impl `compact` as pre-command for sqlness, with http compaction
service.
- update integration tests to test compaction offload.

## Test Plan

---------

Co-authored-by: kamille <[email protected]>

integration_tests/Makefile

@@ -21,6 +21,7 @@ HORAEDB_DATA_DIR = /tmp/horaedb
 HORAEDB_DATA_DIR_0 = /tmp/horaedb0
 HORAEDB_DATA_DIR_1 = /tmp/horaedb1
 HORAEMETA_DATA_DIR = /tmp/horaemeta
+HORAEDB_DATA_DIR_2 = /tmp/compaction-offload
 
 export HORAEDB_TEST_CASE_PATH ?= $(ROOT)/cases/env
 export HORAEDB_TEST_BINARY ?= $(ROOT)/../target/$(MODE)/horaedb-test
@@ -42,13 +43,17 @@ export CLUSTER_HORAEDB_STDOUT_FILE_0 ?= /tmp/horaedb-stdout-0.log
 export CLUSTER_HORAEDB_STDOUT_FILE_1 ?= /tmp/horaedb-stdout-1.log
 export RUST_BACKTRACE=1
 
+# Environment variables for compaction offload
+export HORAEDB_STDOUT_FILE_2 ?= /tmp/horaedb-stdout-2.log
+export HORAEDB_CONFIG_FILE_2 ?= $(ROOT)/config/compaction-offload.toml
+
 # Whether update related repos
 # We don't want to rebuild the binaries and data on sometimes(e.g. debugging in local),
 # and we can set it to false.
 export UPDATE_REPOS_TO_LATEST ?= true
 
 clean:
-	rm -rf $(HORAEDB_DATA_DIR) $(HORAEDB_DATA_DIR_0) $(HORAEDB_DATA_DIR_1) $(HORAEMETA_DATA_DIR)
+	rm -rf $(HORAEDB_DATA_DIR) $(HORAEDB_DATA_DIR_0) $(HORAEDB_DATA_DIR_1) $(HORAEMETA_DATA_DIR) $(HORAEDB_DATA_DIR_2) 
 
 build-meta:
 	./build_meta.sh
@@ -80,15 +85,20 @@ run-horaedb-cluster: build-horaedb
 	nohup ${HORAEDB_BINARY_PATH} --config ${HORAEDB_CONFIG_FILE_1} > ${CLUSTER_HORAEDB_STDOUT_FILE_1} 2>&1 &
 	sleep 30
 
-run: prepare build-meta
-	$(HORAEDB_TEST_BINARY)
+run:
+	make run-local
+	make run-cluster
+	make run-compaction-offload
 
 run-local: prepare
 	HORAEDB_ENV_FILTER=local $(HORAEDB_TEST_BINARY)
 
 run-cluster: prepare build-meta
 	HORAEDB_ENV_FILTER=cluster $(HORAEDB_TEST_BINARY)
 
+run-compaction-offload: prepare
+	HORAEDB_ENV_FILTER=compaction_offload $(HORAEDB_TEST_BINARY)
+
 run-java:
 	java -version
 	cd sdk/java && MAVEN_OPTS="--add-opens=java.base/java.nio=ALL-UNNAMED" mvn clean compile exec:java

integration_tests/README.md

@@ -12,6 +12,9 @@ make run-local
 
 # Only cluster env
 make run-cluster
+
+# Only compaction offload env
+make run-compaction-offload
 ```
 
 `horaedb-test` will recursively find all the files end with `.sql` and run it. Each file will be treated as a case. A file can contain multiple SQLs. When finished it will tell how many cases it run, and display the diff set if there is any. An example with one case:

integration_tests/cases/env/compaction_offload/compact/compact.result

@@ -0,0 +1,110 @@
+--
+-- 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.
+--
+DROP TABLE IF EXISTS `compact_table1`;
+
+affected_rows: 0
+
+CREATE TABLE `compact_table1` (
+    `timestamp` timestamp NOT NULL,
+    `value` double,
+    `dic` string dictionary,
+    timestamp KEY (timestamp)) ENGINE=Analytic
+WITH(
+	 enable_ttl='false',
+     update_mode='OVERWRITE'
+);
+
+affected_rows: 0
+
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (1, 100, "d1"), (2, 200, "d2"), (3, 300, "d3");
+
+affected_rows: 3
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (1, 100, "update_d1"), (2, 200, "update_d2"), (3, 300, "update_d3");
+
+affected_rows: 3
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (4, 400, "d4"), (5, 500, "d5"), (6, 600, "d6");
+
+affected_rows: 3
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (4, 400, "update_d4"), (5, 500, "update_d5"), (6, 600, "update_d6");
+
+affected_rows: 3
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (7, 700, "d7"), (8, 800, "d8"), (9, 900, "d9");
+
+affected_rows: 3
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (7, 700, "update_d7"), (8, 800, "update_d8"), (9, 900, "update_d9");
+
+affected_rows: 3
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (10, 1000, "d10"), (11, 1100, "d11"), (12, 1200, "d12");
+
+affected_rows: 3
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (10, 1000, "update_d10"), (11, 1100, "update_d11"), (12, 1200, "update_d12");
+
+affected_rows: 3
+
+-- trigger manual compaction after flush memtable
+-- SQLNESS ARG pre_cmd=flush
+-- SQLNESS ARG pre_cmd=compact
+SELECT
+    *
+FROM
+    `compact_table1`
+ORDER BY
+    `value` ASC;
+
+tsid,timestamp,value,dic,
+UInt64(0),Timestamp(1),Double(100.0),String("update_d1"),
+UInt64(0),Timestamp(2),Double(200.0),String("update_d2"),
+UInt64(0),Timestamp(3),Double(300.0),String("update_d3"),
+UInt64(0),Timestamp(4),Double(400.0),String("update_d4"),
+UInt64(0),Timestamp(5),Double(500.0),String("update_d5"),
+UInt64(0),Timestamp(6),Double(600.0),String("update_d6"),
+UInt64(0),Timestamp(7),Double(700.0),String("update_d7"),
+UInt64(0),Timestamp(8),Double(800.0),String("update_d8"),
+UInt64(0),Timestamp(9),Double(900.0),String("update_d9"),
+UInt64(0),Timestamp(10),Double(1000.0),String("update_d10"),
+UInt64(0),Timestamp(11),Double(1100.0),String("update_d11"),
+UInt64(0),Timestamp(12),Double(1200.0),String("update_d12"),
+
+
+DROP TABLE `compact_table1`;
+
+affected_rows: 0
+

integration_tests/cases/env/compaction_offload/compact/compact.sql

@@ -0,0 +1,76 @@
+--
+-- 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.
+--
+
+DROP TABLE IF EXISTS `compact_table1`;
+
+CREATE TABLE `compact_table1` (
+    `timestamp` timestamp NOT NULL,
+    `value` double,
+    `dic` string dictionary,
+    timestamp KEY (timestamp)) ENGINE=Analytic
+WITH(
+	 enable_ttl='false',
+     update_mode='OVERWRITE'
+);
+
+
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (1, 100, "d1"), (2, 200, "d2"), (3, 300, "d3");
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (1, 100, "update_d1"), (2, 200, "update_d2"), (3, 300, "update_d3");
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (4, 400, "d4"), (5, 500, "d5"), (6, 600, "d6");
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (4, 400, "update_d4"), (5, 500, "update_d5"), (6, 600, "update_d6");
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (7, 700, "d7"), (8, 800, "d8"), (9, 900, "d9");
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (7, 700, "update_d7"), (8, 800, "update_d8"), (9, 900, "update_d9");
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (10, 1000, "d10"), (11, 1100, "d11"), (12, 1200, "d12");
+
+-- SQLNESS ARG pre_cmd=flush
+INSERT INTO `compact_table1` (`timestamp`, `value`, `dic`)
+    VALUES (10, 1000, "update_d10"), (11, 1100, "update_d11"), (12, 1200, "update_d12");
+
+
+-- trigger manual compaction after flush memtable
+-- SQLNESS ARG pre_cmd=flush
+-- SQLNESS ARG pre_cmd=compact
+SELECT
+    *
+FROM
+    `compact_table1`
+ORDER BY
+    `value` ASC;
+
+
+DROP TABLE `compact_table1`;

integration_tests/cases/env/compaction_offload/config.toml

@@ -0,0 +1,44 @@
+# 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.
+
+[server]
+bind_addr = "127.0.0.1"
+http_port = 5440
+grpc_port = 8831
+
+[query_engine]
+read_parallelism = 8
+
+[analytic.wal]
+type = "RocksDB"
+data_dir = "/tmp/horaedb"
+
+[analytic.storage]
+mem_cache_capacity = '1G'
+mem_cache_partition_bits = 0
+disk_cache_dir = "/tmp/horaedb"
+disk_cache_capacity = '2G'
+disk_cache_page_size = '1M'
+
+[analytic.storage.object_store]
+type = "Local"
+data_dir = "/tmp/horaedb"
+
+[analytic.compaction_mode]
+compaction_mode = "Offload"
+node_picker = "Local"
+endpoint = "127.0.0.1:8831"

integration_tests/config/compaction-offload.toml

@@ -0,0 +1,44 @@
+# 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.
+
+[server]
+bind_addr = "0.0.0.0"
+http_port = 5440
+grpc_port = 8831
+postgresql_port = 5433
+
+[logger]
+level = "info"
+
+[tracing]
+dir = "/tmp/compaction-offload"
+
+[analytic.storage.object_store]
+type = "Local"
+data_dir = "/tmp/compaction-offload"
+
+[analytic.wal]
+type = "Local"
+data_dir = "/tmp/compaction-offload"
+
+[analytic.compaction_mode]
+compaction_mode = "Offload"
+node_picker = "Local"
+endpoint = "127.0.0.1:8831"
+
+[analytic]
+enable_primary_key_sampling = true