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 edb30dac61f..4e540c15af5 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 00000000000..f6580abae2b --- /dev/null +++ b/backends-velox/src/main/resources/org/apache/gluten/proto/IcebergReadExtension.proto @@ -0,0 +1,18 @@ +// 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 ColumnDefault { + string name = 1; + int32 field_id = 2; + string initial_default = 3; + } + + repeated ColumnDefault column_defaults = 1; +} 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 14150196817..77d3b015793 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 47ad778dc8a..9c150e40d7e 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,26 @@ 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() + 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 22a7c49ac37..82c1f290368 100644 --- a/cpp/velox/compute/VeloxPlanConverter.cc +++ b/cpp/velox/compute/VeloxPlanConverter.cc @@ -137,9 +137,10 @@ std::shared_ptr parseDeltaSplitInfo( std::shared_ptr parseScanSplitInfo( const facebook::velox::config::ConfigBase* veloxCfg, - const google::protobuf::RepeatedPtrField& fileList) { + const substrait::ReadRel_LocalFiles& localFiles) { using SubstraitFileFormatCase = ::substrait::ReadRel_LocalFiles_FileOrFiles::FileFormatCase; + const auto& fileList = localFiles.items(); auto splitInfo = std::make_shared(); splitInfo->leafType = SplitInfo::LeafType::TABLE_SCAN; splitInfo->paths.reserve(fileList.size()); @@ -193,7 +194,8 @@ std::shared_ptr 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> 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 07c40e6e1c7..68d50fafdd8 100644 --- a/cpp/velox/compute/iceberg/IcebergPlanConverter.cc +++ b/cpp/velox/compute/iceberg/IcebergPlanConverter.cc @@ -17,10 +17,13 @@ #include "IcebergPlanConverter.h" +#include "IcebergReadExtension.pb.h" + namespace gluten { std::shared_ptr IcebergPlanConverter::parseIcebergSplitInfo( substrait::ReadRel_LocalFiles_FileOrFiles file, + const substrait::extensions::AdvancedExtension& extension, std::shared_ptr splitInfo) { using SubstraitFileFormatCase = ::substrait::ReadRel_LocalFiles_FileOrFiles::IcebergReadOptions::FileFormatCase; using SubstraitDeleteFileFormatCase = @@ -40,6 +43,20 @@ std::shared_ptr 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_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()); + } + } + } if (icebergReadOption.delete_files_size() > 0) { auto deleteFiles = icebergReadOption.delete_files(); std::vector deletes; diff --git a/cpp/velox/compute/iceberg/IcebergPlanConverter.h b/cpp/velox/compute/iceberg/IcebergPlanConverter.h index d634a861fe3..af2631200ee 100644 --- a/cpp/velox/compute/iceberg/IcebergPlanConverter.h +++ b/cpp/velox/compute/iceberg/IcebergPlanConverter.h @@ -17,14 +17,24 @@ #pragma once +#include +#include +#include + #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 initialDefault; +}; + struct IcebergSplitInfo : SplitInfo { std::vector> deleteFilesVec; + std::unordered_map 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 parseIcebergSplitInfo( substrait::ReadRel_LocalFiles_FileOrFiles file, + const substrait::extensions::AdvancedExtension& extension, std::shared_ptr splitInfo); }; diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc b/cpp/velox/substrait/SubstraitToVeloxPlan.cc index fedd950374f..1d3caf31a1b 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 outNames; outNames.reserve(colNameList.size()); connector::ColumnHandleMap assignments; + const auto icebergSplitInfo = std::dynamic_pointer_cast(splitInfo); for (int idx = 0; idx < colNameList.size(); idx++) { auto outName = SubstraitParser::makeNodeName(planNodeId_, idx); auto columnType = columnTypes[idx]; - assignments[outName] = std::make_shared( - 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 the + // Iceberg handle, which expects epoch-day partition dates, only when its + // initial-default metadata is needed by Velox. + if (icebergColumn && icebergColumn->initialDefault.has_value()) { + assignments[outName] = std::make_shared( + colNameList[idx], + columnType, + veloxTypeList[idx], + facebook::velox::parquet::ParquetFieldId(icebergColumn->fieldId), + std::vector{}, + icebergColumn->initialDefault); + } else { + assignments[outName] = std::make_shared( + colNameList[idx], columnType, veloxTypeList[idx], veloxTypeList[idx]); + } outNames.emplace_back(outName); } auto outputType = ROW(std::move(outNames), std::move(veloxTypeList)); diff --git a/gluten-iceberg/src-iceberg10/main/java/org/apache/gluten/IcebergDefaultValueUtil.java b/gluten-iceberg/src-iceberg10/main/java/org/apache/gluten/IcebergDefaultValueUtil.java new file mode 100644 index 00000000000..0d50c905b88 --- /dev/null +++ b/gluten-iceberg/src-iceberg10/main/java/org/apache/gluten/IcebergDefaultValueUtil.java @@ -0,0 +1,27 @@ +/* + * 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. + */ +package org.apache.gluten; + +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/gluten-iceberg/src-iceberg3/main/java/org/apache/gluten/IcebergDefaultValueUtil.java b/gluten-iceberg/src-iceberg3/main/java/org/apache/gluten/IcebergDefaultValueUtil.java new file mode 100644 index 00000000000..fff8cb444f1 --- /dev/null +++ b/gluten-iceberg/src-iceberg3/main/java/org/apache/gluten/IcebergDefaultValueUtil.java @@ -0,0 +1,27 @@ +/* + * 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. + */ +package org.apache.gluten; + +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-iceberg5/main/java/org/apache/gluten/IcebergDefaultValueUtil.java b/gluten-iceberg/src-iceberg5/main/java/org/apache/gluten/IcebergDefaultValueUtil.java new file mode 100644 index 00000000000..fff8cb444f1 --- /dev/null +++ b/gluten-iceberg/src-iceberg5/main/java/org/apache/gluten/IcebergDefaultValueUtil.java @@ -0,0 +1,27 @@ +/* + * 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. + */ +package org.apache.gluten; + +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 eef98f77845..ca4597b48ef 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 static IcebergLocalFilesNode makeIcebergLocalFiles( LocalFilesNode.ReadFileFormat fileFormat, List preferredLocations, List> deleteFilesList, - List> metadataColumns) { + List> metadataColumns, + Map fieldIds, + Map initialDefaults) { return new IcebergLocalFilesNode( index, paths, @@ -41,6 +43,8 @@ public static IcebergLocalFilesNode makeIcebergLocalFiles( 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 cca3a4f79db..e5150a0b6d7 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; @@ -26,6 +30,8 @@ public class IcebergLocalFilesNode extends LocalFilesNode { private final List> deleteFilesList; + private final Map fieldIds; + private final Map initialDefaults; IcebergLocalFilesNode( Integer index, @@ -36,7 +42,9 @@ public class IcebergLocalFilesNode extends LocalFilesNode { ReadFileFormat fileFormat, List preferredLocations, List> deleteFilesList, - List> metadataColumns) { + List> metadataColumns, + Map fieldIds, + Map initialDefaults) { super( index, paths, @@ -51,6 +59,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 cc665ba4eb8..9317be9ae9c 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 @@ -150,6 +150,12 @@ case class IcebergScanTransformer( return ValidationResult.succeeded } val metadata = baseTable.operations().current() + if ( + !BackendsApiManager.getSettings.supportIcebergInitialDefaultRead() && + GlutenIcebergSourceUtil.hasInitialDefault(scan) + ) { + return ValidationResult.failed("Iceberg initial-default reads are not supported") + } if (metadata.formatVersion() >= 3) { val hasUnsupportedDelete = finalPartitions.exists { case p: SparkDataSourceRDDPartition => @@ -208,7 +214,7 @@ case class IcebergScanTransformer( metadataColumnNames: Seq[String]): SplitInfo = { val splitInfo = partition match { case p: SparkDataSourceRDDPartition => - GlutenIcebergSourceUtil.genSplitInfo(p, getPartitionSchema, metadataColumnNames) + GlutenIcebergSourceUtil.genSplitInfo(p, scan, getPartitionSchema, metadataColumnNames) 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 1a658a78925..5b9a09f2fb2 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 @@ -56,6 +57,7 @@ object GlutenIcebergSourceUtil { def genSplitInfo( partition: SparkDataSourceRDDPartition, + sparkScan: Scan, readPartitionSchema: StructType, metadataColumnNames: Seq[String]): SplitInfo = { val paths = new JArrayList[String]() @@ -103,10 +105,52 @@ object GlutenIcebergSourceUtil { .toList .asJava, deleteFilesList, - metadataColumns + metadataColumns, + getFieldIds(sparkScan), + getInitialDefaults(sparkScan) ) } + private 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 + } + + private 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 + } + + def hasInitialDefault(sparkScan: Scan): Boolean = sparkScan match { + case scan: SparkBatchQueryScan => + scan.table().schema().columns().asScala.exists { + field => IcebergDefaultValueUtil.getInitialDefault(field) != null + } + case _ => false + } + private def genMetadataColumns( metadataColumnNames: Seq[String], filePath: String, 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 85b146e1956..cf2026af630 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 a8555376644..c35a85398a5 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 = {}