Databricks のマルチテーブルトランザクション機能の基本的な検証結果

Databricks

Last updated at 2026-04-16Posted at 2026-04-15

概要

Databricks のマルチテーブルトランザクション機能の基本的な検証結果を共有します。

下記の処理を実行して、テーブルの履歴を確認する手順となっております。

単一テーブルで INSERT を 3 回実行
単一テーブルで INSERT + MERGE を実行
単一テーブルで INSERT + DELETE を実行
複数テーブルで INSERT + DELETE を実行

operationがTRANSACTIONという履歴が残ることを確認できました。

複数テーブルでのマルチテーブルトランザクションをためしたところ、同一時刻（例:2026-04-14T15:57:26.192+00:00）でのコミットとなっていることを確認でき、複数テーブルでのトランザクションも想定通りに動作することを確認できました。

検証

事前準備

カタログとスキーマを作成します。

%sql
CREATE CATALOG IF NOT EXISTS multi_tran_01;
CREATE SCHEMA IF NOT EXISTS multi_tran_01.schema_01;

予測最適化をオフにします。これはテーブルの履歴に OPTIMIZE が含まれないようにするためです。

%sql
-- 予測最適化をオフ
ALTER CATALOG multi_tran_01 DISABLE PREDICTIVE OPTIMIZATION;

現在のカタログとスキーマを指定します。

%sql
USE CATALOG multi_tran_01;
USE multi_tran_01.schema_01;

1. 単一テーブルで INSERT を 3 回実行

%sql
DROP TABLE IF EXISTS txn_target_01;
CREATE TABLE txn_target_01 (
  id         INT,
  value      STRING,
  op_version INT,
  updated_at TIMESTAMP
)
USING DELTA
TBLPROPERTIES (
  'delta.feature.catalogManaged' = 'supported'
);

%sql
BEGIN ATOMIC
  INSERT INTO txn_target_01
  VALUES (1, 'ins_1', 0, current_timestamp());
  INSERT INTO txn_target_01
  VALUES (2, 'ins_2', 0, current_timestamp());
  INSERT INTO txn_target_01
  VALUES (3, 'ins_3', 0, current_timestamp());
END;

%sql
SELECT
  *
    EXCEPT (
      userID,userName
    )
FROM
  (
    DESC HISTORY txn_target_01
  )

[
    {
        "operationName": "WRITE",
        "operationParameters": {
            "mode": "Append",
            "statsOnLoad": false,
            "partitionBy": "[]",
        },
    },
    {
        "operationName": "WRITE",
        "operationParameters": {
            "mode": "Append",
            "statsOnLoad": false,
            "partitionBy": "[]",
        },
    },
    {
        "operationName": "WRITE",
        "operationParameters": {
            "mode": "Append",
            "statsOnLoad": false,
            "partitionBy": "[]",
        },
    },
]

[
	{
		"operationName": "WRITE",
		"operationMetrics": {
			"numFiles": "1",
			"numOutputRows": "1",
			"numOutputBytes": "1241"
		}
	},
	{
		"operationName": "WRITE",
		"operationMetrics": {
			"numFiles": "1",
			"numOutputRows": "1",
			"numOutputBytes": "1241"
		}
	},
	{
		"operationName": "WRITE",
		"operationMetrics": {
			"numFiles": "1",
			"numOutputRows": "1",
			"numOutputBytes": "1241"
		}
	}
]

2. 単一テーブルで INSERT + MERGE を実行

%sql
DROP TABLE IF EXISTS txn_target_02;
CREATE TABLE txn_target_02 (
  id         INT,
  value      STRING,
  op_version INT,
  updated_at TIMESTAMP
)
USING DELTA
TBLPROPERTIES (
  'delta.feature.catalogManaged' = 'supported'
);

%sql
BEGIN ATOMIC
  INSERT INTO txn_target_02
    VALUES (1, 'ins_1', 0, current_timestamp()),
          (2, 'ins_2', 0, current_timestamp());
  MERGE INTO
    txn_target_02 AS t
  USING (SELECT
      2 AS id,
      'merge_1_u2' AS value,
      1 AS op_version,
      current_timestamp() AS updated_at
    UNION ALL
    SELECT
      4 AS id,
      'merge_1_i4' AS value,
      1 AS op_version,
      current_timestamp() AS updated_at
  ) AS s
  ON
    t.id = s.id
  WHEN MATCHED THEN UPDATE SET
    t.value = s.value, t.op_version = s.op_version, t.updated_at = s.updated_at
  WHEN NOT MATCHED THEN INSERT (id, value, op_version, updated_at)
    VALUES (s.id, s.value, s.op_version, s.updated_at);
END;

%sql
SELECT
  *
    EXCEPT (
      userID,userName
    )
FROM
  (
    DESC HISTORY txn_target_02
  );

[
    {
        "operationName": "WRITE",
        "operationParameters": {
            "mode": "Append",
            "statsOnLoad": false,
            "partitionBy": "[]",
        },
    },
    {
        "operationName": "MERGE",
        "operationParameters": {
            "predicate": '["(id#14557 = id#14449)"]',
            "clusterBy": "[]",
            "matchedPredicates": '[{"actionType":"update"}]',
            "statsOnLoad": false,
            "notMatchedBySourcePredicates": "[]",
            "notMatchedPredicates": '[{"actionType":"insert"}]',
        },
    },
]

[
	{
		"operationName": "WRITE",
		"operationMetrics": {
			"numFiles": "1",
			"numOutputRows": "2",
			"numOutputBytes": "1327"
		}
	},
	{
		"operationName": "MERGE",
		"operationMetrics": {
			"numTargetRowsCopied": "0",
			"numTargetRowsDeleted": "0",
			"numTargetBytesRemoved": "0",
			"numTargetDeletionVectorsAdded": "1",
			"numTargetRowsMatchedUpdated": "1",
			"numTargetRowsMatchedDeleted": "0",
			"numTargetRowsUpdated": "1",
			"numTargetChangeFilesAdded": "0",
			"numTargetRowsNotMatchedBySourceDeleted": "0",
			"rewriteTimeMs": "1913",
			"numTargetFilesAdded": "2",
			"numTargetBytesAdded": "3914",
			"executionTimeMs": "3121",
			"materializeSourceTimeMs": "1",
			"numTargetRowsInserted": "1",
			"numTargetDeletionVectorsUpdated": "0",
			"scanTimeMs": "1198",
			"numOutputRows": "2",
			"numTargetDeletionVectorsRemoved": "0",
			"numTargetRowsNotMatchedBySourceUpdated": "0",
			"numSourceRows": "2",
			"numTargetFilesRemoved": "0"
		}
	}
]

3. 単一テーブルで INSERT + DELETE を実行

%sql
DROP TABLE IF EXISTS txn_target_03;
CREATE TABLE txn_target_03 (
  id         INT,
  value      STRING,
  op_version INT,
  updated_at TIMESTAMP
)
USING DELTA
TBLPROPERTIES (
  'delta.feature.catalogManaged' = 'supported'
);

%sql

BEGIN ATOMIC
  INSERT INTO txn_target_03
    VALUES (1, 'ins_1', 0, current_timestamp()),
          (2, 'ins_2', 0, current_timestamp());
  DELETE FROM txn_target_03
  WHERE id = 2;
END;

%sql
SELECT
  *
    EXCEPT (
      userID,userName
    )
FROM
  (
    DESC HISTORY txn_target_03
  );

[
    {
        "operationName": "DELETE",
        "operationParameters": {"predicate": '["(id#15302 = 2)"]'},
    }
]

[
	{
		"operationName": "DELETE",
		"operationMetrics": {
			"numRemovedFiles": "0",
			"numRemovedBytes": "0",
			"numCopiedRows": "0",
			"numDeletionVectorsAdded": "0",
			"executionTimeMs": "481",
			"numDeletionVectorsUpdated": "0",
			"numAddedFiles": "0",
			"numDeletionVectorsRemoved": "0",
			"numAddedChangeFiles": "0",
			"numDeletedRows": "0",
			"scanTimeMs": "455",
			"numAddedBytes": "0",
			"rewriteTimeMs": "0"
		}
	}
]

TRUNCATE TABLE はサポートされていないため、 DELETE で記述する必要があります。

[UNSUPPORTED_FEATURE.SQL_SCRIPTING_WITH_UNSUPPORTED_COMMANDS_IN_ATOMIC_BODY] The feature is not supported: Command TRUNCATETABLE is currently unsupported in Atomic blocks. Only DML commands (such as INSERT, UPDATE, DELETE, MERGE), SELECT and variable assignments are allowed within atomic transaction blocks in SQL Scripting. SQLSTATE: 0A000

4. 複数テーブルで INSERT + DELETE を実行

%sql
DROP TABLE IF EXISTS txn_target_10_source;
CREATE TABLE txn_target_10_source (
  id         INT,
  value      STRING,
  op_version INT,
  updated_at TIMESTAMP
)
USING DELTA
TBLPROPERTIES (
  'delta.feature.catalogManaged' = 'supported'
);

DROP TABLE IF EXISTS txn_target_10_target;
CREATE TABLE txn_target_10_target (
  id         INT,
  value      STRING,
  op_version INT,
  updated_at TIMESTAMP
)
USING DELTA
TBLPROPERTIES (
  'delta.feature.catalogManaged' = 'supported'
);

%sql
INSERT INTO txn_target_10_source
  VALUES (1, 'ins_1', 0, current_timestamp()), (2, 'ins_2', 0, current_timestamp());ｚ

%sql
BEGIN ATOMIC
  INSERT INTO txn_target_10_target
  SELECT * FROM txn_target_10_source;
  DELETE FROM txn_target_10_source;
END;

%sql
SELECT
  * EXCEPT (userID, userName)
FROM
  (
    DESC HISTORY txn_target_10_source
  );


%sql
SELECT
  * EXCEPT (userID, userName)
FROM
  (
    DESC HISTORY txn_target_10_target
  );

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up