- Extract SQL Generation

- Adjust Cursor function names to camel case to align with rest
- Update create table method

Author: Georg Traar

src/CrateDBClient.ts

@@ -11,8 +11,12 @@ import {
   CrateDBBulkResponse,
   CrateDBRecord,
   CrateDBErrorResponse,
+  OptimizeOptions,
+  ColumnDefinition,
+  TableOptions,
 } from './interfaces';
 import { CrateDBError, DeserializationError, RequestError } from './utils/Error.js';
+import { StatementGenerator } from './StatementGenerator.js';
 
 // Configuration options with CrateDB-specific environment variables
 const defaultConfig: CrateDBConfig = {
@@ -162,23 +166,6 @@ export class CrateDBClient {
     }
   }
 
-  // Convenience methods for common SQL operations
-  private _generateInsertQuery(tableName: string, keys: string[], primaryKeys: string[] | null): string {
-    const placeholders = keys.map(() => '?').join(', ');
-    let query = `INSERT INTO ${tableName} (${keys.map((key) => `"${key}"`).join(', ')}) VALUES (${placeholders})`;
-
-    if (primaryKeys && primaryKeys.length > 0) {
-      const keysWithoutPrimary = keys.filter((key) => !primaryKeys.includes(key));
-      const updates = keysWithoutPrimary.map((key) => `"${key}" = excluded."${key}"`).join(', ');
-      query += ` ON CONFLICT (${primaryKeys.map((key) => `"${key}"`).join(', ')}) DO UPDATE SET ${updates}`;
-    } else {
-      query += ' ON CONFLICT DO NOTHING';
-    }
-
-    query += ';'; // Ensure the query ends with a semicolon
-    return query;
-  }
-
   async insert(
     tableName: string,
     obj: Record<string, unknown>,
@@ -196,7 +183,7 @@ export class CrateDBClient {
     }
 
     const keys = Object.keys(obj);
-    const query = this._generateInsertQuery(tableName, keys, primaryKeys);
+    const query = StatementGenerator.insert(tableName, keys, primaryKeys);
     const args = Object.values(obj);
 
     // Execute the query
@@ -233,7 +220,7 @@ export class CrateDBClient {
       uniqueKeys.map((key) => (Object.prototype.hasOwnProperty.call(obj, key) ? obj[key] : null))
     );
 
-    const query = this._generateInsertQuery(tableName, uniqueKeys, primaryKeys);
+    const query = StatementGenerator.insert(tableName, uniqueKeys, primaryKeys);
 
     // Execute the query with bulk arguments
     const response = await this.executeMany(query, bulkArgs);
@@ -245,32 +232,69 @@ export class CrateDBClient {
 
   async update(tableName: string, options: Record<string, unknown>, whereClause: string): Promise<CrateDBResponse> {
     const { keys, values, args } = this._prepareOptions(options);
-    const setClause = keys.map((key, i) => `${key}=${values[i]}`).join(', ');
-    const query = `UPDATE ${tableName} SET ${setClause} WHERE ${whereClause}`;
+    const query = StatementGenerator.update(tableName, options, whereClause);
     return this.execute(query, args);
   }
 
   async delete(tableName: string, whereClause: string): Promise<CrateDBResponse> {
-    const query = `DELETE FROM ${tableName} WHERE ${whereClause}`;
+    const query = StatementGenerator.delete(tableName, whereClause);
     return this.execute(query);
   }
 
+  /**
+   * Drops a table if it exists in CrateDB.
+   *
+   * Constructs and executes a `DROP TABLE IF EXISTS` SQL statement.
+   *
+   * @param {string} tableName - The name of the table to drop.
+   * @returns {Promise<CrateDBResponse>} A promise resolving to the response from CrateDB.
+   */
   async drop(tableName: string): Promise<CrateDBResponse> {
-    const query = `DROP TABLE IF EXISTS ${tableName}`;
+    const query = StatementGenerator.dropTable(tableName);
     return this.execute(query);
   }
 
+  async createTable(
+    tableName: string,
+    schema: Record<string, ColumnDefinition>,
+    options?: TableOptions
+  ): Promise<CrateDBResponse> {
+    const query = StatementGenerator.createTable(tableName, schema, options);
+    return this.execute(query);
+  }
+
+  /**
+   * Refreshes a given table by refreshing it in CrateDB.
+   *
+   * The `REFRESH TABLE` command makes recently committed changes available for querying
+   * without waiting for automatic refresh intervals.
+   *
+   * @param {string} tableName - The name of the table to refresh.
+   * @returns {Promise<CrateDBResponse>} A promise resolving to the response from CrateDB.
+   */
   async refresh(tableName: string): Promise<CrateDBResponse> {
-    const query = `REFRESH TABLE ${tableName}`;
+    const query = StatementGenerator.refresh(tableName);
     return this.execute(query);
   }
 
-  async createTable(schema: Record<string, Record<string, string>>): Promise<CrateDBResponse> {
-    const tableName = Object.keys(schema)[0];
-    const columns = Object.entries(schema[tableName])
-      .map(([col, type]) => `"${col}" ${type}`)
-      .join(', ');
-    const query = `CREATE TABLE ${tableName} (${columns})`;
+  /**
+   * Optimizes a given table or specific partitions in CrateDB by merging table segments.
+   *
+   * The `OPTIMIZE TABLE` command reduces the number of segments in a table, improving
+   * query performance and reducing storage overhead. It supports optimizing the entire table
+   * or specific partitions and allows additional optimization parameters.
+   *
+   * @param {string} tableName - The name of the table to optimize.
+   * @param {OptimizeOptions} [options] - Optional parameters for table optimization.
+   * @param {Record<string, string | number>} [partitions] - Optional key-value pairs specifying partition columns and values.
+   * @returns {Promise<CrateDBResponse>} A promise resolving to the response from CrateDB.
+   */
+  async optimize(
+    tableName: string,
+    options?: OptimizeOptions,
+    partitions?: Record<string, string | number>
+  ): Promise<CrateDBResponse> {
+    const query = StatementGenerator.optimize(tableName, options, partitions);
     return this.execute(query);
   }
 

src/Cursor.ts

@@ -40,13 +40,13 @@ export class Cursor {
     this.isOpen = true;
   }
 
-  async fetchone(): Promise<CrateDBRecord | null> {
+  async fetchOne(): Promise<CrateDBRecord | null> {
     this._ensureOpen();
     const result = await this._execute(`FETCH NEXT FROM ${this.cursorName}`);
     return result.length > 0 ? result[0] : null; // Return the first row or null
   }
 
-  async fetchmany(size = 10): Promise<Array<CrateDBRecord>> {
+  async fetchMany(size = 10): Promise<Array<CrateDBRecord>> {
     if (size < 1) {
       // Return an empty array if size is less than 1
       return [];
@@ -55,7 +55,7 @@ export class Cursor {
     return await this._execute(`FETCH ${size} FROM ${this.cursorName}`);
   }
 
-  async fetchall(): Promise<Array<CrateDBRecord>> {
+  async fetchAll(): Promise<Array<CrateDBRecord>> {
     this._ensureOpen();
     return await this._execute(`FETCH ALL FROM ${this.cursorName}`);
   }
@@ -64,7 +64,7 @@ export class Cursor {
     this._ensureOpen();
 
     while (true) {
-      const rows = await this.fetchmany(size);
+      const rows = await this.fetchMany(size);
 
       if (!rows || rows.length === 0) {
         break; // Stop iteration when no more rows are returned

src/StatementGenerator.ts

@@ -0,0 +1,140 @@
+import { OptimizeOptions, ColumnDefinition, TableOptions } from './interfaces';
+import { Serializer } from './Serializer';
+
+export class StatementGenerator {
+  public static createTable(
+    tableName: string,
+    schema: Record<string, ColumnDefinition>,
+    options?: TableOptions
+  ): string {
+    // Build column definitions
+    const columns = Object.entries(schema)
+      .map(([col, definition]) => {
+        let colDef = `"${col}" ${definition.type.toUpperCase()}`;
+        if (definition.notNull) {
+          colDef += ' NOT NULL';
+        }
+        if (definition.defaultValue !== undefined) {
+          colDef += ` DEFAULT ${definition.defaultValue}`;
+        }
+        return colDef;
+      })
+      .join(', ');
+
+    // Build primary key clause if any
+    const primaryKeys = Object.keys(schema)
+      .filter((col) => schema[col].primaryKey)
+      .map((col) => `"${col}"`)
+      .join(', ');
+    const primaryKeyClause = primaryKeys ? `, PRIMARY KEY(${primaryKeys})` : '';
+
+    // Start query construction
+    let query = `CREATE TABLE ${this.quoteIdentifier(tableName)} (${columns}${primaryKeyClause})`;
+
+    // Partition clause
+    if (options?.partitionedBy?.length) {
+      const partitionCols = options.partitionedBy.map((col) => `"${col}"`).join(', ');
+      query += ` PARTITIONED BY (${partitionCols})`;
+    }
+
+    // Cluster clause
+    if (options?.clusteredBy || options?.numberOfShards) {
+      if (options.clusteredBy && options.numberOfShards) {
+        query += ` CLUSTERED BY ("${options.clusteredBy}") INTO ${options.numberOfShards} SHARDS`;
+      } else if (options.clusteredBy) {
+        query += ` CLUSTERED BY ("${options.clusteredBy}")`;
+      } else if (options.numberOfShards) {
+        query += ` CLUSTERED INTO ${options.numberOfShards} SHARDS`;
+      }
+    }
+
+    // Replicas clause
+    if (options?.numberOfReplicas !== undefined) {
+      query += ` WITH (number_of_replicas = '${options.numberOfReplicas}')`;
+    }
+
+    return query + ';';
+  }
+
+  public static dropTable(tableName: string): string {
+    return `DROP TABLE ${this.quoteIdentifier(tableName)};`;
+  }
+
+  public static delete(tableName: string, whereClause?: string): string {
+    return `DELETE FROM ${this.quoteIdentifier(tableName)} WHERE ${whereClause};`;
+  }
+
+  public static update(tableName: string, options: Record<string, unknown>, whereClause: string): string {
+    const { keys, values } = this._prepareOptions(options);
+    const setClause = keys.map((key, i) => `${key}=${values[i]}`).join(', ');
+    return `UPDATE ${this.quoteIdentifier(tableName)} SET ${setClause} WHERE ${whereClause};`;
+  }
+
+  public static insert(tableName: string, keys: string[], primaryKeys: string[] | null): string {
+    const placeholders = keys.map(() => '?').join(', ');
+    const columns = keys.map((key) => `"${key}"`).join(', ');
+    let query = `INSERT INTO ${this.quoteIdentifier(tableName)} (${columns}) VALUES (${placeholders})`;
+
+    if (primaryKeys && primaryKeys.length > 0) {
+      const nonPrimaryKeys = keys.filter((key) => !primaryKeys.includes(key));
+      const updates = nonPrimaryKeys.map((key) => `"${key}" = excluded."${key}"`).join(', ');
+      const pkClause = primaryKeys.map((key) => `"${key}"`).join(', ');
+      query += ` ON CONFLICT (${pkClause}) DO UPDATE SET ${updates}`;
+    } else {
+      query += ' ON CONFLICT DO NOTHING';
+    }
+
+    return query + ';';
+  }
+
+  public static refresh(tableName: string): string {
+    return `REFRESH TABLE ${this.quoteIdentifier(tableName)};`;
+  }
+
+  public static optimize(
+    tableName: string,
+    options?: OptimizeOptions,
+    partitions?: Record<string, string | number>
+  ): string {
+    let query = `OPTIMIZE TABLE ${this.quoteIdentifier(tableName)}`;
+
+    // Build options clause
+    if (options && Object.keys(options).length > 0) {
+      const optionsClauses = Object.entries(options)
+        .map(([key, value]) => `${key}=${Serializer.serialize(value)}`)
+        .join(', ');
+      query += ` WITH (${optionsClauses})`;
+    }
+
+    // Build partitions clause
+    if (partitions && Object.keys(partitions).length > 0) {
+      const partitionClauses = Object.entries(partitions)
+        .map(([key, value]) => {
+          const val = typeof value === 'string' ? `'${value}'` : value;
+          return `${key}=${val}`;
+        })
+        .join(', ');
+      query += ` PARTITION (${partitionClauses})`;
+    }
+
+    return query + ';';
+  }
+
+  private static quoteIdentifier(tableName: string): string {
+    return tableName
+      .split('.')
+      .map((part) => `"${part}"`)
+      .join('.');
+  }
+
+  private static _prepareOptions(options: Record<string, unknown>): {
+    keys: string[];
+    values: string[];
+    args: unknown[];
+  } {
+    const keys = Object.keys(options).map((key) => `"${key}"`);
+    const values = keys.map(() => '?');
+    const args = Object.values(options);
+    return { keys, values, args };
+  }
+}

src/interfaces.ts

@@ -67,3 +67,29 @@ export interface CrateDBErrorResponse {
 }
 
 export type CrateDBRecord = Record<string, unknown>;
+
+export type OptimizeOptions = {
+  max_num_segments?: number; // Defines the number of segments to merge to
+  only_expunge_deletes?: boolean; // If true, only segments with deletes are merged
+  flush?: boolean; // If false, prevents automatic flushing after optimization
+};
+
+/**
+ * Defines the structure of a column in a CrateDB table.
+ */
+export type ColumnDefinition = {
+  type: string;
+  primaryKey?: boolean;
+  notNull?: boolean;
+  defaultValue?: unknown;
+};
+
+/**
+ * Defines additional table options such as clustering, partitioning, and replicas.
+ */
+export type TableOptions = {
+  clusteredBy?: string; // Column to cluster by
+  partitionedBy?: string[]; // List of partition key columns
+  numberOfShards?: number; // Replication factor
+  numberOfReplicas?: string | number; // Replication factor
+};