# 低レベルksqlDBクライアント

低レベルksqlDBクライアントは、ksqlDBの全機能に直接アクセスできる強力なAPIです。生のSQL文を実行し、ストリーミング処理、複雑な集約、ウィンドウ関数などのksqlDBの高度な機能を活用できます。

## 特徴

- 🚀 **フル機能** - ksqlDBのすべての機能にアクセス
- 📊 **ストリーミング** - リアルタイムデータ処理
- 🔄 **プッシュクエリ** - 継続的なデータストリーミング
- ⚡ **プルクエリ** - 一度だけの高速データ取得
- 🛠️ **DDL/DML** - スキーマ作成・データ操作
- 🔧 **柔軟性** - カスタムSQL文の実行
- ♾️ **自動Retention設定** - デフォルトでトピックの保持期間をinfinityに設定

## セットアップ

```typescript
import { 
  initializeKsqlDbClient,
  executeQuery,
  executePullQuery,
  executePushQuery,
  executeDDL,
  executeDDLWithOptions 
} from '@gftdcojp/gftd-ksqldb-orm';

// クライアント初期化
initializeKsqlDbClient({
  url: 'http://localhost:8088',
  apiKey: 'your-api-key',
  apiSecret: 'your-api-secret'
});
```

## ♾️ Retention Infinity機能

このORMでは、CREATE STREAM/TABLEで作成されるすべてのトピックが**デフォルトでinfinite retention（無期限保持）**に設定されます。

### 自動Retention設定

```typescript
// 通常のDDL実行
await executeDDL(`
  CREATE STREAM user_events (
    user_id INT,
    event_type STRING,
    timestamp STRING
  ) WITH (
    kafka_topic='user_events',
    value_format='JSON'
  );
`);
// ↓ 自動的に'retention.ms'='-1'が追加される
```

### カスタムRetention設定

```typescript
import { executeDDLWithOptions, type DDLExecutionOptions } from '@gftdcojp/gftd-ksqldb-orm';

// retention設定の無効化
const options: DDLExecutionOptions = {
  disableAutoInfinityRetention: true
};
await executeDDLWithOptions(ddl, options);

// カスタムretention（1日）
const customOptions: DDLExecutionOptions = {
  customRetentionMs: 86400000 // 1日間
};
await executeDDLWithOptions(ddl, customOptions);

// 追加のトピック設定
const advancedOptions: DDLExecutionOptions = {
  additionalWithSettings: {
    'cleanup.policy': 'compact',
    'segment.ms': '3600000'
  }
};
await executeDDLWithOptions(ddl, advancedOptions);
```

### ヘルパー関数

便利なヘルパー関数も提供されています：

```typescript
import { 
  createStreamWithInfinityRetention,
  createTableWithInfinityRetention 
} from '@gftdcojp/gftd-ksqldb-orm';

// ストリーム作成ヘルパー
await createStreamWithInfinityRetention(
  'user_events',
  {
    user_id: 'INT',
    event_type: 'STRING',
    timestamp: 'STRING'
  },
  {
    topic: 'events',
    valueFormat: 'AVRO',
    keyField: 'user_id',
    partitions: 6,
    replicas: 3
  }
);

// テーブル作成ヘルパー
await createTableWithInfinityRetention(
  'user_summary',
  'user_events',
  'user_id, COUNT(*) as event_count',
  'user_id'
);
```

### スキーマ定義でのRetention設定

スキーマ定義からもretention設定を制御できます：

```typescript
import { defineSchema, generateDDLFromSchema, createStreamFromSchema } from '@gftdcojp/gftd-ksqldb-orm';
import { string, int } from '@gftdcojp/gftd-ksqldb-orm/field-types';

// デフォルトでinfinite retention
const userSchema = defineSchema('User', {
  id: int().primaryKey(),
  name: string().notNull(),
  email: string()
});

// カスタムretention設定
const customSchema = defineSchema('Event', {
  id: string(),
  data: string()
}, {
  retentionMs: 86400000, // 1日
  cleanupPolicy: 'compact',
  partitions: 12
});

// スキーマからストリーム作成
await createStreamFromSchema('User', 'STREAM');
```

## DDL操作（スキーマ定義）

### ストリーム作成

```typescript
// 基本的なストリーム
await executeDDL(`
  CREATE STREAM users_stream (
    id INT,
    name VARCHAR,
    email VARCHAR,
    created_at VARCHAR
  ) WITH (
    kafka_topic='users',
    value_format='JSON',
    partitions=3
  );
`);

// 複雑なスキーマのストリーム
await executeDDL(`
  CREATE STREAM transaction_events (
    transaction_id VARCHAR,
    user_id INT,
    amount DECIMAL(10,2),
    currency VARCHAR,
    merchant_data STRUCT<
      name VARCHAR,
      category VARCHAR,
      location STRUCT<
        lat DOUBLE,
        lng DOUBLE
      >
    >,
    tags ARRAY<VARCHAR>,
    metadata MAP<VARCHAR, VARCHAR>,
    timestamp VARCHAR
  ) WITH (
    kafka_topic='transactions',
    value_format='AVRO',
    key='transaction_id',
    partitions=6,
    replicas=3
  );
`);

// タイムスタンプ列の指定
await executeDDL(`
  CREATE STREAM user_activities (
    user_id INT,
    activity_type VARCHAR,
    page_url VARCHAR,
    session_id VARCHAR,
    event_time VARCHAR
  ) WITH (
    kafka_topic='user_activities',
    value_format='JSON',
    timestamp='event_time',
    timestamp_format='yyyy-MM-dd HH:mm:ss'
  );
`);
```

### テーブル作成（マテリアライズドビュー）

```typescript
// 基本的な集約テーブル
await executeDDL(`
  CREATE TABLE user_counts AS
  SELECT 
    status,
    COUNT(*) as total_users,
    COUNT_DISTINCT(email) as unique_emails
  FROM users_stream
  GROUP BY status
  EMIT CHANGES;
`);

// 複雑な集約
await executeDDL(`
  CREATE TABLE user_stats AS
  SELECT 
    id,
    LATEST_BY_OFFSET(name) as name,
    LATEST_BY_OFFSET(email) as email,
    COUNT(*) as event_count,
    EARLIEST_BY_OFFSET(created_at) as first_seen,
    LATEST_BY_OFFSET(created_at) as last_seen,
    COLLECT_LIST(activity_type) as activities
  FROM users_stream
  GROUP BY id
  EMIT CHANGES;
`);

// ウィンドウ集約テーブル
await executeDDL(`
  CREATE TABLE sales_hourly AS
  SELECT 
    WINDOWSTART as window_start,
    WINDOWEND as window_end,
    product_id,
    SUM(amount) as total_sales,
    COUNT(*) as order_count,
    AVG(amount) as avg_order_value,
    MAX(amount) as max_order_value
  FROM orders_stream
  WINDOW TUMBLING (SIZE 1 HOUR)
  GROUP BY product_id
  EMIT CHANGES;
`);
```

### JOIN操作

```typescript
// ストリーム-テーブルJOIN
await executeDDL(`
  CREATE STREAM enriched_orders AS
  SELECT 
    o.order_id,
    o.user_id,
    o.amount,
    u.name as user_name,
    u.email as user_email,
    u.status as user_status
  FROM orders_stream o
  LEFT JOIN users_table u ON o.user_id = u.id
  EMIT CHANGES;
`);

// ストリーム-ストリームJOIN（ウィンドウ内）
await executeDDL(`
  CREATE STREAM user_journey AS
  SELECT 
    a.user_id,
    a.page_url as current_page,
    a.timestamp as current_time,
    b.page_url as previous_page,
    b.timestamp as previous_time
  FROM user_activities a
  INNER JOIN user_activities b 
  WITHIN 10 MINUTES
  ON a.user_id = b.user_id
  WHERE a.timestamp > b.timestamp
  EMIT CHANGES;
`);
```

## DML操作（データ操作）

### データ挿入

```typescript
// 単一レコード挿入
await executeQuery(`
  INSERT INTO users_stream (id, name, email, created_at)
  VALUES (1, 'John Doe', 'john@example.com', '2024-01-01T00:00:00Z');
`);

// 複数レコード挿入
await executeQuery(`
  INSERT INTO users_stream (id, name, email, status) VALUES 
    (1, 'Alice', 'alice@example.com', 'active'),
    (2, 'Bob', 'bob@example.com', 'active'),
    (3, 'Charlie', 'charlie@example.com', 'inactive');
`);

// 複雑なデータ構造の挿入
await executeQuery(`
  INSERT INTO transaction_events (
    transaction_id,
    user_id,
    amount,
    merchant_data,
    tags,
    metadata
  ) VALUES (
    'tx_12345',
    100,
    99.99,
    STRUCT(
      name := 'Best Electronics',
      category := 'electronics',
      location := STRUCT(lat := 35.6762, lng := 139.6503)
    ),
    ARRAY['online', 'credit_card'],
    MAP('channel' := 'web', 'campaign' := 'summer_sale')
  );
`);
```

## プルクエリ（一度だけ取得）

### 基本的なプルクエリ

```typescript
// 単純な選択
const result = await executePullQuery(`
  SELECT * FROM users_table WHERE id = 123;
`);

// 複雑な条件
const users = await executePullQuery(`
  SELECT id, name, email, status
  FROM users_table 
  WHERE created_at > '2024-01-01'
  AND status = 'active'
  ORDER BY created_at DESC
  LIMIT 100;
`);

// 集約結果取得
const stats = await executePullQuery(`
  SELECT 
    status, 
    total_users,
    unique_emails
  FROM user_counts
  ORDER BY total_users DESC;
`);
```

### 高度なプルクエリ

```typescript
// 時間範囲指定
const recentSales = await executePullQuery(`
  SELECT 
    window_start,
    window_end,
    product_id,
    total_sales,
    order_count
  FROM sales_hourly
  WHERE window_start >= '2024-01-01T00:00:00'
  AND window_start < '2024-01-02T00:00:00'
  ORDER BY total_sales DESC
  LIMIT 10;
`);

// 複雑な条件とサブクエリ
const topUsers = await executePullQuery(`
  SELECT 
    u.id,
    u.name,
    u.event_count,
    u.last_seen
  FROM user_stats u
  WHERE u.event_count > (
    SELECT AVG(event_count) FROM user_stats
  )
  ORDER BY u.event_count DESC
  LIMIT 20;
`);
```

## プッシュクエリ（リアルタイムストリーミング）

### 基本的なストリーミング

```typescript
// リアルタイムユーザーイベント監視
executePushQuery(
  `SELECT * FROM users_stream EMIT CHANGES;`,
  (data) => {
    console.log('New user event:', data);
    // リアルタイム処理ロジック
    processUserEvent(data);
  },
  (error) => {
    console.error('Stream error:', error);
    // エラーハンドリング
  },
  () => {
    console.log('Stream ended');
    // 終了処理
  }
);

// 条件付きストリーミング
executePushQuery(
  `
  SELECT 
    user_id,
    amount,
    currency,
    timestamp
  FROM transaction_events 
  WHERE amount > 1000
  EMIT CHANGES;
  `,
  (data) => {
    // 高額取引のアラート
    sendHighValueTransactionAlert(data);
  }
);
```

### ウィンドウ関数を使った集約ストリーミング

```typescript
// 1分間隔のアクティブユーザー数
executePushQuery(
  `
  SELECT 
    WINDOWSTART as window_start,
    WINDOWEND as window_end,
    COUNT_DISTINCT(user_id) as active_users,
    COUNT(*) as total_events
  FROM user_activities
  WINDOW TUMBLING (SIZE 1 MINUTE)
  GROUP BY 1
  EMIT CHANGES;
  `,
  (data) => {
    console.log(`Active users in minute ${data.window_start}: ${data.active_users}`);
    updateDashboard('active_users', data);
  }
);

// スライディングウィンドウでの移動平均
executePushQuery(
  `
  SELECT 
    WINDOWSTART,
    WINDOWEND,
    product_id,
    AVG(amount) as moving_avg_price,
    COUNT(*) as orders_in_window
  FROM orders_stream
  WINDOW HOPPING (SIZE 10 MINUTES, ADVANCE BY 1 MINUTE)
  GROUP BY product_id
  EMIT CHANGES;
  `,
  (data) => {
    updatePriceMonitoring(data);
  }
);

// セッションウィンドウ
executePushQuery(
  `
  SELECT 
    user_id,
    COUNT(*) as session_events,
    MIN(timestamp) as session_start,
    MAX(timestamp) as session_end,
    COLLECT_LIST(page_url) as pages_visited
  FROM user_activities
  WINDOW SESSION (60 SECONDS)
  GROUP BY user_id
  EMIT CHANGES;
  `,
  (data) => {
    analyzeUserSession(data);
  }
);
```

## 高度なクエリパターン

### 異常検知

```typescript
// 急激な取引量増加の検知
executePushQuery(
  `
  SELECT 
    WINDOWSTART,
    user_id,
    COUNT(*) as transaction_count,
    SUM(amount) as total_amount,
    AVG(amount) as avg_amount
  FROM transaction_events
  WINDOW TUMBLING (SIZE 5 MINUTES)
  GROUP BY user_id
  HAVING COUNT(*) > 10 OR SUM(amount) > 5000
  EMIT CHANGES;
  `,
  (data) => {
    // 不審な取引パターンのアラート
    triggerFraudAlert(data);
  }
);

// 価格異常の検知
await executeDDL(`
  CREATE STREAM price_anomalies AS
  SELECT 
    product_id,
    current_price,
    avg_price,
    (current_price - avg_price) / avg_price * 100 as price_change_percent
  FROM (
    SELECT 
      product_id,
      price as current_price,
      AVG(price) OVER (
        PARTITION BY product_id 
        RANGE 1 HOUR PRECEDING
      ) as avg_price
    FROM product_price_stream
  )
  WHERE ABS((current_price - avg_price) / avg_price * 100) > 20
  EMIT CHANGES;
`);
```

### イベント駆動アーキテクチャ

```typescript
// カスケード処理の例
await executeDDL(`
  CREATE STREAM order_events AS
  SELECT 
    order_id,
    user_id,
    total_amount,
    status,
    CASE 
      WHEN total_amount > 1000 THEN 'HIGH_VALUE'
      WHEN total_amount > 100 THEN 'MEDIUM_VALUE'
      ELSE 'LOW_VALUE'
    END as order_tier
  FROM orders_stream
  EMIT CHANGES;
`);

// 各レベルでの処理
executePushQuery(
  `SELECT * FROM order_events WHERE order_tier = 'HIGH_VALUE' EMIT CHANGES;`,
  (data) => {
    // VIP顧客向け特別処理
    processVipOrder(data);
    sendVipNotification(data);
  }
);

executePushQuery(
  `SELECT * FROM order_events WHERE status = 'COMPLETED' EMIT CHANGES;`,
  (data) => {
    // 在庫更新トリガー
    updateInventory(data);
    // レコメンデーションエンジン更新
    updateRecommendations(data);
  }
);
```

## 複雑なデータ型の操作

### 配列操作

```typescript
// 配列要素の検索
const result = await executePullQuery(`
  SELECT 
    user_id,
    activities,
    ARRAY_LENGTH(activities) as activity_count
  FROM user_stats
  WHERE ARRAY_CONTAINS(activities, 'purchase');
`);

// 配列の展開
await executeDDL(`
  CREATE STREAM user_activity_flat AS
  SELECT 
    user_id,
    EXPLODE(activities) as activity
  FROM user_stats
  EMIT CHANGES;
`);
```

### JSON/構造体操作

```typescript
// 構造体フィールドアクセス
const result = await executePullQuery(`
  SELECT 
    transaction_id,
    merchant_data->name as merchant_name,
    merchant_data->location->lat as latitude,
    merchant_data->location->lng as longitude
  FROM transaction_events
  WHERE merchant_data->category = 'electronics';
`);

// マップ操作
const metadata = await executePullQuery(`
  SELECT 
    transaction_id,
    metadata['channel'] as channel,
    metadata['campaign'] as campaign
  FROM transaction_events
  WHERE metadata['channel'] IS NOT NULL;
`);
```

## スキーマ管理

### テーブル・ストリーム情報取得

```typescript
// 全テーブル一覧
const tables = await executeQuery('LIST TABLES EXTENDED;');

// 全ストリーム一覧
const streams = await executeQuery('LIST STREAMS EXTENDED;');

// スキーマ確認
const schema = await executeQuery('DESCRIBE users_table;');

// 実行中のクエリ確認
const queries = await executeQuery('LIST QUERIES;');
```

### リソース管理

```typescript
// ストリーム削除
await executeDDL('DROP STREAM users_stream DELETE TOPIC;');

// テーブル削除
await executeDDL('DROP TABLE user_counts DELETE TOPIC;');

// クエリ終了
await executeQuery('TERMINATE QUERY_ID;');

// 全クエリ終了
await executeQuery('TERMINATE ALL;');
```

## 実用的なサンプル

### リアルタイム分析ダッシュボード

```typescript
// ダッシュボード用メトリクス
class RealTimeDashboard {
  constructor() {
    this.setupMetrics();
  }

  setupMetrics() {
    // アクティブユーザー数
    executePushQuery(
      `
      SELECT 
        WINDOWSTART,
        COUNT_DISTINCT(user_id) as active_users
      FROM user_activities
      WINDOW TUMBLING (SIZE 30 SECONDS)
      GROUP BY 1
      EMIT CHANGES;
      `,
      (data) => {
        this.updateMetric('active_users', data.active_users);
      }
    );

    // 収益メトリクス
    executePushQuery(
      `
      SELECT 
        WINDOWSTART,
        SUM(amount) as revenue,
        COUNT(*) as order_count,
        AVG(amount) as avg_order_value
      FROM orders_stream
      WHERE status = 'completed'
      WINDOW TUMBLING (SIZE 1 MINUTE)
      GROUP BY 1
      EMIT CHANGES;
      `,
      (data) => {
        this.updateMetric('revenue', data.revenue);
        this.updateMetric('order_count', data.order_count);
        this.updateMetric('avg_order_value', data.avg_order_value);
      }
    );

    // エラー率監視
    executePushQuery(
      `
      SELECT 
        WINDOWSTART,
        SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as error_count,
        COUNT(*) as total_requests,
        CAST(SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) * 100 as error_rate
      FROM api_requests_stream
      WINDOW TUMBLING (SIZE 1 MINUTE)
      GROUP BY 1
      EMIT CHANGES;
      `,
      (data) => {
        this.updateMetric('error_rate', data.error_rate);
        if (data.error_rate > 5) {
          this.triggerAlert('High error rate detected', data);
        }
      }
    );
  }

  updateMetric(name: string, value: any) {
    // ダッシュボード更新ロジック
    console.log(`Metric ${name}: ${value}`);
  }

  triggerAlert(message: string, data: any) {
    // アラート送信ロジック
    console.error(`ALERT: ${message}`, data);
  }
}
```

### A/Bテスト分析

```typescript
// A/Bテスト結果のリアルタイム集計
await executeDDL(`
  CREATE TABLE ab_test_results AS
  SELECT 
    experiment_id,
    variant,
    COUNT(*) as participant_count,
    SUM(CASE WHEN conversion = true THEN 1 ELSE 0 END) as conversions,
    CAST(SUM(CASE WHEN conversion = true THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) as conversion_rate,
    AVG(revenue) as avg_revenue_per_user
  FROM ab_test_events
  GROUP BY experiment_id, variant
  EMIT CHANGES;
`);

// 統計的有意性の監視
executePushQuery(
  `
  SELECT 
    experiment_id,
    COLLECT_LIST(
      STRUCT(
        variant := variant,
        conversion_rate := conversion_rate,
        participant_count := participant_count
      )
    ) as variants
  FROM ab_test_results
  GROUP BY experiment_id
  EMIT CHANGES;
  `,
  (data) => {
    // 統計的有意性をチェック
    const { isSignificant, winner } = analyzeStatisticalSignificance(data.variants);
    
    if (isSignificant) {
      console.log(`Experiment ${data.experiment_id} has significant results. Winner: ${winner}`);
      // 実験終了の提案
      suggestExperimentConclusion(data.experiment_id, winner);
    }
  }
);
```

### ログ分析とアラート

```typescript
// エラーログのパターン検出
await executeDDL(`
  CREATE STREAM error_patterns AS
  SELECT 
    WINDOWSTART,
    error_type,
    service_name,
    COUNT(*) as error_count,
    COLLECT_LIST(error_message) as error_messages
  FROM application_logs
  WHERE log_level = 'ERROR'
  WINDOW TUMBLING (SIZE 5 MINUTES)
  GROUP BY error_type, service_name
  HAVING COUNT(*) > 10
  EMIT CHANGES;
`);

// アラート生成
executePushQuery(
  `SELECT * FROM error_patterns EMIT CHANGES;`,
  (data) => {
    const alert = {
      type: 'ERROR_SPIKE',
      service: data.service_name,
      errorType: data.error_type,
      count: data.error_count,
      timeWindow: data.WINDOWSTART,
      samples: data.error_messages.slice(0, 3)
    };
    
    // アラートシステムに送信
    sendAlert(alert);
    
    // 自動スケーリングトリガー
    if (data.error_count > 50) {
      triggerAutoScaling(data.service_name);
    }
  }
);
```

## パフォーマンス最適化

### クエリ最適化のベストプラクティス

```typescript
// 1. 適切なパーティショニング
await executeDDL(`
  CREATE STREAM optimized_events (
    user_id INT,
    event_data VARCHAR,
    timestamp VARCHAR
  ) WITH (
    kafka_topic='events',
    value_format='JSON',
    partitions=12,  -- 適切なパーティション数
    key='user_id'   -- 効率的なパーティショニング
  );
`);

// 2. インデックス的な使用パターン
await executeDDL(`
  CREATE TABLE user_lookup AS
  SELECT 
    id,
    LATEST_BY_OFFSET(name) as name,
    LATEST_BY_OFFSET(email) as email
  FROM users_stream
  GROUP BY id
  EMIT CHANGES;
`);

// 3. 効率的なウィンドウサイズ
// 小さすぎる → 高頻度更新でオーバーヘッド
// 大きすぎる → メモリ使用量増加
await executeDDL(`
  CREATE TABLE balanced_metrics AS
  SELECT 
    WINDOWSTART,
    metric_name,
    AVG(value) as avg_value,
    COUNT(*) as count
  FROM metrics_stream
  WINDOW TUMBLING (SIZE 1 MINUTE)  -- バランスの取れたウィンドウサイズ
  GROUP BY metric_name
  EMIT CHANGES;
`);
```

## エラーハンドリングとトラブルシューティング

### 一般的なエラーパターン

```typescript
// リトライ機構付きクエリ実行
const executeWithRetry = async (query: string, maxRetries: number = 3) => {
  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    try {
      return await executeQuery(query);
    } catch (error: any) {
      console.error(`Attempt ${attempt} failed:`, error.message);
      
      if (attempt === maxRetries) {
        throw new Error(`Query failed after ${maxRetries} attempts: ${error.message}`);
      }
      
      // 指数バックオフ
      const delay = Math.pow(2, attempt) * 1000;
      await new Promise(resolve => setTimeout(resolve, delay));
    }
  }
};

// 接続状態監視
const monitorConnection = () => {
  const checkInterval = setInterval(async () => {
    try {
      await executeQuery('LIST STREAMS;');
      console.log('ksqlDB connection healthy');
    } catch (error) {
      console.error('ksqlDB connection lost:', error);
      // 再接続ロジック
      reconnectKsqlDb();
    }
  }, 30000); // 30秒間隔

  return checkInterval;
};
```

### デバッグとモニタリング

```typescript
// クエリ実行時間測定
const executeWithTiming = async (query: string) => {
  const startTime = Date.now();
  try {
    const result = await executeQuery(query);
    const duration = Date.now() - startTime;
    console.log(`Query executed in ${duration}ms`);
    return result;
  } catch (error) {
    const duration = Date.now() - startTime;
    console.error(`Query failed after ${duration}ms:`, error);
    throw error;
  }
};

// システムメトリクス監視
const getSystemMetrics = async () => {
  const serverInfo = await executeQuery('SHOW PROPERTIES;');
  const queryStatus = await executeQuery('LIST QUERIES;');
  
  return {
    serverInfo,
    activeQueries: queryStatus.length,
    timestamp: new Date().toISOString()
  };
};
```