Merge branch 'main' of https://github.com/risingwavelabs/risingwave i…

…nto li0k/storage_split

.gitignore

@@ -71,12 +71,12 @@ e2e_test/generated/*
 scale-test.tar.zst
 simulation-it-test.tar.zst
 
-
 # hummock-trace
 .trace
 
 # spark binary
 e2e_test/iceberg/spark-*-bin*
+e2e_test/iceberg/metastore_db
 
 **/poetry.lock
 

.typos.toml

@@ -36,4 +36,5 @@ extend-exclude = [
     # We don't want to fix "fals" here, but may want in other places.
     # Ideally, we should just ignore that line: https://github.com/crate-ci/typos/issues/316
     "src/common/src/cast/mod.rs",
+    "src/tests/simulation/tests/integration_tests/scale/shared_source.rs",
 ]

Cargo.toml

@@ -135,7 +135,8 @@ tonic-build = { package = "madsim-tonic-build", version = "0.5" }
 otlp-embedded = { git = "https://github.com/risingwavelabs/otlp-embedded", rev = "e6cd165b9bc85783b42c106e99186b86b73e3507" }
 prost = { version = "0.13" }
 prost-build = { version = "0.13" }
-icelake = { git = "https://github.com/risingwavelabs/icelake.git", rev = "1860eb315183a5f3f72b4097c1e40d49407f8373", features = [
+# branch dylan/fix_parquet_nested_type_field_id
+icelake = { git = "https://github.com/risingwavelabs/icelake.git", rev = "3f4724158acee37a4785f56670a1427993a58739", features = [
     "prometheus",
 ] }
 arrow-array-iceberg = { package = "arrow-array", version = "52" }

README.md

@@ -50,7 +50,7 @@ RisingWave is a Postgres-compatible SQL database engineered to provide the <i><b
 
 RisingWave can <b>ingest</b> millions of events per second, continuously <b>join and analyze</b> live data streams with historical tables, <b>serve</b> ad-hoc queries in real-time, and <b>deliver</b> fresh, consistent results wherever needed.
 
-![RisingWave](./docs/dev/src/images/architecture_20240814.png)
+![RisingWave](./docs/dev/src/images/architecture_20240908.png)
 
 ## Try it out in 60 seconds
 

ci/scripts/e2e-iceberg-sink-v2-test.sh

@@ -45,6 +45,7 @@ poetry run python main.py -t ./test_case/partition_upsert.toml
 poetry run python main.py -t ./test_case/range_partition_append_only.toml
 poetry run python main.py -t ./test_case/range_partition_upsert.toml
 poetry run python main.py -t ./test_case/append_only_with_checkpoint_interval.toml
+poetry run python main.py -t ./test_case/iceberg_select_empty_table.toml
 
 
 echo "--- Kill cluster"

ci/scripts/e2e-source-test.sh

@@ -130,7 +130,7 @@ echo "> inserted new rows into postgres"
 
 # start cluster w/o clean-data
 unset RISINGWAVE_CI
-export RUST_LOG="risingwave_stream=debug,risingwave_batch=info,risingwave_storage=info" \
+export RUST_LOG="risingwave_stream=debug,risingwave_batch=info,risingwave_storage=info"
 
 risedev dev ci-1cn-1fe-with-recovery
 echo "> wait for cluster recovery finish"

docker/docker-compose-distributed-etcd.yml

@@ -1,7 +1,7 @@
 ---
 version: "3"
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v1.10.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0-rc.1}
 services:
   compactor-0:
     <<: *image

docker/docker-compose-distributed.yml

@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v1.10.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0-rc.1}
 services:
   compactor-0:
     <<: *image

docker/docker-compose-etcd.yml

@@ -1,7 +1,7 @@
 ---
 version: "3"
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v1.10.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0-rc.1}
 services:
   risingwave-standalone:
     <<: *image

docker/docker-compose-with-azblob.yml

@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v1.10.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0-rc.1}
 services:
   risingwave-standalone:
     <<: *image

docker/docker-compose-with-gcs.yml

@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v1.10.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0-rc.1}
 services:
   risingwave-standalone:
     <<: *image

docker/docker-compose-with-local-fs.yml

@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v1.10.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0-rc.1}
 services:
   risingwave-standalone:
     <<: *image

docker/docker-compose-with-obs.yml

@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v1.10.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0-rc.1}
 services:
   risingwave-standalone:
     <<: *image

docker/docker-compose-with-oss.yml

@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v1.10.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0-rc.1}
 services:
   risingwave-standalone:
     <<: *image

docker/docker-compose-with-s3.yml

@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v1.10.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0-rc.1}
 services:
   risingwave-standalone:
     <<: *image

docker/docker-compose-with-sqlite.yml

@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v1.10.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0-rc.1}
 services:
   risingwave-standalone:
     <<: *image

docker/docker-compose.yml

@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v1.10.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0-rc.1}
 services:
   risingwave-standalone:
     <<: *image
@@ -60,7 +60,7 @@ services:
       ENABLE_TELEMETRY: ${ENABLE_TELEMETRY:-true}
       RW_TELEMETRY_TYPE: ${RW_TELEMETRY_TYPE:-"docker-compose"}
       RW_SECRET_STORE_PRIVATE_KEY_HEX: ${RW_SECRET_STORE_PRIVATE_KEY_HEX:-0123456789abcdef}
-      RW_LICENSE_KEY: ${RW_LICENSE_KEY:-""}
+      RW_LICENSE_KEY: ${RW_LICENSE_KEY:-}
     container_name: risingwave-standalone
     healthcheck:
       test:
@@ -113,7 +113,7 @@ services:
       - "./grafana-risedev-datasource.yml:/etc/grafana/provisioning/datasources/grafana-risedev-datasource.yml"
       - "./grafana-risedev-dashboard.yml:/etc/grafana/provisioning/dashboards/grafana-risedev-dashboard.yml"
       - "./dashboards:/dashboards"
-    environment: { }
+    environment: {}
     container_name: grafana-0
     healthcheck:
       test:
@@ -187,7 +187,7 @@ services:
     volumes:
       - "prometheus-0:/prometheus"
       - "./prometheus.yaml:/etc/prometheus/prometheus.yml"
-    environment: { }
+    environment: {}
     container_name: prometheus-0
     healthcheck:
       test:
@@ -229,7 +229,7 @@ services:
     depends_on: [ ]
     volumes:
       - "message_queue:/var/lib/redpanda/data"
-    environment: { }
+    environment: {}
     container_name: message_queue
     healthcheck:
       test: curl -f localhost:9644/v1/status/ready

e2e_test/backup_restore/tpch_snapshot_create.slt

@@ -1,5 +1,8 @@
 include ../tpch/create_tables.slt.part
 
+statement ok
+CREATE SECRET secret1 WITH (backend = 'meta') AS 'demo-secret'
+
 # First, insert the data into the tables
 include ../tpch/insert_customer.slt.part
 include ../tpch/insert_lineitem.slt.part

e2e_test/backup_restore/tpch_snapshot_drop.slt

@@ -1,3 +1,6 @@
+statement ok
+DROP SECRET secret1;
+
 statement ok
 drop materialized view tpch_q7;
 

e2e_test/batch/catalog/pg_settings.slt.part

@@ -22,6 +22,7 @@ user backfill_rate_limit
 user background_ddl
 user batch_enable_distributed_dml
 user batch_parallelism
+user bypass_cluster_limits
 user bytea_output
 user cdc_source_wait_streaming_start_timeout
 user client_encoding

e2e_test/commands/risectl

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+RUST_LOG="error" .risingwave/bin/risingwave/risectl "$@"

e2e_test/iceberg/main.py

@@ -55,16 +55,23 @@ def execute_slt(args, slt):
 
 def verify_result(args, verify_sql, verify_schema, verify_data):
     tc = unittest.TestCase()
-    print(f"Executing sql: {verify_sql}")
+
+    time.sleep(3)
+    print(f"verify_result:\nExecuting sql: {verify_sql}")
     spark = get_spark(args)
     df = spark.sql(verify_sql).collect()
+    print(f"Result:")
+    print(f"================")
     for row in df:
         print(row)
+    print(f"================")
     rows = verify_data.splitlines()
-    tc.assertEqual(len(df), len(rows))
+    tc.assertEqual(len(df), len(rows), "row length mismatch")
+    tc.assertEqual(len(verify_schema), len(df[0]), "column length mismatch")
     for row1, row2 in zip(df, rows):
         print(f"Row1: {row1}, Row 2: {row2}")
-        row2 = row2.split(",")
+        # New parsing logic for row2
+        row2 = parse_row(row2)
         for idx, ty in enumerate(verify_schema):
             if ty == "int" or ty == "long":
                 tc.assertEqual(row1[idx], int(row2[idx]))
@@ -89,7 +96,7 @@ def verify_result(args, verify_sql, verify_schema, verify_data):
                 else:
                     tc.assertEqual(row1[idx], decimal.Decimal(row2[idx]))
             else:
-                tc.fail(f"Unsupported type {ty}")
+                tc.assertEqual(str(row1[idx]), str(row2[idx]))
 
 def compare_sql(args, cmp_sqls):
     assert len(cmp_sqls) == 2
@@ -113,6 +120,32 @@ def drop_table(args, drop_sqls):
         spark.sql(sql)
 
 
+def parse_row(row):
+    result = []
+    current = ""
+    parenthesis_count = {"{": 0, "[": 0, "(": 0}
+    for char in row:
+        if char in parenthesis_count:
+            parenthesis_count[char] += 1
+        elif char == "}":
+            parenthesis_count["{"] -= 1
+        elif char == "]":
+            parenthesis_count["["] -= 1
+        elif char == ")":
+            parenthesis_count["("] -= 1
+
+        if char == "," and all(value == 0 for value in parenthesis_count.values()):
+            result.append(current.strip())
+            current = ""
+        else:
+            current += char
+
+    if current:
+        result.append(current.strip())
+
+    return result
+
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Test script for iceberg")
     parser.add_argument("-t", dest="test_case", type=str, help="Test case file")
@@ -151,4 +184,3 @@ def drop_table(args, drop_sqls):
             execute_slt(config, verify_slt)
         if drop_sqls is not None and drop_sqls != "":
             drop_table(config, drop_sqls)
-

e2e_test/iceberg/start_spark_connect_server.sh

@@ -1,3 +1,5 @@
+#!/usr/bin/env bash
+
 set -ex
 
 ICEBERG_VERSION=1.4.3

e2e_test/iceberg/test_case/iceberg_select_empty_table.slt

@@ -0,0 +1,60 @@
+statement ok
+set sink_decouple = false;
+
+statement ok
+set streaming_parallelism=4;
+
+statement ok
+CREATE TABLE s1 (i1 int, i2 varchar, i3 varchar);
+
+statement ok
+CREATE MATERIALIZED VIEW mv1 AS SELECT * FROM s1;
+
+statement ok
+CREATE SINK sink1 AS select * from mv1 WITH (
+    connector = 'iceberg',
+    type = 'append-only',
+    force_append_only = 'true',
+    database.name = 'demo_db',
+    table.name = 't1',
+    catalog.name = 'demo',
+    catalog.type = 'storage',
+    warehouse.path = 's3a://icebergdata/demo',
+    s3.endpoint = 'http://127.0.0.1:9301',
+    s3.region = 'us-east-1',
+    s3.access.key = 'hummockadmin',
+    s3.secret.key = 'hummockadmin',
+    commit_checkpoint_interval = 1,
+    create_table_if_not_exists = 'true'
+);
+
+statement ok
+CREATE SOURCE iceberg_t1_source
+WITH (
+    connector = 'iceberg',
+    s3.endpoint = 'http://127.0.0.1:9301',
+    s3.region = 'us-east-1',
+    s3.access.key = 'hummockadmin',
+    s3.secret.key = 'hummockadmin',
+    catalog.type = 'storage',
+    warehouse.path = 's3a://icebergdata/demo',
+    database.name = 'demo_db',
+    table.name = 't1',
+);
+
+statement ok
+flush;
+
+query I
+select count(*) from iceberg_t1_source;
+----
+0
+
+statement ok
+DROP SINK sink1;
+
+statement ok
+DROP SOURCE iceberg_t1_source;
+
+statement ok
+DROP TABLE s1 cascade;