| # Licensed to the Apache Software Foundation (ASF) under one | |
| # or more contributor license agreements. See the NOTICE file | |
| # distributed with this work for additional information | |
| # regarding copyright ownership. The ASF licenses this file | |
| # to you under the Apache License, Version 2.0 (the | |
| # "License"); you may not use this file except in compliance | |
| # with the License. You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, | |
| # software distributed under the License is distributed on an | |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | |
| # KIND, either express or implied. See the License for the | |
| # specific language governing permissions and limitations | |
| # under the License. | |
| """ | |
| SQL commands for Paimon CLI. | |
| This module provides SQL query capability via pypaimon-rust + DataFusion. | |
| """ | |
| import os | |
| import re | |
| import sys | |
| import time | |
| import pyarrow as pa | |
| _PAIMON_BANNER = r""" | |
| ____ _ | |
| / __ \____ _(_)___ ___ ____ ____ | |
| / /_/ / __ `/ / __ `__ \/ __ \/ __ \ | |
| / ____/ /_/ / / / / / / / /_/ / / / / | |
| /_/ \__,_/_/_/ /_/ /_/\____/_/ /_/ | |
| Powered by pypaimon-rust + DataFusion | |
| Type 'help' for usage, 'exit' to quit. | |
| """ | |
| _USE_PATTERN = re.compile( | |
| r"^\s*use\s+(\w+)\s*;?\s*$", | |
| re.IGNORECASE, | |
| ) | |
| _HISTORY_FILE = os.path.expanduser("~/.paimon_history") | |
| _HISTORY_MAX_LENGTH = 1000 | |
| _PROMPT = "paimon> " | |
| _CONTINUATION_PROMPT = " > " | |
| def _get_readline(): | |
| """Get the best available readline module. | |
| Prefers gnureadline (full GNU readline) over the built-in readline | |
| (which is libedit on macOS and may have limited features). | |
| """ | |
| try: | |
| import gnureadline as readline | |
| return readline | |
| except ImportError: | |
| pass | |
| try: | |
| import readline | |
| return readline | |
| except ImportError: | |
| return None | |
| def _is_libedit(rl): | |
| """Check if the readline module is backed by libedit (macOS default).""" | |
| return hasattr(rl, '__doc__') and rl.__doc__ and 'libedit' in rl.__doc__ | |
| def _setup_readline(): | |
| """Enable readline for arrow key support and persistent command history.""" | |
| rl = _get_readline() | |
| if rl is None: | |
| return | |
| rl.set_history_length(_HISTORY_MAX_LENGTH) | |
| if not os.path.exists(_HISTORY_FILE): | |
| return | |
| if _is_libedit(rl): | |
| # libedit escapes spaces as \040 in history files, so we load manually. | |
| with open(_HISTORY_FILE, 'r', encoding='utf-8') as f: | |
| for line in f: | |
| line = line.rstrip('\n') | |
| if line: | |
| rl.add_history(line) | |
| else: | |
| rl.read_history_file(_HISTORY_FILE) | |
| def _save_history(): | |
| """Save readline history to file.""" | |
| rl = _get_readline() | |
| if rl is None: | |
| return | |
| try: | |
| if _is_libedit(rl): | |
| # Write history manually to avoid libedit's \040 escaping. | |
| length = rl.get_current_history_length() | |
| lines = [] | |
| for i in range(1, length + 1): | |
| item = rl.get_history_item(i) | |
| if item is not None: | |
| lines.append(item) | |
| # Keep only the last N entries | |
| lines = lines[-_HISTORY_MAX_LENGTH:] | |
| with open(_HISTORY_FILE, 'w', encoding='utf-8') as f: | |
| for line in lines: | |
| f.write(line + '\n') | |
| else: | |
| rl.write_history_file(_HISTORY_FILE) | |
| except OSError: | |
| pass | |
| def cmd_sql(args): | |
| """ | |
| Execute the 'sql' command. | |
| Runs a SQL query against Paimon tables, or starts an interactive SQL REPL. | |
| Args: | |
| args: Parsed command line arguments. | |
| """ | |
| from pypaimon.cli.cli import load_catalog_config | |
| config_path = args.config | |
| config = load_catalog_config(config_path) | |
| try: | |
| from pypaimon_rust.datafusion import SQLContext | |
| catalog_options = {str(k): str(v) for k, v in config.items()} | |
| ctx = SQLContext() | |
| ctx.register_catalog("paimon", catalog_options) | |
| ctx.set_current_catalog("paimon") | |
| ctx.set_current_database("default") | |
| except Exception as e: | |
| print(f"Error: {e}", file=sys.stderr) | |
| sys.exit(1) | |
| query = args.query | |
| if query: | |
| _execute_query(ctx, query, getattr(args, 'format', 'table')) | |
| else: | |
| _interactive_repl(ctx, getattr(args, 'format', 'table')) | |
| def _execute_query(ctx, query, output_format): | |
| """Execute a single SQL query and print the result.""" | |
| try: | |
| batches = ctx.sql(query) | |
| except Exception as e: | |
| print(f"Error: {e}", file=sys.stderr) | |
| sys.exit(1) | |
| if batches: | |
| _print_table(pa.Table.from_batches(batches), output_format) | |
| else: | |
| print("OK") | |
| def _print_table(table, output_format, elapsed=None): | |
| """Print a PyArrow Table in the requested format.""" | |
| df = table.to_pandas() | |
| if output_format == 'json': | |
| import json | |
| print(json.dumps(df.to_dict(orient='records'), ensure_ascii=False)) | |
| else: | |
| print(df.to_string(index=False)) | |
| if elapsed is not None: | |
| row_count = len(df) | |
| print(f"({row_count} {'row' if row_count == 1 else 'rows'} in {elapsed:.2f}s)") | |
| def _read_multiline_query(): | |
| """Read a potentially multi-line SQL query, terminated by ';'. | |
| Returns the complete query string, or None on EOF/interrupt. | |
| """ | |
| lines = [] | |
| prompt = _PROMPT | |
| while True: | |
| try: | |
| line = input(prompt) | |
| except (EOFError, KeyboardInterrupt): | |
| if lines: | |
| # Cancel current multi-line input | |
| print() | |
| return "" | |
| return None | |
| lines.append(line) | |
| joined = "\n".join(lines).strip() | |
| if not joined: | |
| lines.clear() | |
| prompt = _PROMPT | |
| continue | |
| # Single-word commands that don't need ';' | |
| lower = joined.lower().rstrip(';').strip() | |
| if lower in ('exit', 'quit', 'help'): | |
| return joined | |
| # USE command doesn't strictly need ';' | |
| if _USE_PATTERN.match(joined): | |
| return joined | |
| # For SQL statements, wait for ';' | |
| if joined.endswith(';'): | |
| return joined | |
| prompt = _CONTINUATION_PROMPT | |
| def _handle_use(ctx, match): | |
| """Handle USE <database> command.""" | |
| database = match.group(1) | |
| try: | |
| ctx.set_current_database(database) | |
| print(f"Using database '{database}'.") | |
| except Exception as e: | |
| print(f"Error: {e}", file=sys.stderr) | |
| def _interactive_repl(ctx, output_format): | |
| """Run an interactive SQL REPL.""" | |
| _setup_readline() | |
| print(_PAIMON_BANNER) | |
| try: | |
| while True: | |
| query = _read_multiline_query() | |
| if query is None: | |
| print("\nBye!") | |
| break | |
| if not query: | |
| continue | |
| lower = query.lower().rstrip(';').strip() | |
| if lower in ('exit', 'quit'): | |
| print("Bye!") | |
| break | |
| if lower == 'help': | |
| _print_help() | |
| continue | |
| # Handle USE <database> | |
| use_match = _USE_PATTERN.match(query) | |
| if use_match: | |
| _handle_use(ctx, use_match) | |
| continue | |
| try: | |
| start = time.time() | |
| batches = ctx.sql(query) | |
| elapsed = time.time() - start | |
| if batches: | |
| _print_table(pa.Table.from_batches(batches), output_format, elapsed) | |
| else: | |
| print(f"OK ({elapsed:.2f}s)") | |
| print() | |
| except Exception as e: | |
| print(f"Error: {e}\n", file=sys.stderr) | |
| finally: | |
| _save_history() | |
| def _print_help(): | |
| """Print REPL help information.""" | |
| print(""" | |
| Commands: | |
| USE <database>; Switch the default database | |
| SHOW DATABASES; List all databases | |
| SHOW TABLES; List tables in the current database | |
| SELECT ... FROM <table>; Execute a SQL query | |
| exit / quit Exit the REPL | |
| Table reference: | |
| <table> Table in the current default database | |
| <database>.<table> Table in a specific database | |
| Tips: | |
| - SQL statements end with ';' and can span multiple lines | |
| - Arrow keys are supported for line editing and command history | |
| - Command history is saved across sessions (~/.paimon_history) | |
| """) | |
| def add_sql_subcommand(subparsers): | |
| """ | |
| Add the sql subcommand to the main parser. | |
| Args: | |
| subparsers: The subparsers object from the main argument parser. | |
| """ | |
| sql_parser = subparsers.add_parser( | |
| 'sql', | |
| help='Execute SQL queries on Paimon tables (requires pypaimon-rust)' | |
| ) | |
| sql_parser.add_argument( | |
| 'query', | |
| nargs='?', | |
| default=None, | |
| help='SQL query to execute. If omitted, starts interactive REPL.' | |
| ) | |
| sql_parser.add_argument( | |
| '--format', '-f', | |
| type=str, | |
| choices=['table', 'json'], | |
| default='table', | |
| help='Output format: table (default) or json' | |
| ) | |
| sql_parser.set_defaults(func=cmd_sql) |