You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jo...@apache.org on 2021/03/16 15:45:50 UTC
[impala] branch master updated (573c60a -> 122bd9e)
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git.
from 573c60a IMPALA-10367: Impala-shell internal error - UnboundLocalError, local variable 'retry_msg' referenced before assign
new 768722b IMPALA-10551: Add result sink support for external frontends
new 122bd9e IMPALA-10553: External Frontend CTAS support
The 2 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/exec/hdfs-table-sink.cc | 72 +++++++++++++++++++---
be/src/exec/hdfs-table-sink.h | 15 ++++-
be/src/runtime/coordinator.cc | 37 +++++++++--
be/src/runtime/coordinator.h | 6 ++
be/src/runtime/query-exec-params.cc | 19 +++++-
be/src/runtime/query-exec-params.h | 7 +++
common/thrift/DataSinks.thrift | 10 +++
.../org/apache/impala/planner/HdfsTableSink.java | 45 +++++++++++++-
.../java/org/apache/impala/planner/TableSink.java | 14 ++++-
.../java/org/apache/impala/service/Frontend.java | 21 +++++--
10 files changed, 218 insertions(+), 28 deletions(-)
[impala] 01/02: IMPALA-10551: Add result sink support for external
frontends
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 768722b8686c15d84af3c0a94427edd70e9efb9f
Author: John Sherman <jf...@cloudera.com>
AuthorDate: Thu Jun 11 22:41:19 2020 +0000
IMPALA-10551: Add result sink support for external frontends
- The intended purpose of these changes is to allow external frontends
to receive query results via files rather than streaming the results
through the thrift interface.
- External frontends are expected to provide an FeFsTable implementation
that describes the desired location to store results.
- External frontends are responsible for managing the files after the
query is completed.
- Testing has been manual and through an implementation of an external
frontend.
Change-Id: I024bf41d77bb81f1ab0debdbd31ec3687c83f072
Reviewed-by: Aman Sinha <am...@cloudera.com>
Reviewed-on: http://gerrit.cloudera.org:8080/17144
Reviewed-by: Joe McDonnell <jo...@cloudera.com>
Tested-by: Joe McDonnell <jo...@cloudera.com>
---
be/src/exec/hdfs-table-sink.cc | 5 ++-
be/src/exec/hdfs-table-sink.h | 2 ++
be/src/runtime/coordinator.cc | 37 +++++++++++++++++++---
be/src/runtime/coordinator.h | 6 ++++
be/src/runtime/query-exec-params.cc | 19 ++++++++++-
be/src/runtime/query-exec-params.h | 7 ++++
common/thrift/DataSinks.thrift | 3 ++
.../org/apache/impala/planner/HdfsTableSink.java | 18 +++++++++--
.../java/org/apache/impala/planner/TableSink.java | 14 ++++++--
9 files changed, 99 insertions(+), 12 deletions(-)
diff --git a/be/src/exec/hdfs-table-sink.cc b/be/src/exec/hdfs-table-sink.cc
index 4611080..4ae6c96 100644
--- a/be/src/exec/hdfs-table-sink.cc
+++ b/be/src/exec/hdfs-table-sink.cc
@@ -88,7 +88,8 @@ HdfsTableSink::HdfsTableSink(TDataSinkId sink_id, const HdfsTableSinkConfig& sin
sort_columns_(hdfs_sink.sort_columns),
sorting_order_((TSortingOrder::type)hdfs_sink.sorting_order),
current_clustered_partition_(nullptr),
- partition_key_exprs_(sink_config.partition_key_exprs_) {
+ partition_key_exprs_(sink_config.partition_key_exprs_),
+ is_result_sink_(hdfs_sink.is_result_sink) {
if (hdfs_sink.__isset.write_id) {
write_id_ = hdfs_sink.write_id;
DCHECK_GT(write_id_, 0);
@@ -739,6 +740,8 @@ void HdfsTableSink::Close(RuntimeState* state) {
bool HdfsTableSink::ShouldSkipStaging(RuntimeState* state, OutputPartition* partition) {
if (IsTransactional()) return true;
+ // We skip staging if we are writing query results
+ if (is_result_sink_) return true;
return (IsS3APath(partition->final_hdfs_file_name_prefix.c_str()) && !overwrite_ &&
state->query_options().s3_skip_insert_staging);
}
diff --git a/be/src/exec/hdfs-table-sink.h b/be/src/exec/hdfs-table-sink.h
index c1a6adc..a66aa7c 100644
--- a/be/src/exec/hdfs-table-sink.h
+++ b/be/src/exec/hdfs-table-sink.h
@@ -301,6 +301,8 @@ class HdfsTableSink : public DataSink {
RuntimeProfile::Counter* hdfs_write_timer_;
/// Time spent compressing data
RuntimeProfile::Counter* compress_timer_;
+ /// Will the output of this sink be used for query results
+ const bool is_result_sink_;
};
}
diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc
index 3e1a00a..3723177 100644
--- a/be/src/runtime/coordinator.cc
+++ b/be/src/runtime/coordinator.cc
@@ -792,6 +792,28 @@ void Coordinator::HandleExecStateTransition(
finalized_.Set(true);
}
+Status Coordinator::FinalizeResultSink() {
+ // All instances must have reported their final statuses before finalization, which is a
+ // post-condition of Wait. Result sink file clean up is the responsibility of the
+ // external frontend implementation.
+ DCHECK(has_called_wait_.Load());
+ VLOG_QUERY << "Finalizing result sink: " << PrintId(query_id());
+ SCOPED_TIMER(finalization_timer_);
+ RETURN_IF_ERROR(UpdateExecState(Status::OK(), nullptr, FLAGS_hostname));
+
+ HdfsTableDescriptor* hdfs_table;
+ DCHECK(query_ctx().__isset.desc_tbl_serialized);
+ // When a query has a result sink, it is explicitly assumed that the table descriptor
+ // for id = 0 is a pseudo-table describing the location in which to write results.
+ const TableId result_sink_table_id = 0;
+ RETURN_IF_ERROR(DescriptorTbl::CreateHdfsTblDescriptor(query_ctx().desc_tbl_serialized,
+ result_sink_table_id, obj_pool(), &hdfs_table));
+ DCHECK(hdfs_table != nullptr)
+ << "Result Sink target table not known in descriptor table";
+ hdfs_table->ReleaseResources();
+ return Status::OK();
+}
+
Status Coordinator::FinalizeHdfsDml() {
// All instances must have reported their final statuses before finalization, which is a
// post-condition of Wait. If the query was not successful, still try to clean up the
@@ -866,7 +888,7 @@ Status Coordinator::Wait() {
if (has_called_wait_.Load()) return Status::OK();
has_called_wait_.Store(true);
- if (stmt_type_ == TStmtType::QUERY) {
+ if (stmt_type_ == TStmtType::QUERY && !exec_params_.HasResultSink()) {
DCHECK(coord_instance_ != nullptr);
RETURN_IF_ERROR(UpdateExecState(coord_instance_->WaitForOpen(),
&coord_instance_->runtime_state()->fragment_instance_id(), FLAGS_hostname));
@@ -883,14 +905,19 @@ Status Coordinator::Wait() {
}
return Status::OK();
}
- DCHECK_EQ(stmt_type_, TStmtType::DML);
+ DCHECK(stmt_type_ == TStmtType::DML ||
+ (stmt_type_ == TStmtType::QUERY && exec_params_.HasResultSink()));
// DML finalization can only happen when all backends have completed all side-effects
- // and reported relevant state.
+ // and reported relevant state. We also wait for backends to complete if we are writing
+ // results to a filesystem.
WaitForBackends();
+
if (finalize_params() != nullptr) {
- RETURN_IF_ERROR(UpdateExecState(
- FinalizeHdfsDml(), nullptr, FLAGS_hostname));
+ RETURN_IF_ERROR(UpdateExecState(FinalizeHdfsDml(), nullptr, FLAGS_hostname));
+ } else if (exec_params_.HasResultSink()) {
+ RETURN_IF_ERROR(UpdateExecState(FinalizeResultSink(), nullptr, FLAGS_hostname));
}
+
// DML requests are finished at this point.
RETURN_IF_ERROR(SetNonErrorTerminalState(ExecState::RETURNED_RESULTS));
query_profile_->AddInfoString(
diff --git a/be/src/runtime/coordinator.h b/be/src/runtime/coordinator.h
index 2ba58eb..c982b46 100644
--- a/be/src/runtime/coordinator.h
+++ b/be/src/runtime/coordinator.h
@@ -544,6 +544,12 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// finishing the INSERT in flight.
Status FinalizeHdfsDml() WARN_UNUSED_RESULT;
+ /// Perform any post-query cleanup required for Result Sink writing. Called by Wait()
+ /// only after all fragment instances have returned. It is the responsibility of an
+ /// external frontend to clean up the Result Sink location on success or error of the
+ /// running query.
+ Status FinalizeResultSink() WARN_UNUSED_RESULT;
+
/// Helper for Exec(). Populates backend_states_, starts query execution at all
/// backends in parallel, and blocks until startup completes.
Status StartBackendExec();
diff --git a/be/src/runtime/query-exec-params.cc b/be/src/runtime/query-exec-params.cc
index d453fcc..4c4d980 100644
--- a/be/src/runtime/query-exec-params.cc
+++ b/be/src/runtime/query-exec-params.cc
@@ -49,7 +49,18 @@ QueryExecParams::QueryExecParams(const UniqueIdPB& query_id,
const TExecRequest& exec_request, const QuerySchedulePB& query_schedule)
: query_id_(query_id), exec_request_(exec_request), query_schedule_(query_schedule) {}
-const TPlanFragment* QueryExecParams::GetCoordFragment() const {
+bool QueryExecParams::HasResultSink(const TPlanFragment* fragment) const {
+ return fragment != nullptr &&
+ fragment->output_sink.type == TDataSinkType::TABLE_SINK &&
+ fragment->output_sink.table_sink.type == TTableSinkType::HDFS &&
+ fragment->output_sink.table_sink.hdfs_table_sink.is_result_sink;
+}
+
+bool QueryExecParams::HasResultSink() const {
+ return HasResultSink(GetCoordFragmentImpl());
+}
+
+const TPlanFragment* QueryExecParams::GetCoordFragmentImpl() const {
// Only have coordinator fragment for statements that return rows.
if (query_exec_request().stmt_type != TStmtType::QUERY) return nullptr;
DCHECK_GE(query_exec_request().plan_exec_info.size(), 1);
@@ -59,6 +70,12 @@ const TPlanFragment* QueryExecParams::GetCoordFragment() const {
return fragment;
}
+const TPlanFragment* QueryExecParams::GetCoordFragment() const {
+ const TPlanFragment* fragment = GetCoordFragmentImpl();
+ if (HasResultSink(fragment)) return nullptr;
+ return fragment;
+}
+
int QueryExecParams::GetNumFragmentInstances() const {
int total = 0;
for (const BackendExecParamsPB& backend_exec_params :
diff --git a/be/src/runtime/query-exec-params.h b/be/src/runtime/query-exec-params.h
index f8bd0b7..0825c52 100644
--- a/be/src/runtime/query-exec-params.h
+++ b/be/src/runtime/query-exec-params.h
@@ -54,6 +54,9 @@ class QueryExecParams {
/// Return the coordinator fragment, or nullptr if there isn't one.
const TPlanFragment* GetCoordFragment() const;
+ /// Returns true if the the query has a result sink
+ bool HasResultSink() const;
+
/// Return the total number of instances across all fragments.
int GetNumFragmentInstances() const;
@@ -71,6 +74,10 @@ class QueryExecParams {
/// Owned by the ClientRwquestState and has query lifetime.
const QuerySchedulePB& query_schedule_;
+ const TPlanFragment* GetCoordFragmentImpl() const;
+
+ bool HasResultSink(const TPlanFragment* fragment) const;
+
/// Constructor used for testing only. Does not initialize 'fragments_'.
QueryExecParams(const UniqueIdPB& query_id, const TExecRequest& exec_request,
const QuerySchedulePB& query_schedule);
diff --git a/common/thrift/DataSinks.thrift b/common/thrift/DataSinks.thrift
index 858921e..af887de 100644
--- a/common/thrift/DataSinks.thrift
+++ b/common/thrift/DataSinks.thrift
@@ -87,6 +87,9 @@ struct THdfsTableSink {
// Sorting order. If not lexical, the backend should not populate the
// RowGroup::sorting_columns list in parquet files.
7: required Types.TSortingOrder sorting_order
+
+ // Indicates that this HdfsTableSink is writing query results
+ 8: optional bool is_result_sink = false;
}
// Structure to encapsulate specific options that are passed down to the KuduTableSink
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java b/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java
index 48a50f9..1568a76 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java
@@ -62,6 +62,9 @@ public class HdfsTableSink extends TableSink {
// be opened, written, and closed one by one.
protected final boolean inputIsClustered_;
+ // Indicates that this sink is being used to write query results
+ protected final boolean isResultSink_;
+
private static final Set<HdfsFileFormat> SUPPORTED_FILE_FORMATS = ImmutableSet.of(
HdfsFileFormat.PARQUET, HdfsFileFormat.TEXT, HdfsFileFormat.RC_FILE,
HdfsFileFormat.SEQUENCE_FILE, HdfsFileFormat.AVRO, HdfsFileFormat.ICEBERG);
@@ -83,7 +86,7 @@ public class HdfsTableSink extends TableSink {
public HdfsTableSink(FeTable targetTable, List<Expr> partitionKeyExprs,
List<Expr> outputExprs, boolean overwrite, boolean inputIsClustered,
Pair<List<Integer>, TSortingOrder> sortProperties, long writeId,
- int maxTableSinks) {
+ int maxTableSinks, boolean isResultSink) {
super(targetTable, Op.INSERT, outputExprs);
Preconditions.checkState(targetTable instanceof FeFsTable);
partitionKeyExprs_ = partitionKeyExprs;
@@ -93,6 +96,7 @@ public class HdfsTableSink extends TableSink {
sortingOrder_ = sortProperties.second;
writeId_ = writeId;
maxHdfsSinks_ = maxTableSinks;
+ isResultSink_ = isResultSink;
}
@Override
@@ -215,6 +219,7 @@ public class HdfsTableSink extends TableSink {
}
hdfsTableSink.setSort_columns(sortColumns_);
hdfsTableSink.setSorting_order(sortingOrder_);
+ hdfsTableSink.setIs_result_sink(isResultSink_);
if (writeId_ != -1) hdfsTableSink.setWrite_id(writeId_);
TTableSink tTableSink = new TTableSink(DescriptorTable.TABLE_SINK_ID,
TTableSinkType.HDFS, sinkOp_.toThrift());
@@ -230,9 +235,16 @@ public class HdfsTableSink extends TableSink {
@Override
public void collectExprs(List<Expr> exprs) {
- // Avoid adding any partition exprs redundantly.
if (!(targetTable_ instanceof FeIcebergTable)) exprs.addAll(partitionKeyExprs_);
- exprs.addAll(outputExprs_.subList(0, targetTable_.getNonClusteringColumns().size()));
+ if (isResultSink_) {
+ // It is expected outputExprs_ is fully populated with the expected result schema
+ // when writing query results
+ exprs.addAll(outputExprs_);
+ } else {
+ // Avoid adding any partition exprs redundantly.
+ exprs.addAll(outputExprs_.subList(0,
+ targetTable_.getNonClusteringColumns().size()));
+ }
}
/**
diff --git a/fe/src/main/java/org/apache/impala/planner/TableSink.java b/fe/src/main/java/org/apache/impala/planner/TableSink.java
index f7038fb..3f5d93b 100644
--- a/fe/src/main/java/org/apache/impala/planner/TableSink.java
+++ b/fe/src/main/java/org/apache/impala/planner/TableSink.java
@@ -102,7 +102,7 @@ public abstract class TableSink extends DataSink {
List<Integer> referencedColumns, boolean overwrite,
boolean inputIsClustered, Pair<List<Integer>, TSortingOrder> sortProperties) {
return create(table, sinkAction, partitionKeyExprs, outputExprs, referencedColumns,
- overwrite, inputIsClustered, sortProperties, -1, 0);
+ overwrite, inputIsClustered, sortProperties, -1, 0, false);
}
/**
@@ -113,6 +113,16 @@ public abstract class TableSink extends DataSink {
List<Integer> referencedColumns, boolean overwrite, boolean inputIsClustered,
Pair<List<Integer>, TSortingOrder> sortProperties, long writeId,
int maxTableSinks) {
+ return create(table, sinkAction, partitionKeyExprs, outputExprs, referencedColumns,
+ overwrite, inputIsClustered, sortProperties, writeId, maxTableSinks, false);
+ }
+
+ public static TableSink create(FeTable table, Op sinkAction,
+ List<Expr> partitionKeyExprs, List<Expr> outputExprs,
+ List<Integer> referencedColumns,
+ boolean overwrite, boolean inputIsClustered,
+ Pair<List<Integer>, TSortingOrder> sortProperties, long writeId,
+ int maxTableSinks, boolean isResultSink) {
Preconditions.checkNotNull(partitionKeyExprs);
Preconditions.checkNotNull(referencedColumns);
Preconditions.checkNotNull(sortProperties.first);
@@ -122,7 +132,7 @@ public abstract class TableSink extends DataSink {
// Referenced columns don't make sense for an Hdfs table.
Preconditions.checkState(referencedColumns.isEmpty());
return new HdfsTableSink(table, partitionKeyExprs,outputExprs, overwrite,
- inputIsClustered, sortProperties, writeId, maxTableSinks);
+ inputIsClustered, sortProperties, writeId, maxTableSinks, isResultSink);
} else if (table instanceof FeHBaseTable) {
// HBase only supports inserts.
Preconditions.checkState(sinkAction == Op.INSERT);
[impala] 02/02: IMPALA-10553: External Frontend CTAS support
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 122bd9ea06dc9ebd126fe4f99d62651957016176
Author: John Sherman <jf...@cloudera.com>
AuthorDate: Fri Aug 28 21:16:07 2020 +0000
IMPALA-10553: External Frontend CTAS support
- Adds the concept of an external staging dir to HdfsTableSink
- This allows an external to specify the destination of the
sink
- When this is set, the external frontend is responsible for
for moving and managing the results
- Any DDL related operations are assumed to be handled by
the external frontend
- External frontends may optionally supply a partition
depth which acts as a hint to skip a certain number of
partitions while creating directories for partitions. This
is for when the external frontend has pre-created a
certain number of the directories in staging (usually the
static portion of a partition specification)/
- Modifies delta/base naming to include 0 prefix padding to
match Hive for dynamic partitioning detection
- External frontends are responsible for doing authorization
checks against the staging directory and it is assumed the
external frontend service is not exposed directly to users.
Co-authored-by: Kurt Deschler <kd...@cloudera.com>
Change-Id: Iae0ea4a832d8281c563427d0d7da1623bfce437b
Reviewed-by: Kurt Deschler <kd...@cloudera.com>
Reviewed-on: http://gerrit.cloudera.org:8080/17145
Reviewed-by: Joe McDonnell <jo...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/hdfs-table-sink.cc | 69 ++++++++++++++++++----
be/src/exec/hdfs-table-sink.h | 13 +++-
common/thrift/DataSinks.thrift | 7 +++
.../org/apache/impala/planner/HdfsTableSink.java | 27 +++++++++
.../java/org/apache/impala/service/Frontend.java | 21 +++++--
5 files changed, 120 insertions(+), 17 deletions(-)
diff --git a/be/src/exec/hdfs-table-sink.cc b/be/src/exec/hdfs-table-sink.cc
index 4ae6c96..7ff2419 100644
--- a/be/src/exec/hdfs-table-sink.cc
+++ b/be/src/exec/hdfs-table-sink.cc
@@ -23,6 +23,7 @@
#include "exprs/scalar-expr-evaluator.h"
#include "exprs/scalar-expr.h"
#include "gen-cpp/ImpalaInternalService_constants.h"
+#include "gutil/stringprintf.h"
#include "runtime/hdfs-fs-cache.h"
#include "runtime/mem-tracker.h"
#include "runtime/raw-value.inline.h"
@@ -94,6 +95,15 @@ HdfsTableSink::HdfsTableSink(TDataSinkId sink_id, const HdfsTableSinkConfig& sin
write_id_ = hdfs_sink.write_id;
DCHECK_GT(write_id_, 0);
}
+
+ // Optional output path provided by an external FE
+ if (hdfs_sink.__isset.external_output_dir) {
+ external_output_dir_ = hdfs_sink.external_output_dir;
+ }
+
+ if (hdfs_sink.__isset.external_output_partition_depth) {
+ external_output_partition_depth_ = hdfs_sink.external_output_partition_depth;
+ }
}
Status HdfsTableSink::Prepare(RuntimeState* state, MemTracker* parent_mem_tracker) {
@@ -216,7 +226,7 @@ Status HdfsTableSink::Open(RuntimeState* state) {
void HdfsTableSink::BuildHdfsFileNames(
const HdfsPartitionDescriptor& partition_descriptor,
- OutputPartition* output_partition) {
+ OutputPartition* output_partition, const string &external_partition_name) {
// Create final_hdfs_file_name_prefix and tmp_hdfs_file_name_prefix.
// Path: <hdfs_base_dir>/<partition_values>/<unique_id_str>
@@ -243,7 +253,13 @@ void HdfsTableSink::BuildHdfsFileNames(
output_partition->tmp_hdfs_dir_name, output_partition->partition_name,
query_suffix);
- if (partition_descriptor.location().empty()) {
+ if (HasExternalOutputDir()) {
+ // When an external FE has provided a staging directory we use that directly.
+ // We are trusting that the external frontend implementation has done appropriate
+ // authorization checks on the external output directory.
+ output_partition->final_hdfs_file_name_prefix = Substitute("$0/$1/",
+ external_output_dir_, external_partition_name);
+ } else if (partition_descriptor.location().empty()) {
output_partition->final_hdfs_file_name_prefix = Substitute("$0/$1/",
table_desc_->hdfs_base_dir(), output_partition->partition_name);
} else {
@@ -254,8 +270,23 @@ void HdfsTableSink::BuildHdfsFileNames(
Substitute("$0/", partition_descriptor.location());
}
if (IsHiveAcid()) {
- string acid_dir = Substitute(overwrite_ ? "/base_$0/" : "/delta_$0_$0/", write_id_);
- output_partition->final_hdfs_file_name_prefix += acid_dir;
+ if (HasExternalOutputDir()) {
+ // The 0 padding on base and delta is to match the behavior of Hive since various
+ // systems will expect a certain format for dynamic partition creation. Additionally
+ // include an 0 statement id for delta directory so various Hive AcidUtils detect
+ // the directory (such as AcidUtils.baseOrDeltaSubdir()). Multiple statements in a
+ // single transaction is not supported.
+ if (overwrite_) {
+ output_partition->final_hdfs_file_name_prefix += StringPrintf("/base_%07ld/",
+ write_id_);
+ } else {
+ output_partition->final_hdfs_file_name_prefix += StringPrintf(
+ "/delta_%07ld_%07ld_0000/", write_id_, write_id_);
+ }
+ } else {
+ string acid_dir = Substitute(overwrite_ ? "/base_$0/" : "/delta_$0_$0/", write_id_);
+ output_partition->final_hdfs_file_name_prefix += acid_dir;
+ }
}
if (IsIceberg()) {
//TODO: implement LocationProviders.
@@ -469,12 +500,21 @@ Status HdfsTableSink::InitOutputPartition(RuntimeState* state,
// Build the unique name for this partition from the partition keys, e.g. "j=1/f=foo/"
// etc.
stringstream partition_name_ss;
+ stringstream external_partition_name_ss;
for (int j = 0; j < partition_key_expr_evals_.size(); ++j) {
+ bool is_external_part = HasExternalOutputDir() &&
+ j >= external_output_partition_depth_;
+ if (is_external_part) {
+ external_partition_name_ss << GetPartitionName(j) << "=";
+ }
partition_name_ss << GetPartitionName(j) << "=";
void* value = partition_key_expr_evals_[j]->GetValue(row);
// nullptr partition keys get a special value to be compatible with Hive.
if (value == nullptr) {
partition_name_ss << table_desc_->null_partition_key_value();
+ if (is_external_part) {
+ external_partition_name_ss << table_desc_->null_partition_key_value();
+ }
} else {
string value_str;
partition_key_expr_evals_[j]->PrintValue(value, &value_str);
@@ -488,16 +528,26 @@ Status HdfsTableSink::InitOutputPartition(RuntimeState* state,
// decoded partition key value.
string encoded_str;
UrlEncode(value_str, &encoded_str, true);
+ string part_key_value = (encoded_str.empty() ?
+ table_desc_->null_partition_key_value() : encoded_str);
// If the string is empty, map it to nullptr (mimicking Hive's behaviour)
- partition_name_ss << (encoded_str.empty() ?
- table_desc_->null_partition_key_value() : encoded_str);
+ partition_name_ss << part_key_value;
+ if (is_external_part) {
+ external_partition_name_ss << part_key_value;
+ }
+ }
+ if (j < partition_key_expr_evals_.size() - 1) {
+ partition_name_ss << "/";
+ if (is_external_part) {
+ external_partition_name_ss << "/";
+ }
}
- if (j < partition_key_expr_evals_.size() - 1) partition_name_ss << "/";
}
// partition_name_ss now holds the unique descriptor for this partition,
output_partition->partition_name = partition_name_ss.str();
- BuildHdfsFileNames(partition_descriptor, output_partition);
+ BuildHdfsFileNames(partition_descriptor, output_partition,
+ external_partition_name_ss.str());
if (ShouldSkipStaging(state, output_partition)) {
// We will be writing to the final file if we're skipping staging, so get a connection
@@ -739,9 +789,8 @@ void HdfsTableSink::Close(RuntimeState* state) {
}
bool HdfsTableSink::ShouldSkipStaging(RuntimeState* state, OutputPartition* partition) {
- if (IsTransactional()) return true;
+ if (IsTransactional() || HasExternalOutputDir() || is_result_sink_) return true;
// We skip staging if we are writing query results
- if (is_result_sink_) return true;
return (IsS3APath(partition->final_hdfs_file_name_prefix.c_str()) && !overwrite_ &&
state->query_options().s3_skip_insert_staging);
}
diff --git a/be/src/exec/hdfs-table-sink.h b/be/src/exec/hdfs-table-sink.h
index a66aa7c..2263531 100644
--- a/be/src/exec/hdfs-table-sink.h
+++ b/be/src/exec/hdfs-table-sink.h
@@ -176,7 +176,7 @@ class HdfsTableSink : public DataSink {
/// the partition_key_names_ and the evaluated partition_key_exprs_.
/// The Hdfs file name is the unique_id_str_.
void BuildHdfsFileNames(const HdfsPartitionDescriptor& partition_descriptor,
- OutputPartition* output);
+ OutputPartition* output, const std::string &external_partition_path);
/// Writes all rows referenced by the row index vector in 'partition_pair' to the
/// partition's writer and clears the row index vector afterwards.
@@ -217,6 +217,9 @@ class HdfsTableSink : public DataSink {
/// Returns TRUE for Iceberg tables.
bool IsIceberg() const { return table_desc_->IsIcebergTable(); }
+ /// Returns TRUE if an external output directory was provided.
+ bool HasExternalOutputDir() { return !external_output_dir_.empty(); }
+
/// Descriptor of target table. Set in Prepare().
const HdfsTableDescriptor* table_desc_;
@@ -263,6 +266,14 @@ class HdfsTableSink : public DataSink {
/// <hdfs_table_base_dir>/_impala_insert_staging/ during Prepare()
std::string staging_dir_;
+ /// The directory in which an external FE expects results to be written to.
+ std::string external_output_dir_;
+
+ /// How deep into the partition specification in which to start creating partition
+ // directories. Used in conjunction with external_output_dir_ to inform the table
+ // sink which directories are pre-created.
+ int external_output_partition_depth_ = 0;
+
/// string representation of the unique fragment instance id. Used for per-partition
/// Hdfs file names, and for tmp Hdfs directories. Set in Prepare();
std::string unique_id_str_;
diff --git a/common/thrift/DataSinks.thrift b/common/thrift/DataSinks.thrift
index af887de..07c1651 100644
--- a/common/thrift/DataSinks.thrift
+++ b/common/thrift/DataSinks.thrift
@@ -90,6 +90,13 @@ struct THdfsTableSink {
// Indicates that this HdfsTableSink is writing query results
8: optional bool is_result_sink = false;
+
+ // Indicates that an external FE is expecting results here
+ 9: optional string external_output_dir;
+
+ // Indicates how deep into the partition specification in which to start creating
+ // partition directories
+ 10: optional i32 external_output_partition_depth;
}
// Structure to encapsulate specific options that are passed down to the KuduTableSink
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java b/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java
index 1568a76..96a699d 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java
@@ -83,6 +83,21 @@ public class HdfsTableSink extends TableSink {
// A value of 0 means no limit.
private int maxHdfsSinks_;
+ // Stores the desired location for the sink output. Primarily intended to be used by
+ // external FEs that expect the results at a certain location for the purposes of
+ // post processing and finalization.
+ protected String externalOutputDir_;
+
+ // This specifies the position in the partition specifition in which to start
+ // creating partition directories.
+ // Example:
+ // Partition specification: year, month, day
+ // The external frontend has provided a path that has year, month created
+ // by them and thus provides a depth of 2 which hints to the TableSink to skip the
+ // first two partition keys and create the day directory for new results. This is
+ // usually used by tables with dynamic partition columns.
+ private int externalOutputPartitionDepth_;
+
public HdfsTableSink(FeTable targetTable, List<Expr> partitionKeyExprs,
List<Expr> outputExprs, boolean overwrite, boolean inputIsClustered,
Pair<List<Integer>, TSortingOrder> sortProperties, long writeId,
@@ -99,6 +114,14 @@ public class HdfsTableSink extends TableSink {
isResultSink_ = isResultSink;
}
+ public void setExternalOutputDir(String externalOutputDir) {
+ externalOutputDir_ = externalOutputDir;
+ }
+
+ public void setExternalOutputPartitionDepth(int partitionDepth) {
+ externalOutputPartitionDepth_ = partitionDepth;
+ }
+
@Override
public void computeResourceProfile(TQueryOptions queryOptions) {
PlanNode inputNode = fragment_.getPlanRoot();
@@ -220,6 +243,10 @@ public class HdfsTableSink extends TableSink {
hdfsTableSink.setSort_columns(sortColumns_);
hdfsTableSink.setSorting_order(sortingOrder_);
hdfsTableSink.setIs_result_sink(isResultSink_);
+ if (externalOutputDir_ != null) {
+ hdfsTableSink.setExternal_output_dir(externalOutputDir_);
+ hdfsTableSink.setExternal_output_partition_depth(externalOutputPartitionDepth_);
+ }
if (writeId_ != -1) hdfsTableSink.setWrite_id(writeId_);
TTableSink tTableSink = new TTableSink(DescriptorTable.TABLE_SINK_ID,
TTableSinkType.HDFS, sinkOp_.toThrift());
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 177b9e7..a78ca91 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -1796,19 +1796,28 @@ public class Frontend {
private static void addFinalizationParamsForInsert(
TQueryCtx queryCtx, TQueryExecRequest queryExecRequest, InsertStmt insertStmt) {
FeTable targetTable = insertStmt.getTargetTable();
+ addFinalizationParamsForInsert(queryCtx, queryExecRequest,
+ targetTable, insertStmt.getWriteId(), insertStmt.isOverwrite());
+ }
+
+ // This is public to allow external frontends to utilize this method to fill in the
+ // finalization parameters for externally generated INSERTs.
+ public static void addFinalizationParamsForInsert(
+ TQueryCtx queryCtx, TQueryExecRequest queryExecRequest, FeTable targetTable,
+ long writeId, boolean isOverwrite) {
if (targetTable instanceof FeFsTable) {
TFinalizeParams finalizeParams = new TFinalizeParams();
- finalizeParams.setIs_overwrite(insertStmt.isOverwrite());
- finalizeParams.setTable_name(insertStmt.getTargetTableName().getTbl());
+ finalizeParams.setIs_overwrite(isOverwrite);
+ finalizeParams.setTable_name(targetTable.getTableName().getTbl());
finalizeParams.setTable_id(DescriptorTable.TABLE_SINK_ID);
- String db = insertStmt.getTargetTableName().getDb();
+ String db = targetTable.getTableName().getDb();
finalizeParams.setTable_db(db == null ? queryCtx.session.database : db);
- FeFsTable hdfsTable = (FeFsTable) insertStmt.getTargetTable();
+ FeFsTable hdfsTable = (FeFsTable) targetTable;
finalizeParams.setHdfs_base_dir(hdfsTable.getHdfsBaseDir());
- if (insertStmt.getWriteId() != -1) {
+ if (writeId != -1) {
Preconditions.checkState(queryCtx.isSetTransaction_id());
finalizeParams.setTransaction_id(queryCtx.getTransaction_id());
- finalizeParams.setWrite_id(insertStmt.getWriteId());
+ finalizeParams.setWrite_id(writeId);
} else if (targetTable instanceof FeIcebergTable) {
FeIcebergTable iceTable = (FeIcebergTable)targetTable;
finalizeParams.setSpec_id(iceTable.getDefaultPartitionSpecId());