fix

Author: zhyass

src/query/service/src/sessions/query_ctx.rs

@@ -512,21 +512,6 @@ impl QueryContext {
         table: &str,
         max_batch_size: Option<u64>,
     ) -> Result<Arc<dyn Table>> {
-        let max_batch_size = match max_batch_size {
-            Some(v) => {
-                // use the batch size specified in the statement
-                Some(v)
-            }
-            None => {
-                if let Some(v) = self.get_settings().get_stream_consume_batch_size_hint()? {
-                    info!("Overriding stream max_batch_size with setting value: {}", v);
-                    Some(v)
-                } else {
-                    None
-                }
-            }
-        };
-
         let table = self
             .shared
             .get_table(catalog, database, table, max_batch_size)
@@ -1440,8 +1425,22 @@ impl TableContext for QueryContext {
         database: &str,
         table: &str,
     ) -> Result<Arc<dyn Table>> {
-        self.get_table_with_batch(catalog, database, table, None)
-            .await
+        // Queries to non-internal system_history databases require license checks to be enabled.
+        if database.eq_ignore_ascii_case("system_history")
+            && ThreadTracker::capture_log_settings().is_none()
+        {
+            LicenseManagerSwitch::instance()
+                .check_enterprise_enabled(self.get_license_key(), Feature::SystemHistory)?;
+
+            if GlobalConfig::instance().log.history.is_invisible(table) {
+                return Err(ErrorCode::InvalidArgument(format!(
+                    "history table `{}` is configured as invisible",
+                    table
+                )));
+            }
+        }
+
+        self.get_table_from_shared(catalog, database, table, None).await
     }
 
     fn evict_table_from_cache(&self, catalog: &str, database: &str, table: &str) -> Result<()> {
@@ -1456,41 +1455,39 @@ impl TableContext for QueryContext {
         table: &str,
         max_batch_size: Option<u64>,
     ) -> Result<Arc<dyn Table>> {
-        // Queries to non-internal system_history databases require license checks to be enabled.
-        if database.eq_ignore_ascii_case("system_history")
-            && ThreadTracker::capture_log_settings().is_none()
-        {
-            LicenseManagerSwitch::instance()
-                .check_enterprise_enabled(self.get_license_key(), Feature::SystemHistory)?;
-
-            if GlobalConfig::instance().log.history.is_invisible(table) {
-                return Err(ErrorCode::InvalidArgument(format!(
-                    "history table `{}` is configured as invisible",
-                    table
-                )));
+        let final_batch_size = match max_batch_size {
+            Some(v) => {
+                // use the batch size specified in the statement
+                Some(v)
             }
-        }
+            None => {
+                if let Some(v) = self.get_settings().get_stream_consume_batch_size_hint()? {
+                    info!("Overriding stream max_batch_size with setting value: {}", v);
+                    Some(v)
+                } else {
+                    None
+                }
+            }
+        };
 
         let table = self
-            .get_table_from_shared(catalog, database, table, max_batch_size)
+            .get_table_from_shared(catalog, database, table, final_batch_size)
             .await?;
-        if max_batch_size.is_some() {
-            if table.is_stream() {
-                let stream = StreamTable::try_from_table(table.as_ref())?;
-                let actual_batch_limit = stream.max_batch_size();
-                if actual_batch_limit != max_batch_size {
-                    return Err(ErrorCode::StorageUnsupported(
-                        format!(
-                            "Stream batch size must be consistent within transaction: actual={:?}, requested={:?}",
-                            actual_batch_limit, max_batch_size
-                        )
-                    ));
-                }
-            } else {
+        if table.is_stream() {
+            let stream = StreamTable::try_from_table(table.as_ref())?;
+            let actual_batch_limit = stream.max_batch_size();
+            if actual_batch_limit != final_batch_size {
                 return Err(ErrorCode::StorageUnsupported(
-                    "MAX_BATCH_SIZE parameter only supported for STREAM tables",
+                    format!(
+                        "Stream batch size must be consistent within transaction: actual={:?}, requested={:?}",
+                        actual_batch_limit, final_batch_size
+                    )
                 ));
             }
+        } else if max_batch_size.is_some() {
+            return Err(ErrorCode::StorageUnsupported(
+                "MAX_BATCH_SIZE parameter only supported for STREAM tables",
+            ));
         }
         Ok(table)
     }

src/query/sql/src/planner/planner_cache.rs

@@ -205,7 +205,7 @@ impl TableRefVisitor {
             databend_common_base::runtime::block_on(async move {
                 if let Ok(table_meta) = self
                     .ctx
-                    .get_table(&catalog_name, &database_name, &table_name)
+                    .get_table_with_batch(&catalog_name, &database_name, &table_name, None)
                     .await
                 {
                     if !table_meta.is_temp()

tests/sqllogictests/suites/ee/06_ee_stream/06_0000_stream.test

@@ -172,7 +172,7 @@ drop stream s3
 statement ok
 drop table t2 all
 
-query TTTTT
+query TTTTTT
 select catalog, database, name, mode, table_name, comment from system.streams where database='test_stream' order by name
 ----
 default test_stream s append_only test_stream.t (empty)
@@ -194,7 +194,7 @@ drop stream t1
 statement ok
 drop table t1 all
 
-query T
+query TTTTAA
 show columns from s2
 ----
 a int YES (empty) NULL NULL
@@ -210,12 +210,12 @@ s2 Change tracking is not enabled on table 'test_stream'.'t'
 statement ok
 drop table t all
 
-query TTT
+query TT
 select name, invalid_reason from system.streams where database='test_stream' order by name
 ----
 s2 Unknown table 't'
 
-query T
+query TTTTAA
 show columns from s2
 ----
 
@@ -535,7 +535,7 @@ statement ok
 create stream s8_1 on table t7
 
 # test merge into insert only
-query T
+query I
 merge into t7 using (select a, b from t8) as s on t7.a=s.a when not matched then insert *
 ----
 1
@@ -546,7 +546,7 @@ select a, b, change$action, change$is_update from s8_1 order by a, b
 3 3 INSERT 0
 
 # test merge into matched only
-query TT
+query II
 merge into t7 using (select a, b+1 as b from t8) as s on t7.a=s.a when matched and s.a=2 then update * when matched and s.a=4 then delete
 ----
 1 1
@@ -572,7 +572,7 @@ statement ok
 create stream s9_1 on table t7
 
 # test merge into full operation
-query TTT
+query III
 merge into t7 using t8 on t7.a=t8.a when matched and t8.a=2 then update set t7.b=t8.b when matched and t8.a=3 then delete when not matched then insert (a,b) values(t8.a,t8.b)
 ----
 1 1 1
@@ -673,7 +673,7 @@ merge into t9 a using (select 10,'a') b(id,c1) on a.id=b.id when matched then up
 ----
 1 0
 
-query T
+query ITTB
 select id, c1, change$action, change$is_update from stream_t9;
 ----
 10 a INSERT 0
@@ -693,7 +693,7 @@ merge into t9 a using (select 10,'a') b(id,c1) on a.id=b.id when matched then up
 ----
 0 1
 
-query T
+query ITTTB
 select * from stream_t9;
 ----
 
@@ -810,15 +810,15 @@ insert into t1 values(1,1),(2,3),(3,3);
 statement ok
 explain merge into t1 using t2 on t1.a=t2.a when matched then update set t1.b=t2.b when not matched then insert *;
 
-query TT
+query II
 merge into t1 using t2 on t1.a=t2.a when matched then update set t1.b=t2.b when not matched then insert *;
 ----
 1 2
 
 statement ok
 alter table t1 rename to t1_1
 
-query T
+query TT
 select name, invalid_reason from system.streams where database='test_stream' order by name
 ----
 s1 (empty)

tests/sqllogictests/suites/ee/06_ee_stream/06_0001_stream_status.test

@@ -39,45 +39,45 @@ alter table t set options(change_tracking=true)
 statement ok
 create stream if not exists s on table t
 
-query I
+query B
 select * from stream_status('s')
 ----
 0
 
-query I
+query B
 select * from stream_status('test_stream_status.s')
 ----
 0
 
-query I
+query B
 select * from stream_status('default.test_stream_status.s')
 ----
 0
 
-query I
+query B
 call system$stream_status('s')
 ----
 0
 
-query I
+query B
 call system$stream_status('test_stream_status.s')
 ----
 0
 
-query I
+query B
 call system$stream_status('default.test_stream_status.s')
 ----
 0
 
 statement ok
 insert into t values(2)
 
-query I
+query B
 select * from stream_status('s')
 ----
 1
 
-query I
+query B
 call system$stream_status('s')
 ----
 1

tests/sqllogictests/suites/ee/06_ee_stream/06_0002_stream_txn_consume.test

@@ -371,7 +371,7 @@ insert into t_1 (str) values ('a'), ('b');
 statement ok
 begin;
 
-query I
+query T
 select str from s_1 with consume order by str;
 ----
 a
@@ -383,7 +383,7 @@ select 1/0;
 statement ok
 commit;
 
-query I
+query T
 select str from s_1 order by str;
 ----
 a
@@ -394,20 +394,20 @@ b
 statement ok
 begin;
 
-query I
+query T
 select str from s_1 with consume order by str;
 ----
 a
 b
 
 # inside txn, s_1 is consumed, expects empty result set
-query I
+query T
 select str from s_1;
 
 statement ok
 rollback;
 
-query I
+query T
 select str from s_1 order by str;
 ----
 a
@@ -428,7 +428,7 @@ statement ok
 insert into tmp_sink select str from s_1;
 
 # changes should not be consumed
-query I
+query T
 select str from s_1 order by str;
 ----
 a
@@ -459,7 +459,7 @@ select count() from tmp_sink
 statement ok
 commit;
 
-query I
+query T
 select str from s_1 order by str;
 ----
 
@@ -486,7 +486,7 @@ statement ok
 explain select str from s_1;
 
 # explain should not consume the stream
-query I
+query T
 select str from s_1 order by str;
 ----
 a
@@ -508,7 +508,7 @@ begin;
 statement ok
 insert into target_1 select str from s_1;
 
-query I
+query T
 select str from s_1 with consume order by str;
 ----
 a
@@ -520,13 +520,13 @@ insert into target_2 select str from s_1;
 statement ok
 commit;
 
-query I
+query T
 select str from target_1 order by str;
 ----
 a
 b
 
-query I
+query T
 select str from target_2 order by str;
 ----
 

tests/sqllogictests/suites/ee/06_ee_stream/06_0003_stream_multi_table_insert.test

@@ -14,10 +14,10 @@
 
 # multi table INSERT
 statement ok
-create or replace database test_txn_stream;
+create or replace database test_txn_stream_1;
 
 statement ok
-use test_txn_stream;
+use test_txn_stream_1;
 
 statement ok
 CREATE TABLE t_append_only(a INT);
@@ -194,4 +194,4 @@ SELECT * FROM t_consume_append_only_6_1; -- empty
 ----
 
 statement ok
-drop database test_txn_stream;
+drop database test_txn_stream_1;