退订
At 2026-11-04 20:16:00, "Ramin Gharib (Jira) via dev" <[email protected]> wrote: >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)
