Ramin Gharib created FLINK-41351:
------------------------------------
Summary: Release Testing: Verify FLIP-578: In-place Table to
Materialized Table conversion
Key: FLINK-41351
URL: https://issues.apache.org/jira/browse/FLINK-41351
Project: Flink
Issue Type: Sub-task
Components: Table SQL / Gateway
Reporter: Ramin Gharib
Fix For: 2.4.0
[FLIP-578|https://cwiki.apache.org/confluence/display/FLINK/FLIP-578%3A+In-place+Table+to+Materialized+Table+conversion]
lets {{CREATE OR ALTER MATERIALIZED TABLE}} turn an existing regular table
into a materialized table in place. The catalog entry and its storage stay.
Everything else comes from the statement, and nothing is copied from the old
table. The feature is off unless
{{table.materialized-table.conversion-from-table.enabled: true}} is set in
{{config.yaml}}. If the refresh job cannot start, the table stays a
materialized table in {{SUSPENDED}} status.
Docs: [Converting a Table to a Materialized
Table|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/sql/materialized-table/statements/#converting-a-table-to-a-materialized-table].
The schema rules are covered by unit tests. This ticket is about the
end-to-end path on a real cluster.
h3. 0. Setup
Build {{release-2.4}}, or use the 2.4.0 RC and take
{{flink-table-filesystem-test-utils-2.4.0.jar}} from the staging repository.
{code:bash}
./mvnw clean install -DskipTests -Dfast -Pskip-webui-build -T1C
cd build-target
cp
../flink-test-utils-parent/flink-table-filesystem-test-utils/target/flink-table-filesystem-test-utils-2.4-SNAPSHOT.jar
lib/
mkdir -p /tmp/flip578/catalog/mydb /tmp/flip578/catalog-store
/tmp/flip578/checkpoints /tmp/flip578/savepoints
{code}
In {{conf/config.yaml}}, change {{numberOfTaskSlots: 1}} to {{4}}, and append
the following. Leave the conversion flag out for now.
{code:yaml}
execution.checkpointing.dir: file:///tmp/flip578/checkpoints
execution.checkpointing.savepoint-dir: file:///tmp/flip578/savepoints
table.catalog-store.kind: file
table.catalog-store.file.path: /tmp/flip578/catalog-store
workflow-scheduler.type: embedded
sql-gateway.endpoint.rest.address: 127.0.0.1
sql-gateway.endpoint.rest.port: 8083
{code}
Start the cluster with {{./bin/start-cluster.sh}} and {{./bin/sql-gateway.sh
start}}. Then connect with {{./bin/sql-client.sh gateway --endpoint
http://127.0.0.1:8083}}. Do not set {{table.dml-sync}}, because it makes the
conversion hang.
{code:sql}
SET 'sql-client.execution.result-mode' = 'tableau';
CREATE CATALOG mt_cat WITH ('type' = 'test-filesystem', 'path' =
'/tmp/flip578/catalog', 'default-database' = 'mydb');
USE CATALOG mt_cat;
CREATE TABLE orders (user_id BIGINT NOT NULL, amount BIGINT, order_time
TIMESTAMP(3), ds STRING)
WITH ('format' = 'json', 'source.monitor-interval' = '10s');
INSERT INTO orders VALUES
(1, 10, TIMESTAMP '2026-11-01 10:00:00', '2026-11-01'),
(2, 20, TIMESTAMP '2026-11-01 11:00:00', '2026-11-01'),
(1, 30, TIMESTAMP '2026-11-02 09:00:00', '2026-11-02');
-- Regular tables to convert, each with a row written before the conversion
CREATE TABLE large_orders (
user_id BIGINT NOT NULL, amount BIGINT, order_time TIMESTAMP(3),
WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH ('format' = 'json');
INSERT INTO large_orders VALUES (100, 500, TIMESTAMP '2026-10-31 09:00:00');
CREATE TABLE daily_spending (ds STRING, user_id BIGINT, total_amount BIGINT)
PARTITIONED BY (ds) WITH ('format' = 'json');
INSERT INTO daily_spending VALUES ('2026-10-31', 100, 1000);
{code}
h3. 1. The flag
{code:sql}
CREATE OR ALTER MATERIALIZED TABLE large_orders
WITH ('format' = 'json', 'sink.rolling-policy.rollover-interval' = '10s',
'sink.rolling-policy.check-interval' = '10s')
FRESHNESS = INTERVAL '30' SECOND
AS SELECT user_id, amount, order_time FROM orders WHERE amount >= 20;
-- [ERROR] ... ValidationException: Regular table does not support create or
alter operation.
SET 'table.materialized-table.conversion-from-table.enabled' = 'true';
-- The same statement fails the same way. The flag is read from the cluster
configuration only.
-- SHOW CREATE TABLE large_orders is still CREATE TABLE.
{code}
Add {{table.materialized-table.conversion-from-table.enabled: true}} to
{{config.yaml}}. Restart the gateway, reconnect, and run {{USE CATALOG
mt_cat;}} again.
h3. 2. Continuous mode
Run the {{large_orders}} statement from section 1 again.
{code:sql}
SHOW CREATE MATERIALIZED TABLE large_orders;
-- REFRESH_MODE = CONTINUOUS, and 'path' =
'/tmp/flip578/catalog/mydb/large_orders/data' as before
DESCRIBE large_orders;
-- No watermark. It is not carried over.
SHOW JOBS;
-- Materialized_table_`mt_cat`.`mydb`.`large_orders`_continuous_refresh_job
RUNNING
-- About a minute later, once files are committed on a checkpoint:
SET 'execution.runtime-mode' = 'batch';
SELECT * FROM large_orders ORDER BY order_time;
-- 100 | 500 | 2026-10-31 09:00:00.000 <- written before the conversion
-- 2 | 20 | 2026-11-01 11:00:00.000
-- 1 | 30 | 2026-11-02 09:00:00.000
{code}
h3. 3. Full mode
{code:sql}
SET 'execution.runtime-mode' = 'streaming';
INSERT INTO orders VALUES (3, 7, LOCALTIMESTAMP, CAST(CURRENT_DATE AS STRING));
CREATE OR ALTER MATERIALIZED TABLE daily_spending
PARTITIONED BY (ds)
WITH ('format' = 'json', 'partition.fields.ds.date-formatter' = 'yyyy-MM-dd')
FRESHNESS = INTERVAL '1' MINUTE
REFRESH_MODE = FULL
AS SELECT ds, user_id, SUM(amount) AS total_amount FROM orders GROUP BY ds,
user_id;
ALTER MATERIALIZED TABLE daily_spending REFRESH PARTITION (ds = '2026-11-01');
-- After the next full minute:
SET 'execution.runtime-mode' = 'batch';
SELECT * FROM daily_spending ORDER BY ds, user_id;
-- 2026-10-31 | 100 | 1000 <- written before the conversion, not touched
-- 2026-11-01 | 1 | 10 <- REFRESH PARTITION
-- 2026-11-01 | 2 | 20
-- <today> | 3 | 7 <- the periodic refresh
{code}
h3. 4. Suspend, resume and gateway restart
{code:sql}
SET 'execution.runtime-mode' = 'streaming';
ALTER MATERIALIZED TABLE large_orders SUSPEND; -- the refresh job goes to
FINISHED with a savepoint
ALTER MATERIALIZED TABLE large_orders RESUME; -- a new refresh job restores
from that savepoint
{code}
Restart the gateway and reconnect. {{SUSPEND}} and {{RESUME}} on
{{large_orders}} still work, because the refresh handler is in the catalog.
h3. 5. The refresh job fails to start
{code:sql}
CREATE TABLE spending_json (user_id BIGINT NOT NULL, total_amount BIGINT) WITH
('format' = 'json');
-- 'json' cannot store the updates that a GROUP BY produces
CREATE OR ALTER MATERIALIZED TABLE spending_json
WITH ('format' = 'json')
FRESHNESS = INTERVAL '30' SECOND
AS SELECT user_id, SUM(amount) AS total_amount FROM orders GROUP BY user_id;
-- [ERROR] ... TableException: Table sink 'mt_cat.mydb.spending_json' doesn't
support consuming update changes ...
-- With SET 'sql-client.verbose' = 'true', the trace also says the table "was
converted and left in SUSPENDED status".
SHOW CREATE MATERIALIZED TABLE spending_json; -- now a materialized table
-- Expected per the docs: fix the cause, then RESUME
ALTER MATERIALIZED TABLE spending_json SET ('format' = 'debezium-json');
ALTER MATERIALIZED TABLE spending_json RESUME;
-- Actual: both fail with NullPointerException: Cannot read the array length
because "buf" is null. This is FLINK-41350.
-- Only DROP MATERIALIZED TABLE spending_json works.
{code}
Clean up with {{DROP MATERIALIZED TABLE}} on each converted table. Their
refresh jobs are cancelled and their directories deleted.
h3. Exploratory
* Convert a table while another job is still writing into it.
* Convert a partitioned table with different partition keys, or none.
h3. Reporting
Comment with the commit or RC and give pass or fail per section. File new bugs
as linked issues with Fix Version 2.4.0.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
