ローカルファイル

ローカルファイルConnectorは、ファイルシステムからCSV、Parquet、JSON Lines、Icebergのテーブルを読み、ファイルを一覧してコンテンツとして読み、テーブルとファイルを書き戻し、新しいテーブルを作り、ファイルを削除・移動します。手元のマシンにすでにあるデータ（エクスポート、フィクスチャ、他のツールの出力）のためのConnectorであり、DWHに向ける前の開発段階のプログラムのためのConnectorでもあります。その処理はすべてHost自身で動きます。Siteは常に host:local です。

読み取り元Copy Anchor Link

接続で宣言した各テーブルが読み取り元です。local.orders、local.customers のように参照します。テーブルは、1つのファイル（または同じスキーマを持つファイルのディレクトリ）と形式の組です。

• CSV — 見出しの行が列名になります。型は接続で宣言するか、サンプルから推定したうえで固定します。
• Parquet — 列と型はファイルから取ります。
• JSON Lines — 1行に1オブジェクトです。列と型は宣言します。
• Iceberg — Icebergカタログのディレクトリにあるテーブルです。列と型はテーブルのメタデータから取り、読み取りはスナップショットに固定されます。

テーブルのキーは、接続で宣言したもの（"key": ["id"]）です。キーを宣言していないテーブルにはキーがなく、キーが必要なEffectは check で拒否されます。

テーブルの宣言とは別に、ファイルも読み取り元です。

• local.files — ルートの下のファイル。resource、version（サイズと更新日時のハッシュ。接続で "version": "content" を指定すると内容のハッシュ）、path、size、modified_time。LIST_FILES(local, glob => r'inbound/*.pdf') はパスによる同じ一覧です。一覧の読み取り元が1つなので、接続名だけで書けます。
• local.contents — ファイルの内容。resource、version、path、media_type（拡張子から決まり、接続で上書きできる）、そして content_ref の content。LIST_FILES から resource と version でJOINしてください。

対応するストレージのConnectorが設定されていれば、パスをバケットの中（gs://…、s3://…）に向けることもできます。ファイルはそのConnectorを通して読まれます。

型Copy Anchor Link

Effectと操作Copy Anchor Link

テーブル：

• store rows — EFFECT STORE local.<table> は、プログラムのレコードを宣言済みのテーブルに書きます。CSV、Parquet、JSON Linesのテーブルはレコードで置き換えられ、Icebergのテーブルには1つのスナップショットとして追記されます。テーブルのパスと形式は接続で決まり、プログラムには書きません。
• create table — |> CREATE [OR REPLACE] TABLE local.<name> は、レコードをルートの下に新しいテーブルとして、接続の既定の形式（"default_format": "parquet"）で書き、あとのプログラムが読めるように宣言します。CREATE はテーブルがあれば失敗し、CREATE OR REPLACE は置き換えます。
• delete rows — EFFECT DELETE local.<table> ID key は、テーブルが宣言するキーでレコードを削除します。CSV、Parquet、JSON Linesのテーブルはそれらを除いて書き直され、Icebergのテーブルには削除のスナップショットが1つ付きます。

ファイル：

• write file — EFFECT STORE local.files SET (path = …, content = content [, overwrite = TRUE]) は、レコードの content（SERIALIZE または contents の読み取り元から来たもの）をルートの下のパスに書きます。
• delete file — EFFECT DELETE local.files ID path は、ファイルを削除します。

操作：

• move — EFFECT COMMAND local.move TARGET path ID key INPUT (destination = …) は、ルートの下でファイルの名前を変えるか移動します。宛先は、overwrite = TRUE も指定しない限り存在してはいけません。

パスは root の下で解決されます。プログラムからその外のパスを指定することはできず、Effectからもできません。

セットアップCopy Anchor Link

"local": {
  "connector": "local",
  "root": "./data",
  "default_format": "parquet",
  "tables": {
    "orders": {"path": "orders.parquet", "key": ["id"]},
    "customers": {"path": "customers.csv", "key": ["id"], "columns": {
      "id": "string", "name": "string", "segment": "string", "signed_up": "date"
    }},
    "events": {"path": "warehouse/events", "format": "iceberg"},
    "customer_segments": {"path": "out/customer_segments.parquet", "key": ["id"]}
  }
}

認証情報は不要です。root は設定ファイルからの相対パスです。Hostのプロセスには、読み取り元のためにルートの下の読み取り権限が、Effectのために書き込み権限が必要です。それ以外は要りません。ActQL Cloudでは、local の接続はCloudのワーカー自身のストレージを指し、同じように範囲が限定されます。

CapabilityとプッシュダウンCopy Anchor Link

プレビューと突き合わせCopy Anchor Link

プレビュー。 テーブルへの保存とファイルの書き込みは targeted です。プレビューには、テーブル（またはパス）、その形式、件数またはコンテンツの識別子とバイト数、既存のファイルを置き換えるかどうかが出ます。削除と移動は full_diff です。現在のファイルのサイズ、更新日時、チェックサムを読んで表示し、操作をそれらの条件付きにするので、その間に変わったファイルは項目が failed（conflict）になります。Icebergへの追記は、続くスナップショットを指定し、別の書き込み側が先にコミットしていれば失敗します。

突き合わせ。 書き込みは一時ファイルに行ってから本来の場所にリネームするので、クラッシュしても書きかけのファイルが残ることはありません。Icebergへの追記は1つのスナップショットをコミットします。ローカルファイルで応答が失われることはありませんが、リネームとその記録の間でクラッシュすることはあります。Connectorはリネームの前に項目（id、宛先、コンテンツの識別子）をルートの下のジャーナルに記録し、突き合わせではそのジャーナルと、Icebergならサマリーにidを持つスナップショットを読みます。チェックサムが偶然一致するファイルや、ファイルがないことだけでは証跡になりません。宛先に別のチェックサムのファイルがあっても証跡になりません。項目は unknown のままで、何も書き直しません。

上限と費用Copy Anchor Link

• ファイルはRunの開始時に読まれ、固定されます。Runの途中で変わったファイルは読み直されず、Icebergの読み取りはそのスナップショットにとどまります。
• 読み取りはプログラムの件数とバイト数の上限で抑えられます。バイト数の上限より大きいファイルは、途中で切られるのではなく拒否されます。
• 2つのHostが同じルートの下に同時に書き込んではいけません。Connectorは書き込み中、テーブルごとにアドバイザリロックを取ります。
• 外部への課金はありません。上限はマシンのディスクとメモリ、そしてHostの容量ポリシーです。

例Copy Anchor Link

顧客セグメント別の注文合計です。

FROM local.orders AS o
JOIN local.customers AS c ON o.customer_id = c.id
|> WHERE o.status = 'open'
|> AGGREGATE COUNT(*) AS orders, SUM(o.amount) AS total GROUP BY c.segment
|> ORDER BY total DESC;

顧客をメモから分類し、結果をParquetのテーブルとして書き出します。

FROM local.customers
|> INFER `model-task:segment.v1` INPUT (name, notes) AS seg
|> WHERE seg.outcome = 'ok'
|> SELECT id, seg.value.segment AS segment
|> EFFECT STORE local.customer_segments;

Runが処理し終えた受信ファイルをアーカイブします。

FROM LIST_FILES(local, glob => r'inbound/*.csv') AS f
|> WHERE f.modified_time < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
|> EFFECT COMMAND local.move TARGET path ID path INPUT (destination = 'archive/' || path);