CTEs in PIPELINES

Help Connectors and Pipelines
1

Is there any limitation in using a CTE in a Pipeline?

2

You mean like in SQL called from a stored procedure? This should work.

3

Yes, using a stored procedure with a CTE in a PIPELINE a few things seem to occur.

If the QUERY input is used in the CTE, the following error occurs. (understandable)
Unhandled exception
Type: ER_TEMPORARY_TABLE_INSIDE_CTE
Message: Temporary table DatabaseName.13969829983540540504 cannot appear in the definition of common table expression.
Callstack:
#0 Line 4 in 'DatabaseName'.'Reading_Upser' called from
#1 Line 1 in helper

If a CTE is used without the QUERY input, no record is placed into the table. No errors are logged.
The LATEST_OFFSET increments, the LATEST_LOADED_OFFSET does not in the PIPELINE_OFFSETS table.
The same appears in the PIPELINES_CURSORS table, the LATEST_OFFSET increments, but CURSOR_OFFSET and SUCESSFUL_CURSOR_OFFSET did not increment.
No errors are logged in the PIPELINES_ERRORS table

4

Using temp tables in CTEs is currently not supported. You are not using an explicit temp table are you? It appears that a temp table is getting used internally.

Try using a subquery (nested SELECT in the FROM clause) instead of a CTE if that works for you.

5

When the PIPELINE is a consumer from Kafka, it does not appear to work with a CTE. A basic PIPELINE using Kafka does not allow a CTE.

Below are some of the scripts that are used. Running the PIPELINE via a PROC does seem to work when using Kafka to input the data it does not show the data in the table.

CREATE OR REPLACE PIPELINE GameStats AS
    LOAD DATA KAFKA 'Kafka:9092/GameStats'
    BATCH_INTERVAL 10000
    SKIP PARSER ERRORS INTO PROCEDURE `GameOver`
   FIELDS TERMINATED BY '' ESCAPED BY '' LINES TERMINATED BY ''
    (Player);

CREATE OR REPLACE PROCEDURE GameOver (
    Game QUERY(
        Player varchar(45)
    )
)
AS
BEGIN

	WITH CoinTotals AS (
		SELECT SUM(Coins) AS Coins
						, Player
						
                FROM CoinsKK
				
                GROUP BY Player 
    
    
    )
	INSERT INTO GameStatsKK (Player, TotalCoins)
	SELECT Game.Player
			, CoinTotals.Coins
		FROM Game
			INNER JOIN CoinTotals 
				ON CoinTotals.Player = Game.Player;


	/*INSERT INTO GameStatsKK (Player, TotalCoins)
		SELECT Game.Player
			, Totals.Coins
		FROM Game
			INNER JOIN (
				SELECT SUM(Coins) AS Coins
						, Player
						
                FROM CoinsKK
				
                GROUP BY Player 
            
            ) AS Totals 
				ON Game.Player = Totals.Player;*/
END //

DELIMITER ;