You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jo...@apache.org on 2024/11/19 17:30:20 UTC
(impala) branch master updated (2085edbe1 -> 9a2496dd2)
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 2085edbe1 IMPALA-13503: CustomClusterTestSuite for whole class
new f53322591 IMPALA-13543: single_node_perf_run.py must accept tpcds_partitioned
new 3574fe89b IMPALA-13513: Support decode function
new 9a2496dd2 IMPALA-13511: Calcite planner: support sub-second datetime parts
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
bin/single_node_perf_run.py | 28 ++-
java/calcite-planner/src/main/codegen/config.fmpp | 1 +
.../src/main/codegen/templates/Parser.jj | 150 ++++++++++-----
.../calcite/coercenodes/CoerceOperandShuttle.java | 76 +++++++-
.../impala/calcite/functions/FunctionResolver.java | 27 ++-
.../impala/calcite/functions/RexCallConverter.java | 14 ++
.../operators/ImpalaCustomOperatorTable.java | 2 +
.../calcite/operators/ImpalaDecodeFunction.java | 127 +++++++++++++
.../calcite/operators/ImpalaOperatorTable.java | 3 +
.../calcite/type/ImpalaSqlIntervalQualifier.java | 159 ++++++++++++++++
testdata/bin/generate-schema-statements.py | 208 +++++++++++++--------
.../tpcds_partitioned_schema_template.sql | 48 ++---
tests/util/test_file_parser.py | 11 +-
13 files changed, 690 insertions(+), 164 deletions(-)
create mode 100644 java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaDecodeFunction.java
create mode 100644 java/calcite-planner/src/main/java/org/apache/impala/calcite/type/ImpalaSqlIntervalQualifier.java
(impala) 01/03: IMPALA-13543: single_node_perf_run.py must accept tpcds_partitioned
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit f533225915d78e1a2a3d1e6606cf4f6d064151d2
Author: Riza Suminto <ri...@cloudera.com>
AuthorDate: Sat Nov 9 15:44:24 2024 -0800
IMPALA-13543: single_node_perf_run.py must accept tpcds_partitioned
tpcds_partitioned dataset is a fully-partitioned version of tpcds
dataset (the latter only partition store_sales table). It does not have
the default text format database like tpcds dataset. Instead, it relies
on pre-existence of text format tpcds database, which then INSERT
OVERWRITE INTO tpcds_partitioned database equivalent. It does not have
its own queries set, but instead symlinked to share
testdata/workloads/tpcds/queries. It also have slightly different schema
from tpcds dataset, namely column "c_last_review_date" in tpcds dataset
is "c_last_review_date_sk" in tpcds_partitioned (TPC-DS v2.11.0, section
2.4.7). These reasons make tpcds_partitioned ineligible for
perf-AB-test (single_node_perf_run.py).
This patch update single_node_perf_run.py and related scripts to make
tpcds_partitioned eligible for benchmark dataset. It adds an initial
steps to load the text database from tpcds dataset with selected scale
before running the load script for tpcds_partitioned dataset. Compute
stats step also limited to run one at a time to not overadmit the
cluster with concurrent compute stats queries.
Created helper function build_replacement_params() inside
generate-schema-statements.py for common function.
Testing
- Run perf-AB-test-ub2004 with this commit included and confirm
benchmark works with tpcds_partitioned dataset.
- Run normal data loading. Pass FE tests, and
query_test/test_tpcds_queries.py.
Change-Id: I4b6f435705dcf873696ffd151052ebeab35d9898
Reviewed-on: http://gerrit.cloudera.org:8080/22061
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
bin/single_node_perf_run.py | 28 ++-
testdata/bin/generate-schema-statements.py | 208 +++++++++++++--------
.../tpcds_partitioned_schema_template.sql | 48 ++---
tests/util/test_file_parser.py | 11 +-
4 files changed, 184 insertions(+), 111 deletions(-)
diff --git a/bin/single_node_perf_run.py b/bin/single_node_perf_run.py
index 4255ac7b4..4f77977df 100755
--- a/bin/single_node_perf_run.py
+++ b/bin/single_node_perf_run.py
@@ -88,6 +88,7 @@ import textwrap
from tests.common.test_dimensions import TableFormatInfo
IMPALA_HOME = os.environ["IMPALA_HOME"]
+IMPALA_PERF_RESULTS = os.path.join(IMPALA_HOME, "perf_results")
def configured_call(cmd):
@@ -100,14 +101,17 @@ def configured_call(cmd):
def load_data(db_to_load, table_formats, scale):
"""Loads a database with a particular scale factor."""
+ all_formats = ("text/none," + table_formats if "text/none" not in table_formats
+ else table_formats)
configured_call(["{0}/bin/load-data.py".format(IMPALA_HOME),
"--workloads", db_to_load, "--scale_factor", str(scale),
- "--table_formats", "text/none," + table_formats])
+ "--table_formats", all_formats])
for table_format in table_formats.split(","):
suffix = TableFormatInfo.create_from_string(None, table_format).db_suffix()
db_name = db_to_load + scale + suffix
configured_call(["{0}/tests/util/compute_table_stats.py".format(IMPALA_HOME),
- "--stop_on_error", "--db_names", db_name])
+ "--stop_on_error", "--db_names", db_name,
+ "--parallelism", "1"])
def get_git_hash_for_name(name):
@@ -164,7 +168,7 @@ def run_workload(base_dir, workloads, options):
def report_benchmark_results(file_a, file_b, description):
"""Wrapper around report_benchmark_result.py."""
- result = "{0}/perf_results/latest/performance_result.txt".format(IMPALA_HOME)
+ result = os.path.join(IMPALA_PERF_RESULTS, "latest", "performance_result.txt")
with open(result, "w") as f:
subprocess.check_call(
["{0}/tests/benchmark/report_benchmark_results.py".format(IMPALA_HOME),
@@ -266,7 +270,7 @@ def perf_ab_test(options, args):
hash_a = get_git_hash_for_name(args[0])
# Create the base directory to store the results in
- results_path = os.path.join(IMPALA_HOME, "perf_results")
+ results_path = IMPALA_PERF_RESULTS
if not os.access(results_path, os.W_OK):
os.makedirs(results_path)
@@ -284,12 +288,20 @@ def perf_ab_test(options, args):
start_minicluster()
start_impala(options.num_impalads, options)
- workloads = set(options.workloads.split(","))
+ workloads = options.workloads.split(",")
if options.load:
- WORKLOAD_TO_DATASET = {"tpch": "tpch", "tpcds": "tpcds", "targeted-perf": "tpch",
- "tpcds-unmodified": "tpcds-unmodified"}
- datasets = set([WORKLOAD_TO_DATASET[workload] for workload in workloads])
+ WORKLOAD_TO_DATASET = {
+ "tpch": "tpch",
+ "tpcds": "tpcds",
+ "targeted-perf": "tpch",
+ "tpcds-unmodified": "tpcds-unmodified",
+ "tpcds_partitioned": "tpcds_partitioned"
+ }
+ datasets = [WORKLOAD_TO_DATASET[workload] for workload in workloads]
+ if "tpcds_partitioned" in datasets and "tpcds" not in datasets:
+ # "tpcds_partitioned" require the text "tpcds" database.
+ load_data("tpcds", "text/none", options.scale)
for dataset in datasets:
load_data(dataset, options.table_formats, options.scale)
diff --git a/testdata/bin/generate-schema-statements.py b/testdata/bin/generate-schema-statements.py
index 2361fd2d6..195094c49 100755
--- a/testdata/bin/generate-schema-statements.py
+++ b/testdata/bin/generate-schema-statements.py
@@ -112,7 +112,7 @@ from tests.common.test_dimensions import (
parser = OptionParser()
parser.add_option("-e", "--exploration_strategy", dest="exploration_strategy",
- default="core", help="The exploration strategy for schema gen: 'core',"\
+ default="core", help="The exploration strategy for schema gen: 'core',"
" 'pairwise', or 'exhaustive'")
parser.add_option("--hive_warehouse_dir", dest="hive_warehouse_dir",
default="/test-warehouse",
@@ -122,16 +122,16 @@ parser.add_option("-w", "--workload", dest="workload",
parser.add_option("-s", "--scale_factor", dest="scale_factor", default="",
help="An optional scale factor to generate the schema for")
parser.add_option("-f", "--force_reload", dest="force_reload", action="store_true",
- default= False, help='Skips HDFS exists check and reloads all tables')
+ default=False, help='Skips HDFS exists check and reloads all tables')
parser.add_option("-v", "--verbose", dest="verbose", action="store_true",
- default = False, help="If set, outputs additional logging.")
+ default=False, help="If set, outputs additional logging.")
parser.add_option("-b", "--backend", dest="backend", default="localhost:21000",
help="Backend connection to use, default: localhost:21000")
parser.add_option("--table_names", dest="table_names", default=None,
- help="Only load the specified tables - specified as a comma-seperated "\
+ help="Only load the specified tables - specified as a comma-seperated "
"list of base table names")
parser.add_option("--table_formats", dest="table_formats", default=None,
- help="Override the test vectors and load using the specified table "\
+ help="Override the test vectors and load using the specified table "
"formats. Ex. --table_formats=seq/snap/block,text/none")
parser.add_option("--hdfs_namenode", dest="hdfs_namenode", default="localhost:20500",
help="HDFS name node for Avro schema URLs, default localhost:20500")
@@ -142,11 +142,12 @@ if options.workload is None:
parser.print_help()
sys.exit(1)
-WORKLOAD_DIR = os.path.join(os.environ['IMPALA_HOME'], 'testdata', 'workloads')
-DATASET_DIR = os.path.join(os.environ['IMPALA_HOME'], 'testdata', 'datasets')
+IMPALA_HOME = os.getenv("IMPALA_HOME")
+WORKLOAD_DIR = os.path.join(IMPALA_HOME, 'testdata', 'workloads')
+DATASET_DIR = os.path.join(IMPALA_HOME, 'testdata', 'datasets')
SQL_OUTPUT_DIR = os.environ['IMPALA_DATA_LOADING_SQL_DIR']
AVRO_SCHEMA_DIR = "avro_schemas"
-DEFAULT_FS=os.environ['DEFAULT_FS']
+DEFAULT_FS = os.environ['DEFAULT_FS']
IMPALA_SUPPORTED_INSERT_FORMATS = ['parquet', 'hbase', 'text', 'kudu']
IMPALA_PARQUET_COMPRESSION_MAP = \
@@ -178,18 +179,19 @@ DATASET_IDX = 1
CODEC_IDX = 2
COMPRESSION_TYPE_IDX = 3
-COMPRESSION_MAP = {'def': 'org.apache.hadoop.io.compress.DefaultCodec',
- 'gzip': 'org.apache.hadoop.io.compress.GzipCodec',
- 'bzip': 'org.apache.hadoop.io.compress.BZip2Codec',
- 'snap': 'org.apache.hadoop.io.compress.SnappyCodec',
- 'none': ''
- }
+COMPRESSION_MAP = {
+ 'def': 'org.apache.hadoop.io.compress.DefaultCodec',
+ 'gzip': 'org.apache.hadoop.io.compress.GzipCodec',
+ 'bzip': 'org.apache.hadoop.io.compress.BZip2Codec',
+ 'snap': 'org.apache.hadoop.io.compress.SnappyCodec',
+ 'none': ''
+}
AVRO_COMPRESSION_MAP = {
'def': 'deflate',
'snap': 'snappy',
'none': '',
- }
+}
HIVE_TO_AVRO_TYPE_MAP = {
'STRING': 'string',
@@ -206,7 +208,7 @@ HIVE_TO_AVRO_TYPE_MAP = {
# in our tests.
'TIMESTAMP': 'string',
'BINARY': 'bytes',
- }
+}
PARQUET_ALTER_STATEMENT = "ALTER TABLE %(table_name)s SET\n\
SERDEPROPERTIES ('blocksize' = '1073741824', 'compression' = '%(compression)s');"
@@ -230,9 +232,47 @@ PARTITIONED_INSERT_RE = re.compile(
HINT_SHUFFLE = "/* +shuffle, clustered */"
+# Variable names in the sql statement will be replaced by the actual values.
+# Use build_replacement_params() to create such variable:value map.
+VAR_DB_NAME = "db_name"
+VAR_DB_SUFFIX = "db_suffix"
+VAR_TABLE_NAME = "table_name"
+VAR_HDFS_LOCATION = "hdfs_location"
+VAR_IMPALA_HOME = "impala_home"
+VAR_HINT = "hint"
+VAR_PART_PREDICATE = "part_predicate"
+VAR_FILE_FORMAT = "file_format"
+VAR_SCALE_FACTOR = "scale_factor"
+
+
+def build_replacement_params(
+ db_name, db_suffix, table_name, hdfs_location=None, impala_home=None,
+ hint=None, part_predicate=None, file_format=None, scale_factor=None):
+ """Create a variable:value map for SQL replacement based on given parameters.
+ db_name, db_suffix, and table_name is mandatory. Others are optional."""
+ params = {
+ VAR_DB_NAME: db_name,
+ VAR_DB_SUFFIX: db_suffix,
+ VAR_TABLE_NAME: table_name
+ }
+ if hdfs_location is not None:
+ params[VAR_HDFS_LOCATION] = hdfs_location
+ if impala_home is not None:
+ params[VAR_IMPALA_HOME] = impala_home
+ if hint is not None:
+ params[VAR_HINT] = hint
+ if part_predicate is not None:
+ params[VAR_PART_PREDICATE] = part_predicate
+ if file_format is not None:
+ params[VAR_FILE_FORMAT] = file_format
+ if scale_factor is not None:
+ params[VAR_SCALE_FACTOR] = scale_factor
+ return params
+
+
def build_create_statement(table_template, table_name, db_name, db_suffix,
- file_format, compression, hdfs_location,
- force_reload, is_hive_stmt):
+ file_format, hdfs_location, force_reload, is_hive_stmt,
+ scale_factor=""):
create_stmt = ''
if (force_reload):
tbl_type = 'TABLE'
@@ -247,12 +287,14 @@ def build_create_statement(table_template, table_name, db_name, db_suffix,
# Remove location part from the format string
table_template = table_template.replace("LOCATION '{hdfs_location}'", "")
- stmt = table_template.format(
- db_name=db_name,
- db_suffix=db_suffix,
- table_name=table_name,
+ params = build_replacement_params(
+ db_name,
+ db_suffix,
+ table_name,
file_format=FILE_FORMAT_TO_STORED_AS_MAP[file_format],
- hdfs_location=hdfs_location)
+ hdfs_location=hdfs_location,
+ scale_factor=scale_factor)
+ stmt = table_template.format(**params)
# Apache Hive 3.1 doesn't support "STORED BY ICEBERG STORED AS AVRO" and
# "STORED AS JSONFILE" (HIVE-25162, HIVE-19899)
if is_hive_stmt and os.environ['USE_APACHE_HIVE'] == "true":
@@ -299,9 +341,9 @@ def parse_table_properties(file_format, table_properties):
tblproperties = {}
TABLE_PROPERTY_RE = re.compile(
# Optional '<data-format>:' prefix, capturing just the 'data-format' part.
- r'(?:(\w+):)?' +
+ r'(?:(\w+):)?'
# Required key=value, capturing the key and value
- r'(.+?)=(.*)')
+ + r'(.+?)=(.*)')
for table_property in [_f for _f in table_properties.split("\n") if _f]:
m = TABLE_PROPERTY_RE.match(table_property)
if not m:
@@ -408,13 +450,14 @@ LOCATION '{{hdfs_location}}'
partitioned_by=partitioned_by,
tblproperties=tblproperties_clause,
file_format_string=file_format_string
- ).strip()
+ ).strip()
# Remove empty lines from the stmt string. There is an empty line for
# each of the sections that didn't have anything (e.g. partitioned_by)
stmt = os.linesep.join([s for s in stmt.splitlines() if s])
stmt += ';'
return stmt
+
def build_hbase_create_stmt_in_hive(columns, partition_columns, table_name):
# The hbase create statement differs sufficiently from the generic create to justify a
# separate method. Specifically, STORED AS becomes STORED BY. There is section called
@@ -439,15 +482,16 @@ def build_hbase_create_stmt_in_hive(columns, partition_columns, table_name):
columns=',\n'.join(columns),
hbase_column_mapping=hbase_column_mapping,
tbl_properties=tbl_properties,
- ).strip()
+ ).strip()
return stmt + ';'
+
def avro_schema(columns):
record = {
- "name": "a", # doesn't matter
+ "name": "a", # doesn't matter
"type": "record",
"fields": list()
- }
+ }
for column_spec in columns.strip().split('\n'):
# column_spec looks something like "col_name col_type COMMENT comment"
# (comment may be omitted, we don't use it)
@@ -460,7 +504,7 @@ def avro_schema(columns):
precision = 9
else:
# Parse out scale and precision from decimal type
- m = re.search("DECIMAL\((?P<precision>.*),(?P<scale>.*)\)", column_spec.upper())
+ m = re.search(r"DECIMAL\((?P<precision>.*),(?P<scale>.*)\)", column_spec.upper())
assert m, "Could not parse decimal column spec: " + column_spec
scale = int(m.group('scale'))
precision = int(m.group('precision'))
@@ -477,10 +521,11 @@ def avro_schema(columns):
record["fields"].append(
{'name': name,
- 'type': [type, "null"]}) # all columns nullable
+ 'type': [type, "null"]}) # all columns nullable
return json.dumps(record)
+
def build_compression_codec_statement(codec, compression_type, file_format):
codec = (AVRO_COMPRESSION_MAP if file_format == 'avro' else COMPRESSION_MAP).get(codec)
if not codec:
@@ -488,6 +533,7 @@ def build_compression_codec_statement(codec, compression_type, file_format):
return (AVRO_COMPRESSION_CODEC % codec) if file_format == 'avro' else (
COMPRESSION_TYPE % compression_type.upper() + '\n' + COMPRESSION_CODEC % codec)
+
def build_codec_enabled_statement(codec):
compression_enabled = 'false' if codec == 'none' else 'true'
return COMPRESSION_ENABLED % compression_enabled
@@ -505,25 +551,25 @@ def build_partitioned_load(insert, re_match, can_hint, params):
batch = int(re_match.group(4))
insert = PARTITIONED_INSERT_RE.sub("", insert)
statements = []
- params["hint"] = HINT_SHUFFLE if can_hint else ''
+ params[VAR_HINT] = HINT_SHUFFLE if can_hint else ''
first_part = min_val
while first_part < max_val:
if first_part + batch >= max_val:
# This is the last batch
if first_part == min_val:
# There is only 1 batch in this insert. No predicate needed.
- params["part_predicate"] = ''
+ params[VAR_PART_PREDICATE] = ''
else:
# Insert the remaining partitions + NULL partition
- params["part_predicate"] = "WHERE {0} <= {1} OR {2} IS NULL".format(
+ params[VAR_PART_PREDICATE] = "WHERE {0} <= {1} OR {2} IS NULL".format(
first_part, part_col, part_col)
elif first_part == min_val:
# This is the first batch.
- params["part_predicate"] = "WHERE {0} < {1}".format(
+ params[VAR_PART_PREDICATE] = "WHERE {0} < {1}".format(
part_col, first_part + batch)
else:
# This is the middle batch.
- params["part_predicate"] = "WHERE {0} <= {1} AND {2} < {3}".format(
+ params[VAR_PART_PREDICATE] = "WHERE {0} <= {1} AND {2} < {3}".format(
first_part, part_col, part_col, first_part + batch)
statements.append(insert.format(**params))
first_part += batch
@@ -531,24 +577,24 @@ def build_partitioned_load(insert, re_match, can_hint, params):
def build_insert_into_statement(insert, db_name, db_suffix, table_name, file_format,
- hdfs_path, for_impala=False):
+ hdfs_path, for_impala=False, scale_factor=""):
can_hint = for_impala and (file_format == 'parquet' or is_iceberg_table(file_format))
- params = {
- "db_name": db_name,
- "db_suffix": db_suffix,
- "table_name": table_name,
- "hdfs_location": hdfs_path,
- "impala_home": os.getenv("IMPALA_HOME"),
- "hint": "",
- "part_predicate": ""
- }
+ params = build_replacement_params(
+ db_name,
+ db_suffix,
+ table_name,
+ hdfs_location=hdfs_path,
+ impala_home=IMPALA_HOME,
+ hint="",
+ part_predicate="",
+ scale_factor=scale_factor)
m = PARTITIONED_INSERT_RE.search(insert)
if m:
insert_statement = build_partitioned_load(insert, m, can_hint, params)
else:
if can_hint:
- params["hint"] = HINT_SHUFFLE
+ params[VAR_HINT] = HINT_SHUFFLE
insert_statement = insert.format(**params)
# Kudu tables are managed and don't support OVERWRITE, so we replace OVERWRITE
@@ -573,20 +619,22 @@ def build_insert_into_statement(insert, db_name, db_suffix, table_name, file_for
statement += SET_HIVE_INPUT_FORMAT % "HiveInputFormat"
return statement + insert_statement
+
def build_hbase_insert(db_name, db_suffix, table_name):
hbase_insert = SET_HIVE_HBASE_BULK_LOAD + ';\n'
# For Apache Hive, "hive.hbase.bulk does not exist" exception will be thrown and there
# is only warning in cdp
if os.environ['USE_APACHE_HIVE'] == "true":
hbase_insert = ""
+ params = build_replacement_params(db_name, db_suffix, table_name)
hbase_insert += ("INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}"
- " SELECT * FROM {db_name}.{table_name};\n").\
- format(db_name=db_name, db_suffix=db_suffix,table_name=table_name)
+ " SELECT * FROM {db_name}.{table_name};\n").format(**params)
return hbase_insert
+
def build_insert(insert, db_name, db_suffix, file_format,
codec, compression_type, table_name, hdfs_path, create_hive=False,
- for_impala=False):
+ for_impala=False, scale_factor=""):
# HBASE inserts don't need the hive options to be set, and don't require and HDFS
# file location, so they're handled separately.
if file_format == 'hbase' and not create_hive:
@@ -602,26 +650,25 @@ def build_insert(insert, db_name, db_suffix, file_format,
output += build_impala_parquet_codec_statement(codec) + "\n"
output += build_insert_into_statement(insert, db_name, db_suffix,
table_name, file_format, hdfs_path,
- for_impala) + "\n"
+ for_impala, scale_factor) + "\n"
return output
-def build_load_statement(load_template, db_name, db_suffix, table_name):
+def build_load_statement(load_template, db_name, db_suffix, table_name, scale_factor=""):
# hbase does not need the hdfs path.
if table_name.startswith('hbase'):
- load_template = load_template.format(table_name=table_name,
- db_name=db_name,
- db_suffix=db_suffix)
+ params = build_replacement_params(db_name, db_suffix, table_name)
+ load_template = load_template.format(**params)
else:
- base_load_dir = os.getenv("REMOTE_LOAD", os.getenv("IMPALA_HOME"))
- params = {
- "db_name": db_name,
- "db_suffix": db_suffix,
- "table_name": table_name,
- "impala_home": base_load_dir,
- "hint": "",
- "part_predicate": ""
- }
+ base_load_dir = os.getenv("REMOTE_LOAD", IMPALA_HOME)
+ params = build_replacement_params(
+ db_name,
+ db_suffix,
+ table_name,
+ impala_home=base_load_dir,
+ hint="",
+ part_predicate="",
+ scale_factor=scale_factor)
m = PARTITIONED_INSERT_RE.search(load_template)
if m:
@@ -646,11 +693,12 @@ def build_hbase_create_stmt(db_name, table_name, column_families, region_splits)
create_stmts.append(create_statement)
return create_stmts
+
# Does a hdfs directory listing and returns array with all the subdir names.
def get_hdfs_subdirs_with_data(path):
tmp_file = tempfile.TemporaryFile("w+")
cmd = "hadoop fs -du %s | grep -v '^0' | awk '{print $3}'" % path
- subprocess.call([cmd], shell = True, stderr = open('/dev/null'), stdout = tmp_file)
+ subprocess.call([cmd], shell=True, stderr=open('/dev/null'), stdout=tmp_file)
tmp_file.seek(0)
# Results look like:
@@ -658,6 +706,7 @@ def get_hdfs_subdirs_with_data(path):
# So to get subdirectory names just return everything after the last '/'
return [line[line.rfind('/') + 1:].strip() for line in tmp_file.readlines()]
+
class Statements(object):
"""Simple container object for storing SQL statements to be output to a
file. Useful for ordering the statements correctly."""
@@ -676,6 +725,7 @@ class Statements(object):
def __bool__(self):
return bool(self.create or self.load or self.load_base)
+
def eval_section(section_str):
"""section_str should be the contents of a section (i.e. a string). If section_str
starts with `, evaluates section_str as a shell command and returns the
@@ -690,6 +740,7 @@ def eval_section(section_str):
assert p.returncode == 0
return stdout.strip()
+
def generate_statements(output_name, test_vectors, sections,
schema_include_constraints, schema_exclude_constraints,
schema_only_constraints, convert_orc_to_full_acid):
@@ -801,10 +852,10 @@ def generate_statements(output_name, test_vectors, sections,
tblproperties = parse_table_properties(create_file_format, table_properties)
# ORC tables are full ACID by default.
- if (convert_orc_to_full_acid and
- HIVE_MAJOR_VERSION == 3 and
- create_file_format == 'orc' and
- 'transactional' not in tblproperties):
+ if (convert_orc_to_full_acid
+ and HIVE_MAJOR_VERSION == 3
+ and create_file_format == 'orc'
+ and 'transactional' not in tblproperties):
tblproperties['transactional'] = 'true'
if create_file_format == 'orc' and create_codec != 'def':
# The default value of 'orc.compress' is ZLIB which corresponds to 'def'.
@@ -861,15 +912,15 @@ def generate_statements(output_name, test_vectors, sections,
if file_format == 'avro':
if not os.path.exists(avro_schema_dir):
os.makedirs(avro_schema_dir)
- with open("%s/%s.json" % (avro_schema_dir, table_name),"w") as f:
+ with open("%s/%s.json" % (avro_schema_dir, table_name), "w") as f:
f.write(avro_schema(columns))
else:
table_template = None
if table_template:
output.create.append(build_create_statement(table_template, table_name, db_name,
- db_suffix, create_file_format, create_codec, data_path, force_reload,
- create_hive))
+ db_suffix, create_file_format, data_path, force_reload, create_hive,
+ options.scale_factor))
# HBASE create table
if file_format == 'hbase':
# If the HBASE_COLUMN_FAMILIES section does not exist, default to 'd'
@@ -907,18 +958,20 @@ def generate_statements(output_name, test_vectors, sections,
load_from_json_file = file_format == 'json' and table_name.endswith('_json')
if not db_suffix or load_from_json_file:
if load:
- hive_output.load_base.append(build_load_statement(load, db_name,
- db_suffix, table_name))
+ hive_output.load_base.append(
+ build_load_statement(load, db_name, db_suffix, table_name,
+ options.scale_factor))
else:
print('Empty base table load for %s. Skipping load generation' % table_name)
elif file_format in ['kudu', 'parquet', 'iceberg']:
if insert_hive:
hive_output.load.append(build_insert(insert_hive, db_name, db_suffix,
- file_format, codec, compression_type, table_name, data_path))
+ file_format, codec, compression_type, table_name, data_path,
+ for_impala=False, scale_factor=options.scale_factor))
elif insert:
impala_load.load.append(build_insert(insert, db_name, db_suffix,
file_format, codec, compression_type, table_name, data_path,
- for_impala=True))
+ for_impala=True, scale_factor=options.scale_factor))
else:
print('Empty parquet/kudu load for table %s. Skipping insert generation'
% table_name)
@@ -961,6 +1014,7 @@ def parse_schema_template_file(file_name):
'TABLE_PROPERTIES', 'HBASE_REGION_SPLITS', 'HIVE_MAJOR_VERSION']
return parse_test_file(file_name, VALID_SECTION_NAMES, skip_unknown_sections=False)
+
if __name__ == "__main__":
if options.table_formats is None:
if options.exploration_strategy not in KNOWN_EXPLORATION_STRATEGIES:
@@ -968,7 +1022,7 @@ if __name__ == "__main__":
print('Valid values:', ', '.join(KNOWN_EXPLORATION_STRATEGIES))
sys.exit(1)
- test_vectors = [vector.value for vector in\
+ test_vectors = [vector.value for vector in
load_table_info_dimension(options.workload, options.exploration_strategy)]
else:
table_formats = options.table_formats.split(',')
diff --git a/testdata/datasets/tpcds_partitioned/tpcds_partitioned_schema_template.sql b/testdata/datasets/tpcds_partitioned/tpcds_partitioned_schema_template.sql
index a7ceecda1..6c6b436ff 100644
--- a/testdata/datasets/tpcds_partitioned/tpcds_partitioned_schema_template.sql
+++ b/testdata/datasets/tpcds_partitioned/tpcds_partitioned_schema_template.sql
@@ -35,7 +35,7 @@ t_meal_time string
primary key (t_time_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -73,7 +73,7 @@ d_current_year string
primary key (d_date_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -86,7 +86,7 @@ r_reason_desc string
primary key (r_reason_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -109,7 +109,7 @@ ca_location_type string
primary key (ca_address_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -122,7 +122,7 @@ ib_upper_bound int
primary key (ib_income_band_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -138,7 +138,7 @@ primary key (hd_demo_sk) DISABLE NOVALIDATE RELY
foreign key (hd_income_band_sk) references {db_name}{db_suffix}.income_band (ib_income_band_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -157,7 +157,7 @@ cd_dep_college_count int
primary key (cd_demo_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -173,7 +173,7 @@ sm_contract string
primary key (sm_ship_mode_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -252,7 +252,7 @@ SELECT
i_container,
i_manager_id,
i_product_name
-FROM tpcds.{table_name};
+FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -276,7 +276,7 @@ w_gmt_offset decimal(5,2)
primary key (w_warehouse_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -319,7 +319,7 @@ foreign key (cc_closed_date_sk) references {db_name}{db_suffix}.date_dim (d_date
foreign key (cc_open_date_sk) references {db_name}{db_suffix}.date_dim (d_date_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -392,7 +392,7 @@ SELECT
c_login,
c_email_address,
CAST(c_last_review_date AS INT)
-FROM tpcds.{table_name};
+FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -465,7 +465,7 @@ SELECT
p_channel_details,
p_purpose,
p_discount_active
-FROM tpcds.{table_name};
+FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -505,7 +505,7 @@ primary key (s_store_sk) DISABLE NOVALIDATE RELY
foreign key (s_closed_date_sk) references {db_name}{db_suffix}.date_dim (d_date_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -526,7 +526,7 @@ foreign key (cp_start_date_sk) references {db_name}{db_suffix}.date_dim (d_date_
foreign key (cp_end_date_sk) references {db_name}{db_suffix}.date_dim (d_date_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -553,7 +553,7 @@ foreign key (wp_access_date_sk) references {db_name}{db_suffix}.date_dim (d_date
foreign key (wp_customer_sk) references {db_name}{db_suffix}.customer (c_customer_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -591,7 +591,7 @@ foreign key (web_open_date_sk) references {db_name}{db_suffix}.date_dim (d_date_
foreign key (web_close_date_sk) references {db_name}{db_suffix}.date_dim (d_date_sk) DISABLE NOVALIDATE RELY
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name}
-SELECT * FROM tpcds.{table_name};
+SELECT * FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -629,7 +629,7 @@ SELECT
inv_warehouse_sk,
inv_quantity_on_hand,
inv_date_sk
-FROM tpcds.{table_name};
+FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -762,7 +762,7 @@ SELECT
cs_net_paid_inc_ship_tax,
cs_net_profit,
cs_sold_date_sk
-FROM tpcds.{table_name};
+FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -874,7 +874,7 @@ SELECT
cr_store_credit,
cr_net_loss,
cr_returned_date_sk
-FROM tpcds.{table_name};
+FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -966,7 +966,7 @@ SELECT
ss_net_paid_inc_tax,
ss_net_profit,
ss_sold_date_sk
-FROM tpcds.{table_name};
+FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -1050,7 +1050,7 @@ SELECT
sr_store_credit,
sr_net_loss,
sr_returned_date_sk
-FROM tpcds.{table_name};
+FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -1183,7 +1183,7 @@ SELECT
ws_net_paid_inc_ship_tax,
ws_net_profit,
ws_sold_date_sk
-FROM tpcds.{table_name};
+FROM tpcds{scale_factor}.{table_name};
====
---- DATASET
tpcds_partitioned
@@ -1283,5 +1283,5 @@ SELECT
wr_account_credit,
wr_net_loss,
wr_returned_date_sk
-FROM tpcds.{table_name};
+FROM tpcds{scale_factor}.{table_name};
====
diff --git a/tests/util/test_file_parser.py b/tests/util/test_file_parser.py
index 754854f49..d0136a9cf 100644
--- a/tests/util/test_file_parser.py
+++ b/tests/util/test_file_parser.py
@@ -394,7 +394,7 @@ def load_tpc_queries(workload, include_stress_queries=False, query_name_filters=
# find. Both workload directories contain other queries that are not part of the TPC
# spec.
file_workload = workload
- if workload == "tpcds":
+ if workload in ["tpcds", "tpcds_partitioned"]:
# TPCDS is assumed to always use decimal_v2, which is the default since 3.0
file_workload = "tpcds-decimal_v2"
if include_stress_queries:
@@ -402,7 +402,11 @@ def load_tpc_queries(workload, include_stress_queries=False, query_name_filters=
else:
file_name_pattern = re.compile(r"^{0}-(q.*).test$".format(file_workload))
- query_name_pattern = re.compile(r"^{0}-(.*)$".format(workload.upper()))
+ query_name_prefix = workload.upper()
+ if workload == "tpcds_partitioned":
+ # 'tpcds_partitioned' symlink to the same queries set of 'tpcds'.
+ query_name_prefix = "TPCDS"
+ query_name_pattern = re.compile(r"^{0}-(.*)$".format(query_name_prefix))
if workload == "tpch_nested":
query_name_pattern = re.compile(r"^TPCH-(.*)$")
@@ -426,6 +430,9 @@ def load_tpc_queries(workload, include_stress_queries=False, query_name_filters=
test_cases = parse_query_test_file(file_path)
for test_case in test_cases:
query_sql = remove_comments(test_case["QUERY"])
+ if workload == "tpcds_partitioned":
+ # replace old columns names from old TPC-DS spec with a new one.
+ query_sql = query_sql.replace("c_last_review_date", "c_last_review_date_sk")
if re.match(filter_regex, test_case["QUERY_NAME"]):
query_name_match = query_name_pattern.search(test_case["QUERY_NAME"])
(impala) 02/03: IMPALA-13513: Support decode function
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 3574fe89b0cd583265805f7052cac1075c6845cf
Author: Steve Carlin <sc...@cloudera.com>
AuthorDate: Tue Nov 5 09:40:16 2024 -0800
IMPALA-13513: Support decode function
The decode operator is similar to the case operator. The tricky part
for supporting the decode is that some of the operands are search
parameters and some are return parameters. The search parameters all
need to be compatible with each other. And the return parameters need
to be compatible with each other as well. Much of the code deals
with casting these parameters to compatible types.
Change-Id: Ia3b68fda7cfa14799a41428e35d5bbc5984a801a
Reviewed-on: http://gerrit.cloudera.org:8080/22031
Reviewed-by: Michael Smith <mi...@cloudera.com>
Reviewed-by: Joe McDonnell <jo...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../calcite/coercenodes/CoerceOperandShuttle.java | 70 +++++++++++-
.../impala/calcite/functions/FunctionResolver.java | 27 ++++-
.../impala/calcite/functions/RexCallConverter.java | 14 +++
.../operators/ImpalaCustomOperatorTable.java | 2 +
.../calcite/operators/ImpalaDecodeFunction.java | 127 +++++++++++++++++++++
5 files changed, 233 insertions(+), 7 deletions(-)
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/coercenodes/CoerceOperandShuttle.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/coercenodes/CoerceOperandShuttle.java
index 28e1f84f1..67aac1060 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/coercenodes/CoerceOperandShuttle.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/coercenodes/CoerceOperandShuttle.java
@@ -43,6 +43,7 @@ import org.apache.impala.catalog.ScalarType;
import org.apache.impala.catalog.Type;
import org.apache.impala.calcite.functions.FunctionResolver;
import org.apache.impala.calcite.functions.ImplicitTypeChecker;
+import org.apache.impala.calcite.operators.ImpalaDecodeFunction;
import org.apache.impala.calcite.type.ImpalaTypeConverter;
import java.math.BigDecimal;
@@ -109,6 +110,12 @@ public class CoerceOperandShuttle extends RexShuttle {
return castedOperandsCall;
}
+ // had to special case decode since the casting is different from common
+ // functions.
+ if (castedOperandsCall.getOperator().getName().equals("DECODE")) {
+ return castDecodedFunction(castedOperandsCall, factory, rexBuilder);
+ }
+
Function fn = FunctionResolver.getSupertypeFunction(castedOperandsCall);
if (fn == null) {
@@ -324,17 +331,23 @@ public class CoerceOperandShuttle extends RexShuttle {
return fn.functionName().equals("case");
}
+ private static RexNode castOperand(RexNode node, Type toImpalaType,
+ RelDataTypeFactory factory, RexBuilder rexBuilder) {
+
+ RelDataType toType = ImpalaTypeConverter.getRelDataType(toImpalaType);
+ return castOperand(node, toType, factory, rexBuilder);
+ }
+
/**
* castOperand takes a RexNode and a toType and returns the RexNode
* with a potential cast wrapper. If the types match, then the
* original RexNode is returned. If the toType is an incompatible
* type, this method returns null.
*/
- private static RexNode castOperand(RexNode node, Type toImpalaType,
+ private static RexNode castOperand(RexNode node, RelDataType toType,
RelDataTypeFactory factory, RexBuilder rexBuilder) {
RelDataType fromType = node.getType();
- RelDataType toType = ImpalaTypeConverter.getRelDataType(toImpalaType);
// No need to cast if types are the same
if (fromType.getSqlTypeName().equals(toType.getSqlTypeName())) {
@@ -359,4 +372,57 @@ public class CoerceOperandShuttle extends RexShuttle {
}
return rexBuilder.makeCast(toType, node);
}
+
+ /**
+ * Special casting function for the decode operator. While the case operator
+ * has its operands as boolean types, the operands within the decode operator
+ * can be any compatible type with the search operator. Therefore, all operands
+ * need to be checked to see if they need to be cast to a compatible type, not
+ * just the return parameters.
+ */
+ private static RexNode castDecodedFunction(RexCall decodeCall,
+ RelDataTypeFactory factory, RexBuilder rexBuilder) {
+ List<RexNode> operands = decodeCall.getOperands();
+ List<RelDataType> argTypes =
+ Util.transform(operands, RexNode::getType);
+ List<RexNode> newOperands = new ArrayList<>(operands.size());
+
+ // Grab the compatible search operand from all the search operands
+ RelDataType searchOperand =
+ ImpalaDecodeFunction.getCompatibleSearchOperand(argTypes, factory);
+ // Grab the compatible return operand from all the return operands
+ RelDataType returnType =
+ ImpalaDecodeFunction.getCompatibleReturnType(argTypes, factory);
+
+ boolean hasElse = (argTypes.size() % 2 == 0);
+ int numNonElseParams = hasElse ? argTypes.size() - 1 : argTypes.size();
+
+
+ // start by potentially casting all the search operands. Check the first one outside
+ // of the loop.
+ newOperands.add(castOperand(operands.get(0), searchOperand, factory, rexBuilder));
+
+ // Now we loop through all the operands, alternating between search type
+ // and return type. We don't include the last one in case there is an
+ // else parameter
+ for (int i = 1; i < numNonElseParams; ++i) {
+ // Alternate between return and search
+ RelDataType toType = (i % 2 == 0) ? returnType : searchOperand;
+ newOperands.add(castOperand(operands.get(i), toType, factory, rexBuilder));
+ }
+
+ // Check the else param for casting
+ if (hasElse) {
+ int elseParam = operands.size() - 1;
+ newOperands.add(castOperand(operands.get(elseParam), returnType, factory,
+ rexBuilder));
+ }
+
+ // If any operand was casted, create the new RexNode, otherwise, just send back
+ // the existing one.
+ return operands.equals(newOperands)
+ ? decodeCall
+ : (RexCall) rexBuilder.makeCall(returnType, decodeCall.getOperator(),
+ newOperands);
+ }
}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/FunctionResolver.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/FunctionResolver.java
index 40852e477..1e88f18c5 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/FunctionResolver.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/FunctionResolver.java
@@ -209,21 +209,38 @@ public class FunctionResolver {
// return value as its only operand. This operand needs to be calculated based
// on its parameters (see getCaseArgs method comment) if we are looking for a
// non-exact match (the best case function that would work if casting the operands).
- if (name.equals("case")) {
+ // The decode statement is also special in that the 3rd parameter is going to
+ // contain the return value
+ int caseOperandNum = getCaseOperandNum(name);
+ if (caseOperandNum != -1) {
return exactMatch
- ? Lists.newArrayList(ImpalaTypeConverter.createImpalaType(argTypes.get(1)))
- : getCaseArgs(argTypes);
+ ? Lists.newArrayList(ImpalaTypeConverter.createImpalaType(
+ argTypes.get(caseOperandNum)))
+ : getCaseArgs(name, argTypes);
}
return ImpalaTypeConverter.getNormalizedImpalaTypes(argTypes);
}
+ private static int getCaseOperandNum(String name) {
+ if (name.equals("case")) {
+ return 1;
+ }
+ if (name.equals("decode")) {
+ return 2;
+ }
+ return -1;
+ }
+
/**
* getCaseArgs needs to calculate the proper operand. It walks through
* all the "then" clauses (and the default clause) and calculated the
* most compatible type which it will use as its return value.
*/
- private static List<Type> getCaseArgs(List<RelDataType> argTypes) {
+ private static List<Type> getCaseArgs(String name, List<RelDataType> argTypes) {
+ // Only handles the "case" function, not the "decode".
+ Preconditions.checkState(name.equals("case"));
+
int numOperands = argTypes.size();
Type compatibleType = Type.NULL;
for (int i = 0; i < numOperands; ++i) {
@@ -240,7 +257,7 @@ public class FunctionResolver {
/**
* shouldSkipOperand checks to see if we should skip this operand from casting.
* Currently it only checks for CASE. CASE will have some boolean operands which
- * will never need casting. TODO: support DECODE
+ * will never need casting.
*/
public static boolean shouldSkipOperandForCase(int numOperands, int i) {
// The even number parameters are the boolean checks, but the last parameter
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexCallConverter.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexCallConverter.java
index 4e8354acc..586ffe238 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexCallConverter.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexCallConverter.java
@@ -33,6 +33,7 @@ import org.apache.impala.analysis.BinaryPredicate;
import org.apache.impala.analysis.CaseWhenClause;
import org.apache.impala.analysis.CompoundPredicate;
import org.apache.impala.analysis.Expr;
+import org.apache.impala.analysis.FunctionCallExpr;
import org.apache.impala.analysis.TimestampArithmeticExpr;
import org.apache.impala.calcite.type.ImpalaTypeConverter;
import org.apache.impala.catalog.Function;
@@ -110,6 +111,10 @@ public class RexCallConverter {
impalaRetType);
}
+ if (fn.functionName().equals("decode")) {
+ return createDecodeExpr(fn, params, impalaRetType);
+ }
+
switch (rexCall.getOperator().getKind()) {
case CASE:
return createCaseExpr(fn, params, impalaRetType);
@@ -170,6 +175,15 @@ public class RexCallConverter {
return new AnalyzedCastExpr(impalaRetType, params.get(0));
}
+ private static Expr createDecodeExpr(Function fn, List<Expr> params,
+ Type impalaRetType) throws ImpalaException {
+ // case expression is wrapped with a function, which is how the Impala
+ // analyzer Expr code expects to handle the decode function.
+ FunctionCallExpr decodeExpr =
+ new AnalyzedFunctionCallExpr(fn, params, impalaRetType);
+ return new AnalyzedCaseExpr(fn, impalaRetType, decodeExpr);
+ }
+
private static Expr createCaseExpr(Function fn, List<Expr> params, Type retType) {
List<CaseWhenClause> caseWhenClauses = new ArrayList<>();
Expr whenParam = null;
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaCustomOperatorTable.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaCustomOperatorTable.java
index 2a1a2a8a7..d3e3b45e7 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaCustomOperatorTable.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaCustomOperatorTable.java
@@ -222,6 +222,8 @@ public class ImpalaCustomOperatorTable extends ReflectiveSqlOperatorTable {
// validator needs to search for the explicit_cast function while validating.
public static final ImpalaCastFunction EXPLICIT_CAST = ImpalaCastFunction.INSTANCE;
+ public static final ImpalaDecodeFunction DECODE = ImpalaDecodeFunction.INSTANCE;
+
public static ImpalaCustomOperatorTable instance() {
return INSTANCE.get();
}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaDecodeFunction.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaDecodeFunction.java
new file mode 100644
index 000000000..2f8987f99
--- /dev/null
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaDecodeFunction.java
@@ -0,0 +1,127 @@
+// 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.
+
+package org.apache.impala.calcite.operators;
+
+import com.google.common.base.Preconditions;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperandCountRange;
+import org.apache.calcite.sql.SqlOperatorBinding;
+import org.apache.calcite.sql.SqlSyntax;
+import org.apache.calcite.sql.type.SqlOperandCountRanges;
+import org.apache.impala.calcite.type.ImpalaTypeConverter;
+import org.apache.impala.calcite.type.ImpalaTypeSystemImpl;
+
+import java.util.List;
+
+/**
+ * Special operator for Decode.
+ *
+ * The decode operator has to infer its return type off of a compatible
+ * parameter from all the choices of return parameters. These choices
+ * are the third parameter, the fifth parameter, the seventh, etc...
+ * If there are an even number of parameters, the last parameter is part
+ * of an else clause, so that is also a return parameter.
+ *
+ * This class also ensures that the search parameters are all compatible
+ * as well. These are all the parameters not mentioned; The first, second,
+ * fourth, sixth, and every other parameter except the last one.
+ *
+ * An example of a decode function (in the documentation) is:
+ * DECODE(day_of_week, 1, "Monday", 2, "Tuesday", 3, "Wednesday",
+ * 4, "Thursday", 5, "Friday", 6, "Saturday", 7, "Sunday", "Unknown day")
+ * The return values here are of type string, and the search parameters are of
+ * some numeric type for day_of_week and the choices for day of the week.
+ *
+ */
+public class ImpalaDecodeFunction extends ImpalaOperator {
+
+ public static ImpalaDecodeFunction INSTANCE = new ImpalaDecodeFunction();
+
+ private ImpalaDecodeFunction() {
+ super("DECODE");
+ }
+
+
+ @Override
+ public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
+ List<RelDataType> operandTypes = CommonOperatorFunctions.getOperandTypes(opBinding);
+
+ RexBuilder rexBuilder =
+ new RexBuilder(new JavaTypeFactoryImpl(new ImpalaTypeSystemImpl()));
+ RelDataTypeFactory factory = rexBuilder.getTypeFactory();
+
+ // No need to capture the return value, but an exception will be thrown
+ // if the parameters are not compatible.
+ getCompatibleSearchOperand(operandTypes, factory);
+
+ return getCompatibleReturnType(operandTypes, factory);
+ }
+
+ public static RelDataType getCompatibleSearchOperand(
+ List<RelDataType> operandTypes, RelDataTypeFactory factory) {
+ // check search operands are all compatible, grab the first.
+ RelDataType commonSearchOperand = operandTypes.get(0);
+ int searchOperandsToCheck = operandTypes.size()/2;
+ // Check that the second, fourth, sixth, etc... parameter is compatible
+ // but don't check the last one.
+ for (int i = 1; i < operandTypes.size() - 1; i += 2) {
+ commonSearchOperand = ImpalaTypeConverter.getCompatibleType(commonSearchOperand,
+ operandTypes.get(i), factory);
+ if (commonSearchOperand == null) {
+ throw new IllegalArgumentException("Decode function has incompatible " +
+ "types with search argument and argument number " + i);
+ }
+ }
+ return commonSearchOperand;
+ }
+
+ public static RelDataType getCompatibleReturnType(
+ List<RelDataType> operandTypes, RelDataTypeFactory factory) {
+ // Initialize with the third parameter
+ RelDataType returnType = operandTypes.get(2);
+ returnType = factory.createTypeWithNullability(returnType, true);
+ // Skip over every other parameter starting with the fifth one.
+ for (int i = 4; i < operandTypes.size(); i += 2) {
+ returnType = ImpalaTypeConverter.getCompatibleType(returnType,
+ operandTypes.get(i), factory);
+ if (returnType == null) {
+ throw new IllegalArgumentException("Decode function has incompatible " +
+ "return type (argument number) " + i);
+ }
+ }
+ // If there are an even number of parameters, the last one is the else
+ // clause, so check that one too.
+ if ((operandTypes.size() % 2) == 0) {
+ returnType = ImpalaTypeConverter.getCompatibleType(returnType,
+ operandTypes.get(operandTypes.size() - 1), factory);
+ }
+ return returnType;
+ }
+
+ @Override
+ public SqlOperandCountRange getOperandCountRange() {
+ return SqlOperandCountRanges.from(3);
+ }
+}
(impala) 03/03: IMPALA-13511: Calcite planner: support sub-second datetime parts
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 9a2496dd288ea23857c67cade25f96807c5e61dc
Author: Steve Carlin <sc...@cloudera.com>
AuthorDate: Tue Nov 5 06:35:16 2024 -0800
IMPALA-13511: Calcite planner: support sub-second datetime parts
This commit allows support for milliseconds, microseconds, and
nanoseconds for Calcite.
The ImpalaSqlIntervalQualifier class extends the Calcite
SqlIntervalQualifier class which handles most datetime parts. Some
code was copied from the base class since some of the methods were private,
but in general, the code handling the parts is exactly the same.
Change-Id: I392c3900c70e7754a35ef25fc720ba4a2f2e5dd6
Reviewed-on: http://gerrit.cloudera.org:8080/22029
Reviewed-by: Michael Smith <mi...@cloudera.com>
Reviewed-by: Joe McDonnell <jo...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
java/calcite-planner/src/main/codegen/config.fmpp | 1 +
.../src/main/codegen/templates/Parser.jj | 150 +++++++++++++------
.../calcite/coercenodes/CoerceOperandShuttle.java | 6 +
.../calcite/operators/ImpalaOperatorTable.java | 3 +
.../calcite/type/ImpalaSqlIntervalQualifier.java | 159 +++++++++++++++++++++
5 files changed, 273 insertions(+), 46 deletions(-)
diff --git a/java/calcite-planner/src/main/codegen/config.fmpp b/java/calcite-planner/src/main/codegen/config.fmpp
index 18886790a..9cd56e6c3 100644
--- a/java/calcite-planner/src/main/codegen/config.fmpp
+++ b/java/calcite-planner/src/main/codegen/config.fmpp
@@ -24,6 +24,7 @@ parser: {
# List of additional classes and packages to import.
# Example: "org.apache.calcite.sql.*", "java.util.List".
imports: [
+ "org.apache.impala.calcite.type.ImpalaSqlIntervalQualifier"
]
# List of new keywords. Example: "DATABASES", "TABLES". If the keyword is
diff --git a/java/calcite-planner/src/main/codegen/templates/Parser.jj b/java/calcite-planner/src/main/codegen/templates/Parser.jj
index f21f912c1..128750f30 100644
--- a/java/calcite-planner/src/main/codegen/templates/Parser.jj
+++ b/java/calcite-planner/src/main/codegen/templates/Parser.jj
@@ -121,6 +121,7 @@ import org.apache.calcite.util.Pair;
import org.apache.calcite.util.SourceStringReader;
import org.apache.calcite.util.Util;
import org.apache.calcite.util.trace.CalciteTrace;
+import org.apache.impala.calcite.type.ImpalaSqlIntervalQualifier;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
@@ -4890,7 +4891,7 @@ SqlNode PeriodConstructor() :
SqlLiteral IntervalLiteral() :
{
final String p;
- final SqlIntervalQualifier intervalQualifier;
+ final ImpalaSqlIntervalQualifier intervalQualifier;
int sign = 1;
final Span s;
}
@@ -4914,7 +4915,7 @@ SqlLiteral IntervalLiteral() :
SqlNode IntervalLiteralOrExpression() :
{
final String p;
- final SqlIntervalQualifier intervalQualifier;
+ final ImpalaSqlIntervalQualifier intervalQualifier;
int sign = 1;
final Span s;
SqlNode e;
@@ -5027,7 +5028,34 @@ TimeUnit Second() :
<SECONDS> { return warn(TimeUnit.SECOND); }
}
-SqlIntervalQualifier IntervalQualifier() :
+TimeUnit MilliSecond() :
+{
+}
+{
+ <MILLISECOND> { return TimeUnit.MILLISECOND; }
+|
+ <MILLISECONDS> { return warn(TimeUnit.MILLISECOND); }
+}
+
+TimeUnit MicroSecond() :
+{
+}
+{
+ <MICROSECOND> { return TimeUnit.MICROSECOND; }
+|
+ <MICROSECONDS> { return warn(TimeUnit.MICROSECOND); }
+}
+
+TimeUnit NanoSecond() :
+{
+}
+{
+ <NANOSECOND> { return TimeUnit.NANOSECOND; }
+|
+ <NANOSECONDS> { return warn(TimeUnit.NANOSECOND); }
+}
+
+ImpalaSqlIntervalQualifier IntervalQualifier() :
{
final Span s;
final TimeUnit start;
@@ -5093,15 +5121,39 @@ SqlIntervalQualifier IntervalQualifier() :
| { startPrec = -1; }
)
{ end = null; }
+ |
+ start = MilliSecond() { s = span(); }
+ (
+ <LPAREN> startPrec = UnsignedIntLiteral()
+ <RPAREN>
+ | { startPrec = -1; }
+ )
+ { end = null; }
+ |
+ start = MicroSecond() { s = span(); }
+ (
+ <LPAREN> startPrec = UnsignedIntLiteral()
+ <RPAREN>
+ | { startPrec = -1; }
+ )
+ { end = null; }
+ |
+ start = NanoSecond() { s = span(); }
+ (
+ <LPAREN> startPrec = UnsignedIntLiteral()
+ <RPAREN>
+ | { startPrec = -1; }
+ )
+ { end = null; }
)
{
- return new SqlIntervalQualifier(start, startPrec, end, secondFracPrec,
+ return new ImpalaSqlIntervalQualifier(start, startPrec, end, secondFracPrec,
s.end(this));
}
}
/** Interval qualifier without 'TO unit'. */
-SqlIntervalQualifier IntervalQualifierStart() :
+ImpalaSqlIntervalQualifier IntervalQualifierStart() :
{
final Span s;
final TimeUnit start;
@@ -5118,6 +5170,9 @@ SqlIntervalQualifier IntervalQualifierStart() :
| start = Day()
| start = Hour()
| start = Minute()
+ | start = MilliSecond()
+ | start = MicroSecond()
+ | start = NanoSecond()
)
{ s = span(); }
startPrec = PrecisionOpt()
@@ -5129,14 +5184,14 @@ SqlIntervalQualifier IntervalQualifierStart() :
]
)
{
- return new SqlIntervalQualifier(start, startPrec, null, secondFracPrec,
+ return new ImpalaSqlIntervalQualifier(start, startPrec, null, secondFracPrec,
s.end(this));
}
}
/** Parses a built-in time unit (e.g. "YEAR")
* or user-defined time frame (e.g. "MINUTE15")
- * and in each case returns a {@link SqlIntervalQualifier}.
+ * and in each case returns a {@link ImpalaSqlIntervalQualifier}.
*
* <p>The units are used in several functions, incuding CEIL, FLOOR, EXTRACT.
* Includes NANOSECOND, MILLISECOND, which were previously allowed in EXTRACT
@@ -5149,9 +5204,9 @@ SqlIntervalQualifier IntervalQualifierStart() :
* parsed as identifiers and can be resolved in the validator if they are
* registered as abbreviations in your time frame set.
*/
-SqlIntervalQualifier TimeUnitOrName() : {
+ImpalaSqlIntervalQualifier TimeUnitOrName() : {
final SqlIdentifier unitName;
- final SqlIntervalQualifier intervalQualifier;
+ final ImpalaSqlIntervalQualifier intervalQualifier;
}
{
// When we see a time unit that is also a non-reserved keyword, such as
@@ -5166,13 +5221,13 @@ SqlIntervalQualifier TimeUnitOrName() : {
return intervalQualifier;
}
| unitName = SimpleIdentifier() {
- return new SqlIntervalQualifier(unitName.getSimple(),
+ return new ImpalaSqlIntervalQualifier(unitName.getSimple(),
unitName.getParserPosition());
}
}
/** Parses a built-in time unit (e.g. "YEAR")
- * and returns a {@link SqlIntervalQualifier}.
+ * and returns a {@link ImpalaSqlIntervalQualifier}.
*
* <p>Includes {@code WEEK} and {@code WEEK(SUNDAY)} through
{@code WEEK(SATURDAY)}.
@@ -5181,42 +5236,42 @@ SqlIntervalQualifier TimeUnitOrName() : {
* parsed as identifiers and can be resolved in the validator if they are
* registered as abbreviations in your time frame set.
*/
-SqlIntervalQualifier TimeUnit() : {
+ImpalaSqlIntervalQualifier TimeUnit() : {
final Span span;
final String w;
}
{
- <NANOSECOND> { return new SqlIntervalQualifier(TimeUnit.NANOSECOND, null, getPos()); }
-| <MICROSECOND> { return new SqlIntervalQualifier(TimeUnit.MICROSECOND, null, getPos()); }
-| <MILLISECOND> { return new SqlIntervalQualifier(TimeUnit.MILLISECOND, null, getPos()); }
-| <SECOND> { return new SqlIntervalQualifier(TimeUnit.SECOND, null, getPos()); }
-| <MINUTE> { return new SqlIntervalQualifier(TimeUnit.MINUTE, null, getPos()); }
-| <HOUR> { return new SqlIntervalQualifier(TimeUnit.HOUR, null, getPos()); }
-| <DAY> { return new SqlIntervalQualifier(TimeUnit.DAY, null, getPos()); }
-| <DAYOFWEEK> { return new SqlIntervalQualifier(TimeUnit.DOW, null, getPos()); }
-| <DAYOFYEAR> { return new SqlIntervalQualifier(TimeUnit.DOY, null, getPos()); }
-| <DOW> { return new SqlIntervalQualifier(TimeUnit.DOW, null, getPos()); }
-| <DOY> { return new SqlIntervalQualifier(TimeUnit.DOY, null, getPos()); }
-| <ISODOW> { return new SqlIntervalQualifier(TimeUnit.ISODOW, null, getPos()); }
-| <ISOYEAR> { return new SqlIntervalQualifier(TimeUnit.ISOYEAR, null, getPos()); }
+ <NANOSECOND> { return new ImpalaSqlIntervalQualifier(TimeUnit.NANOSECOND, null, getPos()); }
+| <MICROSECOND> { return new ImpalaSqlIntervalQualifier(TimeUnit.MICROSECOND, null, getPos()); }
+| <MILLISECOND> { return new ImpalaSqlIntervalQualifier(TimeUnit.MILLISECOND, null, getPos()); }
+| <SECOND> { return new ImpalaSqlIntervalQualifier(TimeUnit.SECOND, null, getPos()); }
+| <MINUTE> { return new ImpalaSqlIntervalQualifier(TimeUnit.MINUTE, null, getPos()); }
+| <HOUR> { return new ImpalaSqlIntervalQualifier(TimeUnit.HOUR, null, getPos()); }
+| <DAY> { return new ImpalaSqlIntervalQualifier(TimeUnit.DAY, null, getPos()); }
+| <DAYOFWEEK> { return new ImpalaSqlIntervalQualifier(TimeUnit.DOW, null, getPos()); }
+| <DAYOFYEAR> { return new ImpalaSqlIntervalQualifier(TimeUnit.DOY, null, getPos()); }
+| <DOW> { return new ImpalaSqlIntervalQualifier(TimeUnit.DOW, null, getPos()); }
+| <DOY> { return new ImpalaSqlIntervalQualifier(TimeUnit.DOY, null, getPos()); }
+| <ISODOW> { return new ImpalaSqlIntervalQualifier(TimeUnit.ISODOW, null, getPos()); }
+| <ISOYEAR> { return new ImpalaSqlIntervalQualifier(TimeUnit.ISOYEAR, null, getPos()); }
| <WEEK> { span = span(); }
(
// There is a choice between "WEEK(weekday)" and "WEEK". We prefer
// the former, and the parser will look ahead for '('.
LOOKAHEAD(2)
<LPAREN> w = weekdayName() <RPAREN> {
- return new SqlIntervalQualifier(w, span.end(this));
+ return new ImpalaSqlIntervalQualifier(w, span.end(this));
}
|
- { return new SqlIntervalQualifier(TimeUnit.WEEK, null, getPos()); }
+ { return new ImpalaSqlIntervalQualifier(TimeUnit.WEEK, null, getPos()); }
)
-| <MONTH> { return new SqlIntervalQualifier(TimeUnit.MONTH, null, getPos()); }
-| <QUARTER> { return new SqlIntervalQualifier(TimeUnit.QUARTER, null, getPos()); }
-| <YEAR> { return new SqlIntervalQualifier(TimeUnit.YEAR, null, getPos()); }
-| <EPOCH> { return new SqlIntervalQualifier(TimeUnit.EPOCH, null, getPos()); }
-| <DECADE> { return new SqlIntervalQualifier(TimeUnit.DECADE, null, getPos()); }
-| <CENTURY> { return new SqlIntervalQualifier(TimeUnit.CENTURY, null, getPos()); }
-| <MILLENNIUM> { return new SqlIntervalQualifier(TimeUnit.MILLENNIUM, null, getPos()); }
+| <MONTH> { return new ImpalaSqlIntervalQualifier(TimeUnit.MONTH, null, getPos()); }
+| <QUARTER> { return new ImpalaSqlIntervalQualifier(TimeUnit.QUARTER, null, getPos()); }
+| <YEAR> { return new ImpalaSqlIntervalQualifier(TimeUnit.YEAR, null, getPos()); }
+| <EPOCH> { return new ImpalaSqlIntervalQualifier(TimeUnit.EPOCH, null, getPos()); }
+| <DECADE> { return new ImpalaSqlIntervalQualifier(TimeUnit.DECADE, null, getPos()); }
+| <CENTURY> { return new ImpalaSqlIntervalQualifier(TimeUnit.CENTURY, null, getPos()); }
+| <MILLENNIUM> { return new ImpalaSqlIntervalQualifier(TimeUnit.MILLENNIUM, null, getPos()); }
}
String weekdayName() :
@@ -6098,7 +6153,7 @@ SqlNode BuiltinFunctionCall() :
SqlNode e;
final Span s;
SqlDataTypeSpec dt;
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
final SqlNode node;
final SqlLiteral style; // mssql convert 'style' operand
final SqlFunction f;
@@ -6853,7 +6908,7 @@ SqlCall DateDiffFunctionCall() :
{
final List<SqlNode> args = new ArrayList<SqlNode>();
final Span s;
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
}
{
<DATE_DIFF> { s = span(); }
@@ -6875,7 +6930,7 @@ SqlCall TimestampAddFunctionCall() :
{
final List<SqlNode> args = new ArrayList<SqlNode>();
final Span s;
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
}
{
<TIMESTAMPADD> { s = span(); }
@@ -6898,7 +6953,7 @@ SqlCall TimestampDiffFunctionCall() :
{
final List<SqlNode> args = new ArrayList<SqlNode>();
final Span s;
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
}
{
<TIMESTAMPDIFF> { s = span(); }
@@ -6926,7 +6981,7 @@ SqlCall TimestampDiff3FunctionCall() :
{
final List<SqlNode> args = new ArrayList<SqlNode>();
final Span s;
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
}
{
<TIMESTAMP_DIFF> { s = span(); }
@@ -6948,7 +7003,7 @@ SqlCall DatetimeDiffFunctionCall() :
{
final List<SqlNode> args = new ArrayList<SqlNode>();
final Span s;
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
}
{
<DATETIME_DIFF> { s = span(); }
@@ -6970,7 +7025,7 @@ SqlCall DateTruncFunctionCall() :
{
final List<SqlNode> args = new ArrayList<SqlNode>();
final Span s;
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
}
{
<DATE_TRUNC> { s = span(); }
@@ -6998,7 +7053,7 @@ SqlCall TimestampTruncFunctionCall() :
{
final List<SqlNode> args = new ArrayList<SqlNode>();
final Span s;
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
}
{
<TIMESTAMP_TRUNC> { s = span(); }
@@ -7018,7 +7073,7 @@ SqlCall TimeDiffFunctionCall() :
{
final List<SqlNode> args = new ArrayList<SqlNode>();
final Span s;
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
}
{
<TIME_DIFF> { s = span(); }
@@ -7040,7 +7095,7 @@ SqlNode DatetimeTruncFunctionCall() :
{
final List<SqlNode> args = new ArrayList<SqlNode>();
final Span s;
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
final SqlNode literal;
}
{
@@ -7061,7 +7116,7 @@ SqlCall TimeTruncFunctionCall() :
{
final List<SqlNode> args = new ArrayList<SqlNode>();
final Span s;
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
}
{
<TIME_TRUNC> { s = span(); }
@@ -7376,7 +7431,7 @@ SqlNode StandardFloorCeilOptions(Span s, boolean floorFlag) :
{
SqlNode e;
final List<SqlNode> args = new ArrayList<SqlNode>();
- final SqlIntervalQualifier unit;
+ final ImpalaSqlIntervalQualifier unit;
SqlCall function;
final Span s1;
}
@@ -8187,7 +8242,9 @@ SqlPostfixOperator PostfixRowOperator() :
| < MESSAGE_TEXT: "MESSAGE_TEXT" >
| < METHOD: "METHOD" >
| < MICROSECOND: "MICROSECOND" >
+| < MICROSECONDS: "MICROSECONDS" >
| < MILLISECOND: "MILLISECOND" >
+| < MILLISECONDS: "MILLISECONDS" >
| < MILLENNIUM: "MILLENNIUM" >
| < MIN: "MIN" >
| < MINUTE: "MINUTE" >
@@ -8205,6 +8262,7 @@ SqlPostfixOperator PostfixRowOperator() :
| < NAME: "NAME" >
| < NAMES: "NAMES" >
| < NANOSECOND: "NANOSECOND" >
+| < NANOSECONDS: "NANOSECONDS" >
| < NATIONAL: "NATIONAL" >
| < NATURAL: "NATURAL" >
| < NCHAR: "NCHAR" >
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/coercenodes/CoerceOperandShuttle.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/coercenodes/CoerceOperandShuttle.java
index 67aac1060..8c845e0f5 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/coercenodes/CoerceOperandShuttle.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/coercenodes/CoerceOperandShuttle.java
@@ -349,6 +349,12 @@ public class CoerceOperandShuttle extends RexShuttle {
RelDataType fromType = node.getType();
+ // no need to cast if type is interval, this will get handled when
+ // changing to the physical Expr object.
+ if (SqlTypeUtil.isInterval(fromType)) {
+ return node;
+ }
+
// No need to cast if types are the same
if (fromType.getSqlTypeName().equals(toType.getSqlTypeName())) {
return node;
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaOperatorTable.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaOperatorTable.java
index 8e9e28059..354e33b19 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaOperatorTable.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaOperatorTable.java
@@ -71,6 +71,9 @@ public class ImpalaOperatorTable extends ReflectiveSqlOperatorTable {
.add("minute")
.add("second")
.add("millisecond")
+ .add("dayofmonth")
+ .add("dayofyear")
+ .add("weekofyear")
.add("corr")
.add("covar_pop")
.add("covar_samp")
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/type/ImpalaSqlIntervalQualifier.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/type/ImpalaSqlIntervalQualifier.java
new file mode 100644
index 000000000..da10640d8
--- /dev/null
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/type/ImpalaSqlIntervalQualifier.java
@@ -0,0 +1,159 @@
+// 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.
+
+package org.apache.impala.calcite.type;
+
+import org.apache.calcite.avatica.util.TimeUnit;
+import org.apache.calcite.avatica.util.TimeUnitRange;
+import org.apache.calcite.rel.type.RelDataTypeSystem;
+import org.apache.calcite.runtime.CalciteContextException;
+import org.apache.calcite.sql.SqlUtil;
+import org.apache.calcite.sql.SqlIntervalQualifier;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.util.Util;
+
+import static org.apache.calcite.linq4j.Nullness.castNonNull;
+import static org.apache.calcite.util.Static.RESOURCE;
+import static java.util.Objects.requireNonNull;
+
+import java.math.BigDecimal;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ImpalaSqlIntervalQualifier is an extension of the Calcite SqlIntervalQualifier
+ * class. The Calcite version does not handle sub-millisecond parts (the other parts
+ * are still processed in the super class). In order to handle these parts, some
+ * private methods had to be copied from the base class.
+ */
+public class ImpalaSqlIntervalQualifier extends SqlIntervalQualifier {
+
+ public ImpalaSqlIntervalQualifier(
+ TimeUnit startUnit,
+ int startPrecision,
+ TimeUnit endUnit,
+ int fractionalSecondPrecision,
+ SqlParserPos pos) {
+ super(startUnit, startPrecision, endUnit, fractionalSecondPrecision, pos);
+ }
+
+ public ImpalaSqlIntervalQualifier(
+ TimeUnit startUnit,
+ TimeUnit endUnit,
+ SqlParserPos pos) {
+ super(startUnit, endUnit, pos);
+ }
+
+ public ImpalaSqlIntervalQualifier(String timeFrameName,
+ SqlParserPos pos) {
+ super(timeFrameName, pos);
+ }
+
+ @Override
+ public int[] evaluateIntervalLiteral(String value, SqlParserPos pos,
+ RelDataTypeSystem typeSystem) {
+
+ switch(timeUnitRange) {
+ case MILLISECOND:
+ case MICROSECOND:
+ case NANOSECOND:
+ return super.evaluateIntervalLiteral(value, pos, typeSystem);
+ }
+
+ // save original value for if we have to throw
+ final String value0 = value;
+
+ // First strip off any leading whitespace
+ value = value.trim();
+
+ // check if the sign was explicitly specified. Record
+ // the explicit or implicit sign, and strip it off to
+ // simplify pattern matching later.
+ final int sign = getIntervalSign(value);
+ value = stripLeadingSign(value);
+
+ // If we have an empty or null literal at this point,
+ // it's illegal. Complain and bail out.
+ if (Util.isNullOrEmpty(value)) {
+ throw invalidValueException(pos, value0);
+ }
+
+ return evaluateIntervalLiteralAsSubSeconds(typeSystem, sign, value, value0,
+ pos);
+ }
+
+ private int[] evaluateIntervalLiteralAsSubSeconds(
+ RelDataTypeSystem typeSystem, int sign,
+ String value,
+ String originalValue,
+ SqlParserPos pos) {
+ BigDecimal subseconds;
+
+ // validate the integer decimal pattern used for subseconds
+ String intervalPattern = "(\\d+)";
+
+ Matcher m = Pattern.compile(intervalPattern).matcher(value);
+ if (m.matches()) {
+ // Break out field values
+ try {
+ subseconds = parseField(m, 1);
+ } catch (NumberFormatException e) {
+ throw invalidValueException(pos, originalValue);
+ }
+
+ // package values up for return
+ return fillIntervalValueArray(sign, subseconds);
+ } else {
+ throw invalidValueException(pos, originalValue);
+ }
+ }
+
+ private static String stripLeadingSign(String value) {
+ String unsignedValue = value;
+
+ if (!Util.isNullOrEmpty(value)) {
+ if (('-' == value.charAt(0)) || ('+' == value.charAt(0))) {
+ unsignedValue = value.substring(1);
+ }
+ }
+
+ return unsignedValue;
+ }
+
+ private static BigDecimal parseField(Matcher m, int i) {
+ return new BigDecimal(castNonNull(m.group(i)));
+ }
+
+ private CalciteContextException invalidValueException(SqlParserPos pos,
+ String value) {
+ return SqlUtil.newContextException(pos,
+ RESOURCE.unsupportedIntervalLiteral(
+ "'" + value + "'", "INTERVAL " + toString()));
+ }
+
+ private static int[] fillIntervalValueArray(
+ int sign,
+ BigDecimal subseconds) {
+ int[] ret = new int[3];
+
+ ret[0] = sign;
+ ret[1] = subseconds.intValue();
+ ret[2] = 0;
+
+ return ret;
+ }
+}