Copilot commented on code in PR #12827:
URL: https://github.com/apache/gluten/pull/12827#discussion_r3814730476
##########
gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala:
##########
@@ -55,12 +71,64 @@ object GlutenIcebergSourceUtil {
}
}
+ /**
+ * The per-table S3 credentials the catalog vended for this scan's table,
empty when the table has
+ * none - which is the case whenever the files are readable with the process
credentials.
+ *
+ * The credentials are read on the driver at planning time and travel with
the split; executors
+ * cannot re-vend them, so a scan has to start within the vend TTL.
+ */
+ def vendedReadProperties(sparkScan: Scan): JMap[String, String] = sparkScan
match {
+ case scan: SparkBatchQueryScan =>
+ val table = scan.table()
+ val ioProperties =
+ try {
+ table.io().properties()
+ } catch {
+ // FileIO.properties() is implemented by S3FileIO and
ResolvingFileIO but
+ // defaults to throwing on FileIOs that do not expose their
configuration.
+ // Such a FileIO cannot be carrying vended credentials.
+ case _: UnsupportedOperationException =>
Collections.emptyMap[String, String]()
+ }
+ extractVendedReadProperties(
+ ioProperties,
+
BackendsApiManager.getTransformerApiInstance.encodeFilePathIfNeed(table.location()))
+ case _ => Collections.emptyMap()
+ }
+
+ /**
+ * Keeps the vended credential set only when the access-key/secret pair is
present, carrying the
+ * optional companions verbatim and the table location under [[LocationKey]].
+ */
+ private[source] def extractVendedReadProperties(
+ ioProperties: JMap[String, String],
+ encodedTableLocation: String): JMap[String, String] = {
+ val accessKeyId = ioProperties.get(S3AccessKeyId)
+ val secretAccessKey = ioProperties.get(S3SecretAccessKey)
+ if (accessKeyId == null || secretAccessKey == null) {
+ return Collections.emptyMap()
+ }
+ val readProperties = new JHashMap[String, String]()
+ readProperties.put(S3AccessKeyId, accessKeyId)
+ readProperties.put(S3SecretAccessKey, secretAccessKey)
+ OptionalCredentialKeys.foreach {
+ key =>
+ val value = ioProperties.get(key)
+ if (value != null) {
+ readProperties.put(key, value)
+ }
+ }
+ readProperties.put(LocationKey, encodedTableLocation)
+ readProperties
+ }
Review Comment:
This will serialize *any* `s3.access-key-id`/`s3.secret-access-key` present
in `FileIO.properties()` into the split payload, not just REST-catalog vended
credentials. In setups where users configure static S3 keys via Iceberg/Spark
config, this change could unexpectedly propagate long-lived secrets into
Substrait payloads (and potentially logs/plan dumps). Consider tightening the
detection to a signal that specifically indicates vended/temporary creds (e.g.,
requiring a session token / expiry marker, or a dedicated marker property), or
(at minimum) making the config default opt-in when static keys are detected to
avoid broad secret propagation.
##########
gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java:
##########
@@ -112,6 +116,7 @@ protected LocalFilesNode(LocalFilesNode other,
List<Map<String, Object>> otherMe
this.fileFormat = other.fileFormat;
this.preferredLocations.addAll(other.preferredLocations);
this.fileReadProperties = other.fileReadProperties;
+ this.readProperties = other.readProperties;
this.iterAsInput = other.iterAsInput;
Review Comment:
`readProperties` is assigned by reference in the copy constructor, so
subsequent mutation of the map (through `setReadProperties` or external
references) can unintentionally affect multiple `LocalFilesNode` instances.
Copying into a new map (or defensively wrapping in an immutable map) would make
the node safer to clone and reason about.
##########
ep/build-velox/src/get-velox.sh:
##########
@@ -25,7 +25,9 @@ RUN_SETUP_SCRIPT=ON
ENABLE_ENHANCED_FEATURES=OFF
# Developer use only for testing Velox PR.
-UPSTREAM_VELOX_PR_ID=""
+# TODO: reset to "" once facebookincubator/velox#18570 (per-path S3 credentials
+# from the query TokenProvider) is merged and the Velox pin above includes it.
+UPSTREAM_VELOX_PR_ID="18570"
Review Comment:
Hard-coding `UPSTREAM_VELOX_PR_ID` to a live upstream PR makes builds
non-reproducible and risks unintentionally shipping with an unreleased Velox
dependency. Prefer leaving this empty in-repo and wiring the PR override
through CI/environment (or a dedicated dev-only flag), so merge artifacts
always build from the pinned Velox SHA.
##########
cpp/velox/utils/GlutenS3TokenProvider.cc:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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 "utils/GlutenS3TokenProvider.h"
+
+#include <cstring>
+#include <functional>
+
+#include "velox/common/base/BitUtil.h"
+
+namespace gluten {
+namespace {
+
+// Segment-boundary-safe prefix match: "bucket/tableA" must not claim
+// "bucket/tableAB/part.parquet".
+bool prefixMatches(const std::string& path, const std::string& prefix) {
+ if (prefix.empty() || path.size() < prefix.size() || path.compare(0,
prefix.size(), prefix) != 0) {
+ return false;
+ }
+ return path.size() == prefix.size() || prefix.back() == '/' ||
path[prefix.size()] == '/';
+}
+
+std::string findOrEmpty(const std::unordered_map<std::string, std::string>&
properties, const char* key) {
+ const auto it = properties.find(key);
+ return it == properties.end() ? "" : it->second;
+}
+
+} // namespace
+
+GlutenS3TokenProvider::GlutenS3TokenProvider(std::map<std::string,
S3TableCredentials> credentialsByPrefix)
+ : credentialsByPrefix_(std::move(credentialsByPrefix)) {
+ const std::hash<std::string> hasher;
+ size_t hash = 0;
+ // std::map iteration order is deterministic, so equal contents hash equally.
+ for (const auto& [prefix, credentials] : credentialsByPrefix_) {
+ hash = facebook::velox::bits::hashMix(hash, hasher(prefix));
+ hash = facebook::velox::bits::hashMix(hash,
hasher(credentials.accessKeyId));
+ hash = facebook::velox::bits::hashMix(hash,
hasher(credentials.secretAccessKey));
+ hash = facebook::velox::bits::hashMix(hash,
hasher(credentials.sessionToken));
+ }
+ hash_ = hash;
+}
+
+std::shared_ptr<GlutenS3TokenProvider> GlutenS3TokenProvider::create(
+ const std::vector<std::unordered_map<std::string, std::string>>&
readProperties) {
+ std::map<std::string, S3TableCredentials> credentialsByPrefix;
+ for (const auto& properties : readProperties) {
+ const auto location = findOrEmpty(properties, kReadPropertiesLocation);
+ const auto accessKeyId = findOrEmpty(properties,
kReadPropertiesAccessKeyId);
+ const auto secretAccessKey = findOrEmpty(properties,
kReadPropertiesSecretAccessKey);
+ if (location.empty() || accessKeyId.empty() || secretAccessKey.empty()) {
+ continue;
+ }
+ credentialsByPrefix[normalizeS3Path(location)] =
+ S3TableCredentials{accessKeyId, secretAccessKey,
findOrEmpty(properties, kReadPropertiesSessionToken)};
+ }
+ if (credentialsByPrefix.empty()) {
+ return nullptr;
+ }
+ return
std::make_shared<GlutenS3TokenProvider>(std::move(credentialsByPrefix));
+}
+
+bool GlutenS3TokenProvider::equals(const
facebook::velox::filesystems::TokenProvider& other) const {
+ const auto* typedOther = dynamic_cast<const GlutenS3TokenProvider*>(&other);
+ return typedOther != nullptr && credentialsByPrefix_ ==
typedOther->credentialsByPrefix_;
+}
+
+size_t GlutenS3TokenProvider::hash() const {
+ return hash_;
+}
+
+std::shared_ptr<facebook::velox::filesystems::AccessToken>
GlutenS3TokenProvider::getToken(
+ const facebook::velox::filesystems::AccessTokenKey& key) const {
+ const auto* s3Key = dynamic_cast<const
facebook::velox::filesystems::S3AccessTokenKey*>(&key);
+ if (s3Key == nullptr) {
+ return nullptr;
+ }
+ const auto& path = s3Key->path();
+ const S3TableCredentials* longestMatch = nullptr;
+ size_t longestMatchSize = 0;
+ for (const auto& [prefix, credentials] : credentialsByPrefix_) {
+ if (prefix.size() >= longestMatchSize && prefixMatches(path, prefix)) {
+ longestMatch = &credentials;
+ longestMatchSize = prefix.size();
+ }
+ }
+ if (longestMatch == nullptr) {
+ return nullptr;
+ }
+ return std::make_shared<facebook::velox::filesystems::S3AccessToken>(
+ longestMatch->accessKeyId, longestMatch->secretAccessKey,
longestMatch->sessionToken);
+}
Review Comment:
`getToken()` does an O(numTables) scan on every file open and allocates a
new `S3AccessToken` each time. Since credentials are immutable for the life of
the query, consider precomputing and storing a `shared_ptr<S3AccessToken>` per
prefix during construction (and returning it directly) to reduce
allocations/copies; if table count can grow, also consider a data structure
that avoids linear scans (e.g., ordered prefixes with early break or a trie).
##########
gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto:
##########
@@ -111,6 +111,12 @@ message ReadRel {
repeated FileOrFiles items = 1;
substrait.extensions.AdvancedExtension advanced_extension = 10;
+ // Table-scoped storage properties the native reader needs to open the
+ // listed files, e.g. the S3 credentials an Iceberg REST catalog vends for
+ // one table. A LocalFiles is single-table by construction, so table
+ // granularity is split granularity. Emitted only when present.
+ map<string, string> read_properties = 11;
Review Comment:
Since `read_properties` can carry access keys/secrets, it’s important to
ensure these fields are not emitted in any debug logging, plan explain output,
or error messages that stringify the Substrait plan/splits. Consider adding
explicit redaction in any plan/split pretty-printer used by this repo (or
documenting that `read_properties` must never be logged) to prevent credential
leakage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
