Merge branch 'main' into xxh/doris_sink

.config/hakari.toml

@@ -15,10 +15,10 @@ resolver = "2"
 # Add triples corresponding to platforms commonly used by developers here.
 # https://doc.rust-lang.org/rustc/platform-support.html
 platforms = [
-    "x86_64-unknown-linux-gnu",
-    "aarch64-unknown-linux-gnu",
-    "x86_64-apple-darwin",
-    "aarch64-apple-darwin",
+    # "x86_64-unknown-linux-gnu",
+    # "aarch64-unknown-linux-gnu",
+    # "x86_64-apple-darwin",
+    # "aarch64-apple-darwin",
 ]
 
 # Write out exact versions rather than a semver range. (Defaults to false.)

e2e_test/batch/basic/func.slt.part

@@ -570,6 +570,30 @@ select regexp_replace('💩💩💩💩💩foo🤔️bar亲爱的😭baz这不
 ----
 💩💩💩💩💩foo🤔️bar亲爱的😭这是🥵爱情❤️‍🔥
 
+# Positive Lookahead
+query T
+select regexp_replace('foobarbaz', 'a(?=r)', 'X');
+----
+foobXrbaz
+
+# Negative Lookahead
+query T
+select regexp_replace('chocolate', 'o(?!c)', 'X');
+----
+chocXlate
+
+# Positive Lookbehind
+query T
+select regexp_replace('foobarXaz', '(?<=X)a', 'X');
+----
+foobarXXz
+
+# Negative Lookbehind
+query T
+select regexp_replace('foobarXaz', '(?<!X)a', 'X');
+----
+foobXrXaz
+
 query T
 select regexp_count('ABCABCAXYaxy', 'A.');
 ----

e2e_test/batch/functions/format.slt.part

@@ -7,3 +7,26 @@ query T
 SELECT format('Testing %s, %s, %s, %%', 'one', 'two', 'three');
 ----
 Testing one, two, three, %
+
+query T
+SELECT format('%s %s', a, b) from (values
+    ('Hello', 'World'),
+    ('Rising', 'Wave')
+) as t(a, b);
+----
+Hello World
+Rising Wave
+
+query T
+SELECT format(f, a, b) from (values
+    ('%s %s', 'Hello', 'World'),
+    ('%s%s', 'Hello', null),
+    (null, 'Hello', 'World')
+) as t(f, a, b);
+----
+Hello World
+Hello
+NULL
+
+query error too few arguments for format()
+SELECT format('%s %s', 'Hello');

e2e_test/streaming/bug_fixes/issue_12299.slt

@@ -0,0 +1,31 @@
+# https://github.com/risingwavelabs/risingwave/issues/12299
+# TL;DR When upstream's stream key is not pk and the stream scan does not contain whole pk.
+
+statement ok
+create table t1(
+  id bigint primary key,
+  i bigint
+);
+
+statement ok
+create materialized view mv1 as select id, i from t1 order by id, i;
+
+statement ok
+insert into t1 values(1, 1);
+
+statement ok
+create materialized view mv2 as select id from mv1;
+
+query I
+select * from mv2;
+----
+1
+
+statement ok
+drop materialized view mv2;
+
+statement ok
+drop materialized view mv1;
+
+statement ok
+drop table t1;

e2e_test/udf/udf.slt

@@ -224,6 +224,35 @@ select (extract_tcp_info(E'\\x45000034a8a8400040065b8ac0a8000ec0a80001035d20b6d9
 ----
 192.168.0.14 192.168.0.1 861 8374
 
+# steaming
+# to ensure UDF & UDTF respect visibility
+
+statement ok
+create table t (x int);
+
+statement ok
+create materialized view mv as select gcd(x, x), series(x) from t where x <> 2;
+
+statement ok
+insert into t values (1), (2), (3);
+
+statement ok
+flush;
+
+query II
+select * from mv;
+----
+1 0
+3 0
+3 1
+3 2
+
+statement ok
+drop materialized view mv;
+
+statement ok
+drop table t;
+
 # error handling
 
 statement error

proto/catalog.proto

@@ -32,6 +32,13 @@ enum SchemaRegistryNameStrategy {
   TOPIC_RECORD_NAME_STRATEGY = 2;
 }
 
+enum StreamJobStatus {
+  // Prefixed by `STREAM_JOB_STATUS` due to protobuf namespacing rules.
+  STREAM_JOB_STATUS_UNSPECIFIED = 0;
+  CREATING = 1;
+  CREATED = 2;
+}
+
 message StreamSourceInfo {
   // deprecated
   plan_common.RowFormatType row_format = 1;
@@ -116,6 +123,7 @@ message Sink {
   optional uint64 created_at_epoch = 16;
   string db_name = 17;
   string sink_from_name = 18;
+  StreamJobStatus stream_job_status = 19;
 }
 
 message Connection {
@@ -157,6 +165,7 @@ message Index {
 
   optional uint64 initialized_at_epoch = 10;
   optional uint64 created_at_epoch = 11;
+  StreamJobStatus stream_job_status = 12;
 }
 
 message Function {
@@ -250,6 +259,9 @@ message Table {
   // In older versions we can just initialize without it.
   bool cleaned_by_watermark = 30;
 
+  // Used to filter created / creating tables in meta.
+  StreamJobStatus stream_job_status = 31;
+
   // Per-table catalog version, used by schema change. `None` for internal tables and tests.
   // Not to be confused with the global catalog version for notification service.
   TableVersion version = 100;

proto/hummock.proto

@@ -633,6 +633,19 @@ message ListBranchedObjectResponse {
   repeated BranchedObject branched_objects = 1;
 }
 
+message ListActiveWriteLimitRequest {}
+
+message ListActiveWriteLimitResponse {
+  // < compaction group id, write limit info >
+  map<uint64, WriteLimits.WriteLimit> write_limits = 1;
+}
+
+message ListHummockMetaConfigRequest {}
+
+message ListHummockMetaConfigResponse {
+  map<string, string> configs = 1;
+}
+
 service HummockManagerService {
   rpc UnpinVersionBefore(UnpinVersionBeforeRequest) returns (UnpinVersionBeforeResponse);
   rpc GetCurrentVersion(GetCurrentVersionRequest) returns (GetCurrentVersionResponse);
@@ -664,6 +677,8 @@ service HummockManagerService {
   rpc RiseCtlListCompactionStatus(RiseCtlListCompactionStatusRequest) returns (RiseCtlListCompactionStatusResponse);
   rpc SubscribeCompactionEvent(stream SubscribeCompactionEventRequest) returns (stream SubscribeCompactionEventResponse);
   rpc ListBranchedObject(ListBranchedObjectRequest) returns (ListBranchedObjectResponse);
+  rpc ListActiveWriteLimit(ListActiveWriteLimitRequest) returns (ListActiveWriteLimitResponse);
+  rpc ListHummockMetaConfig(ListHummockMetaConfigRequest) returns (ListHummockMetaConfigResponse);
 }
 
 message CompactionConfig {

src/common/src/array/arrow.rs

@@ -27,6 +27,7 @@ use crate::util::iter_util::ZipEqDebug;
 
 // Implement bi-directional `From` between `DataChunk` and `arrow_array::RecordBatch`.
 
+// note: DataChunk -> arrow RecordBatch will IGNORE the visibilities.
 impl TryFrom<&DataChunk> for arrow_array::RecordBatch {
     type Error = ArrayError;
 
@@ -47,8 +48,9 @@ impl TryFrom<&DataChunk> for arrow_array::RecordBatch {
             .collect();
 
         let schema = Arc::new(Schema::new(fields));
-
-        arrow_array::RecordBatch::try_new(schema, columns)
+        let opts =
+            arrow_array::RecordBatchOptions::default().with_row_count(Some(chunk.capacity()));
+        arrow_array::RecordBatch::try_new_with_options(schema, columns, &opts)
             .map_err(|err| ArrayError::ToArrow(err.to_string()))
     }
 }