You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by bo...@apache.org on 2020/09/08 10:12:23 UTC
[impala] branch master updated (2359a1b -> fb6d96e)
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git.
from 2359a1b IMPALA-10119: Fix impala-shell history duplication test
new dc08b65 IMPALA-7658: Proper codegen for HiveUdfCall
new 75146c9 IMPALA-10012: ds_hll_sketch() results ascii codec decoding error fix
new b7965d8 Revert "IMPALA-10012: ds_hll_sketch() results ascii codec decoding error fix"
new fe6e625 IMPALA-10012: ds_hll_sketch() results ascii codec decoding error fix
new fb6d96e IMPALA-9741: Support querying Iceberg table by impala
The 5 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:
be/src/codegen/codegen-util.h | 28 ++
be/src/codegen/gen_ir_descriptions.py | 12 +-
be/src/codegen/impala-ir.cc | 1 +
be/src/codegen/llvm-codegen.cc | 3 +-
be/src/codegen/llvm-codegen.h | 5 +-
be/src/exprs/CMakeLists.txt | 1 +
be/src/exprs/hive-udf-call-ir.cc | 92 +++++++
be/src/exprs/hive-udf-call.cc | 286 ++++++++++++++++++---
be/src/exprs/hive-udf-call.h | 42 +++
be/src/runtime/descriptors.cc | 1 +
bin/rat_exclude_files.txt | 1 +
common/thrift/CatalogObjects.thrift | 8 +
fe/pom.xml | 20 --
.../org/apache/impala/analysis/AlterTableStmt.java | 5 +
.../java/org/apache/impala/analysis/Analyzer.java | 4 +-
.../apache/impala/analysis/ComputeStatsStmt.java | 5 +
.../apache/impala/analysis/CreateTableStmt.java | 37 +++
.../impala/analysis/IcebergPartitionField.java | 12 +-
.../impala/analysis/IcebergPartitionSpec.java | 12 +-
.../org/apache/impala/analysis/InsertStmt.java | 7 +-
.../org/apache/impala/analysis/ShowFilesStmt.java | 3 +-
.../org/apache/impala/analysis/ShowStatsStmt.java | 36 +--
.../org/apache/impala/analysis/ToSqlUtils.java | 9 +-
.../org/apache/impala/analysis/TruncateStmt.java | 7 +-
.../impala/catalog/CatalogServiceCatalog.java | 5 +-
.../org/apache/impala/catalog/FeIcebergTable.java | 246 +++++++++++++++++-
.../java/org/apache/impala/catalog/HdfsTable.java | 2 +-
.../org/apache/impala/catalog/IcebergTable.java | 110 +++++---
.../impala/catalog/local/LocalFsPartition.java | 10 +-
.../apache/impala/catalog/local/LocalFsTable.java | 2 +-
.../impala/catalog/local/LocalIcebergTable.java | 141 ++++++----
.../org/apache/impala/planner/HdfsScanNode.java | 11 +-
.../org/apache/impala/planner/IcebergScanNode.java | 222 ++++++++++++++++
.../apache/impala/planner/SingleNodePlanner.java | 14 +-
.../apache/impala/service/CatalogOpExecutor.java | 2 +-
.../java/org/apache/impala/service/Frontend.java | 9 +-
.../java/org/apache/impala/util/IcebergUtil.java | 75 +++++-
.../impala/catalog/FileMetadataLoaderTest.java | 34 +++
shell/impala_client.py | 10 +-
shell/impala_shell.py | 11 +-
shell/shell_output.py | 21 +-
testdata/data/README | 8 +
...d44ad-18bc-40f2-9dd6-aeb2cc23457c-00000.parquet | Bin 0 -> 1162 bytes
...b2521-1e8b-40c1-b846-552cd620abce-00000.parquet | Bin 0 -> 1190 bytes
...55628-0544-4833-8b11-1b4add53dfd6-00000.parquet | Bin 0 -> 1169 bytes
...530ef-93b6-4994-b3c8-db957d44848c-00000.parquet | Bin 0 -> 1162 bytes
...b22da-5f10-4cbf-8e4d-160f829b5e48-00000.parquet | Bin 0 -> 1190 bytes
...9a1f7-9024-4bc3-a030-e20861586146-00000.parquet | Bin 0 -> 1162 bytes
...814ae-56cd-486b-af81-18541437da7d-00000.parquet | Bin 0 -> 1189 bytes
...c70a4-bf4d-4a82-8c97-c90e2b4d9dcf-00000.parquet | Bin 0 -> 1190 bytes
...dca7f-c050-407e-b70c-2bd076f83e4e-00000.parquet | Bin 0 -> 1169 bytes
...31a1f-309e-43b3-a5cf-3ef82fa4a87c-00000.parquet | Bin 0 -> 1169 bytes
...38078-244c-4b38-8127-04a5bfbc4695-00000.parquet | Bin 0 -> 1162 bytes
...69895-df25-4ad8-b64d-49c4540d36c9-00000.parquet | Bin 0 -> 1168 bytes
...0c1ea-a2f5-4109-81ec-3ff9c155430f-00000.parquet | Bin 0 -> 1169 bytes
...61b8c-9d9a-4823-b64e-109770c16696-00000.parquet | Bin 0 -> 1190 bytes
...c9d65-bfc3-47c4-8da7-6610b49b3305-00000.parquet | Bin 0 -> 1162 bytes
...9b09d-c730-40a6-ad03-8b36096cb234-00000.parquet | Bin 0 -> 1190 bytes
...b2129-0d18-418a-bcdd-b1b5332ea2a5-00000.parquet | Bin 0 -> 1162 bytes
...51a0c-3d30-4cdb-afcb-e8c1cfa3caa4-00000.parquet | Bin 0 -> 1169 bytes
...7707d-58db-469c-ab96-b77188f25189-00000.parquet | Bin 0 -> 1162 bytes
...f6a62-c326-44f7-bdc3-b1471bd594d7-00000.parquet | Bin 0 -> 1162 bytes
.../9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21-m0.avro | Bin 0 -> 5045 bytes
...973-1-9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21.avro | Bin 0 -> 2538 bytes
.../metadata/v1.metadata.json | 42 +++
.../metadata/v2.metadata.json | 61 +++++
.../metadata/version-hint.text | 1 +
...5a171-0911-47c2-90c8-300f23c28772-00000.parquet | Bin 0 -> 1162 bytes
...6f547-de52-4aad-8c07-5d0f1aaa0558-00000.parquet | Bin 0 -> 1162 bytes
...024d6-8b55-42d3-b25e-5f2a31303026-00000.parquet | Bin 0 -> 1162 bytes
...23269-c8d7-40cf-94c8-8b624969b9c0-00000.parquet | Bin 0 -> 1162 bytes
...a376e-696e-4da3-95f3-7e0aacb11ca3-00000.parquet | Bin 0 -> 1162 bytes
...94175-75de-41ef-b824-3923ebfba0cb-00000.parquet | Bin 0 -> 1162 bytes
...34050-b263-41e8-bc7d-773a47b33c32-00000.parquet | Bin 0 -> 1162 bytes
...1c3ff-7154-4fd1-a090-8fa0b9c2d3c8-00000.parquet | Bin 0 -> 1162 bytes
...9da05-6be1-42e1-90e0-bff8ca08b5af-00000.parquet | Bin 0 -> 1169 bytes
...9012d-85e3-4144-ac83-574ed5046718-00000.parquet | Bin 0 -> 1169 bytes
...e3f98-5f3b-4245-9c01-76cf0f1dd21a-00000.parquet | Bin 0 -> 1169 bytes
...a1955-f01c-4e08-9f19-dcec0cbceadd-00000.parquet | Bin 0 -> 1168 bytes
...32a29-fd5b-4ba5-a0db-2992c73075c2-00000.parquet | Bin 0 -> 1169 bytes
...20e43-f85a-4609-9ebb-4c76269cddb6-00000.parquet | Bin 0 -> 1169 bytes
...7d440-44f5-4a11-bbfa-c72c520ca997-00000.parquet | Bin 0 -> 1190 bytes
...2540c-b002-413c-b422-7ec3c06d404e-00000.parquet | Bin 0 -> 1190 bytes
...8f0e5-7a46-417e-8c0a-50bb00e54a30-00000.parquet | Bin 0 -> 1189 bytes
...4b7c7-164d-4d06-8b4d-24d296906f73-00000.parquet | Bin 0 -> 1190 bytes
...5b9eb-060d-4180-aa1e-2bb69187d8db-00000.parquet | Bin 0 -> 1190 bytes
...b05c6-5407-41fc-a881-60626253ca77-00000.parquet | Bin 0 -> 1190 bytes
.../af797bab-2f2c-44df-a77b-d91c7198fe53-m0.avro | Bin 0 -> 5579 bytes
...308-1-af797bab-2f2c-44df-a77b-d91c7198fe53.avro | Bin 0 -> 2563 bytes
.../iceberg_partitioned/metadata/v1.metadata.json | 62 +++++
.../iceberg_partitioned/metadata/v2.metadata.json | 81 ++++++
.../iceberg_partitioned/metadata/version-hint.text | 1 +
.../functional/functional_schema_template.sql | 26 ++
.../datasets/functional/schema_constraints.csv | 2 +
.../{iceberg_create.test => iceberg-create.test} | 31 +--
.../queries/QueryTest/iceberg-negative.test | 45 ++++
.../queries/QueryTest/iceberg-profile.test | 20 ++
.../queries/QueryTest/iceberg-query.test | 185 +++++++++++++
.../queries/QueryTest/java-udf.test | 23 ++
.../queries/QueryTest/show-create-table.test | 4 +-
.../{custom_cluster => query_test}/test_iceberg.py | 15 +-
tests/query_test/test_scanners.py | 23 ++
102 files changed, 1945 insertions(+), 247 deletions(-)
create mode 100644 be/src/exprs/hive-udf-call-ir.cc
create mode 100644 fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00001-1-5dbd44ad-18bc-40f2-9dd6-aeb2cc23457c-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00003-3-27db2521-1e8b-40c1-b846-552cd620abce-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00004-4-f1b55628-0544-4833-8b11-1b4add53dfd6-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00006-6-f75530ef-93b6-4994-b3c8-db957d44848c-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00007-7-8d9b22da-5f10-4cbf-8e4d-160f829b5e48-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00009-9-f029a1f7-9024-4bc3-a030-e20861586146-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00011-11-f07814ae-56cd-486b-af81-18541437da7d-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00012-12-967c70a4-bf4d-4a82-8c97-c90e2b4d9dcf-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00014-14-d0cdca7f-c050-407e-b70c-2bd076f83e4e-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00015-15-0e931a1f-309e-43b3-a5cf-3ef82fa4a87c-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00017-17-43138078-244c-4b38-8127-04a5bfbc4695-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00019-19-52569895-df25-4ad8-b64d-49c4540d36c9-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00020-20-f160c1ea-a2f5-4109-81ec-3ff9c155430f-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00022-22-c1f61b8c-9d9a-4823-b64e-109770c16696-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00023-23-8cfc9d65-bfc3-47c4-8da7-6610b49b3305-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00025-25-d519b09d-c730-40a6-ad03-8b36096cb234-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00027-27-802b2129-0d18-418a-bcdd-b1b5332ea2a5-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00028-28-70f51a0c-3d30-4cdb-afcb-e8c1cfa3caa4-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00030-30-e887707d-58db-469c-ab96-b77188f25189-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/data/00031-31-facf6a62-c326-44f7-bdc3-b1471bd594d7-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/metadata/9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21-m0.avro
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/metadata/snap-93996984692289973-1-9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21.avro
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/metadata/v1.metadata.json
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/metadata/v2.metadata.json
create mode 100644 testdata/data/iceberg_test/iceberg_non_partitioned/metadata/version-hint.text
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00001-1-b975a171-0911-47c2-90c8-300f23c28772-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00006-6-9bc6f547-de52-4aad-8c07-5d0f1aaa0558-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00009-9-bf3024d6-8b55-42d3-b25e-5f2a31303026-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00017-17-a0d23269-c8d7-40cf-94c8-8b624969b9c0-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00023-23-138a376e-696e-4da3-95f3-7e0aacb11ca3-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00027-27-04494175-75de-41ef-b824-3923ebfba0cb-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00030-30-eba34050-b263-41e8-bc7d-773a47b33c32-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00031-31-0ec1c3ff-7154-4fd1-a090-8fa0b9c2d3c8-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00004-4-8b99da05-6be1-42e1-90e0-bff8ca08b5af-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00014-14-5b39012d-85e3-4144-ac83-574ed5046718-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00015-15-e37e3f98-5f3b-4245-9c01-76cf0f1dd21a-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00019-19-e36a1955-f01c-4e08-9f19-dcec0cbceadd-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00020-20-c9632a29-fd5b-4ba5-a0db-2992c73075c2-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00028-28-31220e43-f85a-4609-9ebb-4c76269cddb6-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00003-3-7e47d440-44f5-4a11-bbfa-c72c520ca997-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00007-7-c902540c-b002-413c-b422-7ec3c06d404e-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00011-11-5558f0e5-7a46-417e-8c0a-50bb00e54a30-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00012-12-ea94b7c7-164d-4d06-8b4d-24d296906f73-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00022-22-0565b9eb-060d-4180-aa1e-2bb69187d8db-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00025-25-1f2b05c6-5407-41fc-a881-60626253ca77-00000.parquet
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/metadata/af797bab-2f2c-44df-a77b-d91c7198fe53-m0.avro
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/metadata/snap-8270633197658268308-1-af797bab-2f2c-44df-a77b-d91c7198fe53.avro
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/metadata/v1.metadata.json
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/metadata/v2.metadata.json
create mode 100644 testdata/data/iceberg_test/iceberg_partitioned/metadata/version-hint.text
rename testdata/workloads/functional-query/queries/QueryTest/{iceberg_create.test => iceberg-create.test} (77%)
create mode 100644 testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test
create mode 100644 testdata/workloads/functional-query/queries/QueryTest/iceberg-profile.test
create mode 100644 testdata/workloads/functional-query/queries/QueryTest/iceberg-query.test
rename tests/{custom_cluster => query_test}/test_iceberg.py (71%)
[impala] 04/05: IMPALA-10012: ds_hll_sketch() results ascii codec
decoding error fix
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit fe6e6257475b521b3ceeae4f6a898aae6ce3eb94
Author: Adam Tamas <ta...@cloudera.com>
AuthorDate: Thu Jul 30 10:10:25 2020 +0200
IMPALA-10012: ds_hll_sketch() results ascii codec decoding error fix
While the ds_hll_sketch() generates a string value as output the data
is not an ascii encoded text but a bitsketch, because of this, when
the shell get this data it disconnect while it tries to decode it.
The issue can be reproduced with a simple method like using unhex
with a wrong input.
Example: SELECT unhex("aa");
This patch contains a solution, where we replace any not UTF-8
decodable characters if we run into an UnicodeDecodeError after
fetching it.
This solution is working with the Thrift 0.9.3 autogenerated gen-py
but still fails with Thrift 0.11.0.
For Thrift 0.11.0 the error is catched and an error message is sent
(not working with beeswax protocol, because it generates a different
error (TypeError) which can come for other reasons too).
Testing:
-manual testing with these protocols: 'hs2-http', 'hs2', 'beeswax'
Change-Id: I0c5f1290356e21aed8ca7f896f953541942aed05
Reviewed-on: http://gerrit.cloudera.org:8080/16418
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Gabor Kaszab <ga...@cloudera.com>
---
shell/impala_client.py | 10 +++++++++-
shell/impala_shell.py | 11 ++++++++++-
shell/shell_output.py | 21 ++++++++++++++++++---
3 files changed, 37 insertions(+), 5 deletions(-)
diff --git a/shell/impala_client.py b/shell/impala_client.py
index cb7cb39..19a570f 100755
--- a/shell/impala_client.py
+++ b/shell/impala_client.py
@@ -1095,7 +1095,15 @@ class ImpalaBeeswaxClient(ImpalaClient):
self.fetch_size))
if rpc_status != RpcStatus.OK:
raise RPCException()
- yield [row.split('\t') for row in result.data]
+
+ def split_row_and_decode_if_needed(row):
+ try:
+ return row.split('\t')
+ except UnicodeDecodeError:
+ return row.decode('utf-8', 'replace').split('\t')
+
+ yield [split_row_and_decode_if_needed(row) for row in result.data]
+
if not result.has_more:
return
diff --git a/shell/impala_shell.py b/shell/impala_shell.py
index 8c18408..6cb9aa7 100755
--- a/shell/impala_shell.py
+++ b/shell/impala_shell.py
@@ -1221,6 +1221,15 @@ class ImpalaShell(cmd.Cmd, object):
except RPCException as e:
# could not complete the rpc successfully
print(e, file=sys.stderr)
+ except UnicodeDecodeError as e:
+ # An error occoured possibly during the fetching.
+ # Depending of which protocol is at use it can come from different places.
+ # Possibly occours because we try to display binary data which contains
+ # undecodable elements.
+ if self.last_query_handle is not None:
+ self.imp_client.close_query(self.last_query_handle)
+ print('UnicodeDecodeError : %s \nPlease check for columns containing binary data '
+ 'to find the possible source of the error.' % (e,), file=sys.stderr)
except QueryStateException as e:
# an exception occurred while executing the query
if self.last_query_handle is not None:
@@ -1348,7 +1357,7 @@ class ImpalaShell(cmd.Cmd, object):
if self.readline and self.readline.get_current_history_length() > 0:
for index in xrange(1, self.readline.get_current_history_length() + 1):
cmd = self.readline.get_history_item(index)
- print('[%d]: %s' % (index, cmd), file=sys.stderr)
+ print('[%d]: %s' % (index, cmd.decode('utf-8', 'replace')), file=sys.stderr)
else:
print(READLINE_UNAVAILABLE_ERROR, file=sys.stderr)
diff --git a/shell/shell_output.py b/shell/shell_output.py
index c6c2e99..31d91a0 100644
--- a/shell/shell_output.py
+++ b/shell/shell_output.py
@@ -35,11 +35,25 @@ class PrettyOutputFormatter(object):
def format(self, rows):
"""Returns string containing representation of the table data."""
+
+ def decode_if_needed(row):
+ # Checking if the values in row is decodable. If it is we just give back the row
+ # as it is, if not we generate a new row and give that back instead where the
+ # undecodable parts are swapped out.
+ try:
+ ''.join(str(row))
+ return row
+ except UnicodeDecodeError:
+ new_row = []
+ for entry in row:
+ new_row.append(entry.decode('UTF-8', 'replace'))
+ return new_row
+
# Clear rows that already exist in the table.
self.prettytable.clear_rows()
try:
for row in rows:
- self.prettytable.add_row(row)
+ self.prettytable.add_row(decode_if_needed(row))
return self.prettytable.get_string()
except Exception as e:
# beeswax returns each row as a tab separated string. If a string column
@@ -49,7 +63,8 @@ class PrettyOutputFormatter(object):
"embedded tabs. Reverting to tab delimited text output")
print(error_msg, file=sys.stderr)
print('{0}: {1}'.format(type(e), str(e)), file=sys.stderr)
- return '\n'.join(['\t'.join(row) for row in rows])
+
+ return '\n'.join(['\t'.join(decode_if_needed(row)) for row in rows])
class DelimitedOutputFormatter(object):
@@ -77,7 +92,7 @@ class DelimitedOutputFormatter(object):
lineterminator='\n', quoting=csv.QUOTE_MINIMAL)
for row in rows:
if sys.version_info.major == 2:
- row = [val.encode('utf-8') for val in row]
+ row = [val.encode('utf-8', 'replace') for val in row]
writer.writerow(row)
rows = temp_buffer.getvalue().rstrip()
temp_buffer.close()
[impala] 02/05: IMPALA-10012: ds_hll_sketch() results ascii codec
decoding error fix
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 75146c9138051758176d3d4482f45a9d1056127b
Author: Adam Tamas <ta...@cloudera.com>
AuthorDate: Thu Jul 30 10:10:25 2020 +0200
IMPALA-10012: ds_hll_sketch() results ascii codec decoding error fix
While the ds_hll_sketch() generates a string value as output the data
is not an ascii encoded text but a bitsketch, because of this, when
the shell get this data it disconnect while it tries to decode it.
The issue can be reproduced with a simple method like using unhex
with a wrong input.
Example: SELECT unhex("aa");
This patch contains a solution, where we replace any not UTF-8
decodable characters if we run into an UnicodeDecodeError after
fetching it.
This solution is working with the Thrift 0.9.3 autogenerated gen-py
but still fails with Thrift 0.11.0.
For Thrift 0.11.0 the error is catched and an error message is sent
(not working with beeswax protocol, because it generates a different
error (TypeError) which can come for other reasons too).
Testing:
-manual testing with these protocols: 'hs2-http', 'hs2', 'beeswax'
Change-Id: Ic5cfb907871ca83e5f04a39ca9d7a8e138d711a8
Reviewed-on: http://gerrit.cloudera.org:8080/16305
Reviewed-by: Csaba Ringhofer <cs...@cloudera.com>
Tested-by: Csaba Ringhofer <cs...@cloudera.com>
---
shell/impala_client.py | 10 +++++++++-
shell/impala_shell.py | 11 ++++++++++-
shell/shell_output.py | 21 ++++++++++++++++++---
3 files changed, 37 insertions(+), 5 deletions(-)
diff --git a/shell/impala_client.py b/shell/impala_client.py
index cb7cb39..19a570f 100755
--- a/shell/impala_client.py
+++ b/shell/impala_client.py
@@ -1095,7 +1095,15 @@ class ImpalaBeeswaxClient(ImpalaClient):
self.fetch_size))
if rpc_status != RpcStatus.OK:
raise RPCException()
- yield [row.split('\t') for row in result.data]
+
+ def split_row_and_decode_if_needed(row):
+ try:
+ return row.split('\t')
+ except UnicodeDecodeError:
+ return row.decode('utf-8', 'replace').split('\t')
+
+ yield [split_row_and_decode_if_needed(row) for row in result.data]
+
if not result.has_more:
return
diff --git a/shell/impala_shell.py b/shell/impala_shell.py
index 8c18408..6cb9aa7 100755
--- a/shell/impala_shell.py
+++ b/shell/impala_shell.py
@@ -1221,6 +1221,15 @@ class ImpalaShell(cmd.Cmd, object):
except RPCException as e:
# could not complete the rpc successfully
print(e, file=sys.stderr)
+ except UnicodeDecodeError as e:
+ # An error occoured possibly during the fetching.
+ # Depending of which protocol is at use it can come from different places.
+ # Possibly occours because we try to display binary data which contains
+ # undecodable elements.
+ if self.last_query_handle is not None:
+ self.imp_client.close_query(self.last_query_handle)
+ print('UnicodeDecodeError : %s \nPlease check for columns containing binary data '
+ 'to find the possible source of the error.' % (e,), file=sys.stderr)
except QueryStateException as e:
# an exception occurred while executing the query
if self.last_query_handle is not None:
@@ -1348,7 +1357,7 @@ class ImpalaShell(cmd.Cmd, object):
if self.readline and self.readline.get_current_history_length() > 0:
for index in xrange(1, self.readline.get_current_history_length() + 1):
cmd = self.readline.get_history_item(index)
- print('[%d]: %s' % (index, cmd), file=sys.stderr)
+ print('[%d]: %s' % (index, cmd.decode('utf-8', 'replace')), file=sys.stderr)
else:
print(READLINE_UNAVAILABLE_ERROR, file=sys.stderr)
diff --git a/shell/shell_output.py b/shell/shell_output.py
index c6c2e99..31d91a0 100644
--- a/shell/shell_output.py
+++ b/shell/shell_output.py
@@ -35,11 +35,25 @@ class PrettyOutputFormatter(object):
def format(self, rows):
"""Returns string containing representation of the table data."""
+
+ def decode_if_needed(row):
+ # Checking if the values in row is decodable. If it is we just give back the row
+ # as it is, if not we generate a new row and give that back instead where the
+ # undecodable parts are swapped out.
+ try:
+ ''.join(str(row))
+ return row
+ except UnicodeDecodeError:
+ new_row = []
+ for entry in row:
+ new_row.append(entry.decode('UTF-8', 'replace'))
+ return new_row
+
# Clear rows that already exist in the table.
self.prettytable.clear_rows()
try:
for row in rows:
- self.prettytable.add_row(row)
+ self.prettytable.add_row(decode_if_needed(row))
return self.prettytable.get_string()
except Exception as e:
# beeswax returns each row as a tab separated string. If a string column
@@ -49,7 +63,8 @@ class PrettyOutputFormatter(object):
"embedded tabs. Reverting to tab delimited text output")
print(error_msg, file=sys.stderr)
print('{0}: {1}'.format(type(e), str(e)), file=sys.stderr)
- return '\n'.join(['\t'.join(row) for row in rows])
+
+ return '\n'.join(['\t'.join(decode_if_needed(row)) for row in rows])
class DelimitedOutputFormatter(object):
@@ -77,7 +92,7 @@ class DelimitedOutputFormatter(object):
lineterminator='\n', quoting=csv.QUOTE_MINIMAL)
for row in rows:
if sys.version_info.major == 2:
- row = [val.encode('utf-8') for val in row]
+ row = [val.encode('utf-8', 'replace') for val in row]
writer.writerow(row)
rows = temp_buffer.getvalue().rstrip()
temp_buffer.close()
[impala] 01/05: IMPALA-7658: Proper codegen for HiveUdfCall
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit dc08b657e86c3bd81af27c7acf4a31b39b5b20e9
Author: Daniel Becker <da...@cloudera.com>
AuthorDate: Tue Aug 4 19:11:03 2020 +0200
IMPALA-7658: Proper codegen for HiveUdfCall
Implementing codegen for HiveUdfCall.
Testing:
Verified that java udf tests pass locally.
Benchmarks:
Used a UDF from TestUdf.java that adds three integers:
create function tpch15_parquet.sum3(int, int, int) returns int
location '/test-warehouse/impala-hive-udfs.jar'
symbol='org.apache.impala.TestUdf';
Used the following query on the master branch and the change's branch:
set num_nodes=1; set mt_dop=1;
select min(tpch15_parquet.sum3(cast(l_orderkey as int),
cast(l_partkey as int), cast(l_suppkey as int)))
from tpch15_parquet.lineitem;
Results averaged over 100 runs after warmup:
Master: 20.6346s, stddev: 0.3132411856765332
This change: 19.0256s, stddev: 0.42039019873436
This is a ~7.8% improvement.
Change-Id: I2f994dac550f297ed3c88491816403f237d4d747
Reviewed-on: http://gerrit.cloudera.org:8080/16314
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/codegen/codegen-util.h | 28 ++
be/src/codegen/gen_ir_descriptions.py | 12 +-
be/src/codegen/impala-ir.cc | 1 +
be/src/codegen/llvm-codegen.cc | 3 +-
be/src/codegen/llvm-codegen.h | 5 +-
be/src/exprs/CMakeLists.txt | 1 +
be/src/exprs/hive-udf-call-ir.cc | 92 +++++++
be/src/exprs/hive-udf-call.cc | 286 ++++++++++++++++++---
be/src/exprs/hive-udf-call.h | 42 +++
.../queries/QueryTest/java-udf.test | 23 ++
10 files changed, 454 insertions(+), 39 deletions(-)
diff --git a/be/src/codegen/codegen-util.h b/be/src/codegen/codegen-util.h
index 0a2199b..d85c5f2 100644
--- a/be/src/codegen/codegen-util.h
+++ b/be/src/codegen/codegen-util.h
@@ -22,6 +22,8 @@
#include <llvm/IR/IRBuilder.h>
#include "codegen/llvm-codegen.h"
+#include "runtime/string-value.h"
+#include "runtime/timestamp-value.h"
namespace impala {
@@ -59,6 +61,32 @@ class CodeGenUtil {
return str;
}
+ /// Return the byte size of the given PrimitiveType.
+ static int64_t GetTypeSize(const PrimitiveType type) {
+ switch (type) {
+ case TYPE_BOOLEAN:
+ case TYPE_TINYINT:
+ return 1;
+ case TYPE_SMALLINT:
+ return 2;
+ case TYPE_INT:
+ case TYPE_FLOAT:
+ case TYPE_DATE:
+ return 4;
+ case TYPE_BIGINT:
+ case TYPE_DOUBLE:
+ return 8;
+ case TYPE_TIMESTAMP:
+ return sizeof(TimestampValue);
+ case TYPE_STRING:
+ case TYPE_VARCHAR:
+ return sizeof(StringValue);
+ default:
+ DCHECK(false) << "NYI";
+ return 0;
+ }
+ }
+
};
}
diff --git a/be/src/codegen/gen_ir_descriptions.py b/be/src/codegen/gen_ir_descriptions.py
index 8da5569..6332dae 100755
--- a/be/src/codegen/gen_ir_descriptions.py
+++ b/be/src/codegen/gen_ir_descriptions.py
@@ -238,7 +238,17 @@ ir_functions = [
["KRPC_DSS_GET_PART_EXPR_EVAL",
"_ZN6impala20KrpcDataStreamSender25GetPartitionExprEvaluatorEi"],
["KRPC_DSS_HASH_AND_ADD_ROWS",
- "_ZN6impala20KrpcDataStreamSender14HashAndAddRowsEPNS_8RowBatchE"]
+ "_ZN6impala20KrpcDataStreamSender14HashAndAddRowsEPNS_8RowBatchE"],
+ ["GET_FUNCTION_CTX",
+ "_ZN6impala11HiveUdfCall18GetFunctionContextEPNS_19ScalarExprEvaluatorEi"],
+ ["GET_JNI_CONTEXT",
+ "_ZN6impala11HiveUdfCall13GetJniContextEPN10impala_udf15FunctionContextE"],
+ ["JNI_CTX_SET_INPUT_NULL_BUFF_ELEM",
+ "_ZN6impala11HiveUdfCall10JniContext26SetInputNullsBufferElementEPS1_ih"],
+ ["JNI_CTX_INPUT_VAL_BUFF_AT_OFFSET",
+ "_ZN6impala11HiveUdfCall10JniContext28GetInputValuesBufferAtOffsetEPS1_i"],
+ ["HIVE_UDF_CALL_CALL_JAVA",
+ "_ZN6impala11HiveUdfCall22CallJavaAndStoreResultEPKNS_10ColumnTypeEPN10impala_udf15FunctionContextEPNS0_10JniContextE"],
]
enums_preamble = '\
diff --git a/be/src/codegen/impala-ir.cc b/be/src/codegen/impala-ir.cc
index c5bc441..e5df8cf 100644
--- a/be/src/codegen/impala-ir.cc
+++ b/be/src/codegen/impala-ir.cc
@@ -47,6 +47,7 @@
#include "exprs/date-functions-ir.cc"
#include "exprs/decimal-functions-ir.cc"
#include "exprs/decimal-operators-ir.cc"
+#include "exprs/hive-udf-call-ir.cc"
#include "exprs/in-predicate-ir.cc"
#include "exprs/is-null-predicate-ir.cc"
#include "exprs/like-predicate-ir.cc"
diff --git a/be/src/codegen/llvm-codegen.cc b/be/src/codegen/llvm-codegen.cc
index e36c22e..4e1f028 100644
--- a/be/src/codegen/llvm-codegen.cc
+++ b/be/src/codegen/llvm-codegen.cc
@@ -618,7 +618,8 @@ llvm::Constant* LlvmCodeGen::GetIntConstant(
return llvm::ConstantInt::get(context(), llvm::APInt(8 * num_bytes, vals));
}
-llvm::Value* LlvmCodeGen::GetStringConstant(LlvmBuilder* builder, char* data, int len) {
+llvm::Value* LlvmCodeGen::GetStringConstant(
+ LlvmBuilder* builder, const char* data, int len) {
// Create a global string with private linkage.
llvm::Constant* const_string =
llvm::ConstantDataArray::getString(context(), llvm::StringRef(data, len), false);
diff --git a/be/src/codegen/llvm-codegen.h b/be/src/codegen/llvm-codegen.h
index 9b3245b..e93ceb6 100644
--- a/be/src/codegen/llvm-codegen.h
+++ b/be/src/codegen/llvm-codegen.h
@@ -531,7 +531,10 @@ class LlvmCodeGen {
llvm::Constant* GetIntConstant(int byte_size, uint64_t low_bits, uint64_t high_bits);
/// Initialise a constant global string and returns an i8* pointer to it.
- llvm::Value* GetStringConstant(LlvmBuilder* builder, char* data, int len);
+ llvm::Value* GetStringConstant(LlvmBuilder* builder, const char* data, int len);
+ llvm::Value* GetStringConstant(LlvmBuilder* builder, const std::string& str) {
+ return GetStringConstant(builder, str.c_str(), str.size());
+ }
/// Returns true/false constants (bool type)
llvm::Constant* true_value() { return true_value_; }
diff --git a/be/src/exprs/CMakeLists.txt b/be/src/exprs/CMakeLists.txt
index 38de9ef..5dc2ba2 100644
--- a/be/src/exprs/CMakeLists.txt
+++ b/be/src/exprs/CMakeLists.txt
@@ -43,6 +43,7 @@ add_library(Exprs
decimal-operators-ir.cc
expr.cc
hive-udf-call.cc
+ hive-udf-call-ir.cc
in-predicate-ir.cc
is-not-empty-predicate.cc
is-null-predicate-ir.cc
diff --git a/be/src/exprs/hive-udf-call-ir.cc b/be/src/exprs/hive-udf-call-ir.cc
new file mode 100644
index 0000000..4b69233
--- /dev/null
+++ b/be/src/exprs/hive-udf-call-ir.cc
@@ -0,0 +1,92 @@
+// 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.
+
+#include "exprs/hive-udf-call.h"
+
+#include <sstream>
+
+#include "exprs/anyval-util.h"
+#include "exprs/scalar-expr-evaluator.h"
+#include "runtime/string-value.h"
+#include "util/jni-util.h"
+
+namespace impala {
+
+FunctionContext* HiveUdfCall::GetFunctionContext(
+ ScalarExprEvaluator* eval, int fn_ctx_idx) {
+ return eval->fn_context(fn_ctx_idx);
+}
+
+HiveUdfCall::JniContext* HiveUdfCall::GetJniContext(FunctionContext* fn_ctx) {
+ JniContext* jni_ctx = reinterpret_cast<JniContext*>(
+ fn_ctx->GetFunctionState(FunctionContext::THREAD_LOCAL));
+ DCHECK(jni_ctx != nullptr);
+ return jni_ctx;
+}
+
+void HiveUdfCall::JniContext::SetInputNullsBufferElement(
+ JniContext* jni_ctx, int index, uint8_t value) {
+ jni_ctx->input_nulls_buffer[index] = value;
+}
+
+uint8_t* HiveUdfCall::JniContext::GetInputValuesBufferAtOffset(
+ JniContext* jni_ctx, int offset) {
+ return &jni_ctx->input_values_buffer[offset];
+}
+
+AnyVal* HiveUdfCall::CallJavaAndStoreResult(const ColumnType* type,
+ FunctionContext* fn_ctx, JniContext* jni_ctx) {
+ DCHECK(jni_ctx != nullptr);
+ JNIEnv* env = jni_ctx->jni_env;
+ DCHECK(env != nullptr);
+
+ // Using this version of Call has the lowest overhead. This eliminates the
+ // vtable lookup and setting up return stacks.
+ env->CallNonvirtualVoidMethodA(
+ jni_ctx->executor, executor_cl_, executor_evaluate_id_, nullptr);
+ Status status = JniUtil::GetJniExceptionMsg(env);
+ if (!status.ok()) {
+ if (!jni_ctx->warning_logged) {
+ std::stringstream ss;
+ ss << "Hive UDF path=" << jni_ctx->hdfs_location << " class="
+ << jni_ctx->scalar_fn_symbol << " failed due to: " << status.GetDetail();
+ fn_ctx->AddWarning(ss.str().c_str());
+ jni_ctx->warning_logged = true;
+ }
+ jni_ctx->output_anyval->is_null = true;
+ return jni_ctx->output_anyval;
+ }
+
+ // Write output_value_buffer to output_anyval
+ if (jni_ctx->output_null_value) {
+ jni_ctx->output_anyval->is_null = true;
+ } else {
+ AnyValUtil::SetAnyVal(jni_ctx->output_value_buffer, *type, jni_ctx->output_anyval);
+
+ if (type->type == TYPE_STRING) {
+ // Copy the string into a result allocation with the usual lifetime for expr
+ // results. Needed because the UDF output buffer is owned by the Java UDF executor
+ // and may be freed or reused by the next call into the Java UDF executor.
+ StringVal* str_val = static_cast<StringVal*>(jni_ctx->output_anyval);
+ const StringVal temp = StringVal::CopyFrom(fn_ctx, str_val->ptr, str_val->len);
+ *str_val = temp;
+ }
+ }
+ return jni_ctx->output_anyval;
+}
+
+} // namespace impala
diff --git a/be/src/exprs/hive-udf-call.cc b/be/src/exprs/hive-udf-call.cc
index 8f8399c..b903891 100644
--- a/be/src/exprs/hive-udf-call.cc
+++ b/be/src/exprs/hive-udf-call.cc
@@ -21,6 +21,9 @@
#include <sstream>
#include <string>
+#include "codegen/codegen-anyval.h"
+#include "codegen/codegen-util.h"
+#include "codegen/llvm-codegen.h"
#include "exprs/anyval-util.h"
#include "exprs/scalar-expr-evaluator.h"
#include "rpc/jni-thrift-util.h"
@@ -46,28 +49,6 @@ jmethodID HiveUdfCall::executor_ctor_id_ = NULL;
jmethodID HiveUdfCall::executor_evaluate_id_ = NULL;
jmethodID HiveUdfCall::executor_close_id_ = NULL;
-struct JniContext {
- jobject executor;
-
- uint8_t* input_values_buffer;
- uint8_t* input_nulls_buffer;
- uint8_t* output_value_buffer;
- uint8_t output_null_value;
- bool warning_logged;
-
- /// AnyVal to evaluate the expression into. Only used as temporary storage during
- /// expression evaluation.
- AnyVal* output_anyval;
-
- JniContext()
- : executor(NULL),
- input_values_buffer(NULL),
- input_nulls_buffer(NULL),
- output_value_buffer(NULL),
- warning_logged(false),
- output_anyval(NULL) {}
-};
-
HiveUdfCall::HiveUdfCall(const TExprNode& node)
: ScalarExpr(node), input_buffer_size_(0) {
DCHECK_EQ(node.node_type, TExprNodeType::FUNCTION_CALL);
@@ -79,23 +60,16 @@ AnyVal* HiveUdfCall::Evaluate(ScalarExprEvaluator* eval, const TupleRow* row) co
FunctionContext* fn_ctx = eval->fn_context(fn_ctx_idx_);
JniContext* jni_ctx = reinterpret_cast<JniContext*>(
fn_ctx->GetFunctionState(FunctionContext::THREAD_LOCAL));
- DCHECK(jni_ctx != NULL);
+ DCHECK(jni_ctx != nullptr);
- JNIEnv* env = JniUtil::GetJNIEnv();
- if (env == NULL) {
- stringstream ss;
- ss << "Hive UDF path=" << fn_.hdfs_location << " class=" << fn_.scalar_fn.symbol
- << " failed due to JNI issue getting the JNIEnv object";
- fn_ctx->SetError(ss.str().c_str());
- jni_ctx->output_anyval->is_null = true;
- return jni_ctx->output_anyval;
- }
+ JNIEnv* env = jni_ctx->jni_env;
+ DCHECK(env != nullptr);
// Evaluate all the children values and put the results in input_values_buffer
for (int i = 0; i < GetNumChildren(); ++i) {
void* v = eval->GetValue(*GetChild(i), row);
- if (v == NULL) {
+ if (v == nullptr) {
jni_ctx->input_nulls_buffer[i] = 1;
} else {
uint8_t* input_ptr = jni_ctx->input_values_buffer + input_byte_offsets_[i];
@@ -134,7 +108,7 @@ AnyVal* HiveUdfCall::Evaluate(ScalarExprEvaluator* eval, const TupleRow* row) co
// Using this version of Call has the lowest overhead. This eliminates the
// vtable lookup and setting up return stacks.
env->CallNonvirtualVoidMethodA(
- jni_ctx->executor, executor_cl_, executor_evaluate_id_, NULL);
+ jni_ctx->executor, executor_cl_, executor_evaluate_id_, nullptr);
Status status = JniUtil::GetJniExceptionMsg(env);
if (!status.ok()) {
if (!jni_ctx->warning_logged) {
@@ -201,7 +175,17 @@ Status HiveUdfCall::OpenEvaluator(FunctionContext::FunctionStateScope scope,
fn_ctx->SetFunctionState(FunctionContext::THREAD_LOCAL, jni_ctx);
JNIEnv* env = JniUtil::GetJNIEnv();
- if (env == NULL) return Status("Failed to get/create JVM");
+ if (env == nullptr) return Status("Failed to get/create JVM");
+
+ // Cache the JNI env pointer here so it can be retrieved by codegen at execution time.
+ // LLVM can't handle thread-local variables (yet) so we cannot call JniUtil::GetJNIEnv()
+ // from codegen code.
+ jni_ctx->jni_env = env;
+
+ // Fields used for error reporting.
+ // This object and thus fn_ are alive when rows are evaluated.
+ jni_ctx->hdfs_location = fn_.hdfs_location.c_str();
+ jni_ctx->scalar_fn_symbol = fn_.scalar_fn.symbol.c_str();
// Add a scoped cleanup jni reference object. This cleans up local refs made below.
JniLocalFrame jni_frame;
@@ -280,8 +264,238 @@ void HiveUdfCall::CloseEvaluator(FunctionContext::FunctionStateScope scope,
ScalarExpr::CloseEvaluator(scope, state, eval);
}
+Status HiveUdfCall::CodegenEvalChildren(LlvmCodeGen* codegen, LlvmBuilder* builder,
+ llvm::Function* function, llvm::Value* (*args)[2], llvm::Value* jni_ctx,
+ llvm::BasicBlock* const first_block, llvm::BasicBlock** next_block) {
+
+ llvm::Function* const set_input_null_buff_elem_fn =
+ codegen->GetFunction(IRFunction::JNI_CTX_SET_INPUT_NULL_BUFF_ELEM, false);
+ llvm::Function* const get_input_val_buff_at_offset_fn =
+ codegen->GetFunction(IRFunction::JNI_CTX_INPUT_VAL_BUFF_AT_OFFSET, false);
+
+ llvm::LLVMContext& context = codegen->context();
+ llvm::BasicBlock* current_eval_child_block = first_block;
+ const int num_children = GetNumChildren();
+ for (int i = 0; i < num_children; ++i) {
+ ScalarExpr* const child_expr = GetChild(i);
+ llvm::Function* child_fn = nullptr;
+ RETURN_IF_ERROR(child_expr->GetCodegendComputeFn(codegen, false, &child_fn));
+
+ builder->SetInsertPoint(current_eval_child_block);
+ llvm::BasicBlock* next_eval_child_block = llvm::BasicBlock::Create(
+ context, "eval_child", function);
+
+ const ColumnType& child_type = child_expr->type();
+ CodegenAnyVal child_wrapped = CodegenAnyVal::CreateCallWrapped(
+ codegen, builder, child_type, child_fn, *args, "child");
+
+ llvm::BasicBlock* const child_not_null_block = llvm::BasicBlock::Create(
+ context, "child_not_null", function, next_eval_child_block);
+
+ llvm::Value* const child_is_null = child_wrapped.GetIsNull("child_is_null");
+ llvm::Value* const child_is_null_i8 = builder->CreateZExtOrTrunc(
+ child_is_null, codegen->i8_type(), "child_is_null_i8");
+ builder->CreateCall(set_input_null_buff_elem_fn,
+ {jni_ctx, codegen->GetI32Constant(i), child_is_null_i8});
+ builder->CreateCondBr(child_is_null, next_eval_child_block, child_not_null_block);
+
+ // Child is not null.
+ builder->SetInsertPoint(child_not_null_block);
+ llvm::Value* const input_ptr = builder->CreateCall(get_input_val_buff_at_offset_fn,
+ {jni_ctx, codegen->GetI32Constant(input_byte_offsets_[i])}, "input_ptr");
+
+ llvm::Value* const child_val_ptr = child_wrapped.ToNativePtr();
+ const std::size_t size = CodeGenUtil::GetTypeSize(child_type.type);
+ codegen->CodegenMemcpy(builder, input_ptr, child_val_ptr, size);
+ builder->CreateBr(next_eval_child_block);
+ current_eval_child_block = next_eval_child_block;
+ }
+
+ *next_block = current_eval_child_block;
+ return Status::OK();
+}
+
+llvm::Value* CastPtrAndLoad(LlvmCodeGen* codegen, LlvmBuilder* builder,
+ const ColumnType& type, llvm::Value* ptr, const std::string& name) {
+ llvm::PointerType* const ptr_type =
+ CodegenAnyVal::GetLoweredType(codegen, type)->getPointerTo();
+ llvm::Value* const ptr_cast =
+ builder->CreateBitCast(ptr, ptr_type, name + "_ptr_cast");
+ return builder->CreateLoad(ptr_cast, name);
+}
+
+/// Sample IR for calling the following Java function:
+/// public String evaluate(String a, String b, String c) {
+/// if (a == null || b == null || c == null) return null;
+/// return a + b + c;
+/// }
+///
+/// define { i64, i8* } @HiveUdfCall(%"class.impala::ScalarExprEvaluator"* %eval,
+/// %"class.impala::TupleRow"* %row) #46 {
+/// entry:
+/// %0 = alloca %"struct.impala::ColumnType"
+/// %1 = alloca %"struct.impala::StringValue"
+/// %2 = alloca %"struct.impala::StringValue"
+/// %3 = alloca %"struct.impala::StringValue"
+/// %fn_ctx = call %"class.impala_udf::FunctionContext"*
+/// @_ZN6impala11HiveUdfCall18GetFunctionContextEPNS_19ScalarExprEvaluatorEi(
+/// %"class.impala::ScalarExprEvaluator"* %eval, i32 0)
+/// %jni_ctx = call %"struct.impala::HiveUdfCall::JniContext"*
+/// @_ZN6impala11HiveUdfCall13GetJniContextEPN10impala_udf15FunctionContextE(
+/// %"class.impala_udf::FunctionContext"* %fn_ctx)
+/// br label %eval_child
+///
+/// eval_child: ; preds = %entry
+/// %child = call { i64, i8* } @GetSlotRef(%"class.impala::ScalarExprEvaluator"* %eval,
+/// %"class.impala::TupleRow"* %row)
+/// %4 = extractvalue { i64, i8* } %child, 0
+/// %child_is_null = trunc i64 %4 to i1
+/// %child_is_null_i8 = zext i1 %child_is_null to i8
+/// call void @_ZN6impala11HiveUdfCall10JniContext26SetInputNullsBufferElementEPS1_ih(
+/// %"struct.impala::HiveUdfCall::JniContext"* %jni_ctx,
+/// i32 0, i8 %child_is_null_i8)
+/// br i1 %child_is_null, label %eval_child1, label %child_not_null
+///
+/// child_not_null: ; preds = %eval_child
+/// %input_ptr = call i8*
+/// @_ZN6impala11HiveUdfCall10JniContext28GetInputValuesBufferAtOffsetEPS1_i(
+/// %"struct.impala::HiveUdfCall::JniContext"* %jni_ctx, i32 0)
+/// %5 = extractvalue { i64, i8* } %child, 0
+/// %6 = ashr i64 %5, 32
+/// %7 = trunc i64 %6 to i32
+/// %8 = insertvalue %"struct.impala::StringValue" zeroinitializer, i32 %7, 1
+/// %child2 = extractvalue { i64, i8* } %child, 1
+/// %9 = insertvalue %"struct.impala::StringValue" %8, i8* %child2, 0
+/// store %"struct.impala::StringValue" %9, %"struct.impala::StringValue"* %3
+/// %10 = bitcast %"struct.impala::StringValue"* %3 to i8*
+/// call void @llvm.memcpy.p0i8.p0i8.i64(i8* %input_ptr, i8* %10,
+/// i64 12, i32 0, i1 false)
+/// br label %eval_child1
+///
+/// eval_child1: ; preds = %child_not_null, %eval_child
+/// %child4 = call { i64, i8* } @"impala::CastFunctions::CastToStringValWrapper"(
+/// %"class.impala::ScalarExprEvaluator"* %eval, %"class.impala::TupleRow"* %row)
+/// %11 = extractvalue { i64, i8* } %child4, 0
+/// %child_is_null6 = trunc i64 %11 to i1
+/// %child_is_null_i87 = zext i1 %child_is_null6 to i8
+/// call void @_ZN6impala11HiveUdfCall10JniContext26SetInputNullsBufferElementEPS1_ih(
+/// %"struct.impala::HiveUdfCall::JniContext"* %jni_ctx,
+/// i32 1, i8 %child_is_null_i87)
+/// br i1 %child_is_null6, label %eval_child3, label %child_not_null5
+///
+/// child_not_null5: ; preds = %eval_child1
+/// %input_ptr8 = call i8*
+/// @_ZN6impala11HiveUdfCall10JniContext28GetInputValuesBufferAtOffsetEPS1_i(
+/// %"struct.impala::HiveUdfCall::JniContext"* %jni_ctx, i32 16)
+/// %12 = extractvalue { i64, i8* } %child4, 0
+/// %13 = ashr i64 %12, 32
+/// %14 = trunc i64 %13 to i32
+/// %15 = insertvalue %"struct.impala::StringValue" zeroinitializer, i32 %14, 1
+/// %child9 = extractvalue { i64, i8* } %child4, 1
+/// %16 = insertvalue %"struct.impala::StringValue" %15, i8* %child9, 0
+/// store %"struct.impala::StringValue" %16, %"struct.impala::StringValue"* %2
+/// %17 = bitcast %"struct.impala::StringValue"* %2 to i8*
+/// call void @llvm.memcpy.p0i8.p0i8.i64(i8* %input_ptr8, i8* %17,
+/// i64 12, i32 0, i1 false)
+/// br label %eval_child3
+///
+/// eval_child3: ; preds = %child_not_null5, %eval_child1
+/// %child11 = call { i64, i8* } @"impala::CastFunctions::CastToStringVal.5Wrapper"(
+/// %"class.impala::ScalarExprEvaluator"* %eval, %"class.impala::TupleRow"* %row)
+/// %18 = extractvalue { i64, i8* } %child11, 0
+/// %child_is_null13 = trunc i64 %18 to i1
+/// %child_is_null_i814 = zext i1 %child_is_null13 to i8
+/// call void @_ZN6impala11HiveUdfCall10JniContext26SetInputNullsBufferElementEPS1_ih(
+/// %"struct.impala::HiveUdfCall::JniContext"* %jni_ctx,
+/// i32 2, i8 %child_is_null_i814)
+/// br i1 %child_is_null13, label %call_java, label %child_not_null12
+///
+/// child_not_null12: ; preds = %eval_child3
+/// %input_ptr15 = call i8*
+/// @_ZN6impala11HiveUdfCall10JniContext28GetInputValuesBufferAtOffsetEPS1_i(
+/// %"struct.impala::HiveUdfCall::JniContext"* %jni_ctx, i32 32)
+/// %19 = extractvalue { i64, i8* } %child11, 0
+/// %20 = ashr i64 %19, 32
+/// %21 = trunc i64 %20 to i32
+/// %22 = insertvalue %"struct.impala::StringValue" zeroinitializer, i32 %21, 1
+/// %child16 = extractvalue { i64, i8* } %child11, 1
+/// %23 = insertvalue %"struct.impala::StringValue" %22, i8* %child16, 0
+/// store %"struct.impala::StringValue" %23, %"struct.impala::StringValue"* %1
+/// %24 = bitcast %"struct.impala::StringValue"* %1 to i8*
+/// call void @llvm.memcpy.p0i8.p0i8.i64(i8* %input_ptr15, i8* %24,
+/// i64 12, i32 0, i1 false)
+/// br label %call_java
+///
+/// call_java: ; preds = %child_not_null12, %eval_child3
+/// store %"struct.impala::ColumnType" {
+/// i32 10, i32 -1, i32 -1, i32 -1,
+/// %"class.std::vector.13" zeroinitializer,
+/// %"class.std::vector.18" zeroinitializer },
+/// %"struct.impala::ColumnType"* %0
+/// %ret_ptr = call %"struct.impala_udf::AnyVal"*
+/// ; The next two lines should be one line but the name of the identifier is too long.
+/// @_ZN6impala11HiveUdfCall22CallJavaAndStoreResultEPKNS_10ColumnTypeEPN10
+///impala_udf15FunctionContextEPNS0_10JniContextE(
+/// %"struct.impala::ColumnType"* %0,
+/// %"class.impala_udf::FunctionContext"* %fn_ctx,
+/// %"struct.impala::HiveUdfCall::JniContext"* %jni_ctx)
+/// %ret_ptr_cast = bitcast %"struct.impala_udf::AnyVal"* %ret_ptr to { i64, i8* }*
+/// %ret = load { i64, i8* }, { i64, i8* }* %ret_ptr_cast
+/// ret { i64, i8* } %ret
+/// }
Status HiveUdfCall::GetCodegendComputeFnImpl(LlvmCodeGen* codegen, llvm::Function** fn) {
- return GetCodegendComputeFnWrapper(codegen, fn);
+ // Cross-compiled functions this hand-crafted function will call.
+ llvm::Function* const get_func_ctx_fn =
+ codegen->GetFunction(IRFunction::GET_FUNCTION_CTX, false);
+ llvm::Function* const get_jni_ctx_fn =
+ codegen->GetFunction(IRFunction::GET_JNI_CONTEXT, false);
+ llvm::Function* const call_java_fn =
+ codegen->GetFunction(IRFunction::HIVE_UDF_CALL_CALL_JAVA, false);
+
+ // Function prototype.
+ llvm::LLVMContext& context = codegen->context();
+ LlvmBuilder builder(context);
+
+ llvm::Value* args[2];
+ llvm::Function* const function =
+ CreateIrFunctionPrototype("HiveUdfCall", codegen, &args);
+
+ // Codegen the initialisation of function context etc.
+ llvm::BasicBlock* const entry_block =
+ llvm::BasicBlock::Create(context, "entry", function);
+ builder.SetInsertPoint(entry_block);
+
+ llvm::Value* const fn_ctx = builder.CreateCall(
+ get_func_ctx_fn, {args[0], codegen->GetI32Constant(fn_ctx_idx_)}, "fn_ctx");
+ llvm::Value* const jni_ctx = builder.CreateCall(get_jni_ctx_fn, {fn_ctx}, "jni_ctx");
+
+ // Codegen the evaluation of children.
+ llvm::BasicBlock* const first_eval_child_block =
+ llvm::BasicBlock::Create(context, "eval_child", function);
+ builder.CreateBr(first_eval_child_block);
+
+ llvm::BasicBlock* call_java_block = nullptr;
+ RETURN_IF_ERROR(CodegenEvalChildren(codegen, &builder, function, &args, jni_ctx,
+ first_eval_child_block, &call_java_block));
+
+ // Codegen the call to Java and returning the result.
+ call_java_block->setName("call_java");
+ builder.SetInsertPoint(call_java_block);
+
+ llvm::Value* const ir_type = type().ToIR(codegen);
+ llvm::Value* const ir_type_ptr = codegen->GetPtrTo(&builder, ir_type);
+
+ llvm::Value* const ret_ptr = builder.CreateCall(call_java_fn,
+ {ir_type_ptr, fn_ctx, jni_ctx}, "ret_ptr");
+
+ llvm::Value* ret_val = CastPtrAndLoad(codegen, &builder, type(), ret_ptr, "ret");
+ builder.CreateRet(ret_val);
+
+ *fn = codegen->FinalizeFunction(function);
+ if (UNLIKELY(*fn == nullptr)) {
+ return Status(TErrorCode::IR_VERIFY_FAILED, "HiveUdfCall");
+ }
+ return Status::OK();
}
string HiveUdfCall::DebugString() const {
diff --git a/be/src/exprs/hive-udf-call.h b/be/src/exprs/hive-udf-call.h
index 19109b9..186857b 100644
--- a/be/src/exprs/hive-udf-call.h
+++ b/be/src/exprs/hive-udf-call.h
@@ -41,6 +41,7 @@ using impala_udf::StringVal;
using impala_udf::DecimalVal;
using impala_udf::DateVal;
+class LlvmBuilder;
class RuntimeState;
class ScalarExprEvaluator;
class TExprNode;
@@ -105,6 +106,14 @@ class HiveUdfCall : public ScalarExpr {
/// error.
AnyVal* Evaluate(ScalarExprEvaluator* eval, const TupleRow* row) const;
+ /// Codegens the code that evaluates and stores the children of this expression.
+ /// 'first_block' is the block codegenning the first child should start. 'next_block' is
+ /// an output parameter that will be set to the block where the function should
+ /// continue.
+ Status CodegenEvalChildren(LlvmCodeGen* codegen, LlvmBuilder* builder,
+ llvm::Function* function, llvm::Value* (*args)[2], llvm::Value* jni_ctx,
+ llvm::BasicBlock* const first_block, llvm::BasicBlock** next_block);
+
/// input_byte_offsets_[i] is the byte offset child ith's input argument should
/// be written to.
std::vector<int> input_byte_offsets_;
@@ -118,6 +127,39 @@ class HiveUdfCall : public ScalarExpr {
static jmethodID executor_ctor_id_;
static jmethodID executor_evaluate_id_;
static jmethodID executor_close_id_;
+
+ struct JniContext {
+ JNIEnv* jni_env = nullptr;
+ jobject executor = nullptr;
+
+ uint8_t* input_values_buffer = nullptr;
+ uint8_t* input_nulls_buffer = nullptr;
+ uint8_t* output_value_buffer = nullptr;
+ uint8_t output_null_value;
+ bool warning_logged = false;
+
+ /// Used for logging errors.
+ const char* hdfs_location = nullptr;
+ const char* scalar_fn_symbol = nullptr;
+
+ /// AnyVal to evaluate the expression into. Only used as temporary storage during
+ /// expression evaluation.
+ AnyVal* output_anyval = nullptr;
+
+ /// These functions are cross-compiled to IR and used by codegen.
+ static void SetInputNullsBufferElement(
+ JniContext* jni_ctx, int index, uint8_t value);
+ static uint8_t* GetInputValuesBufferAtOffset(JniContext* jni_ctx, int offset);
+ };
+
+ /// Static helper functions for codegen.
+ static jclass* GetExecutorClass();
+ static jmethodID* GetExecutorEvaluateId();
+ static FunctionContext* GetFunctionContext(ScalarExprEvaluator* eval, int fn_ctx_idx);
+ static JniContext* GetJniContext(FunctionContext* fn_ctx);
+ static JNIEnv* GetJniEnv(JniContext* jni_ctx);
+ static AnyVal* CallJavaAndStoreResult(const ColumnType* type, FunctionContext* fn_ctx,
+ JniContext* jni_ctx);
};
}
diff --git a/testdata/workloads/functional-query/queries/QueryTest/java-udf.test b/testdata/workloads/functional-query/queries/QueryTest/java-udf.test
index 980cf8d..152ba79 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/java-udf.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/java-udf.test
@@ -317,3 +317,26 @@ string
---- RESULTS
'Hello'
====
+---- QUERY
+# In the interpreted code path of HiveUdfCall, we use ScalarExprEvaluator::GetValue which
+# evaluates the child expression and returns a nullptr if it is NULL. But in the case of
+# TYPE_DATE it also returns nullptr if the date is invalid, so the interpreted path would
+# handle invalid dates as nulls. Java UDFs involving Date are not allowed yet, this test
+# is a reminder that we'll have to handle this case in codegen when we add support for Date.
+create function identity(Date) returns Date
+location '$FILESYSTEM_PREFIX/test-warehouse/impala-hive-udfs.jar'
+symbol='org.apache.impala.TestUdf';
+---- CATCH
+AnalysisException: Type DATE is not supported for Java UDFs.
+====
+---- QUERY
+# Timestamp values are not supported in Java UDFs yet and the implementation may have been
+# written without Timestamps in mind. This test is a reminder that we'll need to review
+# and test the implementation of HiveUdfCall (especially codegen) with Timestamp values
+# when we add support for them.
+create function identity(Timestamp) returns Timestamp
+location '$FILESYSTEM_PREFIX/test-warehouse/impala-hive-udfs.jar'
+symbol='org.apache.impala.TestUdf';
+---- CATCH
+AnalysisException: Type TIMESTAMP is not supported for Java UDFs.
+====
[impala] 03/05: Revert "IMPALA-10012: ds_hll_sketch() results ascii
codec decoding error fix"
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit b7965d8240f3a5cd6972351b7fe1116e4a0835c1
Author: Csaba Ringhofer <cs...@cloudera.com>
AuthorDate: Fri Sep 4 12:27:58 2020 +0000
Revert "IMPALA-10012: ds_hll_sketch() results ascii codec decoding error fix"
This reverts commit 75146c9138051758176d3d4482f45a9d1056127b.
Change-Id: I57f790389a8c847877999d2b9b8185939b416c07
Reviewed-on: http://gerrit.cloudera.org:8080/16417
Reviewed-by: Csaba Ringhofer <cs...@cloudera.com>
Tested-by: Csaba Ringhofer <cs...@cloudera.com>
---
shell/impala_client.py | 10 +---------
shell/impala_shell.py | 11 +----------
shell/shell_output.py | 21 +++------------------
3 files changed, 5 insertions(+), 37 deletions(-)
diff --git a/shell/impala_client.py b/shell/impala_client.py
index 19a570f..cb7cb39 100755
--- a/shell/impala_client.py
+++ b/shell/impala_client.py
@@ -1095,15 +1095,7 @@ class ImpalaBeeswaxClient(ImpalaClient):
self.fetch_size))
if rpc_status != RpcStatus.OK:
raise RPCException()
-
- def split_row_and_decode_if_needed(row):
- try:
- return row.split('\t')
- except UnicodeDecodeError:
- return row.decode('utf-8', 'replace').split('\t')
-
- yield [split_row_and_decode_if_needed(row) for row in result.data]
-
+ yield [row.split('\t') for row in result.data]
if not result.has_more:
return
diff --git a/shell/impala_shell.py b/shell/impala_shell.py
index 6cb9aa7..8c18408 100755
--- a/shell/impala_shell.py
+++ b/shell/impala_shell.py
@@ -1221,15 +1221,6 @@ class ImpalaShell(cmd.Cmd, object):
except RPCException as e:
# could not complete the rpc successfully
print(e, file=sys.stderr)
- except UnicodeDecodeError as e:
- # An error occoured possibly during the fetching.
- # Depending of which protocol is at use it can come from different places.
- # Possibly occours because we try to display binary data which contains
- # undecodable elements.
- if self.last_query_handle is not None:
- self.imp_client.close_query(self.last_query_handle)
- print('UnicodeDecodeError : %s \nPlease check for columns containing binary data '
- 'to find the possible source of the error.' % (e,), file=sys.stderr)
except QueryStateException as e:
# an exception occurred while executing the query
if self.last_query_handle is not None:
@@ -1357,7 +1348,7 @@ class ImpalaShell(cmd.Cmd, object):
if self.readline and self.readline.get_current_history_length() > 0:
for index in xrange(1, self.readline.get_current_history_length() + 1):
cmd = self.readline.get_history_item(index)
- print('[%d]: %s' % (index, cmd.decode('utf-8', 'replace')), file=sys.stderr)
+ print('[%d]: %s' % (index, cmd), file=sys.stderr)
else:
print(READLINE_UNAVAILABLE_ERROR, file=sys.stderr)
diff --git a/shell/shell_output.py b/shell/shell_output.py
index 31d91a0..c6c2e99 100644
--- a/shell/shell_output.py
+++ b/shell/shell_output.py
@@ -35,25 +35,11 @@ class PrettyOutputFormatter(object):
def format(self, rows):
"""Returns string containing representation of the table data."""
-
- def decode_if_needed(row):
- # Checking if the values in row is decodable. If it is we just give back the row
- # as it is, if not we generate a new row and give that back instead where the
- # undecodable parts are swapped out.
- try:
- ''.join(str(row))
- return row
- except UnicodeDecodeError:
- new_row = []
- for entry in row:
- new_row.append(entry.decode('UTF-8', 'replace'))
- return new_row
-
# Clear rows that already exist in the table.
self.prettytable.clear_rows()
try:
for row in rows:
- self.prettytable.add_row(decode_if_needed(row))
+ self.prettytable.add_row(row)
return self.prettytable.get_string()
except Exception as e:
# beeswax returns each row as a tab separated string. If a string column
@@ -63,8 +49,7 @@ class PrettyOutputFormatter(object):
"embedded tabs. Reverting to tab delimited text output")
print(error_msg, file=sys.stderr)
print('{0}: {1}'.format(type(e), str(e)), file=sys.stderr)
-
- return '\n'.join(['\t'.join(decode_if_needed(row)) for row in rows])
+ return '\n'.join(['\t'.join(row) for row in rows])
class DelimitedOutputFormatter(object):
@@ -92,7 +77,7 @@ class DelimitedOutputFormatter(object):
lineterminator='\n', quoting=csv.QUOTE_MINIMAL)
for row in rows:
if sys.version_info.major == 2:
- row = [val.encode('utf-8', 'replace') for val in row]
+ row = [val.encode('utf-8') for val in row]
writer.writerow(row)
rows = temp_buffer.getvalue().rstrip()
temp_buffer.close()
[impala] 05/05: IMPALA-9741: Support querying Iceberg table by
impala
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit fb6d96e001c1a04475a8fd01f757dd0605cf3279
Author: skyyws <sk...@163.com>
AuthorDate: Fri Jun 19 09:48:47 2020 +0800
IMPALA-9741: Support querying Iceberg table by impala
This patch mainly realizes the querying of iceberg table through impala,
we can use the following sql to create an external iceberg table:
CREATE EXTERNAL TABLE default.iceberg_test (
level string,
event_time timestamp,
message string,
)
STORED AS ICEBERG
LOCATION 'hdfs://xxx'
TBLPROPERTIES ('iceberg_file_format'='parquet');
Or just including table name and location like this:
CREATE EXTERNAL TABLE default.iceberg_test
STORED AS ICEBERG
LOCATION 'hdfs://xxx'
TBLPROPERTIES ('iceberg_file_format'='parquet');
'iceberg_file_format' is the file format in iceberg, currently only
support PARQUET, other format would be supported in the future. And
if you don't specify this property in your SQL, default file format
is PARQUET.
We achieved this function by treating the iceberg table as normal
unpartitioned hdfs table. When querying iceberg table, we pushdown
partition column predicates to iceberg to decide which data files
need to be scanned, and then transfer this information to BE to
do the real scan operation.
Testing:
- Unit test for Iceberg in FileMetadataLoaderTest
- Create table tests in functional_schema_template.sql
- Iceberg table query test in test_scanners.py
Change-Id: I856cfee4f3397d1a89cf17650e8d4fbfe1f2b006
Reviewed-on: http://gerrit.cloudera.org:8080/16143
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/runtime/descriptors.cc | 1 +
bin/rat_exclude_files.txt | 1 +
common/thrift/CatalogObjects.thrift | 8 +
fe/pom.xml | 20 --
.../org/apache/impala/analysis/AlterTableStmt.java | 5 +
.../java/org/apache/impala/analysis/Analyzer.java | 4 +-
.../apache/impala/analysis/ComputeStatsStmt.java | 5 +
.../apache/impala/analysis/CreateTableStmt.java | 37 ++++
.../impala/analysis/IcebergPartitionField.java | 12 +-
.../impala/analysis/IcebergPartitionSpec.java | 12 +-
.../org/apache/impala/analysis/InsertStmt.java | 7 +-
.../org/apache/impala/analysis/ShowFilesStmt.java | 3 +-
.../org/apache/impala/analysis/ShowStatsStmt.java | 36 +--
.../org/apache/impala/analysis/ToSqlUtils.java | 9 +-
.../org/apache/impala/analysis/TruncateStmt.java | 7 +-
.../impala/catalog/CatalogServiceCatalog.java | 5 +-
.../org/apache/impala/catalog/FeIcebergTable.java | 246 ++++++++++++++++++++-
.../java/org/apache/impala/catalog/HdfsTable.java | 2 +-
.../org/apache/impala/catalog/IcebergTable.java | 110 +++++----
.../impala/catalog/local/LocalFsPartition.java | 10 +-
.../apache/impala/catalog/local/LocalFsTable.java | 2 +-
.../impala/catalog/local/LocalIcebergTable.java | 141 +++++++-----
.../org/apache/impala/planner/HdfsScanNode.java | 11 +-
.../org/apache/impala/planner/IcebergScanNode.java | 222 +++++++++++++++++++
.../apache/impala/planner/SingleNodePlanner.java | 14 +-
.../apache/impala/service/CatalogOpExecutor.java | 2 +-
.../java/org/apache/impala/service/Frontend.java | 9 +-
.../java/org/apache/impala/util/IcebergUtil.java | 75 ++++++-
.../impala/catalog/FileMetadataLoaderTest.java | 34 +++
testdata/data/README | 8 +
...d44ad-18bc-40f2-9dd6-aeb2cc23457c-00000.parquet | Bin 0 -> 1162 bytes
...b2521-1e8b-40c1-b846-552cd620abce-00000.parquet | Bin 0 -> 1190 bytes
...55628-0544-4833-8b11-1b4add53dfd6-00000.parquet | Bin 0 -> 1169 bytes
...530ef-93b6-4994-b3c8-db957d44848c-00000.parquet | Bin 0 -> 1162 bytes
...b22da-5f10-4cbf-8e4d-160f829b5e48-00000.parquet | Bin 0 -> 1190 bytes
...9a1f7-9024-4bc3-a030-e20861586146-00000.parquet | Bin 0 -> 1162 bytes
...814ae-56cd-486b-af81-18541437da7d-00000.parquet | Bin 0 -> 1189 bytes
...c70a4-bf4d-4a82-8c97-c90e2b4d9dcf-00000.parquet | Bin 0 -> 1190 bytes
...dca7f-c050-407e-b70c-2bd076f83e4e-00000.parquet | Bin 0 -> 1169 bytes
...31a1f-309e-43b3-a5cf-3ef82fa4a87c-00000.parquet | Bin 0 -> 1169 bytes
...38078-244c-4b38-8127-04a5bfbc4695-00000.parquet | Bin 0 -> 1162 bytes
...69895-df25-4ad8-b64d-49c4540d36c9-00000.parquet | Bin 0 -> 1168 bytes
...0c1ea-a2f5-4109-81ec-3ff9c155430f-00000.parquet | Bin 0 -> 1169 bytes
...61b8c-9d9a-4823-b64e-109770c16696-00000.parquet | Bin 0 -> 1190 bytes
...c9d65-bfc3-47c4-8da7-6610b49b3305-00000.parquet | Bin 0 -> 1162 bytes
...9b09d-c730-40a6-ad03-8b36096cb234-00000.parquet | Bin 0 -> 1190 bytes
...b2129-0d18-418a-bcdd-b1b5332ea2a5-00000.parquet | Bin 0 -> 1162 bytes
...51a0c-3d30-4cdb-afcb-e8c1cfa3caa4-00000.parquet | Bin 0 -> 1169 bytes
...7707d-58db-469c-ab96-b77188f25189-00000.parquet | Bin 0 -> 1162 bytes
...f6a62-c326-44f7-bdc3-b1471bd594d7-00000.parquet | Bin 0 -> 1162 bytes
.../9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21-m0.avro | Bin 0 -> 5045 bytes
...973-1-9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21.avro | Bin 0 -> 2538 bytes
.../metadata/v1.metadata.json | 42 ++++
.../metadata/v2.metadata.json | 61 +++++
.../metadata/version-hint.text | 1 +
...5a171-0911-47c2-90c8-300f23c28772-00000.parquet | Bin 0 -> 1162 bytes
...6f547-de52-4aad-8c07-5d0f1aaa0558-00000.parquet | Bin 0 -> 1162 bytes
...024d6-8b55-42d3-b25e-5f2a31303026-00000.parquet | Bin 0 -> 1162 bytes
...23269-c8d7-40cf-94c8-8b624969b9c0-00000.parquet | Bin 0 -> 1162 bytes
...a376e-696e-4da3-95f3-7e0aacb11ca3-00000.parquet | Bin 0 -> 1162 bytes
...94175-75de-41ef-b824-3923ebfba0cb-00000.parquet | Bin 0 -> 1162 bytes
...34050-b263-41e8-bc7d-773a47b33c32-00000.parquet | Bin 0 -> 1162 bytes
...1c3ff-7154-4fd1-a090-8fa0b9c2d3c8-00000.parquet | Bin 0 -> 1162 bytes
...9da05-6be1-42e1-90e0-bff8ca08b5af-00000.parquet | Bin 0 -> 1169 bytes
...9012d-85e3-4144-ac83-574ed5046718-00000.parquet | Bin 0 -> 1169 bytes
...e3f98-5f3b-4245-9c01-76cf0f1dd21a-00000.parquet | Bin 0 -> 1169 bytes
...a1955-f01c-4e08-9f19-dcec0cbceadd-00000.parquet | Bin 0 -> 1168 bytes
...32a29-fd5b-4ba5-a0db-2992c73075c2-00000.parquet | Bin 0 -> 1169 bytes
...20e43-f85a-4609-9ebb-4c76269cddb6-00000.parquet | Bin 0 -> 1169 bytes
...7d440-44f5-4a11-bbfa-c72c520ca997-00000.parquet | Bin 0 -> 1190 bytes
...2540c-b002-413c-b422-7ec3c06d404e-00000.parquet | Bin 0 -> 1190 bytes
...8f0e5-7a46-417e-8c0a-50bb00e54a30-00000.parquet | Bin 0 -> 1189 bytes
...4b7c7-164d-4d06-8b4d-24d296906f73-00000.parquet | Bin 0 -> 1190 bytes
...5b9eb-060d-4180-aa1e-2bb69187d8db-00000.parquet | Bin 0 -> 1190 bytes
...b05c6-5407-41fc-a881-60626253ca77-00000.parquet | Bin 0 -> 1190 bytes
.../af797bab-2f2c-44df-a77b-d91c7198fe53-m0.avro | Bin 0 -> 5579 bytes
...308-1-af797bab-2f2c-44df-a77b-d91c7198fe53.avro | Bin 0 -> 2563 bytes
.../iceberg_partitioned/metadata/v1.metadata.json | 62 ++++++
.../iceberg_partitioned/metadata/v2.metadata.json | 81 +++++++
.../iceberg_partitioned/metadata/version-hint.text | 1 +
.../functional/functional_schema_template.sql | 26 +++
.../datasets/functional/schema_constraints.csv | 2 +
.../{iceberg_create.test => iceberg-create.test} | 31 +--
.../queries/QueryTest/iceberg-negative.test | 45 ++++
.../queries/QueryTest/iceberg-profile.test | 20 ++
.../queries/QueryTest/iceberg-query.test | 185 ++++++++++++++++
.../queries/QueryTest/show-create-table.test | 4 +-
.../{custom_cluster => query_test}/test_iceberg.py | 15 +-
tests/query_test/test_scanners.py | 23 ++
89 files changed, 1454 insertions(+), 203 deletions(-)
diff --git a/be/src/runtime/descriptors.cc b/be/src/runtime/descriptors.cc
index b387f16..d43769b 100644
--- a/be/src/runtime/descriptors.cc
+++ b/be/src/runtime/descriptors.cc
@@ -554,6 +554,7 @@ Status DescriptorTbl::CreateTblDescriptorInternal(const TTableDescriptor& tdesc,
ObjectPool* pool, TableDescriptor** desc) {
*desc = nullptr;
switch (tdesc.tableType) {
+ case TTableType::ICEBERG_TABLE:
case TTableType::HDFS_TABLE: {
HdfsTableDescriptor* hdfs_tbl = pool->Add(new HdfsTableDescriptor(tdesc, pool));
*desc = hdfs_tbl;
diff --git a/bin/rat_exclude_files.txt b/bin/rat_exclude_files.txt
index 316208f..c7e64ba 100644
--- a/bin/rat_exclude_files.txt
+++ b/bin/rat_exclude_files.txt
@@ -151,6 +151,7 @@ testdata/data/text_large_zstd.zst
testdata/data/widerow.txt
testdata/data/local_tbl/00000.txt
testdata/data/hudi_parquet/*
+testdata/data/iceberg_test/*
testdata/datasets/functional/functional_schema_template.sql
testdata/hive_benchmark/grepTiny/part-00000
testdata/tzdb/2017c.zip
diff --git a/common/thrift/CatalogObjects.thrift b/common/thrift/CatalogObjects.thrift
index 744010c..d0a1a2d 100644
--- a/common/thrift/CatalogObjects.thrift
+++ b/common/thrift/CatalogObjects.thrift
@@ -87,6 +87,12 @@ enum THdfsCompression {
LZ4_BLOCKED = 12
}
+// Iceberg table file format identitied by table property 'iceberg_file_format'
+enum TIcebergFileFormat {
+ PARQUET = 0
+ ORC = 1
+}
+
enum TColumnEncoding {
AUTO = 0
PLAIN = 1
@@ -526,6 +532,8 @@ struct TIcebergTable {
// Iceberg file system table location
1: required string table_location
2: required list<TIcebergPartitionSpec> partition_spec
+ // Data file path md5 and it's file descriptor
+ 3: optional map<string,THdfsFileDesc> path_md5_to_file_descriptor
}
// Represents a table or view.
diff --git a/fe/pom.xml b/fe/pom.xml
index 6ea5c8a..3120335 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -414,32 +414,12 @@ under the License.
<groupId>org.apache.iceberg</groupId>
<artifactId>iceberg-api</artifactId>
<version>${iceberg.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hive</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.iceberg</groupId>
<artifactId>iceberg-core</artifactId>
<version>${iceberg.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hive</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
</dependency>
</dependencies>
diff --git a/fe/src/main/java/org/apache/impala/analysis/AlterTableStmt.java b/fe/src/main/java/org/apache/impala/analysis/AlterTableStmt.java
index 249987e..d4c3b13 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AlterTableStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableStmt.java
@@ -21,6 +21,7 @@ import java.util.List;
import org.apache.impala.authorization.Privilege;
import org.apache.impala.catalog.FeDataSourceTable;
+import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.thrift.TAlterTableParams;
@@ -88,6 +89,10 @@ public abstract class AlterTableStmt extends StatementBase {
throw new AnalysisException(String.format(
"ALTER TABLE not allowed on a nested collection: %s", tableName_));
}
+ if (tableRef.getTable() instanceof FeIcebergTable) {
+ throw new AnalysisException(String.format(
+ "ALTER TABLE not allowed on iceberg table: %s", tableName_));
+ }
Preconditions.checkState(tableRef instanceof BaseTableRef);
table_ = tableRef.getTable();
analyzer.checkTableCapability(table_, Analyzer.OperationType.WRITE);
diff --git a/fe/src/main/java/org/apache/impala/analysis/Analyzer.java b/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
index 796429d..3bf0b8e 100644
--- a/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
+++ b/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
@@ -54,7 +54,6 @@ import org.apache.impala.catalog.FeIncompleteTable;
import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.FeView;
-import org.apache.impala.catalog.IcebergTable;
import org.apache.impala.catalog.KuduTable;
import org.apache.impala.catalog.TableLoadingException;
import org.apache.impala.catalog.Type;
@@ -794,8 +793,7 @@ public class Analyzer {
Preconditions.checkState(table instanceof FeFsTable ||
table instanceof FeKuduTable ||
table instanceof FeHBaseTable ||
- table instanceof FeDataSourceTable ||
- table instanceof IcebergTable);
+ table instanceof FeDataSourceTable);
resolvedTableRef = new BaseTableRef(tableRef, resolvedPath);
}
// Only do table masking when authorization is enabled and the authorization
diff --git a/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java b/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
index 7cda266..6bb2b17 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
@@ -35,6 +35,7 @@ import org.apache.impala.catalog.FeCatalogUtils;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeHBaseTable;
+import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
@@ -916,6 +917,10 @@ public class ComputeStatsStmt extends StatementBase {
&& partition.getFileFormat() != HdfsFileFormat.ORC)
return false;
}
+
+ if (table_ instanceof FeIcebergTable) {
+ return FeIcebergTable.Utils.isColumnar((FeIcebergTable) table_);
+ }
return true;
}
diff --git a/fe/src/main/java/org/apache/impala/analysis/CreateTableStmt.java b/fe/src/main/java/org/apache/impala/analysis/CreateTableStmt.java
index 920d725..96ac124 100644
--- a/fe/src/main/java/org/apache/impala/analysis/CreateTableStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/CreateTableStmt.java
@@ -580,6 +580,12 @@ public class CreateTableStmt extends StatementBase {
"managed iceberg table.");
}
+ if ((!isExternal() || Boolean.parseBoolean(getTblProperties().get(
+ Table.TBL_PROP_EXTERNAL_TABLE_PURGE)))) {
+ // Check partition columns for managed iceberg table
+ checkPartitionColumns();
+ }
+
String handler = getTblProperties().get(IcebergTable.KEY_STORAGE_HANDLER);
if (handler != null && !IcebergTable.isIcebergStorageHandler(handler)) {
throw new AnalysisException("Invalid storage handler " +
@@ -587,5 +593,36 @@ public class CreateTableStmt extends StatementBase {
}
putGeneratedKuduProperty(IcebergTable.KEY_STORAGE_HANDLER,
IcebergTable.ICEBERG_STORAGE_HANDLER);
+
+ String fileformat = getTblProperties().get(IcebergTable.ICEBERG_FILE_FORMAT);
+ if (fileformat == null || fileformat.isEmpty()) {
+ putGeneratedKuduProperty(IcebergTable.ICEBERG_FILE_FORMAT, "parquet");
+ }
+ }
+
+ /**
+ * For iceberg table, partition column must be from source column
+ */
+ private void checkPartitionColumns() throws AnalysisException {
+ // This check is unnecessary for iceberg table without partition spec
+ List<IcebergPartitionSpec> specs = tableDef_.getIcebergPartitionSpecs();
+ if (specs == null || specs.isEmpty()) return;
+
+ // Iceberg table only has one partition spec now
+ List<IcebergPartitionField> fields = specs.get(0).getIcebergPartitionFields();
+ Preconditions.checkState(fields != null && !fields.isEmpty());
+ for (IcebergPartitionField field : fields) {
+ String fieldName = field.getFieldName();
+ boolean containFlag = false;
+ for (ColumnDef columnDef : tableDef_.getColumnDefs()) {
+ if (columnDef.getColName().equalsIgnoreCase(fieldName)) {
+ containFlag = true;
+ break;
+ }
+ }
+ if (!containFlag) {
+ throw new AnalysisException("Cannot find source column: " + fieldName);
+ }
+ }
}
}
diff --git a/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionField.java b/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionField.java
index 2cc2d5d..4db74e9 100644
--- a/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionField.java
+++ b/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionField.java
@@ -25,8 +25,8 @@ import org.apache.impala.thrift.TIcebergPartitionTransform;
* Represents a PartitionField of iceberg
*/
public class IcebergPartitionField extends StmtNode {
- // The id of the source field in iceberg table Schema, you can get these source
- // fields by Schema.columns(), the return type is List<NestedField>.
+ // The id of the source column in the Iceberg table schema. The source column is
+ // used as the input for this partition field.
private int sourceId_;
// The field id from Iceberg PartitionField, which across all the table
@@ -51,6 +51,14 @@ public class IcebergPartitionField extends StmtNode {
this(0, 0, fieldName, fieldType);
}
+ public String getFieldName() {
+ return fieldName_;
+ }
+
+ public int getSourceId() {
+ return sourceId_;
+ }
+
@Override
public void analyze(Analyzer analyzer) throws AnalysisException {
return;
diff --git a/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionSpec.java b/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionSpec.java
index 6bf2829..f5c3af0 100644
--- a/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionSpec.java
+++ b/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionSpec.java
@@ -23,8 +23,6 @@ import org.apache.impala.common.AnalysisException;
import org.apache.impala.compat.MetastoreShim;
import org.apache.impala.thrift.TIcebergPartitionSpec;
-import com.google.common.base.Preconditions;
-
/**
* Represents the partitioning of a Iceberg table as defined in the PARTITION BY SPEC
* clause of a CREATE TABLE statement. Iceberg supported kinds of partition.
@@ -52,12 +50,17 @@ public class IcebergPartitionSpec extends StmtNode {
this(0, fields);
}
- public List<IcebergPartitionField> getIcebergPartitionFields_() {
+ public List<IcebergPartitionField> getIcebergPartitionFields() {
return icebergPartitionFields_;
}
+ public boolean hasPartitionFields() {
+ return icebergPartitionFields_ != null && (!icebergPartitionFields_.isEmpty());
+ }
+
@Override
public void analyze(Analyzer analyzer) throws AnalysisException {
+ if (!hasPartitionFields()) return;
for (IcebergPartitionField field : icebergPartitionFields_) {
field.analyze(analyzer);
}
@@ -72,7 +75,7 @@ public class IcebergPartitionSpec extends StmtNode {
public String toSql(ToSqlOptions options) {
StringBuilder builder = new StringBuilder();
builder.append("(");
- if (!icebergPartitionFields_.isEmpty()) {
+ if (hasPartitionFields()) {
builder.append("\n");
for (IcebergPartitionField field : icebergPartitionFields_) {
builder.append(String.format(" %s,\n", field.toSql()));
@@ -85,6 +88,7 @@ public class IcebergPartitionSpec extends StmtNode {
public TIcebergPartitionSpec toThrift() {
TIcebergPartitionSpec result = new TIcebergPartitionSpec();
result.setPartition_id(partitionId_);
+ if (!hasPartitionFields()) return result;
for (IcebergPartitionField field : icebergPartitionFields_) {
result.addToPartition_fields(field.toThrift());
}
diff --git a/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java b/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java
index 2228726..811b817 100644
--- a/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java
@@ -27,6 +27,7 @@ import org.apache.impala.authorization.Privilege;
import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeHBaseTable;
+import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.FeView;
@@ -439,11 +440,15 @@ public class InsertStmt extends StatementBase {
builder.onTable(table_).allOf(privilegeRequired).build());
}
- // We do not support (in|up)serting into views.
+ // We do not support (in|up)serting into views and iceberg table
if (table_ instanceof FeView) {
throw new AnalysisException(
String.format("Impala does not support %sing into views: %s", getOpName(),
table_.getFullName()));
+ } else if (table_ instanceof FeIcebergTable) {
+ throw new AnalysisException(
+ String.format("Impala does not support %sing into iceberg table: %s",
+ getOpName(), table_.getFullName()));
}
Analyzer.ensureTableNotFullAcid(table_, "INSERT");
diff --git a/fe/src/main/java/org/apache/impala/analysis/ShowFilesStmt.java b/fe/src/main/java/org/apache/impala/analysis/ShowFilesStmt.java
index 618cba0..0db0357 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ShowFilesStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ShowFilesStmt.java
@@ -76,8 +76,7 @@ public class ShowFilesStmt extends StatementBase {
table_ = tableRef.getTable();
Preconditions.checkNotNull(table_);
if (!(table_ instanceof FeFsTable)) {
- throw new AnalysisException(String.format(
- "SHOW FILES not applicable to a non hdfs table: %s", tableName_));
+ throw new AnalysisException("SHOW FILES is applicable only to a HDFS table");
}
tableRef.analyze(analyzer);
diff --git a/fe/src/main/java/org/apache/impala/analysis/ShowStatsStmt.java b/fe/src/main/java/org/apache/impala/analysis/ShowStatsStmt.java
index 42fba21..e9e8ce6 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ShowStatsStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ShowStatsStmt.java
@@ -81,7 +81,26 @@ public class ShowStatsStmt extends StatementBase {
"%s not applicable to a view: %s", getSqlPrefix(), table_.getFullName()));
}
if (table_ instanceof FeFsTable) {
- if (table_.getNumClusteringCols() == 0 && op_ == TShowStatsOp.PARTITIONS) {
+ // There two cases here: Non-partitioned hdfs table and non-partitioned
+ // iceberg table
+ boolean partitioned = true;
+ if (op_ == TShowStatsOp.PARTITIONS) {
+ if (table_ instanceof FeIcebergTable) {
+ // We only get latest partition spec from Iceberg now, so this list only
+ // contains one partition spec member.
+ // Iceberg snapshots chosen maybe supported in the future.
+ Preconditions.checkNotNull(((FeIcebergTable) table_).getPartitionSpec());
+ // Partition spec without partition fields is non-partitioned.
+ if (!((FeIcebergTable) table_).getPartitionSpec().get(0).hasPartitionFields()) {
+ partitioned = false;
+ }
+ } else {
+ if (table_.getNumClusteringCols() == 0) {
+ partitioned = false;
+ }
+ }
+ }
+ if (!partitioned) {
throw new AnalysisException("Table is not partitioned: " + table_.getFullName());
}
if (op_ == TShowStatsOp.RANGE_PARTITIONS) {
@@ -95,21 +114,6 @@ public class ShowStatsStmt extends StatementBase {
throw new AnalysisException(getSqlPrefix() + " requested but table does not " +
"have range partitions: " + table_.getFullName());
}
- } else if (table_ instanceof FeIcebergTable) {
- FeIcebergTable icebergTable = (FeIcebergTable) table_;
- if (op_ == TShowStatsOp.PARTITIONS) {
- //Non-partition iceberg table only has an empty PartitionField set
- Preconditions.checkArgument(!icebergTable.getPartitionSpec().isEmpty());
- IcebergPartitionSpec spec = icebergTable.getPartitionSpec().get(0);
- boolean emptySpec = (spec.getIcebergPartitionFields_() == null ||
- spec.getIcebergPartitionFields_().size() == 0);
- if (icebergTable.getPartitionSpec().size() == 1 && emptySpec) {
- throw new AnalysisException("Iceberg table does not have PartitionSpec: "
- + table_.getFullName());
- }
- } else {
- throw new AnalysisException(getSqlPrefix() + " not supported for Iceberg table.");
- }
} else {
if (op_ == TShowStatsOp.RANGE_PARTITIONS) {
throw new AnalysisException(getSqlPrefix() + " must target a Kudu table: " +
diff --git a/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java b/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java
index c983d3c..b4c5310 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java
@@ -392,6 +392,11 @@ public class ToSqlUtils {
colsSql = null;
}
} else if (table instanceof FeFsTable) {
+ if (table instanceof FeIcebergTable) {
+ storageHandlerClassName = null;
+ properties.remove(IcebergTable.KEY_STORAGE_HANDLER);
+ }
+
String inputFormat = msTable.getSd().getInputFormat();
format = HdfsFileFormat.fromHdfsInputFormatClass(inputFormat);
compression = HdfsCompression.fromHdfsInputFormatClass(inputFormat);
@@ -401,10 +406,6 @@ public class ToSqlUtils {
} catch (Exception e) {
throw new CatalogException("Could not get primary key/foreign keys sql.", e);
}
- } else if (table instanceof FeIcebergTable) {
- storageHandlerClassName = null;
- format = HdfsFileFormat.ICEBERG;
- properties.remove(IcebergTable.KEY_STORAGE_HANDLER);
}
HdfsUri tableLocation = location == null ? null : new HdfsUri(location);
diff --git a/fe/src/main/java/org/apache/impala/analysis/TruncateStmt.java b/fe/src/main/java/org/apache/impala/analysis/TruncateStmt.java
index 9945d78..ad0068b 100644
--- a/fe/src/main/java/org/apache/impala/analysis/TruncateStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/TruncateStmt.java
@@ -21,6 +21,7 @@ import java.util.List;
import org.apache.impala.authorization.Privilege;
import org.apache.impala.catalog.FeFsTable;
+import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.thrift.TTruncateParams;
@@ -62,10 +63,14 @@ public class TruncateStmt extends StatementBase {
if (ifExists_) return;
throw e;
}
- // We only support truncating hdfs tables now.
+ // We only support truncating hdfs tables now, we also cannot truncate Iceberg
+ // tables.
if (!(table_ instanceof FeFsTable)) {
throw new AnalysisException(String.format(
"TRUNCATE TABLE not supported on non-HDFS table: %s", table_.getFullName()));
+ } else if (table_ instanceof FeIcebergTable) {
+ throw new AnalysisException(String.format(
+ "TRUNCATE TABLE not supported on iceberg table: %s", table_.getFullName()));
}
Analyzer.ensureTableNotFullAcid(table_, "TRUNCATE");
Analyzer.checkTableCapability(table_, Analyzer.OperationType.WRITE);
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index e51f56f..116b8dd 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -3194,8 +3194,9 @@ public class CatalogServiceCatalog extends Catalog {
// TODO(todd): consider a read-write lock here.
table.getLock().lock();
try {
- if (table instanceof HdfsTable) {
- HdfsTable hdfsTable = (HdfsTable) table;
+ if (table instanceof HdfsTable || table instanceof IcebergTable) {
+ HdfsTable hdfsTable = table instanceof HdfsTable ? (HdfsTable) table :
+ ((IcebergTable) table).getHdfsTable();
missingPartialInfos = Maps.newHashMap();
resp = hdfsTable.getPartialInfo(req, missingPartialInfos);
if (missingPartialInfos.isEmpty()) return resp;
diff --git a/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
index d245d11..c216a72 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
@@ -17,24 +17,64 @@
package org.apache.impala.catalog;
+import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.DataFile;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.types.Types;
import org.apache.impala.analysis.IcebergPartitionField;
import org.apache.impala.analysis.IcebergPartitionSpec;
+import org.apache.impala.analysis.LiteralExpr;
+import org.apache.impala.common.FileSystemUtil;
+import org.apache.impala.common.Reference;
+import org.apache.impala.compat.HdfsShim;
import org.apache.impala.thrift.TColumn;
+import org.apache.impala.thrift.THdfsTable;
+import org.apache.impala.thrift.THdfsPartition;
+import org.apache.impala.thrift.TIcebergFileFormat;
+import org.apache.impala.thrift.TIcebergTable;
+import org.apache.impala.thrift.TNetworkAddress;
import org.apache.impala.thrift.TResultSet;
import org.apache.impala.thrift.TResultSetMetadata;
import org.apache.impala.util.IcebergUtil;
+import org.apache.impala.util.ListMap;
import org.apache.impala.util.TResultRowBuilder;
/**
* Frontend interface for interacting with an Iceberg-backed table.
*/
-public interface FeIcebergTable extends FeTable {
+public interface FeIcebergTable extends FeFsTable {
+
+ /**
+ * FileDescriptor map
+ */
+ Map<String, HdfsPartition.FileDescriptor> getPathMD5ToFileDescMap();
+
+ /**
+ * Return the hdfs table transformed from iceberg table
+ */
+ FeFsTable getFeFsTable();
+
+ /**
+ * Return iceberg file format from table properties
+ */
+ TIcebergFileFormat getIcebergFileFormat();
+
/**
* Return the name of iceberg table name, usually a hdfs location path
*/
@@ -45,10 +85,123 @@ public interface FeIcebergTable extends FeTable {
*/
List<IcebergPartitionSpec> getPartitionSpec();
+ @Override
+ default boolean isCacheable() {
+ return getFeFsTable().isCacheable();
+ }
+
+ @Override
+ default boolean isLocationCacheable() {
+ return getFeFsTable().isLocationCacheable();
+ }
+
+ @Override
+ default boolean isMarkedCached() {
+ return getFeFsTable().isMarkedCached();
+ }
+
+ @Override
+ default String getLocation() {
+ return getFeFsTable().getLocation();
+ }
+
+ @Override
+ default String getNullPartitionKeyValue() {
+ return getFeFsTable().getNullPartitionKeyValue();
+ }
+
+ @Override
+ default String getHdfsBaseDir() {
+ return getFeFsTable().getHdfsBaseDir();
+ }
+
+ @Override
+ default FileSystemUtil.FsType getFsType() {
+ return getFeFsTable().getFsType();
+ }
+
+ @Override
+ default long getTotalHdfsBytes() {
+ return getFeFsTable().getTotalHdfsBytes();
+ }
+
+ @Override
+ default boolean usesAvroSchemaOverride() {
+ return getFeFsTable().usesAvroSchemaOverride();
+ }
+
+ @Override
+ default Set<HdfsFileFormat> getFileFormats() {
+ return getFeFsTable().getFileFormats();
+ }
+
+ @Override
+ default boolean hasWriteAccessToBaseDir() {
+ return getFeFsTable().hasWriteAccessToBaseDir();
+ }
+
+ @Override
+ default String getFirstLocationWithoutWriteAccess() {
+ return getFeFsTable().getFirstLocationWithoutWriteAccess();
+ }
+
+ @Override
+ default TResultSet getTableStats() {
+ return getFeFsTable().getTableStats();
+ }
+
+ @Override
+ default Collection<? extends PrunablePartition> getPartitions() {
+ return getFeFsTable().getPartitions();
+ }
+
+ @Override
+ default Set<Long> getPartitionIds() {
+ return getFeFsTable().getPartitionIds();
+ }
+
+ @Override
+ default Map<Long, ? extends PrunablePartition> getPartitionMap() {
+ return getFeFsTable().getPartitionMap();
+ }
+
+ @Override
+ default TreeMap<LiteralExpr, Set<Long>> getPartitionValueMap(int col) {
+ return getFeFsTable().getPartitionValueMap(col);
+ }
+
+ @Override
+ default Set<Long> getNullPartitionIds(int colIdx) {
+ return getFeFsTable().getNullPartitionIds(colIdx);
+ }
+
+ @Override
+ default List<? extends FeFsPartition> loadPartitions(Collection<Long> ids) {
+ return getFeFsTable().loadPartitions(ids);
+ }
+
+ @Override
+ default SqlConstraints getSqlConstraints() {
+ return getFeFsTable().getSqlConstraints();
+ }
+
+ @Override
+ default ListMap<TNetworkAddress> getHostIndex() {
+ return getFeFsTable().getHostIndex();
+ }
+
/**
* Utility functions
*/
public static abstract class Utils {
+ /**
+ * Returns true if FeIcebergTable file format is columnar: parquet or orc
+ */
+ public static boolean isColumnar(FeIcebergTable table) {
+ HdfsFileFormat format = IcebergUtil.toHdfsFileFormat(table.getIcebergFileFormat());
+ return format == HdfsFileFormat.PARQUET || format == HdfsFileFormat.ORC;
+ }
+
public static TResultSet getPartitionSpecs(FeIcebergTable table)
throws TableLoadingException {
TResultSet result = new TResultSet();
@@ -80,6 +233,85 @@ public interface FeIcebergTable extends FeTable {
return result;
}
+ /**
+ * Get iceberg table file format from hms table properties
+ */
+ public static TIcebergFileFormat getIcebergFileFormat(
+ org.apache.hadoop.hive.metastore.api.Table msTable) {
+ TIcebergFileFormat fileFormat = IcebergUtil.getIcebergFileFormat(
+ msTable.getParameters().get(IcebergTable.ICEBERG_FILE_FORMAT));
+ return fileFormat == null ? TIcebergFileFormat.PARQUET : fileFormat;
+ }
+
+
+ public static TIcebergTable getTIcebergTable(FeIcebergTable icebergTable) {
+ TIcebergTable tIcebergTable = new TIcebergTable();
+ tIcebergTable.setTable_location(icebergTable.getIcebergTableLocation());
+
+ for (IcebergPartitionSpec partitionSpec : icebergTable.getPartitionSpec()) {
+ tIcebergTable.addToPartition_spec(partitionSpec.toThrift());
+ }
+
+ for (Map.Entry<String, HdfsPartition.FileDescriptor> entry :
+ icebergTable.getPathMD5ToFileDescMap().entrySet()) {
+ tIcebergTable.putToPath_md5_to_file_descriptor(entry.getKey(),
+ entry.getValue().toThrift());
+ }
+ return tIcebergTable;
+ }
+
+ /**
+ * Get FileDescriptor by data file location
+ */
+ private static HdfsPartition.FileDescriptor getFileDescriptor(Path fileLoc,
+ Path tableLoc, ListMap<TNetworkAddress> hostIndex) throws IOException {
+ FileSystem fs = FileSystemUtil.getFileSystemForPath(tableLoc);
+ FileStatus fileStatus = fs.getFileStatus(fileLoc);
+ return getFileDescriptor(fs, tableLoc, fileStatus, hostIndex);
+ }
+
+ private static HdfsPartition.FileDescriptor getFileDescriptor(FileSystem fs,
+ Path tableLoc, FileStatus fileStatus, ListMap<TNetworkAddress> hostIndex)
+ throws IOException {
+ Reference<Long> numUnknownDiskIds = new Reference<Long>(Long.valueOf(0));
+ String relPath = FileSystemUtil.relativizePath(fileStatus.getPath(), tableLoc);
+
+ if (!FileSystemUtil.supportsStorageIds(fs)) {
+ return HdfsPartition.FileDescriptor.createWithNoBlocks(fileStatus, relPath);
+ }
+
+ BlockLocation[] locations;
+ if (fileStatus instanceof LocatedFileStatus) {
+ locations = ((LocatedFileStatus)fileStatus).getBlockLocations();
+ } else {
+ locations = fs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
+ }
+
+ return HdfsPartition.FileDescriptor.create(fileStatus, relPath, locations,
+ hostIndex, HdfsShim.isErasureCoded(fileStatus), numUnknownDiskIds);
+ }
+
+ /**
+ * Get all FileDescriptor from iceberg table without any predicates.
+ */
+ public static Map<String, HdfsPartition.FileDescriptor> loadAllPartition(
+ String location, FeIcebergTable table) throws IOException {
+ // Empty predicates
+ List<DataFile> dataFileList = IcebergUtil.getIcebergDataFiles(
+ location, new ArrayList<>());
+
+ Map<String, HdfsPartition.FileDescriptor> fileDescMap = new HashMap<>();
+ for (DataFile file : dataFileList) {
+ HdfsPartition.FileDescriptor fileDesc = getFileDescriptor(
+ new Path(file.path().toString()), new Path(location), table.getHostIndex());
+ fileDescMap.put(IcebergUtil.getDataFileMD5(file), fileDesc);
+ }
+ return fileDescMap;
+ }
+
+ /**
+ * Get iceberg partition spec by iceberg table metadata
+ */
public static List<IcebergPartitionSpec> loadPartitionSpecByIceberg(
TableMetadata metadata) throws TableLoadingException {
List<IcebergPartitionSpec> ret = new ArrayList<>();
@@ -93,5 +325,17 @@ public interface FeIcebergTable extends FeTable {
}
return ret;
}
+
+ /**
+ * Update iceberg table partition file format by table file format
+ */
+ public static void updateIcebergPartitionFileFormat(FeIcebergTable icebergTable,
+ THdfsTable hdfsTable) {
+ for (Map.Entry<Long, THdfsPartition> entry : hdfsTable.getPartitions().entrySet()) {
+ THdfsPartition partition = entry.getValue();
+ partition.setFileFormat(IcebergUtil.toTHdfsFileFormat(
+ icebergTable.getIcebergFileFormat()));
+ }
+ }
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
index bf91168..f4e562a 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -2112,7 +2112,7 @@ public class HdfsTable extends Table implements FeFsTable {
* (e.g. number of blocks, files, etc) in order to compute an estimate of the metadata
* size of this table.
*/
- private THdfsTable getTHdfsTable(ThriftObjectType type, Set<Long> refPartitions) {
+ protected THdfsTable getTHdfsTable(ThriftObjectType type, Set<Long> refPartitions) {
if (type == ThriftObjectType.FULL) {
// "full" implies all partitions should be included.
Preconditions.checkArgument(refPartitions == null);
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
index d448277..9ffc12f 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
@@ -18,7 +18,9 @@
package org.apache.impala.catalog;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import org.apache.hadoop.hive.common.StatsSetupConst;
@@ -26,13 +28,15 @@ import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.metastore.TableType;
-import org.apache.iceberg.PartitionField;
-import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.types.Types;
import org.apache.impala.analysis.IcebergPartitionField;
import org.apache.impala.analysis.IcebergPartitionSpec;
+import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
import org.apache.impala.thrift.TCatalogObjectType;
+import org.apache.impala.thrift.THdfsFileDesc;
+import org.apache.impala.thrift.THdfsTable;
+import org.apache.impala.thrift.TIcebergFileFormat;
import org.apache.impala.thrift.TIcebergPartitionField;
import org.apache.impala.thrift.TIcebergPartitionSpec;
import org.apache.impala.thrift.TIcebergTable;
@@ -60,6 +64,12 @@ public class IcebergTable extends Table implements FeIcebergTable {
public static final String ICEBERG_STORAGE_HANDLER =
"com.expediagroup.hiveberg.IcebergStorageHandler";
+ // Iceberg file format key in tblproperties
+ public static final String ICEBERG_FILE_FORMAT = "iceberg_file_format";
+
+ // Iceberg file format dependend on table properties
+ private TIcebergFileFormat icebergFileFormat_;
+
// The iceberg file system table location
private String icebergTableLocation_;
@@ -69,13 +79,18 @@ public class IcebergTable extends Table implements FeIcebergTable {
// Schema of the iceberg table.
private org.apache.iceberg.Schema icebergSchema_;
- // PartitionSpec of the iceberg table.
- private List<PartitionSpec> icebergPartitionSpecs_;
+ // Key is the DataFile md5, value is FileDescriptor transformed from DataFile
+ private Map<String, FileDescriptor> pathMD5ToFileDescMap_;
+
+ // Treat iceberg table as a non-partitioned hdfs table in backend
+ private HdfsTable hdfsTable_;
protected IcebergTable(org.apache.hadoop.hive.metastore.api.Table msTable,
- Db db, String name, String owner) {
+ Db db, String name, String owner) {
super(msTable, db, name, owner);
icebergTableLocation_ = msTable.getSd().getLocation();
+ icebergFileFormat_ = Utils.getIcebergFileFormat(msTable);
+ hdfsTable_ = new HdfsTable(msTable, db, name, owner);
}
/**
@@ -96,6 +111,10 @@ public class IcebergTable extends Table implements FeIcebergTable {
return msTbl.getTableType().equalsIgnoreCase(TableType.MANAGED_TABLE.toString());
}
+ public HdfsTable getHdfsTable() {
+ return hdfsTable_;
+ }
+
@Override
public TCatalogObjectType getCatalogObjectType() {
return TCatalogObjectType.TABLE;
@@ -114,12 +133,9 @@ public class IcebergTable extends Table implements FeIcebergTable {
return isIcebergStorageHandler(msTbl.getParameters().get(KEY_STORAGE_HANDLER));
}
- public org.apache.iceberg.Schema getIcebergSchema() {
- return icebergSchema_;
- }
-
- public List<PartitionSpec> getIcebergPartitionSpec() {
- return icebergPartitionSpecs_;
+ @Override
+ public TIcebergFileFormat getIcebergFileFormat() {
+ return icebergFileFormat_;
}
@Override
@@ -128,16 +144,27 @@ public class IcebergTable extends Table implements FeIcebergTable {
}
@Override
+ public FeFsTable getFeFsTable() {
+ return hdfsTable_;
+ }
+
+ @Override
public List<IcebergPartitionSpec> getPartitionSpec() {
Preconditions.checkState(partitionSpecs_ != null);
return ImmutableList.copyOf(partitionSpecs_);
}
@Override
+ public Map<String, FileDescriptor> getPathMD5ToFileDescMap() {
+ return pathMD5ToFileDescMap_;
+ }
+
+ @Override
public TTable toThrift() {
TTable table = super.toThrift();
table.setTable_type(TTableType.ICEBERG_TABLE);
- table.setIceberg_table(getTIcebergTable());
+ table.setIceberg_table(Utils.getTIcebergTable(this));
+ table.setHdfs_table(transfromToTHdfsTable(true));
return table;
}
@@ -150,7 +177,7 @@ public class IcebergTable extends Table implements FeIcebergTable {
*/
@Override
public void load(boolean dummy /* not used */, IMetaStoreClient msClient,
- org.apache.hadoop.hive.metastore.api.Table msTbl, String reason)
+ org.apache.hadoop.hive.metastore.api.Table msTbl, String reason)
throws TableLoadingException {
final Timer.Context context =
getMetrics().getTimer(Table.LOAD_DURATION_METRIC).time();
@@ -162,6 +189,12 @@ public class IcebergTable extends Table implements FeIcebergTable {
getMetrics().getTimer(Table.LOAD_DURATION_STORAGE_METADATA).time();
try {
loadSchemaFromIceberg();
+ // Loading hdfs table after loaded schema from Iceberg,
+ // in case we create external Iceberg table skipping column info in sql.
+ hdfsTable_.load(false, msClient, msTable_, true, true, false, null, reason);
+ pathMD5ToFileDescMap_ = Utils.loadAllPartition(msTable_.getSd().getLocation(),
+ this);
+ loadAllColumnStats(msClient);
} catch (Exception e) {
throw new TableLoadingException("Error loading metadata for Iceberg table " +
icebergTableLocation_, e);
@@ -194,9 +227,8 @@ public class IcebergTable extends Table implements FeIcebergTable {
public void loadSchemaFromIceberg() throws TableLoadingException {
TableMetadata metadata = IcebergUtil.getIcebergTableMetadata(icebergTableLocation_);
icebergSchema_ = metadata.schema();
- icebergPartitionSpecs_ = metadata.specs();
loadSchema();
- partitionSpecs_ = buildIcebergPartitionSpec(icebergPartitionSpecs_);
+ partitionSpecs_ = Utils.loadPartitionSpecByIceberg(metadata);
}
/**
@@ -220,29 +252,15 @@ public class IcebergTable extends Table implements FeIcebergTable {
}
}
- /**
- * Build IcebergPartitionSpec list by iceberg PartitionSpec
- */
- private List<IcebergPartitionSpec> buildIcebergPartitionSpec(
- List<PartitionSpec> specs) throws TableLoadingException {
- List<IcebergPartitionSpec> ret = new ArrayList<>();
- for (PartitionSpec spec : specs) {
- List<IcebergPartitionField> fields = new ArrayList<>();
- for (PartitionField field : spec.fields()) {
- fields.add(new IcebergPartitionField(field.sourceId(), field.fieldId(),
- field.name(), IcebergUtil.getPartitionTransform(field)));
- }
- ret.add(new IcebergPartitionSpec(spec.specId(), fields));
- }
- return ret;
- }
-
@Override
protected void loadFromThrift(TTable thriftTable) throws TableLoadingException {
super.loadFromThrift(thriftTable);
TIcebergTable ticeberg = thriftTable.getIceberg_table();
icebergTableLocation_ = ticeberg.getTable_location();
partitionSpecs_ = loadPartitionBySpecsFromThrift(ticeberg.getPartition_spec());
+ pathMD5ToFileDescMap_ = loadFileDescFromThrift(
+ ticeberg.getPath_md5_to_file_descriptor());
+ hdfsTable_.loadFromThrift(thriftTable);
}
private List<IcebergPartitionSpec> loadPartitionBySpecsFromThrift(
@@ -266,21 +284,33 @@ public class IcebergTable extends Table implements FeIcebergTable {
return ret;
}
+ private Map<String, FileDescriptor> loadFileDescFromThrift(
+ Map<String, THdfsFileDesc> tFileDescMap) {
+ Map<String, FileDescriptor> fileDescMap = new HashMap<>();
+ if (tFileDescMap == null) return fileDescMap;
+ for (Map.Entry<String, THdfsFileDesc> entry : tFileDescMap.entrySet()) {
+ fileDescMap.put(entry.getKey(), FileDescriptor.fromThrift(entry.getValue()));
+ }
+ return fileDescMap;
+ }
+
@Override
public TTableDescriptor toThriftDescriptor(int tableId,
- Set<Long> referencedPartitions) {
+ Set<Long> referencedPartitions) {
TTableDescriptor desc = new TTableDescriptor(tableId, TTableType.ICEBERG_TABLE,
getTColumnDescriptors(), numClusteringCols_, name_, db_.getName());
- desc.setIcebergTable(getTIcebergTable());
+ desc.setIcebergTable(Utils.getTIcebergTable(this));
+ desc.setHdfsTable(transfromToTHdfsTable(false));
return desc;
}
- private TIcebergTable getTIcebergTable() {
- TIcebergTable tbl = new TIcebergTable();
- tbl.setTable_location(icebergTableLocation_);
- for (IcebergPartitionSpec partition : partitionSpecs_) {
- tbl.addToPartition_spec(partition.toThrift());
+ private THdfsTable transfromToTHdfsTable(boolean updatePartitionFlag) {
+ THdfsTable hdfsTable = hdfsTable_.getTHdfsTable(ThriftObjectType.FULL, null);
+ if (updatePartitionFlag) {
+ // Iceberg table only has one THdfsPartition, we set this partition
+ // file format by iceberg file format which depend on table properties
+ Utils.updateIcebergPartitionFileFormat(this, hdfsTable);
}
- return tbl;
+ return hdfsTable;
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java
index 2905987..1bed832 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsPartition.java
@@ -33,6 +33,7 @@ import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
import org.apache.impala.catalog.HdfsStorageDescriptor;
import org.apache.impala.catalog.HdfsStorageDescriptor.InvalidStorageDescriptorException;
+import org.apache.impala.catalog.IcebergTable;
import org.apache.impala.catalog.PartitionStatsUtil;
import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.compat.MetastoreShim;
@@ -42,6 +43,7 @@ import org.apache.impala.thrift.TPartitionStats;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
+import org.apache.impala.util.IcebergUtil;
public class LocalFsPartition implements FeFsPartition {
private final LocalFsTable table_;
@@ -197,7 +199,13 @@ public class LocalFsPartition implements FeFsPartition {
@Override
public HdfsFileFormat getFileFormat() {
- return getInputFormatDescriptor().getFileFormat();
+ HdfsFileFormat format = getInputFormatDescriptor().getFileFormat();
+ if (format == HdfsFileFormat.ICEBERG) {
+ String format_str = table_.getMetaStoreTable().getParameters().get(
+ IcebergTable.ICEBERG_FILE_FORMAT);
+ return IcebergUtil.toHdfsFileFormat(format_str);
+ }
+ return format;
}
@Override
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java
index a611fd9..12da212 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalFsTable.java
@@ -369,7 +369,7 @@ public class LocalFsTable extends LocalTable implements FeFsTable {
return false;
}
- private LocalFsPartition createPrototypePartition() {
+ protected LocalFsPartition createPrototypePartition() {
Partition protoMsPartition = new Partition();
// The prototype partition should not have a location set in its storage
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java
index 83b3608..117a621 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java
@@ -17,22 +17,26 @@
package org.apache.impala.catalog.local;
-import java.util.ArrayList;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
-import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.iceberg.Schema;
import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.types.Types;
import org.apache.impala.analysis.IcebergPartitionSpec;
-import org.apache.impala.catalog.Column;
+import org.apache.impala.catalog.CatalogObject;
import org.apache.impala.catalog.FeCatalogUtils;
+import org.apache.impala.catalog.FeFsPartition;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeIcebergTable;
+import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
import org.apache.impala.catalog.TableLoadingException;
-import org.apache.impala.catalog.Type;
-import org.apache.impala.thrift.TIcebergTable;
+import org.apache.impala.thrift.THdfsPartition;
+import org.apache.impala.thrift.THdfsTable;
+import org.apache.impala.thrift.TIcebergFileFormat;;
import org.apache.impala.thrift.TTableDescriptor;
import org.apache.impala.thrift.TTableType;
import org.apache.impala.util.IcebergUtil;
@@ -44,57 +48,64 @@ import com.google.errorprone.annotations.Immutable;
* Iceberg table for LocalCatalog
*/
public class LocalIcebergTable extends LocalTable implements FeIcebergTable {
- private final TableParams tableParams_;
- private final List<IcebergPartitionSpec> partitionSpecs_;
+ private TableParams tableParams_;
+ private TIcebergFileFormat icebergFileFormat_;
+ private List<IcebergPartitionSpec> partitionSpecs_;
+ private Map<String, FileDescriptor> pathMD5ToFileDescMap_;
+ private LocalFsTable localFsTable_;
static LocalTable loadFromIceberg(LocalDb db, Table msTable,
- MetaProvider.TableMetaRef ref)
- throws TableLoadingException {
+ MetaProvider.TableMetaRef ref) throws TableLoadingException {
Preconditions.checkNotNull(db);
Preconditions.checkNotNull(msTable);
- String fullTableName = msTable.getDbName() + "." + msTable.getTableName();
- TableParams params = new TableParams(msTable);
-
- List<Column> cols = new ArrayList<>();
- List<FieldSchema> fieldSchemas = new ArrayList<>();
- TableMetadata metadata =
- IcebergUtil.getIcebergTableMetadata(params.icebergTableName_);
- convertColsFromKIceberg(metadata.schema(), cols, fieldSchemas);
+ try {
+ TableParams params = new TableParams(msTable);
+ TableMetadata metadata =
+ IcebergUtil.getIcebergTableMetadata(params.icebergTableLocation_);
- msTable.getSd().setCols(fieldSchemas);
+ List<IcebergPartitionSpec> partitionSpecs =
+ Utils.loadPartitionSpecByIceberg(metadata);
- List<IcebergPartitionSpec> partitionSpecs =
- Utils.loadPartitionSpecByIceberg(metadata);
-
- ColumnMap cmap = new ColumnMap(cols, /*numClusteringCols=*/0, fullTableName,
- false);
- return new LocalIcebergTable(db, msTable, ref, cmap, partitionSpecs);
- }
-
- private static void convertColsFromKIceberg(Schema schema, List<Column> cols,
- List<FieldSchema> fieldSchemas)
- throws TableLoadingException {
- List<Types.NestedField> columns = schema.columns();
- int pos = 0;
- for (Types.NestedField column : columns) {
- Preconditions.checkNotNull(column);
- Type colType = IcebergUtil.toImpalaType(column.type());
- fieldSchemas.add(new FieldSchema(column.name(), colType.toSql().toLowerCase(),
- column.doc()));
- cols.add(new Column(column.name(), colType, column.doc(), pos++));
+ return new LocalIcebergTable(db, msTable, ref, ColumnMap.fromMsTable(msTable),
+ partitionSpecs);
+ } catch (Exception e) {
+ String fullTableName = msTable.getDbName() + "." + msTable.getTableName();
+ throw new TableLoadingException(
+ String.format("Error opening Icberg table '%s', error: %s",
+ fullTableName, e.getMessage()));
}
}
private LocalIcebergTable(LocalDb db, Table msTable, MetaProvider.TableMetaRef ref,
- ColumnMap cmap, List<IcebergPartitionSpec> partitionSpecs) {
+ ColumnMap cmap, List<IcebergPartitionSpec> partitionSpecs)
+ throws TableLoadingException {
super(db, msTable, ref, cmap);
tableParams_ = new TableParams(msTable);
partitionSpecs_ = partitionSpecs;
+ localFsTable_ = LocalFsTable.load(db, msTable, ref);
+ try {
+ pathMD5ToFileDescMap_ = Utils.loadAllPartition(tableParams_.icebergTableLocation_,
+ this);
+ } catch (IOException e) {
+ throw new TableLoadingException(e.getMessage());
+ }
+
+ icebergFileFormat_ = Utils.getIcebergFileFormat(msTable);
+ }
+
+ @Override
+ public TIcebergFileFormat getIcebergFileFormat() {
+ return icebergFileFormat_;
}
@Override
public String getIcebergTableLocation() {
- return tableParams_.icebergTableName_;
+ return tableParams_.icebergTableLocation_;
+ }
+
+ @Override
+ public FeFsTable getFeFsTable() {
+ return localFsTable_;
}
@Override
@@ -103,33 +114,59 @@ public class LocalIcebergTable extends LocalTable implements FeIcebergTable {
}
@Override
+ public Map<String, FileDescriptor> getPathMD5ToFileDescMap() {
+ return pathMD5ToFileDescMap_;
+ }
+
+ @Override
+ protected void loadColumnStats() {
+ localFsTable_.loadColumnStats();
+ }
+
+ @Override
public TTableDescriptor toThriftDescriptor(int tableId,
- Set<Long> referencedPartitions) {
+ Set<Long> referencedPartitions) {
TTableDescriptor desc = new TTableDescriptor(tableId, TTableType.ICEBERG_TABLE,
FeCatalogUtils.getTColumnDescriptors(this),
getNumClusteringCols(),
name_, db_.getName());
- TIcebergTable tbl = new TIcebergTable();
- tbl.setTable_location(tableParams_.icebergTableName_);
- tbl.partition_spec = new ArrayList<>();
- for (IcebergPartitionSpec spec : partitionSpecs_) {
- tbl.addToPartition_spec(spec.toThrift());
- }
- desc.setIcebergTable(tbl);
+ desc.setIcebergTable(Utils.getTIcebergTable(this));
+ desc.setHdfsTable(transfromToTHdfsTable());
return desc;
}
+ private THdfsTable transfromToTHdfsTable() {
+ Map<Long, THdfsPartition> idToPartition = new HashMap<>();
+ // LocalFsTable transformed from iceberg table only has one partition
+ Collection<? extends FeFsPartition> partitions =
+ FeCatalogUtils.loadAllPartitions(localFsTable_);
+ Preconditions.checkState(partitions.size() == 1);
+ FeFsPartition partition = (FeFsPartition) partitions.toArray()[0];
+ idToPartition.put(partition.getId(),
+ FeCatalogUtils.fsPartitionToThrift(partition,
+ CatalogObject.ThriftObjectType.DESCRIPTOR_ONLY));
+
+ THdfsPartition tPrototypePartition = FeCatalogUtils.fsPartitionToThrift(
+ localFsTable_.createPrototypePartition(),
+ CatalogObject.ThriftObjectType.DESCRIPTOR_ONLY);
+ THdfsTable hdfsTable = new THdfsTable(localFsTable_.getHdfsBaseDir(),
+ getColumnNames(), localFsTable_.getNullPartitionKeyValue(),
+ FeFsTable.DEFAULT_NULL_COLUMN_VALUE, idToPartition, tPrototypePartition);
+ Utils.updateIcebergPartitionFileFormat(this, hdfsTable);
+ return hdfsTable;
+ }
+
@Immutable
private static class TableParams {
- private final String icebergTableName_;
+ private final String icebergTableLocation_;
TableParams(Table msTable) {
String fullTableName = msTable.getDbName() + "." + msTable.getTableName();
if (msTable.getSd().isSetLocation()) {
- icebergTableName_ = msTable.getSd().getLocation();
+ icebergTableLocation_ = msTable.getSd().getLocation();
} else {
- throw new LocalCatalogException("Cannot find iceberg table name for table "
+ throw new LocalCatalogException("Cannot find iceberg table location for table "
+ fullTableName);
}
}
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
index f5d6d4c..0855337 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
@@ -35,7 +35,6 @@ import org.apache.impala.analysis.Analyzer;
import org.apache.impala.analysis.BinaryPredicate;
import org.apache.impala.analysis.DescriptorTable;
import org.apache.impala.analysis.Expr;
-import org.apache.impala.analysis.ExprSubstitutionMap;
import org.apache.impala.analysis.InPredicate;
import org.apache.impala.analysis.IsNotEmptyPredicate;
import org.apache.impala.analysis.LiteralExpr;
@@ -196,7 +195,7 @@ public class HdfsScanNode extends ScanNode {
private final FeFsTable tbl_;
// List of partitions to be scanned. Partitions have been pruned.
- private final List<? extends FeFsPartition> partitions_;
+ protected final List<? extends FeFsPartition> partitions_;
// Parameters for table sampling. Null if not sampling.
private final TableSampleClause sampleParams_;
@@ -838,7 +837,13 @@ public class HdfsScanNode extends ScanNode {
maxScanRangeNumRows_ = -1;
fileFormats_ = new HashSet<>();
for (FeFsPartition partition: partitions_) {
- List<FileDescriptor> fileDescs = partition.getFileDescriptors();
+ List<FileDescriptor> fileDescs;
+ if (this instanceof IcebergScanNode) {
+ fileDescs = ((IcebergScanNode) this).getFileDescriptorByIcebergPredicates();
+ } else {
+ fileDescs = partition.getFileDescriptors();
+ }
+
if (sampledFiles != null) {
// If we are sampling, check whether this partition is included in the sample.
fileDescs = sampledFiles.get(
diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java b/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
new file mode 100644
index 0000000..5da888f
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
@@ -0,0 +1,222 @@
+// 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.planner;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.ListIterator;
+
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Expression.Operation;
+import org.apache.iceberg.expressions.UnboundPredicate;
+import org.apache.impala.analysis.Analyzer;
+import org.apache.impala.analysis.BinaryPredicate;
+import org.apache.impala.analysis.BoolLiteral;
+import org.apache.impala.analysis.DateLiteral;
+import org.apache.impala.analysis.Expr;
+import org.apache.impala.analysis.LiteralExpr;
+import org.apache.impala.analysis.MultiAggregateInfo;
+import org.apache.impala.analysis.NumericLiteral;
+import org.apache.impala.analysis.SlotRef;
+import org.apache.impala.analysis.StringLiteral;
+import org.apache.impala.analysis.TableRef;
+import org.apache.impala.analysis.TupleDescriptor;
+import org.apache.impala.catalog.FeCatalogUtils;
+import org.apache.impala.catalog.FeFsPartition;
+import org.apache.impala.catalog.FeFsTable;
+import org.apache.impala.catalog.FeIcebergTable;
+import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
+import org.apache.impala.common.ImpalaException;
+import org.apache.impala.common.ImpalaRuntimeException;
+import org.apache.impala.util.IcebergUtil;
+
+import com.google.common.base.Preconditions;
+import org.apache.impala.util.KuduUtil;
+
+/**
+ * Scan of a single iceberg table
+ */
+public class IcebergScanNode extends HdfsScanNode {
+ private final FeIcebergTable icebergTable_;
+
+ // Exprs in icebergConjuncts_ converted to UnboundPredicate.
+ private final List<UnboundPredicate> icebergPredicates_ = new ArrayList<>();
+
+ public IcebergScanNode(PlanNodeId id, TupleDescriptor desc, List<Expr> conjuncts,
+ TableRef hdfsTblRef, FeFsTable feFsTable, MultiAggregateInfo aggInfo) {
+ super(id, desc, conjuncts, getIcebergPartition(feFsTable), hdfsTblRef, aggInfo,
+ null, false);
+ icebergTable_ = (FeIcebergTable) desc_.getTable();
+ // Hdfs table transformed from iceberg table only has one partition
+ Preconditions.checkState(partitions_.size() == 1);
+ }
+
+ /**
+ * Get partition info from FeFsTable, we treat iceberg table as an
+ * unpartitioned hdfs table
+ */
+ private static List<? extends FeFsPartition> getIcebergPartition(FeFsTable feFsTable) {
+ Collection<? extends FeFsPartition> partitions =
+ FeCatalogUtils.loadAllPartitions(feFsTable);
+ return new ArrayList<>(partitions);
+ }
+
+ @Override
+ public void init(Analyzer analyzer) throws ImpalaException {
+ extractIcebergConjuncts(analyzer);
+ super.init(analyzer);
+ }
+
+ /**
+ * We need prune hdfs partition FileDescriptor by iceberg predicates
+ */
+ public List<FileDescriptor> getFileDescriptorByIcebergPredicates()
+ throws ImpalaRuntimeException{
+ List<DataFile> dataFileList = IcebergUtil.getIcebergDataFiles(
+ icebergTable_.getIcebergTableLocation(), icebergPredicates_);
+
+ List<FileDescriptor> fileDescList = new ArrayList<>();
+ for (DataFile dataFile : dataFileList) {
+ FileDescriptor fileDesc = icebergTable_.getPathMD5ToFileDescMap()
+ .get(IcebergUtil.getDataFileMD5(dataFile));
+ fileDescList.add(fileDesc);
+ //Todo: how to deal with iceberg metadata update, we need to invalidate manually now
+ if (fileDesc == null) {
+ throw new ImpalaRuntimeException("Cannot find file in cache: " + dataFile.path());
+ }
+ }
+ return fileDescList;
+ }
+
+ /**
+ * Extracts predicates from conjuncts_ that can be pushed down to Iceberg.
+ *
+ * Since Iceberg will filter data files by metadata instead of scan data files,
+ * we pushdown all predicates to Iceberg to get the minimum data files to scan.
+ * Here are three cases for predicate pushdown:
+ * 1.The column is not part of any Iceberg partition expression
+ * 2.The column is part of all partition keys without any transformation (i.e. IDENTITY)
+ * 3.The column is part of all partition keys with transformation (i.e. MONTH/DAY/HOUR)
+ * We can use case 1 and 3 to filter data files, but also need to evaluate it in the
+ * scan, for case 2 we don't need to evaluate it in the scan. So we evaluate all
+ * predicates in the scan to keep consistency. More details about Iceberg scanning,
+ * please refer: https://iceberg.apache.org/spec/#scan-planning
+ */
+ private void extractIcebergConjuncts(Analyzer analyzer) throws ImpalaException {
+ ListIterator<Expr> it = conjuncts_.listIterator();
+ while (it.hasNext()) {
+ tryConvertBinaryIcebergPredicate(analyzer, it.next());
+ }
+ }
+
+ /**
+ * Transform impala binary predicate to iceberg predicate
+ */
+ private boolean tryConvertBinaryIcebergPredicate(Analyzer analyzer, Expr expr)
+ throws ImpalaException {
+ if (! (expr instanceof BinaryPredicate)) return false;
+
+ BinaryPredicate predicate = (BinaryPredicate) expr;
+ Operation op = getIcebergOperator(predicate.getOp());
+ if (op == null) return false;
+
+ if (!(predicate.getChild(0) instanceof SlotRef)) return false;
+ SlotRef ref = (SlotRef) predicate.getChild(0);
+
+ if (!(predicate.getChild(1) instanceof LiteralExpr)) return false;
+ LiteralExpr literal = (LiteralExpr) predicate.getChild(1);
+
+ String colName = ref.getDesc().getColumn().getName();
+ UnboundPredicate unboundPredicate = null;
+ switch (literal.getType().getPrimitiveType()) {
+ case BOOLEAN: {
+ unboundPredicate = Expressions.predicate(op, colName,
+ ((BoolLiteral) literal).getValue());
+ break;
+ }
+ case TINYINT:
+ case SMALLINT:
+ case INT: {
+ unboundPredicate = Expressions.predicate(op, colName,
+ ((NumericLiteral) literal).getIntValue());
+ break;
+ }
+ case BIGINT: {
+ unboundPredicate = Expressions.predicate(op, colName,
+ ((NumericLiteral) literal).getLongValue());
+ break;
+ }
+ case FLOAT: {
+ unboundPredicate = Expressions.predicate(op, colName,
+ (float)((NumericLiteral) literal).getDoubleValue());
+ break;
+ }
+ case DOUBLE: {
+ unboundPredicate = Expressions.predicate(op, colName,
+ ((NumericLiteral) literal).getDoubleValue());
+ break;
+ }
+ case STRING:
+ case DATETIME:
+ case CHAR: {
+ unboundPredicate = Expressions.predicate(op, colName,
+ ((StringLiteral) literal).getUnescapedValue());
+ break;
+ }
+ case TIMESTAMP: {
+ unboundPredicate = Expressions.predicate(op, colName,
+ KuduUtil.timestampToUnixTimeMicros(analyzer, literal));
+ break;
+ }
+ case DATE: {
+ unboundPredicate = Expressions.predicate(op, colName,
+ ((DateLiteral) literal).getValue());
+ break;
+ }
+ case DECIMAL: {
+ unboundPredicate = Expressions.predicate(op, colName,
+ ((NumericLiteral) literal).getValue());
+ break;
+ }
+ default: break;
+ }
+ if (unboundPredicate == null) return false;
+
+ icebergPredicates_.add(unboundPredicate);
+
+ return true;
+ }
+
+ /**
+ * Returns Iceberg operator by BinaryPredicate operator, or null if the operation
+ * is not supported by Iceberg.
+ */
+ private Operation getIcebergOperator(BinaryPredicate.Operator op) {
+ switch (op) {
+ case EQ: return Operation.EQ;
+ case NE: return Operation.NOT_EQ;
+ case LE: return Operation.LT_EQ;
+ case GE: return Operation.GT_EQ;
+ case LT: return Operation.LT;
+ case GT: return Operation.GT;
+ default: return null;
+ }
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
index b046c3e..dd460e4 100644
--- a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
@@ -65,11 +65,11 @@ import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeHBaseTable;
import org.apache.impala.catalog.FeKuduTable;
+import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.HdfsPartition;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
-import org.apache.impala.catalog.IcebergTable;
import org.apache.impala.catalog.ScalarType;
import org.apache.impala.catalog.TableLoadingException;
import org.apache.impala.common.AnalysisException;
@@ -1736,6 +1736,15 @@ public class SingleNodePlanner {
// TODO(todd) introduce FE interfaces for DataSourceTable, HBaseTable, KuduTable
FeTable table = tblRef.getTable();
if (table instanceof FeFsTable) {
+ if (table instanceof FeIcebergTable) {
+ FeFsTable feFsTable = ((FeIcebergTable) tblRef.getDesc().getTable()).
+ getFeFsTable();
+ analyzer.materializeSlots(conjuncts);
+ scanNode = new IcebergScanNode(ctx_.getNextNodeId(), tblRef.getDesc(), conjuncts,
+ tblRef, feFsTable, aggInfo);
+ scanNode.init(analyzer);
+ return scanNode;
+ }
return createHdfsScanPlan(tblRef, aggInfo, conjuncts, analyzer);
} else if (table instanceof FeDataSourceTable) {
scanNode = new DataSourceScanNode(ctx_.getNextNodeId(), tblRef.getDesc(),
@@ -1753,9 +1762,6 @@ public class SingleNodePlanner {
aggInfo);
scanNode.init(analyzer);
return scanNode;
- } else if (tblRef.getTable() instanceof IcebergTable) {
- // This function will be supported in the future
- throw new NotImplementedException("Query is not supported for iceberg table now");
} else {
throw new NotImplementedException(
"Planning not implemented for table ref class: " + tblRef.getClass());
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index 1c54f84..cbd0ea3 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -2527,7 +2527,7 @@ public class CatalogOpExecutor {
}
/**
- * Creates a new Icebrg table.
+ * Creates a new Iceberg table.
*/
private boolean createIcebergTable(org.apache.hadoop.hive.metastore.api.Table newTable,
TCreateTableParams params, TDdlExecResponse response)
diff --git a/fe/src/main/java/org/apache/impala/service/Frontend.java b/fe/src/main/java/org/apache/impala/service/Frontend.java
index 4e1d45e..ba7da76 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -1188,6 +1188,9 @@ public class Frontend {
throws ImpalaException {
FeTable table = getCatalog().getTable(dbName, tableName);
if (table instanceof FeFsTable) {
+ if (table instanceof FeIcebergTable && op == TShowStatsOp.PARTITIONS) {
+ return FeIcebergTable.Utils.getPartitionSpecs((FeIcebergTable) table);
+ }
return ((FeFsTable) table).getTableStats();
} else if (table instanceof FeHBaseTable) {
return ((FeHBaseTable) table).getTableStats();
@@ -1202,12 +1205,6 @@ public class Frontend {
Preconditions.checkState(op == TShowStatsOp.TABLE_STATS);
return FeKuduTable.Utils.getTableStats((FeKuduTable) table);
}
- } else if (table instanceof FeIcebergTable) {
- if (op == TShowStatsOp.PARTITIONS) {
- return FeIcebergTable.Utils.getPartitionSpecs((FeIcebergTable) table);
- } else {
- throw new AnalysisException("Iceberg table not supported table stats now.");
- }
} else {
throw new InternalException("Invalid table class: " + table.getClass());
}
diff --git a/fe/src/main/java/org/apache/impala/util/IcebergUtil.java b/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
index 79963d6..5e79b76 100644
--- a/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
@@ -20,7 +20,13 @@ package org.apache.impala.util;
import java.util.ArrayList;
import java.util.List;
+import com.google.common.hash.Hasher;
+import com.google.common.hash.Hashing;
import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.expressions.UnboundPredicate;
import org.apache.iceberg.hadoop.HadoopTableOperations;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.PartitionField;
@@ -28,8 +34,8 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.types.Types;
-import org.apache.iceberg.types.Type.TypeID;
import org.apache.impala.catalog.ArrayType;
+import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.MapType;
import org.apache.impala.catalog.ScalarType;
import org.apache.impala.catalog.StructField;
@@ -39,6 +45,8 @@ import org.apache.impala.catalog.Type;
import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.common.ImpalaRuntimeException;
import org.apache.impala.thrift.TCreateTableParams;
+import org.apache.impala.thrift.THdfsFileFormat;
+import org.apache.impala.thrift.TIcebergFileFormat;
import org.apache.impala.thrift.TIcebergPartitionField;
import org.apache.impala.thrift.TIcebergPartitionTransform;
@@ -73,8 +81,7 @@ public class IcebergUtil {
* partition columns are all from source columns, this is different from hdfs table.
*/
public static PartitionSpec createIcebergPartition(Schema schema,
- TCreateTableParams params)
- throws ImpalaRuntimeException {
+ TCreateTableParams params) throws ImpalaRuntimeException {
if (params.getPartition_spec() == null) {
return null;
}
@@ -101,6 +108,14 @@ public class IcebergUtil {
}
/**
+ * Get TIcebergFileFormat from a string, usually from table properties
+ */
+ public static TIcebergFileFormat getIcebergFileFormat(String format){
+ if ("PARQUET".equalsIgnoreCase(format)) return TIcebergFileFormat.PARQUET;
+ return null;
+ }
+
+ /**
* Build TIcebergPartitionTransform by iceberg PartitionField
*/
public static TIcebergPartitionTransform getPartitionTransform(PartitionField field)
@@ -133,6 +148,33 @@ public class IcebergUtil {
}
/**
+ * Transform TIcebergFileFormat to THdfsFileFormat
+ */
+ public static THdfsFileFormat toTHdfsFileFormat(TIcebergFileFormat format) {
+ switch (format) {
+ case ORC:
+ return THdfsFileFormat.ORC;
+ case PARQUET:
+ default:
+ return THdfsFileFormat.PARQUET;
+ }
+ }
+
+ /**
+ * Transform TIcebergFileFormat to HdfsFileFormat
+ */
+ public static HdfsFileFormat toHdfsFileFormat(TIcebergFileFormat format) {
+ return HdfsFileFormat.fromThrift(toTHdfsFileFormat(format));
+ }
+
+ /**
+ * Transform TIcebergFileFormat to HdfsFileFormat
+ */
+ public static HdfsFileFormat toHdfsFileFormat(String format) {
+ return HdfsFileFormat.fromThrift(toTHdfsFileFormat(getIcebergFileFormat(format)));
+ }
+
+ /**
* Transform iceberg type to impala type
*/
public static Type toImpalaType(org.apache.iceberg.types.Type t)
@@ -183,4 +225,31 @@ public class IcebergUtil {
"Iceberg type '%s' is not supported in Impala", t.typeId()));
}
}
+
+ /**
+ * Get iceberg data file by file system table location and iceberg predicates
+ */
+ public static List<DataFile> getIcebergDataFiles(String location,
+ List<UnboundPredicate> predicates) {
+ BaseTable table = IcebergUtil.getBaseTable(location);
+ TableScan scan = table.newScan();
+ for (UnboundPredicate predicate : predicates) {
+ scan = scan.filter(predicate);
+ }
+
+ List<DataFile> dataFileList = new ArrayList<>();
+ for (FileScanTask task : scan.planFiles()) {
+ dataFileList.add(task.file());
+ }
+ return dataFileList;
+ }
+
+ /**
+ * Use DataFile path to construct md5 as map key, cached in memory
+ */
+ public static String getDataFileMD5(DataFile dataFile) {
+ Hasher hasher = Hashing.md5().newHasher();
+ hasher.putUnencodedChars(dataFile.path().toString());
+ return hasher.hash().toString();
+ }
}
diff --git a/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java b/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
index 8498a7b..9ccdc0c 100644
--- a/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
@@ -112,6 +112,40 @@ public class FileMetadataLoaderTest {
}
@Test
+ public void testIcebergLoading() throws IOException, CatalogException {
+ ListMap<TNetworkAddress> hostIndex = new ListMap<>();
+ Path table1Path = new Path(
+ "hdfs://localhost:20500/test-warehouse/iceberg_test/iceberg_partitioned");
+ FileMetadataLoader fml1 = new FileMetadataLoader(table1Path, /* recursive=*/true,
+ /* oldFds = */ Collections.emptyList(), hostIndex, null, null,
+ HdfsFileFormat.ICEBERG);
+ fml1.load();
+ assertEquals(25, fml1.getStats().loadedFiles);
+ assertEquals(25, fml1.getLoadedFds().size());
+
+ ArrayList<String> relPaths = new ArrayList<>(
+ Collections2.transform(fml1.getLoadedFds(), FileDescriptor::getRelativePath));
+ Collections.sort(relPaths);
+ assertEquals("data/event_time_hour=2020-01-01-08/action=view/" +
+ "00001-1-b975a171-0911-47c2-90c8-300f23c28772-00000.parquet", relPaths.get(0));
+
+ Path table2Path = new Path(
+ "hdfs://localhost:20500/test-warehouse/iceberg_test/iceberg_non_partitioned");
+ FileMetadataLoader fml2 = new FileMetadataLoader(table2Path, /* recursive=*/true,
+ /* oldFds = */ Collections.emptyList(), hostIndex, null, null,
+ HdfsFileFormat.ICEBERG);
+ fml2.load();
+ assertEquals(25, fml2.getStats().loadedFiles);
+ assertEquals(25, fml2.getLoadedFds().size());
+
+ relPaths = new ArrayList<>(
+ Collections2.transform(fml2.getLoadedFds(), FileDescriptor::getRelativePath));
+ Collections.sort(relPaths);
+ assertEquals("data/00001-1-5dbd44ad-18bc-40f2-9dd6-aeb2cc23457c-00000.parquet",
+ relPaths.get(0));
+ }
+
+ @Test
public void testAcidMinorCompactionLoading() throws IOException, CatalogException {
//TODO(IMPALA-9042): Remove "throws CatalogException"
ListMap<TNetworkAddress> hostIndex = new ListMap<>();
diff --git a/testdata/data/README b/testdata/data/README
index 2d5eb53..575a17c 100644
--- a/testdata/data/README
+++ b/testdata/data/README
@@ -523,6 +523,14 @@ kll_sketches_from_impala.parquet:
This holds the same sketches as kll_sketches_from_hive.parquet but these sketches were
created by Impala instead of Hive.
+iceberg:
+IMPALA-9741: Support querying Iceberg table by impala
+Since impala cannot insert iceberg table now, we generated data by
+spark-shell, version is 2.4.x, and there are two tables' data in
+testdata/data/iceberg_test. Iceberg table location contains
+two directories: metadata, which contains table metadata managed
+by iceberg; data, which contains the data files.
+
hudi_parquet:
IMPALA-8778: Support read Apache Hudi tables
Hudi parquet is a special format of parquet files managed by Apache Hudi
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00001-1-5dbd44ad-18bc-40f2-9dd6-aeb2cc23457c-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00001-1-5dbd44ad-18bc-40f2-9dd6-aeb2cc23457c-00000.parquet
new file mode 100644
index 0000000..80ef7b5
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00001-1-5dbd44ad-18bc-40f2-9dd6-aeb2cc23457c-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00003-3-27db2521-1e8b-40c1-b846-552cd620abce-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00003-3-27db2521-1e8b-40c1-b846-552cd620abce-00000.parquet
new file mode 100644
index 0000000..8e492a3
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00003-3-27db2521-1e8b-40c1-b846-552cd620abce-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00004-4-f1b55628-0544-4833-8b11-1b4add53dfd6-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00004-4-f1b55628-0544-4833-8b11-1b4add53dfd6-00000.parquet
new file mode 100644
index 0000000..3497c05
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00004-4-f1b55628-0544-4833-8b11-1b4add53dfd6-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00006-6-f75530ef-93b6-4994-b3c8-db957d44848c-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00006-6-f75530ef-93b6-4994-b3c8-db957d44848c-00000.parquet
new file mode 100644
index 0000000..e6bcbb0
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00006-6-f75530ef-93b6-4994-b3c8-db957d44848c-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00007-7-8d9b22da-5f10-4cbf-8e4d-160f829b5e48-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00007-7-8d9b22da-5f10-4cbf-8e4d-160f829b5e48-00000.parquet
new file mode 100644
index 0000000..d438fa6
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00007-7-8d9b22da-5f10-4cbf-8e4d-160f829b5e48-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00009-9-f029a1f7-9024-4bc3-a030-e20861586146-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00009-9-f029a1f7-9024-4bc3-a030-e20861586146-00000.parquet
new file mode 100644
index 0000000..bb699f8
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00009-9-f029a1f7-9024-4bc3-a030-e20861586146-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00011-11-f07814ae-56cd-486b-af81-18541437da7d-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00011-11-f07814ae-56cd-486b-af81-18541437da7d-00000.parquet
new file mode 100644
index 0000000..be8c330
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00011-11-f07814ae-56cd-486b-af81-18541437da7d-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00012-12-967c70a4-bf4d-4a82-8c97-c90e2b4d9dcf-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00012-12-967c70a4-bf4d-4a82-8c97-c90e2b4d9dcf-00000.parquet
new file mode 100644
index 0000000..4198171
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00012-12-967c70a4-bf4d-4a82-8c97-c90e2b4d9dcf-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00014-14-d0cdca7f-c050-407e-b70c-2bd076f83e4e-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00014-14-d0cdca7f-c050-407e-b70c-2bd076f83e4e-00000.parquet
new file mode 100644
index 0000000..fd9658c
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00014-14-d0cdca7f-c050-407e-b70c-2bd076f83e4e-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00015-15-0e931a1f-309e-43b3-a5cf-3ef82fa4a87c-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00015-15-0e931a1f-309e-43b3-a5cf-3ef82fa4a87c-00000.parquet
new file mode 100644
index 0000000..b9f2758
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00015-15-0e931a1f-309e-43b3-a5cf-3ef82fa4a87c-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00017-17-43138078-244c-4b38-8127-04a5bfbc4695-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00017-17-43138078-244c-4b38-8127-04a5bfbc4695-00000.parquet
new file mode 100644
index 0000000..be3530f
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00017-17-43138078-244c-4b38-8127-04a5bfbc4695-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00019-19-52569895-df25-4ad8-b64d-49c4540d36c9-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00019-19-52569895-df25-4ad8-b64d-49c4540d36c9-00000.parquet
new file mode 100644
index 0000000..e5c0cc6
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00019-19-52569895-df25-4ad8-b64d-49c4540d36c9-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00020-20-f160c1ea-a2f5-4109-81ec-3ff9c155430f-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00020-20-f160c1ea-a2f5-4109-81ec-3ff9c155430f-00000.parquet
new file mode 100644
index 0000000..53e5944
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00020-20-f160c1ea-a2f5-4109-81ec-3ff9c155430f-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00022-22-c1f61b8c-9d9a-4823-b64e-109770c16696-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00022-22-c1f61b8c-9d9a-4823-b64e-109770c16696-00000.parquet
new file mode 100644
index 0000000..e0bb8db
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00022-22-c1f61b8c-9d9a-4823-b64e-109770c16696-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00023-23-8cfc9d65-bfc3-47c4-8da7-6610b49b3305-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00023-23-8cfc9d65-bfc3-47c4-8da7-6610b49b3305-00000.parquet
new file mode 100644
index 0000000..64b0b2b
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00023-23-8cfc9d65-bfc3-47c4-8da7-6610b49b3305-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00025-25-d519b09d-c730-40a6-ad03-8b36096cb234-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00025-25-d519b09d-c730-40a6-ad03-8b36096cb234-00000.parquet
new file mode 100644
index 0000000..680c94e
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00025-25-d519b09d-c730-40a6-ad03-8b36096cb234-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00027-27-802b2129-0d18-418a-bcdd-b1b5332ea2a5-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00027-27-802b2129-0d18-418a-bcdd-b1b5332ea2a5-00000.parquet
new file mode 100644
index 0000000..a37066f
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00027-27-802b2129-0d18-418a-bcdd-b1b5332ea2a5-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00028-28-70f51a0c-3d30-4cdb-afcb-e8c1cfa3caa4-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00028-28-70f51a0c-3d30-4cdb-afcb-e8c1cfa3caa4-00000.parquet
new file mode 100644
index 0000000..06fe5bb
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00028-28-70f51a0c-3d30-4cdb-afcb-e8c1cfa3caa4-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00030-30-e887707d-58db-469c-ab96-b77188f25189-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00030-30-e887707d-58db-469c-ab96-b77188f25189-00000.parquet
new file mode 100644
index 0000000..abed000
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00030-30-e887707d-58db-469c-ab96-b77188f25189-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/data/00031-31-facf6a62-c326-44f7-bdc3-b1471bd594d7-00000.parquet b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00031-31-facf6a62-c326-44f7-bdc3-b1471bd594d7-00000.parquet
new file mode 100644
index 0000000..791d910
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/data/00031-31-facf6a62-c326-44f7-bdc3-b1471bd594d7-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21-m0.avro b/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21-m0.avro
new file mode 100644
index 0000000..5b7ac32
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21-m0.avro differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/snap-93996984692289973-1-9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21.avro b/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/snap-93996984692289973-1-9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21.avro
new file mode 100644
index 0000000..debfc88
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/snap-93996984692289973-1-9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21.avro differ
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/v1.metadata.json b/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/v1.metadata.json
new file mode 100644
index 0000000..b672f66
--- /dev/null
+++ b/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/v1.metadata.json
@@ -0,0 +1,42 @@
+{
+ "format-version" : 1,
+ "table-uuid" : "cc9afe4b-8922-4b67-8170-2a201e41261e",
+ "location" : "/test-warehouse/iceberg_test/iceberg_non_partitioned",
+ "last-updated-ms" : 1598853012619,
+ "last-column-id" : 4,
+ "schema" : {
+ "type" : "struct",
+ "fields" : [ {
+ "id" : 1,
+ "name" : "id",
+ "required" : false,
+ "type" : "int"
+ }, {
+ "id" : 2,
+ "name" : "user",
+ "required" : true,
+ "type" : "string"
+ }, {
+ "id" : 3,
+ "name" : "action",
+ "required" : true,
+ "type" : "string"
+ }, {
+ "id" : 4,
+ "name" : "event_time",
+ "required" : true,
+ "type" : "timestamptz"
+ } ]
+ },
+ "partition-spec" : [ ],
+ "default-spec-id" : 0,
+ "partition-specs" : [ {
+ "spec-id" : 0,
+ "fields" : [ ]
+ } ],
+ "properties" : { },
+ "current-snapshot-id" : -1,
+ "snapshots" : [ ],
+ "snapshot-log" : [ ],
+ "metadata-log" : [ ]
+}
\ No newline at end of file
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/v2.metadata.json b/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/v2.metadata.json
new file mode 100644
index 0000000..5ea2845
--- /dev/null
+++ b/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/v2.metadata.json
@@ -0,0 +1,61 @@
+{
+ "format-version" : 1,
+ "table-uuid" : "cc9afe4b-8922-4b67-8170-2a201e41261e",
+ "location" : "/test-warehouse/iceberg_test/iceberg_non_partitioned",
+ "last-updated-ms" : 1598853017107,
+ "last-column-id" : 4,
+ "schema" : {
+ "type" : "struct",
+ "fields" : [ {
+ "id" : 1,
+ "name" : "id",
+ "required" : false,
+ "type" : "int"
+ }, {
+ "id" : 2,
+ "name" : "user",
+ "required" : true,
+ "type" : "string"
+ }, {
+ "id" : 3,
+ "name" : "action",
+ "required" : true,
+ "type" : "string"
+ }, {
+ "id" : 4,
+ "name" : "event_time",
+ "required" : true,
+ "type" : "timestamptz"
+ } ]
+ },
+ "partition-spec" : [ ],
+ "default-spec-id" : 0,
+ "partition-specs" : [ {
+ "spec-id" : 0,
+ "fields" : [ ]
+ } ],
+ "properties" : { },
+ "current-snapshot-id" : 93996984692289973,
+ "snapshots" : [ {
+ "snapshot-id" : 93996984692289973,
+ "timestamp-ms" : 1598853017107,
+ "summary" : {
+ "operation" : "append",
+ "spark.app.id" : "local-1598853006476",
+ "added-data-files" : "20",
+ "added-records" : "20",
+ "changed-partition-count" : "1",
+ "total-records" : "20",
+ "total-data-files" : "20"
+ },
+ "manifest-list" : "/test-warehouse/iceberg_test/iceberg_non_partitioned/metadata/snap-93996984692289973-1-9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21.avro"
+ } ],
+ "snapshot-log" : [ {
+ "timestamp-ms" : 1598853017107,
+ "snapshot-id" : 93996984692289973
+ } ],
+ "metadata-log" : [ {
+ "timestamp-ms" : 1598853012619,
+ "metadata-file" : "/test-warehouse/iceberg_test/iceberg_non_partitioned/metadata/v1.metadata.json"
+ } ]
+}
\ No newline at end of file
diff --git a/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/version-hint.text b/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/version-hint.text
new file mode 100644
index 0000000..d8263ee
--- /dev/null
+++ b/testdata/data/iceberg_test/iceberg_non_partitioned/metadata/version-hint.text
@@ -0,0 +1 @@
+2
\ No newline at end of file
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00001-1-b975a171-0911-47c2-90c8-300f23c28772-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00001-1-b975a171-0911-47c2-90c8-300f23c28772-00000.parquet
new file mode 100644
index 0000000..1189a11
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00001-1-b975a171-0911-47c2-90c8-300f23c28772-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00006-6-9bc6f547-de52-4aad-8c07-5d0f1aaa0558-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00006-6-9bc6f547-de52-4aad-8c07-5d0f1aaa0558-00000.parquet
new file mode 100644
index 0000000..991a6c3
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00006-6-9bc6f547-de52-4aad-8c07-5d0f1aaa0558-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00009-9-bf3024d6-8b55-42d3-b25e-5f2a31303026-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00009-9-bf3024d6-8b55-42d3-b25e-5f2a31303026-00000.parquet
new file mode 100644
index 0000000..640e34b
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00009-9-bf3024d6-8b55-42d3-b25e-5f2a31303026-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00017-17-a0d23269-c8d7-40cf-94c8-8b624969b9c0-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00017-17-a0d23269-c8d7-40cf-94c8-8b624969b9c0-00000.parquet
new file mode 100644
index 0000000..ec1c852
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00017-17-a0d23269-c8d7-40cf-94c8-8b624969b9c0-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00023-23-138a376e-696e-4da3-95f3-7e0aacb11ca3-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00023-23-138a376e-696e-4da3-95f3-7e0aacb11ca3-00000.parquet
new file mode 100644
index 0000000..861cb5f
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00023-23-138a376e-696e-4da3-95f3-7e0aacb11ca3-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00027-27-04494175-75de-41ef-b824-3923ebfba0cb-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00027-27-04494175-75de-41ef-b824-3923ebfba0cb-00000.parquet
new file mode 100644
index 0000000..dccd0c7
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00027-27-04494175-75de-41ef-b824-3923ebfba0cb-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00030-30-eba34050-b263-41e8-bc7d-773a47b33c32-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00030-30-eba34050-b263-41e8-bc7d-773a47b33c32-00000.parquet
new file mode 100644
index 0000000..cb8475a
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00030-30-eba34050-b263-41e8-bc7d-773a47b33c32-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00031-31-0ec1c3ff-7154-4fd1-a090-8fa0b9c2d3c8-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00031-31-0ec1c3ff-7154-4fd1-a090-8fa0b9c2d3c8-00000.parquet
new file mode 100644
index 0000000..19f648e
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00031-31-0ec1c3ff-7154-4fd1-a090-8fa0b9c2d3c8-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00004-4-8b99da05-6be1-42e1-90e0-bff8ca08b5af-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00004-4-8b99da05-6be1-42e1-90e0-bff8ca08b5af-00000.parquet
new file mode 100644
index 0000000..4971d1f
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00004-4-8b99da05-6be1-42e1-90e0-bff8ca08b5af-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00014-14-5b39012d-85e3-4144-ac83-574ed5046718-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00014-14-5b39012d-85e3-4144-ac83-574ed5046718-00000.parquet
new file mode 100644
index 0000000..0fea555
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00014-14-5b39012d-85e3-4144-ac83-574ed5046718-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00015-15-e37e3f98-5f3b-4245-9c01-76cf0f1dd21a-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00015-15-e37e3f98-5f3b-4245-9c01-76cf0f1dd21a-00000.parquet
new file mode 100644
index 0000000..c8fb783
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00015-15-e37e3f98-5f3b-4245-9c01-76cf0f1dd21a-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00019-19-e36a1955-f01c-4e08-9f19-dcec0cbceadd-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00019-19-e36a1955-f01c-4e08-9f19-dcec0cbceadd-00000.parquet
new file mode 100644
index 0000000..8781b35
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00019-19-e36a1955-f01c-4e08-9f19-dcec0cbceadd-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00020-20-c9632a29-fd5b-4ba5-a0db-2992c73075c2-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00020-20-c9632a29-fd5b-4ba5-a0db-2992c73075c2-00000.parquet
new file mode 100644
index 0000000..d42f1a8
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00020-20-c9632a29-fd5b-4ba5-a0db-2992c73075c2-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00028-28-31220e43-f85a-4609-9ebb-4c76269cddb6-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00028-28-31220e43-f85a-4609-9ebb-4c76269cddb6-00000.parquet
new file mode 100644
index 0000000..220f4c8
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00028-28-31220e43-f85a-4609-9ebb-4c76269cddb6-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00003-3-7e47d440-44f5-4a11-bbfa-c72c520ca997-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00003-3-7e47d440-44f5-4a11-bbfa-c72c520ca997-00000.parquet
new file mode 100644
index 0000000..0df70ce
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00003-3-7e47d440-44f5-4a11-bbfa-c72c520ca997-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00007-7-c902540c-b002-413c-b422-7ec3c06d404e-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00007-7-c902540c-b002-413c-b422-7ec3c06d404e-00000.parquet
new file mode 100644
index 0000000..3e4eaee
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00007-7-c902540c-b002-413c-b422-7ec3c06d404e-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00011-11-5558f0e5-7a46-417e-8c0a-50bb00e54a30-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00011-11-5558f0e5-7a46-417e-8c0a-50bb00e54a30-00000.parquet
new file mode 100644
index 0000000..a3f03cf
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00011-11-5558f0e5-7a46-417e-8c0a-50bb00e54a30-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00012-12-ea94b7c7-164d-4d06-8b4d-24d296906f73-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00012-12-ea94b7c7-164d-4d06-8b4d-24d296906f73-00000.parquet
new file mode 100644
index 0000000..270a96d
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00012-12-ea94b7c7-164d-4d06-8b4d-24d296906f73-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00022-22-0565b9eb-060d-4180-aa1e-2bb69187d8db-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00022-22-0565b9eb-060d-4180-aa1e-2bb69187d8db-00000.parquet
new file mode 100644
index 0000000..7a17b2d
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00022-22-0565b9eb-060d-4180-aa1e-2bb69187d8db-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00025-25-1f2b05c6-5407-41fc-a881-60626253ca77-00000.parquet b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00025-25-1f2b05c6-5407-41fc-a881-60626253ca77-00000.parquet
new file mode 100644
index 0000000..4eca66e
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00025-25-1f2b05c6-5407-41fc-a881-60626253ca77-00000.parquet differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/metadata/af797bab-2f2c-44df-a77b-d91c7198fe53-m0.avro b/testdata/data/iceberg_test/iceberg_partitioned/metadata/af797bab-2f2c-44df-a77b-d91c7198fe53-m0.avro
new file mode 100644
index 0000000..59634c2
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/metadata/af797bab-2f2c-44df-a77b-d91c7198fe53-m0.avro differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/metadata/snap-8270633197658268308-1-af797bab-2f2c-44df-a77b-d91c7198fe53.avro b/testdata/data/iceberg_test/iceberg_partitioned/metadata/snap-8270633197658268308-1-af797bab-2f2c-44df-a77b-d91c7198fe53.avro
new file mode 100644
index 0000000..64ae021
Binary files /dev/null and b/testdata/data/iceberg_test/iceberg_partitioned/metadata/snap-8270633197658268308-1-af797bab-2f2c-44df-a77b-d91c7198fe53.avro differ
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/metadata/v1.metadata.json b/testdata/data/iceberg_test/iceberg_partitioned/metadata/v1.metadata.json
new file mode 100644
index 0000000..0ea08db
--- /dev/null
+++ b/testdata/data/iceberg_test/iceberg_partitioned/metadata/v1.metadata.json
@@ -0,0 +1,62 @@
+{
+ "format-version" : 1,
+ "table-uuid" : "78e22bda-dd51-4c0d-bfa8-22d2f4fdf714",
+ "location" : "/test-warehouse/iceberg_test/iceberg_partitioned",
+ "last-updated-ms" : 1598853483938,
+ "last-column-id" : 4,
+ "schema" : {
+ "type" : "struct",
+ "fields" : [ {
+ "id" : 1,
+ "name" : "id",
+ "required" : false,
+ "type" : "int"
+ }, {
+ "id" : 2,
+ "name" : "user",
+ "required" : true,
+ "type" : "string"
+ }, {
+ "id" : 3,
+ "name" : "action",
+ "required" : true,
+ "type" : "string"
+ }, {
+ "id" : 4,
+ "name" : "event_time",
+ "required" : true,
+ "type" : "timestamptz"
+ } ]
+ },
+ "partition-spec" : [ {
+ "name" : "event_time_hour",
+ "transform" : "hour",
+ "source-id" : 4,
+ "field-id" : 1000
+ }, {
+ "name" : "action",
+ "transform" : "identity",
+ "source-id" : 3,
+ "field-id" : 1001
+ } ],
+ "default-spec-id" : 0,
+ "partition-specs" : [ {
+ "spec-id" : 0,
+ "fields" : [ {
+ "name" : "event_time_hour",
+ "transform" : "hour",
+ "source-id" : 4,
+ "field-id" : 1000
+ }, {
+ "name" : "action",
+ "transform" : "identity",
+ "source-id" : 3,
+ "field-id" : 1001
+ } ]
+ } ],
+ "properties" : { },
+ "current-snapshot-id" : -1,
+ "snapshots" : [ ],
+ "snapshot-log" : [ ],
+ "metadata-log" : [ ]
+}
\ No newline at end of file
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/metadata/v2.metadata.json b/testdata/data/iceberg_test/iceberg_partitioned/metadata/v2.metadata.json
new file mode 100644
index 0000000..6738660
--- /dev/null
+++ b/testdata/data/iceberg_test/iceberg_partitioned/metadata/v2.metadata.json
@@ -0,0 +1,81 @@
+{
+ "format-version" : 1,
+ "table-uuid" : "78e22bda-dd51-4c0d-bfa8-22d2f4fdf714",
+ "location" : "/test-warehouse/iceberg_test/iceberg_partitioned",
+ "last-updated-ms" : 1598853488440,
+ "last-column-id" : 4,
+ "schema" : {
+ "type" : "struct",
+ "fields" : [ {
+ "id" : 1,
+ "name" : "id",
+ "required" : false,
+ "type" : "int"
+ }, {
+ "id" : 2,
+ "name" : "user",
+ "required" : true,
+ "type" : "string"
+ }, {
+ "id" : 3,
+ "name" : "action",
+ "required" : true,
+ "type" : "string"
+ }, {
+ "id" : 4,
+ "name" : "event_time",
+ "required" : true,
+ "type" : "timestamptz"
+ } ]
+ },
+ "partition-spec" : [ {
+ "name" : "event_time_hour",
+ "transform" : "hour",
+ "source-id" : 4,
+ "field-id" : 1000
+ }, {
+ "name" : "action",
+ "transform" : "identity",
+ "source-id" : 3,
+ "field-id" : 1001
+ } ],
+ "default-spec-id" : 0,
+ "partition-specs" : [ {
+ "spec-id" : 0,
+ "fields" : [ {
+ "name" : "event_time_hour",
+ "transform" : "hour",
+ "source-id" : 4,
+ "field-id" : 1000
+ }, {
+ "name" : "action",
+ "transform" : "identity",
+ "source-id" : 3,
+ "field-id" : 1001
+ } ]
+ } ],
+ "properties" : { },
+ "current-snapshot-id" : 8270633197658268308,
+ "snapshots" : [ {
+ "snapshot-id" : 8270633197658268308,
+ "timestamp-ms" : 1598853488440,
+ "summary" : {
+ "operation" : "append",
+ "spark.app.id" : "local-1598853475123",
+ "added-data-files" : "20",
+ "added-records" : "20",
+ "changed-partition-count" : "3",
+ "total-records" : "20",
+ "total-data-files" : "20"
+ },
+ "manifest-list" : "/test-warehouse/iceberg_test/iceberg_partitioned/metadata/snap-8270633197658268308-1-af797bab-2f2c-44df-a77b-d91c7198fe53.avro"
+ } ],
+ "snapshot-log" : [ {
+ "timestamp-ms" : 1598853488440,
+ "snapshot-id" : 8270633197658268308
+ } ],
+ "metadata-log" : [ {
+ "timestamp-ms" : 1598853483938,
+ "metadata-file" : "/test-warehouse/iceberg_test/iceberg_partitioned/metadata/v1.metadata.json"
+ } ]
+}
\ No newline at end of file
diff --git a/testdata/data/iceberg_test/iceberg_partitioned/metadata/version-hint.text b/testdata/data/iceberg_test/iceberg_partitioned/metadata/version-hint.text
new file mode 100644
index 0000000..d8263ee
--- /dev/null
+++ b/testdata/data/iceberg_test/iceberg_partitioned/metadata/version-hint.text
@@ -0,0 +1 @@
+2
\ No newline at end of file
diff --git a/testdata/datasets/functional/functional_schema_template.sql b/testdata/datasets/functional/functional_schema_template.sql
index 0f4a8dc..0432e6d 100644
--- a/testdata/datasets/functional/functional_schema_template.sql
+++ b/testdata/datasets/functional/functional_schema_template.sql
@@ -2933,3 +2933,29 @@ case when id % 2 = 0 then cast(timestamp_col as date)
else cast(cast(timestamp_col as date) + interval 5 days as date) end date_col
FROM {db_name}{db_suffix}.alltypes where id < 500;
====
+---- DATASET
+functional
+---- BASE_TABLE_NAME
+iceberg_partitioned
+---- CREATE
+CREATE EXTERNAL TABLE IF NOT EXISTS {db_name}{db_suffix}.{table_name}
+STORED AS ICEBERG
+LOCATION '/test-warehouse/iceberg_test/iceberg_partitioned'
+TBLPROPERTIES('iceberg_file_format'='parquet');
+---- DEPENDENT_LOAD
+`hadoop fs -mkdir -p /test-warehouse/iceberg_test && \
+hadoop fs -put -f ${IMPALA_HOME}/testdata/data/iceberg_test/iceberg_partitioned /test-warehouse/iceberg_test/
+====
+---- DATASET
+functional
+---- BASE_TABLE_NAME
+iceberg_non_partitioned
+---- CREATE
+CREATE EXTERNAL TABLE IF NOT EXISTS {db_name}{db_suffix}.{table_name}
+STORED AS ICEBERG
+LOCATION '/test-warehouse/iceberg_test/iceberg_non_partitioned'
+TBLPROPERTIES('iceberg_file_format'='parquet');
+---- DEPENDENT_LOAD
+`hadoop fs -mkdir -p /test-warehouse/iceberg_test && \
+hadoop fs -put -f ${IMPALA_HOME}/testdata/data/iceberg_test/iceberg_non_partitioned /test-warehouse/iceberg_test/
+====
diff --git a/testdata/datasets/functional/schema_constraints.csv b/testdata/datasets/functional/schema_constraints.csv
index bfacbff..39b724f 100644
--- a/testdata/datasets/functional/schema_constraints.csv
+++ b/testdata/datasets/functional/schema_constraints.csv
@@ -61,6 +61,8 @@ table_name:customer_multiblock, constraint:restrict_to, table_format:parquet/non
table_name:hudi_partitioned, constraint:restrict_to, table_format:parquet/none/none
table_name:hudi_non_partitioned, constraint:restrict_to, table_format:parquet/none/none
table_name:hudi_as_parquet, constraint:restrict_to, table_format:parquet/none/none
+table_name:iceberg_partitioned, constraint:restrict_to, table_format:parquet/none/none
+table_name:iceberg_non_partitioned, constraint:restrict_to, table_format:parquet/none/none
# TODO: Support Avro. Data loading currently fails for Avro because complex types
# cannot be converted to the corresponding Avro types yet.
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg_create.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-create.test
similarity index 77%
rename from testdata/workloads/functional-query/queries/QueryTest/iceberg_create.test
rename to testdata/workloads/functional-query/queries/QueryTest/iceberg-create.test
index c2d2603..f751335 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg_create.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-create.test
@@ -60,31 +60,6 @@ DESCRIBE iceberg_test2;
STRING,STRING,STRING
====
---- QUERY
-SHOW PARTITIONS iceberg_test2;
----- CATCH
-AnalysisException: Iceberg table does not have PartitionSpec: $DATABASE.iceberg_test2
-====
----- QUERY
-CREATE TABLE iceberg_test3(
- level STRING
-)
-PARTITION BY SPEC
-(
- level IDENTITY,
- event_time HOUR
-)
-STORED AS ICEBERG;
----- CATCH
-ImpalaRuntimeException: Error making 'createTable' RPC to Hive Metastore:
-CAUSED BY: IllegalArgumentException: Cannot find source column: event_time
-====
----- QUERY
-CREATE TABLE iceberg_test3
-STORED AS ICEBERG;
----- CATCH
-AnalysisException: Table requires at least 1 column for managed iceberg table.
-====
----- QUERY
CREATE EXTERNAL TABLE iceberg_test_external(
level STRING
)
@@ -101,12 +76,12 @@ PARTITION BY SPEC
level IDENTITY
)
STORED AS ICEBERG
-LOCATION '/iceberg_test_with_location';
+LOCATION '/$DATABASE.iceberg_test_with_location';
CREATE EXTERNAL TABLE iceberg_test_external(
level STRING
)
STORED AS ICEBERG
-LOCATION '/iceberg_test_with_location';
+LOCATION '/$DATABASE.iceberg_test_with_location';
---- RESULTS
'Table has been created.'
====
@@ -127,7 +102,7 @@ BIGINT,BIGINT,BIGINT,STRING,STRING
---- QUERY
CREATE EXTERNAL TABLE iceberg_test_external_empty_column
STORED AS ICEBERG
-LOCATION '/iceberg_test_with_location';
+LOCATION '/$DATABASE.iceberg_test_with_location';
---- RESULTS
'Table has been created.'
====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test
new file mode 100644
index 0000000..759880e
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test
@@ -0,0 +1,45 @@
+====
+---- QUERY
+CREATE TABLE iceberg_test1
+STORED AS ICEBERG;
+---- CATCH
+AnalysisException: Table requires at least 1 column for managed iceberg table.
+====
+---- QUERY
+CREATE TABLE iceberg_test2(
+ level STRING
+)
+PARTITION BY SPEC
+(
+ level IDENTITY,
+ event_time HOUR
+)
+STORED AS ICEBERG;
+---- CATCH
+AnalysisException: Cannot find source column: event_time
+====
+---- QUERY
+CREATE TABLE iceberg_test3(
+ level STRING
+)
+STORED AS ICEBERG;
+INSERT INTO iceberg_test3 values('1');
+---- CATCH
+AnalysisException: Impala does not support INSERTing into iceberg table: $DATABASE.iceberg_test3
+====
+---- QUERY
+TRUNCATE iceberg_test3
+---- CATCH
+AnalysisException: TRUNCATE TABLE not supported on iceberg table: $DATABASE.iceberg_test3
+====
+---- QUERY
+ALTER TABLE iceberg_test3 ADD COLUMN event_time TIMESTAMP
+---- CATCH
+AnalysisException: ALTER TABLE not allowed on iceberg table: iceberg_test3
+====
+---- QUERY
+# iceberg_non_partitioned is not partitioned
+SHOW PARTITIONS functional_parquet.iceberg_non_partitioned
+---- CATCH
+AnalysisException: Table is not partitioned: functional_parquet.iceberg_non_partitioned
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-profile.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-profile.test
new file mode 100644
index 0000000..fce7e1a
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-profile.test
@@ -0,0 +1,20 @@
+====
+---- QUERY
+# This query will filter files by partition column predicate
+select * from functional_parquet.iceberg_partitioned
+where event_time > to_timestamp('2020-01-01 09:00:00','yyyy-MM-dd HH:mm:ss')
+---- RUNTIME_PROFILE
+aggregation(SUM, RowsRead): 6
+====
+---- QUERY
+# This query will scan a file which id = 1, filter by predicate for parquet file
+select * from functional_parquet.iceberg_partitioned where id = 1
+---- RUNTIME_PROFILE
+aggregation(SUM, RowsRead): 1
+====
+---- QUERY
+# This query will do a full scan
+select * from functional_parquet.iceberg_partitioned
+---- RUNTIME_PROFILE
+aggregation(SUM, RowsRead): 20
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-query.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-query.test
new file mode 100644
index 0000000..67e081d
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-query.test
@@ -0,0 +1,185 @@
+====
+---- QUERY
+# IMPALA-9741: Support querying Iceberg table by impala
+SELECT count(*) from iceberg_partitioned;
+---- TYPES
+bigint
+---- RESULTS
+20
+====
+---- QUERY
+SELECT count(*) from iceberg_non_partitioned
+---- TYPES
+bigint
+---- RESULTS
+20
+====
+---- QUERY
+# Test non partition column filter count
+SELECT count(*) from iceberg_partitioned
+where id > 10
+---- TYPES
+bigint
+---- RESULTS
+10
+====
+---- QUERY
+SELECT count(*) from iceberg_non_partitioned
+where id > 10
+---- TYPES
+bigint
+---- RESULTS
+10
+====
+---- QUERY
+# Test partition column filter count
+SELECT count(*) from iceberg_partitioned
+where action <> 'click'
+---- TYPES
+bigint
+---- RESULTS
+14
+====
+---- QUERY
+SELECT count(*) from iceberg_non_partitioned
+where action = 'click'
+---- TYPES
+bigint
+---- RESULTS
+6
+====
+---- QUERY
+# Test distinct count
+SELECT count(distinct id),count(distinct action) from iceberg_partitioned
+---- TYPES
+bigint,bigint
+---- RESULTS
+20,3
+====
+---- QUERY
+SELECT count(distinct id),count(distinct action) from iceberg_non_partitioned
+---- TYPES
+bigint,bigint
+---- RESULTS
+20,3
+====
+---- QUERY
+# Test timestamp filter for iceberg derived partition 'HOUR'
+SELECT count(*) from iceberg_partitioned
+where event_time > to_timestamp('2020-01-01 09:00:00','yyyy-MM-dd HH:mm:ss')
+---- TYPES
+bigint
+---- RESULTS
+6
+====
+---- QUERY
+SELECT count(*) from iceberg_non_partitioned
+where event_time > to_timestamp('2020-01-01 09:00:00','yyyy-MM-dd HH:mm:ss')
+---- TYPES
+bigint
+---- RESULTS
+6
+====
+---- QUERY
+# Test non-count query for partition filter
+SELECT * from iceberg_partitioned
+where event_time > to_timestamp('2020-01-01 09:00:00','yyyy-MM-dd HH:mm:ss')
+ORDER BY id
+---- TYPES
+int,string,string,timestamp
+---- RESULTS
+2,'Lisa','download',2020-01-01 10:00:00
+5,'Lisa','download',2020-01-01 10:00:00
+7,'Lisa','download',2020-01-01 10:00:00
+8,'Lisa','download',2020-01-01 10:00:00
+14,'Lisa','download',2020-01-01 10:00:00
+16,'Lisa','download',2020-01-01 10:00:00
+====
+---- QUERY
+# Test non-count query for non-partition filter
+SELECT * from iceberg_non_partitioned
+where id < 5
+ORDER BY id
+---- TYPES
+int,string,string,timestamp
+---- RESULTS
+1,'Alex','view',2020-01-01 08:00:00
+2,'Lisa','download',2020-01-01 10:00:00
+3,'Alan','click',2020-01-01 09:00:00
+4,'Alex','view',2020-01-01 08:00:00
+====
+---- QUERY
+# Test show partitions query for iceberg table
+SHOW PARTITIONS iceberg_partitioned
+---- TYPES
+bigint,bigint,bigint,string,string
+---- RESULTS
+0,4,1000,'event_time_hour','HOUR'
+0,3,1001,'action','IDENTITY'
+====
+---- QUERY
+SHOW FILES in iceberg_partitioned
+---- LABELS
+Path,Size,Partition
+---- RESULTS
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00001-1-b975a171-0911-47c2-90c8-300f23c28772-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00006-6-9bc6f547-de52-4aad-8c07-5d0f1aaa0558-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00009-9-bf3024d6-8b55-42d3-b25e-5f2a31303026-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00017-17-a0d23269-c8d7-40cf-94c8-8b624969b9c0-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00023-23-138a376e-696e-4da3-95f3-7e0aacb11ca3-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00027-27-04494175-75de-41ef-b824-3923ebfba0cb-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00030-30-eba34050-b263-41e8-bc7d-773a47b33c32-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-08/action=view/00031-31-0ec1c3ff-7154-4fd1-a090-8fa0b9c2d3c8-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00004-4-8b99da05-6be1-42e1-90e0-bff8ca08b5af-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00014-14-5b39012d-85e3-4144-ac83-574ed5046718-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00015-15-e37e3f98-5f3b-4245-9c01-76cf0f1dd21a-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00019-19-e36a1955-f01c-4e08-9f19-dcec0cbceadd-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00020-20-c9632a29-fd5b-4ba5-a0db-2992c73075c2-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-09/action=click/00028-28-31220e43-f85a-4609-9ebb-4c76269cddb6-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00003-3-7e47d440-44f5-4a11-bbfa-c72c520ca997-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00007-7-c902540c-b002-413c-b422-7ec3c06d404e-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00011-11-5558f0e5-7a46-417e-8c0a-50bb00e54a30-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00012-12-ea94b7c7-164d-4d06-8b4d-24d296906f73-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00022-22-0565b9eb-060d-4180-aa1e-2bb69187d8db-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/data/event_time_hour=2020-01-01-10/action=download/00025-25-1f2b05c6-5407-41fc-a881-60626253ca77-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/metadata/af797bab-2f2c-44df-a77b-d91c7198fe53-m0.avro',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/metadata/snap-8270633197658268308-1-af797bab-2f2c-44df-a77b-d91c7198fe53.avro',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/metadata/v1.metadata.json',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/metadata/v2.metadata.json',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_partitioned/metadata/version-hint.text',regex:.*,''
+---- TYPES
+STRING,STRING,STRING
+====
+---- QUERY
+SHOW FILES in iceberg_non_partitioned
+---- LABELS
+Path,Size,Partition
+---- RESULTS
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00001-1-5dbd44ad-18bc-40f2-9dd6-aeb2cc23457c-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00003-3-27db2521-1e8b-40c1-b846-552cd620abce-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00004-4-f1b55628-0544-4833-8b11-1b4add53dfd6-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00006-6-f75530ef-93b6-4994-b3c8-db957d44848c-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00007-7-8d9b22da-5f10-4cbf-8e4d-160f829b5e48-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00009-9-f029a1f7-9024-4bc3-a030-e20861586146-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00011-11-f07814ae-56cd-486b-af81-18541437da7d-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00012-12-967c70a4-bf4d-4a82-8c97-c90e2b4d9dcf-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00014-14-d0cdca7f-c050-407e-b70c-2bd076f83e4e-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00015-15-0e931a1f-309e-43b3-a5cf-3ef82fa4a87c-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00017-17-43138078-244c-4b38-8127-04a5bfbc4695-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00019-19-52569895-df25-4ad8-b64d-49c4540d36c9-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00020-20-f160c1ea-a2f5-4109-81ec-3ff9c155430f-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00022-22-c1f61b8c-9d9a-4823-b64e-109770c16696-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00023-23-8cfc9d65-bfc3-47c4-8da7-6610b49b3305-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00025-25-d519b09d-c730-40a6-ad03-8b36096cb234-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00027-27-802b2129-0d18-418a-bcdd-b1b5332ea2a5-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00028-28-70f51a0c-3d30-4cdb-afcb-e8c1cfa3caa4-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00030-30-e887707d-58db-469c-ab96-b77188f25189-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/data/00031-31-facf6a62-c326-44f7-bdc3-b1471bd594d7-00000.parquet',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/metadata/9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21-m0.avro',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/metadata/snap-93996984692289973-1-9b8c72ab-43b9-42fb-a5e9-1dcfa1801a21.avro',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/metadata/v1.metadata.json',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/metadata/v2.metadata.json',regex:.*,''
+'$NAMENODE/test-warehouse/iceberg_test/iceberg_non_partitioned/metadata/version-hint.text',regex:.*,''
+---- TYPES
+STRING,STRING,STRING
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/show-create-table.test b/testdata/workloads/functional-query/queries/QueryTest/show-create-table.test
index daaac65..5ad9bd2 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/show-create-table.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/show-create-table.test
@@ -561,17 +561,19 @@ CREATE TABLE iceberg_test1 (
level STRING
)
STORED AS ICEBERG
+TBLPROPERTIES('iceberg_file_format'='parquet')
---- RESULTS-HIVE
CREATE TABLE show_create_table_test_db.iceberg_test1 (
level STRING
)
STORED AS ICEBERG
LOCATION '$$location_uri$$'
+TBLPROPERTIES('iceberg_file_format'='parquet')
---- RESULTS-HIVE-3
CREATE EXTERNAL TABLE show_create_table_test_db.iceberg_test1 (
level STRING
)
STORED AS ICEBERG
LOCATION '$$location_uri$$'
-TBLPROPERTIES ('external.table.purge'='TRUE')
+TBLPROPERTIES ('external.table.purge'='TRUE', 'iceberg_file_format'='parquet')
====
diff --git a/tests/custom_cluster/test_iceberg.py b/tests/query_test/test_iceberg.py
similarity index 71%
rename from tests/custom_cluster/test_iceberg.py
rename to tests/query_test/test_iceberg.py
index be2857f..dd31ea6 100644
--- a/tests/custom_cluster/test_iceberg.py
+++ b/tests/query_test/test_iceberg.py
@@ -15,18 +15,21 @@
# specific language governing permissions and limitations
# under the License.
-import pytest
+from tests.common.impala_test_suite import ImpalaTestSuite
-from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
-
-class TestCreatingIcebergTable(CustomClusterTestSuite):
+class TestCreatingIcebergTable(ImpalaTestSuite):
"""Test creating iceberg managed and external table"""
@classmethod
def get_workload(cls):
return 'functional-query'
- @pytest.mark.execute_serially
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestCreatingIcebergTable, cls).add_test_dimensions()
+ cls.ImpalaTestMatrix.add_constraint(
+ lambda v: v.get_value('table_format').file_format == 'parquet')
+
def test_create_iceberg_tables(self, vector, unique_database):
- self.run_test_case('QueryTest/iceberg_create', vector, use_db=unique_database)
+ self.run_test_case('QueryTest/iceberg-create', vector, use_db=unique_database)
diff --git a/tests/query_test/test_scanners.py b/tests/query_test/test_scanners.py
index 1ff5e6d..9ce9f5d 100644
--- a/tests/query_test/test_scanners.py
+++ b/tests/query_test/test_scanners.py
@@ -340,6 +340,29 @@ class TestHudiParquet(ImpalaTestSuite):
self.run_test_case('QueryTest/hudi-parquet', vector)
+class TestIceberg(ImpalaTestSuite):
+ @classmethod
+ def get_workload(cls):
+ return 'functional-query'
+
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestIceberg, cls).add_test_dimensions()
+ cls.ImpalaTestMatrix.add_dimension(
+ create_exec_option_dimension(debug_action_options=DEBUG_ACTION_DIMS))
+ cls.ImpalaTestMatrix.add_constraint(
+ lambda v: v.get_value('table_format').file_format == 'parquet')
+
+ def test_iceberg_query(self, vector):
+ self.run_test_case('QueryTest/iceberg-query', vector)
+
+ def test_iceberg_negative(self, vector, unique_database):
+ self.run_test_case('QueryTest/iceberg-negative', vector, use_db=unique_database)
+
+ def test_iceberg_profile(self, vector, unique_database):
+ self.run_test_case('QueryTest/iceberg-profile', vector, use_db=unique_database)
+
+
class TestParquet(ImpalaTestSuite):
@classmethod
def get_workload(cls):