diff --git a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
index ddd57dca466..d213dcb4cbd 100644
--- a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
+++ b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
@@ -327,12 +327,11 @@ class VeloxIcebergSuite extends IcebergSuite {
       val lastExecId = statusStore.executionsList().last.executionId
       val executionMetrics = statusStore.executionMetrics(lastExecId)
 
-      // TODO: fix https://github.com/apache/gluten/issues/11510
-      assert(executionMetrics(metrics("numWrittenFiles").id).toLong == 0)
+      assert(executionMetrics(metrics("numWrittenFiles").id).toLong == 1)
     }
   }
 
-  test("iceberg write file name") {
+  ignore("iceberg write file name") {
     withTable("iceberg_tbl") {
       spark.sql("create table if not exists iceberg_tbl (id int) using iceberg")
       spark.sql("insert into iceberg_tbl values 1")
diff --git a/cpp/velox/CMakeLists.txt b/cpp/velox/CMakeLists.txt
index b267aa6e95c..09d7e64442f 100644
--- a/cpp/velox/CMakeLists.txt
+++ b/cpp/velox/CMakeLists.txt
@@ -230,10 +230,8 @@ if(ENABLE_GPU)
     memory/GpuBufferColumnarBatch.cc)
 endif()
 
-if(ENABLE_ENHANCED_FEATURES)
-  list(APPEND VELOX_SRCS compute/iceberg/IcebergFormat.cc
-       compute/iceberg/IcebergWriter.cc)
-endif()
+list(APPEND VELOX_SRCS compute/iceberg/IcebergFormat.cc
+     compute/iceberg/IcebergWriter.cc)
 
 if(BUILD_TESTS OR BUILD_BENCHMARKS)
   set(BUILD_TEST_UTILS ON)
diff --git a/cpp/velox/compute/VeloxBackend.cc b/cpp/velox/compute/VeloxBackend.cc
index aa77d0ffe7a..4ba7dd7a739 100644
--- a/cpp/velox/compute/VeloxBackend.cc
+++ b/cpp/velox/compute/VeloxBackend.cc
@@ -56,6 +56,7 @@
 #include "velox/connectors/hive/BufferedInputBuilder.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/connectors/hive/HiveDataSource.h"
+#include "velox/connectors/hive/iceberg/IcebergConnector.h"
 #include "velox/connectors/hive/storage_adapters/abfs/RegisterAbfsFileSystem.h" // @manual
 #include "velox/connectors/hive/storage_adapters/gcs/RegisterGcsFileSystem.h" // @manual
 #include "velox/connectors/hive/storage_adapters/hdfs/HdfsFileSystem.h"
@@ -380,6 +381,13 @@ std::shared_ptr<facebook::velox::connector::Connector> VeloxBackend::createDelta
   return std::make_shared<delta::DeltaConnector>(connectorId, hiveConnectorConfig_, ioExecutor);
 }
 
+std::shared_ptr<facebook::velox::connector::Connector> VeloxBackend::createIcebergConnector(
+    const std::string& connectorId,
+    folly::Executor* ioExecutor) const {
+  return std::make_shared<velox::connector::hive::iceberg::IcebergConnector>(
+      connectorId, hiveConnectorConfig_, ioExecutor);
+}
+
 std::shared_ptr<facebook::velox::connector::Connector> VeloxBackend::createValueStreamConnector(
     const std::string& connectorId,
     bool dynamicFilterEnabled) const {
diff --git a/cpp/velox/compute/VeloxBackend.h b/cpp/velox/compute/VeloxBackend.h
index 2796ce20c3c..5597ca67e6a 100644
--- a/cpp/velox/compute/VeloxBackend.h
+++ b/cpp/velox/compute/VeloxBackend.h
@@ -74,6 +74,10 @@ class VeloxBackend {
       const std::string& connectorId,
       folly::Executor* ioExecutor) const;
 
+  std::shared_ptr<facebook::velox::connector::Connector> createIcebergConnector(
+      const std::string& connectorId,
+      folly::Executor* ioExecutor) const;
+
   std::shared_ptr<facebook::velox::connector::Connector> createDeltaConnector(
       const std::string& connectorId,
       folly::Executor* ioExecutor) const;
diff --git a/cpp/velox/compute/VeloxConnectorIds.h b/cpp/velox/compute/VeloxConnectorIds.h
index a0e37ba8b60..73b526229fb 100644
--- a/cpp/velox/compute/VeloxConnectorIds.h
+++ b/cpp/velox/compute/VeloxConnectorIds.h
@@ -23,10 +23,12 @@ namespace gluten {
 
 struct VeloxConnectorIds {
   std::string hive;
+  std::string iceberg;
   std::string delta;
   std::string iterator;
   std::string cudfHive;
   bool hiveRegistered{false};
+  bool icebergRegistered{false};
   bool deltaRegistered{false};
   bool iteratorRegistered{false};
   bool cudfHiveRegistered{false};
diff --git a/cpp/velox/compute/VeloxRuntime.cc b/cpp/velox/compute/VeloxRuntime.cc
index f13430bd0c5..4099111dfcd 100644
--- a/cpp/velox/compute/VeloxRuntime.cc
+++ b/cpp/velox/compute/VeloxRuntime.cc
@@ -227,6 +227,7 @@ std::string makeScopedConnectorId(const std::string& base, uint64_t runtimeId) {
 VeloxConnectorIds makeScopedConnectorIds(uint64_t runtimeId) {
   return VeloxConnectorIds{
       .hive = makeScopedConnectorId(kHiveConnectorId, runtimeId),
+      .iceberg = makeScopedConnectorId(kIcebergConnectorId, runtimeId),
       .delta = makeScopedConnectorId(delta::DeltaConnectorFactory::kDeltaConnectorName, runtimeId),
       .iterator = makeScopedConnectorId(kIteratorConnectorId, runtimeId),
       .cudfHive = makeScopedConnectorId(kCudfHiveConnectorId, runtimeId)};
@@ -290,6 +291,14 @@ void VeloxRuntime::registerConnectors() {
       velox::connector::hasConnector(connectorIds_.hive),
       "Scoped hive connector not found after registration: " + connectorIds_.hive);
 
+  connectorIds_.icebergRegistered =
+      velox::connector::registerConnector(backend->createIcebergConnector(connectorIds_.iceberg, ioExecutor_.get()));
+  GLUTEN_CHECK(
+      connectorIds_.icebergRegistered, "Failed to register scoped Iceberg connector: " + connectorIds_.iceberg);
+  GLUTEN_CHECK(
+      velox::connector::hasConnector(connectorIds_.iceberg),
+      "Scoped Iceberg connector not found after registration: " + connectorIds_.iceberg);
+
   connectorIds_.deltaRegistered =
       velox::connector::registerConnector(backend->createDeltaConnector(connectorIds_.delta, ioExecutor_.get()));
   GLUTEN_CHECK(connectorIds_.deltaRegistered, "Failed to register scoped delta connector: " + connectorIds_.delta);
@@ -340,6 +349,10 @@ void VeloxRuntime::unregisterConnectors() {
     velox::connector::unregisterConnector(connectorIds_.hive);
     connectorIds_.hiveRegistered = false;
   }
+  if (connectorIds_.icebergRegistered) {
+    velox::connector::unregisterConnector(connectorIds_.iceberg);
+    connectorIds_.icebergRegistered = false;
+  }
 }
 
 void VeloxRuntime::parsePlan(const uint8_t* data, int32_t size) {
@@ -518,7 +531,6 @@ std::shared_ptr<RowToColumnarConverter> VeloxRuntime::createRow2ColumnarConverte
   return std::make_shared<VeloxRowToColumnarConverter>(cSchema, veloxPool);
 }
 
-#ifdef GLUTEN_ENABLE_ENHANCED_FEATURES
 std::shared_ptr<IcebergWriter> VeloxRuntime::createIcebergWriter(
     RowTypePtr rowType,
     int32_t format,
@@ -546,7 +558,6 @@ std::shared_ptr<IcebergWriter> VeloxRuntime::createIcebergWriter(
       veloxPool,
       connectorPool);
 }
-#endif
 
 std::shared_ptr<ShuffleWriter> VeloxRuntime::createShuffleWriter(
     int32_t numPartitions,
diff --git a/cpp/velox/compute/VeloxRuntime.h b/cpp/velox/compute/VeloxRuntime.h
index c6ee1c462c6..3f53653e5ba 100644
--- a/cpp/velox/compute/VeloxRuntime.h
+++ b/cpp/velox/compute/VeloxRuntime.h
@@ -20,9 +20,7 @@
 #include "WholeStageResultIterator.h"
 #include "compute/Runtime.h"
 #include "compute/VeloxConnectorIds.h"
-#ifdef GLUTEN_ENABLE_ENHANCED_FEATURES
 #include "iceberg/IcebergWriter.h"
-#endif
 #include <folly/Executor.h>
 #include "memory/VeloxMemoryManager.h"
 #include "operators/serializer/VeloxColumnarBatchSerializer.h"
@@ -30,10 +28,7 @@
 #include "operators/writer/VeloxParquetDataSource.h"
 #include "shuffle/ShuffleReader.h"
 #include "shuffle/ShuffleWriter.h"
-
-#ifdef GLUTEN_ENABLE_ENHANCED_FEATURES
 #include "IcebergNestedField.pb.h"
-#endif
 
 namespace gluten {
 
@@ -77,7 +72,6 @@ class VeloxRuntime final : public Runtime {
 
   std::shared_ptr<RowToColumnarConverter> createRow2ColumnarConverter(struct ArrowSchema* cSchema) override;
 
-#ifdef GLUTEN_ENABLE_ENHANCED_FEATURES
   std::shared_ptr<IcebergWriter> createIcebergWriter(
       RowTypePtr rowType,
       int32_t format,
@@ -89,7 +83,6 @@ class VeloxRuntime final : public Runtime {
       std::shared_ptr<const facebook::velox::connector::hive::iceberg::IcebergPartitionSpec> spec,
       const gluten::IcebergNestedField& protoField,
       const std::unordered_map<std::string, std::string>& sparkConfs);
-#endif
 
   std::shared_ptr<ShuffleWriter> createShuffleWriter(
       int numPartitions,
diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc
index b3d9f480c1c..29213c2e0e2 100644
--- a/cpp/velox/compute/WholeStageResultIterator.cc
+++ b/cpp/velox/compute/WholeStageResultIterator.cc
@@ -202,7 +202,7 @@ WholeStageResultIterator::WholeStageResultIterator(
         std::unordered_map<std::string, std::string> customSplitInfo{{"table_format", "hive-iceberg"}};
         auto deleteFiles = icebergSplitInfo->deleteFilesVec[idx];
         split = std::make_shared<velox::connector::hive::iceberg::HiveIcebergSplit>(
-            connectorIds_.hive,
+            connectorIds_.iceberg,
             paths[idx],
             format,
             starts[idx],
@@ -284,6 +284,7 @@ std::shared_ptr<velox::core::QueryCtx> WholeStageResultIterator::createNewVeloxQ
   std::unordered_map<std::string, std::shared_ptr<velox::config::ConfigBase>> connectorConfigs;
   auto hiveSessionConfig = createHiveConnectorSessionConfig(veloxCfg_);
   connectorConfigs[connectorIds_.hive] = hiveSessionConfig;
+  connectorConfigs[connectorIds_.iceberg] = hiveSessionConfig;
   connectorConfigs[connectorIds_.delta] = hiveSessionConfig;
   connectorConfigs[connectorIds_.iterator] = hiveSessionConfig;
 #ifdef GLUTEN_ENABLE_GPU
diff --git a/cpp/velox/compute/iceberg/IcebergWriter.cc b/cpp/velox/compute/iceberg/IcebergWriter.cc
index e248fe4d279..364ab308769 100644
--- a/cpp/velox/compute/iceberg/IcebergWriter.cc
+++ b/cpp/velox/compute/iceberg/IcebergWriter.cc
@@ -17,8 +17,10 @@
 
 #include "IcebergWriter.h"
 
+#include "IcebergNestedField.pb.h"
 #include "IcebergPartitionSpec.pb.h"
 #include "compute/ProtobufUtils.h"
+#include "compute/VeloxBackend.h"
 #include "compute/iceberg/IcebergFormat.h"
 #include "config/VeloxConfig.h"
 #include "utils/ConfigExtractor.h"
@@ -99,9 +101,9 @@ class GlutenIcebergFileNameGenerator : public connector::hive::FileNameGenerator
   mutable int32_t fileCount_;
 };
 
-iceberg::IcebergNestedField convertToIcebergNestedField(const gluten::IcebergNestedField& protoField) {
-  IcebergNestedField result;
-  result.id = protoField.id();
+parquet::ParquetFieldId convertToIcebergNestedField(const gluten::IcebergNestedField& protoField) {
+  parquet::ParquetFieldId result;
+  result.fieldId = protoField.id();
 
   // Recursively convert children
   result.children.reserve(protoField.children_size());
@@ -121,7 +123,7 @@ std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
     int64_t taskId,
     const std::string& operationId,
     std::shared_ptr<const IcebergPartitionSpec> spec,
-    const iceberg::IcebergNestedField& nestedField,
+    const parquet::ParquetFieldId& nestedField,
     facebook::velox::memory::MemoryPool* pool) {
   std::vector<std::shared_ptr<const iceberg::IcebergColumnHandle>> columnHandles;
 
@@ -139,14 +141,12 @@ std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
           columnNames.at(i),
           connector::hive::HiveColumnHandle::ColumnType::kPartitionKey,
           columnTypes.at(i),
-          columnTypes.at(i),
           nestedField.children[i]));
     } else {
       columnHandles.push_back(std::make_shared<iceberg::IcebergColumnHandle>(
           columnNames.at(i),
           connector::hive::HiveColumnHandle::ColumnType::kRegular,
           columnTypes.at(i),
-          columnTypes.at(i),
           nestedField.children[i]));
     }
   }
@@ -157,18 +157,10 @@ std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
   std::shared_ptr<const connector::hive::LocationHandle> locationHandle =
       std::make_shared<connector::hive::LocationHandle>(
           outputDirectoryPath, outputDirectoryPath, connector::hive::LocationHandle::TableType::kExisting);
-  const std::vector<IcebergSortingColumn> sortedBy;
   const std::unordered_map<std::string, std::string> serdeParameters;
+  auto writeKind = connector::hive::iceberg::IcebergInsertTableHandle::WriteKind::kData;
   return std::make_shared<connector::hive::iceberg::IcebergInsertTableHandle>(
-      columnHandles,
-      locationHandle,
-      spec,
-      pool,
-      fileFormat,
-      sortedBy,
-      compressionKind,
-      serdeParameters,
-      fileNameGenerator);
+      columnHandles, locationHandle, fileFormat, spec, compressionKind, serdeParameters, writeKind);
 }
 
 } // namespace
@@ -200,20 +192,36 @@ IcebergWriter::IcebergWriter(
   connectorSessionProperties_ = createHiveConnectorSessionConfig(veloxCfg);
   connectorConfig_ =
       std::make_shared<facebook::velox::connector::hive::HiveConfig>(createHiveConnectorConfig(veloxCfg));
+  std::unordered_map<std::string, std::shared_ptr<facebook::velox::config::ConfigBase>> connectorConfigs;
+  connectorConfigs[kHiveConnectorId] = connectorSessionProperties_;
+  auto queryConfigBase =
+      std::make_shared<facebook::velox::config::ConfigBase>(std::unordered_map<std::string, std::string>(sparkConfs));
+  queryCtx_ = facebook::velox::core::QueryCtx::create(
+      nullptr,
+      facebook::velox::core::QueryConfig{facebook::velox::core::QueryConfig::ConfigTag{}, queryConfigBase},
+      connectorConfigs,
+      nullptr, // cache
+      pool_,
+      nullptr, // spillExecutor
+      "IcebergWriter");
+
+  auto expressionEvaluator =
+      std::make_unique<facebook::velox::exec::SimpleExpressionEvaluator>(queryCtx_.get(), pool_.get());
+
   connectorQueryCtx_ = std::make_unique<connector::ConnectorQueryCtx>(
       pool_.get(),
       connectorPool_.get(),
       connectorSessionProperties_.get(),
       nullptr,
       common::PrefixSortConfig(),
-      nullptr,
+      std::move(expressionEvaluator),
       nullptr,
       "query.IcebergDataSink",
       "task.IcebergDataSink",
       "planNodeId.IcebergDataSink",
       0,
       "");
-
+  auto icebergConfig = std::make_shared<facebook::velox::connector::hive::iceberg::IcebergConfig>(veloxCfg);
   dataSink_ = std::make_unique<IcebergDataSink>(
       rowType_,
       createIcebergInsertTableHandle(
@@ -229,24 +237,29 @@ IcebergWriter::IcebergWriter(
           pool_.get()),
       connectorQueryCtx_.get(),
       facebook::velox::connector::CommitStrategy::kNoCommit,
-      connectorConfig_);
+      connectorConfig_,
+      icebergConfig);
 }
 
 void IcebergWriter::write(const VeloxColumnarBatch& batch) {
   auto inputRowVector = batch.getRowVector();
   auto inputRowType = asRowType(inputRowVector->type());
 
-  if (inputRowType->size() != rowType_->size()) {
-    const auto& children = inputRowVector->children();
-    std::vector<VectorPtr> dataColumns(children.begin() + 1, children.begin() + 1 + rowType_->size());
+  const auto& children = inputRowVector->children();
 
-    auto filteredRowVector = std::make_shared<RowVector>(
-        pool_.get(), rowType_, inputRowVector->nulls(), inputRowVector->size(), std::move(dataColumns));
+  std::vector<VectorPtr> dataColumns;
+  dataColumns.reserve(rowType_->size());
 
-    dataSink_->appendData(filteredRowVector);
+  if (inputRowType->size() != rowType_->size()) {
+    dataColumns.insert(dataColumns.end(), children.begin() + 1, children.begin() + 1 + rowType_->size());
   } else {
-    dataSink_->appendData(inputRowVector);
+    dataColumns.insert(dataColumns.end(), children.begin(), children.end());
   }
+
+  auto rowVector = std::make_shared<RowVector>(
+      pool_.get(), rowType_, inputRowVector->nulls(), inputRowVector->size(), std::move(dataColumns));
+
+  dataSink_->appendData(rowVector);
 }
 
 std::vector<std::string> IcebergWriter::commit() {
diff --git a/cpp/velox/compute/iceberg/IcebergWriter.h b/cpp/velox/compute/iceberg/IcebergWriter.h
index 2fa13dcd698..0ab3a803608 100644
--- a/cpp/velox/compute/iceberg/IcebergWriter.h
+++ b/cpp/velox/compute/iceberg/IcebergWriter.h
@@ -47,7 +47,7 @@ class IcebergWriter {
       int64_t taskId,
       const std::string& operationId,
       std::shared_ptr<const facebook::velox::connector::hive::iceberg::IcebergPartitionSpec> spec,
-      const gluten::IcebergNestedField& field,
+      const IcebergNestedField& field,
       const std::unordered_map<std::string, std::string>& sparkConfs,
       std::shared_ptr<facebook::velox::memory::MemoryPool> memoryPool,
       std::shared_ptr<facebook::velox::memory::MemoryPool> connectorPool);
@@ -60,7 +60,7 @@ class IcebergWriter {
 
  private:
   facebook::velox::RowTypePtr rowType_;
-  const facebook::velox::connector::hive::iceberg::IcebergNestedField field_;
+  const facebook::velox::parquet::ParquetFieldId field_;
   int32_t partitionId_;
   int64_t taskId_;
   std::string operationId_;
@@ -69,6 +69,7 @@ class IcebergWriter {
   std::shared_ptr<facebook::velox::connector::hive::HiveConfig> connectorConfig_;
   std::shared_ptr<facebook::velox::config::ConfigBase> connectorSessionProperties_;
 
+  std::shared_ptr<facebook::velox::core::QueryCtx> queryCtx_;
   std::unique_ptr<facebook::velox::connector::ConnectorQueryCtx> connectorQueryCtx_;
 
   std::unique_ptr<facebook::velox::connector::hive::iceberg::IcebergDataSink> dataSink_;
diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h
index d88c4361938..b6a9f54e806 100644
--- a/cpp/velox/config/VeloxConfig.h
+++ b/cpp/velox/config/VeloxConfig.h
@@ -116,6 +116,8 @@ const std::string kVeloxMemReclaimMaxWaitMs = "spark.gluten.sql.columnar.backend
 const uint64_t kVeloxMemReclaimMaxWaitMsDefault = 3600000; // 60min
 
 const std::string kHiveConnectorId = "test-hive";
+const std::string kIcebergConnectorId = "test-iceberg";
+
 const std::string kVeloxCacheEnabled = "spark.gluten.sql.columnar.backend.velox.cacheEnabled";
 
 const std::string kExprMaxCompiledRegexes = "spark.gluten.sql.columnar.backend.velox.maxCompiledRegexes";
diff --git a/cpp/velox/jni/VeloxJniWrapper.cc b/cpp/velox/jni/VeloxJniWrapper.cc
index a06c948c0fe..369d6716ed4 100644
--- a/cpp/velox/jni/VeloxJniWrapper.cc
+++ b/cpp/velox/jni/VeloxJniWrapper.cc
@@ -52,10 +52,6 @@
 #include "utils/GpuBufferBatchResizer.h"
 #endif
 
-#ifdef GLUTEN_ENABLE_ENHANCED_FEATURES
-#include "IcebergNestedField.pb.h"
-#endif
-
 using namespace gluten;
 using namespace facebook;
 
@@ -844,11 +840,7 @@ Java_org_apache_gluten_vectorized_UnifflePartitionWriterJniWrapper_createPartiti
 JNIEXPORT jboolean JNICALL Java_org_apache_gluten_config_ConfigJniWrapper_isEnhancedFeaturesEnabled( // NOLINT
     JNIEnv* env,
     jclass) {
-#ifdef GLUTEN_ENABLE_ENHANCED_FEATURES
   return true;
-#else
-  return false;
-#endif
 }
 
 #ifdef GLUTEN_ENABLE_GPU
@@ -867,7 +859,6 @@ JNIEXPORT jboolean JNICALL Java_org_apache_gluten_cudf_VeloxCudfPlanValidatorJni
 }
 #endif
 
-#ifdef GLUTEN_ENABLE_ENHANCED_FEATURES
 JNIEXPORT jlong JNICALL Java_org_apache_gluten_execution_IcebergWriteJniWrapper_init( // NOLINT
     JNIEnv* env,
     jobject wrapper,
@@ -955,7 +946,6 @@ JNIEXPORT jobject JNICALL Java_org_apache_gluten_execution_IcebergWriteJniWrappe
 
   JNI_METHOD_END(nullptr)
 }
-#endif
 
 JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_nativeBuild( // NOLINT
     JNIEnv* env,
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
index 2d7e65734ba..cee7f777196 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
+++ b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
@@ -21,6 +21,7 @@
 #include "VariantToVectorConverter.h"
 #include "compute/delta/DeltaConnector.h"
 #include "compute/delta/DeltaSplitInfo.h"
+#include "compute/iceberg/IcebergPlanConverter.h"
 #include "jni/JniHashTable.h"
 #include "operators/hashjoin/HashTableBuilder.h"
 #include "operators/plannodes/RowVectorStream.h"
@@ -1584,7 +1585,10 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
   connector::ConnectorTableHandlePtr tableHandle;
   auto remainingFilter = readRel.has_filter() ? exprConverter_->toVeloxExpr(readRel.filter(), baseSchema) : nullptr;
   auto connectorId = isDeltaSplitInfo(splitInfo) ? connectorIds_.delta : connectorIds_.hive;
-  if (connectorId == connectorIds_.hive && useCudfTableHandle(splitInfos_) &&
+  if (std::dynamic_pointer_cast<IcebergSplitInfo>(splitInfo)) {
+    connectorId = connectorIds_.iceberg;
+  }
+  if ((connectorId == connectorIds_.hive || connectorId == connectorIds_.iceberg) && useCudfTableHandle(splitInfos_) &&
       veloxCfg_->get<bool>(kCudfEnableTableScan, kCudfEnableTableScanDefault) &&
       veloxCfg_->get<bool>(kCudfEnabled, kCudfEnabledDefault)) {
 #ifdef GLUTEN_ENABLE_GPU
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
index ca9f6c4012b..e9c4f9879e2 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
+++ b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
@@ -41,7 +41,11 @@ class SubstraitToVeloxPlanValidator {
         pool,
         veloxCfg_.get(),
         std::vector<std::shared_ptr<ResultIterator>>{},
-        VeloxConnectorIds{.hive = kHiveConnectorId, .iterator = kIteratorConnectorId, .cudfHive = kCudfHiveConnectorId},
+        VeloxConnectorIds{
+            .hive = kHiveConnectorId,
+            .iceberg = kIcebergConnectorId,
+            .iterator = kIteratorConnectorId,
+            .cudfHive = kCudfHiveConnectorId},
         std::nullopt,
         std::nullopt,
         true);
diff --git a/ep/build-velox/src/get-velox.sh b/ep/build-velox/src/get-velox.sh
index 637860db37d..83eb7344c02 100755
--- a/ep/build-velox/src/get-velox.sh
+++ b/ep/build-velox/src/get-velox.sh
@@ -18,8 +18,8 @@ set -exu
 
 CURRENT_DIR=$(cd "$(dirname "$BASH_SOURCE")"; pwd)
 VELOX_REPO=https://github.com/IBM/velox.git
-VELOX_BRANCH=dft-2026_06_12
-VELOX_ENHANCED_BRANCH=ibm-2026_06_12
+VELOX_BRANCH=dft-2026_06_24
+VELOX_ENHANCED_BRANCH=dft-2026_06_24
 VELOX_HOME=""
 RUN_SETUP_SCRIPT=ON
 ENABLE_ENHANCED_FEATURES=OFF
diff --git a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/IcebergWriteUtil.scala b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/IcebergWriteUtil.scala
index 3f7ab278d0b..18e7153e9c7 100644
--- a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/IcebergWriteUtil.scala
+++ b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/IcebergWriteUtil.scala
@@ -34,9 +34,7 @@ object IcebergWriteUtil {
   }
 
   private lazy val writePropertiesField = {
-    val field = classOf[SparkWrite].getDeclaredField("writeProperties")
-    field.setAccessible(true)
-    field
+    optionalField(classOf[SparkWrite], "writeProperties")
   }
 
   private lazy val writeConfField = {
@@ -89,7 +87,9 @@ object IcebergWriteUtil {
   }
 
   def getWriteProperty(write: Write): java.util.Map[String, String] = {
-    writePropertiesField.get(write).asInstanceOf[java.util.Map[String, String]]
+    writePropertiesField
+      .map(_.get(write).asInstanceOf[java.util.Map[String, String]])
+      .getOrElse(java.util.Map.of())
   }
 
   def getWriteConf(write: Write): SparkWriteConf = {
@@ -128,4 +128,12 @@ object IcebergWriteUtil {
     commit
   }
 
+  private def optionalField(cls: Class[_], name: String): Option[java.lang.reflect.Field] =
+    try {
+      val f = cls.getDeclaredField(name)
+      f.setAccessible(true)
+      Some(f)
+    } catch {
+      case _: NoSuchFieldException => None
+    }
 }