Merge pull request #1255 from utPLSQL/bugfix/fix_1254

jgebal · web-flow · commit fa3b3b0dfb64 · 2023-05-28T17:35:37.000+03:00
Fix output length error and output buffer.
diff --git a/source/core/output_buffers/ut_output_buffer_base.tpb b/source/core/output_buffers/ut_output_buffer_base.tpb
@@ -24,7 +24,6 @@ create or replace type body ut_output_buffer_base is
     self.self_type := coalesce(a_self_type,self.self_type);
     self.output_id := coalesce(a_output_id, self.output_id, sys_guid());
     self.start_date := coalesce(self.start_date, sysdate);
-    self.last_write_message_id := 0;
     select /*+ no_parallel */ count(*) into l_exists from ut_output_buffer_info_tmp where output_id = self.output_id;
     if ( l_exists > 0 ) then
       update  /*+ no_parallel */ ut_output_buffer_info_tmp set start_date = self.start_date where output_id = self.output_id;
diff --git a/source/core/output_buffers/ut_output_buffer_base.tps b/source/core/output_buffers/ut_output_buffer_base.tps
@@ -19,7 +19,6 @@ create or replace type ut_output_buffer_base force authid definer as object(
   output_id                 raw(32),
   is_closed                 number(1,0),
   start_date                date,
-  last_write_message_id     number(38,0),
   lock_handle               varchar2(30  byte),
   self_type                 varchar2(250 byte),
   member procedure init(self in out nocopy ut_output_buffer_base, a_output_id raw := null, a_self_type varchar2 := null),
diff --git a/source/core/output_buffers/ut_output_buffer_tmp.sql b/source/core/output_buffers/ut_output_buffer_tmp.sql
@@ -30,3 +30,4 @@ create table ut_output_buffer_tmp(
 ) organization index nologging initrans 100
   overflow nologging initrans 100;
 
+create sequence ut_output_buffer_tmp_seq cache 20;
diff --git a/source/core/output_buffers/ut_output_bulk_buffer.tpb b/source/core/output_buffers/ut_output_bulk_buffer.tpb
@@ -34,9 +34,8 @@ create or replace type body ut_output_bulk_buffer is
           a_item_type
           );
       else
-        self.last_write_message_id := self.last_write_message_id + 1;
         insert /*+ no_parallel */ into ut_output_buffer_tmp(output_id, message_id, text, item_type)
-        values (self.output_id, self.last_write_message_id, a_text, a_item_type);
+        values (self.output_id, ut_output_buffer_tmp_seq.nextval, a_text, a_item_type);
       end if;
       commit;
     end if;
@@ -46,10 +45,9 @@ create or replace type body ut_output_bulk_buffer is
     pragma autonomous_transaction;
   begin
     insert /*+ no_parallel */ into ut_output_buffer_tmp(output_id, message_id, text, item_type)
-    select /*+ no_parallel */ self.output_id, self.last_write_message_id + rownum, t.column_value, a_item_type
+    select /*+ no_parallel */ self.output_id, ut_output_buffer_tmp_seq.nextval, t.column_value, a_item_type
       from table(a_text_list) t
      where t.column_value is not null or a_item_type is not null;
-    self.last_write_message_id := self.last_write_message_id + sql%rowcount;
     commit;
   end;
 
@@ -65,9 +63,8 @@ create or replace type body ut_output_bulk_buffer is
           a_item_type
           );
       else
-        self.last_write_message_id := self.last_write_message_id + 1;
         insert /*+ no_parallel */ into ut_output_buffer_tmp(output_id, message_id, text, item_type)
-        values (self.output_id, self.last_write_message_id, a_text, a_item_type);
+        values (self.output_id, ut_output_buffer_tmp_seq.nextval, a_text, a_item_type);
       end if;
       commit;
     end if;
diff --git a/source/core/output_buffers/ut_output_clob_buffer_tmp.sql b/source/core/output_buffers/ut_output_clob_buffer_tmp.sql
@@ -45,3 +45,5 @@ begin
   end;
 end;
 /
+
+create sequence ut_output_clob_buffer_tmp_seq cache 20;
diff --git a/source/core/output_buffers/ut_output_clob_table_buffer.tpb b/source/core/output_buffers/ut_output_clob_table_buffer.tpb
@@ -26,9 +26,8 @@ create or replace type body ut_output_clob_table_buffer is
     pragma autonomous_transaction;
   begin
     if a_text is not null or a_item_type is not null then
-      self.last_write_message_id := self.last_write_message_id + 1;
       insert /*+ no_parallel */ into ut_output_clob_buffer_tmp(output_id, message_id, text, item_type)
-      values (self.output_id, self.last_write_message_id, a_text, a_item_type);
+      values (self.output_id, ut_output_clob_buffer_tmp_seq.nextval, a_text, a_item_type);
     end if;
     commit;
   end;
@@ -37,20 +36,18 @@ create or replace type body ut_output_clob_table_buffer is
     pragma autonomous_transaction;
   begin
     insert /*+ no_parallel */ into ut_output_clob_buffer_tmp(output_id, message_id, text, item_type)
-    select /*+ no_parallel */ self.output_id, self.last_write_message_id + rownum, t.column_value, a_item_type
+    select /*+ no_parallel */ self.output_id, ut_output_clob_buffer_tmp_seq.nextval, t.column_value, a_item_type
       from table(a_text_list) t
      where t.column_value is not null or a_item_type is not null;
-    self.last_write_message_id := self.last_write_message_id + SQL%rowcount;
     commit;
   end;
 
   overriding member procedure send_clob(self in out nocopy ut_output_clob_table_buffer, a_text clob, a_item_type varchar2 := null) is
     pragma autonomous_transaction;
   begin
     if a_text is not null and a_text != empty_clob() or a_item_type is not null then
-      self.last_write_message_id := self.last_write_message_id + 1;
       insert /*+ no_parallel */ into ut_output_clob_buffer_tmp(output_id, message_id, text, item_type)
-      values (self.output_id, self.last_write_message_id, a_text, a_item_type);
+      values (self.output_id, ut_output_clob_buffer_tmp_seq.nextval, a_text, a_item_type);
     end if;
     commit;
   end;
@@ -60,33 +57,29 @@ create or replace type body ut_output_clob_table_buffer is
     l_buffer_rowids         ut_varchar2_rows;
     l_buffer_data           ut_output_data_rows;
     l_finished_flags        ut_integer_list;
-    l_last_read_message_id  integer;
     l_already_waited_sec    number(10,2) := 0;
     l_finished              boolean := false;
     l_sleep_time            number(2,1);
     l_lock_status           integer;
     l_producer_started      boolean := false;
     l_producer_finished     boolean := false;
     procedure get_data_from_buffer_table(
-        a_last_read_message_id in out nocopy integer,
         a_buffer_data    out nocopy ut_output_data_rows,
         a_buffer_rowids  out nocopy ut_varchar2_rows,
         a_finished_flags out nocopy ut_integer_list
       ) is
       lc_bulk_limit           constant integer     := 5000;
     begin
-      a_last_read_message_id := coalesce(a_last_read_message_id, 0);
       with ordered_buffer as (
         select  /*+ no_parallel index(a) */ ut_output_data_row(a.text, a.item_type), rowidtochar(a.rowid), is_finished
           from ut_output_clob_buffer_tmp a
          where a.output_id = self.output_id
-           and a.message_id <= a_last_read_message_id + lc_bulk_limit
+           and a.message_id <= (select min(message_id) from ut_output_clob_buffer_tmp o where o.output_id = self.output_id) + lc_bulk_limit
          order by a.message_id
       )
       select /*+ no_parallel */ b.*
         bulk collect into a_buffer_data, a_buffer_rowids, a_finished_flags
         from ordered_buffer b;
-      a_last_read_message_id := a_last_read_message_id + a_finished_flags.count;
     end;
 
     procedure remove_read_data(a_buffer_rowids ut_varchar2_rows) is
@@ -103,7 +96,7 @@ create or replace type body ut_output_clob_table_buffer is
 
       l_sleep_time := case when l_already_waited_sec >= 1 then 0.5 else 0.1 end;
       l_lock_status := self.get_lock_status();
-      get_data_from_buffer_table( l_last_read_message_id, l_buffer_data, l_buffer_rowids, l_finished_flags );
+      get_data_from_buffer_table( l_buffer_data, l_buffer_rowids, l_finished_flags );
 
       if l_buffer_data.count > 0 then
         l_already_waited_sec := 0;
diff --git a/source/core/output_buffers/ut_output_table_buffer.tpb b/source/core/output_buffers/ut_output_table_buffer.tpb
@@ -26,17 +26,16 @@ create or replace type body ut_output_table_buffer is
     pragma autonomous_transaction;
   begin
     if a_text is not null or a_item_type is not null then
-      if length(a_text) > ut_utils.gc_max_storage_varchar2_len then
+      if lengthb(a_text) > ut_utils.gc_max_storage_varchar2_len then
         self.send_lines(
           ut_utils.convert_collection(
             ut_utils.clob_to_table(a_text, ut_utils.gc_max_storage_varchar2_len)
             ),
           a_item_type
           );
       else
-        self.last_write_message_id := self.last_write_message_id + 1;
         insert /*+ no_parallel */ into ut_output_buffer_tmp(output_id, message_id, text, item_type)
-        values (self.output_id, self.last_write_message_id, a_text, a_item_type);
+        values (self.output_id, ut_output_buffer_tmp_seq.nextval, a_text, a_item_type);
       end if;
       commit;
     end if;
@@ -46,28 +45,26 @@ create or replace type body ut_output_table_buffer is
     pragma autonomous_transaction;
   begin
     insert /*+ no_parallel */ into ut_output_buffer_tmp(output_id, message_id, text, item_type)
-    select /*+ no_parallel */ self.output_id, self.last_write_message_id + rownum, t.column_value, a_item_type
+    select /*+ no_parallel */ self.output_id, ut_output_buffer_tmp_seq.nextval, t.column_value, a_item_type
       from table(a_text_list) t
      where t.column_value is not null or a_item_type is not null;
-    self.last_write_message_id := self.last_write_message_id + SQL%rowcount;
     commit;
   end;
 
   overriding member procedure send_clob(self in out nocopy ut_output_table_buffer, a_text clob, a_item_type varchar2 := null) is
     pragma autonomous_transaction;
   begin
     if a_text is not null and a_text != empty_clob() or a_item_type is not null then
-      if length(a_text) > ut_utils.gc_max_storage_varchar2_len then
+      if ut_utils.lengthb_clob(a_text) > ut_utils.gc_max_storage_varchar2_len then
         self.send_lines(
           ut_utils.convert_collection(
             ut_utils.clob_to_table(a_text, ut_utils.gc_max_storage_varchar2_len)
             ),
           a_item_type
           );
       else
-        self.last_write_message_id := self.last_write_message_id + 1;
         insert /*+ no_parallel */ into ut_output_buffer_tmp(output_id, message_id, text, item_type)
-        values (self.output_id, self.last_write_message_id, a_text, a_item_type);
+        values (self.output_id, ut_output_buffer_tmp_seq.nextval, a_text, a_item_type);
       end if;
       commit;
     end if;
@@ -99,7 +96,6 @@ create or replace type body ut_output_table_buffer is
     l_buffer_texts          ut_varchar2_rows;
     l_buffer_item_types     ut_varchar2_rows;
     l_finished_flags        ut_integer_list;
-    l_last_read_message_id  integer;
     l_already_waited_sec    number(10,2) := 0;
     l_finished              boolean := false;
     l_sleep_time            number(2,1);
@@ -108,33 +104,30 @@ create or replace type body ut_output_table_buffer is
     l_producer_finished     boolean := false;
 
   procedure get_data_from_buffer_table(
-    a_last_read_message_id in out nocopy integer,
     a_buffer_texts         out nocopy ut_varchar2_rows,
     a_buffer_item_types    out nocopy ut_varchar2_rows,
     a_finished_flags       out nocopy ut_integer_list
   ) is
     lc_bulk_limit           constant integer     := 20000;
     pragma autonomous_transaction;
   begin
-    a_last_read_message_id := coalesce(a_last_read_message_id,0);
     delete /*+ no_parallel */ from (
                   select /*+ no_parallel */ *
-                    from ut_output_buffer_tmp o
-                   where o.output_id = self.output_id
-                     and o.message_id <= a_last_read_message_id + lc_bulk_limit
-                   order by o.message_id
+                    from ut_output_buffer_tmp a
+                   where a.output_id = self.output_id
+                     and a.message_id <= (select min(message_id) from ut_output_buffer_tmp o where o.output_id = self.output_id) + lc_bulk_limit
+                   order by a.message_id
                 ) d
     returning d.text, d.item_type, d.is_finished
     bulk collect into a_buffer_texts, a_buffer_item_types, a_finished_flags;
-    a_last_read_message_id := a_last_read_message_id + a_finished_flags.count;
     commit;
   end;
   begin
     while not l_finished loop
 
       l_sleep_time := case when l_already_waited_sec >= 1 then 0.5 else 0.1 end;
       l_lock_status := self.get_lock_status();
-      get_data_from_buffer_table( l_last_read_message_id, l_buffer_texts, l_buffer_item_types, l_finished_flags );
+      get_data_from_buffer_table( l_buffer_texts, l_buffer_item_types, l_finished_flags );
 
       if l_buffer_texts.count > 0 then
         l_already_waited_sec := 0;
diff --git a/source/core/ut_utils.pkb b/source/core/ut_utils.pkb
@@ -92,8 +92,8 @@ create or replace package body ut_utils is
     a_max_output_len in number := gc_max_output_string_length
   ) return varchar2 is
     l_result                  varchar2(32767);
-    c_length                  constant integer := coalesce( length( a_value ), 0 );
-    c_max_input_string_length constant integer := a_max_output_len - coalesce( length( a_quote_char ) * 2, 0 );
+    c_length                  constant integer := coalesce( lengthb( a_value ), 0 );
+    c_max_input_string_length constant integer := a_max_output_len - coalesce( lengthb( a_quote_char ) * 2, 0 );
     c_overflow_substr_len     constant integer := c_max_input_string_length - gc_more_data_string_len;
   begin
     if c_length = 0 then
@@ -112,8 +112,8 @@ create or replace package body ut_utils is
     a_max_output_len in number := gc_max_output_string_length
   ) return varchar2 is
     l_result                  varchar2(32767);
-    c_length                  constant integer := coalesce(dbms_lob.getlength(a_value), 0);
-    c_max_input_string_length constant integer := a_max_output_len - coalesce( length( a_quote_char ) * 2, 0 );
+    c_length                  constant integer := coalesce(ut_utils.lengthb_clob(a_value), 0);
+    c_max_input_string_length constant integer := a_max_output_len - coalesce( lengthb( a_quote_char ) * 2, 0 );
     c_overflow_substr_len     constant integer := c_max_input_string_length - gc_more_data_string_len;
   begin
     if a_value is null then
@@ -135,7 +135,7 @@ create or replace package body ut_utils is
   ) return varchar2 is
     l_result                  varchar2(32767);
     c_length                  constant integer := coalesce(dbms_lob.getlength(a_value), 0);
-    c_max_input_string_length constant integer := a_max_output_len - coalesce( length( a_quote_char ) * 2, 0 );
+    c_max_input_string_length constant integer := a_max_output_len - coalesce( lengthb( a_quote_char ) * 2, 0 );
     c_overflow_substr_len     constant integer := c_max_input_string_length - gc_more_data_string_len;
   begin
     if a_value is null then
@@ -412,7 +412,7 @@ create or replace package body ut_utils is
       if a_list is null then
         a_list := ut_varchar2_rows();
       end if;
-      if length(a_item) > gc_max_storage_varchar2_len then
+      if lengthb(a_item) > gc_max_storage_varchar2_len then
         append_to_list(
           a_list,
           ut_utils.convert_collection(
@@ -468,7 +468,7 @@ create or replace package body ut_utils is
       l_result := ut_varchar2_rows();
       for i in 1 .. a_collection.count loop
         l_result.extend();
-        l_result(i) := substr(a_collection(i),1,gc_max_storage_varchar2_len);
+        l_result(i) := substrb(a_collection(i),1,gc_max_storage_varchar2_len);
       end loop;
     end if;
     return l_result;
@@ -990,5 +990,39 @@ create or replace package body ut_utils is
 
     return l_result;
   end;
+
+
+  /*
+  * Inspired by
+  *   https://stackoverflow.com/a/48782891
+  */
+  function lengthb_clob( a_clob clob) return integer is
+    l_blob     blob;
+    l_desc_offset PLS_INTEGER := 1;
+    l_src_offset  PLS_INTEGER := 1;
+    l_lang        PLS_INTEGER := 0;
+    l_warning     PLS_INTEGER := 0;
+    l_result      integer;
+  begin
+    if a_clob = empty_clob() then
+      l_result := 0;
+    elsif a_clob is not null then
+      dbms_lob.createtemporary(l_blob,true);
+      dbms_lob.converttoblob
+          ( l_blob
+          , a_clob
+          , dbms_lob.getlength(a_clob)
+          , l_desc_offset
+          , l_src_offset
+          , dbms_lob.default_csid
+          , l_lang
+          , l_warning
+          );
+      l_result := length(l_blob);
+      dbms_lob.freetemporary(l_blob);
+    end if;
+    return l_result;
+  end;
+
 end ut_utils;
 /
diff --git a/source/core/ut_utils.pks b/source/core/ut_utils.pks
@@ -476,6 +476,11 @@ create or replace package ut_utils authid definer is
   * Return value of interval in plain english
   */    
   function interval_to_text(a_interval yminterval_unconstrained) return varchar2;
+
+  /*
+  * Return length of CLOB in bytes. Null for NULL
+  */
+  function lengthb_clob( a_clob clob) return integer;
   
 end ut_utils;
 /
diff --git a/source/uninstall_objects.sql b/source/uninstall_objects.sql
@@ -333,8 +333,12 @@ drop type ut_output_buffer_base force;
 
 drop table ut_output_buffer_tmp purge;
 
+drop sequence ut_output_buffer_tmp_seq;
+
 drop table ut_output_clob_buffer_tmp purge;
 
+drop sequence ut_output_clob_buffer_tmp_seq;
+
 drop table ut_output_buffer_info_tmp purge;
 
 drop package ut_session_context;
diff --git a/test/ut3_tester/core/test_output_buffer.pkb b/test/ut3_tester/core/test_output_buffer.pkb
diff --git a/test/ut3_tester/core/test_output_buffer.pks b/test/ut3_tester/core/test_output_buffer.pks
diff --git a/test/ut3_tester/core/test_ut_utils.pkb b/test/ut3_tester/core/test_ut_utils.pkb
diff --git a/test/ut3_tester/core/test_ut_utils.pks b/test/ut3_tester/core/test_ut_utils.pks

Original file line number	Diff line number	Diff line change
`@@ -30,3 +30,4 @@ create table ut_output_buffer_tmp(`
`30`	`30`	`) organization index nologging initrans 100`
`31`	`31`	`overflow nologging initrans 100;`
`32`	`32`
	`33`	`+create sequence ut_output_buffer_tmp_seq cache 20;`
Original file line number	Diff line number	Diff line change
`@@ -45,3 +45,5 @@ begin`
`45`	`45`	`end;`
`46`	`46`	`end;`
`47`	`47`	`/`
	`48`	`+`
	`49`	`+create sequence ut_output_clob_buffer_tmp_seq cache 20;`