This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new af7b9dd156 [VL] feat(iceberg): Add V3 initial value reads (#12537)
af7b9dd156 is described below
commit af7b9dd15658faa883ac8de4fa1c9c0ddd9f6d14
Author: inf <[email protected]>
AuthorDate: Wed Aug 19 13:21:15 2026 +0000
[VL] feat(iceberg): Add V3 initial value reads (#12537)
---
.../gluten/execution/VeloxIcebergSuite.scala | 50 +++++++++++++++++++++-
.../apache/gluten/proto/IcebergReadExtension.proto | 24 +++++++++++
.../gluten/backendsapi/velox/VeloxBackend.scala | 2 +
.../backendsapi/velox/VeloxTransformerApi.scala | 30 ++++++++++++-
cpp/velox/compute/VeloxPlanConverter.cc | 9 ++--
cpp/velox/compute/iceberg/IcebergPlanConverter.cc | 26 +++++++++++
cpp/velox/compute/iceberg/IcebergPlanConverter.h | 11 +++++
cpp/velox/substrait/SubstraitToVeloxPlan.cc | 35 ++++++++++++++-
.../org/apache/gluten/IcebergDefaultValueUtil.java | 12 +++++-
.../org/apache/gluten/IcebergDefaultValueUtil.java | 12 +++++-
.../org/apache/gluten/IcebergDefaultValueUtil.java | 12 +++++-
.../substrait/rel/IcebergLocalFilesBuilder.java | 8 +++-
.../substrait/rel/IcebergLocalFilesNode.java | 29 ++++++++++++-
.../gluten/execution/IcebergScanTransformer.scala | 24 ++++++++++-
.../spark/source/GlutenIcebergSourceUtil.scala | 41 +++++++++++++++++-
.../rel/IcebergLocalFilesNodeBoundsTest.java | 4 +-
.../gluten/backendsapi/BackendSettingsApi.scala | 2 +
.../apache/gluten/backendsapi/TransformerApi.scala | 7 +++
18 files changed, 317 insertions(+), 21 deletions(-)
diff --git
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
index edb30dac61..4e540c15af 100644
---
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
+++
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
@@ -16,4 +16,52 @@
*/
package org.apache.gluten.execution
-class VeloxIcebergSuite extends IcebergSuite
+import org.apache.gluten.config.GlutenConfig
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
+
+import org.apache.iceberg.UpdateSchema
+import org.apache.iceberg.expressions.Literal
+import org.apache.iceberg.spark.source.SparkTable
+import org.apache.iceberg.types.{Type, Types}
+
+class VeloxIcebergSuite extends IcebergSuite {
+ testWithMinSparkVersion("iceberg v3 initial default for an added column",
"3.4") {
+ withTable("iceberg_v3_initial_default") {
+ withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false") {
+ spark.sql("""
+ |CREATE TABLE iceberg_v3_initial_default (id INT)
+ |USING iceberg
+ |TBLPROPERTIES ('format-version' = '3')
+ |""".stripMargin)
+ spark.sql("INSERT INTO iceberg_v3_initial_default VALUES (1), (2)")
+
+ val catalog = spark.sessionState.catalogManager
+ .catalog("spark_catalog")
+ .asInstanceOf[TableCatalog]
+ val updateSchema = catalog
+ .loadTable(Identifier.of(Array("default"),
"iceberg_v3_initial_default"))
+ .asInstanceOf[SparkTable]
+ .table()
+ .updateSchema()
+ classOf[UpdateSchema]
+ .getMethod(
+ "addColumn",
+ classOf[String],
+ classOf[Type],
+ classOf[Literal[_]])
+ .invoke(updateSchema, "country", Types.StringType.get(),
Literal.of("IN"))
+ updateSchema.commit()
+ spark.catalog.refreshTable("iceberg_v3_initial_default")
+ }
+
+ runQueryAndCompare(
+ "SELECT id, country FROM iceberg_v3_initial_default ORDER BY id") {
+ df =>
+ checkAnswer(df, Seq(Row(1, "IN"), Row(2, "IN")))
+ checkGlutenPlan[IcebergScanTransformer](df)
+ }
+ }
+ }
+}
diff --git
a/backends-velox/src/main/resources/org/apache/gluten/proto/IcebergReadExtension.proto
b/backends-velox/src/main/resources/org/apache/gluten/proto/IcebergReadExtension.proto
new file mode 100644
index 0000000000..2ea518dd6d
--- /dev/null
+++
b/backends-velox/src/main/resources/org/apache/gluten/proto/IcebergReadExtension.proto
@@ -0,0 +1,24 @@
+// SPDX-License-Identifier: Apache-2.0
+syntax = "proto3";
+
+package gluten;
+
+option java_package = "org.apache.gluten.proto";
+option java_multiple_files = true;
+
+// Iceberg-specific metadata carried in LocalFiles.advanced_extension.
+message IcebergReadExtension {
+ message ColumnFieldId {
+ string name = 1;
+ int32 field_id = 2;
+ }
+
+ message ColumnDefault {
+ string name = 1;
+ int32 field_id = 2;
+ string initial_default = 3;
+ }
+
+ repeated ColumnDefault column_defaults = 1;
+ repeated ColumnFieldId column_field_ids = 2;
+}
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala
index 09deec69e4..27d1dc0a2d 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala
@@ -571,6 +571,8 @@ object VeloxBackendSettings extends BackendSettingsApi {
override def supportIcebergEqualityDeleteRead(): Boolean = false
+ override def supportIcebergInitialDefaultRead(): Boolean = true
+
override def reorderColumnsForPartitionWrite(): Boolean = true
override def enableEnhancedFeatures(): Boolean =
VeloxConfig.get.enableEnhancedFeatures()
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxTransformerApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxTransformerApi.scala
index 47ad778dc8..04df524427 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxTransformerApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxTransformerApi.scala
@@ -21,7 +21,7 @@ import org.apache.gluten.exception.GlutenException
import org.apache.gluten.execution.WriteFilesExecTransformer
import org.apache.gluten.execution.datasource.GlutenFormatFactory
import org.apache.gluten.expression.ConverterUtils
-import org.apache.gluten.proto.ConfigMap
+import org.apache.gluten.proto.{ConfigMap, IcebergReadExtension}
import org.apache.gluten.runtime.Runtimes
import org.apache.gluten.substrait.SubstraitContext
import org.apache.gluten.substrait.expression.{ExpressionBuilder,
ExpressionNode}
@@ -124,6 +124,34 @@ class VeloxTransformerApi extends TransformerApi with
Logging {
override def packPBMessage(message: Message): Any = Any.pack(message, "")
+ override def packIcebergReadExtension(
+ fieldIds: JMap[String, Integer],
+ initialDefaults: JMap[String, String]): Any = {
+ val extensionBuilder = IcebergReadExtension.newBuilder()
+ fieldIds.asScala.toSeq.sortBy(_._1).foreach {
+ case (name, fieldId) =>
+ extensionBuilder.addColumnFieldIds(
+ IcebergReadExtension.ColumnFieldId
+ .newBuilder()
+ .setName(name)
+ .setFieldId(fieldId))
+ }
+ initialDefaults.asScala.toSeq.sortBy(_._1).foreach {
+ case (name, initialDefault) =>
+ val fieldId = fieldIds.get(name)
+ if (fieldId == null) {
+ throw new IllegalArgumentException(s"Missing Iceberg field ID for
column $name")
+ }
+ extensionBuilder.addColumnDefaults(
+ IcebergReadExtension.ColumnDefault
+ .newBuilder()
+ .setName(name)
+ .setFieldId(fieldId)
+ .setInitialDefault(initialDefault))
+ }
+ packPBMessage(extensionBuilder.build())
+ }
+
override def invalidateSQLExecutionResource(executionId: String): Unit = {
GlutenDriverEndpoint.invalidateResourceRelation(executionId)
}
diff --git a/cpp/velox/compute/VeloxPlanConverter.cc
b/cpp/velox/compute/VeloxPlanConverter.cc
index 22a7c49ac3..82c1f29036 100644
--- a/cpp/velox/compute/VeloxPlanConverter.cc
+++ b/cpp/velox/compute/VeloxPlanConverter.cc
@@ -137,9 +137,10 @@ std::shared_ptr<DeltaSplitInfo> parseDeltaSplitInfo(
std::shared_ptr<SplitInfo> parseScanSplitInfo(
const facebook::velox::config::ConfigBase* veloxCfg,
- const
google::protobuf::RepeatedPtrField<substrait::ReadRel_LocalFiles_FileOrFiles>&
fileList) {
+ const substrait::ReadRel_LocalFiles& localFiles) {
using SubstraitFileFormatCase =
::substrait::ReadRel_LocalFiles_FileOrFiles::FileFormatCase;
+ const auto& fileList = localFiles.items();
auto splitInfo = std::make_shared<SplitInfo>();
splitInfo->leafType = SplitInfo::LeafType::TABLE_SCAN;
splitInfo->paths.reserve(fileList.size());
@@ -193,7 +194,8 @@ std::shared_ptr<SplitInfo> parseScanSplitInfo(
splitInfo->format = dwio::common::FileFormat::TEXT;
break;
case SubstraitFileFormatCase::kIceberg:
- splitInfo = IcebergPlanConverter::parseIcebergSplitInfo(file,
std::move(splitInfo));
+ splitInfo =
+ IcebergPlanConverter::parseIcebergSplitInfo(file,
localFiles.advanced_extension(), std::move(splitInfo));
break;
case SubstraitFileFormatCase::kDelta:
splitInfo = parseDeltaSplitInfo(file, std::move(splitInfo));
@@ -237,8 +239,7 @@ void parseLocalFileNodes(
std::vector<std::shared_ptr<SplitInfo>> splitInfos;
splitInfos.reserve(localFiles.size());
for (const auto& localFile : localFiles) {
- const auto& fileList = localFile.items();
- splitInfos.push_back(parseScanSplitInfo(veloxCfg, fileList));
+ splitInfos.push_back(parseScanSplitInfo(veloxCfg, localFile));
}
planConverter->setSplitInfos(std::move(splitInfos));
diff --git a/cpp/velox/compute/iceberg/IcebergPlanConverter.cc
b/cpp/velox/compute/iceberg/IcebergPlanConverter.cc
index 5212381623..097c984357 100644
--- a/cpp/velox/compute/iceberg/IcebergPlanConverter.cc
+++ b/cpp/velox/compute/iceberg/IcebergPlanConverter.cc
@@ -17,6 +17,8 @@
#include "IcebergPlanConverter.h"
+#include "IcebergReadExtension.pb.h"
+
namespace gluten {
namespace {
@@ -39,6 +41,7 @@ std::unordered_map<int32_t, std::string> parseBounds(const
SubstraitDeleteBounds
std::shared_ptr<IcebergSplitInfo> IcebergPlanConverter::parseIcebergSplitInfo(
substrait::ReadRel_LocalFiles_FileOrFiles file,
+ const substrait::extensions::AdvancedExtension& extension,
std::shared_ptr<SplitInfo> splitInfo) {
using SubstraitFileFormatCase =
::substrait::ReadRel_LocalFiles_FileOrFiles::IcebergReadOptions::FileFormatCase;
using SubstraitDeleteFileFormatCase =
@@ -58,6 +61,29 @@ std::shared_ptr<IcebergSplitInfo>
IcebergPlanConverter::parseIcebergSplitInfo(
icebergSplitInfo->format = dwio::common::FileFormat::UNKNOWN;
break;
}
+
+ if (icebergSplitInfo->columns.empty() && extension.has_enhancement() &&
+ extension.enhancement().Is<::gluten::IcebergReadExtension>()) {
+ ::gluten::IcebergReadExtension icebergExtension;
+ VELOX_USER_CHECK(extension.enhancement().UnpackTo(&icebergExtension),
"Failed to unpack Iceberg read extension");
+ for (const auto& column : icebergExtension.column_field_ids()) {
+ auto [it, inserted] =
+ icebergSplitInfo->columns.try_emplace(column.name(),
IcebergColumnInfo{column.field_id(), std::nullopt});
+ if (!inserted) {
+ VELOX_USER_CHECK_EQ(
+ it->second.fieldId, column.field_id(), "Conflicting Iceberg field
IDs for column '{}'", column.name());
+ }
+ }
+ for (const auto& column : icebergExtension.column_defaults()) {
+ auto [it, inserted] = icebergSplitInfo->columns.try_emplace(
+ column.name(), IcebergColumnInfo{column.field_id(),
column.initial_default()});
+ if (!inserted) {
+ VELOX_USER_CHECK_EQ(
+ it->second.fieldId, column.field_id(), "Conflicting Iceberg field
IDs for column '{}'", column.name());
+ it->second.initialDefault = column.initial_default();
+ }
+ }
+ }
if (icebergReadOption.delete_files_size() > 0) {
auto deleteFiles = icebergReadOption.delete_files();
std::vector<IcebergDeleteFile> deletes;
diff --git a/cpp/velox/compute/iceberg/IcebergPlanConverter.h
b/cpp/velox/compute/iceberg/IcebergPlanConverter.h
index d634a861fe..af2631200e 100644
--- a/cpp/velox/compute/iceberg/IcebergPlanConverter.h
+++ b/cpp/velox/compute/iceberg/IcebergPlanConverter.h
@@ -17,14 +17,24 @@
#pragma once
+#include <optional>
+#include <string>
+#include <unordered_map>
+
#include "substrait/SubstraitToVeloxPlan.h"
#include "velox/connectors/hive/iceberg/IcebergDeleteFile.h"
using namespace facebook::velox::connector::hive::iceberg;
namespace gluten {
+struct IcebergColumnInfo {
+ int32_t fieldId;
+ std::optional<std::string> initialDefault;
+};
+
struct IcebergSplitInfo : SplitInfo {
std::vector<std::vector<IcebergDeleteFile>> deleteFilesVec;
+ std::unordered_map<std::string, IcebergColumnInfo> columns;
IcebergSplitInfo(const SplitInfo& splitInfo) : SplitInfo(splitInfo) {
// Reserve the actual size of the deleteFilesVec.
@@ -36,6 +46,7 @@ class IcebergPlanConverter {
public:
static std::shared_ptr<IcebergSplitInfo> parseIcebergSplitInfo(
substrait::ReadRel_LocalFiles_FileOrFiles file,
+ const substrait::extensions::AdvancedExtension& extension,
std::shared_ptr<SplitInfo> splitInfo);
};
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
index 9e85daa206..97324de4e1 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
+++ b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
@@ -26,6 +26,7 @@
#include "operators/hashjoin/HashTableBuilder.h"
#include "operators/plannodes/RowVectorStream.h"
#include "velox/connectors/hive/HiveDataSink.h"
+#include "velox/connectors/hive/iceberg/IcebergColumnHandle.h"
#include "velox/exec/TableWriter.h"
#include "velox/type/Type.h"
@@ -1609,11 +1610,41 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
std::vector<std::string> outNames;
outNames.reserve(colNameList.size());
connector::ColumnHandleMap assignments;
+ const auto icebergSplitInfo =
std::dynamic_pointer_cast<IcebergSplitInfo>(splitInfo);
for (int idx = 0; idx < colNameList.size(); idx++) {
auto outName = SubstraitParser::makeNodeName(planNodeId_, idx);
auto columnType = columnTypes[idx];
- assignments[outName] = std::make_shared<connector::hive::HiveColumnHandle>(
- colNameList[idx], columnType, veloxTypeList[idx], veloxTypeList[idx]);
+ const IcebergColumnInfo* icebergColumn = nullptr;
+ if (icebergSplitInfo) {
+ auto columnIt = icebergSplitInfo->columns.find(colNameList[idx]);
+ if (columnIt != icebergSplitInfo->columns.end()) {
+ icebergColumn = &columnIt->second;
+ } else if (asLowerCase) {
+ for (const auto& [name, column] : icebergSplitInfo->columns) {
+ auto normalizedName = name;
+ folly::toLowerAscii(normalizedName);
+ if (normalizedName == colNameList[idx]) {
+ icebergColumn = &column;
+ break;
+ }
+ }
+ }
+ }
+ // Gluten serializes Iceberg partition dates as ISO strings. Use Iceberg
+ // handles only for regular columns so all data columns are mapped by field
+ // ID together, while partition columns keep the existing Hive conversion.
+ if (icebergColumn && columnType == ColumnType::kRegular) {
+ assignments[outName] =
std::make_shared<connector::hive::iceberg::IcebergColumnHandle>(
+ colNameList[idx],
+ columnType,
+ veloxTypeList[idx],
+ facebook::velox::parquet::ParquetFieldId(icebergColumn->fieldId),
+ std::vector<common::Subfield>{},
+ icebergColumn->initialDefault);
+ } else {
+ assignments[outName] =
std::make_shared<connector::hive::HiveColumnHandle>(
+ colNameList[idx], columnType, veloxTypeList[idx],
veloxTypeList[idx]);
+ }
outNames.emplace_back(outName);
}
auto outputType = ROW(std::move(outNames), std::move(veloxTypeList));
diff --git
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
b/gluten-iceberg/src-iceberg10/main/java/org/apache/gluten/IcebergDefaultValueUtil.java
similarity index 75%
copy from
backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
copy to
gluten-iceberg/src-iceberg10/main/java/org/apache/gluten/IcebergDefaultValueUtil.java
index edb30dac61..0d50c905b8 100644
---
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
+++
b/gluten-iceberg/src-iceberg10/main/java/org/apache/gluten/IcebergDefaultValueUtil.java
@@ -14,6 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.gluten.execution
+package org.apache.gluten;
-class VeloxIcebergSuite extends IcebergSuite
+import org.apache.iceberg.types.Types;
+
+public final class IcebergDefaultValueUtil {
+ private IcebergDefaultValueUtil() {}
+
+ public static Object getInitialDefault(Types.NestedField field) {
+ return field.initialDefault();
+ }
+}
diff --git
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
b/gluten-iceberg/src-iceberg3/main/java/org/apache/gluten/IcebergDefaultValueUtil.java
similarity index 76%
copy from
backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
copy to
gluten-iceberg/src-iceberg3/main/java/org/apache/gluten/IcebergDefaultValueUtil.java
index edb30dac61..fff8cb444f 100644
---
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
+++
b/gluten-iceberg/src-iceberg3/main/java/org/apache/gluten/IcebergDefaultValueUtil.java
@@ -14,6 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.gluten.execution
+package org.apache.gluten;
-class VeloxIcebergSuite extends IcebergSuite
+import org.apache.iceberg.types.Types;
+
+public final class IcebergDefaultValueUtil {
+ private IcebergDefaultValueUtil() {}
+
+ public static Object getInitialDefault(Types.NestedField field) {
+ return null;
+ }
+}
diff --git
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
b/gluten-iceberg/src-iceberg5/main/java/org/apache/gluten/IcebergDefaultValueUtil.java
similarity index 76%
copy from
backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
copy to
gluten-iceberg/src-iceberg5/main/java/org/apache/gluten/IcebergDefaultValueUtil.java
index edb30dac61..fff8cb444f 100644
---
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
+++
b/gluten-iceberg/src-iceberg5/main/java/org/apache/gluten/IcebergDefaultValueUtil.java
@@ -14,6 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.gluten.execution
+package org.apache.gluten;
-class VeloxIcebergSuite extends IcebergSuite
+import org.apache.iceberg.types.Types;
+
+public final class IcebergDefaultValueUtil {
+ private IcebergDefaultValueUtil() {}
+
+ public static Object getInitialDefault(Types.NestedField field) {
+ return null;
+ }
+}
diff --git
a/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesBuilder.java
b/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesBuilder.java
index eef98f7784..ca4597b48e 100644
---
a/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesBuilder.java
+++
b/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesBuilder.java
@@ -31,7 +31,9 @@ public class IcebergLocalFilesBuilder {
LocalFilesNode.ReadFileFormat fileFormat,
List<String> preferredLocations,
List<List<DeleteFile>> deleteFilesList,
- List<Map<String, String>> metadataColumns) {
+ List<Map<String, String>> metadataColumns,
+ Map<String, Integer> fieldIds,
+ Map<String, String> initialDefaults) {
return new IcebergLocalFilesNode(
index,
paths,
@@ -41,6 +43,8 @@ public class IcebergLocalFilesBuilder {
fileFormat,
preferredLocations,
deleteFilesList,
- metadataColumns);
+ metadataColumns,
+ fieldIds,
+ initialDefaults);
}
}
diff --git
a/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNode.java
b/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNode.java
index ac5841226a..74c589143c 100644
---
a/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNode.java
+++
b/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNode.java
@@ -16,6 +16,10 @@
*/
package org.apache.gluten.substrait.rel;
+import org.apache.gluten.backendsapi.BackendsApiManager;
+
+import com.google.protobuf.Any;
+import io.substrait.proto.AdvancedExtension;
import io.substrait.proto.ReadRel;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileContent;
@@ -25,6 +29,8 @@ import java.util.*;
public class IcebergLocalFilesNode extends LocalFilesNode {
private final List<List<DeleteFile>> deleteFilesList;
+ private final Map<String, Integer> fieldIds;
+ private final Map<String, String> initialDefaults;
IcebergLocalFilesNode(
Integer index,
@@ -35,7 +41,9 @@ public class IcebergLocalFilesNode extends LocalFilesNode {
ReadFileFormat fileFormat,
List<String> preferredLocations,
List<List<DeleteFile>> deleteFilesList,
- List<Map<String, String>> metadataColumns) {
+ List<Map<String, String>> metadataColumns,
+ Map<String, Integer> fieldIds,
+ Map<String, String> initialDefaults) {
super(
index,
paths,
@@ -50,6 +58,25 @@ public class IcebergLocalFilesNode extends LocalFilesNode {
new HashMap<>(),
new ArrayList<>());
this.deleteFilesList = deleteFilesList;
+ this.fieldIds = fieldIds;
+ this.initialDefaults = initialDefaults;
+ }
+
+ @Override
+ public ReadRel.LocalFiles toProtobuf() {
+ ReadRel.LocalFiles localFiles = super.toProtobuf();
+ if (initialDefaults.isEmpty()) {
+ return localFiles;
+ }
+
+ Any extension =
+ BackendsApiManager.getTransformerApiInstance()
+ .packIcebergReadExtension(fieldIds, initialDefaults);
+
+ return localFiles
+ .toBuilder()
+
.setAdvancedExtension(AdvancedExtension.newBuilder().setEnhancement(extension))
+ .build();
}
@Override
diff --git
a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
index cc665ba4eb..16018e086d 100644
---
a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
+++
b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
@@ -41,6 +41,7 @@ import org.apache.iceberg.types.{Type, Types}
import org.apache.iceberg.types.Type.TypeID
import org.apache.iceberg.types.Types.{ListType, MapType, NestedField}
+import java.util.{HashMap => JHashMap}
import java.util.Locale
case class IcebergScanTransformer(
@@ -65,6 +66,16 @@ case class IcebergScanTransformer(
// So use Metric to get NumSplits, NumDeletes is not reported by native
metric
private val numSplits = SQLMetrics.createMetric(sparkContext, new
NumSplits().description())
+ private lazy val icebergInitialDefaults =
+ GlutenIcebergSourceUtil.getInitialDefaults(scan)
+
+ private lazy val icebergFieldIds =
+ if (icebergInitialDefaults.isEmpty) {
+ new JHashMap[String, Integer]()
+ } else {
+ GlutenIcebergSourceUtil.getFieldIds(scan)
+ }
+
override def withNewPushdownFilters(filters: Seq[Expression]):
BatchScanExecTransformerBase = {
this.copy(pushDownFilters = Some(filters))
}
@@ -150,6 +161,12 @@ case class IcebergScanTransformer(
return ValidationResult.succeeded
}
val metadata = baseTable.operations().current()
+ if (
+ !BackendsApiManager.getSettings.supportIcebergInitialDefaultRead() &&
+ !icebergInitialDefaults.isEmpty
+ ) {
+ return ValidationResult.failed("Iceberg initial-default reads are not
supported")
+ }
if (metadata.formatVersion() >= 3) {
val hasUnsupportedDelete = finalPartitions.exists {
case p: SparkDataSourceRDDPartition =>
@@ -208,7 +225,12 @@ case class IcebergScanTransformer(
metadataColumnNames: Seq[String]): SplitInfo = {
val splitInfo = partition match {
case p: SparkDataSourceRDDPartition =>
- GlutenIcebergSourceUtil.genSplitInfo(p, getPartitionSchema,
metadataColumnNames)
+ GlutenIcebergSourceUtil.genSplitInfo(
+ p,
+ getPartitionSchema,
+ metadataColumnNames,
+ icebergFieldIds,
+ icebergInitialDefaults)
case _ => throw new GlutenNotSupportException()
}
numSplits.add(splitInfo.asInstanceOf[LocalFilesNode].getPaths.size())
diff --git
a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
index 1a658a7892..db1bb024af 100644
---
a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
+++
b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
@@ -16,6 +16,7 @@
*/
package org.apache.iceberg.spark.source
+import org.apache.gluten.IcebergDefaultValueUtil
import org.apache.gluten.backendsapi.BackendsApiManager
import org.apache.gluten.exception.GlutenNotSupportException
import org.apache.gluten.execution.SparkDataSourceRDDPartition
@@ -57,7 +58,9 @@ object GlutenIcebergSourceUtil {
def genSplitInfo(
partition: SparkDataSourceRDDPartition,
readPartitionSchema: StructType,
- metadataColumnNames: Seq[String]): SplitInfo = {
+ metadataColumnNames: Seq[String],
+ fieldIds: JMap[String, Integer],
+ initialDefaults: JMap[String, String]): SplitInfo = {
val paths = new JArrayList[String]()
val starts = new JArrayList[JLong]()
val lengths = new JArrayList[JLong]()
@@ -103,10 +106,44 @@ object GlutenIcebergSourceUtil {
.toList
.asJava,
deleteFilesList,
- metadataColumns
+ metadataColumns,
+ fieldIds,
+ initialDefaults
)
}
+ def getFieldIds(sparkScan: Scan): JHashMap[String, Integer] = {
+ val fieldIds = new JHashMap[String, Integer]()
+ sparkScan match {
+ case scan: SparkBatchQueryScan =>
+ scan.table().schema().columns().asScala.foreach {
+ field => fieldIds.put(field.name(), field.fieldId())
+ }
+ case _ =>
+ throw new GlutenNotSupportException("Only support iceberg
SparkBatchQueryScan.")
+ }
+ fieldIds
+ }
+
+ def getInitialDefaults(sparkScan: Scan): JHashMap[String, String] = {
+ val initialDefaults = new JHashMap[String, String]()
+ sparkScan match {
+ case scan: SparkBatchQueryScan =>
+ scan.table().schema().columns().asScala.foreach {
+ field =>
+ val defaultValue = IcebergDefaultValueUtil.getInitialDefault(field)
+ if (defaultValue != null) {
+ initialDefaults.put(
+ field.name(),
+ TypeUtil.getPartitionValueString(field.`type`(), defaultValue))
+ }
+ }
+ case _ =>
+ throw new GlutenNotSupportException("Only support iceberg
SparkBatchQueryScan.")
+ }
+ initialDefaults
+ }
+
private def genMetadataColumns(
metadataColumnNames: Seq[String],
filePath: String,
diff --git
a/gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeBoundsTest.java
b/gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeBoundsTest.java
index 01fc50035c..f4fff4d6c4 100644
---
a/gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeBoundsTest.java
+++
b/gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeBoundsTest.java
@@ -59,7 +59,9 @@ public class IcebergLocalFilesNodeBoundsTest {
LocalFilesNode.ReadFileFormat.ParquetReadFormat,
Collections.emptyList(),
Collections.singletonList(Collections.singletonList(deleteFile)),
- Collections.singletonList(Collections.emptyMap()));
+ Collections.singletonList(Collections.emptyMap()),
+ Collections.emptyMap(),
+ Collections.emptyMap());
ReadRel.LocalFiles.FileOrFiles.Builder fileBuilder =
ReadRel.LocalFiles.FileOrFiles.newBuilder();
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala
index 85b146e195..cf2026af63 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala
@@ -152,6 +152,8 @@ trait BackendSettingsApi {
def supportIcebergEqualityDeleteRead(): Boolean = true
+ def supportIcebergInitialDefaultRead(): Boolean = false
+
def reorderColumnsForPartitionWrite(): Boolean = false
def enableEnhancedFeatures(): Boolean = false
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/TransformerApi.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/TransformerApi.scala
index a855537664..c35a85398a 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/TransformerApi.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/TransformerApi.scala
@@ -74,6 +74,13 @@ trait TransformerApi {
def packPBMessage(message: Message): Any
+ /** Packs Iceberg column initial defaults into a backend-specific read
extension. */
+ def packIcebergReadExtension(
+ fieldIds: util.Map[String, Integer],
+ initialDefaults: util.Map[String, String]): Any = {
+ throw new UnsupportedOperationException("Iceberg initial-default reads are
not supported")
+ }
+
/** This method is only used for CH backend tests */
def invalidateSQLExecutionResource(executionId: String): Unit = {}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
