Pipeline into procedure with pipeline_source_file()

Help Connectors and Pipelines
1

Hi Guys,
Can somebody provide example of correct pipeline into procedure which loads csv or tsv files and receives into procedure too pipeline source file ?
Thanks!

2

I’m not sure I understand your question.

There is an example of pipelines to stored procedures with a CSV file (with a .txt extension) here:
SingleStoreDB Cloud · SingleStore Documentation.

If this doesn’t answer your question, please clarify.

3

Hi @hanson, thank, I mean possibility to know in procedure source file. I not was able to see it in procedure with “set” on pipeline creating. So, I ask example of pipeline + procedure…

4

hi Ira,

Did you use this link for example : SingleStoreDB Cloud · SingleStore Documentation

Regards,
Manish Kumar

5

Hi @mkumar , in this link need example missed. Thanks

6

Hi again, any ideas ((( ?

7

I have the same problem. I can get this to work fine with json, but not a csv file. When I test the pipeline I get the following error: ERROR 1261 ER_WARN_TOO_FEW_RECORDS: Leaf Error (localhost:3306): Leaf Error (10.249.41.51:3306): Row 1 doesn’t contain data for all columns

If I remove all of the source_file info, it works fine.
Here’s my code:
Create table trs_scrapdata(
lot_id varchar(50),
die_qty varchar(50),
unit_qty varchar(50),
scrap_time varchar(50),
source_file varchar(1000),
shard key(lot_id)
);

delimiter //
create or replace procedure trs_sp_scrapdata (batch query(lot_id varchar(50), die_qty varchar(50), unit_qty varchar(50), scrap_time varchar(50), source_file varchar(1000)))
as
BEGIN
insert into trs_scrapdata (lot_id, die_qty, unit_qty, scrap_time, source_file)
select
*
from batch;
end //
delimiter ;

create or replace pipeline trs_sp_scrapdata as
load data link sds2 ‘data/treadstone/pti/scrap/’
into procedure trs_sp_scrapdata
columns terminated by ‘,’
OPTIONALLY ENCLOSED BY ‘"’
ignore 1 lines
set source_file = pipeline_source_file();

Thanks for the help!
Tim

8

The error you’re encountering seems to be due to the way the data is being selected from the batch in your procedure.
Instead of using “SELECT *”, which attempts to select all columns, we need to specify the columns individually.
This will ensure that the data is correctly inserted into your table.
You need to do below changes in your procedure and pipeline:

CREATE OR REPLACE PROCEDURE prc_trs_scrapdata (batch QUERY(lot_id varchar(50), die_qty varchar(50), unit_qty varchar(50), scrap_time varchar(50), source_file varchar(1000)))
AS
BEGIN
INSERT INTO trs_scrapdata2 (lot_id, die_qty, unit_qty, scrap_time, source_file)
SELECT lot_id, die_qty, unit_qty, scrap_time, source_file
FROM batch;
END

pipeline changes:

create or replace pipeline trs_sp_scrapdata as
load data link sds2 ‘data/treadstone/pti/scrap/’
into procedure trs_sp_scrapdata
(lot_id, die_qty, unit_qty, scrap_time) – columns in file
columns terminated by ‘,’
OPTIONALLY ENCLOSED BY ‘"’
ignore 1 lines
set source_file = pipeline_source_file(); – column which we want to set in the table

Hope this helps everyone!

9

Thank you!
Here’s the final code:

Create table trs_scrapdata(
lot_id varchar(50),
die_qty varchar(50),
unit_qty varchar(50),
scrap_time varchar(50),
source_file varchar(1000),
supplier varchar(20),
shard key(lot_id)
);

delimiter //
create or replace procedure trs_sp_scrapdata (batch query(lot_id varchar(50), die_qty varchar(50), unit_qty varchar(50), scrap_time varchar(50), source_file varchar(1000), supplier varchar(20)))
as
BEGIN
insert into trs_scrapdata (lot_id, die_qty, unit_qty, scrap_time, source_file, supplier)
select
lot_id,
die_qty,
unit_qty,
scrap_time,
source_file,
supplier
from batch;
end //
delimiter ;

create or replace pipeline trs_pl_scrapdata as
load data link sds2 ‘data/treadstone/pti/scrap/’
into procedure trs_sp_scrapdata
(lot_id, die_qty, unit_qty, scrap_time)
columns terminated by ‘,’
OPTIONALLY ENCLOSED BY ‘"’
ignore 1 lines
set source_file = pipeline_source_file(), supplier = ‘PTI’;

1 Like