You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by db...@apache.org on 2025/02/20 20:40:12 UTC
(impala) branch master updated (0f17ae8fc -> 26b7116c3)
This is an automated email from the ASF dual-hosted git repository.
dbecker pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 0f17ae8fc IMPALA-13771: Fix heap-use-after-free in Cluster Membership Manager
new dacc44793 IMPALA-13768: Redundant Iceberg delete records are shuffled around which cause error "Invalid file path arrived at builder"
new 26b7116c3 IMPALA-13724: Add hostnames for Docker host and gateway to Impala containers
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/iceberg-delete-builder.cc | 21 ++++++++-------
be/src/runtime/krpc-data-stream-sender.cc | 22 +++++++---------
bin/start-impala-cluster.py | 19 +++++++++++---
.../QueryTest/iceberg-delete-partitioned.test | 30 ++++++++++++++++++++++
4 files changed, 67 insertions(+), 25 deletions(-)
(impala) 02/02: IMPALA-13724: Add hostnames for Docker host and gateway to Impala containers
Posted by db...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dbecker pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 26b7116c3bb296615d297af518f29bddcb2b2bd9
Author: Laszlo Gaal <la...@cloudera.com>
AuthorDate: Mon Nov 25 08:58:55 2024 -0800
IMPALA-13724: Add hostnames for Docker host and gateway to Impala containers
Reverse DNS lookup for the Docker container's internal gateway (routing
traffic between code running inside the container and code runnning
natively on the Docker host) happens differently on various operating
systems. Some recent platforms, like RHEL 9 resolve this address to the
default name _gateway. Unfortunately the Java Thrift library within
Impala's frontend considers the underscore character invalid in DNS
names, so it throws an error, preventing the Impala coordinator from
connecting to HMS. This kills Impala on startup, blocking any testing
efforts inside containers.
To avoid this problem this patch adds explicit entries to the container's
/etc/hosts file for the gateway's address as well as the Docker host network.
The name doesn't really matter, as it is used only for Thrift's logging
code, so the mapping uses constant generic name 'gateway'.
The IP address of the gateway is retrieved from the environment variable
INTERNAL_LISTEN_HOST, which is set up by docker/configure_test_network.sh
before the Impala containers are launched.
Tested by a dockerised test run executed on Rocky Linux 9.2, using Rocky
9.2 Docker base images for the Impala containers.
Change-Id: I545607c0bb32f8043a0d3f6045710f28a47bab99
Reviewed-on: http://gerrit.cloudera.org:8080/22438
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
bin/start-impala-cluster.py | 19 ++++++++++++++++---
1 file changed, 16 insertions(+), 3 deletions(-)
diff --git a/bin/start-impala-cluster.py b/bin/start-impala-cluster.py
index 9f56eeeb2..725525d5b 100755
--- a/bin/start-impala-cluster.py
+++ b/bin/start-impala-cluster.py
@@ -1003,13 +1003,25 @@ class DockerMiniClusterOperations(object):
# for config changes to take effect.
conf_dir = os.path.join(IMPALA_HOME, "fe/src/test/resources/")
mount_args = ["--mount", "type=bind,src={0},dst=/opt/impala/conf".format(conf_dir)]
-
# Collect container logs in a unique subdirectory per daemon to avoid any potential
# interaction between containers, which should be isolated.
log_dir = os.path.join(IMPALA_HOME, options.log_dir, host_name)
if not os.path.isdir(log_dir):
os.makedirs(log_dir)
mount_args += ["--mount", "type=bind,src={0},dst=/opt/impala/logs".format(log_dir)]
+ # Add entries to the container's /etc/hosts file for the Docker host and the
+ # gateway from the container to the host. These are needed for stable reverse
+ # name resolution of the host's and the gateway's IP addresses
+ # The local host's name is governed by convention, see
+ # https://docs.docker.com/reference/cli/docker/container/run/#add-host
+ local_host_arg = ["--add-host=host.docker.internal:host-gateway"]
+ # For a dockerised setup INTERNAL_LISTEN_HOST stores the Docker gateway's IP address,
+ # prepared by docker/configure_test_network.sh. If it contains "localhost", then
+ # setup is not complete, so don't add it; it would be invalid as an IP address
+ # anyway.
+ local_gateway_arg = None
+ if INTERNAL_LISTEN_HOST != "localhost":
+ local_gateway_arg = ["--add-host=gateway.internal:{0}".format(INTERNAL_LISTEN_HOST)]
# Create a data cache subdirectory for each daemon and mount at /opt/impala/cache
# in the container.
@@ -1028,8 +1040,9 @@ class DockerMiniClusterOperations(object):
mem_limit_args = ["--memory", str(mem_limit)]
LOG.info("Running container {0}".format(container_name))
run_cmd = (["docker", "run", "-d"] + env_args + port_args + user_args + ["--network",
- self.network_name, "--name", container_name, "--network-alias", host_name] +
- mount_args + mem_limit_args + [image_tag] + args)
+ self.network_name, "--name", container_name, "--network-alias", host_name]
+ + local_host_arg + local_gateway_arg
+ + mount_args + mem_limit_args + [image_tag] + args)
LOG.info("Running command {0}".format(run_cmd))
check_call(run_cmd)
port_mapping = check_output(["docker", "port", container_name],
(impala) 01/02: IMPALA-13768: Redundant Iceberg delete records are shuffled around which cause error "Invalid file path arrived at builder"
Posted by db...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dbecker pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit dacc4479383b8d0c885452abf1f17e2a51de950d
Author: Zoltan Borok-Nagy <bo...@cloudera.com>
AuthorDate: Wed Feb 19 14:11:19 2025 +0100
IMPALA-13768: Redundant Iceberg delete records are shuffled around which cause error "Invalid file path arrived at builder"
IcebergDeleteBuilder assumes that it should only receive delete
records for paths of data files that are scheduled for its
corresponding SCAN operator.
It is not true when any of the following happens:
* number of output channels in sender is 1
(currently no DIRECTED mode, no filtering)
* hit bug in DIRECTED mode, see below
* single node plan is being used (no DIRECTED mode, no filtering)
With this patch, KrpcDataStreamSender::Send() will use DIRECTED mode
even if number of output channels is 1. It also fixes the bug in
DIRECTED mode (which was due to an unused variable 'skipped_prev_row')
and simplified the logic a bit.
The patch also relaxes the assumption in IcebergDeleteBuilder, i.e.
only return error for dangling delete records when we are in a
distributed plan where we can assume DIRECTED distribution mode of
position delete records.
Testing
* added e2e tests
Change-Id: I695c919c9a74edec768e413a02b2ef7dbfa0d6a5
Reviewed-on: http://gerrit.cloudera.org:8080/22500
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/iceberg-delete-builder.cc | 21 ++++++++-------
be/src/runtime/krpc-data-stream-sender.cc | 22 +++++++---------
.../QueryTest/iceberg-delete-partitioned.test | 30 ++++++++++++++++++++++
3 files changed, 51 insertions(+), 22 deletions(-)
diff --git a/be/src/exec/iceberg-delete-builder.cc b/be/src/exec/iceberg-delete-builder.cc
index 4692aca0e..fc4833ace 100644
--- a/be/src/exec/iceberg-delete-builder.cc
+++ b/be/src/exec/iceberg-delete-builder.cc
@@ -261,16 +261,19 @@ Status IcebergDeleteBuilder::AddToDeletedRows(const StringValue& path,
} else if (path.Len() == 0) {
return Status("NULL found as file_path in delete file");
} else {
- // 'deleted_rows_' should contain all data file paths that this join
- // processes.
- stringstream ss;
- ss << "Invalid file path arrived at builder: " << path << " Paths expected: [";
- for (auto& [vec_path, unused] : deleted_rows_) {
- ss << vec_path << ", ";
+ // We received a path of a data file that is not scheduled for this fragment.
+ if (is_separate_build_) {
+ // When the build is in its own fragment we assume that delete records are filtered
+ // in the sender by DIRECTED distribution mode. Therefore we should never see
+ // dangling delete records.
+ stringstream ss;
+ ss << "Invalid file path arrived at builder: " << path << " Paths expected: [";
+ for (auto& [vec_path, unused] : deleted_rows_) {
+ ss << vec_path << ", ";
+ }
+ ss << "]";
+ return Status(ss.str());
}
- ss << "]";
- DCHECK(false) << ss.str();
- return Status(ss.str());
}
return Status::OK();
}
diff --git a/be/src/runtime/krpc-data-stream-sender.cc b/be/src/runtime/krpc-data-stream-sender.cc
index 6a41018c2..127ced2a1 100644
--- a/be/src/runtime/krpc-data-stream-sender.cc
+++ b/be/src/runtime/krpc-data-stream-sender.cc
@@ -1133,7 +1133,8 @@ Status KrpcDataStreamSender::Send(RuntimeState* state, RowBatch* batch) {
}
// At this point no RPCs can still refer to the old in_flight_batch_.
in_flight_batch_.swap(serialization_batch_);
- } else if (partition_type_ == TPartitionType::RANDOM || channels_.size() == 1) {
+ } else if (partition_type_ == TPartitionType::RANDOM ||
+ (channels_.size() == 1 && partition_type_ != TPartitionType::DIRECTED)) {
// Round-robin batches among channels. Wait for the current channel to finish its
// rpc before overwriting its batch.
Channel* current_channel = channels_[current_channel_idx_].get();
@@ -1171,7 +1172,6 @@ Status KrpcDataStreamSender::Send(RuntimeState* state, RowBatch* batch) {
const int num_rows = batch->num_rows();
char* prev_filename_ptr = nullptr;
vector<IcebergPositionDeleteChannel*> prev_channels;
- bool skipped_prev_row = false;
for (int row_idx = 0; row_idx < num_rows; ++row_idx) {
DCHECK_EQ(batch->num_tuples_per_row(), 1);
TupleRow* tuple_row = batch->GetRow(row_idx);
@@ -1182,24 +1182,22 @@ Status KrpcDataStreamSender::Send(RuntimeState* state, RowBatch* batch) {
if (filename_value_ss.ptr == prev_filename_ptr) {
// If the filename pointer is the same as the previous one then we can instantly
// send the row to the same channels as the previous row.
- DCHECK(skipped_prev_row || !prev_channels.empty() ||
- (filename_value_ss.len == 0 && prev_channels.empty()));
for (IcebergPositionDeleteChannel* ch : prev_channels) {
RETURN_IF_ERROR(ch->AddRow(tuple_row));
}
continue;
}
+ prev_channels.clear();
prev_filename_ptr = filename_value_ss.ptr;
string filename(filename_value_ss.ptr, filename_value_ss.len);
const auto filepath_to_hosts_it = filepath_to_hosts_.find(filename);
if (filepath_to_hosts_it == filepath_to_hosts_.end()) {
- // This can happen when e.g. compaction removed some data files from a snapshot
- // but a delete file referencing them remained because it references other data
- // files that remains in the new snapshot.
- // Another use-case is table sampling where we read only a subset of the data
- // files.
- // A third case is when the delete record is invalid.
+ // This can happen due to file pruning, or when compaction removed some data
+ // files from a snapshot but a delete file referencing them remained because
+ // it references other data files that are still present in the new snapshot.
+ // Another case is table sampling where we read only a subset of the data files.
+ // Or, when simply the delete record is invalid.
if (UNLIKELY(filename_value_ss.len == 0)) {
state->LogError(
ErrorMsg(TErrorCode::GENERAL, "NULL found as file_path in delete file"));
@@ -1207,12 +1205,10 @@ Status KrpcDataStreamSender::Send(RuntimeState* state, RowBatch* batch) {
VLOG(3) << "Row from delete file refers to a non-existing data file: " <<
filename;
}
- skipped_prev_row = true;
+ DCHECK(prev_channels.empty());
continue;
}
- skipped_prev_row = false;
- prev_channels.clear();
for (const NetworkAddressPB& host_addr : filepath_to_hosts_it->second) {
const auto channel_map_it = host_to_channel_.find(host_addr);
if (channel_map_it == host_to_channel_.end()) {
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-delete-partitioned.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-delete-partitioned.test
index 0e8d8e973..b81fa89c8 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-delete-partitioned.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-delete-partitioned.test
@@ -803,3 +803,33 @@ delete from ice_alltypes_part_v2 where i=1;
---- TYPES
INT, BOOLEAN, INT, BIGINT, FLOAT, DOUBLE, DECIMAL, DATE, STRING
====
+---- QUERY
+# Create table to test IMPALA-13768. It will have multiple partitions, with
+# multiple data files in each partition, where min/max ranges of column 'bi'
+# is disjunct for data files (in same partition).
+create table ice_invalid_deletes (bi bigint, year int)
+partitioned by spec (year)
+stored as iceberg tblproperties ('format-version'='2');
+insert into ice_invalid_deletes
+ select bigint_col, year from functional.alltypes where month = 10;
+with v as (select max(bi) as max_bi from ice_invalid_deletes)
+insert into ice_invalid_deletes select bi + v.max_bi, year from v, ice_invalid_deletes;
+delete from ice_invalid_deletes where bi % 11 = 0;
+====
+---- QUERY
+# Scan single data file (single channel in EXCHANGE sender) with delete file
+# that have references to other data files.
+select count(*) from ice_invalid_deletes where year=2010 and bi = 180;
+---- RESULTS
+31
+---- TYPES
+BIGINT
+====
+---- QUERY
+# Scan single data file per partition with delete files that reference other data files.
+select count(*) from ice_invalid_deletes where year>2000 and bi = 180;
+---- RESULTS
+62
+---- TYPES
+BIGINT
+====