索引嵌套循环连接忽略 SQL 语句

Question

我正在尝试使用两组数据构建一个基于星型模式的数据仓库，其中包含 5 个维度表和 1 个事实表，MASTERDATA 包含 100 条记录，DATASTREAM 包含 10,000 条记录。

我正在从 DATASTREAM 读取 100 条记录作为游标的输入，然后逐条读取游标记录，然后从索引 product_id 上的 MASTERDATA 检索相关记录作为索引嵌套循环连接。在此之后，我从相关维度和事实表中的事务元组加载新属性。

但是，我有一些错误。我只是在寻求帮助以了解为什么我会遇到我遇到的错误。目前的错误是：

Error(98,6):PL/SQL:SQL Statement Ignored
Error(101,5):PL/SQL: ORA-00933: SQL command not properly ended
Error(105,8):PLS-00103:Encountered the symbol "LOOP" when expecting one of the following:  if
Error(113):PLS-00103:Encountered the symbol "end-of-file" when expecting one of the following:  ;

我的代码：

CREATE OR REPLACE PROCEDURE transactionINLJ AS TYPE t_cursor is ref cursor;
v_cursor t_cursor;
v_cursor_records DATASTREAM%rowtype;
record_100 varchar2(300);
rec number;

v_customer_id masterdata.customer_id%type;
v_customer_account_type masterdata.customer_account_type%type;
v_product_id masterdata.product_id%type;
v_product_name masterdata.product_name%type;
v_supplier_id masterdata.supplier_id%type;
v_supplier_name masterdata.supplier_name%type;
v_outlet_id masterdata.outlet_id%type;
v_outlet_name masterdata.outlet_name%type;
v_sale_price masterdata.sale_price%type;

t_customer_id int;
t_supplier_id int;
t_product_id int;
t_outlet_id int;
t_date_id int;
t_sales_fact int;

BEGIN
    rec := 1;
    WHILE (rec <= 10000)
    LOOP
        record_100 := 'SELECT * FROM datastream WHERE datastream_id between '|| TO_CHAR(rec) ||
        ' and ' || TO_CHAR(rec+99);
        OPEN v_cursor FOR record_100;
            LOOP
                FETCH v_cursor INTO v_cursor_records;
                EXIT WHEN v_cursor%notfound;
                
                SELECT product_id, product_name, supplier_id, supplier_name, sale_price
                INTO v_product_id, v_product_name, v_supplier_id, v_supplier_name, v_sale_price
                FROM masterdata
                WHERE product_id = v_cursor_records.product_id;
                
                SELECT COUNT(0)
                INTO t_product_id
                FROM product_dim
                WHERE product_id = v_cursor_records.product_id;
                IF t_product_id = 0 THEN 
                    INSERT INTO product_dim(product_id, product_name)
                    VALUES (v_cursor_records.product_id, v_cursor_records.product_name);
                END IF;
                
                SELECT COUNT(0)
                INTO t_customer_id
                FROM customer_dim
                WHERE customer_id = v_cursor_records.customer_id;
                IF t_customer_id = 0 THEN 
                    INSERT INTO customer_dim(customer_id, customer_name,customer_account_type)
                    VALUES (v_cursor_records.customer_name, v_cursor_records.customer_account_type, v_cursor_records.customer_account_type);
                END IF;
                
                SELECT COUNT(0)
                INTO t_supplier_id
                FROM supplier_dim
                WHERE supplier_id = v_cursor_records.supplier_id;
                IF t_supplier_id = 0 THEN 
                    INSERT INTO supplier_dim(supplier_id, supplier_name)
                    VALUES (v_cursor_records.supplier_id, v_cursor_records.supplier_name);
                END IF;
                
                SELECT COUNT(0)
                INTO t_outlet_id
                FROM outlet_dim
                WHERE outlet_id = v_cursor_records.outlet_id;
                IF t_outlet_id = 0 THEN 
                    INSERT INTO outlet_dim(outlet_id, outlet_name)
                    VALUES (v_cursor_records.outlet_id, v_cursor_records.outlet_name);
                END IF;
                
                SELECT COUNT(0)
                INTO t_date_id
                FROM date_dim
                WHERE d_date = v_cursor_records.d_date;
                IF t_date_id = 0 THEN 
                    INSERT INTO date_dim(d_date, d_year, d_quater, d_month, d_day)
                    VALUES (v_cursor_records.d_date
                            ,EXTRACT(year FROM v_cursor_records.d_date), TO_CHAR(v_cursor_records.d_date,'Q')
                            ,EXTRACT(month FROM v_cursor_records.d_date)
                            ,EXTRACT(day FROM v_cursor_records.d_date));
                END IF;
                
                SELECT COUNT(0)
                INTO t_sales_fact
                FROM sales_fact
                WHERE product_id = v_cursor_records.product_id
                AND customer_id = v_csr_rec.customer_id
                AND supplier_id = v_csr_rec.supplier_id
                AND outlet_id = v_csr_rec.outlet_id
                AND d_date = v_csr_rec.d_date
                AND sale_price = v_csr_rec.sale_price
                AND quantity_sold = v_csr_rec.quantity_sold;
                IF t_sales_fact = 0 THEN 
                    INSERT INTO sales_fact(customer_id,product_id,outlet_id,supplier_id,d_date,sale_price,total_sale,quantity_sold)
                    VALUES (v_cursor_records.customer_id, v_cursor_records.product_id, v_cursor_records.outlet_id,v_cursor_records.supplier_id,
                        v_cursor_records.d_date, v_cursor_records.sale_price, v_cursor_records.quantity_sold*sale_price, v_cursor_records.quantity_sold)
                END IF;
                
                COMMIT;
                
            END LOOP;
        CLOSE v_cursor;
        COMMIT;
        
        rec := rec+100;
        
    END LOOP;
END;        
                

Answer 1

不幸的是，有时需要进行程序处理。但几乎所有这些都可以通过 SQL 和一点点 PL/SQL 扩展来完成。特别是不需要为任何目标表“选择计数...”，sql 在 INSERT 语句本身上很容易处理。此外，无需在逐行（又名逐行缓慢）过程中遍历游标，而是使用 BULK COLLECT 和 FORALL 来处理整个数组（在本例中为 100 行），所有这些都使用单个 INSERT对于每张桌子。有了它，就不需要循环控制计数器，也不需要计算要检索的 ID 号，也不需要确切的行数（如果你的源表包含 10050 或 9950 行而不是正好 10000 行会发生什么）。作为副作用，您可以获得可观的性能。下面展示了这个过程：

create or replace procedure transactioninlj as 
 
    k_bulk_buffer_size constant integer := 100;
   
    cursor v_cursor is
           select d.customer_id
                , d.outlet_id
                , d.outlet_name
                , d.customer_name  
                , d.customer_account_type
                , d.d_date
                , d.quantity_sold
                , m.product_id
                , m.product_name
                , m.supplier_id
                , m.supplier_name
                , m.sale_price
             from datastream d
             join masterdata m on m.product_id = d.product_id
         ;
    type t_cursor_records is table of v_cursor%rowtype;
    v_cursor_records t_cursor_records;

begin
    open v_cursor; 
    loop
        fetch v_cursor
         bulk collect
         into v_cursor_records
        limit k_bulk_buffer_size; 
        
        forall v_index in 1 .. v_cursor_records.count 
           insert into product_dim(product_id, product_name) 
           select v_cursor_records(v_index).product_id
                , v_cursor_records(v_index).product_name 
             from dual
            where not exists 
                 ( select null 
                     from product_dim
                    where product_id = v_cursor_records(v_index).product_id
                  );
                  
        forall v_index in 1 .. v_cursor_records.count 
           insert into supplier_dim(supplier_id, supplier_name)
           select v_cursor_records(v_index).supplier_id
                 , v_cursor_records(v_index).supplier_name
              from dual
            where not exists 
                 ( select null 
                     from supplier_dim
                    where supplier_id = v_cursor_records(v_index).supplier_id
                  ); 
        
        forall v_index in 1 .. v_cursor_records.count
           insert into customer_dim(customer_id, customer_name,customer_account_type)
              select v_cursor_records(v_index).customer_id
                   , v_cursor_records(v_index).customer_name
                   , v_cursor_records(v_index).customer_account_type 
                from dual
            where not exists 
                 ( select null 
                     from customer_dim
                    where customer_id = v_cursor_records(v_index).customer_id
                  );  
                  
        forall v_index in 1 .. v_cursor_records.count
           insert into outlet_dim(outlet_id, outlet_name)
              select v_cursor_records(v_index).outlet_id
                   , v_cursor_records(v_index).outlet_name 
                from dual
            where not exists 
                 ( select null 
                     from outlet_dim
                    where outlet_id = v_cursor_records(v_index).outlet_id
                  ); 
                  
        forall v_index in 1 .. v_cursor_records.count
           insert into date_dim(d_date, d_year, d_quater, d_month, d_day)
              select v_cursor_records(v_index).d_date
                   , extract(year from v_cursor_records(v_index).d_date)
                   , to_char(v_cursor_records(v_index).d_date,'Q')
                   , extract(month from v_cursor_records(v_index).d_date)
                   , extract(day from v_cursor_records(v_index).d_date)
                from dual
            where not exists 
                 ( select null 
                     from outlet_dim
                    where  outlet_id = v_cursor_records(v_index).outlet_id
                  );                  
 
        forall v_index in 1 .. v_cursor_records.count
           insert into sales_fact( customer_id
                                 , product_id
                                 , outlet_id
                                 , supplier_id
                                 , d_date
                                 , sale_price
                                 , total_sale
                                 , quantity_sold
                                 )
              select v_cursor_records(v_index).customer_id
                   , v_cursor_records(v_index).product_id
                   , v_cursor_records(v_index).outlet_id
                   , v_cursor_records(v_index).supplier_id 
                   , v_cursor_records(v_index).d_date
                   , v_cursor_records(v_index).sale_price
                   , v_cursor_records(v_index).quantity_sold
                     * v_cursor_records(v_index).sale_price
                   , v_cursor_records(v_index).quantity_sold
               from dual
              where not exists 
                  ( select null 
                      from sales_fact
                     where product_id = v_cursor_records(v_index).product_id
                       and customer_id = v_cursor_records(v_index).customer_id
                       and supplier_id = v_cursor_records(v_index).supplier_id
                       and outlet_id = v_cursor_records(v_index).outlet_id
                       and d_date = v_cursor_records(v_index).d_date
                       and sale_price = v_cursor_records(v_index).sale_price
                       and quantity_sold = v_cursor_records(v_index).quantity_sold
                  );
                  
       exit when v_cursor_records.count < k_bulk_buffer_size;              

    end loop; 
 
    close v_cursor;    
    commit;
    
end  transactioninlj; 

注意：源表的 DDL 未包含在您的帖子中，因此我不得不“发明”DATASTREAM 的定义。但是，您只有 2 个源输入：DATASTREAM 和 MASTERDATA。由于您只从 masterdata 中选择 5 列，因此其他所有内容都必须来自数据流。