Databricks DLT にて CTAS が実施されたテーブルをソースにした際の DIFFERENT_DELTA_TABLE_READ_BY_STREAMING_SOURCE エラーへの対応方法

## 概要

Databricks DLT の STREMIGN TABLE の処理にて CTAS が実施されたテーブルをソースとした際に発生する DIFFERENT_DELTA_TABLE_READ_BY_STREAMING_SOURCE エラーへの対応方法を共有します。

```python
table_01 = "table_01"
@dlt.table(
  name = table_01,
)
def customers_bronze_ingest_flow():
  return (
      spark.readStream.table(f"{catalog}.dlt_source_01.table_01")
  )
```

> org.apache.spark.sql.streaming.StreamingQueryException: [STREAM_FAILED] Query [id = 766d828a-ed98-430e-b880-895337ff75ff, runId = 34a0cd29-2085-4f6b-8c9d-2d279e7a7827] terminated with exception: [DIFFERENT_DELTA_TABLE_READ_BY_STREAMING_SOURCE] The streaming query was reading from an unexpected Delta table (id = 'd14151f7-b18c-40fa-bbfa-57e9d7ff147a'). 

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/89aa8b91-7c69-46a1-9338-f0ebf4537381.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cf23af55-3538-4c1e-977f-6c7324626850.png)

対応方法としては下記です。CTAS の実行頻度に応じて対応方法を選択してください。

1. Full refresh により処理を再実行する方法
1. Materialized View に変更する方法

## 事前準備

### カレントカタログを変更

```sql
%sql
USE CATALOG manabian_test_01;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6fa3656e-fc58-4b96-b73c-7b1b4913e509.png)


### ソースのスキーマとテーブルを作成

```sql
%sql
CREATE SCHEMA IF NOT EXISTS dlt_source_01
```

```sql
%sql
CREATE OR REPLACE TABLE dlt_source_01.table_01
AS
SELECT
  1 AS id,
  'John' AS string_col;
SELECT * FROM dlt_source_01.table_01;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0d6d0875-b16b-4f9d-a7b8-e30299557a74.png)


### DLT にて初回の処理を実行

```python
from dlt import *
from pyspark.sql.functions import *


catalog = "manabian_test_01"

# table_01 の処理
table_01 = "table_01"
@dlt.table(
  name = table_01,
)
def customers_bronze_ingest_flow():
  return (
      spark.readStream.table(f"{catalog}.dlt_source_01.table_01")
  )
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/64f9e5db-d261-440c-85d7-5efac8de5ba0.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c811bf7b-dde6-4c91-87c2-c61766564033.png)


### ソーステーブルに対する CTAS 処理の実行

```sql
%sql
CREATE OR REPLACE TABLE dlt_source_01.table_01
AS
SELECT
  2 AS id,
  'Doe' AS string_col;
SELECT * FROM dlt_source_01.table_01;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9f4fc09e-f273-4c1d-9d14-6ee6b4aed331.png)

### DLT のパイプラインを再実行してエラーとなることを確認

> org.apache.spark.sql.streaming.StreamingQueryException: [STREAM_FAILED] Query [id = 766d828a-ed98-430e-b880-895337ff75ff, runId = 34a0cd29-2085-4f6b-8c9d-2d279e7a7827] terminated with exception: [DIFFERENT_DELTA_TABLE_READ_BY_STREAMING_SOURCE] The streaming query was reading from an unexpected Delta table (id = 'd14151f7-b18c-40fa-bbfa-57e9d7ff147a'). 
> It used to read from another Delta table (id = 'b62c7c6a-c58f-44f4-b31f-d3679462ab7b') according to checkpoint. 
> This may happen when you changed the code to read from a new table or you deleted and 
> re-created a table. Please revert your change or delete your streaming query checkpoint 
> to restart from scratch. SQLSTATE: XXKST
> === Streaming Query ===
> Identifier: manabian_test_01.dlt_test_01.__materialization_mat_ede7c5ed_6bd8_4172_abdc_0adabac2252c_table_01_1 [id = 766d828a-ed98-430e-b880-895337ff75ff, runId = 34a0cd29-2085-4f6b-8c9d-2d279e7a7827]

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/89aa8b91-7c69-46a1-9338-f0ebf4537381.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cf23af55-3538-4c1e-977f-6c7324626850.png)


## 対応方法

### 1. Full refresh により処理を再実行する方法

エラーとなったオブジェクトを、 Full refresh により処理を実行します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/28239d06-2aeb-41ec-b3a9-23e6e1da2373.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a0e86227-def4-4f44-9477-60c5f8788c39.png)

ソースのデータが書き込まれたことを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/35cfce2d-2078-4646-a777-c501f59bba68.png)


### 2. 1. Materialized View に変更する方法

既存の STREAMING TABLE を削除します。

```sql
%sql
DROP TABLE IF EXISTS manabian_test_01.dlt_test_01.table_01;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/77e6b5ff-e0fb-4cee-96f7-2e525f4d8c68.png)

DLT のコードを Materialized View に変更して、処理を実行します。

```python
from dlt import *
from pyspark.sql.functions import *


catalog = "manabian_test_01"

# table_01 の処理
table_01 = "table_01"
@dlt.table(
  name = table_01,
)
def customers_bronze_ingest_flow():
  return (
      spark.table(f"{catalog}.dlt_source_01.table_01")
  )
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9f5285f3-eb75-47d4-987a-b3f0fdba0412.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/31e92035-4e4f-4a69-b249-1abef510abb4.png)


再度 CTAS を実行後に、 DLT の処理が正常に実行できることを確認します。

```sql
%sql
CREATE OR REPLACE TABLE dlt_source_01.table_01
AS
SELECT
  3 AS id,
  'Cat' AS string_col;
SELECT * FROM dlt_source_01.table_01;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3cf88f6c-150e-4f46-9207-3fa0d1b6ad5c.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3b461770-f99c-473b-b2fa-babf9f87560b.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2ab3f525-83e8-420b-a7c5-ac5b98a2e54b.png)