You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2021/02/20 03:03:51 UTC
[impala] 07/08: IMPALA-10485: part(1): make ORC column reader
creation independent of schema resolution
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 7c8e3eb4400f5924c2ed12be92899d254d5b412b
Author: Zoltan Borok-Nagy <bo...@cloudera.com>
AuthorDate: Mon Feb 8 12:03:47 2021 +0100
IMPALA-10485: part(1): make ORC column reader creation independent of schema resolution
Currently ORC schema resolution is position-based only. Then the
positional information is reused during ORC column reader creation.
This prevents adding other column resolution strategies, e.g.
column resolution by name or Iceberg field id. It also prevents
schema evolution, as table metadata and file metadata should be
in sync.
This patch makes column reader creation independent of schema
resolution. It does this by creating a mapping between slot/tuple
descriptors and ORC type ids during schema resolution and use
this mapping during column reader creation.
Now further patches just need to add support for other column
resolution strategies, the column readers will be created
accordingly.
Testing:
* no additional tests as it is neither a bug fix nor a new feature
Change-Id: I0f7d521f9397c5188fadc7996cee0bd1650d363e
Reviewed-on: http://gerrit.cloudera.org:8080/17039
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/hdfs-orc-scanner.cc | 8 +--
be/src/exec/hdfs-orc-scanner.h | 8 +--
be/src/exec/orc-column-readers.cc | 109 ++++++++++++++++++++------------------
be/src/exec/orc-column-readers.h | 23 ++++++--
be/src/exec/orc-metadata-utils.cc | 62 ----------------------
be/src/exec/orc-metadata-utils.h | 10 ----
tests/query_test/test_scanners.py | 2 +-
7 files changed, 82 insertions(+), 140 deletions(-)
diff --git a/be/src/exec/hdfs-orc-scanner.cc b/be/src/exec/hdfs-orc-scanner.cc
index 0a706a4..d3dbe2f 100644
--- a/be/src/exec/hdfs-orc-scanner.cc
+++ b/be/src/exec/hdfs-orc-scanner.cc
@@ -240,12 +240,6 @@ Status HdfsOrcScanner::Open(ScannerContext* context) {
// blob more efficiently.
row_reader_options_.setEnableLazyDecoding(true);
- // Build 'col_id_path_map_' that maps from ORC column ids to their corresponding
- // SchemaPath in the table. The map is used in the constructors of OrcColumnReaders
- // where we resolve SchemaPaths of the descriptors.
- RETURN_IF_ERROR(schema_resolver_->BuildSchemaPaths(scan_node_->num_partition_keys(),
- &col_id_path_map_));
-
// To create OrcColumnReaders, we need the selected orc schema. It's a subset of the
// file schema: a tree of selected orc types and can only be got from an orc::RowReader
// (by orc::RowReader::getSelectedType).
@@ -382,6 +376,7 @@ Status HdfsOrcScanner::ResolveColumns(const TupleDescriptor& tuple_desc,
return Status(Substitute("Could not find nested column '$0' in file '$1'.",
PrintPath(*scan_node_->hdfs_table(), tuple_desc.tuple_path()), filename()));
}
+ tuple_to_col_id_.insert({&tuple_desc, node->getColumnId()});
if (tuple_desc.byte_size() == 0) {
// Don't need to materialize any slots but just generate an empty tuple for each
// (collection) row. (E.g. count(*) or 'exists' on results of subquery).
@@ -435,6 +430,7 @@ Status HdfsOrcScanner::ResolveColumns(const TupleDescriptor& tuple_desc,
missing_field_slots_.insert(slot_desc);
continue;
}
+ slot_to_col_id_.insert({slot_desc, node->getColumnId()});
if (pos_field) {
DCHECK(pos_slot_desc == nullptr)
<< "There should only be one position slot per tuple";
diff --git a/be/src/exec/hdfs-orc-scanner.h b/be/src/exec/hdfs-orc-scanner.h
index a2c88ff..3af37de 100644
--- a/be/src/exec/hdfs-orc-scanner.h
+++ b/be/src/exec/hdfs-orc-scanner.h
@@ -209,10 +209,10 @@ class HdfsOrcScanner : public HdfsColumnarScanner {
/// orc::RowReaderOptions.includeTypes() expects a list
std::list<uint64_t> selected_type_ids_;
- /// Map from orc::Type id to SchemaPath. See more in descriptors.h for the definition
- /// of SchemaPath. The map is used in the constructors of OrcColumnReaders where we
- /// resolve SchemaPaths of the descriptors into ORC columns. Built in 'Open'.
- std::vector<SchemaPath> col_id_path_map_;
+ /// Mappings from slot/tuple descriptors to ORC type id. We save this information during
+ /// column resolution and use it during column reader creation.
+ std::unordered_map<const SlotDescriptor*, uint64_t> slot_to_col_id_;
+ std::unordered_map<const TupleDescriptor*, uint64_t> tuple_to_col_id_;
/// Scan range for the metadata (file tail).
const io::ScanRange* metadata_range_ = nullptr;
diff --git a/be/src/exec/orc-column-readers.cc b/be/src/exec/orc-column-readers.cc
index 138fa5a..0b22f71 100644
--- a/be/src/exec/orc-column-readers.cc
+++ b/be/src/exec/orc-column-readers.cc
@@ -272,8 +272,9 @@ Status OrcDecimal16ColumnReader::ReadValue(int row_idx, Tuple* tuple, MemPool* p
OrcComplexColumnReader::OrcComplexColumnReader(const orc::Type* node,
const TupleDescriptor* table_tuple_desc, HdfsOrcScanner* scanner)
: OrcBatchedReader(node, nullptr, scanner) {
- SchemaPath& path = scanner->col_id_path_map_[node->getColumnId()];
- if (path == table_tuple_desc->tuple_path()) tuple_desc_ = table_tuple_desc;
+ uint64_t node_col_id = node->getColumnId();
+ uint64_t tuple_desc_col_id = GetColId(table_tuple_desc);
+ if (node_col_id == tuple_desc_col_id) tuple_desc_ = table_tuple_desc;
materialize_tuple_ = (tuple_desc_ != nullptr);
VLOG(3) << "Created top level ComplexColumnReader for " << PrintNode(node)
<< ": tuple_desc_=" << (tuple_desc_ ? tuple_desc_->DebugString() : "null");
@@ -287,23 +288,37 @@ bool OrcStructReader::EndOfBatch() {
return vbatch_ == nullptr || row_idx_ == NumElements();
}
-inline bool PathContains(const SchemaPath& path, const SchemaPath& sub_path) {
- return path.size() >= sub_path.size() &&
- std::equal(sub_path.begin(), sub_path.end(), path.begin());
-}
-
-inline const SchemaPath& GetTargetColPath(const SlotDescriptor* slot_desc) {
+inline uint64_t OrcComplexColumnReader::GetTargetColId(
+ const SlotDescriptor* slot_desc) const {
return slot_desc->type().IsCollectionType() ?
- slot_desc->collection_item_descriptor()->tuple_path(): slot_desc->col_path();
-}
-
-bool OrcStructReader::FindChild(const orc::Type& parent, const SchemaPath& child_path,
+ GetColId(slot_desc->collection_item_descriptor()):
+ GetColId(slot_desc);
+}
+
+/**
+ * Returns true if 'candidate_col_id' is under 'node' in the type hierarchy.
+ */
+inline bool IsDescendant(const orc::Type& node, uint64_t candidate_col_id) {
+ uint64_t node_col_id = node.getColumnId();
+ uint64_t node_max_col_id = node.getMaximumColumnId();
+ return node_col_id <= candidate_col_id && candidate_col_id <= node_max_col_id;
+}
+
+/**
+ * 'parent' is a complex type that might have multiple children, i.e. STRUCT or MAP.
+ * 'descendant_col_id' refers to an ORC type somewhere under 'parent' in the type
+ * hieararchy. This method selects the direct child of 'parent' that is also the ancestor
+ * of 'descendant_col_id'. Return the result in '*child' and its index in '*field'.
+ * Returns false for not found.
+ */
+bool FindChild(const orc::Type& parent, uint64_t descendant_col_id,
const orc::Type** child, int* field) {
+ DCHECK(parent.getKind() == orc::TypeKind::STRUCT ||
+ parent.getKind() == orc::TypeKind::MAP);
int size = parent.getSubtypeCount();
for (int c = 0; c < size; ++c) {
const orc::Type* node = parent.getSubtype(c);
- const SchemaPath& node_path = scanner_->col_id_path_map_[node->getColumnId()];
- if (PathContains(child_path, node_path)) {
+ if (node && IsDescendant(*node, descendant_col_id)) {
*child = node;
*field = c;
return true;
@@ -319,7 +334,7 @@ void OrcStructReader::CreateChildForSlot(const orc::Type* curr_node,
// Create a child reader and pass down 'slot_desc'.
const orc::Type* child_node;
int field;
- if (!FindChild(*curr_node, GetTargetColPath(slot_desc), &child_node, &field)) {
+ if (!FindChild(*curr_node, GetTargetColId(slot_desc), &child_node, &field)) {
DCHECK(false) << PrintNode(curr_node) << " has no children selected for "
<< slot_desc->DebugString();
}
@@ -495,11 +510,10 @@ Status OrcStructReader::UpdateInputBatch(orc::ColumnVectorBatch* orc_batch) {
OrcCollectionReader::OrcCollectionReader(const orc::Type* node,
const SlotDescriptor* slot_desc, HdfsOrcScanner* scanner)
: OrcComplexColumnReader(node, slot_desc, scanner) {
- const SchemaPath& path = scanner->col_id_path_map_[node->getColumnId()];
if (slot_desc->type().IsCollectionType() &&
- slot_desc->collection_item_descriptor()->tuple_path() == path) {
- // This is a collection SlotDescriptor whose item TupleDescriptor matches our
- // SchemaPath. We should materialize the slot (creating a CollectionValue) and its
+ node->getColumnId() == GetTargetColId(slot_desc)) {
+ // This is a collection SlotDescriptor whose item TupleDescriptor matches
+ // 'node'. We should materialize the slot (creating a CollectionValue) and its
// collection tuples (see more in HdfsOrcScanner::AssembleCollection).
tuple_desc_ = slot_desc->collection_item_descriptor();
materialize_tuple_ = true;
@@ -586,15 +600,16 @@ Status OrcListReader::SetPositionSlot(int row_idx, Tuple* tuple) {
void OrcListReader::CreateChildForSlot(const orc::Type* node,
const SlotDescriptor* slot_desc) {
- int depth = scanner_->col_id_path_map_[node->getColumnId()].size();
- const SchemaPath& target_path = GetTargetColPath(slot_desc);
- DCHECK_GT(target_path.size(), depth);
- int field = target_path[depth];
- if (field == SchemaPathConstants::ARRAY_POS) {
+ uint64_t slot_col_id = GetTargetColId(slot_desc);
+ DCHECK(IsDescendant(*node, slot_col_id));
+ // We have a position slot descriptor if it refers to this LIST ORC type, but it isn't
+ // a collection slot.
+ bool is_pos_slot = slot_col_id == node->getColumnId() &&
+ slot_desc->collection_item_descriptor() == nullptr;
+ if (is_pos_slot) {
DCHECK(pos_slot_desc_ == nullptr) << "Should have unique pos slot";
pos_slot_desc_ = slot_desc;
} else {
- DCHECK_EQ(field, SchemaPathConstants::ARRAY_ITEM);
OrcColumnReader* child = OrcColumnReader::Create(node->getSubtype(0), slot_desc,
scanner_);
children_.push_back(child);
@@ -672,26 +687,17 @@ Status OrcListReader::ReadChildrenValue(int row_idx, int tuple_idx, Tuple* tuple
void OrcMapReader::CreateChildForSlot(const orc::Type* node,
const SlotDescriptor* slot_desc) {
- const SchemaPath& path = scanner_->col_id_path_map_[node->getColumnId()];
- const SchemaPath& target_path = GetTargetColPath(slot_desc);
- int depth = path.size();
- // The target of 'slot_desc' matches a descendant so its SchemaPath should be deeper
- DCHECK_GT(target_path.size(), depth);
- int field = target_path[depth];
const orc::Type* child_type;
- if (field == SchemaPathConstants::MAP_KEY) {
- child_type = node->getSubtype(0);
- } else {
- DCHECK_EQ(field, SchemaPathConstants::MAP_VALUE);
- child_type = node->getSubtype(1);
+ int field;
+ if (!FindChild(*node, GetTargetColId(slot_desc), &child_type, &field)) {
+ DCHECK(false) << PrintNode(node) << " has no children selected for "
+ << slot_desc->DebugString();
}
- DCHECK(child_type != nullptr) << Substitute(
- "$0 matches an empty child of $1: path=$2, target_path=$3",
- slot_desc->DebugString(), PrintNode(node), PrintNumericPath(path),
- PrintNumericPath(target_path));
+ // Map type only has two children.
+ DCHECK_LE(field, 1);
OrcColumnReader* child = OrcColumnReader::Create(child_type, slot_desc, scanner_);
children_.push_back(child);
- if (field == SchemaPathConstants::MAP_KEY) {
+ if (field == 0) {
key_readers_.push_back(child);
} else {
value_readers_.push_back(child);
@@ -707,20 +713,19 @@ OrcMapReader::OrcMapReader(const orc::Type* node,
CreateChildForSlot(node, child_slot);
}
} else {
- // 'table_tuple_desc' should match to a descendant of 'node'
- int depth = scanner->col_id_path_map_[node->getColumnId()].size();
- DCHECK_GT(table_tuple_desc->tuple_path().size(), depth);
- // Create a child corresponding to the subtype in the path to the descendant.
- int field = table_tuple_desc->tuple_path()[depth];
- DCHECK(field == SchemaPathConstants::MAP_KEY ||
- field == SchemaPathConstants::MAP_VALUE);
- bool key_selected = (field == SchemaPathConstants::MAP_KEY);
- const orc::Type* child_type =
- key_selected ? node->getSubtype(0) : node->getSubtype(1);
+ const orc::Type* child_type;
+ int field;
+ if (!FindChild(*node, GetColId(table_tuple_desc),
+ &child_type, &field)) {
+ DCHECK(false) << PrintNode(node) << " has no children selected for "
+ << table_tuple_desc->DebugString();
+ }
+ // Map type only has two children.
+ DCHECK_LE(field, 1);
OrcComplexColumnReader* child = OrcComplexColumnReader::CreateTopLevelReader(
child_type, table_tuple_desc, scanner);
children_.push_back(child);
- if (key_selected) {
+ if (field == 0) {
key_readers_.push_back(child);
} else {
value_readers_.push_back(child);
diff --git a/be/src/exec/orc-column-readers.h b/be/src/exec/orc-column-readers.h
index e50c216..8b8ce73 100644
--- a/be/src/exec/orc-column-readers.h
+++ b/be/src/exec/orc-column-readers.h
@@ -134,6 +134,20 @@ class OrcColumnReader {
friend class OrcStructReader;
friend class OrcCollectionReader;
+ /// Returns the ORC type column id corresponding to the given slot descriptor.
+ uint64_t GetColId(const SlotDescriptor* desc) const {
+ auto it = scanner_->slot_to_col_id_.find(desc);
+ DCHECK(it != scanner_->slot_to_col_id_.end());
+ return it->second;
+ }
+
+ /// Returns the ORC type column id corresponding to the given tuple descriptor.
+ uint64_t GetColId(const TupleDescriptor* desc) const {
+ auto it = scanner_->tuple_to_col_id_.find(desc);
+ DCHECK(it != scanner_->tuple_to_col_id_.end());
+ return it->second;
+ }
+
/// Convenient field for debug. We can't keep the pointer of orc::Type since they'll be
/// destroyed after orc::RowReader was released. Only keep the id orc::Type here.
uint64_t orc_column_id_;
@@ -513,6 +527,10 @@ class OrcComplexColumnReader : public OrcBatchedReader<OrcComplexColumnReader> {
/// delegates its job to its children. It might still write a position slot though.
bool DirectReader() const { return slot_desc_ != nullptr; }
protected:
+ /// Return target column id. For collection slots it returns the column id corresponding
+ /// to the collection tuple descriptor
+ uint64_t GetTargetColId(const SlotDescriptor* slot_desc) const;
+
vector<OrcColumnReader*> children_;
/// Holds the TupleDescriptor if we should materialize its tuples
@@ -604,11 +622,6 @@ class OrcStructReader : public OrcComplexColumnReader {
void CreateChildForSlot(const orc::Type* curr_node, const SlotDescriptor* slot_desc);
- /// Find which children of 'curr_node' matches the 'child_path'. Return the result in
- /// '*child' and its index inside the children. Returns false for not found.
- inline bool FindChild(const orc::Type& curr_node, const SchemaPath& child_path,
- const orc::Type** child, int* field);
-
Status ReadAndValidateRows(ScratchTupleBatch* scratch_batch, MemPool* pool);
};
diff --git a/be/src/exec/orc-metadata-utils.cc b/be/src/exec/orc-metadata-utils.cc
index 400bba0..b198bae 100644
--- a/be/src/exec/orc-metadata-utils.cc
+++ b/be/src/exec/orc-metadata-utils.cc
@@ -26,68 +26,6 @@ using boost::algorithm::iequals;
namespace impala {
-Status OrcSchemaResolver::BuildSchemaPaths(int num_partition_keys,
- vector<SchemaPath>* col_id_path_map) const {
- if (root_->getKind() != orc::TypeKind::STRUCT) {
- return Status(TErrorCode::ORC_TYPE_NOT_ROOT_AT_STRUCT, "file", root_->toString(),
- filename_);
- }
- bool synthetic_acid_schema = is_table_full_acid_ && !is_file_full_acid_;
- SchemaPath path;
- col_id_path_map->push_back(path);
- int num_columns = root_->getSubtypeCount();
- // Original files don't have "row" field, let's add it manually.
- if (synthetic_acid_schema) path.push_back(ACID_FIELD_ROW + num_partition_keys);
- for (int i = 0; i < num_columns; ++i) {
- // For synthetic ACID schema these columns are not top-level, so don't need to
- // adjust them by 'num_part_keys'.
- int field_offset = synthetic_acid_schema ? 0 : num_partition_keys;
- path.push_back(i + field_offset);
- BuildSchemaPathHelper(*root_->getSubtype(i), &path, col_id_path_map);
- path.pop_back();
- }
- return Status::OK();
-}
-
-void OrcSchemaResolver::BuildSchemaPathHelper(const orc::Type& node, SchemaPath* path,
- vector<SchemaPath>* col_id_path_map) {
- DCHECK_EQ(col_id_path_map->size(), node.getColumnId()) << Substitute(
- "Failed building map from ORC type ids to SchemaPaths. $0 are built but current "
- "ORC type id is $1, missing ORC type with id=$0.",
- col_id_path_map->size(), node.getColumnId());
- // Map ORC type with id=size() to 'path'.
- col_id_path_map->push_back(*path);
- // Deal with subtypes recursively.
- if (node.getKind() == orc::TypeKind::STRUCT
- || node.getKind() == orc::TypeKind::UNION) {
- // It's possible the file schema contains more columns than the table schema.
- // We should deal with UNION type here in case it's not in the table schema and the
- // query is not reading them. Otherwise the map will miss some ORC types.
- int size = node.getSubtypeCount();
- for (int i = 0; i < size; ++i) {
- path->push_back(i);
- const orc::Type* child = node.getSubtype(i);
- BuildSchemaPathHelper(*child, path, col_id_path_map);
- path->pop_back();
- }
- } else if (node.getKind() == orc::TypeKind::LIST) {
- DCHECK_EQ(node.getSubtypeCount(), 1);
- const orc::Type* child = node.getSubtype(0);
- path->push_back(SchemaPathConstants::ARRAY_ITEM);
- BuildSchemaPathHelper(*child, path, col_id_path_map);
- path->pop_back();
- } else if (node.getKind() == orc::TypeKind::MAP) {
- DCHECK_EQ(node.getSubtypeCount(), 2);
- const orc::Type* key_child = node.getSubtype(0);
- const orc::Type* value_child = node.getSubtype(1);
- path->push_back(SchemaPathConstants::MAP_KEY);
- BuildSchemaPathHelper(*key_child, path, col_id_path_map);
- (*path)[path->size() - 1] = SchemaPathConstants::MAP_VALUE;
- BuildSchemaPathHelper(*value_child, path, col_id_path_map);
- path->pop_back();
- }
-}
-
Status OrcSchemaResolver::ResolveColumn(const SchemaPath& col_path,
const orc::Type** node, bool* pos_field, bool* missing_field) const {
const ColumnType* table_col_type = nullptr;
diff --git a/be/src/exec/orc-metadata-utils.h b/be/src/exec/orc-metadata-utils.h
index 5450617..4c2dfb2 100644
--- a/be/src/exec/orc-metadata-utils.h
+++ b/be/src/exec/orc-metadata-utils.h
@@ -54,11 +54,6 @@ class OrcSchemaResolver {
Status ResolveColumn(const SchemaPath& col_path, const orc::Type** node,
bool* pos_field, bool* missing_field) const;
- /// Build a map from each orc::Type id to a SchemaPath. The map will be used in
- /// creating OrcColumnReaders. It contains all ORC types including unmaterialized ones.
- Status BuildSchemaPaths(int num_partition_keys,
- std::vector<SchemaPath>* col_id_path_map) const;
-
/// Returns true if file schema corresponds to full ACIDv2 format.
bool HasFullAcidV2Schema() const { return is_file_full_acid_; }
@@ -108,10 +103,5 @@ class OrcSchemaResolver {
/// Validate whether the ColumnType is compatible with the orc type
Status ValidateType(const ColumnType& type, const orc::Type& orc_type) const
WARN_UNUSED_RESULT;
-
- /// Helper function to build 'col_id_path_map' start from 'node' whose corresponding
- /// SchemaPath is 'path'.
- static void BuildSchemaPathHelper(const orc::Type& node, SchemaPath* path,
- std::vector<SchemaPath>* col_id_path_map);
};
}
diff --git a/tests/query_test/test_scanners.py b/tests/query_test/test_scanners.py
index e4d9e0b..b38908d 100644
--- a/tests/query_test/test_scanners.py
+++ b/tests/query_test/test_scanners.py
@@ -1582,7 +1582,7 @@ class TestOrc(ImpalaTestSuite):
self._run_invalid_schema_test(unique_database, "corrupt_schema",
"Encountered parse error during schema selection")
self._run_invalid_schema_test(unique_database, "corrupt_root_type",
- "Root of the file type returned by the ORC lib is not STRUCT: boolean.")
+ "Root of the selected type returned by the ORC lib is not STRUCT: boolean.")
def test_date_out_of_range_orc(self, vector, unique_database):