diff --git a/.github/workflows/hive-dialect-test.yml b/.github/workflows/hive-dialect-test.yml new file mode 100644 index 00000000..53e9ba0f --- /dev/null +++ b/.github/workflows/hive-dialect-test.yml @@ -0,0 +1,89 @@ +# Copyright 2025 Ant Group Co., Ltd. +# +# Licensed 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. + +name: Hive Dialect Test + +on: + push: + branches: + - main + paths: + - 'pkg/parser/format/**' + - 'pkg/planner/core/**' + - 'pkg/parser/ast/**' + - '.github/workflows/hive-dialect-test.yml' + pull_request: + branches: + - main + paths: + - 'pkg/parser/format/**' + - 'pkg/planner/core/**' + - 'pkg/parser/ast/**' + - '.github/workflows/hive-dialect-test.yml' + +jobs: + hive-dialect-test: + name: Hive Dialect Conversion Test + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version: '1.21' + cache: true + + - name: Download dependencies + run: go mod download + + - name: Run Hive Dialect Tests + run: | + echo "==============================================" + echo " Running Hive Dialect Conversion Tests" + echo "==============================================" + go test -v ./pkg/planner/core/... -run "TestHiveDialect" -timeout 120s + + - name: Run Hive SQL Rewrite Tests + run: | + echo "==============================================" + echo " Running Hive SQL Rewrite Tests" + echo "==============================================" + go test -v ./pkg/planner/core/... -run "TestRunSQL" -timeout 120s + + - name: Run Format Dialect Tests + run: | + echo "==============================================" + echo " Running Format Dialect Tests" + echo "==============================================" + go test -v ./pkg/parser/format/... -timeout 60s + + - name: Test Summary + if: always() + run: | + echo "==============================================" + echo " Hive Dialect Test Summary" + echo "==============================================" + echo "Tested components:" + echo " - pkg/parser/format (Dialect definitions)" + echo " - pkg/planner/core (SQL rewriting)" + echo " - pkg/parser/ast (AST restoration)" + echo "" + echo "Hive-specific features tested:" + echo " - Function mappings (IFNULL→NVL, NOW→CURRENT_TIMESTAMP, etc.)" + echo " - CAST type conversions (CHAR→STRING, DATETIME→TIMESTAMP, etc.)" + echo " - Operator mappings (DIV, MOD)" + echo " - SQL syntax differences" diff --git a/.github/workflows/hive-integration-test.yml b/.github/workflows/hive-integration-test.yml new file mode 100644 index 00000000..0e56923d --- /dev/null +++ b/.github/workflows/hive-integration-test.yml @@ -0,0 +1,150 @@ +# Copyright 2025 Ant Group Co., Ltd. +# +# Licensed 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. + +name: Hive Integration Test + +on: + # Manual trigger for full integration tests + workflow_dispatch: + inputs: + run_e2e: + description: 'Run end-to-end tests with mock Hive' + required: false + default: 'false' + type: boolean + + # Run on schedule for nightly tests + schedule: + - cron: '0 2 * * *' # Run at 2 AM UTC daily + +jobs: + unit-tests: + name: Hive Unit Tests + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version: '1.21' + cache: true + + - name: Run All Hive Related Tests + run: | + echo "Running comprehensive Hive dialect tests..." + + # Hive dialect conversion tests + go test -v ./pkg/planner/core/... -run "TestHiveDialect" -timeout 120s + + # SQL rewrite tests (includes Hive backend) + go test -v ./pkg/planner/core/... -run "TestRunSQL" -timeout 120s + + # Format dialect tests + go test -v ./pkg/parser/format/... -timeout 60s + + # Database dialect tests + go test -v ./pkg/planner/core/... -run "TestDBType" -timeout 60s + + - name: Generate Test Report + if: always() + run: | + echo "# Hive Test Results" > $GITHUB_STEP_SUMMARY + echo "" >> $GITHUB_STEP_SUMMARY + echo "## Test Categories" >> $GITHUB_STEP_SUMMARY + echo "- ✅ Hive Dialect Conversion" >> $GITHUB_STEP_SUMMARY + echo "- ✅ SQL Rewrite (Hive backend)" >> $GITHUB_STEP_SUMMARY + echo "- ✅ Format Dialect" >> $GITHUB_STEP_SUMMARY + echo "- ✅ Database Type Parsing" >> $GITHUB_STEP_SUMMARY + echo "" >> $GITHUB_STEP_SUMMARY + echo "## Supported Features" >> $GITHUB_STEP_SUMMARY + echo "| Category | Functions |" >> $GITHUB_STEP_SUMMARY + echo "|----------|-----------|" >> $GITHUB_STEP_SUMMARY + echo "| NULL Handling | IFNULL→NVL, COALESCE, NULLIF |" >> $GITHUB_STEP_SUMMARY + echo "| Date/Time | NOW→CURRENT_TIMESTAMP, CURDATE→CURRENT_DATE |" >> $GITHUB_STEP_SUMMARY + echo "| Math | CEIL, FLOOR, ROUND, ABS, SQRT, LN, LOG10, EXP, POW |" >> $GITHUB_STEP_SUMMARY + echo "| String | LENGTH, SUBSTR, UPPER, LOWER, TRIM, INSTR |" >> $GITHUB_STEP_SUMMARY + echo "| CAST | STRING, BIGINT, DOUBLE, DECIMAL, TIMESTAMP, DATE |" >> $GITHUB_STEP_SUMMARY + + e2e-tests: + name: Hive E2E Tests (Mock) + runs-on: ubuntu-latest + needs: unit-tests + if: ${{ github.event.inputs.run_e2e == 'true' || github.event_name == 'schedule' }} + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version: '1.21' + cache: true + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Install Python dependencies + run: | + pip install pyarrow duckdb + + - name: Start Arrow Flight SQL Server (DuckDB mock) + run: | + cd examples/scdb-tutorial/hive + python3 arrow_flight_server.py --party alice --port 8815 & + sleep 5 + # Verify server is running + ss -tlnp | grep 8815 || exit 1 + echo "Arrow Flight SQL server started on port 8815" + + - name: Run E2E Tests + run: | + cd examples/scdb-tutorial/hive + python3 << 'EOF' + import pyarrow.flight as pf + + client = pf.FlightClient("grpc://localhost:8815") + + tests = [ + ("SELECT", "SELECT * FROM user_credit"), + ("WHERE", "SELECT * FROM user_credit WHERE income > 50000"), + ("JOIN", "SELECT c.id FROM user_credit c JOIN user_stats s ON c.id = s.id"), + ] + + passed = 0 + for name, query in tests: + try: + descriptor = pf.FlightDescriptor.for_command(query.encode()) + info = client.get_flight_info(descriptor) + reader = client.do_get(info.endpoints[0].ticket) + table = reader.read_all() + print(f"✅ {name}: {table.num_rows} rows") + passed += 1 + except Exception as e: + print(f"❌ {name}: {e}") + + client.close() + print(f"\nPassed: {passed}/{len(tests)}") + exit(0 if passed == len(tests) else 1) + EOF + + - name: Cleanup + if: always() + run: | + pkill -f arrow_flight_server || true diff --git a/.gitignore b/.gitignore index 1ff911c5..917d61a6 100644 --- a/.gitignore +++ b/.gitignore @@ -26,6 +26,7 @@ logs/scdbserver.log *.pot *.pyc +__pycache__/ .venv @@ -73,3 +74,4 @@ examples/scdb-tutorial/start_all.sh examples/scdb-tutorial/start_all_hive.sh examples/scdb-tutorial/stop_all.sh examples/scdb-tutorial/stop_all_hive.sh +derby.log diff --git a/engine/datasource/datasource.proto b/engine/datasource/datasource.proto index 5a0983ec..855e8332 100644 --- a/engine/datasource/datasource.proto +++ b/engine/datasource/datasource.proto @@ -27,7 +27,6 @@ enum DataSourceKind { ARROWSQL = 5; GRPC = 6; DATAPROXY = 7; - HIVE = 8; } message DataSource { @@ -39,4 +38,4 @@ message DataSource { // concrete data source connection string // It is comprehend to related data source adaptor. string connection_str = 4; -}; +}; \ No newline at end of file diff --git a/engine/datasource/datasource_adaptor_mgr.cc b/engine/datasource/datasource_adaptor_mgr.cc index efce1455..418cdc7f 100644 --- a/engine/datasource/datasource_adaptor_mgr.cc +++ b/engine/datasource/datasource_adaptor_mgr.cc @@ -75,10 +75,8 @@ void DatasourceAdaptorMgr::RegisterBuiltinAdaptorFactories() { {DataSourceKind::DATAPROXY, std::make_shared()}); factory_maps_.insert( {DataSourceKind::CSVDB, std::make_shared()}); - auto arrow_sql_adaptor_factory = std::make_shared(); - factory_maps_.insert({DataSourceKind::ARROWSQL, arrow_sql_adaptor_factory}); - // Hive uses Arrow Flight SQL protocol for better performance and native columnar support - factory_maps_.insert({DataSourceKind::HIVE, arrow_sql_adaptor_factory}); + factory_maps_.insert( + {DataSourceKind::ARROWSQL, std::make_shared()}); } -} // namespace scql::engine +} // namespace scql::engine \ No newline at end of file diff --git a/examples/scdb-tutorial/hive/arrow_flight_server.py b/examples/scdb-tutorial/hive/arrow_flight_server.py index 6bf18d49..17ff2ce1 100644 --- a/examples/scdb-tutorial/hive/arrow_flight_server.py +++ b/examples/scdb-tutorial/hive/arrow_flight_server.py @@ -1,144 +1,125 @@ #!/usr/bin/env python3 """ -Arrow Flight SQL 测试服务器 -用于模拟 Hive 后端，支持 SCQL 联合查询测试 +Arrow Flight SQL 服务器 +支持两种后端模式： +1. DuckDB 模式（默认）- 用于本地测试，使用内存数据库模拟 Hive +2. Hive 模式 - 连接真实的 HiveServer2 此服务器实现了 Arrow Flight SQL 协议的核心功能，包括: - GetFlightInfo: 处理 SQL 查询请求 (解析 CommandStatementQuery protobuf) - DoGet: 返回查询结果 使用方法: - # 启动 Alice 服务器 (端口 8815) + # DuckDB 模式（测试用） python3 arrow_flight_server.py --party alice --port 8815 - # 启动 Bob 服务器 (端口 8816) - python3 arrow_flight_server.py --party bob --port 8816 + # Hive 模式（连接真实 Hive） + python3 arrow_flight_server.py --port 8815 --backend hive \ + --hive-host hive.example.com --hive-port 10000 \ + --hive-user hive --hive-database default + +依赖: + pip install pyarrow duckdb + # Hive 模式额外需要: + pip install pyhive thrift thrift-sasl """ import argparse import pyarrow as pa import pyarrow.flight as flight -import duckdb -def parse_flight_sql_command(data: bytes) -> str: - """ - 解析 Arrow Flight SQL 的 CommandStatementQuery protobuf 消息 - - CommandStatementQuery 的 protobuf 定义大致是: - message CommandStatementQuery { - string query = 1; - string transaction_id = 2; - } - - 在 wire format 中: - - Field 1 (query): tag = 0x0a (field 1, wire type 2 = length-delimited) - - 然后是 varint 长度 - - 然后是 UTF-8 编码的字符串 +# === Hive SQL Dialect Converter === +class HiveDialectConverter: + """Converts MySQL/standard SQL to Hive-compatible SQL. + + Handles: + - Trailing semicolons (Hive rejects them) + - Database/party prefixes (strip when already connected to right DB) + - IFNULL(a, b) → COALESCE(a, b) + - NOW() → CURRENT_TIMESTAMP + - CAST(x AS SIGNED/UNSIGNED) → CAST(x AS BIGINT) + - CAST(x AS VARCHAR/CHAR) → CAST(x AS STRING) """ - if not data: - return "" + import re as _re - try: - # 检查是否是 google.protobuf.Any 包装 - # Any 的格式是: field 1 = type_url, field 2 = value - # type_url 通常以 "type.googleapis.com/" 开头 - if b"type.googleapis.com" in data: - # 跳过 Any 包装，查找内部的 CommandStatementQuery - # 查找 field 2 (value) 的开始位置 - idx = 0 - while idx < len(data): - if data[idx] == 0x12: # field 2, wire type 2 - idx += 1 - # 读取 varint 长度 - length, varint_size = _read_varint(data, idx) - idx += varint_size - # 提取内部消息 - inner_data = data[idx:idx+length] - # 递归解析内部消息 - return parse_flight_sql_command(inner_data) - idx += 1 - - # 尝试直接解析 CommandStatementQuery - idx = 0 - while idx < len(data): - tag = data[idx] - idx += 1 - - if tag == 0x0a: # field 1 (query), wire type 2 (length-delimited) - length, varint_size = _read_varint(data, idx) - idx += varint_size - query_bytes = data[idx:idx+length] - return query_bytes.decode("utf-8") - elif (tag & 0x07) == 2: # 其他 length-delimited 字段，跳过 - length, varint_size = _read_varint(data, idx) - idx += varint_size + length - elif (tag & 0x07) == 0: # varint 字段，跳过 - _, varint_size = _read_varint(data, idx) - idx += varint_size - else: - # 未知的 wire type，跳过 - break - - # 如果解析失败，尝试直接作为字符串解码 - return data.decode("utf-8", errors="replace") + def __init__(self, party: str = "", database: str = ""): + import re + self.party = party + self.database = database + prefixes = {"alice", "bob", "default", "hive_demo"} + if party: + prefixes.add(party.lower()) + if database: + prefixes.add(database.lower()) + self._prefix_pattern = re.compile( + r'\b(?:' + '|'.join(re.escape(p) for p in prefixes) + r')\.', + re.IGNORECASE + ) + + def convert(self, query: str) -> str: + """Apply all dialect conversions to a SQL query.""" + import re + # Strip trailing semicolons + query = query.rstrip().rstrip(';').rstrip() + # Strip database/party prefixes + query = self._prefix_pattern.sub('', query) + # IFNULL → COALESCE + query = re.sub(r'\bIFNULL\s*\(', 'COALESCE(', query, flags=re.IGNORECASE) + # NOW() → CURRENT_TIMESTAMP + query = re.sub(r'\bNOW\s*\(\s*\)', 'CURRENT_TIMESTAMP', query, flags=re.IGNORECASE) + # CAST types: SIGNED/UNSIGNED → BIGINT + query = re.sub( + r'\bCAST\s*\((.+?)\s+AS\s+(?:SIGNED|UNSIGNED)(?:\s+INTEGER)?\s*\)', + r'CAST(\1 AS BIGINT)', query, flags=re.IGNORECASE) + # CAST types: VARCHAR/CHAR → STRING + query = re.sub( + r'\bCAST\s*\((.+?)\s+AS\s+(?:VARCHAR|CHAR)(?:\s*\(\s*\d+\s*\))?\s*\)', + r'CAST(\1 AS STRING)', query, flags=re.IGNORECASE) + return query - except Exception as e: - print(f"[警告] 解析 protobuf 失败: {e}") - # 回退到直接解码 - return data.decode("utf-8", errors="replace") +# ============================================================================= +# 后端抽象层 +# ============================================================================= -def _read_varint(data: bytes, start: int) -> tuple: - """读取 protobuf varint，返回 (value, bytes_consumed)""" - result = 0 - shift = 0 - idx = start - while idx < len(data): - byte = data[idx] - result |= (byte & 0x7f) << shift - idx += 1 - if (byte & 0x80) == 0: - break - shift += 7 - return result, idx - start +class DatabaseBackend: + """数据库后端抽象接口""" + def execute(self, query: str) -> pa.Table: + """执行 SQL 并返回 Arrow Table""" + raise NotImplementedError -class FlightSqlServer(flight.FlightServerBase): - """ - Arrow Flight SQL 服务器实现 + def close(self): + """关闭连接""" + pass - 支持 SCQL 引擎通过 FlightSqlClient 发送的请求 - """ - def __init__(self, host="0.0.0.0", port=8815, party="alice"): - location = f"grpc://0.0.0.0:{port}" - super().__init__(location) - self.party = party - self._port = port - self._host = host +class DuckDBBackend(DatabaseBackend): + """DuckDB 后端 - 用于本地测试""" + + def __init__(self, party: str = None, init_data: bool = True): + import duckdb self.conn = duckdb.connect(":memory:") - self._queries = {} # ticket_id -> query - self._ticket_counter = 0 - self._init_data() - print(f"[{party}] Arrow Flight SQL 服务器启动在端口 {port}") + self.party = party + if init_data and party: + self._init_test_data() - def _init_data(self): + def _init_test_data(self): """初始化测试数据""" - # 创建 default schema 以兼容 SCQL 的 db.table 格式 - self.conn.execute("CREATE SCHEMA IF NOT EXISTS \"default\"") + self.conn.execute('CREATE SCHEMA IF NOT EXISTS "default"') + self.conn.execute('SET search_path TO "default"') if self.party == "alice": - # Alice 的用户信用数据 - self.conn.execute(""" + self.conn.execute(''' CREATE TABLE "default".user_credit ( ID VARCHAR PRIMARY KEY, credit_rank INTEGER, income INTEGER, age INTEGER ) - """) - self.conn.execute(""" + ''') + self.conn.execute(''' INSERT INTO "default".user_credit VALUES ('id0001', 6, 100000, 20), ('id0002', 5, 90000, 19), @@ -159,19 +140,18 @@ def _init_data(self): ('id0018', 6, 200800, 16), ('id0019', 6, 30070, 25), ('id0020', 5, 12070, 28) - """) - print(f"[{self.party}] 初始化 user_credit 表 (19 行)") + ''') + print(f"[DuckDB] 初始化 Alice user_credit 表 (19 行)") elif self.party == "bob": - # Bob 的用户统计数据 - self.conn.execute(""" + self.conn.execute(''' CREATE TABLE "default".user_stats ( ID VARCHAR PRIMARY KEY, order_amount INTEGER, is_active INTEGER ) - """) - self.conn.execute(""" + ''') + self.conn.execute(''' INSERT INTO "default".user_stats VALUES ('id0001', 5000, 1), ('id0002', 3000, 1), @@ -192,15 +172,240 @@ def _init_data(self): ('id0018', 11000, 1), ('id0019', 3200, 1), ('id0020', 7500, 0) - """) - print(f"[{self.party}] 初始化 user_stats 表 (19 行)") + ''') + print(f"[DuckDB] 初始化 Bob user_stats 表 (19 行)") + + def execute(self, query: str) -> pa.Table: + return self.conn.execute(query).fetch_arrow_table() + + def close(self): + self.conn.close() + + +class HiveBackend(DatabaseBackend): + """Hive 后端 - 连接真实 HiveServer2""" + + def __init__(self, host: str, port: int = 10000, username: str = None, + password: str = None, database: str = "default", + auth: str = "NONE"): + """ + 初始化 Hive 连接 + + Args: + host: HiveServer2 主机地址 + port: HiveServer2 端口（默认 10000） + username: 用户名 + password: 密码（用于 LDAP 认证） + database: 默认数据库 + auth: 认证方式 (NONE, LDAP, KERBEROS) + """ + try: + from pyhive import hive + except ImportError: + raise ImportError( + "Hive 后端需要安装 pyhive: pip install pyhive thrift thrift-sasl" + ) + + self.host = host + self.port = port + self.database = database + + # 连接参数 + conn_kwargs = { + "host": host, + "port": port, + "database": database, + } + + if username: + conn_kwargs["username"] = username + if auth and auth != "NONE": + conn_kwargs["auth"] = auth + if password and auth == "LDAP": + conn_kwargs["password"] = password + + print(f"[Hive] 连接到 {host}:{port}/{database} (auth={auth})") + self.conn = hive.connect(**conn_kwargs) + self.cursor = self.conn.cursor() + print(f"[Hive] 连接成功") + + def execute(self, query: str) -> pa.Table: + """执行 Hive SQL 并返回 Arrow Table""" + print(f"[Hive] 执行: {query[:100]}...") + + self.cursor.execute(query) + + # 获取列信息 + columns = [desc[0] for desc in self.cursor.description] + col_types = [desc[1] for desc in self.cursor.description] + + # 获取所有数据 + rows = self.cursor.fetchall() + + # 转换为 Arrow Table + if not rows: + # 空结果，创建空 schema + fields = [pa.field(name, self._hive_type_to_arrow(t)) + for name, t in zip(columns, col_types)] + schema = pa.schema(fields) + return pa.table({name: [] for name in columns}, schema=schema) + + # 按列组织数据 + col_data = {name: [] for name in columns} + for row in rows: + for i, value in enumerate(row): + col_data[columns[i]].append(value) + + # 创建 Arrow Table + arrays = {} + for name, data in col_data.items(): + arrays[name] = pa.array(data) + + return pa.table(arrays) + + def _hive_type_to_arrow(self, hive_type: str) -> pa.DataType: + """将 Hive 类型映射到 Arrow 类型""" + hive_type = hive_type.upper() + type_map = { + "STRING": pa.string(), + "VARCHAR": pa.string(), + "CHAR": pa.string(), + "INT": pa.int32(), + "INTEGER": pa.int32(), + "BIGINT": pa.int64(), + "SMALLINT": pa.int16(), + "TINYINT": pa.int8(), + "FLOAT": pa.float32(), + "DOUBLE": pa.float64(), + "DECIMAL": pa.float64(), + "BOOLEAN": pa.bool_(), + "BINARY": pa.binary(), + "TIMESTAMP": pa.timestamp("us"), + "DATE": pa.date32(), + } + return type_map.get(hive_type, pa.string()) + + def close(self): + self.cursor.close() + self.conn.close() + print("[Hive] 连接已关闭") + + +def create_backend(args) -> DatabaseBackend: + """根据参数创建数据库后端""" + if args.backend == "hive": + if not args.hive_host: + raise ValueError("Hive 模式需要指定 --hive-host") + return HiveBackend( + host=args.hive_host, + port=args.hive_port, + username=args.hive_user, + password=args.hive_password, + database=args.hive_database, + auth=args.hive_auth, + ) + else: + return DuckDBBackend(party=args.party, init_data=True) + + +# ============================================================================= +# ============================================================================= +# Protobuf parsing (using google.protobuf library) +# ============================================================================= + +def parse_flight_sql_command(data: bytes) -> str: + """ + Parse Arrow Flight SQL CommandStatementQuery protobuf message. + Uses google.protobuf to unwrap Any wrapper if present, + then extracts the query string from field 1. + """ + if not data: + return "" + + try: + from google.protobuf import descriptor_pb2 + from google.protobuf.any_pb2 import Any as AnyProto + from google.protobuf.descriptor_pool import DescriptorPool + from google.protobuf.message_factory import GetMessageClass + + if b"type.googleapis.com" in data: + any_msg = AnyProto() + any_msg.ParseFromString(data) + data = any_msg.value + + file_desc_proto = descriptor_pb2.FileDescriptorProto( + name="flight_sql.proto", + package="arrow.flight.protocol.sql", + message_type=[descriptor_pb2.DescriptorProto( + name="CommandStatementQuery", + field=[ + descriptor_pb2.FieldDescriptorProto( + name="query", number=1, type=9, label=1, + ), + descriptor_pb2.FieldDescriptorProto( + name="transaction_id", number=2, type=9, label=1, + ), + ], + )], + ) + pool = DescriptorPool() + pool.Add(file_desc_proto) + desc = pool.FindMessageTypeByName( + "arrow.flight.protocol.sql.CommandStatementQuery" + ) + CmdClass = GetMessageClass(desc) + msg = CmdClass() + msg.ParseFromString(data) + return msg.query + + except Exception as e: + print(f"[warning] protobuf parse failed, falling back to raw decode: {e}") + return data.decode("utf-8", errors="replace") + + +# ============================================================================= +# Arrow Flight SQL 服务器 +# ============================================================================= + +class FlightSqlServer(flight.FlightServerBase): + """ + Arrow Flight SQL 服务器实现 + 支持 DuckDB（测试）和 Hive（生产）两种后端 + """ + + def __init__(self, backend: DatabaseBackend, host="0.0.0.0", port=8815, + party="unknown"): + location = f"grpc://0.0.0.0:{port}" + super().__init__(location) + self.backend = backend + self.party = party + self._port = port + self._host = host + self._queries = {} # ticket_id -> query + self._ticket_counter = 0 + print(f"[{party}] Arrow Flight SQL 服务器启动在端口 {port}") def _preprocess_query(self, query: str) -> str: - """预处理 SQL 查询，将 default.table 转换为 "default".table""" + """ + Preprocess SQL query for the target backend. + + Uses HiveDialectConverter for Hive backend (full dialect conversion). + For DuckDB: strips party name prefixes and semicolons. + """ import re - # 匹配 default.tablename 并替换为 "default".tablename - query = re.sub(r'\bdefault\.(\w+)', r'"default".\1', query, flags=re.IGNORECASE) - return query + + if isinstance(self.backend, HiveBackend): + # Full Hive dialect conversion + if not hasattr(self, '_dialect'): + self._dialect = HiveDialectConverter( + party=self.party, database=self.backend.database + ) + return self._dialect.convert(query) + else: + # DuckDB: strip semicolons and party name prefixes + query = query.rstrip().rstrip(';').rstrip() + query = re.sub(r'\b(?:alice|bob|default|hive_demo)\.', '', query, flags=re.IGNORECASE) + return query def _generate_ticket(self, query: str) -> bytes: """生成唯一的 ticket ID""" @@ -212,36 +417,30 @@ def _generate_ticket(self, query: str) -> bytes: def get_flight_info(self, context, descriptor): """ 处理 GetFlightInfo 请求 - - Arrow Flight SQL 客户端通过此方法发送 SQL 查询。 - 命令被编码在 descriptor.command 中，格式为 CommandStatementQuery protobuf。 + Arrow Flight SQL 客户端通过此方法发送 SQL 查询 """ - # 从 descriptor 中提取 SQL 查询 if descriptor.descriptor_type == flight.DescriptorType.CMD: query = parse_flight_sql_command(descriptor.command) elif descriptor.descriptor_type == flight.DescriptorType.PATH: - # 表名查询 - table_name = "/".join(p.decode() if isinstance(p, bytes) else p for p in descriptor.path) + table_name = "/".join( + p.decode() if isinstance(p, bytes) else p + for p in descriptor.path + ) query = f"SELECT * FROM {table_name}" else: raise flight.FlightUnavailableError("Unsupported descriptor type") - # 预处理查询：将 default.table_name 转换为 "default".table_name query = self._preprocess_query(query) - print(f"[{self.party}] GetFlightInfo - Query: {query[:100]}...") - # 执行查询获取 schema try: - result = self.conn.execute(query).fetch_arrow_table() + result = self.backend.execute(query) schema = result.schema num_rows = result.num_rows - # 保存查询以供 DoGet 使用 ticket_bytes = self._generate_ticket(query) ticket = flight.Ticket(ticket_bytes) - # 创建 endpoint location = flight.Location.for_grpc_tcp("localhost", self._port) endpoint = flight.FlightEndpoint(ticket, [location]) @@ -250,10 +449,10 @@ def get_flight_info(self, context, descriptor): descriptor, [endpoint], num_rows, - -1 # 未知的字节数 + -1 ) - print(f"[{self.party}] FlightInfo created - rows: {num_rows}, columns: {len(schema)}") + print(f"[{self.party}] FlightInfo - rows: {num_rows}, columns: {len(schema)}") return info except Exception as e: @@ -261,50 +460,25 @@ def get_flight_info(self, context, descriptor): raise flight.FlightServerError(f"Query execution failed: {e}") def do_get(self, context, ticket): - """ - 处理 DoGet 请求，返回查询结果 - - ticket 包含查询 ID 或直接是 SQL 查询 - """ + """处理 DoGet 请求，返回查询结果""" ticket_data = ticket.ticket.decode("utf-8") - # 检查是否是保存的 ticket ID if ticket_data in self._queries: - query = self._queries[ticket_data] - # 清理已使用的 ticket - del self._queries[ticket_data] + query = self._queries.pop(ticket_data) else: - # 直接使用 ticket 作为查询 query = ticket_data - # 预处理查询 query = self._preprocess_query(query) - print(f"[{self.party}] DoGet - Query: {query[:100]}...") try: - result = self.conn.execute(query).fetch_arrow_table() + result = self.backend.execute(query) print(f"[{self.party}] 返回 {result.num_rows} 行, {result.num_columns} 列") return flight.RecordBatchStream(result) except Exception as e: print(f"[{self.party}] 查询错误: {e}") raise flight.FlightServerError(f"Query execution failed: {e}") - def list_flights(self, context, criteria): - """列出可用的表""" - tables = self.conn.execute("SHOW TABLES").fetchall() - for table in tables: - table_name = table[0] - descriptor = flight.FlightDescriptor.for_path(table_name) - schema = self.conn.execute(f"SELECT * FROM {table_name} LIMIT 0").fetch_arrow_table().schema - yield flight.FlightInfo( - schema, - descriptor, - [], - -1, - -1 - ) - def do_action(self, context, action): """处理 Action 请求""" action_type = action.type @@ -313,36 +487,101 @@ def do_action(self, context, action): if action_type == "healthcheck": yield flight.Result(b"ok") else: - # Flight SQL 使用各种 action，这里返回空结果 yield flight.Result(b"") def list_actions(self, context): """列出支持的 actions""" - return [ - ("healthcheck", "Health check"), - ] + return [("healthcheck", "Health check")] + + def shutdown(self): + """关闭服务器和后端连接""" + self.backend.close() + super().shutdown() def main(): - parser = argparse.ArgumentParser(description="Arrow Flight SQL 测试服务器") - parser.add_argument("--party", type=str, default="alice", choices=["alice", "bob"], - help="参与方名称 (alice 或 bob)") + parser = argparse.ArgumentParser( + description="Arrow Flight SQL 服务器 - 支持 DuckDB（测试）和 Hive（生产）后端", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +示例: + # DuckDB 模式（本地测试） + python3 arrow_flight_server.py --party alice --port 8815 + + # Hive 模式（连接真实 Hive） + python3 arrow_flight_server.py --port 8815 --backend hive \\ + --hive-host hive.example.com --hive-port 10000 \\ + --hive-user hive --hive-database default + """ + ) + + # 基本参数 + parser.add_argument("--party", type=str, default="alice", + help="参与方名称 (用于日志标识)") parser.add_argument("--port", type=int, default=8815, - help="服务端口") + help="Arrow Flight 服务端口 (默认: 8815)") parser.add_argument("--host", type=str, default="0.0.0.0", - help="监听地址") + help="监听地址 (默认: 0.0.0.0)") + + # 后端选择 + parser.add_argument("--backend", type=str, default="duckdb", + choices=["duckdb", "hive"], + help="数据库后端 (默认: duckdb)") + + # Hive 连接参数 + hive_group = parser.add_argument_group("Hive 连接参数") + hive_group.add_argument("--hive-host", type=str, + help="HiveServer2 主机地址") + hive_group.add_argument("--hive-port", type=int, default=10000, + help="HiveServer2 端口 (默认: 10000)") + hive_group.add_argument("--hive-user", type=str, + help="Hive 用户名") + hive_group.add_argument("--hive-password", type=str, + help="Hive 密码 (LDAP 认证时使用)") + hive_group.add_argument("--hive-database", type=str, default="default", + help="Hive 数据库 (默认: default)") + hive_group.add_argument("--hive-auth", type=str, default="NONE", + choices=["NONE", "LDAP", "KERBEROS"], + help="Hive 认证方式 (默认: NONE)") + args = parser.parse_args() - server = FlightSqlServer(host=args.host, port=args.port, party=args.party) - print(f"Arrow Flight SQL 服务器 [{args.party}] 正在运行...") - print(f"连接地址: grpc://localhost:{args.port}") + # 创建后端 + print("=" * 60) + print("Arrow Flight SQL 服务器") + print("=" * 60) + + try: + backend = create_backend(args) + except Exception as e: + print(f"[错误] 创建后端失败: {e}") + return 1 + + # 创建并启动服务器 + server = FlightSqlServer( + backend=backend, + host=args.host, + port=args.port, + party=args.party + ) + + print(f"后端: {args.backend.upper()}") + if args.backend == "hive": + print(f"Hive: {args.hive_host}:{args.hive_port}/{args.hive_database}") + print(f"监听: grpc://{args.host}:{args.port}") + print("-" * 60) print("按 Ctrl+C 停止服务器") + print() try: server.serve() except KeyboardInterrupt: - print(f"\n[{args.party}] 服务器已停止") + print(f"\n[{args.party}] 正在关闭服务器...") + server.shutdown() + print(f"[{args.party}] 服务器已停止") + + return 0 if __name__ == "__main__": - main() + exit(main()) diff --git a/examples/scdb-tutorial/hive/init_hive_test_data.sh b/examples/scdb-tutorial/hive/init_hive_test_data.sh new file mode 100755 index 00000000..109deb8b --- /dev/null +++ b/examples/scdb-tutorial/hive/init_hive_test_data.sh @@ -0,0 +1,84 @@ +#!/bin/bash +# 在本地 Hive 中初始化 SCQL 测试数据 +# 复用 initdb/ 目录下的 HQL 文件 + +HIVE_HOME=${HIVE_HOME:-/opt/hive} +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +INITDB_DIR="$SCRIPT_DIR/initdb" + +# 颜色定义 +GREEN='\033[0;32m' +RED='\033[0;31m' +BLUE='\033[0;34m' +NC='\033[0m' + +log_info() { echo -e "${BLUE}[INFO]${NC} $1"; } +log_success() { echo -e "${GREEN}[SUCCESS]${NC} $1"; } +log_error() { echo -e "${RED}[ERROR]${NC} $1"; } + +echo "==========================================" +echo " 初始化 Hive 测试数据" +echo "==========================================" +echo "" + +# 检查 HiveServer2 是否运行 +if ! nc -z localhost 10000 2>/dev/null; then + log_error "HiveServer2 未运行 (端口 10000)" + echo "请先运行: bash $SCRIPT_DIR/start_local_hive.sh" + exit 1 +fi + +# 检查 initdb 文件 +if [ ! -f "$INITDB_DIR/alice_init.hql" ] || [ ! -f "$INITDB_DIR/bob_init.hql" ]; then + log_error "找不到初始化文件: $INITDB_DIR/alice_init.hql 或 bob_init.hql" + exit 1 +fi + +BEELINE="$HIVE_HOME/bin/beeline -u jdbc:hive2://localhost:10000 --silent=true" + +# 初始化 Alice 数据 +log_info "初始化 Alice 数据 (alice_init.hql)..." +echo " 表: alice.user_credit" +$HIVE_HOME/bin/beeline -u 'jdbc:hive2://localhost:10000' --silent=true -f "$INITDB_DIR/alice_init.hql" + +if [ $? -eq 0 ]; then + log_success "✓ Alice 数据初始化成功" +else + log_error "✗ Alice 数据初始化失败" + exit 1 +fi + +# 初始化 Bob 数据 +log_info "初始化 Bob 数据 (bob_init.hql)..." +echo " 表: bob.user_stats" +$HIVE_HOME/bin/beeline -u 'jdbc:hive2://localhost:10000' --silent=true -f "$INITDB_DIR/bob_init.hql" + +if [ $? -eq 0 ]; then + log_success "✓ Bob 数据初始化成功" +else + log_error "✗ Bob 数据初始化失败" + exit 1 +fi + +echo "" +log_info "验证数据..." +echo "" + +echo "Alice - user_credit 表:" +$HIVE_HOME/bin/beeline -u 'jdbc:hive2://localhost:10000/alice' --silent=true -e "SELECT COUNT(*) as count FROM user_credit;" + +echo "" +echo "Bob - user_stats 表:" +$HIVE_HOME/bin/beeline -u 'jdbc:hive2://localhost:10000/bob' --silent=true -e "SELECT COUNT(*) as count FROM user_stats;" + +echo "" +echo "==========================================" +echo " 测试数据初始化完成" +echo "==========================================" +echo "" +echo "已创建的表:" +echo " - alice.user_credit (ID, credit_rank, income, age) - 19 行" +echo " - bob.user_stats (ID, order_amount, is_active) - 19 行" +echo "" +echo "下一步:" +echo " bash $SCRIPT_DIR/start_scql_with_real_hive.sh" diff --git a/examples/scdb-tutorial/hive/initdb/bob_init.hql b/examples/scdb-tutorial/hive/initdb/bob_init.hql index bf3b0f54..bb2b646f 100644 --- a/examples/scdb-tutorial/hive/initdb/bob_init.hql +++ b/examples/scdb-tutorial/hive/initdb/bob_init.hql @@ -7,29 +7,27 @@ USE bob; DROP TABLE IF EXISTS user_stats; CREATE TABLE user_stats ( ID STRING, - credit_rank INT, - income INT, - age INT -) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' STORED AS TEXTFILE; + order_amount INT, + is_active INT +) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',', STORED AS TEXTFILE; INSERT INTO user_stats VALUES - ('id0001', 6, 100000, 20), - ('id0002', 5, 90000, 19), - ('id0003', 6, 89700, 32), - ('id0005', 6, 607000, 30), - ('id0006', 5, 30070, 25), - ('id0007', 6, 12070, 28), - ('id0008', 6, 200800, 50), - ('id0009', 6, 607000, 30), - ('id0010', 5, 30070, 25), - ('id0011', 5, 12070, 28), - ('id0012', 6, 200800, 50), - ('id0013', 5, 30070, 25), - ('id0014', 5, 12070, 28), - ('id0015', 6, 200800, 18), - ('id0016', 5, 30070, 26), - ('id0017', 5, 12070, 27), - ('id0018', 6, 200800, 16), - ('id0019', 6, 30070, 25), - ('id0020', 5, 12070, 28); - + ('id0001', 5000, 1), + ('id0002', 3000, 1), + ('id0003', 8000, 0), + ('id0005', 12000, 1), + ('id0006', 1500, 1), + ('id0007', 2500, 0), + ('id0008', 9500, 1), + ('id0009', 7000, 1), + ('id0010', 500, 0), + ('id0011', 3500, 1), + ('id0012', 15000, 1), + ('id0013', 2000, 0), + ('id0014', 4500, 1), + ('id0015', 6500, 1), + ('id0016', 1000, 0), + ('id0017', 8500, 1), + ('id0018', 11000, 1), + ('id0019', 3200, 1), + ('id0020', 7500, 0); diff --git a/examples/scdb-tutorial/hive/start_local_hive.sh b/examples/scdb-tutorial/hive/start_local_hive.sh new file mode 100755 index 00000000..4e62a7f1 --- /dev/null +++ b/examples/scdb-tutorial/hive/start_local_hive.sh @@ -0,0 +1,140 @@ +#!/bin/bash +# 启动本地 Hive 服务脚本 +# 包括 Hive Metastore 和 HiveServer2 + +set -e + +HIVE_HOME=${HIVE_HOME:-/opt/hive} +HADOOP_HOME=${HADOOP_HOME:-/opt/hadoop} +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +LOG_DIR="$SCRIPT_DIR/logs" + +# 颜色定义 +GREEN='\033[0;32m' +RED='\033[0;31m' +YELLOW='\033[1;33m' +BLUE='\033[0;34m' +NC='\033[0m' + +log_info() { echo -e "${BLUE}[INFO]${NC} $1"; } +log_success() { echo -e "${GREEN}[SUCCESS]${NC} $1"; } +log_error() { echo -e "${RED}[ERROR]${NC} $1"; } +log_warning() { echo -e "${YELLOW}[WARNING]${NC} $1"; } + +echo "==========================================" +echo " 启动本地 Hive 服务" +echo "==========================================" +echo "" + +# 检查 Hive 安装 +if [ ! -f "$HIVE_HOME/bin/hive" ]; then + log_error "找不到 Hive: $HIVE_HOME/bin/hive" + exit 1 +fi + +# 创建日志目录 +mkdir -p "$LOG_DIR" + +# 停止可能正在运行的服务 +log_info "清理旧服务..." +pkill -f "HiveServer2" 2>/dev/null || true +pkill -f "HiveMetaStore" 2>/dev/null || true +sleep 2 + +# 清理 Derby 锁文件 +rm -f /tmp/hive/metastore_db/*.lck 2>/dev/null || true + +# 检查是否需要初始化 schema +if [ ! -d "/tmp/hive/metastore_db" ]; then + log_info "首次启动，初始化 Metastore schema..." + $HIVE_HOME/bin/schematool -dbType derby -initSchema 2>/dev/null || true +fi + +# 启动 Hive Metastore (监听 9083 端口) +log_info "启动 Hive Metastore (端口 9083)..." +nohup $HIVE_HOME/bin/hive --service metastore \ + > "$LOG_DIR/hive_metastore.log" 2>&1 & +METASTORE_PID=$! +echo $METASTORE_PID > "$LOG_DIR/metastore.pid" +log_success "Hive Metastore 启动 (PID: $METASTORE_PID)" + +# 等待 Metastore 就绪 +log_info "等待 Metastore 就绪..." +for i in {1..60}; do + if nc -z localhost 9083 2>/dev/null; then + log_success "Metastore 已就绪 (端口 9083)" + break + fi + echo -n "." + sleep 1 +done +echo "" + +if ! nc -z localhost 9083 2>/dev/null; then + log_error "Metastore 启动超时" + cat "$LOG_DIR/hive_metastore.log" + exit 1 +fi + +# 等待额外时间确保 Metastore 完全初始化 +sleep 5 + +# 启动 HiveServer2 (监听 10000 端口) +log_info "启动 HiveServer2 (端口 10000)..." +nohup $HIVE_HOME/bin/hiveserver2 \ + > "$LOG_DIR/hiveserver2.log" 2>&1 & +HS2_PID=$! +echo $HS2_PID > "$LOG_DIR/hiveserver2.pid" +log_success "HiveServer2 启动 (PID: $HS2_PID)" + +# 等待 HiveServer2 就绪 +log_info "等待 HiveServer2 就绪 (可能需要 1-2 分钟)..." +for i in {1..120}; do + if nc -z localhost 10000 2>/dev/null; then + log_success "HiveServer2 已就绪 (端口 10000)" + break + fi + echo -n "." + sleep 1 +done +echo "" + +# 验证服务状态 +echo "" +log_info "检查服务状态..." + +if nc -z localhost 9083 2>/dev/null; then + log_success "✓ Hive Metastore 运行在端口 9083" +else + log_error "✗ Hive Metastore 未响应" +fi + +if nc -z localhost 10000 2>/dev/null; then + log_success "✓ HiveServer2 运行在端口 10000" +else + log_warning "✗ HiveServer2 未响应 (端口 10000)" + log_info "检查日志: $LOG_DIR/hiveserver2.log" +fi + +echo "" +echo "==========================================" +echo " Hive 服务启动完成" +echo "==========================================" +echo "" +echo "连接信息:" +echo " Metastore: thrift://localhost:9083" +echo " HiveServer2: jdbc:hive2://localhost:10000" +echo " 用户: $(whoami)" +echo "" +echo "日志文件:" +echo " $LOG_DIR/hive_metastore.log" +echo " $LOG_DIR/hiveserver2.log" +echo "" +echo "测试连接:" +echo " $HIVE_HOME/bin/beeline -u 'jdbc:hive2://localhost:10000/default' -e 'show databases;'" +echo "" +echo "初始化测试数据:" +echo " bash $SCRIPT_DIR/init_hive_test_data.sh" +echo "" +echo "停止服务:" +echo " bash $SCRIPT_DIR/stop_local_hive.sh" diff --git a/examples/scdb-tutorial/hive/stop_local_hive.sh b/examples/scdb-tutorial/hive/stop_local_hive.sh new file mode 100755 index 00000000..0926e175 --- /dev/null +++ b/examples/scdb-tutorial/hive/stop_local_hive.sh @@ -0,0 +1,17 @@ +#!/bin/bash +# 停止本地 Hive 服务 + +echo "停止 Hive 服务..." + +# 停止 HiveServer2 +pkill -f "HiveServer2" && echo "✓ Stopped HiveServer2" || echo "HiveServer2 未运行" + +# 停止 Metastore +pkill -f "HiveMetaStore" && echo "✓ Stopped Hive Metastore" || echo "Hive Metastore 未运行" + +# 清理 PID 文件 +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +rm -f "$SCRIPT_DIR/logs/metastore.pid" "$SCRIPT_DIR/logs/hiveserver2.pid" 2>/dev/null + +echo "" +echo "Hive 服务已停止" diff --git a/examples/scdb-tutorial/start_all.sh b/examples/scdb-tutorial/start_all.sh index fefeb013..14813e70 100755 --- a/examples/scdb-tutorial/start_all.sh +++ b/examples/scdb-tutorial/start_all.sh @@ -1,8 +1,8 @@ #!/bin/bash # SCQL 本地启动脚本 -PROJECT_ROOT="/root/autodl-tmp/scql" -TUTORIAL_DIR="/root/autodl-tmp/scql/examples/scdb-tutorial" +PROJECT_ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd) +TUTORIAL_DIR="$PROJECT_ROOT/examples/scdb-tutorial" BIN_DIR="$PROJECT_ROOT/bin" echo "=========================================" diff --git a/examples/scdb-tutorial/start_all_hive.sh b/examples/scdb-tutorial/start_all_hive.sh index c329f460..37c8ec3f 100755 --- a/examples/scdb-tutorial/start_all_hive.sh +++ b/examples/scdb-tutorial/start_all_hive.sh @@ -2,8 +2,8 @@ # SCQL 本地启动脚本 - Hive 后端版本 # 使用 Arrow Flight SQL 服务器模拟 Hive 数据源 -PROJECT_ROOT="/root/autodl-tmp/scql" -TUTORIAL_DIR="/root/autodl-tmp/scql/examples/scdb-tutorial" +PROJECT_ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd) +TUTORIAL_DIR="$PROJECT_ROOT/examples/scdb-tutorial" HIVE_DIR="$TUTORIAL_DIR/hive" BIN_DIR="$PROJECT_ROOT/bin" diff --git a/pkg/parser/ast/functions.go b/pkg/parser/ast/functions.go index 52e3e0b9..9392f9c8 100644 --- a/pkg/parser/ast/functions.go +++ b/pkg/parser/ast/functions.go @@ -452,6 +452,49 @@ func (n *FuncCallExpr) RestoreDateFuncWithPostgresDialect(ctx *RestoreCtx) (err return nil } +func (n *FuncCallExpr) RestoreDateFuncWithHiveDialect(ctx *RestoreCtx) (err error) { + switch n.FnName.L { + case Curdate: + ctx.WriteKeyWord(ctx.Dialect.GetSpecialFuncName(n.FnName.L)) + ctx.WritePlain("()") + case Now: + ctx.WriteKeyWord(ctx.Dialect.GetSpecialFuncName(n.FnName.L)) + ctx.WritePlain("()") + case AddDate, SubDate, DateAdd, DateSub: + ctx.WriteKeyWord(n.FnName.O) + ctx.WritePlain("(") + if err = n.Args[0].Restore(ctx); err != nil { + return errors.Annotatef(err, "An error occurred while restore FuncCallExpr.Args[0]") + } + ctx.WritePlain(", ") + ctx.WriteKeyWord("INTERVAL ") + if err := n.Args[1].Restore(ctx); err != nil { + return errors.Annotatef(err, "An error occurred while restore FuncCallExpr.Args[1]") + } + ctx.WritePlain(" ") + old_ctx_flags := ctx.Flags + ctx.Flags &= ^format.RestoreStringSingleQuotes + if err := n.Args[2].Restore(ctx); err != nil { + return errors.Annotatef(err, "An error occurred while restore FuncCallExpr.Args[2]") + } + ctx.Flags = old_ctx_flags + ctx.WritePlain(")") + default: + ctx.WriteKeyWord(n.FnName.O) + ctx.WritePlain("(") + for i, argv := range n.Args { + if i != 0 { + ctx.WritePlain(", ") + } + if err := argv.Restore(ctx); err != nil { + return errors.Annotatef(err, "An error occurred while restore FuncCallExpr.Args %d", i) + } + } + ctx.WritePlain(")") + } + return nil +} + func (n *FuncCallExpr) RestoreDateFuncWithCSVDBDialect(ctx *RestoreCtx) (err error) { switch n.FnName.L { case Curdate: @@ -530,6 +573,8 @@ func (n *FuncCallExpr) Restore(ctx *RestoreCtx) error { return n.RestoreDateFuncWithPostgresDialect(ctx) case *CVSDBDialect: return n.RestoreDateFuncWithCSVDBDialect(ctx) + case *HiveDialect: + return n.RestoreDateFuncWithHiveDialect(ctx) default: return n.RestoreDateFuncWithMysqlDialect(ctx) } diff --git a/pkg/parser/format/format_dialect.go b/pkg/parser/format/format_dialect.go index 749f58b7..131cd6e6 100644 --- a/pkg/parser/format/format_dialect.go +++ b/pkg/parser/format/format_dialect.go @@ -376,15 +376,77 @@ func (d *OdpsDialect) NeedParenthesesForCmpOperand() bool { type HiveDialect struct { MySQLDialect funcNameMap map[string]string + operatorMap map[string]string } func NewHiveDialect() Dialect { return &HiveDialect{ funcNameMap: map[string]string{ - "ifnull": "nvl", + // NULL handling + "ifnull": "nvl", + + // Date/Time functions + "now": "current_timestamp", + "curdate": "current_date", + "current_date": "current_date", + "sysdate": "current_timestamp", + "unix_timestamp": "unix_timestamp", + "from_unixtime": "from_unixtime", + + // Math functions "truncate": "trunc", - "now": "current_timestamp", - "curdate": "current_date", + "ceil": "ceil", + "ceiling": "ceil", + "floor": "floor", + "round": "round", + "abs": "abs", + "pow": "pow", + "power": "power", + "sqrt": "sqrt", + "ln": "ln", + "log": "log", + "log10": "log10", + "log2": "log2", + "exp": "exp", + "rand": "rand", + "sign": "sign", + + // String functions + "char_length": "length", + "character_length": "length", + "octet_length": "length", + "lcase": "lower", + "ucase": "upper", + "substr": "substr", + "substring": "substr", + "concat": "concat", + "concat_ws": "concat_ws", + "trim": "trim", + "ltrim": "ltrim", + "rtrim": "rtrim", + "lpad": "lpad", + "rpad": "rpad", + "reverse": "reverse", + "repeat": "repeat", + "replace": "regexp_replace", // Hive uses regexp_replace + "upper": "upper", + "lower": "lower", + "length": "length", + "instr": "instr", + "space": "space", + "ascii": "ascii", + + // Conditional functions + "coalesce": "coalesce", + "nullif": "nullif", + "greatest": "greatest", + "least": "least", + }, + operatorMap: map[string]string{ + // Hive uses DIV for integer division (same as MySQL) + " DIV ": " DIV ", + // MOD operator + " MOD ": " % ", }, } } @@ -402,8 +464,11 @@ func (d *HiveDialect) GetSpecialFuncName(originName string) string { func (d *HiveDialect) ConvertCastTypeToString(asType byte, flen int, decimal int, flag uint) (keyword string, plainWord string, err error) { switch asType { - case mysql.TypeVarString, mysql.TypeVarchar: + // String types -> STRING + case mysql.TypeVarString, mysql.TypeVarchar, mysql.TypeString: keyword = "STRING" + + // Decimal type case mysql.TypeNewDecimal: keyword = "DECIMAL" if flen > 0 && decimal > 0 { @@ -411,24 +476,63 @@ func (d *HiveDialect) ConvertCastTypeToString(asType byte, flen int, decimal int } else if flen > 0 { plainWord = fmt.Sprintf("(%d)", flen) } + + // Integer types + case mysql.TypeTiny: + // TINYINT in Hive, or BOOLEAN if it's a boolean flag + if flag&mysql.IsBooleanFlag != 0 { + keyword = "BOOLEAN" + } else { + keyword = "TINYINT" + } + case mysql.TypeShort: + if flag&mysql.UnsignedFlag != 0 { + err = fmt.Errorf("unsupported cast as unsigned smallint in Hive") + return + } + keyword = "SMALLINT" + case mysql.TypeLong, mysql.TypeInt24: + if flag&mysql.UnsignedFlag != 0 { + err = fmt.Errorf("unsupported cast as unsigned int in Hive") + return + } + keyword = "INT" case mysql.TypeLonglong: if flag&mysql.UnsignedFlag != 0 { - err = fmt.Errorf("unsupported cast as data type %+v", asType) + err = fmt.Errorf("unsupported cast as unsigned bigint in Hive") return } keyword = "BIGINT" - case mysql.TypeDouble, mysql.TypeFloat: + + // Floating point types + case mysql.TypeFloat: + keyword = "FLOAT" + case mysql.TypeDouble: keyword = "DOUBLE" - case mysql.TypeDate: + + // Date/Time types + case mysql.TypeDate, mysql.TypeNewDate: keyword = "DATE" - case mysql.TypeDatetime: + case mysql.TypeDatetime, mysql.TypeTimestamp: keyword = "TIMESTAMP" + + // Binary types -> BINARY + case mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob: + keyword = "BINARY" + default: - return d.MySQLDialect.ConvertCastTypeToString(asType, flen, decimal, flag) + err = fmt.Errorf("unsupported cast as data type in Hive: %+v", asType) } return } +func (d *HiveDialect) GetOperator(originName string) string { + if res, ok := d.operatorMap[originName]; ok { + return res + } + return originName +} + func (d *HiveDialect) NeedParenthesesForCmpOperand() bool { return true } diff --git a/pkg/planner/core/hive_dialect_test.go b/pkg/planner/core/hive_dialect_test.go new file mode 100644 index 00000000..b847e129 --- /dev/null +++ b/pkg/planner/core/hive_dialect_test.go @@ -0,0 +1,294 @@ +// Copyright 2023 Ant Group Co., Ltd. +// +// Licensed 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. + +package core + +import ( + "bytes" + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/secretflow/scql/pkg/infoschema" + "github.com/secretflow/scql/pkg/parser" + "github.com/secretflow/scql/pkg/parser/format" + "github.com/secretflow/scql/pkg/util/mock" + + _ "github.com/secretflow/scql/pkg/types/parser_driver" +) + +// TestHiveDialectConversion tests that SQL is properly converted to Hive dialect +func TestHiveDialectConversion(t *testing.T) { + mockTables, err := mock.MockAllTables() + require.NoError(t, err) + is := infoschema.MockInfoSchema(mockTables) + ctx := mock.MockContext() + p := parser.New() + dialect := NewHiveDialect() + + testCases := []struct { + name string + inputSQL string + expectedSQL string + skip bool + }{ + { + name: "Simple select", + inputSQL: "select plain_float_0 from alice.tbl_1", + expectedSQL: "select tbl_1.plain_float_0 from alice.tbl_1", + }, + { + name: "IFNULL to NVL conversion", + inputSQL: "select ifnull(plain_float_0, plain_float_1) from alice.tbl_1", + expectedSQL: "select nvl(tbl_1.plain_float_0, tbl_1.plain_float_1) as expr_121 from alice.tbl_1", + }, + { + name: "TRUNCATE to TRUNC conversion", + inputSQL: "select truncate(plain_float_0, 2) from alice.tbl_1", + expectedSQL: "select trunc(tbl_1.plain_float_0, 2) as expr_121 from alice.tbl_1", + }, + { + name: "NOW to CURRENT_TIMESTAMP conversion", + inputSQL: "select plain_datetime_0 < now() from alice.tbl_1", + expectedSQL: "select tbl_1.plain_datetime_0 0 and plain_float_1 < 100", + expectedSQL: "select tbl_1.plain_float_0 from alice.tbl_1 where (tbl_1.plain_float_0>0) and (tbl_1.plain_float_1<100)", + }, + { + name: "LIMIT and OFFSET", + inputSQL: "select plain_float_0 from alice.tbl_1 limit 10 offset 5", + expectedSQL: "select tbl_1.plain_float_0 from alice.tbl_1 limit 10 offset 5", + }, + // Extended function mappings tests (only using SCQL-supported functions) + { + name: "SUBSTRING to SUBSTR", + inputSQL: "select substring(plain_string_0, 1, 5) from alice.tbl_0", + expectedSQL: "select substr(tbl_0.plain_string_0, 1, 5) as expr_121 from alice.tbl_0", + }, + { + name: "Math function CEIL", + inputSQL: "select ceil(plain_float_0) from alice.tbl_1", + expectedSQL: "select ceil(tbl_1.plain_float_0) as expr_121 from alice.tbl_1", + }, + { + name: "Math function FLOOR", + inputSQL: "select floor(plain_float_0) from alice.tbl_1", + expectedSQL: "select floor(tbl_1.plain_float_0) as expr_121 from alice.tbl_1", + }, + { + name: "Math function ROUND", + inputSQL: "select round(plain_float_0, 2) from alice.tbl_1", + expectedSQL: "select round(tbl_1.plain_float_0, 2) as expr_121 from alice.tbl_1", + }, + { + name: "Math function ABS", + inputSQL: "select abs(plain_float_0) from alice.tbl_1", + expectedSQL: "select abs(tbl_1.plain_float_0) as expr_121 from alice.tbl_1", + }, + { + name: "TRIM function", + inputSQL: "select trim(plain_string_0) from alice.tbl_0", + expectedSQL: "select trim(tbl_0.plain_string_0) as expr_121 from alice.tbl_0", + }, + { + name: "COALESCE function", + inputSQL: "select coalesce(plain_float_0, plain_float_1, 0) from alice.tbl_1", + expectedSQL: "select coalesce(tbl_1.plain_float_0, tbl_1.plain_float_1, 0) as expr_121 from alice.tbl_1", + }, + { + name: "Math function SQRT", + inputSQL: "select sqrt(plain_float_0) from alice.tbl_1", + expectedSQL: "select sqrt(tbl_1.plain_float_0) as expr_121 from alice.tbl_1", + }, + { + name: "Math function LN", + inputSQL: "select ln(plain_float_0) from alice.tbl_1", + expectedSQL: "select ln(tbl_1.plain_float_0) as expr_121 from alice.tbl_1", + }, + { + name: "Math function LOG10", + inputSQL: "select log10(plain_float_0) from alice.tbl_1", + expectedSQL: "select log10(tbl_1.plain_float_0) as expr_121 from alice.tbl_1", + }, + { + name: "Math function EXP", + inputSQL: "select exp(plain_float_0) from alice.tbl_1", + expectedSQL: "select exp(tbl_1.plain_float_0) as expr_121 from alice.tbl_1", + }, + { + name: "Math function POW", + inputSQL: "select pow(plain_float_0, 2) from alice.tbl_1", + expectedSQL: "select pow(tbl_1.plain_float_0, 2) as expr_121 from alice.tbl_1", + }, + { + name: "LENGTH function", + inputSQL: "select length(plain_string_0) from alice.tbl_0", + expectedSQL: "select length(tbl_0.plain_string_0) as expr_121 from alice.tbl_0", + }, + { + name: "REPLACE function (maps to regexp_replace in Hive)", + inputSQL: "select replace(plain_string_0, 'a', 'b') from alice.tbl_0", + expectedSQL: "select regexp_replace(tbl_0.plain_string_0, 'a', 'b') as expr_121 from alice.tbl_0", + }, + { + name: "INSTR function", + inputSQL: "select instr(plain_string_0, 'test') from alice.tbl_0", + expectedSQL: "select instr(tbl_0.plain_string_0, 'test') as expr_121 from alice.tbl_0", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + if tc.skip { + t.Skip("Test case skipped") + return + } + + stmt, err := p.ParseOneStmt(tc.inputSQL, "", "") + require.NoError(t, err, "Failed to parse SQL: %s", tc.inputSQL) + + err = Preprocess(ctx, stmt, is) + require.NoError(t, err, "Failed to preprocess SQL: %s", tc.inputSQL) + + lp, _, err := BuildLogicalPlanWithOptimization(context.Background(), ctx, stmt, is) + require.NoError(t, err, "Failed to build logical plan: %s", tc.inputSQL) + + sqlCtx, err := BuildChildCtx(dialect, lp) + require.NoError(t, err, "Failed to build SQL context") + + newStmt, err := sqlCtx.GetSQLStmt() + require.NoError(t, err, "Failed to get SQL statement") + + b := new(bytes.Buffer) + err = newStmt.Restore(format.NewRestoreCtxWithDialect( + format.RestoreStringSingleQuotes|format.RestoreKeyWordLowercase, + b, + dialect.GetFormatDialect(), + )) + require.NoError(t, err, "Failed to restore SQL") + + actualSQL := b.String() + assert.Equal(t, tc.expectedSQL, actualSQL, "SQL conversion mismatch for input: %s", tc.inputSQL) + }) + } +} + +// TestHiveDialectProperties tests the properties of HiveDialect +func TestHiveDialectProperties(t *testing.T) { + dialect := NewHiveDialect() + + t.Run("SupportAnyValue should be false", func(t *testing.T) { + assert.False(t, dialect.SupportAnyValue(), "Hive should not support ANY_VALUE") + }) + + t.Run("GetRestoreFlags", func(t *testing.T) { + flags := dialect.GetRestoreFlags() + assert.True(t, flags&format.RestoreStringSingleQuotes != 0, "Should use single quotes for strings") + assert.True(t, flags&format.RestoreKeyWordLowercase != 0, "Should use lowercase keywords") + }) + + t.Run("GetFormatDialect should not be nil", func(t *testing.T) { + assert.NotNil(t, dialect.GetFormatDialect()) + }) +} + +// TestDBTypeHive tests DBType constants and parsing +func TestDBTypeHive(t *testing.T) { + t.Run("DBTypeHive constant", func(t *testing.T) { + assert.Equal(t, "hive", DBTypeHive.String()) + }) + + t.Run("ParseDBType for hive", func(t *testing.T) { + dbType, err := ParseDBType("hive") + require.NoError(t, err) + assert.Equal(t, DBTypeHive, dbType) + }) + + t.Run("ParseDBType case insensitive", func(t *testing.T) { + dbType, err := ParseDBType("HIVE") + require.NoError(t, err) + assert.Equal(t, DBTypeHive, dbType) + }) + + t.Run("DBDialectMap contains Hive", func(t *testing.T) { + dialect, ok := DBDialectMap[DBTypeHive] + assert.True(t, ok, "DBDialectMap should contain Hive") + assert.NotNil(t, dialect) + }) +} diff --git a/pkg/planner/core/logicalplan_to_stmt_test.go b/pkg/planner/core/logicalplan_to_stmt_test.go index 5ca7b019..c689fec6 100644 --- a/pkg/planner/core/logicalplan_to_stmt_test.go +++ b/pkg/planner/core/logicalplan_to_stmt_test.go @@ -38,8 +38,8 @@ import ( var _ = Suite(&testRunSQLSuite{}) -// Note: Hive is not included in testBackEnds for now, add test cases to runsql_in.json before enabling -var testBackEnds = []string{MySQL, Postgres, ODPS, CSV} +// Hive is now included in testBackEnds +var testBackEnds = []string{MySQL, Postgres, ODPS, CSV, Hive} const ( MySQL = "MYSQL" diff --git a/pkg/planner/core/testdata/runsql_in.json b/pkg/planner/core/testdata/runsql_in.json index 1429ae02..658fcd24 100644 --- a/pkg/planner/core/testdata/runsql_in.json +++ b/pkg/planner/core/testdata/runsql_in.json @@ -95,7 +95,8 @@ "rewritten_sql": "select ifnull(alice.plain_float_0, alice.plain_float_1) as fl from alice.tbl_1 as alice", "rewritten_sql_pg": "select coalesce(alice.plain_float_0, alice.plain_float_1) as fl from alice.tbl_1 as alice", "rewritten_sql_odps": "select nvl(alice.plain_float_0, alice.plain_float_1) as fl from alice.tbl_1 as alice", - "rewritten_sql_csv": "select coalesce(alice.plain_float_0, alice.plain_float_1) as fl from alice.tbl_1 as alice" + "rewritten_sql_csv": "select coalesce(alice.plain_float_0, alice.plain_float_1) as fl from alice.tbl_1 as alice", + "rewritten_sql_hive": "select nvl(alice.plain_float_0, alice.plain_float_1) as fl from alice.tbl_1 as alice" }, { "sql": "select plain_float_0 is null as fl from alice.tbl_1 where plain_float_1 is not null;", @@ -131,7 +132,8 @@ // this not really work for pg, it is just for test, since pg can't compare int with boolean "rewritten_sql_pg": "select (tbl_1.plain_float_0=tbl_1.plain_float_1)=1 as fl from alice.tbl_1", "rewritten_sql_csv": "select (tbl_1.plain_float_0=tbl_1.plain_float_1)=1 as fl from alice.tbl_1", - "rewritten_sql_odps": "select (tbl_1.plain_float_0=tbl_1.plain_float_1)=1 as fl from alice.tbl_1" + "rewritten_sql_odps": "select (tbl_1.plain_float_0=tbl_1.plain_float_1)=1 as fl from alice.tbl_1", + "rewritten_sql_hive": "select (tbl_1.plain_float_0=tbl_1.plain_float_1)=1 as fl from alice.tbl_1" }, { "sql": "select plain_int_0 % plain_int_0 as fl from alice.tbl_1 where tbl_1.plain_int_0 != 0;", @@ -191,7 +193,8 @@ "rewritten_sql": "select any_value(tbl_1.plain_float_0) as expr_242 from alice.tbl_1 join alice.tbl_2 on tbl_1.plain_int_0=tbl_2.plain_int_0 group by tbl_1.plain_float_0 having count(1)>0", "rewritten_sql_pg": "select tbl_1.plain_float_0 from alice.tbl_1 join alice.tbl_2 on tbl_1.plain_int_0=tbl_2.plain_int_0 group by tbl_1.plain_float_0 having count(1)>0", "rewritten_sql_odps": "select tbl_1.plain_float_0 from alice.tbl_1 join alice.tbl_2 on tbl_1.plain_int_0=tbl_2.plain_int_0 group by tbl_1.plain_float_0 having count(1)>0", - "rewritten_sql_csv": "select tbl_1.plain_float_0 from alice.tbl_1 join alice.tbl_2 on tbl_1.plain_int_0=tbl_2.plain_int_0 group by tbl_1.plain_float_0 having count(1)>0" + "rewritten_sql_csv": "select tbl_1.plain_float_0 from alice.tbl_1 join alice.tbl_2 on tbl_1.plain_int_0=tbl_2.plain_int_0 group by tbl_1.plain_float_0 having count(1)>0", + "rewritten_sql_hive": "select tbl_1.plain_float_0 from alice.tbl_1 join alice.tbl_2 on tbl_1.plain_int_0=tbl_2.plain_int_0 group by tbl_1.plain_float_0 having count(1)>0" }, { "sql": "select tbl_1.plain_float_0 from alice.tbl_1 join alice.tbl_2 on tbl_1.plain_int_0 = tbl_2.plain_int_0 group by tbl_1.plain_float_0 having count(*) > 0 and sum(tbl_1.plain_float_0) > 100 and tbl_1.plain_float_0 > 3;", @@ -199,7 +202,8 @@ "rewritten_sql": "select any_value(t_0.plain_float_0) as expr_243 from (select tbl_1.plain_float_0,tbl_1.plain_int_0 from alice.tbl_1 where tbl_1.plain_float_0>3) as t_0 join alice.tbl_2 on t_0.plain_int_0=tbl_2.plain_int_0 group by t_0.plain_float_0 having (count(1)>0) and (sum(t_0.plain_float_0)>100)", "rewritten_sql_pg": "select t_0.plain_float_0 from (select tbl_1.plain_float_0,tbl_1.plain_int_0 from alice.tbl_1 where tbl_1.plain_float_0>3) as t_0 join alice.tbl_2 on t_0.plain_int_0=tbl_2.plain_int_0 group by t_0.plain_float_0 having (count(1)>0) and (sum(t_0.plain_float_0)>100)", "rewritten_sql_odps": "select t_0.plain_float_0 from (select tbl_1.plain_float_0,tbl_1.plain_int_0 from alice.tbl_1 where tbl_1.plain_float_0>3) as t_0 join alice.tbl_2 on t_0.plain_int_0=tbl_2.plain_int_0 group by t_0.plain_float_0 having (count(1)>0) and (sum(t_0.plain_float_0)>100)", - "rewritten_sql_csv": "select t_0.plain_float_0 from (select tbl_1.plain_float_0,tbl_1.plain_int_0 from alice.tbl_1 where tbl_1.plain_float_0>3) as t_0 join alice.tbl_2 on t_0.plain_int_0=tbl_2.plain_int_0 group by t_0.plain_float_0 having (count(1)>0) and (sum(t_0.plain_float_0)>100)" + "rewritten_sql_csv": "select t_0.plain_float_0 from (select tbl_1.plain_float_0,tbl_1.plain_int_0 from alice.tbl_1 where tbl_1.plain_float_0>3) as t_0 join alice.tbl_2 on t_0.plain_int_0=tbl_2.plain_int_0 group by t_0.plain_float_0 having (count(1)>0) and (sum(t_0.plain_float_0)>100)", + "rewritten_sql_hive": "select t_0.plain_float_0 from (select tbl_1.plain_float_0,tbl_1.plain_int_0 from alice.tbl_1 where tbl_1.plain_float_0>3) as t_0 join alice.tbl_2 on t_0.plain_int_0=tbl_2.plain_int_0 group by t_0.plain_float_0 having (count(1)>0) and (sum(t_0.plain_float_0)>100)" }, { "sql": "select count(*) as c from alice.tbl_1 join alice.tbl_2 on tbl_1.plain_int_0 = tbl_2.plain_int_0 group by tbl_1.plain_float_0;", @@ -312,7 +316,8 @@ "rewritten_sql": "select count(tt.expr_244) as expr_243 from ((select count(1) as expr_244,any_value(t.encrypt_float_0) as expr_241 from alice.tbl_1 as t group by t.encrypt_float_0) union all (select count(1) as expr_244,any_value(t1.encrypt_float_0) as expr_241 from alice.tbl_1 as t1 group by t1.encrypt_float_0)) as tt group by tt.expr_241", "rewritten_sql_pg": "select count(tt.expr_244) as expr_243 from ((select count(1) as expr_244,t.encrypt_float_0 from alice.tbl_1 as t group by t.encrypt_float_0) union all (select count(1) as expr_244,t1.encrypt_float_0 from alice.tbl_1 as t1 group by t1.encrypt_float_0)) as tt group by tt.encrypt_float_0", "rewritten_sql_odps": "select count(tt.expr_244) as expr_243 from ((select count(1) as expr_244,t.encrypt_float_0 from alice.tbl_1 as t group by t.encrypt_float_0) union all (select count(1) as expr_244,t1.encrypt_float_0 from alice.tbl_1 as t1 group by t1.encrypt_float_0)) as tt group by tt.encrypt_float_0", - "rewritten_sql_csv": "select count(tt.expr_244) as expr_243 from ((select count(1) as expr_244,t.encrypt_float_0 from alice.tbl_1 as t group by t.encrypt_float_0) union all (select count(1) as expr_244,t1.encrypt_float_0 from alice.tbl_1 as t1 group by t1.encrypt_float_0)) as tt group by tt.encrypt_float_0" + "rewritten_sql_csv": "select count(tt.expr_244) as expr_243 from ((select count(1) as expr_244,t.encrypt_float_0 from alice.tbl_1 as t group by t.encrypt_float_0) union all (select count(1) as expr_244,t1.encrypt_float_0 from alice.tbl_1 as t1 group by t1.encrypt_float_0)) as tt group by tt.encrypt_float_0", + "rewritten_sql_hive": "select count(tt.expr_244) as expr_243 from ((select count(1) as expr_244,t.encrypt_float_0 from alice.tbl_1 as t group by t.encrypt_float_0) union all (select count(1) as expr_244,t1.encrypt_float_0 from alice.tbl_1 as t1 group by t1.encrypt_float_0)) as tt group by tt.encrypt_float_0" }, { "sql": "select min(alice.compare_float_0*alice.compare_float_1) as m from alice.tbl_0 as alice;;", @@ -335,7 +340,8 @@ "rewritten_sql": "select cast(alice.compare_string_0 as datetime) as m from alice.tbl_0 as alice", "rewritten_sql_pg": "select cast(alice.compare_string_0 as timestamp) as m from alice.tbl_0 as alice", "rewritten_sql_csv": "select cast(alice.compare_string_0 as timestamp) as m from alice.tbl_0 as alice", - "rewritten_sql_odps": "select cast(alice.compare_string_0 as datetime) as m from alice.tbl_0 as alice" + "rewritten_sql_odps": "select cast(alice.compare_string_0 as datetime) as m from alice.tbl_0 as alice", + "rewritten_sql_hive": "select cast(alice.compare_string_0 as timestamp) as m from alice.tbl_0 as alice" }, { "sql": "select cast(alice.compare_string_0 as CHAR(100)) as m from alice.tbl_0 as alice;;", @@ -343,7 +349,8 @@ "rewritten_sql": "select cast(alice.compare_string_0 as char(100)) as m from alice.tbl_0 as alice", "rewritten_sql_odps": "select cast(alice.compare_string_0 as string) as m from alice.tbl_0 as alice", "rewritten_sql_pg": "select cast(alice.compare_string_0 as varchar) as m from alice.tbl_0 as alice", - "rewritten_sql_csv": "select cast(alice.compare_string_0 as varchar) as m from alice.tbl_0 as alice" + "rewritten_sql_csv": "select cast(alice.compare_string_0 as varchar) as m from alice.tbl_0 as alice", + "rewritten_sql_hive": "select cast(alice.compare_string_0 as string) as m from alice.tbl_0 as alice" }, { "sql": "select cast(alice.compare_float_0 as double) as m from alice.tbl_0 as alice;;", @@ -351,7 +358,8 @@ "rewritten_sql": "select cast(alice.compare_float_0 as decimal(64,30)) as m from alice.tbl_0 as alice", "rewritten_sql_pg": "select cast(alice.compare_float_0 as double precision) as m from alice.tbl_0 as alice", "rewritten_sql_csv": "select cast(alice.compare_float_0 as double precision) as m from alice.tbl_0 as alice", - "rewritten_sql_odps": "select cast(alice.compare_float_0 as double) as m from alice.tbl_0 as alice" + "rewritten_sql_odps": "select cast(alice.compare_float_0 as double) as m from alice.tbl_0 as alice", + "rewritten_sql_hive": "select cast(alice.compare_float_0 as double) as m from alice.tbl_0 as alice" }, { "sql": "select cast(alice.compare_float_0 as decimal(11)) as m from alice.tbl_0 as alice;;", @@ -367,7 +375,8 @@ "rewritten_sql": "select cast(alice.compare_float_0 as signed) as m from alice.tbl_0 as alice", "rewritten_sql_pg": "select cast(alice.compare_float_0 as integer) as m from alice.tbl_0 as alice", "rewritten_sql_csv": "select cast(alice.compare_float_0 as integer) as m from alice.tbl_0 as alice", - "rewritten_sql_odps": "select cast(alice.compare_float_0 as bigint) as m from alice.tbl_0 as alice" + "rewritten_sql_odps": "select cast(alice.compare_float_0 as bigint) as m from alice.tbl_0 as alice", + "rewritten_sql_hive": "select cast(alice.compare_float_0 as bigint) as m from alice.tbl_0 as alice" }, { "sql": "select cast(alice.compare_float_0 as unsigned) as m from alice.tbl_0 as alice;;", @@ -375,6 +384,7 @@ "skip_pg_test": true, "skip_csv_test": true, "skip_odps_test": true, + "skip_hive_test": true, "rewritten_sql": "select cast(alice.compare_float_0 as unsigned) as m from alice.tbl_0 as alice" }, { @@ -389,7 +399,8 @@ "skip_odps_test": true, "rewritten_sql": "select cast(alice.compare_datetime_0 as datetime) as m from alice.tbl_0 as alice", "rewritten_sql_pg": "select cast(alice.compare_datetime_0 as timestamp) as m from alice.tbl_0 as alice", - "rewritten_sql_csv": "select cast(alice.compare_datetime_0 as timestamp) as m from alice.tbl_0 as alice" + "rewritten_sql_csv": "select cast(alice.compare_datetime_0 as timestamp) as m from alice.tbl_0 as alice", + "rewritten_sql_hive": "select cast(alice.compare_datetime_0 as timestamp) as m from alice.tbl_0 as alice" }, { "sql": "SELECT 'David!' LIKE 'David\\_' as tt from alice.tbl_0 as alice;;", @@ -423,7 +434,8 @@ "rewritten_sql": "select tbl_1.plain_int_0,3+tbl_1.plain_int_1=2 or 3+tbl_1.plain_int_1=5 or 3+tbl_1.plain_int_1=15 as ee from alice.tbl_1 where (tbl_1.plain_int_0=2 or tbl_1.plain_int_0=5 or tbl_1.plain_int_0=15) and (not(tbl_1.plain_int_1=2 or tbl_1.plain_int_1=5 or tbl_1.plain_int_1=15))", "rewritten_sql_pg": "select tbl_1.plain_int_0,(3+tbl_1.plain_int_1=2) or (3+tbl_1.plain_int_1=5) or (3+tbl_1.plain_int_1=15) as ee from alice.tbl_1 where ((tbl_1.plain_int_0=2) or (tbl_1.plain_int_0=5) or (tbl_1.plain_int_0=15)) and (not((tbl_1.plain_int_1=2) or (tbl_1.plain_int_1=5) or (tbl_1.plain_int_1=15)))", "rewritten_sql_csv": "select tbl_1.plain_int_0,(3+tbl_1.plain_int_1=2) or (3+tbl_1.plain_int_1=5) or (3+tbl_1.plain_int_1=15) as ee from alice.tbl_1 where ((tbl_1.plain_int_0=2) or (tbl_1.plain_int_0=5) or (tbl_1.plain_int_0=15)) and (not((tbl_1.plain_int_1=2) or (tbl_1.plain_int_1=5) or (tbl_1.plain_int_1=15)))", - "rewritten_sql_odps": "select tbl_1.plain_int_0,(3+tbl_1.plain_int_1=2) or (3+tbl_1.plain_int_1=5) or (3+tbl_1.plain_int_1=15) as ee from alice.tbl_1 where ((tbl_1.plain_int_0=2) or (tbl_1.plain_int_0=5) or (tbl_1.plain_int_0=15)) and (not((tbl_1.plain_int_1=2) or (tbl_1.plain_int_1=5) or (tbl_1.plain_int_1=15)))" + "rewritten_sql_odps": "select tbl_1.plain_int_0,(3+tbl_1.plain_int_1=2) or (3+tbl_1.plain_int_1=5) or (3+tbl_1.plain_int_1=15) as ee from alice.tbl_1 where ((tbl_1.plain_int_0=2) or (tbl_1.plain_int_0=5) or (tbl_1.plain_int_0=15)) and (not((tbl_1.plain_int_1=2) or (tbl_1.plain_int_1=5) or (tbl_1.plain_int_1=15)))", + "rewritten_sql_hive": "select tbl_1.plain_int_0,(3+tbl_1.plain_int_1=2) or (3+tbl_1.plain_int_1=5) or (3+tbl_1.plain_int_1=15) as ee from alice.tbl_1 where ((tbl_1.plain_int_0=2) or (tbl_1.plain_int_0=5) or (tbl_1.plain_int_0=15)) and (not((tbl_1.plain_int_1=2) or (tbl_1.plain_int_1=5) or (tbl_1.plain_int_1=15)))" }, { "sql": "select plain_int_0 + plain_int_0 + 5, (plain_int_1 + plain_int_2 + plain_int_0) > plain_int_1 and plain_int_1 > plain_int_0 and plain_int_1 < plain_int_0 from alice.tbl_1;;", @@ -431,7 +443,8 @@ "rewritten_sql": "select tbl_1.plain_int_0+tbl_1.plain_int_0+5 as expr_121,tbl_1.plain_int_1+tbl_1.plain_int_2+tbl_1.plain_int_0>tbl_1.plain_int_1 and tbl_1.plain_int_1>tbl_1.plain_int_0 and tbl_1.plain_int_1tbl_1.plain_int_1) and (tbl_1.plain_int_1>tbl_1.plain_int_0) and (tbl_1.plain_int_1tbl_1.plain_int_1) and (tbl_1.plain_int_1>tbl_1.plain_int_0) and (tbl_1.plain_int_1tbl_1.plain_int_1) and (tbl_1.plain_int_1>tbl_1.plain_int_0) and (tbl_1.plain_int_1tbl_1.plain_int_1) and (tbl_1.plain_int_1>tbl_1.plain_int_0) and (tbl_1.plain_int_1tbl_1.plain_int_1) and (tbl_1.plain_int_1>tbl_1.plain_int_0) and (tbl_1.plain_int_1 if(plain_int_0, plain_int_1, plain_int_2) as res from alice.tbl_1;;", @@ -445,7 +458,8 @@ "rewritten_sql": "select tbl_1.plain_int_0,tbl_1.plain_int_0>1 and tbl_1.plain_int_0>=1 as and_log,not(tbl_1.plain_int_0<5) as not_log,tbl_1.plain_int_0<=5 or tbl_1.plain_int_0!=8 as or_log from carol.tbl_1", "rewritten_sql_pg": "select tbl_1.plain_int_0,(tbl_1.plain_int_0>1) and (tbl_1.plain_int_0>=1) as and_log,not(tbl_1.plain_int_0<5) as not_log,(tbl_1.plain_int_0<=5) or (tbl_1.plain_int_0!=8) as or_log from carol.tbl_1", "rewritten_sql_csv": "select tbl_1.plain_int_0,(tbl_1.plain_int_0>1) and (tbl_1.plain_int_0>=1) as and_log,not(tbl_1.plain_int_0<5) as not_log,(tbl_1.plain_int_0<=5) or (tbl_1.plain_int_0!=8) as or_log from carol.tbl_1", - "rewritten_sql_odps": "select tbl_1.plain_int_0,(tbl_1.plain_int_0>1) and (tbl_1.plain_int_0>=1) as and_log,not(tbl_1.plain_int_0<5) as not_log,(tbl_1.plain_int_0<=5) or (tbl_1.plain_int_0!=8) as or_log from carol.tbl_1" + "rewritten_sql_odps": "select tbl_1.plain_int_0,(tbl_1.plain_int_0>1) and (tbl_1.plain_int_0>=1) as and_log,not(tbl_1.plain_int_0<5) as not_log,(tbl_1.plain_int_0<=5) or (tbl_1.plain_int_0!=8) as or_log from carol.tbl_1", + "rewritten_sql_hive": "select tbl_1.plain_int_0,(tbl_1.plain_int_0>1) and (tbl_1.plain_int_0>=1) as and_log,not(tbl_1.plain_int_0<5) as not_log,(tbl_1.plain_int_0<=5) or (tbl_1.plain_int_0!=8) as or_log from carol.tbl_1" }, { "sql": "select -(plain_int_0 + plain_int_2) * plain_int_1 as res from carol.tbl_1;", @@ -463,7 +477,8 @@ "rewritten_sql": "select truncate(tbl_1.plain_float_0, 2) as a,truncate(tbl_1.plain_float_0, 0) as b,truncate(3.1415926, 3) as c from alice.tbl_1", "rewritten_sql_pg": "select trunc(tbl_1.plain_float_0, 2) as a,trunc(tbl_1.plain_float_0, 0) as b,trunc(3.1415926, 3) as c from alice.tbl_1", "rewritten_sql_csv": "select trunc(tbl_1.plain_float_0, 2) as a,trunc(tbl_1.plain_float_0, 0) as b,trunc(3.1415926, 3) as c from alice.tbl_1", - "rewritten_sql_odps": "select trunc(tbl_1.plain_float_0, 2) as a,trunc(tbl_1.plain_float_0, 0) as b,trunc(3.1415926, 3) as c from alice.tbl_1" + "rewritten_sql_odps": "select trunc(tbl_1.plain_float_0, 2) as a,trunc(tbl_1.plain_float_0, 0) as b,trunc(3.1415926, 3) as c from alice.tbl_1", + "rewritten_sql_hive": "select trunc(tbl_1.plain_float_0, 2) as a,trunc(tbl_1.plain_float_0, 0) as b,trunc(3.1415926, 3) as c from alice.tbl_1" }, { "sql": "select abs(aggregate_float_0) as a from alice.tbl_1;", @@ -508,14 +523,16 @@ { "sql": "select plain_datetime_0 < now() as a from alice.tbl_1;", "skip_projection": false, - "rewritten_sql": "select tbl_1.plain_datetime_0=4", "rewritten_sql_pg": "select tbl_1.plain_string_0 as b4 from alice.tbl_1 where tbl_1.plain_string_0='123' group by tbl_1.plain_string_0 having count(1)>=4", "rewritten_sql_csv": "select tbl_1.plain_string_0 as b4 from alice.tbl_1 where tbl_1.plain_string_0='123' group by tbl_1.plain_string_0 having count(1)>=4", - "rewritten_sql_odps": "select tbl_1.plain_string_0 as b4 from alice.tbl_1 where tbl_1.plain_string_0='123' group by tbl_1.plain_string_0 having count(1)>=4" + "rewritten_sql_odps": "select tbl_1.plain_string_0 as b4 from alice.tbl_1 where tbl_1.plain_string_0='123' group by tbl_1.plain_string_0 having count(1)>=4", + "rewritten_sql_hive": "select tbl_1.plain_string_0 as b4 from alice.tbl_1 where tbl_1.plain_string_0='123' group by tbl_1.plain_string_0 having count(1)>=4" }, { "sql": "SELECT CASE WHEN (t_0.b4='1' OR t_0.b4='1' OR t_0.b4='2') THEN '3' ELSE '4' END AS bank_type, count(1) AS expr_24 FROM (SELECT str_to_date(plain_string_0, '%Y-%m-%d') AS p0, plain_string_1 as b4, plain_string_2 as b5 FROM alice.tbl_1 WHERE (str_to_date(plain_string_0, '%Y-%m-%d')=str_to_date('2023-01-10', '%Y-%m-%d'))) AS t_0 GROUP BY t_0.b4 HAVING count(1)>=4;;", @@ -579,6 +597,7 @@ "skip_odps_test": true, "skip_csv_test": true, "skip_pg_test": true, + "skip_hive_test": true, "rewritten_sql": "select case when (any_value(tbl_1.plain_string_1)='1' or any_value(tbl_1.plain_string_1)='1' or any_value(tbl_1.plain_string_1)='2') then '3' else '4' end as bank_type,count(1) as expr_24 from alice.tbl_1 where str_to_date(tbl_1.plain_string_0, '%Y-%m-%d')=str_to_date('2023-01-10', '%Y-%m-%d') group by tbl_1.plain_string_1 having count(1)>=4" }, { @@ -592,6 +611,7 @@ "skip_odps_test": true, "skip_pg_test": true, "skip_csv_test": true, + "skip_hive_test": true, "rewritten_sql": "select any_value(t_0.bank_type) as bank_type,count(1) as expr_123 from (select case when (tbl_1.plain_string_1='1' or tbl_1.plain_string_1='2' or tbl_1.plain_string_1='3') then '1' else '2' end as bank_type from alice.tbl_1 where str_to_date(tbl_1.plain_string_0, '%Y-%m-%d')='2022-02-02') as t_0 group by t_0.bank_type" }, { @@ -600,7 +620,8 @@ "rewritten_sql": "select a.plain_int_0,b.expr_217 from alice.tbl_0 as a left join (select any_value(tbl_1.plain_int_0) as expr_217 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.expr_217", "rewritten_sql_pg": "select a.plain_int_0,b.plain_int_0 from alice.tbl_0 as a left join (select tbl_1.plain_int_0 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.plain_int_0", "rewritten_sql_csv": "select a.plain_int_0,b.plain_int_0 from alice.tbl_0 as a left join (select tbl_1.plain_int_0 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.plain_int_0", - "rewritten_sql_odps": "select a.plain_int_0,b.plain_int_0 from alice.tbl_0 as a left join (select tbl_1.plain_int_0 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.plain_int_0" + "rewritten_sql_odps": "select a.plain_int_0,b.plain_int_0 from alice.tbl_0 as a left join (select tbl_1.plain_int_0 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.plain_int_0", + "rewritten_sql_hive": "select a.plain_int_0,b.plain_int_0 from alice.tbl_0 as a left join (select tbl_1.plain_int_0 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.plain_int_0" }, { "sql": "select a.plain_int_0 from alice.tbl_0 a left join (select distinct plain_int_0, plain_int_1 from alice.tbl_1) b on a.plain_int_0 = b.plain_int_0;", @@ -608,7 +629,8 @@ "rewritten_sql": "select a.plain_int_0 from alice.tbl_0 as a left join (select any_value(tbl_1.plain_int_0) as expr_217 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.expr_217", "rewritten_sql_pg": "select a.plain_int_0 from alice.tbl_0 as a left join (select tbl_1.plain_int_0 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.plain_int_0", "rewritten_sql_csv": "select a.plain_int_0 from alice.tbl_0 as a left join (select tbl_1.plain_int_0 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.plain_int_0", - "rewritten_sql_odps": "select a.plain_int_0 from alice.tbl_0 as a left join (select tbl_1.plain_int_0 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.plain_int_0" + "rewritten_sql_odps": "select a.plain_int_0 from alice.tbl_0 as a left join (select tbl_1.plain_int_0 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.plain_int_0", + "rewritten_sql_hive": "select a.plain_int_0 from alice.tbl_0 as a left join (select tbl_1.plain_int_0 from alice.tbl_1 group by tbl_1.plain_int_0,tbl_1.plain_int_1) as b on a.plain_int_0=b.plain_int_0" }, { "sql": "select plain_int_0, 'a' from alice.tbl_0 UNION ALL select plain_int_0, 'a' from alice.tbl_0;", @@ -626,7 +648,8 @@ { "sql": "select replace(plain_string_0, 'a', 'b') as res1 from alice.tbl_0;", "skip_projection": false, - "rewritten_sql": "select replace(tbl_0.plain_string_0, 'a', 'b') as res1 from alice.tbl_0" + "rewritten_sql": "select replace(tbl_0.plain_string_0, 'a', 'b') as res1 from alice.tbl_0", + "rewritten_sql_hive": "select regexp_replace(tbl_0.plain_string_0, 'a', 'b') as res1 from alice.tbl_0" }, { "sql": "SELECT a.SVC_NUM AS SVC_NUM, a.plain_string_2, IF((d.plain_int_1 <= 30) OR d.plain_int_1 IS NULL, '1', IF(d.plain_int_1 <= 100, '2', '3')) AS plain_int_1, IF(a.compare_string_0 = '1', '是', '否') AS compare_string_0, IF((a.compare_int_1 <= 21) OR a.compare_int_1 IS NULL, '1', '2') AS compare_int_1 FROM (SELECT a.plain_int_2 AS SVC_NUM, a.plain_string_2, c.compare_string_0, c.compare_int_1 FROM alice.tbl_2 AS a LEFT JOIN (SELECT plain_int_0, compare_string_0, compare_int_1 FROM alice.tbl_0 WHERE plain_string_1 = '202505') AS c ON a.plain_int_2 = c.plain_int_0) AS a LEFT JOIN (SELECT plain_int_0, plain_int_1 FROM alice.tbl_1 WHERE plain_string_1 = '202505') AS d ON a.SVC_NUM = d.plain_int_0;", @@ -634,7 +657,8 @@ "rewritten_sql": "select a.plain_int_2 as SVC_NUM,a.plain_string_2,if(d.plain_int_1<=30 or d.plain_int_1 is null, '1', if(d.plain_int_1<=100, '2', '3')) as plain_int_1,if(a.compare_string_0='1', '是', '否') as compare_string_0,if(a.compare_int_1<=21 or a.compare_int_1 is null, '1', '2') as compare_int_1 from (select a.plain_int_2,a.plain_string_2,c.compare_int_1,c.compare_string_0 from (alice.tbl_2 as a left join (select tbl_0.compare_int_1,tbl_0.compare_string_0,tbl_0.plain_int_0 from alice.tbl_0 where tbl_0.plain_string_1='202505') as c on a.plain_int_2=c.plain_int_0)) as a left join (select tbl_1.plain_int_0,tbl_1.plain_int_1 from alice.tbl_1 where tbl_1.plain_string_1='202505') as d on a.plain_int_2=d.plain_int_0", "rewritten_sql_pg": "select a.plain_int_2 as SVC_NUM,a.plain_string_2,if((d.plain_int_1<=30) or d.plain_int_1 is null, '1', if(d.plain_int_1<=100, '2', '3')) as plain_int_1,if(a.compare_string_0='1', '是', '否') as compare_string_0,if((a.compare_int_1<=21) or a.compare_int_1 is null, '1', '2') as compare_int_1 from (select a.plain_int_2,a.plain_string_2,c.compare_int_1,c.compare_string_0 from (alice.tbl_2 as a left join (select tbl_0.compare_int_1,tbl_0.compare_string_0,tbl_0.plain_int_0 from alice.tbl_0 where tbl_0.plain_string_1='202505') as c on a.plain_int_2=c.plain_int_0)) as a left join (select tbl_1.plain_int_0,tbl_1.plain_int_1 from alice.tbl_1 where tbl_1.plain_string_1='202505') as d on a.plain_int_2=d.plain_int_0", "rewritten_sql_csv": "select a.plain_int_2 as SVC_NUM,a.plain_string_2,if((d.plain_int_1<=30) or d.plain_int_1 is null, '1', if(d.plain_int_1<=100, '2', '3')) as plain_int_1,if(a.compare_string_0='1', '是', '否') as compare_string_0,if((a.compare_int_1<=21) or a.compare_int_1 is null, '1', '2') as compare_int_1 from (select a.plain_int_2,a.plain_string_2,c.compare_int_1,c.compare_string_0 from (alice.tbl_2 as a left join (select tbl_0.compare_int_1,tbl_0.compare_string_0,tbl_0.plain_int_0 from alice.tbl_0 where tbl_0.plain_string_1='202505') as c on a.plain_int_2=c.plain_int_0)) as a left join (select tbl_1.plain_int_0,tbl_1.plain_int_1 from alice.tbl_1 where tbl_1.plain_string_1='202505') as d on a.plain_int_2=d.plain_int_0", - "rewritten_sql_odps": "select a.plain_int_2 as SVC_NUM,a.plain_string_2,if((d.plain_int_1<=30) or d.plain_int_1 is null, '1', if(d.plain_int_1<=100, '2', '3')) as plain_int_1,if(a.compare_string_0='1', '是', '否') as compare_string_0,if((a.compare_int_1<=21) or a.compare_int_1 is null, '1', '2') as compare_int_1 from (select a.plain_int_2,a.plain_string_2,c.compare_int_1,c.compare_string_0 from (alice.tbl_2 as a left join (select tbl_0.compare_int_1,tbl_0.compare_string_0,tbl_0.plain_int_0 from alice.tbl_0 where tbl_0.plain_string_1='202505') as c on a.plain_int_2=c.plain_int_0)) as a left join (select tbl_1.plain_int_0,tbl_1.plain_int_1 from alice.tbl_1 where tbl_1.plain_string_1='202505') as d on a.plain_int_2=d.plain_int_0" + "rewritten_sql_odps": "select a.plain_int_2 as SVC_NUM,a.plain_string_2,if((d.plain_int_1<=30) or d.plain_int_1 is null, '1', if(d.plain_int_1<=100, '2', '3')) as plain_int_1,if(a.compare_string_0='1', '是', '否') as compare_string_0,if((a.compare_int_1<=21) or a.compare_int_1 is null, '1', '2') as compare_int_1 from (select a.plain_int_2,a.plain_string_2,c.compare_int_1,c.compare_string_0 from (alice.tbl_2 as a left join (select tbl_0.compare_int_1,tbl_0.compare_string_0,tbl_0.plain_int_0 from alice.tbl_0 where tbl_0.plain_string_1='202505') as c on a.plain_int_2=c.plain_int_0)) as a left join (select tbl_1.plain_int_0,tbl_1.plain_int_1 from alice.tbl_1 where tbl_1.plain_string_1='202505') as d on a.plain_int_2=d.plain_int_0", + "rewritten_sql_hive": "select a.plain_int_2 as SVC_NUM,a.plain_string_2,if((d.plain_int_1<=30) or d.plain_int_1 is null, '1', if(d.plain_int_1<=100, '2', '3')) as plain_int_1,if(a.compare_string_0='1', '是', '否') as compare_string_0,if((a.compare_int_1<=21) or a.compare_int_1 is null, '1', '2') as compare_int_1 from (select a.plain_int_2,a.plain_string_2,c.compare_int_1,c.compare_string_0 from (alice.tbl_2 as a left join (select tbl_0.compare_int_1,tbl_0.compare_string_0,tbl_0.plain_int_0 from alice.tbl_0 where tbl_0.plain_string_1='202505') as c on a.plain_int_2=c.plain_int_0)) as a left join (select tbl_1.plain_int_0,tbl_1.plain_int_1 from alice.tbl_1 where tbl_1.plain_string_1='202505') as d on a.plain_int_2=d.plain_int_0" } ] }