# 機能設計書：Table API Scala

## 1. 機能概要

### 1.1 処理概要

Table API Scalaは、Apache Flinkにおけるリレーショナルクエリを記述するためのScala言語向けAPIである。Scalaのimplicit変換機能を活用し、より簡潔で表現力豊かなDSLを提供する。

**重要な注意事項**: 本機能はFLIP-265により非推奨（deprecated）となっており、将来のFlinkメジャーバージョンで削除される予定である。新規開発ではJava版のTable API（No.13）の使用が推奨される。

主要コンポーネント：
1. **ImplicitExpressionConversions**: Scalaリテラルから式への暗黙変換
2. **ImplicitExpressionOperations**: 式に対する演算子の拡張
3. **ImplicitTypeConversions**: Scala型からFlink型への変換
4. **package object api**: 全暗黙変換を提供するパッケージオブジェクト

### 1.2 業務上の目的・役割

| 項目 | 内容 |
|------|------|
| 目的 | Scalaの言語機能を活用したより自然なクエリ記述 |
| 役割 | Scala開発者向けの流暢なDSLの提供 |
| 主な利用場面 | 既存Scalaコードベースとの統合、インタラクティブ分析 |
| 非推奨状態 | FLIP-265により非推奨、Java APIへの移行推奨 |

### 1.3 利用シーン・ユースケース

1. **既存Scalaプロジェクト**
   - Scalaで書かれた既存のFlinkアプリケーションの保守
   - 段階的なJava APIへの移行作業

2. **インタラクティブ開発**
   - Scala REPLでのアドホッククエリ
   - ノートブック環境でのデータ探索

3. **簡潔なDSL記述**（非推奨だが従来サポート）
   - Symbolベースのカラム参照（'columnName）
   - 演算子オーバーロード（===, +, -等）

## 2. 入出力仕様

### 2.1 入力仕様

#### 2.1.1 パッケージインポート

```scala
// 全暗黙変換のインポート
import org.apache.flink.table.api._
```

#### 2.1.2 カラム参照方法

| 方式 | 構文 | 説明 |
|------|------|------|
| Symbol | 'columnName | Scala Symbolによるカラム参照 |
| 文字列 | $"columnName" | 文字列補間によるカラム参照 |

### 2.2 出力仕様

Java版Table APIと同一のTable、TableResult等を出力する。

## 3. 処理フロー

### 3.1 暗黙変換階層

```mermaid
flowchart TD
    subgraph ImplicitConversions["暗黙変換"]
        S[Symbol] --> |UnresolvedFieldExpression| E[Expression]
        L[Literal] --> |LiteralXxxExpression| E
        E --> |WithOperations| O[ImplicitExpressionOperations]
    end

    subgraph Operations["演算子拡張"]
        O --> |===| EQ[isEqual]
        O --> |+| PLUS[plus]
        O --> |-| MINUS[minus]
        O --> |&&| AND[and]
        O --> |"||"| OR[or]
    end

    subgraph Result["結果"]
        EQ --> R[Expression]
        PLUS --> R
        MINUS --> R
        AND --> R
        OR --> R
    end
```

### 3.2 式評価フロー

```mermaid
sequenceDiagram
    participant User as Scalaコード
    participant Implicit as Implicit Conversions
    participant Expr as Expression
    participant Table as Table API

    User->>Implicit: 'column + 10
    Implicit->>Implicit: UnresolvedFieldExpression('column)
    Implicit->>Implicit: LiteralIntExpression(10)
    Implicit->>Expr: plus(valueLiteral(10))
    Expr->>Table: Expression AST
```

## 4. 業務ルール・条件

### 4.1 演算子マッピング

| Scala演算子 | Java API相当 | 説明 |
|------------|--------------|------|
| === | isEqual | 等価比較 |
| !== | isNotEqual | 非等価比較 |
| > | isGreater | 大なり |
| >= | isGreaterOrEqual | 大なりイコール |
| < | isLess | 小なり |
| <= | isLessOrEqual | 小なりイコール |
| && | and | 論理AND |
| \|\| | or | 論理OR |
| ! | NOT | 論理NOT |
| + | plus | 加算/文字列連結 |
| - | minus | 減算 |
| * | times | 乗算 |
| / | dividedBy | 除算 |
| % | mod | 剰余 |

### 4.2 暗黙リテラル変換

| Scala型 | Expression変換 |
|---------|---------------|
| Long | LiteralLongExpression |
| Int | LiteralIntExpression |
| Short | LiteralShortExpression |
| Byte | LiteralByteExpression |
| Float | LiteralFloatExpression |
| Double | LiteralDoubleExpression |
| String | LiteralStringExpression |
| Boolean | LiteralBooleanExpression |
| BigDecimal | LiteralScalaDecimalExpression |
| java.sql.Date | LiteralSqlDateExpression |
| java.sql.Time | LiteralSqlTimeExpression |
| java.sql.Timestamp | LiteralSqlTimestampExpression |

### 4.3 Over Window定数

```scala
// 暗黙値として提供
implicit val UNBOUNDED_ROW: Expression   // 行ベース無限範囲開始
implicit val UNBOUNDED_RANGE: Expression // 時間ベース無限範囲開始
implicit val CURRENT_ROW: Expression     // 現在行
implicit val CURRENT_RANGE: Expression   // 現在範囲
```

### 4.4 三項演算子

```scala
// 条件式
($"f0" > 5).?("A", "B")  // if f0 > 5 then "A" else "B"
```

### 4.5 範囲指定

```scala
// カラム範囲選択
withColumns(1 to 3)  // カラム1から3を選択
```

## 5. コードリーディングガイド

### 5.1 プログラム呼び出し階層図

```
package object api
├── extends ImplicitExpressionConversions
│   ├── implicit class WithOperations(Expression)
│   ├── implicit class UnresolvedFieldExpression(Symbol)
│   ├── implicit class LiteralLongExpression(Long)
│   ├── implicit class LiteralIntExpression(Int)
│   ├── implicit class LiteralStringExpression(String)
│   └── ... (他のリテラル変換)
│
├── extends ImplicitTypeConversions
│   └── Scala型 → TypeInformation変換
│
└── trait ImplicitExpressionOperations
    ├── def ===(other): Expression = isEqual(other)
    ├── def !==(other): Expression = isNotEqual(other)
    ├── def &&(other): Expression = and(other)
    ├── def ||(other): Expression = or(other)
    ├── def +(other): Expression = plus(other)
    ├── def -(other): Expression = minus(other)
    ├── def *(other): Expression = times(other)
    ├── def /(other): Expression = dividedBy(other)
    ├── def %(other): Expression = mod(other)
    └── def ?(ifTrue, ifFalse): Expression = ifThenElse(...)
```

### 5.2 データフロー図

```
[Scala リテラル/Symbol]
        │
        ▼
[暗黙変換 implicit conversion]
        │
        ├── Symbol → UnresolvedFieldExpression
        │              └── unresolvedRef(s.name)
        │
        ├── Int/Long/String/... → LiteralXxxExpression
        │                          └── valueLiteral(value)
        │
        └── Expression → WithOperations
                          └── ImplicitExpressionOperations拡張
                                    │
                                    ▼
                          [演算子呼び出し]
                                    │
                                    ├── === → isEqual()
                                    ├── + → plus()
                                    └── ...
                                    │
                                    ▼
                          [Expression AST生成]
                                    │
                                    ▼
                          [Table API処理]
```

### 5.3 関連ファイル一覧

| ファイルパス | 役割 |
|-------------|------|
| `flink-table-api-scala/.../package.scala` | パッケージオブジェクト（全暗黙変換の集約） |
| `flink-table-api-scala/.../ImplicitExpressionConversions.scala` | リテラル→Expression暗黙変換 |
| `flink-table-api-scala/.../ImplicitExpressionOperations.scala` | Expression演算子拡張 |
| `flink-table-api-scala/.../ImplicitTypeConversions.scala` | 型変換 |
| `flink-table-api-scala/.../typeutils/CaseClassTypeInfo.scala` | Case Class型情報 |
| `flink-table-api-scala/.../typeutils/OptionTypeInfo.scala` | Option型情報 |
| `flink-table-api-scala/.../typeutils/TryTypeInfo.scala` | Try型情報 |
| `flink-table-api-scala/.../typeutils/EitherTypeInfo.scala` | Either型情報 |
| `flink-table-api-scala/.../codegen/TypeInformationGen.scala` | 型情報マクロ生成 |

### 5.4 コードリーディング手順

#### Step 1: パッケージオブジェクトの理解

```scala
// flink-table-api-scala/.../package.scala (48-54行目)
package object api extends ImplicitExpressionConversions with ImplicitTypeConversions {
  val FLIP_265_WARNING: String = "All Flink Scala APIs are deprecated..."
}
```

- `import org.apache.flink.table.api._` で全暗黙変換が利用可能

#### Step 2: Symbol変換の理解

```scala
// ImplicitExpressionConversions.scala (92-94行目)
implicit class UnresolvedFieldExpression(s: Symbol) extends ImplicitExpressionOperations {
  def expr: Expression = unresolvedRef(s.name)
}
```

- `'columnName` が `unresolvedRef("columnName")` に変換される

#### Step 3: 演算子オーバーロードの理解

```scala
// ImplicitExpressionOperations.scala (61-82行目)
def &&(other: Expression): Expression = and(other)
def ||(other: Expression): Expression = or(other)
def ===(other: Expression): Expression = isEqual(other)
def !==(other: Expression): Expression = isNotEqual(other)
```

- Scalaの演算子構文でJava API相当の操作を実現

#### Step 4: リテラル変換の理解

```scala
// ImplicitExpressionConversions.scala (101-131行目)
implicit class LiteralLongExpression(l: Long) extends ImplicitExpressionOperations {
  def expr: Expression = valueLiteral(l)
}
implicit class LiteralStringExpression(str: String) extends ImplicitExpressionOperations {
  def expr: Expression = valueLiteral(str)
}
```

- 各Scala型からExpressionへの暗黙変換

## 6. 関連機能・API

### 6.1 関連機能

| 機能名 | 関連種別 | 説明 |
|--------|---------|------|
| Table API Java（No.13） | 推奨代替 | Java言語向けTable API（移行先） |
| 型変換（No.15） | 連携機能 | DataStreamとの相互変換 |
| 組み込み関数（No.16） | 関数提供 | 標準関数の提供 |
| UDF（No.17） | 拡張機能 | ユーザー定義関数 |

### 6.2 使用例

```scala
import org.apache.flink.table.api._

// TableEnvironment作成
val tEnv = TableEnvironment.create(EnvironmentSettings.newInstance().build())

// クエリ例（非推奨だが動作可能）
val result = tEnv.from("MyTable")
  .filter('age > 18)
  .select('name, 'age + 1 as 'nextAge)
  .where('name === "Alice" || 'name === "Bob")

// 三項演算子
val categorized = tEnv.from("Sales")
  .select('amount, ('amount > 1000).?("High", "Low") as 'category)
```

## 7. 設計上の考慮事項

### 7.1 非推奨について

**FLIP-265による非推奨宣言**:
- 全てのFlink Scala APIは非推奨であり、将来のメジャーバージョンで削除予定
- 新規開発ではJava版APIを使用すること
- 既存コードはScalaで記述可能だが、Java APIへの移行を推奨

```scala
@deprecated(since = "1.18.0")
@PublicEvolving
trait ImplicitExpressionOperations extends BaseExpressions[Expression, Expression] {
  // ...
}
```

### 7.2 移行ガイダンス

| Scala API | Java API代替 |
|-----------|-------------|
| 'column | $("column") |
| 'a === 'b | $("a").isEqual($("b")) |
| 'a + 10 | $("a").plus(10) |
| 'cond.?(a, b) | ifThenElse($("cond"), a, b) |

### 7.3 型情報サポート

Scala固有の型に対する型情報サポート：
- **Option[T]**: OptionTypeInfo
- **Try[T]**: TryTypeInfo
- **Either[L, R]**: EitherTypeInfo
- **Case Class**: CaseClassTypeInfo
- **Enumeration**: EnumValueTypeInfo

### 7.4 パフォーマンス考慮事項

| 考慮点 | 説明 |
|--------|------|
| 暗黙変換オーバーヘッド | コンパイル時に解決されるため実行時オーバーヘッドは最小 |
| Case Class直列化 | ScalaCaseClassSerializerによる効率的な直列化 |
| マクロによる型情報生成 | TypeInformationGenマクロでコンパイル時に型情報生成 |

## 8. 用語集

| 用語 | 説明 |
|------|------|
| implicit conversion | Scalaの暗黙型変換機能 |
| implicit class | 暗黙変換を提供するクラス |
| Symbol | Scalaのシンボルリテラル（'name形式） |
| package object | パッケージレベルで定義・エクスポートを行うオブジェクト |
| DSL | Domain Specific Language（ドメイン固有言語） |
| FLIP-265 | Scala API非推奨化を定めたFlink Improvement Proposal |
| Case Class | Scalaのデータクラス（不変、パターンマッチ対応） |
| TypeInformation | Flinkの型情報抽象 |
